diff --git a/modules/field/field.api.php b/modules/field/field.api.php
index 6fd2cef32c7766e5264be23d02dbd0cc59e9659c..380444af11862c48e641b8ea8f3aa5d6b1fbbfd4 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.
  */
 
 /**
@@ -1096,6 +1099,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.
  *
@@ -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) {
 }
 
 /**
@@ -1185,26 +1232,6 @@ function hook_field_storage_delete_revision($obj_type, $object) {
 function hook_field_storage_query($field_name, $conditions, $count, &$cursor = NULL, $age) {
 }
 
-/**
- * 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.
  *
@@ -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 7571ee4fc851b55d1776d42cf7e3cc5e097ce0b6..e2726a7760d4b627b155c7510527b40a45550905 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 210dad13626defac9ea4e002cf45c258dda6320b..88e153a03dd5ddd44b58210f5f877eefdc4cd95c 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 aa5b4d08bd884e81f9c9e822de39eda170c081a0..8b273da6a7fb51addae4556a2d0848ec362088f8 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();
@@ -410,6 +427,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.
  *
@@ -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,14 +629,28 @@ 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);
   return isset($info['settings']) ? $info['settings'] : array();
 }
 
+/**
+ * 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 ba3db1df5ae0de07a3292f6587cc96c195e38f7b..c112ef1b67691569ed09fd4efa4051716d051042 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 75dc3d43111f702e48a07868902cc3a9bf90bfee..4d031e935889178989dfb5d6f2d1192ca7bba3ce 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 2523c8b721409db6332775ca053d0440e1280e6b..c77d790c92b44b63624da083baa875a87074fae0 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.
@@ -211,6 +223,56 @@ class FieldAttachStorageTestCase extends FieldAttachTestCase {
     $this->assert(!isset($entity->{$field_names[3]}), t('Entity %index: field %field_name is not loaded.', array('%index' => 3, '%field_name' => $field_names[3])));
   }
 
+  /**
+   * 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.
    */
@@ -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 67b7e4afc08f3c10b422b32e988e82ef1d261cd7..7ab6a8cf9aec5a841f91e7d2df7e943056c43d0b 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 97d372f3bb27aed82d58a6b6273d6e0b447d287c..09dee368a302bf71262d24d2b3b8afb8165d394c 100644
--- a/modules/field/modules/field_sql_storage/field_sql_storage.module
+++ b/modules/field/modules/field_sql_storage/field_sql_storage.module
@@ -17,6 +17,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.
  *
@@ -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 5f8a30a42a4abee3bb99db19d0bc90b67208f9c0..7108913da0cebf4558b3b1f90cc0ccbe32fef9df 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 1f21e5dc44d25fd9f3ac5172eda15121577527ab..93b8084e5b36c03dc4e097c6f1424d49172f3c9b 100644
--- a/modules/simpletest/tests/field_test.module
+++ b/modules/simpletest/tests/field_test.module
@@ -640,6 +640,436 @@ function field_test_entity_info_translatable($obj_type = NULL, $translatable = N
   return $stored_value;
 }
 
+/**
+ *
+ * '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.
  *
@@ -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');
-}