engines.run_sharded()

The plan’s result on all the data, computed shard by shard on Polars.

Usage

engines.run_sharded(plan: str | dict, shards: Sequence[Mapping[str, pa.Table]], sharded: Sequence[str], broadcast: Mapping[str, pa.Table] | None = None, keys: Mapping[str, Sequence[str]] | None = None, unique: Mapping[str, Sequence[Sequence[str]]] | None = None, float_order: FloatOrder = 'exact', max_workers: int | None = None, tolerance: Literal[False] = False) -> pa.Table
 
engines.run_sharded(plan: str | dict, shards: Sequence[Mapping[str, pa.Table]], sharded: Sequence[str], broadcast: Mapping[str, pa.Table] | None = None, keys: Mapping[str, Sequence[str]] | None = None, unique: Mapping[str, Sequence[Sequence[str]]] | None = None, float_order: FloatOrder = 'exact', max_workers: int | None = None, tolerance: Literal[True]) -> Sharded

This is sverify.engines.run_sharded.

Each shard maps every table of sharded to its rows on that shard; broadcast holds the other tables. The result equals the plan’s on the concatenated shards, as a bag (row order is not kept), unless float_order="any" was needed: sums of doubles then round as the split makes them, within a proved tolerance that tolerance=True reports.

Before running, the split keys and unique column sets are checked on the data.

Parameters

plan: str | dict: A Substrait JSON plan, as a string or parsed.
shards: Sequence[Mapping[str, pa.Table]]: One mapping per shard, from each table of sharded to its rows on that shard.
sharded: Sequence[str]: The names of the tables split across shards.
broadcast: Mapping[str, pa.Table] | None = None: The other tables the plan reads, each the same on every shard. Default: none.
keys: Mapping[str, Sequence[str]] | None = None: Split keys, as a mapping from sharded table name to columns: rows with equal values on them are always on the same shard. Default: none.
unique: Mapping[str, Sequence[Sequence[str]]] | None = None: Keys of whole tables, as a mapping from table name to a list of column sets. They let windows run on each shard. Default: none.
float_order: FloatOrder = "exact": "exact" (the default) refuses SUM and AVG of doubles; "any" splits them anyway, within a proved tolerance.
max_workers: int | None = None: The number of threads running shards in parallel. Default: ThreadPoolExecutor’s default.
tolerance: bool = False: Return a Sharded: the result, with the tolerance τ of each split double per row (see the module docstring). Default: False.

Returns

pa.Table | Sharded: The plan’s result, as an Arrow table, or with tolerance=True a
Sharded holding it.

Raises

ValueError: The plan’s result would depend on how the rows are split (sverify.partition.NotShardable, a ValueError, saying why); there are no shards; a shard’s tables are not exactly sharded; rows with equal split keys are on two shards; a unique column set has two equal rows; or keys or unique name unknown columns.