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/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'
+
diff --git a/app/Commands/Cloning/RunCommand.php b/app/Commands/Cloning/RunCommand.php
index 89a6bda..3e034d8 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 ($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 ($indicator !== null) {
- $this->output->write($indicator);
- $dotColumn++;
+ 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);
@@ -926,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;
@@ -973,10 +1075,156 @@ 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().'…' : '';
+ }
+
+ /**
+ * 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
+ {
+ $eta = 'ETA '.$this->formatDuration($stats->estimatedSecondsRemaining);
+ $overall = trim($this->formatThroughput($stats->loopAggregate->latestPacePerMillion));
+
+ if (! $verbose) {
+ return $overall === '—' ? $eta : $eta.' · '.$overall;
+ }
+
+ return sprintf(
+ '%s · all %s · sel %s · tr %s · ins %s',
+ $eta,
+ $overall,
+ trim($this->formatThroughput($stats->selectAggregate->latestPacePerMillion)),
+ trim($this->formatThroughput($stats->transformAggregate->latestPacePerMillion)),
+ trim($this->formatThroughput($stats->insertAggregate->latestPacePerMillion)),
+ );
+ }
+
+ 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->pacePerMillion),
+ ];
+ }
+
+ $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 +1233,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));
}
}
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 row across all samples,
+ * or null when no rows have been processed.
+ */
+ public ?float $pace {
+ get => $this->rowsProcessed > 0
+ ? ($this->sum / $this->rowsProcessed)
+ : null;
+ }
+
+ /**
+ * 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
+ : 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;
+ }
+
+ 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..d1a1e5d
--- /dev/null
+++ b/app/Data/Cloning/StatsTableTransferData.php
@@ -0,0 +1,173 @@
+ */
+ 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;
+ }
+
+ /**
+ * 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;
+
+ 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,
+ selectPacePerMillion: $this->selectAggregate->pacePerMillion,
+ transformPacePerMillion: $this->transformAggregate->pacePerMillion,
+ insertPacePerMillion: $this->insertAggregate->pacePerMillion,
+ loopPacePerMillion: $this->loopAggregate->pacePerMillion,
+ ));
+ }
+
+ 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';
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