function SqlBase::initializeIterator

Overrides SourcePluginBase::initializeIterator

File

core/modules/migrate/src/Plugin/migrate/source/SqlBase.php, line 249

Class

SqlBase

Sources whose data may be fetched via a database connection.

Namespace

Code

protected function initializeIterator() {
    // Initialize the batch size.
    if ($this->batchSize == 0 && isset($this->configuration['batch_size'])) {
        // Valid batch sizes are integers >= 0.
        if (is_int($this->configuration['batch_size']) && $this->configuration['batch_size'] >= 0) {
            $this->batchSize = $this->configuration['batch_size'];
        }
        else {
            throw new MigrateException("batch_size must be greater than or equal to zero");
        }
    }
    // If a batch has run the query is already setup.
    if ($this->batch == 0) {
        $this->prepareQuery();
        // The rules for determining what conditions to add to the query are as
        // follows (applying first applicable rule):
        // 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;
        $added_fields = [];
        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'] . '.' . $this->query
                        ->escapeField($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;
            // 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}");
                $added_fields[] = "{$alias}.{$map_key}";
            }
            if ($n = count($this->migration
                ->getDestinationIds())) {
                for ($count = 1; $count <= $n; $count++) {
                    $map_key = 'destid' . $count++;
                    $this->query
                        ->addField($alias, $map_key, "migrate_map_{$map_key}");
                    $added_fields[] = "{$alias}.{$map_key}";
                }
            }
            $this->query
                ->addField($alias, 'source_row_status', 'migrate_map_source_row_status');
            $added_fields[] = "{$alias}.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 ($this->getHighWaterProperty()) {
            $high_water_field = $this->getHighWaterField();
            $high_water = $this->getHighWater();
            // We check against NULL because 0 is an acceptable value for the high
            // water mark.
            if ($high_water !== NULL) {
                $conditions->condition($high_water_field, $high_water, '>');
                $condition_added = TRUE;
            }
            // Always sort by the high water field, to ensure that the first run
            // (before we have a high water value) also has the results in a
            // consistent order.
            $this->query
                ->orderBy($high_water_field);
        }
        if ($condition_added) {
            $this->query
                ->condition($conditions);
        }
        // If the query has a group by, our added fields need it too, to keep the
        // query valid.
        // @see https://dev.mysql.com/doc/refman/8.0/en/group-by-handling.html
        $group_by = $this->query
            ->getGroupBy();
        if ($group_by && $added_fields) {
            foreach ($added_fields as $added_field) {
                $this->query
                    ->groupBy($added_field);
            }
        }
    }
    // Download data in batches for performance.
    if ($this->batchSize > 0) {
        $this->query
            ->range($this->batch * $this->batchSize, $this->batchSize);
    }
    $statement = $this->query
        ->execute();
    $statement->setFetchMode(\PDO::FETCH_ASSOC);
    return new \IteratorIterator($statement);
}