wip

bin/mysql2jsonl

@@ -1,4 +1,18 @@
-#!env php
+#!/usr/bin/env php
 <?php
 
+use EcomDev\MySQL2JSONL\Command\ExportCommand;
+use Symfony\Component\Console\Application;
+
 require_once __DIR__ . '/../vendor/autoload.php';
+
+$application = new Application(
+    basename(__FILE__),
+    '1.0.0'
+);
+
+$application->addCommands([
+    new ExportCommand()
+]);
+
+$application->run();

composer.json

@@ -7,6 +7,7 @@
     "amphp/file": "~3.2",
     "amphp/mysql": "~3.0",
     "amphp/amp": "~3.0",
+    "amphp/pipeline": "~1.2",
     "revolt/event-loop": "~1.0",
     "symfony/console": "~7.2",
     "justinrainbow/json-schema": "^6.0"
@@ -29,6 +30,9 @@
     }
   },
   "autoload-dev": {
+    "files": [
+      "tests/fixtures.php"
+    ],
     "psr-4": {
       "EcomDev\\MySQL2JSONL\\": "tests/"
     }

schema.json

@@ -10,7 +10,7 @@
     "maxConnections": {
       "type": "integer",
       "description": "Maximum number of open connection by tool",
-      "default": 50
+      "default": 10
     },
     "idleTimeout": {
       "type": "integer",
@@ -26,6 +26,7 @@
       "description": "List of tables to exclude from the database dump"
     }
   },
+  "required": ["connection"],
   "additionalProperties": false,
   "definitions": {
     "matchExpression": {

src/Command/ExportCommand.php

@@ -0,0 +1,86 @@
+<?php
+
+namespace EcomDev\MySQL2JSONL\Command;
+
+use Amp\Pipeline\Queue;
+use EcomDev\MySQL2JSONL\Configuration;
+use EcomDev\MySQL2JSONL\ConfigurationException;
+use EcomDev\MySQL2JSONL\ExportTableFactory;
+use EcomDev\MySQL2JSONL\Progress\ExportProgressNotifier;
+use Revolt\EventLoop;
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Helper\FormatterHelper;
+use Symfony\Component\Console\Input\InputArgument;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use function Amp\async;
+use function Amp\delay;
+use function Amp\Future\awaitAll;
+
+#[AsCommand(name: 'export', description: 'Export data from MySQL to a directory with JSONL files')]
+class ExportCommand extends Command
+{
+    public function configure()
+    {
+        $this->addOption(
+            'config',
+            'c', InputOption::VALUE_REQUIRED,
+            'Configuration file',
+            'config.json'
+        );
+
+        $this->addArgument(
+            'directory',
+            InputArgument::OPTIONAL,
+            'Directory to export data to',
+            './data-dump'
+        );
+    }
+
+    public function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $file = $input->getOption('config');
+        if (!file_exists($file)) {
+            /* @var FormatterHelper $formatter*/
+            $formatter = $this->getHelper('formatter');
+            $output->write($formatter->formatSection('Error', sprintf(
+                'Configuration file %s does not exist',
+                $file
+            ), 'error'));
+            return 1;
+        }
+        try {
+            $config = Configuration::fromJSON(file_get_contents($file));
+        } catch (ConfigurationException $error) {
+            $error->output($output);
+            return 1;
+        }
+
+        $notifiers = new ExportProgressNotifier($output);
+        $futures = [];
+        $connectionPool = $config->createConnectionPool();
+        $factory = new ExportTableFactory($connectionPool, $notifiers);
+        foreach ($factory->tablesToExport($config) as $table) {
+            $queue = new Queue(100);
+
+            $futures[] = async(function () use ($factory, $table, $queue) {
+                $factory->createExport($table)->run($queue);
+            });
+
+            $futures[] = async(function () use ($queue, $output) {
+                foreach ($queue->iterate() as $item) {
+                }
+            });
+
+            if (count($futures) >= 100) {
+                awaitAll($futures);
+                $futures = [];
+            }
+        }
+
+        EventLoop::run();
+        return 0;
+    }
+}

src/Condition/AnyTableCondition.php

@@ -15,7 +15,6 @@
 {
     public function __construct(private array $conditions)
     {
-
     }
 
     public function isSatisfiedBy(string $tableName, int $rows): bool
@@ -33,4 +32,4 @@ public function withCondition(TableCondition $condition): TableCondition
         $conditions[] = $condition;
         return new self($conditions);
     }
-}
+}

src/Condition/StaticTableCondition.php

@@ -34,4 +34,4 @@ public function isSatisfiedBy(string $tableName, int $rows): bool
     {
         return $this->result;
     }
-}
+}

src/Condition/TableNameCondition.php

@@ -62,4 +62,4 @@ public function isSatisfiedBy(string $tableName, int $rows): bool
             default => false
         };
     }
-}
+}

src/Condition/TableRowsCondition.php

@@ -21,9 +21,7 @@ class TableRowsCondition implements TableCondition
     public function __construct(
         private readonly int $rowsCount,
         private readonly string $type
-    )
-    {
-
+    ) {
     }
 
     public static function minRows(int $minRows): self
