diff --git a/core/lib/Drupal/Core/Cache/DatabaseBackend.php b/core/lib/Drupal/Core/Cache/DatabaseBackend.php
index e0fa59e..cd51e91 100644
--- a/core/lib/Drupal/Core/Cache/DatabaseBackend.php
+++ b/core/lib/Drupal/Core/Cache/DatabaseBackend.php
@@ -74,7 +74,7 @@ public function getMultiple(&$cids, $allow_invalid = FALSE) {
     // is used here only due to the performance overhead we would incur
     // otherwise. When serving an uncached page, the overhead of using
     // ::select() is a much smaller proportion of the request.
-    $result = $this->connection->query('SELECT cid, data, created, expire, serialized, tags, checksum_invalidations, checksum_deletions FROM {' . $this->connection->escapeTable($this->bin) . '} WHERE cid IN (:cids)', array(':cids' => array_keys($cid_mapping)));
+    $result = $this->connection->query('SELECT cid, data, created, expire, serialized, tags, checksum_invalidations, checksum_deletions FROM {' . $this->connection->escapeTable($this->bin) . '} WHERE cid IN (:cids)', array(':cids' => array_keys($cid_mapping)), array('schema_provider' => $this));
     $cache = array();
     foreach ($result as $item) {
       // Map the cache ID back to the original.
@@ -176,7 +176,7 @@ public function set($cid, $data, $expire = Cache::PERMANENT, array $tags = array
       $fields['serialized'] = 0;
     }
 
-    $this->connection->merge($this->bin)
+    $this->connection->merge($this->bin, array('schema_provider' => $this))
       ->key('cid', $this->normalizeCid($cid))
       ->fields($fields)
       ->execute();
@@ -199,7 +199,7 @@ public function setMultiple(array $items) {
       $this->deleteMultiple(array_keys($items));
 
       $query = $this->connection
-        ->insert($this->bin)
+        ->insert($this->bin, array('schema_provider' => $this))
         ->fields(array('cid', 'data', 'expire', 'created', 'serialized', 'tags', 'checksum_invalidations', 'checksum_deletions'));
 
       foreach ($items as $cid => $item) {
@@ -271,7 +271,7 @@ public function deleteMultiple(array $cids) {
     $cids = array_values(array_map(array($this, 'normalizeCid'), $cids));
     // Delete in chunks when a large array is passed.
     foreach (array_chunk($cids, 1000) as $cids_chunk) {
-      $this->connection->delete($this->bin)
+      $this->connection->delete($this->bin, array('schema_provider' => $this))
         ->condition('cid', $cids_chunk, 'IN')
         ->execute();
     }
@@ -290,7 +290,7 @@ public function deleteTags(array $tags) {
       }
       $deleted_tags[$tag] = TRUE;
       unset($tag_cache[$tag]);
-      $this->connection->merge('cachetags')
+      $this->connection->merge('cachetags', array('schema_provider' => $this))
         ->insertFields(array('deletions' => 1))
         ->expression('deletions', 'deletions + 1')
         ->key('tag', $tag)
@@ -302,7 +302,7 @@ public function deleteTags(array $tags) {
    * Implements Drupal\Core\Cache\CacheBackendInterface::deleteAll().
    */
   public function deleteAll() {
-    $this->connection->truncate($this->bin)->execute();
+    $this->connection->truncate($this->bin, array('schema_provider' => $this))->execute();
   }
 
   /**
@@ -319,7 +319,7 @@ public function invalidateMultiple(array $cids) {
     $cids = array_values(array_map(array($this, 'normalizeCid'), $cids));
     // Update in chunks when a large array is passed.
     foreach (array_chunk($cids, 1000) as $cids_chunk) {
-      $this->connection->update($this->bin)
+      $this->connection->update($this->bin, array('schema_provider' => $this))
         ->fields(array('expire' => REQUEST_TIME - 1))
         ->condition('cid', $cids_chunk, 'IN')
         ->execute();
@@ -339,7 +339,7 @@ public function invalidateTags(array $tags) {
       }
       $invalidated_tags[$tag] = TRUE;
       unset($tag_cache[$tag]);
-      $this->connection->merge('cachetags')
+      $this->connection->merge('cachetags', array('schema_provider' => $this))
         ->insertFields(array('invalidations' => 1))
         ->expression('invalidations', 'invalidations + 1')
         ->key('tag', $tag)
@@ -351,7 +351,7 @@ public function invalidateTags(array $tags) {
    * Implements Drupal\Core\Cache\CacheBackendInterface::invalidateAll().
    */
   public function invalidateAll() {
-    $this->connection->update($this->bin)
+    $this->connection->update($this->bin, array('schema_provider' => $this))
       ->fields(array('expire' => REQUEST_TIME - 1))
       ->execute();
   }
@@ -392,7 +392,7 @@ protected function checksumTags(array $tags) {
 
     $query_tags = array_diff($tags, array_keys($tag_cache));
     if ($query_tags) {
-      $db_tags = $this->connection->query('SELECT tag, invalidations, deletions FROM {cachetags} WHERE tag IN (:tags)', array(':tags' => $query_tags))->fetchAllAssoc('tag', \PDO::FETCH_ASSOC);
+      $db_tags = $this->connection->query('SELECT tag, invalidations, deletions FROM {cachetags} WHERE tag IN (:tags)', array(':tags' => $query_tags), array('schema_provider' => $this))->fetchAllAssoc('tag', \PDO::FETCH_ASSOC);
       $tag_cache += $db_tags;
 
       // Fill static cache with empty objects for tags not found in the database.
diff --git a/core/lib/Drupal/Core/Config/DatabaseStorage.php b/core/lib/Drupal/Core/Config/DatabaseStorage.php
index 211d62d..bd115fe 100644
--- a/core/lib/Drupal/Core/Config/DatabaseStorage.php
+++ b/core/lib/Drupal/Core/Config/DatabaseStorage.php
@@ -63,7 +63,7 @@ class DatabaseStorage implements StorageInterface, SchemaProviderInterface {
   public function __construct(Connection $connection, $table, array $options = array(), $collection = StorageInterface::DEFAULT_COLLECTION) {
     $this->connection = $connection;
     $this->table = $table;
-    $this->options = $options;
+    $this->options = $options + ['schema_provider' => $this];
     $this->collection = $collection;
   }
 
@@ -204,7 +204,7 @@ public function decode($raw) {
    */
   public function listAll($prefix = '') {
     try {
-      $query = $this->connection->select($this->table);
+      $query = $this->connection->select($this->table, NULL, $this->options);
       $query->fields($this->table, array('name'));
       $query->condition('collection', $this->collection, '=');
       $query->condition('name', $prefix . '%', 'LIKE');
@@ -260,7 +260,7 @@ public function getAllCollectionNames() {
     try {
       return $this->connection->query('SELECT DISTINCT collection FROM {' . $this->connection->escapeTable($this->table) . '} WHERE collection <> :collection ORDER by collection', array(
         ':collection' => StorageInterface::DEFAULT_COLLECTION)
-      )->fetchCol();
+      , array(), $this->options)->fetchCol();
     }
     catch (\Exception $e) {
       return array();
diff --git a/core/lib/Drupal/Core/Database/Connection.php b/core/lib/Drupal/Core/Database/Connection.php
index b39a836..d5b56cd 100644
--- a/core/lib/Drupal/Core/Database/Connection.php
+++ b/core/lib/Drupal/Core/Database/Connection.php
@@ -603,17 +603,17 @@ protected function handleMissingTable($e, $query, $args, $options) {
         // early instead.
         return new StatementEmpty();
       }
-      if (is_string($options['create_missing_table'])) {
+      if (is_string($options['create_missing_table']) && isset($options['schema_provider'])) {
         $table = $options['create_missing_table'];
-        foreach (debug_backtrace(DEBUG_BACKTRACE_PROVIDE_OBJECT) as $bt) {
-          if (isset($bt['object']) && $bt['object'] instanceof SchemaProviderInterface && ($schema = $bt['object']->getSchema($table))) {
-            if ($this->schema()->ensureTableExists($table, $schema)) {
-              // Theoretically the table must exist at this point so this
-              // should not matter but still, databases. This makes sure
-              // no infinite recursion happens.
-              $options['create_missing_table'] = FALSE;
-              return $this->query($query, $args, $options);
-            }
+        /** @var \Drupal\Core\Database\SchemaProviderInterface $schema_provider */
+        if ($schema_provider = $options['schema_provider']) {
+          if ($this->schema()->ensureTableExists($table, $schema_provider->getSchema($table))) {
+            // Theoretically the table must exist at this point so this
+            // should not matter but still, databases. This makes sure
+            // no infinite recursion happens.
+            $options['create_missing_table'] = FALSE;
+            unset($options['schema_provider']);
+            return $this->query($query, $args, $options);
           }
         }
       }
diff --git a/core/lib/Drupal/Core/Database/Query/Merge.php b/core/lib/Drupal/Core/Database/Query/Merge.php
index cbff7aa..447f1d0 100644
--- a/core/lib/Drupal/Core/Database/Query/Merge.php
+++ b/core/lib/Drupal/Core/Database/Query/Merge.php
@@ -440,12 +440,13 @@ public function execute() {
       if (!count($this->condition)) {
         throw new InvalidMergeQueryException(t('Invalid merge query: no conditions'));
       }
-      $select = $this->connection->select($this->conditionTable)
+      $options = isset($this->queryOptions['schema_provider']) ? array('schema_provider' => $this->queryOptions['schema_provider']) : array();
+      $select = $this->connection->select($this->conditionTable, NULL, $options)
         ->condition($this->condition);
       $select->addExpression('1');
       if (!$select->execute()->fetchField()) {
         try {
-          $insert = $this->connection->insert($this->table)->fields($this->insertFields);
+          $insert = $this->connection->insert($this->table, $options)->fields($this->insertFields);
           if ($this->defaultFields) {
             $insert->useDefaults($this->defaultFields);
           }
@@ -463,7 +464,7 @@ public function execute() {
         }
       }
       if ($this->needsUpdate) {
-        $update = $this->connection->update($this->table)
+        $update = $this->connection->update($this->table, $options)
           ->fields($this->updateFields)
           ->condition($this->condition);
         if ($this->expressionFields) {
diff --git a/core/lib/Drupal/Core/Menu/MenuTreeStorage.php b/core/lib/Drupal/Core/Menu/MenuTreeStorage.php
index f8735ad..6558022 100644
--- a/core/lib/Drupal/Core/Menu/MenuTreeStorage.php
+++ b/core/lib/Drupal/Core/Menu/MenuTreeStorage.php
@@ -117,7 +117,7 @@ public function __construct(Connection $connection, CacheBackendInterface $menu_
     $this->connection = $connection;
     $this->menuCacheBackend = $menu_cache_backend;
     $this->table = $table;
-    $this->options = $options;
+    $this->options = $options + ['schema_provider' => $this];
   }
 
   /**
@@ -432,7 +432,7 @@ public function getSubtreeHeight($id) {
    *   Returns the relative depth.
    */
   protected function doFindChildrenRelativeDepth(array $original) {
-    $query = $this->connection->select($this->table, $this->options);
+    $query = $this->connection->select($this->table, NULL, $this->options);
     $query->addField($this->table, 'depth');
     $query->condition('menu_name', $original['menu_name']);
     $query->orderBy('depth', 'DESC');
@@ -567,7 +567,7 @@ protected function updateParentalStatus(array $link) {
     // If parent is empty, there is nothing to update.
     if (!empty($link['parent'])) {
       // Check if at least one visible child exists in the table.
-      $query = $this->connection->select($this->table, $this->options);
+      $query = $this->connection->select($this->table, NULL, $this->options);
       $query->addExpression('1');
       $query->range(0, 1);
       $query
@@ -609,7 +609,7 @@ protected function prepareLink(array $link, $intersect = FALSE) {
    * {@inheritdoc}
    */
   public function loadByProperties(array $properties) {
-    $query = $this->connection->select($this->table, $this->options);
+    $query = $this->connection->select($this->table, NULL, $this->options);
     $query->fields($this->table, $this->definitionFields());
     foreach ($properties as $name => $value) {
       if (!in_array($name, $this->definitionFields(), TRUE)) {
@@ -636,7 +636,7 @@ public function loadByRoute($route_name, array $route_parameters = array(), $men
     // @todo Standardize an efficient way to load by route name and parameters
     //   in place of system path. https://www.drupal.org/node/2302139
     $param_key = $route_parameters ? UrlHelper::buildQuery($route_parameters) : '';
-    $query = $this->connection->select($this->table, $this->options);
+    $query = $this->connection->select($this->table, NULL, $this->options);
     $query->fields($this->table, $this->definitionFields());
     $query->condition('route_name', $route_name);
     $query->condition('route_param_key', $param_key);
@@ -661,7 +661,7 @@ public function loadMultiple(array $ids) {
     $missing_ids = array_diff($ids, array_keys($this->definitions));
 
     if ($missing_ids) {
-      $query = $this->connection->select($this->table, $this->options);
+      $query = $this->connection->select($this->table, NULL, $this->options);
       $query->fields($this->table, $this->definitionFields());
       $query->condition('id', $missing_ids, 'IN');
       $loaded = $query->execute()->fetchAllAssoc('id', \PDO::FETCH_ASSOC);
@@ -707,7 +707,7 @@ protected function loadFull($id) {
    *   The loaded menu link definitions.
    */
   protected function loadFullMultiple(array $ids) {
-    $query = $this->connection->select($this->table, $this->options);
+    $query = $this->connection->select($this->table, NULL, $this->options);
     $query->fields($this->table);
     $query->condition('id', $ids, 'IN');
     $loaded = $query->execute()->fetchAllAssoc('id', \PDO::FETCH_ASSOC);
@@ -723,7 +723,7 @@ protected function loadFullMultiple(array $ids) {
    * {@inheritdoc}
    */
   public function getRootPathIds($id) {
-    $subquery = $this->connection->select($this->table, $this->options);
+    $subquery = $this->connection->select($this->table, NULL, $this->options);
     // @todo Consider making this dynamic based on static::MAX_DEPTH or from the
     //   schema if that is generated using static::MAX_DEPTH.
     //   https://www.drupal.org/node/2302043
@@ -732,7 +732,7 @@ public function getRootPathIds($id) {
     $result = current($subquery->execute()->fetchAll(\PDO::FETCH_ASSOC));
     $ids = array_filter($result);
     if ($ids) {
-      $query = $this->connection->select($this->table, $this->options);
+      $query = $this->connection->select($this->table, NULL, $this->options);
       $query->fields($this->table, array('id'));
       $query->orderBy('depth', 'DESC');
       $query->condition('mlid', $ids, 'IN');
@@ -750,7 +750,7 @@ public function getExpanded($menu_name, array $parents) {
     // @todo Go back to tracking in state or some other way which menus have
     //   expanded links? https://www.drupal.org/node/2302187
     do {
-      $query = $this->connection->select($this->table, $this->options);
+      $query = $this->connection->select($this->table, NULL, $this->options);
       $query->fields($this->table, array('id'));
       $query->condition('menu_name', $menu_name);
       $query->condition('expanded', 1);
@@ -836,7 +836,7 @@ public function loadTreeData($menu_name, MenuTreeParameters $parameters) {
    *   depth-first.
    */
   protected function loadLinks($menu_name, MenuTreeParameters $parameters) {
-    $query = $this->connection->select($this->table, $this->options);
+    $query = $this->connection->select($this->table, NULL, $this->options);
     $query->fields($this->table);
 
     // Allow a custom root to be specified for loading a menu link tree. If
@@ -979,7 +979,7 @@ public function loadSubtreeData($id, $max_relative_depth = NULL) {
    * {@inheritdoc}
    */
   public function menuNameInUse($menu_name) {
-    $query = $this->connection->select($this->table, $this->options);
+    $query = $this->connection->select($this->table, NULL, $this->options);
     $query->addField($this->table, 'mlid');
     $query->condition('menu_name', $menu_name);
     $query->range(0, 1);
@@ -990,7 +990,7 @@ public function menuNameInUse($menu_name) {
    * {@inheritdoc}
    */
   public function getMenuNames() {
-    $query = $this->connection->select($this->table, $this->options);
+    $query = $this->connection->select($this->table, NULL, $this->options);
     $query->addField($this->table, 'menu_name');
     $query->distinct();
     return $query->execute()->fetchAllKeyed(0, 0);
@@ -1000,7 +1000,7 @@ public function getMenuNames() {
    * {@inheritdoc}
    */
   public function countMenuLinks($menu_name = NULL) {
-    $query = $this->connection->select($this->table, $this->options);
+    $query = $this->connection->select($this->table, NULL, $this->options);
     if ($menu_name) {
       $query->condition('menu_name', $menu_name);
     }
@@ -1015,7 +1015,7 @@ public function getAllChildIds($id) {
     if (!$root) {
       return array();
     }
-    $query = $this->connection->select($this->table, $this->options);
+    $query = $this->connection->select($this->table, NULL, $this->options);
     $query->fields($this->table, array('id'));
     $query->condition('menu_name', $root['menu_name']);
     for ($i = 1; $i <= $root['depth']; $i++) {
diff --git a/core/modules/system/tests/modules/error_test/src/Controller/ErrorTestController.php b/core/modules/system/tests/modules/error_test/src/Controller/ErrorTestController.php
index a753d35..e2e819f 100644
--- a/core/modules/system/tests/modules/error_test/src/Controller/ErrorTestController.php
+++ b/core/modules/system/tests/modules/error_test/src/Controller/ErrorTestController.php
@@ -69,7 +69,7 @@ public function triggerException() {
    */
   public function triggerPDOException() {
     define('SIMPLETEST_COLLECT_ERRORS', FALSE);
-    $this->database->query('SELECT * FROM bananas_are_awesome', [], ['create_missing_table' => FALSE]);
+    $this->database->query('SELECT * FROM bananas_are_awesome', []);
   }
 
 }