feat(flow-php/etl): the run report shows the peak memory of every process

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)

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";
 ```

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'))

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

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'],
+                ],
         );
     }
 }

src/cli/tests/Flow/CLI/Tests/Unit/Formatter/PipelineReportFormatterTest.php

@@ -0,0 +1,52 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\CLI\Tests\Unit\Formatter;
+
+use Flow\CLI\Command\PipelineRunCommand;
+use Flow\CLI\Formatter\PipelineReportFormatter;
+use Flow\ETL\Runtime\Metrics\Processes;
+use Flow\ETL\Tests\FlowTestCase;
+use Flow\ETL\Tests\Mother\ReportMother;
+use Flow\ETL\Values\Memory\Unit;
+use Symfony\Component\Console\Input\ArrayInput;
+use Symfony\Component\Console\Output\BufferedOutput;
+use Symfony\Component\Console\Style\SymfonyStyle;
+
+final class PipelineReportFormatterTest extends FlowTestCase
+{
+    public function test_a_launched_run_prints_the_worker_peak_and_the_peak_sum(): void
+    {
+        $input = new ArrayInput(['pipeline-file' => '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);
+    }
+}

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);
         }

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<int, Rows, mixed, PhysicalPlan> 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<int, Rows>
      */
-    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

src/core/etl/src/Flow/ETL/Runtime/Executor/SourceCounts.php → 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<Extractor, SourceCount>
@@ -29,6 +35,11 @@ final class SourceCounts
      */
     private SplObjectStorage $declared;
 
+    /**
+     * @var array<int, int> the peak bytes of each worker, by pid, in the order worker() first saw them
+     */
+    private array $workers = [];
+
     public function __construct()
     {
         /** @var SplObjectStorage<Extractor, SourceCount> $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)));
     }
 
     /**