feat: introduction of encoder/hydrator row contract

bin/docs.php

@@ -57,6 +57,7 @@ public function execute(InputInterface $input, OutputInterface $output): int
         $paths = [
             __DIR__ . '/../src/core/etl/src/Flow/ETL/DSL/functions.php',
             __DIR__ . '/../src/core/etl/src/Flow/Floe/DSL/functions.php',
+            __DIR__ . '/../src/core/etl/src/Flow/Serializer/DSL/functions.php',
             __DIR__ . '/../src/adapter/etl-adapter-avro/src/Flow/ETL/Adapter/Avro/functions.php',
             __DIR__ . '/../src/adapter/etl-adapter-chartjs/src/Flow/ETL/Adapter/ChartJS/functions.php',
             __DIR__ . '/../src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/functions.php',

composer.json

@@ -247,6 +247,7 @@
             "src/cli/src/Flow/CLI/DSL/functions.php",
             "src/core/etl/src/Flow/ETL/DSL/functions.php",
             "src/core/etl/src/Flow/Floe/DSL/functions.php",
+            "src/core/etl/src/Flow/Serializer/DSL/functions.php",
             "src/functions.php",
             "src/lib/array-dot/src/Flow/ArrayDot/array_dot.php",
             "src/lib/azure-sdk/src/Flow/Azure/SDK/DSL/functions.php",

documentation/components/core/building-blocks.md

@@ -41,7 +41,7 @@ $rows = array_to_rows([
     ['id' => 2, 'name' => 'user_02', 'active' => false],
     ['id' => 3, 'name' => 'user_03', 'active' => true],
     ['id' => 4, 'name' => 'user_04', 'active' => false],
-], flow_context(config())->entryFactory());
+]);
 ```
 
 ## Entry Types
@@ -56,6 +56,7 @@ $rows = array_to_rows([
 - [Json](/src/core/etl/src/Flow/ETL/Row/Entry/JsonEntry.php)
 - [List](/src/core/etl/src/Flow/ETL/Row/Entry/ListEntry.php)
 - [Map](/src/core/etl/src/Flow/ETL/Row/Entry/MapEntry.php)
+- [Null](/src/core/etl/src/Flow/ETL/Row/Entry/NullEntry.php)
 - [String](/src/core/etl/src/Flow/ETL/Row/Entry/StringEntry.php)
 - [Structure](/src/core/etl/src/Flow/ETL/Row/Entry/StructureEntry.php)
 - [Uuid](/src/core/etl/src/Flow/ETL/Row/Entry/UuidEntry.php)
@@ -64,7 +65,8 @@ $rows = array_to_rows([
 - [XML](/src/core/etl/src/Flow/ETL/Row/Entry/XMLEntry.php)
 - [XMLElement](/src/core/etl/src/Flow/ETL/Row/Entry/XMLElementEntry.php)
 
-Internally, Flow is using [EntryFactory](/src/core/etl/src/Flow/ETL/Row/EntryFactory.php) to create entries.
-It will try to detect and create the most appropriate entry type based on the value.
+Internally, Flow is using a [Hydrator](/src/core/etl/src/Flow/ETL/Row/Hydrator.php) to turn raw values
+into rows and an [EntryFactory](/src/core/etl/src/Flow/ETL/Row/EntryFactory.php) to create single entries.
+Both will try to detect and create the most appropriate entry type based on the value.
 
 Flow Entries are based on [Flow Types Library](/documentation/components/libs/types.md)

documentation/components/core/caching.md

@@ -63,33 +63,40 @@ but it does not come with any out of the box.
 
 ## Serialization
 
-Cache entries are stored as [Floe](/documentation/components/core/floe.md) files, streamed in both
-directions: writes go straight to the cache file batch by batch (the payload string is never
-materialized), reads decode one batch at a time. The batch size (default 1000 rows) bounds how many
-rows cross the engine at once:
+Persisting caches (`FilesystemCache`, `PSRSimpleCache`) turn `Rows` into bytes through a
+[`Serializer`](/src/core/etl/src/Flow/Serializer/Serializer.php). The default is
+[`FloeSerializer`](/src/core/etl/src/Flow/Floe/FloeSerializer.php), which encodes to the
+[Floe](/documentation/components/core/floe.md) binary format.
+
+`Cache::get()` returns one fully materialized `Rows` per key. The caching processor writes exactly
+one batch per cache key, so each key holds a single batch bounded by `DataFrame::cache(cacheBatchSize:)`
+(falling back to `DataFrame::batchSize()`):
 
 ```php
 <?php
 
-use function Flow\ETL\DSL\{config_builder, filesystem_cache};
-
-// through the config builder (default FilesystemCache)
-config_builder()
-    ->cacheSerializerBatchSize(1000);
+use function Flow\ETL\DSL\{data_frame, from_array};
 
