Rows

Rows[T] adds record operations to a lazy Flow: field selection, joins, grouping, reshaping, and record I/O.

Enter it with flow(source).rows(), construct it with rows(source), or use one of the adapter factories. A Rows view shares its Flow's plan and source ownership; it does not inspect the source or make a one-shot input reusable.

Inspect field inputs (2.5.0)

The source version supports static field dependencies through the existing explanation object. It inspects the logical plan without opening the source:

from fpstreams import col, rows

query = rows([{"price": 4, "tax": 2}]).with_columns(
    total=col("price") + col("tax")
).select("total")
metadata = query.explain().to_dict(include_field_dependencies=True)["field_dependencies"]
assert metadata["fields"]["total"] == [
    {"source": "source_0", "field": "price"},
    {"source": "source_0", "field": "tax"},
]

Unknown dependencies are None; an empty list means no declared field input. This is logical field metadata. It does not trace individual records or permit operator reordering. See static field dependencies and the plan dictionary reference.

Entering and leaving Rows

Nonconflicting record methods can be called directly on Flow. select(), with_columns(), rename(), cast(), fill_nulls(), drop_nulls(), explode(), unnest(), unpivot(), and pivot() return Rows, while group_by() returns GroupedRows.

Four relation-building names already have Flow meanings. Use .rows() before them when you intend these relational forms:

Flow call	Flow meaning	Explicit Rows call	Rows meaning
`drop(count)`	Skip leading items	`rows().drop(*columns)`	Remove record fields
`join(separator)`	Join string representations	`rows().join(other, ...)`	Relational join
`aggregate(...)`	Execute and return a dictionary	`rows().aggregate(...)`	Build a lazy one-row relation
`where(predicate)`	Alias of `filter`	`rows().where(predicate, **equalities)`	Filter records, including field equalities

Flow also has same-named output methods. Enter .rows() before to_csv(), to_pandas(), or to_df() when you need the record writer or Rows-specific materialization options.

Rows.map() and Rows.flat_map() return Flow because their result may no longer be record-shaped. If they do emit records, calling a nonconflicting record method enters Rows again.

Rows.to_flow() returns the exact underlying Flow without adding an operation or copying data. Rows.explain() delegates to that Flow and does not consume the source, so relational plans can be inspected without using private attributes.

Rows.concat(*sources) drains each source in order and stays lazy. It preserves records exactly as received; unlike a dataframe union, it does not align fields, fill missing columns, or infer a common dtype.

Creating rows

Call	Source
`rows(source)`	Records, a `Flow`, supported tabular objects, or standard tabular-protocol providers
`rows.from_csv(path)`	Compatible CSV rows from a path, text handle, or opener
`rows.scan_csv(path)`	Typed, streaming Arrow CSV batches
`rows.from_jsonl(path)`	JSON objects from a path, text/binary handle, or opener
`rows.from_arrow(source)`	An Arrow `Table`, `RecordBatch`, `RecordBatchReader`, or `__arrow_c_stream__` provider
`rows.from_columns(columns)`	Equal-length named columns retained as an Arrow table
`rows.from_numpy(array, columns=...)`	An explicit two-dimensional array converted lazily to named records
`rows.from_dataframe(frame)`	The dataframe interchange protocol
`rows.from_polars(frame)`	A Polars `DataFrame` or `LazyFrame`
`rows.from_parquet(source)`	Parquet data with optional projection and filtering
`rows.from_db(connect, query)`	A DB-API query using a connection factory
`rows.from_sqlite(database, query)`	A SQLite query

Path, opener, and database adapters open their resources only when the pipeline executes. An already-open CSV or JSONL handle is caller-owned and one-shot; fpstreams neither rewinds nor closes it. A zero-argument opener is replayable, and fpstreams closes each handle it returns. Dataframe interchange conversion and Polars LazyFrame collection are also deferred. An Arrow C stream provider is the exception: its stream is imported once at construction and treated as one-shot. Adapter docstrings list the exact options and return types.

Use from_csv() when Python csv.DictReader compatibility and string-valued cells matter. Use scan_csv() for typed inference and wide analytical scans. Query projection follows conversion of every caller-visible CSV column. A select() query still reports conversion errors in columns it does not return. Pass ConvertOptions(include_columns=[...]) explicitly to restrict which columns the CSV reader converts. Arrow's incremental reader is single-threaded and freezes inferred types after its first byte block, so pass ReadOptions or ConvertOptions.column_types when the default inference is not appropriate.

The primary flow(source) entry automatically recognizes PyArrow, pandas, and Polars objects and standard __arrow_c_stream__ or __dataframe__ providers. rows(source) uses the same dispatch path, including Arrow priority for an object that implements both protocols. The named factories remain useful when you want to make the adapter or its options explicit. flow.scan_csv() and flow.from_parquet() return Flow, while the corresponding Rows factories return the explicit relational view.

Bounded JSONL and spreadsheet-safe CSV

rows.from_jsonl() limits each physical line to 8 MiB by default. An oversized record raises BufferLimitError before JSON parsing; binary inputs are checked before decoding as well. Adjust the limit with max_record_bytes; use None only when unlimited records from a trusted local file are intentional.

CSV writers preserve raw cells by default. Use to_csv(..., spreadsheet_safe=True) for untrusted text that will be opened in a spreadsheet. A string whose first non-whitespace character is =, +, -, or @ receives a leading single quote. Non-string values are unchanged. CSV and JSON/JSONL writers return None.

Since 2.2.0, this protection also covers header cells without changing the keys used to read records. to_csv() and to_jsonl() also accept atomic=True and if_exists="error"; a failed export preserves the old file. to_jsonl(default=...) can serialize dates or other custom values. These additions were introduced in 2.2.0; see file output for examples and limits.

Rows to_csv() is the record writer and exposes fieldnames, include_header, and extrasaction. Flow to_csv() instead writes arbitrary items and accepts an optional header.

Column collection and custom iteration (2.5.0)

to_columns() collects fields in encounter order and fills missing cells with None:

from fpstreams import rows

assert rows([{"b": 1}, {"a": 2}]).to_columns() == {
    "b": [1, None],
    "a": [None, 2],
}

In the current source checkout, custom Rows or Flow iteration remains authoritative. to_columns() closes the iterator it opens on normal completion or conversion failure. A conversion error remains primary if closing also fails; the cleanup error is attached as a note. A close failure without a conversion error is propagated. This custom-iterator close fix is included in 2.5.0 and is not included in the 2.4.1 package.

Arrow and dataframe output

Rows accepts Arrow inputs and can also produce Arrow and dataframe outputs:

arrow_batches() emits bounded PyArrow RecordBatches lazily;
to_arrow() materializes a PyArrow Table;
__arrow_c_stream__() exports the standard Arrow PyCapsule stream protocol;
to_pandas() and to_polars() materialize their respective dataframe types;
to_parquet() writes bounded row groups and publishes the local file atomically.

Direct retained Arrow sources can reuse native batches. Joins and aggregates are evaluated first because their output has no equivalent linear source view.

