## engines.run_sharded()


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


Usage

``` python
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](engines.run_sharded.md#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](unique.md#sverify.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](unique.md#sverify.unique) column set has two equal rows; or `keys` or [unique](unique.md#sverify.unique) name unknown columns.