opdavies/drupalcampbristol
Archived
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Update to Drupal 8.0.0-beta15. For more information, see: https://www.drupal.org/node/2563023

Browse source
This commit is contained in:
Pantheon Automation 2015-09-04 13:20:09 -07:00 • committed by Greg Anderson
parent 2720a9ec4b
commit f3791f1da3
1898 changed files with 54300 additions and 11481 deletions
Download patch file Download diff file Expand all files Collapse all files

31
core/modules/migrate/src/Plugin/MigrateBuilderInterface.php Normal file
View file

@ -0,0 +1,31 @@
<?php
/**
* @file
* Contains \Drupal\migrate\Plugin\MigrateBuilderInterface.
*/
namespace Drupal\migrate\Plugin;
/**
* Defines the builder plugin type.
*
* Builder plugins implement custom logic to generate migration entities from
* migration templates. For example, a migration may need to be customized
* based on data that's present in the source database; such customization is
* implemented by builders.
*/
interface MigrateBuilderInterface {
/**
* Builds migration entities based on a template.
*
* @param array $template
* The parsed template.
*
* @return \Drupal\migrate\Entity\MigrationInterface[]
* The unsaved migrations generated from the template.
*/
public function buildMigrations(array $template);
}

27
core/modules/migrate/src/Plugin/MigrateDestinationInterface.php
View file

@ -55,33 +55,6 @@ interface MigrateDestinationInterface extends PluginInspectionInterface {
*/
public function fields(MigrationInterface $migration = NULL);
/**
* Allows pre-processing of an import.
*
* Derived classes may implement preImport() to do any processing they need
* done before over all source rows.
*/
public function preImport();
/**
* Allows pre-processing of a rollback.
*/
public function preRollback();
/**
* Allows post-processing of an import.
*
* Derived classes may implement postImport(), to do any processing they need
* done after looping over all source rows.
*/
public function postImport();
/**
* Allows post-processing of a rollback.
*/
public function postRollback();
/**
* Import the row.
*

14
core/modules/migrate/src/Plugin/MigrateIdMapInterface.php
View file

@ -65,6 +65,20 @@ interface MigrateIdMapInterface extends \Iterator, PluginInspectionInterface {
*/
public function saveMessage(array $source_id_values, $message, $level = MigrationInterface::MESSAGE_ERROR);
/**
* Retrieves an iterator over messages relate to source records.
*
* @param array $source_id_values
* (optional) The source identifier values of a specific record to retrieve.
* If empty, all messages are retrieved.
* @param int $level
* (optional) Message severity. If NULL, retrieve messages of all severities.
*
* @return \Iterator
* Retrieves an iterator over the message rows.
*/
public function getMessageIterator(array $source_id_values = [], $level = NULL);
/**
* Prepares to run a full update.
*

52
core/modules/migrate/src/Plugin/migrate/builder/BuilderBase.php Normal file
View file

@ -0,0 +1,52 @@
<?php
/**
* @file
* Contains \Drupal\migrate\Plugin\migrate\builder\BuilderBase.
*/
namespace Drupal\migrate\Plugin\migrate\builder;
use Drupal\Core\Plugin\PluginBase;
use Drupal\migrate\Entity\Migration;
use Drupal\migrate\Plugin\MigrateBuilderInterface;
/**
* Base class for builder plugins.
*/
abstract class BuilderBase extends PluginBase implements MigrateBuilderInterface {
/**
* Returns a fully initialized instance of a source plugin.
*
* @param string $plugin_id
* The plugin ID.
* @param array $configuration
* (optional) Additional configuration for the plugin.
*
* @return \Drupal\migrate\Plugin\MigrateSourceInterface
* The fully initialized source plugin.
*/
protected function getSourcePlugin($plugin_id, array $configuration = []) {
$configuration['plugin'] = $plugin_id;
// By default, SqlBase subclasses will try to join on a map table. But in
// this case we're trying to use the source plugin as a detached iterator
// over the source data, so we don't want to join on (or create) the map
// table.
// @see SqlBase::initializeIterator()
$configuration['ignore_map'] = TRUE;
// Source plugins are tightly coupled to migration entities, so we need
// to create a fake migration in order to properly initialize the plugin.
$values = [
'id' => uniqid(),
'source' => $configuration,
// Since this isn't a real migration, we don't want a real destination --
// the 'null' destination is perfect for this.
'destination' => [
'plugin' => 'null',
],
];
return Migration::create($values)->getSourcePlugin();
}
}

49
core/modules/migrate/src/Plugin/migrate/destination/DestinationBase.php
View file

@ -59,34 +59,6 @@ abstract class DestinationBase extends PluginBase implements MigrateDestinationI
}
}
/**
* Modify the Row before it is imported.
*/
public function preImport() {
// By default we do nothing.
}
/**
* Modify the Row before it is rolled back.
*/
public function preRollback() {
// By default we do nothing.
}
/**
* {@inheritdoc}
*/
public function postImport() {
// By default we do nothing.
}
/**
* {@inheritdoc}
*/
public function postRollback() {
// By default we do nothing.
}
/**
* {@inheritdoc}
*/
@ -94,25 +66,4 @@ abstract class DestinationBase extends PluginBase implements MigrateDestinationI
// By default we do nothing.
}
/**
* {@inheritdoc}
*/
public function getCreated() {
// TODO: Implement getCreated() method.
}
/**
* {@inheritdoc}
*/
public function getUpdated() {
// TODO: Implement getUpdated() method.
}
/**
* {@inheritdoc}
*/
public function resetStats() {
// TODO: Implement resetStats() method.
}
}

