first import

sites/all/modules/migrate/plugins/sources/csv.inc

@@ -0,0 +1,208 @@
+<?php
+
+/**
+ * @file
+ * Define a MigrateSource for importing from comma separated values files.
+ */
+
+/**
+ * Implementation of MigrateSource, to handle imports from CSV files.
+ *
+ * If the CSV file contains non-ASCII characters, make sure it includes a
+ * UTF BOM (Byte Order Marker) so they are interpreted correctly.
+ */
+class MigrateSourceCSV extends MigrateSource {
+  /**
+   * List of available source fields.
+   *
+   * @var array
+   */
+  protected $fields = array();
+
+  /**
+   * Parameters for the fgetcsv() call.
+   *
+   * @var array
+   */
+  protected $fgetcsv = array();
+
+  /**
+   * File handle for the CSV file being iterated.
+   *
+   * @var resource
+   */
+  protected $csvHandle = NULL;
+
+  /**
+   * The number of rows in the CSV file before the data starts.
+   *
+   * @var integer
+   */
+  protected $headerRows = 0;
+
+  /**
+   * Simple initialization.
+   *
+   * @param string $path
+   *  The path to the source file
+   * @param array $csvcolumns
+   *  Keys are integers. values are array(field name, description).
+   * @param array $options
+   *  Options applied to this source.
+   * @param array $fields
+   *  Optional - keys are field names, values are descriptions. Use to override
+   *  the default descriptions, or to add additional source fields which the
+   *  migration will add via other means (e.g., prepareRow()).
+   */
+  public function __construct($path, array $csvcolumns = array(), array $options = array(), array $fields = array()) {
+    parent::__construct($options);
+    $this->file = $path;
+    if (!empty($options['header_rows'])) {
+      $this->headerRows = $options['header_rows'];
+    }
+    else {
+      $this->headerRows = 0;
+    }
+    $this->options = $options;
+    $this->fields = $fields;
+    // fgetcsv specific options
+    foreach (array('length' => NULL, 'delimiter' => ',', 'enclosure' => '"', 'escape' => '\\') as $key => $default) {
+      $this->fgetcsv[$key] = isset($options[$key]) ? $options[$key] : $default;
+    }
+    // One can either pass in an explicit list of column names to use, or if we have
+    // a header row we can use the names from that
+    if ($this->headerRows && empty($csvcolumns)) {
+      $this->csvcolumns = array();
+      $this->csvHandle = fopen($this->file, 'r');
+      // Skip all but the last header
+      for ($i = 0; $i < $this->headerRows - 1; $i++) {
+        $this->getNextLine();
+      }
+
+      $row = $this->getNextLine();
+      foreach ($row as $header) {
+        $header = trim($header);
+        $this->csvcolumns[] = array($header, $header);
+      }
+      fclose($this->csvHandle);
+      unset($this->csvHandle);
+    }
+    else {
+      $this->csvcolumns = $csvcolumns;
+    }
+  }
+
+  /**
+   * Return a string representing the source query.
+   *
+   * @return string
+   */
+  public function __toString() {
+    return $this->file;
+  }
+
+  /**
+   * Returns a list of fields available to be mapped from the source query.
+   *
+   * @return array
+   *  Keys: machine names of the fields (to be passed to addFieldMapping)
+   *  Values: Human-friendly descriptions of the fields.
+   */
+  public function fields() {
+    $fields = array();
+    foreach ($this->csvcolumns as $values) {
+      $fields[$values[0]] = $values[1];
+    }
+
+    // Any caller-specified fields with the same names as extracted fields will
+    // override them; any others will be added
+    if ($this->fields) {
+      $fields = $this->fields + $fields;
+    }
+
+    return $fields;
+  }
+
+  /**
+   * Return a count of all available source records.
+   */
+  public function computeCount() {
+    // If the data may have embedded newlines, the file line count won't reflect
+    // the number of CSV records (one record will span multiple lines). We need
+    // to scan with fgetcsv to get the true count.
+    if (!empty($this->options['embedded_newlines'])) {
+      $result = fopen($this->file, 'r');
+      // Skip all but the last header
+      for ($i = 0; $i < $this->headerRows; $i++) {
+        fgets($result);
+      }
+      $count = 0;
+      while ($this->getNextLine()) {
+        $count++;
+      }
+      fclose($result);
+    }
+    else {
+      // TODO. If this takes too much time/memory, use exec('wc -l')
+      $count = count(file($this->file));
+      $count -= $this->headerRows;
+    }
+    return $count;
+  }
+
+  /**
+   * Implementation of MigrateSource::performRewind().
+   *
+   * @return void
+   */
+  public function performRewind() {
+    // Close any previously-opened handle
+    if (!is_null($this->csvHandle)) {
+      fclose($this->csvHandle);
+    }
+    // Load up the first row, skipping the header(s) if necessary
+    $this->csvHandle = fopen($this->file, 'r');
+    for ($i = 0; $i < $this->headerRows; $i++) {
+      $this->getNextLine();
+    }
+  }
+
+  /**
+   * Implementation of MigrateSource::getNextRow().
+   * Return the next line of the source CSV file as an object.
+   *
+   * @return null|object
+   */
+  public function getNextRow() {
+    $row = $this->getNextLine();
+    if ($row) {
+      // Set meaningful keys for the columns mentioned in $this->csvcolumns().
+      foreach ($this->csvcolumns as $int => $values) {
+        list($key, $description) = $values;
+        // Copy value to more descriptive string based key and then unset original.
+        $row[$key] = isset($row[$int]) ? $row[$int] : NULL;
+        unset($row[$int]);
+      }
+      return (object)$row;
+    }
+    else {
+      fclose($this->csvHandle);
+      $this->csvHandle = NULL;
+      return NULL;
+    }
+  }
+
+  protected function getNextLine() {
+    // escape parameter was added in PHP 5.3.
+    if (version_compare(phpversion(), '5.3', '<')) {
+      $row = fgetcsv($this->csvHandle, $this->fgetcsv['length'],
+        $this->fgetcsv['delimiter'], $this->fgetcsv['enclosure']);
+    }
+    else {
+      $row = fgetcsv($this->csvHandle, $this->fgetcsv['length'],
+        $this->fgetcsv['delimiter'], $this->fgetcsv['enclosure'],
+        $this->fgetcsv['escape']);
+    }
+    return $row;
+  }
+}

sites/all/modules/migrate/plugins/sources/files.inc

