- Patch #443422 by yched, bjaspan | chx, merlinofchaos, Scott Reynolds, plach, profix898, mattyoung: added support for pluggable 'per field' storage engine. Comes with documentation and tests.

The Field Attach API uses the Field Storage API to perform all "database access". Each Field Storage API hook function defines a primitive database operation such as read, write, or delete. The default field storage module, field_sql_storage.module, uses the local SQL database to implement these operations, but alternative field storage backends can choose to represent the data in SQL differently or use a completely different storage mechanism such as a cloud-based database.

11 files changed, 1140 insertions, 305 deletions

diff --git a/modules/field/field.api.php b/modules/field/field.api.php
index 6fd2cef32..380444af1 100644
--- a/modules/field/field.api.php
+++ b/modules/field/field.api.php
@@ -70,10 +70,10 @@ function hook_field_extra_fields($bundle) {
 /**
  * @defgroup field_types Field Types API
  * @{
- * Define field types, widget types, and display formatter types.
+ * Define field types, widget types, display formatter types, storage types.
  *
  * The bulk of the Field Types API are related to field types. A field type
- * represents a particular data storage type (integer, string, date, etc.) that
+ * represents a particular type of data (integer, string, date, etc.) that
  * can be attached to a fieldable object. hook_field_info() defines the basic
  * properties of a field type, and a variety of other field hooks are called by
  * the Field Attach API to perform field-type-specific actions.
@@ -97,6 +97,9 @@ function hook_field_extra_fields($bundle) {
  * behavior of existing field types.
  * @see hook_field_widget_info().
  * @see hook_field_formatter_info().
+ *
+ * A third kind of pluggable handlers, storage backends, is defined by the
+ * @link field_storage Field Storage API @endlink.
  */
 
 /**
@@ -1097,6 +1100,45 @@ function hook_field_attach_delete_bundle($bundle, $instances) {
  */
 
 /**
+ * Expose Field API storage backends.
+ *
+ * @return
+ *   An array describing the storage backends implemented by the module.
+ *   The keys are storage backend names. To avoid name clashes, storage backend
+ *   names should be prefixed with the name of the module that exposes them.
+ *   The values are arrays describing the storage backend, with the following
+ *   key/value pairs:
+ *   - label: The human-readable name of the storage backend.
+ *   - description: A short description for the storage backend.
+ *   - settings: An array whose keys are the names of the settings available
+ *     for the storage backend, and whose values are the default values for
+ *     those settings.
+ */
+function hook_field_storage_info() {
+  return array(
+    'field_sql_storage' => array(
+      'label' => t('Default SQL storage'),
+      'description' => t('Stores fields in the local SQL database, using per-field tables.'),
+      'settings' => array(),
+    ),
+  );
+}
+
+/**
+ * Perform alterations on Field API storage types.
+ *
+ * @param $info
+ *   Array of informations on storage types exposed by
+ *   hook_field_field_storage_info() implementations.
+ */
+function hook_field_storage_info_alter(&$info) {
+  // Add a setting to a storage type.
+  $info['field_sql_storage']['settings'] += array(
+    'mymodule_additional_setting' => 'default value',
+  );
+}
+
+/**
  * Load field data for a set of objects.
  *
  * @param $obj_type
@@ -1107,15 +1149,15 @@ function hook_field_attach_delete_bundle($bundle, $instances) {
  *   FIELD_LOAD_CURRENT to load the most recent revision for all
  *   fields, or FIELD_LOAD_REVISION to load the version indicated by
  *   each object.
- * @param $skip_fields
- *   An array keyed by field ids whose data has already been loaded and
- *   therefore should not be loaded again. The values associated to these keys
- *   are not specified.
+ * @param $fields
+ *   An array listing the fields to be loaded. The keys of the array are field
+ *   ids, the values of the array are the object ids (or revision ids,
+ *   depending on the $age parameter) to be loaded for each field.
  * @return
  *   Loaded field values are added to $objects. Fields with no values should be
  *   set as an empty array.
  */
-function hook_field_storage_load($obj_type, $objects, $age, $skip_fields) {
+function hook_field_storage_load($obj_type, $objects, $age, $fields) {
 }
 
 /**
@@ -1128,12 +1170,11 @@ function hook_field_storage_load($obj_type, $objects, $age, $skip_fields) {
  * @param $op
  *   FIELD_STORAGE_UPDATE when updating an existing object,
  *   FIELD_STORAGE_INSERT when inserting a new object.
- * @param $skip_fields
- *   An array keyed by field ids whose data has already been written and
- *   therefore should not be written again. The values associated to these keys
- *   are not specified.
+ * @param $fields
+ *   An array listing the fields to be written. The keys and values of the
+ *   array are field ids.
  */
-function hook_field_storage_write($obj_type, $object, $op, $skip_fields) {
+function hook_field_storage_write($obj_type, $object, $op, $fields) {
 }
 
 /**
@@ -1143,8 +1184,11 @@ function hook_field_storage_write($obj_type, $object, $op, $skip_fields) {
  *   The entity type of object, such as 'node' or 'user'.
  * @param $object
  *   The object on which to operate.
+ * @param $fields
+ *   An array listing the fields to delete. The keys and values of the
+ *   array are field ids.
  */
-function hook_field_storage_delete($obj_type, $object) {
+function hook_field_storage_delete($obj_type, $object, $fields) {
 }
 
 /**
@@ -1159,8 +1203,11 @@ function hook_field_storage_delete($obj_type, $object) {
  *   The object on which to operate. The revision to delete is
  *   indicated by the object's revision id property, as identified by
  *   hook_fieldable_info() for $obj_type.
+ * @param $fields
+ *   An array listing the fields to delete. The keys and values of the
+ *   array are field ids.
  */
-function hook_field_storage_delete_revision($obj_type, $object) {
+function hook_field_storage_delete_revision($obj_type, $object, $fields) {
 }
 
 /**
@@ -1186,26 +1233,6 @@ function hook_field_storage_query($field_name, $conditions, $count, &$cursor = N
 }
 
 /**
- * Act on creation of a new bundle.
- *
- * @param $bundle
- *   The name of the bundle being created.
- */
-function hook_field_storage_create_bundle($bundle) {
-}
-
-/**
- * Act on a bundle being renamed.
- *
- * @param $bundle_old
- *   The old name of the bundle.
- * @param $bundle_new
- *   The new name of the bundle.
- */
-function hook_field_storage_rename_bundle($bundle_old, $bundle_new) {
-}
-
-/**
  * Act on creation of a new field.
  *
  * @param $field
@@ -1217,21 +1244,19 @@ function hook_field_storage_create_field($field) {
 /**
  * Act on deletion of a field.
  *
- * @param $field_name
- *   The name of the field being deleted.
+ * @param $field
+ *   The field being deleted.
  */
-function hook_field_storage_delete_field($field_name) {
+function hook_field_storage_delete_field($field) {
 }
 
 /**
  * Act on deletion of a field instance.
  *
- * @param $field_name
- *   The name of the field in the new instance.
- * @param $bundle
- *   The name of the bundle in the new instance.
+ * @param $instance
+ *   The instance being deleted.
  */
-function hook_field_storage_delete_instance($field_name, $bundle) {
+function hook_field_storage_delete_instance($instance) {
 }
 
 /**
diff --git a/modules/field/field.attach.inc b/modules/field/field.attach.inc
index 7571ee4fc..e2726a776 100644
--- a/modules/field/field.attach.inc
+++ b/modules/field/field.attach.inc
@@ -44,17 +44,16 @@ class FieldQueryException extends FieldException {}
  * @{
  * Implement a storage engine for Field API data.
  *
- * The Field Attach API uses the Field Storage API to perform all
- * "database access". Each Field Storage API hook function defines a
- * primitive database operation such as read, write, or delete. The
- * default field storage module, field_sql_storage.module, uses the
- * local SQL database to implement these operations, but alternative
- * field storage engines can choose to represent the data in SQL
- * differently or use a completely different storage mechanism such as
- * a cloud-based database.
+ * The Field Attach API uses the Field Storage API to perform all "database
+ * access". Each Field Storage API hook function defines a primitive database
+ * operation such as read, write, or delete. The default field storage module,
+ * field_sql_storage.module, uses the local SQL database to implement these
+ * operations, but alternative field storage backends can choose to represent
+ * the data in SQL differently or use a completely different storage mechanism
+ * such as a cloud-based database.
  *
- * The Drupal system variable field_storage_module identifies the
- * field storage module to use.
+ * Each field defines which storage backend it uses. The Drupal system variable
+ * 'field_default_storage' identifies the storage backend used by default.
  */
 
 /**
@@ -525,9 +524,8 @@ function field_attach_form($obj_type, $object, &$form, &$form_state, $langcode =
  *  - 'deleted': If TRUE, the function will operate on deleted fields
  *    as well as non-deleted fields. If unset or FALSE, only
  *    non-deleted fields are operated on.
- * @returns
- *   Loaded field values are added to $objects. Fields with no values should be
- *   set as an empty array.
+ * @return
+ *   Loaded field values are added to $objects.
  */
 function field_attach_load($obj_type, $objects, $age = FIELD_LOAD_CURRENT, $options = array()) {
   $load_current = $age == FIELD_LOAD_CURRENT;
@@ -578,7 +576,7 @@ function field_attach_load($obj_type, $objects, $age = FIELD_LOAD_CURRENT, $opti
   if ($queried_objects) {
     // The invoke order is:
     // - hook_field_attach_pre_load()
-    // - storage engine's hook_field_storage_load()
+    // - storage backend's hook_field_storage_load()
     // - field-type module's hook_field_load()
     // - hook_field_attach_load()
 
@@ -590,9 +588,39 @@ function field_attach_load($obj_type, $objects, $age = FIELD_LOAD_CURRENT, $opti
       $function($obj_type, $queried_objects, $age, $skip_fields, $options);
     }
 
-    // Invoke the storage engine's hook_field_storage_load(): the field storage
-    // engine loads the rest.
-    module_invoke(variable_get('field_storage_module', 'field_sql_storage'), 'field_storage_load', $obj_type, $queried_objects, $age, $skip_fields, $options);
+    // Collect the storage backends used by the remaining fields in the objects.
+    $storages = array();
+    foreach ($queried_objects as $obj) {
+      list($id, $vid, $bundle) = field_extract_ids($obj_type, $obj);
+      if ($options['deleted']) {
+        $instances = field_read_instances(array('bundle' => $bundle), array('include_deleted' => $options['deleted']));
+      }
+      else {
+        $instances = field_info_instances($bundle);
+      }
+
+      foreach ($instances as $instance) {
+        if (!isset($options['field_id']) || $options['field_id'] == $instance['field_id']) {
+          $field_name = $instance['field_name'];
+          $field_id = $instance['field_id'];
+          // Make sure all fields are present at least as empty arrays.
+          if (!isset($queried_objects[$id]->{$field_name})) {
+            $queried_objects[$id]->{$field_name} = array();
+          }
+          // Collect the storage backend if the field has not been loaded yet.
+          if (!isset($skip_fields[$field_id])) {
+            $field = field_info_field_by_id($field_id);
+            $storages[$field['storage']['type']][$field_id][] = $load_current ? $id : $vid;
+          }
+        }
+      }
+    }
+
+    // Invoke hook_field_storage_load() on the relevant storage backends.
+    foreach ($storages as $storage => $fields) {
+      $storage_info = field_info_storage_types($storage);
+      module_invoke($storage_info['module'], 'field_storage_load', $obj_type, $queried_objects, $age, $fields, $options);
+    }
 
     // Invoke field-type module's hook_field_load().
     _field_invoke_multiple('load', $obj_type, $queried_objects, $age, $options);
@@ -791,6 +819,8 @@ function field_attach_insert($obj_type, $object) {
   _field_invoke_default('insert', $obj_type, $object);
   _field_invoke('insert', $obj_type, $object);
 
+  list($id, $vid, $bundle, $cacheable) = field_extract_ids($obj_type, $object);
+
   // Let other modules act on inserting the object, accumulating saved
   // fields along the way.
   $skip_fields = array();
@@ -799,10 +829,26 @@ function field_attach_insert($obj_type, $object) {
     $function($obj_type, $object, $skip_fields);
   }
 
-  // Field storage module saves any remaining unsaved fields.
-  module_invoke(variable_get('field_storage_module', 'field_sql_storage'), 'field_storage_write', $obj_type, $object, FIELD_STORAGE_INSERT, $skip_fields);
+  // Collect the storage backends used by the remaining fields in the objects.
+  $storages = array();
+  foreach (field_info_instances($bundle) as $instance) {
+    $field = field_info_field_by_id($instance['field_id']);
+    $field_id = $field['id'];
+    $field_name = $field['field_name'];
+    if (!empty($object->$field_name)) {
+      // Collect the storage backend if the field has not been written yet.
+      if (!isset($skip_fields[$field_id])) {
+        $storages[$field['storage']['type']][$field_id] = $field_id;
+      }
+    }
+  }
+
+  // Field storage backends save any remaining unsaved fields.
+  foreach ($storages as $storage => $fields) {
+    $storage_info = field_info_storage_types($storage);
+    module_invoke($storage_info['module'], 'field_storage_write', $obj_type, $object, FIELD_STORAGE_INSERT, $fields);
+  }
 
-  list($id, $vid, $bundle, $cacheable) = field_extract_ids($obj_type, $object);
   if ($cacheable) {
     cache_clear_all("field:$obj_type:$id", 'cache_field');
   }
@@ -819,6 +865,8 @@ function field_attach_insert($obj_type, $object) {
 function field_attach_update($obj_type, $object) {
   _field_invoke('update', $obj_type, $object);
 
+  list($id, $vid, $bundle, $cacheable) = field_extract_ids($obj_type, $object);
+
   // Let other modules act on updating the object, accumulating saved
   // fields along the way.
   $skip_fields = array();
@@ -827,10 +875,30 @@ function field_attach_update($obj_type, $object) {
     $function($obj_type, $object, $skip_fields);
   }
 
-  // Field storage module saves any remaining unsaved fields.
-  module_invoke(variable_get('field_storage_module', 'field_sql_storage'), 'field_storage_write', $obj_type, $object, FIELD_STORAGE_UPDATE, $skip_fields);
+  // Collect the storage backends used by the remaining fields in the objects.
+  $storages = array();
+  foreach (field_info_instances($bundle) as $instance) {
+    $field = field_info_field_by_id($instance['field_id']);
+    $field_id = $field['id'];
+    $field_name = $field['field_name'];
+    // Leave the field untouched if $object comes with no $field_name property,
+    // but empty the field if it comes as a NULL value or an empty array.
+    // Function property_exists() is slower, so we catch the more frequent
+    // cases where it's an empty array with the faster isset().
+    if (isset($object->$field_name) || property_exists($object, $field_name)) {
+      // Collect the storage backend if the field has not been written yet.
+      if (!isset($skip_fields[$field_id])) {
+        $storages[$field['storage']['type']][$field_id] = $field_id;
+      }
+    }
+  }
+
+  // Field storage backends save any remaining unsaved fields.
+  foreach ($storages as $storage => $fields) {
+    $storage_info = field_info_storage_types($storage);
+    module_invoke($storage_info['module'], 'field_storage_write', $obj_type, $object, FIELD_STORAGE_UPDATE, $fields);
+  }
 
-  list($id, $vid, $bundle, $cacheable) = field_extract_ids($obj_type, $object);
   if ($cacheable) {
     cache_clear_all("field:$obj_type:$id", 'cache_field');
   }
@@ -847,7 +915,22 @@ function field_attach_update($obj_type, $object) {
  */
 function field_attach_delete($obj_type, $object) {
   _field_invoke('delete', $obj_type, $object);
-  module_invoke(variable_get('field_storage_module', 'field_sql_storage'), 'field_storage_delete', $obj_type, $object);
+
+  list($id, $vid, $bundle, $cacheable) = field_extract_ids($obj_type, $object);
+
+  // Collect the storage backends used by the fields in the objects.
+  $storages = array();
+  foreach (field_info_instances($bundle) as $instance) {
+    $field = field_info_field_by_id($instance['field_id']);
+    $field_id = $field['id'];
+    $storages[$field['storage']['type']][$field_id] = $field_id;
+  }
+
+  // Field storage backends delete their data.
+  foreach ($storages as $storage => $fields) {
+    $storage_info = field_info_storage_types($storage);
+    module_invoke($storage_info['module'], 'field_storage_delete', $obj_type, $object, $fields);
+  }
 
   // Let other modules act on deleting the object.
   foreach (module_implements('field_attach_delete') as $module) {
@@ -855,7 +938,6 @@ function field_attach_delete($obj_type, $object) {
     $function($obj_type, $object);
   }
 
-  list($id, $vid, $bundle, $cacheable) = field_extract_ids($obj_type, $object);
   if ($cacheable) {
     cache_clear_all("field:$obj_type:$id", 'cache_field');
   }
@@ -872,7 +954,22 @@ function field_attach_delete($obj_type, $object) {
  */
 function field_attach_delete_revision($obj_type, $object) {
   _field_invoke('delete_revision', $obj_type, $object);
-  module_invoke(variable_get('field_storage_module', 'field_sql_storage'), 'field_storage_delete_revision', $obj_type, $object);
+
+  list($id, $vid, $bundle, $cacheable) = field_extract_ids($obj_type, $object);
+
+  // Collect the storage backends used by the fields in the objects.
+  $storages = array();
+  foreach (field_info_instances($bundle) as $instance) {
+    $field = field_info_field_by_id($instance['field_id']);
+    $field_id = $field['id'];
+    $storages[$field['storage']['type']][$field_id] = $field_id;
+  }
+
+  // Field storage backends delete their data.
+  foreach ($storages as $storage => $fields) {
+    $storage_info = field_info_storage_types($storage);
+    module_invoke($storage_info['module'], 'field_storage_delete_revision', $obj_type, $object, $fields);
+  }
 
   // Let other modules act on deleting the revision.
   foreach (module_implements('field_attach_delete_revision') as $module) {
@@ -977,7 +1074,8 @@ function field_attach_query($field_id, $conditions, $count, &$cursor = NULL, $ag
   }
   // If the request hasn't been handled, let the storage engine handle it.
   if (!$skip_field) {
-    $function = variable_get('field_storage_module', 'field_sql_storage') . '_field_storage_query';
+    $field = field_info_field_by_id($field_id);
+    $function = $field['storage']['module'] . '_field_storage_query';
     $results = $function($field_id, $conditions, $count, $cursor, $age);
   }
 
@@ -1194,8 +1292,6 @@ function field_attach_prepare_translation($node) {
  *   The name of the newly created bundle.
  */
 function field_attach_create_bundle($bundle) {
-  module_invoke(variable_get('field_storage_module', 'field_sql_storage'), 'field_storage_create_bundle', $bundle);
-
   // Clear the cache.
   field_cache_clear();
 
@@ -1214,7 +1310,6 @@ function field_attach_create_bundle($bundle) {
  *   The new name of the bundle.
  */
 function field_attach_rename_bundle($bundle_old, $bundle_new) {
-  module_invoke(variable_get('field_storage_module', 'field_sql_storage'), 'field_storage_rename_bundle', $bundle_old, $bundle_new);
   db_update('field_config_instance')
     ->fields(array('bundle' => $bundle_new))
     ->condition('bundle', $bundle_old)
@@ -1243,12 +1338,15 @@ function field_attach_rename_bundle($bundle_old, $bundle_new) {
  *   The bundle to delete.
  */
 function field_attach_delete_bundle($bundle) {
-  // Delete the instances themseves
+  // First, delete the instances themseves.
   $instances = field_info_instances($bundle);
   foreach ($instances as $instance) {
     field_delete_instance($instance['field_name'], $bundle);
   }
 
+  // Clear the cache.
+  field_cache_clear();
+
   // Let other modules act on deleting the bundle.
   foreach (module_implements('field_attach_delete_bundle') as $module) {
     $function = $module . '_field_attach_delete_bundle';
diff --git a/modules/field/field.crud.inc b/modules/field/field.crud.inc
index 210dad136..88e153a03 100644
--- a/modules/field/field.crud.inc
+++ b/modules/field/field.crud.inc
@@ -76,6 +76,20 @@
  * - settings (array)
  *     A sub-array of key/value pairs of field-type-specific settings. Each
  *     field type module defines and documents its own field settings.
+ * - storage (array)
+ *     A sub-array of key/value pairs identifying the storage backend to use for
+ *     the for the field.
+ *     - type (string)
+ *         The storage backend used by the field. Storage backends are defined
+ *         by modules that implement hook_field_storage_info().
+ *     - module (string, read-only)
+ *         The name of the module that implements the storage backend.
+ *     - active (integer, read-only)
+ *         TRUE if the module that implements the storage backend is currently
+ *         enabled, FALSE otherwise.
+ *     - settings (array)
+ *         A sub-array of key/value pairs of settings. Each storage backend
+ *         defines and documents its own settings.
  *
  * Field Instance objects are (currently) represented as an array of
  * key/value pairs. The object properties are:
@@ -196,6 +210,11 @@
  *     carefully, for it might seriously affect the site's performance.
  *   - settings: each omitted setting is given the default value defined in
  *     hook_field_info().
+ *   - storage:
+ *     - type: the storage backend specified in the 'field_default_storage'
+ *       system variable.
+ *     - settings: each omitted setting is given the default value specified in
+ *       hook_field_storage_info().
  * @return
  *   The $field structure with the id property filled in.
  * @throw
@@ -223,12 +242,6 @@ function field_create_field($field) {
       array('%name' => $field['field_name'])));
   }
 
-  // Check that the field type is known.
-  $field_type = field_info_field_types($field['type']);
-  if (!$field_type) {
-    throw new FieldException(t('Attempt to create a field of unknown type %type.', array('%type' => $field['type'])));
-  }
-
   // Ensure the field name is unique over active and disabled fields.
   // We do not care about deleted fields.
   $prior_field = field_read_field($field['field_name'], array('include_inactive' => TRUE));
@@ -253,22 +266,40 @@ function field_create_field($field) {
     'translatable' => FALSE,
     'locked' => FALSE,
     'settings' => array(),
+    'storage' => array(),
+    'deleted' => 0,
   );
 
+  // Check that the field type is known.
+  $field_type = field_info_field_types($field['type']);
+  if (!$field_type) {
+    throw new FieldException(t('Attempt to create a field of unknown type %type.', array('%type' => $field['type'])));
+  }
   // Create all per-field-type properties (needed here as long as we have
   // settings that impact column definitions).
   $field['settings'] += field_info_field_settings($field['type']);
   $field['module'] = $field_type['module'];
   $field['active'] = 1;
-  $field['deleted'] = 0;
 
+  // Provide default storage.
+  $field['storage'] += array(
+    'type' => variable_get('field_storage_default', 'field_sql_storage'),
+    'settings' => array(),
+  );
+  // Check that the storage type is known.
+  $storage_type = field_info_storage_types($field['storage']['type']);
+  if (!$storage_type) {
+    throw new FieldException(t('Attempt to create a field with unknown storage type %type.', array('%type' => $field['storage']['type'])));
+  }
+  // Provide default storage settings.
+  $field['storage']['settings'] += field_info_storage_settings($field['storage']['type']);
+  $field['storage']['module'] = $storage_type['module'];
+  $field['storage']['active'] = 1;
   // Collect storage information.
   $schema = (array) module_invoke($field['module'], 'field_schema', $field);
   $schema += array('columns' => array(), 'indexes' => array());
-
   // 'columns' are hardcoded in the field type.
   $field['columns'] = $schema['columns'];
-
   // 'indexes' can be both hardcoded in the field type, and specified in the
   // incoming $field definition.
   $field += array(
@@ -280,19 +311,32 @@ function field_create_field($field) {
   // have its own column and is not automatically populated when the field is
   // read.
   $data = $field;
-  unset($data['columns'], $data['field_name'], $data['type'], $data['locked'], $data['module'], $data['cardinality'], $data['active'], $data['deleted']);
-  $field['data'] = $data;
+  unset($data['columns'], $data['field_name'], $data['type'], $data['active'], $data['module'], $data['storage_type'], $data['storage_active'], $data['storage_module'], $data['locked'], $data['cardinality'], $data['deleted']);
 
-  // Store the field and create the id.
-  drupal_write_record('field_config', $field);
+  $record = array(
+    'field_name' => $field['field_name'],
+    'type' => $field['type'],
+    'module' => $field['module'],
+    'active' => $field['active'],
+    'storage_type' => $field['storage']['type'],
+    'storage_module' => $field['storage']['module'],
+    'storage_active' => $field['storage']['active'],
+    'locked' => $field['locked'],
+    'data' => $data,
+    'cardinality' => $field['cardinality'],
+    'deleted' => $field['deleted'],
+  );
 
-  // The 'data' property is not part of the public field record.
-  unset($field['data']);
+  // Store the field and get the id back.
+  drupal_write_record('field_config', $record);
+  $field['id'] = $record['id'];
 
   // Invoke hook_field_storage_create_field after the field is
   // complete (e.g. it has its id).
   try {
-    module_invoke(variable_get('field_storage_module', 'field_sql_storage'), 'field_storage_create_field', $field);
+    // Invoke hook_field_storage_create_field after
+    // drupal_write_record() sets the field id.
+    module_invoke($storage_type['module'], 'field_storage_create_field', $field);
   }
   catch (Exception $e) {
     // If storage creation failed, remove the field_config record before
@@ -344,10 +388,13 @@ function field_update_field($field) {
   $field += $prior_field;
   $field['settings'] += $prior_field['settings'];
 
-  // Field type cannot be changed.
+  // Some updates are always disallowed.
   if ($field['type'] != $prior_field['type']) {
     throw new FieldException("Cannot change an existing field's type.");
   }
+  if ($field['storage']['type'] != $prior_field['storage']['type']) {
+    throw new FieldException("Cannot change an existing field's storage type.");
+  }
 
   // Collect the new storage information, since what is in
   // $prior_field may no longer be right.
@@ -442,7 +489,9 @@ function field_read_fields($params = array(), $include_additional = array()) {
     $query->condition($key, $value);
   }
   if (!isset($include_additional['include_inactive']) || !$include_additional['include_inactive']) {
-    $query->condition('fc.active', 1);
+    $query
+      ->condition('fc.active', 1)
+      ->condition('fc.storage_active', 1);
   }
   $include_deleted = (isset($include_additional['include_deleted']) && $include_additional['include_deleted']);
   if (!$include_deleted) {
@@ -451,11 +500,20 @@ function field_read_fields($params = array(), $include_additional = array()) {
 
   $fields = array();
   $results = $query->execute();
-  foreach ($results as $field) {
-    // Extract serialized data.
-    $data = unserialize($field['data']);
-    unset($field['data']);
-    $field += $data;
+  foreach ($results as $record) {
+    $field = unserialize($record['data']);
+    $field['id'] = $record['id'];
+    $field['field_name'] = $record['field_name'];
+    $field['type'] = $record['type'];
+    $field['module'] = $record['module'];
+    $field['active'] = $record['active'];
+    $field['storage']['type'] = $record['storage_type'];
+    $field['storage']['module'] = $record['storage_module'];
+    $field['storage']['active'] = $record['storage_active'];
+    $field['locked'] = $record['locked'];
+    $field['cardinality'] = $record['cardinality'];
+    $field['translatable'] = $record['translatable'];
+    $field['deleted'] = $record['deleted'];
 
     module_invoke_all('field_read_field', $field);
 
@@ -489,8 +547,8 @@ function field_delete_field($field_name) {
     }
   }
 
-  // Mark field storage for deletion.
-  module_invoke(variable_get('field_storage_module', 'field_sql_storage'), 'field_storage_delete_field', $field_name);
+  // Mark field data for deletion.
+  module_invoke($field['storage']['module'], 'field_storage_delete_field', $field);
 
   // Mark the field for deletion.
   db_update('field_config')
@@ -677,7 +735,7 @@ function _field_write_instance($instance, $update = FALSE) {
   // not have its own column and is not automatically populated when the
   // instance is read.
   $data = $instance;
-  unset($data['id'], $data['field_id'], $data['field_name'], $data['bundle'], $data['widget']['type'], $data['deleted']);
+  unset($data['id'], $data['field_id'], $data['field_name'], $data['bundle'], $data['widget']['type'], $data['widget']['module'], $data['widget']['active'], $data['deleted']);
 
   $record = array(
     'field_id' => $instance['field_id'],
@@ -751,8 +809,10 @@ function field_read_instances($params = array(), $include_additional = array())
     $query->condition('fci.' . $key, $value);
   }
   if (!isset($include_additional['include_inactive']) || !$include_additional['include_inactive']) {
-    $query->condition('fc.active', 1);
-    $query->condition('fci.widget_active', 1);
+    $query
+      ->condition('fc.active', 1)
+      ->condition('fc.storage_active', 1)
+      ->condition('fci.widget_active', 1);
   }
   if (!isset($include_additional['include_deleted']) || !$include_additional['include_deleted']) {
     $query->condition('fc.deleted', 0);
@@ -799,8 +859,10 @@ function field_delete_instance($field_name, $bundle) {
     ->condition('bundle', $bundle)
     ->execute();
 
-  // Mark all data associated with the field for deletion.
-  module_invoke(variable_get('field_storage_module', 'field_sql_storage'), 'field_storage_delete_instance', $field_name, $bundle);
+  // Mark instance data for deletion.
+  $field = field_info_field($field_name);
+  module_invoke($field['storage']['module'], 'field_storage_delete_instance', $instance);
+
   // Clear the cache.
   field_cache_clear();
 
@@ -958,7 +1020,7 @@ function field_purge_data($obj_type, $object, $field, $instance) {
   _field_invoke('delete', $obj_type, $object, $dummy, $dummy, $options);
 
   // Tell the field storage system to purge the data.
-  module_invoke(variable_get('field_storage_module', 'field_sql_storage'), 'field_storage_purge', $obj_type, $object, $field, $instance);
+  module_invoke($field['storage']['module'], 'field_storage_purge', $obj_type, $object, $field, $instance);
 
   // Let other modules act on purging the data.
   foreach (module_implements('field_attach_purge') as $module) {
@@ -982,7 +1044,8 @@ function field_purge_instance($instance) {
     ->execute();
 
   // Notify the storage engine.
-  module_invoke(variable_get('field_storage_module', 'field_sql_storage'), 'field_storage_purge_instance', $instance);
+  $field = field_info_field_by_id($instance['field_id']);
+  module_invoke($field['storage']['module'], 'field_storage_purge_instance', $instance);
 
   // Clear the cache.
   field_info_cache_clear();
@@ -1011,7 +1074,7 @@ function field_purge_field($field) {
     ->execute();
 
   // Notify the storage engine.
-  module_invoke(variable_get('field_storage_module', 'field_sql_storage'), 'field_storage_purge_field', $field);
+  module_invoke($field['storage']['module'], 'field_storage_purge_field', $field);
 
   // Clear the cache.
   field_info_cache_clear();
diff --git a/modules/field/field.info.inc b/modules/field/field.info.inc
index aa5b4d08b..8b273da6a 100644
--- a/modules/field/field.info.inc
+++ b/modules/field/field.info.inc
@@ -78,6 +78,7 @@ function _field_info_collate_types($reset = FALSE) {
         'field types' => array(),
         'widget types' => array(),
         'formatter types' => array(),
+        'storage types' => array(),
         'fieldable types' => array(),
       );
 
@@ -110,7 +111,7 @@ function _field_info_collate_types($reset = FALSE) {
       }
       drupal_alter('field_widget_info', $info['widget types']);
 
-      // Populate formatters.
+      // Populate formatter types.
       foreach (module_implements('field_formatter_info') as $module) {
         $formatter_types = (array) module_invoke($module, 'field_formatter_info');
         foreach ($formatter_types as $name => $formatter_info) {
@@ -124,6 +125,20 @@ function _field_info_collate_types($reset = FALSE) {
       }
       drupal_alter('field_formatter_info', $info['formatter types']);
 
+      // Populate storage types.
+      foreach (module_implements('field_storage_info') as $module) {
+        $storage_types = (array) module_invoke($module, 'field_storage_info');
+        foreach ($storage_types as $name => $storage_info) {
+          // Provide defaults.
+          $storage_info += array(
+            'settings' => array(),
+          );
+          $info['storage types'][$name] = $storage_info;
+          $info['storage types'][$name]['module'] = $module;
+        }
+      }
+      drupal_alter('field_storage_info', $info['storage types']);
+
       // Populate information about 'fieldable' entities.
       foreach (module_implements('entity_info') as $module) {
         $entities = (array) module_invoke($module, 'entity_info');
@@ -246,6 +261,7 @@ function _field_info_collate_fields($reset = FALSE) {
 function _field_info_prepare_field($field) {
   // Make sure all expected field settings are present.
   $field['settings'] += field_info_field_settings($field['type']);
+  $field['storage']['settings'] += field_info_storage_settings($field['storage']['type']);
 
   return $field;
 }
@@ -371,8 +387,8 @@ function field_info_field_types($field_type = NULL) {
  *   returned.
  * @return
  *   Either a widget type description, as provided by
- *   hook_field_widget_info(), or an array of all existing widget
- *   types, keyed by widget type name.
+ *   hook_field_widget_info(), or an array of all existing widget types, keyed
+ *   by widget type name.
  */
 function field_info_widget_types($widget_type = NULL) {
   $info = _field_info_collate_types();
@@ -394,8 +410,9 @@ function field_info_widget_types($widget_type = NULL) {
  *   (optional) A formatter type name. If ommitted, all formatter types will be
  *   returned.
  * @return
- *   Either a formatter type description, as provided by hook_field_formatter_info(),
- *   or an array of all existing widget types, keyed by widget type name.
+ *   Either a formatter type description, as provided by
+ *   hook_field_formatter_info(), or an array of all existing formatter types,
+ *   keyed by formatter type name.
  */
 function field_info_formatter_types($formatter_type = NULL) {
   $info = _field_info_collate_types();
@@ -411,6 +428,30 @@ function field_info_formatter_types($formatter_type = NULL) {
 }
 
 /**
+ * Return hook_field_storage_info() data.
+ *
+ * @param $storage_type
+ *   (optional) A storage type name. If ommitted, all storage types will be
+ *   returned.
+ * @return
+ *   Either a storage type description, as provided by
+ *   hook_field_storage_info(), or an array of all existing storage types,
+ *   keyed by storage type name.
+ */
+function field_info_storage_types($storage_type = NULL) {
+  $info = _field_info_collate_types();
+  $storage_types = $info['storage types'];
+  if ($storage_type) {
+    if (isset($storage_types[$storage_type])) {
+      return $storage_types[$storage_type];
+    }
+  }
+  else {
+    return $storage_types;
+  }
+}
+
+/**
  * Return hook_fieldable_info() data.
  *
  * @param $obj_type
@@ -574,8 +615,8 @@ function field_info_instance_settings($type) {
  * @param $type
  *   A widget type name.
  * @return
- *   The field type's default settings, as provided by hook_field_info(), or an
- *   empty array.
+ *   The widget type's default settings, as provided by
+ *   hook_field_widget_info(), or an empty array.
  */
 function field_info_widget_settings($type) {
   $info = field_info_widget_types($type);
@@ -588,8 +629,8 @@ function field_info_widget_settings($type) {
  * @param $type
  *   A field formatter type name.
  * @return
- *   The field formatter's default settings, as provided by
- *   hook_field_info(), or an empty array.
+ *   The formatter type's default settings, as provided by
+ *   hook_field_formatter_info(), or an empty array.
  */
 function field_info_formatter_settings($type) {
   $info = field_info_formatter_types($type);
@@ -597,5 +638,19 @@ function field_info_formatter_settings($type) {
 }
 
 /**
+ * Return a field formatter's default settings.
+ *
+ * @param $type
+ *   A field storage type name.
+ * @return
+ *   The storage type's default settings, as provided by
+ *   hook_field_storage_info(), or an empty array.
+ */
+function field_info_storage_settings($type) {
+  $info = field_info_storage_types($type);
+  return isset($info['settings']) ? $info['settings'] : array();
+}
+
+/**
  * @} End of "defgroup field_info"
  */
diff --git a/modules/field/field.install b/modules/field/field.install
index ba3db1df5..c112ef1b6 100644
--- a/modules/field/field.install
+++ b/modules/field/field.install
@@ -28,41 +28,63 @@ function field_schema() {
         'type' => 'varchar',
         'length' => 128,
         'not null' => TRUE,
-        'description' => 'The type of this field, coming from a field module',
+        'description' => 'The type of this field.',
       ),
-      'locked' => array(
+     'module' => array(
+        'type' => 'varchar',
+        'length' => 128,
+        'not null' => TRUE,
+        'default' => '',
+        'description' => 'The module that implements the field type.',
+      ),
+      'active' => array(
         'type' => 'int',
         'size' => 'tiny',
         'not null' => TRUE,
         'default' => 0,
-        'description' => '@TODO',
+        'description' => 'Boolean indicating whether the module that implements the field type is enabled.',
       ),
-      'data' => array(
-        'type' => 'text',
-        'size' => 'medium',
+      'storage_type' => array(
+        'type' => 'varchar',
+        'length' => 128,
         'not null' => TRUE,
-        'serialize' => TRUE,
-        'description' => 'Field specific settings, for example maximum length',
+        'description' => 'The storage backend for the field.',
       ),
-      'module' => array(
+      'storage_module' => array(
         'type' => 'varchar',
         'length' => 128,
         'not null' => TRUE,
         'default' => '',
+        'description' => 'The module that implements the storage backend.',
       ),
-      'cardinality' => array(
+      'storage_active' => array(
         'type' => 'int',
         'size' => 'tiny',
         'not null' => TRUE,
         'default' => 0,
+        'description' => 'Boolean indicating whether the module that implements the storage backend is enabled.',
       ),
-      'translatable' => array(
+      'locked' => array(
         'type' => 'int',
         'size' => 'tiny',
         'not null' => TRUE,
         'default' => 0,
+        'description' => '@TODO',
       ),
-      'active' => array(
+      'data' => array(
+        'type' => 'text',
+        'size' => 'medium',
+        'not null' => TRUE,
+        'serialize' => TRUE,
+        'description' => 'Serialized data containing the field properties that do not warrant a dedicated column.',
+      ),
+      'cardinality' => array(
+        'type' => 'int',
+        'size' => 'tiny',
+        'not null' => TRUE,
+        'default' => 0,
+      ),
+      'translatable' => array(
         'type' => 'int',
         'size' => 'tiny',
         'not null' => TRUE,
@@ -77,14 +99,17 @@ function field_schema() {
     ),
     'primary key' => array('id'),
     'indexes' => array(
-      // used by field_delete_field() among others
       'field_name' => array('field_name'),
-      // used by field_read_fields()
-      'active_deleted' => array('active', 'deleted'),
-      // used by field_modules_disabled()
+      // Used by field_read_fields().
+      'active' => array('active'),
+      'storage_active' => array('storage_active'),
+      'deleted' => array('deleted'),
+      // Used by field_modules_disabled().
       'module' => array('module'),
-      // used by field_associate_fields()
+      'storage_module' => array('storage_module'),
+      // Used by field_associate_fields().
       'type' => array('type'),
+      'storage_type' => array('storage_type'),
     ),
   );
   $schema['field_config_instance'] = array(
@@ -124,13 +149,14 @@ function field_schema() {
     ),
     'primary key' => array('id'),
     'indexes' => array(
-      // used by field_delete_instance()
+      // Used by field_delete_instance().
       'field_name_bundle' => array('field_name', 'bundle'),
-      // used by field_read_instances()
-      'widget_active_deleted' => array('widget_active', 'deleted'),
-      // used by field_modules_disabled()
+      // Used by field_read_instances().
+      'widget_active' => array('widget_active'),
+      'deleted' => array('deleted'),
+      // Used by field_modules_disabled().
       'widget_module' => array('widget_module'),
-      // used by field_associate_fields()
+      // Used by field_associate_fields().
       'widget_type' => array('widget_type'),
     ),
   );
diff --git a/modules/field/field.module b/modules/field/field.module
index 75dc3d431..4d031e935 100644
--- a/modules/field/field.module
+++ b/modules/field/field.module
@@ -219,6 +219,10 @@ function field_modules_disabled($modules) {
       ->fields(array('active' => 0))
       ->condition('module', $module)
       ->execute();
+    db_update('field_config')
+      ->fields(array('storage_active' => 0))
+      ->condition('storage_module', $module)
+      ->execute();
     db_update('field_config_instance')
       ->fields(array('widget_active' => 0))
       ->condition('widget_module', $module)
@@ -234,25 +238,32 @@ function field_modules_disabled($modules) {
  *   The name of the module to update on.
  */
 function field_associate_fields($module) {
-  $module_fields = module_invoke($module, 'field_info');
-  if ($module_fields) {
-    foreach ($module_fields as $name => $field_info) {
-      watchdog('field', 'Updating field type %type with module %module.', array('%type' => $name, '%module' => $module));
-      db_update('field_config')
-        ->fields(array('module' => $module, 'active' => 1))
-        ->condition('type', $name)
-        ->execute();
-    }
+  // Associate field types.
+  $field_types =(array) module_invoke($module, 'field_info');
+  foreach ($field_types as $name => $field_info) {
+    watchdog('field', 'Updating field type %type with module %module.', array('%type' => $name, '%module' => $module));
+    db_update('field_config')
+      ->fields(array('module' => $module, 'active' => 1))
+      ->condition('type', $name)
+      ->execute();
   }
-  $module_widgets = module_invoke($module, 'widget_info');
-  if ($module_widgets) {
-    foreach ($module_widgets as $name => $widget_info) {
-      watchdog('field', 'Updating widget type %type with module %module.', array('%type' => $name, '%module' => $module));
-      db_update('field_config_instance')
-        ->fields(array('widget_module' => $module, 'widget_active' => 1))
-        ->condition('widget_type', $name)
-        ->execute();
-    }
+  // Associate storage backends.
+  $storage_types = (array) module_invoke($module, 'field_storage_info');
+  foreach ($storage_types as $name => $storage_info) {
+    watchdog('field', 'Updating field storage %type with module %module.', array('%type' => $name, '%module' => $module));
+    db_update('field_config')
+      ->fields(array('storage_module' => $module, 'storage_active' => 1))
+      ->condition('storage_type', $name)
+      ->execute();
+  }
+  // Associate widget types.
+  $widget_types = (array) module_invoke($module, 'field_widget_info');
+  foreach ($widget_types as $name => $widget_info) {
+    watchdog('field', 'Updating widget type %type with module %module.', array('%type' => $name, '%module' => $module));
+    db_update('field_config_instance')
+      ->fields(array('widget_module' => $module, 'widget_active' => 1))
+      ->condition('widget_type', $name)
+      ->execute();
   }
 }
 
diff --git a/modules/field/field.test b/modules/field/field.test
index 2523c8b72..c77d790c9 100644
--- a/modules/field/field.test
+++ b/modules/field/field.test
@@ -10,6 +10,18 @@
  * Parent class for Field API tests.
  */
 class FieldTestCase extends DrupalWebTestCase {
+  var $default_storage = 'field_sql_storage';
+
+  /**
+   * Set the default field storage backend for fields created during tests.
+   */
+  function setUp() {
+    // Call parent::setUp().
+    $args = func_get_args();
+    call_user_func_array(array('parent', 'setUp'), $args);
+    // Set default storage backend.
+    variable_set('field_storage_default', $this->default_storage);
+  }
 
   /**
    * Generate random values for a field_test field.
@@ -212,6 +224,56 @@ class FieldAttachStorageTestCase extends FieldAttachTestCase {
   }
 
   /**
+   * Test saving and loading fields using different storage backends.
+   */
+  function testFieldAttachSaveLoadDifferentStorage() {
+    $entity_type = 'test_entity';
+    $langcode = FIELD_LANGUAGE_NONE;
+
+    // Create two fields using different storage backends, and their instances.
+    $fields = array(
+      array(
+        'field_name' => 'field_1',
+        'type' => 'test_field',
+        'cardinality' => 4,
+        'storage' => array('type' => 'field_sql_storage')
+      ),
+      array(
+        'field_name' => 'field_2',
+        'type' => 'test_field',
+        'cardinality' => 4,
+        'storage' => array('type' => 'field_test_storage')
+      ),
+    );
+    foreach ($fields as $field) {
+      field_create_field($field);
+      $instance = array(
+        'field_name' => $field['field_name'],
+        'bundle' => 'test_bundle',
+      );
+      field_create_instance($instance);
+    }
+
+    $entity_init = field_test_create_stub_entity();
+
+    // Create entity and insert random values.
+    $entity = clone($entity_init);
+    $values = array();
+    foreach ($fields as $field) {
+      $values[$field['field_name']] = $this->_generateTestFieldValues($this->field['cardinality']);
+      $entity->{$field['field_name']}[$langcode] = $values[$field['field_name']];
+    }
+    field_attach_insert($entity_type, $entity);
+
+    // Check that values are loaded as expected.
+    $entity = clone($entity_init);
+    field_attach_load($entity_type, array($entity->ftid => $entity));
+    foreach ($fields as $field) {
+      $this->assertEqual($values[$field['field_name']], $entity->{$field['field_name']}[$langcode], t('%storage storage: expected values were found.', array('%storage' => $field['storage']['type'])));
+    }
+  }
+
+  /**
    * Tests insert and update with missing or NULL fields.
    */
   function testFieldAttachSaveMissingData() {
@@ -225,17 +287,17 @@ class FieldAttachStorageTestCase extends FieldAttachTestCase {
 
     $entity = clone($entity_init);
     field_attach_load($entity_type, array($entity->ftid => $entity));
-    $this->assertTrue(empty($entity->{$this->field_name}[$langcode]), t('Insert: missing field results in no value saved'));
+    $this->assertTrue(empty($entity->{$this->field_name}), t('Insert: missing field results in no value saved'));
 
     // Insert: Field is NULL.
     field_cache_clear();
     $entity = clone($entity_init);
-    $entity->{$this->field_name}[$langcode] = NULL;
+    $entity->{$this->field_name} = NULL;
     field_attach_insert($entity_type, $entity);
 
     $entity = clone($entity_init);
     field_attach_load($entity_type, array($entity->ftid => $entity));
-    $this->assertTrue(empty($entity->{$this->field_name}[$langcode]), t('Insert: NULL field results in no value saved'));
+    $this->assertTrue(empty($entity->{$this->field_name}), t('Insert: NULL field results in no value saved'));
 
     // Add some real data.
     field_cache_clear();
@@ -260,12 +322,33 @@ class FieldAttachStorageTestCase extends FieldAttachTestCase {
     // Update: Field is NULL. Data should be wiped.
     field_cache_clear();
     $entity = clone($entity_init);
-    $entity->{$this->field_name}[$langcode] = NULL;
+    $entity->{$this->field_name} = NULL;
     field_attach_update($entity_type, $entity);
 
     $entity = clone($entity_init);
     field_attach_load($entity_type, array($entity->ftid => $entity));
-    $this->assertTrue(empty($entity->{$this->field_name}[$langcode]), t('Update: NULL field removes existing values'));
+    $this->assertTrue(empty($entity->{$this->field_name}), t('Update: NULL field removes existing values'));
+
+    // Re-add some data.
+    field_cache_clear();
+    $entity = clone($entity_init);
+    $values = $this->_generateTestFieldValues(1);
+    $entity->{$this->field_name}[$langcode] = $values;
+    field_attach_update($entity_type, $entity);
+
+    $entity = clone($entity_init);
+    field_attach_load($entity_type, array($entity->ftid => $entity));
+    $this->assertEqual($entity->{$this->field_name}[$langcode], $values, t('Field data saved'));
+
+    // Update: Field is empty array. Data should be wiped.
+    field_cache_clear();
+    $entity = clone($entity_init);
+    $entity->{$this->field_name} = array();
+    field_attach_update($entity_type, $entity);
+
+    $entity = clone($entity_init);
+    field_attach_load($entity_type, array($entity->ftid => $entity));
+    $this->assertTrue(empty($entity->{$this->field_name}), t('Update: empty array removes existing values'));
   }
 
   /**
@@ -391,7 +474,7 @@ class FieldAttachStorageTestCase extends FieldAttachTestCase {
     $this->assertIdentical($this->instance['bundle'], $new_bundle, "Bundle name has been updated in the instance.");
 
     // Verify the field data is present on load.
-    $entity = field_test_create_stub_entity(0, 0, $this->instance['bundle']);
+    $entity = field_test_create_stub_entity(0, 0, $new_bundle);
     field_attach_load($entity_type, array(0 => $entity));
     $this->assertEqual(count($entity->{$this->field_name}[$langcode]), $this->field['cardinality'], "Bundle name has been updated in the field storage");
   }
@@ -956,6 +1039,7 @@ class FieldInfoTestCase extends FieldTestCase {
     $field_test_info = field_test_field_info();
     $formatter_info = field_test_field_formatter_info();
     $widget_info = field_test_field_widget_info();
+    $storage_info = field_test_field_storage_info();
 
     $info = field_info_field_types();
     foreach ($field_test_info as $t_key => $field_type) {
@@ -981,6 +1065,14 @@ class FieldInfoTestCase extends FieldTestCase {
       $this->assertEqual($info[$w_key]['module'], 'field_test',  t("Widget type field_test module appears"));
     }
 
+    $info = field_info_storage_types();
+    foreach ($storage_info as $s_key => $storage) {
+      foreach ($storage as $key => $val) {
+        $this->assertEqual($info[$s_key][$key], $val, t("Storage type $s_key key $key is $val"));
+      }
+      $this->assertEqual($info[$s_key]['module'], 'field_test',  t("Storage type field_test module appears"));
+    }
+
     // Verify that no unexpected instances exist.
     $core_fields = field_info_fields();
     $instances = field_info_instances(FIELD_TEST_BUNDLE);
@@ -1473,6 +1565,9 @@ class FieldCrudTestCase extends FieldTestCase {
     $field_type = field_info_field_types($field_definition['type']);
     $this->assertIdentical($record['data']['settings'], $field_type['settings'], t('Default field settings have been written.'));
 
+    // Ensure that default storage was set.
+    $this->assertEqual($record['storage_type'], variable_get('field_storage_default'), t('The field type is properly saved.'));
+
     // Guarantee that the name is unique.
     try {
       field_create_field($field_definition);
@@ -1565,16 +1660,13 @@ class FieldCrudTestCase extends FieldTestCase {
    */
   function testCreateFieldFail() {
     $field_name = 'duplicate';
-    $field_definition = array('field_name' => $field_name, 'type' => 'test_field');
+    $field_definition = array('field_name' => $field_name, 'type' => 'test_field', 'storage' => array('type' => 'field_test_storage_failure'));
     $query = db_select('field_config')->condition('field_name', $field_name)->countQuery();
 
     // The field does not appear in field_config.
     $count = $query->execute()->fetchField();
     $this->assertEqual($count, 0, 'A field_config row for the field does not exist.');
 
-    // Make field creation fail.
-    variable_set('field_storage_module', 'field_test');
-    
     // Try to create the field.
     try {
       $field = field_create_field($field_definition);
@@ -1679,8 +1771,7 @@ class FieldCrudTestCase extends FieldTestCase {
 
     // Make sure that the field is marked as deleted when it is specifically
     // loaded.
-    $fields = field_read_fields(array(), array('include_deleted' => TRUE));
-    $field = current($field);
+    $field = field_read_field($this->field['field_name'], array('include_deleted' => TRUE));
     $this->assertTrue(!empty($field['deleted']), t('A deleted field is marked for deletion.'));
 
     // Make sure that this field's instance is marked as deleted when it is
@@ -1829,6 +1920,69 @@ class FieldCrudTestCase extends FieldTestCase {
       $this->pass(t("An unchangeable setting cannot be updated."));
     }
   }
+
+  /**
+   * Test that fields are properly marked active or inactive.
+   */
+  function testActive() {
+    $field_definition = array(
+      'field_name' => 'field_1',
+      'type' => 'test_field',
+      // For this test, we need a storage backend provided by a different
+      // module than field_test.module.
+      'storage' => array(
+        'type' => 'field_sql_storage',
+      ),
+    );
+    field_create_field($field_definition);
+
+    // Test disabling and enabling:
+    // - the field type module,
+    // - the storage module,
+    // - both.
+    $this->_testActiveHelper($field_definition, array('field_test'));
+    $this->_testActiveHelper($field_definition, array('field_sql_storage'));
+    $this->_testActiveHelper($field_definition, array('field_test', 'field_sql_storage'));
+  }
+
+  /**
+   * Helper function for testActive().
+   *
+   * Test dependency between a field and a set of modules.
+   *
+   * @param $field_definition
+   *   A field definition.
+   * @param $modules
+   *   An aray of module names. The field will be tested to be inactive as long
+   *   as any of those modules is disabled.
+   */
+  function _testActiveHelper($field_definition, $modules) {
+    $field_name = $field_definition['field_name'];
+
+    // Read the field.
+    $field = field_read_field($field_name);
+    $this->assertTrue($field_definition <= $field, t('The field was properly read.'));
+
+    module_disable($modules);
+
+    $fields = field_read_fields(array('field_name' => $field_name), array('include_inactive' => TRUE));
+    $this->assertTrue(isset($fields[$field_name]) && $field_definition < $field, t('The field is properly read when explicitly fetching inactive fields.'));
+
+    // Re-enable modules one by one, and check that the field is still inactive
+    // while some modules remain disabled.
+    while ($modules) {
+      $field = field_read_field($field_name);
+      $this->assertTrue(empty($field), t('%modules disabled. The field is marked inactive.', array('%modules' => implode(', ', $modules))));
+
+      $module = array_shift($modules);
+      module_enable(array($module));
+    }
+
+    // Check that the field is active again after all modules have been
+    // enabled.
+    $field = field_read_field($field_name);
+    $this->assertTrue($field_definition <= $field, t('The field was was marked active.'));
+  }
 }
 
 class FieldInstanceCrudTestCase extends FieldTestCase {
@@ -1844,6 +1998,7 @@ class FieldInstanceCrudTestCase extends FieldTestCase {
 
   function setUp() {
     parent::setUp('field_test');
+
     $this->field = array(
       'field_name' => drupal_strtolower($this->randomName()),
       'type' => 'test_field',
@@ -2430,7 +2585,7 @@ class FieldBulkDeleteTestCase extends FieldTestCase {
 
     // The field still exists, not deleted, because it has a second instance.
     $fields = field_read_fields(array('id' => $field['id']), array('include_deleted' => 1, 'include_inactive' => 1));
-    $this->assertEqual($field, $fields[$field['id']], 'The field exists and is not deleted');
+    $this->assertTrue(isset($fields[$field['id']]), 'The field exists and is not deleted');
   }
 
   /**
@@ -2452,7 +2607,7 @@ class FieldBulkDeleteTestCase extends FieldTestCase {
 
       // The field still exists, not deleted, because it was never deleted.
       $fields = field_read_fields(array('id' => $field['id']), array('include_deleted' => 1, 'include_inactive' => 1));
-      $this->assertEqual($field, $fields[$field['id']], 'The field exists and is not deleted');
+      $this->assertTrue(isset($fields[$field['id']]), 'The field exists and is not deleted');
     }
 
     // Delete the field.
diff --git a/modules/field/modules/field_sql_storage/field_sql_storage.install b/modules/field/modules/field_sql_storage/field_sql_storage.install
index 67b7e4afc..7ab6a8cf9 100644
--- a/modules/field/modules/field_sql_storage/field_sql_storage.install
+++ b/modules/field/modules/field_sql_storage/field_sql_storage.install
@@ -37,7 +37,9 @@ function field_sql_storage_schema() {
     $fields = field_read_fields(array(), array('include_deleted' => TRUE, 'include_inactive' => TRUE));
     drupal_load('module', 'field_sql_storage');
     foreach ($fields as $field) {
-      $schema += _field_sql_storage_schema($field);
+      if ($field['storage']['type'] == 'field_sql_storage') {
+        $schema += _field_sql_storage_schema($field);
+      }
     }
   }
   return $schema;
diff --git a/modules/field/modules/field_sql_storage/field_sql_storage.module b/modules/field/modules/field_sql_storage/field_sql_storage.module
index 97d372f3b..09dee368a 100644
--- a/modules/field/modules/field_sql_storage/field_sql_storage.module
+++ b/modules/field/modules/field_sql_storage/field_sql_storage.module
@@ -18,6 +18,18 @@ function field_sql_storage_help($path, $arg) {
 }
 
 /**
+ * Implement hook_field_storage_info().
+ */
+function field_sql_storage_field_storage_info() {
+  return array(
+    'field_sql_storage' => array(
+      'label' => t('Default SQL storage'),
+      'description' => t('Stores fields in the local SQL database, using per-field tables.'),
+    ),
+  );
+}
+
+/**
  * Generate a table name for a field data table.
  *
  * @param $field
@@ -264,9 +276,7 @@ function field_sql_storage_field_storage_update_field($field, $prior_field, $has
 /**
  * Implement hook_field_storage_delete_field().
  */
-function field_sql_storage_field_storage_delete_field($field_name) {
-  $field = field_info_field($field_name);
-
+function field_sql_storage_field_storage_delete_field($field) {
   // Mark all data associated with the field for deletion.
   $field['deleted'] = 0;
   $table = _field_sql_storage_tablename($field);
@@ -288,34 +298,11 @@ function field_sql_storage_field_storage_delete_field($field_name) {
 /**
  * Implement hook_field_storage_load().
  */
-function field_sql_storage_field_storage_load($obj_type, $objects, $age, $skip_fields, $options) {
+function field_sql_storage_field_storage_load($obj_type, $objects, $age, $fields, $options) {
   $etid = _field_sql_storage_etid($obj_type);
   $load_current = $age == FIELD_LOAD_CURRENT;
 
-  // Gather ids needed for each field.
-  $field_ids = array();
-  $delta_count = array();
-  foreach ($objects as $obj) {
-    list($id, $vid, $bundle) = field_extract_ids($obj_type, $obj);
-
-    if ($options['deleted']) {
-      $instances = field_read_instances(array('bundle' => $bundle), array('include_deleted' => $options['deleted']));
-    }
-    else {
-      $instances = field_info_instances($bundle);
-    }
-
-    foreach ($instances as $instance) {
-      $field_name = $instance['field_name'];
-      if (!isset($skip_fields[$instance['field_id']]) && (!isset($options['field_id']) || $options['field_id'] == $instance['field_id'])) {
-        $objects[$id]->{$field_name} = array();
-        $field_ids[$instance['field_id']][] = $load_current ? $id : $vid;
-        $delta_count[$id][$field_name] = array();
-      }
-    }
-  }
-
-  foreach ($field_ids as $field_id => $ids) {
+  foreach ($fields as $field_id => $ids) {
     $field = field_info_field_by_id($field_id);
     $field_name = $field['field_name'];
     $table = $load_current ? _field_sql_storage_tablename($field) : _field_sql_storage_revision_tablename($field);
@@ -333,12 +320,13 @@ function field_sql_storage_field_storage_load($obj_type, $objects, $age, $skip_f
 
     $results = $query->execute();
 
+    $delta_count = array();
     foreach ($results as $row) {
-      if (!isset($delta_count[$row->entity_id][$field_name][$row->language])) {
-        $delta_count[$row->entity_id][$field_name][$row->language] = 0;
+      if (!isset($delta_count[$row->entity_id][$row->language])) {
+        $delta_count[$row->entity_id][$row->language] = 0;
       }
 
-      if ($field['cardinality'] == FIELD_CARDINALITY_UNLIMITED || $delta_count[$row->entity_id][$field_name][$row->language] < $field['cardinality']) {
+      if ($field['cardinality'] == FIELD_CARDINALITY_UNLIMITED || $delta_count[$row->entity_id][$row->language] < $field['cardinality']) {
         $item = array();
         // For each column declared by the field, populate the item
         // from the prefixed database column.
@@ -349,7 +337,7 @@ function field_sql_storage_field_storage_load($obj_type, $objects, $age, $skip_f
 
         // Add the item to the field values for the entity.
         $objects[$row->entity_id]->{$field_name}[$row->language][] = $item;
-        $delta_count[$row->entity_id][$field_name][$row->language]++;
+        $delta_count[$row->entity_id][$row->language]++;
       }
     }
   }
@@ -358,41 +346,30 @@ function field_sql_storage_field_storage_load($obj_type, $objects, $age, $skip_f
 /**
  * Implement hook_field_storage_write().
  */
-function field_sql_storage_field_storage_write($obj_type, $object, $op, $skip_fields) {
+function field_sql_storage_field_storage_write($obj_type, $object, $op, $fields) {
   list($id, $vid, $bundle) = field_extract_ids($obj_type, $object);
   $etid = _field_sql_storage_etid($obj_type);
 
-  $instances = field_info_instances($bundle);
-  foreach ($instances as $instance) {
-    $field_name = $instance['field_name'];
-    if (isset($skip_fields[$instance['field_id']])) {
-      continue;
-    }
-
-    $field = field_info_field($field_name);
+  foreach ($fields as $field_id) {
+    $field = field_info_field_by_id($field_id);
+    $field_name = $field['field_name'];
     $table_name = _field_sql_storage_tablename($field);
     $revision_name = _field_sql_storage_revision_tablename($field);
 
-    // Leave the field untouched if $object comes with no $field_name property.
-    // Empty the field if $object->$field_name is NULL or an empty array.
-
-    // Function property_exists() is slower, so we catch the more frequent cases
-    // where it's an empty array with the faster isset().
-    if (isset($object->$field_name) || property_exists($object, $field_name)) {
-      $available_languages = field_multilingual_available_languages($obj_type, $field);
-      $available_translations = is_array($object->$field_name) ? array_intersect($available_languages, array_keys($object->$field_name)) : FALSE;
-
-      // Delete and insert, rather than update, in case a value was added.
-      // If no translation is available, empty the field for all the available languages.
-      if ($op == FIELD_STORAGE_UPDATE && count($available_translations)) {
-        $languages = empty($object->$field_name) ? $available_languages : $available_translations;
+    $all_languages = field_multilingual_available_languages($obj_type, $field);
+    $field_languages = array_intersect($all_languages, array_keys((array) $object->$field_name));
 
+    // Delete and insert, rather than update, in case a value was added.
+    if ($op == FIELD_STORAGE_UPDATE) {
+      // Delete languages present in the incoming $object->$field_name.
+      // Delete all languages if $object->$field_name is empty.
+      $languages = !empty($object->$field_name) ? $field_languages : $all_languages;
+      if ($languages) {
         db_delete($table_name)
           ->condition('etid', $etid)
           ->condition('entity_id', $id)
           ->condition('language', $languages, 'IN')
           ->execute();
-
         if (isset($vid)) {
           db_delete($revision_name)
             ->condition('etid', $etid)
@@ -402,46 +379,48 @@ function field_sql_storage_field_storage_write($obj_type, $object, $op, $skip_fi
             ->execute();
         }
       }
+    }
 
-      if (!empty($available_translations)) {
-        // Prepare the multi-insert query.
-        $columns = array('etid', 'entity_id', 'revision_id', 'bundle', 'delta', 'language');
+    // Prepare the multi-insert query.
+    $do_insert = FALSE;
+    $columns = array('etid', 'entity_id', 'revision_id', 'bundle', 'delta', 'language');
+    foreach ($field['columns'] as $column => $attributes) {
+      $columns[] = _field_sql_storage_columnname($field_name, $column);
+    }
+    $query = db_insert($table_name)->fields($columns);
+    if (isset($vid)) {
+      $revision_query = db_insert($revision_name)->fields($columns);
+    }
+
+    foreach ($field_languages as $langcode) {
+      $items = (array) $object->{$field_name}[$langcode];
+      $delta_count = 0;
+      foreach ($items as $delta => $item) {
+        // We now know we have someting to insert.
+        $do_insert = TRUE;
+        $record = array(
+          'etid' => $etid,
+          'entity_id' => $id,
+          'revision_id' => $vid,
+          'bundle' => $bundle,
+          'delta' => $delta,
+          'language' => $langcode,
+        );
         foreach ($field['columns'] as $column => $attributes) {
-          $columns[] = _field_sql_storage_columnname($field_name, $column);
+          $record[_field_sql_storage_columnname($field_name, $column)] = isset($item[$column]) ? $item[$column] : NULL;
         }
-        $query = db_insert($table_name)->fields($columns);
+        $query->values($record);
         if (isset($vid)) {
-          $revision_query = db_insert($revision_name)->fields($columns);
+          $revision_query->values($record);
         }
 
-        foreach ($available_translations as $langcode) {
-          if ($items = $object->{$field_name}[$langcode]) {
-            $delta_count = 0;
-            foreach ($items as $delta => $item) {
-              $record = array(
-                'etid' => $etid,
-                'entity_id' => $id,
-                'revision_id' => $vid,
-                'bundle' => $bundle,
-                'delta' => $delta,
-                'language' => $langcode,
-              );
-              foreach ($field['columns'] as $column => $attributes) {
-                $record[_field_sql_storage_columnname($field_name, $column)] = isset($item[$column]) ? $item[$column] : NULL;
-              }
-              $query->values($record);
-              if (isset($vid)) {
-                $revision_query->values($record);
-              }
-
-              if ($field['cardinality'] != FIELD_CARDINALITY_UNLIMITED && ++$delta_count == $field['cardinality']) {
-                break;
-              }
-            }
-          }
+        if ($field['cardinality'] != FIELD_CARDINALITY_UNLIMITED && ++$delta_count == $field['cardinality']) {
+          break;
         }
+      }
 
-        // Execute the insert.
+      // Execute the query if we have values to insert.
+      if ($do_insert) {
         $query->execute();
         if (isset($vid)) {
           $revision_query->execute();
@@ -456,14 +435,15 @@ function field_sql_storage_field_storage_write($obj_type, $object, $op, $skip_fi
  *
  * This function deletes data for all fields for an object from the database.
  */
-function field_sql_storage_field_storage_delete($obj_type, $object) {
+function field_sql_storage_field_storage_delete($obj_type, $object, $fields) {
   list($id, $vid, $bundle) = field_extract_ids($obj_type, $object);
   $etid = _field_sql_storage_etid($obj_type);
 
-  $instances = field_info_instances($bundle);
-  foreach ($instances as $instance) {
-    $field = field_info_field($instance['field_name']);
-    field_sql_storage_field_storage_purge($obj_type, $object, $field, $instance);
+  foreach (field_info_instances($bundle) as $instance) {
+    if (isset($fields[$instance['field_id']])) {
+      $field = field_info_field_by_id($instance['field_id']);
+      field_sql_storage_field_storage_purge($obj_type, $object, $field, $instance);
+    }
   }
 }
 
@@ -477,7 +457,6 @@ function field_sql_storage_field_storage_purge($obj_type, $object, $field, $inst
   list($id, $vid, $bundle) = field_extract_ids($obj_type, $object);
   $etid = _field_sql_storage_etid($obj_type);
 
-  $field = field_info_field_by_id($field['id']);
   $table_name = _field_sql_storage_tablename($field);
   $revision_name = _field_sql_storage_revision_tablename($field);
   db_delete($table_name)
@@ -553,16 +532,16 @@ function field_sql_storage_field_storage_query($field_id, $conditions, $count, &
         $value = _field_sql_storage_etid($value);
       }
     }
-
-    $query->condition($column, $value, $operator);
-
+    // Track condition on 'deleted'.
     if ($column == 'deleted') {
-      $deleted = $value;
+      $condition_deleted = TRUE;
     }
+
+    $query->condition($column, $value, $operator);
   }
 
   // Exclude deleted data unless we have a condition on it.
-  if (!isset($deleted)) {
+  if (!isset($condition_deleted)) {
     $query->condition('deleted', 0);
   }
 
@@ -610,15 +589,13 @@ function field_sql_storage_field_storage_query($field_id, $conditions, $count, &
  *
  * This function actually deletes the data from the database.
  */
-function field_sql_storage_field_storage_delete_revision($obj_type, $object) {
+function field_sql_storage_field_storage_delete_revision($obj_type, $object, $fields) {
   list($id, $vid, $bundle) = field_extract_ids($obj_type, $object);
   $etid = _field_sql_storage_etid($obj_type);
 
   if (isset($vid)) {
-    $instances = field_info_instances($bundle);
-    foreach ($instances as $instance) {
-      $field_name = $instance['field_name'];
-      $field = field_read_field($field_name);
+    foreach ($fields as $field_id) {
+      $field = field_info_field_by_id($field_id);
       $revision_name = _field_sql_storage_revision_tablename($field);
       db_delete($revision_name)
         ->condition('etid', $etid)
@@ -634,37 +611,40 @@ function field_sql_storage_field_storage_delete_revision($obj_type, $object) {
  *
  * This function simply marks for deletion all data associated with the field.
  */
-function field_sql_storage_field_storage_delete_instance($field_name, $bundle) {
-  $field = field_read_field($field_name);
+function field_sql_storage_field_storage_delete_instance($instance) {
+  $field = field_info_field($instance['field_name']);
   $table_name = _field_sql_storage_tablename($field);
   $revision_name = _field_sql_storage_revision_tablename($field);
   db_update($table_name)
     ->fields(array('deleted' => 1))
-    ->condition('bundle', $bundle)
+    ->condition('bundle', $instance['bundle'])
     ->execute();
   db_update($revision_name)
     ->fields(array('deleted' => 1))
-    ->condition('bundle', $bundle)
+    ->condition('bundle', $instance['bundle'])
     ->execute();
 }
 
 /**
- * Implement hook_field_storage_rename_bundle().
+ * Implement hook_field_attach_rename_bundle().
  */
-function field_sql_storage_field_storage_rename_bundle($bundle_old, $bundle_new) {
-  $instances = field_info_instances($bundle_old);
+function field_sql_storage_field_attach_rename_bundle($bundle_old, $bundle_new) {
+  // We need to account for deleted or inactive fields and instances.
+  $instances = field_read_instances(array('bundle' => $bundle_new), array('include_deleted' => TRUE, 'include_inactive' => TRUE));
   foreach ($instances as $instance) {
-    $field = field_read_field($instance['field_name']);
-    $table_name = _field_sql_storage_tablename($field);
-    $revision_name = _field_sql_storage_revision_tablename($field);
-    db_update($table_name)
-      ->fields(array('bundle' => $bundle_new))
-      ->condition('bundle', $bundle_old)
-      ->execute();
-    db_update($revision_name)
-      ->fields(array('bundle' => $bundle_new))
-      ->condition('bundle', $bundle_old)
-      ->execute();
+    $field = field_info_field_by_id($instance['field_id']);
+    if ($field['storage']['type'] == 'field_sql_storage') {
+      $table_name = _field_sql_storage_tablename($field);
+      $revision_name = _field_sql_storage_revision_tablename($field);
+      db_update($table_name)
+        ->fields(array('bundle' => $bundle_new))
+        ->condition('bundle', $bundle_old)
+        ->execute();
+      db_update($revision_name)
+        ->fields(array('bundle' => $bundle_new))
+        ->condition('bundle', $bundle_old)
+        ->execute();
+    }
   }
 }
 
diff --git a/modules/field/modules/field_sql_storage/field_sql_storage.test b/modules/field/modules/field_sql_storage/field_sql_storage.test
index 5f8a30a42..7108913da 100644
--- a/modules/field/modules/field_sql_storage/field_sql_storage.test
+++ b/modules/field/modules/field_sql_storage/field_sql_storage.test
@@ -183,8 +183,8 @@ class FieldSqlStorageTestCase extends DrupalWebTestCase {
     $this->assertTrue(empty($rev_values), "All values for all revisions are stored in revision table {$this->revision_table}");
 
     // Check that update leaves the field data untouched if
-    // $object->{$field_name} has no language key.
-    unset($entity->{$this->field_name}[$langcode]);
+    // $object->{$field_name} is absent.
+    unset($entity->{$this->field_name});
     field_attach_update($entity_type, $entity);
     $rows = db_select($this->table, 't')->fields('t')->execute()->fetchAllAssoc('delta', PDO::FETCH_ASSOC);
     foreach ($values as $delta => $value) {
@@ -194,7 +194,7 @@ class FieldSqlStorageTestCase extends DrupalWebTestCase {
     }
 
     // Check that update with an empty $object->$field_name empties the field.
-    $entity->{$this->field_name}[$langcode] = NULL;
+    $entity->{$this->field_name} = NULL;
     field_attach_update($entity_type, $entity);
     $rows = db_select($this->table, 't')->fields('t')->execute()->fetchAllAssoc('delta', PDO::FETCH_ASSOC);
     $this->assertEqual(count($rows), 0, t("Update with an empty field_name entry empties the field."));
@@ -217,7 +217,7 @@ class FieldSqlStorageTestCase extends DrupalWebTestCase {
     $this->assertEqual($count, 0, 'Missing field results in no inserts');
 
     // Insert: Field is NULL
-    $entity->{$this->field_name}[$langcode] = NULL;
+    $entity->{$this->field_name} = NULL;
     field_attach_insert($entity_type, $entity);
     $count = db_select($this->table)
       ->countQuery()
diff --git a/modules/simpletest/tests/field_test.module b/modules/simpletest/tests/field_test.module
index 1f21e5dc4..93b8084e5 100644
--- a/modules/simpletest/tests/field_test.module
+++ b/modules/simpletest/tests/field_test.module
@@ -641,6 +641,436 @@ function field_test_entity_info_translatable($obj_type = NULL, $translatable = N
 }
 
 /**
+ *
+ * 'Field storage' API.
+ *
+ */
+
+/**
+ * Implement hook_field_storage_info().
+ */
+function field_test_field_storage_info() {
+  return array(
+    'field_test_storage' => array(
+      'label' => t('Test storage'),
+      'description' => t('Dummy test storage backend. Stores field values in the variable table.'),
+    ),
+    'field_test_storage_failure' => array(
+      'label' => t('Test storage failure'),
+      'description' => t('Dummy test storage backend. Always fails to create fields.'),
+    ),
+  );
+}
+
+/**
+ * Helper function: store or retrieve data from the 'storage backend'.
+ */
+function _field_test_storage_data($data = NULL) {
+  if (is_null($data)) {
+    return variable_get('field_test_storage_data', array());
+  }
+  else {
+    variable_set('field_test_storage_data', $data);
+  }
+}
+
+/**
+ * Implement hook_field_storage_load().
+ */
+function field_test_field_storage_load($obj_type, $objects, $age, $fields, $options) {
+  $data = _field_test_storage_data();
+
+  $load_current = $age == FIELD_LOAD_CURRENT;
+
+  foreach ($fields as $field_id => $ids) {
+    $field = field_info_field_by_id($field_id);
+    $field_name = $field['field_name'];
+    $field_data = $data[$field['id']];
+    $sub_table = $load_current ? 'current' : 'revisions';
+    $delta_count = array();
+    foreach ($field_data[$sub_table] as $row) {
+      if ($row->type == $obj_type && (!$row->deleted || $options['deleted'])) {
+        if (($load_current && in_array($row->entity_id, $ids)) || (!$load_current && in_array($row->revision_id, $ids))) {
+          if (in_array($row->language, field_multilingual_available_languages($obj_type, $field))) {
+            if (!isset($delta_count[$row->entity_id][$row->language])) {
+              $delta_count[$row->entity_id][$row->language] = 0;
+            }
+            if ($field['cardinality'] == FIELD_CARDINALITY_UNLIMITED || $delta_count[$row->entity_id][$row->language] < $field['cardinality']) {
+              $item = array();
+              foreach ($field['columns'] as $column => $attributes) {
+                $item[$column] = $row->{$column};
+              }
+              $objects[$row->entity_id]->{$field_name}[$row->language][] = $item;
+              $delta_count[$row->entity_id][$row->language]++;
+            }
+          }
+        }
+      }
+    }
+  }
+}
+
+/**
+ * Implement hook_field_storage_write().
+ */
+function field_test_field_storage_write($obj_type, $object, $op, $fields) {
+  $data = _field_test_storage_data();
+
+  list($id, $vid, $bundle) = field_extract_ids($obj_type, $object);
+
+  foreach ($fields as $field_id) {
+    $field = field_info_field_by_id($field_id);
+    $field_name = $field['field_name'];
+    $field_data = &$data[$field_id];
+
+    $all_languages = field_multilingual_available_languages($obj_type, $field);
+    $field_languages = array_intersect($all_languages, array_keys((array) $object->$field_name));
+
+    // Delete and insert, rather than update, in case a value was added.
+    if ($op == FIELD_STORAGE_UPDATE) {
+      // Delete languages present in the incoming $object->$field_name.
+      // Delete all languages if $object->$field_name is empty.
+      $languages = !empty($object->$field_name) ? $field_languages : $all_languages;
+      if ($languages) {
+        foreach ($field_data['current'] as $key => $row) {
+          if ($row->type == $obj_type && $row->entity_id == $id && in_array($row->language, $languages)) {
+            unset($field_data['current'][$key]);
+          }
+        }
+        if (isset($vid)) {
+          foreach ($field_data['revisions'] as $key => $row) {
+            if ($row->type == $obj_type && $row->revision_id == $vid) {
+              unset($field_data['revisions'][$key]);
+            }
+          }
+        }
+      }
+    }
+
+    foreach ($field_languages as $langcode) {
+      $items = (array) $object->{$field_name}[$langcode];
+      $delta_count = 0;
+      foreach ($items as $delta => $item) {
+        $row = (object) array(
+          'field_id' => $field_id,
+          'type' => $obj_type,
+          'entity_id' => $id,
+          'revision_id' => $vid,
+          'bundle' => $bundle,
+          'delta' => $delta,
+          'deleted' => FALSE,
+          'language' => $langcode,
+        );
+        foreach ($field['columns'] as $column => $attributes) {
+          $row->{$column} = isset($item[$column]) ? $item[$column] : NULL;
+        }
+
+        $field_data['current'][] = $row;
+        if (isset($vid)) {
+          $field_data['revisions'][] = $row;
+        }
+
+        if ($field['cardinality'] != FIELD_CARDINALITY_UNLIMITED && ++$delta_count == $field['cardinality']) {
+          break;
+        }
+      }
+    }
+  }
+
+  _field_test_storage_data($data);
+}
+
+/**
+ * Implement hook_field_storage_delete().
+ */
+function field_test_field_storage_delete($obj_type, $object, $fields) {
+  list($id, $vid, $bundle) = field_extract_ids($obj_type, $object);
+
+  // Note: reusing field_test_storage_purge(), like field_sql_storage.module
+  // does, is highly inefficient in our case...
+  foreach (field_info_instances($bundle) as $instance) {
+    if (isset($fields[$instance['field_id']])) {
+      $field = field_info_field_by_id($instance['field_id']);
+      field_test_field_storage_purge($obj_type, $object, $field, $instance);
+    }
+  }
+}
+
+/**
+ * Implement hook_field_storage_purge().
+ */
+function field_test_field_storage_purge($obj_type, $object, $field, $instance) {
+  $data = _field_test_storage_data();
+
+  list($id, $vid, $bundle) = field_extract_ids($obj_type, $object);
+
+  $field_data = &$data[$field['id']];
+  foreach (array('current', 'revisions') as $sub_table) {
+    foreach ($field_data[$sub_table] as $key => $row) {
+      if ($row->type == $obj_type && $row->entity_id == $id) {
+        unset($field_data[$sub_table][$key]);
+      }
+    }
+  }
+
+  _field_test_storage_data($data);
+}
+
+/**
+ * Implement hook_field_storage_delete_revision().
+ */
+function field_test_field_storage_delete_revision($obj_type, $object, $fields) {
+  $data = _field_test_storage_data();
+
+  list($id, $vid, $bundle) = field_extract_ids($obj_type, $object);
+  foreach ($fields as $field_id) {
+    $field_data = &$data[$field_id];
+    foreach (array('current', 'revisions') as $sub_table) {
+      foreach ($field_data[$sub_table] as $key => $row) {
+        if ($row->type == $obj_type && $row->entity_id == $id && $row->revision_id == $vid) {
+          unset($field_data[$sub_table][$key]);
+        }
+      }
+    }
+  }
+
+  _field_test_storage_data($data);
+}
+
+/**
+ * Implement hook_field_storage_query().
+ */
+function field_test_field_storage_query($field_id, $conditions, $count, &$cursor = NULL, $age) {
+  $data = _field_test_storage_data();
+
+  $load_current = $age == FIELD_LOAD_CURRENT;
+
+  $field = field_info_field_by_id($field_id);
+  $field_columns = array_keys($field['columns']);
+
+  $field_data = $data[$field['id']];
+  $sub_table = $load_current ? 'current' : 'revisions';
+  // We need to sort records by object type and object id.
+  usort($field_data[$sub_table], '_field_test_field_storage_query_sort_helper');
+
+    // Initialize results array.
+  $return = array();
+  $obj_count = 0;
+  $rows_count = 0;
+  $rows_total = count($field_data[$sub_table]);
+  $skip = $cursor;
+  $skipped = 0;
+
+  foreach ($field_data[$sub_table] as $row) {
+    if ($count != FIELD_QUERY_NO_LIMIT && $obj_count >= $count) {
+      break;
+    }
+
+    if ($row->field_id == $field['id']) {
+      $match = TRUE;
+      $condition_deleted = FALSE;
+      // Add conditions.
+      foreach ($conditions as $condition) {
+        @list($column, $value, $operator) = $condition;
+        if (empty($operator)) {
+          $operator = is_array($value) ? 'IN' : '=';
+        }
+        switch ($operator) {
+          case '=':
+            $match = $match && $row->{$column} == $value;
+            break;
+          case '!=':
+          case '<':
+          case '<=':
+          case '>':
+          case '>=':
+            eval('$match = $match && '. $row->{$column} . ' ' . $operator . ' '. $value);
+            break;
+          case 'IN':
+            $match = $match && in_array($row->{$column}, $value);
+            break;
+          case 'NOT IN':
+            $match = $match && !in_array($row->{$column}, $value);
+            break;
+          case 'BETWEEN':
+            $match = $match && $row->{$column} >= $value[0] && $row->{$column} <= $value[1];
+            break;
+          case 'STARTS_WITH':
+          case 'ENDS_WITH':
+          case 'CONTAINS':
+            // Not supported.
+            $match = FALSE;
+            break;
+        }
+        // Track condition on 'deleted'.
+        if ($column == 'deleted') {
+          $condition_deleted = TRUE;
+        }
+      }
+
+      // Exclude deleted data unless we have a condition on it.
+      if (!$condition_deleted && $row->deleted) {
+        $match = FALSE;
+      }
+
+      if ($match) {
+        if (is_null($skip) || $skipped >= $skip) {
+          $cursor++;
+          // If querying all revisions and the entity type has revisions, we need
+          // to key the results by revision_ids.
+          $entity_type = field_info_fieldable_types($row->type);
+          $id = ($load_current || empty($entity_type['object keys']['revision'])) ? $row->entity_id : $row->revision_id;
+
+          if (!isset($return[$row->type][$id])) {
+            $return[$row->type][$id] = field_create_stub_entity($row->type, array($row->entity_id, $row->revision_id, $row->bundle));
+            $obj_count++;
+          }
+        }
+        else {
+          $skipped++;
+        }
+      }
+    }
+    $rows_count++;
+
+    // The query is complete if we walked the whole array.
+    if ($count != FIELD_QUERY_NO_LIMIT && $rows_count >= $rows_total) {
+      $cursor = FIELD_QUERY_COMPLETE;
+    }
+  }
+
+  return $return;
+}
+
+/**
+ * Sort helper for field_test_field_storage_query().
+ *
+ * Sort by object type and object id.
+ */
+function _field_test_field_storage_query_sort_helper($a, $b) {
+  if ($a->type == $b->type) {
+    if ($a->entity_id == $b->entity_id) {
+      return 0;
+    }
+    else {
+      return $a->entity_id < $b->entity_id ? -1 : 1;
+    }
+  }
+  else {
+    return $a->type < $b->type ? -1 : 1;
+  }
+}
+
+/**
+ * Implement hook_field_storage_create_field().
+ */
+function field_test_field_storage_create_field($field) {
+  if ($field['storage']['type'] == 'field_test_storage_failure') {
+    throw new Exception('field_test_storage_failure engine always fails to create fields');
+  }
+    
+  $data = _field_test_storage_data();
+
+  $data[$field['id']] = array(
+    'current' => array(),
+    'revisions' => array(),
+  );
+
+  _field_test_storage_data($data);
+}
+
+/**
+ * Implement hook_field_storage_delete_field().
+ */
+function field_test_field_storage_delete_field($field) {
+  $data = _field_test_storage_data();
+
+  $field_data = &$data[$field['id']];
+  foreach (array('current', 'revisions') as $sub_table) {
+    foreach ($field_data[$sub_table] as &$row) {
+      $row->deleted = TRUE;
+    }
+  }
+
+  _field_test_storage_data($data);
+}
+
+/**
+ * Implement hook_field_storage_delete_instance().
+ */
+function field_test_field_storage_delete_instance($instance) {
+  $data = _field_test_storage_data();
+
+  $field = field_info_field($instance['field_name']);
+  $field_data = &$data[$field['id']];
+  foreach (array('current', 'revisions') as $sub_table) {
+    foreach ($field_data[$sub_table] as &$row) {
+      if ($row->bundle == $instance['bundle']) {
+        $row->deleted = TRUE;
+      }
+    }
+  }
+
+  _field_test_storage_data($data);
+}
+
+/**
+ * Implement hook_field_attach_create_bundle().
+ */
+function field_test_field_attach_create_bundle($bundle) {
+  // We don't need to do anything here.
+}
+
+/**
+ * Implement hook_field_attach_rename_bundle().
+ */
+function field_test_field_attach_rename_bundle($bundle_old, $bundle_new) {
+  $data = _field_test_storage_data();
+
+  // We need to account for deleted or inactive fields and instances.
+  $instances = field_read_instances(array('bundle' => $bundle_new), array('include_deleted' => TRUE, 'include_inactive' => TRUE));
+  foreach ($instances as $field_name => $instance) {
+    $field = field_info_field_by_id($instance['field_id']);
+    if ($field['storage']['type'] == 'field_test_storage') {
+      $field_data = &$data[$field['id']];
+      foreach (array('current', 'revisions') as $sub_table) {
+        foreach ($field_data[$sub_table] as &$row) {
+          if ($row->bundle == $bundle_old) {
+            $row->bundle = $bundle_new;
+          }
+        }
+      }
+    }
+  }
+
+  _field_test_storage_data($data);
+}
+
+/**
+ * Implement hook_field_attach_delete_bundle().
+ */
+function field_test_field_attach_delete_bundle($bundle, $instances) {
+  $data = _field_test_storage_data();
+
+  $instances = field_info_instances($bundle);
+  foreach ($instances as $field_name => $instance) {
+    $field = field_info_field($field_name);
+    if ($field['storage']['type'] == 'field_test_storage') {
+      $field_data = &$data[$field['id']];
+      foreach (array('current', 'revisions') as $sub_table) {
+        foreach ($field_data[$sub_table] as &$row) {
+          if ($row->bundle == $bundle_old) {
+            $row->deleted = TRUE;
+          }
+        }
+      }
+    }
+  }
+
+  _field_test_storage_data($data);
+}
+
+/**
  * Store and retrieve keyed data for later verification by unit tests.
  *
  * This function is a simple in-memory key-value store with the
@@ -725,13 +1155,3 @@ function field_test_field_delete($obj_type, $object, $field, $instance, $items)
   $args = func_get_args();
   field_test_memorize(__FUNCTION__, $args);
 }
-
-/**
- *
- * 'Field storage' API.
- *
- */
-
-function field_test_field_storage_create_field($field) {
-  throw new Exception('field_test storage module always fails to create fields');
-}