Extractor statistics

documentation/components/cli/docs.md

@@ -262,6 +262,7 @@ Options:
       --schema-union-by-name                               Union the column sets of every sniffed source instead of taking the first one.
       --stats-schema[=STATS-SCHEMA]                        Prints schema of executed data transformation pipeline. [default: false]
       --stats-columns[=STATS-COLUMNS]                      Prints number of rows in dataset. [default: false]
+      --stats-sources[=STATS-SOURCES]                      Prints the rows every source declared next to the rows it yielded. [default: false]
   -h, --help                                               Display help for the given command. When no command is given display help for the list command
       --silent                                             Do not output any message
   -q, --quiet                                              Only errors are displayed. All other output is suppressed
@@ -406,6 +407,7 @@ Options:
       --config=CONFIG                  Path to a local php file that MUST return instance of: Flow\ETL\Config
       --stats-schema[=STATS-SCHEMA]    Prints schema of executed data transformation pipeline. [default: false]
       --stats-columns[=STATS-COLUMNS]  Prints number of rows in dataset. [default: false]
+      --stats-sources[=STATS-SOURCES]  Prints the rows every source declared next to the rows it yielded. [default: false]
   -h, --help                           Display help for the given command. When no command is given display help for the list command
       --silent                         Do not output any message
   -q, --quiet                          Only errors are displayed. All other output is suppressed

documentation/components/core/core.md

@@ -48,7 +48,8 @@ tree read from the bottom up: a node's children are where its rows come from, an
 rows reach them - the source is `#1`. What a node does is listed under it.
 
 The verbs build a plan with no consumer on top; **the trigger adds the one it needs**, so `explain()` takes the
-trigger it should print - `Trigger::rows` by default, `Trigger::run` for the plan `run()` executes.
+trigger it should print - `Trigger::rows` by default, `Trigger::run` for the plan `run()` executes, `Trigger::count`
+for the plan `count()` executes.
 
 ```php
 echo data_frame()
@@ -154,6 +155,9 @@ step sits in is where rows stop flowing through, and every step lists the settin
 storage among them, which the logical stages cannot know because the planner picks it.
 
 ```php
+$users = [['id' => 1, 'name' => 'Alice'], ['id' => 2, 'name' => 'Bob']];
+$emails = [['id' => 1, 'email' => 'alice@example.com']];
+
 echo data_frame()
     ->read(from_array($users))
     ->join(data_frame()->read(from_array($emails)), join_on(['id' => 'id'], join_prefix: 'joined_'), Join::left)
@@ -164,13 +168,14 @@ echo data_frame()
 
 ```text
 Physical plan
-│  Schema: derived
+│  Columns: id, name, joined_id, joined_email
 └─ Pipeline #1
    │  Processor: CollectingProcessor
    │     Schema: declared
    │  Loader: StreamLoader
    └─ Pipeline #0
       │  Extractor: ArrayExtractor
+      │     Statistics: rows exact 2 · size unknown
       │  Processor: HashJoinProcessor
       │     Join: left
       │     On: id = id
@@ -180,9 +185,12 @@ Physical plan
       │     Batch: 1000
       └─ Right side: Pipeline #0
             Extractor: ArrayExtractor
+               Statistics: rows exact 1 · size unknown
 ```
 
 A joined frame is planned apart, so its pipelines are numbered apart - `Right side:` says which plan they belong to.
+Every source lists what it declares about itself before a row is read - `exact n`, `≤ n` (a guaranteed bound),
+`~n ±e%` (an estimate) or `unknown`. A source that declares nothing lists no `Statistics:` line.
 Reaching this stage plans the frame, so a source that infers its schema by reading is read here; the logical stages
 never read a row.
 
@@ -235,6 +243,7 @@ Before a frame runs, the optimizer rewrites its plan. `Optimizer::default()` run
 | `CombineSortAndLimit`  | `sortBy()` followed by `limit()` keeps only the top rows instead of sorting all  |
 | `PushLimitIntoSource`  | the extractor stops reading once the limit (plus any `offset()`) is reached      |
 | `PushFilterIntoSource` | a `filter()` on partition columns skips whole partition directories              |