In the 2.5.0 source, arrow_batches(schema=None) and explicit-schema calls both respect custom Rows/Flow iteration. The native-path decision is made when consumption starts, so an iteration method changed after construction but before consumption is observed. Unmodified native inputs can still reuse buffers; custom iteration uses the existing bounded row-to-batch conversion. Without a schema, the first output batch determines the schema for later batches.

from contextlib import closing
import pyarrow as pa
from fpstreams import Rows, rows


class AdjustedRows(Rows):
    def __iter__(self):
        with closing(super().__iter__()) as values:
            for row in values:
                yield {"value": row["value"] + 10}


table = pa.table({"value": [1, 2, 3]})
source = AdjustedRows(rows.from_arrow(table).to_flow())
for schema in (None, table.schema):
    batches = source.arrow_batches(batch_size=2, schema=schema).to_list()
    assert [batch.num_rows for batch in batches] == [2, 1]
    assert pa.Table.from_batches(batches).to_pylist() == [
        {"value": 11}, {"value": 12}, {"value": 13},
    ]

Terminal consumption closes owned iterators on exhaustion, failure, or short circuit. When manually iterating batches, close the returned iterator on early exit. A custom iterator must also close any iterator it opens itself, as in the closing block above. The default-schema iteration fix is included in 2.5.0.

Record transformations

rename(a="x", b="y") renames both top-level fields; names absent from a record are ignored. A destination that duplicates another output field raises ValueError when that record is consumed.

For ordinary Python dictionary rows, with_columns() makes a shallow top-level copy: existing nested values retain their identities. For dataclass rows it uses dataclasses.asdict(), which recursively copies values and can propagate copy errors. Computed selectors read the original row, not earlier additions in the same call. Other row types and columnar adapters may have different conversion behavior.

Dotted string selectors split at each dot. Use expression indexing for a list index or a literal key containing a dot:

from fpstreams import col, rows

record = {"items": [{"id": 42}], "a": {"d.e": 7}}
assert rows([record]).select(
    first=col("items")[0]["id"], dotted=col("a")["d.e"]
).to_list() == [{"first": 42, "dotted": 7}]

Nulls, missing fields, and NaN

In ordinary Python row comparisons, col("x") == None matches a present None, and col("x").isin([1, 2, None]) includes None. This differs from SQL's three-valued null comparisons. A missing x raises SelectionError in these expressions. Explicit dataset filters supplied to an adapter follow that adapter's semantics; do not assume a Python expression and an external filter use the same null rules.

Rows.fill_nulls(x=value) fills both a missing x and a present None; col("x").fill_null(value) first selects x, so a missing field raises SelectionError. fillna and dropna are aliases of fill_nulls and drop_nulls: for ordinary Python rows they handle None, not floating-point NaN. Unlike pandas' similarly named methods, these aliases do not treat NaN as missing.

In ordinary in-memory joins, None keys match one another. In a left join, a right-hand field whose value is None cannot by itself distinguish a matching row containing None from an unmatched row padded with None; retain a known non-null right-side marker when that distinction matters.

For ordinary Python dictionary rows, grouping and joining NaN keys can depend on object identity: repeated use of one NaN object can match, while distinct NaN objects do not. Arrow and Polars adapters can materialize separate Python NaN objects, so grouping after conversion may differ. These examples describe Python row behavior and conversions, not a guarantee for every source, backend, or NaN type.

Safer joins

A duplicate lookup key can multiply records without warning. Declare the relationship you expect when duplicates would indicate bad data:

In the 2.5.0 in-memory Python join path, errors from a key's custom hash or equality operation propagate unchanged. Unhashable keys raise TypeError.

customers = [
    {"customer_id": 1, "name": "Ada"},
    {"customer_id": 2, "name": "Lin"},
]

enriched = (
    rows(orders)
    .join(
        customers,
        on="customer_id",
        how="left",
        validate="m:1",
    )
    .to_list()
)

This many-to-one join permits several orders per customer and requires unique customer keys. Use 1:m for a unique left side, 1:1 when both sides must be unique, or m:m to permit duplicates on both sides. Partitioned joins use the same checks, and errors name the duplicate side and key.

In the ordinary in-memory path, inner, left, semi, and anti joins index the entire right input before reading the left; even an empty left input or a downstream take(1) does not avoid that right-side read. The right input must therefore be finite for these operations to produce output. Semi and anti joins retain qualifying left records, including duplicates. Ordinary right and full joins materialize both sides. Spill and other execution paths have their own resource behavior.

Partitioning is governed by SpillLimits. Its defaults are:

Limit	Default
Rows in one loaded partition	100,000
Serialized bytes in one loaded partition	64 MiB
Matches for one join key	100,000
Total output rows	1,000,000
Recursive repartition levels	3

Oversized partitions are repartitioned with a new deterministic salt. If skew still exceeds the configured depth, or match, group-state, or output expansion exceeds another limit, the operation raises BufferLimitError before loading an unbounded bucket. Temporary files are removed. Pass a custom SpillLimits to join(..., partitions=..., limits=...) or group_by(...).spill(limits=...).

Highly skewed or many-to-many input may exceed these limits even with spill enabled.

Null filtering and lookup failures (2.5.0)

The missing argument is a source-candidate addition and is not in 2.4.1. For drop_nulls(), the default missing="legacy" treats missing selected fields as None. A custom callable selector's SelectionError propagates unchanged. For a RowExpr, errors raised by map() callbacks and directly supplied evaluators also propagate, while missing generated fields are treated as None. Attribute lookup TypeError is preserved as a SelectionError with its original cause; an attribute getter's own SelectionError propagates unchanged. This applies to string field and path lookups as well. A RowExpr subclass that overrides __call__ is treated as a user callback. Errors from overloaded operators propagate, including nested expression failures. Attribute AttributeError and KeyError raised inside a mapping lookup may remain indistinguishable from a missing field. This compatibility policy cannot validate arbitrary row protocols.

Use missing="raise" to propagate every selected lookup failure. None remains null and how="any" / how="all" retain their left-to-right short circuiting, so a later selector may never run. String and generated RowExpr lookups keep their usual SelectionError and original exception in __cause__; explicit callables propagate the exception object they raise. This policy disables the direct-field optimization that treats missing keys as null. With no selectors, both policies inspect the record's values as before.

To migrate a custom protocol, supply a callable that returns None for its explicit absence signal and lets other failures propagate. For example, assume record.lookup("value") raises the protocol's AbsentField only for absence.

def optional_value(record):
    try:
        return record.lookup("value")
    except AbsentField:
        return None


clean = rows(records).drop_nulls(optional_value, missing="raise")

An internal AttributeError or KeyError then propagates unchanged. If the protocol uses the same exception for absence and internal failure, the callable must use another reliable signal supplied by that protocol. missing="raise" does not infer the difference. Existing calls keep missing="legacy" behavior; migration is opt-in. Each reached selector is called once per row.

Methods

fpstreams.Rows

Bases: RowsIOMixin[T], Generic[T]

