fpstreams 2.4.0

fpstreams 2.4.0 adds stateful pipeline operations, more ways to compose collectors, and shared execution limits. It also makes execution reports easier to export and fixes several cases where optimized columnar paths could hide conversion errors.

Stateful pipelines and collectors

• flow.unfold() and aflow.unfold() generate values from explicit state transitions. Return (value, next_state) to continue or None to stop.
• map_accum(initial, function) threads state through a pipeline and emits one output per input. Its callback returns (next_state, output).
• window_reduce(size, combine, step=1) reuses partial aggregates across count-based windows. The combine function must be associative and must not modify its arguments. Overlapping windows cache at most size + 1 entries, with amortized constant combine work per input.
• collapse(predicate, collector) reduces adjacent runs through existing Collector state without buffering the raw run. The binary merger form remains available.
• Named aggregation accepts composed Collector instances. agg.min_by() and agg.max_by() retain a whole winning item, while agg.top() and agg.bottom() keep a bounded ranking per group. Equal keys preserve encounter order.

from operator import add
from fpstreams import flow

countdown = flow.unfold(4, lambda n: (n, n - 1) if n else None)
assert countdown.to_list() == [4, 3, 2, 1]
assert flow([2, 3, 5, 7]).window_reduce(3, add).to_list() == [10, 15]

Shared limits and file recovery

BoundedExecutor(pool, max_pending=...) shares an unfinished-task limit across queries using one thread or process pool. Queries clean up their own work, and the caller retains ownership of the pool. The limit counts tasks, not retained result bytes.

FileCheckpoint(directory, version=...) reuses completed file-partition outputs after checking input and output hashes. Processing versions are explicit. This provides file recovery, not transactions for external callback side effects.

Execution metadata

run_with_report() records up to 64 child plan and route observations without replacing the outer route. Reports mark additional observations as truncated.

ExecutionReport.to_dict() and ExecutionEvent.to_dict() export plain metadata. Reports and plan dictionaries carry an independent schema version. Reports include metric units, query-owned resource scope, and event limits.

import json
from fpstreams import flow

observed = flow([1, 2, 3]).run_with_report("sum")
assert observed.value == 6
metadata = observed.report.to_dict()
assert metadata["schema_version"] == 1
serialized = json.dumps(metadata)

Export does not rerun the source or include the terminal value. Applications can use the metadata in an OpenLineage integration and supply their own identities and lifecycle events.

Correctness and execution changes

• Prefix counts, Collector counts, and task/file limits use Python's integer index protocol. Floats, NaN, and infinity are rejected before opening sources or acquiring resources where applicable.
• Collectors.first() and head() preserve completed state inside grouped and partitioned collectors. Partial-state merging and reducer law checks also preserve iterator cleanup and reject missing combiners before invoking merge callbacks.
• Flow.collect() and AsyncFlow.collect() preserve a single Collector's result type in type checkers.
• Arrow and Parquet optimizations preserve visible conversion errors, residual filter order, and source boundaries. CSV scans validate caller-visible columns before query projection. Zero-column Arrow and Polars exports preserve row counts.
• Supported primitive Arrow comparison chains share a selection mask. Offline Optuna tuning informed the thresholds without adding a runtime dependency.
• filter_map() removes tuple adaptation in synchronous pipelines and uses existing async serial fusion. It calls the mapper once and discards only None. Spill decoding releases encoded buffers and completed frames earlier.

Python 3.11 or newer is required. See the changelog for the full release details and the documentation for examples and resource contracts.