diff --git a/core/core.services.yml b/core/core.services.yml
index d72d30c..f89f330 100644
--- a/core/core.services.yml
+++ b/core/core.services.yml
@@ -233,6 +233,9 @@ services:
   state:
     class: Drupal\Core\State\State
     arguments: ['@keyvalue']
+  schema:
+    class: Drupal\Core\Schema\Schema
+    arguments: ['@module_handler', '@cache.default', '@keyvalue', '@database']
   queue:
     class: Drupal\Core\Queue\QueueFactory
     arguments: ['@settings']
diff --git a/core/includes/schema.inc b/core/includes/schema.inc
index 4722cf8..9e8aed2 100644
--- a/core/includes/schema.inc
+++ b/core/includes/schema.inc
@@ -15,8 +15,10 @@
 
 /**
  * Indicates that a module has not been installed yet.
+ *
+ * @deprecated Use \Drupal\Core\Schema\SchemaInterface::UNINSTALLED.
  */
-const SCHEMA_UNINSTALLED = -1;
+const SCHEMA_UNINSTALLED = \Drupal\Core\Schema\SchemaInterface::UNINSTALLED;
 
 /**
  * Gets the schema definition of a table, or the whole database schema.
@@ -28,23 +30,11 @@
  *   The name of the table. If not given, the schema of all tables is returned.
  * @param bool $rebuild
  *   If TRUE, the schema will be rebuilt instead of retrieved from the cache.
+ *
+ * @deprecated Use \Drupal::service('schema')->get().
  */
 function drupal_get_schema($table = NULL, $rebuild = FALSE) {
-  static $schema;
-
-  if ($rebuild || !isset($schema)) {
-    $schema = drupal_get_complete_schema($rebuild);
-  }
-
-  if (!isset($table)) {
-    return $schema;
-  }
-  if (isset($schema[$table])) {
-    return $schema[$table];
-  }
-  else {
-    return FALSE;
-  }
+  return \Drupal::service('schema')->get($table, $rebuild);
 }
 
 /**
@@ -55,44 +45,11 @@ function drupal_get_schema($table = NULL, $rebuild = FALSE) {
  *
  * @param bool $rebuild
  *   If TRUE, the schema will be rebuilt instead of retrieved from the cache.
+ *
+ * @deprecated Use \Drupal::service('schema')->getComplete().
  */
 function drupal_get_complete_schema($rebuild = FALSE) {
-  static $schema;
-
-  if (!isset($schema) || $rebuild) {
-    // Try to load the schema from cache.
-    if (!$rebuild && $cached = \Drupal::cache()->get('schema')) {
-      $schema = $cached->data;
-    }
-    // Otherwise, rebuild the schema cache.
-    else {
-      $schema = array();
-      // Load the .install files to get hook_schema.
-      \Drupal::moduleHandler()->loadAllIncludes('install');
-
-      require_once __DIR__ . '/common.inc';
-      // Invoke hook_schema for all modules.
-      foreach (\Drupal::moduleHandler()->getImplementations('schema') as $module) {
-        // Cast the result of hook_schema() to an array, as a NULL return value
-        // would cause array_merge() to set the $schema variable to NULL as well.
-        // That would break modules which use $schema further down the line.
-        $current = (array) \Drupal::moduleHandler()->invoke($module, 'schema');
-        // Set 'module' and 'name' keys for each table, and remove descriptions,
-        // as they needlessly slow down \Drupal::cache()->get() for every single request.
-        _drupal_schema_initialize($current, $module);
-        $schema = array_merge($schema, $current);
-      }
-      \Drupal::moduleHandler()->alter('schema', $schema);
-
-      // If the schema is empty, avoid saving it: some database engines require
-      // the schema to perform queries, and this could lead to infinite loops.
-      if (!empty($schema)) {
-        \Drupal::cache()->set('schema', $schema, Cache::PERMANENT);
-      }
-    }
-  }
-
-  return $schema;
+  return \Drupal::service('schema')->getComplete($rebuild);
 }
 
 /**
@@ -104,38 +61,11 @@ function drupal_get_complete_schema($rebuild = FALSE) {
  * @return array|bool
  *   If the module has updates, an array of available updates sorted by
  *   version. Otherwise, FALSE.
+ *
+ * @deprecated Use \Drupal::service('schema')->getVersions().
  */
 function drupal_get_schema_versions($module) {
-  $updates = &drupal_static(__FUNCTION__, NULL);
-  if (!isset($updates[$module])) {
-    $updates = array();
-    foreach (\Drupal::moduleHandler()->getModuleList() as $loaded_module => $filename) {
-      $updates[$loaded_module] = array();
-    }
-
-    // Prepare regular expression to match all possible defined hook_update_N().
-    $regexp = '/^(?<module>.+)_update_(?<version>\d+)$/';
-    $functions = get_defined_functions();
-    // Narrow this down to functions ending with an integer, since all
-    // hook_update_N() functions end this way, and there are other
-    // possible functions which match '_update_'. We use preg_grep() here
-    // instead of foreaching through all defined functions, since the loop
-    // through all PHP functions can take significant page execution time
-    // and this function is called on every administrative page via
-    // system_requirements().
-    foreach (preg_grep('/_\d+$/', $functions['user']) as $function) {
-      // If this function is a module update function, add it to the list of
-      // module updates.
-      if (preg_match($regexp, $function, $matches)) {
-        $updates[$matches['module']][] = $matches['version'];
-      }
-    }
-    // Ensure that updates are applied in numerical order.
-    foreach ($updates as &$module_updates) {
-      sort($module_updates, SORT_NUMERIC);
-    }
-  }
-  return empty($updates[$module]) ? FALSE : $updates[$module];
+  return \Drupal::service('schema')->getVersions($module);
 }
 
 /**
@@ -152,26 +82,11 @@ function drupal_get_schema_versions($module) {
  * @return string|int
  *   The currently installed schema version, or SCHEMA_UNINSTALLED if the
  *   module is not installed.
+ *
+ * @deprecated Use \Drupal::service('schema')->getInstalledVersion().
  */
 function drupal_get_installed_schema_version($module, $reset = FALSE, $array = FALSE) {
-  static $versions = array();
-
-  if ($reset) {
-    $versions = array();
-  }
-
-  if (!$versions) {
-    if (!$versions = \Drupal::keyValue('system.schema')->getAll()) {
-      $versions = array();
-    }
-  }
-
-  if ($array) {
-    return $versions;
-  }
-  else {
-    return isset($versions[$module]) ? $versions[$module] : SCHEMA_UNINSTALLED;
-  }
+  return \Drupal::service('schema')->getInstalledVersion($module, $reset, $array);
 }
 
 /**
@@ -181,11 +96,11 @@ function drupal_get_installed_schema_version($module, $reset = FALSE, $array = F
  *   A module name.
  * @param string $version
  *   The new schema version.
+ *
+ * @deprecated Use \Drupal::service('schema')->setInstalledVersion().
  */
 function drupal_set_installed_schema_version($module, $version) {
-  \Drupal::keyValue('system.schema')->set($module, $version);
-  // Reset the static cache of module schema versions.
-  drupal_get_installed_schema_version(NULL, TRUE);
+  \Drupal::service('schema')->setInstalledVersion($module, $version);
 }
 
 /**
@@ -197,14 +112,11 @@ function drupal_set_installed_schema_version($module, $version) {
  *
  * @param string $module
  *   The module for which the tables will be created.
+ *
+ * @deprecated Use \Drupal::service('schema')->install().
  */
 function drupal_install_schema($module) {
-  $schema = drupal_get_schema_unprocessed($module);
-  _drupal_schema_initialize($schema, $module, FALSE);
-
-  foreach ($schema as $name => $table) {
-    db_create_table($name, $table);
-  }
+  \Drupal::service('schema')->install($module);
 }
 
 /**
@@ -222,16 +134,11 @@ function drupal_install_schema($module) {
  *    - success: a boolean indicating whether the query succeeded.
  *    - query: the SQL query(s) executed, passed through
  *      \Drupal\Component\Utility\String::checkPlain().
+ *
+ * @deprecated Use \Drupal::service('schema')->uninstall().
  */
 function drupal_uninstall_schema($module) {
-  $schema = drupal_get_schema_unprocessed($module);
-  _drupal_schema_initialize($schema, $module, FALSE);
-
-  foreach ($schema as $table) {
-    if (db_table_exists($table['name'])) {
-      db_drop_table($table['name']);
-    }
-  }
+  \Drupal::service('schema')->uninstall($module);
 }
 
 /**
@@ -257,53 +164,11 @@ function drupal_uninstall_schema($module) {
  * @param string $table
  *   The name of the table. If not given, the module's complete schema
  *   is returned.
- */
-function drupal_get_schema_unprocessed($module, $table = NULL) {
-  // Load the .install file to get hook_schema.
-  module_load_install($module);
-  $schema = \Drupal::moduleHandler()->invoke($module, 'schema');
-
-  if (isset($table)) {
-    if (isset($schema[$table])) {
-      return $schema[$table];
-    }
-    return array();
-  }
-  elseif (!empty($schema)) {
-    return $schema;
-  }
-  return array();
-}
-
-/**
- * Fills in required default values for table definitions from hook_schema().
  *
- * @param array $schema
- *   The schema definition array as it was returned by the module's
- *   hook_schema().
- * @param string $module
- *   The module for which hook_schema() was invoked.
- * @param bool $remove_descriptions
- *   (optional) Whether to additionally remove 'description' keys of all tables
- *   and fields to improve performance of serialize() and unserialize().
- *   Defaults to TRUE.
+ * @deprecated Use \Drupal::service('schema')->getUnprocessed().
  */