@@ -44,4 +42,4 @@ public function isSatisfiedBy(string $tableName, int $rows): bool
             default => false,
         };
     }
-}
+}

src/Configuration.php

@@ -10,6 +10,7 @@
 namespace EcomDev\MySQL2JSONL;
 
 use Amp\Mysql\MysqlConfig;
+use Amp\Mysql\MysqlConnectionPool;
 use EcomDev\MySQL2JSONL\Condition\AndTableCondition;
 use EcomDev\MySQL2JSONL\Condition\StaticTableCondition;
 use EcomDev\MySQL2JSONL\Condition\TableNameCondition;
@@ -45,6 +46,7 @@ public static function fromJSON(string $json): Configuration
             $json,
             flags: JSON_THROW_ON_ERROR
         );
+
         $result = $validator->validate($data, $schema, Constraint::CHECK_MODE_APPLY_DEFAULTS);
 
         if ($result !== Validator::ERROR_NONE) {
@@ -102,7 +104,7 @@ function ($errors, $error) {
         );
     }
 
-    public static function fromMySQLConfig(MysqlConfig $config, int $maxConnections = 50, int $idleTimeout = 60): self
+    public static function fromMySQLConfig(MysqlConfig $config, int $maxConnections = 10, int $idleTimeout = 60): self
     {
         return new self(
             $config,
@@ -155,4 +157,13 @@ private static function buildCondition(string|object $condition): TableCondition
             };
         }
     }
+
+    public function createConnectionPool(): MysqlConnectionPool
+    {
+        return new MysqlConnectionPool(
+            $this->connection,
+            $this->maxConnections,
+            $this->idleTimeout
+        );
+    }
 }

src/ConfigurationException.php

@@ -47,4 +47,4 @@ public function output(ConsoleOutputInterface $output): void
             $helper->formatSection($this->getMessage(), PHP_EOL . implode(PHP_EOL, $messages), 'error')
         );
     }
-}
+}

src/ExportTable.php

@@ -9,15 +9,43 @@
 
 namespace EcomDev\MySQL2JSONL;
 
+use Amp\Future;
 use Amp\Mysql\MysqlConnectionPool;
+use Amp\Pipeline\Queue;
+use function Amp\async;
+use function Amp\delay;
 
-class ExportTable
+final class ExportTable
 {
     public function __construct(
         private readonly string $tableName,
         private readonly MysqlConnectionPool $connectionPool,
         private readonly ProgressNotifier $progressNotifier,
     ) {
+    }
+
+    public function run(Queue $queue): void
+    {
+        $total = $this->connectionPool
+            ->execute(sprintf('SELECT COUNT(*) as `total` FROM `%s`', $this->tableName))
+            ->fetchRow();
+
+        $this->progressNotifier->start($this->tableName, $total['total']);
+
+        $result = $this->connectionPool->execute(sprintf('SELECT * FROM `%s`', $this->tableName));
+        $header = [];
+        foreach ($result->getColumnDefinitions() as $column) {
+            $header[] = $column->getName();
+        }
+        $queue->push($header);
+
+        $index = 0;
+        foreach ($result as $row) {
+            $queue->push(array_values($row));
+            $this->progressNotifier->update($this->tableName, ++$index);
+        }
 
+        $queue->complete();
+        $this->progressNotifier->finish($this->tableName);
     }
-}
+}

src/ExportTableFactory.php

@@ -22,6 +22,25 @@ public function __construct(
 
     public function createExport(string $tableName): ExportTable
     {
-        return new ExportTable($tableName, $this->connectionPool, $por);
+        return new ExportTable($tableName, $this->connectionPool, $this->progressNotifier);
+    }
+
+    public function tablesToExport(Configuration $configuration): array
+    {
+        $result = $this->connectionPool->execute(
+            'SELECT TABLE_NAME, TABLE_ROWS FROM information_schema.tables'
+            . ' WHERE TABLE_SCHEMA = SCHEMA() AND TABLE_TYPE = ?',
+            ['BASE TABLE']
+        );
+
+        $tables = [];
+        foreach ($result as $row) {
+            if (!$configuration->includeCondition->isSatisfiedBy($row['TABLE_NAME'], $row['TABLE_ROWS'])
+                || $configuration->excludeCondition->isSatisfiedBy($row['TABLE_NAME'], $row['TABLE_ROWS'])) {
+                continue;
+            }
+            $tables[] = $row['TABLE_NAME'];
+        }
+        return $tables;
     }
 }

src/ProgressNotifier.php

@@ -16,4 +16,4 @@ public function start(string $name, int $total): void;
     public function update(string $name, int $current): void;
 
     public function finish(string $name): void;
-}
+}

src/Sql/InsertOnDuplicate.php

@@ -44,4 +44,4 @@ public function generate(string $tableName, array $columns, int $rowCount, $onUp
 
         return $sql;
     }
-}
+}

src/TableCondition.php

@@ -14,4 +14,4 @@ interface TableCondition
     public function isSatisfiedBy(string $tableName, int $rows): bool;
 
     public function withCondition(TableCondition $condition): TableCondition;
-}
+}