@@ -0,0 +1,175 @@
+<?php
+
+/**
+ * @file
+ * Support for migration from files sources.
+ */
+
+/**
+ * Implementation of MigrateList, for retrieving a list of IDs to be migrated
+ * from a directory listing. Each item is a file, it's ID is the path.
+ */
+class MigrateListFiles extends MigrateList {
+
+  protected $listDirs;
+  protected $baseDir;
+  protected $fileMask;
+  protected $directoryOptions;
+
+  /**
+   * Constructor.
+   *
+   * @param $list_dirs
+   *   Array of directory paths that will be scanned for files. No trailing
+   *   slash. For example:
+   *     array(
+   *       '/var/html_source/en/news',
+   *       '/var/html_source/fr/news',
+   *       '/var/html_source/zh/news',
+   *     );
+   * @param $base_dir
+   *   The base dir is the part of the path that will be excluded when making
+   *   an ID for each file. To continue the example from above, you want base_dir
+   *   to be = '/var/html_source', so that the files will have IDs in the format
+   *   '/en/news/news_2011_03_4.html'.
+   * @param $file_mask
+   *   Passed on and used to filter for certain types of files. Use a regular
+   *   expression, for example '/(.*\.htm$|.*\.html$)/i' to match all .htm and
+   *   .html files, case insensitive.
+   * @param $options
+   *   Options that will be passed on to file_scan_directory(). See docs of that
+   *   core Drupal function for more information.
+   */
+  public function __construct($list_dirs, $base_dir, $file_mask = NULL, $options = array()) {
+    parent::__construct();
+    $this->listDirs = $list_dirs;
+    $this->baseDir = $base_dir;
+    $this->fileMask = $file_mask;
+    $this->directoryOptions = $options;
+  }
+
+  /**
+   * Our public face is the directories we're getting items from.
+   */
+  public function __toString() {
+    if (is_array($this->listDirs)) {
+      return implode(',', $this->listDirs);
+    }
+    else {
+      return $this->listDirs;
+    }
+  }
+
+  /**
+   * Retrieve a list of files based on parameters passed for the migration.
+   */
+  public function getIdList() {
+    $files = array();
+    foreach ($this->listDirs as $dir) {
+      migrate_instrument_start("Retrieve $dir");
+      $files = array_merge(file_scan_directory($dir, $this->fileMask, $this->directoryOptions), $files);
+      migrate_instrument_stop("Retrieve $dir");
+    }
+
+    if (isset($files)) {
+      return $this->getIDsFromFiles($files);
+    }
+    Migration::displayMessage(t('Loading of !listuri failed:', array('!listuri' => $this->listUri)));
+    return NULL;
+  }
+
+  /**
+   * Given an array generated from file_scan_directory(), parse out the IDs for
+   * processing and return them as an array.
+   */
+  protected function getIDsFromFiles(array $files) {
+    $ids = array();
+    foreach ($files as $file) {
+      $ids[] = str_replace($this->baseDir, '', (string) $file->uri);
+    }
+    return array_unique($ids);
+  }
+
+  /**
+   * Return a count of all available IDs from the source listing.
+   */
+  public function computeCount() {
+    $count = 0;
+    $files = $this->getIdList();
+    if ($files) {
+      $count = count($files);
+    }
+    return $count;
+  }
+
+}
+
+/**
+ * Implementation of MigrateItem, for retrieving a file from the file system
+ * based on source directory and an ID provided by a MigrateList class.
+ */
+class MigrateItemFile extends MigrateItem {
+
+  protected $baseDir;
+  protected $getContents;
+
+  /**
+   * Constructor.
+   *
+   * @param $base_dir
+   *   The base directory from which all file paths are calculated.
+   * @param $get_contents
+   *   TRUE if we should try load the contents of each file (in the case
+   *   of a text file), or FALSE if we just want to confirm it exists (binary).
+   */
+  public function __construct($base_dir, $get_contents = TRUE) {
+    parent::__construct();
+    $this->baseDir = $base_dir;
+    $this->getContents = $get_contents;
+  }
+
+  /**
+   * Return an object representing a file.
+   *
+   * @param $id
+   *   The file id, which is the file URI.
+   *
+   * @return object
+   *   The item object for migration.
+   */
+  public function getItem($id) {
+    $item_uri = $this->baseDir . $id;
+    // Get the file data at the specified URI
+    $data = $this->loadFile($item_uri);
+    if (is_string($data)) {
+      $return = new stdClass;
+      $return->filedata = $data;
+      return $return;
+    }
+    elseif ($data === TRUE) {
+      $return = new stdClass;
+      return $return;
+    }
+    else {
+      $migration = Migration::currentMigration();
+      $message = t('Loading of !objecturi failed:', array('!objecturi' => $item_uri));
+      $migration->getMap()->saveMessage(
+              array($id), $message, MigrationBase::MESSAGE_ERROR);
+      return NULL;
+    }
+  }
+
+  /**
+   * Default file loader.
+   */
+  protected function loadFile($item_uri) {
+    // Only try load the contents if we have this flag set.
+    if ($this->getContents) {
+      $data = file_get_contents($item_uri);
+    }
+    else {
+      $data = file_exists($item_uri);
+    }
+    return $data;
+  }
+}

sites/all/modules/migrate/plugins/sources/json.inc

@@ -0,0 +1,172 @@
+<?php
+
+/**
+ * @file
+ * Support for migration from JSON sources.
+ */
+
+/**
+ * Implementation of MigrateList, for retrieving a list of IDs to be migrated
+ * from a JSON object.
+ */
+class MigrateListJSON extends MigrateList {
+  /**
+   * A URL pointing to an JSON object containing a list of IDs to be processed.
+   *
+   * @var string
+   */
+  protected $listUrl;
+
+  protected $httpOptions;
+
+  public function __construct($list_url, $http_options = array()) {
+    parent::__construct();
+    $this->listUrl = $list_url;
+    $this->httpOptions = $http_options;
+  }
+
+  /**
+   * Our public face is the URL we're getting items from
+   *
+   * @return string
+   */
+  public function __toString() {
+    return $this->listUrl;
+  }
+
+  /**
+   * Load the JSON at the given URL, and return an array of the IDs found within it.
+   *
+   * @return array
+   */
+  public function getIdList() {
+    migrate_instrument_start("Retrieve $this->listUrl");
+    if (empty($this->httpOptions)) {
+      $json = file_get_contents($this->listUrl);
+    }
+    else {
+      $response = drupal_http_request($this->listUrl, $this->httpOptions);
+      $json = $response->data;
+    }
+    migrate_instrument_stop("Retrieve $this->listUrl");
+    if ($json) {
+      $data = drupal_json_decode($json);
+      if ($data) {
+        return $this->getIDsFromJSON($data);
+      }
+    }
+    Migration::displayMessage(t('Loading of !listurl failed:',
+        array('!listurl' => $this->listUrl)));
+    return NULL;
+  }
+
+  /**
+   * Given an array generated from JSON, parse out the IDs for processing
+   * and return them as an array. The default implementation assumes the IDs are
+   * simply the values of the top-level elements - in most cases, you will need
+   * to override this to reflect your particular JSON structure.
+   *
+   * @param array $data
+   *
+   * @return array
+   */
+  protected function getIDsFromJSON(array $data) {
+    return $data;
+  }
+
+  /**
+   * Return a count of all available IDs from the source listing. The default
+   * implementation assumes the count of top-level elements reflects the number
+   * of IDs available - in many cases, you will need to override this to reflect
+   * your particular JSON structure.
+   */
+  public function computeCount() {
+    $count = 0;
+    if (empty($this->httpOptions)) {
+      $json = file_get_contents($this->listUrl);
+    }
+    else {
+      $response = drupal_http_request($this->listUrl, $this->httpOptions);
+      $json = $response->data;
+    }
+    if ($json) {
+      $data = drupal_json_decode($json);
+      if ($data) {
+        $count = count($data);
+      }
+    }
+    return $count;
+  }
+}
+
+/**
+ * Implementation of MigrateItem, for retrieving a parsed JSON object given
+ * an ID provided by a MigrateList class.
+ */
+class MigrateItemJSON extends MigrateItem {
+  /**
+   * A URL pointing to a JSON object containing the data for one item to be
+   * migrated.
+   *
+   * @var string
+   */
+  protected $itemUrl;
+
+  protected $httpOptions;
+
+  public function __construct($item_url, $http_options) {
+    parent::__construct();
+    $this->itemUrl = $item_url;
+    $this->httpOptions = $http_options;
+  }
+
+  /**
+   * Implementors are expected to return an object representing a source item.
+   *
+   * @param mixed $id
+   *
+   * @return stdClass
+   */
+  public function getItem($id) {
+    $item_url = $this->constructItemUrl($id);
+    // Get the JSON object at the specified URL
+    $json = $this->loadJSONUrl($item_url);
+    if ($json) {
+      return $json;
+    }
+    else {
+      $migration = Migration::currentMigration();
+      $message =  t('Loading of !objecturl failed:', array('!objecturl' => $item_url));
+      $migration->getMap()->saveMessage(
+        array($id), $message, MigrationBase::MESSAGE_ERROR);
+      return NULL;
+    }
+  }
+
+  /**
+   * The default implementation simply replaces the :id token in the URL with
+   * the ID obtained from MigrateListJSON. Override if the item URL is not
+   * so easily expressed from the ID.
+   *
+   * @param mixed $id
+   */
+  protected function constructItemUrl($id) {
+    return str_replace(':id', $id, $this->itemUrl);
+  }
+
+  /**
+   * Default JSON loader - just pull and decode. This can be overridden for
+   * preprocessing of JSON (removal of unwanted elements, caching of JSON if the
+   * source service is slow, etc.)
+   */
+  protected function loadJSONUrl($item_url) {
+    if (empty($this->httpOptions)) {
+      $json = file_get_contents($item_url);
+    }
+    else {
+      $response = drupal_http_request($item_url, $this->httpOptions);
+      $json = $response->data;
+    }
+    return json_decode($json);
+  }
+}

sites/all/modules/migrate/plugins/sources/list.inc

