diff --git a/core/lib/Drupal/Core/Entity/Sql/SqlContentEntityStorageSchema.php b/core/lib/Drupal/Core/Entity/Sql/SqlContentEntityStorageSchema.php
index 3349ebd30d..bfcab8b53f 100644
--- a/core/lib/Drupal/Core/Entity/Sql/SqlContentEntityStorageSchema.php
+++ b/core/lib/Drupal/Core/Entity/Sql/SqlContentEntityStorageSchema.php
@@ -2,6 +2,7 @@
 
 namespace Drupal\Core\Entity\Sql;
 
+use Drupal\Component\Utility\NestedArray;
 use Drupal\Core\Database\Connection;
 use Drupal\Core\Database\DatabaseExceptionWrapper;
 use Drupal\Core\DependencyInjection\DependencySerializationTrait;
@@ -717,7 +718,9 @@ protected function getEntitySchema(ContentEntityTypeInterface $entity_type, $res
           elseif ($table_mapping->allowsSharedTableStorage($this->fieldStorageDefinitions[$field_name])) {
             $column_names = $table_mapping->getColumnNames($field_name);
             $storage_definition = $this->fieldStorageDefinitions[$field_name];
-            $schema[$table_name] = array_merge_recursive($schema[$table_name], $this->getSharedTableFieldSchema($storage_definition, $table_name, $column_names));
+            // Preserve integer keys when merging the schema so that composite
+            // primary keys are not merged.
+            $schema[$table_name] = NestedArray::mergeDeepArray([$schema[$table_name], $this->getSharedTableFieldSchema($storage_definition, $table_name, $column_names)], TRUE);
           }
         }
       }
@@ -1050,15 +1053,12 @@ protected function initializeBaseTable(ContentEntityTypeInterface $entity_type)
 
     $schema = [
       'description' => "The base table for $entity_type_id entities.",
-      'primary key' => [$entity_type->getKey('id')],
       'indexes' => [],
       'foreign keys' => [],
     ];
 
     if ($entity_type->hasKey('revision')) {
       $revision_key = $entity_type->getKey('revision');
-      $key_name = $this->getEntityIndexName($entity_type, $revision_key);
-      $schema['unique keys'][$key_name] = [$revision_key];
       $schema['foreign keys'][$entity_type_id . '__revision'] = [
         'table' => $this->storage->getRevisionTable(),
         'columns' => [$revision_key => $revision_key],
@@ -1082,11 +1082,9 @@ protected function initializeBaseTable(ContentEntityTypeInterface $entity_type)
   protected function initializeRevisionTable(ContentEntityTypeInterface $entity_type) {
     $entity_type_id = $entity_type->id();
     $id_key = $entity_type->getKey('id');
-    $revision_key = $entity_type->getKey('revision');
 
     $schema = [
       'description' => "The revision table for $entity_type_id entities.",
-      'primary key' => [$revision_key],
       'indexes' => [],
       'foreign keys' => [
         $entity_type_id . '__revisioned' => [
@@ -1096,8 +1094,6 @@ protected function initializeRevisionTable(ContentEntityTypeInterface $entity_ty
       ],
     ];
 
-    $schema['indexes'][$this->getEntityIndexName($entity_type, $id_key)] = [$id_key];
-
     $this->addTableDefaults($schema);
 
     return $schema;
@@ -1118,10 +1114,6 @@ protected function initializeDataTable(ContentEntityTypeInterface $entity_type)
 
     $schema = [
       'description' => "The data table for $entity_type_id entities.",
-      'primary key' => [$id_key, $entity_type->getKey('langcode')],
-      'indexes' => [
-        $entity_type_id . '__id__default_langcode__langcode' => [$id_key, $entity_type->getKey('default_langcode'), $entity_type->getKey('langcode')],
-      ],
       'foreign keys' => [
         $entity_type_id => [
           'table' => $this->storage->getBaseTable(),
@@ -1130,11 +1122,6 @@ protected function initializeDataTable(ContentEntityTypeInterface $entity_type)
       ],
     ];
 
-    if ($entity_type->hasKey('revision')) {
-      $key = $entity_type->getKey('revision');
-      $schema['indexes'][$this->getEntityIndexName($entity_type, $key)] = [$key];
-    }
-
     $this->addTableDefaults($schema);
 
     return $schema;
@@ -1156,10 +1143,6 @@ protected function initializeRevisionDataTable(ContentEntityTypeInterface $entit
 
     $schema = [
       'description' => "The revision data table for $entity_type_id entities.",
-      'primary key' => [$revision_key, $entity_type->getKey('langcode')],
-      'indexes' => [
-        $entity_type_id . '__id__default_langcode__langcode' => [$id_key, $entity_type->getKey('default_langcode'), $entity_type->getKey('langcode')],
-      ],
       'foreign keys' => [
         $entity_type_id => [
           'table' => $this->storage->getBaseTable(),
@@ -1286,6 +1269,8 @@ protected function processFieldStorageSchema(array &$field_storage_schema) {
         unset($field_storage_schema[$table_name]['fields'][$key]['initial']);
         unset($field_storage_schema[$table_name]['fields'][$key]['initial_from_field']);
       }
+
+      $this->addTableDefaults($field_storage_schema[$table_name]);
     }
   }
 
@@ -1358,23 +1343,49 @@ protected function createSharedTableSchema(FieldStorageDefinitionInterface $stor
           // Create field columns.
           $schema[$table_name] = $this->getSharedTableFieldSchema($storage_definition, $table_name, $column_names);
           if (!$only_save) {
+            // The entity schema needs to be checked because the field schema is
+            // potentially incomplete.
             foreach ($schema[$table_name]['fields'] as $name => $specifier) {
+              // Check if the field is part of the primary keys and pass along
+              // this information when adding the field.
+              // @see \Drupal\Core\Database\Schema::addField()
+              $new_keys = [];
+              if (!empty($schema[$table_name]['primary key']) && in_array($name, $schema[$table_name]['primary key'], TRUE)) {
+                $new_keys = ['primary key' => $schema[$table_name]['primary key']];
+
+                // If the field is part of a primary key, it might be a serial
+                // field.
+                if (($specifier['type'] === 'int') && $this->isSerialField($schema[$table_name], $table_name, $created_field_name)) {
+                  $specifier['type'] = 'serial';
+                }
+              }
+
               // Check if the field exists because it might already have been
               // created as part of the earlier entity type update event.
               if (!$schema_handler->fieldExists($table_name, $name)) {
-                $schema_handler->addField($table_name, $name, $specifier);
+                $schema_handler->addField($table_name, $name, $specifier, $new_keys);
               }
             }
             if (!empty($schema[$table_name]['indexes'])) {
               foreach ($schema[$table_name]['indexes'] as $name => $specifier) {
+                // When creating a composite index, the field schema of all
+                // participating fields must be passed to the schema handler.
+                $index_schema = $schema[$table_name];
+                foreach ($specifier as $index_key => $index_field) {
+                  // Get the name of the field from the index specification.
+                  $other_column_name = is_array($index_field) ? $index_field[0] : $index_field;
+                  if (!isset($index_schema['fields'][$other_column_name])) {
+                    $this->mergeFieldSchema($index_schema, $table_mapping, $table_name, $other_column_name);
+                  }
+                }
                 // Check if the index exists because it might already have been
                 // created as part of the earlier entity type update event.
-                $this->addIndex($table_name, $name, $specifier, $schema[$table_name]);
+                $this->addIndex($table_name, $name, $specifier, $index_schema);
               }
             }
             if (!empty($schema[$table_name]['unique keys'])) {
               foreach ($schema[$table_name]['unique keys'] as $name => $specifier) {
-                $schema_handler->addUniqueKey($table_name, $name, $specifier);
+                $this->addUniqueKey($table_name, $name, $specifier);
               }
             }
           }
@@ -1439,7 +1450,9 @@ protected function deleteSharedTableSchema(FieldStorageDefinitionInterface $stor
         if ($field_name == $deleted_field_name) {
           $schema = $this->getSharedTableFieldSchema($storage_definition, $table_name, $column_names);
 
-          // Drop indexes and unique keys first.
+          // Drop the primary key, indexes and unique keys first.
+          $this->dropPrimaryKey($storage_definition, $table_name, $schema, $column_names);
+
           if (!empty($schema['indexes'])) {
             foreach ($schema['indexes'] as $name => $specifier) {
               $schema_handler->dropIndex($table_name, $name);
@@ -1625,7 +1638,9 @@ protected function updateSharedTableSchema(FieldStorageDefinitionInterface $stor
               }
             }
 
-            // Drop original indexes and unique keys.
+            // Drop the original primary key, indexes and unique keys first.
+            $this->dropPrimaryKey($storage_definition, $table_name, $schema[$table_name], $column_names);
+
             if (!empty($original_schema[$table_name]['indexes'])) {
               foreach ($original_schema[$table_name]['indexes'] as $name => $specifier) {
                 $schema_handler->dropIndex($table_name, $name);
@@ -1636,18 +1651,50 @@ protected function updateSharedTableSchema(FieldStorageDefinitionInterface $stor
                 $schema_handler->dropUniqueKey($table_name, $name);
               }
             }
-            // Create new indexes and unique keys.
+            // Create new primary key, indexes and unique keys.
+            foreach ($schema[$table_name]['fields'] as $name => $specifier) {
+              // Check if the field is part of the primary keys.
+              if (isset($schema[$table_name]['primary key']) && array_intersect($column_names, $schema[$table_name]['primary key'])) {
+                $schema_handler->addPrimaryKey($table_name, $schema[$table_name]['primary key']);
+
+                // If the field being updated is a serial field, it will have
+                // been changed to an integer field by
+                // SqlContentEntityStorageSchema::dropPrimaryKey() above.
+                // Convert it back now that the primary key has been recreated.
+                if ($this->isSerialField($schema[$table_name], $table_name, $updated_field_name)) {
+                  foreach ($schema[$table_name]['primary key'] as $column_name) {
+                    $spec = $schema[$table_name]['fields'][$column_name];
+                    if ($spec['type'] === 'int') {
+                      $spec['type'] = 'serial';
+                      $schema_handler->changeField($table_name, $column_name, $column_name, $spec);
+                    }
+                  }
+                }
+
+                break;
+              }
+            }
             if (!empty($schema[$table_name]['indexes'])) {
               foreach ($schema[$table_name]['indexes'] as $name => $specifier) {
+                // When creating a composite index, the field schema of all
+                // participating fields must be passed to the schema handler.
+                $index_schema = $schema[$table_name];
+                foreach ($specifier as $index_key => $index_field) {
+                  // Get the name of the field from the index specification.
+                  $other_column_name = is_array($index_field) ? $index_field[0] : $index_field;
+                  if (!isset($index_schema['fields'][$other_column_name])) {
+                    $this->mergeFieldSchema($index_schema, $table_mapping, $table_name, $other_column_name);
+                  }
+                }
                 // Check if the index exists because it might already have been
                 // created as part of the earlier entity type update event.
-                $this->addIndex($table_name, $name, $specifier, $schema[$table_name]);
+                $this->addIndex($table_name, $name, $specifier, $index_schema);
 
               }
             }
             if (!empty($schema[$table_name]['unique keys'])) {
               foreach ($schema[$table_name]['unique keys'] as $name => $specifier) {
-                $schema_handler->addUniqueKey($table_name, $name, $specifier);
+                $this->addUniqueKey($table_name, $name, $specifier);
               }
             }
             // After deleting the field schema skip to the next table.
@@ -1898,6 +1945,53 @@ protected function getSharedTableFieldSchema(FieldStorageDefinitionInterface $st
       $schema['foreign keys'] = $this->getFieldForeignKeys($field_name, $field_schema, $column_mapping);
     }
 
+    // Add primary keys and indexes for key fields.
+    if ($table_name === $this->storage->getBaseTable()) {
+      if ($field_name === $this->entityType->getKey('id')) {
+        $schema['primary key'] = [$this->entityType->getKey('id')];
+      }
+      elseif ($field_name === $this->entityType->getKey('revision')) {
+        $key_name = $this->getEntityIndexName($this->entityType, $this->entityType->getKey('revision'));
+        $schema['unique keys'][$key_name] = [$this->entityType->getKey('revision')];
+      }
+    }
+    elseif ($table_name === $this->storage->getRevisionTable()) {
+      if ($field_name === $this->entityType->getKey('id')) {
+        $schema['indexes'][$this->getEntityIndexName($this->entityType, $this->entityType->getKey('id'))] = [$field_name];
+      }
+      elseif ($field_name === $this->entityType->getKey('revision')) {
+        $schema['primary key'] = [$this->entityType->getKey('revision')];
+      }
+    }
+    elseif ($table_name === $this->storage->getDataTable()) {
+      if (in_array($field_name, [$this->entityType->getKey('id'), $this->entityType->getKey('langcode')], TRUE)) {
+        $schema['primary key'] = [
+          $this->entityType->getKey('id'),
+          $this->entityType->getKey('langcode'),
+        ];
+      }
+      elseif ($field_name === $this->entityType->getKey('revision')) {
+        $key_name = $this->getEntityIndexName($this->entityType, $this->entityType->getKey('revision'));
+        $schema['indexes'][$key_name] = [$this->entityType->getKey('revision')];
+      }
+    }
+    elseif ($table_name === $this->storage->getRevisionDataTable()) {
+      if (in_array($field_name, [$this->entityType->getKey('revision'), $this->entityType->getKey('langcode')], TRUE)) {
+        $schema['primary key'] = [$this->entityType->getKey('revision'), $this->entityType->getKey('langcode')];
+      }
+    }
+    if (in_array($table_name, [$this->storage->getDataTable(), $this->storage->getRevisionDataTable()], TRUE)) {
+      if (in_array($field_name, [$this->entityType->getKey('id'), $this->entityType->getKey('default_langcode'), $this->entityType->getKey('langcode')], TRUE)) {
+        $schema['indexes'][$this->entityType->id() . '__id__default_langcode__langcode'] = [
+          $this->entityType->getKey('id'),
+          $this->entityType->getKey('default_langcode'),
+          $this->entityType->getKey('langcode'),
+        ];
+      }
+    }
+
+    $this->addTableDefaults($schema);
+
     return $schema;
   }
 
@@ -2148,6 +2242,34 @@ protected function getDedicatedTableSchema(FieldStorageDefinitionInterface $stor
     return $dedicated_table_schema;
   }
 
+  /**
+   * Merges schema with the field schema for a given column.
+   *
+   * This looks up the field that the column belongs to and merges the field
+   * schema of that field into the passed-in schema.
+   *
+   * @param array $schema
+   *   The schema to merge the other field schema to.
+   * @param \Drupal\Core\Entity\Sql\TableMappingInterface $table_mapping
+   *   The table mapping.
+   * @param string $table_name
+   *   The table name.
+   * @param string $column_name
+   *   The column name to merge the field schema for.
+   */
+  protected function mergeFieldSchema(&$schema, TableMappingInterface $table_mapping, $table_name, $column_name) {
+    foreach ($table_mapping->getFieldNames($table_name) as $search_field_name) {
+      $search_column_names = $table_mapping->getColumnNames($search_field_name);
+      foreach ($search_column_names as $search_column_name) {
+        if ($search_column_name === $column_name) {
+          $other_storage_definition = $this->fieldStorageDefinitions[$search_field_name];
+          $other_schema = $this->getSharedTableFieldSchema($other_storage_definition, $table_name, $search_column_names);
+          $schema = NestedArray::mergeDeepArray([$schema, $other_schema], TRUE);
+        }
+      }
+    }
+  }
+
   /**
    * Gets the name to be used for the given entity index.
    *
@@ -2286,4 +2408,69 @@ protected function addUniqueKey($table, $name, array $specifier) {
     $schema_handler->addUniqueKey($table, $name, $specifier);
   }
 
+  /**
+   * Removes the primary key of a table if the passed columns are part of it.
+   *
+   * @param \Drupal\Core\Field\FieldStorageDefinitionInterface $storage_definition
+   *   The storage definition the primary key belongs to.
+   * @param string $table_name
+   *   The name of the table whose primary key is to be deleted.
+   * @param array $schema
+   *   The field schema for this table.
+   * @param string[] $column_names
+   *   The list of columns to check whether they are part of the primary key.
+   *
+   * @internal
+   */
+  private function dropPrimaryKey(FieldStorageDefinitionInterface $storage_definition, $table_name, array $schema, $column_names) {
+    $schema_handler = $this->database->schema();
+    if (!empty($schema['primary key']) && array_intersect($column_names, $schema['primary key'])) {
+      // Dropping a primary key must not leave a serial key, so we need to
+      // change any serial field to integer first.
+      $field_name = $storage_definition->getName();
+      if ($this->isSerialField($schema, $table_name, $field_name)) {
+        foreach ($schema['primary key'] as $column_name) {
+          if ($schema['fields'][$column_name]['type'] === 'int') {
+            $schema_handler->changeField($table_name, $column_name, $column_name, $schema['fields'][$column_name]);
+          }
+        }
+      }
+      $schema_handler->dropPrimaryKey($table_name);
+    }
+  }
+
+  /**
+   * Checks whether the a field is a serial field in a given table.
+   *
+   * Because the field schema is incorrectly stored as 'int' we have to manually
+   * call SqlContentEntityStorageSchema::processIdentifierSchema() assuming the
+   * conditions are correct.
+   *
+   * @param array $schema
+   *   The table schema.
+   * @param $table_name
+   *   The table name.
+   * @param $field_name
+   *   The field name.
+   *
+   * @return bool
+   *   TRUE if the field is a serial field in the given table; FALSE otherwise.
+   *
+   * @internal
+   *
+   * @todo Remove this in https://www.drupal.org/project/drupal/issues/2928906
+   */
+  private function isSerialField(array $schema, $table_name, $field_name) {
+    // The ID field is serial in the base table.
+    $is_base_table = ($table_name === $this->storage->getBaseTable());
+    $is_revision_table = ($table_name === $this->storage->getRevisionTable());
+    $is_id = ($field_name === $this->entityType->getKey('id'));
+    $is_revision_id = ($field_name === $this->entityType->getKey('revision'));
+    if (($is_base_table && $is_id) || ($is_revision_table && $is_revision_id)) {
+      $this->processIdentifierSchema($schema, $field_name);
+      return $schema['fields'][$field_name]['type'] === 'serial';
+    }
+    return FALSE;
+  }
+
 }
diff --git a/core/modules/node/node.install b/core/modules/node/node.install
index 6469245c49..651928462d 100644
--- a/core/modules/node/node.install
+++ b/core/modules/node/node.install
@@ -255,3 +255,24 @@ function node_update_8400() {
   $schema['fields']['realm']['description'] = 'The realm in which the user must possess the grant ID. Modules can define one or more realms by implementing hook_node_grants().';
   Database::getConnection()->schema()->changeField('node_access', 'realm', 'realm', $schema['fields']['realm']);
 }
+
+/**
+ * Update the stored schema data of various node fields.
+ */
+function node_update_8601() {
+  $definition_update_manager = \Drupal::entityDefinitionUpdateManager();
+
+  $field_names = [
+    'changed',
+    'created',
+    'promote',
+    'status',
+    'sticky',
+    'title',
+    'type',
+  ];
+  foreach ($field_names as $field_name) {
+    $field_storage_definition = $definition_update_manager->getFieldStorageDefinition($field_name, 'node');
+    $definition_update_manager->updateFieldStorageDefinition($field_storage_definition);
+  }
+}
diff --git a/core/modules/node/src/NodeStorageSchema.php b/core/modules/node/src/NodeStorageSchema.php
index ac45bb5d1e..b0c4f17b14 100644
--- a/core/modules/node/src/NodeStorageSchema.php
+++ b/core/modules/node/src/NodeStorageSchema.php
@@ -2,7 +2,6 @@
 
 namespace Drupal\node;
 
-use Drupal\Core\Entity\ContentEntityTypeInterface;
 use Drupal\Core\Entity\Sql\SqlContentEntityStorageSchema;
 use Drupal\Core\Field\FieldStorageDefinitionInterface;
 
@@ -11,20 +10,6 @@
  */
 class NodeStorageSchema extends SqlContentEntityStorageSchema {
 
-  /**
-   * {@inheritdoc}
-   */
-  protected function getEntitySchema(ContentEntityTypeInterface $entity_type, $reset = FALSE) {
-    $schema = parent::getEntitySchema($entity_type, $reset);
-
-    $schema['node_field_data']['indexes'] += [
-      'node__frontpage' => ['promote', 'status', 'sticky', 'created'],
-      'node__title_type' => ['title', ['type', 4]],
-    ];
-
-    return $schema;
-  }
-
   /**
    * {@inheritdoc}
    */
@@ -45,22 +30,17 @@ protected function getSharedTableFieldSchema(FieldStorageDefinitionInterface $st
     }
 
     if ($table_name == 'node_field_data') {
-      switch ($field_name) {
-        case 'promote':
-        case 'status':
-        case 'sticky':
-        case 'title':
-          // Improves the performance of the indexes defined
-          // in getEntitySchema().
-          $schema['fields'][$field_name]['not null'] = TRUE;
-          break;
-
-        case 'changed':
-        case 'created':
-          // @todo Revisit index definitions:
-          //   https://www.drupal.org/node/2015277.
-          $this->addSharedTableFieldIndex($storage_definition, $schema, TRUE);
-          break;
+      if (in_array($field_name, ['changed', 'created'], TRUE)) {
+        $this->addSharedTableFieldIndex($storage_definition, $schema, TRUE);
+      }
+      if (in_array($field_name, ['promote', 'status', 'sticky', 'title'], TRUE)) {
+        $schema['fields'][$field_name]['not null'] = TRUE;
+      }
+      if (in_array($field_name, ['promote', 'status', 'sticky', 'created'], TRUE)) {
+        $schema['node_field_data']['indexes']['node__frontpage'] = ['promote', 'status', 'sticky', 'created'];
+      }
+      if (in_array($field_name, ['title', 'type'], TRUE)) {
+        $schema['node_field_data']['indexes']['node__title_type'] = ['title', ['type', 4]];
       }
     }
 
diff --git a/core/modules/system/system.install b/core/modules/system/system.install
index 1838326451..b38c2437c8 100644
--- a/core/modules/system/system.install
+++ b/core/modules/system/system.install
@@ -2135,3 +2135,33 @@ function system_update_8501() {
     }
   }
 }
+
+/**
+ * Update the stored schema data of ID and language fields.
+ */
+function system_update_8601() {
+  $entity_type_manager = \Drupal::entityTypeManager();
+  $definition_update_manager = \Drupal::entityDefinitionUpdateManager();
+  /** @var \Drupal\Core\Entity\EntityLastInstalledSchemaRepositoryInterface $entity_last_installed_schema_respository */
+  $entity_last_installed_schema_respository = \Drupal::service('entity.last_installed_schema.repository');
+
+  foreach ($entity_type_manager->getDefinitions() as $entity_type_id => $entity_type) {
+    $original_entity_type = $entity_last_installed_schema_respository->getLastInstalledDefinition($entity_type_id);
+    if (!$original_entity_type) {
+      continue;
+    }
+    $field_storage_definitions = $entity_last_installed_schema_respository->getLastInstalledFieldStorageDefinitions($entity_type_id);
+
+    $field_names = array_filter([
+      $original_entity_type->getKey('id'),
+      $original_entity_type->getKey('revision'),
+      $original_entity_type->getKey('langcode'),
+      $original_entity_type->getKey('default_langcode'),
+    ]);
+    foreach ($field_names as $field_name) {
+      if (isset($field_storage_definitions[$field_name])) {
+        $definition_update_manager->updateFieldStorageDefinition($field_storage_definitions[$field_name]);
+      }
+    }
+  }
+}
diff --git a/core/modules/system/tests/src/Functional/Entity/Update/SqlContentEntityStorageSchemaConverterTest.php b/core/modules/system/tests/src/Functional/Entity/Update/SqlContentEntityStorageSchemaConverterTest.php
index 625ee190ed..639be89565 100644
--- a/core/modules/system/tests/src/Functional/Entity/Update/SqlContentEntityStorageSchemaConverterTest.php
+++ b/core/modules/system/tests/src/Functional/Entity/Update/SqlContentEntityStorageSchemaConverterTest.php
@@ -220,6 +220,47 @@ public function testMakeRevisionableErrorHandling() {
     foreach ($new_storage_definitions as $storage_definition) {
       $new_field_schema_data[$storage_definition->getName()] = $this->installedStorageSchema->get('entity_test_update.field_schema_data.' . $storage_definition->getName(), []);
     }
+    // Before https://www.drupal.org/project/drupal/issues/2929120 various keys
+    // were missing from the field schema, so we explicitly add those before
+    // checking equality.
+    $this->assertFalse(isset($original_field_schema_data['id']['entity_test_update']['primary key']));
+    $original_field_schema_data['id']['entity_test_update']['primary key'] = ['id'];
+    $this->assertFalse(isset($original_field_schema_data['id']['entity_test_update_data']['primary key']));
+    $original_field_schema_data['id']['entity_test_update_data']['primary key'] = ['id', 'langcode'];
+    $this->assertFalse(isset($original_field_schema_data['id']['entity_test_update_data']['indexes']['entity_test_update__id__default_langcode__langcode']));
+    $original_field_schema_data['id']['entity_test_update_data']['indexes']['entity_test_update__id__default_langcode__langcode'] = ['id', 'default_langcode', 'langcode'];
+    $this->assertFalse(isset($original_field_schema_data['langcode']['entity_test_update_data']['primary key']));
+    $original_field_schema_data['langcode']['entity_test_update_data']['primary key'] = ['id', 'langcode'];
+    $this->assertFalse(isset($original_field_schema_data['langcode']['entity_test_update_data']['indexes']['entity_test_update__id__default_langcode__langcode']));
+    $original_field_schema_data['langcode']['entity_test_update_data']['indexes']['entity_test_update__id__default_langcode__langcode'] = ['id', 'default_langcode', 'langcode'];
+    $this->assertFalse(isset($original_field_schema_data['default_langcode']['entity_test_update_data']['indexes']['entity_test_update__id__default_langcode__langcode']));
+    $original_field_schema_data['default_langcode']['entity_test_update_data']['indexes']['entity_test_update__id__default_langcode__langcode'] = ['id', 'default_langcode', 'langcode'];
+    // Also account for the fact that the field schema now contains default
+    // values for DX.
+    $this->assertFalse(isset($original_field_schema_data['id']['entity_test_update']['unique keys']));
+    $original_field_schema_data['id']['entity_test_update']['unique keys'] = [];
+    $this->assertFalse(isset($original_field_schema_data['id']['entity_test_update']['indexes']));
+    $original_field_schema_data['id']['entity_test_update']['indexes'] = [];
+    $this->assertFalse(isset($original_field_schema_data['id']['entity_test_update']['foreign keys']));
+    $original_field_schema_data['id']['entity_test_update']['foreign keys'] = [];
+    $this->assertFalse(isset($original_field_schema_data['id']['entity_test_update_data']['unique keys']));
+    $original_field_schema_data['id']['entity_test_update_data']['unique keys'] = [];
+    $this->assertFalse(isset($original_field_schema_data['id']['entity_test_update_data']['foreign keys']));
+    $original_field_schema_data['id']['entity_test_update_data']['foreign keys'] = [];
+    $this->assertFalse(isset($original_field_schema_data['langcode']['entity_test_update']['unique keys']));
+    $original_field_schema_data['langcode']['entity_test_update']['unique keys'] = [];
+    $this->assertFalse(isset($original_field_schema_data['langcode']['entity_test_update']['indexes']));
+    $original_field_schema_data['langcode']['entity_test_update']['indexes'] = [];
+    $this->assertFalse(isset($original_field_schema_data['langcode']['entity_test_update']['foreign keys']));
+    $original_field_schema_data['langcode']['entity_test_update']['foreign keys'] = [];
+    $this->assertFalse(isset($original_field_schema_data['langcode']['entity_test_update_data']['unique keys']));
+    $original_field_schema_data['langcode']['entity_test_update_data']['unique keys'] = [];
+    $this->assertFalse(isset($original_field_schema_data['langcode']['entity_test_update_data']['foreign keys']));
+    $original_field_schema_data['langcode']['entity_test_update_data']['foreign keys'] = [];
+    $this->assertFalse(isset($original_field_schema_data['default_langcode']['entity_test_update_data']['unique keys']));
+    $original_field_schema_data['default_langcode']['entity_test_update_data']['unique keys'] = [];
+    $this->assertFalse(isset($original_field_schema_data['default_langcode']['entity_test_update_data']['foreign keys']));
+    $original_field_schema_data['default_langcode']['entity_test_update_data']['foreign keys'] = [];
     $this->assertEqual($original_field_schema_data, $new_field_schema_data);
 
     // Check that temporary tables have been removed.
diff --git a/core/modules/user/src/UserStorageSchema.php b/core/modules/user/src/UserStorageSchema.php
index 5e590b8958..fae75fbb79 100644
--- a/core/modules/user/src/UserStorageSchema.php
+++ b/core/modules/user/src/UserStorageSchema.php
@@ -2,7 +2,6 @@
 
 namespace Drupal\user;
 
-use Drupal\Core\Entity\ContentEntityTypeInterface;
 use Drupal\Core\Entity\Sql\SqlContentEntityStorageSchema;
 use Drupal\Core\Field\FieldStorageDefinitionInterface;
 
@@ -11,19 +10,6 @@
  */
 class UserStorageSchema extends SqlContentEntityStorageSchema {
 
-  /**
-   * {@inheritdoc}
-   */
-  protected function getEntitySchema(ContentEntityTypeInterface $entity_type, $reset = FALSE) {
-    $schema = parent::getEntitySchema($entity_type, $reset);
-
-    $schema['users_field_data']['unique keys'] += [
-      'user__name' => ['name', 'langcode'],
-    ];
-
-    return $schema;
-  }
-
   /**
    * {@inheritdoc}
    */
@@ -42,24 +28,22 @@ protected function getSharedTableFieldSchema(FieldStorageDefinitionInterface $st
     $field_name = $storage_definition->getName();
 
     if ($table_name == 'users_field_data') {
-      switch ($field_name) {
-        case 'name':
-          // Improves the performance of the user__name index defined
-          // in getEntitySchema().
-          $schema['fields'][$field_name]['not null'] = TRUE;
-          // Make sure the field is no longer than 191 characters so we can
-          // add a unique constraint in MySQL.
-          $schema['fields'][$field_name]['length'] = USERNAME_MAX_LENGTH;
-          break;
-
-        case 'mail':
-          $this->addSharedTableFieldIndex($storage_definition, $schema);
-          break;
-
-        case 'access':
-        case 'created':
-          $this->addSharedTableFieldIndex($storage_definition, $schema, TRUE);
-          break;
+      if ($field_name === 'name') {
+        // Improves the performance of the user__name index defined
+        // in getEntitySchema().
+        $schema['fields'][$field_name]['not null'] = TRUE;
+        // Make sure the field is no longer than 191 characters so we can
+        // add a unique constraint in MySQL.
+        $schema['fields'][$field_name]['length'] = UserInterface::USERNAME_MAX_LENGTH;
+      }
+      if ($field_name === 'mail') {
+        $this->addSharedTableFieldIndex($storage_definition, $schema);
+      }
+      if (in_array($field_name, ['access', 'created'], TRUE)) {
+        $this->addSharedTableFieldIndex($storage_definition, $schema, TRUE);
+      }
+      if (in_array($field_name, ['name', 'langcode'], TRUE)) {
+        $schema['users_field_data']['unique keys']['user__name'] = ['name', 'langcode'];
       }
     }
 
diff --git a/core/modules/user/user.install b/core/modules/user/user.install
index 0af797a43f..aa0a2a4be6 100644
--- a/core/modules/user/user.install
+++ b/core/modules/user/user.install
@@ -98,3 +98,16 @@ function user_update_8100() {
     $config->set('status_blocked', $mail)->save(TRUE);
   }
 }
+
+/**
+ * Update the stored schema data of various node fields.
+ */
+function user_update_8601() {
+  $definition_update_manager = \Drupal::entityDefinitionUpdateManager();
+
+  $field_storage_definition = $definition_update_manager->getFieldStorageDefinition('name', 'user');
+  $definition_update_manager->updateFieldStorageDefinition($field_storage_definition);
+
+  $entity_type = $definition_update_manager->getEntityType('user');
+  $definition_update_manager->updateEntityType($entity_type);
+}
diff --git a/core/tests/Drupal/KernelTests/Core/Database/AssertSchemaTrait.php b/core/tests/Drupal/KernelTests/Core/Database/AssertSchemaTrait.php
new file mode 100644
index 0000000000..8ba2564707
--- /dev/null
+++ b/core/tests/Drupal/KernelTests/Core/Database/AssertSchemaTrait.php
@@ -0,0 +1,55 @@
+<?php
+
+namespace Drupal\KernelTests\Core\Database;
+
+use Drupal\Core\Database\Database;
+
+/**
+ * Provides a trait for schema-related assertions in tests.
+ */
+trait AssertSchemaTrait {
+
+  /**
+   * Tests the primary keys of a table.
+   *
+   * @param string $table_name
+   *   The name of the table to check.
+   * @param array $primary_key
+   *   The expected key column specifier for a table's primary key.
+   */
+  protected function assertPrimaryKeyColumns($table_name, array $primary_key = []) {
+    $connection = Database::getConnection();
+    $db_type = $connection->databaseType();
+
+    switch ($db_type) {
+      case 'mysql':
+        $result = $connection->query("SHOW KEYS FROM {" . $table_name . "} WHERE Key_name = 'PRIMARY'")->fetchAllAssoc('Column_name');
+        $this->assertSame($primary_key, array_keys($result));
+
+        break;
+      case 'pgsql':
+        $result = $connection->query("SELECT a.attname, format_type(a.atttypid, a.atttypmod) AS data_type
+          FROM pg_index i
+          JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
+          WHERE i.indrelid = '{" . $table_name . "}'::regclass AND i.indisprimary")
+          ->fetchAllAssoc('attname');
+        $this->assertSame($primary_key, array_keys($result));
+
+        break;
+      case 'sqlite':
+        // For SQLite we need access to the protected
+        // \Drupal\Core\Database\Driver\sqlite\Schema::introspectSchema() method
+        // because we have no other way of getting the table prefixes needed for
+        // running a straight PRAGMA query.
+        $schema_object = $connection->schema();
+        $reflection = new \ReflectionMethod($schema_object, 'introspectSchema');
+        $reflection->setAccessible(TRUE);
+
+        $table_info = $reflection->invoke($schema_object, $table_name);
+        $this->assertSame($primary_key, $table_info['primary key']);
+
+        break;
+    }
+  }
+
+}
diff --git a/core/tests/Drupal/KernelTests/Core/Database/SchemaTest.php b/core/tests/Drupal/KernelTests/Core/Database/SchemaTest.php
index e239098715..608c748606 100644
--- a/core/tests/Drupal/KernelTests/Core/Database/SchemaTest.php
+++ b/core/tests/Drupal/KernelTests/Core/Database/SchemaTest.php
@@ -16,6 +16,8 @@
  */
 class SchemaTest extends KernelTestBase {
 
+  use AssertSchemaTrait;
+
   /**
    * A global counter for table and field creation.
    */
@@ -826,46 +828,4 @@ public function testFindTables() {
     Database::setActiveConnection('default');
   }
 
-  /**
-   * Tests the primary keys of a table.
-   *
-   * @param string $table_name
-   *   The name of the table to check.
-   * @param array $primary_key
-   *   The expected key column specifier for a table's primary key.
-   */
-  protected function assertPrimaryKeyColumns($table_name, array $primary_key = []) {
-    $db_type = Database::getConnection()->databaseType();
-
-    switch ($db_type) {
-      case 'mysql':
-        $result = Database::getConnection()->query("SHOW KEYS FROM {" . $table_name . "} WHERE Key_name = 'PRIMARY'")->fetchAllAssoc('Column_name');
-        $this->assertSame($primary_key, array_keys($result));
-
-        break;
-      case 'pgsql':
-        $result = Database::getConnection()->query("SELECT a.attname, format_type(a.atttypid, a.atttypmod) AS data_type
-          FROM pg_index i
-          JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
-          WHERE i.indrelid = '{" . $table_name . "}'::regclass AND i.indisprimary")
-          ->fetchAllAssoc('attname');
-        $this->assertSame($primary_key, array_keys($result));
-
-        break;
-      case 'sqlite':
-        // For SQLite we need access to the protected
-        // \Drupal\Core\Database\Driver\sqlite\Schema::introspectSchema() method
-        // because we have no other way of getting the table prefixes needed for
-        // running a straight PRAGMA query.
-        $schema_object = Database::getConnection()->schema();
-        $reflection = new \ReflectionMethod($schema_object, 'introspectSchema');
-        $reflection->setAccessible(TRUE);
-
-        $table_info = $reflection->invoke($schema_object, $table_name);
-        $this->assertSame($primary_key, $table_info['primary key']);
-
-        break;
-    }
-  }
-
 }
diff --git a/core/tests/Drupal/KernelTests/Core/Entity/EntitySchemaTest.php b/core/tests/Drupal/KernelTests/Core/Entity/EntitySchemaTest.php
index 3e285ff69b..a31228eff8 100644
--- a/core/tests/Drupal/KernelTests/Core/Entity/EntitySchemaTest.php
+++ b/core/tests/Drupal/KernelTests/Core/Entity/EntitySchemaTest.php
@@ -3,14 +3,19 @@
 namespace Drupal\KernelTests\Core\Entity;
 
 use Drupal\Component\Utility\SafeMarkup;
+use Drupal\Core\Entity\EntityTypeInterface;
+use Drupal\Core\Field\BaseFieldDefinition;
+use Drupal\KernelTests\Core\Database\AssertSchemaTrait;
 
 /**
- * Tests adding a custom bundle field.
+ * Tests the default entity storage schema handler.
  *
- * @group system
+ * @group Entity
  */
 class EntitySchemaTest extends EntityKernelTestBase {
 
+  use AssertSchemaTrait;
+
   /**
    * The database connection used.
    *
@@ -109,6 +114,163 @@ public function testEntitySchemaUpdate() {
     $this->assertTrue($schema_handler->tableExists($dedicated_tables[0]), SafeMarkup::format('Field schema correct for the @table table.', ['@table' => $table]));
   }
 
+  /**
+   * Tests deleting and creating a field that is part of a primary key.
+   *
+   * @param string $entity_type_id
+   *   The ID of the entity type whose schema is being tested.
+   * @param string $field_name
+   *   The name of the field that is being re-installed.
+   *
+   * @dataProvider providerTestPrimaryKeyUpdate
+   */
+  public function testPrimaryKeyUpdate($entity_type_id, $field_name) {
+    // EntityKernelTestBase::setUp() already installs the schema for the
+    // 'entity_test' entity type.
+    if ($entity_type_id !== 'entity_test') {
+      $this->installEntitySchema($entity_type_id);
+    }
+
+    /* @var \Drupal\Core\Entity\EntityDefinitionUpdateManagerInterface $update_manager */
+    $update_manager = $this->container->get('entity.definition_update_manager');
+    $entity_type = $update_manager->getEntityType($entity_type_id);
+
+    /* @see \Drupal\Core\Entity\ContentEntityBase::baseFieldDefinitions() */
+    switch ($field_name) {
+      case 'id':
+        $field = BaseFieldDefinition::create('integer')
+          ->setLabel('ID')
+          ->setReadOnly(TRUE)
+          ->setSetting('unsigned', TRUE);
+        break;
+
+      case 'revision_id':
+        $field = BaseFieldDefinition::create('integer')
+          ->setLabel('Revision ID')
+          ->setReadOnly(TRUE)
+          ->setSetting('unsigned', TRUE);
+        break;
+
+      case 'langcode':
+        $field = BaseFieldDefinition::create('language')
+          ->setLabel('Language');
+        if ($entity_type->isRevisionable()) {
+          $field->setRevisionable(TRUE);
+        }
+        if ($entity_type->isTranslatable()) {
+          $field->setTranslatable(TRUE);
+        }
+        break;
+    }
+
+    $field
+      ->setName($field_name)
+      ->setTargetEntityTypeId($entity_type_id)
+      ->setProvider($entity_type->getProvider());
+
+    // First test explicitly deleting and re-installing a field.
+    // Make sure all primary keys are there to start with.
+    $this->assertPrimaryKeys($entity_type);
+    // Then uninstall the field and make sure all primary keys that the field
+    // is part of have been removed.
+    $update_manager->uninstallFieldStorageDefinition($field);
+    $this->assertPrimaryKeys($entity_type, $field_name);
+    // Finally reinstall the field and make sure the primary keys have been
+    // recreated.
+    $update_manager->installFieldStorageDefinition($field->getName(), $entity_type_id, $field->getProvider(), $field);
+    $this->assertPrimaryKeys($entity_type);
+
+    // Now test updating a field without data. This will end up deleting
+    // and re-creating the field, similar to the code above.
+    $update_manager->updateFieldStorageDefinition($field);
+    $this->assertPrimaryKeys($entity_type);
+
+    // Now test updating a field with data.
+    /* @var \Drupal\Core\Entity\FieldableEntityStorageInterface $storage */
+    $storage = $this->entityManager->getStorage($entity_type_id);
+    // The schema of ID fields is incorrectly recreated as 'int' instead of
+    // 'serial', so we manually have to specify an ID.
+    // @todo Remove this in https://www.drupal.org/project/drupal/issues/2928906
+    $storage->create(['id' => 1, 'revision_id' => 1])->save();
+    $this->assertTrue($storage->countFieldData($field, TRUE));
+    $update_manager->updateFieldStorageDefinition($field);
+    $this->assertPrimaryKeys($entity_type);
+    $this->assertTrue($storage->countFieldData($field, TRUE));
+  }
+
+  /**
+   * Asserts that the primary keys are as expected for a given entity type.
+   *
+   * @param \Drupal\Core\Entity\EntityTypeInterface $entity_type
+   *   The entity type whose primary keys are being checked.
+   * @param string $removed_field
+   *   (optional) The name of a field that has been removed. If passed, any
+   *   primary key that would otherwise have contained this field will be
+   *   checked to have been dropped.
+   */
+  protected function assertPrimaryKeys(EntityTypeInterface $entity_type, $removed_field = NULL) {
+    $base_table = $entity_type->getBaseTable();
+    $revision_table = $entity_type->getRevisionTable();
+    $data_table = $entity_type->getDataTable();
+    $revision_data_table = $entity_type->getRevisionDataTable();
+
+    $id_key = $entity_type->getKey('id');
+    $revision_key = $entity_type->getKey('revision');
+    $langcode_key = $entity_type->getKey('langcode');
+
+    // Build up a map of primary keys depending on the entity type
+    // configuration.
+    $key_map = [];
+    $key_map[$base_table] = [$id_key];
+    if ($entity_type->isRevisionable()) {
+      $key_map[$revision_table] = [$revision_key];
+    }
+    if ($entity_type->isTranslatable()) {
+      $key_map[$data_table] = [$id_key, $langcode_key];
+    }
+    if ($entity_type->isRevisionable() && $entity_type->isTranslatable()) {
+      $key_map[$revision_data_table] = [$revision_key, $langcode_key];
+    }
+
+    foreach ($key_map as $table => $primary_key) {
+      // Assert that the primary key has been dropped if a field name was passed
+      // and the given primary key contains this field.
+      if (!$removed_field || !in_array($removed_field, $primary_key, TRUE)) {
+        $this->assertPrimaryKeyColumns($table, $primary_key);
+      }
+      else {
+        $this->assertPrimaryKeyColumns($table, []);
+      }
+    }
+  }
+
+  /**
+   * Provides test cases for EntitySchemaTest::testPrimaryKeyUpdate()
+   *
+   * @return array
+   *   An array of test cases consisting of an entity type ID and a field name.
+   */
+  public function providerTestPrimaryKeyUpdate() {
+    // Build up test cases for all possible entity type configurations.
+    // For each entity type we test reinstalling each field that is part of
+    // any table's primary key.
+    $tests = [];
+
+    $tests['entity_test:id'] = ['entity_test', 'id'];
+
+    $tests['entity_test_rev:id'] = ['entity_test_rev', 'id'];
+    $tests['entity_test_rev:revision_id'] = ['entity_test_rev', 'revision_id'];
+
+    $tests['entity_test_mul:id'] = ['entity_test_mul', 'id'];
+    $tests['entity_test_mul:langcode'] = ['entity_test_mul', 'langcode'];
+
+    $tests['entity_test_mulrev:id'] = ['entity_test_mulrev', 'id'];
+    $tests['entity_test_mulrev:revision_id'] = ['entity_test_mulrev', 'revision_id'];
+    $tests['entity_test_mulrev:langcode'] = ['entity_test_mulrev', 'langcode'];
+
+    return $tests;
+  }
+
   /**
    * {@inheritdoc}
    */
diff --git a/core/tests/Drupal/Tests/Core/Entity/Sql/SqlContentEntityStorageSchemaTest.php b/core/tests/Drupal/Tests/Core/Entity/Sql/SqlContentEntityStorageSchemaTest.php
index d16ece2097..56852869e5 100644
--- a/core/tests/Drupal/Tests/Core/Entity/Sql/SqlContentEntityStorageSchemaTest.php
+++ b/core/tests/Drupal/Tests/Core/Entity/Sql/SqlContentEntityStorageSchemaTest.php
@@ -397,7 +397,7 @@ public function testGetSchemaRevisionable() {
       ],
     ]);
 
-    $this->storage->expects($this->exactly(2))
+    $this->storage->expects($this->exactly(6))
       ->method('getRevisionTable')
       ->will($this->returnValue('entity_test_revision'));
 
@@ -604,13 +604,13 @@ public function testGetSchemaRevisionableTranslatable() {
       ],
     ]);
 
-    $this->storage->expects($this->exactly(3))
+    $this->storage->expects($this->exactly(25))
       ->method('getRevisionTable')
       ->will($this->returnValue('entity_test_revision'));
-    $this->storage->expects($this->once())
+    $this->storage->expects($this->exactly(45))
       ->method('getDataTable')
       ->will($this->returnValue('entity_test_field_data'));
-    $this->storage->expects($this->once())
+    $this->storage->expects($this->exactly(37))
       ->method('getRevisionDataTable')
       ->will($this->returnValue('entity_test_revision_field_data'));
 
diff --git a/core/tests/Drupal/Tests/Core/Entity/Sql/SqlContentEntityStorageTest.php b/core/tests/Drupal/Tests/Core/Entity/Sql/SqlContentEntityStorageTest.php
index bf3fd46eb6..d7e23a1d87 100644
--- a/core/tests/Drupal/Tests/Core/Entity/Sql/SqlContentEntityStorageTest.php
+++ b/core/tests/Drupal/Tests/Core/Entity/Sql/SqlContentEntityStorageTest.php
@@ -327,6 +327,14 @@ public function testOnEntityTypeCreate() {
     $this->entityType->expects($this->once())
       ->method('getKeys')
       ->will($this->returnValue(['id' => 'id']));
+    $this->entityType->expects($this->any())
+      ->method('hasKey')
+      ->will($this->returnValueMap([
+        // SqlContentEntityStorageSchema::initializeBaseTable()
+        ['revision', FALSE],
+        // SqlContentEntityStorageSchema::processBaseTable()
+        ['id', TRUE],
+      ]));
     $this->entityType->expects($this->any())
       ->method('getKey')
       ->will($this->returnValueMap([