-// or constructing the cache directly
-filesystem_cache(serializer_batch_size: 1000);
+data_frame()
+    ->read(from_array($data))
+    ->cache('my-dataset', cacheBatchSize: 1000)
+    ->run();
 ```
 
-`Cache::get()` always returns a fully materialized `Rows` — the raw payload bytes are never held
-next to the decoded rows, but the decoded rows themselves are unbounded. To keep the decoded side
-bounded too, use `Cache::read()`:
+A `Serializer` writes to and reads from streams, so the default `FilesystemCache` streams the payload
+directly to and from the cache file. `PSRSimpleCache` materializes the payload string — its PSR-16
+backend stores string values — bounded by `cacheBatchSize`.
+
+You can swap the serializer per cache; the default `FilesystemCache` shares the same serializer as the
+rest of the pipeline (so it uses the context hydrator):
 
 ```php
 <?php
 
-foreach ($cache->read('my-dataset') as $rows) {
-    // FilesystemCache: one batch of Rows at a time
-    // InMemory/PSRSimpleCache: the whole value, yielded once
-}
-```
+use function Flow\ETL\DSL\filesystem_cache;
+
+filesystem_cache(serializer: new MyCustomSerializer());
+```
+
+`FilesystemCache` stores each entry in a file named after the cache key, with no extension — the
+on-disk bytes are whatever the configured serializer produced.

documentation/components/core/floe.md

@@ -38,6 +38,11 @@ data_frame()
     ->run();
 ```
 
+When the [`flow_php` extension](/documentation/components/extensions/flow-php-ext.md) is loaded, the
+strict read and the write fuse frame-split + value decode/encode + hydrate/dehydrate into one native
+call per batch. It is transparent — the on-disk format is unchanged and the rows are byte-for-byte
+identical to the pure-PHP engine; the salvage path (`FloeStreamReader::recover()`) always stays pure PHP.
+
 ## Save Modes
 
 `to_floe()` honors every [Save Mode](/documentation/components/core/save-mode.md) through the same
@@ -68,9 +73,11 @@ data_frame()
     ->run();
 ```
 
-> Floe additionally supports appending seamlessly into a **single** file with schema evolution
-> through the low-level `Flow\Floe\FloeWriter::append()` API; the DataFrame `Append` save mode uses
-> the sibling-file behavior for consistency with the rest of Flow.
+> Floe additionally supports appending into a **single** file through the low-level
+> `Flow\Floe\FloeWriter::append()` API — appended batches must match the file's schema (a drifted
+> batch throws `IncompatibleSchemaException`; schema evolution is only available through
+> `merge_floe()`). The DataFrame `Append` save mode uses the sibling-file behavior for consistency
+> with the rest of Flow.
 
 ## Partitioning
 
@@ -174,19 +181,21 @@ merge_floe(
 );
 ```
 
-All sources must share one partition combination and evolve the running merged schema cleanly —
-otherwise `IncompatibleSchemaException` is thrown before anything is written. `merge_floe()` works on
-the local filesystem; for other filesystems use `Flow\Floe\FloeMerger` directly.
+Sources must evolve the running merged schema cleanly — otherwise `IncompatibleSchemaException` is
+thrown before anything is written. Each source's per-section partition combinations are preserved
+(deduped into the merged footer table), so sources with differing combinations merge without error.
+`merge_floe()` works on the local filesystem; for other filesystems use `Flow\Floe\FloeMerger` directly.
 
 ## Whole-Value Serialization
 
 The whole-value serialization paths — the [cache](/documentation/components/core/caching.md) and
-`Flow\Floe\FloeSerializer` (the config default serializer) — stream: encode goes through
-`FloeWriter` and decode through `FloeFile::recover()` in batches of `batchSize` rows (default
-1000), so the engine holds one batch at a time. The batch size never changes the produced bytes —
-only the memory bound. A whole-value decode verifies the recovered row count against the footer
-and rejects torn payloads. Numbers and guidance live in the
-[caching documentation](/documentation/components/core/caching.md).
+`Flow\Floe\FloeSerializer` (the config default serializer) — stream: `serialize(Rows, DestinationStream)`
+goes through `FloeStreamWriter` and `unserialize(SourceStream)` through the strict `FloeStreamReader::rows()`
+read in batches of `batchSize` rows (default 1000), so the engine holds one batch at a time. The batch size
+never changes the produced bytes — only the memory bound. String payloads round-trip through the
+`Flow\Serializer\DSL` helpers `serialize_to_string()` / `unserialize_from_string()`. A whole-value
+`unserialize()` verifies the decoded row count against the footer and rejects torn payloads. Numbers and
+guidance live in the [caching documentation](/documentation/components/core/caching.md).
 
 ## On-Disk Layout
 
@@ -199,8 +208,8 @@ reader can locate from the last 8 bytes without scanning the body. All multi-byt
 │ HEADER                              6 bytes              │
 ├──────────────────────────────────────────────────────────┤
 │ FRAMES  (repeated, in write order)                       │
-│    PARTITIONS  (0x03)   once, partitioned files only     │
-│    SCHEMA      (0x01)   emitted when the schema changes  │
+│    PARTITIONS  (0x03)   emitted when combination changes │
+│    SCHEMA      (0x01)   written once, before first ROW   │
 │    ROW         (0x02)   one frame per row                │
 │    ROW         (0x02)                                    │
 │    …                                                     │
@@ -234,19 +243,23 @@ Every frame — `SCHEMA`, `ROW`, `PARTITIONS`, `FOOTER` — shares the same enve
         0x01 SCHEMA   0x02 ROW   0x03 PARTITIONS   0x06 FOOTER
 ```
 