+| `CountFromStatistics`  | `count()` over a source that knows its rows exactly reads that number, no rows   |
 
 Rules live in `Flow\ETL\Optimizer\Rule` and are configured through `config_builder()->optimizer()`:
 
@@ -286,6 +295,7 @@ For detailed information about specific DataFrame operations, see the following
 - **[Partitioning](/documentation/components/core/partitioning.md)** - Data partitioning for efficient processing
 - **[Caching](/documentation/components/core/caching.md)** - Performance optimization through caching
 - **[Floe File Format](/documentation/components/core/floe.md)** - Flow's native self-describing binary row format
+- **[Source Statistics](/documentation/components/core/statistics.md)** - What a source declares about its rows and bytes before it is read
 - **[Data Retrieval](/documentation/components/core/data-retrieval.md)** - Methods for getting processed data
 
 ### Data Quality & Validation

documentation/components/core/data-retrieval.md

@@ -99,9 +99,17 @@ $totalCount = $dataFrame->count();
 echo "Total rows: $totalCount\n";
 ```
 
-> **Performance Warning**: The `count()` method must process the entire dataset to return the total count, which can
-> be expensive for large datasets. Consider whether you actually need the exact count or if an approximation would
-> suffice.
+A frame that only reads one source is counted from the source's statistics when they are exact - a Parquet or Floe
+file answers from its footer without reading a row (`explain(Trigger::count)` shows it):
+
+```php
+<?php
+
+$totalCount = data_frame()->read(from_parquet('orders.parquet'))->count();
+```
+
+> **Performance Warning**: Any other frame - a filter, a limit, a new column, a join, an estimated source - is executed
+> in full to count its rows, which can be expensive for large datasets.
 
 ## Iteration with Callback
 
@@ -124,4 +132,34 @@ sinks and returns a `Report` when asked to analyze.
 $dataFrame->write(to_json('out.json'))->run();
 
 $report = $dataFrame->write(to_json('out.json'))->run(analyze: analyze()->withSchema());
+```
+
+`withSourceStatistics()` puts the rows every source declared next to the rows it actually yielded - the output row
+count alone cannot tell them apart once a filter or a join sits in between:
+
+```php
+<?php
+
+$report = data_frame()
+    ->read(from_parquet('orders/*.parquet'))
+    ->filter(ref('total')->greaterThan(lit(100)))
+    ->run(analyze: analyze()->withSourceStatistics());
+
+foreach ($report->sources() as $source) {
+    echo $source->extractor;                // ParquetExtractor
+    echo $source->declared->rows->estimate; // 20000 - extrapolated from the first footer
+    echo $source->rows;                     // 29000 - read
+    echo $source->rowsError();              // 0.31 - |declared - read| / read
+}
+```
+
+`rowsError()` is null without an estimate, and for a read that did not cover the whole source - a pushed `limit()` or
+partition filter, or a read stopped early - since the declaration describes the whole source:
+
+```php
+<?php
+
+foreach ($report->sources() as $source) {
+    $source->complete;                      // false after a pushed limit() or a stopped read
+}
 ```

documentation/components/core/floe.md

@@ -355,11 +355,15 @@ scanning rows**:
   "schema":     { /* the file's single schema */ },
   "sections":   [ { "offset": 6, "partitionsId": 0, "rowCount": 2 } ],
   "partitions": [ { "country": "PL" } ],
