feat: cleanup of table data is now an external service

Beacuse the clean operation was producing a deadlock on table metadata it was decided to move cleanup on main thread with consecutive table cleanup. Moving changes to main thread allowed to un-block full pottential of multiple workers to import data in parallel

Author: IvanChepurnyi

src/Command/ImportCommand.php

@@ -98,8 +98,13 @@ public function execute(InputInterface $input, OutputInterface $output): int
         )->create();
 
         $inputSource = new JsonImportSourceFactory($input->getArgument('directory'), $config);
+        $tables = $inputSource->listTables();
 
-        foreach ($inputSource->listTables() as $table) {
+        foreach ($tables as $table) {
+            $importer->cleanTable($table);
+        }
+
+        foreach ($tables as $table) {
             $importer->importTable($table, $progressNotifier);
         }
 

src/Export/TableListService.php

@@ -11,6 +11,7 @@
 
 use EcomDev\MySQL2JSONL\Configuration;
 use EcomDev\MySQL2JSONL\TableEntry;
+use PDO;
 
 final readonly class TableListService
 {
@@ -21,12 +22,12 @@ public function __construct(private Configuration $configuration)
     public function tablesToExport(): array
     {
         $connection = $this->configuration->createPDOConnection();
-        $connection->prepare('SET information_schema_stats_expiry=0')->execute();
 
         $result = $connection->prepare(
             'SELECT TABLE_NAME, TABLE_ROWS FROM information_schema.tables'
             . ' WHERE TABLE_SCHEMA = SCHEMA() AND TABLE_TYPE = ?'
         );
+
         $result->execute(['BASE TABLE']);
 
         $tables = [];

src/Import/BlockingImportCleanupService.php

@@ -0,0 +1,39 @@
+<?php
+
+/**
+ * Copyright © EcomDev B.V. All rights reserved.
+ * See LICENSE for license details.
+ */
+
+declare(strict_types=1);
+
+namespace EcomDev\MySQL2JSONL\Import;
+
+use EcomDev\MySQL2JSONL\Configuration;
+use EcomDev\MySQL2JSONL\TableEntry;
+use PDO;
+
+final readonly class BlockingImportCleanupService implements ImportCleanupService
+{
+    private function __construct(
+        private ImportMode $importMode,
+        private PDO $connection,
+    ) {
+    }
+
+    public static function create(
+        ImportMode $importMode,
+        PDO $connection
+    ): BlockingImportCleanupService {
+        return new self($importMode, $connection);
+    }
+
+    public function cleanTable(TableEntry $table): void
+    {
+        if ($this->importMode->isTruncate()) {
+            $this->connection->query(
+                sprintf('TRUNCATE `%s`', $table->name)
+            );
+        }
+    }
+}

src/Import/BlockingImportService.php

@@ -7,13 +7,13 @@
 use EcomDev\MySQL2JSONL\TableEntry;
 use PDO;
 
-final readonly class BlockingImportService implements ImportService
+final readonly class BlockingImportService implements ImportService, ImportCleanupService
 {
     public function __construct(
         private PDO $connection,
         private InsertOnDuplicate $insertOnDuplicate,
         private ImportSourceFactory $sourceFactory,
-        private ImportMode $mode,
+        private BlockingImportCleanupService $cleanupService,
         private int $batchSize
     ) {
     }
@@ -24,10 +24,6 @@ public function importTable(TableEntry $table, ProgressNotifier $notifier): void
 
         $columns = $source->header();
 
-        if ($this->mode->isTruncate()) {
-            $this->connection->prepare(sprintf('TRUNCATE TABLE `%s`', $table->name))->execute();
-        }
-
         $statementCache = InsertStatementCache::create(
             $this->connection,
             $this->insertOnDuplicate,
@@ -61,4 +57,9 @@ public function importTable(TableEntry $table, ProgressNotifier $notifier): void
         $this->connection->commit();
         $notifier->finish($table->name);
     }
+
+    public function cleanTable(TableEntry $table): void
+    {
+        $this->cleanupService->cleanTable($table);
+    }
 }

src/Import/BlockingImportServiceFactory.php

@@ -19,14 +19,18 @@ public static function createFromConfiguration(Configuration $configuration, Imp
         return new self($configuration, $importSourceFactory, new InsertOnDuplicate());
     }
 
-    public function create(): ImportService
+    public function create(): BlockingImportService
     {
+        $connection = $this->configuration->createPDOConnection();
         return new BlockingImportService(
-            $this->configuration->createPDOConnection(),
+            $connection,
             $this->insertOnDuplicate,
             $this->importSourceFactory,
-            $this->configuration->importMode,
-            $this->configuration->batchSize
+            BlockingImportCleanupService::create(
+                $this->configuration->importMode,
+                $connection,
+            ),
+            $this->configuration->batchSize,
         );
     }
 }

src/Import/ImportCleanupService.php

@@ -0,0 +1,20 @@
+<?php
+
+/**
+ * Copyright © EcomDev B.V. All rights reserved.
+ * See LICENSE for license details.
+ */
+
+declare(strict_types=1);
+
+namespace EcomDev\MySQL2JSONL\Import;
+
+use EcomDev\MySQL2JSONL\TableEntry;
+
+/**
+ * Service to clean-up table data on import
+ */
+interface ImportCleanupService
+{
+    public function cleanTable(TableEntry $table): void;
+}

src/Import/WorkerImportService.php

@@ -18,13 +18,14 @@
 
 use function Amp\async;
 
-final readonly class WorkerImportService implements ImportService
+final readonly class WorkerImportService implements ImportService, ImportCleanupService
 {
     public function __construct(
         private Configuration $configuration,
         private WorkerPool $workerPool,
         private string $inputDirectory,
         private PendingExecution $pendingExecution,
+        private ImportCleanupService $cleanupService,
     ) {
     }
 
@@ -51,4 +52,10 @@ public function await(): void
         $this->pendingExecution->await();
         $this->workerPool->shutdown();
     }
+
+
+    public function cleanTable(TableEntry $table): void
+    {
+        $this->cleanupService->cleanTable($table);
+    }
 }

src/Import/WorkerImportServiceFactory.php

@@ -19,7 +19,7 @@
     private function __construct(
         private Configuration $configuration,
         private WorkerPool    $workerPool,
-        private string        $inputDirectory,
+        private string        $inputDirectory
     ) {
     }
 
@@ -38,7 +38,11 @@ public function create(): WorkerImportService
             $this->configuration,
             $this->workerPool,
             $this->inputDirectory,
-            new PendingExecution()
+            new PendingExecution(),
+            BlockingImportCleanupService::create(
+                $this->configuration->importMode,
+                $this->configuration->createPDOConnection()
+            ),
         );
     }
 }