A lazy record pipeline with joins, grouping, and data-system adapters.

to_columns

to_columns() -> dict[str, list[Any]]

Transpose rows into encounter-ordered, None-padded column lists.

Returns:

Type	Description
`dict[str, list[Any]]`	A field-to-list mapping with one aligned entry for every consumed row.

to_pandas

to_pandas(
    *, batch_size: int = 65536, schema: Any = None
) -> Any

Materialize all rows as a pandas DataFrame through bounded Arrow conversion.

Parameters:

Name	Type	Description	Default
`batch_size`	`int`	Maximum source rows converted in each intermediate Arrow batch.	`65536`
`schema`	`Any`	Optional Arrow schema fixing field order, types, and allowed columns.	`None`

Returns:

Type	Description
`Any`	A pandas DataFrame containing all rows; the Rows pipeline is fully consumed.

to_numpy

to_numpy(
    *selectors: Selector,
    dtype: Any = None,
    copy: bool | None = None,
) -> Any

Materialize selected record values as a two-dimensional NumPy array.

Without selectors, mapping fields are aligned in first-seen order and missing fields become None. With selectors, every selector follows the same field, index, path, expression, and SelectionError behavior as the rest of Rows.

Parameters:

Name	Type	Description	Default
`*selectors`	`Selector`	Optional selectors defining output columns in encounter order.	`()`
`dtype`	`Any`	Optional dtype forwarded to NumPy conversion.	`None`
`copy`	`bool \| None`	NumPy copy policy: `None` copies only as needed, `True` always copies, and `False` requests no copy. NumPy 2.x raises when that request cannot be honored; NumPy 1.x treats it as a best-effort preference.	`None`

Returns:

Type	Description
`Any`	A two-dimensional ndarray with one row per consumed record.

arrow_batches

arrow_batches(
    *, batch_size: int = 65536, schema: Any = None
) -> Flow[Any]

Return a lazy Flow of bounded PyArrow RecordBatch objects.

Parameters:

Name	Type	Description	Default
`batch_size`	`int`	Maximum source rows retained for one emitted RecordBatch.	`65536`
`schema`	`Any`	Optional fixed schema; later fields outside it are rejected.	`None`

Returns:

Type	Description
`Flow[Any]`	A Flow that closes its upstream iterator when exhausted, failed, or short-circuited.

to_arrow

to_arrow(
    *, batch_size: int = 65536, schema: Any = None
) -> Any

Materialize all rows as one PyArrow Table.

Parameters:

Name	Type	Description	Default
`batch_size`	`int`	Maximum rows converted in each intermediate RecordBatch.	`65536`
`schema`	`Any`	Optional schema used for conversion and for an empty result.	`None`

Returns:

Type	Description
`Any`	A Table containing every row; direct Arrow sources may reuse native batches.

polars_batches

polars_batches(
    *, batch_size: int = 65536, schema: Any = None
) -> Flow[Any]

Return a lazy Flow of Polars DataFrames converted from bounded Arrow batches.

Parameters:

Name	Type	Description	Default
`batch_size`	`int`	Maximum source rows represented by each emitted DataFrame.	`65536`
`schema`	`Any`	Optional Arrow schema fixing conversion fields and types.	`None`

Returns:

Type	Description
`Flow[Any]`	A Flow that preserves batch order and inherits upstream cleanup behavior.

to_polars

to_polars(
    *, batch_size: int = 65536, schema: Any = None
) -> Any

Materialize all rows as one Polars DataFrame through Arrow.

Parameters:

Name	Type	Description	Default
`batch_size`	`int`	Maximum rows converted in each intermediate Arrow batch.	`65536`
`schema`	`Any`	Optional Arrow schema fixing conversion fields and types.	`None`

Returns:

Type	Description
`Any`	A non-rechunked Polars DataFrame containing the fully consumed pipeline.

to_csv

to_csv(
    path: str | PathLike[str],
    *,
    fieldnames: Iterable[str] | None = None,
    encoding: str = "utf-8",
    include_header: bool = True,
    extrasaction: Literal["raise", "ignore"] = "raise",
    spreadsheet_safe: bool = False,
    atomic: bool = False,
    if_exists: Literal["replace", "error"] = "replace",
) -> None

Consume rows into a CSV file incrementally and return None.

Parameters:

Name	Type	Description	Default
`path`	`str \| PathLike[str]`	Destination file opened for text overwrite.	required
`fieldnames`	`Iterable[str] \| None`	Output order; inferred from the first row when omitted.	`None`
`encoding`	`str`	Text encoding used to write the destination.	`'utf-8'`
`include_header`	`bool`	Write fieldnames before data rows when true.	`True`
`extrasaction`	`Literal['raise', 'ignore']`	Raise or ignore fields absent from fieldnames.	`'raise'`
`spreadsheet_safe`	`bool`	Prefix formula-like cells, including headers, without changing record field lookup.	`False`
`atomic`	`bool`	Publish after the writer and source close successfully. Writing or source-cleanup failures preserve the old target.	`False`
`if_exists`	`Literal['replace', 'error']`	With atomic=True, replace the target or fail if it already exists.	`'replace'`

to_jsonl

to_jsonl(
    path: str | PathLike[str],
    *,
    encoding: str = "utf-8",
    ensure_ascii: bool = False,
    default: Callable[[Any], Any] | None = None,
    atomic: bool = False,
    if_exists: Literal["replace", "error"] = "replace",
) -> None

Consume rows into an overwritten file as one JSON object per line.

Parameters:

Name	Type	Description	Default
`path`	`str \| PathLike[str]`	Destination JSON Lines file.	required
`encoding`	`str`	Text encoding used to write the destination.	`'utf-8'`
`ensure_ascii`	`bool`	Escape non-ASCII code points when true.	`False`
`default`	`Callable[[Any], Any] \| None`	Optional serializer called for objects the JSON encoder cannot handle.	`None`
`atomic`	`bool`	Publish after the writer and source close successfully. Writing or source-cleanup failures preserve the old target.	`False`
`if_exists`	`Literal['replace', 'error']`	With atomic=True, replace the target or fail if it already exists.	`'replace'`

to_parquet

to_parquet(
    path: str | PathLike[str],
    *,
    if_exists: Literal["error", "replace"] = "error",
    batch_size: int = 65536,
    schema: Any = None,
    compression: Any = "zstd",
    use_dictionary: Any = True,
    write_statistics: Any = True,
    writer_options: Mapping[str, Any] | None = None,
) -> int

Stream rows to a local temporary Parquet file and publish it atomically.

The destination changes only after all bounded row groups are written; failures remove the temporary file.

Parameters:

Name	Type	Description	Default
`path`	`str \| PathLike[str]`	Local destination path; URI-style paths are rejected.	required
`if_exists`	`Literal['error', 'replace']`	Error before consuming rows, or atomically replace the target.	`'error'`
`batch_size`	`int`	Maximum rows converted and written per row group.	`65536`
`schema`	`Any`	Optional Arrow schema, required to write an empty pipeline. Nonempty rows require at least one column to preserve their row count.	`None`
`compression`	`Any`	Compression setting passed to PyArrow ParquetWriter.	`'zstd'`
`use_dictionary`	`Any`	Dictionary-encoding setting passed to ParquetWriter.	`True`
`write_statistics`	`Any`	Column-statistics setting passed to ParquetWriter.	`True`
`writer_options`	`Mapping[str, Any] \| None`	Extra ParquetWriter options except reserved arguments.	`None`

Returns:

Type	Description
`int`	The number of source rows published.

to_db

to_db(
    connect: ConnectionFactory,
    statement: str,
    *,
    parameters: Callable[[T], Any] | None = None,
    batch_size: int = 1000,
) -> int

Submit mapped rows with DB-API executemany batches in one transaction.

Commit after all rows succeed or roll back on error; always close the iterator, cursor, and connection.

Parameters:

Name	Type	Description	Default
`connect`	`ConnectionFactory`	Zero-argument factory for the transaction's connection.	required
`statement`	`str`	Statement passed to cursor.executemany().	required
`parameters`	`Callable[[T], Any] \| None`	Optional callable mapping each row to bound parameters.	`None`
`batch_size`	`int`	Maximum parameter sets in each executemany() call.	`1000`

Returns:

Type	Description
`int`	The number of source rows submitted, independent of cursor.rowcount.

to_sqlite

to_sqlite(
    database: str | PathLike[str],
    table: str,
    *,
    if_exists: Literal[
        "append", "fail", "replace"
    ] = "append",
    conflict: Literal[
        "error", "ignore", "replace"
    ] = "error",
    columns: Iterable[str] | None = None,
    schema: Mapping[str, str] | None = None,
    batch_size: int = 1000,
    timeout: float = 5.0,
    uri: bool = False,
) -> int

Insert rows into a SQLite table in bounded batches and one transaction.

Replace-mode DDL and inserts roll back together; all owned resources close on every exit path. Table and column identifiers use SQLite's ASCII case rules; source record field lookup keeps the configured names. Duplicate aliases for one column raise DuplicateKeyError before inserts.

Parameters:

Name	Type	Description	Default
`database`	`str \| PathLike[str]`	SQLite path or URI passed to sqlite3.connect().	required
`table`	`str`	Destination table identifier, validated and quoted as data.	required
`if_exists`	`Literal['append', 'fail', 'replace']`	Append, fail if present, or transactionally replace the table.	`'append'`
`conflict`	`Literal['error', 'ignore', 'replace']`	Use ordinary INSERT, INSERT OR IGNORE, or INSERT OR REPLACE.	`'error'`
`columns`	`Iterable[str] \| None`	Optional ordered projection; extra source fields are omitted.	`None`
`schema`	`Mapping[str, str] \| None`	Optional mapping from field names to supported SQLite type names.	`None`
`batch_size`	`int`	Maximum bindings in each executemany() call.	`1000`
`timeout`	`float`	Seconds sqlite3 waits for a locked database.	`5.0`
`uri`	`bool`	Interpret database as a SQLite URI when true.	`False`

Returns:

Type	Description
`int`	Source records submitted, including records ignored by SQLite conflicts.

from_csv `staticmethod`

from_csv(
    path: str
    | PathLike[str]
    | TextIO
    | Callable[[], TextIO],
    *,
    encoding: str = "utf-8",
    **format_parameters: Any,
) -> Rows[dict[str, Any]]

Read CSV rows lazily from a path, caller-owned handle, or owned opener.

Parameters:

Name	Type	Description	Default
`path`	`str \| PathLike[str] \| TextIO \| Callable[[], TextIO]`	A path reopened for every execution, an already-open text handle consumed once without being closed, or a zero-argument opener whose returned handle is closed after each execution.	required
`encoding`	`str`	Text encoding used only when fpstreams opens a path.	`'utf-8'`
`**format_parameters`	`Any`	Keyword options forwarded to csv.DictReader, such as dialect, delimiter, or quoting.	`{}`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy dictionaries keyed by the unique CSV header. Open handles can be consumed once; paths and opener functions can be read again.

scan_csv `staticmethod`

scan_csv(
    path: str | PathLike[str],
    *,
    batch_size: int = 65536,
    read_options: Any = None,
    parse_options: Any = None,
    convert_options: Any = None,
    memory_pool: Any = None,
) -> Rows[dict[str, Any]]

Lazily scan typed CSV batches using Arrow conversion.

Unlike from_csv(), this Arrow reader infers non-string types. PyArrow options can fix parsing and conversion behavior when inference is unsuitable. The incremental Arrow reader is single-threaded and freezes inferred types after its first byte block; use read_options or convert_options to control those choices. Query projection runs after conversion of every caller-visible column. Selecting fewer columns does not suppress conversion errors in other columns. Use convert_options to restrict the reader's columns explicitly when that behavior is intended.

Parameters:

Name	Type	Description	Default
`path`	`str \| PathLike[str]`	Local CSV path reopened for each iteration.	required
`batch_size`	`int`	Maximum rows exposed by each retained Arrow batch.	`65536`
`read_options`	`Any`	Optional `pyarrow.csv.ReadOptions`.	`None`
`parse_options`	`Any`	Optional `pyarrow.csv.ParseOptions`.	`None`
`convert_options`	`Any`	Optional `pyarrow.csv.ConvertOptions`.	`None`
`memory_pool`	`Any`	Optional PyArrow memory pool used by the reader.	`None`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Reusable typed rows. This adapter requires the `arrow` extra.

from_jsonl `staticmethod`

from_jsonl(
    path: str
    | PathLike[str]
    | TextIO
    | BinaryIO
    | Callable[[], TextIO | BinaryIO],
    *,
    encoding: str = "utf-8",
    max_record_bytes: int | None = 8 * 1024 * 1024,
) -> Rows[dict[str, Any]]

Read JSON objects lazily from a path, caller-owned handle, or owned opener.

Parameters:

Name	Type	Description	Default
`path`	`str \| PathLike[str] \| TextIO \| BinaryIO \| Callable[[], TextIO \| BinaryIO]`	A path reopened for every execution, an already-open text or binary handle consumed once without being closed, or a zero-argument opener whose returned handle is closed after each execution.	required
`encoding`	`str`	Encoding used for paths and binary handles, and for byte accounting on text handles.	`'utf-8'`
`max_record_bytes`	`int \| None`	Encoded-byte limit per line, or None for no limit.	`8 * 1024 * 1024`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy dictionary rows. Open handles can be consumed once; paths and opener functions can be read again. Duplicate keys and non-object records raise errors during consumption.

from_arrow `staticmethod`

from_arrow(
    source: Any, *, batch_size: int = 65536
) -> Rows[dict[str, Any]]

Adapt an Arrow object or C Stream provider to dictionary rows.

Parameters:

Name	Type	Description	Default
`source`	`Any`	Reusable PyArrow Table/RecordBatch, one-shot RecordBatchReader, or an object implementing `__arrow_c_stream__`. A C Stream provider is imported once at construction and treated as one-shot.	required
`batch_size`	`int`	Maximum rows converted from each Arrow batch slice.	`65536`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy Rows; reader-backed inputs may be consumed only once and are closed afterward.

from_columns `staticmethod`

from_columns(
    columns: Mapping[str, Any], *, batch_size: int = 65536
) -> Rows[dict[str, Any]]

Adapt an explicit mapping of independent columns through a retained Arrow table.

from_numpy `staticmethod`

from_numpy(
    array: Any, *, columns: Iterable[str] | None = None
) -> Rows[dict[str, Any]]

Adapt a two-dimensional NumPy array to replayable dictionary rows.

NumPy conversion happens at construction, but each array row is converted to Python scalar values only when consumed. An existing ndarray is retained by reference, while other array-like inputs follow numpy.asarray conversion semantics.

Parameters:

Name	Type	Description	Default
`array`	`Any`	Two-dimensional ndarray or array-like input accepted by `numpy.asarray`.	required
`columns`	`Iterable[str] \| None`	Unique non-empty string names matching the array width. Defaults to `"0"`, `"1"`, and so on.	`None`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy, replayable Rows whose records follow the retained array's row order.

from_dataframe `staticmethod`

from_dataframe(
    frame: Any,
    *,
    batch_size: int = 65536,
    allow_copy: bool = True,
) -> Rows[dict[str, Any]]

Adapt an object implementing the dataframe interchange protocol through PyArrow.

Parameters:

Name	Type	Description	Default
`frame`	`Any`	Object providing dataframe(), optionally with an Arrow C stream.	required
`batch_size`	`int`	Maximum rows converted from each Arrow batch.	`65536`
`allow_copy`	`bool`	Permit interchange conversion to allocate copied buffers.	`True`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy Rows that perform dataframe-to-Arrow conversion when iterated.

from_polars `staticmethod`

from_polars(
    frame: Any,
    *,
    batch_size: int = 65536,
    maintain_order: bool = True,
    engine: Any = "auto",
) -> Rows[dict[str, Any]]

Adapt an eager Polars DataFrame or batch-collected LazyFrame to dictionary rows.

Parameters:

Name	Type	Description	Default
`frame`	`Any`	Polars DataFrame or LazyFrame to slice or collect.	required
`batch_size`	`int`	Rows requested per eager slice or lazy collection batch.	`65536`
`maintain_order`	`bool`	Preserve LazyFrame row order while collecting batches.	`True`
`engine`	`Any`	Polars engine used only for LazyFrame batch collection.	`'auto'`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy reusable Rows; a LazyFrame is collected again for each iteration.

from_parquet `staticmethod`

from_parquet(
    source: Any,
    *,
    columns: Iterable[str] | None = None,
    filter: Any = None,
    batch_size: int = 65536,
    use_threads: bool = True,
    filesystem: Any = None,
    partitioning: Any = None,
) -> Rows[dict[str, Any]]

Build reusable rows from a fresh PyArrow dataset scanner per iteration.

Parameters:

Name	Type	Description	Default
`source`	`Any`	PyArrow Dataset or dataset source accepted by pyarrow.dataset().	required
`columns`	`Iterable[str] \| None`	Unique projected column names, or None for all columns.	`None`
`filter`	`Any`	PyArrow dataset expression pushed into the scanner.	`None`
`batch_size`	`int`	Maximum rows requested from each scanner batch.	`65536`
`use_threads`	`bool`	Allow the PyArrow scanner to use worker threads.	`True`
`filesystem`	`Any`	Optional PyArrow filesystem for resolving the source.	`None`
`partitioning`	`Any`	Optional dataset partitioning specification.	`None`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy dictionary rows with projection and filtering performed by PyArrow.

from_db `staticmethod`

from_db(
    connect: ConnectionFactory,
    query: str,
    parameters: DBParameters = None,
    *,
    batch_size: int = 1000,
) -> Rows[dict[str, Any]]

Build a reiterable DB-API query source that owns its connections and cursors.

Parameters:

Name	Type	Description	Default
`connect`	`ConnectionFactory`	Zero-argument factory called once per iteration for a new connection.	required
`query`	`str`	Statement executed by each newly opened cursor.	required
`parameters`	`DBParameters`	Optional mapping or positional values passed to cursor.execute().	`None`
`batch_size`	`int`	Maximum rows requested by each cursor.fetchmany() call.	`1000`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy rows that close the cursor and connection on exhaustion, error, or early stop.

from_sqlite `staticmethod`

from_sqlite(
    database: str | PathLike[str],
    query: str,
    parameters: DBParameters = None,
    *,
    batch_size: int = 1000,
    timeout: float = 5.0,
    uri: bool = False,
) -> Rows[dict[str, Any]]

Build a reiterable SQLite query source that owns one connection per iteration.

Parameters:

Name	Type	Description	Default
`database`	`str \| PathLike[str]`	SQLite path or URI passed to sqlite3.connect().	required
`query`	`str`	Statement executed by each newly opened cursor.	required
`parameters`	`DBParameters`	Optional mapping or positional values passed to cursor.execute().	`None`
`batch_size`	`int`	Maximum rows requested by each cursor.fetchmany() call.	`1000`
`timeout`	`float`	Seconds sqlite3 waits for a locked database.	`5.0`
`uri`	`bool`	Interpret database as a SQLite URI when true.	`False`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy dictionary rows that close their cursor and connection after iteration.

to_flow

to_flow() -> Flow[T]

Return the underlying Flow without copying data or changing the plan.

Returns:

Type	Description
`Flow[T]`	The same lazy Flow that owns this Rows view's source and operations.

explain

explain(
    terminal: TerminalName = "iterate",
) -> PlanExplanation

Describe this record pipeline's execution plan without consuming it.

Parameters:

Name	Type	Description	Default
`terminal`	`TerminalName`	Terminal operation included in engine selection and validation.	`'iterate'`

Returns:

Type	Description
`PlanExplanation`	The same structured explanation produced by the underlying Flow.

to_list

to_list() -> list[T]

Execute the record pipeline and collect its rows.

Returns:

Type	Description
`list[T]`	A list containing the consumed results in encounter order.

run_with_report

run_with_report(
    terminal: str, /, *args: Any, **kwargs: Any
) -> ExecutionResult[Any]

Run one eager Rows terminal and return its value with query-owned metrics.

count

count() -> int

Consume the pipeline and count every emitted row.

Returns:

Type	Description
`int`	The number of rows remaining after all lazy transformations.

with_engine

with_engine(engine: Engine) -> Rows[T]

Return equivalent lazy Rows requesting auto, Python, or native Flow execution.

concat

concat(*others: Iterable[T] | Flow[T] | Rows[T]) -> Rows[T]

Emit these rows followed by each supplied record source in order.

Concatenation is lazy and preserves every input record as-is. It does not align fields, fill missing values, or infer a common schema.

Parameters:

