www.aleph1.co.uk Git - yaffs-website/blob - web/core/modules/migrate/src/Plugin/migrate/id_map/Sql.php

   1 <?php
   2
   3 namespace Drupal\migrate\Plugin\migrate\id_map;
   4
   5 use Drupal\Component\Utility\Unicode;
   6 use Drupal\Core\Field\BaseFieldDefinition;
   7 use Drupal\Core\Plugin\ContainerFactoryPluginInterface;
   8 use Drupal\Core\Plugin\PluginBase;
   9 use Drupal\migrate\MigrateMessage;
  10 use Drupal\migrate\Plugin\MigrationInterface;
  11 use Drupal\migrate\Event\MigrateIdMapMessageEvent;
  12 use Drupal\migrate\MigrateException;
  13 use Drupal\migrate\MigrateMessageInterface;
  14 use Drupal\migrate\Plugin\MigrateIdMapInterface;
  15 use Drupal\migrate\Row;
  16 use Drupal\migrate\Event\MigrateEvents;
  17 use Drupal\migrate\Event\MigrateMapSaveEvent;
  18 use Drupal\migrate\Event\MigrateMapDeleteEvent;
  19 use Symfony\Component\DependencyInjection\ContainerInterface;
  20 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
  21
  22 /**
  23  * Defines the sql based ID map implementation.
  24  *
  25  * It creates one map and one message table per migration entity to store the
  26  * relevant information.
  27  *
  28  * @PluginID("sql")
  29  */
  30 class Sql extends PluginBase implements MigrateIdMapInterface, ContainerFactoryPluginInterface {
  31
  32   /**
  33    * Column name of hashed source id values.
  34    */
  35   const SOURCE_IDS_HASH = 'source_ids_hash';
  36
  37   /**
  38    * An event dispatcher instance to use for map events.
  39    *
  40    * @var \Symfony\Component\EventDispatcher\EventDispatcherInterface
  41    */
  42   protected $eventDispatcher;
  43
  44   /**
  45    * The migration map table name.
  46    *
  47    * @var string
  48    */
  49   protected $mapTableName;
  50
  51   /**
  52    * The message table name.
  53    *
  54    * @var string
  55    */
  56   protected $messageTableName;
  57
  58   /**
  59    * The migrate message service.
  60    *
  61    * @var \Drupal\migrate\MigrateMessageInterface
  62    */
  63   protected $message;
  64
  65   /**
  66    * The database connection for the map/message tables on the destination.
  67    *
  68    * @var \Drupal\Core\Database\Connection
  69    */
  70   protected $database;
  71
  72   /**
  73    * The select query.
  74    *
  75    * @var \Drupal\Core\Database\Query\SelectInterface
  76    */
  77   protected $query;
  78
  79   /**
  80    * The migration being done.
  81    *
  82    * @var \Drupal\migrate\Plugin\MigrationInterface
  83    */
  84   protected $migration;
  85
  86   /**
  87    * The source ID fields.
  88    *
  89    * @var array
  90    */
  91   protected $sourceIdFields;
  92
  93   /**
  94    * The destination ID fields.
  95    *
  96    * @var array
  97    */
  98   protected $destinationIdFields;
  99
 100   /**
 101    * Whether the plugin is already initialized.
 102    *
 103    * @var bool
 104    */
 105   protected $initialized;
 106
 107   /**
 108    * The result.
 109    *
 110    * @var null
 111    */
 112   protected $result = NULL;
 113
 114   /**
 115    * The source identifiers.
 116    *
 117    * @var array
 118    */
 119   protected $sourceIds = [];
 120
 121   /**
 122    * The destination identifiers.
 123    *
 124    * @var array
 125    */
 126   protected $destinationIds = [];
 127
 128   /**
 129    * The current row.
 130    *
 131    * @var null
 132    */
 133   protected $currentRow = NULL;
 134
 135   /**
 136    * The current key.
 137    *
 138    * @var array
 139    */
 140   protected $currentKey = [];
 141
 142   /**
 143    * Constructs an SQL object.
 144    *
 145    * Sets up the tables and builds the maps,
 146    *
 147    * @param array $configuration
 148    *   The configuration.
 149    * @param string $plugin_id
 150    *   The plugin ID for the migration process to do.
 151    * @param mixed $plugin_definition
 152    *   The configuration for the plugin.
 153    * @param \Drupal\migrate\Plugin\MigrationInterface $migration
 154    *   The migration to do.
 155    */
 156   public function __construct(array $configuration, $plugin_id, $plugin_definition, MigrationInterface $migration, EventDispatcherInterface $event_dispatcher) {
 157     parent::__construct($configuration, $plugin_id, $plugin_definition);
 158     $this->migration = $migration;
 159     $this->eventDispatcher = $event_dispatcher;
 160     $this->message = new MigrateMessage();
 161   }
 162
 163   /**
 164    * {@inheritdoc}
 165    */
 166   public static function create(ContainerInterface $container, array $configuration, $plugin_id, $plugin_definition, MigrationInterface $migration = NULL) {
 167     return new static(
 168       $configuration,
 169       $plugin_id,
 170       $plugin_definition,
 171       $migration,
 172       $container->get('event_dispatcher')
 173     );
 174   }
 175
 176   /**
 177    * Retrieves the hash of the source identifier values.
 178    *
 179    * @internal
 180    *
 181    * @param array $source_id_values
 182    *   The source identifiers
 183    *
 184    * @return string
 185    *   An hash containing the hashed values of the source identifiers.
 186    */
 187   public function getSourceIDsHash(array $source_id_values) {
 188     // When looking up the destination ID we require an array with both the
 189     // source key and value, e.g. ['nid' => 41]. In this case, $source_id_values
 190     // need to be ordered the same order as $this->sourceIdFields().
 191     // However, the Migration process plugin doesn't currently have a way to get
 192     // the source key so we presume the values have been passed through in the
 193     // correct order.
 194     if (!isset($source_id_values[0])) {
 195       $source_id_values_keyed = [];
 196       foreach ($this->sourceIdFields() as $field_name => $source_id) {
 197         $source_id_values_keyed[] = $source_id_values[$field_name];
 198       }
 199       $source_id_values = $source_id_values_keyed;
 200     }
 201     return hash('sha256', serialize(array_map('strval', $source_id_values)));
 202   }
 203
 204   /**
 205    * The source ID fields.
 206    *
 207    * @return array
 208    *   The source ID fields.
 209    */
 210   protected function sourceIdFields() {
 211     if (!isset($this->sourceIdFields)) {
 212       // Build the source and destination identifier maps.
 213       $this->sourceIdFields = [];
 214       $count = 1;
 215       foreach ($this->migration->getSourcePlugin()->getIds() as $field => $schema) {
 216         $this->sourceIdFields[$field] = 'sourceid' . $count++;
 217       }
 218     }
 219     return $this->sourceIdFields;
 220   }
 221
 222   /**
 223    * The destination ID fields.
 224    *
 225    * @return array
 226    *   The destination ID fields.
 227    */
 228   protected function destinationIdFields() {
 229     if (!isset($this->destinationIdFields)) {
 230       $this->destinationIdFields = [];
 231       $count = 1;
 232       foreach ($this->migration->getDestinationPlugin()->getIds() as $field => $schema) {
 233         $this->destinationIdFields[$field] = 'destid' . $count++;
 234       }
 235     }
 236     return $this->destinationIdFields;
 237   }
 238
 239   /**
 240    * The name of the database map table.
 241    *
 242    * @return string
 243    *   The map table name.
 244    */
 245   public function mapTableName() {
 246     $this->init();
 247     return $this->mapTableName;
 248   }
 249
 250   /**
 251    * The name of the database message table.
 252    *
 253    * @return string
 254    *   The message table name.
 255    */
 256   public function messageTableName() {
 257     $this->init();
 258     return $this->messageTableName;
 259   }
 260
 261   /**
 262    * Get the fully qualified map table name.
 263    *
 264    * @return string
 265    *   The fully qualified map table name.
 266    */
 267   public function getQualifiedMapTableName() {
 268     return $this->getDatabase()->getFullQualifiedTableName($this->mapTableName);
 269   }
 270
 271   /**
 272    * Gets the database connection.
 273    *
 274    * @return \Drupal\Core\Database\Connection
 275    *   The database connection object.
 276    */
 277   public function getDatabase() {
 278     if (!isset($this->database)) {
 279       $this->database = \Drupal::database();
 280     }
 281     $this->init();
 282     return $this->database;
 283   }
 284
 285   /**
 286    * Initialize the plugin.
 287    */
 288   protected function init() {
 289     if (!$this->initialized) {
 290       $this->initialized = TRUE;
 291       // Default generated table names, limited to 63 characters.
 292       $machine_name = str_replace(':', '__', $this->migration->id());
 293       $prefix_length = strlen($this->getDatabase()->tablePrefix());
 294       $this->mapTableName = 'migrate_map_' . Unicode::strtolower($machine_name);
 295       $this->mapTableName = Unicode::substr($this->mapTableName, 0, 63 - $prefix_length);
 296       $this->messageTableName = 'migrate_message_' . Unicode::strtolower($machine_name);
 297       $this->messageTableName = Unicode::substr($this->messageTableName, 0, 63 - $prefix_length);
 298       $this->ensureTables();
 299     }
 300   }
 301
 302   /**
 303    * {@inheritdoc}
 304    */
 305   public function setMessage(MigrateMessageInterface $message) {
 306     $this->message = $message;
 307   }
 308
 309   /**
 310    * Create the map and message tables if they don't already exist.
 311    */
 312   protected function ensureTables() {
 313     if (!$this->getDatabase()->schema()->tableExists($this->mapTableName)) {
 314       // Generate appropriate schema info for the map and message tables,
 315       // and map from the source field names to the map/msg field names.
 316       $count = 1;
 317       $source_id_schema = [];
 318       $indexes = [];
 319       foreach ($this->migration->getSourcePlugin()->getIds() as $id_definition) {
 320         $mapkey = 'sourceid' . $count++;
 321         $indexes['source'][] = $mapkey;
 322         $source_id_schema[$mapkey] = $this->getFieldSchema($id_definition);
 323         $source_id_schema[$mapkey]['not null'] = TRUE;
 324       }
 325
 326       $source_ids_hash[static::SOURCE_IDS_HASH] = [
 327         'type' => 'varchar',
 328         'length' => '64',
 329         'not null' => TRUE,
 330         'description' => 'Hash of source ids. Used as primary key',
 331       ];
 332       $fields = $source_ids_hash + $source_id_schema;
 333
 334       // Add destination identifiers to map table.
 335       // @todo How do we discover the destination schema?
 336       $count = 1;
 337       foreach ($this->migration->getDestinationPlugin()->getIds() as $id_definition) {
 338         // Allow dest identifier fields to be NULL (for IGNORED/FAILED cases).
 339         $mapkey = 'destid' . $count++;
 340         $fields[$mapkey] = $this->getFieldSchema($id_definition);
 341         $fields[$mapkey]['not null'] = FALSE;
 342       }
 343       $fields['source_row_status'] = [
 344         'type' => 'int',
 345         'size' => 'tiny',
 346         'unsigned' => TRUE,
 347         'not null' => TRUE,
 348         'default' => MigrateIdMapInterface::STATUS_IMPORTED,
 349         'description' => 'Indicates current status of the source row',
 350       ];
 351       $fields['rollback_action'] = [
 352         'type' => 'int',
 353         'size' => 'tiny',
 354         'unsigned' => TRUE,
 355         'not null' => TRUE,
 356         'default' => MigrateIdMapInterface::ROLLBACK_DELETE,
 357         'description' => 'Flag indicating what to do for this item on rollback',
 358       ];
 359       $fields['last_imported'] = [
 360         'type' => 'int',
 361         'unsigned' => TRUE,
 362         'not null' => TRUE,
 363         'default' => 0,
 364         'description' => 'UNIX timestamp of the last time this row was imported',
 365       ];
 366       $fields['hash'] = [
 367         'type' => 'varchar',
 368         'length' => '64',
 369         'not null' => FALSE,
 370         'description' => 'Hash of source row data, for detecting changes',
 371       ];
 372       $schema = [
 373         'description' => 'Mappings from source identifier value(s) to destination identifier value(s).',
 374         'fields' => $fields,
 375         'primary key' => [static::SOURCE_IDS_HASH],
 376         'indexes' => $indexes,
 377       ];
 378       $this->getDatabase()->schema()->createTable($this->mapTableName, $schema);
 379
 380       // Now do the message table.
 381       if (!$this->getDatabase()->schema()->tableExists($this->messageTableName())) {
 382         $fields = [];
 383         $fields['msgid'] = [
 384           'type' => 'serial',
 385           'unsigned' => TRUE,
 386           'not null' => TRUE,
 387         ];
 388         $fields += $source_ids_hash;
 389
 390         $fields['level'] = [
 391           'type' => 'int',
 392           'unsigned' => TRUE,
 393           'not null' => TRUE,
 394           'default' => 1,
 395         ];
 396         $fields['message'] = [
 397           'type' => 'text',
 398           'size' => 'medium',
 399           'not null' => TRUE,
 400         ];
 401         $schema = [
 402           'description' => 'Messages generated during a migration process',
 403           'fields' => $fields,
 404           'primary key' => ['msgid'],
 405         ];
 406         $this->getDatabase()->schema()->createTable($this->messageTableName(), $schema);
 407       }
 408     }
 409     else {
 410       // Add any missing columns to the map table.
 411       if (!$this->getDatabase()->schema()->fieldExists($this->mapTableName,
 412                                                     'rollback_action')) {
 413         $this->getDatabase()->schema()->addField($this->mapTableName, 'rollback_action',
 414           [
 415             'type' => 'int',
 416             'size' => 'tiny',
 417             'unsigned' => TRUE,
 418             'not null' => TRUE,
 419             'default' => 0,
 420             'description' => 'Flag indicating what to do for this item on rollback',
 421           ]
 422         );
 423       }
 424       if (!$this->getDatabase()->schema()->fieldExists($this->mapTableName, 'hash')) {
 425         $this->getDatabase()->schema()->addField($this->mapTableName, 'hash',
 426           [
 427             'type' => 'varchar',
 428             'length' => '64',
 429             'not null' => FALSE,
 430             'description' => 'Hash of source row data, for detecting changes',
 431           ]
 432         );
 433       }
 434       if (!$this->getDatabase()->schema()->fieldExists($this->mapTableName, static::SOURCE_IDS_HASH)) {
 435         $this->getDatabase()->schema()->addField($this->mapTableName, static::SOURCE_IDS_HASH, [
 436           'type' => 'varchar',
 437           'length' => '64',
 438           'not null' => TRUE,
 439           'description' => 'Hash of source ids. Used as primary key',
 440         ]);
 441       }
 442     }
 443   }
 444
 445   /**
 446    * Creates schema from an ID definition.
 447    *
 448    * @param array $id_definition
 449    *   The definition of the field having the structure as the items returned by
 450    *   MigrateSourceInterface or MigrateDestinationInterface::getIds().
 451    *
 452    * @return array
 453    *   The database schema definition.
 454    *
 455    * @see \Drupal\migrate\Plugin\MigrateSourceInterface::getIds()
 456    * @see \Drupal\migrate\Plugin\MigrateDestinationInterface::getIds()
 457    */
 458   protected function getFieldSchema(array $id_definition) {
 459     $type_parts = explode('.', $id_definition['type']);
 460     if (count($type_parts) == 1) {
 461       $type_parts[] = 'value';
 462     }
 463     unset($id_definition['type']);
 464
 465     // Get the field storage definition.
 466     $definition = BaseFieldDefinition::create($type_parts[0]);
 467
 468     // Get a list of setting keys belonging strictly to the field definition.
 469     $default_field_settings = $definition->getSettings();
 470     // Separate field definition settings from custom settings. Custom settings
 471     // are settings passed in $id_definition that are not part of field storage
 472     // definition settings.
 473     $field_settings = array_intersect_key($id_definition, $default_field_settings);
 474     $custom_settings = array_diff_key($id_definition, $default_field_settings);
 475
 476     // Resolve schema from field storage definition settings.
 477     $schema = $definition
 478       ->setSettings($field_settings)
 479       ->getColumns()[$type_parts[1]];
 480
 481     // Merge back custom settings.
 482     return $schema + $custom_settings;
 483   }
 484
 485   /**
 486    * {@inheritdoc}
 487    */
 488   public function getRowBySource(array $source_id_values) {
 489     $query = $this->getDatabase()->select($this->mapTableName(), 'map')
 490       ->fields('map');
 491     $query->condition(static::SOURCE_IDS_HASH, $this->getSourceIDsHash($source_id_values));
 492     $result = $query->execute();
 493     return $result->fetchAssoc();
 494   }
 495
 496   /**
 497    * {@inheritdoc}
 498    */
 499   public function getRowByDestination(array $destination_id_values) {
 500     $query = $this->getDatabase()->select($this->mapTableName(), 'map')
 501       ->fields('map');
 502     foreach ($this->destinationIdFields() as $field_name => $destination_id) {
 503       $query->condition("map.$destination_id", $destination_id_values[$field_name], '=');
 504     }
 505     $result = $query->execute();
 506     return $result->fetchAssoc();
 507   }
 508
 509   /**
 510    * {@inheritdoc}
 511    */
 512   public function getRowsNeedingUpdate($count) {
 513     $rows = [];
 514     $result = $this->getDatabase()->select($this->mapTableName(), 'map')
 515       ->fields('map')
 516       ->condition('source_row_status', MigrateIdMapInterface::STATUS_NEEDS_UPDATE)
 517       ->range(0, $count)
 518       ->execute();
 519     foreach ($result as $row) {
 520       $rows[] = $row;
 521     }
 522     return $rows;
 523   }
 524
 525   /**
 526    * {@inheritdoc}
 527    */
 528   public function lookupSourceID(array $destination_id_values) {
 529     $source_id_fields = $this->sourceIdFields();
 530     $query = $this->getDatabase()->select($this->mapTableName(), 'map');
 531     foreach ($source_id_fields as $source_field_name => $idmap_field_name) {
 532       $query->addField('map', $idmap_field_name, $source_field_name);
 533     }
 534     foreach ($this->destinationIdFields() as $field_name => $destination_id) {
 535       $query->condition("map.$destination_id", $destination_id_values[$field_name], '=');
 536     }
 537     $result = $query->execute();
 538     return $result->fetchAssoc() ?: [];
 539   }
 540
 541   /**
 542    * {@inheritdoc}
 543    */
 544   public function lookupDestinationId(array $source_id_values) {
 545     $results = $this->lookupDestinationIds($source_id_values);
 546     return $results ? reset($results) : [];
 547   }
 548
 549   /**
 550    * {@inheritdoc}
 551    */
 552   public function lookupDestinationIds(array $source_id_values) {
 553     if (empty($source_id_values)) {
 554       return [];
 555     }
 556
 557     // Canonicalize the keys into a hash of DB-field => value.
 558     $is_associative = !isset($source_id_values[0]);
 559     $conditions = [];
 560     foreach ($this->sourceIdFields() as $field_name => $db_field) {
 561       if ($is_associative) {
 562         // Associative $source_id_values can have fields out of order.
 563         if (isset($source_id_values[$field_name])) {
 564           $conditions[$db_field] = $source_id_values[$field_name];
 565           unset($source_id_values[$field_name]);
 566         }
 567       }
 568       else {
 569         // For non-associative $source_id_values, we assume they're the first
 570         // few fields.
 571         if (empty($source_id_values)) {
 572           break;
 573         }
 574         $conditions[$db_field] = array_shift($source_id_values);
 575       }
 576     }
 577
 578     if (!empty($source_id_values)) {
 579       throw new MigrateException("Extra unknown items in source IDs");
 580     }
 581
 582     $query = $this->getDatabase()->select($this->mapTableName(), 'map')
 583       ->fields('map', $this->destinationIdFields());
 584     if (count($this->sourceIdFields()) === count($conditions)) {
 585       // Optimization: Use the primary key.
 586       $query->condition(self::SOURCE_IDS_HASH, $this->getSourceIDsHash(array_values($conditions)));
 587     }
 588     else {
 589       foreach ($conditions as $db_field => $value) {
 590         $query->condition($db_field, $value);
 591       }
 592     }
 593
 594     return $query->execute()->fetchAll(\PDO::FETCH_NUM);
 595   }
 596
 597   /**
 598    * {@inheritdoc}
 599    */
 600   public function saveIdMapping(Row $row, array $destination_id_values, $source_row_status = MigrateIdMapInterface::STATUS_IMPORTED, $rollback_action = MigrateIdMapInterface::ROLLBACK_DELETE) {
 601     // Construct the source key.
 602     $source_id_values = $row->getSourceIdValues();
 603     // Construct the source key and initialize to empty variable keys.
 604     $fields = [];
 605     foreach ($this->sourceIdFields() as $field_name => $key_name) {
 606       // A NULL key value is usually an indication of a problem.
 607       if (!isset($source_id_values[$field_name])) {
 608         $this->message->display($this->t(
 609           'Did not save to map table due to NULL value for key field @field',
 610           ['@field' => $field_name]), 'error');
 611         return;
 612       }
 613       $fields[$key_name] = $source_id_values[$field_name];
 614     }
 615
 616     if (!$fields) {
 617       return;
 618     }
 619
 620     $fields += [
 621       'source_row_status' => (int) $source_row_status,
 622       'rollback_action' => (int) $rollback_action,
 623       'hash' => $row->getHash(),
 624     ];
 625     $count = 0;
 626     foreach ($destination_id_values as $dest_id) {
 627       $fields['destid' . ++$count] = $dest_id;
 628     }
 629     if ($count && $count != count($this->destinationIdFields())) {
 630       $this->message->display(t('Could not save to map table due to missing destination id values'), 'error');
 631       return;
 632     }
 633     if ($this->migration->getTrackLastImported()) {
 634       $fields['last_imported'] = time();
 635     }
 636     $keys = [static::SOURCE_IDS_HASH => $this->getSourceIDsHash($source_id_values)];
 637     // Notify anyone listening of the map row we're about to save.
 638     $this->eventDispatcher->dispatch(MigrateEvents::MAP_SAVE, new MigrateMapSaveEvent($this, $fields));
 639     $this->getDatabase()->merge($this->mapTableName())
 640       ->key($keys)
 641       ->fields($fields)
 642       ->execute();
 643   }
 644
 645   /**
 646    * {@inheritdoc}
 647    */
 648   public function saveMessage(array $source_id_values, $message, $level = MigrationInterface::MESSAGE_ERROR) {
 649     foreach ($this->sourceIdFields() as $field_name => $source_id) {
 650       // If any key value is not set, we can't save.
 651       if (!isset($source_id_values[$field_name])) {
 652         return;
 653       }
 654     }
 655     $fields[static::SOURCE_IDS_HASH] = $this->getSourceIDsHash($source_id_values);
 656     $fields['level'] = $level;
 657     $fields['message'] = $message;
 658     $this->getDatabase()->insert($this->messageTableName())
 659       ->fields($fields)
 660       ->execute();
 661
 662     // Notify anyone listening of the message we've saved.
 663     $this->eventDispatcher->dispatch(MigrateEvents::IDMAP_MESSAGE,
 664       new MigrateIdMapMessageEvent($this->migration, $source_id_values, $message, $level));
 665   }
 666
 667   /**
 668    * {@inheritdoc}
 669    */
 670   public function getMessageIterator(array $source_id_values = [], $level = NULL) {
 671     $query = $this->getDatabase()->select($this->messageTableName(), 'msg')
 672       ->fields('msg');
 673     if ($source_id_values) {
 674       $query->condition(static::SOURCE_IDS_HASH, $this->getSourceIDsHash($source_id_values));
 675     }
 676
 677     if ($level) {
 678       $query->condition('level', $level);
 679     }
 680     return $query->execute();
 681   }
 682
 683   /**
 684    * {@inheritdoc}
 685    */
 686   public function prepareUpdate() {
 687     $this->getDatabase()->update($this->mapTableName())
 688       ->fields(['source_row_status' => MigrateIdMapInterface::STATUS_NEEDS_UPDATE])
 689       ->execute();
 690   }
 691
 692   /**
 693    * {@inheritdoc}
 694    */
 695   public function processedCount() {
 696     return $this->getDatabase()->select($this->mapTableName())
 697       ->countQuery()
 698       ->execute()
 699       ->fetchField();
 700   }
 701
 702   /**
 703    * {@inheritdoc}
 704    */
 705   public function importedCount() {
 706     return $this->getDatabase()->select($this->mapTableName())
 707       ->condition('source_row_status', [MigrateIdMapInterface::STATUS_IMPORTED, MigrateIdMapInterface::STATUS_NEEDS_UPDATE], 'IN')
 708       ->countQuery()
 709       ->execute()
 710       ->fetchField();
 711   }
 712
 713   /**
 714    * {@inheritdoc}
 715    */
 716   public function updateCount() {
 717     return $this->countHelper(MigrateIdMapInterface::STATUS_NEEDS_UPDATE);
 718   }
 719
 720   /**
 721    * {@inheritdoc}
 722    */
 723   public function errorCount() {
 724     return $this->countHelper(MigrateIdMapInterface::STATUS_FAILED);
 725   }
 726
 727   /**
 728    * {@inheritdoc}
 729    */
 730   public function messageCount() {
 731     return $this->countHelper(NULL, $this->messageTableName());
 732   }
 733
 734   /**
 735    * Counts records in a table.
 736    *
 737    * @param int $status
 738    *   An integer for the source_row_status column.
 739    * @param string $table
 740    *   (optional) The table to work. Defaults to NULL.
 741    *
 742    * @return int
 743    *   The number of records.
 744    */
 745   protected function countHelper($status, $table = NULL) {
 746     $query = $this->getDatabase()->select($table ?: $this->mapTableName());
 747     if (isset($status)) {
 748       $query->condition('source_row_status', $status);
 749     }
 750     return $query->countQuery()->execute()->fetchField();
 751   }
 752
 753   /**
 754    * {@inheritdoc}
 755    */
 756   public function delete(array $source_id_values, $messages_only = FALSE) {
 757     if (empty($source_id_values)) {
 758       throw new MigrateException('Without source identifier values it is impossible to find the row to delete.');
 759     }
 760
 761     if (!$messages_only) {
 762       $map_query = $this->getDatabase()->delete($this->mapTableName());
 763       $map_query->condition(static::SOURCE_IDS_HASH, $this->getSourceIDsHash($source_id_values));
 764       // Notify anyone listening of the map row we're about to delete.
 765       $this->eventDispatcher->dispatch(MigrateEvents::MAP_DELETE, new MigrateMapDeleteEvent($this, $source_id_values));
 766       $map_query->execute();
 767     }
 768     $message_query = $this->getDatabase()->delete($this->messageTableName());
 769     $message_query->condition(static::SOURCE_IDS_HASH, $this->getSourceIDsHash($source_id_values));
 770     $message_query->execute();
 771   }
 772
 773   /**
 774    * {@inheritdoc}
 775    */
 776   public function deleteDestination(array $destination_id_values) {
 777     $map_query = $this->getDatabase()->delete($this->mapTableName());
 778     $message_query = $this->getDatabase()->delete($this->messageTableName());
 779     $source_id_values = $this->lookupSourceID($destination_id_values);
 780     if (!empty($source_id_values)) {
 781       foreach ($this->destinationIdFields() as $field_name => $destination_id) {
 782         $map_query->condition($destination_id, $destination_id_values[$field_name]);
 783       }
 784       // Notify anyone listening of the map row we're about to delete.
 785       $this->eventDispatcher->dispatch(MigrateEvents::MAP_DELETE, new MigrateMapDeleteEvent($this, $source_id_values));
 786       $map_query->execute();
 787
 788       $message_query->condition(static::SOURCE_IDS_HASH, $this->getSourceIDsHash($source_id_values));
 789       $message_query->execute();
 790     }
 791   }
 792
 793   /**
 794    * {@inheritdoc}
 795    */
 796   public function setUpdate(array $source_id_values) {
 797     if (empty($source_id_values)) {
 798       throw new MigrateException('No source identifiers provided to update.');
 799     }
 800     $query = $this->getDatabase()
 801       ->update($this->mapTableName())
 802       ->fields(['source_row_status' => MigrateIdMapInterface::STATUS_NEEDS_UPDATE]);
 803
 804     foreach ($this->sourceIdFields() as $field_name => $source_id) {
 805       $query->condition($source_id, $source_id_values[$field_name]);
 806     }
 807     $query->execute();
 808   }
 809
 810   /**
 811    * {@inheritdoc}
 812    */
 813   public function clearMessages() {
 814     $this->getDatabase()->truncate($this->messageTableName())->execute();
 815   }
 816
 817   /**
 818    * {@inheritdoc}
 819    */
 820   public function destroy() {
 821     $this->getDatabase()->schema()->dropTable($this->mapTableName());
 822     $this->getDatabase()->schema()->dropTable($this->messageTableName());
 823   }
 824
 825   /**
 826    * Implementation of \Iterator::rewind().
 827    *
 828    * This is called before beginning a foreach loop.
 829    */
 830   public function rewind() {
 831     $this->currentRow = NULL;
 832     $fields = [];
 833     foreach ($this->sourceIdFields() as $field) {
 834       $fields[] = $field;
 835     }
 836     foreach ($this->destinationIdFields() as $field) {
 837       $fields[] = $field;
 838     }
 839     $this->result = $this->getDatabase()->select($this->mapTableName(), 'map')
 840       ->fields('map', $fields)
 841       ->orderBy('destid1')
 842       ->execute();
 843     $this->next();
 844   }
 845
 846   /**
 847    * Implementation of \Iterator::current().
 848    *
 849    * This is called when entering a loop iteration, returning the current row.
 850    */
 851   public function current() {
 852     return $this->currentRow;
 853   }
 854
 855   /**
 856    * Implementation of \Iterator::key().
 857    *
 858    * This is called when entering a loop iteration, returning the key of the
 859    * current row. It must be a scalar - we will serialize to fulfill the
 860    * requirement, but using getCurrentKey() is preferable.
 861    */
 862   public function key() {
 863     return serialize($this->currentKey);
 864   }
 865
 866   /**
 867    * {@inheritdoc}
 868    */
 869   public function currentDestination() {
 870     if ($this->valid()) {
 871       $result = [];
 872       foreach ($this->destinationIdFields() as $destination_field_name => $idmap_field_name) {
 873         if (!is_null($this->currentRow[$idmap_field_name])) {
 874           $result[$destination_field_name] = $this->currentRow[$idmap_field_name];
 875         }
 876       }
 877       return $result;
 878     }
 879     else {
 880       return NULL;
 881     }
 882   }
 883
 884   /**
 885    * @inheritdoc
 886    */
 887   public function currentSource() {
 888     if ($this->valid()) {
 889       $result = [];
 890       foreach ($this->sourceIdFields() as $field_name => $source_id) {
 891         $result[$field_name] = $this->currentKey[$source_id];
 892       }
 893       return $result;
 894     }
 895     else {
 896       return NULL;
 897     }
 898   }
 899
 900   /**
 901    * Implementation of \Iterator::next().
 902    *
 903    * This is called at the bottom of the loop implicitly, as well as explicitly
 904    * from rewind().
 905    */
 906   public function next() {
 907     $this->currentRow = $this->result->fetchAssoc();
 908     $this->currentKey = [];
 909     if ($this->currentRow) {
 910       foreach ($this->sourceIdFields() as $map_field) {
 911         $this->currentKey[$map_field] = $this->currentRow[$map_field];
 912         // Leave only destination fields.
 913         unset($this->currentRow[$map_field]);
 914       }
 915     }
 916   }
 917
 918   /**
 919    * Implementation of \Iterator::valid().
 920    *
 921    * This is called at the top of the loop, returning TRUE to process the loop
 922    * and FALSE to terminate it.
 923    */
 924   public function valid() {
 925     return $this->currentRow !== FALSE;
 926   }
 927
 928 }