tests/Export/TableListServiceTest.php

@@ -26,15 +26,30 @@ public function usesConfigurationToListExportableTables()
             ->withExcludeCondition(TableNameCondition::endsWith('_varchar'))
             ->withExcludeCondition(TableNameCondition::contains('gallery'));
 
-        $this->assertEquals(
-            [
-                TableEntry::fromName('catalog_product_entity_datetime')->withRows(6),
-                TableEntry::fromName('catalog_product_entity_decimal')->withRows(3903),
-                TableEntry::fromName('catalog_product_entity_int')->withRows(12480),
-                TableEntry::fromName('catalog_product_entity_text')->withRows(2652),
-                TableEntry::fromName('catalog_product_entity_tier_price')->withRows(0)
-            ],
-            (new TableListService($config))->tablesToExport()
-        );
+
+        $expectedTableValues = [
+            TableEntry::fromName('catalog_product_entity_datetime')->withRows(6),
+            TableEntry::fromName('catalog_product_entity_decimal')->withRows(3903),
+            TableEntry::fromName('catalog_product_entity_int')->withRows(12480),
+            TableEntry::fromName('catalog_product_entity_text')->withRows(2652),
+            TableEntry::fromName('catalog_product_entity_tier_price')->withRows(0)
+        ];
+
+        $actualTableValues = (new TableListService($config))->tablesToExport();
+
+        foreach ($expectedTableValues as $index => $expectedTable) {
+            $this->assertArrayHasKey(
+                $index,
+                $actualTableValues,
+                'Table list does not match'
+            );
+
+            $this->assertEquals($expectedTable->name, $actualTableValues[$index]->name, 'Table name does not match expected');
+            $this->assertEqualsWithDelta(
+                $expectedTable->rowCount,
+                $actualTableValues[$index]->rowCount,
+                $expectedTable->rowCount * 0.05
+            );
+        }
     }
 }

tests/Import/ImportTableTest.php

@@ -43,6 +43,8 @@ public function importsTableDataFromSourceWithTruncate()
         );
 
         $import = $factory->create();
+
+        $import->cleanTable($source->table);
         $import->importTable($source->table, $this);
 
         $connection = $this->containerWithConfig->configuration->createPDOConnection();
@@ -73,6 +75,7 @@ public function appendsExistingStoresWhenUpdateStrategyIsUsed()
         );
 
         $import = $factory->create();
+        $import->cleanTable($source->table);
         $import->importTable($source->table, $this);
 
         $connection = $this->containerWithConfig->configuration->createPDOConnection();
@@ -105,6 +108,7 @@ public function notifiesOfTheImportPorgress()
         );
 
         $import = $factory->create();
+        $import->cleanTable($source->table);
         $import->importTable($source->table, $this);
 
         $this->assertEquals(
@@ -138,6 +142,7 @@ public function batchesLongInputs()
         );
 
         $import = $factory->create();
+        $import->cleanTable($source->table);
         $import->importTable($source->table, $this);
 
         $connection = $this->containerWithConfig->configuration->createPDOConnection();

tests/Import/WorkerImportTableTest.php

@@ -38,6 +38,7 @@ public function importsDataFromSource()
         $service = WorkerImportServiceFactory::fromConfiguration($container->configuration, $this->path)
             ->create();
 
+        $service->cleanTable(TableEntry::fromName('store'));
         $service->importTable(TableEntry::fromName('store')->withRows(2), $this);
         $service->await();