Index: modules/field/field.test
===================================================================
RCS file: /cvs/drupal/drupal/modules/field/field.test,v
retrieving revision 1.38
diff -u -r1.38 field.test
--- modules/field/field.test	10 Aug 2009 21:19:42 -0000	1.38
+++ modules/field/field.test	11 Aug 2009 04:53:30 -0000
@@ -184,6 +184,55 @@
   }
 
   /**
+   * Test saving and loading fields using different storage backends.
+   */
+  function testFieldAttachSaveLoadDifferentStorage() {
+    $entity_type = 'test_entity';
+
+    // 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']} = $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']}, t('%storage storage: expected values were found.', array('%storage' => $field['storage']['type'])));
+    }
+  }
+
+  /**
    * Tests insert and update with missing or NULL fields.
    */
   function testFieldAttachSaveMissingData() {
@@ -1598,6 +1647,64 @@
       $this->assertEqual($entity->{$field['field_name']}[$delta]['value'], $values[$delta]['value'], "Data in previously deleted field saves and loads correctly");
     }
   }
+
+  /**
+   * Test that fields are properly marked active or inactive.
+   */
+  function testActive() {
+    $field_definition = array(
+      'field_name' => 'field_1',
+      'type' => 'test_field',
+    );
+    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 DrupalWebTestCase {
Index: modules/field/field.info.inc
===================================================================
RCS file: /cvs/drupal/drupal/modules/field/field.info.inc,v
retrieving revision 1.11
diff -u -r1.11 field.info.inc
--- modules/field/field.info.inc	2 Aug 2009 11:24:21 -0000	1.11
+++ modules/field/field.info.inc	11 Aug 2009 04:53:29 -0000
@@ -64,6 +64,7 @@
         'field types' => array(),
         'widget types' => array(),
         'formatter types' => array(),
+        'storage types' => array(),
         'fieldable types' => array(),
       );
 
@@ -96,7 +97,7 @@
       }
       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) {
@@ -110,6 +111,20 @@
       }
       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('fieldable_info') as $module) {
         $fieldable_types = (array) module_invoke($module, 'fieldable_info');
@@ -212,6 +227,7 @@
 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;
 }
@@ -332,8 +348,8 @@
  *   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();
@@ -355,8 +371,9 @@
  *   (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();
@@ -372,6 +389,30 @@
 }
 
 /**
+ * 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
@@ -516,8 +557,8 @@
  * @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);
@@ -530,8 +571,8 @@
  * @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);
@@ -539,5 +580,19 @@
 }
 
 /**
+ * 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"
  */
Index: modules/field/field.install
===================================================================
RCS file: /cvs/drupal/drupal/modules/field/field.install,v
retrieving revision 1.9
diff -u -r1.9 field.install
--- modules/field/field.install	28 May 2009 10:05:32 -0000	1.9
+++ modules/field/field.install	11 Aug 2009 04:53:29 -0000
@@ -35,35 +35,57 @@
         '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.',
       ),
-      'active' => array(
+      'locked' => array(
+        'type' => 'int',
+        'size' => 'tiny',
+        'not null' => TRUE,
+        'default' => 0,
+        'description' => '@TODO',
+      ),
+      'data' => array(
+        'type' => 'text',
+        'size' => 'medium',
+        'not null' => TRUE,
+        'serialize' => TRUE,
+        'description' => 'Field specific settings, for example maximum length',
+      ),
+      'cardinality' => array(
         'type' => 'int',
         'size' => 'tiny',
         'not null' => TRUE,
@@ -78,14 +100,17 @@
     ),
     '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(
@@ -133,11 +158,13 @@
       'field_id_bundle' => array('field_id', 'bundle'),
     ),
     'indexes' => array(
-      // used by field_read_instances()
-      'widget_active_deleted' => array('widget_active', 'deleted'),
-      // used by field_modules_disabled()
+      // TODO : missing indexes on field_name and bundle ?
+      // 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'),
     ),
   );