@@ -0,0 +1,195 @@
+<?php
+
+/**
+ * @file
+ * Support for migration from sources with distinct means of listing items to
+ * import and obtaining the items themselves.
+ *
+ * TODO: multiple-field source keys
+ */
+
+
+/**
+ * Extend the MigrateList class to provide a means to obtain a list of IDs to
+ * be migrated from a given source (e.g., MigrateListXML extends MigrateList to
+ * obtain a list of IDs from an XML document).
+ */
+abstract class MigrateList {
+  public function __construct() {}
+
+  /**
+   * Implementors are expected to return a string representing where the listing
+   * is obtained from (a URL, file directory, etc.)
+   *
+   * @return string
+   */
+  abstract public function __toString();
+
+  /**
+   * Implementors are expected to return an array of unique IDs, suitable for
+   * passing to the MigrateItem class to retrieve the data for a single item.
+   *
+   * @return Mixed, iterator or array
+   */
+  abstract public function getIdList();
+
+  /**
+   * Implementors are expected to return a count of IDs available to be migrated.
+   *
+   * @return int
+   */
+  abstract public function computeCount();
+}
+
+/**
+ * Extend the MigrateItem class to provide a means to obtain the data for a
+ * given migratable item given its ID as provided by the MigrateList class.
+ */
+abstract class MigrateItem {
+  public function __construct() {}
+
+  /**
+   * Implementors are expected to return an object representing a source item.
+   *
+   * @param mixed $id
+   *
+   * @return stdClass
+   */
+  abstract public function getItem($id);
+}
+
+/**
+ * Implementation of MigrateSource, providing the semantics of iterating over
+ * IDs provided by a MigrateList and retrieving data from a MigrateItem.
+ */
+class MigrateSourceList extends MigrateSource {
+  /**
+   * MigrateList object used to obtain ID lists.
+   *
+   * @var MigrateList
+   */
+  protected $listClass;
+
+  /**
+   * MigrateItem object used to obtain the source object for a given ID.
+   *
+   * @var MigrateItem
+   */
+  protected $itemClass;
+
+  /**
+   * Iterator of IDs from the listing class.
+   *
+   * @var Iterator
+   */
+  protected $idIterator;
+
+  /**
+   * List of available source fields.
+   *
+   * @var array
+   */
+  protected $fields = array();
+
+  /**
+   * Simple initialization.
+   */
+  public function __construct(MigrateList $list_class, MigrateItem $item_class, $fields = array(),
+      $options = array()) {
+    parent::__construct($options);
+    $this->listClass = $list_class;
+    $this->itemClass = $item_class;
+    $this->fields = $fields;
+  }
+
+  /**
+   * Return a string representing the source.
+   *
+   * @return string
+   */
+  public function __toString() {
+    return (string) $this->listClass;
+  }
+
+  /**
+   * Returns a list of fields available to be mapped from the source query.
+   * Since we can't reliably figure out what "fields" are in the source,
+   * it's up to the implementing Migration constructor to fill them in.
+   *
+   * @return array
+   *  Keys: machine names of the fields (to be passed to addFieldMapping)
+   *  Values: Human-friendly descriptions of the fields.
+   */
+  public function fields() {
+    return $this->fields;
+  }
+
+  /**
+   * It's the list class that knows how many records are available, so ask it.
+   *
+   * @return int
+   */
+  public function computeCount() {
+    // @API: Support old count method for now.
+    if (method_exists($this->listClass, 'computeCount')) {
+      return $this->listClass->computeCount();
+    }
+    else {
+      return $this->listClass->count();
+    }
+  }
+
+  /**
+   * Implementation of MigrateSource::performRewind().
+   *
+   * @return void
+   */
+  public function performRewind() {
+    // If there isn't a specific ID list passed in, get it from the list class.
+    if ($this->idList) {
+      $this->idsToProcess = $this->idList;
+    }
+    else {
+      $this->idsToProcess = $this->listClass->getIdList();
+    }
+    $this->idIterator = ($this->idsToProcess instanceof Iterator) ?
+      $this->idsToProcess : new ArrayIterator($this->idsToProcess);
+    $this->idIterator->rewind();
+  }
+
+  /**
+   * Implementation of MigrateSource::getNextRow().
+   *
+   * @return null|stdClass
+   */
+  public function getNextRow() {
+    $row = NULL;
+    while ($this->idIterator->valid()) {
+      $ids = $this->idIterator->current();
+      $this->idIterator->next();
+
+      // Skip empty IDs
+      if (empty($ids)) {
+        continue;
+      }
+
+      // Got a good ID, get the data and get out.
+      $row = $this->itemClass->getItem($ids);
+      if ($row) {
+        // No matter what $ids is, be it a string, integer, object, or array, we
+        // cast it to an array so that it can be properly mapped to the source
+        // keys as specified by the map. This is done after getItem is called so
+        // that the ItemClass doesn't have to care about this requirement.
+        $ids = (array) $ids;
+        foreach (array_keys($this->activeMap->getSourceKey()) as $key_name) {
+          // Grab the first id and advance the array cursor. Then save the ID
+          // using the map source key - it will be used for mapping.
+          list(, $id) = each($ids);
+          $row->$key_name = $id;
+        }
+      }
+      break;
+    }
+    return $row;
+  }
+}

sites/all/modules/migrate/plugins/sources/mssql.inc

@@ -0,0 +1,206 @@
+<?php
+
+/**
+ * @file
+ * Define a MigrateSource for importing from Microsoft SQL Server databases.
+ */
+
+/**
+ * Implementation of MigrateSource, to handle imports from remote MS SQL Server db servers.
+ */
+class MigrateSourceMSSQL extends MigrateSource {
+  /**
+   * Array containing information for connecting to SQL Server:
+   *  servername - Hostname of the SQL Server
+   *  username - Username to connect as
+   *  password - Password for logging in
+   *  database (optional) - Database to select after connecting
+   *
+   * @var array
+   */
+  protected $configuration;
+
+  /**
+   * The active MS SQL Server connection for this source.
+   *
+   * @var resource
+   */
+  protected $connection;
+
+  /**
+   * The SQL query from which to obtain data. Is a string.
+   */
+  protected $query;
+
+  /**
+   * The result object from executing the query - traversed to process the
+   * incoming data.
+   */
+  protected $result;
+
+  /**
+   * By default, mssql_query fetches all results - severe memory problems with
+   * big tables. So, we will fetch a batch at a time.
+   *
+   * @var int
+   */
+  protected $batchSize;
+
+  /**
+   * Return an options array for MS SQL sources.
+   *
+   * @param int $batch_size
+   *  Number of rows to pull at once (defaults to 500).
+   * @param boolean $cache_counts
+   *  Indicates whether to cache counts of source records.
+   */
+  static public function options($batch_size, $cache_counts) {
+    return compact('batch_size', 'cache_counts');
+  }
+
+  /**
+   * Simple initialization.
+   */
+  public function __construct(array $configuration, $query, $count_query,
+      array $fields, array $options = array()) {
+    parent::__construct($options);
+    $this->query = $query;
+    $this->countQuery = $count_query;
+    $this->configuration = $configuration;
+    $this->fields = $fields;
+    $this->batchSize = isset($options['batch_size']) ? $options['batch_size'] : 500;
+  }
+
+  /**
+   * Return a string representing the source query.
+   *
+   * @return string
+   */
+  public function __toString() {
+    return $this->query;
+  }
+
+  /**
+   * Connect lazily to the DB server.
+   */
+  protected function connect() {
+    if (!isset($this->connection)) {
+      if (!extension_loaded('mssql')) {
+        throw new Exception(t('You must configure the mssql extension in PHP.'));
+      }
+
+      if (isset($this->configuration['port'])) {
+        $host = $this->configuration['servername'] . ':' . $this->configuration['port'];
+      }
+      else {
+        $host = $this->configuration['servername'];
+      }
+      $this->connection = mssql_connect(
+        $host,
+        $this->configuration['username'],
+        $this->configuration['password'],
+        TRUE);
+      if (isset($this->configuration['database'])) {
+        return mssql_select_db($this->configuration['database'], $this->connection);
+      }
+    }
+  }
+
+  /**
+   * Returns a list of fields available to be mapped from the source query.
+   *
+   * @return array
+   *  Keys: machine names of the fields (to be passed to addFieldMapping)
+   *  Values: Human-friendly descriptions of the fields.
+   */
+  public function fields() {
+    // The fields are passed to the constructor for this plugin.
+    return $this->fields;
+  }
+
+  /**
+   * Return a count of all available source records.
+   */
+  public function computeCount() {
+    migrate_instrument_start('MigrateSourceMSSQL count');
+    if ($this->connect()) {
+      $result = mssql_query($this->countQuery);
+      $count = reset(mssql_fetch_object($result));
+    }
+    else {
+      // Do something else?
+      $count = FALSE;
+    }
+    migrate_instrument_stop('MigrateSourceMSSQL count');
+    return $count;
+  }
+
+  /**
+   * Implementation of MigrateSource::performRewind().
+   */
+  public function performRewind() {
+    /*
+     * Replace :criteria placeholder with idlist or highwater clauses. We
+     * considered supporting both but it is not worth the complexity. Run twice
+     * instead.
+     */
+    if (!empty($this->idList)) {
+      $keys = array();
+      foreach ($this->activeMap->getSourceKey() as $field_name => $field_schema) {
+        // Allow caller to provide an alias to table containing the primary key.
+        if (!empty($field_schema['alias'])) {
+          $field_name = $field_schema['alias'] . '.' . $field_name;
+        }
+        $keys[] = $field_name;
+      }
+
+      // TODO: Sanitize. not critical as this is admin supplied data in drush.
+      $this->query = str_replace(':criteria',
+        $keys[0] . ' IN (' . implode(',', $this->idList) . ')', $this->query);
+    }
+    else {
+      if (isset($this->highwaterField['name']) && $highwater = $this->activeMigration->getHighwater()) {
+        if (empty($this->highwaterField['alias'])) {
+          $highwater_name = $this->highwaterField['name'];
+        }
+        else {
+          $highwater_name = $this->highwaterField['alias'] . '.' . $this->highwaterField['name'];
+        }
+        $this->query = str_replace(':criteria', "$highwater_name > '$highwater'", $this->query);
+      }
+      else {
+        // No idlist or highwater. Replace :criteria placeholder with harmless WHERE
+        // clause instead of empty since we don't know if an AND follows.
+        $this->query = str_replace(':criteria', '1=1', $this->query);
+      }
+    }
+
+    migrate_instrument_start('mssql_query');
+    $this->connect();
+    $this->result = mssql_query($this->query, $this->connection, $this->batchSize);
+    migrate_instrument_stop('mssql_query');
+  }
+
+  /**
+   * Implementation of MigrateSource::getNextRow().
+   *
+   * Returns the next row of the result set as an object, dealing with the
+   * difference between the end of the batch and the end of all data.
+   */
+  public function getNextRow() {
+    $row = mssql_fetch_object($this->result);
+
+    // Might be totally out of data, or just out of this batch - request another
+    // batch and see
+    if (!is_object($row)) {
+      mssql_fetch_batch($this->result);
+      $row = mssql_fetch_object($this->result);
+    }
+    if (is_object($row)) {
+      return $row;
+    }
+    else {
+      return NULL;
+    }
+  }
+}

