Logical plan best practices: should I do pre-work at plan time, or Exec time?

[rickbatka]

I'm building a TableProvider that represents a custom external data source that stores data in partitions in a consistent hash ring. The partitions are trivial to discern from the filters provided.

I'm trying to implement filter pushdown and I can't seem to find good "best practices" guidance on how much work should be done in the logical versus physical stages.

1. Should I determine which shards I need to scan in my logical plan and pass it into the new physical plan I create in the constructor? Or should the logical plan just hand off the raw filters and leave it to the physical plan to sort out?
2. Should I represent each shard scan as a separate Exec node in the physical plan? For example, a logical plan could determine it needs to talk to shards 1, 3, and 5 and therefore create 3 Exec nodes - for each individual scan from a shard. Or should I treat my sharding as a black box, and just put all this logic into the single Exec node to be determined at runtime?
3. In general, how much "pre-work" should I do in the logical plan? As much as possible? As little as possible?

Any links to readings or presentations on this subject would be appreciated.

Thanks in advance!

[ryux1]

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:

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.