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.

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 with query column pruning
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. With a plain local path and default reader options, direct select() queries can prune columns during Arrow CSV conversion. 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.

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.

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.

Safer joins

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

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.

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.

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,
) -> 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 cells that spreadsheet software may execute.	False

to_jsonl

to_jsonl(
    path: str | PathLike[str],
    *,
    encoding: str = "utf-8",
    ensure_ascii: bool = False,
) -> 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

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.	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.

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 with optional query-level column pruning.

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 column pruning intentionally avoids conversion work, including conversion errors, in columns the query does not read.

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"
) -> 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'

Returns:

Type	Description
Rows[T]	Lazy Rows; a missing selected field is treated as None.

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: Aggregator,
) -> 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	Aggregator	Named aggregators 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.

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.

In the unreleased version, 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. These fixes are not included in 2.1.0.

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: Aggregator,
) -> 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	Aggregator	Named aggregators evaluated during the same traversal.	{}

Returns:

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