sites/all/modules/migrate/plugins/sources/multiitems.inc

@@ -0,0 +1,186 @@
+<?php
+
+/**
+ * @file
+ * Support for migration from sources where data spans multiple lines
+ * (ex. xml, json) and IDs for the items are part of each item and multiple
+ * items reside in a single file.
+ */
+
+/**
+ * Extend the MigrateItems class to provide a means to obtain a list of IDs to
+ * be migrated from a given source (e.g., MigrateItemsXML extends MigrateItem to
+ * obtain a list of IDs from an XML document).  This class also provides a means
+ * to obtain the data for a given migratable item given its ID.
+ */
+abstract class MigrateItems {
+  public function __construct() {}
+
+  /**
+   * Implementors are expected to return a string representing where the listing
+   * is obtained from (a URL, file directory, etc.)
+   *
+   * @return string
+   */
+  abstract public function __toString();
+
+  /**
+   * Implementors are expected to return an array of unique IDs, suitable for
+   * passing to the MigrateItem class to retrieve the data for a single item.
+   *
+   * @return Mixed, iterator or array
+   */
+  abstract public function getIdList();
+
+  /**
+   * Implementors are expected to return a count of IDs available to be migrated.
+   *
+   * @return int
+   */
+  abstract public function computeCount();
+
+  /**
+   * Implementors are expected to return an object representing a source item.
+   *
+   * @param mixed $id
+   *
+   * @return stdClass
+   */
+  abstract public function getItem($id);
+}
+
+
+/**
+ * Implementation of MigrateItems, for providing a list of IDs and for
+ * retrieving a parsed XML document given an ID from this list.
+ */
+
+/**
+ * Implementation of MigrateSource, providing the semantics of iterating over
+ * IDs provided by a MigrateItems and retrieving data from a MigrateItems.
+ */
+class MigrateSourceMultiItems extends MigrateSource {
+  /**
+   * MigrateItems object used to obtain the list of IDs and source for
+   * all objects.
+   *
+   * @var MigrateItems
+   */
+  protected $itemsClass;
+
+  /**
+   * List of available source fields.
+   *
+   * @var array
+   */
+  protected $fields = array();
+
+  /**
+   * Iterator of IDs from the listing class.
+   *
+   * @var Iterator
+   */
+  protected $idIterator;
+
+  /**
+   * List of item IDs to iterate.
+   *
+   * @var array
+   */
+  protected $idsToProcess = array();
+
+  /**
+   * Simple initialization.
+   */
+  public function __construct(MigrateItems $items_class, $fields = array(), $options = array()) {
+    parent::__construct($options);
+
+    $this->itemsClass = $items_class;
+    $this->fields = $fields;
+  }
+
+  /**
+   * Return a string representing the source.
+   *
+   * @return string
+   */
+  public function __toString() {
+    return (string) $this->itemsClass;
+  }
+
+  /**
+   * Returns a list of fields available to be mapped from the source query.
+   * Since we can't reliably figure out what "fields" are in the source,
+   * it's up to the implementing Migration constructor to fill them in.
+   *
+   * @return array
+   *  Keys: machine names of the fields (to be passed to addFieldMapping)
+   *  Values: Human-friendly descriptions of the fields.
+   */
+  public function fields() {
+    return $this->fields;
+  }
+
+  /**
+   * It's the list class that knows how many records are available, so ask it.
+   *
+   * @return int
+   */
+  public function computeCount() {
+      // @API: Support old count method for now.
+    if (method_exists($this->itemsClass, 'computeCount')) {
+      return $this->itemsClass->computeCount();
+    }
+    else {
+      return $this->itemsClass->count();
+    }
+  }
+
+  /**
+   * Implementation of MigrateSource::performRewind().
+   *
+   * @return void
+   */
+  public function performRewind() {
+    // If there isn't a specific ID list passed in, get it from the list class.
+    if ($this->idList) {
+      $this->idsToProcess = $this->idList;
+    }
+    else {
+      $this->idsToProcess = $this->itemsClass->getIdList();
+    }
+    $this->idIterator = ($this->idsToProcess instanceof Iterator) ?
+      $this->idsToProcess : new ArrayIterator($this->idsToProcess);
+    $this->idIterator->rewind();
+  }
+
+  /**
+   * Implementation of MigrateSource::getNextRow().
+   *
+   * @return null|stdClass
+   */
+  public function getNextRow() {
+    $row = NULL;
+    while ($this->idIterator->valid()) {
+      $id = $this->idIterator->current();
+      $this->idIterator->next();
+
+      // Skip empty IDs
+      if (empty($id)) {
+        continue;
+      }
+
+      // Got a good ID, get the data and get out.
+      $row = $this->itemsClass->getItem($id);
+      if ($row) {
+        // Save the ID using the map source key - it will be used for mapping
+        $sourceKey = $this->activeMap->getSourceKey();
+        $key_name = key($sourceKey);
+        $row->$key_name = $id;
+      }
+      break;
+    }
+    return $row;
+  }
+}
+

sites/all/modules/migrate/plugins/sources/oracle.inc