69
core/modules/migrate/src/Plugin/migrate/id_map/Sql.php
View file

@ -9,13 +9,20 @@ namespace Drupal\migrate\Plugin\migrate\id_map;
use Drupal\Component\Utility\Unicode;
use Drupal\Core\Database\Connection;
use Drupal\Core\Entity\EntityTypeInterface;
use Drupal\Core\Field\BaseFieldDefinition;
use Drupal\Core\Plugin\ContainerFactoryPluginInterface;
use Drupal\Core\Plugin\PluginBase;
use Drupal\migrate\Entity\MigrationInterface;
use Drupal\migrate\MigrateException;
use Drupal\migrate\MigrateMessageInterface;
use Drupal\migrate\Plugin\MigrateIdMapInterface;
use Drupal\migrate\Row;
use Drupal\migrate\Event\MigrateEvents;
use Drupal\migrate\Event\MigrateMapSaveEvent;
use Drupal\migrate\Event\MigrateMapDeleteEvent;
use Symfony\Component\DependencyInjection\ContainerInterface;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;
/**
* Defines the sql based ID map implementation.
@ -25,7 +32,14 @@ use Drupal\migrate\Row;
*
* @PluginID("sql")
*/
class Sql extends PluginBase implements MigrateIdMapInterface {
class Sql extends PluginBase implements MigrateIdMapInterface, ContainerFactoryPluginInterface {
/**
* An event dispatcher instance to use for map events.
*
* @var \Symfony\Component\EventDispatcher\EventDispatcherInterface
*/
protected $eventDispatcher;
/**
* The migration map table name.
@ -137,10 +151,23 @@ class Sql extends PluginBase implements MigrateIdMapInterface {
* @param \Drupal\migrate\Entity\MigrationInterface $migration
* The migration to do.
*/
public function __construct(array $configuration, $plugin_id, $plugin_definition, MigrationInterface $migration) {
public function __construct(array $configuration, $plugin_id, $plugin_definition, MigrationInterface $migration, EventDispatcherInterface $event_dispatcher) {
parent::__construct($configuration, $plugin_id, $plugin_definition);
$this->migration = $migration;
$this->eventDispatcher = $event_dispatcher;
}
/**
* {@inheritdoc}
*/
public static function create(ContainerInterface $container, array $configuration, $plugin_id, $plugin_definition, MigrationInterface $migration = NULL) {
return new static(
$configuration,
$plugin_id,
$plugin_definition,
$migration,
$container->get('event_dispatcher')
);
}
/**
@ -504,6 +531,8 @@ class Sql extends PluginBase implements MigrateIdMapInterface {
$fields['last_imported'] = time();
}
if ($keys) {
// Notify anyone listening of the map row we're about to save.
$this->eventDispatcher->dispatch(MigrateEvents::MAP_SAVE, new MigrateMapSaveEvent($this, $keys + $fields));
$this->getDatabase()->merge($this->mapTableName())
->key($keys)
->fields($fields)
@ -530,6 +559,22 @@ class Sql extends PluginBase implements MigrateIdMapInterface {
->execute();
}
/**
* {@inheritdoc}
*/
public function getMessageIterator(array $source_id_values = [], $level = NULL) {
$query = $this->getDatabase()->select($this->messageTableName(), 'msg')
->fields('msg');
$count = 1;
foreach ($source_id_values as $id_value) {
$query->condition('sourceid' . $count++, $id_value);
}
if ($level) {
$query->condition('level', $level);
}
return $query->execute();
}
/**
* {@inheritdoc}
*/
@ -620,6 +665,8 @@ class Sql extends PluginBase implements MigrateIdMapInterface {
}
if (!$messages_only) {
// Notify anyone listening of the map row we're about to delete.
$this->eventDispatcher->dispatch(MigrateEvents::MAP_DELETE, new MigrateMapDeleteEvent($this, $source_id_values));
$map_query->execute();
}
$message_query->execute();
@ -638,6 +685,8 @@ class Sql extends PluginBase implements MigrateIdMapInterface {
$map_query->condition('destid' . $count, $key_value);
$count++;
}
// Notify anyone listening of the map row we're about to delete.
$this->eventDispatcher->dispatch(MigrateEvents::MAP_DELETE, new MigrateMapDeleteEvent($this, $source_id));
$map_query->execute();
$count = 1;
foreach ($source_id as $key_value) {
@ -673,6 +722,8 @@ class Sql extends PluginBase implements MigrateIdMapInterface {
if (count($this->migration->getSourcePlugin()->getIds()) == 1) {
$sourceids = array();
foreach ($source_id_values as $source_id) {
// Notify anyone listening of the map rows we're about to delete.
$this->eventDispatcher->dispatch(MigrateEvents::MAP_DELETE, new MigrateMapDeleteEvent($this, $source_id));
$sourceids[] = $source_id;
}
$this->getDatabase()->delete($this->mapTableName())
@ -684,6 +735,8 @@ class Sql extends PluginBase implements MigrateIdMapInterface {
}
else {
foreach ($source_id_values as $source_id) {
// Notify anyone listening of the map rows we're deleting.
$this->eventDispatcher->dispatch(MigrateEvents::MAP_DELETE, new MigrateMapDeleteEvent($this, $source_id));
$map_query = $this->getDatabase()->delete($this->mapTableName());
$message_query = $this->getDatabase()->delete($this->messageTableName());
$count = 1;
@ -716,8 +769,6 @@ class Sql extends PluginBase implements MigrateIdMapInterface {
* Implementation of Iterator::rewind().
*
* This is called before beginning a foreach loop.
*
* @todo Support idlist, itemlimit.
*/
public function rewind() {
$this->currentRow = NULL;
@ -728,13 +779,6 @@ class Sql extends PluginBase implements MigrateIdMapInterface {
foreach ($this->destinationIdFields() as $field) {
$fields[] = $field;
}
// @todo Make this work.
/*
if (isset($this->options['itemlimit'])) {
$query = $query->range(0, $this->options['itemlimit']);
}
*/
$this->result = $this->getDatabase()->select($this->mapTableName(), 'map')
->fields('map', $fields)
->execute();
@ -786,7 +830,6 @@ class Sql extends PluginBase implements MigrateIdMapInterface {
* and FALSE to terminate it.
*/
public function valid() {
// @todo Check numProcessed against itemlimit.
return $this->currentRow !== FALSE;
}

7
core/modules/migrate/src/Plugin/migrate/process/Extract.php
View file

@ -33,7 +33,12 @@ class Extract extends ProcessPluginBase {
}
$new_value = NestedArray::getValue($value, $this->configuration['index'], $key_exists);
if (!$key_exists) {
throw new MigrateException('Array index missing, extraction failed.');
if (isset($this->configuration['default'])) {
$new_value = $this->configuration['default'];
}
else {
throw new MigrateException('Array index missing, extraction failed.');
}
}
return $new_value;
}

2
core/modules/migrate/src/Plugin/migrate/process/Migration.php
View file

@ -11,7 +11,6 @@ namespace Drupal\migrate\Plugin\migrate\process;
use Drupal\Core\Entity\EntityStorageInterface;
use Drupal\Core\Plugin\ContainerFactoryPluginInterface;
use Drupal\migrate\MigrateSkipProcessException;
use Drupal\migrate\MigrateSkipRowException;
use Drupal\migrate\Plugin\MigratePluginManager;
use Drupal\migrate\ProcessPluginBase;
use Drupal\migrate\Entity\MigrationInterface;
@ -145,7 +144,6 @@ class Migration extends ProcessPluginBase implements ContainerFactoryPluginInter
return $destination_ids;
}
}
throw new MigrateSkipRowException();
}
/**

1
core/modules/migrate/src/Plugin/migrate/process/Route.php
View file

@ -47,6 +47,7 @@ class Route extends ProcessPluginBase implements ContainerFactoryPluginInterface
$container->get('path.validator')
);
}
/**
* {@inheritdoc}
*

89
core/modules/migrate/src/Plugin/migrate/source/EmbeddedDataSource.php Normal file
View file

@ -0,0 +1,89 @@
<?php
/**
* @file
* Contains \Drupal\migrate\Plugin\migrate\source\EmbeddedDataSource.
*/
namespace Drupal\migrate\Plugin\migrate\source;
use Drupal\migrate\Entity\MigrationInterface;
/**
* Source which takes its data directly from the plugin config.
*
* @MigrateSource(
* id = "embedded_data"
* )
*/
class EmbeddedDataSource extends SourcePluginBase {
/**
* Data obtained from the source plugin configuration.
*
* @var array[]
* Array of data rows, each one an array of values keyed by field names.
*/
protected $dataRows = [];
/**
* Description of the unique ID fields for this source.
*
* @var array[]
* Each array member is keyed by a field name, with a value that is an
* array with a single member with key 'type' and value a column type such
* as 'integer'.
*/
protected $ids = [];
/**
* {@inheritdoc}
*/
public function __construct(array $configuration, $plugin_id, $plugin_definition, MigrationInterface $migration) {
parent::__construct($configuration, $plugin_id, $plugin_definition, $migration);
$this->dataRows = $configuration['data_rows'];
$this->ids = $configuration['ids'];
}
/**
* {@inheritdoc}
*/
public function fields() {
if ($this->count() > 0) {
$first_row = reset($this->dataRows);
$field_names = array_keys($first_row);
return array_combine($field_names, $field_names);
}
else {
return [];
}
}
/**
* {@inheritdoc}
*/
public function initializeIterator() {
return new \ArrayIterator($this->dataRows);
}
/**
* {@inheritdoc}
*/
public function __toString() {
return 'Embedded data';
}
/**
* {@inheritdoc}
*/
public function getIds() {
return $this->ids;
}
/**
* {@inheritdoc}
*/
public function count() {
return count($this->dataRows);
}
}

106
core/modules/migrate/src/Plugin/migrate/source/SourcePluginBase.php
View file

@ -10,6 +10,8 @@ namespace Drupal\migrate\Plugin\migrate\source;
use Drupal\Core\Plugin\PluginBase;
use Drupal\migrate\Entity\MigrationInterface;
use Drupal\migrate\MigrateException;
use Drupal\migrate\MigrateExecutableInterface;
use Drupal\migrate\MigrateSkipRowException;
use Drupal\migrate\Plugin\MigrateIdMapInterface;
use Drupal\migrate\Plugin\MigrateSourceInterface;
use Drupal\migrate\Row;
@ -59,20 +61,6 @@ abstract class SourcePluginBase extends PluginBase implements MigrateSourceInter
*/
protected $currentSourceIds;
/**
* Number of rows intentionally ignored (prepareRow() returned FALSE)
*
* @var int
*/
protected $numIgnored = 0;
/**
* Number of rows we've at least looked at.
*
* @var int
*/
protected $numProcessed = 0;
/**
* The high water mark at the beginning of the import operation.
*
@ -83,13 +71,6 @@ abstract class SourcePluginBase extends PluginBase implements MigrateSourceInter
*/
protected $originalHighWater;
/**
* List of source IDs to process.
*
* @var array
*/
protected $idList = array();
/**
* Whether this instance should cache the source count.
*
@ -144,8 +125,12 @@ abstract class SourcePluginBase extends PluginBase implements MigrateSourceInter
*/
protected $iterator;
// @TODO, find out how to remove this.
// @see https://www.drupal.org/node/2443617
/**
* @TODO, find out how to remove this.
* @see https://www.drupal.org/node/2443617
*
* @var MigrateExecutableInterface
*/
public $migrateExecutable;
/**
@ -166,10 +151,6 @@ abstract class SourcePluginBase extends PluginBase implements MigrateSourceInter
$this->originalHighWater = $this->migration->getHighWater();
}
if ($id_list = $this->migration->get('idlist')) {
$this->idList = $id_list;
}
// Don't allow the use of both highwater and track changes together.
if ($this->highWaterProperty && $this->trackChanges) {
throw new MigrateException('You should either use a highwater mark or track changes not both. They are both designed to solve the same problem');
@ -201,22 +182,31 @@ abstract class SourcePluginBase extends PluginBase implements MigrateSourceInter
* {@inheritdoc}
*/
public function prepareRow(Row $row) {
$result = TRUE;
$result_hook = $this->getModuleHandler()->invokeAll('migrate_prepare_row', array($row, $this, $this->migration));
$result_named_hook = $this->getModuleHandler()->invokeAll('migrate_' . $this->migration->id() . '_prepare_row', array($row, $this, $this->migration));
try {
$result_hook = $this->getModuleHandler()->invokeAll('migrate_prepare_row', array($row, $this, $this->migration));
$result_named_hook = $this->getModuleHandler()->invokeAll('migrate_' . $this->migration->id() . '_prepare_row', array($row, $this, $this->migration));
// We will skip if any hook returned FALSE.
$skip = ($result_hook && in_array(FALSE, $result_hook)) || ($result_named_hook && in_array(FALSE, $result_named_hook));
$save_to_map = TRUE;
}
catch (MigrateSkipRowException $e) {
$skip = TRUE;
$save_to_map = $e->getSaveToMap();
}
// We're explicitly skipping this row - keep track in the map table.
if (($result_hook && in_array(FALSE, $result_hook)) || ($result_named_hook && in_array(FALSE, $result_named_hook))) {
if ($skip) {
// Make sure we replace any previous messages for this item with any
// new ones.
$id_map = $this->migration->getIdMap();
$id_map->delete($this->currentSourceIds, TRUE);
$this->migrateExecutable->saveQueuedMessages();
$id_map->saveIdMapping($row, array(), MigrateIdMapInterface::STATUS_IGNORED, $this->migrateExecutable->rollbackAction);
$this->numIgnored++;
$this->currentRow = NULL;
$this->currentSourceIds = NULL;
if ($save_to_map) {
$id_map->saveIdMapping($row, array(), MigrateIdMapInterface::STATUS_IGNORED, $this->migrateExecutable->rollbackAction);
$this->currentRow = NULL;
$this->currentSourceIds = NULL;
}
$result = FALSE;
}
elseif ($this->trackChanges) {
@ -226,7 +216,6 @@ abstract class SourcePluginBase extends PluginBase implements MigrateSourceInter
// after hashes).
$row->rehash();
}
$this->numProcessed++;
return $result;
}
@ -280,8 +269,6 @@ abstract class SourcePluginBase extends PluginBase implements MigrateSourceInter
*/
public function rewind() {
$this->idMap = $this->migration->getIdMap();
$this->numProcessed = 0;
$this->numIgnored = 0;
$this->getIterator()->rewind();
$this->next();
}
@ -320,25 +307,17 @@ abstract class SourcePluginBase extends PluginBase implements MigrateSourceInter
$row->setIdMap($id_map);
}
// In case we have specified an ID list, but the ID given by the source is
// not in there, we skip the row.
$id_in_the_list = $this->idList && in_array(reset($this->currentSourceIds), $this->idList);
if ($this->idList && !$id_in_the_list) {
continue;
}
// Preparing the row gives source plugins the chance to skip.
if ($this->prepareRow($row) === FALSE) {
continue;
}
// Check whether the row needs processing.
// 1. Explicitly specified IDs.
// 2. This row has not been imported yet.
// 3. Explicitly set to update.
// 4. The row is newer than the current highwater mark.
// 5. If no such property exists then try by checking the hash of the row.
if ($id_in_the_list || !$row->getIdMap() || $row->needsUpdate() || $this->aboveHighwater($row) || $this->rowChanged($row) ) {
// 1. This row has not been imported yet.
// 2. Explicitly set to update.
// 3. The row is newer than the current highwater mark.
// 4. If no such property exists then try by checking the hash of the row.
if (!$row->getIdMap() || $row->needsUpdate() || $this->aboveHighwater($row) || $this->rowChanged($row) ) {
$this->currentRow = $row->freezeSource();
}
}
@ -377,27 +356,6 @@ abstract class SourcePluginBase extends PluginBase implements MigrateSourceInter
return $this->currentSourceIds;
}
/**
* Getter for numIgnored data member.
*/
public function getIgnored() {
return $this->numIgnored;
}
/**
* Getter for numProcessed data member.
*/
public function getProcessed() {
return $this->numProcessed;
}
/**
* Reset numIgnored back to 0.
*/
public function resetStats() {
$this->numIgnored = 0;
}
/**
* Get the source count.
*
@ -423,7 +381,7 @@ abstract class SourcePluginBase extends PluginBase implements MigrateSourceInter
// class to get the count from the source.
if ($refresh || !$this->cacheCounts) {
$count = $this->getIterator()->count();
$this->getCache()->set($this->cacheKey, $count, 'cache');
$this->getCache()->set($this->cacheKey, $count);
}
else {
// Caching is in play, first try to retrieve a cached count.
@ -436,7 +394,7 @@ abstract class SourcePluginBase extends PluginBase implements MigrateSourceInter
// No cached count, ask the derived class to count 'em up, and cache
// the result.
$count = $this->getIterator()->count();
$this->getCache()->set($this->cacheKey, $count, 'cache');
$this->getCache()->set($this->cacheKey, $count);
}
}
return $count;

112
core/modules/migrate/src/Plugin/migrate/source/SqlBase.php
View file

@ -28,11 +28,6 @@ abstract class SqlBase extends SourcePluginBase {
*/
protected $query;
/**
* @var \Drupal\migrate\Entity\MigrationInterface
*/
protected $migration;
/**
* @var \Drupal\Core\Database\Connection
*/
@ -116,76 +111,69 @@ abstract class SqlBase extends SourcePluginBase {
$this->prepareQuery();
$high_water_property = $this->migration->get('highWaterProperty');
// Get the key values, for potential use in joining to the map table, or
// enforcing idlist.
// Get the key values, for potential use in joining to the map table.
$keys = array();
// 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 ($id_list = $this->migration->get('idlist')) {
$this->query->condition($keys[0], $id_list, '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 high water fields are in play, we want to accept all rows
// above the high water 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 high water).
$conditions = $this->query->orConditionGroup();
$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;
$map_join = '';
$delimiter = '';
foreach ($this->getIds() as $field_name => $field_schema) {
if (isset($field_schema['alias'])) {
$field_name = $field_schema['alias'] . '.' . $field_name;
}
$map_join .= "$delimiter$field_name = map.sourceid" . $count++;
$delimiter = ' AND ';
// 1. 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 high water fields are in play, we want to accept all rows
// above the high water 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 high water).
$conditions = $this->query->orConditionGroup();
$condition_added = FALSE;
if (empty($this->configuration['ignore_map']) && $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;
$map_join = '';
$delimiter = '';
foreach ($this->getIds() as $field_name => $field_schema) {
if (isset($field_schema['alias'])) {
$field_name = $field_schema['alias'] . '.' . $field_name;
}
$map_join .= "$delimiter$field_name = map.sourceid" . $count++;
$delimiter = ' AND ';
}
$alias = $this->query->leftJoin($this->migration->getIdMap()->getQualifiedMapTableName(), 'map', $map_join);
$conditions->isNull($alias . '.sourceid1');
$conditions->condition($alias . '.source_row_status', MigrateIdMapInterface::STATUS_NEEDS_UPDATE);
$condition_added = TRUE;
$alias = $this->query->leftJoin($this->migration->getIdMap()->getQualifiedMapTableName(), 'map', $map_join);
$conditions->isNull($alias . '.sourceid1');
$conditions->condition($alias . '.source_row_status', MigrateIdMapInterface::STATUS_NEEDS_UPDATE);
$condition_added = TRUE;
// And as long as we have the map table, add its data to the row.
$n = count($this->getIds());
// And as long as we have the map table, add its data to the row.
$n = count($this->getIds());
for ($count = 1; $count <= $n; $count++) {
$map_key = 'sourceid' . $count;
$this->query->addField($alias, $map_key, "migrate_map_$map_key");
}
if ($n = count($this->migration->get('destinationIds'))) {
for ($count = 1; $count <= $n; $count++) {
$map_key = 'sourceid' . $count;
$map_key = 'destid' . $count++;
$this->query->addField($alias, $map_key, "migrate_map_$map_key");
}
if ($n = count($this->migration->get('destinationIds'))) {
for ($count = 1; $count <= $n; $count++) {
$map_key = 'destid' . $count++;
$this->query->addField($alias, $map_key, "migrate_map_$map_key");
}
}
$this->query->addField($alias, 'source_row_status', 'migrate_map_source_row_status');
}
// 3. If we are using high water marks, also include rows above the mark.
// But, include all rows if the high water mark is not set.
if (isset($high_water_property['name']) && ($high_water = $this->migration->getHighWater()) !== '') {
if (isset($high_water_property['alias'])) {
$high_water = $high_water_property['alias'] . '.' . $high_water_property['name'];
}
else {
$high_water = $high_water_property['name'];
}
$conditions->condition($high_water, $high_water, '>');
$condition_added = TRUE;
$this->query->addField($alias, 'source_row_status', 'migrate_map_source_row_status');
}
// 2. If we are using high water marks, also include rows above the mark.
// But, include all rows if the high water mark is not set.
if (isset($high_water_property['name']) && ($high_water = $this->migration->getHighWater()) !== '') {
if (isset($high_water_property['alias'])) {
$high_water = $high_water_property['alias'] . '.' . $high_water_property['name'];
}
if ($condition_added) {
$this->query->condition($conditions);
else {
$high_water = $high_water_property['name'];
}
$conditions->condition($high_water, $high_water, '>');
$condition_added = TRUE;
}
if ($condition_added) {
$this->query->condition($conditions);
}
return new \IteratorIterator($this->query->execute());