Name	Type	Description	Default
`*others`	`Iterable[T] \| Flow[T] \| Rows[T]`	Rows, Flow, or record iterables opened only after earlier inputs finish.	`()`

Returns:

Type	Description
`Rows[T]`	A Rows view over the ordered concatenation, or this same view when no source is given.

first

first() -> T

Return the first row and close upstream without requesting an unnecessary tail.

Returns:

Type	Description
`T`	The first emitted row.

Raises:

Type	Description
`EmptyFlowError`	If the pipeline emits no rows.

last

last() -> T

Consume the pipeline and return its final row, raising EmptyFlowError when empty.

Returns:

Type	Description
`T`	The last emitted row.

take

take(count: int) -> Rows[T]

Return a lazy prefix that stops and closes upstream after at most count rows.

Parameters:

Name	Type	Description	Default
`count`	`int`	Nonnegative maximum number of rows to emit.	required

Returns:

Type	Description
`Rows[T]`	New Rows preserving encounter order; zero emits nothing.

skip

skip(count: int) -> Rows[T]

Return lazy rows after discarding the first count upstream items.

Parameters:

Name	Type	Description	Default
`count`	`int`	Nonnegative number of rows to consume without emitting.	required

Returns:

Type	Description
`Rows[T]`	New Rows containing the remaining encounter-ordered rows.

unique_by

unique_by(selector: Selector) -> Rows[T]

Keep the first row for each distinct selected key in encounter order.

Parameters:

Name	Type	Description	Default
`selector`	`Selector`	Field, path, index, expression, or callable producing a hashable key.	required

Returns:

Type	Description
`Rows[T]`	Lazy Rows whose later duplicate keys are omitted.

filter

filter(predicate: Callable[[T], bool]) -> Rows[T]

Keep rows for which predicate returns a truthy result.

The predicate runs lazily in encounter order, and the parent Rows pipeline remains unchanged.

Parameters:

Name	Type	Description	Default
`predicate`	`Callable[[T], bool]`	Callable evaluated once for each upstream row reached.	required

Returns:

Type	Description
`Rows[T]`	New lazy Rows containing only matching rows.

map

map(function: Callable[[T], R]) -> Flow[R]

Map rows into an ordinary Flow whose output may have any shape.

Parameters:

Name	Type	Description	Default
`function`	`Callable[[T], R]`	Lazily transforms each row into one output value.	required

Returns:

Type	Description
`Flow[R]`	A Flow of transformed values. Call row operations on that Flow to re-enter Rows.

flat_map

flat_map(function: Callable[[T], Iterable[R]]) -> Flow[R]

Map rows to iterables and flatten them into an ordinary Flow.

Parameters:

Name	Type	Description	Default
`function`	`Callable[[T], Iterable[R]]`	Lazily transforms each row into zero or more output values.	required

Returns:

Type	Description
`Flow[R]`	A Flow of flattened values. Call row operations on that Flow to re-enter Rows.

where

where(
    predicate: Callable[[T], bool] | None = None,
    **equalities: Any,
) -> Rows[T]

Require the optional predicate and every named field equality.

Named fields are compiled once, then selected lazily from each consumed row.

Parameters:

Name	Type	Description	Default
`predicate`	`Callable[[T], bool] \| None`	Optional callable that must return truthy for a row.	`None`
`**equalities`	`Any`	Top-level or dotted field paths mapped to required values.	`{}`

Returns:

Type	Description
`Rows[T]`	New lazy Rows containing rows that satisfy all supplied conditions.

with_columns

with_columns(**columns: Selector) -> Rows[dict[str, Any]]

Copy each row and add or replace fields evaluated against the original row.

Parameters:

Name	Type	Description	Default
`**columns`	`Selector`	Output field names mapped to selectors or RowExpr values.	`{}`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy dictionary Rows; computed columns do not observe earlier additions.

rename

rename(**columns: str) -> Rows[dict[str, Any]]

Rename top-level fields while rejecting collisions in each output record.

Parameters:

Name	Type	Description	Default
`**columns`	`str`	Existing field names mapped to nonempty destination names.	`{}`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy copied dictionaries; unmapped fields retain their names and order.

drop

drop(*columns: str) -> Rows[dict[str, Any]]

Copy each row without the named top-level fields.

Parameters:

Name	Type	Description	Default
`*columns`	`str`	Field names to omit; absent names are ignored.	`()`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy dictionary Rows preserving the order of retained fields.

cast

cast(
    **columns: Callable[[Any], Any],
) -> Rows[dict[str, Any]]

Convert existing named fields with one callable per field.

Parameters:

Name	Type	Description	Default
`**columns`	`Callable[[Any], Any]`	Field names mapped to callable value converters.	`{}`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy copied dictionaries; a missing field raises SelectionError when consumed.

fill_nulls

fill_nulls(**replacements: object) -> Rows[dict[str, Any]]

Replace missing or None named fields with constants or RowExpr results.

Parameters:

Name	Type	Description	Default
`**replacements`	`object`	Field names mapped to literal values or row expressions.	`{}`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy copied dictionaries; non-None existing values are preserved.

drop_nulls

drop_nulls(
    *selectors: Selector,
    how: Literal["any", "all"] = "any",
    missing: Literal["legacy", "raise"] = "legacy",
) -> Rows[T]

Drop rows according to None values in selected fields or the whole record.

Parameters:

Name	Type	Description	Default
`*selectors`	`Selector`	Fields to inspect; omitted means every field in each record.	`()`
`how`	`Literal['any', 'all']`	"any" drops on one null; "all" requires every inspected value to be null.	`'any'`
`missing`	`Literal['legacy', 'raise']`	"legacy" retains historical missing-field handling. "raise" propagates lookup failures; use a callable returning None to define actual absence for custom row protocols. None values remain null.	`'legacy'`

Returns:

Type	Description
`Rows[T]`	Lazy Rows using the selected null and lookup-failure policies.

explode

explode(
    selector: Selector,
    *,
    into: str | None = None,
    outer: bool = False,
) -> Rows[dict[str, Any]]

Expand a selected iterable into one copied row per element.

Parameters:

Name	Type	Description	Default
`selector`	`Selector`	Selector returning a non-string iterable or None.	required
`into`	`str \| None`	Output field name; required for non-top-level selectors.	`None`
`outer`	`bool`	Emit one row with None when the selected value is None or empty.	`False`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy flattened dictionary Rows that close upstream on downstream stop.

unnest

unnest(
    column: str, *, prefix: str = ""
) -> Rows[dict[str, Any]]

Replace one top-level nested record with its fields.

Parameters:

Name	Type	Description	Default
`column`	`str`	Non-dotted field name containing a supported record-like value.	required
`prefix`	`str`	Text prepended to every promoted nested field.	`''`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy copied dictionaries; output-name collisions raise DuplicateKeyError.

unpivot

unpivot(
    *columns: str,
    names_to: str = "variable",
    values_to: str = "value",
) -> Rows[dict[str, Any]]

Convert selected top-level fields from wide form into name/value rows.

Parameters:

Name	Type	Description	Default
`*columns`	`str`	Unique fields expanded in the given order.	`()`
`names_to`	`str`	Noncolliding output field for each former column name.	`'variable'`
`values_to`	`str`	Noncolliding output field for each former column value.	`'value'`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy Rows emitting len(columns) records per input row.

pivot

pivot(
    *,
    index: Selector | tuple[Selector, ...],
    columns: Selector,
    values: Selector,
    aggregate: str | Callable[[Any, Any], Any] = "error",
    fill: Any = None,
) -> Rows[dict[str, Any]]

Materialize long-form rows into encounter-ordered wide records.

Parameters:

Name	Type	Description	Default
`index`	`Selector \| tuple[Selector, ...]`	Selector or selector tuple defining each output row and its key fields.	required
`columns`	`Selector`	Selector whose values become dynamic output field names.	required
`values`	`Selector`	Selector producing each pivot cell value.	required
`aggregate`	`str \| Callable[[Any, Any], Any]`	Duplicate-cell policy: error, first, last, sum, or a reducer callable.	`'error'`
`fill`	`Any`	Value inserted for missing cells among discovered columns.	`None`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy pipeline that builds the full pivot only when consumed.

select

select(
    *selectors: str | int, **named: Selector
) -> Rows[dict[str, Any]]

Project positional and named selectors into new dictionaries.

Parameters:

Name	Type	Description	Default
`*selectors`	`str \| int`	String paths or integer indexes; output names are derived automatically.	`()`
`**named`	`Selector`	Explicit output names mapped to any supported selector.	`{}`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy projected Rows; duplicate derived or explicit names are rejected immediately.

sort_by

sort_by(
    selector: Selector,
    *,
    reverse: bool = False,
    buffer_size: int | None = None,
    tempdir: str | PathLike[str] | None = None,
) -> Rows[T]

Sort rows by a selected key, in memory or through bounded external runs.

Parameters:

Name	Type	Description	Default
`selector`	`Selector`	Field, path, index, expression, or callable producing the sort key.	required
`reverse`	`bool`	Emit descending order when true.	`False`
`buffer_size`	`int \| None`	None for full in-memory sort, or positive rows per spilled run.	`None`
`tempdir`	`str \| PathLike[str] \| None`	Parent directory for automatically cleaned external-sort files.	`None`

Returns:

Type	Description
`Rows[T]`	Lazy stably sorted Rows.

external_sort_by

external_sort_by(
    selector: Selector,
    *,
    reverse: bool = False,
    buffer_size: int = 100000,
    tempdir: str | PathLike[str] | None = None,
) -> Rows[T]

Sort rows stably with bounded in-memory runs and temporary files.

Each run holds at most buffer_size rows; the lazy merge closes upstream and removes temporary files after completion, failure, or downstream short-circuit.

Parameters:

Name	Type	Description	Default
`selector`	`Selector`	Field, path, index, expression, or callable producing the sort key.	required
`reverse`	`bool`	Emit descending order when true.	`False`
`buffer_size`	`int`	Positive maximum rows held in each sorted run.	`100000`
`tempdir`	`str \| PathLike[str] \| None`	Parent directory for automatically cleaned run files.	`None`

Returns:

Type	Description
`Rows[T]`	Lazy externally sorted Rows.

aggregate

aggregate(
    **aggregations: Collector[Any, Any, Any],
) -> Rows[dict[str, Any]]

Run named Aggregators and return a one-row pipeline.

The computation is deferred and produces a Rows pipeline containing one result record.

Parameters:

Name	Type	Description	Default
`**aggregations`	`Collector[Any, Any, Any]`	Named collectors evaluated during the same traversal.	`{}`

Returns:

Type	Description
`Rows[dict[str, Any]]`	A lazy one-row pipeline containing the named results.

group_by

group_by(
    *selectors: Selector, **named: Selector
) -> GroupedRows[T]

Describe grouped aggregation by positional and/or explicitly named selectors.

Parameters:

Name	Type	Description	Default
`*selectors`	`Selector`	Keys named from field paths or as key_N for other selector types.	`()`
`**named`	`Selector`	Explicit output key names mapped to supported selectors.	`{}`

Returns:

Type	Description
`GroupedRows[T]`	GroupedRows configuration; no source rows are read until aggregate() is consumed.

group_by_sorted

group_by_sorted(
    *selectors: Selector, **named: Selector
) -> GroupedRows[T]

Group adjacent records whose selected keys are already ascending.

Keys must be exact int, str, bytes, or nonempty flat tuples of these types. Their shape must stay fixed. Type and order checks cover consumed rows, including one row of lookahead. No sorting is performed.

Parameters:

Name	Type	Description	Default
`*selectors`	`Selector`	Keys named with the same rules as group_by().	`()`
`**named`	`Selector`	Explicit output key names and selectors.	`{}`

Returns:

Type	Description
`GroupedRows[T]`	A deferred grouping using current aggregate state plus one lookahead row. Aggregators that collect values can retain additional state. Spill is unsupported.

join_sorted

join_sorted(
    other: Iterable[Any] | Flow[Any] | Rows[Any],
    *,
    on: JoinSelector | None = None,
    left_on: JoinSelector | None = None,
    right_on: JoinSelector | None = None,
    how: str = "inner",
    suffix: str = "_right",
    validate: JoinValidation = "m:m",
    max_right_group_rows: int = 100000,
    max_matches_per_left: int = 100000,
    max_output_rows: int = 1000000,
) -> Rows[dict[str, Any]]

Join ascending inputs while buffering one right key group.

Keys must be exact int, str, bytes, or flat tuples of these types. Inner/left output columns come from the first right record; later new columns fail. Ordering and cardinality checks cover consumed rows only. Finite row budgets raise BufferLimitError before the excess output. Added in version 2.2.0.

join

join(
    other: Iterable[Any] | Flow[Any] | Rows[Any],
    *,
    on: JoinSelector | None = None,
    left_on: JoinSelector | None = None,
    right_on: JoinSelector | None = None,
    how: str = "inner",
    suffix: str = "_right",
    validate: JoinValidation = "m:m",
    partitions: int | None = None,
    tempdir: str | PathLike[str] | None = None,
    limits: SpillLimits | None = None,
) -> Rows[dict[str, Any]]

Join this record pipeline with another source.

Joins are lazy and preserve stable input order. Inner, left, semi, and anti joins stream the left source after indexing the right source. Right and full joins materialize both sides. Set partitions to use bounded-memory hash partitioning through temporary files.

Parameters:

Name	Type	Description	Default
`other`	`Iterable[Any] \| Flow[Any] \| Rows[Any]`	The record iterable, Flow, or Rows pipeline to join.	required
`on`	`JoinSelector \| None`	A selector used for both left and right keys.	`None`
`left_on`	`JoinSelector \| None`	The left key selector when the two sides use different fields.	`None`
`right_on`	`JoinSelector \| None`	The right key selector when the two sides use different fields.	`None`
`how`	`str`	One of inner, left, right, full, semi, or anti.	`'inner'`
`suffix`	`str`	Text appended to conflicting right-side field names.	`'_right'`
`validate`	`JoinValidation`	Expected key cardinality. 1:m requires unique left keys, m:1 requires unique right keys, 1:1 requires both, and m:m permits duplicates on both sides.	`'m:m'`
`partitions`	`int \| None`	Number of hash partitions for bounded-memory execution. Must be between 2 and 256.	`None`
`tempdir`	`str \| PathLike[str] \| None`	Parent directory for temporary partition files. Requires partitions.	`None`
`limits`	`SpillLimits \| None`	Finite partition, match, and output budgets for spilled execution.	`None`

Returns:

Type	Description
`Rows[dict[str, Any]]`	Lazy dictionary Rows that execute the selected in-memory or spilled join when consumed.

Raises:

Type	Description
`ValueError`	If selectors, modes, partition options, or key cardinality are invalid.
`TypeError`	If a key is unhashable or spilled data cannot be serialized.
`DuplicateKeyError`	If suffixing would create an ambiguous output field.
`BufferLimitError`	If spilled execution exceeds a configured resource budget.

GroupedRows

group_by() returns a grouped plan. Call aggregate() directly, or call spill() first to use partitioned temporary storage.

Ordinary Python grouping is global, not adjacent: keys in separated rows can belong to the same group. In ordinary dictionary-row samples a None key is kept, while a missing selected key raises SelectionError. group_by_sorted() instead groups adjacent ascending keys and applies the constraints below.

In the 2.5.0 in-memory Python grouping path, errors from a key's custom hash or equality operation propagate unchanged. Unhashable keys raise TypeError.

Since 2.2.0, Python grouping uses the current selector if a source or callback changes its code or closure. The previous field shortcut could keep using an old field and merge distinct groups.

Grouping with one collector also uses the current step after truth-testing a custom done result. Temporary states and unused keys are released at the same points as in the general collector path, including when the output iterator is closed early. Release callbacks can therefore affect the next row or finisher as expected.

fpstreams.tabular.GroupedRows

Bases: Generic[T]

A deferred grouping that chooses in-memory or partitioned aggregation.

spill

spill(
    partitions: int = 32,
    *,
    tempdir: str | PathLike[str] | None = None,
    limits: SpillLimits | None = None,
) -> GroupedRows[T]

Return grouping configured to aggregate through bounded temporary partitions.

Parameters:

Name	Type	Description	Default
`partitions`	`int`	Hash-partition count from 2 through 256.	`32`
`tempdir`	`str \| PathLike[str] \| None`	Parent directory for automatically removed spill files.	`None`
`limits`	`SpillLimits \| None`	Partition, group-state, output, and repartition budgets.	`None`

Returns:

Type	Description
`GroupedRows[T]`	A new GroupedRows configuration; call aggregate() to obtain a lazy pipeline.

aggregate

aggregate(
    **aggregations: Collector[Any, Any, Any],
) -> Rows[dict[str, Any]]

Compute named aggregations independently for each group.

Grouping and aggregation remain deferred until the returned Rows pipeline is consumed.

Parameters:

Name	Type	Description	Default
`**aggregations`	`Collector[Any, Any, Any]`	Named collectors evaluated during the same traversal.	`{}`

Returns:

Type	Description
`Rows[dict[str, Any]]`	A lazy row pipeline containing one aggregate record per group.

Grouping sorted records

group_by_sorted() uses the same key names and .aggregate() API as group_by(). The input must already be ascending by exact builtin int, str, bytes, or nonempty flat tuple keys with a fixed type at each position. It checks consumed rows and one row of lookahead; an unordered or unsupported key raises during consumption.

from fpstreams import agg, flow

result = (
    flow([{"region": "eu", "amount": 2}, {"region": "eu", "amount": 3}])
    .rows().group_by_sorted("region").aggregate(total=agg.sum("amount"))
    .to_list()
)
assert result == [{"region": "eu", "total": 5}]

Added in 2.2.0. It buffers no complete group. Aggregators that collect values still grow their own state; .spill() raises ValueError in this mode. Ordinary group_by() is unchanged.

Joining sorted records

join_sorted(other, on=..., how="inner", validate="m:m") accepts ascending inputs and supports inner, left, semi, and anti joins. It preserves left order and the order of matching right records. Both sides must contain records, including semi and anti joins. The supported key types are the same as group_by_sorted().

The keyword-only budgets are max_right_group_rows=100_000, max_matches_per_left=100_000, and max_output_rows=1_000_000. Each must be a positive integer. Exceeding a budget raises BufferLimitError; already yielded rows remain yielded. Semi joins enumerate at most one match per left row; anti output does not count as a match. The right-group limit applies to every mode.

Inner and left joins take the right column list from the first right record. Later records may omit columns but cannot introduce new ones. This restriction allows output to start without scanning the entire right source. Suffixes and shared key columns follow ordinary join() rules. Validation supports m:m, m:1, 1:m, and 1:1, and checks the consumed prefix. Neither method infers sorting or changes ordinary join() execution. This API was added in 2.2.0.

Like merge_sorted(), a sorted join acquires source iterators in its composed branches before pulling rows. It rejects iterator aliases and closes acquired iterators once; source-opening callbacks therefore run before key callbacks. Before emitting a match, it buffers the matching right key group and may read the first row of the next group. A short-circuited consumer only checks ordering and cardinality for the rows actually consumed, including any lookahead.

Rows

Inspect field inputs (2.5.0)

Entering and leaving Rows

Creating rows

Bounded JSONL and spreadsheet-safe CSV

Column collection and custom iteration (2.5.0)

Arrow and dataframe output

Record transformations

Nulls, missing fields, and NaN

Safer joins

Null filtering and lookup failures (2.5.0)

Methods

fpstreams.Rows

to_columns

to_pandas

to_numpy

arrow_batches

to_arrow

polars_batches

to_polars

to_csv

to_jsonl

to_parquet

to_db

to_sqlite

from_csv staticmethod

scan_csv staticmethod

from_jsonl staticmethod

from_arrow staticmethod

from_columns staticmethod

from_numpy staticmethod

from_dataframe staticmethod

from_polars staticmethod

from_parquet staticmethod

from_db staticmethod

from_sqlite staticmethod

to_flow

explain

to_list

run_with_report

count

with_engine

concat

first

last

take

skip

unique_by

filter

map

flat_map

where

with_columns

rename

drop

cast

fill_nulls

drop_nulls

explode

unnest

unpivot

pivot

select

sort_by

external_sort_by

aggregate

group_by

group_by_sorted

join_sorted

join

GroupedRows

fpstreams.tabular.GroupedRows

spill

aggregate

Grouping sorted records

Joining sorted records

from_csv `staticmethod`

scan_csv `staticmethod`

from_jsonl `staticmethod`

from_arrow `staticmethod`

from_columns `staticmethod`

from_numpy `staticmethod`

from_dataframe `staticmethod`

from_polars `staticmethod`

from_parquet `staticmethod`

from_db `staticmethod`

from_sqlite `staticmethod`