-function _drupal_schema_initialize(&$schema, $module, $remove_descriptions = TRUE) {
-  // Set the name and module key for all tables.
-  foreach ($schema as $name => &$table) {
-    if (empty($table['module'])) {
-      $table['module'] = $module;
-    }
-    if (!isset($table['name'])) {
-      $table['name'] = $name;
-    }
-    if ($remove_descriptions) {
-      unset($table['description']);
-      foreach ($table['fields'] as &$field) {
-        unset($field['description']);
-      }
-    }
-  }
+function drupal_get_schema_unprocessed($module, $table = NULL) {
+  return \Drupal::service('schema')->getUnprocessed($module, $table);
 }
 
 /**
@@ -320,21 +185,11 @@ function _drupal_schema_initialize(&$schema, $module, $remove_descriptions = TRU
  *
  * @return mixed
  *   The converted value.
+ *
+ * @deprecated Use \Drupal::service('schema')->getFieldValue().
  */
 function drupal_schema_get_field_value(array $info, $value) {
-  // Preserve legal NULL values.
-  if (isset($value) || !empty($info['not null'])) {
-    if ($info['type'] == 'int' || $info['type'] == 'serial') {
-      $value = (int) $value;
-    }
-    elseif ($info['type'] == 'float') {
-      $value = (float) $value;
-    }
-    elseif (!is_array($value)) {
-      $value = (string) $value;
-    }
-  }
-  return $value;
+  return \Drupal::service('schema')->getFieldValue($info, $value);
 }
 
 /**
diff --git a/core/includes/update.inc b/core/includes/update.inc
index a81d20f..a6fa9cb 100644
--- a/core/includes/update.inc
+++ b/core/includes/update.inc
@@ -103,7 +103,7 @@ function update_settings_file_requirements() {
 function update_system_schema_requirements() {
   $requirements = array();
 
-  $system_schema = drupal_get_installed_schema_version('system');
+  $system_schema = \Drupal::service('schema')->getInstalledVersion('system');
 
   $requirements['minimum schema']['title'] = 'Minimum schema version';
   if ($system_schema >= \Drupal::CORE_MINIMUM_SCHEMA_VERSION) {
@@ -245,7 +245,7 @@ function update_do_one($module, $number, $dependency_map, &$context) {
 
   // Record the schema update if it was completed successfully.
   if ($context['finished'] == 1 && empty($ret['#abort'])) {
-    drupal_set_installed_schema_version($module, $number);
+    \Drupal::service('schema')->setInstalledVersion($module, $number);
   }
 
   $context['message'] = 'Updating ' . String::checkPlain($module) . ' module';
@@ -300,7 +300,7 @@ function update_get_update_list() {
   // Make sure that the system module is first in the list of updates.
   $ret = array('system' => array());
 
-  $modules = drupal_get_installed_schema_version(NULL, FALSE, TRUE);
+  $modules = \Drupal::service('schema')->getInstalledVersion(NULL, FALSE, TRUE);
   foreach ($modules as $module => $schema_version) {
     // Skip uninstalled and incompatible modules.
     if ($schema_version == SCHEMA_UNINSTALLED || update_check_incompatibility($module)) {
@@ -313,7 +313,7 @@ function update_get_update_list() {
       continue;
     }
     // Otherwise, get the list of updates defined by this module.
-    $updates = drupal_get_schema_versions($module);
+    $updates = \Drupal::service('schema')->getVersions($module);
     if ($updates !== FALSE) {
       // \Drupal::moduleHandler()->invoke() returns NULL for non-existing hooks,
       // so if no updates are removed, it will == 0.
@@ -451,7 +451,7 @@ function update_get_update_function_list($starting_updates) {
   $update_functions = array();
   foreach ($starting_updates as $module => $version) {
     $update_functions[$module] = array();
-    $updates = drupal_get_schema_versions($module);
+    $updates = \Drupal::service('schema')->getVersions($module);
     if ($updates !== FALSE) {
       $max_version = max($updates);
       if ($version <= $max_version) {
@@ -581,7 +581,7 @@ function update_is_missing($module, $number, $update_functions) {
  *   performed; FALSE otherwise.
  */
 function update_already_performed($module, $number) {
-  return $number <= drupal_get_installed_schema_version($module);
+  return $number <= \Drupal::service('schema')->getInstalledVersion($module);
 }
 
 /**
diff --git a/core/lib/Drupal/Core/Extension/ModuleHandler.php b/core/lib/Drupal/Core/Extension/ModuleHandler.php
index 92e9748..2f3ac9a 100644
--- a/core/lib/Drupal/Core/Extension/ModuleHandler.php
+++ b/core/lib/Drupal/Core/Extension/ModuleHandler.php
@@ -801,14 +801,16 @@ public function install(array $module_list, $enable_dependencies = TRUE) {
           $kernel->updateModules($module_filenames, $module_filenames);
         }
 
+        /** @var \Drupal\Core\Schema\Schema $schema */
+        $schema = \Drupal::service('schema');
         // Refresh the schema to include it.
-        drupal_get_schema(NULL, TRUE);
+        $schema->get(NULL, TRUE);
 
         // Allow modules to react prior to the installation of a module.
         $this->invokeAll('module_preinstall', array($module));
 
         // Now install the module's schema if necessary.
-        drupal_install_schema($module);
+        $schema->install($module);
 
         // Clear plugin manager caches.
         \Drupal::getContainer()->get('plugin.cache_clearer')->clearCachedDefinitions();
@@ -816,7 +818,7 @@ public function install(array $module_list, $enable_dependencies = TRUE) {
         // Set the schema version to the number of the last update provided by
         // the module, or the minimum core schema version.
         $version = \Drupal::CORE_MINIMUM_SCHEMA_VERSION;
-        $versions = drupal_get_schema_versions($module);
+        $versions = $schema->getVersions($module);
         if ($versions) {
           $version = max(max($versions), $version);
         }
@@ -853,7 +855,7 @@ public function install(array $module_list, $enable_dependencies = TRUE) {
         if ($last_removed = $this->invoke($module, 'update_last_removed')) {
           $version = max($version, $last_removed);
         }
-        drupal_set_installed_schema_version($module, $version);
+        $schema->setInstalledVersion($module, $version);
 
         // Record the fact that it was installed.
         $modules_installed[] = $module;
@@ -937,6 +939,8 @@ public function uninstall(array $module_list, $uninstall_dependents = TRUE) {
     // necessarily installed.
     $schema_store = \Drupal::keyValue('system.schema');
     $entity_manager = \Drupal::entityManager();
+    /** @var \Drupal\Core\Schema\Schema $schema */
+    $schema = \Drupal::service('schema');
     foreach ($module_list as $module) {
 
       // Clean up all entity bundles (including field instances) of every entity
@@ -970,7 +974,7 @@ public function uninstall(array $module_list, $uninstall_dependents = TRUE) {
       }
 
       // Remove the schema.
-      drupal_uninstall_schema($module);
+      $schema->uninstall($module);
 
       // Remove the module's entry from the config.
       $extension_config->clear("module.$module")->save();
@@ -1008,7 +1012,7 @@ public function uninstall(array $module_list, $uninstall_dependents = TRUE) {
 
       $schema_store->delete($module);
     }
-    drupal_get_installed_schema_version(NULL, TRUE);
+    $schema->getInstalledVersion(NULL, TRUE);
 
     // Let other modules react.
     $this->invokeAll('modules_uninstalled', array($module_list));
diff --git a/core/lib/Drupal/Core/Field/FieldStorageDefinitionInterface.php b/core/lib/Drupal/Core/Field/FieldStorageDefinitionInterface.php
index 6ace3fa..a65dcbb 100644
--- a/core/lib/Drupal/Core/Field/FieldStorageDefinitionInterface.php
+++ b/core/lib/Drupal/Core/Field/FieldStorageDefinitionInterface.php
@@ -239,7 +239,7 @@ public function getSchema();
    *
    * @return array[]
    *   The array of field columns, keyed by column name, in the same format
-   *   returned by getSchema().
+   *   returned by get().
    *
    * @see \Drupal\Core\Field\FieldStorageDefinitionInterface::getSchema()
    */
diff --git a/core/lib/Drupal/Core/Installer/Form/SiteConfigureForm.php b/core/lib/Drupal/Core/Installer/Form/SiteConfigureForm.php
index c8fdd19..419796c 100644
--- a/core/lib/Drupal/Core/Installer/Form/SiteConfigureForm.php
+++ b/core/lib/Drupal/Core/Installer/Form/SiteConfigureForm.php
@@ -11,6 +11,7 @@
 use Drupal\Core\Form\FormBase;
 use Drupal\Core\Form\FormStateInterface;
 use Drupal\Core\Locale\CountryManagerInterface;
+use Drupal\Core\Schema\Schema;
 use Drupal\Core\State\StateInterface;
 use Drupal\user\UserStorageInterface;
 use Symfony\Component\DependencyInjection\ContainerInterface;
@@ -59,12 +60,15 @@ class SiteConfigureForm extends FormBase {
    *   The module handler.
    * @param \Drupal\Core\Locale\CountryManagerInterface $country_manager
    *   The country manager.
+   * @param \Drupal\Core\Schema\Schema $schema
+   *   The schema service.
    */
-  public function __construct(UserStorageInterface $user_storage, StateInterface $state, ModuleHandlerInterface $module_handler, CountryManagerInterface $country_manager) {
+  public function __construct(UserStorageInterface $user_storage, StateInterface $state, ModuleHandlerInterface $module_handler, CountryManagerInterface $country_manager, Schema $schema) {
     $this->userStorage = $user_storage;
     $this->state = $state;
     $this->moduleHandler = $module_handler;
     $this->countryManager = $country_manager;
+    $this->schema = $schema;
   }
 
   /**
@@ -75,7 +79,8 @@ public static function create(ContainerInterface $container) {
       $container->get('entity.manager')->getStorage('user'),
       $container->get('state'),
       $container->get('module_handler'),
-      $container->get('country_manager')
+      $container->get('country_manager'),
+      $container->get('schema')
     );
   }
 
@@ -122,7 +127,7 @@ public function buildForm(array $form, FormStateInterface $form_state) {
     // try to generate one but with no loaded modules will return nothing.
     //
     // @todo Move this to the 'install_finished' task?
-    drupal_get_schema(NULL, TRUE);
+    $this->schema->get(NULL, TRUE);
 
     $form['site_information'] = array(
       '#type' => 'fieldgroup',
diff --git a/core/lib/Drupal/Core/Schema/Schema.php b/core/lib/Drupal/Core/Schema/Schema.php
new file mode 100644
index 0000000..a9da688
--- /dev/null
+++ b/core/lib/Drupal/Core/Schema/Schema.php
@@ -0,0 +1,351 @@
+<?php
+
+/**
+ * @file
+ * Contains \Drupal\Core\Schema\Schema.
+ */
+
+namespace Drupal\Core\Schema;
+
+use Drupal\Core\Cache\Cache;
+use Drupal\Core\Cache\CacheBackendInterface;
+use Drupal\Core\Database\Connection;
+use Drupal\Core\Extension\ModuleHandlerInterface;
+use Drupal\Core\KeyValueStore\KeyValueFactoryInterface;
+
+/**
+ * Provides database schema handling.
+ */
+class Schema implements SchemaInterface {
+
+  /**
+   * Statically cached schema data.
+   *
+   * @var array
+   */
+  protected $schema;
+
+  /**
+   * Statically cached complete schema data.
+   *
+   * @var array
+   */
+  protected $completeSchema;
+
+  /**
+   * A static cache of schema versions per module.
+   *
+   * @var array
+   */
+  protected $updates;
+
+  /**
+   * A static cache of current module schema versions.
+   *
+   * @var array
+   */
+  protected $versions;
+
+  /**
+   * The module handler.
+   *
+   * @var \Drupal\Core\Extension\ModuleHandlerInterface
+   */
+  protected $moduleHandler;
+
+  /**
+   * The cache backend.
+   *
+   * @var \Drupal\Core\Cache\CacheBackendInterface
+   */
+  protected $cacheBackend;
+
+  /**
+   * The key value factory.
+   *
+   * @var \Drupal\Core\KeyValueStore\KeyValueFactory
+   */
+  protected $keyValue;
+
+  /**
+   * The database connection.
+   *
+   * @var \Drupal\Core\Database\Connection
+   */
+  protected $connection;
+
+  /**
+   * Constructs a Schema object.
+   *
+   * @param \Drupal\Core\Extension\ModuleHandlerInterface $module_handler
+   *   The module handler.
+   * @param \Drupal\Core\Cache\CacheBackendInterface $cache_backend
+   *   The cache backend.
+   * @param \Drupal\Core\KeyValueStore\KeyValueFactory $key_value_factory
+   *   The key value factory.
+   * @param \Drupal\Core\Database\Connection $connection
+   *   The database connection.
+   */
+  public function __construct(ModuleHandlerInterface $module_handler, CacheBackendInterface $cache_backend, KeyValueFactoryInterface $key_value_factory, Connection $connection) {
+    $this->moduleHandler = $module_handler;
+    $this->cacheBackend = $cache_backend;
+    $this->keyValue = $key_value_factory;
+    $this->connection = $connection;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function get($table = NULL, $rebuild = FALSE) {
+    if ($rebuild || !isset($table)) {
+      $this->schema = $this->getComplete($rebuild);
+    }
+
+    if (!isset($table)) {
+      return $this->schema;
+    }
+    if (isset($this->schema[$table])) {
+      return $this->schema[$table];
+    }
+    else {
+      return FALSE;
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getComplete($rebuild = FALSE) {
+    if (!isset($this->completeSchema) || $rebuild) {
+      // Try to load the schema from cache.
+      if (!$rebuild && ($cached = $this->cacheBackend->get('schema'))) {
+        $this->completeSchema = $cached->data;
+      }
+      // Otherwise, rebuild the schema cache.
+      else {
+        $this->completeSchema = array();
+        // Load the .install files to get hook_schema.
+        $this->moduleHandler->loadAllIncludes('install');
+
+        // Invoke hook_schema for all modules.
+        foreach ($this->moduleHandler->getImplementations('schema') as $module) {
+          // Cast the result of hook_schema() to an array, as a NULL return value
+          // would cause array_merge() to set the $schema variable to NULL as well.
+          // That would break modules which use $schema further down the line.
+          $current = (array) $this->moduleHandler->invoke($module, 'schema');
+          // Set 'module' and 'name' keys for each table, and remove descriptions,
+          // as they needlessly slow down cache()->get() for every single request.
+          $this->initialize($current, $module);
+          $this->completeSchema = array_merge($this->completeSchema, $current);
+        }
+
+        $this->moduleHandler->alter('schema', $this->completeSchema);
+
+        if ($rebuild) {
+          $this->cacheBackend->deleteTags(array('schema' => TRUE));
+        }
+        // If the schema is empty, avoid saving it: some database engines require
+        // the schema to perform queries, and this could lead to infinite loops.
+        if (!empty($this->completeSchema) && ($this->drupalGetBootstrapPhase() == DRUPAL_BOOTSTRAP_FULL)) {
+          $this->cacheBackend->set('schema', $this->completeSchema, Cache::PERMANENT, array('schema' => TRUE));
+        }
+      }
+    }
+
+    return $this->completeSchema;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getVersions($module) {
+    if (!isset($this->updates[$module])) {
+      $this->updates = array();
+      foreach ($this->moduleHandler->getModuleList() as $loaded_module => $filename) {
+        $this->updates[$loaded_module] = array();
+      }
+
+      // Prepare regular expression to match all possible defined hook_update_N().
+      $regexp = '/^(?<module>.+)_update_(?<version>\d+)$/';
+      $functions = get_defined_functions();
+      // Narrow this down to functions ending with an integer, since all
+      // hook_update_N() functions end this way, and there are other
+      // possible functions which match '_update_'. We use preg_grep() here
+      // instead of foreaching through all defined functions, since the loop
+      // through all PHP functions can take significant page execution time
+      // and this function is called on every administrative page via
+      // system_requirements().
+      foreach (preg_grep('/_\d+$/', $functions['user']) as $function) {
+        // If this function is a module update function, add it to the list of
+        // module updates.
+        if (preg_match($regexp, $function, $matches)) {
+          $this->updates[$matches['module']][] = $matches['version'];
+        }
+      }
+      // Ensure that updates are applied in numerical order.
+      foreach ($this->updates as &$module_updates) {
+        sort($module_updates, SORT_NUMERIC);
+      }
+    }
+
+    return empty($this->updates[$module]) ? FALSE : $this->updates[$module];
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getInstalledVersion($module, $reset = FALSE, $array = FALSE) {
+    if ($reset) {
+      $this->versions = array();
+    }
+
+    if (!$this->versions) {
+      if (!$this->versions = $this->keyValue->get('system.schema')->getAll()) {
+        $this->versions = array();
+      }
+    }
+
+    if ($array) {
+      return $this->versions;
+    }
+    else {
+      return isset($this->versions[$module]) ? $this->versions[$module] : static::UNINSTALLED;
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function setInstalledVersion($module, $version) {
+    $this->keyValue->get('system.schema')->set($module, $version);
+    // Reset the static cache of module schema versions.
+    $this->getInstalledVersion(NULL, TRUE);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function install($module) {
+    $schema = $this->getUnprocessed($module);
+    $this->initialize($schema, $module, FALSE);
+
+    foreach ($schema as $name => $table) {
+      $this->connection->schema()->createTable($name, $table);
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function uninstall($module) {
+    $schema = $this->getUnprocessed($module);
+    $this->initialize($schema, $module, FALSE);
+    $connection_schema = $this->connection->schema();
+
+    foreach ($schema as $table) {
+      if ($connection_schema->tableExists($table['name'])) {
+        $connection_schema->dropTable($table['name']);
+      }
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getUnprocessed($module, $table = NULL) {
+    // Load the .install file to get hook_schema.
+    $this->moduleHandler->loadInclude($module, 'install');
+    $schema = $this->moduleHandler->invoke($module, 'schema');
+
+    if (isset($table)) {
+      if (isset($schema[$table])) {
+        return $schema[$table];
+      }
+      return array();
+    }
+    elseif (!empty($schema)) {
+      return $schema;
+    }
+
+    return array();
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  function getFieldsSql($table, $prefix = NULL) {
+    if (!$schema = $this->get($table)) {
+      return array();
+    }
+    $fields = array_keys($schema['fields']);
+    if ($prefix) {
+      $columns = array();
+      foreach ($fields as $field) {
+        $columns[] = "$prefix.$field";
+      }
+      return $columns;
+    }
+    else {
+      return $fields;
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getFieldValue(array $info, $value) {
+    // Preserve legal NULL values.
+    if (isset($value) || !empty($info['not null'])) {
+      if ($info['type'] == 'int' || $info['type'] == 'serial') {
+        $value = (int) $value;
+      }
+      elseif ($info['type'] == 'float') {
+        $value = (float) $value;
+      }
+      else {
+        $value = (string) $value;
+      }
+    }
+
+    return $value;
+  }
+
+  /**
+   * Fills in required default values for table definitions from hook_schema().
+   *
+   * @param array $schema
+   *   The schema definition array as it was returned by the module's
+   *   hook_schema().
+   * @param string $module
+   *   The module for which hook_schema() was invoked.
+   * @param bool $remove_descriptions
+   *   (optional) Whether to additionally remove 'description' keys of all tables
+   *   and fields to improve performance of serialize() and unserialize().
+   *   Defaults to TRUE.
+   */
+  protected function initialize(&$schema, $module, $remove_descriptions = TRUE) {
+    // Set the name and module key for all tables.
+    foreach ($schema as $name => &$table) {
+      if (empty($table['module'])) {
+        $table['module'] = $module;
+      }
+      if (!isset($table['name'])) {
+        $table['name'] = $name;
+      }
+      if ($remove_descriptions) {
+        unset($table['description']);
+        foreach ($table['fields'] as &$field) {
+          unset($field['description']);
+        }
+      }
+    }
+  }
+
+  /**
+   * Wraps drupal_get_bootstrap_phase().
+   */
+  protected function drupalGetBootstrapPhase() {
+    return drupal_get_bootstrap_phase();
+  }
+
+}
diff --git a/core/lib/Drupal/Core/Schema/SchemaInterface.php b/core/lib/Drupal/Core/Schema/SchemaInterface.php
new file mode 100644
index 0000000..8b779d8
--- /dev/null
+++ b/core/lib/Drupal/Core/Schema/SchemaInterface.php
@@ -0,0 +1,172 @@
+<?php
+
+/**
+ * @file
+ * Contains \Drupal\Core\Schema\SchemaInterface.
+ */
+
+
+namespace Drupal\Core\Schema;
+
+/**
+ * Provides an interface for database schema handling.
+ */
+interface SchemaInterface {
+
+  /**
+   * Indicates that a module schema has not been installed yet.
+   */
+  const UNINSTALLED = -1;
+
+  /**
+   * Gets the schema definition of a table, or the whole database schema.
+   *
+   * The returned schema will include any modifications made by any
+   * module that implements hook_schema_alter().
+   *
+   * @param string $table
+   *   The name of the table. If not given, the schema of all tables is returned.
+   * @param bool $rebuild
+   *   If TRUE, the schema will be rebuilt instead of retrieved from the cache.
+   */
+  public function get($table = NULL, $rebuild = FALSE);
+
+  /**
+   * Gets the whole database schema.
+   *
+   * The returned schema will include any modifications made by any
+   * module that implements hook_schema_alter().
+   *
+   * @param bool $rebuild
+   *   If TRUE, the schema will be rebuilt instead of retrieved from the cache.
+   */
+  public function getComplete($rebuild = FALSE);
+
+  /**
+   * Returns an array of available schema versions for a module.
+   *
+   * @param string $module
+   *   A module name.
+   *
+   * @return array|bool
+   *   If the module has updates, an array of available updates sorted by
+   *   version. Otherwise, FALSE.
+   */
+  public function getVersions($module);
+
+  /**
+   * Returns the currently installed schema version for a module.
+   *
+   * @param string $module
+   *   A module name.
+   * @param bool $reset
+   *   Set to TRUE after installing or uninstalling an extension.
+   * @param bool $array
+   *   Set to TRUE if you want to get information about all modules in the
+   *   system.
+   *
+   * @return string|int
+   *   The currently installed schema version, or static::UNINSTALLED if the
+   *   module is not installed.
+   */
+  public function getInstalledVersion($module, $reset = FALSE, $array = FALSE);
+
+  /**
+   * Updates the installed version information for a module.
+   *
+   * @param string $module
+   *   A module name.
+   * @param string $version
+   *   The new schema version.
+   */
+  public function setInstalledVersion($module, $version);
+
+  /**
+   * Creates all tables defined in a module's hook_schema().
+   *
+   * Note: This function does not pass the module's schema through
+   * hook_schema_alter(). The module's tables will be created exactly as the
+   * module defines them.
+   *
+   * @param string $module
+   *   The module for which the tables will be created.
+   */
+  public function install($module);
+
+  /**
+   * Removes all tables defined in a module's hook_schema().
+   *
+   * Note: This function does not pass the module's schema through
+   * hook_schema_alter(). The module's tables will be created exactly as the
+   * module defines them.
+   *
+   * @param string $module
+   *   The module for which the tables will be removed.
+   *
+   * @return array
+   *   An array of arrays with the following key/value pairs:
+   *    - success: a boolean indicating whether the query succeeded.
+   *    - query: the SQL query(s) executed, passed through
+   *      \Drupal\Component\Utility\String::checkPlain().
+   */
+  public function uninstall($module);
+
+  /**
+   * Returns the unprocessed and unaltered version of a module's schema.
+   *
+   * Use this function only if you explicitly need the original
+   * specification of a schema, as it was defined in a module's
+   * hook_schema(). No additional default values will be set,
+   * hook_schema_alter() is not invoked and these unprocessed
+   * definitions won't be cached.
+   *
+   * This function can be used to retrieve a schema specification in
+   * hook_schema(), so it allows you to derive your tables from existing
+   * specifications.
+   *
+   * It is also used by drupal_install_schema() and
+   * drupal_uninstall_schema() to ensure that a module's tables are
+   * created exactly as specified without any changes introduced by a
+   * module that implements hook_schema_alter().
+   *
+   * @param string $module
+   *   The module to which the table belongs.
+   * @param string $table
+   *   The name of the table. If not given, the module's complete schema
+   *   is returned.
+   */
+  public function getUnprocessed($module, $table = NULL);
+
+  /**
+   * Retrieves a list of fields from a table schema.
+   *
+   * The returned list is suitable for use in an SQL query.
+   *
+   * @param string $table
+   *   The name of the table from which to retrieve fields.
+   * @param string $prefix
+   *   An optional prefix to to all fields.
+   *
+   * @return array
+   *   An array of fields.
+   */
+  function getFieldsSql($table, $prefix = NULL);
+
+  /**
+   * Typecasts values to proper datatypes.
+   *
+   * MySQL PDO silently casts, e.g. FALSE and '' to 0, when inserting the value
+   * into an integer column, but PostgreSQL PDO does not. Look up the schema
+   * information and use that to correctly typecast the value.
+   *
+   * @param array $info
+   *   An array describing the schema field info.
+   * @param mixed $value
+   *   The value to be converted.
+   *
+   * @return mixed
+   *   The converted value.
+   */
+  public function getFieldValue(array $info, $value);
+
+}
diff --git a/core/modules/simpletest/src/Tests/KernelTestBaseTest.php b/core/modules/simpletest/src/Tests/KernelTestBaseTest.php
index 73cee71..de3f86f 100644
--- a/core/modules/simpletest/src/Tests/KernelTestBaseTest.php
+++ b/core/modules/simpletest/src/Tests/KernelTestBaseTest.php
@@ -125,7 +125,7 @@ function testEnableModulesInstallContainer() {
   }
 
   /**
-   * Tests expected behavior of installSchema().
+   * Tests expected behavior of install().
    */
   function testInstallSchema() {
     $module = 'entity_test';
diff --git a/core/modules/system/src/Controller/DbUpdateController.php b/core/modules/system/src/Controller/DbUpdateController.php
index da7e4c3..08b890d 100644
--- a/core/modules/system/src/Controller/DbUpdateController.php
+++ b/core/modules/system/src/Controller/DbUpdateController.php
@@ -13,6 +13,7 @@
 use Drupal\Core\Extension\ModuleHandlerInterface;
 use Drupal\Core\KeyValueStore\KeyValueExpirableFactoryInterface;
 use Drupal\Core\Page\DefaultHtmlPageRenderer;
+use Drupal\Core\Schema\Schema;
 use Drupal\Core\Session\AccountInterface;
 use Drupal\Core\Site\Settings;
 use Drupal\Core\State\StateInterface;
@@ -69,6 +70,13 @@ class DbUpdateController extends ControllerBase {
   protected $entityDefinitionUpdateManager;
 
   /**
+   * The schema service.
+   *
+   * @var \Drupal\Core\Schema\Schema
+   */
+  protected $schema;
+
+  /**
    * Constructs a new UpdateController.
    *
    * @param \Drupal\Core\KeyValueStore\KeyValueExpirableFactoryInterface $key_value_expirable_factory
@@ -83,14 +91,17 @@ class DbUpdateController extends ControllerBase {
    *   The current user.
    * @param \Drupal\Core\Entity\EntityDefinitionUpdateManagerInterface $entity_definition_update_manager
    *   The entity definition update manager.
+   * @param \Drupal\Core\Schema\Schema $schema
+   *  The schema service.
    */
-  public function __construct(KeyValueExpirableFactoryInterface $key_value_expirable_factory, CacheBackendInterface $cache, StateInterface $state, ModuleHandlerInterface $module_handler, AccountInterface $account, EntityDefinitionUpdateManagerInterface $entity_definition_update_manager) {
+  public function __construct(KeyValueExpirableFactoryInterface $key_value_expirable_factory, CacheBackendInterface $cache, StateInterface $state, ModuleHandlerInterface $module_handler, AccountInterface $account, EntityDefinitionUpdateManagerInterface $entity_definition_update_manager, Schema $schema) {
     $this->keyValueExpirableFactory = $key_value_expirable_factory;
     $this->cache = $cache;
     $this->state = $state;
     $this->moduleHandler = $module_handler;
     $this->account = $account;
     $this->entityDefinitionUpdateManager = $entity_definition_update_manager;
+    $this->schema = $schema;
   }
 
   /**
@@ -103,7 +114,8 @@ public static function create(ContainerInterface $container) {
       $container->get('state'),
       $container->get('module_handler'),
       $container->get('current_user'),
-      $container->get('entity.definition_update_manager')
+      $container->get('entity.definition_update_manager'),
+      $container->get('schema')
     );
   }
 
@@ -560,7 +572,7 @@ protected function triggerBatch(Request $request) {
         // correct place. (The updates are already sorted, so we can simply base
         // this on the first one we come across in the above foreach loop.)
         if (isset($start[$update['module']])) {
-          drupal_set_installed_schema_version($update['module'], $update['number'] - 1);
+          $this->schema->setInstalledVersion($update['module'], $update['number'] - 1);
           unset($start[$update['module']]);
         }
         // Add this update function to the batch.
diff --git a/core/modules/views/tests/src/Unit/EntityViewsDataTest.php b/core/modules/views/tests/src/Unit/EntityViewsDataTest.php
index 0dc3cd4..1dcf633 100644
--- a/core/modules/views/tests/src/Unit/EntityViewsDataTest.php
+++ b/core/modules/views/tests/src/Unit/EntityViewsDataTest.php
@@ -205,23 +205,23 @@ public function testRevisionTable() {
   protected function setupFieldStorageDefinition() {
     $id_field_storage_definition = $this->getMock('Drupal\Core\Field\FieldStorageDefinitionInterface');
     $id_field_storage_definition->expects($this->any())
-      ->method('getSchema')
+      ->method('get')
       ->willReturn(IntegerItem::schema($id_field_storage_definition));
     $uuid_field_storage_definition = $this->getMock('Drupal\Core\Field\FieldStorageDefinitionInterface');
     $uuid_field_storage_definition->expects($this->any())
-      ->method('getSchema')
+      ->method('get')
       ->willReturn(UuidItem::schema($uuid_field_storage_definition));
     $type_field_storage_definition = $this->getMock('Drupal\Core\Field\FieldStorageDefinitionInterface');
     $type_field_storage_definition->expects($this->any())
-      ->method('getSchema')
+      ->method('get')
       ->willReturn(StringItem::schema($type_field_storage_definition));
     $langcode_field_storage_definition = $this->getMock('Drupal\Core\Field\FieldStorageDefinitionInterface');
     $langcode_field_storage_definition->expects($this->any())
-      ->method('getSchema')
+      ->method('get')
       ->willReturn(LanguageItem::schema($langcode_field_storage_definition));
     $name_field_storage_definition = $this->getMock('Drupal\Core\Field\FieldStorageDefinitionInterface');
     $name_field_storage_definition->expects($this->any())
-      ->method('getSchema')
+      ->method('get')
       ->willReturn(StringItem::schema($name_field_storage_definition));
 
     // Setup the user_id entity reference field.
@@ -237,12 +237,12 @@ protected function setupFieldStorageDefinition() {
       ->with('target_type')
       ->willReturn('user');
     $user_id_field_storage_definition->expects($this->any())
-      ->method('getSchema')
+      ->method('get')
       ->willReturn(EntityReferenceItem::schema($user_id_field_storage_definition));
 
     $revision_id_field_storage_definition = $this->getMock('Drupal\Core\Field\FieldStorageDefinitionInterface');
     $revision_id_field_storage_definition->expects($this->any())
-      ->method('getSchema')
+      ->method('get')
       ->willReturn(IntegerItem::schema($revision_id_field_storage_definition));
 
     $this->entityManager->expects($this->any())
diff --git a/core/tests/Drupal/Tests/Core/Entity/Sql/SqlContentEntityStorageSchemaTest.php b/core/tests/Drupal/Tests/Core/Entity/Sql/SqlContentEntityStorageSchemaTest.php
index 788744f..e41907d 100644
--- a/core/tests/Drupal/Tests/Core/Entity/Sql/SqlContentEntityStorageSchemaTest.php
+++ b/core/tests/Drupal/Tests/Core/Entity/Sql/SqlContentEntityStorageSchemaTest.php
@@ -81,7 +81,7 @@ protected function setUp() {
    * Tests the schema for non-revisionable, non-translatable entities.
    *
    * @covers ::__construct()
-   * @covers ::getSchema()
+   * @covers ::get()
    * @covers ::getEntitySchemaTables()
    * @covers ::initializeBaseTable()
    * @covers ::addTableDefaults()
@@ -389,7 +389,7 @@ public function testGetSchemaBase() {
    * Tests the schema for revisionable, non-translatable entities.
    *
    * @covers ::__construct()
-   * @covers ::getSchema()
+   * @covers ::get()
    * @covers ::getEntitySchemaTables()
    * @covers ::initializeBaseTable()
    * @covers ::initializeRevisionTable()
@@ -489,7 +489,7 @@ public function testGetSchemaRevisionable() {
    * Tests the schema for non-revisionable, translatable entities.
    *
    * @covers ::__construct()
-   * @covers ::getSchema()
+   * @covers ::get()
    * @covers ::getEntitySchemaTables()
    * @covers ::initializeDataTable()
    * @covers ::addTableDefaults()
@@ -579,7 +579,7 @@ public function testGetSchemaTranslatable() {
    * Tests the schema for revisionable, translatable entities.
    *
    * @covers ::__construct()
-   * @covers ::getSchema()
+   * @covers ::get()
    * @covers ::getEntitySchemaTables()
    * @covers ::initializeDataTable()
    * @covers ::addTableDefaults()
@@ -1120,7 +1120,7 @@ protected function setUpStorageSchema(array $expected = array()) {
    *   The field name.
    * @param array $schema
    *   The schema array of the field definition, as returned from
-   *   FieldStorageDefinitionInterface::getSchema().
+   *   FieldStorageDefinitionInterface::get().
    */
   public function setUpStorageDefinition($field_name, array $schema) {
     $this->storageDefinitions[$field_name] = $this->getMock('Drupal\Tests\Core\Field\TestBaseFieldDefinitionInterface');
@@ -1135,9 +1135,9 @@ public function setUpStorageDefinition($field_name, array $schema) {
     $this->storageDefinitions[$field_name]->expects($this->any())
       ->method('getDescription')
       ->will($this->returnValue("The $field_name field."));
-    // getSchema() is called once for each table.
+    // get() is called once for each table.
     $this->storageDefinitions[$field_name]->expects($this->any())
-      ->method('getSchema')
+      ->method('get')
       ->will($this->returnValue($schema));
     $this->storageDefinitions[$field_name]->expects($this->any())
       ->method('getColumns')
diff --git a/core/tests/Drupal/Tests/Core/Entity/Sql/SqlContentEntityStorageTest.php b/core/tests/Drupal/Tests/Core/Entity/Sql/SqlContentEntityStorageTest.php
index 6a6cef5..4228e78 100644
--- a/core/tests/Drupal/Tests/Core/Entity/Sql/SqlContentEntityStorageTest.php
+++ b/core/tests/Drupal/Tests/Core/Entity/Sql/SqlContentEntityStorageTest.php
@@ -285,7 +285,7 @@ public function testOnEntityTypeCreate() {
       ->method('getColumns')
       ->will($this->returnValue($columns));
     $this->fieldDefinitions['id']->expects($this->once())
-      ->method('getSchema')
+      ->method('get')
       ->will($this->returnValue(array('columns' => $columns)));
 
     $this->entityType->expects($this->once())