@@ -0,0 +1,221 @@
+<?php
+
+/**
+ * @file
+ * Define a MigrateSource class for importing from Oracle databases.
+ */
+
+/**
+ * Implementation of MigrateSource, to handle imports from remote Oracle servers.
+ */
+class MigrateSourceOracle extends MigrateSource {
+  /**
+   * Array containing information for connecting to Oracle:
+   *  username - Username to connect as
+   *  password - Password for logging in
+   *  connection_string - See http://us.php.net/manual/en/function.oci-connect.php.
+   *
+   * @var array
+   */
+  protected $configuration;
+
+  /**
+   * The active Oracle connection for this source.
+   *
+   * @var resource
+   */
+  protected $connection;
+  public function getConnection() {
+    return $this->connection;
+  }
+
+  /**
+   * The SQL query from which to obtain data. Is a string.
+   */
+  protected $query;
+
+  /**
+   * The result object from executing the query - traversed to process the
+   * incoming data.
+   */
+  protected $result;
+
+  /**
+   * Character set to use in retrieving data.
+   *
+   * @var string
+   */
+  protected $characterSet;
+
+  /**
+   * Return an options array for Oracle sources.
+   *
+   * @param string $character_set
+   *  Character set to use in retrieving data. Defaults to 'UTF8'.
+   * @param boolean $cache_counts
+   *  Indicates whether to cache counts of source records.
+   */
+  static public function options($character_set = 'UTF8', $cache_counts = FALSE) {
+    return compact('character_set', 'cache_counts');
+  }
+
+  /**
+   * Simple initialization.
+   */
+  public function __construct(array $configuration, $query, $count_query,
+      array $fields, array $options = array()) {
+    parent::__construct($options);
+    $this->query = $query;
+    $this->countQuery = $count_query;
+    $this->configuration = $configuration;
+    $this->fields = $fields;
+    if (empty($options['character_set'])) {
+      $this->characterSet = 'UTF8';
+    }
+    else {
+      $this->characterSet = $options['character_set'];
+    }
+  }
+
+  /**
+   * Return a string representing the source query.
+   *
+   * @return string
+   */
+  public function __toString() {
+    return $this->query;
+  }
+
+  /**
+   * Connect lazily to the DB server.
+   */
+  protected function connect() {
+    if (!isset($this->connection)) {
+      if (!extension_loaded('oci8')) {
+        throw new Exception(t('You must configure the oci8 extension in PHP.'));
+      }
+      $this->connection = oci_connect($this->configuration['username'],
+        $this->configuration['password'], $this->configuration['connection_string'],
+        $this->characterSet);
+    }
+    if ($this->connection) {
+      return TRUE;
+    }
+    else {
+      $e = oci_error();
+      throw new Exception($e['message']);
+      return FALSE;
+    }
+  }
+
+  /**
+   * Returns a list of fields available to be mapped from the source query.
+   *
+   * @return array
+   *  Keys: machine names of the fields (to be passed to addFieldMapping)
+   *  Values: Human-friendly descriptions of the fields.
+   */
+  public function fields() {
+    // The fields are passed to the constructor for this plugin.
+    return $this->fields;
+  }
+
+  /**
+   * Return a count of all available source records.
+   */
+  public function computeCount() {
+    migrate_instrument_start('MigrateSourceOracle count');
+    if ($this->connect()) {
+      $statement = oci_parse($this->connection, $this->countQuery);
+      if (!$statement) {
+        $e = oci_error($this->connection);
+        throw new Exception($e['message'] . "\n" . $e['sqltext']);
+      }
+      $result = oci_execute($statement);
+      if (!$result) {
+        $e = oci_error($statement);
+        throw new Exception($e['message'] . "\n" . $e['sqltext']);
+      }
+      $count_array = oci_fetch_array($statement);
+      $count = reset($count_array);
+    }
+    else {
+      // Do something else?
+      $count = FALSE;
+    }
+    migrate_instrument_stop('MigrateSourceOracle count');
+    return $count;
+  }
+
+  /**
+   * Implementation of MigrateSource::performRewind().
+   */
+  public function performRewind() {
+    $keys = array();
+    foreach ($this->activeMap->getSourceKey() as $field_name => $field_schema) {
+      // Allow caller to provide an alias to table containing the primary key.
+      if (!empty($field_schema['alias'])) {
+        $field_name = $field_schema['alias'] . '.' . $field_name;
+      }
+      $keys[] = $field_name;
+    }
+
+    /*
+     * Replace :criteria placeholder with idlist or highwater clauses. We
+     * considered supporting both but it is not worth the complexity. Run twice
+     * instead.
+     */
+    if (!empty($this->idList)) {
+      // TODO: Sanitize. not critical as this is admin supplied data in drush.
+      $this->query = str_replace(':criteria',
+        $keys[0] . ' IN (' . implode(',', $this->idList) . ')', $this->query);
+    }
+    else {
+      if (isset($this->highwaterField['name']) && $highwater = $this->activeMigration->getHighwater()) {
+        if (empty($this->highwaterField['alias'])) {
+          $highwater_name = $this->highwaterField['name'];
+        }
+        else {
+          $highwater_name = $this->highwaterField['alias'] . '.' . $this->highwaterField['name'];
+        }
+        $this->query = str_replace(':criteria', "$highwater_name > '$highwater'", $this->query);
+      }
+      else {
+        // No idlist or highwater. Replace :criteria placeholder with harmless WHERE
+        // clause instead of empty since we don't know if an AND follows.
+        $this->query = str_replace(':criteria', '1=1', $this->query);
+      }
+    }
+
+    migrate_instrument_start('oracle_query');
+    $this->connect();
+    $this->result = oci_parse($this->connection, $this->query);
+    if (!$this->result) {
+      $e = oci_error($this->connection);
+      throw new Exception($e['message'] . "\n" . $e['sqltext']);
+    }
+    $status = oci_execute($this->result);
+    if (!$status) {
+      $e = oci_error($this->result);
+      throw new Exception($e['message'] . "\n" . $e['sqltext']);
+    }
+    migrate_instrument_stop('oracle_query');
+  }
+
+  /**
+   * Implementation of MigrateSource::getNextRow().
+   *
+   * Returns the next row of the result set as an object, making sure NULLs are
+   * represented as PHP NULLs and that LOBs are returned directly without special
+   * handling.
+   */
+  public function getNextRow() {
+    $row = oci_fetch_array($this->result, OCI_ASSOC | OCI_RETURN_NULLS | OCI_RETURN_LOBS);
+    if (!empty($row)) {
+      return (object)$row;
+    }
+    else {
+      return FALSE;
+    }
+  }
+}

sites/all/modules/migrate/plugins/sources/sql.inc

