diff --git a/core/lib/Drupal/Core/Entity/ContentEntityDatabaseStorage.php b/core/lib/Drupal/Core/Entity/ContentEntityDatabaseStorage.php
index b21958b..a650fce 100644
--- a/core/lib/Drupal/Core/Entity/ContentEntityDatabaseStorage.php
+++ b/core/lib/Drupal/Core/Entity/ContentEntityDatabaseStorage.php
@@ -18,9 +18,8 @@
 use Drupal\Core\Field\FieldStorageDefinitionInterface;
 use Drupal\Core\Language\Language;
 use Drupal\field\FieldConfigUpdateForbiddenException;
-use Drupal\field\FieldConfigInterface;
-use Drupal\field\FieldInstanceConfigInterface;
 use Drupal\field\Entity\FieldConfig;
+use Drupal\field\FieldInstanceConfigInterface;
 use Symfony\Component\DependencyInjection\ContainerInterface;
 
 /**
@@ -954,7 +953,7 @@ protected function doLoadFieldItems($entities, $age) {
     foreach ($bundles as $bundle => $v) {
       foreach ($this->entityManager->getFieldDefinitions($this->entityTypeId, $bundle) as $field_name => $instance) {
         if ($instance instanceof FieldInstanceConfigInterface) {
-          $fields[$field_name] = $instance->getField();
+          $fields[$field_name] = $instance;
         }
       }
     }
@@ -1023,9 +1022,8 @@ protected function doSaveFieldItems(EntityInterface $entity, $update) {
       if (!($instance instanceof FieldInstanceConfigInterface)) {
         continue;
       }
-      $field = $instance->getField();
-      $table_name = static::_fieldTableName($field);
-      $revision_name = static::_fieldRevisionTableName($field);
+      $table_name = static::_fieldTableName($instance);
+      $revision_name = static::_fieldRevisionTableName($instance);
 
       // Delete and insert, rather than update, in case a value was added.
       if ($update) {
@@ -1045,13 +1043,13 @@ protected function doSaveFieldItems(EntityInterface $entity, $update) {
       // Prepare the multi-insert query.
       $do_insert = FALSE;
       $columns = array('entity_id', 'revision_id', 'bundle', 'delta', 'langcode');
-      foreach ($field->getColumns() as $column => $attributes) {
-        $columns[] = static::_fieldColumnName($field, $column);
+      foreach ($instance->getColumns() as $column => $attributes) {
+        $columns[] = static::_fieldColumnName($instance, $column);
       }
       $query = $this->database->insert($table_name)->fields($columns);
       $revision_query = $this->database->insert($revision_name)->fields($columns);
 
-      $langcodes = $field->isTranslatable() ? $translation_langcodes : array($default_langcode);
+      $langcodes = $instance->isTranslatable() ? $translation_langcodes : array($default_langcode);
       foreach ($langcodes as $langcode) {
         $delta_count = 0;
         $items = $entity->getTranslation($langcode)->get($field_name);
@@ -1066,15 +1064,15 @@ protected function doSaveFieldItems(EntityInterface $entity, $update) {
             'delta' => $delta,
             'langcode' => $langcode,
           );
-          foreach ($field->getColumns() as $column => $attributes) {
-            $column_name = static::_fieldColumnName($field, $column);
+          foreach ($instance->getColumns() as $column => $attributes) {
+            $column_name = static::_fieldColumnName($instance, $column);
             // Serialize the value if specified in the column schema.
             $record[$column_name] = !empty($attributes['serialize']) ? serialize($item->$column) : $item->$column;
           }
           $query->values($record);
           $revision_query->values($record);
 
-          if ($field->getCardinality() != FieldStorageDefinitionInterface::CARDINALITY_UNLIMITED && ++$delta_count == $field->getCardinality()) {
+          if ($instance->getCardinality() != FieldStorageDefinitionInterface::CARDINALITY_UNLIMITED && ++$delta_count == $instance->getCardinality()) {
             break;
           }
         }
@@ -1100,9 +1098,8 @@ protected function doDeleteFieldItems(EntityInterface $entity) {
       if (!($instance instanceof FieldInstanceConfigInterface)) {
         continue;
       }
-      $field = $instance->getField();
-      $table_name = static::_fieldTableName($field);
-      $revision_name = static::_fieldRevisionTableName($field);
+      $table_name = static::_fieldTableName($instance);
+      $revision_name = static::_fieldRevisionTableName($instance);
       $this->database->delete($table_name)
         ->condition('entity_id', $entity->id())
         ->execute();
@@ -1122,7 +1119,7 @@ protected function doDeleteFieldItemsRevision(EntityInterface $entity) {
         if (!($instance instanceof FieldInstanceConfigInterface)) {
           continue;
         }
-        $revision_name = static::_fieldRevisionTableName($instance->getField());
+        $revision_name = static::_fieldRevisionTableName($instance);
         $this->database->delete($revision_name)
           ->condition('entity_id', $entity->id())
           ->condition('revision_id', $vid)
@@ -1134,8 +1131,8 @@ protected function doDeleteFieldItemsRevision(EntityInterface $entity) {
   /**
    * {@inheritdoc}
    */
