GitHub user ryux1 added a comment to the discussion: Logical plan best practices: should I do pre-work at plan time, or Exec time?
The clean boundary is usually: 1. classify supported predicates in `TableProvider::supports_filters_pushdown`; 2. choose the shards in `TableProvider::scan`, using the `filters` passed to it; and 3. return an `ExecutionPlan` whose output partitions represent the selected shards. I would not add a custom logical node just to carry shard selection. `scan` is already the logical-to-physical boundary for a table source: it receives projection, pushed filters, and limit and returns the physical scan. It is async, so this is also the appropriate place to resolve routing metadata if the ring cannot be read synchronously. For the physical shape, a single custom `ShardScanExec` with one output partition per selected shard is generally simpler than one Exec node per shard: ```text ShardScanExec shards = [1, 3, 5] output_partitioning = UnknownPartitioning(3) execute(0) -> stream shard 1 execute(1) -> stream shard 3 execute(2) -> stream shard 5 ``` DataFusion schedules those output partitions independently, so this still gives parallel shard reads. Use separate child Exec nodes plus `UnionExec` when the shards genuinely need different physical subplans, not merely because they are separate endpoints. One important detail is to keep **partition pruning** separate from **predicate evaluation**. If `key = 42` tells you which shard can contain the row, but that shard also contains other keys, pruning the other shards has not fully evaluated the predicate. Report that filter as `Inexact` and let DataFusion retain a `FilterExec` above the scan (or evaluate it remotely as well). Report `Exact` only when every row returned by the provider is guaranteed to satisfy the predicate. Reporting `Exact` for routing alone can return incorrect rows. I would therefore keep these phases: - `supports_filters_pushdown`: cheap, synchronous capability classification; no network work. - `scan`: inspect the accepted expressions, compute the selected shard list, and construct plan properties that describe the real partition count/order. - `execute(partition)`: open the network stream for exactly that selected shard. Avoid starting reads before the returned stream is polled, and make dropping the stream cancel the request. Use `UnknownPartitioning(n)` unless the ring distribution really satisfies DataFusion's declared hash-partitioning semantics. A custom consistent hash ring should not be advertised as `Hash(...)` merely because it is hash based. The current trait documentation describes `scan` as responsible for scanning source partitions in a streaming, parallelized fashion and defines the `Exact`/`Inexact` contract: - https://github.com/apache/datafusion/blob/main/datafusion/session/src/table.rs - https://github.com/apache/datafusion/blob/main/datafusion/physical-plan/src/execution_plan.rs - `UnionExec` is a useful comparison for mapping an output partition index to one underlying input: https://github.com/apache/datafusion/blob/main/datafusion/physical-plan/src/union.rs This keeps optimizer-visible facts fixed at plan construction while deferring the expensive data access to execution. GitHub link: https://github.com/apache/datafusion/discussions/18156#discussioncomment-18322240 ---- This is an automatically sent email for [email protected]. To unsubscribe, please send an email to: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