@@ -0,0 +1,337 @@
+<?php
+
+/**
+ * @file
+ * Define a MigrateSource for importing from Drupal connections
+ */
+
+/**
+ * Implementation of MigrateSource, to handle imports from Drupal connections.
+ */
+class MigrateSourceSQL extends MigrateSource {
+  /**
+   * The SQL query objects from which to obtain data, and counts of data
+   *
+   * @var SelectQueryInterface
+   */
+  protected $originalQuery, $query, $countQuery;
+
+  /**
+   * The result object from executing the query - traversed to process the
+   * incoming data.
+   *
+   * @var DatabaseStatementInterface
+   */
+  protected $result;
+
+  /**
+   * Number of eligible rows processed so far (used for itemlimit checking)
+   *
+   * @var int
+   */
+  protected $numProcessed = 0;
+
+  /**
+   * List of available source fields.
+   *
+   * @var array
+   */
+  protected $fields = array();
+
+  /**
+   * If the map is a MigrateSQLMap, and the table is compatible with the
+   * source query, we can join directly to the map and make things much faster
+   * and simpler.
+   *
+   * @var boolean
+   */
+  protected $mapJoinable = FALSE;
+  // Dynamically set whether the map is joinable - not really for production use,
+  // this is primarily to support simpletests
+  public function setMapJoinable($map_joinable) {
+    $this->mapJoinable = $map_joinable;
+  }
+
+  /**
+   * Whether this source is configured to use a highwater mark, and there is
+   * a highwater mark present to use.
+   *
+   * @var boolean
+   */
+  protected $usingHighwater = FALSE;
+
+  /**
+   * Whether, in the current iteration, we have reached the highwater mark.
+   *
+   * @var boolen
+   */
+  protected $highwaterSeen = FALSE;
+
+  /**
+   * Return an options array for PDO sources.
+   *
+   * @param boolean $map_joinable
+   *  Indicates whether the map table can be joined directly to the source query.
+   * @param boolean $cache_counts
+   *  Indicates whether to cache counts of source records.
+   */
+  static public function options($map_joinable, $cache_counts) {
+    return compact('map_joinable', 'cache_counts');
+  }
+
+  /**
+   * Simple initialization.
+   *
+   * @param SelectQueryInterface $query
+   *  The query we are iterating over.
+   * @param array $fields
+   *  Optional - keys are field names, values are descriptions. Use to override
+   *  the default descriptions, or to add additional source fields which the
+   *  migration will add via other means (e.g., prepareRow()).
+   * @param SelectQueryInterface $count_query
+   *  Optional - an explicit count query, primarily used when counting the
+   *  primary query is slow.
+   * @param boolean $options
+   *  Options applied to this source.
+   */
+  public function __construct(SelectQueryInterface $query, array $fields = array(),
+      SelectQueryInterface $count_query = NULL, array $options = array()) {
+    parent::__construct($options);
+    $this->originalQuery = $query;
+    $this->query = clone $query;
+    $this->fields = $fields;
+    if (is_null($count_query)) {
+      $this->countQuery = clone $query->countQuery();
+    }
+    else {
+      $this->countQuery = $count_query;
+    }
+
+    if (isset($options['map_joinable'])) {
+      $this->mapJoinable = $options['map_joinable'];
+    }
+    else {
+      // TODO: We want to automatically determine if the map table can be joined
+      // directly to the query, but this won't work unless/until
+      // http://drupal.org/node/802514 is committed, assume joinable for now
+      $this->mapJoinable = TRUE;
+/*      // To be able to join the map directly, it must be a PDO map on the same
+      // connection, or a compatible connection
+      $map = $migration->getMap();
+      if (is_a($map, 'MigrateSQLMap')) {
+        $map_options = $map->getConnection()->getConnectionOptions();
+        $query_options = $this->query->connection()->getConnectionOptions();
+
+        // Identical options means it will work
+        if ($map_options == $query_options) {
+          $this->mapJoinable = TRUE;
+        }
+        else {
+          // Otherwise, the one scenario we know will work is if it's MySQL and
+          // the credentials match (SQLite too?)
+          if ($map_options['driver'] == 'mysql' && $query_options['driver'] == 'mysql') {
+            if ($map_options['host'] == $query_options['host'] &&
+                $map_options['port'] == $query_options['port'] &&
+                $map_options['username'] == $query_options['username'] &&
+                $map_options['password'] == $query_options['password']) {
+              $this->mapJoinable = TRUE;
+            }
+          }
+        }
+      }*/
+    }
+  }
+
+
+  /**
+   * Return a string representing the source query.
+   *
+   * @return string
+   */
+  public function __toString() {
+    return (string) $this->query;
+  }
+
+  /**
+   * Returns a list of fields available to be mapped from the source query.
+   *
+   * @return array
+   *  Keys: machine names of the fields (to be passed to addFieldMapping)
+   *  Values: Human-friendly descriptions of the fields.
+   */
+  public function fields() {
+    $fields = array();
+    $queryFields = $this->query->getFields();
+
+    if ($queryFields) {
+      // Not much we can do in terms of describing the fields without manual intervention
+      foreach ($queryFields as $field_name => $field_info) {
+        // Lower case, because Drupal forces lowercase on fetch
+        $fields[drupal_strtolower($field_name)] = drupal_strtolower(
+          $field_info['table'] . '.' . $field_info['field']);
+      }
+    }
+    else {
+      // Detect available fields
+      $detection_query = clone $this->query;
+      $result = $detection_query->range(0, 1)->execute();
+      $row = $result->fetchAssoc();
+      if (is_array($row)) {
+        foreach ($row as $field_name => $field_value) {
+          // Lower case, because Drupal forces lowercase on fetch
+          $fields[drupal_strtolower($field_name)] = t('Example Content: !value',
+            array('!value' => $field_value));
+        }
+      }
+    }
+
+/*
+ * Handle queries without explicit field lists
+ * TODO: Waiting on http://drupal.org/node/814312
+    $info = Database::getConnectionInfo($query->getConnection());
+    $database = $info['default']['database'];
+    foreach ($this->query->getTables() as $table) {
+      if (isset($table['all_fields']) && $table['all_fields']) {
+
+        $database = 'plants';
+        $table = $table['table'];
+        $sql = 'SELECT column_name
+                FROM information_schema.columns
+                WHERE table_schema=:database AND table_name = :table
+                ORDER BY ordinal_position';
+        $result = dbtng_query($sql, array(':database' => $database, ':table' => $table));
+        foreach ($result as $row) {
+          $fields[drupal_strtolower($row->column_name)] = drupal_strtolower(
+            $table . '.' . $row->column_name);
+        }
+      }
+    }*/
+    $expressionFields = $this->query->getExpressions();
+    foreach ($expressionFields as $field_name => $field_info) {
+      // Lower case, because Drupal forces lowercase on fetch
+      $fields[drupal_strtolower($field_name)] = drupal_strtolower($field_info['alias']);
+    }
+
+    // Any caller-specified fields with the same names as extracted fields will
+    // override them; any others will be added
+    if ($this->fields) {
+      $fields = $this->fields + $fields;
+    }
+
+    return $fields;
+  }
+
+  /**
+   * Return a count of all available source records.
+   */
+  public function computeCount() {
+    $count = $this->countQuery->execute()->fetchField();
+    return $count;
+  }
+
+  /**
+   * Implementation of MigrateSource::performRewind().
+   *
+   * We could simply execute the query and be functionally correct, but
+   * we will take advantage of the PDO-based API to optimize the query up-front.
+   */
+  public function performRewind() {
+    $this->result = NULL;
+    $this->query = clone $this->originalQuery;
+
+    // Get the key values, for potential use in joining to the map table, or
+    // enforcing idlist.
+    $keys = array();
+    foreach ($this->activeMap->getSourceKey() as $field_name => $field_schema) {
+      if (isset($field_schema['alias'])) {
+        $field_name = $field_schema['alias'] . '.' . $field_name;
+      }
+      $keys[] = $field_name;
+    }
+
+    // The rules for determining what conditions to add to the query are as
+    // follows (applying first applicable rule)
+    // 1. If idlist is provided, then only process items in that list (AND key
+    //    IN (idlist)). Only applicable with single-value keys.
+    if ($this->idList) {
+      $this->query->condition($keys[0], $this->idList, 'IN');
+    }
+    else {
+      // 2. If the map is joinable, join it. We will want to accept all rows
+      //    which are either not in the map, or marked in the map as NEEDS_UPDATE.
+      //    Note that if highwater fields are in play, we want to accept all rows
+      //    above the highwater mark in addition to those selected by the map
+      //    conditions, so we need to OR them together (but AND with any existing
+      //    conditions in the query). So, ultimately the SQL condition will look
+      //    like (original conditions) AND (map IS NULL OR map needs update
+      //      OR above highwater).
+      $conditions = db_or();
+      $condition_added = FALSE;
+      if ($this->mapJoinable) {
+        // Build the join to the map table. Because the source key could have
+        // multiple fields, we need to build things up.
+        $count = 1;
+
+        foreach ($this->activeMap->getSourceKey() as $field_name => $field_schema) {
+          if (isset($field_schema['alias'])) {
+            $field_name = $field_schema['alias'] . '.' . $field_name;
+          }
+          $map_key = 'sourceid' . $count++;
+          if (!isset($map_join)) {
+            $map_join = '';
+          }
+          else {
+            $map_join .= ' AND ';
+          }
+          $map_join .= "$field_name = map.$map_key";
+        }
+
+        $alias = $this->query->leftJoin($this->activeMap->getQualifiedMapTable(),
+                                        'map', $map_join);
+        $conditions->isNull($alias . '.sourceid1');
+        $conditions->condition($alias . '.needs_update', MigrateMap::STATUS_NEEDS_UPDATE);
+        $condition_added = TRUE;
+
+        // And as long as we have the map table, add its data to the row.
+        $count = 1;
+        foreach ($this->activeMap->getSourceKey() as $field_name => $field_schema) {
+          $map_key = 'sourceid' . $count++;
+          $this->query->addField($alias, $map_key, "migrate_map_$map_key");
+        }
+        $count = 1;
+        foreach ($this->activeMap->getDestinationKey() as $field_name => $field_schema) {
+          $map_key = 'destid' . $count++;
+          $this->query->addField($alias, $map_key, "migrate_map_$map_key");
+        }
+        $this->query->addField($alias, 'needs_update', 'migrate_map_needs_update');
+      }
+      // 3. If we are using highwater marks, also include rows above the mark.
+      if (isset($this->highwaterField['name'])) {
+        if (isset($this->highwaterField['alias'])) {
+          $highwater = $this->highwaterField['alias'] . '.' . $this->highwaterField['name'];
+        }
+        else {
+          $highwater = $this->highwaterField['name'];
+        }
+        $conditions->condition($highwater, $this->activeMigration->getHighwater(), '>');
+        $condition_added = TRUE;
+      }
+      if ($condition_added) {
+        $this->query->condition($conditions);
+      }
+    }
+
+    migrate_instrument_start('MigrateSourceSQL execute');
+    $this->result = $this->query->execute();
+    migrate_instrument_stop('MigrateSourceSQL execute');
+  }
+
+  /**
+   * Implementation of MigrateSource::getNextRow().
+   *
+   * @return object
+   */
+  public function getNextRow() {
+    return $this->result->fetchObject();
+  }
+}

sites/all/modules/migrate/plugins/sources/sqlmap.inc