-  "totalRows":  2,
+  "statistics": { "rows": 2, "byteSize": 118 },
   "metadata":   { /* typed key/value, Schema\Metadata */ }
 }
 ```
 
+- **`statistics`** holds the file's row count and the uncompressed bytes of its data frames (header and
+  footer excluded), so a reader can size the file before reading a row. The footer and its sections
+  ignore keys they do not know, so a field added later does not break an older reader.
+
 - **`sections`** map a byte `offset` → `partitionsId` + `rowCount`, so a reader can skip whole sections
   (offset/limit pushdown) and know each section's partition combination up front. Sections bound
   partition combinations and appends only; every section shares the file's one schema.

documentation/components/core/statistics.md

@@ -0,0 +1,100 @@
+# Source Statistics
+
+[DOC_LINK:/documentation/components/core/core.md]
+
+[TOC]
+
+Every extractor says what it knows about its data before a row is read - how many rows, how many bytes:
+
+```php
+<?php
+
+use function Flow\ETL\Adapter\Parquet\from_parquet;
+
+$statistics = from_parquet('orders.parquet')->statistics();
+
+$statistics->rows; // Flow\ETL\Cardinality
+$statistics->size; // Flow\ETL\Cardinality, bytes
+```
+
+The answer describes the whole source, with no `limit()` or partition filter applied.
+
+## Cardinality
+
+A `Cardinality` carries a guarantee and a guess side by side:
+
+```php
+<?php
+
+use Flow\ETL\Cardinality;
+
+Cardinality::exact(1_000);                          // atMost 1000, estimate 1000, relativeError 0.0
+Cardinality::atMost(1_000);                         // atMost 1000 - a guaranteed upper bound, no estimate
+Cardinality::approximately(1_000);                  // estimate 1000, relativeError 0.5 (DEFAULT_RELATIVE_ERROR)
+Cardinality::unknown();                             // nothing known
+
+Cardinality::exact(1_000)->exactly();               // 1000 - null unless the count is exact
+Cardinality::approximately(1_000)->confident(0.25); // null - the estimate is looser than 25%
+Cardinality::unknown()->isUnknown();                // true
+```
+
+`atMost` is never exceeded. `estimate` may be wrong in either direction, by roughly `relativeError`.
+
+## What each source declares
+
+| Source                                                                 | rows                                                                                                                      | size                         |
+|------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------|------------------------------|
+| `from_parquet()`, `from_floe()`                                        | exact from the footers; an estimate when a glob's footers were not all read                                               | exact, or estimate on a glob |
+| `from_csv()`, `from_json_lines()`                                      | unknown before the schema is sniffed; exact when the sniff read every file whole; else an estimate from the mean row size | exact listed bytes           |
+| `from_json()`, `from_excel()`                                          | unknown before the schema is sniffed; exact when the sniff read every file whole; else unknown                            | exact listed bytes           |
+| `from_text()`, `from_xml()`                                            | unknown                                                                                                                   | exact listed bytes           |
+| `files()`                                                              | exact listed file count                                                                                                   | exact listed bytes           |
+| `from_path_partitions()`                                               | exact listed file count                                                                                                   | unknown                      |
+| `from_array()`, `from_rows()`, `from_sequence_*()`                     | exact                                                                                                                     | unknown                      |
+| `from_cache()`                                                         | exact from the cache index                                                                                                | unknown                      |
+| `from_pgsql_*()`, `from_dbal_*()`                                      | estimate from `EXPLAIN` (never `ANALYZE`), bounded by `withMaximum()`                                                     | unknown                      |
+| `from_google_sheet()`                                                  | bounded by the sheet's grid size once the schema is sniffed                                                               | unknown                      |
+| `from_all()`, `batches()`, `batched_by()`                              | what the wrapped extractors declare, merged                                                                               | same                         |
+| `from_avro()`, `from_*_http_*()`, `from_memory()`, `from_data_frame()` | unknown                                                                                                                   | unknown                      |
+
+## Where statistics are used
+
+- `explain()->toString(Stage::physical)` lists them under every source: `Statistics: rows exact 1 000 · size exact 327 527 B`.
+- `count()` over a source with exact rows reads the number instead of the rows - see
+  [Data Retrieval](/documentation/components/core/data-retrieval.md).
+- `run(analyze: analyze()->withSourceStatistics())` puts the declared rows next to the rows each source yielded.
+
+## A custom extractor
+
+`statistics()` is part of the `Extractor` contract. A source that knows nothing returns `new Statistics()`:
+
+```php
+<?php
+
+use Flow\ETL\Cardinality;
+use Flow\ETL\Extractor;
+use Flow\ETL\Extractor\Statistics;
+
+final class ApiExtractor implements Extractor
+{
+    private ?Statistics $statistics = null;
+
+    public function __construct(private readonly ApiClient $api)
+    {
+    }
+
+    public function statistics(): Statistics
+    {
+        return $this->statistics ??= new Statistics(
+            rows: Cardinality::approximately($this->api->totalHint(), Cardinality::DEFAULT_RELATIVE_ERROR),
+        );
+    }
+
+    // extract(), schema(), withSchema()
+}
+```
+
+`statistics()` is called at most once per run, only when something needs the answer. Memoise what it reads. It
+may spend only what the run spends anyway - a listing it must produce, metadata of files `schema()` already opens -
+and never open an extra file or make an extra remote call. A database source may ask its planner (`EXPLAIN`), never
+run the query.

documentation/upgrading.md

@@ -238,6 +238,46 @@ after it shift by one. `Stage::physical` is new - see the core documentation.
 | `CSVFileReader::samples()` yields `Generator`s                             | yields `CSVFileSample` (`IteratorAggregate`); `$unit->getIterator()` for the generator         |
 | `CSVFileReader::sample($source)`                                           | `new CSVFileSample($opener, $source)`                                                          |
 
+### 24) `flow-php/etl` - Floe footer carries a statistics block, existing `.floe` files must be rewritten
+
+| Before                                          | After                                                           |
+|-------------------------------------------------|-----------------------------------------------------------------|
+| footer key `totalRows`                          | `statistics.rows`                                               |
+| -                                               | `statistics.byteSize` - uncompressed data bytes                 |
+| footer/section parsers rejected any unknown key | unknown keys ignored, so later additions are not a break        |
+| reading a file written by 0.44.x                | `Floe footer is malformed: ... "statistics"` - rewrite the file |
+
+Header version stays `0x02`: the change is in the footer, not the frame layout. Files written by 0.44.x
+cannot be read. Regenerate them from their source, or export them with 0.44.x to another format before
+upgrading.
+
+### 25) `flow-php/etl` - every `Extractor` declares `statistics()`
+
+| Before                                                  | After                                                                                           |
+|---------------------------------------------------------|-------------------------------------------------------------------------------------------------|
+| `Extractor`: `extract()`, `schema()`, `withSchema()`    | `+ statistics(): Statistics` - a custom extractor that knows nothing returns `new Statistics()` |
+| `SequenceGenerator`: `generate()`                       | `+ rows(): Cardinality` - how many items `generate()` yields                                    |
+| `SelfDescribingFile`: `close()`, `schema()`, `source()` | `+ statistics(): Statistics` - that one file's rows and bytes from its own metadata             |
+| `CacheIndex` rows: `key`                                | `key`, `rows` - an index written by 0.44.x reads fine, its row count is unknown                 |
+| `JsonFileReader::samples()` yields `Generator`s         | yields `JsonFileSample` (`IteratorAggregate`)                                                   |
+| `WorkbookSampler::samples()` yields `Generator`s        | yields `WorkbookSheetSample` (`IteratorAggregate`)                                              |
+
+```php
+final class MyExtractor implements Extractor
+{
+    public function statistics(): Statistics
+    {
+        return new Statistics();
+    }
+}
+```
+
+### 26) `flow-php/etl` - `Report` takes the source statistics of the run
+
+| Before                                           | After                                                                                                |
+|--------------------------------------------------|------------------------------------------------------------------------------------------------------|
+| `new Report(?Schema $schema, Statistics $stats)` | `new Report(?Schema $schema, Statistics $stats, ?array $sources)` - `null` unless analyzed with them |
+
 ---
 
 ## Upgrading from 0.43.x to 0.44.x

src/adapter/etl-adapter-avro/src/Flow/ETL/Adapter/Avro/FlixTech/AvroExtractor.php

@@ -9,6 +9,7 @@
 use Flow\ETL\Extractor;
 use Flow\ETL\Extractor\FileExtractor;
 use Flow\ETL\Extractor\PathFiltering;
+use Flow\ETL\Extractor\Statistics;
 use Flow\ETL\FlowContext;
 use Flow\ETL\Rows;
 use Flow\ETL\Schema;
@@ -65,4 +66,9 @@ public function withSchema(Schema $schema): static
 
         return $this;
     }
+
+    public function statistics(): Statistics
+    {
+        return new Statistics();
+    }
 }