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