diff --git a/core/lib/Drupal/Core/Cache/DatabaseBackend.php b/core/lib/Drupal/Core/Cache/DatabaseBackend.php
index ff12845..57d86dd 100644
--- a/core/lib/Drupal/Core/Cache/DatabaseBackend.php
+++ b/core/lib/Drupal/Core/Cache/DatabaseBackend.php
@@ -9,7 +9,7 @@
 
 use Drupal\Component\Utility\Crypt;
 use Drupal\Core\Database\Connection;
-use Drupal\Core\Database\SchemaObjectExistsException;
+use Drupal\Core\Database\SchemaProviderInterface;
 
 /**
  * Defines a default cache implementation.
@@ -19,7 +19,7 @@
  *
  * @ingroup cache
  */
-class DatabaseBackend implements CacheBackendInterface {
+class DatabaseBackend implements CacheBackendInterface, SchemaProviderInterface {
 
   /**
    * @var string
@@ -84,13 +84,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 = array();
-    try {
-      $result = $this->connection->query('SELECT cid, data, created, expire, serialized, tags, checksum FROM {' . $this->connection->escapeTable($this->bin) . '} WHERE cid IN ( :cids[] ) ORDER BY cid', array(':cids[]' => array_keys($cid_mapping)));
-    }
-    catch (\Exception $e) {
-      // Nothing to do.
-    }
+    $result = $this->connection->query('SELECT cid, data, created, expire, serialized, tags, checksum FROM {' . $this->connection->escapeTable($this->bin) . '} WHERE cid IN ( :cids[] ) ORDER BY cid', array(':cids[]' => array_keys($cid_mapping)), array('schema_provider' => $this));
     $cache = array();
     foreach ($result as $item) {
       // Map the cache ID back to the original.
@@ -154,29 +148,6 @@ public function set($cid, $data, $expire = Cache::PERMANENT, array $tags = array
     $tags = array_unique($tags);
     // Sort the cache tags so that they are stored consistently in the database.
     sort($tags);
-    $try_again = FALSE;
-    try {
-      // The bin might not yet exist.
-      $this->doSet($cid, $data, $expire, $tags);
-    }
-    catch (\Exception $e) {
-      // If there was an exception, try to create the bins.
-      if (!$try_again = $this->ensureBinExists()) {
-        // If the exception happened for other reason than the missing bin
-        // table, propagate the exception.
-        throw $e;
-      }
-    }
-    // Now that the bin has been created, try again if necessary.
-    if ($try_again) {
-      $this->doSet($cid, $data, $expire, $tags);
-    }
-  }
-
-  /**
-   * Actually set the cache.
-   */
-  protected function doSet($cid, $data, $expire, $tags) {
     $fields = array(
       'created' => round(microtime(TRUE), 3),
       'expire' => $expire,
@@ -192,7 +163,7 @@ protected function doSet($cid, $data, $expire, $tags) {
       $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();
@@ -246,7 +217,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', 'expire', 'created', 'tags', 'checksum', 'data', 'serialized'));
       foreach ($values as $fields) {
         // Only pass the values since the order of $fields matches the order of
@@ -276,21 +247,11 @@ public function delete($cid) {
    */
   public function deleteMultiple(array $cids) {
     $cids = array_values(array_map(array($this, 'normalizeCid'), $cids));
-    try {
-      // Delete in chunks when a large array is passed.
-      foreach (array_chunk($cids, 1000) as $cids_chunk) {
-        $this->connection->delete($this->bin)
-          ->condition('cid', $cids_chunk, 'IN')
-          ->execute();
-      }
-    }
-    catch (\Exception $e) {
-      // Create the cache table, which will be empty. This fixes cases during
-      // core install where a cache table is cleared before it is set
-      // with {cache_render} and {cache_data}.
-      if (!$this->ensureBinExists()) {
-        $this->catchException($e);
-      }
+    // Delete in chunks when a large array is passed.
+    foreach (array_chunk($cids, 1000) as $cids_chunk) {
+      $this->connection->delete($this->bin, array('schema_provider' => $this))
+        ->condition('cid', $cids_chunk, 'IN')
+        ->execute();
     }
   }
 
@@ -298,17 +259,7 @@ public function deleteMultiple(array $cids) {
    * Implements Drupal\Core\Cache\CacheBackendInterface::deleteAll().
    */
   public function deleteAll() {
-    try {
-      $this->connection->truncate($this->bin)->execute();
-    }
-    catch (\Exception $e) {
-      // Create the cache table, which will be empty. This fixes cases during
-      // core install where a cache table is cleared before it is set
-      // with {cache_render} and {cache_data}.
-      if (!$this->ensureBinExists()) {
-        $this->catchException($e);
-      }
-    }
+    $this->connection->truncate($this->bin, array('schema_provider' => $this))->execute();
   }
 
   /**
@@ -323,17 +274,12 @@ public function invalidate($cid) {
    */
   public function invalidateMultiple(array $cids) {
     $cids = array_values(array_map(array($this, 'normalizeCid'), $cids));
-    try {
-      // Update in chunks when a large array is passed.
-      foreach (array_chunk($cids, 1000) as $cids_chunk) {
-        $this->connection->update($this->bin)
-          ->fields(array('expire' => REQUEST_TIME - 1))
-          ->condition('cid', $cids_chunk, 'IN')
-          ->execute();
-      }
-    }
-    catch (\Exception $e) {
-      $this->catchException($e);
+    // Update in chunks when a large array is passed.
+    foreach (array_chunk($cids, 1000) as $cids_chunk) {
+      $this->connection->update($this->bin, array('schema_provider' => $this))
+        ->fields(array('expire' => REQUEST_TIME - 1))
+        ->condition('cid', $cids_chunk, 'IN')
+        ->execute();
     }
   }
 
@@ -341,14 +287,9 @@ public function invalidateMultiple(array $cids) {
    * Implements Drupal\Core\Cache\CacheBackendInterface::invalidateAll().
    */
   public function invalidateAll() {
-    try {
-      $this->connection->update($this->bin)
-        ->fields(array('expire' => REQUEST_TIME - 1))
-        ->execute();
-    }
-    catch (\Exception $e) {
-      $this->catchException($e);
-    }
+    $this->connection->update($this->bin, array('schema_provider' => $this))
+      ->fields(array('expire' => REQUEST_TIME - 1))
+      ->execute();
   }
 
   /**
@@ -376,49 +317,10 @@ public function removeBin() {
       $this->connection->schema()->dropTable($this->bin);
     }
     catch (\Exception $e) {
-      $this->catchException($e);
-    }
-  }
-
-  /**
-   * Check if the cache bin exists and create it if not.
-   */
-  protected function ensureBinExists() {
-    try {
-      $database_schema = $this->connection->schema();
-      if (!$database_schema->tableExists($this->bin)) {
-        $schema_definition = $this->schemaDefinition();
-        $database_schema->createTable($this->bin, $schema_definition);
-        return TRUE;
+      if ($this->connection->schema()->tableExists($this->bin)) {
+        throw $e;
       }
     }
-    // If another process has already created the cache table, attempting to
-    // recreate it will throw an exception. In this case just catch the
-    // exception and do nothing.
-    catch (SchemaObjectExistsException $e) {
-      return TRUE;
-    }
-    return FALSE;
-  }
-
-  /**
-   * Act on an exception when cache might be stale.
-   *
-   * If the table does not yet exist, that's fine, but if the table exists and
-   * yet the query failed, then the cache is stale and the exception needs to
-   * propagate.
-   *
-   * @param $e
-   *   The exception.
-   * @param string|null $table_name
-   *   The table name. Defaults to $this->bin.
-   *
-   * @throws \Exception
-   */
-  protected function catchException(\Exception $e, $table_name = NULL) {
-    if ($this->connection->schema()->tableExists($table_name ?: $this->bin)) {
-      throw $e;
-    }
   }
 
   /**
@@ -448,8 +350,9 @@ protected function normalizeCid($cid) {
   /**
    * Defines the schema for the {cache_*} bin tables.
    */
-  public function schemaDefinition() {
-    $schema = array(
+  public function getSchema($table_name) {
+    // {cache_*} bin tables all have same schema.
+    return array(
       'description' => 'Storage for the cache API.',
       'fields' => array(
         'cid' => array(
@@ -505,6 +408,6 @@ public function schemaDefinition() {
       ),
       'primary key' => array('cid'),
     );
-    return $schema;
   }
+
 }
diff --git a/core/lib/Drupal/Core/Config/DatabaseStorage.php b/core/lib/Drupal/Core/Config/DatabaseStorage.php
index 646a1d5..8efcc07 100644
--- a/core/lib/Drupal/Core/Config/DatabaseStorage.php
+++ b/core/lib/Drupal/Core/Config/DatabaseStorage.php
@@ -9,13 +9,14 @@
 
 use Drupal\Core\Database\Database;
 use Drupal\Core\Database\Connection;
-use Drupal\Core\Database\SchemaObjectExistsException;
+use Drupal\Core\Database\SchemaProviderInterface;
 use Drupal\Core\DependencyInjection\DependencySerializationTrait;
 
 /**
  * Defines the Database storage.
  */
-class DatabaseStorage implements StorageInterface {
+class DatabaseStorage implements StorageInterface, SchemaProviderInterface {
+
   use DependencySerializationTrait;
 
   /**
@@ -62,7 +63,7 @@ class DatabaseStorage implements StorageInterface {
   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;
   }
 
@@ -70,11 +71,14 @@ public function __construct(Connection $connection, $table, array $options = arr
    * Implements Drupal\Core\Config\StorageInterface::exists().
    */
   public function exists($name) {
+    $options = array(
+      'create_missing_table' => FALSE,
+    ) + $this->options;
     try {
       return (bool) $this->connection->queryRange('SELECT 1 FROM {' . $this->connection->escapeTable($this->table) . '} WHERE collection = :collection AND name = :name', 0, 1, array(
         ':collection' => $this->collection,
         ':name' => $name,
-      ), $this->options)->fetchField();
+      ), $options)->fetchField();
     }
     catch (\Exception $e) {
       // If we attempt a read without actually having the database or the table
@@ -87,36 +91,16 @@ public function exists($name) {
    * {@inheritdoc}
    */
   public function read($name) {
-    $data = FALSE;
-    try {
-      $raw = $this->connection->query('SELECT data FROM {' . $this->connection->escapeTable($this->table) . '} WHERE collection = :collection AND name = :name', array(':collection' => $this->collection, ':name' => $name), $this->options)->fetchField();
-      if ($raw !== FALSE) {
-        $data = $this->decode($raw);
-      }
-    }
-    catch (\Exception $e) {
-      // If we attempt a read without actually having the database or the table
-      // available, just return FALSE so the caller can handle it.
-    }
-    return $data;
+    $raw = $this->connection->query('SELECT data FROM {' . $this->connection->escapeTable($this->table) . '} WHERE collection = :collection AND name = :name', array(':collection' => $this->collection, ':name' => $name), $this->options)->fetchField();
+    return $raw !== FALSE ? $this->decode($raw) : FALSE;
   }
 
   /**
    * {@inheritdoc}
    */
   public function readMultiple(array $names) {
-    $list = array();
-    try {
-      $list = $this->connection->query('SELECT name, data FROM {' . $this->connection->escapeTable($this->table) . '} WHERE collection = :collection AND name IN ( :names[] )', array(':collection' => $this->collection, ':names[]' => $names), $this->options)->fetchAllKeyed();
-      foreach ($list as &$data) {
-        $data = $this->decode($data);
-      }
-    }
-    catch (\Exception $e) {
-      // If we attempt a read without actually having the database or the table
-      // available, just return an empty array so the caller can handle it.
-    }
-    return $list;
+    $list = $this->connection->query('SELECT name, data FROM {' . $this->connection->escapeTable($this->table) . '} WHERE collection = :collection AND name IN ( :names[] )', array(':collection' => $this->collection, ':names[]' => $names), $this->options)->fetchAllKeyed();
+    return array_map([$this, 'decode'], $list);
   }
 
   /**
@@ -124,30 +108,6 @@ public function readMultiple(array $names) {
    */
   public function write($name, array $data) {
     $data = $this->encode($data);
-    try {
-      return $this->doWrite($name, $data);
-    }
-    catch (\Exception $e) {
-      // If there was an exception, try to create the table.
-      if ($this->ensureTableExists()) {
-        return $this->doWrite($name, $data);
-      }
-      // Some other failure that we can not recover from.
-      throw $e;
-    }
-  }
-
-  /**
-   * Helper method so we can re-try a write.
-   *
-   * @param string $name
-   *   The config name.
-   * @param string $data
-   *   The config data, already dumped to a string.
-   *
-   * @return bool
-   */
-  protected function doWrite($name, $data) {
     $options = array('return' => Database::RETURN_AFFECTED) + $this->options;
     return (bool) $this->connection->merge($this->table, $options)
       ->keys(array('collection', 'name'), array($this->collection, $name))
@@ -156,37 +116,9 @@ protected function doWrite($name, $data) {
   }
 
   /**
-   * Check if the config table exists and create it if not.
-   *
-   * @return bool
-   *   TRUE if the table was created, FALSE otherwise.
-   *
-   * @throws \Drupal\Core\Config\StorageException
-   *   If a database error occurs.
-   */
-  protected function ensureTableExists()  {
-    try {
-      if (!$this->connection->schema()->tableExists($this->table)) {
-        $this->connection->schema()->createTable($this->table, static::schemaDefinition());
-        return TRUE;
-      }
-    }
-    // If another process has already created the config table, attempting to
-    // recreate it will throw an exception. In this case just catch the
-    // exception and do nothing.
-    catch (SchemaObjectExistsException $e) {
-      return TRUE;
-    }
-    catch (\Exception $e) {
-      throw new StorageException($e->getMessage(), NULL, $e);
-    }
-    return FALSE;
-  }
-
-  /**
    * Defines the schema for the configuration table.
    */
-  protected static function schemaDefinition() {
+  public function getSchema($table_name) {
     $schema = array(
       'description' => 'The base table for configuration data.',
       'fields' => array(
@@ -213,7 +145,7 @@ protected static function schemaDefinition() {
       ),
       'primary key' => array('collection', 'name'),
     );
-    return $schema;
+    return $this->table == $table_name ? $schema : FALSE;
   }
 
   /**
@@ -224,7 +156,10 @@ protected static function schemaDefinition() {
    * @todo Ignore replica targets for data manipulation operations.
    */
   public function delete($name) {
-    $options = array('return' => Database::RETURN_AFFECTED) + $this->options;
+    $options = array(
+      'return' => Database::RETURN_AFFECTED,
+      'create_missing_table' => FALSE,
+    ) + $this->options;
     return (bool) $this->connection->delete($this->table, $options)
       ->condition('collection', $this->collection)
       ->condition('name', $name)
@@ -269,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');
@@ -286,7 +221,10 @@ public function listAll($prefix = '') {
    */
   public function deleteAll($prefix = '') {
     try {
-      $options = array('return' => Database::RETURN_AFFECTED) + $this->options;
+      $options = array(
+        'return' => Database::RETURN_AFFECTED,
+        'create_missing_table' => FALSE,
+      ) + $this->options;
       return (bool) $this->connection->delete($this->table, $options)
         ->condition('name', $prefix . '%', 'LIKE')
         ->condition('collection', $this->collection)
@@ -323,7 +261,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 af91674..5fad447 100644
--- a/core/lib/Drupal/Core/Database/Connection.php
+++ b/core/lib/Drupal/Core/Database/Connection.php
@@ -215,6 +215,12 @@ public function destroy() {
    *   - Database::RETURN_NULL: Do not return anything, as there is no
    *     meaningful value to return. That is the case for INSERT queries on
    *     tables that do not contain a serial column.
+   * - create_missing_table: Boolean value, whether to create a missing table.
+   * - missing_table_name: In case create_missing_table is TRUE, this will
+   *   be the table name used for the automatic table creation.
+   * - schema_provider: An object implementing
+   *   \Drupal\Core\Schema\SchemaProviderInterface . This will be called when
+   *   both create_missing_table and missing_table_name is set.
    * - throw_exception: By default, the database system will catch any errors
    *   on a query as an Exception, log it, and then rethrow it so that code
    *   further up the call chain can take an appropriate action. To suppress
@@ -229,6 +235,7 @@ protected function defaultOptions() {
       'target' => 'default',
       'fetch' => \PDO::FETCH_OBJ,
       'return' => Database::RETURN_STATEMENT,
+      'create_missing_table' => TRUE,
       'throw_exception' => TRUE,
     );
   }
@@ -567,6 +574,10 @@ public function query($query, array $args = array(), $options = array()) {
       }
     }
     catch (\PDOException $e) {
+      $handled = $this->handleMissingTable($e, $query, $args, $options);
+      if (isset($handled)) {
+        return $handled;
+      }
       // Most database drivers will return NULL here, but some of them
       // (e.g. the SQLite driver) may need to re-run the query, so the return
       // value will be the same as for static::query().
@@ -616,6 +627,52 @@ protected function handleQueryException(\PDOException $e, $query, array $args =
   }
 
   /**
+   * Handles a missing table exception.
+   *
+   * @param \PDOException $e
+   *   The exception thrown by static::query().
+   * @param $query
+   *   The query executed by static::query().
+   * @param array $args
+   *   An array of arguments for the prepared statement.
+   * @param array $options
+   *   An associative array of options to control how the query is run.
+   *
+   * @return \Drupal\Core\Database\StatementInterface|bool
+   *   This method re-runs the query after creating the missing table. In this
+   *   case it returns a \Drupal\Core\Database\StatementInterface object or
+   *   FALSE if the schema provider is not specified in the $options.
+   */
+  protected function handleMissingTable($e, $query, $args, $options) {
+    if ($this->isTableMissingException($e) && $options['create_missing_table']) {
+      if ($options['return'] == Database::RETURN_STATEMENT) {
+        // This is a SELECT on a missing table. Creating the table and
+        // re-running the query will result in an empty result so return that
+        // early instead.
+        return new StatementEmpty();
+      }
+      if (!empty($options['missing_table_name']) && isset($options['schema_provider'])) {
+        $table = $options['missing_table_name'];
+        /** @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['missing_table_name']);
+            unset($options['schema_provider']);
+            return $this->query($query, $args, $options);
+          }
+        }
+      }
+      elseif ($options['create_missing_table'] === TRUE) {
+        return FALSE;
+      }
+    }
+  }
+
+  /**
    * Expands out shorthand placeholders.
    *
    * Drupal supports an alternate syntax for doing arrays of values. We
@@ -1300,4 +1357,37 @@ public function __sleep() {
     throw new \LogicException('The database connection is not serializable. This probably means you are serializing an object that has an indirect reference to the database connection. Adjust your code so that is not necessary. Alternatively, look at DependencySerializationTrait as a temporary solution.');
   }
 
+  /**
+   * Extracts the SQLSTATE error from the PDOException.
+   *
+   * @param \Exception $e
+   *   The exception
+   *
+   * @return string
+   *   The five character error code.
+   */
+  protected function getSQLState(\Exception $e) {
+    // The PDOException code is not always reliable, try to see whether the
+    // message has something usable.
+    if (!$e->getCode() && preg_match('/^SQLSTATE\[(\w{5})\]/', $e->getMessage(), $matches)) {
+      return $matches[1];
+    }
+    else {
+      return $e->getCode();
+    }
+  }
+
+  /**
+   * Determines whether an exception means missing table.
+   *
+   * @param \Exception $e
+   *   The exception being investigated. Use
+   *   \Drupal\Core\Database\Database::getSqlState($e) instead of
+   *   $e->getCode() to get the SQLSTATE.
+   *
+   * @return bool
+   *   TRUE if this exception means table missing, FALSE otherwise.
+   */
+  abstract public function isTableMissingException(\Exception $e);
+
 }
diff --git a/core/lib/Drupal/Core/Database/Driver/fake/FakeConnection.php b/core/lib/Drupal/Core/Database/Driver/fake/FakeConnection.php
index 1be5147..f50cd38 100644
--- a/core/lib/Drupal/Core/Database/Driver/fake/FakeConnection.php
+++ b/core/lib/Drupal/Core/Database/Driver/fake/FakeConnection.php
@@ -164,4 +164,10 @@ public function nextId($existing_id = 0) {
     throw new \Exception('Method not supported');
   }
 
+  /**
+   * {@inheritdoc}
+   */
+  public function isTableMissingException(\Exception $e) {
+  }
+
 }
diff --git a/core/lib/Drupal/Core/Database/Driver/mysql/Connection.php b/core/lib/Drupal/Core/Database/Driver/mysql/Connection.php
index 7dad008..bf6cc7f 100644
--- a/core/lib/Drupal/Core/Database/Driver/mysql/Connection.php
+++ b/core/lib/Drupal/Core/Database/Driver/mysql/Connection.php
@@ -278,6 +278,16 @@ protected function popCommittableTransactions() {
       }
     }
   }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function isTableMissingException(\Exception $e) {
+    // "42S02" is the MySQL error code for "Unknown table".
+    // @see http://dev.mysql.com/doc/refman/5.5/en/error-messages-server.html
+    return $this->getSQLState($e) === '42S02';
+  }
+
 }
 
 
diff --git a/core/lib/Drupal/Core/Database/Driver/mysql/Schema.php b/core/lib/Drupal/Core/Database/Driver/mysql/Schema.php
index 8237c79..01dd588 100644
--- a/core/lib/Drupal/Core/Database/Driver/mysql/Schema.php
+++ b/core/lib/Drupal/Core/Database/Driver/mysql/Schema.php
@@ -487,7 +487,8 @@ public function tableExists($table) {
     // @todo This override should be removed once we require a version of MySQL
     //   that has that bug fixed.
     try {
-      $this->connection->queryRange("SELECT 1 FROM {" . $table . "}", 0, 1);
+      $options['create_missing_table'] = FALSE;
+      $this->connection->queryRange("SELECT 1 FROM {" . $table . "}", 0, 1, [], $options);
       return TRUE;
     }
     catch (\Exception $e) {
diff --git a/core/lib/Drupal/Core/Database/Driver/pgsql/Connection.php b/core/lib/Drupal/Core/Database/Driver/pgsql/Connection.php
index 118974c..f95ba1c 100644
--- a/core/lib/Drupal/Core/Database/Driver/pgsql/Connection.php
+++ b/core/lib/Drupal/Core/Database/Driver/pgsql/Connection.php
@@ -350,6 +350,18 @@ public function rollbackSavepoint($savepoint_name = 'mimic_implicit_commit') {
       $this->rollback($savepoint_name);
     }
   }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function isTableMissingException(\Exception $e) {
+    // "42P01" is the PostgreSQL error code for "undefined_table".
+    // @see http://www.postgresql.org/docs/9.3/static/errcodes-appendix.html
+    return $this->getSQLState($e) === '42P01';
+  }
+
+
+
 }
 
 /**
diff --git a/core/lib/Drupal/Core/Database/Driver/sqlite/Connection.php b/core/lib/Drupal/Core/Database/Driver/sqlite/Connection.php
index 247d653..f37ce3f 100644
--- a/core/lib/Drupal/Core/Database/Driver/sqlite/Connection.php
+++ b/core/lib/Drupal/Core/Database/Driver/sqlite/Connection.php
@@ -395,4 +395,11 @@ public function getFullQualifiedTableName($table) {
     return $prefix . $table;
   }
 
+  /**
+   * {@inheritdoc}
+   */
+  public function isTableMissingException(\Exception $e) {
+    return preg_match('/^SQLSTATE\[HY000\]: General error: 1 no such table/', $e->getMessage());
+  }
+
 }
diff --git a/core/lib/Drupal/Core/Database/Query/Delete.php b/core/lib/Drupal/Core/Database/Query/Delete.php
index 8848391..1a0ad07 100644
--- a/core/lib/Drupal/Core/Database/Query/Delete.php
+++ b/core/lib/Drupal/Core/Database/Query/Delete.php
@@ -139,7 +139,6 @@ public function execute() {
       $this->condition->compile($this->connection, $this);
       $values = $this->condition->arguments();
     }
-
     return $this->connection->query((string) $this, $values, $this->queryOptions);
   }
 
diff --git a/core/lib/Drupal/Core/Database/Query/Insert.php b/core/lib/Drupal/Core/Database/Query/Insert.php
index d9ddbc0..b9fa167 100644
--- a/core/lib/Drupal/Core/Database/Query/Insert.php
+++ b/core/lib/Drupal/Core/Database/Query/Insert.php
@@ -72,9 +72,11 @@ class Insert extends Query {
    *   Array of database options.
    */
   public function __construct($connection, $table, array $options = array()) {
-    if (!isset($options['return'])) {
-      $options['return'] = Database::RETURN_INSERT_ID;
-    }
+    $options += array(
+      'return' => Database::RETURN_INSERT_ID,
+      'create_missing_table' => TRUE,
+      'missing_table_name' => $table,
+    );
     parent::__construct($connection, $options);
     $this->table = $table;
   }
diff --git a/core/lib/Drupal/Core/Database/Query/Merge.php b/core/lib/Drupal/Core/Database/Query/Merge.php
index dfa50e3..ff8bac8 100644
--- a/core/lib/Drupal/Core/Database/Query/Merge.php
+++ b/core/lib/Drupal/Core/Database/Query/Merge.php
@@ -135,6 +135,10 @@ class Merge extends Query implements ConditionInterface {
    */
   public function __construct(Connection $connection, $table, array $options = array()) {
     $options['return'] = Database::RETURN_AFFECTED;
+    if (!isset($options['create_missing_table'])) {
+      $options['create_missing_table'] = TRUE;
+      $options['missing_table_name'] = $table;
+    }
     parent::__construct($connection, $options);
     $this->table = $table;
     $this->conditionTable = $table;
@@ -437,12 +441,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);
           }
@@ -460,7 +465,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/Database/Query/Select.php b/core/lib/Drupal/Core/Database/Query/Select.php
index d925a07..244691e 100644
--- a/core/lib/Drupal/Core/Database/Query/Select.php
+++ b/core/lib/Drupal/Core/Database/Query/Select.php
@@ -647,6 +647,9 @@ public function rightJoin($table, $alias = NULL, $condition = NULL, $arguments =
    * {@inheritdoc}
    */
   public function addJoin($type, $table, $alias = NULL, $condition = NULL, $arguments = array()) {
+    if (isset($type)) {
+      $this->queryOptions['create_missing_table'] = FALSE;
+    }
     if (empty($alias)) {
       if ($table instanceof SelectInterface) {
         $alias = 'subquery';
diff --git a/core/lib/Drupal/Core/Database/Query/Truncate.php b/core/lib/Drupal/Core/Database/Query/Truncate.php
index 857e199..f160c78 100644
--- a/core/lib/Drupal/Core/Database/Query/Truncate.php
+++ b/core/lib/Drupal/Core/Database/Query/Truncate.php
@@ -60,7 +60,16 @@ public function compiled() {
    *   Return value is dependent on the database type.
    */
   public function execute() {
-    return $this->connection->query((string) $this, array(), $this->queryOptions);
+    try {
+      return $this->connection->query((string) $this, array(), $this->queryOptions);
+    }
+    catch (\Exception $e) {
+      if ($this->connection->isTableMissingException($e)) {
+        // The table is quite empty. It doesn't even exist but that's fine.
+        return 0;
+      }
+      throw $e;
+    }
   }
 
   /**
diff --git a/core/lib/Drupal/Core/Database/Schema.php b/core/lib/Drupal/Core/Database/Schema.php
index c4d27c3..33fcec7 100644
--- a/core/lib/Drupal/Core/Database/Schema.php
+++ b/core/lib/Drupal/Core/Database/Schema.php
@@ -7,7 +7,6 @@
 
 namespace Drupal\Core\Database;
 
-use Drupal\Core\Database\SchemaObjectExistsException;
 use Drupal\Core\Database\Query\Condition;
 use Drupal\Core\Database\Query\PlaceholderInterface;
 
@@ -521,6 +520,34 @@ public function createTable($name, $table) {
   }
 
   /**
+   * Checks a table for existence and creates it as necessary.
+   *
+   * @param $table_name
+   *   The name of the table to check and create.
+   * @param array $schema
+   *   A Schema API table definition array.
+   *
+   * @return bool
+   *   TRUE if the table already existed or now exists, FALSE if there was an
+   *   error and it does not.
+   */
+  public function ensureTableExists($table_name, array $schema) {
+    try {
+      if (!$this->tableExists($table_name)) {
+        $this->createTable($table_name, $schema);
+      }
+      return TRUE;
+    }
+    catch (SchemaObjectExistsException $e) {
+      // If another process has already created the table, attempting to
+      // recreate it will throw an exception. In this case just catch the
+      // exception and do nothing.
+      return TRUE;
+    }
+    return FALSE;
+  }
+
+  /**
    * Return an array of field names from an array of key/index column specifiers.
    *
    * This is usually an identity function but if a key/index uses a column prefix
diff --git a/core/lib/Drupal/Core/Database/SchemaProviderInterface.php b/core/lib/Drupal/Core/Database/SchemaProviderInterface.php
new file mode 100644
index 0000000..0ae6b00
--- /dev/null
+++ b/core/lib/Drupal/Core/Database/SchemaProviderInterface.php
@@ -0,0 +1,32 @@
+<?php
+
+/**
+ * @file
+ * Contains \Drupal\Core\Database\SchemaProviderInterface.
+ */
+
+namespace Drupal\Core\Database;
+
+/**
+ * This interface is a helper for queries creating tables on demand.
+ *
+ * An INSERT or MERGE query will automatically find if the caller implements
+ * this interface and create the table on demand.
+ */
+interface SchemaProviderInterface {
+
+  /**
+   * A schema API array.
+   *
+   * @param string $table_name
+   *   The name of the table being created. The implementation must check this.
+   *
+   * @return array|FALSE
+   *   A schema API array or FALSE if this implementation doesn't know about
+   *   the $table_name table.
+   *
+   * @see hook_schema()
+   * @see \Drupal\Core\Cache\DatabaseBackend
+   */
+  public function getSchema($table_name);
+}
diff --git a/core/lib/Drupal/Core/Database/StatementEmpty.php b/core/lib/Drupal/Core/Database/StatementEmpty.php
index e6cd08d..269be65 100644
--- a/core/lib/Drupal/Core/Database/StatementEmpty.php
+++ b/core/lib/Drupal/Core/Database/StatementEmpty.php
@@ -52,7 +52,7 @@ public function fetch($mode = NULL, $cursor_orientation = NULL, $cursor_offset =
   }
 
   public function fetchField($index = 0) {
-    return NULL;
+    return FALSE;
   }
 
   public function fetchObject() {
diff --git a/core/lib/Drupal/Core/Menu/MenuTreeStorage.php b/core/lib/Drupal/Core/Menu/MenuTreeStorage.php
index 659d6be..d2aa71f 100644
--- a/core/lib/Drupal/Core/Menu/MenuTreeStorage.php
+++ b/core/lib/Drupal/Core/Menu/MenuTreeStorage.php
@@ -15,13 +15,12 @@
 use Drupal\Core\Cache\CacheTagsInvalidatorInterface;
 use Drupal\Core\Database\Connection;
 use Drupal\Core\Database\Database;
-use Drupal\Core\Database\Query\SelectInterface;
-use Drupal\Core\Database\SchemaObjectExistsException;
+use Drupal\Core\Database\SchemaProviderInterface;
 
 /**
  * Provides a menu tree storage using the database.
  */
-class MenuTreeStorage implements MenuTreeStorageInterface {
+class MenuTreeStorage implements MenuTreeStorageInterface, SchemaProviderInterface {
 
   /**
    * The maximum depth of a menu links tree.
@@ -129,7 +128,7 @@ public function __construct(Connection $connection, CacheBackendInterface $menu_
     $this->menuCacheBackend = $menu_cache_backend;
     $this->cacheTagsInvalidator = $cache_tags_invalidator;
     $this->table = $table;
-    $this->options = $options;
+    $this->options = $options + ['schema_provider' => $this];
   }
 
   /**
@@ -220,33 +219,6 @@ protected function purgeMultiple(array $ids) {
   }
 
   /**
-   * Executes a select query while making sure the database table exists.
-   *
-   * @param \Drupal\Core\Database\Query\SelectInterface $query
-   *   The select object to be executed.
-   *
-   * @return \Drupal\Core\Database\StatementInterface|null
-   *   A prepared statement, or NULL if the query is not valid.
-   *
-   * @throws \Exception
-   *   Thrown if the table could not be created or the database connection
-   *   failed.
-   */
-  protected function safeExecuteSelect(SelectInterface $query) {
-    try {
-      return $query->execute();
-    }
-    catch (\Exception $e) {
-      // If there was an exception, try to create the table.
-      if ($this->ensureTableExists()) {
-        return $query->execute();
-      }
-      // Some other failure that we can not recover from.
-      throw $e;
-    }
-  }
-
-  /**
    * {@inheritdoc}
    */
   public function save(array $link) {
@@ -333,7 +305,7 @@ protected function doSave(array $link) {
   protected function preSave(array &$link, array $original) {
     static $schema_fields, $schema_defaults;
     if (empty($schema_fields)) {
-      $schema = static::schemaDefinition();
+      $schema = $this->getSchema($this->table);
       $schema_fields = $schema['fields'];
       foreach ($schema_fields as $name => $spec) {
         if (isset($spec['default'])) {
@@ -427,7 +399,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');
@@ -437,7 +409,7 @@ protected function doFindChildrenRelativeDepth(array $original) {
       $query->condition("p$i", $original["p$i"]);
     }
 
-    $max_depth = $this->safeExecuteSelect($query)->fetchField();
+    $max_depth = $query->execute()->fetchField();
 
     return ($max_depth > $original['depth']) ? $max_depth - $original['depth'] : 0;
   }
@@ -590,7 +562,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
@@ -632,7 +604,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)) {
@@ -641,7 +613,7 @@ public function loadByProperties(array $properties) {
       }
       $query->condition($name, $value);
     }
-    $loaded = $this->safeExecuteSelect($query)->fetchAllAssoc('id', \PDO::FETCH_ASSOC);
+    $loaded = $query->execute()->fetchAllAssoc('id', \PDO::FETCH_ASSOC);
     foreach ($loaded as $id => $link) {
       $loaded[$id] = $this->prepareLink($link);
     }
@@ -659,7 +631,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);
@@ -670,7 +642,7 @@ public function loadByRoute($route_name, array $route_parameters = array(), $men
     $query->orderBy('depth');
     $query->orderBy('weight');
     $query->orderBy('id');
-    $loaded = $this->safeExecuteSelect($query)->fetchAllAssoc('id', \PDO::FETCH_ASSOC);
+    $loaded = $query->execute()->fetchAllAssoc('id', \PDO::FETCH_ASSOC);
     foreach ($loaded as $id => $link) {
       $loaded[$id] = $this->prepareLink($link);
     }
@@ -684,10 +656,10 @@ 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 = $this->safeExecuteSelect($query)->fetchAllAssoc('id', \PDO::FETCH_ASSOC);
+      $loaded = $query->execute()->fetchAllAssoc('id', \PDO::FETCH_ASSOC);
       foreach ($loaded as $id => $link) {
         $this->definitions[$id] = $this->prepareLink($link);
       }
@@ -730,10 +702,10 @@ 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 = $this->safeExecuteSelect($query)->fetchAllAssoc('id', \PDO::FETCH_ASSOC);
+    $loaded = $query->execute()->fetchAllAssoc('id', \PDO::FETCH_ASSOC);
     foreach ($loaded as &$link) {
       foreach ($this->serializedFields() as $name) {
         $link[$name] = unserialize($link[$name]);
@@ -746,7 +718,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
@@ -755,13 +727,13 @@ 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');
       // @todo Cache this result in memory if we find it is being used more
       //   than once per page load. https://www.drupal.org/node/2302185
-      return $this->safeExecuteSelect($query)->fetchAllKeyed(0, 0);
+      return $query->execute()->fetchAllKeyed(0, 0);
     }
     return array();
   }
@@ -773,7 +745,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);
@@ -781,7 +753,7 @@ public function getExpanded($menu_name, array $parents) {
       $query->condition('enabled', 1);
       $query->condition('parent', $parents, 'IN');
       $query->condition('id', $parents, 'NOT IN');
-      $result = $this->safeExecuteSelect($query)->fetchAllKeyed(0, 0);
+      $result = $query->execute()->fetchAllKeyed(0, 0);
       $parents += $result;
     } while (!empty($result));
     return $parents;
@@ -859,7 +831,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
@@ -939,7 +911,7 @@ protected function loadLinks($menu_name, MenuTreeParameters $parameters) {
       }
     }
 
-    $links = $this->safeExecuteSelect($query)->fetchAllAssoc('id', \PDO::FETCH_ASSOC);
+    $links = $query->execute()->fetchAllAssoc('id', \PDO::FETCH_ASSOC);
 
     return $links;
   }
@@ -1002,32 +974,32 @@ 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);
-    return (bool) $this->safeExecuteSelect($query);
+    return (bool) $query->execute();
   }
 
   /**
    * {@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 $this->safeExecuteSelect($query)->fetchAllKeyed(0, 0);
+    return $query->execute()->fetchAllKeyed(0, 0);
   }
 
   /**
    * {@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);
     }
-    return $this->safeExecuteSelect($query->countQuery())->fetchField();
+    return $query->countQuery()->execute()->fetchField();
   }
 
   /**
@@ -1038,7 +1010,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++) {
@@ -1046,7 +1018,7 @@ public function getAllChildIds($id) {
     }
     // The next p column should not be empty. This excludes the root link.
     $query->condition("p$i", 0, '>');
-    return $this->safeExecuteSelect($query)->fetchAllKeyed(0, 0);
+    return $query->execute()->fetchAllKeyed(0, 0);
   }
 
   /**
@@ -1126,34 +1098,6 @@ protected function treeDataRecursive(array &$links, array $parents, $depth) {
   }
 
   /**
-   * Checks if the tree table exists and create it if not.
-   *
-   * @return bool
-   *   TRUE if the table was created, FALSE otherwise.
-   *
-   * @throws \Drupal\Component\Plugin\Exception\PluginException
-   *   If a database error occurs.
-   */
-  protected function ensureTableExists() {
-    try {
-      if (!$this->connection->schema()->tableExists($this->table)) {
-        $this->connection->schema()->createTable($this->table, static::schemaDefinition());
-        return TRUE;
-      }
-    }
-    catch (SchemaObjectExistsException $e) {
-      // If another process has already created the config table, attempting to
-      // recreate it will throw an exception. In this case just catch the
-      // exception and do nothing.
-      return TRUE;
-    }
-    catch (\Exception $e) {
-      throw new PluginException($e->getMessage(), NULL, $e);
-    }
-    return FALSE;
-  }
-
-  /**
    * Determines serialized fields in the storage.
    *
    * @return array
@@ -1161,7 +1105,7 @@ protected function ensureTableExists() {
    */
   protected function serializedFields() {
     if (empty($this->serializedFields)) {
-      $schema = static::schemaDefinition();
+      $schema = $this->getSchema($this->table);
       foreach ($schema['fields'] as $name => $field) {
         if (!empty($field['serialize'])) {
           $this->serializedFields[] = $name;
@@ -1187,7 +1131,7 @@ protected function definitionFields() {
    * @return array
    *   The schema API definition for the SQL storage table.
    */
-  protected static function schemaDefinition() {
+  public function getSchema($table_name) {
     $schema = array(
       'description' => 'Contains the menu tree hierarchy.',
       'fields' => array(
@@ -1434,7 +1378,7 @@ protected static function schemaDefinition() {
       ),
     );
 
-    return $schema;
+    return $this->table == $table_name ? $schema: FALSE;
   }
 
   /**
diff --git a/core/modules/system/src/Tests/Cache/DatabaseBackendTagTest.php b/core/modules/system/src/Tests/Cache/DatabaseBackendTagTest.php
index 24cef4d..d05c6da 100644
--- a/core/modules/system/src/Tests/Cache/DatabaseBackendTagTest.php
+++ b/core/modules/system/src/Tests/Cache/DatabaseBackendTagTest.php
@@ -8,6 +8,7 @@
 namespace Drupal\system\Tests\Cache;
 
 use Drupal\Core\Cache\Cache;
+use Drupal\Core\Database\Database;
 use Drupal\Core\DependencyInjection\ContainerBuilder;
 use Drupal\simpletest\KernelTestBase;
 use Symfony\Component\DependencyInjection\Reference;
@@ -47,8 +48,9 @@ public function testTagInvalidations() {
       $bin->set('test', 'value', Cache::PERMANENT, $tags);
       $this->assertTrue($bin->get('test'), 'Cache item was set in bin.');
     }
-
-    $invalidations_before = intval(db_select('cachetags')->fields('cachetags', array('invalidations'))->condition('tag', 'test_tag:2')->execute()->fetchField());
+    // Just setting cache data does not create the cachetags table, that only
+    // happens when tags are invalidated.
+    $this->assertFalse(Database::getConnection()->schema()->tableExists('cachetags'));
     Cache::invalidateTags(array('test_tag:2'));
 
     // Test that cache entry has been invalidated in multiple bins.
@@ -59,7 +61,7 @@ public function testTagInvalidations() {
 
     // Test that only one tag invalidation has occurred.
     $invalidations_after = intval(db_select('cachetags')->fields('cachetags', array('invalidations'))->condition('tag', 'test_tag:2')->execute()->fetchField());
-    $this->assertEqual($invalidations_after, $invalidations_before + 1, 'Only one addition cache tag invalidation has occurred after invalidating a tag used in multiple bins.');
+    $this->assertEqual($invalidations_after, 1, 'Only one addition cache tag invalidation has occurred after invalidating a tag used in multiple bins.');
   }
 
 }
diff --git a/core/modules/system/src/Tests/Database/SchemaTest.php b/core/modules/system/src/Tests/Database/SchemaTest.php
index 9a234ed..774e159 100644
--- a/core/modules/system/src/Tests/Database/SchemaTest.php
+++ b/core/modules/system/src/Tests/Database/SchemaTest.php
@@ -123,6 +123,8 @@ function testSchema() {
     // Try to drop the table.
     db_drop_table('test_table2');
     $this->assertFalse(db_table_exists('test_table2'), 'The dropped table does not exist.');
+    // Try to ensure the table exists.
+    $this->assertTrue(Database::getConnection()->schema()->ensureTableExists('test_table2', drupal_get_schema_unprocessed('database_test', 'test')));
 
     // Recreate the table.
     db_create_table('test_table', $table_specification);
diff --git a/core/modules/system/src/Tests/Database/SelectTest.php b/core/modules/system/src/Tests/Database/SelectTest.php
index 7937e27..1f5e9fd 100644
--- a/core/modules/system/src/Tests/Database/SelectTest.php
+++ b/core/modules/system/src/Tests/Database/SelectTest.php
@@ -7,13 +7,14 @@
 
 namespace Drupal\system\Tests\Database;
 use Drupal\Core\Database\InvalidQueryException;
+use Drupal\Core\Database\SchemaProviderInterface;
 
 /**
  * Tests the Select query builder.
  *
  * @group Database
  */
-class SelectTest extends DatabaseTestBase {
+class SelectTest extends DatabaseTestBase implements SchemaProviderInterface {
 
   /**
    * Tests rudimentary SELECT statements.
@@ -28,6 +29,30 @@ function testSimpleSelect() {
   }
 
   /**
+   * Tests rudimentary SELECT statements with table dropping and recreation.
+   */
+  function testSimpleSelectWithEnsuringTable() {
+    db_drop_table('test');
+    $query = db_select('test');
+    $query->addField('test', 'name');
+    $query->addField('test', 'age', 'age');
+    $num_records = $query->countQuery()->execute()->fetchField();
+    // The real test is not getting a fatal error from the exception over the
+    // missing table.
+    $this->assertEqual($num_records, 0, 'Returned the correct number of rows.');
+    // The table should be recreated, add back the sample data and re-run the
+    // test.
+    try {
+      \Drupal::database()->schema()->ensureTableExists('test', $this->getSchema('test'));
+      static::addSampleData();
+    }
+    catch (\Exception $e) {
+      // Ignore adding rows to tables already filled.
+    }
+    $this->testSimpleSelect();
+  }
+
+  /**
    * Tests rudimentary SELECT statement with a COMMENT.
    */
   function testSimpleComment() {
@@ -453,7 +478,7 @@ function testInvalidSelectCount() {
 
     try {
       // This query will fail because the table does not exist.
-      db_select('some_table_that_doesnt_exist', 't')
+      db_select('some_table_that_doesnt_exist', 't', ['create_missing_table' => FALSE])
         ->fields('t')
         ->countQuery()
         ->execute();
@@ -494,4 +519,12 @@ function testEmptyInCondition() {
     }
   }
 
+  /**
+   * A schema API array.
+   *
+   * @return array
+   */
+  public function getSchema($table_name) {
+    return $table_name == 'test' ? drupal_get_schema_unprocessed('database_test', 'test') : FALSE;
+  }
 }
diff --git a/core/modules/system/src/Tests/Menu/MenuTreeStorageTest.php b/core/modules/system/src/Tests/Menu/MenuTreeStorageTest.php
index 1df4050..a4f47ad 100644
--- a/core/modules/system/src/Tests/Menu/MenuTreeStorageTest.php
+++ b/core/modules/system/src/Tests/Menu/MenuTreeStorageTest.php
@@ -75,7 +75,7 @@ protected function doTestTable() {
     // that selecting from the storage creates the table.
     $tree_storage = new MenuTreeStorage($this->container->get('database'), $this->container->get('cache.menu'), $this->container->get('cache_tags.invalidator'), 'test_menu_tree');
     $this->assertFalse($this->connection->schema()->tableExists('test_menu_tree'), 'Test table is not yet created');
-    $tree_storage->countMenuLinks();
+    $tree_storage->save($this->getLinkArray($this->randomMachineName()));
     $this->assertTrue($this->connection->schema()->tableExists('test_menu_tree'), 'Test table was created');
   }
 
@@ -349,17 +349,7 @@ public function testLoadByProperties() {
    * Adds a link with the given ID and supply defaults.
    */
   protected function addMenuLink($id, $parent = '', $route_name = 'test', $route_parameters = array(), $menu_name = 'tools', $extra = array()) {
-    $link = array(
-      'id' => $id,
-      'menu_name' => $menu_name,
-      'route_name' => $route_name,
-      'route_parameters' => $route_parameters,
-      'title_arguments' => array(),
-      'title' => 'test',
-      'parent' => $parent,
-      'options' => array(),
-      'metadata' => array(),
-    ) + $extra;
+    $link = $this->getLinkArray($id, $parent, $route_name, $route_parameters, $menu_name) + $extra;
     $this->treeStorage->save($link);
   }
 
@@ -432,4 +422,26 @@ protected function assertMenuLink($id, array $expected_properties, array $parent
     $this->assertEqual(array_intersect($children, $found_children), $children, 'Child IDs match');
   }
 
+  /**
+   * @param $id
+   * @param $parent
+   * @param $route_name
+   * @param $route_parameters
+   * @param $menu_name
+   * @return array
+   */
+  protected function getLinkArray($id, $parent = '', $route_name = 'test', $route_parameters = array(), $menu_name = 'tools') {
+    return array(
+      'id' => $id,
+      'menu_name' => $menu_name,
+      'route_name' => $route_name,
+      'route_parameters' => $route_parameters,
+      'title_arguments' => array(),
+      'title' => 'test',
+      'parent' => $parent,
+      'options' => array(),
+      'metadata' => array(),
+    );
+  }
+
 }
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 25ef1c5..d6da367 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
@@ -80,7 +80,7 @@ public function triggerException() {
    */
   public function triggerPDOException() {
     define('SIMPLETEST_COLLECT_ERRORS', FALSE);
-    $this->database->query('SELECT * FROM bananas_are_awesome');
+    $this->database->query('SELECT * FROM bananas_are_awesome', [], ['create_missing_table' => FALSE]);
   }
 
   /**
diff --git a/core/tests/Drupal/Tests/Core/Database/Stub/StubConnection.php b/core/tests/Drupal/Tests/Core/Database/Stub/StubConnection.php
index d36a8d1..f9f2fb9 100644
--- a/core/tests/Drupal/Tests/Core/Database/Stub/StubConnection.php
+++ b/core/tests/Drupal/Tests/Core/Database/Stub/StubConnection.php
@@ -74,4 +74,9 @@ public function nextId($existing_id = 0) {
     return 0;
   }
 
+  /**
+   * {@inheritdoc}
+   */
+  public function isTableMissingException(\Exception $e) {
+  }
 }