From 3b9e4837f9ce83124a0daa440be487880e7e38f1 Mon Sep 17 00:00:00 2001 From: Norbert Orzechowicz Date: Sat, 31 Oct 2026 15:37:21 -0500 Subject: [PATCH] feat(flow-php/etl): the run report shows the peak memory of every process - Statistics::$memory is now $heap; the new $processes holds the OS peak of the parent and of every worker - SourceCounts is now RunCounts; launchers report worker peaks into it, with worker() and absorbWorkers() - the Report reads the memory and row count that TelemetryContext keeps; it no longer measures them again - Report::normalize() gives the JSON form for CI; INF and NaN are encoded, never dropped - column statistics: min and max skip NaN, as in Parquet and arrow-rs - the CLI prints heap, process, worker and summed peaks; analyze() warns when it stops parallel writes --- .../bridges/symfony-http-foundation-bridge.md | 6 +- .../components/core/batch-processing.md | 3 +- .../components/core/corrupted-records.md | 3 + documentation/upgrading.md | 13 ++- .../CLI/Formatter/PipelineReportFormatter.php | 15 ++- .../Formatter/PipelineReportFormatterTest.php | 52 +++++++++ src/core/etl/src/Flow/ETL/API/DataFrame.php | 2 +- .../Flow/ETL/Runtime/Executor/Executor.php | 21 ++-- .../{SourceCounts.php => RunCounts.php} | 36 ++++++- .../Runtime/Executor/StatisticsCollector.php | 43 ++++---- .../Flow/ETL/Runtime/Executor/StepRunner.php | 6 +- .../Flow/ETL/Runtime/Launcher/Launcher.php | 6 +- .../Flow/ETL/Runtime/Metrics/Processes.php | 62 +++++++++++ .../src/Flow/ETL/Runtime/Metrics/Report.php | 18 ++++ .../ETL/Runtime/Metrics/SourceStatistics.php | 20 ++++ .../Flow/ETL/Runtime/Metrics/Statistics.php | 17 ++- .../ETL/Runtime/Metrics/Statistics/Column.php | 31 +++++- .../Runtime/Metrics/Statistics/Columns.php | 9 ++ .../Metrics/Statistics/ExecutionTime.php | 13 +++ .../Runtime/Telemetry/TelemetryContext.php | 21 +++- .../etl/src/Flow/ETL/Values/Cardinality.php | 16 +++ .../Flow/ETL/Values/Memory/Consumption.php | 8 ++ .../etl/src/Flow/ETL/Values/Memory/Unit.php | 13 +++ .../ETL/Tests/Double/RecordingLauncher.php | 10 +- .../Flow/ETL/Tests/Double/StubLauncher.php | 15 ++- .../Integration/DataFrame/AnalyzeTest.php | 51 ++++++++- .../Flow/ETL/Tests/Mother/ReportMother.php | 49 +++++++++ .../Unit/Planning/Physical/PlannerTest.php | 4 +- .../Unit/Runtime/Executor/ExecutorTest.php | 6 +- ...SourceCountsTest.php => RunCountsTest.php} | 90 ++++++++++++---- .../Executor/StatisticsCollectorTest.php | 100 ++++++++++++++++-- .../Unit/Runtime/Metrics/ProcessesTest.php | 50 +++++++++ .../Tests/Unit/Runtime/Metrics/ReportTest.php | 85 +++++++++++++++ .../Runtime/Metrics/SourceStatisticsTest.php | 36 +++++++ .../Runtime/Metrics/Statistics/ColumnTest.php | 55 ++++++++++ .../Metrics/Statistics/ColumnsTest.php | 10 ++ .../Metrics/Statistics/ExecutionTimeTest.php | 29 +++++ .../Telemetry/TelemetryContextTest.php | 32 ++++++ .../ETL/Tests/Unit/Values/CardinalityTest.php | 21 ++++ .../Unit/Values/Memory/ConsumptionTest.php | 14 +++ .../ETL/Tests/Unit/Values/Memory/UnitTest.php | 11 ++ 41 files changed, 1011 insertions(+), 91 deletions(-) create mode 100644 src/cli/tests/Flow/CLI/Tests/Unit/Formatter/PipelineReportFormatterTest.php rename src/core/etl/src/Flow/ETL/Runtime/Executor/{SourceCounts.php => RunCounts.php} (76%) create mode 100644 src/core/etl/src/Flow/ETL/Runtime/Metrics/Processes.php create mode 100644 src/core/etl/tests/Flow/ETL/Tests/Mother/ReportMother.php rename src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Executor/{SourceCountsTest.php => RunCountsTest.php} (77%) create mode 100644 src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/ProcessesTest.php create mode 100644 src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/ReportTest.php create mode 100644 src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/Statistics/ExecutionTimeTest.php diff --git a/documentation/components/bridges/symfony-http-foundation-bridge.md b/documentation/components/bridges/symfony-http-foundation-bridge.md index c48aa63e6..f5ac23120 100644 --- a/documentation/components/bridges/symfony-http-foundation-bridge.md +++ b/documentation/components/bridges/symfony-http-foundation-bridge.md @@ -184,7 +184,8 @@ final class ReportsController extends AbstractController $this->logger->info('Stream completed', [ 'total_rows' => $report->statistics()->totalRows(), 'execution_time_seconds' => $report->statistics()->executionTime->inSeconds(), - 'memory_peak_mb' => $report->statistics()->memory->max()->inMb(), + 'heap_peak_mb' => $report->statistics()->heap->max()->inMb(), + 'process_peak_mb' => $report->statistics()->processes->parent->inMb(), ]); })) ->as('orders.json') @@ -218,6 +219,7 @@ The `Report` object provides access to: - `$report->statistics()->executionTime->inSeconds()` - Execution duration - `$report->statistics()->executionTime->startedAt` - Start timestamp - `$report->statistics()->executionTime->finishedAt` - End timestamp -- `$report->statistics()->memory->max()` - Peak memory usage +- `$report->statistics()->heap->max()` - Peak PHP heap plus backend memory, sampled once per batch +- `$report->statistics()->processes` - OS peak of the process and of every worker: `workerPeak()`, `peakSum()` - `$report->schema()` - Dataset schema (when `withSchema()` is enabled) - `$report->statistics()->columns` - Column statistics (when `withColumnStatistics()` is enabled) diff --git a/documentation/components/core/batch-processing.md b/documentation/components/core/batch-processing.md index 913ac57f4..95061fd15 100644 --- a/documentation/components/core/batch-processing.md +++ b/documentation/components/core/batch-processing.md @@ -115,5 +115,6 @@ $report = data_frame() ->write($loader) ->run(analyze: analyze()); -echo "Peak memory usage: " . $report->statistics()->memory->max()->inMb() . " MB\n"; +echo "Peak heap: " . $report->statistics()->heap->max()->inMb() . " MB\n"; +echo "Process peak: " . $report->statistics()->processes->parent->inMb() . " MB\n"; ``` \ No newline at end of file diff --git a/documentation/components/core/corrupted-records.md b/documentation/components/core/corrupted-records.md index 912125fed..15e291ec6 100644 --- a/documentation/components/core/corrupted-records.md +++ b/documentation/components/core/corrupted-records.md @@ -24,6 +24,9 @@ data_frame() ->write(to_branch(not($valid), to_json('rejects/invalid.json'))) ->run(); +// count: fail CI above a limit +$rejected = data_frame()->read(from_json_lines('rejects/broken.jsonl'))->count(); + // report data_frame() ->read(from_json_lines('rejects/broken.jsonl')) diff --git a/documentation/upgrading.md b/documentation/upgrading.md index 58040a836..5a1780e83 100644 --- a/documentation/upgrading.md +++ b/documentation/upgrading.md @@ -1535,8 +1535,8 @@ final class MyTransformer implements Transformer | Before | After | |---------------------------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `Planner::plan(LogicalPlan $logical, FlowContext $context, ?SourceRows $sources)` | `plan(LogicalPlan $logical)` (row 103); the counters go to `Executor::execute()` | -| `Executor::execute(PhysicalPlan $plan)`, `fetch(PhysicalPlan $plan)` | `execute(LogicalPlan $logical, FlowContext $context, ?SourceCounts $sources = null)`, `fetch(LogicalPlan $logical, FlowContext $context)`: the executor plans, and `execute()` returns the `PhysicalPlan` (`getReturn()`) | -| `Executor::executePlan(PhysicalPlan $plan)`, `executePipeline(Pipeline $pipeline)` | `executePlan(PhysicalPlan $plan, FlowContext $context, ?SourceCounts $sources = null)`, `executePipeline(Pipeline $pipeline, FlowContext $context, ?Extractor $source = null)` | +| `Executor::execute(PhysicalPlan $plan)`, `fetch(PhysicalPlan $plan)` | `execute(LogicalPlan $logical, FlowContext $context, ?RunCounts $sources = null)`, `fetch(LogicalPlan $logical, FlowContext $context)`: the executor plans, and `execute()` returns the `PhysicalPlan` (`getReturn()`) | +| `Executor::executePlan(PhysicalPlan $plan)`, `executePipeline(Pipeline $pipeline)` | `executePlan(PhysicalPlan $plan, FlowContext $context, ?RunCounts $sources = null)`, `executePipeline(Pipeline $pipeline, FlowContext $context, ?Extractor $source = null)` | | `Executor::merge(Generator $batches, PhysicalPlan $plan)` | `merge(Generator $batches, Backend $backend)`: an empty result takes its schema from the `PhysicalPlan` `$batches` returns | | `Flow\ETL\Planner\SinkFeedFactory` (built the sink pipelines at plan time) | `Flow\ETL\Runtime\Executor\WriteLoaderFactory`: `feed(BranchPipeline, ErrorHandler, FlowContext): BranchLoader`, `transactional(TransactionGroup, FlowContext): TransactionLoader`; the plan holds `BranchPipeline`/`TransactionGroup` and the runtime builds the loaders each run | | `new SinkRun(Pipeline $side, Executor $executor)` | `new BranchFiber(Pipeline $side, Executor $executor, FlowContext $context, FeedExtractor $feed)` | @@ -1682,7 +1682,7 @@ public function apply(LogicalPlan $plan, Backend $backend): LogicalPlan | `Flow\ETL\Executor` | `Flow\ETL\Runtime\Executor\Executor` | | `Flow\ETL\Executor\{AdoptedBatches, SinkFeed, StatisticsCollector, TransactionalSinks}` | `Flow\ETL\Runtime\Executor\{AdoptedBatches, BranchLoader, StatisticsCollector, TransactionLoader}` | | `Flow\ETL\Executor\TransactionRollback` | removed: `TransactionLoader` rolls back itself and logs a failing rollback | -| `Flow\ETL\Executor\SourceRows` | `Flow\ETL\Runtime\Executor\SourceCounts` | +| `Flow\ETL\Executor\SourceRows` | `Flow\ETL\Runtime\Executor\RunCounts` | | `new Flow\ETL\Exception\TransactionRolledBack(Loader $loader, Throwable $cause)` | removed: the rolled-back error is offered once to the plan's error handler | ### 106) `flow-php/etl` - the serialize transformers lose their mode flags @@ -1914,6 +1914,13 @@ by Spark's hidden-path rule. | `CSVOpenSource::batches($schema, $batchSize, $backend)` | `+ ?int $sampleSize = null`; `JsonOpenSource::batches()` also `+ ?Schema $known = null`; both yield `CorruptedRecordException\|Rows` | | `to_date_time(null)` threw `InvalidArgumentException` | returns null; a null format or time zone still throws | +### 128) `flow-php/etl` - the Report's memory is named for what it measures, and every process has a peak + +| Before | After | +|---------------------------------|------------------------------------------------------------------------------------------------------------------| +| `$report->statistics()->memory` | `$report->statistics()->heap`: the parent's PHP heap plus the backend's bytes, sampled once per batch, as before | +| - | `$report->statistics()->processes`: the OS peak of the parent and of every worker, `workerPeak()`, `peakSum()` | + ## Upgrading from 0.45.x to 0.46.x ### 1) `flow-php/etl-adapter-postgresql`, `flow-php/etl-adapter-doctrine` - keyset reads reject NULL keys up front diff --git a/src/cli/src/Flow/CLI/Formatter/PipelineReportFormatter.php b/src/cli/src/Flow/CLI/Formatter/PipelineReportFormatter.php index bea4f5b5e..943137673 100644 --- a/src/cli/src/Flow/CLI/Formatter/PipelineReportFormatter.php +++ b/src/cli/src/Flow/CLI/Formatter/PipelineReportFormatter.php @@ -120,12 +120,23 @@ public function format(): void ->render(); } + $statistics = $this->report->statistics(); + $workerPeak = $statistics->processes->workerPeak(); + $this->style->newLine(); $this->style->definitionList( 'Statistics', new TableSeparator(), - ['Analyzed Rows' => number_format($this->report->statistics()->totalRows())], - ['Execution Time' => $this->report->statistics()->executionTime->highResolutionTime->toString()], + ['Analyzed Rows' => number_format($statistics->totalRows())], + ['Execution Time' => $statistics->executionTime->highResolutionTime->toString()], + ['Heap Peak' => $statistics->heap->max()->inMb() . ' MB'], + ['Process Peak' => $statistics->processes->parent->inMb() . ' MB'], + ...$workerPeak === null + ? [] + : [ + ['Worker Peak' => $workerPeak->inMb() . ' MB'], + ['Peak Sum' => $statistics->processes->peakSum()->inMb() . ' MB'], + ], ); } } diff --git a/src/cli/tests/Flow/CLI/Tests/Unit/Formatter/PipelineReportFormatterTest.php b/src/cli/tests/Flow/CLI/Tests/Unit/Formatter/PipelineReportFormatterTest.php new file mode 100644 index 000000000..d4cd15146 --- /dev/null +++ b/src/cli/tests/Flow/CLI/Tests/Unit/Formatter/PipelineReportFormatterTest.php @@ -0,0 +1,52 @@ + 'pipeline.php'], (new PipelineRunCommand('run'))->getDefinition()); + $output = new BufferedOutput(); + + (new PipelineReportFormatter(ReportMother::of(new Processes(Unit::fromMb(25), [ + Unit::fromMb(40), + Unit::fromMb(41), + ])), new SymfonyStyle($input, $output), $input))->format(); + + $display = $output->fetch(); + static::assertMatchesRegularExpression('/Heap Peak\s+[\d.]+ MB/', $display); + static::assertMatchesRegularExpression('/Process Peak\s+25 MB/', $display); + static::assertMatchesRegularExpression('/Worker Peak\s+41 MB/', $display); + static::assertMatchesRegularExpression('/Peak Sum\s+106 MB/', $display); + } + + public function test_a_serial_run_prints_no_worker_rows(): void + { + $input = new ArrayInput(['pipeline-file' => 'pipeline.php'], (new PipelineRunCommand('run'))->getDefinition()); + $output = new BufferedOutput(); + + (new PipelineReportFormatter( + ReportMother::of(new Processes(Unit::fromMb(25), [])), + new SymfonyStyle($input, $output), + $input, + ))->format(); + + $display = $output->fetch(); + static::assertMatchesRegularExpression('/Process Peak\s+25 MB/', $display); + static::assertStringNotContainsString('Worker Peak', $display); + static::assertStringNotContainsString('Peak Sum', $display); + } +} diff --git a/src/core/etl/src/Flow/ETL/API/DataFrame.php b/src/core/etl/src/Flow/ETL/API/DataFrame.php index f1b7343c9..556d54259 100644 --- a/src/core/etl/src/Flow/ETL/API/DataFrame.php +++ b/src/core/etl/src/Flow/ETL/API/DataFrame.php @@ -665,7 +665,7 @@ public function run(bool|Analyze $analyze = false): ?Report ->execute( ($collector->readsColumns() ? Trigger::rows : Trigger::run)->plan($this->root, $this->writes), $this->context, - $collector->sources(), + $collector->counts(), ) as $rows) { $collector->capture($rows); } diff --git a/src/core/etl/src/Flow/ETL/Runtime/Executor/Executor.php b/src/core/etl/src/Flow/ETL/Runtime/Executor/Executor.php index ab00912f6..35430f704 100644 --- a/src/core/etl/src/Flow/ETL/Runtime/Executor/Executor.php +++ b/src/core/etl/src/Flow/ETL/Runtime/Executor/Executor.php @@ -42,13 +42,14 @@ public function __construct() /** * Plans and runs a whole plan inside one DataFrame span of $context: a planning failure closes the span as failed. * - * @param null|SourceCounts $sources counts the rows every source yields, a joined frame's included + * @param null|RunCounts $counts counts the rows every source yields, a joined frame's included, and the peak of + * every launched worker * * @throws InvalidLogicException when $context is already being advanced * * @return Generator returns the physical plan it ran */ - public function execute(LogicalPlan $logical, FlowContext $context, ?SourceCounts $sources = null): Generator + public function execute(LogicalPlan $logical, FlowContext $context, ?RunCounts $counts = null): Generator { if ($this->advancing->offsetExists($context)) { throw InvalidLogicException::cyclicPlanOnRun(); @@ -61,7 +62,7 @@ public function execute(LogicalPlan $logical, FlowContext $context, ?SourceCount try { $plan = $context->config->planner()->plan($logical); - foreach ($this->executePlan($plan, $context, $sources) as $rows) { + foreach ($this->executePlan($plan, $context, $counts) as $rows) { // disarmed across our own yield: while parked we are not advancing, so a second read // arriving here is another reader of the same plan, not recursion $this->advancing->offsetUnset($context); @@ -133,13 +134,13 @@ public function merge(Generator $batches, Backend $backend): Rows * Drives a plan with no DataFrame span - $context belongs to whoever runs it. The config's launcher may launch the * head of the plan's chain; without one, or when the launch does not pay, the plan runs here. * - * @param null|SourceCounts $sources counts the rows the plan's source yields + * @param null|RunCounts $counts counts the rows the plan's source yields and the peak of every launched worker * * @return Generator */ - public function executePlan(PhysicalPlan $plan, FlowContext $context, ?SourceCounts $sources = null): Generator + public function executePlan(PhysicalPlan $plan, FlowContext $context, ?RunCounts $counts = null): Generator { - return $this->run($plan->root(), $context, $plan, $sources, null, new StepRunner()); + return $this->run($plan->root(), $context, $plan, $counts, null, new StepRunner()); } /** @@ -170,7 +171,7 @@ private function run( Pipeline $pipeline, FlowContext $context, ?PhysicalPlan $plan, - ?SourceCounts $sources, + ?RunCounts $counts, ?Extractor $fed, StepRunner $runner, ): Generator { @@ -183,7 +184,7 @@ private function run( default => $fed ?? throw InvalidLogicException::because('A pipeline that reads no source must be fed'), }; $criteria = $read instanceof Read ? $read->criteria() : new Criteria(); - $counted = $read instanceof Read ? $sources : null; + $counted = $read instanceof Read ? $counts : null; $launch = $plan !== null && $source instanceof SplittableExtractor ? $context->config->launcher()?->read($plan, $source, $criteria, $context, $counted) @@ -207,8 +208,8 @@ private function run( $generator = $position === $last && $launch !== null - ? $runner->steps($launch->rest, $generator, $context, sources: $sources) - : $runner->steps($stage->steps, $generator, $context, $position === 0 ? $source : null, $sources); + ? $runner->steps($launch->rest, $generator, $context, counts: $counts) + : $runner->steps($stage->steps, $generator, $context, $position === 0 ? $source : null, $counts); } // a foreach, never `yield from`: the re-yield must not forward the consumer's sent signal; every stage routes diff --git a/src/core/etl/src/Flow/ETL/Runtime/Executor/SourceCounts.php b/src/core/etl/src/Flow/ETL/Runtime/Executor/RunCounts.php similarity index 76% rename from src/core/etl/src/Flow/ETL/Runtime/Executor/SourceCounts.php rename to src/core/etl/src/Flow/ETL/Runtime/Executor/RunCounts.php index 2c443d9ec..d810fee51 100644 --- a/src/core/etl/src/Flow/ETL/Runtime/Executor/SourceCounts.php +++ b/src/core/etl/src/Flow/ETL/Runtime/Executor/RunCounts.php @@ -8,13 +8,19 @@ use Flow\ETL\IO\Extractor; use Flow\ETL\IO\Extractor\Signal; use Flow\ETL\IO\Extractor\Statistics; +use Flow\ETL\Runtime\Metrics\Processes; use Flow\ETL\Runtime\Metrics\SourceStatistics; +use Flow\ETL\Values\Memory\Unit; use Flow\ETL\Values\Rows; use Flow\Telemetry\ObjectExtractor; use Generator; use SplObjectStorage; -final class SourceCounts +use function array_map; +use function array_values; +use function max; + +final class RunCounts { /** * @var SplObjectStorage @@ -29,6 +35,11 @@ final class SourceCounts */ private SplObjectStorage $declared; + /** + * @var array the peak bytes of each worker, by pid, in the order worker() first saw them + */ + private array $workers = []; + public function __construct() { /** @var SplObjectStorage $reads */ @@ -87,6 +98,29 @@ public function absorb(self $other): void $this->reads[$extractor]->absorb($other->reads[$extractor]); } + + $this->absorbWorkers($other); + } + + /** + * Takes over the worker peaks $other saw and none of its reads: an abandoned attempt's workers used that memory, + * while its source is read again. + */ + public function absorbWorkers(self $other): void + { + foreach ($other->workers as $pid => $peakBytes) { + $this->worker($pid, $peakBytes); + } + } + + public function worker(int $pid, int $peakBytes): void + { + $this->workers[$pid] = max($this->workers[$pid] ?? 0, $peakBytes); + } + + public function processes(Unit $parent): Processes + { + return new Processes($parent, array_map(Unit::fromBytes(...), array_values($this->workers))); } /** diff --git a/src/core/etl/src/Flow/ETL/Runtime/Executor/StatisticsCollector.php b/src/core/etl/src/Flow/ETL/Runtime/Executor/StatisticsCollector.php index 2ec45d103..8d059988a 100644 --- a/src/core/etl/src/Flow/ETL/Runtime/Executor/StatisticsCollector.php +++ b/src/core/etl/src/Flow/ETL/Runtime/Executor/StatisticsCollector.php @@ -12,11 +12,13 @@ use Flow\ETL\Runtime\Metrics\Statistics; use Flow\ETL\Runtime\Metrics\Statistics\Columns; use Flow\ETL\Runtime\Metrics\Statistics\ExecutionTime; -use Flow\ETL\Values\Memory\Consumption; +use Flow\ETL\Values\Memory\Unit; use Flow\ETL\Values\Rows; use Flow\ETL\Values\Schema; use function gc_collect_cycles; +use function implode; +use function sprintf; /** * @template T of Analyze|bool|null @@ -27,18 +29,14 @@ final class StatisticsCollector private ?Columns $columnStatistics = null; - private readonly Consumption $memory; + private readonly RunCounts $counts; private ?Schema $schema = null; - private ?SourceCounts $sources = null; - private readonly DateTimeImmutable $startedAt; private readonly HighResolutionTime $startTime; - private int $totalRows = 0; - /** * @param T $analyze */ @@ -57,27 +55,34 @@ public function __construct( if ($this->analyze !== null) { $this->columnStatistics = $this->analyze->collectColumnStatistics() ? new Columns() : null; $this->schema = $this->analyze->collectSchema() ? new Schema() : null; - $this->sources = $this->analyze->collectSourceStatistics() ? new SourceCounts() : null; + } + + $this->counts = new RunCounts(); + + if ($this->readsColumns() && $this->context->config->launcher() !== null) { + $this->context + ->telemetry() + ->logger() + ->warn(sprintf('analyze()->%s sends every row to the parent process, so the writes do not run in parallel. ' + . 'Drop it to keep parallel writes.', implode('->', [ + ...($this->schema === null ? [] : ['withSchema()']), + ...($this->columnStatistics === null ? [] : ['withColumnStatistics()']), + ]))); } gc_collect_cycles(); - $this->memory = new Consumption($this->context->backend()); $this->startedAt = $this->context->config->clock()->now(); $this->startTime = HighResolutionTime::now(); } public function capture(Rows $rows): void { - $this->totalRows += $rows->count(); - $this->context->telemetry()->dataFrameBatchProcessed($rows); if ($this->analyze === null) { return; } - $this->memory->capture(); - if ($this->schema !== null) { $this->schema = $this->schema->merge($rows->schema()); } @@ -99,11 +104,12 @@ public function readsColumns(): bool } /** - * Handed to the planner, so the pipelines that read a source count its rows. + * Handed to the executor under analyze, so the pipelines that read a source count its rows and a launcher its + * workers' peaks. */ - public function sources(): ?SourceCounts + public function counts(): ?RunCounts { - return $this->sources; + return $this->analyze === null ? null : $this->counts; } /** @@ -121,12 +127,13 @@ public function report(): ?Report return new Report( $this->schema, new Statistics( - $this->totalRows, + $this->context->telemetry()->totalRows(), new ExecutionTime($this->startedAt, $endedAt, $this->startTime->diff($endTime)), - $this->memory, + $this->context->telemetry()->memory(), $this->columnStatistics, + $this->counts->processes(Unit::processPeak()), ), - $this->sources?->statistics(), + $this->analyze->collectSourceStatistics() ? $this->counts->statistics() : null, ); } } diff --git a/src/core/etl/src/Flow/ETL/Runtime/Executor/StepRunner.php b/src/core/etl/src/Flow/ETL/Runtime/Executor/StepRunner.php index 1b7b14ec2..ca66987c6 100644 --- a/src/core/etl/src/Flow/ETL/Runtime/Executor/StepRunner.php +++ b/src/core/etl/src/Flow/ETL/Runtime/Executor/StepRunner.php @@ -54,7 +54,7 @@ public function __construct( * @param Generator $input * @param null|Extractor|Reader $source the source $input comes from directly, so its failures reach the error * handler and its corrupted records its reject loader - * @param null|SourceCounts $sources counts the rows the sources of a join's right plan yield + * @param null|RunCounts $counts counts the rows the sources of a join's right plan yield * * @return Generator */ @@ -63,7 +63,7 @@ public function steps( Generator $input, FlowContext $context, Extractor|Reader|null $source = null, - ?SourceCounts $sources = null, + ?RunCounts $counts = null, ): Generator { $run = []; @@ -77,7 +77,7 @@ public function steps( $step->join::class, $step->join->join( $upstream, - $context->config->executor()->executePlan($step->right, $context, $sources), + $context->config->executor()->executePlan($step->right, $context, $counts), $context->backend(), ), $context, diff --git a/src/core/etl/src/Flow/ETL/Runtime/Launcher/Launcher.php b/src/core/etl/src/Flow/ETL/Runtime/Launcher/Launcher.php index 3aca1f8c7..9206d18b1 100644 --- a/src/core/etl/src/Flow/ETL/Runtime/Launcher/Launcher.php +++ b/src/core/etl/src/Flow/ETL/Runtime/Launcher/Launcher.php @@ -8,7 +8,7 @@ use Flow\ETL\IO\Extractor\SplittableExtractor; use Flow\ETL\IO\Inference\Sniffs; use Flow\ETL\Planning\Physical\PhysicalPlan; -use Flow\ETL\Runtime\Executor\SourceCounts; +use Flow\ETL\Runtime\Executor\RunCounts; use Flow\ETL\Runtime\FlowContext; interface Launcher extends Sniffs @@ -16,13 +16,13 @@ interface Launcher extends Sniffs /** * Decides before the first batch whether launching the head of $plan's chain pays; null when it does not and the * executor reads $source. Otherwise the launch: its rows are identical to the launched steps run over the source's - * own scan of $criteria, with the source's rows counted into $sources. + * own scan of $criteria, with the source's rows and the workers' peaks counted into $counts. */ public function read( PhysicalPlan $plan, SplittableExtractor $source, Criteria $criteria, FlowContext $context, - ?SourceCounts $sources = null, + ?RunCounts $counts = null, ): ?Launch; } diff --git a/src/core/etl/src/Flow/ETL/Runtime/Metrics/Processes.php b/src/core/etl/src/Flow/ETL/Runtime/Metrics/Processes.php new file mode 100644 index 000000000..efdf6e3bc --- /dev/null +++ b/src/core/etl/src/Flow/ETL/Runtime/Metrics/Processes.php @@ -0,0 +1,62 @@ + $workers the peak of each worker, in the order they first completed a task; none for a serial + * run + */ + public function __construct( + public Unit $parent, + public array $workers, + ) {} + + /** + * @return array{parent: int, workers: list, worker_peak: null|int, peak_sum: int} in bytes + */ + public function normalize(): array + { + return [ + 'parent' => $this->parent->inBytes(), + 'workers' => array_map(static fn(Unit $worker): int => $worker->inBytes(), $this->workers), + 'worker_peak' => $this->workerPeak()?->inBytes(), + 'peak_sum' => $this->peakSum()->inBytes(), + ]; + } + + /** + * Parent plus every worker. An upper bound for sizing, not the memory the run used: the peaks happen at different + * times and each process counts its shared libraries. + */ + public function peakSum(): Unit + { + $bytes = $this->parent->inBytes(); + + foreach ($this->workers as $worker) { + $bytes += $worker->inBytes(); + } + + return Unit::fromBytes($bytes); + } + + public function workerPeak(): ?Unit + { + $peak = null; + + foreach ($this->workers as $worker) { + if ($peak === null || $worker->isGreaterThan($peak)) { + $peak = $worker; + } + } + + return $peak; + } +} diff --git a/src/core/etl/src/Flow/ETL/Runtime/Metrics/Report.php b/src/core/etl/src/Flow/ETL/Runtime/Metrics/Report.php index 944582c63..c2ec4cfd8 100644 --- a/src/core/etl/src/Flow/ETL/Runtime/Metrics/Report.php +++ b/src/core/etl/src/Flow/ETL/Runtime/Metrics/Report.php @@ -6,6 +6,8 @@ use Flow\ETL\Values\Schema; +use function array_map; + final readonly class Report { /** @@ -17,6 +19,22 @@ public function __construct( private ?array $sources, ) {} + /** + * The JSON form for CI: the same values as the typed getters, memory in bytes. + * + * @return array{schema: null|array, statistics: array, sources: null|list>} + */ + public function normalize(): array + { + return [ + 'schema' => $this->schema?->normalize(), + 'statistics' => $this->statistics->normalize(), + 'sources' => $this->sources === null + ? null + : array_map(static fn(SourceStatistics $source): array => $source->normalize(), $this->sources), + ]; + } + /** * @return null|list */ diff --git a/src/core/etl/src/Flow/ETL/Runtime/Metrics/SourceStatistics.php b/src/core/etl/src/Flow/ETL/Runtime/Metrics/SourceStatistics.php index 5d30eecad..568484296 100644 --- a/src/core/etl/src/Flow/ETL/Runtime/Metrics/SourceStatistics.php +++ b/src/core/etl/src/Flow/ETL/Runtime/Metrics/SourceStatistics.php @@ -7,6 +7,7 @@ use Flow\ETL\IO\Extractor\Statistics; use function abs; +use function is_finite; use const INF; @@ -23,6 +24,25 @@ public function __construct( public bool $complete, ) {} + /** + * rows_error: rowsError() when it is a finite number; null without an estimate, for an incomplete read, or when + * the estimate was positive against 0 rows measured (infinitely off). + * + * @return array{extractor: string, declared: array{rows: array, size: array}, rows: int, complete: bool, rows_error: null|float} + */ + public function normalize(): array + { + $rowsError = $this->rowsError(); + + return [ + 'extractor' => $this->extractor, + 'declared' => ['rows' => $this->declared->rows->normalize(), 'size' => $this->declared->size->normalize()], + 'rows' => $this->rows, + 'complete' => $this->complete, + 'rows_error' => $rowsError !== null && is_finite($rowsError) ? $rowsError : null, + ]; + } + /** * How far the declared estimate was off, relative to the rows measured. Null without an estimate: an upper bound * alone is a guarantee, not a guess. Null for an incomplete read too: the declaration describes the whole source, diff --git a/src/core/etl/src/Flow/ETL/Runtime/Metrics/Statistics.php b/src/core/etl/src/Flow/ETL/Runtime/Metrics/Statistics.php index 7179f3ea9..7a664c654 100644 --- a/src/core/etl/src/Flow/ETL/Runtime/Metrics/Statistics.php +++ b/src/core/etl/src/Flow/ETL/Runtime/Metrics/Statistics.php @@ -13,10 +13,25 @@ public function __construct( private int $totalRows, public ExecutionTime $executionTime, - public Consumption $memory, + public Consumption $heap, public ?Columns $columns, + public Processes $processes, ) {} + /** + * @return array + */ + public function normalize(): array + { + return [ + 'total_rows' => $this->totalRows, + 'execution_time' => $this->executionTime->normalize(), + 'heap' => $this->heap->normalize(), + 'processes' => $this->processes->normalize(), + 'columns' => $this->columns?->normalize(), + ]; + } + public function totalRows(): int { return $this->totalRows; diff --git a/src/core/etl/src/Flow/ETL/Runtime/Metrics/Statistics/Column.php b/src/core/etl/src/Flow/ETL/Runtime/Metrics/Statistics/Column.php index 202c956d4..f0c88b8b8 100644 --- a/src/core/etl/src/Flow/ETL/Runtime/Metrics/Statistics/Column.php +++ b/src/core/etl/src/Flow/ETL/Runtime/Metrics/Statistics/Column.php @@ -29,7 +29,9 @@ use function is_bool; use function is_countable; use function is_float; +use function is_infinite; use function is_int; +use function is_nan; use function is_string; use function max; use function mb_strlen; @@ -141,6 +143,32 @@ public function name(): string return $this->reference->name(); } + /** + * A date bound is RFC 3339 text; an infinite float bound, which JSON cannot hold, is "INF" or "-INF". + * + * @return array + */ + public function normalize(): array + { + $bound = static fn(int|float|DateTimeInterface|bool|null $value): int|float|string|bool|null => match (true) { + $value instanceof DateTimeInterface => $value->format(DateTimeInterface::RFC3339_EXTENDED), + is_float($value) && is_infinite($value) => $value > 0 ? 'INF' : '-INF', + default => $value, + }; + + return [ + 'name' => $this->name(), + 'nulls_count' => $this->nullsCount, + 'distinct_count' => $this->distinctCount(), + 'min' => $bound($this->min), + 'max' => $bound($this->max), + 'min_length' => $this->minLength, + 'max_length' => $this->maxLength, + 'min_elements_count' => $this->minElementsCount, + 'max_elements_count' => $this->maxElementsCount, + ]; + } + public function nullCount(): int { return $this->nullsCount; @@ -191,7 +219,8 @@ private function scalars(array $physicals): void { // @mago-ignore analysis:mixed-assignment foreach ($physicals as $value) { - if (is_int($value) || is_float($value) || is_bool($value)) { + // NaN is unordered: as in Parquet statistics and arrow-rs, it is never a bound + if (is_int($value) || is_float($value) && !is_nan($value) || is_bool($value)) { $this->min = min($this->min ?? $value, $value); $this->max = max($this->max ?? $value, $value); } diff --git a/src/core/etl/src/Flow/ETL/Runtime/Metrics/Statistics/Columns.php b/src/core/etl/src/Flow/ETL/Runtime/Metrics/Statistics/Columns.php index e453efdc9..b81d52930 100644 --- a/src/core/etl/src/Flow/ETL/Runtime/Metrics/Statistics/Columns.php +++ b/src/core/etl/src/Flow/ETL/Runtime/Metrics/Statistics/Columns.php @@ -10,6 +10,7 @@ use InvalidArgumentException; use function array_key_exists; +use function array_map; use function array_values; use function sprintf; @@ -58,4 +59,12 @@ public function get(string|Reference $ref): Column return $this->columns[$ref]; } + + /** + * @return list> + */ + public function normalize(): array + { + return array_map(static fn(Column $column): array => $column->normalize(), array_values($this->columns)); + } } diff --git a/src/core/etl/src/Flow/ETL/Runtime/Metrics/Statistics/ExecutionTime.php b/src/core/etl/src/Flow/ETL/Runtime/Metrics/Statistics/ExecutionTime.php index 1a7c02339..999197b69 100644 --- a/src/core/etl/src/Flow/ETL/Runtime/Metrics/Statistics/ExecutionTime.php +++ b/src/core/etl/src/Flow/ETL/Runtime/Metrics/Statistics/ExecutionTime.php @@ -6,6 +6,7 @@ use DateInterval; use DateTimeImmutable; +use DateTimeInterface; use Flow\ETL\Foundation\Clock\HighResolutionTime; use Flow\ETL\Foundation\Exception\InvalidArgumentException; @@ -30,4 +31,16 @@ public function inSeconds(): int { return $this->finishedAt->getTimestamp() - $this->startedAt->getTimestamp(); } + + /** + * @return array{started_at: string, finished_at: string, duration_ns: int} + */ + public function normalize(): array + { + return [ + 'started_at' => $this->startedAt->format(DateTimeInterface::RFC3339_EXTENDED), + 'finished_at' => $this->finishedAt->format(DateTimeInterface::RFC3339_EXTENDED), + 'duration_ns' => $this->highResolutionTime->toNanoseconds(), + ]; + } } diff --git a/src/core/etl/src/Flow/ETL/Runtime/Telemetry/TelemetryContext.php b/src/core/etl/src/Flow/ETL/Runtime/Telemetry/TelemetryContext.php index 6055a9847..8f2d6461c 100644 --- a/src/core/etl/src/Flow/ETL/Runtime/Telemetry/TelemetryContext.php +++ b/src/core/etl/src/Flow/ETL/Runtime/Telemetry/TelemetryContext.php @@ -137,9 +137,7 @@ public function dataFrameCompleted(array $attributes = []): void } $this->dataFrameExecutionTime = null; - $this->totalRowsProcessed = 0; $this->dataFrameSpan = null; - $this->memory = new Consumption($this->backend); } /** @@ -195,9 +193,7 @@ public function dataFrameFailed(Throwable $exception, array $attributes = []): v } $this->dataFrameExecutionTime = null; - $this->totalRowsProcessed = 0; $this->dataFrameSpan = null; - $this->memory = new Consumption($this->backend); } /** @@ -242,6 +238,7 @@ public function dataFrameStarted(array $configuration): void $this->dataFrameExecutionTime = HighResolutionTime::now(); $this->totalRowsProcessed = 0; + $this->memory = new Consumption($this->backend); } /** @@ -257,6 +254,22 @@ public function logger(): Logger return $this->logger; } + /** + * The heap of the data frame started last, sampled once per batch; it stays readable after the data frame ends. + */ + public function memory(): Consumption + { + return $this->memory; + } + + /** + * The rows that reached the end of the data frame started last; it stays readable after the data frame ends. + */ + public function totalRows(): int + { + return $this->totalRowsProcessed; + } + /** * Completes the span stepStarted() opened; a span already completed or drained is left alone. * diff --git a/src/core/etl/src/Flow/ETL/Values/Cardinality.php b/src/core/etl/src/Flow/ETL/Values/Cardinality.php index 21892f4d9..193ac8ae5 100644 --- a/src/core/etl/src/Flow/ETL/Values/Cardinality.php +++ b/src/core/etl/src/Flow/ETL/Values/Cardinality.php @@ -6,6 +6,8 @@ use Flow\ETL\Foundation\Exception\InvalidArgumentException; +use function is_finite; + final readonly class Cardinality { public const float DEFAULT_RELATIVE_ERROR = 0.5; @@ -105,4 +107,18 @@ public function merge(self $other): self relativeError: max($this->relativeError, $other->relativeError), ); } + + /** + * relative_error is null when it is not a finite number. + * + * @return array{at_most: null|int<0, max>, estimate: null|int<0, max>, relative_error: null|float} + */ + public function normalize(): array + { + return [ + 'at_most' => $this->atMost, + 'estimate' => $this->estimate, + 'relative_error' => is_finite($this->relativeError) ? $this->relativeError : null, + ]; + } } diff --git a/src/core/etl/src/Flow/ETL/Values/Memory/Consumption.php b/src/core/etl/src/Flow/ETL/Values/Memory/Consumption.php index 3ec6a9fff..97032caec 100644 --- a/src/core/etl/src/Flow/ETL/Values/Memory/Consumption.php +++ b/src/core/etl/src/Flow/ETL/Values/Memory/Consumption.php @@ -69,4 +69,12 @@ public function minDiff(): Unit { return $this->min()->diff($this->initial()); } + + /** + * @return array{initial: int, min: int, max: int} in bytes + */ + public function normalize(): array + { + return ['initial' => $this->initial->inBytes(), 'min' => $this->min->inBytes(), 'max' => $this->max->inBytes()]; + } } diff --git a/src/core/etl/src/Flow/ETL/Values/Memory/Unit.php b/src/core/etl/src/Flow/ETL/Values/Memory/Unit.php index fffb9f215..be2268bed 100644 --- a/src/core/etl/src/Flow/ETL/Values/Memory/Unit.php +++ b/src/core/etl/src/Flow/ETL/Values/Memory/Unit.php @@ -5,9 +5,11 @@ namespace Flow\ETL\Values\Memory; use Flow\ETL\Foundation\Exception\InvalidArgumentException; +use Flow\ETL\Foundation\Exception\RuntimeException; use function abs; use function ctype_digit; +use function getrusage; use function round; use function str_replace; use function strtoupper; @@ -58,6 +60,17 @@ public static function fromString(string $memoryString): self }; } + /** + * The peak resident set size of this process so far, as the OS counts it: Rust allocations and shared libraries + * included. getrusage() reports it in bytes on macOS and in KiB elsewhere. + */ + public static function processPeak(): self + { + $maxRss = (getrusage() ?: throw new RuntimeException('getrusage() failed'))['ru_maxrss']; + + return new self(PHP_OS_FAMILY === 'Darwin' ? $maxRss : $maxRss * 1024); + } + public function absolute(): self { return new self(abs($this->bytes)); diff --git a/src/core/etl/tests/Flow/ETL/Tests/Double/RecordingLauncher.php b/src/core/etl/tests/Flow/ETL/Tests/Double/RecordingLauncher.php index 7a1b30011..ded93420d 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Double/RecordingLauncher.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Double/RecordingLauncher.php @@ -8,7 +8,7 @@ use Flow\ETL\IO\Extractor\SplittableExtractor; use Flow\ETL\IO\Inference\Sniffer; use Flow\ETL\Planning\Physical\PhysicalPlan; -use Flow\ETL\Runtime\Executor\SourceCounts; +use Flow\ETL\Runtime\Executor\RunCounts; use Flow\ETL\Runtime\FlowContext; use Flow\ETL\Runtime\Launcher\Launch; use Flow\ETL\Runtime\Launcher\Launcher; @@ -19,7 +19,7 @@ final class RecordingLauncher implements Launcher { /** - * @var list + * @var list */ public array $reads = []; @@ -40,11 +40,11 @@ public function read( SplittableExtractor $source, Criteria $criteria, FlowContext $context, - ?SourceCounts $sources = null, + ?RunCounts $counts = null, ): ?Launch { - $this->reads[] = [$plan, $source, $criteria, $context, $sources]; + $this->reads[] = [$plan, $source, $criteria, $context, $counts]; - return $this->delegate?->read($plan, $source, $criteria, $context, $sources); + return $this->delegate?->read($plan, $source, $criteria, $context, $counts); } public function sniff(Sniffer $sniffer, int $units): ?array diff --git a/src/core/etl/tests/Flow/ETL/Tests/Double/StubLauncher.php b/src/core/etl/tests/Flow/ETL/Tests/Double/StubLauncher.php index beadfd8d9..b01c8bbdb 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Double/StubLauncher.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Double/StubLauncher.php @@ -10,24 +10,27 @@ use Flow\ETL\Planning\Physical\PhysicalPlan; use Flow\ETL\Planning\Physical\PipelineSteps; use Flow\ETL\Runtime\Executor\AdoptedBatches; -use Flow\ETL\Runtime\Executor\SourceCounts; +use Flow\ETL\Runtime\Executor\RunCounts; use Flow\ETL\Runtime\Executor\StepRunner; use Flow\ETL\Runtime\FlowContext; use Flow\ETL\Runtime\Launcher\Launch; use Flow\ETL\Runtime\Launcher\Launcher; /** - * Launches every plan the same way: $steps run over the source here, the parent resumes with $rest. + * Launches every plan the same way: $steps run over the source here, the parent resumes with $rest, and every + * worker in $workers reports its peak. */ final readonly class StubLauncher implements Launcher { /** * @param int<1, max> $consumed + * @param array $workers the peak bytes of each worker, by pid */ public function __construct( private PipelineSteps $steps = new PipelineSteps(), private PipelineSteps $rest = new PipelineSteps(), private int $consumed = 1, + private array $workers = [], ) {} public function read( @@ -35,16 +38,20 @@ public function read( SplittableExtractor $source, Criteria $criteria, FlowContext $context, - ?SourceCounts $sources = null, + ?RunCounts $counts = null, ): Launch { $batches = (new AdoptedBatches($context->backend()))->of($source->extract($context->backend(), $criteria)); + foreach ($this->workers as $pid => $peakBytes) { + $counts?->worker($pid, $peakBytes); + } + return new Launch( $this->consumed, $this->rest, (new StepRunner())->steps( $this->steps, - $sources === null ? $batches : $sources->count($source, $batches, $criteria->narrows()), + $counts === null ? $batches : $counts->count($source, $batches, $criteria->narrows()), $context, $source, ), diff --git a/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/AnalyzeTest.php b/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/AnalyzeTest.php index 61d34731f..730355c82 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/AnalyzeTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/AnalyzeTest.php @@ -12,13 +12,19 @@ use Flow\ETL\Runtime\Metrics\SourceStatistics; use Flow\ETL\Runtime\Metrics\Statistics\Columns; use Flow\ETL\Tests\Context\FloeFilesContext; +use Flow\ETL\Tests\Context\MemoryTelemetryContext; use Flow\ETL\Tests\Double\InlineLoader; +use Flow\ETL\Tests\Double\RecordingSplittableExtractor; +use Flow\ETL\Tests\Double\StubLauncher; use Flow\ETL\Tests\FlowIntegrationTestCase; use Flow\ETL\Tests\Mother\MarketRowsMother; +use Flow\ETL\Tests\Mother\RowsMother; use Flow\ETL\Values\Cardinality; use Flow\ETL\Values\Column\Backend; use Flow\ETL\Values\Join\JoinType; +use Flow\ETL\Values\Memory\Unit; use Flow\ETL\Values\Rows; +use Flow\Telemetry\Logger\Severity; use function Flow\ETL\DSL\analyze; use function Flow\ETL\DSL\config_builder; @@ -64,7 +70,7 @@ public function test_analyzing_an_inferred_array_source(): void $report->schema(), ); static::assertSame(7, $report->schema()?->count()); - static::assertGreaterThan(0, $report->statistics()->memory->max()->inBytes()); + static::assertGreaterThan(0, $report->statistics()->heap->max()->inBytes()); static::assertInstanceOf(DateTimeImmutable::class, $report->statistics()->executionTime->startedAt); static::assertInstanceOf(DateTimeImmutable::class, $report->statistics()->executionTime->finishedAt); static::assertGreaterThanOrEqual( @@ -77,6 +83,49 @@ public function test_analyzing_an_inferred_array_source(): void static::assertInstanceOf(Columns::class, $report->statistics()->columns); } + public function test_analyzing_without_telemetry_reports_the_rows_and_heap_of_each_run(): void + { + $dataFrame = df(config_builder()->build())->read(from_array([['id' => 1], ['id' => 2]])); + + $first = $dataFrame->run(analyze()); + $second = $dataFrame->run(analyze()); + + static::assertSame(2, $first->statistics()->totalRows()); + static::assertSame(2, $second->statistics()->totalRows()); + static::assertGreaterThan(0, $first->statistics()->heap->max()->inBytes()); + static::assertGreaterThan(0, $second->statistics()->heap->max()->inBytes()); + static::assertNotSame($first->statistics()->heap, $second->statistics()->heap); + } + + public function test_a_launched_run_reports_the_peak_of_every_worker(): void + { + $report = df(config_builder()->launcher(new StubLauncher(workers: [ + 101 => 40_000_000, + 102 => 41_000_000, + ]))->build())->read(new RecordingSplittableExtractor(schema(int_schema('id')), [RowsMother::sequentialIds( + 3, + )]))->run(analyze()); + + static::assertSame(3, $report->statistics()->totalRows()); + static::assertEquals(Unit::fromBytes(41_000_000), $report->statistics()->processes->workerPeak()); + static::assertSame( + $report->statistics()->processes->parent->inBytes() + 81_000_000, + $report->statistics()->processes->peakSum()->inBytes(), + ); + } + + public function test_a_launched_run_that_collects_the_schema_warns_once_and_still_reports_it(): void + { + $telemetry = new MemoryTelemetryContext(); + + $report = df(config_builder()->withTelemetry($telemetry->telemetry)->launcher(new StubLauncher())->build()) + ->read(new RecordingSplittableExtractor(schema(int_schema('id')), [RowsMother::sequentialIds(3)])) + ->run(analyze()->withSchema()); + + static::assertCount(1, $telemetry->logs->entriesWithSeverity(Severity::WARN)); + static::assertSame(1, $report->schema()?->count()); + } + public function test_analyzing_csv_file_with_limit(): void { $report = df() diff --git a/src/core/etl/tests/Flow/ETL/Tests/Mother/ReportMother.php b/src/core/etl/tests/Flow/ETL/Tests/Mother/ReportMother.php new file mode 100644 index 000000000..92cc67ab8 --- /dev/null +++ b/src/core/etl/tests/Flow/ETL/Tests/Mother/ReportMother.php @@ -0,0 +1,49 @@ + $sources + */ + public static function of( + Processes $processes, + int $totalRows = 0, + ?Schema $schema = null, + ?array $sources = null, + ?Columns $columns = null, + ): Report { + return new Report( + $schema, + new Statistics( + $totalRows, + new ExecutionTime( + new DateTimeImmutable('2025-01-01 00:00:00 UTC'), + new DateTimeImmutable('2025-01-01 00:00:01 UTC'), + new HighResolutionTime(1, 0), + ), + new Consumption(new PhpBackend()), + $columns, + $processes, + ), + $sources, + ); + } +} diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Planning/Physical/PlannerTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Planning/Physical/PlannerTest.php index 60114c2eb..89353bd8d 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Planning/Physical/PlannerTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Planning/Physical/PlannerTest.php @@ -51,7 +51,7 @@ use Flow\ETL\Planning\Physical\Planner; use Flow\ETL\Planning\Physical\TransactionGroup; use Flow\ETL\Runtime\Executor\Executor; -use Flow\ETL\Runtime\Executor\SourceCounts; +use Flow\ETL\Runtime\Executor\RunCounts; use Flow\ETL\Runtime\Metrics\SourceStatistics; use Flow\ETL\Storage\Memory\ArrayMemory; use Flow\ETL\Storage\MemoryRowsStorage; @@ -354,7 +354,7 @@ public function test_a_joins_right_side_counts_its_source_rows_too(): void 'r_', ); $context = NodeMother::context(config()); - $sources = new SourceCounts(); + $sources = new RunCounts(); $plan = $context->config->planner()->plan(new LogicalPlan(new Result($node))); iterator_to_array($context->config->executor()->executePlan($plan, $context, $sources)); diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Executor/ExecutorTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Executor/ExecutorTest.php index 0c0c9e429..7f776e12f 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Executor/ExecutorTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Executor/ExecutorTest.php @@ -27,7 +27,7 @@ use Flow\ETL\Planning\Physical\Pipeline; use Flow\ETL\Planning\Physical\PipelineSteps; use Flow\ETL\Runtime\Executor\Executor; -use Flow\ETL\Runtime\Executor\SourceCounts; +use Flow\ETL\Runtime\Executor\RunCounts; use Flow\ETL\Tests\Context\ExecutedPlan; use Flow\ETL\Tests\Context\GroupByContext; use Flow\ETL\Tests\Context\MemoryTelemetryContext; @@ -791,7 +791,7 @@ public function test_the_launcher_gets_the_sources_and_the_parent_resumes_after_ { $extractor = new RecordingSplittableExtractor(schema(int_schema('id')), [RowsMother::sequentialIds(4)]); $filter = new ScalarFunctionFilterTransformer(ref('id')->greaterThan(lit(2))); - $sources = new SourceCounts(); + $sources = new RunCounts(); $launcher = new RecordingLauncher(new StubLauncher(new PipelineSteps([$filter]))); $rows = iterator_to_array((new Executor())->executePlan( @@ -829,7 +829,7 @@ public function test_the_rest_replaces_the_leaf_pipelines_steps(): void public function test_only_the_pipeline_that_reads_the_source_counts_its_rows(): void { - $sources = new SourceCounts(); + $sources = new RunCounts(); $context = NodeMother::context(); iterator_to_array((new Executor())->execute( diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Executor/SourceCountsTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Executor/RunCountsTest.php similarity index 77% rename from src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Executor/SourceCountsTest.php rename to src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Executor/RunCountsTest.php index 6dfc680cc..5ec2d6ad0 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Executor/SourceCountsTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Executor/RunCountsTest.php @@ -7,11 +7,13 @@ use Flow\ETL\IO\Extractor\Criteria; use Flow\ETL\IO\Extractor\Signal; use Flow\ETL\IO\Extractor\Statistics; -use Flow\ETL\Runtime\Executor\SourceCounts; +use Flow\ETL\Runtime\Executor\RunCounts; +use Flow\ETL\Runtime\Metrics\Processes; use Flow\ETL\Runtime\Metrics\SourceStatistics; use Flow\ETL\Tests\Double\CountingExtractor; use Flow\ETL\Tests\FlowTestCase; use Flow\ETL\Values\Cardinality; +use Flow\ETL\Values\Memory\Unit; use Flow\ETL\Values\Rows; use function Flow\ETL\DSL\array_to_rows; @@ -21,12 +23,12 @@ use function Flow\ETL\DSL\schema; use function iterator_to_array; -final class SourceCountsTest extends FlowTestCase +final class RunCountsTest extends FlowTestCase { public function test_every_batch_passes_through_unchanged_and_is_counted(): void { $extractor = from_array([['id' => 1], ['id' => 2], ['id' => 3]])->withBatchSize(2); - $sources = new SourceCounts(); + $sources = new RunCounts(); $batches = iterator_to_array( $sources->count( @@ -50,8 +52,8 @@ public function test_every_batch_passes_through_unchanged_and_is_counted(): void public function test_an_empty_one_absorbs_every_read_with_what_its_source_declared(): void { $extractor = from_array([['id' => 1], ['id' => 2], ['id' => 3]]); - $attempt = new SourceCounts(); - $sources = new SourceCounts(); + $attempt = new RunCounts(); + $sources = new RunCounts(); iterator_to_array($attempt->count( $extractor, FlowTestCase::extracted($extractor, flow_context()->backend(), new Criteria()), @@ -66,11 +68,63 @@ public function test_an_empty_one_absorbs_every_read_with_what_its_source_declar ); } + public function test_a_worker_keeps_its_largest_peak_in_the_order_workers_were_first_seen(): void + { + $counts = new RunCounts(); + $counts->worker(202, 30); + $counts->worker(101, 50); + $counts->worker(202, 40); + $counts->worker(101, 10); + + static::assertEquals(new Processes(Unit::fromBytes(7), [ + Unit::fromBytes(40), + Unit::fromBytes(50), + ]), $counts->processes(Unit::fromBytes(7))); + } + + public function test_a_run_without_workers_reports_the_parent_only(): void + { + static::assertEquals(new Processes(Unit::fromBytes(7), []), (new RunCounts())->processes(Unit::fromBytes(7))); + } + + public function test_absorb_takes_over_the_worker_peaks_with_the_reads(): void + { + $extractor = from_array([['id' => 1]]); + $attempt = new RunCounts(); + $attempt->open($extractor, false)->counted(1); + $attempt->worker(101, 50); + $counts = new RunCounts(); + $counts->worker(101, 60); + $counts->worker(102, 20); + + $counts->absorb($attempt); + + static::assertSame(1, $counts->statistics()[0]->rows); + static::assertEquals( + [Unit::fromBytes(60), Unit::fromBytes(20)], + $counts->processes(Unit::fromBytes(0))->workers, + ); + } + + public function test_absorb_workers_takes_over_the_worker_peaks_and_no_read(): void + { + $extractor = from_array([['id' => 1]]); + $abandoned = new RunCounts(); + $abandoned->open($extractor, false)->counted(1); + $abandoned->worker(101, 50); + $counts = new RunCounts(); + + $counts->absorbWorkers($abandoned); + + static::assertSame([], $counts->statistics()); + static::assertEquals([Unit::fromBytes(50)], $counts->processes(Unit::fromBytes(0))->workers); + } + public function test_an_absorbed_read_of_a_source_already_counted_adds_its_rows(): void { $extractor = from_array([['id' => 1], ['id' => 2]]); - $attempt = new SourceCounts(); - $sources = new SourceCounts(); + $attempt = new RunCounts(); + $sources = new RunCounts(); iterator_to_array($sources->count( $extractor, FlowTestCase::extracted($extractor, flow_context()->backend(), new Criteria()), @@ -93,8 +147,8 @@ public function test_an_absorbed_open_or_narrowed_read_leaves_the_source_incompl { $open = from_array([['id' => 1]]); $narrowed = from_array([['id' => 2]]); - $attempt = new SourceCounts(); - $sources = new SourceCounts(); + $attempt = new RunCounts(); + $sources = new RunCounts(); $attempt->open($open, false)->counted(1); iterator_to_array($attempt->count( $narrowed, @@ -115,7 +169,7 @@ public function test_a_stop_reaches_the_source_and_ends_the_count(): void array_to_rows([['id' => 1], ['id' => 2], ['id' => 3]], schema(int_schema('id'))), ); $extractor->withBatchSize(1); - $sources = new SourceCounts(); + $sources = new RunCounts(); $counted = $sources->count( $extractor, FlowTestCase::extracted($extractor, flow_context()->backend(), new Criteria()), @@ -137,7 +191,7 @@ public function test_a_stop_reaches_the_source_and_ends_the_count(): void public function test_a_narrowed_read_is_not_complete(): void { $extractor = from_array([['id' => 1], ['id' => 2]]); - $sources = new SourceCounts(); + $sources = new RunCounts(); iterator_to_array($sources->count( $extractor, @@ -152,7 +206,7 @@ public function test_a_narrowed_read_is_not_complete(): void public function test_one_narrowed_read_makes_the_source_incomplete(): void { $extractor = from_array([['id' => 1]]); - $sources = new SourceCounts(); + $sources = new RunCounts(); iterator_to_array($sources->count( $extractor, @@ -171,7 +225,7 @@ public function test_one_narrowed_read_makes_the_source_incomplete(): void public function test_interleaved_reads_of_one_source_add_up(): void { $extractor = from_array([['id' => 1], ['id' => 2]])->withBatchSize(1); - $sources = new SourceCounts(); + $sources = new RunCounts(); $first = $sources->count( $extractor, FlowTestCase::extracted($extractor, flow_context()->backend(), new Criteria()), @@ -197,7 +251,7 @@ public function test_interleaved_reads_of_one_source_add_up(): void public function test_a_source_read_twice_is_listed_once_with_both_counts(): void { $extractor = from_array([['id' => 1], ['id' => 2]]); - $sources = new SourceCounts(); + $sources = new RunCounts(); iterator_to_array($sources->count( $extractor, @@ -218,7 +272,7 @@ public function test_sources_are_listed_in_the_order_they_were_first_read(): voi { $first = from_array([['id' => 1]]); $second = from_array([['id' => 1], ['id' => 2]]); - $sources = new SourceCounts(); + $sources = new RunCounts(); iterator_to_array($sources->count( $second, @@ -237,7 +291,7 @@ public function test_sources_are_listed_in_the_order_they_were_first_read(): voi public function test_a_source_that_yields_nothing_is_listed_with_zero_rows(): void { $extractor = from_array([]); - $sources = new SourceCounts(); + $sources = new RunCounts(); iterator_to_array($sources->count( $extractor, @@ -251,7 +305,7 @@ public function test_a_source_that_yields_nothing_is_listed_with_zero_rows(): vo public function test_a_source_never_read_is_not_listed(): void { $extractor = from_array([['id' => 1]]); - $sources = new SourceCounts(); + $sources = new RunCounts(); $sources->count( $extractor, @@ -266,7 +320,7 @@ public function test_open_returns_the_same_read_per_extractor(): void { $extractor = from_array([['id' => 1]]); $other = from_array([['id' => 2]]); - $sources = new SourceCounts(); + $sources = new RunCounts(); $read = $sources->open($extractor, false); $read->counted(5); diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Executor/StatisticsCollectorTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Executor/StatisticsCollectorTest.php index c87f82bde..d964e5bf7 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Executor/StatisticsCollectorTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Executor/StatisticsCollectorTest.php @@ -14,17 +14,25 @@ use Flow\ETL\Runtime\Metrics\Report; use Flow\ETL\Runtime\Metrics\SourceStatistics; use Flow\ETL\Runtime\Metrics\Statistics\Columns; +use Flow\ETL\Tests\Context\MemoryTelemetryContext; +use Flow\ETL\Tests\Double\StubLauncher; use Flow\ETL\Tests\FlowTestCase; use Flow\ETL\Values\Cardinality; +use Flow\ETL\Values\Memory\Unit; +use Flow\Telemetry\Logger\LogEntry; +use Flow\Telemetry\Logger\Severity; +use function array_map; use function Flow\ETL\DSL\analyze; use function Flow\ETL\DSL\array_to_rows; +use function Flow\ETL\DSL\config_builder; use function Flow\ETL\DSL\flow_context; use function Flow\ETL\DSL\from_array; use function Flow\ETL\DSL\int_schema; use function Flow\ETL\DSL\schema; use function Flow\ETL\DSL\str_schema; use function iterator_to_array; +use function memory_get_peak_usage; final class StatisticsCollectorTest extends FlowTestCase { @@ -170,17 +178,17 @@ public function test_report_returns_report_with_memory_consumption(): void $report = $collector->report(); static::assertNotNull($report); - static::assertGreaterThanOrEqual(0, $report->statistics()->memory->initial()->inBytes()); + static::assertGreaterThanOrEqual(0, $report->statistics()->heap->initial()->inBytes()); } public function test_sources_are_counted_and_reported_when_enabled(): void { $collector = new StatisticsCollector(analyze()->withSourceStatistics(), flow_context()); $extractor = from_array([['id' => 1], ['id' => 2]]); - $sources = $collector->sources(); - static::assertNotNull($sources); + $counts = $collector->counts(); + static::assertNotNull($counts); - iterator_to_array($sources->count( + iterator_to_array($counts->count( $extractor, $extractor->extract(flow_context()->backend(), new Criteria()), false, @@ -192,17 +200,45 @@ public function test_sources_are_counted_and_reported_when_enabled(): void ); } - public function test_sources_are_not_counted_when_not_enabled(): void + public function test_sources_are_counted_but_not_reported_when_not_enabled(): void { $collector = new StatisticsCollector(analyze(), flow_context()); + $extractor = from_array([['id' => 1]]); + $counts = $collector->counts(); + static::assertNotNull($counts); + + iterator_to_array($counts->count( + $extractor, + $extractor->extract(flow_context()->backend(), new Criteria()), + false, + )); - static::assertNull($collector->sources()); static::assertNull($collector->report()->sources()); } - public function test_sources_are_not_counted_when_analyze_is_false(): void + public function test_nothing_is_counted_when_analyze_is_false(): void { - static::assertNull((new StatisticsCollector(false, flow_context()))->sources()); + static::assertNull((new StatisticsCollector(false, flow_context()))->counts()); + } + + public function test_a_serial_run_reports_the_parent_peak_and_no_workers(): void + { + $processes = (new StatisticsCollector(analyze(), flow_context()))->report()->statistics()->processes; + + static::assertSame([], $processes->workers); + static::assertGreaterThanOrEqual(memory_get_peak_usage(true), $processes->parent->inBytes()); + } + + public function test_workers_counted_under_analyze_reach_the_report(): void + { + $collector = new StatisticsCollector(analyze(), flow_context()); + $collector->counts()?->worker(101, 40_000_000); + $collector->counts()?->worker(102, 41_000_000); + + static::assertEquals( + [Unit::fromBytes(40_000_000), Unit::fromBytes(41_000_000)], + $collector->report()->statistics()->processes->workers, + ); } public function test_schema_is_null_when_not_enabled(): void @@ -228,4 +264,52 @@ public function test_reads_columns(): void ); static::assertFalse((new StatisticsCollector(false, flow_context()))->readsColumns()); } + + public function test_reading_columns_under_a_launcher_warns_that_writes_do_not_run_in_parallel(): void + { + $telemetry = new MemoryTelemetryContext(); + $config = config_builder()->withTelemetry($telemetry->telemetry)->launcher(new StubLauncher())->build(); + + new StatisticsCollector(analyze()->withSchema()->withColumnStatistics(), flow_context($config)); + + static::assertSame( + [ + 'analyze()->withSchema()->withColumnStatistics() sends every row to the parent process, so the writes do not ' + . 'run in parallel. Drop it to keep parallel writes.', + ], + array_map( + static fn(LogEntry $entry): string => $entry->record->body, + $telemetry->logs->entriesWithSeverity(Severity::WARN), + ), + ); + } + + public function test_column_statistics_alone_under_a_launcher_names_only_them(): void + { + $telemetry = new MemoryTelemetryContext(); + $config = config_builder()->withTelemetry($telemetry->telemetry)->launcher(new StubLauncher())->build(); + + new StatisticsCollector(analyze()->withColumnStatistics(), flow_context($config)); + + static::assertStringStartsWith( + 'analyze()->withColumnStatistics() sends', + $telemetry->logs->entriesWithSeverity(Severity::WARN)[0]->record->body, + ); + } + + public function test_nothing_is_warned_without_a_launcher_or_without_reading_columns(): void + { + $telemetry = new MemoryTelemetryContext(); + + new StatisticsCollector( + analyze()->withSchema(), + flow_context(config_builder()->withTelemetry($telemetry->telemetry)->build()), + ); + new StatisticsCollector( + analyze()->withSourceStatistics(), + flow_context(config_builder()->withTelemetry($telemetry->telemetry)->launcher(new StubLauncher())->build()), + ); + + static::assertSame([], $telemetry->logs->entriesWithSeverity(Severity::WARN)); + } } diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/ProcessesTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/ProcessesTest.php new file mode 100644 index 000000000..571b0b5c7 --- /dev/null +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/ProcessesTest.php @@ -0,0 +1,50 @@ +workerPeak()); + static::assertEquals(Unit::fromBytes(25), $processes->peakSum()); + } + + public function test_the_worker_peak_is_the_largest_worker(): void + { + static::assertEquals(Unit::fromBytes(41), (new Processes(Unit::fromBytes(25), [ + Unit::fromBytes(40), + Unit::fromBytes(41), + Unit::fromBytes(39), + ]))->workerPeak()); + } + + public function test_the_peak_sum_adds_the_parent_and_every_worker(): void + { + static::assertEquals(Unit::fromBytes(145), (new Processes(Unit::fromBytes(25), [ + Unit::fromBytes(40), + Unit::fromBytes(41), + Unit::fromBytes(39), + ]))->peakSum()); + } + + public function test_normalize_gives_every_peak_in_bytes(): void + { + static::assertSame( + ['parent' => 25, 'workers' => [40, 41], 'worker_peak' => 41, 'peak_sum' => 106], + (new Processes(Unit::fromBytes(25), [Unit::fromBytes(40), Unit::fromBytes(41)]))->normalize(), + ); + static::assertSame( + ['parent' => 25, 'workers' => [], 'worker_peak' => null, 'peak_sum' => 25], + (new Processes(Unit::fromBytes(25), []))->normalize(), + ); + } +} diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/ReportTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/ReportTest.php new file mode 100644 index 000000000..bed4e2171 --- /dev/null +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/ReportTest.php @@ -0,0 +1,85 @@ +normalize(); + + static::assertSame( + [ + 'schema' => schema(int_schema('id'))->normalize(), + 'statistics' => [ + 'total_rows' => 2, + 'execution_time' => $report->statistics()->executionTime->normalize(), + 'heap' => $report->statistics()->heap->normalize(), + 'processes' => ['parent' => 25, 'workers' => [40], 'worker_peak' => 40, 'peak_sum' => 65], + 'columns' => null, + ], + 'sources' => [$source->normalize()], + ], + $normalized, + ); + // JSON keeps the values, not a float's ".0" + static::assertEquals($normalized, json_decode( + json_encode($normalized, JSON_THROW_ON_ERROR), + true, + flags: JSON_THROW_ON_ERROR, + )); + } + + public function test_a_report_without_schema_or_sources_normalizes_them_to_null(): void + { + $normalized = ReportMother::of(new Processes(Unit::fromBytes(25), []))->normalize(); + + static::assertNull($normalized['schema']); + static::assertNull($normalized['sources']); + } + + public function test_a_report_holding_infinite_and_nan_values_encodes_as_json(): void + { + $definition = float_schema('f'); + $columns = new Columns(); + $columns->add($definition, ColumnMother::of($definition, [INF, NAN, -INF])); + $report = ReportMother::of( + new Processes(Unit::fromBytes(25), []), + columns: $columns, + sources: [new SourceStatistics('ArrayExtractor', new Statistics(new Cardinality(null, 5, INF)), 0, true)], + ); + + $json = json_encode($report->normalize(), JSON_THROW_ON_ERROR); + + static::assertStringContainsString('"rows_error":null', $json); + static::assertStringContainsString('"relative_error":null', $json); + static::assertStringContainsString('"max":"', $json); + } +} diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/SourceStatisticsTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/SourceStatisticsTest.php index e8938b036..0128591c0 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/SourceStatisticsTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/SourceStatisticsTest.php @@ -76,4 +76,40 @@ public function test_a_positive_estimate_against_zero_rows_is_infinitely_off(): (new SourceStatistics('Source', new Statistics(Cardinality::approximately(500)), 0, true))->rowsError(), ); } + + public function test_normalize(): void + { + static::assertSame( + [ + 'extractor' => 'Source', + 'declared' => [ + 'rows' => ['at_most' => 5, 'estimate' => 5, 'relative_error' => 0.0], + 'size' => ['at_most' => null, 'estimate' => null, 'relative_error' => 0.5], + ], + 'rows' => 5, + 'complete' => true, + 'rows_error' => 0.0, + ], + (new SourceStatistics('Source', new Statistics(Cardinality::exact(5)), 5, true))->normalize(), + ); + } + + public function test_normalize_gives_no_rows_error_when_it_is_infinite_or_unknown(): void + { + static::assertNull( + (new SourceStatistics('Source', new Statistics(Cardinality::exact(5)), 0, true))->normalize()['rows_error'], + ); + static::assertNull( + (new SourceStatistics( + 'Source', + new Statistics(Cardinality::exact(5)), + 2, + false, + ))->normalize()['rows_error'], + ); + static::assertSame( + 0.5, + (new SourceStatistics('Source', new Statistics(Cardinality::exact(3)), 2, true))->normalize()['rows_error'], + ); + } } diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/Statistics/ColumnTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/Statistics/ColumnTest.php index 5238e1cd0..639466d06 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/Statistics/ColumnTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/Statistics/ColumnTest.php @@ -30,6 +30,7 @@ use function Flow\Types\DSL\type_structure; use function Flow\Types\DSL\type_time_zone; +use const INF; use const NAN; final class ColumnTest extends FlowTestCase @@ -257,4 +258,58 @@ public function test_nans_are_one_distinct_value(): void (new Column($definition, ColumnMother::of($definition, [NAN, NAN, 1.0])))->distinctCount(), ); } + + public function test_nan_is_never_a_bound_whatever_its_position(): void + { + $definition = float_schema('f', true); + + $first = new Column($definition, ColumnMother::of($definition, [NAN, 1.0])); + $last = new Column($definition, ColumnMother::of($definition, [1.0, NAN])); + $onlyNan = new Column($definition, ColumnMother::of($definition, [NAN, null, NAN])); + + static::assertSame([1.0, 1.0], [$first->min(), $first->max()]); + static::assertSame([1.0, 1.0], [$last->min(), $last->max()]); + static::assertSame([null, null], [$onlyNan->min(), $onlyNan->max()]); + } + + public function test_normalize_gives_every_statistic_and_dates_as_text(): void + { + $definition = datetime_schema('at', true); + + static::assertSame( + [ + 'name' => 'at', + 'nulls_count' => 1, + 'distinct_count' => 2, + 'min' => '2024-01-01T00:00:00.000+00:00', + 'max' => '2024-01-02T00:00:00.000+00:00', + 'min_length' => null, + 'max_length' => null, + 'min_elements_count' => null, + 'max_elements_count' => null, + ], + (new Column($definition, ColumnMother::of($definition, [ + new DateTimeImmutable('2024-01-02 00:00:00 UTC'), + null, + new DateTimeImmutable('2024-01-01 00:00:00 UTC'), + ])))->normalize(), + ); + } + + public function test_normalize_keeps_scalar_bounds(): void + { + $definition = str_schema('name'); + $normalized = (new Column($definition, ColumnMother::of($definition, ['a', 'abc'])))->normalize(); + + static::assertSame([1, 3], [$normalized['min_length'], $normalized['max_length']]); + } + + public function test_normalize_names_float_bounds_json_cannot_hold(): void + { + $definition = float_schema('f'); + + $infinite = (new Column($definition, ColumnMother::of($definition, [INF, NAN, -INF, 1.0])))->normalize(); + + static::assertSame(['-INF', 'INF'], [$infinite['min'], $infinite['max']]); + } } diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/Statistics/ColumnsTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/Statistics/ColumnsTest.php index f7a307311..2a54c2c93 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/Statistics/ColumnsTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/Statistics/ColumnsTest.php @@ -9,6 +9,7 @@ use Flow\ETL\Tests\Mother\ColumnMother; use InvalidArgumentException; +use function array_column; use function Flow\ETL\DSL\integer_schema; use function Flow\ETL\DSL\string_schema; @@ -38,4 +39,13 @@ public function test_get_non_existing_column(): void $columns->get('non_existing'); } + + public function test_normalize_lists_every_column(): void + { + $columns = new Columns(); + $columns->add(integer_schema('a'), ColumnMother::of(integer_schema('a'), [1])); + $columns->add(string_schema('b'), ColumnMother::of(string_schema('b'), ['x'])); + + static::assertSame(['a', 'b'], array_column($columns->normalize(), 'name')); + } } diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/Statistics/ExecutionTimeTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/Statistics/ExecutionTimeTest.php new file mode 100644 index 000000000..867e711a9 --- /dev/null +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Metrics/Statistics/ExecutionTimeTest.php @@ -0,0 +1,29 @@ + '2025-01-01T00:00:00.000+00:00', + 'finished_at' => '2025-01-01T00:05:00.250+00:00', + 'duration_ns' => 300_250_000_000, + ], + (new ExecutionTime( + new DateTimeImmutable('2025-01-01 00:00:00 UTC'), + new DateTimeImmutable('2025-01-01 00:05:00.250 UTC'), + new HighResolutionTime(300, 250_000_000), + ))->normalize(), + ); + } +} diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Telemetry/TelemetryContextTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Telemetry/TelemetryContextTest.php index afe91a992..6adcafd79 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Telemetry/TelemetryContextTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Runtime/Telemetry/TelemetryContextTest.php @@ -48,6 +48,38 @@ final class TelemetryContextTest extends FlowTestCase { + public function test_a_finished_dataframe_keeps_its_rows_and_memory_until_the_next_one_starts(): void + { + $context = new MemoryTelemetryContext(); + + $context->telemetryContext->dataFrameStarted([]); + $context->telemetryContext->dataFrameBatchProcessed(array_to_rows([ + ['id' => 1], + ['id' => 2], + ], schema(int_schema('id')))); + $finished = $context->telemetryContext->memory(); + $context->telemetryContext->dataFrameCompleted(); + + static::assertSame(2, $context->telemetryContext->totalRows()); + static::assertSame($finished, $context->telemetryContext->memory()); + + $context->telemetryContext->dataFrameStarted([]); + + static::assertSame(0, $context->telemetryContext->totalRows()); + static::assertNotSame($finished, $context->telemetryContext->memory()); + } + + public function test_a_failed_dataframe_keeps_its_rows_until_the_next_one_starts(): void + { + $context = new MemoryTelemetryContext(); + + $context->telemetryContext->dataFrameStarted([]); + $context->telemetryContext->dataFrameBatchProcessed(array_to_rows([['id' => 1]], schema(int_schema('id')))); + $context->telemetryContext->dataFrameFailed(new RuntimeException('failed')); + + static::assertSame(1, $context->telemetryContext->totalRows()); + } + public function test_dataframe_batch_processed_drains_abandoned_spans(): void { $context = new MemoryTelemetryContext(telemetry_options(trace_transformations: true)); diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Values/CardinalityTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Values/CardinalityTest.php index d757f5186..45a480ac4 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Values/CardinalityTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Values/CardinalityTest.php @@ -13,6 +13,9 @@ use function str_repeat; use function strlen; +use const INF; +use const NAN; + final class CardinalityTest extends FlowTestCase { public function test_a_bound_survives_only_if_both_sides_have_one(): void @@ -165,4 +168,22 @@ public function test_a_validated_count_reads_back_as_a_non_negative_int(): void static::assertSame(5, strlen(str_repeat('x', Cardinality::atMost(5)->atMost ?? 0))); static::assertSame(7, strlen(str_repeat('x', Cardinality::approximately(7)->estimate ?? 0))); } + + public function test_normalize(): void + { + static::assertSame( + ['at_most' => 10, 'estimate' => 5, 'relative_error' => 0.25], + (new Cardinality(10, 5, 0.25))->normalize(), + ); + static::assertSame( + ['at_most' => null, 'estimate' => null, 'relative_error' => 0.5], + Cardinality::unknown()->normalize(), + ); + } + + public function test_normalize_gives_no_relative_error_when_it_is_not_finite(): void + { + static::assertNull((new Cardinality(null, 5, INF))->normalize()['relative_error']); + static::assertNull((new Cardinality(null, 5, NAN))->normalize()['relative_error']); + } } diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Values/Memory/ConsumptionTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Values/Memory/ConsumptionTest.php index ff255dd7e..a8153bece 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Values/Memory/ConsumptionTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Values/Memory/ConsumptionTest.php @@ -29,4 +29,18 @@ public function test_consumption_adds_the_backend_allocations(): void static::assertSame(1_048_576, $native->initial()->inBytes() - $php->initial()->inBytes()); static::assertSame(1_048_576, $native->capture()->inBytes() - $php->capture()->inBytes()); } + + public function test_normalize_gives_every_figure_in_bytes(): void + { + $consumption = new Consumption(new SpyBackend(1_048_576)); + + static::assertSame( + [ + 'initial' => $consumption->initial()->inBytes(), + 'min' => $consumption->min()->inBytes(), + 'max' => $consumption->max()->inBytes(), + ], + $consumption->normalize(), + ); + } } diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Values/Memory/UnitTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Values/Memory/UnitTest.php index c83a0e5f5..0af61352d 100644 --- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Values/Memory/UnitTest.php +++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Values/Memory/UnitTest.php @@ -7,8 +7,19 @@ use Flow\ETL\Tests\FlowTestCase; use Flow\ETL\Values\Memory\Unit; +use function memory_get_peak_usage; + final class UnitTest extends FlowTestCase { + public function test_the_process_peak_is_in_bytes_on_every_system(): void + { + $peak = Unit::processPeak()->inBytes(); + + // the resident set holds the whole PHP heap; a KiB value read as bytes, or bytes scaled as KiB, misses by 1024x + static::assertGreaterThanOrEqual(memory_get_peak_usage(true), $peak); + static::assertLessThan(memory_get_peak_usage(true) * 64, $peak); + } + public function test_calculating_percentage_from_value(): void { static::assertEquals(Unit::fromString('10M'), Unit::fromString('100M')->percentage(10));