-  public function onFieldCreate(FieldConfigInterface $field) {
-    $schema = $this->_fieldSqlSchema($field);
+  public function onFieldCreate(FieldStorageDefinitionInterface $storage_definition) {
+    $schema = $this->_fieldSqlSchema($storage_definition);
     foreach ($schema as $name => $table) {
       $this->database->schema()->createTable($name, $table);
     }
@@ -1144,10 +1141,10 @@ public function onFieldCreate(FieldConfigInterface $field) {
   /**
    * {@inheritdoc}
    */
-  public function onFieldUpdate(FieldConfigInterface $field) {
-    $original = $field->original;
+  public function onFieldUpdate(FieldStorageDefinitionInterface $storage_definition) {
+    $original = $storage_definition->original;
 
-    if (!$field->hasData()) {
+    if (!$storage_definition->hasData()) {
       // There is no data. Re-create the tables completely.
 
       if ($this->database->supportsTransactionalDDL()) {
@@ -1161,7 +1158,7 @@ public function onFieldUpdate(FieldConfigInterface $field) {
         foreach ($original_schema as $name => $table) {
           $this->database->schema()->dropTable($name, $table);
         }
-        $schema = $this->_fieldSqlSchema($field);
+        $schema = $this->_fieldSqlSchema($storage_definition);
         foreach ($schema as $name => $table) {
           $this->database->schema()->createTable($name, $table);
         }
@@ -1183,7 +1180,7 @@ public function onFieldUpdate(FieldConfigInterface $field) {
       }
     }
     else {
-      if ($field->getColumns() != $original->getColumns()) {
+      if ($storage_definition->getColumns() != $original->getColumns()) {
         throw new FieldConfigUpdateForbiddenException("The SQL storage cannot change the schema for an existing field with data.");
       }
       // There is data, so there are no column changes. Drop all the prior
@@ -1192,33 +1189,33 @@ public function onFieldUpdate(FieldConfigInterface $field) {
       $table = static::_fieldTableName($original);
       $revision_table = static::_fieldRevisionTableName($original);
 
-      $schema = $field->getSchema();
+      $schema = $storage_definition->getSchema();
       $original_schema = $original->getSchema();
 
       foreach ($original_schema['indexes'] as $name => $columns) {
         if (!isset($schema['indexes'][$name]) || $columns != $schema['indexes'][$name]) {
-          $real_name = static::_fieldIndexName($field, $name);
+          $real_name = static::_fieldIndexName($storage_definition, $name);
           $this->database->schema()->dropIndex($table, $real_name);
           $this->database->schema()->dropIndex($revision_table, $real_name);
         }
       }
-      $table = static::_fieldTableName($field);
-      $revision_table = static::_fieldRevisionTableName($field);
+      $table = static::_fieldTableName($storage_definition);
+      $revision_table = static::_fieldRevisionTableName($storage_definition);
       foreach ($schema['indexes'] as $name => $columns) {
         if (!isset($original_schema['indexes'][$name]) || $columns != $original_schema['indexes'][$name]) {
-          $real_name = static::_fieldIndexName($field, $name);
+          $real_name = static::_fieldIndexName($storage_definition, $name);
           $real_columns = array();
           foreach ($columns as $column_name) {
             // Indexes can be specified as either a column name or an array with
             // column name and length. Allow for either case.
             if (is_array($column_name)) {
               $real_columns[] = array(
-                static::_fieldColumnName($field, $column_name[0]),
+                static::_fieldColumnName($storage_definition, $column_name[0]),
                 $column_name[1],
               );
             }
             else {
-              $real_columns[] = static::_fieldColumnName($field, $column_name);
+              $real_columns[] = static::_fieldColumnName($storage_definition, $column_name);
             }
           }
           $this->database->schema()->addIndex($table, $real_name, $real_columns);
@@ -1231,20 +1228,18 @@ public function onFieldUpdate(FieldConfigInterface $field) {
   /**
    * {@inheritdoc}
    */
-  public function onFieldDelete(FieldConfigInterface $field) {
+  public function onFieldDelete(FieldStorageDefinitionInterface $storage_definition) {
     // Mark all data associated with the field for deletion.
-    $table = static::_fieldTableName($field);
-    $revision_table = static::_fieldRevisionTableName($field);
+    $table = static::_fieldTableName($storage_definition);
+    $revision_table = static::_fieldRevisionTableName($storage_definition);
     $this->database->update($table)
       ->fields(array('deleted' => 1))
       ->execute();
 
     // Move the table to a unique name while the table contents are being
     // deleted.
-    $deleted_field = clone $field;
-    $deleted_field->deleted = TRUE;
-    $new_table = static::_fieldTableName($deleted_field);
-    $revision_new_table = static::_fieldRevisionTableName($deleted_field);
+    $new_table = static::_fieldTableName($storage_definition, TRUE);
+    $revision_new_table = static::_fieldRevisionTableName($storage_definition, TRUE);
     $this->database->schema()->renameTable($table, $new_table);
     $this->database->schema()->renameTable($revision_table, $revision_new_table);
   }
@@ -1252,17 +1247,16 @@ public function onFieldDelete(FieldConfigInterface $field) {
   /**
    * {@inheritdoc}
    */
-  public function onInstanceDelete(FieldInstanceConfigInterface $instance) {
-    $field = $instance->getField();
-    $table_name = static::_fieldTableName($field);
-    $revision_name = static::_fieldRevisionTableName($field);
+  public function onInstanceDelete(FieldDefinitionInterface $field_definition) {
+    $table_name = static::_fieldTableName($field_definition);
+    $revision_name = static::_fieldRevisionTableName($field_definition);
     $this->database->update($table_name)
       ->fields(array('deleted' => 1))
-      ->condition('bundle', $instance->bundle)
+      ->condition('bundle', $field_definition->bundle)
       ->execute();
     $this->database->update($revision_name)
       ->fields(array('deleted' => 1))
-      ->condition('bundle', $instance->bundle)
+      ->condition('bundle', $field_definition->bundle)
       ->execute();
   }
 
@@ -1275,9 +1269,8 @@ public function onBundleRename($bundle, $bundle_new) {
     // using the old bundle name.
     $instances = entity_load_multiple_by_properties('field_instance_config', array('entity_type' => $this->entityTypeId, 'bundle' => $bundle, 'include_deleted' => TRUE));
     foreach ($instances as $instance) {
-      $field = $instance->getField();
-      $table_name = static::_fieldTableName($field);
-      $revision_name = static::_fieldRevisionTableName($field);
+      $table_name = static::_fieldTableName($instance);
+      $revision_name = static::_fieldRevisionTableName($instance);
       $this->database->update($table_name)
         ->fields(array('bundle' => $bundle_new))
         ->condition('bundle', $bundle)
@@ -1292,54 +1285,109 @@ public function onBundleRename($bundle, $bundle_new) {
   /**
    * {@inheritdoc}
    */
-  protected function readFieldItemsToPurge(EntityInterface $entity, FieldInstanceConfigInterface $instance) {
-    $field = $instance->getField();
-    $table_name = static::_fieldTableName($field);
+  protected function readFieldItemsToPurge(FieldDefinitionInterface $field_definition, $batch_size) {
+    // Check whether the whole field storage definition is gone, or just some
+    // bundle fields.
+    $is_deleted = !array_key_exists($field_definition->getName(), $this->entityManager->getFieldStorageDefinitions($this->entityTypeId));
+    $table_name = static::_fieldTableName($field_definition, $is_deleted);
     $query = $this->database->select($table_name, 't', array('fetch' => \PDO::FETCH_ASSOC))
-      ->condition('entity_id', $entity->id())
-      ->orderBy('delta');
-    foreach ($field->getColumns() as $column_name => $data) {
-      $query->addField('t', static::_fieldColumnName($field, $column_name), $column_name);
+      ->fields('t')
+      ->condition('bundle', $field_definition->bundle)
+      ->orderBy('entity_id')
+      ->orderBy('revision_id')
+      ->orderBy('delta')
+      ->range(0, $batch_size);
+
+    // Create a map of field data table column names to field column names.
+    $column_map = array();
+    foreach ($field_definition->getColumns() as $column_name => $data) {
+      $column_map[static::_fieldColumnName($field_definition, $column_name)] = $column_name;
+    }
+
+    $entities = array();
+    $items_by_entity = array();
+    foreach ($query->execute() as $row) {
+      if (!isset($entities[$row['revision_id']])) {
+        // Create entity with the right revision id and entity id combination.
+        $row['entity_type'] = $this->entityTypeId;
+        // @todo: Replace this by an entity object created via an entity
+        // factory, see https://drupal.org/node/1867228.
+        $entities[$row['revision_id']] = _field_create_entity_from_ids((object) $row);
+      }
+      $item = array();
+      foreach ($column_map as $db_column => $field_column) {
+        $item[$field_column] = $row[$db_column];
+      }
+      $items_by_entity[$row['revision_id']][] = $item;
     }
-    return $query->execute()->fetchAll();
+    // Create field item objects and return.
+    foreach ($items_by_entity as $revision_id => $values) {
+      $items_by_entity[$revision_id] = \Drupal::typedDataManager()->create($field_definition, $values, $field_definition->getName(), $entities[$revision_id]);
+    }
+    return $items_by_entity;
   }
 
   /**
    * {@inheritdoc}
    */
-  public function purgeFieldItems(EntityInterface $entity, FieldInstanceConfigInterface $instance) {
-    $field = $instance->getField();
-    $table_name = static::_fieldTableName($field);
-    $revision_name = static::_fieldRevisionTableName($field);
+  protected function purgeFieldItems(ContentEntityInterface $entity, FieldDefinitionInterface $field_definition) {
+    $is_deleted = !array_key_exists($field_definition->getName(), $this->entityManager->getFieldStorageDefinitions($this->entityTypeId));
+    $table_name = static::_fieldTableName($field_definition, $is_deleted);
+    $revision_name = static::_fieldRevisionTableName($field_definition, $is_deleted);
+    $revision_id = $entity->getRevisionId() !== NULL ? $entity->getRevisionId() : $entity->id();
     $this->database->delete($table_name)
-      ->condition('entity_id', $entity->id())
+      ->condition('revision_id', $revision_id)
       ->execute();
     $this->database->delete($revision_name)
-      ->condition('entity_id', $entity->id())
+      ->condition('revision_id', $revision_id)
       ->execute();
   }
 
   /**
    * {@inheritdoc}
    */
-  public function onFieldPurge(FieldConfigInterface $field) {
-    $table_name = static::_fieldTableName($field);
-    $revision_name = static::_fieldRevisionTableName($field);
+  public function finalizePurge(FieldStorageDefinitionInterface $storage_definition) {
+    $table_name = static::_fieldTableName($storage_definition, TRUE);
+    $revision_name = static::_fieldRevisionTableName($storage_definition, TRUE);
     $this->database->schema()->dropTable($table_name);
     $this->database->schema()->dropTable($revision_name);
   }
 
   /**
+   * {@inheritdoc}
+   */
+  public function countFieldData($storage_definition, $as_bool = FALSE) {
+    $is_deleted = !array_key_exists($storage_definition->getName(), $this->entityManager->getFieldStorageDefinitions($this->entityTypeId));
+    $table_name = static::_fieldTableName($storage_definition, $is_deleted);
+
+    $query = $this->database->select($table_name, 't');
+    $or = $query->orConditionGroup();
+    foreach ($storage_definition->getColumns() as $column_name => $data) {
+      $or->isNotNull(static::_fieldColumnName($storage_definition, $column_name));
+    }
+    // If we are performing the query just to check if the field has data
+    // limit the number of rows.
+    if ($as_bool) {
+      $query->range(0, 1);
+    }
+    $count = $query->countQuery()->execute()->fetchField();
+    return $as_bool ? (bool) $count : (int) $count;
+  }
+
+  /**
    * Gets the SQL table schema.
    *
    * @private Calling this function circumvents the entity system and is
    * strongly discouraged. This function is not considered part of the public
    * API and modules relying on it might break even in minor releases.
    *
-   * @param \Drupal\field\FieldConfigInterface $field
-   *   The field object
+   * @param \Drupal\Core\Field\FieldStorageDefinitionInterface $storage_definition
+   *   The field storage definition.
    * @param array $schema
    *   The field schema array. Mandatory for upgrades, omit otherwise.
+   * @param bool $deleted
+   *   (optional) Whether the schema of the table holding the values of a
+   *   deleted field should be returned.
    *
    * @return array
    *   The same as a hook_schema() implementation for the data and the
@@ -1347,17 +1395,20 @@ public function onFieldPurge(FieldConfigInterface $field) {
    *
    * @see hook_schema()
    */
-  public static function _fieldSqlSchema(FieldConfigInterface $field, array $schema = NULL) {
-    if ($field->deleted) {
-      $description_current = "Data storage for deleted field {$field->uuid()} ({$field->entity_type}, {$field->getName()}).";
-      $description_revision = "Revision archive storage for deleted field {$field->uuid()} ({$field->entity_type}, {$field->getName()}).";
+  public static function _fieldSqlSchema(FieldStorageDefinitionInterface $storage_definition, array $schema = NULL, $deleted = FALSE) {
+    if ($storage_definition->deleted && !$deleted) {
+      throw new \Exception("Field schema for deleted field to be returned.");
+    }
+    if ($deleted) {
+      $description_current = "Data storage for deleted field {$storage_definition->getUniqueStorageIdentifier()} ({$storage_definition->getTargetEntityTypeId()}, {$storage_definition->getName()}).";
+      $description_revision = "Revision archive storage for deleted field {$storage_definition->getUniqueStorageIdentifier()} ({$storage_definition->getTargetEntityTypeId()}, {$storage_definition->getName()}).";
     }
     else {
-      $description_current = "Data storage for {$field->entity_type} field {$field->getName()}.";
-      $description_revision = "Revision archive storage for {$field->entity_type} field {$field->getName()}.";
+      $description_current = "Data storage for {$storage_definition->getTargetEntityTypeId()} field {$storage_definition->getName()}.";
+      $description_revision = "Revision archive storage for {$storage_definition->getTargetEntityTypeId()} field {$storage_definition->getName()}.";
     }
 
-    $entity_type_id = $field->entity_type;
+    $entity_type_id = $storage_definition->getTargetEntityTypeId();
     $entity_manager = \Drupal::entityManager();
     $entity_type = $entity_manager->getDefinition($entity_type_id);
     $definitions = $entity_manager->getBaseFieldDefinitions($entity_type_id);
@@ -1445,39 +1496,39 @@ public static function _fieldSqlSchema(FieldConfigInterface $field, array $schem
     );
 
     if (!$schema) {
-      $schema = $field->getSchema();
+      $schema = $storage_definition->getSchema();
     }
 
     // Add field columns.
     foreach ($schema['columns'] as $column_name => $attributes) {
-      $real_name = static::_fieldColumnName($field, $column_name);
+      $real_name = static::_fieldColumnName($storage_definition, $column_name);
       $current['fields'][$real_name] = $attributes;
     }
 
     // Add indexes.
     foreach ($schema['indexes'] as $index_name => $columns) {
-      $real_name = static::_fieldIndexName($field, $index_name);
+      $real_name = static::_fieldIndexName($storage_definition, $index_name);
       foreach ($columns as $column_name) {
         // Indexes can be specified as either a column name or an array with
         // column name and length. Allow for either case.
         if (is_array($column_name)) {
           $current['indexes'][$real_name][] = array(
-            static::_fieldColumnName($field, $column_name[0]),
+            static::_fieldColumnName($storage_definition, $column_name[0]),
             $column_name[1],
           );
         }
         else {
-          $current['indexes'][$real_name][] = static::_fieldColumnName($field, $column_name);
+          $current['indexes'][$real_name][] = static::_fieldColumnName($storage_definition, $column_name);
         }
       }
     }
 
     // Add foreign keys.
     foreach ($schema['foreign keys'] as $specifier => $specification) {
-      $real_name = static::_fieldIndexName($field, $specifier);
+      $real_name = static::_fieldIndexName($storage_definition, $specifier);
       $current['foreign keys'][$real_name]['table'] = $specification['table'];
       foreach ($specification['columns'] as $column_name => $referenced) {
-        $sql_storage_column = static::_fieldColumnName($field, $column_name);
+        $sql_storage_column = static::_fieldColumnName($storage_definition, $column_name);
         $current['foreign keys'][$real_name]['columns'][$sql_storage_column] = $referenced;
       }
     }
@@ -1490,8 +1541,8 @@ public static function _fieldSqlSchema(FieldConfigInterface $field, array $schem
     $revision['fields']['revision_id']['description'] = 'The entity revision id this data is attached to';
 
     return array(
-      static::_fieldTableName($field) => $current,
-      static::_fieldRevisionTableName($field) => $revision,
+      static::_fieldTableName($storage_definition) => $current,
+      static::_fieldRevisionTableName($storage_definition) => $revision,
     );
   }
 
@@ -1505,23 +1556,26 @@ public static function _fieldSqlSchema(FieldConfigInterface $field, array $schem
    * support. Always call entity_load() before using the data found in the
    * table.
    *
-   * @param \Drupal\field\FieldConfigInterface $field
-   *   The field object.
+   * @param \Drupal\Core\Field\FieldStorageDefinitionInterface $storage_definition
+   *   The field storage definition.
+   * @param bool $is_deleted
+   *   (optional) Whether the table name holding the values of a deleted field
+   *   should be returned.
    *
    * @return string
    *   A string containing the generated name for the database table.
-   *
    */
-  static public function _fieldTableName(FieldConfigInterface $field) {
-    if ($field->deleted) {
+  static public function _fieldTableName(FieldStorageDefinitionInterface $storage_definition, $is_deleted = FALSE) {
+    if ($is_deleted) {
       // When a field is a deleted, the table is renamed to
       // {field_deleted_data_FIELD_UUID}. To make sure we don't end up with
-      // table names longer than 64 characters, we hash the uuid and return the
-      // first 10 characters so we end up with a short unique ID.
-      return "field_deleted_data_" . substr(hash('sha256', $field->uuid()), 0, 10);
+      // table names longer than 64 characters, we hash the unique storage
+      // identifier and return the first 10 characters so we end up with a short
+      // unique ID.
+      return "field_deleted_data_" . substr(hash('sha256', $storage_definition->getUniqueStorageIdentifier()), 0, 10);
     }
     else {
-      return static::_generateFieldTableName($field, FALSE);
+      return static::_generateFieldTableName($storage_definition, FALSE);
     }
   }
 
@@ -1535,22 +1589,26 @@ static public function _fieldTableName(FieldConfigInterface $field) {
    * support. Always call entity_load() before using the data found in the
    * table.
    *
-   * @param \Drupal\field\FieldConfigInterface $field
-   *   The field object.
+   * @param \Drupal\Core\Field\FieldStorageDefinitionInterface $storage_definition
+   *   The field storage definition.
+   * @param bool $is_deleted
+   *   (optional) Whether the table name holding the values of a deleted field
+   *   should be returned.
    *
    * @return string
    *   A string containing the generated name for the database table.
    */
-  static public function _fieldRevisionTableName(FieldConfigInterface $field) {
-    if ($field->deleted) {
+  static public function _fieldRevisionTableName(FieldStorageDefinitionInterface $storage_definition, $is_deleted = FALSE) {
+    if ($is_deleted) {
       // When a field is a deleted, the table is renamed to
       // {field_deleted_revision_FIELD_UUID}. To make sure we don't end up with
-      // table names longer than 64 characters, we hash the uuid and return the
-      // first 10 characters so we end up with a short unique ID.
-      return "field_deleted_revision_" . substr(hash('sha256', $field->uuid()), 0, 10);
+      // table names longer than 64 characters, we hash the unique storage
+      // identifier and return the first 10 characters so we end up with a short
+      // unique ID.
+      return "field_deleted_revision_" . substr(hash('sha256', $storage_definition->getUniqueStorageIdentifier()), 0, 10);
     }
     else {
-      return static::_generateFieldTableName($field, TRUE);
+      return static::_generateFieldTableName($storage_definition, TRUE);
     }
   }
 
@@ -1560,17 +1618,17 @@ static public function _fieldRevisionTableName(FieldConfigInterface $field) {
    * The method accounts for a maximum table name length of 64 characters, and
    * takes care of disambiguation.
    *
-   * @param \Drupal\field\FieldConfigInterface $field
-   *   The field object.
+   * @param \Drupal\Core\Field\FieldStorageDefinitionInterface $storage_definition
+   *   The field storage definition.
    * @param bool $revision
    *   TRUE for revision table, FALSE otherwise.
    *
    * @return string
    *   The final table name.
    */
-  static protected function _generateFieldTableName(FieldConfigInterface $field, $revision) {
+  static protected function _generateFieldTableName(FieldStorageDefinitionInterface $storage_definition, $revision) {
     $separator = $revision ? '_revision__' : '__';
-    $table_name = $field->entity_type . $separator .  $field->name;
+    $table_name = $storage_definition->getTargetEntityTypeId() . $separator .  $storage_definition->getName();
     // Limit the string to 48 characters, keeping a 16 characters margin for db
     // prefixes.
     if (strlen($table_name) > 48) {
@@ -1578,8 +1636,8 @@ static protected function _generateFieldTableName(FieldConfigInterface $field, $
       // field UUID.
       $separator = $revision ? '_r__' : '__';
       // Truncate to the same length for the current and revision tables.
-      $entity_type = substr($field->entity_type, 0, 34);
-      $field_hash = substr(hash('sha256', $field->uuid), 0, 10);
+      $entity_type = substr($storage_definition->getTargetEntityTypeId(), 0, 34);
+      $field_hash = substr(hash('sha256', $storage_definition->getUniqueStorageIdentifier()), 0, 10);
       $table_name = $entity_type . $separator . $field_hash;
     }
     return $table_name;
@@ -1592,8 +1650,8 @@ static protected function _generateFieldTableName(FieldConfigInterface $field, $
    * strongly discouraged. This function is not considered part of the public
    * API and modules relying on it might break even in minor releases.
    *
-   * @param \Drupal\field\FieldConfigInterface $field
-   *   The field structure
+   * @param \Drupal\Core\Field\FieldStorageDefinitionInterface $storage_definition
+   *   The field storage definition.
    * @param string $index
    *   The name of the index.
    *
@@ -1601,8 +1659,8 @@ static protected function _generateFieldTableName(FieldConfigInterface $field, $
    *   A string containing a generated index name for a field data table that is
    *   unique among all other fields.
    */
-  static public function _fieldIndexName(FieldConfigInterface $field, $index) {
-    return $field->getName() . '_' . $index;
+  static public function _fieldIndexName(FieldStorageDefinitionInterface $storage_definition, $index) {
+    return $storage_definition->getName() . '_' . $index;
   }
 
   /**
@@ -1615,8 +1673,8 @@ static public function _fieldIndexName(FieldConfigInterface $field, $index) {
    * support. Always call entity_load() before using the data found in the
    * table.
    *
-   * @param \Drupal\field\FieldConfigInterface $field
-   *   The field object.
+   * @param \Drupal\Core\Field\FieldStorageDefinitionInterface $storage_definition
+   *   The field storage definition.
    * @param string $column
    *   The name of the column.
    *
@@ -1624,8 +1682,8 @@ static public function _fieldIndexName(FieldConfigInterface $field, $index) {
    *   A string containing a generated column name for a field data table that is
    *   unique among all other fields.
    */
-  static public function _fieldColumnName(FieldConfigInterface $field, $column) {
-    return in_array($column, FieldConfig::getReservedColumns()) ? $column : $field->getName() . '_' . $column;
+  static public function _fieldColumnName(FieldStorageDefinitionInterface $storage_definition, $column) {
+    return in_array($column, FieldConfig::getReservedColumns()) ? $column : $storage_definition->getName() . '_' . $column;
   }
 
 }
diff --git a/core/lib/Drupal/Core/Entity/ContentEntityNullStorage.php b/core/lib/Drupal/Core/Entity/ContentEntityNullStorage.php
index 98dbd2f..12aeca3 100644
--- a/core/lib/Drupal/Core/Entity/ContentEntityNullStorage.php
+++ b/core/lib/Drupal/Core/Entity/ContentEntityNullStorage.php
@@ -8,7 +8,7 @@
 namespace Drupal\Core\Entity;
 
 use Drupal\Core\Entity\Query\QueryException;
-use Drupal\field\FieldInstanceConfigInterface;
+use Drupal\Core\Field\FieldDefinitionInterface;
 
 /**
  * Defines a null entity storage.
@@ -109,13 +109,14 @@ protected function doDeleteFieldItemsRevision(EntityInterface $entity) {
   /**
    * {@inheritdoc}
    */
-  protected function readFieldItemsToPurge(EntityInterface $entity, FieldInstanceConfigInterface $instance) {
+  protected function readFieldItemsToPurge(FieldDefinitionInterface $field_definition, $batch_size) {
+    return array();
   }
 
   /**
    * {@inheritdoc}
    */
-  protected function purgeFieldItems(EntityInterface $entity, FieldInstanceConfigInterface $instance) {
+  protected function purgeFieldItems(ContentEntityInterface $entity, FieldDefinitionInterface $field_definition) {
   }
 
   /**
@@ -130,4 +131,11 @@ protected function doSave($id, EntityInterface $entity) {
   protected function has($id, EntityInterface $entity) {
   }
 
+  /**
+   * {@inheritdoc}
+   */
+  public function countFieldData($storage_definition, $as_bool = FALSE) {
+    return $as_bool ? FALSE : 0;
+  }
+
 }
diff --git a/core/lib/Drupal/Core/Entity/ContentEntityStorageBase.php b/core/lib/Drupal/Core/Entity/ContentEntityStorageBase.php
index b75e9bd..946b033 100644
--- a/core/lib/Drupal/Core/Entity/ContentEntityStorageBase.php
+++ b/core/lib/Drupal/Core/Entity/ContentEntityStorageBase.php
@@ -8,9 +8,10 @@
 namespace Drupal\Core\Entity;
 
 use Drupal\Component\Utility\String;
+use Drupal\Core\Field\FieldDefinitionInterface;
+use Drupal\Core\Field\FieldStorageDefinitionInterface;
 use Drupal\Core\Cache\Cache;
 use Drupal\Core\Field\PrepareCacheInterface;
-use Drupal\field\FieldConfigInterface;
 use Drupal\field\FieldInstanceConfigInterface;
 use Symfony\Component\DependencyInjection\ContainerInterface;
 
@@ -278,32 +279,32 @@ protected function deleteFieldItemsRevision(EntityInterface $entity) {
   /**
    * {@inheritdoc}
    */
-  public function onFieldCreate(FieldConfigInterface $field) { }
+  public function onFieldCreate(FieldStorageDefinitionInterface $storage_definition) { }
 
   /**
    * {@inheritdoc}
    */
-  public function onFieldUpdate(FieldConfigInterface $field) { }
+  public function onFieldUpdate(FieldStorageDefinitionInterface $storage_definition) { }
 
   /**
    * {@inheritdoc}
    */
-  public function onFieldDelete(FieldConfigInterface $field) { }
+  public function onFieldDelete(FieldStorageDefinitionInterface $storage_definition) { }
 
   /**
    * {@inheritdoc}
    */
-  public function onInstanceCreate(FieldInstanceConfigInterface $instance) { }
+  public function onInstanceCreate(FieldDefinitionInterface $field_definition) { }
 
   /**
    * {@inheritdoc}
    */
-  public function onInstanceUpdate(FieldInstanceConfigInterface $instance) { }
+  public function onInstanceUpdate(FieldDefinitionInterface $field_definition) { }
 
   /**
    * {@inheritdoc}
    */
-  public function onInstanceDelete(FieldInstanceConfigInterface $instance) { }
+  public function onInstanceDelete(FieldDefinitionInterface $field_definition) { }
 
   /**
    * {@inheritdoc}
@@ -323,45 +324,47 @@ public function onBundleDelete($bundle) { }
   /**
    * {@inheritdoc}
    */
-  public function onFieldItemsPurge(EntityInterface $entity, FieldInstanceConfigInterface $instance) {
-    if ($values = $this->readFieldItemsToPurge($entity, $instance)) {
-      $items = \Drupal::typedDataManager()->create($instance, $values, $instance->getName(), $entity);
+  public function purgeFieldData(FieldDefinitionInterface $field_definition, $batch_size) {
+    $items_by_entity = $this->readFieldItemsToPurge($field_definition, $batch_size);
+    $count = 0;
+    foreach ($items_by_entity as $items) {
       $items->delete();
+      $this->purgeFieldItems($items->getEntity(), $field_definition);
+      $count += $items->count();
     }
-    $this->purgeFieldItems($entity, $instance);
+    return $count;
   }
 
   /**
-   * Reads values to be purged for a single field of a single entity.
+   * Reads values to be purged for a single field.
    *
    * This method is called during field data purge, on fields for which
    * onFieldDelete() or onFieldInstanceDelete() has previously run.
    *
-   * @param \Drupal\Core\Entity\EntityInterface $entity
-   *   The entity.
-   * @param \Drupal\field\FieldInstanceConfigInterface $instance
-   *   The field instance.
+   * @param \Drupal\Core\Field\FieldDefinitionInterface $field_definition
+   *   The field definition.
+   * @param $batch_size
+   *   The maximum number of field data records to purge before returning.
    *
-   * @return array
-   *   The field values, in their canonical array format (numerically indexed
-   *   array of items, each item being a property/value array).
+   * @return \Drupal\Core\Field\FieldItemListInterface[]
+   *   An array of field item lists, keyed by entity revision id.
    */
-  abstract protected function readFieldItemsToPurge(EntityInterface $entity, FieldInstanceConfigInterface $instance);
+  abstract protected function readFieldItemsToPurge(FieldDefinitionInterface $field_definition, $batch_size);
 
   /**
-   * Removes field data from storage during purge.
+   * Removes field items from storage per entity during purge.
    *
-   * @param EntityInterface $entity
-   *   The entity whose values are being purged.
-   * @param FieldInstanceConfigInterface $instance
+   * @param ContentEntityInterface $entity
+   *   The entity revision, whose values are being purged.
+   * @param \Drupal\Core\Field\FieldDefinitionInterface $field_definition
    *   The field whose values are bing purged.
    */
-  abstract protected function purgeFieldItems(EntityInterface $entity, FieldInstanceConfigInterface $instance);
+  abstract protected function purgeFieldItems(ContentEntityInterface $entity, FieldDefinitionInterface $field_definition);
 
   /**
    * {@inheritdoc}
    */
-  public function onFieldPurge(FieldConfigInterface $field) { }
+  public function finalizePurge(FieldStorageDefinitionInterface $storage_definition) { }
 
   /**
    * Checks translation statuses and invoke the related hooks if needed.
diff --git a/core/lib/Drupal/Core/Entity/FieldableEntityStorageInterface.php b/core/lib/Drupal/Core/Entity/FieldableEntityStorageInterface.php
index 2998c21..5245d8f 100644
--- a/core/lib/Drupal/Core/Entity/FieldableEntityStorageInterface.php
+++ b/core/lib/Drupal/Core/Entity/FieldableEntityStorageInterface.php
@@ -7,68 +7,68 @@
 
 namespace Drupal\Core\Entity;
 
-use Drupal\field\FieldConfigInterface;
-use Drupal\field\FieldInstanceConfigInterface;
+use Drupal\Core\Field\FieldDefinitionInterface;
+use Drupal\Core\Field\FieldStorageDefinitionInterface;
 
 interface FieldableEntityStorageInterface extends EntityStorageInterface {
 
   /**
-   * Allows reaction to the creation of a configurable field.
+   * Allows reaction to the creation of a field.
    *
-   * @param \Drupal\field\FieldConfigInterface $field
+   * @param \Drupal\Core\Field\FieldStorageDefinitionInterface $storage_definition
    *   The field being created.
    */
-  public function onFieldCreate(FieldConfigInterface $field);
+  public function onFieldCreate(FieldStorageDefinitionInterface $storage_definition);
 
   /**
-   * Allows reaction to the update of a configurable field.
+   * Allows reaction to the update of a field.
    *
-   * @param \Drupal\field\FieldConfigInterface $field
+   * @param \Drupal\Core\Field\FieldStorageDefinitionInterface $storage_definition
    *   The field being updated.
    */
-  public function onFieldUpdate(FieldConfigInterface $field);
+  public function onFieldUpdate(FieldStorageDefinitionInterface $storage_definition);
 
   /**
-   * Allows reaction to the deletion of a configurable field.
+   * Allows reaction to the deletion of a field.
    *
    * Stored values should not be wiped at once, but marked as 'deleted' so that
    * they can go through a proper purge process later on.
    *
-   * @param \Drupal\field\FieldConfigInterface $field
+   * @param \Drupal\Core\Field\FieldStorageDefinitionInterface $storage_definition
    *   The field being deleted.
    *
    * @see fieldPurgeData()
    */
-  public function onFieldDelete(FieldConfigInterface $field);
+  public function onFieldDelete(FieldStorageDefinitionInterface $storage_definition);
 
   /**
-   * Allows reaction to the creation of a configurable field instance.
+   * Allows reaction to the creation of a field.
    *
-   * @param \Drupal\field\FieldInstanceConfigInterface $instance
-   *   The instance being created.
+   * @param \Drupal\Core\Field\FieldDefinitionInterface $field_definition
+   *   The field being created.
    */
-  public function onInstanceCreate(FieldInstanceConfigInterface $instance);
+  public function onInstanceCreate(FieldDefinitionInterface $field_definition);
 
   /**
-   * Allows reaction to the update of a configurable field instance.
+   * Allows reaction to the update of a field.
    *
-   * @param \Drupal\field\FieldInstanceConfigInterface $instance
-   *   The instance being updated.
+   * @param \Drupal\Core\Field\FieldDefinitionInterface $field_definition
+   *   The field being updated.
    */
-  public function onInstanceUpdate(FieldInstanceConfigInterface $instance);
+  public function onInstanceUpdate(FieldDefinitionInterface $field_definition);
 
   /**
-   * Allows reaction to the deletion of a configurable field instance.
+   * Allows reaction to the deletion of a field.
    *
    * Stored values should not be wiped at once, but marked as 'deleted' so that
    * they can go through a proper purge process later on.
    *
-   * @param \Drupal\field\FieldInstanceConfigInterface $instance
-   *   The instance being deleted.
+   * @param \Drupal\Core\Field\FieldDefinitionInterface $field_definition
+   *   The field being deleted.
    *
    * @see fieldPurgeData()
    */
-  public function onInstanceDelete(FieldInstanceConfigInterface $instance);
+  public function onInstanceDelete(FieldDefinitionInterface $field_definition);
 
   /**
    * Allows reaction to a bundle being created.
@@ -81,8 +81,7 @@ public function onBundleCreate($bundle);
   /**
    * Allows reaction to a bundle being renamed.
    *
-   * This method runs before field instance definitions are updated with the new
-   * bundle name.
+   * This method runs before fields are updated with the new bundle name.
    *
    * @param string $bundle
    *   The name of the bundle being renamed.
@@ -94,7 +93,7 @@ public function onBundleRename($bundle, $bundle_new);
   /**
    * Allows reaction to a bundle being deleted.
    *
-   * This method runs before field and instance definitions are deleted.
+   * This method runs before fields are deleted.
    *
    * @param string $bundle
    *   The name of the bundle being deleted.
@@ -102,24 +101,39 @@ public function onBundleRename($bundle, $bundle_new);
   public function onBundleDelete($bundle);
 
   /**
-   * Purges the field data for a single field on a single entity.
+   * Purges a batch of field data.
+   *
+   * @param \Drupal\Core\Field\FieldDefinitionInterface $field_definition
+   *   The deleted field whose data is being purged.
+   * @param $batch_size
+   *   The maximum number of field data records to purge before returning.
+   *
+   * @return int
+   *   The number of field data records that have been purged.
+   */
+  public function purgeFieldData(FieldDefinitionInterface $field_definition, $batch_size);
+
+  /**
+   * Determines the number of field data records of a field.
    *
-   * The entity itself is not being deleted, and it is quite possible that
-   * other field data will remain attached to it.
+   * @param \Drupal\Core\Field\FieldStorageDefinitionInterface $storage_definition
+   *   The field for which to count data records.
+   * @param bool $as_bool
+   *   (Optional) Optimises the query for checking whether there are any records
+   *   or not. Defaults to FALSE.
    *
-   * @param \Drupal\Core\Entity\EntityInterface $entity
-   *   The entity whose field data is being purged.
-   * @param \Drupal\field\FieldInstanceConfigInterface $instance
-   *   The deleted field instance whose data is being purged.
+   * @return bool|int
+   *   The number of field data records. If $as_bool parameter is TRUE then the
+   *   value will either be TRUE or FALSE.
    */
-  public function onFieldItemsPurge(EntityInterface $entity, FieldInstanceConfigInterface $instance);
+  public function countFieldData($storage_definition, $as_bool = FALSE);
 
   /**
-   * Performs final cleanup after all data on all instances has been purged.
+   * Performs final cleanup after all data of a field has been purged.
    *
-   * @param \Drupal\field\FieldConfigInterface $instance
+   * @param \Drupal\Core\Field\FieldStorageDefinitionInterface $storage_definition
    *   The field being purged.
    */
-  public function onFieldPurge(FieldConfigInterface $field);
+  public function finalizePurge(FieldStorageDefinitionInterface $storage_definition);
 
 }
diff --git a/core/lib/Drupal/Core/Entity/Query/Sql/Tables.php b/core/lib/Drupal/Core/Entity/Query/Sql/Tables.php
index 54664d1..c530541 100644
--- a/core/lib/Drupal/Core/Entity/Query/Sql/Tables.php
+++ b/core/lib/Drupal/Core/Entity/Query/Sql/Tables.php
@@ -106,15 +106,7 @@ public function addField($field, $type, $langcode) {
       // This can either be the name of an entity base field or a configurable
       // field.
       $specifier = $specifiers[$key];
-      // Normally it is a field name, but field_purge_batch() is passing in
-      // id:$field_id so check that first.
-      /* @var \Drupal\Core\Field\FieldDefinitionInterface $field */
-      if (substr($specifier, 0, 3) == 'id:') {
-        if ($fields = entity_load_multiple_by_properties('field_config', array('uuid' => substr($specifier, 3), 'include_deleted' => TRUE))) {
-          $field = current($fields);
-        }
-      }
-      elseif (isset($field_storage_definitions[$specifier])) {
+      if (isset($field_storage_definitions[$specifier])) {
         $field = $field_storage_definitions[$specifier];
       }
       else {
diff --git a/core/lib/Drupal/Core/Field/FieldDefinition.php b/core/lib/Drupal/Core/Field/FieldDefinition.php
index cfeccf8..a7837a0 100644
--- a/core/lib/Drupal/Core/Field/FieldDefinition.php
+++ b/core/lib/Drupal/Core/Field/FieldDefinition.php
@@ -543,4 +543,12 @@ public function setCustomStorage($custom_storage) {
     return $this;
   }
 
+  /**
+   * {@inheritdoc}
+   */
+  public function getUniqueStorageIdentifier() {
+    return $this->getTargetEntityTypeId() . '-' . $this->getName();
+  }
+
+
 }
diff --git a/core/lib/Drupal/Core/Field/FieldStorageDefinitionInterface.php b/core/lib/Drupal/Core/Field/FieldStorageDefinitionInterface.php
index aa1d90e..b8961dc 100644
--- a/core/lib/Drupal/Core/Field/FieldStorageDefinitionInterface.php
+++ b/core/lib/Drupal/Core/Field/FieldStorageDefinitionInterface.php
@@ -285,4 +285,10 @@ public function getProvider();
    */
   public function hasCustomStorage();
 
+  /**
+   * @todo
+   * @return string
+   */
+  public function getUniqueStorageIdentifier();
+
 }
diff --git a/core/modules/field/field.purge.inc b/core/modules/field/field.purge.inc
index a19621d..fbeb80d 100644
--- a/core/modules/field/field.purge.inc
+++ b/core/modules/field/field.purge.inc
@@ -80,7 +80,6 @@ function field_purge_batch($batch_size, $field_uuid = NULL) {
   else {
     $instances = entity_load_multiple_by_properties('field_instance_config', array('deleted' => TRUE, 'include_deleted' => TRUE));
   }
-  $factory = \Drupal::service('entity.query');
   $info = \Drupal::entityManager()->getDefinitions();
   foreach ($instances as $instance) {
     $entity_type = $instance->entity_type;
@@ -92,36 +91,16 @@ function field_purge_batch($batch_size, $field_uuid = NULL) {
       continue;
     }
 
-    $ids = (object) array(
-      'entity_type' => $entity_type,
-      'bundle' => $instance->bundle,
-    );
-    // Retrieve some entities.
-    $query = $factory->get($entity_type)
-      ->condition('id:' . $instance->getField()->uuid() . '.deleted', 1)
-      ->range(0, $batch_size);
-    // If there's no bundle key, all results will have the same bundle.
-    if ($bundle_key = $info[$entity_type]->getKey('bundle')) {
-      $query->condition($bundle_key, $ids->bundle);
-    }
-    $results = $query->execute();
-    if ($results) {
-      foreach ($results as $revision_id => $entity_id) {
-        $ids->revision_id = $revision_id;
-        $ids->entity_id = $entity_id;
-        $entity = _field_create_entity_from_ids($ids);
-        \Drupal::entityManager()->getStorage($entity_type)->onFieldItemsPurge($entity, $instance);
-        $batch_size--;
-      }
-      // Only delete up to the maximum number of records.
-      if ($batch_size == 0) {
-        break;
-      }
-    }
-    else {
+    $count_purged = \Drupal::entityManager()->getStorage($entity_type)->purgeFieldData($instance, $batch_size);
+    if ($count_purged < $batch_size) {
       // No field data remains for the instance, so we can remove it.
       field_purge_instance($instance);
     }
+    $batch_size -= $count_purged;
+    // Only delete up to the maximum number of records.
+    if ($batch_size == 0) {
+      break;
+    }
   }
 
   // Retrieve all deleted fields. Any that have no instances can be purged.
@@ -187,7 +166,7 @@ function field_purge_field($field) {
   $state->set('field.field.deleted', $deleted_fields);
 
   // Notify the storage layer.
-  \Drupal::entityManager()->getStorage($field->entity_type)->onFieldPurge($field);
+  \Drupal::entityManager()->getStorage($field->entity_type)->finalizePurge($field);
 
   // Invoke external hooks after the cache is cleared for API consistency.
   \Drupal::moduleHandler()->invokeAll('field_purge_field', array($field));
diff --git a/core/modules/field/src/ConfigImporterFieldPurger.php b/core/modules/field/src/ConfigImporterFieldPurger.php
index db3d8b9..a83ff52 100644
--- a/core/modules/field/src/ConfigImporterFieldPurger.php
+++ b/core/modules/field/src/ConfigImporterFieldPurger.php
@@ -75,7 +75,8 @@ protected static function initializeSandbox(array &$context, ConfigImporter $con
     $context['sandbox']['field']['steps_to_delete'] = 0;
     $fields = static::getFieldsToPurge($context['sandbox']['field']['extensions'], $config_importer->getUnprocessedConfiguration('delete'));
     foreach ($fields as $field) {
-      $row_count = $field->entityCount();
+      $row_count = \Drupal::entityManager()->getStorage($field->getTargetEntityTypeId())
+        ->countFieldData($field);
       if ($row_count > 0) {
         // The number of steps to delete each field is determined by the
         // purge_batch_size setting. For example if the field has 9 rows and the
diff --git a/core/modules/field/src/Entity/FieldConfig.php b/core/modules/field/src/Entity/FieldConfig.php
index 3f5b787..838488d 100644
--- a/core/modules/field/src/Entity/FieldConfig.php
+++ b/core/modules/field/src/Entity/FieldConfig.php
@@ -661,59 +661,7 @@ public static function getReservedColumns() {
    *   TRUE if the field has data for any entity; FALSE otherwise.
    */
   public function hasData() {
-    return $this->entityCount(TRUE);
-  }
-
-  /**
-   * Determines the number of entities that have field data.
-   *
-   * @param bool $as_bool
-   *   (Optional) Optimises query for hasData(). Defaults to FALSE.
-   *
-   * @return bool|int
-   *   The number of entities that have field data. If $as_bool parameter is
-   *   TRUE then the value will either be TRUE or FALSE.
-   */
-  public function entityCount($as_bool = FALSE) {
-    $count = 0;
-    $factory = \Drupal::service('entity.query');
-    $entity_type = \Drupal::entityManager()->getDefinition($this->entity_type);
-    // Entity Query throws an exception if there is no base table.
-    if ($entity_type->getBaseTable()) {
-      if ($this->deleted) {
-        $query = $factory->get($this->entity_type)
-          ->condition('id:' . $this->uuid() . '.deleted', 1);
-      }
-      elseif ($this->getBundles()) {
-        $storage_details = $this->getSchema();
-        $columns = array_keys($storage_details['columns']);
-        $query = $factory->get($this->entity_type);
-        $group = $query->orConditionGroup();
-        foreach ($columns as $column) {
-          $group->exists($this->name . '.' . $column);
-        }
-        $query = $query->condition($group);
-      }
-
-      if (isset($query)) {
-        $query
-          ->count()
-          ->accessCheck(FALSE);
-        // If we are performing the query just to check if the field has data
-        // limit the number of rows returned by the subquery.
-        if ($as_bool) {
-          $query->range(0, 1);
-        }
-        $count = $query->execute();
-      }
-    }
-
-    if ($as_bool) {
-      return (bool) $count;
-    }
-    else {
-      return (int) $count;
-    }
+    return \Drupal::entityManager()->getStorage($this->entity_type)->countFieldData($this, TRUE);
   }
 
   /**
@@ -790,6 +738,14 @@ public function getMainPropertyName() {
   }
 
   /**
+   * {@inheritdoc}
+   */
+  public function getUniqueStorageIdentifier() {
+    return $this->uuid();
+  }
+
+
+  /**
    * Helper to retrieve the field item class.
    */
   protected function getFieldItemClass() {
diff --git a/core/modules/field/src/Entity/FieldInstanceConfig.php b/core/modules/field/src/Entity/FieldInstanceConfig.php
index f655a18..2b4ff56 100644
--- a/core/modules/field/src/Entity/FieldInstanceConfig.php
+++ b/core/modules/field/src/Entity/FieldInstanceConfig.php
@@ -821,4 +821,11 @@ public static function loadByName($entity_type_id, $bundle, $field_name) {
     return \Drupal::entityManager()->getStorage('field_instance_config')->load($entity_type_id . '.' . $bundle . '.' . $field_name);
   }
 
+  /**
+   * {@inheritdoc}
+   */
+  public function getUniqueStorageIdentifier() {
+    return $this->field->uuid();
+  }
+
 }
diff --git a/core/modules/field/src/Tests/BulkDeleteTest.php b/core/modules/field/src/Tests/BulkDeleteTest.php
index 542139e..055aa3a 100644
--- a/core/modules/field/src/Tests/BulkDeleteTest.php
+++ b/core/modules/field/src/Tests/BulkDeleteTest.php
@@ -308,7 +308,7 @@ function testPurgeField() {
     $this->checkHooksInvocations($hooks, $actual_hooks);
 
     // Purge again to purge the instance.
-    field_purge_batch(0);
+    field_purge_batch(1);
 
     // The field still exists, not deleted.
     $fields = entity_load_multiple_by_properties('field_config', array('uuid' => $field->uuid, 'include_deleted' => TRUE));
@@ -340,7 +340,7 @@ function testPurgeField() {
     $this->assertTrue(isset($fields[$field->uuid]) && $fields[$field->uuid]->deleted, 'The field exists and is deleted');
 
     // Purge again to purge the instance and the field.
-    field_purge_batch(0);
+    field_purge_batch(1);
 
     // The field is gone.
     $fields = entity_load_multiple_by_properties('field_config', array('uuid' => $field->uuid, 'include_deleted' => TRUE));
diff --git a/core/modules/field/src/Tests/FieldDataCountTest.php b/core/modules/field/src/Tests/FieldDataCountTest.php
new file mode 100644
index 0000000..5dc854a
--- /dev/null
+++ b/core/modules/field/src/Tests/FieldDataCountTest.php
@@ -0,0 +1,111 @@
+<?php
+
+/**
+ * @file
+ * Contains \Drupal\field\Tests\FieldDataCountTest.
+ */
+
+namespace Drupal\field\Tests;
+
+use Drupal\Core\Entity\ContentEntityDatabaseStorage;
+
+/**
+ * Tests counting field data records.
+ *
+ * @see \Drupal\Core\Entity\FieldableEntityStorageInterface::countFieldData()
+ * @see \Drupal\field\Entity\FieldConfig::hasData()
+ */
+class FieldDataCountTest extends FieldUnitTestBase {
+
+  /**
+   * @var \Drupal\Core\Entity\FieldableEntityStorageInterface
+   */
+  protected $storage;
+
+  /**
+   * {@inheritdoc}
+   */
+  public static function getInfo() {
+    return array(
+      'name' => 'Field config hasData() tests.',
+      'description' => 'Tests counting field data records and the hasData() method on FieldConfig entity.',
+      'group' => 'Field API',
+    );
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function setUp() {
+    parent::setUp();
+    $this->storage = \Drupal::entityManager()->getStorage('entity_test');
+  }
+
+  /**
+   * Tests entityCount() and hadData() methods.
+   */
+  public function testEntityCountAndHasData() {
+    // Create a field with a cardinality of 2 to show that we are counting
+    // entities and not rows in a table.
+    /** @var \Drupal\field\Entity\FieldConfig $field */
+    $field = entity_create('field_config', array(
+      'name' => 'field_int',
+      'entity_type' => 'entity_test',
+      'type' => 'integer',
+      'cardinality' => 2,
+    ));
+    $field->save();
+    entity_create('field_instance_config', array(
+      'entity_type' => 'entity_test',
+      'field_name' => 'field_int',
+      'bundle' => 'entity_test',
+    ))->save();
+
+    $this->assertIdentical($field->hasdata(), FALSE, 'There are no entities with field data.');
+    $this->assertIdentical($this->storage->countFieldData($field), 0, 'There are 0 entities with field data.');
+
+    // Create 1 entity without the field.
+    $entity = entity_create('entity_test');
+    $entity->name->value = $this->randomName();
+    $entity->save();
+
+    $this->assertIdentical($field->hasdata(), FALSE, 'There are no entities with field data.');
+    $this->assertIdentical($this->storage->countFieldData($field), 0, 'There are 0 entities with field data.');
+
+    // Create 12 entities to ensure that the purging works as expected.
+    for ($i=0; $i < 12; $i++) {
+      $entity = entity_create('entity_test');
+      $value = mt_rand(1,99);
+      $value2 = mt_rand(1,99);
+      $entity->field_int[0]->value = $value;
+      $entity->field_int[1]->value = $value2;
+      $entity->name->value = $this->randomName();
+      $entity->save();
+    }
+
+    $storage = \Drupal::entityManager()->getStorage('entity_test');
+    if ($storage instanceof ContentEntityDatabaseStorage) {
+      // Count the actual number of rows in the field table.
+      $field_table_name = $storage->_fieldTableName($field);
+      $result = db_select($field_table_name, 't')
+        ->fields('t')
+        ->countQuery()
+        ->execute()
+        ->fetchField();
+      $this->assertEqual($result, 24, 'The field table has 24 rows.');
+    }
+
+    $this->assertIdentical($field->hasdata(), TRUE, 'There are entities with field data.');
+    $this->assertEqual($this->storage->countFieldData($field), 24, 'There are 24 rows of field data.');
+
+    // Ensure the methods work on deleted fields.
+    $field->delete();
+    $this->assertIdentical($field->hasdata(), TRUE, 'There are entities with deleted field data.');
+    $this->assertEqual($this->storage->countFieldData($field), 24, 'There are 24 rows of deleted field data.');
+
+    field_purge_batch(6);
+    $this->assertIdentical($field->hasdata(), TRUE, 'There are entities with deleted field data.');
+    $this->assertEqual($this->storage->countFieldData($field), 18, 'There are 18 rows of deleted field data.');
+  }
+
+}
diff --git a/core/modules/field/src/Tests/FieldEntityCountTest.php b/core/modules/field/src/Tests/FieldEntityCountTest.php
deleted file mode 100644
index f08cee4..0000000
--- a/core/modules/field/src/Tests/FieldEntityCountTest.php
+++ /dev/null
@@ -1,95 +0,0 @@
-<?php
-
-/**
- * @file
- * Contains \Drupal\field\Tests\FieldEntityCountTest.
- */
-
-namespace Drupal\field\Tests;
-
-use Drupal\Core\Entity\ContentEntityDatabaseStorage;
-
-/**
- * Tests entityCount() and hasData() methods on FieldConfig entity.
- *
- * @see \Drupal\field\Entity\FieldConfig::entityCount()
- * @see \Drupal\field\Entity\FieldConfig::hasData()
- */
-class FieldEntityCountTest extends FieldUnitTestBase {
-
-  public static function getInfo() {
-    return array(
-      'name' => 'Field config entityCount() and hasData() tests.',
-      'description' => 'Tests entityCount() and hasData() methods on FieldConfig entity.',
-      'group' => 'Field API',
-    );
-  }
-
-  /**
-   * Tests entityCount() and hadData() methods.
-   */
-  public function testEntityCountAndHasData() {
-    // Create a field with a cardinality of 2 to show that we are counting
-    // entities and not rows in a table.
-    /** @var \Drupal\field\Entity\FieldConfig $field */
-    $field = entity_create('field_config', array(
-      'name' => 'field_int',
-      'entity_type' => 'entity_test',
-      'type' => 'integer',
-      'cardinality' => 2,
-    ));
-    $field->save();
-    entity_create('field_instance_config', array(
-      'entity_type' => 'entity_test',
-      'field_name' => 'field_int',
-      'bundle' => 'entity_test',
-    ))->save();
-
-    $this->assertIdentical($field->hasdata(), FALSE, 'There are no entities with field data.');
-    $this->assertIdentical($field->entityCount(), 0, 'There are 0 entities with field data.');
-
-    // Create 1 entity without the field.
-    $entity = entity_create('entity_test');
-    $entity->name->value = $this->randomName();
-    $entity->save();
-
-    $this->assertIdentical($field->hasdata(), FALSE, 'There are no entities with field data.');
-    $this->assertIdentical($field->entityCount(), 0, 'There are 0 entities with field data.');
-
-    // Create 12 entities to ensure that the purging works as expected.
-    for ($i=0; $i < 12; $i++) {
-      $entity = entity_create('entity_test');
-      $value = mt_rand(1,99);
-      $value2 = mt_rand(1,99);
-      $entity->field_int[0]->value = $value;
-      $entity->field_int[1]->value = $value2;
-      $entity->name->value = $this->randomName();
-      $entity->save();
-    }
-
-    $storage = \Drupal::entityManager()->getStorage('entity_test');
-    if ($storage instanceof ContentEntityDatabaseStorage) {
-      // Count the actual number of rows in the field table.
-      $field_table_name = $storage->_fieldTableName($field);
-      $result = db_select($field_table_name, 't')
-        ->fields('t')
-        ->countQuery()
-        ->execute()
-        ->fetchField();
-      $this->assertEqual($result, 24, 'The field table has 24 rows.');
-    }
-
-    $this->assertIdentical($field->hasdata(), TRUE, 'There are entities with field data.');
-    $this->assertEqual($field->entityCount(), 12, 'There are 12 entities with field data.');
-
-    // Ensure the methods work on deleted fields.
-    $field->delete();
-    $this->assertIdentical($field->hasdata(), TRUE, 'There are entities with deleted field data.');
-    $this->assertEqual($field->entityCount(), 12, 'There are 12 entities with deleted field data.');
-
-    field_purge_batch(6);
-    $this->assertIdentical($field->hasdata(), TRUE, 'There are entities with deleted field data.');
-    $this->assertEqual($field->entityCount(), 6, 'There are 6 entities with deleted field data.');
-  }
-
-}
diff --git a/core/modules/system/src/Tests/Entity/FieldSqlStorageTest.php b/core/modules/system/src/Tests/Entity/FieldSqlStorageTest.php
index c252acd..ab07227 100644
--- a/core/modules/system/src/Tests/Entity/FieldSqlStorageTest.php
+++ b/core/modules/system/src/Tests/Entity/FieldSqlStorageTest.php
@@ -563,9 +563,9 @@ public function testTableNames() {
       'deleted' => TRUE,
     ));
     $expected = 'field_deleted_data_' . substr(hash('sha256', $field->uuid), 0, 10);
-    $this->assertEqual(ContentEntityDatabaseStorage::_fieldTableName($field), $expected);
+    $this->assertEqual(ContentEntityDatabaseStorage::_fieldTableName($field, TRUE), $expected);
     $expected = 'field_deleted_revision_' . substr(hash('sha256', $field->uuid), 0, 10);
-    $this->assertEqual(ContentEntityDatabaseStorage::_fieldRevisionTableName($field), $expected);
+    $this->assertEqual(ContentEntityDatabaseStorage::_fieldRevisionTableName($field, TRUE), $expected);
   }
 
 }
diff --git a/core/tests/Drupal/Tests/Core/Entity/ContentEntityDatabaseStorageTest.php b/core/tests/Drupal/Tests/Core/Entity/ContentEntityDatabaseStorageTest.php
index cc77d6d..fcc85df 100644
--- a/core/tests/Drupal/Tests/Core/Entity/ContentEntityDatabaseStorageTest.php
+++ b/core/tests/Drupal/Tests/Core/Entity/ContentEntityDatabaseStorageTest.php
@@ -955,15 +955,17 @@ public function testFieldSqlSchemaForEntityWithStringIdentifier() {
       ->method('getBaseFieldDefinitions')
       ->will($this->returnValue($this->fieldDefinitions));
 
-    // Define a field definition for a test_field field.
-    $field = $this->getMock('\Drupal\field\FieldConfigInterface');
+    // Define a field definition for a test_field fuuidield.
+    $field = $this->getMock('\Drupal\Core\Field\FieldStorageDefinitionInterface');
     $field->deleted = FALSE;
-    $field->entity_type = 'test_entity';
-    $field->name = 'test_field';
 
     $field->expects($this->any())
       ->method('getName')
-      ->will($this->returnValue('test'));
+      ->will($this->returnValue('test_field'));
+
+    $field->expects($this->any())
+      ->method('getTargetEntityTypeId')
+      ->will($this->returnValue('test_entity'));
 
     $field_schema = array(
       'columns' => array(