-- **SCHEMA (`0x01`)** — body is the JSON schema for the section that follows. A section's schema is
-  **grow-only**: it starts as the first row's schema and a new `SCHEMA` frame is written only when a row
-  introduces a new column or an incompatible type. Rows narrower than the section ride it (see the ROW
-  absent flag), so a heterogeneous stream needs far fewer sections than one frame per distinct shape.
-- **ROW (`0x02`)** — one row encoded **by column** against the current section schema: for each section
+- **SCHEMA (`0x01`)** — body is the JSON schema for the file. A file carries **exactly one schema**,
+  fixed at session start (explicit, or the first batch's union), written once before the first section.
+  Rows narrower than the schema ride it (see the ROW absent flag). A later batch that introduces a new
+  column or an incompatible type throws `IncompatibleSchemaException` — schema evolution lives only in
+  `merge_floe()`.
+- **ROW (`0x02`)** — one row encoded **by column** against the file schema: for each schema
   column, in order, a one-byte presence flag — `0x01` present (followed by the value encoded per the
   column's schema type, no per-value tag), `0x00` null, `0x02` null-from-null, or `0x03` **absent** (the
   row has no such column). Only dynamically typed values (mixed/union columns, dynamic map keys) carry a
   one-byte type tag (`NULL`, `INTEGER`, `FLOAT`, `BOOLEAN`, `STRING`, `ARRAY`, `DATETIME`, `UUID`,
   `JSON`). A row whose columns exactly match the section produces the same bytes as a plain positional
   encode. One frame per row.
-- **PARTITIONS (`0x03`)** — for a partitioned file, the partition key/value pairs, written once before
-  the first section: a 4-byte count followed by repeated `[nameLen(4), name, valueLen(4), value]`.
+- **PARTITIONS (`0x03`)** — the partition key/value pairs for the section that follows: a 4-byte count
+  followed by repeated `[nameLen(4), name, valueLen(4), value]`. Written at the start of every section
+  whose combination differs from the previous one (mirror of the `SCHEMA`-frame dedup); the reader
+  starts at the empty combination, so an unpartitioned first section emits none and a later change back
+  to unpartitioned emits a `count=0` frame.
 - **FOOTER (`0x06`)** — the footer JSON followed by the trailer (below).
 
 ### Footer
@@ -258,23 +271,24 @@ scanning rows**:
 {
   "version":    1,
   "writer":     "1.x-dev",
-  "schemas":    [ /* schema body per schemaId */ ],
-  "fileSchema": { /* merged schema of every section */ },
-  "sections":   [ { "offset": 6, "schemaId": 0, "rowCount": 2 } ],
-  "partitions": { "country": "PL" },
+  "schema":     { /* the file's single schema */ },
+  "sections":   [ { "offset": 6, "partitionsId": 0, "rowCount": 2 } ],
+  "partitions": [ { "country": "PL" } ],
   "totalRows":  2,
   "metadata":   { /* typed key/value, Schema\Metadata */ }
 }
 ```
 
-- **`sections`** map a byte `offset` → `schemaId` + `rowCount`, so a reader can skip whole sections
-  (offset/limit pushdown) and know each section's schema up front.
-- **`schemas`** is the deduplicated list of every schema written; **`fileSchema`** is their merge —
-  the source schema, available from the footer alone. On a seamless read, a row narrower than
-  `fileSchema` (an absent column, or a column added by a later section) is padded with null entries so
-  every yielded row conforms; `recover()` reads rows exactly as written, absent columns omitted.
-- A new column or incompatible type grows the section (new `SCHEMA` frame); this is also how a single
-  file evolves its schema across appended sections.
+- **`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.
+- **`partitions`** is the deduplicated, order-preserving table of partition combinations, indexed by
+  `partitionsId`; the unpartitioned combination is an empty object at its own id. A file can hold many
+  combinations (one per section).
+- **`schema`** is the file's single schema, available from the footer alone. On a seamless read, a row
+  narrower than it (an absent column) is padded with null entries so every yielded row conforms;
+  `recover()` reads rows exactly as written, absent columns omitted. `merge_floe()` re-encodes drifted
+  sources to their union, so a merged file is still one schema.
 
 ### Trailer (last 8 bytes)
 

documentation/components/core/group-by.md

@@ -41,4 +41,27 @@ However, the result of this operation is not very useful. It will just return a
 
 To make it more useful, you need to use one of the aggregation functions.
 
-[➡️ Aggregations](/documentation/components/core/aggregations.md)
+[➡️ Aggregations](/documentation/components/core/aggregations.md)
+
+## Buckets Storage
+
+`groupBy()` partitions grouping state into buckets through a `BucketsCache` — the same abstraction
+used by [external sort](/documentation/components/core/sort.md) and
+[join](/documentation/components/core/join.md). The default is `FilesystemBucketsCache`: buckets are
+spilled to the local filesystem cache directory (as Floe files), so memory usage is bounded by the
+largest bucket instead of the whole grouped dataset.
+
+```php
+<?php
+
+data_frame(
+    config_builder()
+        ->groupingBucketsCount(64)                  // number of buckets (default 64)
+        ->groupingBatchSize(1000)                   // rows per batch when reading buckets back (default 1000)
+        ->groupingCache(new InMemoryBucketsCache()) // keep buckets in memory instead of on disk
+)
+    ->read(from_parquet('orders.parquet'))
+    ->groupBy(ref('country'))
+    ->aggregate(sum(ref('total')))
+    ->run();
+```

documentation/components/core/join.md

@@ -12,7 +12,7 @@ rows from the left DataFrame.
 
 ### join()
 
-Main join method that loads the right DataFrame into memory as a hash table for efficient lookups.
+Main join method that partitions the right DataFrame into buckets and probes them with left rows through a hash table.
 
 ### crossJoin() - Cartesian Product
 
@@ -33,7 +33,44 @@ Flow PHP supports four join types with specific behaviors:
 | **Right Join** (`Join::right`)         | Returns all rows from right DataFrame with matching rows from left (or NULL)               |
 | **Left Anti Join** (`Join::left_anti`) | Returns rows from left DataFrame that have NO match in right DataFrame                     |
 
-> Flow uses hash join implementation where hashes are stored in sorted buckets to optimize memory usage and performance.
+Joins follow SQL semantics - every matching pair of rows produces one output row, so a left row
+that matches multiple right rows is emitted multiple times.
+
+> Flow uses hash join implementation where hashes are stored in buckets to optimize memory usage and performance.
+> Rows are bucketed by the values of the join columns and every candidate pair is verified against
+> the join expression, so non-equality expressions (like `compare_any()`) are supported as well.
+
+## Buckets Storage
+
+Like group by, `join()` always partitions the right DataFrame into buckets through a `BucketsCache` -
+the same abstraction used by external sort and group by - and the configured implementation decides
+how the join executes:
+
+| Buckets cache                          | Behavior                                                                                                                                                                                        |
+|----------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| `FilesystemBucketsCache` - **Default** | Both sides are partitioned by join key into buckets spilled to disk (Floe files), then joined bucket by bucket. Memory usage is bounded by the largest bucket, left row order is not preserved. |
+| `InMemoryBucketsCache`                 | Right DataFrame is held in memory, left rows are streamed through a hash table and keep their order. Memory usage is bounded by the right side.                                                 |
+
+The implementation is swapped with `config_builder()->joinCache(BucketsCache $cache)`; any cache
+implementing `ResidentBucketsCache` (like `InMemoryBucketsCache`) enables the streaming,
+order-preserving execution:
+
+```php
+<?php
+
+data_frame(
+    config_builder()
+        ->joinCache(new InMemoryBucketsCache()) // right side fits in memory, keep left row order
+        ->joinBucketsCount(64)                  // number of disk buckets
+        ->joinBatchSize(1000)                   // rows per batch when reading buckets back
+)
+    ->read(from_parquet('orders.parquet'))
+    ->join(
+        data_frame()->read(from_parquet('sellers.parquet')),
+        join_on(['seller_id' => 'id'], join_prefix: 'seller_'),
+    )
+    ->run();
+```
 
 ## Example
 

documentation/components/core/partitioning.md

@@ -279,6 +279,10 @@ data_frame()
     ->run();
 ```
 
+Partition values from the path become regular row columns. When the extractor is given a schema that
+does not declare a partition column, the column is appended to the schema as a **string** column —
+declare it explicitly (e.g. `int_schema('date')`) to read partition values as a different type.
+
 ### Partition Pruning
 
 Skip entire partitions without reading their contents using `filterPartitions()`: