From 0d27d3defff30fa512062842a50aea98644a08d6 Mon Sep 17 00:00:00 2001 From: Hagen Pommer Date: Fri, 24 Jul 2026 16:31:24 +0200 Subject: [PATCH 01/15] feat(dev): add local dev environment via docker compose --- .env.example | 2 ++ compose.yaml | 12 ++++++++++++ 2 files changed, 14 insertions(+) create mode 100644 compose.yaml diff --git a/.env.example b/.env.example index ca9d384..fb3cc33 100644 --- a/.env.example +++ b/.env.example @@ -1 +1,3 @@ APP_KEY= +# DEV_PHP_VERSION=8.5-dev-macos +DEV_PHP_VERSION=8.5-dev diff --git a/compose.yaml b/compose.yaml new file mode 100644 index 0000000..069ef1b --- /dev/null +++ b/compose.yaml @@ -0,0 +1,12 @@ +services: + php: + image: "wodby/php:8.5-dev-macos" + environment: + PHP_EXTENSIONS_DISABLE: 'xhprof,spx' + healthcheck: + test: ["CMD", "php", "-v"] + interval: 10s + timeout: 5s + retries: 3 + volumes: + - .:/var/www/html From 653e5540b0161d43d4c53452290b139aaf155a5f Mon Sep 17 00:00:00 2001 From: Hagen Pommer Date: Fri, 24 Jul 2026 16:31:41 +0200 Subject: [PATCH 02/15] feat(dev): add make targets for local development --- Makefile | 3 +++ Makefile.dev.mk | 17 +++++++++++++++++ 2 files changed, 20 insertions(+) create mode 100644 Makefile.dev.mk diff --git a/Makefile b/Makefile index 14f12ed..92c3523 100644 --- a/Makefile +++ b/Makefile @@ -122,3 +122,6 @@ docker-test: docker-build docker-shell: docker-build docker run --rm -it -v "$$(pwd)":/workspace --entrypoint sh $(DOCKER_IMAGE) + + +include Makefile.dev.mk diff --git a/Makefile.dev.mk b/Makefile.dev.mk new file mode 100644 index 0000000..92df658 --- /dev/null +++ b/Makefile.dev.mk @@ -0,0 +1,17 @@ +up: + docker compose up -d + +down: + docker compose down + +restart: down up + +bash: + docker compose exec php bash + +test: + docker compose exec php bash -c 'XDEBUG_MODE=coverage composer test' + +check: + docker compose exec php bash -c 'composer lint && composer test:types' + From 833b1e675b0605a9bc6926f6e2c858a9036af5dd Mon Sep 17 00:00:00 2001 From: Hagen Pommer Date: Fri, 7 Aug 2026 10:54:41 +0200 Subject: [PATCH 03/15] test(init): make IO-error tests user independent --- tests/Feature/Commands/InitCommandTest.php | 39 ++++++++++------------ 1 file changed, 17 insertions(+), 22 deletions(-) diff --git a/tests/Feature/Commands/InitCommandTest.php b/tests/Feature/Commands/InitCommandTest.php index 95447a3..729df04 100644 --- a/tests/Feature/Commands/InitCommandTest.php +++ b/tests/Feature/Commands/InitCommandTest.php @@ -232,40 +232,35 @@ it('returns an IO error when a new .env cannot be written', function (): void { // No APP_KEY anywhere -> the command tries to create a fresh .env. - // Making the disk root read-only forces Storage::put() to fail, which - // bubbles up as a RuntimeException and is reported as an IO error. + // Forcing Storage::put() to return false (as it would on a failed write) + // bubbles up as a RuntimeException and is reported as an IO error. We mock + // rather than chmod because chmod is a no-op under root (e.g. in Docker). putenv('APP_KEY'); unset($_ENV['APP_KEY'], $_SERVER['APP_KEY']); - $root = Storage::path(''); - chmod($root, 0500); + Storage::shouldReceive('exists')->with('.env')->andReturn(false); + Storage::shouldReceive('put')->with('.env', Mockery::type('string'))->andReturn(false); - try { - $this->artisan('init') - ->expectsOutputToContain('permission denied') - ->assertExitCode(ExitCode::IoError->value); - } finally { - chmod($root, 0755); - } + $this->artisan('init') + ->expectsOutputToContain('permission denied') + ->assertExitCode(ExitCode::IoError->value); }); it('returns an IO error when an existing .env cannot be overwritten', function (): void { // An existing .env without an APP_KEY would normally be appended to, but - // making the file itself read-only forces the write to fail. + // forcing Storage::put() to return false makes the rewrite fail. We mock + // rather than chmod because chmod is a no-op under root (e.g. in Docker). putenv('APP_KEY'); unset($_ENV['APP_KEY'], $_SERVER['APP_KEY']); - Storage::put('.env', "DB_HOST=localhost\n"); - $path = Storage::path('.env'); - chmod($path, 0400); + Storage::shouldReceive('exists')->with('.env')->andReturn(true); + Storage::shouldReceive('get')->with('.env')->andReturn("DB_HOST=localhost\n"); + Storage::shouldReceive('path')->with('.env')->andReturn('/tmp/clonio-readonly.env'); + Storage::shouldReceive('put')->with('.env', Mockery::type('string'))->andReturn(false); - try { - $this->artisan('init') - ->expectsOutputToContain('permission denied') - ->assertExitCode(ExitCode::IoError->value); - } finally { - chmod($path, 0644); - } + $this->artisan('init') + ->expectsOutputToContain('permission denied') + ->assertExitCode(ExitCode::IoError->value); }); it('returns an IO error when an existing .env cannot be read', function (): void { From 66881c1d7f0d714c0aa1364654f8421a8abf6221 Mon Sep 17 00:00:00 2001 From: Hagen Pommer Date: Mon, 27 Jul 2026 14:31:38 +0200 Subject: [PATCH 04/15] feat(cloning): add per-table timing and progress tracking data model --- app/Data/Cloning/StatsLoopData.php | 27 ++++ app/Data/Cloning/StatsLoopSnapshotData.php | 28 ++++ app/Data/Cloning/StatsPhaseAggregateData.php | 69 ++++++++ app/Data/Cloning/StatsTableTransferData.php | 158 +++++++++++++++++++ app/Data/Cloning/TableRunPhase.php | 47 ++++++ app/Data/Cloning/TableRunStatus.php | 1 + 6 files changed, 330 insertions(+) create mode 100644 app/Data/Cloning/StatsLoopData.php create mode 100644 app/Data/Cloning/StatsLoopSnapshotData.php create mode 100644 app/Data/Cloning/StatsPhaseAggregateData.php create mode 100644 app/Data/Cloning/StatsTableTransferData.php create mode 100644 app/Data/Cloning/TableRunPhase.php diff --git a/app/Data/Cloning/StatsLoopData.php b/app/Data/Cloning/StatsLoopData.php new file mode 100644 index 0000000..2121daf --- /dev/null +++ b/app/Data/Cloning/StatsLoopData.php @@ -0,0 +1,27 @@ + $this->count > 0 ? $this->sum / $this->count : null; + } + + /** + * Aggregate seconds per 1,000,000 rows across all samples, + * or null when no rows have been processed. + */ + public ?float $secondsPerMillionRows { + get => $this->rowsProcessed > 0 + ? ($this->sum / $this->rowsProcessed) * 1_000_000.0 + : null; + } + + /** Latest-sample seconds per 1,000,000 rows, or null when unavailable. */ + public ?float $latestSecondsPerMillionRows { + get => $this->last !== null && $this->lastRows !== null && $this->lastRows > 0 + ? ($this->last / $this->lastRows) * 1_000_000.0 + : null; + } + + public static function withRecord(float $seconds, int $rows): self + { + $instance = new self; + + $instance->record($seconds, $rows); + + return $instance; + } + + public function record(float $seconds, int $rows): void + { + $this->count++; + $this->sum += $seconds; + $this->min = $this->min === null ? $seconds : min($this->min, $seconds); + $this->max = $this->max === null ? $seconds : max($this->max, $seconds); + $this->last = $seconds; + $this->lastRows = $rows; + $this->rowsProcessed += max(0, $rows); + } +} diff --git a/app/Data/Cloning/StatsTableTransferData.php b/app/Data/Cloning/StatsTableTransferData.php new file mode 100644 index 0000000..9c2f545 --- /dev/null +++ b/app/Data/Cloning/StatsTableTransferData.php @@ -0,0 +1,158 @@ + */ + public private(set) Collection $loops; + + /** @var Collection */ + public private(set) Collection $statsOverTime; + + public private(set) int $totalRows = 0; + + public private(set) int $rowsDone = 0; + + public private(set) int $rowsSkipped = 0; + + /** Rows accounted for so far (transferred + skipped). */ + public int $rowsProcessed { + get => $this->rowsDone + $this->rowsSkipped; + } + + /** Rows still outstanding against `totalRows` (never negative). */ + public int $rowsRemaining { + get => max(0, $this->totalRows - $this->rowsProcessed); + } + + /** Completion ratio in the range [0, 100], or null when the total is unknown. */ + public ?float $percentComplete { + get => $this->totalRows > 0 + ? min(100.0, ($this->rowsProcessed / $this->totalRows) * 100.0) + : null; + } + + public private(set) StatsPhaseAggregateData $selectAggregate; + + public private(set) StatsPhaseAggregateData $transformAggregate; + + public private(set) StatsPhaseAggregateData $insertAggregate; + + /** + * Per-loop aggregate over each chunk's wall-clock time + * (`StatsLoopData::$overallSeconds`, rows = chunkRows). Useful for + * loop throughput including inter-phase overhead. + */ + public private(set) StatsPhaseAggregateData $loopAggregate; + + public function __construct() + { + $this->loops = new Collection; + $this->statsOverTime = new Collection; + $this->selectAggregate = new StatsPhaseAggregateData; + $this->transformAggregate = new StatsPhaseAggregateData; + $this->insertAggregate = new StatsPhaseAggregateData; + $this->loopAggregate = new StatsPhaseAggregateData; + } + + public function setStatus(TableRunPhase $status): void + { + $this->status = $status; + } + + public function setTotalRows(int $totalRows): void + { + $this->totalRows = max(0, $totalRows); + } + + public function recordCountingRows(float $seconds): void + { + $this->countingRowsSeconds = $seconds; + } + + public function recordDisableFk(float $seconds): void + { + $this->disableFkSeconds = $seconds; + } + + public function recordClearTable(float $seconds): void + { + $this->clearTableSeconds = $seconds; + } + + /** + * Append a completed chunk loop, update running aggregates in O(1), + * and append a stats-over-time snapshot. + */ + public function recordLoop(StatsLoopData $loop): void + { + $this->loops->push($loop); + + $this->selectAggregate->record($loop->selectSeconds, $loop->chunkRows); + $this->transformAggregate->record($loop->transformSeconds, $loop->chunkRows); + $this->insertAggregate->record($loop->insertSeconds, $loop->chunkRows); + $this->loopAggregate->record($loop->overallSeconds, $loop->chunkRows); + + $this->rowsDone += $loop->rowsDone; + $this->rowsSkipped += $loop->rowsSkipped; + + $this->statsOverTime->push(new StatsLoopSnapshotData( + loopIndex: $loop->loopIndex, + loopsRecorded: $this->loops->count(), + rowsDoneCumulative: $this->rowsDone, + rowsSkippedCumulative: $this->rowsSkipped, + percentComplete: $this->percentComplete, + selectSecondsPerMillionRows: $this->selectAggregate->secondsPerMillionRows, + transformSecondsPerMillionRows: $this->transformAggregate->secondsPerMillionRows, + insertSecondsPerMillionRows: $this->insertAggregate->secondsPerMillionRows, + loopSecondsPerMillionRows: $this->loopAggregate->secondsPerMillionRows, + )); + } + + public function aggregate(TableRunPhase $phase): StatsPhaseAggregateData + { + return match ($phase) { + TableRunPhase::CountingRows => $this->oneShotAggregate($this->countingRowsSeconds), + TableRunPhase::DisableFkChecks => $this->oneShotAggregate($this->disableFkSeconds), + TableRunPhase::Clear => $this->oneShotAggregate($this->clearTableSeconds), + TableRunPhase::Select => $this->selectAggregate, + TableRunPhase::Transform => $this->transformAggregate, + TableRunPhase::Insert => $this->insertAggregate, + TableRunPhase::Loop => $this->loopAggregate, + }; + } + + /** + * Wrap a one-shot phase duration as a single-sample aggregate. A phase that + * never ran (null) yields an empty (count-0) aggregate so consumers can skip it. + */ + private function oneShotAggregate(?float $seconds): StatsPhaseAggregateData + { + return $seconds === null + ? new StatsPhaseAggregateData + : StatsPhaseAggregateData::withRecord($seconds, $this->totalRows); + } +} diff --git a/app/Data/Cloning/TableRunPhase.php b/app/Data/Cloning/TableRunPhase.php new file mode 100644 index 0000000..d9d3065 --- /dev/null +++ b/app/Data/Cloning/TableRunPhase.php @@ -0,0 +1,47 @@ + true, + default => false, + }; + } + + /** Human-readable, present-continuous label for live progress display. */ + public function label(): string + { + return match ($this) { + self::CountingRows => 'counting rows', + self::DisableFkChecks => 'disabling FK checks', + self::Clear => 'clearing data', + self::Select => 'selecting', + self::Transform => 'transforming', + self::Insert => 'inserting', + self::Loop => 'writing', + }; + } +} diff --git a/app/Data/Cloning/TableRunStatus.php b/app/Data/Cloning/TableRunStatus.php index ee60bc7..0363316 100644 --- a/app/Data/Cloning/TableRunStatus.php +++ b/app/Data/Cloning/TableRunStatus.php @@ -7,6 +7,7 @@ enum TableRunStatus: string { case Transferred = 'transferred'; + case InProgress = 'in_progress'; case SkippedByFlag = 'skipped_by_flag'; case SkippedByCascade = 'skipped_by_cascade'; case NotFound = 'not_found'; From 89446d03a6c838d0b3fecbaf75f5914a27ec7797 Mon Sep 17 00:00:00 2001 From: Hagen Pommer Date: Mon, 27 Jul 2026 14:31:38 +0200 Subject: [PATCH 05/15] test(cloning): cover per-table timing and progress data model --- .../Cloning/StatsTableTransferDataTest.php | 29 +++++++++++++++++++ tests/Unit/Data/Cloning/TableRunPhaseTest.php | 26 +++++++++++++++++ 2 files changed, 55 insertions(+) create mode 100644 tests/Unit/Data/Cloning/StatsTableTransferDataTest.php create mode 100644 tests/Unit/Data/Cloning/TableRunPhaseTest.php diff --git a/tests/Unit/Data/Cloning/StatsTableTransferDataTest.php b/tests/Unit/Data/Cloning/StatsTableTransferDataTest.php new file mode 100644 index 0000000..f8e3a8b --- /dev/null +++ b/tests/Unit/Data/Cloning/StatsTableTransferDataTest.php @@ -0,0 +1,29 @@ +aggregate($phase)->count)->toBe(0); + } +}); + +it('records a single-sample aggregate once a one-shot phase runs', function (): void { + $stats = new StatsTableTransferData; + $stats->setTotalRows(1000); + $stats->recordClearTable(0.5); + + $cleared = $stats->aggregate(TableRunPhase::Clear); + expect($cleared->count)->toBe(1); + expect($cleared->sum)->toBe(0.5); + + // A sibling one-shot phase that still never ran stays at count 0. + expect($stats->aggregate(TableRunPhase::DisableFkChecks)->count)->toBe(0); +}); diff --git a/tests/Unit/Data/Cloning/TableRunPhaseTest.php b/tests/Unit/Data/Cloning/TableRunPhaseTest.php new file mode 100644 index 0000000..9e20048 --- /dev/null +++ b/tests/Unit/Data/Cloning/TableRunPhaseTest.php @@ -0,0 +1,26 @@ +isOneShot())->toBeTrue(); + expect(TableRunPhase::DisableFkChecks->isOneShot())->toBeTrue(); + expect(TableRunPhase::Clear->isOneShot())->toBeTrue(); + + expect(TableRunPhase::Select->isOneShot())->toBeFalse(); + expect(TableRunPhase::Transform->isOneShot())->toBeFalse(); + expect(TableRunPhase::Insert->isOneShot())->toBeFalse(); + expect(TableRunPhase::Loop->isOneShot())->toBeFalse(); +}); + +it('gives every phase a human-readable label distinct from its raw value', function (): void { + foreach (TableRunPhase::cases() as $phase) { + expect($phase->label())->not->toBe('') + ->and($phase->label())->not->toBe($phase->value); + } + + expect(TableRunPhase::CountingRows->label())->toBe('counting rows'); + expect(TableRunPhase::DisableFkChecks->label())->toBe('disabling FK checks'); +}); From ce61d7d3b2223abfa752c0472ccb563496a5735a Mon Sep 17 00:00:00 2001 From: Hagen Pommer Date: Mon, 27 Jul 2026 14:31:38 +0200 Subject: [PATCH 06/15] feat(cloning): emit per-table timing and progress events from run orchestrator --- .../Cloning/CloningRunOrchestrator.php | 140 ++++++++++++++++-- 1 file changed, 125 insertions(+), 15 deletions(-) diff --git a/app/Services/Cloning/CloningRunOrchestrator.php b/app/Services/Cloning/CloningRunOrchestrator.php index c6c7185..ccb8c52 100644 --- a/app/Services/Cloning/CloningRunOrchestrator.php +++ b/app/Services/Cloning/CloningRunOrchestrator.php @@ -9,7 +9,10 @@ use App\Data\Cloning\ColumnCloningConfigData; use App\Data\Cloning\KeyRemappingConfigData; use App\Data\Cloning\RunResultData; +use App\Data\Cloning\StatsLoopData; +use App\Data\Cloning\StatsTableTransferData; use App\Data\Cloning\TableCloningConfigData; +use App\Data\Cloning\TableRunPhase; use App\Data\Cloning\TableRunResultData; use App\Data\Cloning\TableRunStatus; use App\Data\ConnectionData; @@ -36,8 +39,14 @@ public function __construct( /** * @param list $skipTables Tables to exclude (already validated as mutually exclusive with onlyTables) * @param list $onlyTables If non-empty, only these tables are transferred - * @param callable(string, TableRunStatus, int, int, list): void $onProgress + * @param callable(string, TableRunStatus, int, int, list, ?StatsTableTransferData=): void $onProgress * @param (callable(string): void)|null $onTableStart Optional. Fires once per table that enters `transferTable()`, regardless of whether the transfer ultimately succeeds (`Transferred`) or fails (`Failed`). Does NOT fire for tables resolved as `SkippedByFlag`, `SkippedByCascade`, `NotFound`, or `SkippedBySchemaFailure`. + * @param (callable(int): void)|null $onStart Optional. Fires exactly once, before the transfer loop, with the number of tables the loop will iterate. Note: when `$breakOnFailure` aborts early, fewer terminal `onProgress` events fire than this count. Useful for sizing an overall progress bar. + * + * The `$onProgress` `$timings` argument is a live, mutable object reused across + * every event for a table — read it synchronously inside the callback; do not + * retain the reference expecting a point-in-time snapshot. + * @param bool $trackRowTotals When true, run a `SELECT COUNT(*)` per table (respecting the row limit) to size per-table progress. Off by default so non-interactive runs don't pay for a count nobody consumes. */ public function run( CloningConfigData $config, @@ -52,6 +61,8 @@ public function run( bool $breakOnFailure = false, ?callable $onTableStart = null, ?SqlDumpService $dumpSink = null, + ?callable $onStart = null, + bool $trackRowTotals = false, ): RunResultData { $start = microtime(true); $tableNames = array_map(static fn (TableCloningConfigData $t): string => $t->tableName, $config->tables); @@ -76,6 +87,12 @@ public function run( $sortedTables = $this->resolver->sort($sourceSchema, $remaining); + // Announce the number of tables that will be attempted (each emits a + // terminal onProgress event) so callers can size an overall progress bar. + if ($onStart !== null) { + $onStart(count($sortedTables)); + } + // Replicate schema if not skipping /** @var array $schemaFailures */ $schemaFailures = []; @@ -182,8 +199,8 @@ public function run( )) : []; - [$rows, $skipped, $failed, $reason, $skippedRows] = $dumpSink instanceof SqlDumpService - ? $this->dumpTable( + [$rows, $skipped, $failed, $reason, $skippedRows, $timings] = $dumpSink instanceof SqlDumpService + ? [...$this->dumpTable( $config->options, $tableConfig, $source, @@ -191,7 +208,7 @@ public function run( $dumpSink, $keyRemapping, $config->keyRemapping, - ) + ), null] : $this->transferTable( $config->options, $tableConfig, @@ -201,6 +218,8 @@ public function run( $engine, $keyRemapping, $config->keyRemapping, + $onProgress, + $trackRowTotals, ); $tableDuration = microtime(true) - $tableStart; @@ -217,7 +236,7 @@ public function run( $tableResults[] = new TableRunResultData($tableName, $status, $rows, $skipped, $tableDuration, $reason); $totalRows += $rows; $totalSkipped += $skipped; - ($onProgress)($tableName, $status, $rows, $skipped, $skippedRows); + ($onProgress)($tableName, $status, $rows, $skipped, $skippedRows, $timings); if ($failed && $breakOnFailure) { break; @@ -273,10 +292,10 @@ private function findIntegerPkColumn(ConnectionData $target, TableSchemaData $ta } /** - * Transfer a single table. Returns [rowsTransferred, rowsSkipped, hasFailed, failureReason, skippedRows]. + * Transfer a single table. Returns [rowsTransferred, rowsSkipped, hasFailed, failureReason, skippedRows, timings]. * * @param list $pkColumns - * @return array{int, int, bool, ?string, list} + * @return array{int, int, bool, ?string, list, StatsTableTransferData} */ private function transferTable( CloningOptionsData $options, @@ -285,9 +304,12 @@ private function transferTable( ConnectionData $target, array $pkColumns, AnonymizationEngine $engine, - ?KeyRemappingService $keyRemapping = null, - ?KeyRemappingConfigData $keyRemappingConfig = null, + ?KeyRemappingService $keyRemapping, + ?KeyRemappingConfigData $keyRemappingConfig, + callable $onProgress, + bool $trackRowTotals = false, ): array { + $sourceConn = $this->connector->open($source); $targetConn = $this->connector->open($target); @@ -300,31 +322,68 @@ private function transferTable( /** @var list $skippedRows */ $skippedRows = []; + $stats = new StatsTableTransferData; + + // Only pay for the row count when a consumer actually wants per-table + // progress totals; otherwise skip it (totalRows stays 0 → indeterminate). + if ($trackRowTotals) { + $stats->setStatus(TableRunPhase::CountingRows); + ($onProgress)($tableConfig->tableName, TableRunStatus::InProgress, $rows, $skipped, $skippedRows, $stats); + + $tCount = microtime(true); + $stats->setTotalRows($this->countSourceRows($sourceConn, $tableConfig, $source)); + $stats->recordCountingRows(microtime(true) - $tCount); + } + + $loopIndex = 0; + try { if ($options->disableForeignKeyChecks) { + $stats->setStatus(TableRunPhase::DisableFkChecks); + ($onProgress)($tableConfig->tableName, TableRunStatus::InProgress, $rows, $skipped, $skippedRows, $stats); + + $t0 = microtime(true); $this->disableFkChecks($targetConn, $target); + $stats->recordDisableFk(microtime(true) - $t0); } if ($tableConfig->rows->clear !== ClearMode::None) { + $stats->setStatus(TableRunPhase::Clear); + ($onProgress)($tableConfig->tableName, TableRunStatus::InProgress, $rows, $skipped, $skippedRows, $stats); + + $t0 = microtime(true); $this->clearTable($targetConn, $tableConfig->tableName, $tableConfig->rows->clear, $target); + $stats->recordClearTable(microtime(true) - $t0); } do { + $tOverall = microtime(true); + $stats->setStatus(TableRunPhase::Select); + $tSelect = microtime(true); /** @var list $chunk */ $chunk = DB::connection($sourceConn)->select( $this->buildChunkQuery($tableConfig, $source, $offset, $chunkSize) ); + $selectSeconds = microtime(true) - $tSelect; if ($chunk === []) { break; } + $stats->setStatus(TableRunPhase::Transform); + $tTransform = microtime(true); $transformed = $this->transformChunk($chunk, $tableConfig, $engine, $keyRemapping, $keyRemappingConfig); + $transformSeconds = microtime(true) - $tTransform; + $chunkRowsAttempted = count($transformed); + $loopRowsDone = 0; + $loopRowsSkipped = 0; + $stats->setStatus(TableRunPhase::Insert); + $insertStart = microtime(true); // Bulk insert into target try { DB::connection($targetConn)->table($tableConfig->tableName)->insert($transformed); - $rows += count($transformed); + $loopRowsDone = $chunkRowsAttempted; } catch (Throwable $bulkError) { if ($firstInsertError === null) { $firstInsertError = $bulkError->getMessage(); @@ -334,9 +393,9 @@ private function transferTable( foreach ($transformed as $rowIndexInChunk => $row) { try { DB::connection($targetConn)->table($tableConfig->tableName)->insert($row); - $rows++; + $loopRowsDone++; } catch (Throwable $rowError) { - $skipped++; + $loopRowsSkipped++; /** @var array $sourceRow */ $sourceRow = (array) $chunk[$rowIndexInChunk]; $pkSnapshot = $this->extractPkSnapshot($sourceRow, $pkColumns); @@ -359,7 +418,29 @@ private function transferTable( } } + $insertSeconds = microtime(true) - $insertStart; + + $rows += $loopRowsDone; + $skipped += $loopRowsSkipped; + + $overallSeconds = microtime(true) - $tOverall; + + $stats->recordLoop(new StatsLoopData( + loopIndex: $loopIndex, + chunkRows: $chunkRowsAttempted, + selectSeconds: $selectSeconds, + transformSeconds: $transformSeconds, + insertSeconds: $insertSeconds, + overallSeconds: $overallSeconds, + rowsDone: $loopRowsDone, + rowsSkipped: $loopRowsSkipped, + totalRows: $stats->totalRows, + )); + + ($onProgress)($tableConfig->tableName, TableRunStatus::InProgress, $rows, $skipped, [], $stats); + $offset += count($chunk); + $loopIndex++; } while (count($chunk) === $chunkSize); if ($rows === 0 && $skipped > 0) { @@ -368,12 +449,12 @@ private function transferTable( $reason .= sprintf(': %s', $firstInsertError); } - return [0, $skipped, true, $reason, $skippedRows]; + return [0, $skipped, true, $reason, $skippedRows, $stats]; } - return [$rows, $skipped, false, null, $skippedRows]; + return [$rows, $skipped, false, null, $skippedRows, $stats]; } catch (Throwable $throwable) { - return [$rows, $skipped, true, $throwable->getMessage(), $skippedRows]; + return [$rows, $skipped, true, $throwable->getMessage(), $skippedRows, $stats]; } finally { if ($options->disableForeignKeyChecks) { $this->enableFkChecks($targetConn, $target); @@ -491,6 +572,35 @@ private function extractPkSnapshot(array $sourceRow, array $pkColumns): ?array return $snapshot === [] ? null : $snapshot; } + /** + * Best-effort SELECT COUNT(*) on the source table respecting the row + * strategy limit. Returns 0 on failure so progress stays optional. + */ + private function countSourceRows(string $sourceConn, TableCloningConfigData $config, ConnectionData $source): int + { + try { + $quoted = $this->quoteTable($config->tableName, $source->type); + $result = DB::connection($sourceConn)->selectOne(sprintf('SELECT COUNT(*) AS c FROM %s', $quoted)); + $count = 0; + if (is_object($result) && property_exists($result, 'c') && is_numeric($result->c)) { + $count = (int) $result->c; + } elseif (is_array($result) && array_key_exists('c', $result) && is_numeric($result['c'])) { + $count = (int) $result['c']; + } + + $limit = $config->rows->limit; + if ($limit !== null && $limit >= 0 && $limit < $count) { + return $limit; + } + + return $count; + } catch (Throwable $throwable) { + Log::warning('source_row_count_failed', ['table' => $config->tableName, 'error' => $throwable->getMessage()]); + + return 0; + } + } + private function buildChunkQuery(TableCloningConfigData $config, ConnectionData $source, int $offset, int $limit): string { $table = $config->tableName; From b472571829a0398d1896e7c3a9cac9d8526bb7c3 Mon Sep 17 00:00:00 2001 From: Hagen Pommer Date: Mon, 27 Jul 2026 14:31:38 +0200 Subject: [PATCH 07/15] test(cloning): cover progress and timing events in run orchestrator --- .../Cloning/CloningRunOrchestratorTest.php | 192 ++++++++++++++++++ 1 file changed, 192 insertions(+) diff --git a/tests/Unit/Services/Cloning/CloningRunOrchestratorTest.php b/tests/Unit/Services/Cloning/CloningRunOrchestratorTest.php index 654c412..9678c10 100644 --- a/tests/Unit/Services/Cloning/CloningRunOrchestratorTest.php +++ b/tests/Unit/Services/Cloning/CloningRunOrchestratorTest.php @@ -4,8 +4,11 @@ use App\Data\Cloning\CloningConfigData; use App\Data\Cloning\CloningOptionsData; +use App\Data\Cloning\StatsLoopData; +use App\Data\Cloning\StatsTableTransferData; use App\Data\Cloning\TableCloningConfigData; use App\Data\Cloning\TableRowConfigData; +use App\Data\Cloning\TableRunPhase; use App\Data\Cloning\TableRunResultData; use App\Data\Cloning\TableRunStatus; use App\Data\ConnectionData; @@ -674,12 +677,94 @@ static function (string $tbl, TableRunStatus $status, int $rows, int $skipped, a }, ); + // Row totals are not tracked by default, so no counting-rows event: start, + // then InProgress for the one chunk, then the terminal Transferred event. expect($events)->toBe([ ['start', 'users'], ['progress', 'users'], + ['progress', 'users'], ]); }); +it('fires onStart exactly once with the planned table count before the transfer loop', function (): void { + $source = makeOrchestratorConnection('source'); + $target = makeOrchestratorConnection('target'); + $schema = makeOrchestratorSchema(); + $config = makeOrchestratorConfig(); + + DB::shouldReceive('connection')->andReturnSelf(); + DB::shouldReceive('select')->andReturn([(object) ['id' => 1]], []); + DB::shouldReceive('table')->andReturnSelf(); + DB::shouldReceive('insert')->andReturnTrue(); + DB::shouldReceive('purge')->andReturnNull(); + + $events = []; + $orchestrator = makeOrchestrator(); + $orchestrator->run( + $config, + $source, + $target, + $schema, + true, + [], + [], + static function (string $tbl, TableRunStatus $status) use (&$events): void { + $events[] = ['progress', $tbl]; + }, + onStart: static function (int $total) use (&$events): void { + $events[] = ['start', $total]; + }, + ); + + // onStart fires first (before any progress), exactly once, with the number + // of tables that will be attempted. + expect($events[0])->toBe(['start', 1]); + expect(array_values(array_filter($events, static fn (array $e): bool => $e[0] === 'start'))) + ->toBe([['start', 1]]); +}); + +it('announces the one-shot phases via the timings status on InProgress before the row loop', function (): void { + $source = makeOrchestratorConnection('source'); + $target = makeOrchestratorConnection('target'); + $schema = makeOrchestratorSchema(); + $config = makeOrchestratorConfig(clear: ClearMode::Truncate); + + DB::shouldReceive('connection')->andReturnSelf(); + DB::shouldReceive('selectOne')->andReturn((object) ['c' => 1]); + DB::shouldReceive('select')->andReturn([(object) ['id' => 1]], []); + DB::shouldReceive('statement')->andReturnTrue(); + DB::shouldReceive('table')->andReturnSelf(); + DB::shouldReceive('insert')->andReturnTrue(); + DB::shouldReceive('purge')->andReturnNull(); + + /** @var list $phases */ + $phases = []; + $orchestrator = makeOrchestrator(); + $orchestrator->run( + $config, + $source, + $target, + $schema, + true, + [], + [], + static function (string $tbl, TableRunStatus $status, int $rows, int $skipped, array $skippedRows, ?StatsTableTransferData $timings = null) use (&$phases): void { + // Read the phase at emit time (the stats object is mutated in place). + if ($status === TableRunStatus::InProgress && $timings?->status instanceof TableRunPhase) { + $phases[] = $timings->status; + } + }, + trackRowTotals: true, + ); + + // Counting rows is announced first (before the count), then clearing the target, + // both ahead of the per-chunk loop phase (Insert). No FK-disable phase here. + expect($phases[0])->toBe(TableRunPhase::CountingRows); + expect($phases[1])->toBe(TableRunPhase::Clear); + expect($phases[0]->isOneShot())->toBeTrue(); + expect($phases[2])->toBe(TableRunPhase::Insert); +}); + it('does not fire onTableStart for tables skipped by --skip flag', function (): void { $source = makeOrchestratorConnection('source'); $target = makeOrchestratorConnection('target'); @@ -899,3 +984,110 @@ static function (string $tbl, TableRunStatus $status, int $rows, int $skipped, a expect($progressArgs['skippedRows'])->toHaveCount(1); expect($progressArgs['skippedRows'][0]->sqlError)->toBe('SQLSTATE[23000]: row failure on chunk 1'); }); + +it('provides TableTransferTimingsData with per-loop entries, stats-over-time and throughput to onProgress', function (): void { + $source = makeOrchestratorConnection('source'); + $target = makeOrchestratorConnection('target'); + $schema = makeOrchestratorSchema(); + $config = new CloningConfigData( + version: '1', + connectionName: 'source', + options: new CloningOptionsData( + chunkSize: 2, + enforceColumnTypes: false, + dropUnknownTables: false, + dropExtraColumns: false, + disableForeignKeyChecks: false, + fakerLocale: 'en_US', + ), + tables: [ + new TableCloningConfigData( + tableName: 'users', + rows: new TableRowConfigData(strategy: 'full', limit: null, sortBy: null, clear: ClearMode::None), + columns: [], + ), + ], + ); + + DB::shouldReceive('connection')->andReturnSelf(); + DB::shouldReceive('select')->andReturn( + [(object) ['id' => 1], (object) ['id' => 2]], + [(object) ['id' => 3]], + ); + DB::shouldReceive('selectOne')->andReturn((object) ['c' => 3]); + DB::shouldReceive('table')->andReturnSelf(); + DB::shouldReceive('insert')->andReturnTrue(); + DB::shouldReceive('purge')->andReturnNull(); + + /** @var list $events */ + $events = []; + $orchestrator = makeOrchestrator(); + $orchestrator->run( + $config, + $source, + $target, + $schema, + true, + [], + [], + static function (string $tbl, TableRunStatus $status, int $rows, int $skipped, array $skippedRows, ?StatsTableTransferData $timings = null) use (&$events): void { + $events[] = ['status' => $status, 'rows' => $rows, 'timings' => $timings]; + }, + trackRowTotals: true, + ); + + $pending = array_values(array_filter($events, static fn (array $e): bool => $e['status'] === TableRunStatus::InProgress)); + $final = array_values(array_filter($events, static fn (array $e): bool => $e['status'] === TableRunStatus::Transferred)); + + // One InProgress for the counting-rows phase plus one per chunk (2 chunks). + expect($pending)->toHaveCount(3); + expect($final)->toHaveCount(1); + + $timings = $final[0]['timings']; + expect($timings)->toBeInstanceOf(StatsTableTransferData::class); + expect($timings->totalRows)->toBe(3); + expect($timings->rowsDone)->toBe(3); + expect($timings->rowsSkipped)->toBe(0); + expect($timings->rowsRemaining)->toBe(0); + expect($timings->percentComplete)->toBe(100.0); + + expect($timings->loops->count())->toBe(2); + expect($timings->statsOverTime->count())->toBe(2); + + /** @var StatsLoopData $loop0 */ + $loop0 = $timings->loops->get(0); + /** @var StatsLoopData $loop1 */ + $loop1 = $timings->loops->get(1); + expect($loop0->loopIndex)->toBe(0); + expect($loop0->chunkRows)->toBe(2); + expect($loop0->rowsDone)->toBe(2); + expect($loop0->rowsSkipped)->toBe(0); + expect($loop0->totalRows)->toBe(3); + expect($loop1->loopIndex)->toBe(1); + expect($loop1->chunkRows)->toBe(1); + expect($loop1->rowsDone)->toBe(1); + + $snap0 = $timings->statsOverTime->get(0); + $snap1 = $timings->statsOverTime->get(1); + expect($snap0->rowsDoneCumulative)->toBe(2); + expect($snap1->rowsDoneCumulative)->toBe(3); + expect($snap0->loopsRecorded)->toBe(1); + expect($snap1->loopsRecorded)->toBe(2); + expect($snap1->percentComplete)->toBe(100.0); + // Snapshot holds immutable scalars captured at record time, unaffected by later loops. + expect($snap0->insertSecondsPerMillionRows)->not->toBeNull(); + + $insertAgg = $timings->aggregate(TableRunPhase::Insert); + expect($insertAgg->count)->toBe(2); + expect($insertAgg->min)->toBeLessThanOrEqual($insertAgg->max); + expect($insertAgg->averageSeconds)->toBeGreaterThanOrEqual(0.0); + + expect($insertAgg->secondsPerMillionRows)->not->toBeNull(); + expect($insertAgg->latestSecondsPerMillionRows)->not->toBeNull(); +}); + +it('returns null throughput and percent when total rows is zero on StatsTableTransferData', function (): void { + $timings = new StatsTableTransferData; + expect($timings->aggregate(TableRunPhase::Insert)->secondsPerMillionRows)->toBeNull(); + expect($timings->percentComplete)->toBeNull(); +}); From 0894076d48c631362bc3f667291c8ea8c21811dc Mon Sep 17 00:00:00 2001 From: Hagen Pommer Date: Mon, 27 Jul 2026 14:31:38 +0200 Subject: [PATCH 08/15] feat(cloning): render nested progress bars for cloning:run at -v/-vv/-vvv --- app/Commands/Cloning/RunCommand.php | 329 ++++++++++++++++++++++++---- 1 file changed, 284 insertions(+), 45 deletions(-) diff --git a/app/Commands/Cloning/RunCommand.php b/app/Commands/Cloning/RunCommand.php index 89a6bda..5cd0d11 100644 --- a/app/Commands/Cloning/RunCommand.php +++ b/app/Commands/Cloning/RunCommand.php @@ -10,7 +10,9 @@ use App\Data\Cloning\DryRunResultData; use App\Data\Cloning\DryRunTableData; use App\Data\Cloning\KeyRemappingConfigData; +use App\Data\Cloning\StatsTableTransferData; use App\Data\Cloning\TableCloningConfigData; +use App\Data\Cloning\TableRunPhase; use App\Data\Cloning\TableRunResultData; use App\Data\Cloning\TableRunStatus; use App\Data\ConnectionData; @@ -55,6 +57,11 @@ use Illuminate\Support\Facades\Storage; use LaravelZero\Framework\Commands\Command; use RuntimeException; +use Symfony\Component\Console\Helper\ProgressBar; +use Symfony\Component\Console\Helper\Table; +use Symfony\Component\Console\Helper\TableSeparator; +use Symfony\Component\Console\Output\BufferedOutput; +use Symfony\Component\Console\Output\ConsoleOutput; use Symfony\Component\Console\Output\OutputInterface; use Symfony\Component\Yaml\Yaml; use Throwable; @@ -103,10 +110,29 @@ public function handle( $verbosity = $this->getOutput()->getVerbosity(); $isVerbose = $verbosity >= OutputInterface::VERBOSITY_VERBOSE; $isVeryVerbose = $verbosity >= OutputInterface::VERBOSITY_VERY_VERBOSE; + $isDebug = $verbosity >= OutputInterface::VERBOSITY_DEBUG; + + // Live nested progress bars are driven by verbosity on an interactive TTY: + // `-v` shows the bars (compact), `-vv` adds full per-phase throughput to the + // per-table bar, `-vvv` additionally prints the per-table timing summary. + $out = $this->output->getOutput(); + $showProgress = $isVerbose + && ! $ci + && $this->output->isDecorated() + && $out instanceof ConsoleOutput; // Map Symfony verbosity to stderr log threshold so `-v` / `-vv` surface - // info / debug events on stderr without needing a separate live-output path. - $stderrLevel = $isVeryVerbose ? 'debug' : ($isVerbose ? 'info' : ($ci ? 'error' : 'warning')); + // info / debug events on stderr when piped. When the bars are live they own + // the terminal, so the stderr stream is silenced to avoid corrupting them — + // skips/failures still surface as result lines and in the final summary. + // (Set before the channel is first resolved, so it actually takes effect.) + $stderrLevel = match (true) { + $showProgress => 'emergency', + $isVeryVerbose => 'debug', + $isVerbose => 'info', + $ci => 'error', + default => 'warning', + }; config(['logging.channels.stderr.level' => $stderrLevel]); $step = new VerboseStepRenderer($this->output, $ci); @@ -428,6 +454,23 @@ public function handle( $dotColumn = 0; $maxDotColumns = 70; + $overallBar = null; + $tableBar = null; + $logSection = null; + + if ($showProgress) { + // Live bars own the terminal: every write must go through a section, or + // Symfony's cursor math desyncs and re-prints the bars. Sections created + // first render higher up — the results log on top, the per-table bar in + // the middle, the overall bar pinned at the bottom. All reused for the run. + $logSection = $out->section(); + $tableBar = new ProgressBar($out->section()); + $overallBar = new ProgressBar($out->section()); + // Leading newline keeps a blank line between the per-table bar and the + // overall bar (the section owns both lines, so it stays stable on redraw). + $overallBar->setFormat("\n tables [%bar%] %current%/%max%"); + } + $dumpSink = $targetIsDump ? new SqlDumpService(new DumpDialectFactory, new DumpArchiver) : null; @@ -440,69 +483,106 @@ public function handle( skipSchema: $skipSchema, skipTables: $skipTables, onlyTables: $onlyTables, - onProgress: function (string $tableName, TableRunStatus $status, int $rows, int $skipped, array $skippedRows) use ($step, $isVerbose, $ci, &$notFoundTables, &$schemaFailureTables, &$dotColumn, $maxDotColumns): void { + onProgress: function (string $tableName, TableRunStatus $status, int $rows, int $skipped, array $skippedRows, ?StatsTableTransferData $timings = null) use ($isVerbose, $isVeryVerbose, $isDebug, $ci, &$notFoundTables, &$schemaFailureTables, &$dotColumn, $maxDotColumns, $showProgress, $overallBar, $tableBar, $logSection): void { + // Record pre-skip outcomes for the final summary (every mode). + if ($status === TableRunStatus::NotFound) { + $notFoundTables[] = $tableName; + } elseif ($status === TableRunStatus::SkippedBySchemaFailure) { + $schemaFailureTables[] = $tableName; + } + if ($ci) { - if ($status === TableRunStatus::NotFound) { - $notFoundTables[] = $tableName; - } elseif ($status === TableRunStatus::SkippedBySchemaFailure) { - $schemaFailureTables[] = $tableName; + return; + } + + if ($status === TableRunStatus::InProgress) { + if ($showProgress && $timings instanceof StatsTableTransferData) { + $this->advanceTableBar($tableBar, $timings, $isVeryVerbose); } return; } - if ($isVerbose) { - if ($status === TableRunStatus::Transferred) { - $suffix = sprintf('(%s rows%s)', number_format($rows), $skipped > 0 ? ', '.$skipped.' skipped' : ''); - $step->success($suffix); - $this->renderSkipGroups($step, $skippedRows); - } elseif ($status === TableRunStatus::Failed) { - $step->fail(); - $this->renderSkipGroups($step, $skippedRows); - } elseif ($status === TableRunStatus::NotFound) { - $this->line(sprintf(' ? %s — not found in source, skipped', $tableName)); - $notFoundTables[] = $tableName; - } elseif ($status === TableRunStatus::SkippedBySchemaFailure) { - $this->line(sprintf(' S %s — schema replication failed, skipped', $tableName)); - $schemaFailureTables[] = $tableName; + // ── Terminal status ── + // Quiet mode (not verbose): compact one-char dot indicators. + if (! $isVerbose) { + $indicator = match ($status) { + TableRunStatus::Transferred => $skipped > 0 ? 'F' : '.', + TableRunStatus::Failed => 'E', + TableRunStatus::NotFound => '?', + TableRunStatus::SkippedBySchemaFailure => 'S', + default => null, + }; + + if ($indicator !== null) { + $this->output->write($indicator); + $dotColumn++; + + if ($dotColumn >= $maxDotColumns) { + $this->output->writeln(''); + $dotColumn = 0; + } } return; } - // Normal mode: dot indicators wrapped at 70 chars - if ($status === TableRunStatus::NotFound) { - $notFoundTables[] = $tableName; - } elseif ($status === TableRunStatus::SkippedBySchemaFailure) { - $schemaFailureTables[] = $tableName; + // Verbose. Scrolling detail goes to the log section when bars are live + // (so it never corrupts them), otherwise straight to the console. + $sink = $showProgress ? $logSection : $this->output; + + // Only Transferred/Failed tables ever started a per-table bar + // (via onTableStart); NotFound/SchemaFailure never did. + if ($showProgress && ($status === TableRunStatus::Transferred || $status === TableRunStatus::Failed)) { + $tableBar->finish(); } - $indicator = match ($status) { - TableRunStatus::Transferred => $skipped > 0 ? 'F' : '.', - TableRunStatus::Failed => 'E', - TableRunStatus::NotFound => '?', - TableRunStatus::SkippedBySchemaFailure => 'S', - default => null, - }; + $suffix = $skipped > 0 ? ', '.number_format($skipped).' skipped' : ''; - if ($indicator !== null) { - $this->output->write($indicator); - $dotColumn++; + if ($status === TableRunStatus::Transferred) { + $sink->writeln(sprintf(' ✓ %s (%s rows%s)', $tableName, number_format($rows), $suffix)); + } elseif ($status === TableRunStatus::Failed) { + $sink->writeln(sprintf(' ✗ %s — transfer failed', $tableName)); + } elseif ($status === TableRunStatus::NotFound) { + $sink->writeln(sprintf(' ? %s — not found in source, skipped', $tableName)); + } elseif ($status === TableRunStatus::SkippedBySchemaFailure) { + $sink->writeln(sprintf(' S %s — schema replication failed, skipped', $tableName)); + } + + if ($status === TableRunStatus::Transferred || $status === TableRunStatus::Failed) { + $this->renderSkipGroups($sink, $skippedRows); - if ($dotColumn >= $maxDotColumns) { - $this->output->writeln(''); - $dotColumn = 0; + // `-vvv`: per-table timing summary table. + if ($isDebug && $timings instanceof StatsTableTransferData) { + $this->renderTimingSummary($sink, $timings); } } + + if ($showProgress) { + $overallBar->advance(); + } }, keyRemapping: $keyRemappingService, breakOnFailure: (bool) $this->option('break-on-failure'), - onTableStart: $isVerbose && ! $ci - ? fn (string $tableName) => $step->start(' '.$tableName) - : null, + onTableStart: $showProgress + ? function (string $tableName) use ($tableBar): void { + $this->startTableBar($tableBar, $tableName); + } + : null, dumpSink: $dumpSink, + onStart: $showProgress + ? function (int $totalTables) use ($overallBar): void { + $overallBar->setMaxSteps($totalTables); + $overallBar->start(); + } + : null, + trackRowTotals: $showProgress, ); + if ($overallBar instanceof ProgressBar) { + $overallBar->finish(); + } + $finishedAt = new DateTimeImmutable('now', new DateTimeZone('UTC')); // Finalise the dump: write postamble, compress to ZIP, delete the .sql. @@ -665,6 +745,21 @@ public function handle( )); } + // Data-transfer failures with their reason. In live-bar mode the ✗ line is + // terse and the stderr log is silenced, so surface the reason here. + $failedTables = array_values(array_filter($result->tables, static fn (TableRunResultData $t): bool => $t->status === TableRunStatus::Failed)); + + if ($failedTables !== []) { + $this->line(''); + foreach ($failedTables as $failedTable) { + $this->line(sprintf( + ' Error: table %s failed%s', + $failedTable->tableName, + $failedTable->failureReason !== null ? ' — '.$failedTable->failureReason : '', + )); + } + } + $transferredCount = count(array_filter($result->tables, static fn (TableRunResultData $t): bool => $t->status === TableRunStatus::Transferred)); $totalCount = count($result->tables); $duration = $this->formatDuration($result->durationSeconds); @@ -973,10 +1068,154 @@ private function aggregateSkipReasons(array $rows): array return $result; } + /** + * (Re)initialise the nested per-table progress bar for a new table. The row + * total is not known yet (it arrives with the first activity or loop event), + * so the bar starts in the indeterminate format. The `%message%` slot carries + * the one-shot activity first, then per-chunk throughput. + */ + private function startTableBar(ProgressBar $bar, string $tableName): void + { + $bar->setFormat(' %table% [%bar%] %message%'); + $bar->setMessage($tableName, 'table'); + $bar->setMessage('', 'message'); + $bar->start(0); + } + + /** + * Advance the per-table bar from a chunk's cumulative timings. On the first + * event with a known row total it switches to the determinate bar format. + */ + private function advanceTableBar(ProgressBar $bar, StatsTableTransferData $stats, bool $verbose): void + { + $isOneShot = $stats->status?->isOneShot(); + + $message = match ($isOneShot) { + true => $this->formatStatus($stats), + false => $this->formatProgress($stats, $verbose), + null => '' + }; + + $bar->setMessage($message, 'message'); + + if ($isOneShot) { + $bar->display(); + + return; + } + + if ($stats->totalRows > 0) { + if ($bar->getMaxSteps() !== $stats->totalRows) { + $bar->setMaxSteps($stats->totalRows); + $bar->setFormat(' %table% [%bar%] %current%/%max% (%percent:3s%%) %message%'); + } + + $bar->setProgress(min($stats->rowsProcessed, $stats->totalRows)); + + return; + } + + $bar->setProgress($stats->rowsProcessed); + } + + private function formatStatus(StatsTableTransferData $stats): string + { + return $stats->status instanceof TableRunPhase ? $stats->status->label().'…' : ''; + } + + /** + * Throughput text shown on the per-table bar: overall only in default mode, + * full per-phase breakdown in verbose mode. + */ + private function formatProgress(StatsTableTransferData $stats, bool $verbose): string + { + $overall = trim($this->formatThroughput($stats->loopAggregate->latestSecondsPerMillionRows)); + + if (! $verbose) { + return $overall === '—' ? '' : $overall; + } + + return sprintf( + 'all %s · sel %s · tr %s · ins %s', + $overall, + trim($this->formatThroughput($stats->selectAggregate->latestSecondsPerMillionRows)), + trim($this->formatThroughput($stats->transformAggregate->latestSecondsPerMillionRows)), + trim($this->formatThroughput($stats->insertAggregate->latestSecondsPerMillionRows)), + ); + } + + private function renderTimingSummary(OutputInterface $sink, StatsTableTransferData $timings): void + { + if ($timings->loops->isEmpty()) { + return; + } + + $rows = []; + + foreach (TableRunPhase::cases() as $phase) { + $agg = $timings->aggregate($phase); + if ($agg->count === 0) { + continue; + } + + if ($phase === TableRunPhase::Loop || $phase === TableRunPhase::Select) { + $rows[] = new TableSeparator; + } + + $rows[] = [ + $phase->value, + (string) $agg->count, + $this->formatSeconds($agg->min ?? 0.0), + $this->formatSeconds($agg->max ?? 0.0), + $this->formatSeconds($agg->averageSeconds ?? 0.0), + $this->formatSeconds($agg->sum), + $this->formatThroughput($agg->secondsPerMillionRows), + ]; + } + + $indent = str_repeat(' ', 6); + $sink->writeln(sprintf('%s── timing summary ──', $indent)); + $buffer = new BufferedOutput( + $this->output->getVerbosity(), + $this->output->isDecorated(), + $this->output->getFormatter(), + ); + $table = new Table($buffer); + $table->setHeaders(['phase', 'chunks', 'min', 'max', 'avg', 'total', 's/1M rows']); + $table->setRows($rows); + $table->render(); + + foreach (explode("\n", rtrim($buffer->fetch(), "\n")) as $line) { + $sink->writeln($indent.$line); + } + } + + private function formatSeconds(float $seconds): string + { + if ($seconds < 1.0) { + return sprintf('%6.1f ms', $seconds * 1000.0); + } + + return sprintf('%5.1f s', $seconds); + } + + private function formatThroughput(?float $secondsPerMillion): string + { + if ($secondsPerMillion === null) { + return '—'; + } + + if ($secondsPerMillion >= 1.0) { + return sprintf('%5.1f s/M', $secondsPerMillion); + } + + return sprintf('%5.1f ms/M', $secondsPerMillion * 1000.0); + } + /** * @param list $skippedRows */ - private function renderSkipGroups(VerboseStepRenderer $step, array $skippedRows): void + private function renderSkipGroups(OutputInterface $sink, array $skippedRows): void { if ($skippedRows === []) { return; @@ -985,12 +1224,12 @@ private function renderSkipGroups(VerboseStepRenderer $step, array $skippedRows) $groups = $this->aggregateSkipReasons($skippedRows); $shown = array_slice($groups, 0, 10); foreach ($shown as $group) { - $step->note(sprintf(' └ %d× %s', $group['count'], $group['message'])); + $sink->writeln(sprintf(' └ %d× %s', $group['count'], $group['message'])); } $rest = count($groups) - count($shown); if ($rest > 0) { - $step->note(sprintf(' └ … and %d more error types', $rest)); + $sink->writeln(sprintf(' └ … and %d more error types', $rest)); } } From 80ae18b3ad3899f242dd5484ea88f6534d85fe03 Mon Sep 17 00:00:00 2001 From: Hagen Pommer Date: Mon, 27 Jul 2026 14:31:38 +0200 Subject: [PATCH 09/15] test(cloning): cover progress rendering and timing summary in cloning:run --- .../Commands/Cloning/RunCommandTest.php | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/tests/Feature/Commands/Cloning/RunCommandTest.php b/tests/Feature/Commands/Cloning/RunCommandTest.php index 40e82cf..26b37ce 100644 --- a/tests/Feature/Commands/Cloning/RunCommandTest.php +++ b/tests/Feature/Commands/Cloning/RunCommandTest.php @@ -1599,6 +1599,26 @@ function sqliteCloningYaml(bool $withMissing = false, string $strategy = 'full', @unlink($target); }); +it('prints the per-table timing summary at -vvv on a real run', function (): void { + Storage::fake('local'); + $source = sys_get_temp_dir().'/clonio_run_src_'.uniqid().'.db'; + $target = sys_get_temp_dir().'/clonio_run_tgt_'.uniqid().'.db'; + makeSqliteDb($source, rows: 3); + makeSqliteDb($target, rows: 0); + writeSqliteClonioJson($source, $target); + Storage::disk('local')->put('test.cloning.yaml', sqliteCloningYaml()); + + // Live bars need a real TTY; test output is a non-decorated BufferedOutput, so + // -vvv takes the fallback path but still emits the per-table timing summary. + $this->artisan('cloning:run test.cloning.yaml --target=staging -vvv') + ->expectsOutputToContain('timing summary') + ->expectsOutputToContain('Tables:') + ->assertExitCode(ExitCode::Success->value); + + @unlink($source); + @unlink($target); +}); + it('renders the verbose schema-comparison phase on a real run', function (): void { Storage::fake('local'); $source = sys_get_temp_dir().'/clonio_run_src_'.uniqid().'.db'; From 2b977c78fca125882faba9f5a28cfbde5ed35127 Mon Sep 17 00:00:00 2001 From: Hagen Pommer Date: Fri, 7 Aug 2026 16:04:16 +0200 Subject: [PATCH 10/15] feat(cloning): show per-table ETA on the progress bar in verbose modes --- app/Commands/Cloning/RunCommand.php | 27 ++++++++++++------ app/Data/Cloning/StatsLoopSnapshotData.php | 8 +++--- app/Data/Cloning/StatsPhaseAggregateData.php | 28 +++++++++++++++---- app/Data/Cloning/StatsTableTransferData.php | 23 ++++++++++++--- .../Cloning/StatsTableTransferDataTest.php | 28 +++++++++++++++++++ .../Cloning/CloningRunOrchestratorTest.php | 8 +++--- 6 files changed, 95 insertions(+), 27 deletions(-) diff --git a/app/Commands/Cloning/RunCommand.php b/app/Commands/Cloning/RunCommand.php index 5cd0d11..3e034d8 100644 --- a/app/Commands/Cloning/RunCommand.php +++ b/app/Commands/Cloning/RunCommand.php @@ -1021,6 +1021,13 @@ private function formatDuration(float $seconds): string return sprintf('%.0fs', $seconds); } + if ($seconds >= 3600) { + $hours = (int) ($seconds / 3600); + $minutes = (int) (fmod($seconds, 3600) / 60); + + return sprintf('%dh %dm', $hours, $minutes); + } + $minutes = (int) ($seconds / 60); $remaining = $seconds % 60; @@ -1124,23 +1131,25 @@ private function formatStatus(StatsTableTransferData $stats): string } /** - * Throughput text shown on the per-table bar: overall only in default mode, - * full per-phase breakdown in verbose mode. + * Progress detail shown on the per-table bar: ETA plus the overall pace in every + * verbose mode, expanded to the full per-phase pace breakdown in very verbose mode. */ private function formatProgress(StatsTableTransferData $stats, bool $verbose): string { - $overall = trim($this->formatThroughput($stats->loopAggregate->latestSecondsPerMillionRows)); + $eta = 'ETA '.$this->formatDuration($stats->estimatedSecondsRemaining); + $overall = trim($this->formatThroughput($stats->loopAggregate->latestPacePerMillion)); if (! $verbose) { - return $overall === '—' ? '' : $overall; + return $overall === '—' ? $eta : $eta.' · '.$overall; } return sprintf( - 'all %s · sel %s · tr %s · ins %s', + '%s · all %s · sel %s · tr %s · ins %s', + $eta, $overall, - trim($this->formatThroughput($stats->selectAggregate->latestSecondsPerMillionRows)), - trim($this->formatThroughput($stats->transformAggregate->latestSecondsPerMillionRows)), - trim($this->formatThroughput($stats->insertAggregate->latestSecondsPerMillionRows)), + trim($this->formatThroughput($stats->selectAggregate->latestPacePerMillion)), + trim($this->formatThroughput($stats->transformAggregate->latestPacePerMillion)), + trim($this->formatThroughput($stats->insertAggregate->latestPacePerMillion)), ); } @@ -1169,7 +1178,7 @@ private function renderTimingSummary(OutputInterface $sink, StatsTableTransferDa $this->formatSeconds($agg->max ?? 0.0), $this->formatSeconds($agg->averageSeconds ?? 0.0), $this->formatSeconds($agg->sum), - $this->formatThroughput($agg->secondsPerMillionRows), + $this->formatThroughput($agg->pacePerMillion), ]; } diff --git a/app/Data/Cloning/StatsLoopSnapshotData.php b/app/Data/Cloning/StatsLoopSnapshotData.php index 5629842..93ceaa5 100644 --- a/app/Data/Cloning/StatsLoopSnapshotData.php +++ b/app/Data/Cloning/StatsLoopSnapshotData.php @@ -20,9 +20,9 @@ public function __construct( public int $rowsDoneCumulative, public int $rowsSkippedCumulative, public ?float $percentComplete, - public ?float $selectSecondsPerMillionRows, - public ?float $transformSecondsPerMillionRows, - public ?float $insertSecondsPerMillionRows, - public ?float $loopSecondsPerMillionRows, + public ?float $selectPacePerMillion, + public ?float $transformPacePerMillion, + public ?float $insertPacePerMillion, + public ?float $loopPacePerMillion, ) {} } diff --git a/app/Data/Cloning/StatsPhaseAggregateData.php b/app/Data/Cloning/StatsPhaseAggregateData.php index 045dc41..5980f05 100644 --- a/app/Data/Cloning/StatsPhaseAggregateData.php +++ b/app/Data/Cloning/StatsPhaseAggregateData.php @@ -31,19 +31,35 @@ final class StatsPhaseAggregateData } /** - * Aggregate seconds per 1,000,000 rows across all samples, + * Aggregate seconds per row across all samples, * or null when no rows have been processed. */ - public ?float $secondsPerMillionRows { + public ?float $pace { get => $this->rowsProcessed > 0 - ? ($this->sum / $this->rowsProcessed) * 1_000_000.0 + ? ($this->sum / $this->rowsProcessed) : null; } - /** Latest-sample seconds per 1,000,000 rows, or null when unavailable. */ - public ?float $latestSecondsPerMillionRows { + /** + * Aggregate seconds per 1,000,000 rows across all samples, + * or null when no rows have been processed. + */ + public ?float $pacePerMillion { + get => ($pace = $this->pace) !== null + ? $pace * 1_000_000.0 + : null; + } + + public ?float $latestPace { get => $this->last !== null && $this->lastRows !== null && $this->lastRows > 0 - ? ($this->last / $this->lastRows) * 1_000_000.0 + ? $this->last / $this->lastRows + : null; + } + + /** Latest-sample seconds per 1,000,000 rows, or null when unavailable. */ + public ?float $latestPacePerMillion { + get => ($pace = $this->latestPace) !== null + ? $pace * 1_000_000.0 : null; } diff --git a/app/Data/Cloning/StatsTableTransferData.php b/app/Data/Cloning/StatsTableTransferData.php index 9c2f545..d1a1e5d 100644 --- a/app/Data/Cloning/StatsTableTransferData.php +++ b/app/Data/Cloning/StatsTableTransferData.php @@ -55,6 +55,21 @@ final class StatsTableTransferData : null; } + /** + * Estimated wall-clock seconds until this table finishes, from the latest + * loop pace × outstanding rows. 0.0 once no rows remain, and 0.0 until a + * row total and at least one completed loop are known. + */ + public float $estimatedSecondsRemaining { + get { + if ($this->rowsRemaining <= 0) { + return 0.0; + } + + return $this->loopAggregate->latestPace * $this->rowsRemaining; + } + } + public private(set) StatsPhaseAggregateData $selectAggregate; public private(set) StatsPhaseAggregateData $transformAggregate; @@ -125,10 +140,10 @@ public function recordLoop(StatsLoopData $loop): void rowsDoneCumulative: $this->rowsDone, rowsSkippedCumulative: $this->rowsSkipped, percentComplete: $this->percentComplete, - selectSecondsPerMillionRows: $this->selectAggregate->secondsPerMillionRows, - transformSecondsPerMillionRows: $this->transformAggregate->secondsPerMillionRows, - insertSecondsPerMillionRows: $this->insertAggregate->secondsPerMillionRows, - loopSecondsPerMillionRows: $this->loopAggregate->secondsPerMillionRows, + selectPacePerMillion: $this->selectAggregate->pacePerMillion, + transformPacePerMillion: $this->transformAggregate->pacePerMillion, + insertPacePerMillion: $this->insertAggregate->pacePerMillion, + loopPacePerMillion: $this->loopAggregate->pacePerMillion, )); } diff --git a/tests/Unit/Data/Cloning/StatsTableTransferDataTest.php b/tests/Unit/Data/Cloning/StatsTableTransferDataTest.php index f8e3a8b..9a190f2 100644 --- a/tests/Unit/Data/Cloning/StatsTableTransferDataTest.php +++ b/tests/Unit/Data/Cloning/StatsTableTransferDataTest.php @@ -2,6 +2,7 @@ declare(strict_types=1); +use App\Data\Cloning\StatsLoopData; use App\Data\Cloning\StatsTableTransferData; use App\Data\Cloning\TableRunPhase; @@ -27,3 +28,30 @@ // A sibling one-shot phase that still never ran stays at count 0. expect($stats->aggregate(TableRunPhase::DisableFkChecks)->count)->toBe(0); }); + +it('estimates remaining time from the latest loop pace', function (): void { + $stats = new StatsTableTransferData; + $stats->setTotalRows(100); + $stats->recordLoop(new StatsLoopData( + loopIndex: 0, + chunkRows: 10, + selectSeconds: 0.2, + transformSeconds: 0.1, + insertSeconds: 0.7, + overallSeconds: 1.0, + rowsDone: 10, + rowsSkipped: 0, + totalRows: 100, + )); + + // 90 rows remaining × (1.0s / 10 rows) = 9.0s. + expect($stats->estimatedSecondsRemaining)->toBe(9.0); +}); + +it('reports a zero ETA when nothing remains or no pace is known yet', function (): void { + $stats = new StatsTableTransferData; + expect($stats->estimatedSecondsRemaining)->toBe(0.0); // nothing to do + + $stats->setTotalRows(100); + expect($stats->estimatedSecondsRemaining)->toBe(0.0); // rows remain but no completed loop → no pace +}); diff --git a/tests/Unit/Services/Cloning/CloningRunOrchestratorTest.php b/tests/Unit/Services/Cloning/CloningRunOrchestratorTest.php index 9678c10..2315679 100644 --- a/tests/Unit/Services/Cloning/CloningRunOrchestratorTest.php +++ b/tests/Unit/Services/Cloning/CloningRunOrchestratorTest.php @@ -1075,19 +1075,19 @@ static function (string $tbl, TableRunStatus $status, int $rows, int $skipped, a expect($snap1->loopsRecorded)->toBe(2); expect($snap1->percentComplete)->toBe(100.0); // Snapshot holds immutable scalars captured at record time, unaffected by later loops. - expect($snap0->insertSecondsPerMillionRows)->not->toBeNull(); + expect($snap0->insertPacePerMillion)->not->toBeNull(); $insertAgg = $timings->aggregate(TableRunPhase::Insert); expect($insertAgg->count)->toBe(2); expect($insertAgg->min)->toBeLessThanOrEqual($insertAgg->max); expect($insertAgg->averageSeconds)->toBeGreaterThanOrEqual(0.0); - expect($insertAgg->secondsPerMillionRows)->not->toBeNull(); - expect($insertAgg->latestSecondsPerMillionRows)->not->toBeNull(); + expect($insertAgg->pacePerMillion)->not->toBeNull(); + expect($insertAgg->latestPacePerMillion)->not->toBeNull(); }); it('returns null throughput and percent when total rows is zero on StatsTableTransferData', function (): void { $timings = new StatsTableTransferData; - expect($timings->aggregate(TableRunPhase::Insert)->secondsPerMillionRows)->toBeNull(); + expect($timings->aggregate(TableRunPhase::Insert)->pacePerMillion)->toBeNull(); expect($timings->percentComplete)->toBeNull(); }); From 714d35750f9fe4b273536bafbf337b5d2ad65816 Mon Sep 17 00:00:00 2001 From: Hagen Pommer Date: Fri, 7 Aug 2026 11:04:59 +0200 Subject: [PATCH 11/15] perf(cloning): use keyset pagination for chunked reads to avoid deep OFFSET scans --- .../Cloning/CloningRunOrchestrator.php | 156 +++++++++++++++++- .../Cloning/CloningRunOrchestratorTest.php | 57 +++++++ 2 files changed, 206 insertions(+), 7 deletions(-) diff --git a/app/Services/Cloning/CloningRunOrchestrator.php b/app/Services/Cloning/CloningRunOrchestrator.php index ccb8c52..af17d01 100644 --- a/app/Services/Cloning/CloningRunOrchestrator.php +++ b/app/Services/Cloning/CloningRunOrchestrator.php @@ -208,6 +208,7 @@ public function run( $dumpSink, $keyRemapping, $config->keyRemapping, + $pkColumns, ), null] : $this->transferTable( $config->options, @@ -322,6 +323,9 @@ private function transferTable( /** @var list $skippedRows */ $skippedRows = []; + /** @var list $sortKeys */ + $sortKeys = []; + $stats = new StatsTableTransferData; // Only pay for the row count when a consumer actually wants per-table @@ -360,10 +364,11 @@ private function transferTable( $tOverall = microtime(true); $stats->setStatus(TableRunPhase::Select); $tSelect = microtime(true); + $seeking = $sortKeys !== []; + $sql = $this->buildChunkQuery($tableConfig, $source, $offset, $chunkSize, $pkColumns, $sortKeys); + $bindings = $seeking ? array_map(static fn (array $k): mixed => $k['lastValue'], $sortKeys) : []; /** @var list $chunk */ - $chunk = DB::connection($sourceConn)->select( - $this->buildChunkQuery($tableConfig, $source, $offset, $chunkSize) - ); + $chunk = DB::connection($sourceConn)->select($sql, $bindings); $selectSeconds = microtime(true) - $tSelect; if ($chunk === []) { @@ -439,6 +444,14 @@ private function transferTable( ($onProgress)($tableConfig->tableName, TableRunStatus::InProgress, $rows, $skipped, [], $stats); + // Advance the keyset cursor to the last row of this chunk (if seeking). + if ($sortKeys !== []) { + $lastRow = (array) array_last($chunk); + foreach ($sortKeys as $i => $sortKey) { + $sortKeys[$i]['lastValue'] = $lastRow[$sortKey['key']] ?? null; + } + } + $offset += count($chunk); $loopIndex++; } while (count($chunk) === $chunkSize); @@ -470,6 +483,7 @@ private function transferTable( * INSERT batches into the dump file. Mirrors transferTable() minus the live * target connection. Returns [rowsWritten, 0, hasFailed, failureReason, []]. * + * @param list $pkColumns * @return array{int, int, bool, ?string, list} */ private function dumpTable( @@ -480,6 +494,7 @@ private function dumpTable( SqlDumpService $dumpSink, ?KeyRemappingService $keyRemapping = null, ?KeyRemappingConfigData $keyRemappingConfig = null, + array $pkColumns = [], ): array { $sourceConn = $this->connector->open($source); @@ -487,12 +502,16 @@ private function dumpTable( $offset = 0; $chunkSize = $options->chunkSize; + /** @var list $sortKeys */ + $sortKeys = []; + try { do { + $seeking = $sortKeys !== []; + $sql = $this->buildChunkQuery($tableConfig, $source, $offset, $chunkSize, $pkColumns, $sortKeys); + $bindings = $seeking ? array_map(static fn (array $k): mixed => $k['lastValue'], $sortKeys) : []; /** @var list $chunk */ - $chunk = DB::connection($sourceConn)->select( - $this->buildChunkQuery($tableConfig, $source, $offset, $chunkSize) - ); + $chunk = DB::connection($sourceConn)->select($sql, $bindings); if ($chunk === []) { break; @@ -501,6 +520,14 @@ private function dumpTable( $transformed = $this->transformChunk($chunk, $tableConfig, $engine, $keyRemapping, $keyRemappingConfig); $dumpSink->writeRows($tableConfig->tableName, $transformed); $rows += count($transformed); + + if ($sortKeys !== []) { + $lastRow = (array) array_last($chunk); + foreach ($sortKeys as $i => $sortKey) { + $sortKeys[$i]['lastValue'] = $lastRow[$sortKey['key']] ?? null; + } + } + $offset += count($chunk); } while (count($chunk) === $chunkSize); @@ -601,7 +628,117 @@ private function countSourceRows(string $sourceConn, TableCloningConfigData $con } } - private function buildChunkQuery(TableCloningConfigData $config, ConnectionData $source, int $offset, int $limit): string + /** + * Build the SELECT for the next chunk. + * + * When the driver and row strategy allow keyset (seek) pagination and the source + * has a usable ordering, `$sortKeys` is populated with the ordering columns on the + * first call (each `lastValue` null) and the caller fills every `lastValue` from the + * last fetched row before the next call; the query then seeks past that row with a + * bound row-value comparison instead of a growing OFFSET. When keyset does not apply + * (no PK, SQL Server) `$sortKeys` is cleared and the legacy LIMIT/OFFSET SQL is used. + * + * @param list $pkColumns + * @param list $sortKeys by-reference seek cursor + */ + private function buildChunkQuery(TableCloningConfigData $config, ConnectionData $source, int $offset, int $limit, array $pkColumns, array &$sortKeys): string + { + $keyColumns = $this->keysetColumns($config, $source->type, $pkColumns); + + // Keyset not applicable → legacy OFFSET pagination. + if ($keyColumns === []) { + $sortKeys = []; + + return $this->offsetChunkQuery($config, $source, $offset, $limit); + } + + $rows = $config->rows; + $quotedTable = $this->quoteTable($config->tableName, $source->type); + $direction = $rows->strategy === 'last' ? 'DESC' : 'ASC'; + $orderBy = implode(', ', array_map( + fn (string $c): string => $this->quoteIdentifier($c, $source->type).' '.$direction, + $keyColumns, + )); + + // Respect the row-strategy limit for the ordered (non-full) strategies. + $chunkLimit = $limit; + if ($rows->strategy !== 'full') { + $totalLimit = $rows->limit ?? PHP_INT_MAX; + $chunkLimit = min($limit, max(0, $totalLimit - $offset)); + + if ($chunkLimit <= 0) { + $sortKeys = []; + + return sprintf('SELECT * FROM %s WHERE 1=0', $quotedTable); + } + } + + // First call: establish the cursor columns and page from the start (no WHERE). + if ($sortKeys === []) { + $sortKeys = array_map(static fn (string $c): array => ['key' => $c, 'lastValue' => null], $keyColumns); + + return sprintf('SELECT * FROM %s ORDER BY %s LIMIT %d', $quotedTable, $orderBy, $chunkLimit); + } + + // Subsequent calls: seek past the last fetched row via a bound row-value comparison. + $columns = implode(', ', array_map(fn (array $k): string => $this->quoteIdentifier($k['key'], $source->type), $sortKeys)); + $placeholders = implode(', ', array_fill(0, count($sortKeys), '?')); + $operator = $direction === 'DESC' ? '<' : '>'; + + return sprintf( + 'SELECT * FROM %s WHERE (%s) %s (%s) ORDER BY %s LIMIT %d', + $quotedTable, + $columns, + $operator, + $placeholders, + $orderBy, + $chunkLimit, + ); + } + + /** + * The ordering columns to seek by, or [] when keyset pagination does not apply + * (SQL Server, or no primary key to guarantee a total order). + * + * @param list $pkColumns + * @return list + */ + private function keysetColumns(TableCloningConfigData $config, DatabaseConnectionType $type, array $pkColumns): array + { + if ($pkColumns === [] || ! $this->supportsKeysetPagination($type)) { + return []; + } + + if ($config->rows->strategy === 'full') { + return $pkColumns; + } + + // Ordered strategies: sort column first, PK columns appended as a tiebreaker so + // the ordering is total even when the sort column has duplicate values. + $keys = [$config->rows->sortBy ?? 'id']; + foreach ($pkColumns as $pk) { + if (! in_array($pk, $keys, true)) { + $keys[] = $pk; + } + } + + return $keys; + } + + private function supportsKeysetPagination(DatabaseConnectionType $type): bool + { + return in_array($type, [ + DatabaseConnectionType::Mysql, + DatabaseConnectionType::MariaDB, + DatabaseConnectionType::PostgreSQL, + DatabaseConnectionType::Sqlite, + ], true); + } + + /** + * Legacy LIMIT/OFFSET (or OFFSET/FETCH) chunk query — used when keyset does not apply. + */ + private function offsetChunkQuery(TableCloningConfigData $config, ConnectionData $source, int $offset, int $limit): string { $table = $config->tableName; $rows = $config->rows; @@ -635,6 +772,11 @@ private function buildChunkQuery(TableCloningConfigData $config, ConnectionData } private function quoteTable(string $name, DatabaseConnectionType $driver): string + { + return $this->quoteIdentifier($name, $driver); + } + + private function quoteIdentifier(string $name, DatabaseConnectionType $driver): string { return match ($driver) { DatabaseConnectionType::Mysql, DatabaseConnectionType::MariaDB => '`'.$name.'`', diff --git a/tests/Unit/Services/Cloning/CloningRunOrchestratorTest.php b/tests/Unit/Services/Cloning/CloningRunOrchestratorTest.php index 2315679..b811a2a 100644 --- a/tests/Unit/Services/Cloning/CloningRunOrchestratorTest.php +++ b/tests/Unit/Services/Cloning/CloningRunOrchestratorTest.php @@ -765,6 +765,63 @@ static function (string $tbl, TableRunStatus $status, int $rows, int $skipped, a expect($phases[2])->toBe(TableRunPhase::Insert); }); +it('uses keyset (seek) pagination instead of OFFSET when a primary key exists', function (): void { + $source = makeOrchestratorConnection('source'); + $target = makeOrchestratorConnection('target'); + $schema = makeOrchestratorSchema(); // users, PK id + $config = new CloningConfigData( + version: '1', + connectionName: 'source', + options: new CloningOptionsData( + chunkSize: 2, + enforceColumnTypes: false, + dropUnknownTables: false, + dropExtraColumns: false, + disableForeignKeyChecks: false, + fakerLocale: 'en_US', + ), + tables: [ + new TableCloningConfigData( + tableName: 'users', + rows: new TableRowConfigData(strategy: 'full', limit: null, sortBy: null, clear: ClearMode::None), + columns: [], + ), + ], + ); + + /** @var list}> $queries */ + $queries = []; + DB::shouldReceive('connection')->andReturnSelf(); + DB::shouldReceive('select')->andReturnUsing(function (string $sql, array $bindings = []) use (&$queries): array { + $queries[] = ['sql' => $sql, 'bindings' => $bindings]; + + return match (count($queries)) { + 1 => [(object) ['id' => 1], (object) ['id' => 2]], + 2 => [(object) ['id' => 3]], + default => [], + }; + }); + DB::shouldReceive('table')->andReturnSelf(); + DB::shouldReceive('insert')->andReturnTrue(); + DB::shouldReceive('purge')->andReturnNull(); + + makeOrchestrator()->run($config, $source, $target, $schema, true, [], [], static function (): void {}); + + // Two chunks: full window (2), then the trailing short window (1) which ends the loop. + expect($queries)->toHaveCount(2); + + // First chunk: ordered + limited, no OFFSET, no WHERE, no bindings. + expect($queries[0]['sql'])->toContain('ORDER BY')->toContain('LIMIT') + ->not->toContain('OFFSET') + ->not->toContain('WHERE'); + expect($queries[0]['bindings'])->toBe([]); + + // Second chunk: seek past the last fetched id via a bound row-value comparison. + expect($queries[1]['sql'])->toContain('WHERE (`id`) > (?)')->toContain('ORDER BY') + ->not->toContain('OFFSET'); + expect($queries[1]['bindings'])->toBe([2]); +}); + it('does not fire onTableStart for tables skipped by --skip flag', function (): void { $source = makeOrchestratorConnection('source'); $target = makeOrchestratorConnection('target'); From 6e618ed6011191856d920d6c9db46053af83a07b Mon Sep 17 00:00:00 2001 From: Hagen Pommer Date: Wed, 12 Aug 2026 12:25:07 +0200 Subject: [PATCH 12/15] feat(cloning): add support for regex table names in cloning configuration --- app/Commands/Cloning/RunCommand.php | 10 ++ app/Services/Cloning/CloningYamlValidator.php | 7 + app/Services/Cloning/TableConfigResolver.php | 141 ++++++++++++++++++ docs/cloning-yaml.md | 42 +++++- specs/PRD-cloning-yaml-schema.md | 20 +++ 5 files changed, 219 insertions(+), 1 deletion(-) create mode 100644 app/Services/Cloning/TableConfigResolver.php diff --git a/app/Commands/Cloning/RunCommand.php b/app/Commands/Cloning/RunCommand.php index 3e034d8..9357d39 100644 --- a/app/Commands/Cloning/RunCommand.php +++ b/app/Commands/Cloning/RunCommand.php @@ -41,6 +41,7 @@ use App\Services\Cloning\EncryptedFileKeyRemappingStore; use App\Services\Cloning\KeyRemappingService; use App\Services\Cloning\SkippedRow; +use App\Services\Cloning\TableConfigResolver; use App\Services\Config\ConfigService; use App\Services\Database\DatabaseConnectionService; use App\Services\Output\VerboseStepRenderer; @@ -105,6 +106,7 @@ public function handle( SchemaInspector $inspector, CloningRunOrchestrator $orchestrator, AuditBuffer $auditBuffer, + TableConfigResolver $tableResolver, ): int { $ci = (bool) $this->option('ci'); $verbosity = $this->getOutput()->getVerbosity(); @@ -365,6 +367,14 @@ public function handle( return ExitCode::ConnectionError->value; } + // Expand regex table keys against the real source schema. From here on + // $config->tables carry concrete table names, so the orchestrator, key + // mapping and audit code all keep working against literal names. + $config = $tableResolver->resolve($config, $sourceSchema); + // Regex keys may have expanded rows.strategy: skip / skip patterns into + // concrete names — re-merge them into the skip list. + $skipTables = array_values(array_unique(array_merge($skipTables, $config->skipTables))); + // ─── Phase 5b: Key Mapping Generation ───────────────────────────────── $keyRemappingService = null; $skipRemapping = (bool) $this->option('skip-remapping-keys'); diff --git a/app/Services/Cloning/CloningYamlValidator.php b/app/Services/Cloning/CloningYamlValidator.php index cc4b294..564e3a7 100644 --- a/app/Services/Cloning/CloningYamlValidator.php +++ b/app/Services/Cloning/CloningYamlValidator.php @@ -144,6 +144,13 @@ private function validateTables(array $tables): array foreach ($tables as $tableName => $tableConfig) { $prefix = sprintf("Table '%s'", $tableName); + // A table key may be a slash-delimited regex; reject it early if it + // does not compile so the user hears about it before any DB work. + $key = (string) $tableName; + if (str_starts_with($key, '/') && @preg_match($key, '') === false) { + $errors[] = sprintf('%s: invalid regex pattern', $prefix); + } + if (! is_array($tableConfig)) { $errors[] = sprintf('%s: must be an object', $prefix); diff --git a/app/Services/Cloning/TableConfigResolver.php b/app/Services/Cloning/TableConfigResolver.php new file mode 100644 index 0000000..29c404d --- /dev/null +++ b/app/Services/Cloning/TableConfigResolver.php @@ -0,0 +1,141 @@ + $t->name, + $schema->tables, + ); + + /** @var array $resolved */ + $resolved = []; + + // 1. Expand every real table against the config entries (last match wins). + foreach ($realNames as $realName) { + $winner = null; + + foreach ($config->tables as $entry) { + if ($this->matches($entry->tableName, $realName)) { + $winner = $entry; + } + } + + if ($winner instanceof TableCloningConfigData) { + $resolved[$realName] = new TableCloningConfigData( + tableName: $realName, + rows: $winner->rows, + columns: $winner->columns, + ); + } + } + + // 2. Preserve literal entries that match no source table, so a mistyped + // or dropped table still reports NotFound as before. (Regex keys that + // match nothing were never a concrete table — drop them.) + foreach ($config->tables as $entry) { + if ($this->isRegex($entry->tableName)) { + continue; + } + + $matchedReal = array_any( + $realNames, + fn (string $realName): bool => $this->matches($entry->tableName, $realName), + ); + + if ($matchedReal) { + continue; + } + + if (isset($resolved[$entry->tableName])) { + continue; + } + + $resolved[$entry->tableName] = $entry; + } + + return new CloningConfigData( + version: $config->version, + connectionName: $config->connectionName, + options: $config->options, + tables: array_values($resolved), + keyRemapping: $config->keyRemapping, + skipTables: $this->resolveSkipTables($config, $realNames, array_values($resolved)), + ); + } + + /** + * @param list $realNames + * @param list $resolvedTables + * @return list + */ + private function resolveSkipTables(CloningConfigData $config, array $realNames, array $resolvedTables): array + { + /** @var list $skip */ + $skip = []; + + // Carry over the configured skip list, expanding any regex entries. + foreach ($config->skipTables as $entry) { + if ($this->isRegex($entry)) { + foreach ($realNames as $realName) { + if ($this->matches($entry, $realName) && ! in_array($realName, $skip, true)) { + $skip[] = $realName; + } + } + + continue; + } + + if (! in_array($entry, $skip, true)) { + $skip[] = $entry; + } + } + + // Any resolved table whose winning rule is `rows.strategy: skip`. + foreach ($resolvedTables as $table) { + if ($table->rows->strategy === 'skip' && ! in_array($table->tableName, $skip, true)) { + $skip[] = $table->tableName; + } + } + + return $skip; + } + + /** + * Match a config key against a concrete table name: slash-delimited regex, + * otherwise a case-insensitive literal comparison. + */ + private function matches(string $key, string $tableName): bool + { + if ($this->isRegex($key)) { + return preg_match($key, $tableName) === 1; + } + + return strcasecmp($key, $tableName) === 0; + } + + private function isRegex(string $key): bool + { + return str_starts_with($key, '/'); + } +} diff --git a/docs/cloning-yaml.md b/docs/cloning-yaml.md index 7de42db..fb4f01a 100644 --- a/docs/cloning-yaml.md +++ b/docs/cloning-yaml.md @@ -78,7 +78,8 @@ The three schema-control options together determine how closely the target schem ## Tables -Each key under `tables` is the exact table name in the source database. +Each key under `tables` is either the exact table name in the source database +(a **literal**) or a **regex** that matches one or more table names. ```yaml tables: @@ -99,6 +100,45 @@ tables: sort_by: created_at ``` +### Matching table names with regex + +Wrap a key in `/…/` to treat it as a [PCRE regex](https://www.php.net/manual/en/reference.pcre.pattern.syntax.php) +(trailing flags allowed, e.g. `/…/i`). At run time it is expanded against the +actual tables in the source database, so one entry can configure a whole family +of tables: + +```yaml +tables: + # Every monthly archive table gets the same rule + "/^application_logs_archive_\d{2}_\d{4}$/": + rows: + strategy: last + limit: 1 + clear: delete +``` + +- **Keys without a leading `/` are always literal.** There is no glob/`*` + wildcard — use a regex instead (`.*` covers what `*` would). +- **Matching is case-insensitive** (both literal and regex keys). +- **The last matching entry in the file wins.** List a broad regex first, then + override individual tables with a more specific regex or literal below it: + + ```yaml + tables: + "/^app_logs_.*/": # default: keep only the newest row + rows: {strategy: last, limit: 1} + app_logs_critical: # …but copy this one in full + rows: {strategy: full} + ``` + +- A literal key that names a table **not** present in the source is still + reported as *not found* during the run. A regex that matches no table is + simply skipped. + +> **Note.** Key remapping (inline `strategy: remapping` columns and the legacy +> `key_remapping:` section) and the `--skip-tables` / `--only-tables` flags +> operate on **literal** table names only; regex keys are not expanded for them. + ### `rows` Controls which rows are transferred and whether the target table is cleared first. diff --git a/specs/PRD-cloning-yaml-schema.md b/specs/PRD-cloning-yaml-schema.md index 6dd65f6..e640678 100644 --- a/specs/PRD-cloning-yaml-schema.md +++ b/specs/PRD-cloning-yaml-schema.md @@ -70,8 +70,28 @@ options: ### 4.3 `tables.` +A `tables` key is either a **literal** table name or a **regex** (a key wrapped +in `/…/`, trailing flags allowed). Regex keys are expanded against the actual +source tables at run time, so a single entry can configure a whole family of +tables: + +- Keys without a leading `/` are always literal — there is no glob/`*` wildcard. +- Matching is case-insensitive for both literal and regex keys. +- When several entries match the same table, the **last** entry in file order + wins (list a broad regex first, then override specific tables below it). +- A literal key naming an absent source table is still reported *not found*; a + regex that matches nothing is skipped. +- Regex keys are **not** expanded for key remapping or the + `--skip-tables` / `--only-tables` flags — those remain literal-only. + ```yaml tables: + "/^application_logs_archive_\d{2}_\d{4}$/": # regex: all monthly archives + rows: + strategy: last + limit: 1 + clear: delete + users: rows: strategy: full # required; full | first | last — no default From 6a431ce397f66da3e739dfebcb19e611fb8a2ff3 Mon Sep 17 00:00:00 2001 From: Hagen Pommer Date: Wed, 12 Aug 2026 12:27:19 +0200 Subject: [PATCH 13/15] test(cloning): expect to validate config with regex table names --- .../Cloning/CloningYamlValidatorTest.php | 22 +++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/tests/Unit/Services/Cloning/CloningYamlValidatorTest.php b/tests/Unit/Services/Cloning/CloningYamlValidatorTest.php index d6f874e..2dd9447 100644 --- a/tests/Unit/Services/Cloning/CloningYamlValidatorTest.php +++ b/tests/Unit/Services/Cloning/CloningYamlValidatorTest.php @@ -832,3 +832,25 @@ function makeValidConfig(): array ->toContain('key_remapping.tables[0].foreign_keys[0]: must be an object') ->toContain("key_remapping.tables[0].foreign_keys[1]: 'column' is required"); }); + +it('accepts a valid regex table key', function (): void { + $config = makeValidConfig(); + $config['tables'] = [ + '/^application_logs_archive_\d{2}_\d{4}$/' => ['rows' => ['strategy' => 'last', 'limit' => 1]], + ]; + + $errors = (new CloningYamlValidator)->validate($config); + + expect($errors)->toBe([]); +}); + +it('flags an invalid regex table key', function (): void { + $config = makeValidConfig(); + $config['tables'] = [ + '/^app_logs_(unterminated/' => ['rows' => ['strategy' => 'full']], + ]; + + $errors = (new CloningYamlValidator)->validate($config); + + expect($errors)->toContain("Table '/^app_logs_(unterminated/': invalid regex pattern"); +}); From 27c9c2481390461a9a3e9cd78f3e95760a90d66d Mon Sep 17 00:00:00 2001 From: Hagen Pommer Date: Wed, 12 Aug 2026 12:27:23 +0200 Subject: [PATCH 14/15] test(cloning): expect to match regex table names --- .../Cloning/TableConfigResolverTest.php | 198 ++++++++++++++++++ 1 file changed, 198 insertions(+) create mode 100644 tests/Unit/Services/Cloning/TableConfigResolverTest.php diff --git a/tests/Unit/Services/Cloning/TableConfigResolverTest.php b/tests/Unit/Services/Cloning/TableConfigResolverTest.php new file mode 100644 index 0000000..0ab12d2 --- /dev/null +++ b/tests/Unit/Services/Cloning/TableConfigResolverTest.php @@ -0,0 +1,198 @@ + $names + */ +function schemaWith(array $names): DatabaseSchemaData +{ + return new DatabaseSchemaData( + databaseName: 'test', + tables: array_map( + static fn (string $name): TableSchemaData => new TableSchemaData($name, [], []), + $names, + ), + ); +} + +/** + * @param list $tables + * @param list $skip + */ +function configWith(array $tables, array $skip = []): CloningConfigData +{ + return new CloningConfigData( + version: '1', + connectionName: 'source', + options: new CloningOptionsData(1000, false, false, false, true, 'en_US'), + tables: $tables, + keyRemapping: null, + skipTables: $skip, + ); +} + +function tableEntry(string $name, string $strategy = 'full', ?int $limit = null): TableCloningConfigData +{ + return new TableCloningConfigData( + tableName: $name, + rows: new TableRowConfigData(strategy: $strategy, limit: $limit, sortBy: null, clear: ClearMode::None), + columns: [], + ); +} + +it('expands a regex key to every matching source table', function (): void { + $config = configWith([ + tableEntry('/^application_logs_archive_\d{2}_\d{4}$/', 'last', 1), + ]); + + $schema = schemaWith([ + 'application_logs_archive_02_2026', + 'application_logs_archive_03_2026', + 'application_logs_archive_04_2026', + 'users', + ]); + + $resolved = (new TableConfigResolver)->resolve($config, $schema); + + $names = array_map(fn (TableCloningConfigData $t): string => $t->tableName, $resolved->tables); + + expect($names)->toBe([ + 'application_logs_archive_02_2026', + 'application_logs_archive_03_2026', + 'application_logs_archive_04_2026', + ]); + + foreach ($resolved->tables as $table) { + expect($table->rows->strategy)->toBe('last'); + expect($table->rows->limit)->toBe(1); + } +}); + +it('lets the last matching entry win over an earlier regex', function (): void { + $config = configWith([ + tableEntry('/^app_logs_.*/', 'last', 1), + tableEntry('app_logs_critical', 'full'), + ]); + + $schema = schemaWith(['app_logs_2026', 'app_logs_critical']); + + $resolved = (new TableConfigResolver)->resolve($config, $schema); + + $byName = []; + foreach ($resolved->tables as $table) { + $byName[$table->tableName] = $table; + } + + expect($byName['app_logs_2026']->rows->strategy)->toBe('last'); + expect($byName['app_logs_critical']->rows->strategy)->toBe('full'); +}); + +it('preserves a literal entry absent from the source for NotFound reporting', function (): void { + $config = configWith([ + tableEntry('users'), + tableEntry('legacy_orders'), + ]); + + $schema = schemaWith(['users']); + + $resolved = (new TableConfigResolver)->resolve($config, $schema); + + $names = array_map(fn (TableCloningConfigData $t): string => $t->tableName, $resolved->tables); + + expect($names)->toContain('users'); + expect($names)->toContain('legacy_orders'); +}); + +it('drops a regex key that matches nothing', function (): void { + $config = configWith([ + tableEntry('/^no_such_.*/', 'last', 1), + tableEntry('users'), + ]); + + $schema = schemaWith(['users']); + + $resolved = (new TableConfigResolver)->resolve($config, $schema); + + $names = array_map(fn (TableCloningConfigData $t): string => $t->tableName, $resolved->tables); + + expect($names)->toBe(['users']); +}); + +it('ignores source tables that match no config entry', function (): void { + $config = configWith([tableEntry('users')]); + + $schema = schemaWith(['users', 'sessions', 'cache']); + + $resolved = (new TableConfigResolver)->resolve($config, $schema); + + $names = array_map(fn (TableCloningConfigData $t): string => $t->tableName, $resolved->tables); + + expect($names)->toBe(['users']); +}); + +it('expands a regex skip:strategy rule into concrete skipTables', function (): void { + $config = configWith([ + tableEntry('/^tmp_.*/', 'skip'), + tableEntry('users'), + ]); + + $schema = schemaWith(['tmp_a', 'tmp_b', 'users']); + + $resolved = (new TableConfigResolver)->resolve($config, $schema); + + expect($resolved->skipTables)->toContain('tmp_a'); + expect($resolved->skipTables)->toContain('tmp_b'); +}); + +it('expands a regex entry in the top-level skip list', function (): void { + $config = configWith([tableEntry('users')], ['/^tmp_.*/']); + + $schema = schemaWith(['tmp_a', 'tmp_b', 'users']); + + $resolved = (new TableConfigResolver)->resolve($config, $schema); + + expect($resolved->skipTables)->toContain('tmp_a'); + expect($resolved->skipTables)->toContain('tmp_b'); +}); + +it('matches literal keys case-insensitively', function (): void { + $config = configWith([tableEntry('Users')]); + + $schema = schemaWith(['users']); + + $resolved = (new TableConfigResolver)->resolve($config, $schema); + + expect($resolved->tables)->toHaveCount(1); + expect($resolved->tables[0]->tableName)->toBe('users'); +}); + +it('leaves an all-literal config behaviourally unchanged', function (): void { + $config = configWith([ + tableEntry('users', 'full'), + tableEntry('audit_logs', 'first', 100), + ]); + + $schema = schemaWith(['users', 'audit_logs']); + + $resolved = (new TableConfigResolver)->resolve($config, $schema); + + $byName = []; + foreach ($resolved->tables as $table) { + $byName[$table->tableName] = $table; + } + + expect($byName)->toHaveKeys(['users', 'audit_logs']); + expect($byName['users']->rows->strategy)->toBe('full'); + expect($byName['audit_logs']->rows->strategy)->toBe('first'); + expect($byName['audit_logs']->rows->limit)->toBe(100); +}); From cede05efb21a4b3712ebae4cada1adebbfb6c3bf Mon Sep 17 00:00:00 2001 From: Hagen Pommer Date: Mon, 10 Aug 2026 12:55:39 +0200 Subject: [PATCH 15/15] docs(cloning): add row strategy `skip` to docs --- docs/cloning-yaml.md | 7 ++++--- docs/commands/cloning-dump.md | 1 + 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/docs/cloning-yaml.md b/docs/cloning-yaml.md index fb4f01a..4fb621b 100644 --- a/docs/cloning-yaml.md +++ b/docs/cloning-yaml.md @@ -153,10 +153,11 @@ Controls which rows are transferred and whether the target table is cleared firs #### Row strategies | Strategy | Behaviour | -|----------|-----------| -| `full` | Copy all rows from the source table. | +|---------|-----------| +| `full` | Copy all rows from the source table. | | `first` | Copy the first `limit` rows (ordered by `sort_by` ascending). | -| `last` | Copy the last `limit` rows (ordered by `sort_by` descending). | +| `last` | Copy the last `limit` rows (ordered by `sort_by` descending). | +| `skip` | Skip this table entirely. | #### `clear` values diff --git a/docs/commands/cloning-dump.md b/docs/commands/cloning-dump.md index 0027c83..dc04338 100644 --- a/docs/commands/cloning-dump.md +++ b/docs/commands/cloning-dump.md @@ -202,6 +202,7 @@ tables: | `full` | Copy all rows | | `first` | Copy the first N rows (requires `limit`) | | `last` | Copy the last N rows (requires `limit`) | +| `skip` | Skip this table entirely | ## PII detection