@@ -0,0 +1,622 @@
+<?php
+
+/**
+ * @file
+ * Defines a Drupal db-based implementation of MigrateMap.
+ */
+
+class MigrateSQLMap extends MigrateMap {
+  /**
+   * Names of tables created for tracking the migration.
+   *
+   * @var string
+   */
+  protected $mapTable, $messageTable;
+  public function getMapTable() {
+    return $this->mapTable;
+  }
+  public function getMessageTable() {
+    return $this->messageTable;
+  }
+
+  /**
+   * Qualifying the map table name with the database name makes cross-db joins
+   * possible. Note that, because prefixes are applied after we do this (i.e.,
+   * it will prefix the string we return), we do not qualify the table if it has
+   * a prefix. This will work fine when the source data is in the default
+   * (prefixed) database (in particular, for simpletest), but not if the primary
+   * query is in an external database.
+   *
+   * @return string
+   */
+  public function getQualifiedMapTable() {
+    $options = $this->connection->getConnectionOptions();
+    $prefix = $this->connection->tablePrefix($this->mapTable);
+    if ($prefix) {
+      return $this->mapTable;
+    }
+    else {
+      return $options['database'] . '.' . $this->mapTable;
+    }
+  }
+
+  /**
+   * sourceKey and destinationKey arrays are keyed by the field names; values
+   * are the Drupal schema definition for the field.
+   *
+   * @var array
+   */
+  public function getSourceKey() {
+    return $this->sourceKey;
+  }
+  public function getDestinationKey() {
+    return $this->destinationKey;
+  }
+
+  /**
+   * Drupal connection object on which to create the map/message tables
+   * @var DatabaseConnection
+   */
+  protected $connection;
+  public function getConnection() {
+    return $this->connection;
+  }
+
+  /**
+   * We don't need to check the tables more than once per request.
+   *
+   * @var boolean
+   */
+  protected $ensured;
+
+  public function __construct($machine_name, array $source_key,
+      array $destination_key, $connection_key = 'default') {
+    // Default generated table names, limited to 63 characters
+    $this->mapTable = 'migrate_map_' . drupal_strtolower($machine_name);
+    $this->mapTable = drupal_substr($this->mapTable, 0, 63);
+    $this->messageTable = 'migrate_message_' . drupal_strtolower($machine_name);
+    $this->messageTable = drupal_substr($this->messageTable, 0, 63);
+    $this->sourceKey = $source_key;
+    $this->destinationKey = $destination_key;
+    $this->connection = Database::getConnection('default', $connection_key);
+    // Build the source and destination key maps
+    $this->sourceKeyMap = array();
+    $count = 1;
+    foreach ($source_key as $field => $schema) {
+      $this->sourceKeyMap[$field] = 'sourceid' . $count++;
+    }
+    $this->destinationKeyMap = array();
+    $count = 1;
+    foreach ($destination_key as $field => $schema) {
+      $this->destinationKeyMap[$field] = 'destid' . $count++;
+    }
+    $this->ensureTables();
+  }
+
+  /**
+   * Create the map and message tables if they don't already exist.
+   */
+  protected function ensureTables() {
+    if (!$this->ensured) {
+      if (!$this->connection->schema()->tableExists($this->mapTable)) {
+        // Generate appropriate schema info for the map and message tables,
+        // and map from the source field names to the map/msg field names
+        $count = 1;
+        $source_key_schema = array();
+        $pks = array();
+        foreach ($this->sourceKey as $field_schema) {
+          $mapkey = 'sourceid' . $count++;
+          $source_key_schema[$mapkey] = $field_schema;
+          $pks[] = $mapkey;
+        }
+
+        $fields = $source_key_schema;
+
+        // Add destination keys to map table
+        // TODO: How do we discover the destination schema?
+        $count = 1;
+        foreach ($this->destinationKey as $field_schema) {
+          // Allow dest key fields to be NULL (for IGNORED/FAILED cases)
+          $field_schema['not null'] = FALSE;
+          $mapkey = 'destid' . $count++;
+          $fields[$mapkey] = $field_schema;
+        }
+        $fields['needs_update'] = array(
+          'type' => 'int',
+          'size' => 'tiny',
+          'unsigned' => TRUE,
+          'not null' => TRUE,
+          'default' => MigrateMap::STATUS_IMPORTED,
+          'description' => 'Indicates current status of the source row',
+        );
+        $fields['last_imported'] = array(
+          'type' => 'int',
+          'unsigned' => TRUE,
+          'not null' => TRUE,
+          'default' => 0,
+          'description' => 'UNIX timestamp of the last time this row was imported',
+        );
+        $schema = array(
+          'description' => t('Mappings from source key to destination key'),
+          'fields' => $fields,
+          'primary key' => $pks,
+        );
+        $this->connection->schema()->createTable($this->mapTable, $schema);
+
+        // Now for the message table
+        $fields = array();
+        $fields['msgid'] = array(
+          'type' => 'serial',
+          'unsigned' => TRUE,
+          'not null' => TRUE,
+        );
+        $fields += $source_key_schema;
+
+        $fields['level'] = array(
+          'type' => 'int',
+          'unsigned' => TRUE,
+          'not null' => TRUE,
+          'default' => 1,
+        );
+        $fields['message'] = array(
+          'type' => 'text',
+          'size' => 'medium',
+          'not null' => TRUE,
+        );
+        $schema = array(
+          'description' => t('Messages generated during a migration process'),
+          'fields' => $fields,
+          'primary key' => array('msgid'),
+          'indexes' => array('sourcekey' => $pks),
+        );
+        $this->connection->schema()->createTable($this->messageTable, $schema);
+      }
+      $this->ensured = TRUE;
+    }
+  }
+
+  /**
+   * Retrieve a row from the map table, given a source ID
+   *
+   * @param array $source_id
+   */
+  public function getRowBySource(array $source_id) {
+    migrate_instrument_start('mapRowBySource');
+    $query = $this->connection->select($this->mapTable, 'map')
+              ->fields('map');
+    foreach ($this->sourceKeyMap as $key_name) {
+      $query = $query->condition("map.$key_name", array_shift($source_id), '=');
+    }
+    $result = $query->execute();
+    migrate_instrument_stop('mapRowBySource');
+    return $result->fetchAssoc();
+  }
+
+  /**
+   * Retrieve a row from the map table, given a destination ID
+   *
+   * @param array $source_id
+   */
+  public function getRowByDestination(array $destination_id) {
+    migrate_instrument_start('mapRowByDestination');
+    $query = $this->connection->select($this->mapTable, 'map')
+              ->fields('map');
+    foreach ($this->destinationKeyMap as $key_name) {
+      $query = $query->condition("map.$key_name", array_shift($destination_id), '=');
+    }
+    $result = $query->execute();
+    migrate_instrument_stop('mapRowByDestination');
+    return $result->fetchAssoc();
+  }
+
+  /**
+   * Retrieve an array of map rows marked as needing update.
+   *
+   * @param int $count
+   *  Maximum rows to return; defaults to 10,000
+   * @return array
+   *  Array of map row objects with needs_update==1.
+   */
+  public function getRowsNeedingUpdate($count) {
+    $rows = array();
+    $result = db_select($this->mapTable, 'map')
+                      ->fields('map')
+                      ->condition('needs_update', MigrateMap::STATUS_NEEDS_UPDATE)
+                      ->range(0, $count)
+                      ->execute();
+    foreach ($result as $row) {
+      $rows[] = $row;
+    }
+    return $rows;
+  }
+
+  /**
+   * Given a (possibly multi-field) destination key, return the (possibly multi-field)
+   * source key mapped to it.
+   *
+   * @param array $destination_id
+   *  Array of destination key values.
+   * @return array
+   *  Array of source key values, or NULL on failure.
+   */
+  public function lookupSourceID(array $destination_id) {
+    migrate_instrument_start('lookupSourceID');
+    $query = $this->connection->select($this->mapTable, 'map')
+              ->fields('map', $this->sourceKeyMap);
+    foreach ($this->destinationKeyMap as $key_name) {
+      $query = $query->condition("map.$key_name", array_shift($destination_id), '=');
+    }
+    $result = $query->execute();
+    $source_id = $result->fetchAssoc();
+    migrate_instrument_stop('lookupSourceID');
+    return $source_id;
+  }
+
+  /**
+   * Given a (possibly multi-field) source key, return the (possibly multi-field)
+   * destination key it is mapped to.
+   *
+   * @param array $source_id
+   *  Array of source key values.
+   * @return array
+   *  Array of destination key values, or NULL on failure.
+   */
+  public function lookupDestinationID(array $source_id) {
+    migrate_instrument_start('lookupDestinationID');
+    $query = $this->connection->select($this->mapTable, 'map')
+              ->fields('map', $this->destinationKeyMap);
+    foreach ($this->sourceKeyMap as $key_name) {
+      $query = $query->condition("map.$key_name", array_shift($source_id), '=');
+    }
+    $result = $query->execute();
+    $destination_id = $result->fetchAssoc();
+    migrate_instrument_stop('lookupDestinationID');
+    return $destination_id;
+  }
+
+  /**
+   * Called upon successful import of one record, we record a mapping from
+   * the source key to the destination key. Also may be called, setting the
+   * third parameter to NEEDS_UPDATE, to signal an existing record should be remigrated.
+   *
+   * @param stdClass $source_row
+   *  The raw source data. We use the key map derived from the source object
+   *  to get the source key values.
+   * @param array $dest_ids
+   *  The destination key values.
+   * @param int $needs_update
+   *  Status of the source row in the map. Defaults to STATUS_IMPORTED.
+   */
+  public function saveIDMapping(stdClass $source_row, array $dest_ids, $needs_update = MigrateMap::STATUS_IMPORTED) {
+    migrate_instrument_start('saveIDMapping');
+    // Construct the source key
+    $keys = array();
+    foreach ($this->sourceKeyMap as $field_name => $key_name) {
+      // A NULL key value will fail.
+      if (is_null($source_row->$field_name)) {
+        Migration::displayMessage(t(
+          'Could not save to map table due to NULL value for key field !field',
+          array('!field' => $field_name)));
+        migrate_instrument_stop('saveIDMapping');
+        return;
+      }
+      $keys[$key_name] = $source_row->$field_name;
+    }
+
+    $fields = array('needs_update' => (int)$needs_update);
+    $count = 1;
+    foreach ($dest_ids as $dest_id) {
+      $fields['destid' . $count++] = $dest_id;
+    }
+    if ($this->trackLastImported) {
+      $fields['last_imported'] = time();
+    }
+    $this->connection->merge($this->mapTable)
+      ->key($keys)
+      ->fields($fields)
+      ->execute();
+    migrate_instrument_stop('saveIDMapping');
+  }
+
+  /**
+   * Record a message in the migration's message table.
+   *
+   * @param array $source_key
+   *  Source ID of the record in error
+   * @param string $message
+   *  The message to record.
+   * @param int $level
+   *  Optional message severity (defaults to MESSAGE_ERROR).
+   */
+  public function saveMessage($source_key, $message, $level = Migration::MESSAGE_ERROR) {
+    // Source IDs as arguments
+    $count = 1;
+    if (is_array($source_key)) {
+      foreach ($source_key as $key_value) {
+        $fields['sourceid' . $count++] = $key_value;
+        // If any key value is empty, we can't save - print out and abort
+        if (empty($key_value)) {
+          print($message);
+          return;
+        }
+      }
+      $fields['level'] = $level;
+      $fields['message'] = $message;
+      $this->connection->insert($this->messageTable)
+        ->fields($fields)
+        ->execute();
+    }
+    else {
+      // TODO: What else can we do?
+      Migration::displayMessage($message);
+    }
+  }
+
+  /**
+   * Prepares this migration to run as an update - that is, in addition to
+   * unmigrated content (source records not in the map table) being imported,
+   * previously-migrated content will also be updated in place.
+   */
+  public function prepareUpdate() {
+    $this->connection->update($this->mapTable)
+    ->fields(array('needs_update' => MigrateMap::STATUS_NEEDS_UPDATE))
+    ->execute();
+  }
+
+  /**
+   * Returns a count of records in the map table (i.e., the number of
+   * source records which have been processed for this migration).
+   *
+   * @return int
+   */
+  public function processedCount() {
+    $query = $this->connection->select($this->mapTable);
+    $query->addExpression('COUNT(*)', 'count');
+    $count = $query->execute()->fetchField();
+    return $count;
+  }
+
+  /**
+   * Returns a count of imported records in the map table.
+   *
+   * @return int
+   */
+  public function importedCount() {
+    $query = $this->connection->select($this->mapTable);
+    $query->addExpression('COUNT(*)', 'count');
+    $query->condition('needs_update', array(MigrateMap::STATUS_IMPORTED, MigrateMap::STATUS_NEEDS_UPDATE), 'IN');
+    $count = $query->execute()->fetchField();
+    return $count;
+  }
+
+  /**
+   * Returns a count of records which are marked as needing update.
+   *
+   * @return int
+   */
+  public function updateCount() {
+    $query = $this->connection->select($this->mapTable);
+    $query->addExpression('COUNT(*)', 'count');
+    $query->condition('needs_update', MigrateMap::STATUS_NEEDS_UPDATE);
+    $count = $query->execute()->fetchField();
+    return $count;
+  }
+
+  /**
+   * Get the number of source records which failed to import.
+   *
+   * @return int
+   *  Number of records errored out.
+   */
+  public function errorCount() {
+    $query = $this->connection->select($this->mapTable);
+    $query->addExpression('COUNT(*)', 'count');
+    $query->condition('needs_update', MigrateMap::STATUS_FAILED);
+    $count = $query->execute()->fetchField();
+    return $count;
+  }
+
+  /**
+   * Get the number of messages saved.
+   *
+   * @return int
+   *  Number of messages.
+   */
+  public function messageCount() {
+    $query = $this->connection->select($this->messageTable);
+    $query->addExpression('COUNT(*)', 'count');
+    $count = $query->execute()->fetchField();
+    return $count;
+  }
+
+  /**
+   * Delete the map entry and any message table entries for the specified source row.
+   *
+   * @param array $source_key
+   */
+  public function delete(array $source_key, $messages_only = FALSE) {
+    if (!$messages_only) {
+      $map_query = $this->connection->delete($this->mapTable);
+    }
+    $message_query = $this->connection->delete($this->messageTable);
+    $count = 1;
+    foreach ($source_key as $key_value) {
+      if (!$messages_only) {
+        $map_query->condition('sourceid' . $count, $key_value);
+      }
+      $message_query->condition('sourceid' . $count, $key_value);
+      $count++;
+    }
+
+    if (!$messages_only) {
+      $map_query->execute();
+    }
+    $message_query->execute();
+  }
+
+  /**
+   * Delete the map entry and any message table entries for the specified destination row.
+   *
+   * @param array $destination_key
+   */
+  public function deleteDestination(array $destination_key) {
+    $map_query = $this->connection->delete($this->mapTable);
+    $message_query = $this->connection->delete($this->messageTable);
+    $source_key = $this->lookupSourceID($destination_key);
+    if (!empty($source_key)) {
+      $count = 1;
+      foreach ($destination_key as $key_value) {
+        $map_query->condition('destid' . $count, $key_value);
+        $count++;
+      }
+      $map_query->execute();
+      $count = 1;
+      foreach ($source_key as $key_value) {
+        $message_query->condition('sourceid' . $count, $key_value);
+        $count++;
+      }
+      $message_query->execute();
+    }
+  }
+
+  /**
+   * Set the specified row to be updated, if it exists.
+   */
+  public function setUpdate(array $source_key) {
+    $query = $this->connection->update($this->mapTable)
+                              ->fields(array('needs_update' => MigrateMap::STATUS_NEEDS_UPDATE));
+    $count = 1;
+    foreach ($source_key as $key_value) {
+      $query->condition('sourceid' . $count++, $key_value);
+    }
+    $query->execute();
+  }
+
+  /**
+   * Delete all map and message table entries specified.
+   *
+   * @param array $source_keys
+   *  Each array member is an array of key fields for one source row.
+   */
+  public function deleteBulk(array $source_keys) {
+    // If we have a single-column key, we can shortcut it
+    if (count($this->sourceKey) == 1) {
+      $sourceids = array();
+      foreach ($source_keys as $source_key) {
+        $sourceids[] = $source_key;
+      }
+      $this->connection->delete($this->mapTable)
+        ->condition('sourceid1', $sourceids, 'IN')
+        ->execute();
+      $this->connection->delete($this->messageTable)
+        ->condition('sourceid1', $sourceids, 'IN')
+        ->execute();
+    }
+    else {
+      foreach ($source_keys as $source_key) {
+        $map_query = $this->connection->delete($this->mapTable);
+        $message_query = $this->connection->delete($this->messageTable);
+        $count = 1;
+        foreach ($source_key as $key_value) {
+          $map_query->condition('sourceid' . $count, $key_value);
+          $message_query->condition('sourceid' . $count++, $key_value);
+        }
+        $map_query->execute();
+        $message_query->execute();
+      }
+    }
+  }
+
+  /**
+   * Clear all messages from the message table.
+   */
+  public function clearMessages() {
+    $this->connection->truncate($this->messageTable)
+                     ->execute();
+  }
+
+  /**
+   * Remove the associated map and message tables.
+   */
+  public function destroy() {
+    $this->connection->schema()->dropTable($this->mapTable);
+    $this->connection->schema()->dropTable($this->messageTable);
+  }
+
+  protected $result = NULL;
+  protected $currentRow = NULL;
+  protected $currentKey = array();
+  public function getCurrentKey() {
+    return $this->currentKey;
+  }
+
+  /**
+   * Implementation of Iterator::rewind() - called before beginning a foreach loop.
+   * TODO: Support idlist, itemlimit
+   */
+  public function rewind() {
+    $this->currentRow = NULL;
+    $fields = array();
+    foreach ($this->sourceKeyMap as $field) {
+      $fields[] = $field;
+    }
+    foreach ($this->destinationKeyMap as $field) {
+      $fields[] = $field;
+    }
+
+    /* TODO
+    if (isset($this->options['itemlimit'])) {
+      $query = $query->range(0, $this->options['itemlimit']);
+    }
+    */
+    $this->result = $this->connection->select($this->mapTable, 'map')
+      ->fields('map', $fields)
+      ->execute();
+    $this->next();
+  }
+
+  /**
+   * Implementation of Iterator::current() - called when entering a loop
+   * iteration, returning the current row
+   */
+  public function current() {
+    return $this->currentRow;
+  }
+
+  /**
+   * Implementation of Iterator::key - called when entering a loop iteration, returning
+   * the key of the current row. It must be a scalar - we will serialize
+   * to fulfill the requirement, but using getCurrentKey() is preferable.
+   */
+  public function key() {
+    return serialize($this->currentKey);
+  }
+
+  /**
+   * Implementation of Iterator::next() - called at the bottom of the loop implicitly,
+   * as well as explicitly from rewind().
+   */
+  public function next() {
+    $this->currentRow = $this->result->fetchObject();
+    $this->currentKey = array();
+    if (!is_object($this->currentRow)) {
+      $this->currentRow = NULL;
+    }
+    else {
+      foreach ($this->sourceKeyMap as $map_field) {
+        $this->currentKey[$map_field] = $this->currentRow->$map_field;
+        // Leave only destination fields
+        unset($this->currentRow->$map_field);
+      }
+    }
+  }
+
+  /**
+   * Implementation of Iterator::valid() - called at the top of the loop, returning
+   * TRUE to process the loop and FALSE to terminate it
+   */
+  public function valid() {
+    // TODO: Check numProcessed against itemlimit
+    return !is_null($this->currentRow);
+  }
+}