diff --git a/core/lib/Drupal/Core/Cache/DatabaseBackend.php b/core/lib/Drupal/Core/Cache/DatabaseBackend.php
index 813c00a..df2b876 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
@@ -74,13 +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 = array();
-    try {
-      $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)));
-    }
-    catch (\Exception $e) {
-      // Nothing to do.
-    }
+    $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.
@@ -151,23 +145,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);
-    }
   }
 
   /**
@@ -205,7 +182,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();
@@ -228,7 +205,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) {
@@ -298,21 +275,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();
     }
   }
 
@@ -329,16 +296,11 @@ public function deleteTags(array $tags) {
       }
       $deleted_tags[$tag] = TRUE;
       unset($tag_cache[$tag]);
-      try {
-        $this->connection->merge('cachetags')
-          ->insertFields(array('deletions' => 1))
-          ->expression('deletions', 'deletions + 1')
-          ->key('tag', $tag)
-          ->execute();
-      }
-      catch (\Exception $e) {
-        $this->catchException($e, 'cachetags');
-      }
+      $this->connection->merge('cachetags', array('schema_provider' => $this))
+        ->insertFields(array('deletions' => 1))
+        ->expression('deletions', 'deletions + 1')
+        ->key('tag', $tag)
+        ->execute();
     }
   }
 
@@ -346,17 +308,7 @@ public function deleteTags(array $tags) {
    * 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();
   }
 
   /**
@@ -371,17 +323,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();
     }
   }
 
@@ -389,25 +336,20 @@ public function invalidateMultiple(array $cids) {
    * Implements Drupal\Core\Cache\CacheBackendInterface::invalidateTags().
    */
   public function invalidateTags(array $tags) {
-    try {
-      $tag_cache = &drupal_static('Drupal\Core\Cache\CacheBackendInterface::tagCache', array());
-      $invalidated_tags = &drupal_static('Drupal\Core\Cache\DatabaseBackend::invalidatedTags', array());
-      foreach ($tags as $tag) {
-        // Only invalidate tags once per request unless they are written again.
-        if (isset($invalidated_tags[$tag])) {
-          continue;
-        }
-        $invalidated_tags[$tag] = TRUE;
-        unset($tag_cache[$tag]);
-        $this->connection->merge('cachetags')
-          ->insertFields(array('invalidations' => 1))
-          ->expression('invalidations', 'invalidations + 1')
-          ->key('tag', $tag)
-          ->execute();
+    $tag_cache = &drupal_static('Drupal\Core\Cache\CacheBackendInterface::tagCache', array());
+    $invalidated_tags = &drupal_static('Drupal\Core\Cache\DatabaseBackend::invalidatedTags', array());
+    foreach ($tags as $tag) {
+      // Only invalidate tags once per request unless they are written again.
+      if (isset($invalidated_tags[$tag])) {
+        continue;
       }
-    }
-    catch (\Exception $e) {
-      $this->catchException($e, 'cachetags');
+      $invalidated_tags[$tag] = TRUE;
+      unset($tag_cache[$tag]);
+      $this->connection->merge('cachetags', array('schema_provider' => $this))
+        ->insertFields(array('invalidations' => 1))
+        ->expression('invalidations', 'invalidations + 1')
+        ->key('tag', $tag)
+        ->execute();
     }
   }
 
@@ -415,14 +357,9 @@ public function invalidateTags(array $tags) {
    * 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();
   }
 
   /**
@@ -461,7 +398,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.
@@ -484,51 +421,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['bin']);
-        // If the bin doesn't exist, the cache tags table may also not exist.
-        if (!$database_schema->tableExists('cachetags')) {
-          $database_schema->createTable('cachetags', $schema_definition['cachetags']);
-        }
-        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 {cachetags} 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. Can be cachetags.
-   */
-  protected function catchException(\Exception $e, $table_name = NULL) {
-    if ($this->connection->schema()->tableExists($table_name ?: $this->bin)) {
-      throw $e;
-    }
   }
 
   /**
@@ -554,8 +450,8 @@ protected function normalizeCid($cid) {
   /**
    * Defines the schema for the {cache_*} bin and {cachetags} tables.
    */
-  public function schemaDefinition() {
-    $schema['bin'] = array(
+  public function getSchema($table_name) {
+    $schema[$this->bin] = array(
       'description' => 'Storage for the cache API.',
       'fields' => array(
         'cid' => array(
@@ -642,6 +538,6 @@ public function schemaDefinition() {
       ),
       'primary key' => array('tag'),
     );
-    return $schema;
+    return isset($schema[$table_name]) ? $schema[$table_name] : FALSE;
   }
 }
diff --git a/core/lib/Drupal/Core/Config/DatabaseStorage.php b/core/lib/Drupal/Core/Config/DatabaseStorage.php
index 471590a..adc1efd 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;
   }
 
@@ -87,36 +88,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 +105,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 +113,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 +142,7 @@ protected static function schemaDefinition() {
       ),
       'primary key' => array('collection', 'name'),
     );
-    return $schema;
+    return $this->table == $table_name ? $schema : FALSE;
   }
 
   /**
@@ -269,7 +198,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');
diff --git a/core/lib/Drupal/Core/Database/Connection.php b/core/lib/Drupal/Core/Database/Connection.php
index e0c1714..a9a4302 100644
--- a/core/lib/Drupal/Core/Database/Connection.php
+++ b/core/lib/Drupal/Core/Database/Connection.php
@@ -218,6 +218,8 @@ 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.
+   * - schema_provider: An object implementing
+   *   \Drupal\Core\Schema\SchemaProviderInterface.
    * - 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
@@ -576,6 +578,42 @@ public function query($query, array $args = array(), $options = array()) {
   }
 
   /**
+   * Handle a missing table exception.
+   *
+   * @param $e
+   * @param $query
+   * @param $args
+   * @param $options
+   * @return bool|StatementInterface
+   */
+  protected function handleMissingTable($e, $query, $args, $options) {
+    if ($this->isTableMissingException($e) && isset($options['schema_provider'])) {
+      if ($options['return'] == Database::RETURN_STATEMENT) {
+        // This is a SELECT on a missing table. Creating the table and
+        // re-runing the query will result in an empty result so return that
+        // early instead.
+        return new StatementEmpty();
+      }
+      if (isset($options['schema_provider'])) {
+        $table = $options['schema_provider'];
+        /** @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.
+            unset($options['schema_provider']);
+            return $this->query($query, $args, $options);
+          }
+        }
+      }
+      elseif ($options['schema_provider'] === TRUE) {
+        return FALSE;
+      }
+    }
+  }
+
+  /**
    * Expands out shorthand placeholders.
    *
    * Drupal supports an alternate syntax for doing arrays of values. We
@@ -1270,4 +1308,38 @@ public function unserialize($serialized) {
     }
   }
 
+  /**
+   * 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 64e1511..6843ade 100644
--- a/core/lib/Drupal/Core/Database/Driver/mysql/Connection.php
+++ b/core/lib/Drupal/Core/Database/Driver/mysql/Connection.php
@@ -271,6 +271,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 0945fd4..89b8224 100644
--- a/core/lib/Drupal/Core/Database/Driver/mysql/Schema.php
+++ b/core/lib/Drupal/Core/Database/Driver/mysql/Schema.php
@@ -514,7 +514,9 @@ public function fieldExists($table, $column) {
     // @todo This override should be removed once we require a version of MySQL
     //   that has that bug fixed.
     try {
-      $this->connection->queryRange("SELECT $column FROM {" . $table . "}", 0, 1);
+      if (isset($options['schema_provider'])) {
+        $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 cacae24..c7c68dd 100644
--- a/core/lib/Drupal/Core/Database/Driver/pgsql/Connection.php
+++ b/core/lib/Drupal/Core/Database/Driver/pgsql/Connection.php
@@ -324,6 +324,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 bca72c4..0209687 100644
--- a/core/lib/Drupal/Core/Database/Driver/sqlite/Connection.php
+++ b/core/lib/Drupal/Core/Database/Driver/sqlite/Connection.php
@@ -408,4 +408,11 @@ public function popTransaction($name) {
     }
   }
 
+  /**
+   * {@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 33727a7..920e70b 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/Merge.php b/core/lib/Drupal/Core/Database/Query/Merge.php
index be71de3..34b51a2 100644
--- a/core/lib/Drupal/Core/Database/Query/Merge.php
+++ b/core/lib/Drupal/Core/Database/Query/Merge.php
@@ -437,12 +437,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 +461,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/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 b588756..2fedd97 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;
 
@@ -684,6 +683,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/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 0898091..6a1e6af 100644
--- a/core/lib/Drupal/Core/Menu/MenuTreeStorage.php
+++ b/core/lib/Drupal/Core/Menu/MenuTreeStorage.php
@@ -14,13 +14,12 @@
 use Drupal\Core\Cache\CacheBackendInterface;
 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.
@@ -118,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];
   }
 
   /**
@@ -209,33 +208,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) {
@@ -322,7 +294,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'])) {
@@ -443,7 +415,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');
@@ -453,7 +425,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;
   }
@@ -578,7 +550,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
@@ -620,7 +592,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)) {
@@ -629,7 +601,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);
     }
@@ -647,7 +619,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);
@@ -658,7 +630,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);
     }
@@ -672,10 +644,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);
       }
@@ -718,10 +690,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]);
@@ -734,7 +706,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
@@ -743,13 +715,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();
   }
@@ -761,7 +733,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);
@@ -769,7 +741,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;
@@ -847,7 +819,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
@@ -927,7 +899,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;
   }
@@ -990,32 +962,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();
   }
 
   /**
@@ -1026,7 +998,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++) {
@@ -1034,7 +1006,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);
   }
 
   /**
@@ -1114,34 +1086,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
@@ -1149,7 +1093,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;
@@ -1175,7 +1119,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(
@@ -1422,7 +1366,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 e2b9d37..3a0b616 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;
@@ -48,7 +49,9 @@ public function testTagInvalidations() {
       $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 +62,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.');
   }
 
   public function testTagDeletions() {
@@ -72,7 +75,7 @@ public function testTagDeletions() {
       $this->assertTrue($bin->get('test'), 'Cache item was set in bin.');
     }
 
-    $deletions_before = intval(db_select('cachetags')->fields('cachetags', array('deletions'))->condition('tag', 'test_tag:2')->execute()->fetchField());
+    $this->assertFalse(Database::getConnection()->schema()->tableExists('cachetags'));
     Cache::deleteTags(array('test_tag:2'));
 
     // Test that cache entry has been deleted in multiple bins.
@@ -83,7 +86,7 @@ public function testTagDeletions() {
 
     // Test that only one tag deletion has occurred.
     $deletions_after = intval(db_select('cachetags')->fields('cachetags', array('deletions'))->condition('tag', 'test_tag:2')->execute()->fetchField());
-    $this->assertEqual($deletions_after, $deletions_before + 1, 'Only one addition cache tag deletion has occurred after deleting a tag used in multiple bins.');
+    $this->assertEqual($deletions_after, 1, 'Only one addition cache tag deletion has occurred after deleting 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 9009e72..79f860a 100644
--- a/core/modules/system/src/Tests/Database/SchemaTest.php
+++ b/core/modules/system/src/Tests/Database/SchemaTest.php
@@ -135,6 +135,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 26dd109..fc13a43 100644
--- a/core/modules/system/src/Tests/Database/SelectTest.php
+++ b/core/modules/system/src/Tests/Database/SelectTest.php
@@ -6,14 +6,16 @@
  */
 
 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 +30,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() {
@@ -494,4 +520,13 @@ 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 39c403c..f19e96e 100644
--- a/core/modules/system/src/Tests/Menu/MenuTreeStorageTest.php
+++ b/core/modules/system/src/Tests/Menu/MenuTreeStorageTest.php
@@ -76,7 +76,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'), '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');
   }
 
@@ -329,17 +329,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);
   }
 
@@ -412,4 +402,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/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) {
+  }
 }