Fixed duplicated ids to index by mview changelog

app/code/Magento/Eav/Model/Mview/ChangeLogBatchWalker/IdsFetcher.php

@@ -0,0 +1,36 @@
+<?php declare(strict_types=1);
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+
+namespace Magento\Eav\Model\Mview\ChangeLogBatchWalker;
+
+use Magento\Framework\App\ResourceConnection;
+use Magento\Framework\DB\Select;
+use Magento\Framework\Mview\View\ChangeLogBatchWalker\IdsFetcherInterface;
+
+class IdsFetcher implements IdsFetcherInterface
+{
+    /**
+     * @var \Magento\Framework\App\ResourceConnection
+     */
+    private ResourceConnection $resourceConnection;
+
+    /**
+     * @param \Magento\Framework\App\ResourceConnection $resourceConnection
+     */
+    public function __construct(
+        ResourceConnection $resourceConnection
+    ) {
+        $this->resourceConnection = $resourceConnection;
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function fetch(Select $select): array
+    {
+        return $this->resourceConnection->getConnection()->fetchAll($select);
+    }
+}

app/code/Magento/Eav/Model/Mview/ChangeLogBatchWalker.php → app/code/Magento/Eav/Model/Mview/ChangeLogBatchWalker/IdsSelectBuilder.php

@@ -1,47 +1,66 @@
-<?php
+<?php declare(strict_types=1);
 /**
  * Copyright © Magento, Inc. All rights reserved.
  * See COPYING.txt for license details.
  */
 
-namespace Magento\Eav\Model\Mview;
+namespace Magento\Eav\Model\Mview\ChangeLogBatchWalker;
 
 use Magento\Framework\App\ResourceConnection;
+use Magento\Framework\DB\Select;
 use Magento\Framework\DB\Sql\Expression;
-use Magento\Framework\Mview\View\ChangeLogBatchWalkerInterface;
+use Magento\Framework\Mview\View\ChangeLogBatchWalker\IdsSelectBuilderInterface;
 use Magento\Framework\Mview\View\ChangelogInterface;
 
-/**
- * Class BatchIterator
- */
-class ChangeLogBatchWalker implements ChangeLogBatchWalkerInterface
+class IdsSelectBuilder implements IdsSelectBuilderInterface
 {
     private const GROUP_CONCAT_MAX_VARIABLE = 'group_concat_max_len';
     /** ID is defined as small int. Default size of it is 5 */
     private const DEFAULT_ID_SIZE = 5;
-
     /**
-     * @var ResourceConnection
+     * @var \Magento\Framework\App\ResourceConnection
      */
-    private $resourceConnection;
-
+    private ResourceConnection $resourceConnection;
     /**
      * @var array
      */
-    private $entityTypeCodes;
+    private array $entityTypeCodes;
 
     /**
-     * @param ResourceConnection $resourceConnection
+     * @param \Magento\Framework\App\ResourceConnection $resourceConnection
      * @param array $entityTypeCodes
      */
     public function __construct(
         ResourceConnection $resourceConnection,
-        array $entityTypeCodes = []
+        array              $entityTypeCodes = []
     ) {
         $this->resourceConnection = $resourceConnection;
         $this->entityTypeCodes = $entityTypeCodes;
     }
 
+    /**
+     * @inheritdoc
+     */
+    public function build(ChangelogInterface $changelog): Select
+    {
+        $numberOfAttributes = $this->calculateEavAttributeSize($changelog);
+        $this->setGroupConcatMax($numberOfAttributes);
+
+        $changelogTableName = $this->resourceConnection->getTableName($changelog->getName());
+
+        $connection = $this->resourceConnection->getConnection();
+
+        $columns = [
+            $changelog->getColumnName(),
+            'attribute_ids' => new Expression('GROUP_CONCAT(attribute_id)'),
+            'store_id'
+        ];
+
+        return $connection->select()
+            ->from($changelogTableName, $columns)
+            ->group([$changelog->getColumnName(), 'store_id']);
+    }
+
     /**
      * Calculate EAV attributes size
      *
@@ -54,7 +73,7 @@ private function calculateEavAttributeSize(ChangelogInterface $changelog): int
         $connection = $this->resourceConnection->getConnection();
 
         if (!isset($this->entityTypeCodes[$changelog->getViewId()])) {
-            throw new \Exception('Entity type for view was not defined');
+            throw new \InvalidArgumentException('Entity type for view was not defined');
         }
 
         $select = $connection->select();
@@ -63,12 +82,12 @@ private function calculateEavAttributeSize(ChangelogInterface $changelog): int
             new Expression('COUNT(*)')
         )
             ->joinInner(
-              ['type' => $connection->getTableName('eav_entity_type')],
+                ['type' => $connection->getTableName('eav_entity_type')],
                 'type.entity_type_id=eav_attribute.entity_type_id'
             )
             ->where('type.entity_type_code = ?', $this->entityTypeCodes[$changelog->getViewId()]);
 
-        return (int) $connection->fetchOne($select);
+        return (int)$connection->fetchOne($select);
     }
 
     /**
@@ -87,34 +106,4 @@ private function setGroupConcatMax(int $numberOfAttributes): void
             $numberOfAttributes * (self::DEFAULT_ID_SIZE + 1)
         ));
     }
-
-    /**
-     * @inheritdoc
-     * @throws \Exception
-     */
-    public function walk(ChangelogInterface $changelog, int $fromVersionId, int $toVersion, int $batchSize)
-    {
-        $connection = $this->resourceConnection->getConnection();
-        $numberOfAttributes = $this->calculateEavAttributeSize($changelog);
-        $this->setGroupConcatMax($numberOfAttributes);
-        $select = $connection->select()->distinct(true)
-            ->where(
-                'version_id > ?',
-                (int) $fromVersionId
-            )
-            ->where(
-                'version_id <= ?',
-                $toVersion
-            )
-            ->group([$changelog->getColumnName(), 'store_id'])
-            ->limit($batchSize);
-
-        $columns = [
-            $changelog->getColumnName(),
-            'attribute_ids' => new Expression('GROUP_CONCAT(attribute_id)'),
-            'store_id'
-        ];
-        $select->from($changelog->getName(), $columns);
-        return $connection->fetchAll($select);
-    }
 }

app/code/Magento/Eav/Model/Mview/ChangeLogBatchWalker/IdsTableBuilder.php

@@ -0,0 +1,50 @@
+<?php declare(strict_types=1);
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+
+namespace Magento\Eav\Model\Mview\ChangeLogBatchWalker;
+
+use Magento\Framework\DB\Adapter\AdapterInterface;
+use Magento\Framework\DB\Ddl\Table;
+use Magento\Framework\Mview\View\ChangeLogBatchWalker\IdsTableBuilder as BaseIdsTableBuilder;
+use Magento\Framework\Mview\View\ChangelogInterface;
+
+class IdsTableBuilder extends BaseIdsTableBuilder
+{
+    /**
+     * @inheritdoc
+     */
+    public function build(ChangelogInterface $changelog): Table
+    {
+        $table = parent::build($changelog);
+        $table->addColumn(
+            'attribute_ids',
+            Table::TYPE_TEXT,
+            null,
+            ['unsigned' => true, 'nullable' => false],
+            'Attribute IDs'
+        );
+        $table->addColumn(
+            'store_id',
+            Table::TYPE_INTEGER,
+            null,
+            ['unsigned' => true, 'nullable' => false],
+            'Store ID'
+        );
+        $table->addIndex(
+            self::INDEX_NAME_UNIQUE,
+            [
+                $changelog->getColumnName(),
+                'attribute_ids',
+                'store_id'
+            ],
+            [
+                'type' => AdapterInterface::INDEX_TYPE_UNIQUE
+            ]
+        );
+
+        return $table;
+    }
+}

app/code/Magento/Eav/etc/di.xml

@@ -221,4 +221,11 @@
             <argument name="cache" xsi:type="object">configured_eav_cache</argument>
         </arguments>
     </type>
+    <virtualType name="Magento\Eav\Model\Mview\ChangeLogBatchWalker" type="Magento\Framework\Mview\View\ChangeLogBatchWalker">
+        <arguments>
+            <argument name="idsTableBuilder" xsi:type="object">Magento\Eav\Model\Mview\ChangeLogBatchWalker\IdsTableBuilder</argument>
+            <argument name="idsSelectBuilder" xsi:type="object">Magento\Eav\Model\Mview\ChangeLogBatchWalker\IdsSelectBuilder</argument>
+            <argument name="idsFetcher" xsi:type="object">Magento\Eav\Model\Mview\ChangeLogBatchWalker\IdsFetcher</argument>
+        </arguments>
+    </virtualType>
 </config>

lib/internal/Magento/Framework/Mview/Test/Unit/ViewTest.php

@@ -8,6 +8,7 @@
 
 namespace Magento\Framework\Mview\Test\Unit;
 
+use ArrayIterator;
 use Magento\Framework\Mview\ActionFactory;
 use Magento\Framework\Mview\ActionInterface;
 use Magento\Framework\Mview\ConfigInterface;
@@ -365,7 +366,7 @@ public function testUpdate()
         );
 
         $actionMock = $this->getMockForAbstractClass(ActionInterface::class);
