This document provides guidelines for upgrading between versions of Flow PHP. Please follow the instructions for your
specific version to ensure a smooth upgrade process.
flow-php/parquet - statistics are written and read as arrow-rs does; one physical schema for one logical schema on
both engines
Before
After
PHP writer string bounds by PHP < ('9' below '10'), whole strings, no exactness flags, no nan_count, no column_orders, deprecated min/max for every type
unsigned-byte order, truncated to Option::STATISTICS_TRUNCATE_LENGTH (64) bytes, exactness flags, nan_count for floats, column_orders per leaf, deprecated min/max only for signed sort orders
statistics of files written by older flow versions read as written
ColumnOrder::Absent: no bounds are trusted, so such files are never pruned; a statistic of the wrong width (older INT64 bounds are 16 bytes) reads as null
StatisticsReader read BYTE_ARRAY statistics with a length prefix, a FLBA decimal as a float, an INT64 decimal as the unscaled int
raw bytes for BYTE_ARRAY/FLBA (decimals as unscaled two's complement bytes, uuid as 16 bytes), the unscaled int for INT32/INT64 decimals
FlatColumn::decimal() of precision up to 18 written as FIXED_LEN_BYTE_ARRAY
INT32 up to precision 9, INT64 up to 18, FIXED_LEN_BYTE_ARRAY above; older files still read
the packer/unpacker encoded FLBA/BYTE_ARRAY decimals and UUIDs; DecimalConverter took INT-backed decimals only
new DecimalConverter($physicalType, $precision, $scale, $byteLength) and UuidConverter convert to the physical bytes; the packer/unpacker take them as-is
FlatColumn::int32() / int64() wrote converted INT_32 / INT_64
plain INT32 / INT64, as arrow-rs
a column with a logical type and no converted type written without one (other readers read a logical UINT_64 as signed)
the converted type derived from the logical type, as arrow-rs; a FlatColumn whose converted and logical types disagree is refused, on read and write
the arrow writer named map entries m.entries.key
m.key_value.key (LogicalTypes.md): COLUMNS_* options keyed by ...key_value.key address the same leaf on both engines
the arrow writer wrote FlatColumn::enum() as STRING
ENUM
Option::ARROW_BATCH_SIZE, read by nothing
removed
Flow\Parquet\Dremel\Statistics\Comparator
removed: Flow\Parquet\Data\PhysicalOrder
new StatisticsCounter($column)
new StatisticsCounter($column, new StatisticsTruncation($length))
new UnsignedIntegerConverter($path, $bitWidth)
new UnsignedInt32Converter(); UINT_64 is UnsignedInt64Converter
UINT_64 reads as a decimal string on both engines (reverses rows 84 and 85 of 0.44→0.45 for UINT_64)
Before
After
UINT_64 read as int, refused above PHP_INT_MAX on both engines
read as the decimal string ('18446744073709551615') on both engines; UINT_8/16/32 stay int
from_parquet() typed a UINT_64 column integer
numeric_string
writes accepted an int from 0 to PHP_INT_MAX into UINT_64
an int >= 0 or a canonical digit string up to 18446744073709551615
to_parquet() refused a numeric_string column (NumericStringType is not supported)
written as a parquet STRING
#4) flow-php/parquet - the arrow extension must be rebuilt (arrow-rs 60, ABI 2)
Before
After
arrow-ext built against arrow-rs 59, FLOW_ARROW_ABI 1
arrow-rs 60, FLOW_ARROW_ABI / ArrowExtension::ABI 2 (covering every contract change below): an older build is refused by ArrowExtension::available()
#5) flow-php/parquet, flow-php/etl-adapter-parquet - reads take a Query (or, below the file, a
ReadPlan) instead of limit/offset, the PHP writer writes the page index, the reader exposes its Footer; the
arrow and flow_php extensions must be rebuilt
ParquetFileReader::footer(): Footer holds them (and rowGroupChunks()); a footer taken before close() stays readable
ParquetFileReader::readColumns(array $columns, int $batchSize, ?int $limit, ?int $offset)
readColumns(array $columns, int $batchSize, ReadPlan $plan), returning the rows $plan->selection selects that match $plan->rowFilter, at most $plan->limit after $plan->offset matches are skipped, one SkippedMatches first when the offset is above 0; build the plan with ReadPlan::forFilter(...) or ReadPlan::window($rows, $file->indexes(), $offset, $limit)
a ParquetFileReader / ParquetFileWriter implementation
also readStoredColumns(array $columns, int $batchSize, ReadPlan $plan): Iterator and writeStoredColumns(array $columns): void: DATE, TIMESTAMP, TIME and UUID values as stored (days, µs since the epoch, µs since midnight, 16 bytes)
the arrow writer refused a value outside the range arrow stores it in with RuntimeException
InvalidArgumentException, as the PHP writer refuses it (e.g. a DateTime outside TIMESTAMP(NANOS))
ParquetOpenSource::batches(Schema $schema, int $batchSize, ?int $offset, ?int $limit, Backend $backend)
batches(Schema $schema, int $batchSize, ReadPlan $plan, Backend $backend)
RustBatchReader / RustColumnsReader::__construct(..., int $batchSize, ?int $offset, ?int $limit)
readRowGroup(Column, RowGroup, int $group, SourceStream, RowSelection, Indexes, array $decoded = []), one row group at a time; $decoded pages, keyed by their first row, are served instead of read
new ColumnReader(ColumnChunkReader $chunks, DremelAssembler $assembler)
new ColumnReader(new SelectedPages($chunks), $assembler)
ColumnChunkReader::read(ColumnChunk, FlatColumn, SourceStream) yielded a list of pages
pages keyed by their first row in the row group; ?OffsetIndex $offsetIndex with list<int> $pages reads only those pages of it, ?Dictionary $dictionary replaces the dictionary page, the generator returns the dictionary
new PlainFlatColumnChunkBuilder($column, $options, $compression) (and the RLE_DICTIONARY, DELTA_BINARY_PACKED builders, ColumnChunkBuilderFactory::createBuilder())
a 4th argument, StatisticsOptions::fromOptions($options)
ColumnChunk::dataPageOffset(): ?int
int: a column chunk always records it
the PHP writer wrote no column index or offset index
both, by default (Option::STATISTICS_LEVEL, StatisticsLevel::PAGE; CHUNK drops the column index, NONE all statistics)
the PHP writer wrote page-header statistics on DATA_PAGE_V2
none, as arrow-rs: Option::PAGE_HEADER_STATISTICS (true) writes them, on v1 and v2 pages
Option::PAGE_MAXIMUM_ROWS_COUNT (default 1000) was read by nothing
the PHP writer closes a page at exactly that many rows; the arrow writer treats it as a target (arrow-rs data_page_row_count_limit, checked between write batches), so its pages can hold more; below 1 is refused (InvalidArgumentException)
the arrow writer skipped a COLUMNS_COMPRESSIONS / COLUMNS_ENCODINGS entry whose column path is digits only ('2024', an int array key)
applies it; a key that is not UTF-8 is refused (InvalidArgumentException) instead of crashing the process
Option::ROW_GROUP_SIZE_BYTES below 1 was taken by both writers
refused (InvalidArgumentException) by both writers
the arrow writer threw Error (a Rust panic) on PLAIN_DICTIONARY / RLE_DICTIONARY in COLUMNS_ENCODINGS
enables the column's dictionary: RLE_DICTIONARY data pages and a PLAIN dictionary page, as the PHP writer writes
the arrow writer skipped a COLUMNS_COMPRESSIONS / COLUMNS_ENCODINGS value that is not a string
refused (InvalidArgumentException)
files with an RLE_DICTIONARY column spanning more than one page, written by the PHP writer
were corrupt (later pages decoded to wrong values): rewrite them; one dictionary per chunk, PLAIN past Option::DICTIONARY_PAGE_SIZE
Option::DICTIONARY_PAGE_MIN_CARDINALITY_RATION
removed: it was never read; the dictionary falls back to PLAIN on Option::DICTIONARY_PAGE_SIZE only, as arrow-rs
Options::set() / get() took scalars and arrays
also a UnitEnum; Options::getEnum(Option $option, string $class)
no column index truncation option
Option::COLUMN_INDEX_TRUNCATE_LENGTH (64): BYTE_ARRAY/FIXED_LEN_BYTE_ARRAY page bounds in the column index
DataPageHeaderV2::statistics(Options $options)
statistics(), as DataPageHeader::statistics()
ParquetFile::pageHeaders() repeated every chunk of a row group under each column
each column's own chunk pages, once
FLOW_PHP_ABI / FlowPhpExtension::ABI 1
2 (covering every contract change below): rebuild the flow_php extension (and arrow, item 4); an older build is refused by available()
#6) flow-php/etl, flow-php/types - string comparisons in filters use byte order, as sort already does
Before
After
ref('a')->lessThan(lit('10')) on '9': true (PHP < compares numeric strings as numbers)
false: lessThan, lessThanEqual, greaterThan, greaterThanEqual, between, greatest and least order two strings by bytes (strcmp), as sortBy() does
greatest(lit('9'), lit('10')): '10'
'9'
a numeric_string column was comparable only with another numeric_string
comparable with a string literal too (ref('id')->greaterThan(lit('1000'))), and ordered numerically; still not comparable with an integer
$extractor->extract($backend, new Criteria(pathFilter: $filter))
Read::limit(), Read::pathFilter(), new Read($extractor, $limit, $pathFilter)
Read::criteria()->limit, ->pathFilter, ->conjuncts; new Read($extractor, new Criteria($limit, $pathFilter))
Read::withPathFilter() on any extractor
refused (InvalidArgumentException) on an extractor that is no FileExtractor, as withFilter()
Executor\Pipeline::limit(), pathFilter(), new Pipeline(..., $limit, $pathFilter, $sources)
new Pipeline($id, Plan\Node\Read|Pipeline|null $input, PipelineSteps $steps); the read is the Read node, its criteria $pipeline->input->criteria(); the counters are an Executor::execute() argument (row 95)
a filter() over a file source stayed a Filter step
each deterministic part the source uses is pushed into it (Criteria::$conjuncts) and leaves the step: a part over partition columns only, on every source, and a part FileExtractor::prunes() accepts (Parquet: a part it translates into a Parquet filter); an emptied step is removed; a limit() after it is pushed beside it, so the read stops after that many matching rows; explain() prints each pushed part as Filter: ... under Read; a part after one kept in the step is pushed too, so the kept parts, in written order, can run on fewer rows; a >/>= on a float column is pushed only into files whose statistics prove no NaN
a custom FileExtractor
implements prunes(ScalarFunction $conjunct): bool: true only when the source skips data with the conjunct before it decodes rows; false keeps the conjunct in its Filter step
new FileReadLoop($fileColumns, $schema), read($sources, $batches, $batchSize, Backend $backend, $window)
new FileReadLoop($fileColumns, $schema, $criteria->conjuncts), read($sources, $batches, $batchSize, Backend $backend, $window)
Optimizer::default() ran PushLimitIntoSource before PushFilterIntoSource
after it
Rewrite::of(Node $node)
of(Node $original, Node $rebuilt): the node as it was and rebuilt over its rewritten children
new Flow\ETL\Plan\ReplaceLeaf($target, $replacement)
ReplaceNodes::one($target, $replacement), or new ReplaceNodes($replacements, $removals) for several in one rewrite (a removed node gives way to its child)
new ScalarFunctionFilterTransformer($function, ?ScalarFunction $resolved)
the second argument is ?Functions\Evaluation\Filter; bind() sets it, as before
explain(): Select printed Columns: ..., other nodes Defines columns: ...
Keeps only: ... and Changes columns: ... (a column defined, renamed or dropped)
Redefined::$unknown
Redefined::isUnknown(); Redefined::isAllBut() tells an allBut() apart
A custom file extractor:
// before
public function extract(FlowContext $context, ?int $limit = null, Filter $pathFilter = new OnlyFiles()): Generator
{
// reads with $limit and $pathFilter
}
// after
public function extract(Backend $backend, Criteria $criteria): Generator
{
// reads with $criteria->limit and $criteria->pathFilter, returns only rows matching every one of
// $criteria->conjuncts
}
public function prunes(ScalarFunction $conjunct): bool
{
return false; // true only when the source skips data with $conjunct
}
flow-php/symfony-telemetry-bundle - HTTP URLs are redacted, stack traces carry no argument values
Before
After
url.full (PSR-18 and Symfony HttpClient), url.query (HttpKernel), exception.message and the client span status description held the raw URL
userinfo and the values of UrlSanitizer::DEFAULT_SENSITIVE_QUERY_PARAMETERS (OpenTelemetry, OAuth 2.0 / OpenID Connect, API key and credential keys) are REDACTED; configure the list with url_sanitizer([...]) or flow_telemetry.instrumentation.sanitization.url.sensitive_query_parameters (replaces the default)
Symfony HttpClient url.full, url.scheme, server.address, server.port came from the $url argument (/me, unknown with base_uri)
they come from the effective URL, base_uri and the query option included
exception.stacktrace was getTraceAsString(), with argument values when zend.exception_ignore_args is off
StackTrace::format(): every frame renders (), whatever zend.exception_ignore_args is
flow-php/parquet - the PHP engine reads and writes DELTA_LENGTH_BYTE_ARRAY / DELTA_BYTE_ARRAY; both engines write the
same default encodings (dictionary on, arrow-rs fallbacks); COLUMNS_ENCODINGS is literal on both engines
Before
After
WRITER_VERSION 2 fallback encoding of BYTE_ARRAY / FIXED_LEN_BYTE_ARRAY: PLAIN
DELTA_BYTE_ARRAY, as arrow-rs and parquet-java
dictionary overflow past DICTIONARY_PAGE_SIZE: the rest of the chunk PLAIN
the WRITER_VERSION fallback: under v2 DELTA_BINARY_PACKED (INT32/INT64), DELTA_BYTE_ARRAY (BYTE_ARRAY/FLBA); PLAIN under v1
arrow writer: a non-dictionary COLUMNS_ENCODINGS entry kept the column's dictionary, used only once it fell back
turns the column's dictionary off: the entry is the column's encoding, as the PHP writer writes it
DeltaBinaryPackedDecoder::decode(string $data, int $valueCount)
decode(BinaryBufferReader $reader, int $valueCount), leaving the reader at the end of its stream
new PlainFlatColumnChunkBuilder($column, $options, $compression, $statistics); DeltaBinaryPackedColumnChunkBuilder
new FlatColumnChunkBuilder($column, $options, $compression, $statistics, ValueStorage $valueStorage); DeltaBinaryPackedColumnChunkBuilder removed
ValueStorage without an encoding
ValueStorage::encoding(): Encodings; ValueStorage::size() renamed to plainSize(), the PLAIN-encoded byte size
new RLEDictionaryChunkBuilder($column, $options, $compression, $statistics)
new RLEDictionaryChunkBuilder($column, $options, $compression, $statistics, ValueStorage $fallback)
a failed Writer::write() / Writer::open() left a 4-byte PAR1 file behind
the file is removed; the exception is rethrown
PHP writer: columns without COLUMNS_ENCODINGS written without a dictionary
RLE_DICTIONARY (except BOOLEAN, and FIXED_LEN_BYTE_ARRAY under v1), as the arrow writer, arrow-rs and parquet-java; Option::DICTIONARY_ENABLEDfalse restores the fallback encodings
-
Option::DICTIONARY_ENABLED (default true), on both engines
PHP writer WRITER_VERSION 2: BOOLEAN written PLAIN
RLE, as the arrow writer
RLE_DICTIONARY on a FIXED_LEN_BYTE_ARRAY column threw
allowed under WRITER_VERSION 2; still refused under 1
PHP writer fell back once the dictionary exceeded DICTIONARY_PAGE_SIZE
once it reaches it (>=), as arrow-rs
PHP writer v2 chunks listed only the data page encoding
also RLE for the levels, as arrow-rs
PHP writer DELTA_BINARY_PACKED: a miniblock whose largest delta sits at or just above a power of two from about 2^47, or needs more than 55 bits, was packed wrong and read back wrong values
packed exactly; rewrite WRITER_VERSION 2 files written by earlier versions whose INT32/INT64 columns hold such deltas
COLUMNS_ENCODINGSRLE_DICTIONARY on a BOOLEAN column: the PHP writer wrote a dictionary, the arrow writer ignored it
refused by both engines (InvalidArgumentException), as arrow-rs and parquet-java never dictionary-encode booleans
arrow writer: a COLUMNS_ENCODINGS entry the column type cannot take panicked (\Error: Rust panic) or was ignored (FIXED_LEN_BYTE_ARRAY dictionary under v1)
InvalidArgumentException with the PHP writer's message
INT32 DELTA_BINARY_PACKED with consecutive values more than 2^31 apart: written with 64-bit deltas arrow-rs refuses, read back as wrong values
32-bit wrapping deltas, as arrow-rs and parquet-java; such INT32 files from earlier versions must be rewritten
flow-php/etl-adapter-google-sheet - SchemaSampler + SniffsColumnTypes merged into Sniffer
Schema inference samples a source through one contract. Each adapter builds its Sniffer inside schema(); a unit is
one
listed file (one sheet range, one in-memory dataset). A bounded sampleSize still drains one budget across the units in
listing order; everyRow() sniffs every unit whole and independently and merges the partials in unit order, so the
result does not depend on the order the units ran in.
new SchemaInferrer($inference, $typer)->infer($names, $samples)
new SchemaInferrer($inference)->infer($sniffer): Inferred (->schema, ->units) - the narrower is the sniffer's; everyRow() sniffs through infer_schema()->sniffWith($launcher), in this process by default
removed: Flow\ETL\IO\Inference\SampledUnits (the row estimate over every sampled file) and Flow\ETL\IO\Inference\Sample (one file's sniff: columns, bytes, complete)
flow-php/etl-adapter-parquet and every file adapter - file sources say where they may be cut
Additive. A SplittableExtractor cuts its source into FileSplits at plan time by arithmetic alone - one split per
file for JSON documents, text, XML, Excel and listings; byte windows for CSV, JSON lines, Parquet and Floe - and reads
one through scan() with the split on its Criteria. Splits cover every row exactly once, in source order. A CSV
window resyncs both ends to a record start, so a quoted newline never splits a record; a JSON lines window resyncs both
ends to a line start; a Parquet or Floe window owns the row groups or sections whose midpoint it holds. Through a
Launcher with more than one slot, set with config_builder()->launcher(), the splits are read
in worker processes. One launcher serves the whole run, joined frames included.
Criteria(..., ?FileSplit $split = null), withSplit(FileSplit); narrows() is true with a split; a read with a split lists nothing but the split's file
every file extractor read whole files only
SplittableExtractor extends Extractor { splits(Criteria, int $targetSplitBytes): non-empty-list<Split>; reader(): Reader } on files(), from_csv(), from_json(), from_json_lines(), from_text(), from_xml(), from_excel(), from_parquet(), from_floe(); each split is read with extract($backend, $criteria->withSplit($split)), or through the config's launcher
each gains ?array $headers = null - the names of a stream that carries no header record (a window past a file's first record); withHeader must be false with them
#14) flow-php/filesystem-async-aws-bridge, -azure-bridge, -sftp-bridge - filesystems take a client source
Before
After
new AsyncAWSS3Filesystem($mount, $bucket, S3Client $s3Client, $options)
new AsyncAWSS3Filesystem($mount, $bucket, S3ClientSource $client, $options) - new LiveS3Client($s3Client) or new ConfiguredS3Client($configuration)
new AzureBlobFilesystem($mount, BlobServiceInterface $blobService, $options)
new AzureBlobFilesystem($mount, BlobServiceSource $blobService, $options) - new LiveBlobService(...) or ConfiguredBlobService::fromArray([...])
new SFTPFilesystem($mount, SFTP $sftp, $options)
new SFTPFilesystem($mount, SFTPClientSource $client, $options) - new LiveSFTPClient($sftp) or ConfiguredSFTPClient::fromArray([...])
async-aws Options::withBlockSize() refused 5 MiB
5 MiB, S3's inclusive minimum part size, is accepted
aws_s3_filesystem(), azure_filesystem() and sftp_filesystem() keep their parameter names and also accept a plain
configuration array; only a filesystem built from configuration can be described for a worker process. The Symfony
filesystem bundle builds the configured form unless an HTTP client, a factory or a logger is injected.
#15) flow-php/filesystem, flow-php/parquet - every filesystem and parquet engine is configurable
Before
After
interface Filesystem (10 methods)
+ normalize(): ?array, + static fromArray(array $configuration): static - the plain configuration a worker rebuilds the filesystem from; null / throw when the instance holds live state (a client, memory). Core wraps it in Flow\ETL\Foundation\Describe\ConfiguredFilesystem
interface ParquetEngine (3 methods)
the same two methods (fromArray(): self); PhpParquetEngine and AdaptiveParquetEngine normalize {byte_order, options}, RustParquetEngine{options}; wrapped by Flow\ETL\Adapter\Parquet\IO\ConfiguredParquetEngine
Path
+ normalize(): PathData, + static fromArray(array); NativeLocalFileBlocksFactory::blockLocation(); OptionalType::assert() is documented as returning null |T
+ close(string $bucketId): void - finishes the bucket's pending writes so another process can read it; a no-op without any. FilesystemBuckets implements Describable
#17) flow-php/etl - the default batch size is 2048 rows
Before
After
batchSize() is 100 for every extractor using Batches
2048 - the per-batch cost dominated the read at 100 rows (a parallel read handed rows back in 100-row Floe frames; 10 GiB CSV: 51 s to 39 s single-process, 28 s to 9 s at 8 workers). withBatchSize() is unchanged; PostgreSql*/Dbal* extractors keep 1000. FilesystemBuckets defaults to 2048 as well, so a handed-back batch is never re-sliced
#18) flow-php/etl and every adapter - functions, transformers, loaders and error handlers describe themselves
each extends Describable: implement describe(): ?Description and static fromDescription(array $data, Rebuild $rebuild): static; return null / throw when the instance holds live state - it then always runs in the parent process
#19) flow-php/etl, flow-php/etl-adapter-* - FileLoader copies itself to another destination;
removed: FileLoader takes partitionBy(Partitioning $partitioning): static and partitioning(): Partitioning
{part} in a destination named a partition column part
{part} is the part number: to_parquet('out/orders-{part}.parquet') writes out/orders-00000.parquet in one process and one part per task when a launcher splits the read
partitionBy('part') with a '.../{part}/...' destination
throws InvalidArgumentException when the plan is built: rename the column
SaveMode on a {part} destination checked per file
checked over every existing part (orders-*.parquet): ExceptionIfExists/Ignore see any part, Overwrite removes every stale part, Append names carry a run id
#20) flow-php/filesystem - NativeLocalFilesystem::mv() creates the target's parent directory
Before
After
mv() into a missing directory fails (rename() error)
the missing parent directories are created, then rename()
#21) flow-php/etl - aggregating and window functions describe themselves and expose a partial state
extends FunctionTree, Describable; + partialType(): StructureType, + partial(): array, + mergePartial(array $partial, FlowContext $context): void - every custom aggregator implements them (a group by read by workers merges partials)
each extends Describable; return null / throw when the instance holds live state - it then runs in the parent process
Aggregators::cloned()
Aggregators::fresh() - withChildren(children()) per aggregator, never clone: a custom aggregator's withChildren() must return an instance without state
#25) flow-php/etl - count() without a reference copies itself fresh
Before
After
Count::withChildren([]) on count() returns the same instance
returns new Count(null, $window) without state, like every other aggregator's withChildren()
#26) flow-php/etl - Flow\Floe\Statistics renamed to FooterStatistics
Before
After
Flow\Floe\Statistics
Flow\ETL\Storage\Floe\FooterStatistics
The footer struct (rows, byteSize) is no longer a third Statistics beside Flow\ETL\IO\Extractor\Statistics and
Flow\ETL\Runtime\Metrics\Statistics. The Floe bytes are unchanged.
#27) flow-php/filesystem, flow-php/etl - RangeSourceStream exposes its window
Before
After
RangeSourceStream reads only through iterate()/read()
stream(), start(), end() (exclusive) name the inner stream and the window it is cut from
The native CSV reader opens a NativeLocalSourceStream (or a RangeSourceStream over one) by its path and reads the
window itself; every other stream is still read through its PHP iterate(). NativeLocalFilesystem::readFrom() no
longer creates the file's parent directory, and NativeLocalSourceStream::size() is the open handle's size.
#28) flow-php/etl - a Floe BATCH frame is decoded by the Backend
Before
After
(new FrameDecoder($codec))->decode($body, $schema, $backend)
$backend->decodeFrame($body, $schema, $codec); Flow\Floe\FrameDecoder is removed
also add, subtract, multiply, modulo, each (Column $left, Column $right): Column
-
also sort(Rows $rows, list<SortColumn> $columns): Rows, rows equal on every column kept in input order
-
also groupedTotals(): GroupedTotals: running sums and counts per group id, as NumericTotal keeps them per group
-
also groupedCounts(): GroupedCounts: non-null value counts per group id, as count(ref) keeps them
-
also groupTable(list<string> $columns, Schema $declared): GroupTable: dense group ids by key
PhpBackend::decodeFrame() is the former FrameDecoder walk. RustBackend::decodeFrame() reads the frame directory
natively and points the Arrow buffers into the body string instead of copying every buffer through decode().
PhpBackend answers comparisons, null tests, arithmetic and sorts as Comparison, IsNull, IsNotNull, Plus,
Minus, Multiply, Mod and the sort processors did; RustBackend answers the ones it can natively and the rest
through PhpBackend. Divide and the exact mode of Plus, Minus and Multiply stay in their functions. A custom
backend can delegate and adopt the result:
public function compare(Column $left, Operator $operator, Column $right, string $leftName, string $rightName): Column
{
return $this->adopt(
Definitions::of('value', type_boolean(), true),
(new PhpBackend())->compare($left, $operator, $right, $leftName, $rightName),
);
}
#29) flow-php/etl - a pipeline ends at a processor that needs every row; Segment, Segments and
Materialization removed
Before
After
Processor implementors: bind, process
also breaks(): bool - true when process() needs every input row before it emits one; such a processor ends its pipeline
a pipeline ended after a blocking node
it ends at a processor that breaks(): a loader after collect() runs in the next pipeline, a window's WindowProcessor opens the pipeline after its repartition
final readonly class PhysicalPlan(Pipeline $root, ?Schema $schema, ?SchemaNotDerivableException $refusal = null, bool $countsOnly = false); the config's Launcher::read(PhysicalPlan $plan, ...): ?Launch cuts the launch at run time
HashJoinProcessor, GroupByAggregationProcessor, PivotProcessor, DuplicateRowTransformer: final class, bind() set fields on a copy
final readonly class; the bound state is a trailing optional constructor argument (?Schema $left, ?Schema $declaredRight; ?GroupByShape $shape; ?PivotShape $shape; $resolved, ?Schema $output)
new DuplicateRowTransformer($condition, WithColumn ...$columns)
new DuplicateRowTransformer($condition, array $columns)
#39) flow-php/etl - join rules live on the join type; one join schema, one probe loop
| Flow\ETL\Extractor\BatchSize | Flow\ETL\BatchSize |
| new Transformation\BatchSize(int $size) | new Transformation\BatchSize(BatchSize $size); batch_size(int $size) unchanged |
| int $batchSize in new Joiner(), new ExternalSort(), new KWayMerge(), new PivotAggregation(), new ExpandingProcessor(), new SpillingBuckets(), PivotedTable::flush() | BatchSize $batchSize |
| new BatchingProcessor(int $size), new Plan\Node\Batch($input, int $size), int $size property | BatchSize $size |
| these threw InvalidArgumentExceptionBatch size must be greater than 0, given: 0 | new BatchSize(0) throws Batch size must be greater than 0, got 0 |
| TopNProcessor::followedBy($kept, $rest), (new BoundedRead($limit, $backend))->followedBy($read, $rest) | BoundedRead::followedBy($read, $rest) |
| DerivedColumns::declare($schema, $definition) | $schema->put($definition) |
| a transformer or loader opened its own telemetry span | the pipeline opens one span per step, user transformers and loaders included |
| FileWriteFrame::write($rows, $context, $loader) | FileWrite::write(Rows $rows, Backend $backend) |
| - | Flow\ETL\Foundation\Telemetry\SpanAttributes::spanAttributes(): extra attributes for a step's span |
| OnEach::typeOver(Type $arrayType) | OnEach::typeOver(Type $arrayType, ScalarFunction $body) |
| Hashes::add(), Hashes::exists() | removed: Hashes::keep() |
| GroupBy\PivotSchema | removed: GroupBy::pivotSchema(Schema $input, array $pivotColumns, AggregatingFunction $aggregation) |
| new ScalarFunctionTransformer($column, $function, $resolved, $derived, $output) | a sixth argument, ?Schema $declared |
| GroupBy::references() | GroupBy::refs()->all() |
#42) flow-php/etl, adapters - file extractor and reader trims
Before
After
declaredPartitionTypes() on the CSV, JSON, Text, XML, Excel, Parquet and Floe extractors, FileColumnsSource::partitionTypes()
removed
new AdaptiveCSVOpenSource($filesystem, $source, $options, $extension)
(new CSVSourceOpener($filesystem, $options, extension: $extension))->open($source)
ParquetSchemaPassFile
removed: DerivedFromFiles::fold() leaves the first file open for its owner
hash('xxh128', $string), or (new NativePHPHash())->hash($string)
#45) flow-php/etl - foundation and memory accounting moved to their layer namespaces
Flow\Calculator, Flow\Clock and Flow\ArrayComparison move under Flow\ETL\Foundation, memory accounting
under Flow\ETL\Values\Memory. Class names do not change.
#46) flow-php/etl - value types moved to the Flow\ETL\Values namespace
Rows, schemas, definitions, columns and their physical layouts move under Flow\ETL\Values; the Floe frame codec
contract moves to Flow\ETL\Values\Column\Frame. Class names do not change.
#47) flow-php/etl - columns are addressed by name in Schema, Definition and Rows::reduceToArray()
Schema methods take column names (string), no longer Reference. Definition::reference() becomes
Definition::name(), Schema::references() becomes Schema::names().
Before
After
$definition->reference()->name()
$definition->name()
$schema->references()
$schema->names() - list<string>
$schema->get(ref('id')), also findDefinition, keep, remove, reorder
$schema->get('id')
$schema->moveAfter(ref('a'), ref('b')), also addAfter, moveTo, rename
$schema->moveAfter('a', 'b')
$rows->reduceToArray(ref('id'))
$rows->reduceToArray('id')
#48) flow-php/etl - Rows::concat() is static, Rows::sortBy() is removed
Before
After
$rows->concat($backend, $other)
Rows::concat($schema, $backend, $rows, $other)
$rows->sortBy(ref('id')->desc())
$rows->gather($order->permutation($order->keys($rows), $rows->count())) with $order = new RowOrder([ref('id')->desc()])
Rows::concat() takes the schema every part must carry, so an empty list of parts still has one.
#49) flow-php/etl - RowsBuilder::append() and RowsBuilder::appendProjected() are removed
Before
After
$builder->append($row)
$builder->appendRows([$row])
$builder->appendProjected($rows)
$builder->appendRows($rows) - records keyed by position fill the column at that position
LayoutFor, PhysicalKind and PhysicalBuilderFor merge into PhysicalFor.
Before
After
(new PhysicalFor())->type($type)
(new PhysicalFor())->physical($type)
(new LayoutFor())->type($type)
(new PhysicalFor())->layout($type)
(new PhysicalKind())->accepts($type, $v)
(new PhysicalFor())->accepts($type, $v)
(new PhysicalKind())->all($type, $values)
(new PhysicalFor())->acceptsAll($type, $values)
(new PhysicalKind())->elements($type, $v)
(new PhysicalFor())->acceptsElements($type, $v)
(new PhysicalBuilderFor())->type($type)
(new PhysicalFor())->builder($type)
#51) flow-php/etl - definition merging goes through Definitions
Definitions::of() maps a type to its definition (what definition_from_type() does, which now delegates to it),
Definitions::merged() widens two definitions of one column. CommonType and ValueMatch are removed.
Before
After
CommonType
removed - Definitions::merged($left, $right)
ValueMatch
removed - $definition->matches($value)
#52) flow-php/etl - schema errors past an inference sample report the row in the source
InferredRows is removed: memory extractors and the native CSV / JSON readers report a refused row at its 0-based
position in the whole source, not in the batch.
Before
After
InferredRows
removed - the extractors' RecordBatch keeps the source offset
#54) flow-php/etl - storage types moved to the Flow\ETL\Storage namespace
Batch storage, the Floe file encoding, the serializers and Memory move under Flow\ETL\Storage;
TelemetryAttributes moves to Flow\ETL\Foundation\Telemetry. The bucket storages are renamed after the one
contract they implement (row 55). The Floe source and sink (FloeExtractor, FloeLoader) move to
Flow\ETL\IO\Floe (row 112); the Flow\Floe\DSL and Flow\Serializer\DSL functions keep their namespace.
#55) flow-php/etl - one RowsStorage contract for the cache, spilling and shuffles
Cache and BucketsStorage merge into Flow\ETL\Storage\RowsStorage: a key holds batches appended in order.
cache(), from_cache(), config_builder()->cache() and filesystem_cache() keep their names and take or build a
RowsStorage. The default cache writes one Floe file per key with the configured backend, so the configured
Serializer no longer reaches it. Entries cached by 0.46 are not read: they are written again on the next run.
The filesystem cache moves from <dir>/<xx>/<xx>/<xx>/<xx>/<key> to <dir>/flow-php-rows/<xxh128 of the key>/<key>.floe.
Remove the old entries under <dir> by hand.
#61) flow-php/etl - = and join keys of two types compare in one type
When the plan is built, comparisons and join keys of different types are cast to one type. Numbers
widen to float, a date widens to a datetime, and a string casts to the other side's type. An integral string casts
exactly, other numeric strings round half away from zero for an integer side. A value that cannot be cast now throws
instead of not matching. === and !== still compare without a cast.
Before
After
ref('id')->equals(lit('x')) on an int column: false
throws EvaluationException: Column "id" value 'x' cannot be cast to integer
join of an int key and a string key 'x': no match
throws the same exception
join keys keyed loosely (KeyValues)
join keys cast to one type, then keyed exactly (KeyEncoder)
#62) flow-php/etl - NaN is the largest value, uuid, json, xml, enum and html sort by their stored form
NaN equals NaN and orders after every number. This holds in =, >, < and between. It also holds
in sortBy(), groupBy(), batchBy() and window peers.
Before
After
f > 2.0 drops a NaN row
keeps it
f = NAN matches nothing
matches NaN
batchBy('f') puts two NaN rows in two batches
one batch
sortBy() on uuid, json, xml, enum keeps input order
sorts by bytes: uuid by its 16 bytes, json and xml by their text, enum by case name
rank() over a uuid order: input order
the uuid's byte order
#63) flow-php/etl - one set of seen keys for distinct rows, unique constraints and statistics
SeenKeys holds every key a stream brought. It counts the distinct values in statistics and in the CLI
file:analyze output ("Distinct Values"). The counts now tell types apart and keep datetime microseconds. They count
-0.0 as 0.0, and they count every column type, json included.
#64) flow-php/etl - one fold for aggregate partials
A custom AggregatingFunction drops merge(AggregatingFunction $other, FlowContext $context), and mergePartial()
is renamed merge(). Group by merges a spilled aggregator through its partial().
// before
public function mergePartial(array $partial, FlowContext $context): void { /* fold $partial */ }
public function merge(AggregatingFunction $other, FlowContext $context): void
{
$this->mergePartial($other->partial(), $context);
}
// after
public function merge(array $partial, FlowContext $context): void { /* fold $partial */ }
#65) flow-php/etl - kernels take column names and values
Kernels take column names, SortColumn and Backend, never references or the flow context. Hashing runs once per
distinct key.
extract(Backend $backend, Criteria $criteria), the limit is $criteria->limit
Loader::load(Rows $rows, FlowContext $context)
load(Rows $rows, Backend $backend)
Loader\Closure::closure(FlowContext $context)
closure()
Loader\Discardable::discard(FlowContext $context)
discard()
$extractor->extract(flow_context())
$extractor->extract(flow_context()->backend(), new Criteria())
A custom extractor and loader:
// before
public function extract(FlowContext $context, ?int $limit = null): Generator
{
yield rows(...); // stops after $limit rows
}
public function load(Rows $rows, FlowContext $context): void
{
$this->write($rows->toArray());
}
// after
public function extract(Backend $backend, Criteria $criteria): Generator
{
yield rows(...); // stops after $criteria->limit rows
}
public function load(Rows $rows, Backend $backend): void
{
$this->write($rows->toArray());
}
new SerializeTransformer($target, $serializer, $standalone)
new UnserializeTransformer($source, $target, $merge, $mergePrefix)
new UnserializeTransformer($source, $target, $serializer, $merge, $mergePrefix)
new CachingProcessor(?string $id = null, ?RowsStorage $cache = null)
new CachingProcessor(string $id, RowsStorage $cache)
explain() of a cache step always prints the storage in use as Cache: <storage>.
A transformer that read the config at run time takes what it needs in its constructor.
Before:
final class MyTransformer implements Transformer
{
public function transform(Rows $rows, FlowContext $context): Rows
{
$serializer = $context->config->serializer();
// ...
}
}
After:
final class MyTransformer implements Transformer
{
public function __construct(private Serializer $serializer)
{
}
public function transform(Rows $rows, Backend $backend): Rows
{
// ...
}
}
#91) flow-php/etl - operator types moved to Flow\ETL\Operators, their configs to Storage, Kernels and
ConstrainedProcessor refuses a non-constraint with Pipeline constraints must be of type Flow\ETL\Operators\Constraint.
#92) flow-php/etl - the *Steps factories are removed
Before
After
Flow\ETL\GroupBy\GroupBySteps::of()
removed: the planner builds [PruneColumnsTransformer, GroupByAggregationProcessor] (the prune only when GroupBy::readColumns(): ?list<string> is not null) or [PivotProcessor]
Flow\ETL\Join\JoinSteps::of()
removed: the planner builds a JoinStep over a HashJoin (row 94)
Flow\ETL\Repartition\RepartitionSteps::of()
removed: new RepartitionProcessor(References $by, BucketingConfig $config, RandomValueGenerator $random)
new RepartitionProcessor($by, HashBucketing $bucketing, Buckets $buckets, Unit $memoryLimit, Hasher $hasher)
new RepartitionProcessor($by, BucketingConfig $config, RandomValueGenerator $random): it builds its bucketing and its spilling buckets from $config; $memoryLimit is $config->memoryLimit
Flow\ETL\Sort\SortSteps::of()
removed: new MemorySortProcessor($columns) or new MergeSortProcessor($config, $columns, $random)
#94) flow-php/etl - the join operators are fed by the runtime
Before
After
Flow\ETL\Processor\HashJoinProcessor (a Processor)
Flow\ETL\Operators\Join\HashJoin, not a Processor: join(Generator $left, Generator $right, Backend $backend): Generator, bind(Schema $left, Schema $right): BoundStep
new HashJoinProcessor(PhysicalPlan $right, Executor $executor, $on, $type, $config, $random, $backend)
new HashJoin($on, $type, $config, $random); the right plan is not the operator's
Flow\ETL\Transformer\CrossJoinRowsTransformer (a Transformer)
Flow\ETL\Operators\Join\CrossJoin, not a Transformer: join(Generator $left, Generator $right, Backend $backend): Generator, bind(Schema $left, Schema $right): BoundStep
new CrossJoinRowsTransformer(PhysicalPlan $right, Executor $executor, string $prefix)
new CrossJoin(string $prefix)
a join step holds its right plan
the plan holds Flow\ETL\Planning\Physical\JoinStep(HashJoin|CrossJoin $join, PhysicalPlan $right); the runtime runs the right plan once and hands its rows to join()
crossJoin(): a failure in it is offered to the error handler (onTransformation())
a failure in it propagates like any processor's, a failure of the right side included; the error handler is not asked
crossJoin(): one transformation span per batch
no span, as for every processor and for join()
BoundStep::$step is Processor|Transformer
Processor|Transformer|HashJoin|CrossJoin, typed by @template: Processor::bind() returns BoundStep<Processor>, Transformer::bind()BoundStep<Transformer>
explain() prints Processor: HashJoinProcessor, Transformer: CrossJoinRowsTransformer
Join: HashJoin, Join: CrossJoin
#95) flow-php/etl - the physical plan holds plan data, the runtime holds the run
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)
explain() prints Loader: SinkFeed, Loader: TransactionalSinks for a sink with steps of its own, a transaction
// before
public function apply(LogicalPlan $plan, FlowContext $context): LogicalPlan
// after
public function apply(LogicalPlan $plan, Backend $backend): LogicalPlan
#104) flow-php/etl - TelemetryContext takes the data frame id and name, one step trio for every step kind
Before
After
new TelemetryContext($backend, $logger, $tracer, $meter, $options)
new TelemetryContext(string $dataFrameId, string $dataFrameName, $backend, $logger, $tracer, $meter, $options)
dataFrameStarted(FlowContext $context)
dataFrameStarted(array $configuration), the map Config::telemetryAttributes() returns
#114) flow-php/etl - the user's side path is a Branch, a transaction group is not a branch
"Sink" is gone: Sink is Branch, the plan's write roots are Writes. Transactional no longer implements Branch
and has no write(): pass it to DataFrame::write(), to_branch() or to_transformation(). A Transactional
directly inside a Transactional is a TypeError. The named argument of DataFrame::write(), to_branch(),
to_transformation() is $write, of to_dbal_transaction() and to_pgsql_transaction() it is $writes.
#115) flow-php/etl - a file loader writes through a FileWriter
One format written to one open stream is a FileWriter, made by a FileWriterFactory. The factory takes the schema of
the stream's first batch, so a writer opens its file at once. ParquetFileWriter conforms and writes on the PHP lane and
RustParquetFileWriter does the same on the native lane. Both implement FileWriter, so ParquetOpenSink and
PhpParquetOpenSink are gone. A batch without rows opens nothing, for every file loader (to_floe() wrote an empty file
before).
Every file write runs through one FileCommit. Files that replace others (Overwrite) and every parallel task's files
are written under a temporary output inside the destination, then moved into place; the other modes write in place.
The save mode decides once for the whole destination. Readers skip the temporary output of a running or crashed write
because it sits under a . or _ path segment.
Before
After
FilesSink
FileCommit: a task stages its files, the job commits them; a serial write is one task
Overwrite wrote ._flow_php_tmp.<file> beside each file
every Overwrite and every parallel task writes under <destination dir>/.flow-tmp-<run>-<hash>/ first; an aborted parallel write leaves the destination directory behind, empty
Append beside an existing plain file: <name>_<random>.<ext>
<name>-<run>.<ext>
Ignore / ExceptionIfExists on a partitioned plain destination: decided per partition file
decided once over <dir>/**/<name>.<ext>: an existing file in any partition skips (Ignore) or refuses (ExceptionIfExists) the whole write
Overwrite removed <name>*.<ext> in each written partition (data_backup.csv too)
removes only <name>.<ext> and <name>-<run>.<ext>, partitioned or not; files an older Append named <name>_<random>.<ext> stay
Overwrite removed the stale files, then moved the new ones
moves first, then removes: a failed commit leaves duplicates, never loses the old files
a write without rows touched nothing
Overwrite empties an unpartitioned destination (a partitioned one keeps every partition); ExceptionIfExists over an existing destination throws; Ignore and Append write nothing
to_excel()->saveMode(ignore()) over an existing workbook overwrote it
keeps it
file extractors and files() list every file the path matches
skip a file under a . or _ path segment below the path's static part, except segments with =, _metadata and _common_metadata; a path you name yourself is read
#118) flow-php/etl - a limit ends the stream without an exception
Before
After
Flow\ETL\Operators\LimitReachedException thrown by transform() to end the stream
removed: a Limiting transformer returns the trimmed batch and reached() returns true
-
Flow\ETL\Operators\Transformer\Limiting extends Transformer { reached(): bool }, implemented by LimitTransformer and UntilTransformer
Limit reached log attribute limit (0 for until())
attribute step: the class of the transformer that reached it
they may also yield a CorruptedRecordException, one per record that cannot become a row; a direct caller routes or refuses it (instanceof Rows)
a reader refused a corrupted record itself
the run routes it to withCorruptedRecords($loader), loaded, closed and discarded like the run's other loaders; without a loader it fails the run, whatever the ErrorHandler - a corrupted record is not an extraction error
ExtractionError::$extractor was an Extractor
Extractor|Reader: a task's source is a Reader
StepRunner::steps(), run(): ?Extractor $source
Extractor|Reader|null $source; a source implementing CorruptedRecordsRouter routes its corrupted records
#127) flow-php/etl, flow-php/etl-adapter-csv, -json, -excel, -google-sheet - a record that cannot become a row throws
Before
After
a CSV line or a sheet row with fewer cells than the header was padded with nulls
CorruptedRecordException; ->withMissingColumnsAsNull(true) pads it (CSV, Google Sheet), Excel pads by default (->withMissingColumnsAsNull(false) refuses)
a CSV line or an Excel row with more cells than the header lost them
CorruptedRecordException for a non-empty extra cell; an empty one is dropped
Google Sheet ->withDropExtraColumns(bool)
removed: a non-empty extra cell is a corrupted record; ->withMissingColumnsAsNull(bool) pads a short row
an empty CSV line was a row of nulls
skipped
a value that does not cast to the type ->withSchema() declares threw SchemaMismatchException
CorruptedRecordException (route it with ->withCorruptedRecords($loader))
a value or a JSON key a file shows only after the inference sample: SchemaMismatchException, or the key was dropped
InferredSchemaException::outsideTheSample(), naming the file, offset, line and sample size; from_array() and sequences throw InferredSchemaException
a ->withSchema() column missing from a CSV, Excel or Google Sheet header was null on every row
SchemaValidationException
new RecordDecoder($withHeader, $emptyToNull, $dropExtraColumns, $headers), decode(list $batch): list
new RecordDecoder($withHeader, $emptyToNull, $missingColumnsAsNull, $headers), decode(array $cells): null|array|string (null: header or empty record, string: why it is corrupted)
CSVDecoder::decode(list<string> $lines)
CSVDecoder::decode(CSVDecoder::cells(string $line)), one record
+ ?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()
#129) flow-php/etl - DataFrame is immutable, every verb returns a new frame
Before
After
$df->limit(10); limits $df
does nothing, keep the result: $df = $df->limit(10);
$df->onError($handler); sets the handler on $df
$df = $df->onError($handler);
$df->fetch(10), display(), printRows() limit $df
leave $df unchanged
Branch::write(DataFrame $prefix): void
Branch::write(DataFrame $prefix): DataFrame
new DataFrame(Extractor $extractor, Config|FlowContext $context)
df($config)->read($extractor)
new DataFrame(Extractor $extractor, FlowContext $context)
new DataFrame(new \Flow\ETL\Planning\Logical\Node\Read($extractor), new \Flow\ETL\Planning\Logical\Writes(), $context)
PHP 8.5 warns about a dropped result (#[\NoDiscard]); PHP 8.3 and 8.4 do not.
Before:
public function write(DataFrame $prefix): void
{
$prefix->filter($this->condition)->write($this->loader);
}
public function transform(DataFrame $dataFrame): DataFrame
{
foreach ($this->columns as $column) {
$dataFrame->withColumn($column, lit('***'));
}
return $dataFrame;
}
After:
public function write(DataFrame $prefix): DataFrame
{
return $prefix->filter($this->condition)->write($this->loader);
}
public function transform(DataFrame $dataFrame): DataFrame
{
foreach ($this->columns as $column) {
$dataFrame = $dataFrame->withColumn($column, lit('***'));
}
return $dataFrame;
}
$qb->setMaxResults(5) - silently overwritten by the page size
InvalidArgumentExceptionKeyset pagination sets its own LIMIT and OFFSET, ...
$qb->setFirstResult(5) - applied to every page
same InvalidArgumentException
Remove them from the builder; use withMaximum() or DataFrame::limit().
#6) flow-php/postgresql, flow-php/etl-adapter-postgresql - a query's own ORDER BY must equal the keys
Before
After
'SELECT ... ORDER BY name' with key id - kept, pages wrong
PaginationExceptionKeyset pagination requires ORDER BY item #1 to be the key "id ASC" ...
ORDER BY id NULLS FIRST, ORDER BY 1, ORDER BY lower(name), ORDER BY id USING <
same PaginationException
Drop the ORDER BY (it is generated from the keys) or make it list exactly the keys, in order and direction.
#7) flow-php/postgresql - KeysetPaginationConfig / sql_to_keyset_query() reject a NULL cursor value
Before
After
new KeysetPaginationConfig(10, $columns, [null, 7]) - predicate that matches nothing
PaginationExceptionKeyset cursor value #1 is NULL; key columns must be non-null
Page by non-null key columns.
#8) flow-php/postgresql - same-direction composite keys emit a row comparison
Before
After
created_at > $1 OR (created_at = $1 AND id > $2)
(created_at, id) > ($1, $2)
all DESC: created_at < $1 OR (created_at = $1 AND id < $2)
(created_at, id) < ($1, $2)
Nothing to change; the row comparison is an index range bound. Mixed ASC/DESC keys keep the OR form. Update assertions
on the generated SQL.
#9) flow-php/etl-adapter-doctrine - a non-scalar keyset value is rejected
Before
After
a key value that is not string/int/float/bool - bound as a parameter
RuntimeExceptionUnsupported value type "<type>" in column "<key>" for keyset pagination
Page by scalar columns.
#10) flow-php/etl-adapter-doctrine - pagination_key_set() rejects zero keys
Before
After
pagination_key_set() built; from_dbal_key_set_qb() threw KeySet must contain at least one key for pagination
pagination_key_set() throws InvalidArgumentExceptionKeySet requires at least one key
Pass at least one key.
#11) flow-php/filesystem, flow-php/etl - datetime and float partition values keep their full value
Before
After
partition_by('date') over datetime2026-10-02 13:00:00 UTC - date=2026-10-02
date=2026-10-02T13%3A00%3A00%2B00%3A00
reading it back as datetime (writeColumns() or partition_types(date: type_datetime())) - 2026-10-02 00:00:00
2026-10-02 13:00:00
partition_by('x') over float1.0E-9 - x=0.000000, read back 0.0
x=1.0e-9, read back 1.0E-9
from_path_partitions() / files() / an undeclared partition column - '2026-10-02'
'2026-10-02T13:00:00+00:00'
Directories written before keep reading as midnight. Writing into a dataset that has them adds a second directory per
day, and overwrite() replaces only the new one - rewrite the dataset, or partition by a date column to keep
date=2026-10-02:
// Before - now throws "transform() cannot contain array_expand() ..."
$df->transform(new ScalarFunctionTransformer('item', array_expand(ref('items'))));
// After
$df->withEntry('item', array_expand(ref('items')));
#17) flow-php/etl - a serialized Rows holds a list of frames
A Rows serialized by 0.45 (e.g. a FilesystemCache entry) is not read by 0.46: clear such caches after upgrading. A
batch
over 2 GiB of strings in one column is now split across frames instead of throwing OffsetOverflow.
Flow\ETL\Pipeline (class, with Pipeline::has()), Pipeline\Optimizer, Pipeline\Optimizer\Optimization, Pipeline\Optimizer\LimitOptimization, Pipeline\BoundPlan, Pipeline\PlanBinder
removed - an Optimizer\Rule rewrites the Plan\LogicalPlan: apply(LogicalPlan $plan, FlowContext $context): LogicalPlan; drop one with Optimizer::default()->without(Rule::class), add one with ->with(new MyRule())
new Pipeline\Optimizer(Optimization ...), ->disabled(), ->optimizations()
new Optimizer(Rule ...), new Optimizer() (no rules - nothing is rewritten), ->rules(); Optimizer::default() holds the built-in rules, ->without(Rule::class) / ->with(new MyRule()) change them
-
Flow\ETL\Executor\PhysicalPlan interface (root(): Executor\Pipeline, schema(): Schema): Executor\Described returns its schema, Executor\Raw throws its SchemaNotDerivableException
$df->read(files($glob))->filterPartitions(...), same over from_path_partitions($glob)
filter() on the partition column - both sources now emit one string column per key=value directory (next to partitions), and the filter is pushed into the listing
a partition-predicate error was thrown by the verb
thrown at plan / bind time - an optimizer rule can fail a plan
->write($sink)->filter(...) narrowed the earlier sink
a filter is pushed only when every root reaches the source through it - the sink gets every row
#10) flow-php/etl - frames are snapshotted when embedded, fewer limits are pushed
Before
After
$left->join($right, ...); $right->select('id'); - the select() was part of the join's right side
ignored - a frame is snapshotted at join() / from_data_frame() time
$frame->join(df()->read(from_data_frame($frame)), ...) threw InvalidLogicException "Cannot describe this plan:"
runs as a self-join against the snapshot; a run-time cycle still throws cyclicPlanOnRun()
->rows($t)->limit(n), ->transform($t)->limit(n), ->void()->limit(n) pushed the limit into the source for seven allow-listed transformers
not pushed - ->withEntry(...)->limit(n) still pushes; with several sinks the widest limit is pushed, a limit(3) inside a sink limits the whole run
-
schema() followed by a terminal verb plans twice, Extractor::schema() is called once per planning - memoise a sniffing extractor
add_row_index() on a frame run twice continued counting ([0,1,2] then [3,4,5])
starts again on every run; a Transformer keeping state between batches implements Flow\ETL\Transformer\Stateful (fresh()) to do the same
a joined frame ran with its own config and onError()
runs with the outer frame's config, optimizer and error handler; its own onError() is ignored - a from_data_frame() frame still runs with its own
each joined frame opened its own DataFrame span
one DataFrame span per run; a from_data_frame() frame still opens its own
#11) flow-php/etl - one balanced DataFrame telemetry span per run
Before
After
the DataFrame span started when the frame was built
starts when the plan executes - Executor::execute() owns both ends
a planning failure under schema() emitted no span
dataFrameStarted + dataFrameFailed for every verb
an abandoned get*() generator left its span open
closes it
a failure inside a verb's own loop body (a forEach callback, the formatter) closed the span as failed
closed as completed
a from_data_frame() frame emitted no span
one balanced span per run
to_dbal_transaction() / to_pgsql_transaction() emitted their own span
no span
#12) flow-php/etl - to_branch() / to_transformation() return a Sink, the wrapper loaders are removed
Before
After
to_branch($condition, $loader, $transformation) returned BranchingLoader
a Transformation calling $df->fetch() / count() / schema() / run() inside to_transformation() or to_branch() threw InvalidLogicException
executes the prefix plan, as from_data_frame() executes a frame
onError() inside a sink's Transformation
throws InvalidLogicException at write() - set it on the frame
an unresolved column or a non-boolean condition in a sink failed at the first batch
fails at plan time; one undescribable sink operation makes the whole plan run raw; schema() still describes the frame's own rows
a transformer failing inside a non-transactional sink was a LOADING error
a TRANSFORMATION error offered to onTransformation() once: SkipRows / IgnoreError drop the batch (after a blocking operation, the whole buffered batch); a limit() completing the sink mid-load is a LOADING error on Executor\SinkFeed
write_with_retries($loader) around to_transformation(...) or to_branch(...) threw at the first load()
removed with the retry surface, see 14)
telemetry flow.etl.loading.rows counted the rows offered to the branch
counts the rows the branch's loader writes; an empty batch never reaches the loader
SkipRows: a drain failure of to_transformation() / to_branch() at closure() was rethrown
offered to onTransformation() once - SkipRows drops the buffered batch
new Transactional(DbalTransaction::fromConnection($connection)->withIsolationLevel($level), ...$sinks) - returns a new instance; the same on PostgreSqlTransaction, which runs SET TRANSACTION after BEGIN and rolls back a failed SET
a failure a non-throwing onError() handler suppressed still committed the batch
the batch (or the closure drain) rolls back and is never re-delivered; the handler only decides whether the run continues - the failing sink restarts on the next batch, its buffered rows lost
LoadingError::$loader for a begin() / commit() failure was TransactionalDbalLoader
Executor\TransactionalSinks; for a sink failure, the sink's own loader when bare, Executor\SinkFeed otherwise; the exception is always the cause, never TransactionRolledBack
withIsolationLevel() on to_dbal_transaction() / to_pgsql_transaction() applied to every transaction the wrapper opens
set on DbalTransaction / PostgreSqlTransaction; applies to every transaction opened
#14) flow-php/etl - write_with_retries() and the retry surface are removed
#15) flow-php/etl-adapter-doctrine - a DBAL read the driver cannot describe keeps the database error
Before
After
a query the driver cannot describe (multi-statement, data-modifying CTE, INSERT ... RETURNING) threw SchemaNotDerivableException
a missing table or column throws what the read throws (TableNotFoundException, InvalidFieldNameException, on SQLite DriverException); any other query the driver refuses to describe throws SchemaNotDerivableException with the DBAL exception as getPrevious()
#16) flow-php/etl-adapter-postgresql - a PostgreSQL read keeps the database error, reads that write are refused
Before
After
a query that cannot run as a subquery (multi-statement, data-modifying CTE, INSERT ... RETURNING) threw SchemaNotDerivableException
a missing table, column, function, type or privilege throws PostgreSQL's QueryException, as the read does; a query that writes (a data-modifying WITH, SELECT ... INTO) throws InvalidArgumentException before any query runs
#17) flow-php/flow-php-ext - the flow_php extension is versioned with flow-php/etl
Before
After
flow_php extension 0.3.0, required by this flow-php/etl
same version as this flow-php/etl release
#18) flow-php/etl - DataFrame has no @internal methods
Before
After
DataFrame::extractor() (@internal)
removed - (new Repeatability())->ofPlan($dataFrame->explain()->logical) answers whether every source the frame reads can be read twice
new Plan\Node\Outputs(Result $result, Sinks $sinks)
new Plan\Node\Outputs(Result|Write|Transaction ...$consumers) - two or more
InvalidLogicException::resultRewritten()
consumerRewritten(); a Transaction as the first consumer is refused
#21) flow-php/etl - explain layouts render a described entry, a joined frame is drawn without its Result
Plan\Explain\Outline now applies Details while it builds, so a layout prints what the entry carries instead of
asking for it. A joined frame's Result is no longer drawn: the join reads its right side directly, and the numbers
after it shift by one. Stage::physical is new - see the core documentation.
Before
After
new Explain\Entry($node, $number, $shared, $children)
new Explain\Entry($source, $name, $lines, $number, $shared, $children, $suffix)
$entry->node
$entry->source
$entry->title($name)
$entry->title() - the name is the entry's own
new Explain\TreeLayout($details, declarations: true)
new Explain\Outline(declarations: true), the layout takes no arguments
new Explain\BoxLayout($details)
new Explain\BoxLayout()
#22) flow-php/etl-adapter-csv - an escaped or bare enclosure no longer merges a record with the next line
Input (3 lines)
Before
After
a,b / "x\"y",1 / "p",2
1 row: {"a":"x\\\"y","b":"1\n\"p\""}
2 rows: {"a":"x\\\"y","b":1}, {"a":"p","b":2}
a,b / x"y,1 / "p",2
1 row: {"a":"x\"y","b":"1\n\"p\""}
2 rows: {"a":"x\"y","b":1}, {"a":"p","b":2}
#23) flow-php/etl-adapter-csv - CSVOpenSource is an interface, CSVSourceOpener removed,
CSVLineReader takes the separator and escape
Before
After
new CSVLineReader($enclosure, $charactersReadInLine, $removeBOM)
new CSVLineReader($enclosure, $separator, $escape, $charactersReadInLine, $removeBOM)
final class CSVOpenSource, new CSVOpenSource($stream, $dialect, $encoder, $lineReader)
interface CSVOpenSource, implemented by PhpCSVOpenSource (new PhpCSVOpenSource($stream, $decoder, $lineReader), a CSVDecoder), RustCSVOpenSource (flow_php) and AdaptiveCSVOpenSource
(new CSVSourceOpener($filesystem, $options))->open($source)
new AdaptiveCSVOpenSource($filesystem, $source, $options)
new CSVFileReader($opener, $sources)
new CSVFileReader($filesystem, $options, $sources)
CSVFileReader::samples() yields Generators
yields CSVFileSample (IteratorAggregate); $unit->getIterator() for the generator
CSVFileReader::sample($source)
new CSVFileSample($filesystem, $options, $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 Generators
yields JsonFileSample (IteratorAggregate)
WorkbookSampler::samples() yields Generators
yields WorkbookSheetSample (IteratorAggregate)
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
#27) flow-php/postgresql - Traverser visits every node
Before
After
sql_query_tables('CREATE TABLE x AS SELECT * FROM t') - []
[t, x]
sql_query_tables('CREATE VIEW v AS SELECT a FROM src') - []
[v, src]
sql_query_tables('SELECT a INTO new_t FROM src') - [src]
[new_t, src]
sql_query_columns('... ON CONFLICT (name) DO UPDATE SET name = excluded.name') - []
[excluded.name]
OrderBy of SELECT a, row_number() OVER (ORDER BY b) FROM t ORDER BY a - 1 clause
2 clauses - window ORDER BY included
sql_query_tables('SELECT (SELECT x FROM a) FROM b') - [b, a]
[a, b] - descriptor (PostgreSQL walker) order
sql_query_depth(): EXPLAIN SELECT 1 0, CREATE VIEW v AS SELECT 1 0, window subquery 1
1, 1, 2
sql_to_keyset_query() cursor on ... WHERE $1 IN (SELECT ...) - $1, clashing with the user's
$2
TypeCastStripper left casts under a SubLink test, a window and COLLATE
new UnionDefinition('a', type_union(type_integer(), type_string()))
str_schema('a'), or json_schema('a') for dynamic data
Removed with them: UnionMembers, UnionTypeNormalizer, Row\EntryTypeResolver, TypeProjection::union().
TypeProjection now takes the column's Reference; TypeFloor no longer takes a TypeProjection.
#37) flow-php/etl - a union inside a list, map or structure column type
InvalidArgumentException - an IANA region name, UTC or +HH:MM
to_timezone($v, lit('UTC'))
TypeError - pass 'UTC' or a DateTimeZone
to_timezone($ref, ref('tz')) - a per-row zone
TypeError - the zone of to_timezone() and new ToTimeZone() is DateTimeZone|string
now($tz), to_date_time(..., $tz), to_timezone(..., $tz) return a datetime (UTC) column
a datetime column in $tz; a ScalarFunction zone in now() / to_date_time() yields UTC; now(new DateTimeZone('EST')), to_date_time(..., new DateTimeZone('PST')) throw InvalidArgumentException (abbreviation zones)
to_date_time($dateTimeObject) truncates to midnight
keeps the time, converted to the zone
merging datetime columns ignores zones
equal zones kept, different zones become UTC
merging time with date / datetime gives datetime
string
?T inside a list, map or structure passes a valid value through cast() unchanged
cast like T: a ?datetime moves to its zone, a ?structure keeps only its declared keys, in declared order
A value with a key its structure does not declare is invalid. Declare every element, or use type_map() /
type_json() when the keys are dynamic. flow-php 0.44.x cannot read a Floe file, cached schema or schema_to_json()
document written by 0.45.x that has a structure column. Clear APCu and natively serialized caches after upgrading.
#49) flow-php/etl - Rows is a columnar batch, Row is a view
Row::get() of a datetime / date / time / uuid / json / xml / html cell - the instance handed in
a new object per call - compare with assertEquals, not assertSame
a DOMElement / HTMLElement cell keeps its document (parent and siblings reachable)
the cell is the element's own markup - navigate from the document column (xpath())
an optional and nullable structure element given null
reads back absent (the element's validity bit)
$row == $other - compares the rows' values
compares both views' whole batches and indices - compare toArray()
serialize(Rows) payload {schema, rows}
{schema, frame} - the Floe BATCH body; payloads of 0.44.x cannot be unserialized, a corrupt payload throws FloeException
Rows::drop() / dropRight() / take() / takeRight() with a negative size - array_slice() semantics
InvalidArgumentException
#50) flow-php/etl - row(), InferredBatch and DataFrame::getEach() removed
Before
After
row($values)
array_to_row($values, $schema)
rows($schema, row($a), row($b))
array_to_rows([$a, $b], $schema)
Flow\ETL\Row\InferredBatch
removed - array_to_rows($data, $schema), or from_array($data) to infer one
DataFrame::getEach()
getEachAsArray() (values) / get() (batches with their schema)
array_to_row() / array_to_rows() cast values to the declared type and refuse keys the schema does not declare
(SchemaMismatchException naming the row and the column; before, those keys were silently dropped).
#51) flow-php/etl - date and time columns store days and microseconds
Before
After
a date cell - the value as given
midnight UTC (zone UTC) of its calendar day
a time cell - d kept
d folded into hours, $days is false
a time string with months / years ('P1M') - accepted by the hydrate path
SchemaMismatchException "could not convert 'P1M' (string) to time"
a datetime / time beyond ±9 223 372 036 854.775807 s (about year ±292 000) - stored
refused
a date beyond the i32 day range (about year ±5 880 000) - stored
refused
a PHP 8.4 \Dom\Element in an xml_element column - kept
read back as a DOMElement
#52) flow-php/etl - batch columns refuse wildcard enums, mixed elements and foreign definitions
#56) flow-php/etl-adapter-json - untyped JSON Schema arrays and objects become json
Before
After
{"type": "array"} without items → list<mixed>
json
{"type": "object"} without properties / additionalProperties → map<string, mixed>
json
schema_to_json_schema(schema_from_json_schema(...)) of an untyped array / object
{"type": ["object", "array"]}
#57) flow-php/etl - a join name collision suggests a prefix
Before
After
SchemaDefinitionNotUniqueException: Entry definitions must be unique, duplicated entries: [id], all: [id, id]
the same message with . Please consider using join prefix option appended - the text before it is unchanged
Every join shape (joinCross(), join(), joinEach()) reports the collision through JoinSchema::of(); the original
exception is the new one's previous. Only an exact match on the whole message is affected.
SchemaMismatchException "could not convert 2026-01-01T00:00:00.000000+01:00 (datetime) to date"
type_date()->cast($warsawMidnight)
unchanged - the value's own calendar day
A date column stores days and reads back midnight UTC (row 51), so a midnight in another zone was silently moved to a
different instant; it now stays a datetime unless it is cast to a date explicitly.
#59) flow-php/types - Comparator takes no arguments
Before
After
new Comparator(StructureComparison $structures)
new Comparator()
#60) flow-php/etl - the row-shaped reader/writer API is removed
Project with from_parquet(..., columns: [...]). Change a type after reading, e.g. a JSON column another tool wrote as
string: ->withEntry('payload', ref('payload')->cast(type_json())). Re-zone a datetime after reading:
->withEntry('at', to_timezone(ref('at'), 'Europe/Warsaw')).
plus the configured backend's allocatedBytes() (native columns are not bound by memory_limit)
new Consumption(bool $realMemory = true)
new Consumption(Backend, bool $realMemory = true)
new TelemetryContext(Logger, Tracer, Meter, TelemetryOptions)
new TelemetryContext(Backend, Logger, Tracer, Meter, TelemetryOptions)
new KWayMerge(References, int $batchSize = 1000)
new KWayMerge(References, Backend, int $batchSize = 1000)
new Joiner(Expression, Join, int $batchSize = 1000)
new Joiner(Expression, Join, Backend, int $batchSize = 1000)
AggregatedGroups::flush(int $batchSize)
flush(int $batchSize, FlowContext)
BucketingConfigBuilder::build(Path), SortAlgorithmBuilder, GroupByAlgorithmBuilder, JoinAlgorithmBuilder, RepartitionAlgorithmBuilder and their implementations build(Path)
build(Path, Backend) - custom algorithm builders implement the new signature
new FilesystemBuckets(Filesystem, Path, int $batchSize = 1000)
new FilesystemBuckets(Filesystem, Path, Backend, int $batchSize = 1000)
every limit defaults to Config\MemoryLimit::default(): FLOW_MAX_MEMORY, otherwise 70% of memory_limit, 1 GiB when memory_limit is -1
-
a limit is compared with the whole process (memory_get_usage() + the backend's native memory), not with the step's own buffer - a sort and a join in one pipeline stay under it together
hash_join() with a disk storage partitioned both sides into buckets
the right side stays in memory under the limit and the left side streams in its own order; past it both sides are partitioned
groupBy() / window partitions always hash-partitioned their input into buckets
under the limit a group-by aggregates in one pass and a window groups its partitions in memory - no bucket is written; past it the rest of the stream is partitioned (group-by merges the groups aggregated before the limit)
group-by / repartition / join wrote one bucket frame per bucket per input batch
buckets stay in memory under the limit; past it every frame holds batchSize rows
flow_php and arrow-ext loaded and no engine given, Parquet is read and written as native columns
Before
After
from_parquet($path) / to_parquet($path) - AdaptiveParquetEngine (arrow-ext when loaded, else the PHP engine)
with flow_php and arrow-ext loaded: arrow-ext's batches straight into native columns and back (Arrow C Data Interface); flow_php alone: the PHP engine, its values adopted into native columns; without flow_php: unchanged
engine: new PhpParquetEngine() / new ArrowParquetEngine()
engine: new PhpParquetEngine() opts out; new AdaptiveParquetEngine() reads and writes native columns when both extensions are loaded, new RustParquetEngine() reads them
a column type arrow-ext could not read failed inside arrow-ext
refused before the first batch: Flow\Parquet\Exception\RuntimeExceptionParquet column "<name>" (<type>) is not supported by the arrow Parquet reader; read the file with \Flow\Parquet\Reader::php()
flow-php/etl-adapter-parquet installable with any flow_php or arrow-ext version
conflict: ext-flow_php <0.45, ext-arrow <0.45 - load extensions of this release or none
#80) flow-php/etl-adapter-parquet - Flow float columns are written as Parquet DOUBLE
Before
After
float_schema('amount') written as Parquet FLOAT (32-bit): 0.14 read back as 0.14000000059604645
written as DOUBLE (64-bit), read back exactly; readers of the file see a DOUBLE column
flow_php and arrow-ext, a string that is not valid UTF-8 is refused on write
Before
After
without arrow-ext, PhpParquetEngine wrote "\xff\xfe" into a STRING column other readers reject (DuckDB: is not valid UTF8)
Flow\Parquet\Exception\RuntimeExceptionParquet column "name" row 0 holds a string that is not valid UTF-8; Parquet STRING columns require UTF-8 - the row is 0-based within the batch written
with arrow-ext: Column 'name': expected string, got string
the same refusal; invalid UTF-8 into STRING / JSON / ENUM is refused by both engines; Option::VALIDATE_DATA = false on PhpParquetEngine writes it unvalidated (a file other readers reject); arrow-ext always refuses
#82) flow-php/parquet - DELTA_BINARY_PACKED follows the spec; affected files written by
PhpParquetEngine must be rewritten
Before
After
PhpParquetEngine wrote DELTA_BINARY_PACKED blocks that other readers misread (a partial last block without all miniblock bit widths, zero-width miniblocks with 4 padding bytes)
spec-conformant blocks; DuckDB, arrow-rs and flow_php read the values back
it read other writers' DELTA_BINARY_PACKEDINT64 wrong after an outlier (9223372036854645807 read as 9223372036788279061, garbage after)
reads them as DuckDB and arrow-rs do
affected: INT32/INT64 columns PhpParquetEngine wrote with Option::WRITER_VERSION2 or COLUMNS_ENCODINGSDELTA_BINARY_PACKED (the default is version 1, PLAIN)
no spec-conformant reader returns their values: read them with the Flow version that wrote them and write them again (or re-export from the source)
UINT_64 above PHP_INT_MAX read as a numeric string
refused, as by PhpParquetEngine
writes accepted an int into FLOAT / DOUBLE / DATE / TIMESTAMP / TIME, a numeric string into DECIMAL / UINT_64
refused - writes accept what PhpParquetEngine accepts; decimals are written from floats (or ints) as PhpParquetEngine writes them; an object with toString() into UUID is accepted
TIME writes dropped the interval's d
kept
an offset read decoded every row group before the offset; page headers were read to the end of the file
only the row groups a read needs, each column chunk read once
an INTERVAL, a path into a LIST / MAP, LZO or INT96 under Option::INT_96_AS_DATETIME = false failed inside arrow-ext
refused before the first batch naming the column and \Flow\Parquet\Reader::php()
Reader::read() returned without touching the file; an invalid file failed at metadata()
the footer is read when the file is opened: Reader::read() / readStream() throw InvalidArgumentExceptionGiven file is not valid Parquet file: ...
Flow\Arrow\Parquet\Reader / Writer / Exception
removed: Flow\Parquet\Engine\{RustParquetFileReader, RustParquetFileWriter}, Flow\Arrow\Parquet\{RustColumnsReader, RustBatchReader}; refusals are Flow\Parquet\Exception\*
flow-php/parquet installable with any arrow-ext version
conflict: ext-arrow <0.45
#85) flow-php/parquet - PhpParquetEngine reads unsigned integers and INT96 exactly;
FlatColumn::enum() writes the ENUM logical type; values it wrote wrongly are refused
Before
After
UINT_32 4000000000 read as -294967296; UINT_64 18446744073709551615 read as -1
4000000000; a UINT_64 above PHP_INT_MAX is refused
INT96 1960-01-01 23:59:59.999999999 read as 1960-01-02 00:00:00 (rounded)
Validator::validate(Column $column, mixed $data, int $row): void - custom validators add the parameter (the 0-based row within the writer)
logical INTEGER read without its width and sign; written back as an IntType missing both (a footer arrow-rs refuses)
LogicalType::integerData(): ?LogicalType\Integer keeps them and writes them back; LogicalType::integer() → LogicalType::integer(int $bitWidth, bool $isSigned)
Flow\Arrow\Parquet\Reader / Writer - PHP values per call (readRowGroup(), writeBatch())
Flow\Parquet\Engine\{RustParquetEngine, RustParquetFileReader, RustParquetFileWriter} implement ParquetEngine, ParquetFileReader, ParquetFileWriter - the values PhpParquetEngine reads and accepts; Flow\Arrow\Parquet\RustBatchReader and RustParquetFileWriter::writeArrowBatch() hand batches to other extensions through the Arrow C Data Interface
#87) flow-php/etl-adapter-csv - to_csv() writes a boolean as true / false
Before
After
true → 1, false → an empty field, the same bytes as null
true / false; null stays an empty field
a NOT NULL boolean column could not be read back by from_csv()
reads back equal; a nullable column keeps false and null apart
1.0, 1.0e+25, -0.0 - always a fraction or an exponent
NAN, INF, -INF
unchanged; refused inside a nested CSV cell (JsonException)
JSON: the digits followed the serialize_precision ini (17 → 0.10000000000000001)
the shortest digits under any ini value
Writing a float column forces serialize_precision = -1 for the render and restores the ini afterwards. When the ini
holds another value, ini_set() must be allowed; otherwise the writers throw
RuntimeException('Writing floats requires serialize_precision = -1 and ini_set() cannot change it').
$columns holds one list per top-level column, every list of one length; a schema column it lacks is written as nulls,
a key the schema lacks is ignored. Writer::writeColumns() is the public door.
#95) flow-php/etl-adapter-json, -csv - the JSON shape follows the column type, never the data
Before
After
map<int, string> holding the keys 0, 1: ["a","b"]
{"0":"a","1":"b"}
an empty map: []
{}
a json column holding {}: []
{}
a row whose column names are 0, 1, ...: a list
an object
a list
[...], unchanged
Applies to to_json(), to_json_lines() and to a list / map / structure cell of to_csv().
JSON lines and a document holding a top-level array are read by the extension, which refuses what the PHP reader
tolerated, with Flow\ETL\Exception\RuntimeException. A pointer, or a document that is not an array, is read as
before.
Before
After
\v, \f or BOM bytes between tokens ([{"id":1}\v,{"id":2}])
read
Malformed JSON in "<uri>" at element 0: ...
a BOM at the start of a later line
read
Malformed JSON in "<uri>" at line N: ...
content after a line's record ({"id":1}{"id":9})
the first record, the rest ignored
Malformed JSON in "<uri>" at line N: ...
content after the closing ] ([{"id":1}] x)
ignored
Malformed JSON in "<uri>" at element N: ...
a scalar JSON line (5)
JsonMachine\Exception\SyntaxErrorException
A JSON record must be an object or an array, int given in "<uri>".
#105) flow-php/parquet - ArrowParquetEngine, ArrowParquetFileWriter and the arrow() factories removed
Before
After
new ArrowParquetEngine($options)
new RustParquetEngine($options) - arrow-ext's class
Reader::arrow($options)
new Reader(options: $options, engine: new RustParquetEngine($options))
Writer::arrow($compression, $options)
new Writer($compression, $options, new RustParquetEngine())
ArrowParquetEngine::mapCompression($compression)
removed - RustParquetFileWriter takes the Compressions case
new ArrowParquetFileWriter(...)
RustParquetEngine::openForWrite(...), a RustParquetFileWriter
new Reader(ByteOrder::BIG_ENDIAN) with arrow loaded read through arrow
reads through PhpParquetEngine
AdaptiveParquetEngine stays the default of Reader and Writer. Reader::php() / Writer::php() stay.
flow_php throws Flow\ETL\Exception\RuntimeException, plain \Exception when flow-php/etl is not autoloadable.
FloeStreamReader wraps it as FloeException, FloeSerializer as SerializationException, both with the original
as previous; FloeStreamWriter lets it through. Both catch the base Flow\ETL\Exception\RuntimeException now, so
its subclasses thrown while decoding are wrapped as well.
#107) flow-php/etl - RowsBuffer, RowEquality and eleven Rows methods removed, Rows::join*() moved to
concat() requires every part to carry the receiver's schema, an empty receiver too - merge() adopted the other
batch's schema when the receiver was empty. Start an accumulator from null instead of rows(schema()):
// before
$all = rows(schema());
foreach ($batches as $batch) {
$all = $all->merge($batch);
}
// after
$all = null;
foreach ($batches as $batch) {
$all = $all === null ? $batch : $all->concat($context->backend(), $batch);
}
-google-sheet - one record rule: header names, absent cells, blank leading rows
Header names follow the CSV rule in all three: scalar as text, trimmed, a blank cell named e + its position.
header row [' id ', '', 'name']
Excel before [' id ', '', 'name'] after ['id', 'e01', 'name']
Google Sheet before [' id ', '', 'name'] after ['id', 'e01', 'name']
Before
After
CSV withEmptyToNull(false), short line 1,a under id,name,age
['1', 'a', '']
['1', 'a', null] - an absent cell is null, emptyToNull only turns a present '' into null
$loader->load($rows, $context) - partitions come from partitionBy()
#116) flow-php/etl - the configured Backend reaches every batch: Rows, joins, Floe writers
Before
After
$rows->matchTo($schema)
$rows->matchTo($schema, $context->backend())
$rows->project($schema)
$rows->project($schema, $context->backend())
rows($schema) (DSL)
rows($schema) - builds in new AdaptiveBackend(); rows($schema, $backend) builds in $backend
$rows->project($rows->schema()->keep(...$names)) to keep columns
$rows->select(...$names) - the columns as they are, in schema order, no conversion
new FloeWriter($filesystem, $schema, $options)
new FloeWriter($filesystem, $schema, $backend, $options)
new FloeStreamWriter($schema, $options)
new FloeStreamWriter($schema, $backend, $options)
Every batch an extractor yields is adopted into the configured backend before the first step sees it - a custom
extractor that builds PhpBackend rows reads as RustColumns under flow_php with the default backend, and native
rows read as PHP columns under config_builder()->backend(new PhpBackend()). $rows->concat($backend, ...) returns
columns of $backend too.
Rows::matchTo() restamps a changed column whose type change the types prove (same kind at every level, no nested
optional → required) and reads values only for any other change; a null under NOT NULL is reported at its row.
PostgreSQL xml with a DOCTYPE, comments or processing instructions outside the root
the root element only - they are dropped, as DBAL already did
#124) flow-php/etl - ResultColumn takes the Backend; functions build in the configured backend
Before
After
(new ResultColumn())->of($this, $values) in a custom function
(new ResultColumn($context->backend()))->of($this, $values)
a function's result under flow_php - a PHP column, copied into the extension's buffers after the step
a column of the configured backend (RustColumn under flow_php)
lit() / on_each() columns
built by the configured backend
#125) flow-php/etl - file reading and writing plumbing moved to Extractor\File and Loader\File
Class names are unchanged; only the namespace moves. Statistics, FilesExtractor, PathPartitionsExtractor,
Partitioning, PartitioningLoader and SaveMode stay where they are. FilesSink::touched() and
FilesSink::openStreams() are removed, no replacement.
Before
After
Flow\ETL\Extractor\FileExtractor
Flow\ETL\Extractor\File\FileExtractor
Flow\ETL\Extractor\FileReading
Flow\ETL\Extractor\File\FileReading
Flow\ETL\Extractor\SourceFile
Flow\ETL\Extractor\File\SourceFile
Flow\ETL\Extractor\FileColumns
Flow\ETL\Extractor\File\FileColumns
Flow\ETL\Extractor\FileConstants
Flow\ETL\Extractor\File\FileConstants
Flow\ETL\Extractor\PartitionColumns
Flow\ETL\Extractor\File\PartitionColumns
Flow\ETL\Extractor\PartitionTypes
Flow\ETL\Extractor\File\PartitionTypes
Flow\ETL\Extractor\DeclaresPartitionTypes
Flow\ETL\Extractor\File\DeclaresPartitionTypes
Flow\ETL\Extractor\PathFiltering
Flow\ETL\Extractor\File\PathFiltering
Flow\ETL\Extractor\SelfDescribingFile
Flow\ETL\Extractor\File\SelfDescribingFile
Flow\ETL\Extractor\MetadataColumns
Flow\ETL\Extractor\File\MetadataColumns
Flow\ETL\Extractor\MetadataColumnsExtractor
Flow\ETL\Extractor\File\MetadataColumnsExtractor
Flow\ETL\Loader\FileLoader
Flow\ETL\Loader\File\FileLoader
Flow\ETL\Loader\PartitionRouter
Flow\ETL\Loader\File\PartitionRouter
Flow\ETL\Filesystem\FilesSink
Flow\ETL\Loader\File\FilesSink
Flow\ETL\Extractor\SpillState
Flow\ETL\Extractor\Memory\SpillState
Flow\ETL\Extractor\SpilledRows
Flow\ETL\Extractor\Memory\SpilledRows
Flow\ETL\Extractor\InMemoryRows
Flow\ETL\Extractor\Memory\InMemoryRows
Flow\ETL\Extractor\InferredRows
Flow\ETL\Extractor\Memory\InferredRows
#126) flow-php/etl - FloeSerializer and the caches take the backend and the serializer explicitly
Before
After
new FloeSerializer(), new FloeSerializer($batchSize)
new FloeSerializer($backend), new FloeSerializer($backend, $batchSize)
new FilesystemCache($filesystem, $cacheDir, $serializer)
new FilesystemCache($filesystem, $serializer, $cacheDir)
new PSRSimpleCache($cache, $ttl, $serializer)
new PSRSimpleCache($cache, $serializer, $ttl)
new PSRCacheBuckets($cache, $prefix, $ttl, $serializer)
new PSRCacheBuckets($cache, $serializer, $prefix, $ttl)
new ApcuCache($serializer, $namespace) - entries stored as serialized strings
CacheIndex::toRows()
toRows(Backend)
new Config(..., HashRepartitionConfig, Calculator, ...)
new Config(..., HashRepartitionConfig, Backend, Calculator, ...) - Backend required
To only move the cache directory, keep the default cache - it shares the pipeline's serializer and backend:
// before
config_builder()->cache(filesystem_cache($dir));
// after
config_builder()->cacheDir($dir);
Build a cache yourself only for a custom serializer or filesystem; it has no default serializer any more and takes it
from its caller, e.g. filesystem_cache(new FloeSerializer(new AdaptiveBackend()), $dir).
#127) flow-php/filesystem - Path::uri() of a local file keeps the root slash
Before
After
path('/data/orders.csv')->uri()
file://data/orders.csv
file:///data/orders.csv
path('/')->uri()
file://
file:///
_input_file_uri, explain()Source: of a local file
file://data/orders.csv
file:///data/orders.csv
Other protocols are unchanged (memory://path, s3://bucket/key). Windows drive paths stay file://C:/....
new Config(..., FilesystemStreams $filesystemStreams, ...) constructor parameter
removed
Build Config through Config::builder() / Config::default(). A Loader holding per-run state (an open stream, a
writer, a counter) implements Discardable and drops that state in discard().
#7) flow-php/etl - RetryLoader no longer retries InvalidLogicException by default
Before
After
new RetryLoader($loader) default strategy new AnyThrowable(3)
new AnyThrowableExcept([InvalidLogicException::class], 3)
write_with_retries($loader) default strategy new AnyThrowable(3)
new AnyThrowableExcept([InvalidLogicException::class], 3)
an InvalidLogicException was attempted 4 times with delays between
#8) flow-php/etl - operations inside a Transformation answer for the whole stream
Before
After
$df->sortBy(ref('id')) inside a Transformation sorted each batch on its own
sorts the whole stream
$df->aggregate(...) / groupBy()->aggregate() / pivot() / window functions inside a Transformation answered per batch
answer for the whole stream
$df->offset(2) inside a Transformation lost rows
skips exactly the offset across the stream
$df->cache($id) inside a Transformation persisted one batch
persists the whole stream
$df->batchBy(...) / batch_size(...) cut chunks at the incoming batches
cut chunks over the stream
$df->join(...) inside a Transformation emitted rows in input order
emits rows grouped by key
$df->partitionBy(...) inside a Transformation
removed - $df->repartition(...) groups the whole stream by key, see 60)
a Transformation calling $df->fetch() / count() / schema() silently answered over an empty stream
throws InvalidLogicException
write_with_retries($loader) around to_transformation(...) (any wrapped step) or around a to_branch(...) armed with withTransformation(...), at any nesting depth
throws InvalidLogicException at the first load(), use to_transformation(..., write_with_retries($loader)) or to_branch(..., write_with_retries($loader))->withTransformation(...)
Flow\ETL\Extractor\SwappableRowsExtractor
removed
#9) flow-php/etl - to_branch()->withTransformation() drives its Transformation once over the whole stream
Before
After
the Transformation ran on each filtered batch in its own DataFrame
one nested pipeline spans the stream
$df->sortBy(...) in a branch transformation sorted each batch alone
sorts the whole branch stream
$df->aggregate(...) / limit() / other Processor-backed operations answered per batch
answer once for the stream
a Transformation calling $df->fetch() / count() / schema() returned per-batch answers
throws InvalidLogicException
the wrapped loader received exactly one load() per outer batch
receives output as the transformation produces it; blocking operations deliver at closure()
removed - return the plain value, declare the type in returns()
FrameAccumulator::value(): mixed
value(): float|int|null
Parameter::as*() over a malformed value - null / false / $default
throws InvalidArgumentException
Parameter::asBoolean(): bool
?bool
Parameter::asType()
removed
Before:
final class Shout extends ScalarFunctionChain
{
public function __construct(private readonly ScalarFunction $value) {}
public function eval(Row $row, FlowContext $context): ?string
{
$value = (new Parameter($this->value))->asString($row, $context);
return $value === null ? null : strtoupper($value);
}
}
After:
final class Shout implements ScalarFunction
{
use ScalarFunctionChain;
use ResolvesFromChildren;
public function __construct(private readonly ScalarFunction $value) {}
public function eval(Row $row, FlowContext $context): ?string
{
$value = (new Parameter($this->value))->asString($row, $context);
return $value === null ? null : strtoupper($value);
}
public function returns(): Type
{
return type_optional(type_string());
}
public function children(): array
{
return [$this->value];
}
public function withChildren(array $children): static
{
return new self($children[0]);
}
}
#27) flow-php/etl - user callbacks removed, call() takes a ScalarFunction and a return type
#31) flow-php/etl - Rows holds only rows that match its schema
Before
After
any row accepted
new Rows($schema, ...$rows) throws Flow\ETL\Exception\SchemaMismatchException on a wrong-typed value, a null in a NOT NULL column, an undeclared column or a missing NOT NULL column
-
Flow\ETL\Exception\ColumnMismatchException, the getPrevious() of SchemaMismatchException
a row without a declared nullable column
padded with null
$rows->toArray() - keys in each row's insertion order
from_csv($path)->withBatchSize(500) - on each extractor
Config::extractorBatchSize()
removed - $extractor->batchSize()
FlowContext::entryFactory()
removed
new Config(...) with $filesystemTable, $putInputIntoRows, $extractorBatchSize, $entryFactory
those parameters removed, HashRepartitionConfig $repartition added - build through Config::builder()
withMetadataColumns() is on from_csv(), from_json(), from_json_lines(), from_parquet(), from_text(),
from_xml(), from_excel(), from_google_sheet(), from_google_sheet_columns() and from_floe().
up to batchSize() rows per Rows: 100 by default, 1000 for DbalLimitOffsetExtractor, DbalKeySetExtractor, PostgreSqlLimitOffsetExtractor, PostgreSqlKeySetExtractor and PostgreSqlCursorExtractor
Flow\ETL\Pipeline\Optimizer\BatchSizeOptimization re-cut batches to 1000 rows before DbalLoader / PostgreSqlLoader
removed - $df->batchSize(1000) before write()
new BatchExtractor($extractor, chunkSize: 10)
new BatchExtractor($extractor, batchSize: 10) - a size below 1 throws
$df->duplicateRow(...) appended the copies after the batch's last row
each copy follows the row it duplicates
#35) flow-php/etl - LimitableExtractor replaced by LimitPushDown
Before
After
Flow\ETL\Extractor\LimitableExtractor
Flow\ETL\Extractor\LimitPushDown
Flow\ETL\Extractor\Limitable (trait)
Flow\ETL\Extractor\PushesLimit
changeLimit(int $limit): void
pushLimit(int $limit): void - a second push can only lower the limit
parameter removed - a missing key is null under a nullable column, throws under NOT NULL
new HashBucketing(..., nullOnMissing: true)
parameter removed
join() over a row without the join column - threw
the row is unmatched
#51) flow-php/etl - a join producing a duplicate column throws SchemaDefinitionNotUniqueException
Before
After
$df->join($right, join_on(['id' => 'id'])), both sides carry name - JoinException (RuntimeException): Merged entries names must be unique, given: [id, name] + [name] try to use a different join prefix than: ""
SchemaDefinitionNotUniqueException (InvalidArgumentException): Entry definitions must be unique, duplicated entries: [name], all: [id, name, name]
thrown when the first pair of rows was merged
thrown when the pipeline starts, before any row is read
$rows->joinCross($right, '') over shared columns - InvalidArgumentException: ... + [id, name]. Please consider using join prefix option
SchemaDefinitionNotUniqueException: Entry definitions must be unique, duplicated entries: [id, name], all: [id, name, id, name]
catch (JoinException $e) around a join
catch (SchemaDefinitionNotUniqueException $e), or give the right side a prefix: join_on(['id' => 'id'], 'joined_')
#52) flow-php/etl - Joiner takes JoinSide and no EntryFactory, Expression drop helpers removed
frames() - a truncated stream always throws FloeException
#67) flow-php/etl / flow-php/etl-adapter-parquet - a Floe or Parquet glob checks every file against the first
Before
After
from_parquet($glob) / from_floe($glob), a later file with other columns or column types - rows kept that file's columns
throws InferredSchemaException naming both files - read with ->unionByName() or ->withSchema(...)
same read, a later file with the same columns in another order - rows kept that file's order
rows follow the first file's column order
-
->unionByName() reads every file under the merged schema; a column a file lacks is null
#68) flow-php/etl and every package with DSL functions - documentation attributes leave the packages
Before
After
Flow\ETL\Attribute\DocumentationDSL
Flow\Documentation\Attribute\DocumentationDSL
Flow\ETL\Attribute\DocumentationExample
Flow\Documentation\Attribute\DocumentationExample
Flow\ETL\Attribute\Module
Flow\Documentation\Attribute\Module
Flow\ETL\Attribute\Type
Flow\Documentation\Attribute\Type
No installable package ships Flow\Documentation\Attribute\*: ReflectionAttribute::newInstance() on them throws
Attribute class "..." not found. Read them with getName() / getArguments().
#69) flow-php/etl / flow-php/types - TypeMerge moved to flow-php/types as TypeWidener
Before
After
Flow\ETL\Schema\Definition\TypeMerge
Flow\Types\Type\TypeWidener
merge($left, $right)
widen($left, $right)
mergeLists() / mergeMaps() / mergeStructures()
widenLists() / widenMaps() / widenStructures()
#70) flow-php/types - structure optional fields are declared inline with structure_element()
a custom Flow\Doctrine\Bulk\Dialect\Dialect implementation
adds public function maxBindParameters(): int - the platform's bind-parameter limit, at least 1
#80) flow-php/etl-adapter-csv, -json, -excel, -google-sheet - a read without a schema infers one schema
Before
After
schema of a read without withSchema()
each yielded Rows typed from its own values
one schema for the whole read, inferred from a sample
nullability of an inferred column
nullable only where that Rows held a null
always nullable (?integer, ?string, ...)
sample
-
first 20,480 rows over the first 10 files; Google Sheet: first 100 rows
value past the sample that the inferred type refuses
read, typed on its own
throws Flow\ETL\Exception\SchemaMismatchException
CSV/Excel file or Google Sheet whose header differs from the inferred one (header-only files too)
read with its own columns
throws Flow\ETL\Exception\InferredSchemaException
row key order
the row's own; a body column named like a partition is overwritten in place
the schema's: body columns (JSON: first seen across the sample), then partition columns
flow file:schema over CSV / Excel
every column string, nullable: false unless a value was empty
typed, every column nullable: true
infer from every row
-
->inferSchema(infer_schema()->sampleSize(-1))
infer from every file
-
->inferSchema(infer_schema()->filesToSniff(-1))
every column ?string
-
->inferSchema(infer_schema()->allStrings())
files with different columns
-
->inferSchema(infer_schema()->unionByName())
no inference
-
->withSchema($schema)
On from_google_sheet(), ->inferSchema(infer_schema()) without ->sampleSize() samples 20,480 rows, not 100.
#81) flow-php/etl-adapter-csv - cells are typed, not strings
Cell
Before
After
1
string
?integer
1.5
string
?float
true / false
string
?boolean
20240101
string
?integer
2024-01-01 / 2024-01-01 10:00:00
string
?date / ?datetime
uuid / JSON text
string
?uuid / ?json
01234, 2024-01, 12 (padded)
string
?string
Keep every column a string: from_csv($path)->inferSchema(infer_schema()->allStrings()).
#82) flow-php/etl-adapter-doctrine - from_dbal_*() type columns from the query
Driver
Column
Before
After
pgsql
NUMERIC
string
float
pgsql
DATE, TIMESTAMP, TIMESTAMPTZ
string
DateTimeImmutable
pgsql
TIME
string
DateInterval
pgsql
UUID
string
Flow\Types\Value\Uuid
pgsql
JSON, JSONB
string
Flow\Types\Value\Json
pgsql
XML
string
DOMDocument
pgsql
any other type (INTERVAL, INET, MONEY, OID, arrays, ranges, enums, ...)
string
throws SchemaNotDerivableException
mysqli
DECIMAL
string
float
mysqli
DATE, DATETIME, TIMESTAMP
string
DateTimeImmutable
mysqli
TIME
string
DateInterval
mysqli
JSON
string
Flow\Types\Value\Json
mysqli
any other type (BIT, GEOMETRY, ...)
string
throws SchemaNotDerivableException
sqlite3, pdo_sqlite
INTEGER, REAL
1, 1.5
'1', '1.5' - every column is ?string
pdo_pgsql, pdo_mysql, any other
every column
driver values
throws SchemaNotDerivableException - use the pgsql / mysqli driver or ->withSchema(...)
Applies to from_dbal_query(), from_dbal_queries(), from_dbal_limit_offset(), from_dbal_limit_offset_qb() and
from_dbal_key_set_qb() without ->withSchema(). A query the driver cannot describe (multi-statement, data-modifying
CTE, INSERT ... RETURNING) also throws. Declare ->withSchema(...) to pick the types.
#83) flow-php/etl-adapter-doctrine, -postgresql - withPageSize() / withFetchSize() become withBatchSize()
an empty or whitespace-only line - threw SyntaxErrorException
skipped, in schema() and extract()
#91) flow-php/etl-adapter-parquet - to_parquet() does not validate values against the Parquet schema
Before
After
to_parquet($path) wrote with Option::VALIDATE_DATAtrue
false
-
to_parquet($path)->withOptions(Options::default()) validates again
#92) flow-php/etl-adapter-postgresql - from_pgsql_*() type columns from the query instead of reading strings
PostgreSQL column
Before
After
numeric
string
float
date, timestamp, timestamptz
string
DateTimeImmutable
time
string
DateInterval
timetz
string with offset
DateInterval, offset dropped
uuid
string
Flow\Types\Value\Uuid
json, jsonb
string
Flow\Types\Value\Json
xml
string
DOMDocument
oid
string
int
one-dimensional array (int4[], text[], ...)
string'{1,NULL,3}'
[1, null, 3], null elements kept
multi-dimensional array
string
throws while hydrating
record, point, line, lseg, box, path, polygon, circle
string
throws SchemaNotDerivableException
Applies to from_pgsql_cursor(), from_pgsql_limit_offset() and from_pgsql_key_set() without ->withSchema(). A
query that cannot run as a subquery (multi-statement, data-modifying CTE, INSERT ... RETURNING) also throws
SchemaNotDerivableException. Declare ->withSchema(...) to pick the types.
The worker-cycle span and the trace: worker/both modes are gone; a consumed message is a process span and a
produced one a send span. The messenger:consume worker loop is suppressed via
instrumentation.console.exclude_commands - which now fully suppresses matching commands (previously it only
skipped the console span) and defaults to ['messenger:consume'] - so transport-poll/idle-tick work does not surface
and the long-lived messenger:consume console span is dropped. Per-message handler traces are still recorded. Set
console.exclude_commands: [] to trace the worker loop.
#2) flow-php/symfony-telemetry-bundle - cache span names unified to dotted lowercase
Cache spans now match the DBAL/messenger convention (dotted lowercase, low cardinality). The {key} and {pool}
that were baked into the span name move out of it - they were already available as the cache.key and cache.pool
attributes. Rename these series in dashboards and alerts, and update any filters on the camelCase cache.operation
values.
#3) flow-php/telemetry - Sampler::shouldSample() receives the parent Context
Custom Sampler implementations (including a sampler: { type: service } service in
flow-php/symfony-telemetry-bundle) must update the signature and forward $parentContext to any delegated sampler.
The context storage is no longer tagged kernel.reset in flow-php/symfony-telemetry-bundle; scope balance is
maintained by attach/detach alone. A custom context_storage service no longer needs a reset() method.
#5) flow-php/symfony-telemetry-bundle - messenger tracing middleware auto-injected into all buses
The tracing middleware is now injected into every message bus automatically. If you previously added
flow.telemetry.messenger.middleware to a bus's framework.messenger.buses.*.middleware list by hand, remove it to
avoid duplicate spans.
#6) flow-php/symfony-telemetry-bundle - http_kernel.exclude_paths now suppresses the whole request
Before
After
Excluded path only skips its own request span
Excluded path suppresses tracing for the entire request
DBAL/cache/kernel.terminate work still recorded
DBAL/cache/kernel.terminate work produces no spans
An excluded path now attaches the OpenTelemetry suppression key for the request's duration (through
kernel.terminate), so lower-level auto-instrumentation and terminate-phase database writes no longer emit orphan root
spans (e.g. the /_wdt toolbar fetch writing an audit row after its response). If you relied on those child spans being
recorded for an excluded path, remove the path from exclude_paths.
#7) flow-php/postgresql, flow-php/symfony-postgresql-bundle - traceTransactions bool replaced by
TransactionSpanMode::PER_OPERATION emits a short span per BEGIN/COMMIT/ROLLBACK; GROUPED (default) keeps the
single long-lived transaction span.
The db.client.operation.duration metric now uses only low-cardinality dimensions: queries are tagged with
db.system.name, db.namespace, db.operation.name, db.collection.name; transactions with db.system.name,
db.namespace, db.operation.name (begin/commit/rollback), db.transaction.nesting_level. db.query.text,
db.query.parameter.*, server.address and db.transaction.savepoint are no longer metric dimensions.
#8) flow-php/symfony-telemetry-bundle - DBAL spans, metrics and config aligned with the PostgreSQL client
Query spans now carry db.system.name, db.namespace, server.address, server.port, db.operation.name,
db.collection.name, db.response.returned_rows and (on error) db.response.status_code, and the instrumentation
emits db.client.operation.duration and db.client.response.returned_rows metrics. If you matched DBAL spans by their
doctrine.dbal.* names, switch to the semantic names above.
#9) flow-php/symfony-telemetry-bundle - custom attribute keys moved out of reserved OTel namespaces
Before
After
controller
flow.symfony.controller
controller.argument
flow.symfony.controller.argument
code.namespace + code.function
code.function.name (Class::method)
command.name
flow.symfony.command.name
command.class
flow.symfony.command.class
process.signal
flow.symfony.command.signal
process.exit_code
process.exit.code
db.connection.name
flow.db.connection.name
db.transaction.nesting_level
flow.db.transaction.nesting_level
http.client.name
flow.http.client.name
log.channel
flow.log.channel
#10) flow-php/symfony-telemetry-bundle - HTTP spans aligned with the stable HTTP semconv
Before
After
server span url.full
removed
server span url.path = URI incl. query
path only; query moves to url.query
-
server span user_agent.original
client span name {method} {host}
{method}
client span server.port only when non-default
always set (scheme default 80/443 filled in)
Applies to the HttpKernel server span and to the http_client/psr18_client client spans.
#11) flow-php/symfony-telemetry-bundle - messenger destination is the transport; metrics_duration_unit removed
Custom SchemaValidator implementations must return a Flow\ETL\Schema\Validator\ValidationContext
built from the missing, mismatched (MismatchedDefinition), and unexpected definitions they reject.
#18) flow-php/filesystem - Partition name and value forbid { and }
Before
After
new Partition('na{me', 'a}b')
throws InvalidArgumentException
partitionBy() values with {}
throws InvalidArgumentException
{name} in a path is now a partition placeholder resolved from partitionBy() columns; strip braces from partition
values before partitioning.
#19) flow-php/etl - Cache stores only Rows; cache indexes are stored as Rows
Before
After
Cache::get(string): Row|Rows|CacheIndex
Cache::get(string): Rows
Cache::set(string, Row|Rows|CacheIndex): void
Cache::set(string, Rows): void
$cache->set($id, $row)
$cache->set($id, rows($row))
$cache->set($id, $cacheIndex)
$cache->set($id, $cacheIndex->toRows())
$cache->get($id) returning CacheIndex
CacheIndex::fromRows($id, $cache->get($id))
Custom Cache implementations must adopt the Rows-only signatures.
#20) flow-php/etl - cache and serialization use the Floe (.floe) binary format; datetime subclasses rejected
filesystem_cache(..., Serializer $serializer = new NativePHPSerializer())
filesystem_cache(..., Serializer $serializer = new FloeSerializer())
new FilesystemCache($filesystem, $serializer, $cacheDir)
new FilesystemCache($filesystem, $cacheDir, $serializer)
caching or serializing a DateTime/DateTimeImmutable subclass (e.g. Carbon)
throws Flow\Floe\Exception\FloeException
Delete cache directories written by 0.41 - the old serialized payloads are unreadable. Convert
DateTime/DateTimeImmutable subclasses to DateTime/DateTimeImmutable before caching or serializing.
A Floe file carries exactly one schema, and sort / join / group-by spill batches through it - a pipeline with a drifting
or schemaless source must declare column types at the source so nullable columns carry a concrete type
(DataFrame::match($schema) validates but does not retype values):
array_to_rows($data, Hydrator $hydrator = new AdaptiveRowHydrator(), ...)
cast() coerces untrusted values through the schema (null schema infers types); hydrate() instantiates trusted,
already-typed values and requires a schema. A custom extractor reading raw values (CSV, JSON, XML, Excel, Text, Google
Sheet, PostgreSQL, Doctrine) calls $context->hydrator()->cast(...); a self-describing source (Parquet, Floe)
calls $context->hydrator()->hydrate(...).
Under a schema, entries emit in schema-definition order (was data-key order with missing columns appended); a schema
column missing from the source hydrates as a typed null (was a StringEntry null); a column absent from the schema is
dropped. When the flow_php extension is loaded, AdaptiveRowHydrator runs hydration natively - pin the pure-PHP
engine with config_builder()->hydrator(new PhpRowHydrator()).
withDateFormat() on CSVLoader, JsonLoader, JsonLinesLoader, XMLLoader, ExcelLoader, SealLoader; SealLoader::withDateTimeFormat()
Behavioural changes: CSV/JSON/XML/Excel/Seal date columns render with dateFormat (Y-m-d, was dateTimeFormat);
XML date/time columns render (was InvalidArgumentException); Seal time columns render as microseconds (was
null); Google Sheet _spread_sheet_id/_sheet_name are appended to the schema under putInputIntoRows() (were
dropped when a schema was set); Excel now appends _input_file_uri under putInputIntoRows().
#29) flow-php/etl - null is a first-class NullEntry/NullDefinition; from_null metadata removed
Custom aggregators must implement references() - return the references the aggregator reads, or
null when they cannot be statically enumerated (disables spill column pruning).
#31) flow-php/etl - DataFrame::pivot() removed; pivot is declared between groupBy() and aggregate()
Assign the result of every Schema mutator - add, addAfter, addBefore, addMetadata, gracefulRemove,
insertAt, keep, makeNullable, merge, moveAfter, moveBefore, moveTo, prepend, remove, rename,
reorder, replace, setMetadata, sort - and of Definition::addMetadata() / Definition::setMetadata().
Discarding it is a silent no-op.
The Elasticsearch adapter has been removed from Flow PHP and replaced by the SEAL
adapter (flow-php/etl-adapter-seal), a search engine abstraction layer that supports Elasticsearch, OpenSearch,
Meilisearch, Solr, Typesense, Algolia, RediSearch and Loupe.
To migrate, install the SEAL adapter together with the engine adapter for your backend:
Then build a CmsIg\Seal\Engine and pass it to to_seal_upsert() instead of the previous to_es_bulk_index() (or
Meilisearch) DSL functions:
use CmsIg\Seal\Engine;
use CmsIg\Seal\Adapter\Elasticsearch\ElasticsearchAdapter;
use function Flow\ETL\Adapter\Seal\to_seal_upsert;
$engine = new Engine(
new ElasticsearchAdapter($client),
$schema,
);
data_frame()
->read(/* ... */)
->write(to_seal_upsert($engine, 'index_name'))
->run();
flow-php/symfony-http-foundation-telemetry-bridge is now an optional dependency
Before
After
installed transitively by the bundle
install explicitly to enable HTTP trace context propagation
instrumentation.http_kernel.context_propagation is silently disabled when the bridge is absent. To keep extracting
incoming and injecting outgoing W3C trace headers:
grouped under one messenger.receive root span per pass
Remove the link_to_worker key from config. Idle passes (no message received) are marked
messaging.symfony.worker.idle: true.
#9) flow-php/symfony-telemetry-bundle - messenger instrumentation also emits messaging metrics
Before
After
messenger instrumentation emitted spans only
also emits messaging.client.consumed.messages, messaging.client.sent.messages, messaging.process.duration
Disable with instrumentation.messenger.metrics: false; set the messaging.process.duration unit via
instrumentation.messenger.metrics_duration_unit (s default, or ms).
Remove propagation_style from config - continue mode is gone (it made the consumer span absorb the queue wait time).
trace selects which spans are emitted (handlers = the process/send message spans; none = metrics only);
link selects the consumer span's links. link: worker/both requires trace to include the worker, otherwise the
config is rejected. When trace excludes the worker, the transport's poll instrumentation (Doctrine DBAL, HTTP
client, ...) is suppressed during the receive loop so it does not surface as orphan spans - no messenger.receive span,
no orphans either way.
Sponsor