Index: modules/field/field.attach.inc
===================================================================
RCS file: /cvs/drupal/drupal/modules/field/field.attach.inc,v
retrieving revision 1.34
diff -u -r1.34 field.attach.inc
--- modules/field/field.attach.inc	11 Aug 2009 04:46:29 -0000	1.34
+++ modules/field/field.attach.inc	11 Aug 2009 04:53:28 -0000
@@ -44,17 +44,16 @@
  * @{
  * 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.
  */
 
 /**
@@ -492,7 +491,7 @@
   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()
 
@@ -504,9 +503,34 @@
       $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_attach_extract_ids($obj_type, $obj);
+      foreach (field_info_instances($bundle) as $field_name => $instance) {
+        if (!isset($options['field_name']) || $options['field_name'] == $field_name) {
+          // 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_name])) {
+            $field = field_info_field($field_name);
+            // TODO : revisit after 'bulk delete' gets in.
+            $storages[$field['storage']['type']][$field_name][] = $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);
+      $function = $storage_info['module'] . '_field_storage_load';
+      if (drupal_function_exists($function)) {
+        $function($obj_type, $queried_objects, $age, $fields);
+      }
+    }
 
     // Invoke field-type module's hook_field_load().
     _field_invoke_multiple('load', $obj_type, $queried_objects, $age, $options);
@@ -713,8 +737,11 @@
     $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);
+  // Field storage backends save any remaining unsaved fields.
+  foreach (module_implements('field_storage_write') as $module) {
+    $function = $module . '_field_storage_write';
+    $function($obj_type, $object, FIELD_STORAGE_INSERT, $skip_fields);
+  }
 
   list($id, $vid, $bundle, $cacheable) = field_attach_extract_ids($obj_type, $object);
   if ($cacheable) {
@@ -741,8 +768,11 @@
     $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);
+  // Field storage backends save any remaining unsaved fields.
+  foreach (module_implements('field_storage_write') as $module) {
+    $function = $module . '_field_storage_write';
+    $function($obj_type, $object, FIELD_STORAGE_UPDATE, $skip_fields);
+  }
 
   list($id, $vid, $bundle, $cacheable) = field_attach_extract_ids($obj_type, $object);
   if ($cacheable) {
@@ -761,7 +791,12 @@
  */
 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);
+
+  // Field storage backends delete their data.
+  foreach (module_implements('field_storage_delete') as $module) {
+    $function = $module . '_field_storage_delete';
+    $function($obj_type, $object);
+  }
 
   // Let other modules act on deleting the object.
   foreach (module_implements('field_attach_delete') as $module) {
@@ -786,7 +821,12 @@
  */
 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);
+
+  // Field storage backends delete their data.
+  foreach (module_implements('field_storage_delete_revision') as $module) {
+    $function = $module . '_field_storage_delete_revision';
+    $function($obj_type, $object);
+  }
 
   // Let other modules act on deleting the revision.
   foreach (module_implements('field_attach_delete_revision') as $module) {
@@ -889,7 +929,8 @@
   }
   // 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($field_name);
+    $function = $field['storage']['module'] . '_field_storage_query';
     $results = $function($field_name, $conditions, $count, $cursor, $age);
   }
 
@@ -1029,8 +1070,6 @@
  *   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();
 
@@ -1049,7 +1088,6 @@
  *   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)
@@ -1078,12 +1116,15 @@
  *   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';
Index: modules/field/field.crud.inc
===================================================================
RCS file: /cvs/drupal/drupal/modules/field/field.crud.inc,v
retrieving revision 1.23
diff -u -r1.23 field.crud.inc
--- modules/field/field.crud.inc	10 Aug 2009 21:19:42 -0000	1.23
+++ modules/field/field.crud.inc	11 Aug 2009 04:53:29 -0000
@@ -74,6 +74,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:
@@ -193,6 +207,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
@@ -220,12 +239,6 @@
       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.
   // TODO : do we want specific messages when clashing with a disabled or inactive field ?
@@ -238,14 +251,34 @@
     'cardinality' => 1,
     '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);
@@ -265,15 +298,29 @@
   // 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'],
+  );
 
-  // Invoke hook_field_storage_create_field after the field is
-  // complete (e.g. it has its id).
-  module_invoke(variable_get('field_storage_module', 'field_sql_storage'), 'field_storage_create_field', $field);
+  // 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 drupal_write_record() sets the
+  // field id.
+  module_invoke($storage_type['module'], 'field_storage_create_field', $field);
 
   // Clear caches
   field_cache_clear(TRUE);
@@ -331,7 +378,9 @@
     $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) {
@@ -370,8 +419,10 @@
  *   The field name to delete.
  */
 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);
+  // TODO: this should accept a $field structure, and act on field_id.
+  // Mark field data for deletion.
+  $field = field_read_field($field_name);
+  module_invoke($field['storage']['module'], 'field_storage_delete_field', $field_name);
 
   // Mark any instances of the field for deletion.
   db_update('field_config_instance')
@@ -558,7 +609,7 @@
   // 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['weight'], $data['deleted']);
+  unset($data['id'], $data['field_id'], $data['field_name'], $data['bundle'], $data['widget']['type'], $data['widget']['module'], $data['widget']['active'], $data['weight'], $data['deleted']);
 
   $record = array(
     'field_id' => $instance['field_id'],
@@ -679,8 +730,10 @@
     ->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_read_field($field_name);
+  module_invoke($field['storage']['module'], 'field_storage_delete_instance', $field_name, $bundle);
+
   // Clear the cache.
   field_cache_clear();
 }
Index: modules/field/field.api.php
===================================================================
RCS file: /cvs/drupal/drupal/modules/field/field.api.php,v
retrieving revision 1.24
diff -u -r1.24 field.api.php
--- modules/field/field.api.php	1 Aug 2009 06:03:12 -0000	1.24
+++ modules/field/field.api.php	11 Aug 2009 04:53:28 -0000
@@ -107,10 +107,10 @@
 /**
  * @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.
@@ -133,6 +133,9 @@
  * 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.
  */
 
 /**
@@ -1070,6 +1073,45 @@
  */
 
 /**
+ * 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
@@ -1080,15 +1122,13 @@
  *   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 names of fields whose data has already been loaded and
- *   therefore should not be loaded again. The values associated to these keys
- *   are not specified.
+ * @param $fields
+ *   TODO
  * @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) {
 }
 
 /**
@@ -1159,26 +1199,6 @@
 }
 
 /**
- * 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
Index: modules/field/field.module
===================================================================
RCS file: /cvs/drupal/drupal/modules/field/field.module,v
retrieving revision 1.21
diff -u -r1.21 field.module
--- modules/field/field.module	2 Aug 2009 11:24:21 -0000	1.21
+++ modules/field/field.module	11 Aug 2009 04:53:29 -0000
@@ -214,6 +214,10 @@
       ->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)
@@ -229,25 +233,32 @@
  *   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();
   }
 }
 
Index: modules/field/modules/field_sql_storage/field_sql_storage.module
===================================================================
RCS file: /cvs/drupal/drupal/modules/field/modules/field_sql_storage/field_sql_storage.module,v
retrieving revision 1.17
diff -u -r1.17 field_sql_storage.module
--- modules/field/modules/field_sql_storage/field_sql_storage.module	15 Jul 2009 17:55:18 -0000	1.17
+++ modules/field/modules/field_sql_storage/field_sql_storage.module	11 Aug 2009 04:53:31 -0000
@@ -18,6 +18,18 @@
 }
 
 /**
+ * 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
@@ -102,79 +114,85 @@
  *   One or more tables representing the schema for the field.
  */
 function _field_sql_storage_schema($field) {
-  $deleted = $field['deleted'] ? 'deleted ' : '';
-  $current = array(
-    'description' => "Data storage for {$deleted}field {$field['id']} ({$field['field_name']})",
-    'fields' => array(
-      'etid' => array(
-        'type' => 'int',
-        'unsigned' => TRUE,
-        'not null' => TRUE,
-        'description' => 'The entity type id this data is attached to',
-      ),
-      'bundle' => array(
-        'type' => 'varchar',
-        'length' => 32,
-        'not null' => TRUE,
-        'default' => '',
-        'description' => 'The field instance bundle to which this row belongs, used when deleting a field instance',
-      ),
-      'deleted' => array(
-        'type' => 'int',
-        'size' => 'tiny',
-        'not null' => TRUE,
-        'default' => 0,
-        'description' => 'A boolean indicating whether this data item has been deleted'
-      ),
-      'entity_id' => array(
-        'type' => 'int',
-        'unsigned' => TRUE,
-        'not null' => TRUE,
-        'description' => 'The entity id this data is attached to',
-      ),
-      'revision_id' => array(
-        'type' => 'int',
-        'unsigned' => TRUE,
-        'not null' => FALSE,
-        'description' => 'The entity revision id this data is attached to, or NULL if the entity type is not versioned',
-      ),
-      'delta' => array(
-        'type' => 'int',
-        'unsigned' => TRUE,
-        'not null' => TRUE,
-        'description' => 'The sequence number for this data item, used for multi-value fields',
+  $schema = array();
+
+  if ($field['storage']['type'] == 'field_sql_storage') {
+    $deleted = $field['deleted'] ? 'deleted ' : '';
+    $current = array(
+      'description' => "Data storage for {$deleted}field {$field['id']} ({$field['field_name']})",
+      'fields' => array(
+        'etid' => array(
+          'type' => 'int',
+          'unsigned' => TRUE,
+          'not null' => TRUE,
+          'description' => 'The entity type id this data is attached to',
+        ),
+        'bundle' => array(
+          'type' => 'varchar',
+          'length' => 32,
+          'not null' => TRUE,
+          'default' => '',
+          'description' => 'The field instance bundle to which this row belongs, used when deleting a field instance',
+        ),
+        'deleted' => array(
+          'type' => 'int',
+          'size' => 'tiny',
+          'not null' => TRUE,
+          'default' => 0,
+          'description' => 'A boolean indicating whether this data item has been deleted'
+        ),
+        'entity_id' => array(
+          'type' => 'int',
+          'unsigned' => TRUE,
+          'not null' => TRUE,
+          'description' => 'The entity id this data is attached to',
+        ),
+        'revision_id' => array(
+          'type' => 'int',
+          'unsigned' => TRUE,
+          'not null' => FALSE,
+          'description' => 'The entity revision id this data is attached to, or NULL if the entity type is not versioned',
+        ),
+        'delta' => array(
+          'type' => 'int',
+          'unsigned' => TRUE,
+          'not null' => TRUE,
+          'description' => 'The sequence number for this data item, used for multi-value fields',
+        ),
       ),
-    ),
-    'primary key' => array('etid', 'entity_id', 'deleted', 'delta'),
-    // TODO : index on 'bundle'
-  );
+      'primary key' => array('etid', 'entity_id', 'deleted', 'delta'),
+      // TODO : index on 'bundle'
+    );
+
+    // Add field columns.
+    foreach ((array) $field['columns'] as $column_name => $attributes) {
+      $real_name = _field_sql_storage_columnname($field['field_name'], $column_name);
+      $current['fields'][$real_name] = $attributes;
+    }
 
-  // Add field columns.
-  foreach ((array) $field['columns'] as $column_name => $attributes) {
-    $real_name = _field_sql_storage_columnname($field['field_name'], $column_name);
-    $current['fields'][$real_name] = $attributes;
-  }
-
-  // Add indexes.
-  foreach ((array) $field['indexes'] as $index_name => $columns) {
-    $real_name = _field_sql_storage_indexname($field['field_name'], $index_name);
-    foreach ($columns as $column_name) {
-      $current['indexes'][$real_name][] = _field_sql_storage_columnname($field['field_name'], $column_name);
+    // Add indexes.
+    foreach ((array) $field['indexes'] as $index_name => $columns) {
+      $real_name = _field_sql_storage_indexname($field['field_name'], $index_name);
+      foreach ($columns as $column_name) {
+        $current['indexes'][$real_name][] = _field_sql_storage_columnname($field['field_name'], $column_name);
+      }
     }
-  }
 
-  // Construct the revision table. The primary key includes
-  // revision_id but not entity_id so that multiple revision loads can
-  // use the IN operator.
-  $revision = $current;
-  $revision['description'] = "Revision archive storage for {$deleted}field {$field['id']} ({$field['field_name']})";
-  $revision['revision_id']['description'] = 'The entity revision id this data is attached to';
-  $revision['primary key'] = array('etid', 'revision_id', 'deleted', 'delta');
+    // Construct the revision table. The primary key includes
+    // revision_id but not entity_id so that multiple revision loads can
+    // use the IN operator.
+    $revision = $current;
+    $revision['description'] = "Revision archive storage for {$deleted}field {$field['id']} ({$field['field_name']})";
+    $revision['revision_id']['description'] = 'The entity revision id this data is attached to';
+    $revision['primary key'] = array('etid', 'revision_id', 'deleted', 'delta');
+
+    $schema = array(
+      _field_sql_storage_tablename($field) => $current,
+      _field_sql_storage_revision_tablename($field) => $revision,
+    );
+  }
 
-  return array(
-    _field_sql_storage_tablename($field) => $current,
-    _field_sql_storage_revision_tablename($field) => $revision,
-  );
+  return $schema;
 }
 
 /**
@@ -202,25 +220,11 @@
 /**
  * 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) {
   $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_attach_extract_ids($obj_type, $obj);
-    foreach (field_info_instances($bundle) as $field_name => $instance) {
-      if (!isset($skip_fields[$field_name]) && (!isset($options['field_name']) || $options['field_name'] == $instance['field_name'])) {
-        $objects[$id]->{$field_name} = array();
-        $field_ids[$field_name][] = $load_current ? $id : $vid;
-        $delta_count[$id][$field_name] = 0;
-      }
-    }
-  }
-
-  foreach ($field_ids as $field_name => $ids) {
+  foreach ($fields as $field_name => $ids) {
     $field = field_info_field($field_name);
     $table = $load_current ? _field_sql_storage_tablename($field) : _field_sql_storage_revision_tablename($field);
 
@@ -233,8 +237,12 @@
 
     $results = $query->execute();
 
+    $delta_count = array();
     foreach ($results as $row) {
-      if ($field['cardinality'] == FIELD_CARDINALITY_UNLIMITED || $delta_count[$row->entity_id][$field_name] < $field['cardinality']) {
+      if (!isset($delta_count[$row->entity_id])) {
+        $delta_count[$row->entity_id] = 0;
+      }
+      if ($field['cardinality'] == FIELD_CARDINALITY_UNLIMITED || $delta_count[$row->entity_id] < $field['cardinality']) {
         $item = array();
         // For each column declared by the field, populate the item
         // from the prefixed database column.
@@ -245,7 +253,7 @@
 
         // Add the item to the field values for the entity.
         $objects[$row->entity_id]->{$field_name}[] = $item;
-        $delta_count[$row->entity_id][$field_name]++;
+        $delta_count[$row->entity_id]++;
       }
     }
   }
@@ -259,67 +267,64 @@
   $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[$field_name])) {
-      continue;
-    }
-
+  foreach ($instances as $field_name => $instance) {
     $field = field_info_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.
+    if ($field['storage']['type'] == 'field_sql_storage' && !isset($skip_fields[$field_name])) {
+      $table_name = _field_sql_storage_tablename($field);
+      $revision_name = _field_sql_storage_revision_tablename($field);
 
-    // 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)) {
-      // Delete and insert, rather than update, in case a value was added.
-      if ($op == FIELD_STORAGE_UPDATE) {
-        db_delete($table_name)->condition('etid', $etid)->condition('entity_id', $id)->execute();
-        if (isset($vid)) {
-          db_delete($revision_name)->condition('etid', $etid)->condition('entity_id', $id)->condition('revision_id', $vid)->execute();
-        }
-      }
+      // 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.
 
-      if ($object->$field_name) {
-        // Prepare the multi-insert query.
-        $columns = array('etid', 'entity_id', 'revision_id', 'bundle', 'delta');
-        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);
+      // 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)) {
+        // Delete and insert, rather than update, in case a value was added.
+        if ($op == FIELD_STORAGE_UPDATE) {
+          db_delete($table_name)->condition('etid', $etid)->condition('entity_id', $id)->execute();
+          if (isset($vid)) {
+            db_delete($revision_name)->condition('etid', $etid)->condition('entity_id', $id)->condition('revision_id', $vid)->execute();
+          }
         }
 
-        $delta_count = 0;
-        foreach ($object->$field_name as $delta => $item) {
-          $record = array(
-            'etid' => $etid,
-            'entity_id' => $id,
-            'revision_id' => $vid,
-            'bundle' => $bundle,
-            'delta' => $delta,
-          );
+        if ($object->$field_name) {
+          // Prepare the multi-insert query.
+          $columns = array('etid', 'entity_id', 'revision_id', 'bundle', 'delta');
           foreach ($field['columns'] as $column => $attributes) {
-            $record[_field_sql_storage_columnname($field_name, $column)] = isset($item[$column]) ? $item[$column] : NULL;
+            $columns[] = _field_sql_storage_columnname($field_name, $column);
           }
-          $query->values($record);
+          $query = db_insert($table_name)->fields($columns);
           if (isset($vid)) {
-            $revision_query->values($record);
+            $revision_query = db_insert($revision_name)->fields($columns);
           }
 
-          if ($field['cardinality'] != FIELD_CARDINALITY_UNLIMITED && ++$delta_count == $field['cardinality']) {
-            break;
+          $delta_count = 0;
+          foreach ($object->$field_name as $delta => $item) {
+            $record = array(
+              'etid' => $etid,
+              'entity_id' => $id,
+              'revision_id' => $vid,
+              'bundle' => $bundle,
+              'delta' => $delta,
+            );
+            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;
+            }
           }
-        }
 
-        // Execute the insert.
-        $query->execute();
-        if (isset($vid)) {
-          $revision_query->execute();
+          // Execute the insert.
+          $query->execute();
+          if (isset($vid)) {
+            $revision_query->execute();
+          }
         }
       }
     }
@@ -336,22 +341,47 @@
   $etid = _field_sql_storage_etid($obj_type);
 
   $instances = field_info_instances($bundle);
-  foreach ($instances as $instance) {
-    $field_name = $instance['field_name'];
+  foreach ($instances as $field_name => $instance) {
     $field = field_read_field($field_name);
-    $table_name = _field_sql_storage_tablename($field);
-    $revision_name = _field_sql_storage_revision_tablename($field);
-    db_delete($table_name)
-      ->condition('etid', $etid)
-      ->condition('entity_id', $id)
-      ->execute();
-    db_delete($revision_name)
-      ->condition('etid', $etid)
-      ->condition('entity_id', $id)
-      ->execute();
+    if ($field['storage']['type'] == 'field_sql_storage') {
+      $table_name = _field_sql_storage_tablename($field);
+      $revision_name = _field_sql_storage_revision_tablename($field);
+      db_delete($table_name)
+        ->condition('etid', $etid)
+        ->condition('entity_id', $id)
+        ->execute();
+      db_delete($revision_name)
+        ->condition('etid', $etid)
+        ->condition('entity_id', $id)
+        ->execute();
+    }
   }
 }
 
+/**
+ * Implement hook_field_storage_delete_revision().
+ *
+ * This function actually deletes the data from the database.
+ */
+function field_sql_storage_field_storage_delete_revision($obj_type, $object) {
+  list($id, $vid, $bundle) = field_attach_extract_ids($obj_type, $object);
+  $etid = _field_sql_storage_etid($obj_type);
+
+  if (isset($vid)) {
+    $instances = field_info_instances($bundle);
+    foreach ($instances as $field_name => $instance) {
+      $field = field_read_field($field_name);
+      if ($field['storage']['type'] == 'field_sql_storage') {
+        $revision_name = _field_sql_storage_revision_tablename($field);
+        db_delete($revision_name)
+          ->condition('etid', $etid)
+          ->condition('entity_id', $id)
+          ->condition('revision_id', $vid)
+          ->execute();
+      }
+    }
+  }
+}
 
 /**
  * Implement hook_field_storage_query().
@@ -402,7 +432,7 @@
     $query->condition($column, $value, $operator);
   }
 
-  // Initialize results array
+  // Initialize results array.
   $return = array();
 
   // Getting $count objects possibly requires reading more than $count rows
@@ -443,64 +473,44 @@
 }
 
 /**
- * Implement hook_field_storage_delete_revision().
- *
- * This function actually deletes the data from the database.
- */
-function field_sql_storage_field_storage_delete_revision($obj_type, $object) {
-  list($id, $vid, $bundle) = field_attach_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);
-      $revision_name = _field_sql_storage_revision_tablename($field);
-      db_delete($revision_name)
-        ->condition('etid', $etid)
-        ->condition('entity_id', $id)
-        ->condition('revision_id', $vid)
-        ->execute();
-    }
-  }
-}
-
-/**
  * Implement hook_field_storage_delete_instance().
  *
  * 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);
-  $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)
-    ->execute();
-  db_update($revision_name)
-    ->fields(array('deleted' => 1))
-    ->condition('bundle', $bundle)
-    ->execute();
-}
-
-/**
- * Implement hook_field_storage_rename_bundle().
- */
-function field_sql_storage_field_storage_rename_bundle($bundle_old, $bundle_new) {
-  $instances = field_info_instances($bundle_old);
-  foreach ($instances as $instance) {
-    $field = field_read_field($instance['field_name']);
+  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)
+      ->fields(array('deleted' => 1))
+      ->condition('bundle', $bundle)
       ->execute();
     db_update($revision_name)
-      ->fields(array('bundle' => $bundle_new))
-      ->condition('bundle', $bundle_old)
+      ->fields(array('deleted' => 1))
+      ->condition('bundle', $bundle)
       ->execute();
   }
+}
+
+/**
+ * Implement hook_field_attach_rename_bundle().
+ */
+function field_sql_storage_field_attach_rename_bundle($bundle_old, $bundle_new) {
+  $instances = field_info_instances($bundle_new);
+  foreach ($instances as $instance) {
+    $field = field_read_field($instance['field_name']);
+    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();
+    }
+  }
 }
\ No newline at end of file
Index: modules/simpletest/tests/field_test.module
===================================================================
RCS file: /cvs/drupal/drupal/modules/simpletest/tests/field_test.module,v
retrieving revision 1.13
diff -u -r1.13 field_test.module
--- modules/simpletest/tests/field_test.module	10 Jul 2009 05:58:13 -0000	1.13
+++ modules/simpletest/tests/field_test.module	11 Aug 2009 04:53:31 -0000
@@ -581,6 +581,409 @@
 }
 
 /**
+ *
+ * '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.'),
+    ),
+  );
+}
+
+/**
+ * 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) {
+  $data = _field_test_storage_data();
+
+  $load_current = $age == FIELD_LOAD_CURRENT;
+
+  foreach ($fields as $field_name => $ids) {
+    $field = field_info_field($field_name);
+    $field_data = $data[$field['id']];
+    $sub_table = $load_current ? 'current' : 'revisions';
+    $rows = $load_current ? $data[$field['id']]['current'] : $data[$field['id']]['revisions'];
+    $delta_count = array();
+    foreach ($field_data[$sub_table] as $row) {
+      if ($row['type'] == $obj_type && $row['deleted'] == FALSE) {
+        if (($load_current && in_array($row['entity_id'], $ids)) || (!$load_current && in_array($row['revision_id'], $ids))) {
+          if (!isset($delta_count[$row['entity_id']])) {
+            $delta_count[$row['entity_id']] = 0;
+          }
+          if ($field['cardinality'] == FIELD_CARDINALITY_UNLIMITED || $delta_count[$row['entity_id']] < $field['cardinality']) {
+            $item = array();
+            foreach ($field['columns'] as $column => $attributes) {
+              $item[$column] = $row[$column];
+            }
+            $objects[$row['entity_id']]->{$field_name}[] = $item;
+            $delta_count[$row['entity_id']]++;
+          }
+        }
+      }
+    }
+  }
+}
+
+/**
+ * Implement hook_field_storage_write().
+ */
+function field_test_field_storage_write($obj_type, $object, $op, $skip_fields) {
+  $data = _field_test_storage_data();
+
+  list($id, $vid, $bundle) = field_attach_extract_ids($obj_type, $object);
+
+  $instances = field_info_instances($bundle);
+  foreach ($instances as $field_name => $instance) {
+    $field = field_info_field($field_name);
+    // TODO : $skip_fields[$field_id] after http://drupal.org/node/367753 get in.
+    if ($field['storage']['type'] == 'field_test_storage' && !isset($skip_fields[$field_name])) {
+      // 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)) {
+        $field_data = &$data[$field['id']];
+
+        // Delete and insert, rather than update, in case a value was added.
+        if ($op == FIELD_STORAGE_UPDATE) {
+          foreach ($field_data['current'] as $key => $row) {
+            if ($row['type'] == $obj_type && $row['entity_id'] == $id) {
+              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]);
+              }
+            }
+          }
+        }
+
+        if ($object->$field_name) {
+          $delta_count = 0;
+          foreach ($object->$field_name as $delta => $item) {
+            $row = array(
+              'field_id' => $field['id'],
+              'type' => $obj_type,
+              'entity_id' => $id,
+              'revision_id' => $vid,
+              'bundle' => $bundle,
+              'delta' => $delta,
+              'deleted' => FALSE,
+            );
+            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) {
+  $data = _field_test_storage_data();
+
+  list($id, $vid, $bundle) = field_attach_extract_ids($obj_type, $object);
+  $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 $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) {
+  $data = _field_test_storage_data();
+
+  list($id, $vid, $bundle) = field_attach_extract_ids($obj_type, $object);
+  $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 $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_name, $conditions, $count, &$cursor = NULL, $age) {
+  $data = _field_test_storage_data();
+
+  $load_current = $age == FIELD_LOAD_CURRENT;
+
+  $field = field_info_field($field_name);
+  $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'] && $row['deleted'] == FALSE) {
+      $match = TRUE;
+      // 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 &= in_array($row[$column], $value);
+            break;
+          case 'NOT IN':
+            $match &= !in_array($row[$column], $value);
+            break;
+          case 'BETWEEN':
+            $match &= $row[$column] >= $value[0] && $row[$column] <= $value[1];
+            break;
+          case 'STARTS_WITH':
+          case 'ENDS_WITH':
+          case 'CONTAINS':
+            // Not supported.
+            $match &= FALSE;
+            break;
+        }
+      }
+      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_attach_create_stub_object($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) {
+  $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_name) {
+  $data = _field_test_storage_data();
+
+  $field = field_info_field($field_name);
+  $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($field_name, $bundle) {
+  $data = _field_test_storage_data();
+
+  $field = field_info_field($field_name);
+  $field_data = &$data[$field['id']];
+  foreach (array('current', 'revisions') as $sub_table) {
+    foreach ($field_data[$sub_table] as &$row) {
+      if ($row['bundle'] == $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();
+
+  $instances = field_info_instances($bundle_new);
+  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['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