-        $this->iteratorMock->expects($this->once())->method('walk')->willReturn($listId);
+        $this->iteratorMock->expects($this->once())->method('walk')->willReturn(new ArrayIterator([$listId]));
         $actionMock->expects($this->once())->method('execute')->with($listId)->willReturnSelf();
         $this->actionFactoryMock->expects(
             $this->once()
@@ -411,7 +412,10 @@ public function testUpdateEx(): void
             ->expects($this->once())
             ->method('getVersion')
             ->willReturn($currentVersionId);
-        $this->iteratorMock->expects($this->any())->method('walk')->willReturn($this->generateChangeLog(150, 1, 150));
+        $this->iteratorMock
+            ->expects($this->any())
+            ->method('walk')
+            ->willReturn(new ArrayIterator([$this->generateChangeLog(150, 1, 150)]));
         $this->changelogMock->method('getList')
             ->willReturnMap(
                 [
@@ -495,7 +499,7 @@ public function testUpdateWithException()
         );
         $this->iteratorMock->expects($this->any())
             ->method('walk')
-            ->willReturn([2, 3]);
+            ->willReturn(new ArrayIterator([[2, 3]]));
 
         $actionMock = $this->createPartialMock(ActionInterface::class, ['execute']);
         $actionMock->expects($this->once())->method('execute')->with($listId)->willReturnCallback(

lib/internal/Magento/Framework/Mview/View.php

@@ -302,15 +302,9 @@ private function executeAction(ActionInterface $action, int $lastVersionId, int
             ? (int) $this->changelogBatchSize[$this->getChangelog()->getViewId()]
             : self::DEFAULT_BATCH_SIZE;
 
-        $vsFrom = $lastVersionId;
-        while ($vsFrom < $currentVersionId) {
-            $walker = $this->getWalker();
-            $ids = $walker->walk($this->getChangelog(), $vsFrom, $currentVersionId, $batchSize);
+        $batches = $this->getWalker()->walk($this->getChangelog(), $lastVersionId, $currentVersionId, $batchSize);
 
-            if (empty($ids)) {
-                break;
-            }
-            $vsFrom += $batchSize;
+        foreach ($batches as $ids) {
             $action->execute($ids);
         }
     }