Push down preferred sorts into TableScan logical plan node

[adriangb]

• Part of Control ordering of file opened based on statistics #17271
• See how this all fits together in Add TableProvider::scan_with_args to support pushdown sorting #17273 (comment).

This will enable TableProvider's to produce files in an order and partitioning that optimizes query execution, e.g. to make a TopK operator stop earlier via dynamic filters or to completely optimize away a sort if the files can be ordered to do so.

Note that this is:

1. A preference and not an absolute.
2. Does not remove the sort node.
   Because physical optimizer rules should remove unnecessary sorts later on.
   This avoids complexity of negotiating sort order with the TableProvider: ExecutionPlan (the thing TableProvider returns) already has APIs to negotiate sort orders.
   So TableProvider can encode into the ExecutionPlan that it returns that "the sort is completely handled in the scan" and then phyiscal optimizer rules will remove the SortExec.

[adriangb]

@berkaysynnada do you think this is the right information to pass down? Or is there a world where it makes sense to pass down some sort of "equivalence" information?

cc @alamb

[alamb]

I also think @suremarc and @ozankabak may be interested in this

[alamb]

A more future-proof API (that we could change the internal representation) might be something like

/// Preferred ordering
///
/// Preferred orderings can potentially help DataFusion optimize queries, even in cases
/// when the output does not completely follow that order. This is information passed 
/// to the scan about what might help. 
///
/// For example, a query with `ORDER BY time DESC LIMIT 10`, DataFusion's dynamic
/// predicates and TopK operator will work better if the data is roughly ordered by descending
/// time (more recent data first)
struct PreferredOrdering {
  exprs: Vec<SortExpr>
}

And then change this API to

Suggested change

	/// Optional preferred ordering for the scan
	pub preferred_ordering: Option<Vec<SortExpr>>,
	/// Optional preferred ordering for the scan
	pub preferred_ordering: Option<PreferredOrdering>,

[berkaysynnada]

@berkaysynnada do you think this is the right information to pass down? Or is there a world where it makes sense to pass down some sort of "equivalence" information?

cc @alamb

When we are registering the sources, we can provide multiple orderings if the table supports them. However, the requirements are singular, and I don't think there would be any meaning in ordering the table for both col_a and col_b simultaneously. So, I've always thought that requirements need only one ordering, but specs should be capable of having multiple orderings. So there isn't any obvious advantage of using equivalences here, IMO

[alamb]

🤖 ./gh_compare_branch_bench.sh Benchmark Script Running
Linux aal-dev 6.14.0-1014-gcp #15~24.04.1-Ubuntu SMP Fri Jul 25 23:26:08 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing add-preferred-ordering-to-table-scan (c84cf55) to b6a8a0e diff
BENCH_NAME=sql_planner
BENCH_COMMAND=cargo bench --bench sql_planner
BENCH_FILTER=
BENCH_BRANCH_NAME=add-preferred-ordering-to-table-scan
Results will be posted here when complete

[alamb]

I think this looks good but I was confused that the tests don't seem to show the preferred ordering. I think we should fix those tests before merging -- I also expect it to show that some of the pushdown isn't working quite as expected (aka pushing through a projection or filter)

I also recommend putting the prefered sort expressions in their own struct, but that is not required in my mind.

As I understand the plan, in the next few PRs, @adriangb will update the various APIs so that this preferred sort is provided to TableProvider::scan (really via scan_with_args)

I also wonder if we should wait for the DataFusion 50 release before merging this or if it is ok to merge now.

[alamb]

A more future-proof API (that we could change the internal representation) might be something like

/// Preferred ordering
///
/// Preferred orderings can potentially help DataFusion optimize queries, even in cases
/// when the output does not completely follow that order. This is information passed 
/// to the scan about what might help. 
///
/// For example, a query with `ORDER BY time DESC LIMIT 10`, DataFusion's dynamic
/// predicates and TopK operator will work better if the data is roughly ordered by descending
/// time (more recent data first)
struct PreferredOrdering {
  exprs: Vec<SortExpr>
}

And then change this API to

Suggested change

	/// Optional preferred ordering for the scan
	pub preferred_ordering: Option<Vec<SortExpr>>,
	/// Optional preferred ordering for the scan
	pub preferred_ordering: Option<PreferredOrdering>,

[alamb]

mybe we can move some of this comment to PreferredOrdering if we go with the struct approach

[alamb]

when would there be existing preferences?

[alamb]

is this a fundamental limitation? I ask because @pepijnve was asking about "column only" support the other day at

• External table with complex order expression not allowed #17411

[alamb]

as much as I love examples, I am not sure this one adds much

[alamb]

I think this would be clearer and less error prone, if you could do something that only change the field of interest. Perhaps like this:

Suggested change

	let new_table_scan = TableScan {
	table_name: table_scan.table_name.clone(),
	source: Arc::clone(&table_scan.source),
	projection: table_scan.projection.clone(),
	projected_schema: Arc::clone(&table_scan.projected_schema),
	filters: table_scan.filters.clone(),
	fetch: table_scan.fetch,
	preferred_ordering: Some(sort.expr.clone()),
	};
	let new_table_scan = table_scan.clone()
	.with_preferred_ordering(Some(sort.expr.clone()))

[alamb]

these tests don't really show that the preferred ordering is pushed through. Perhaps we can update the plan to show any preferred ordering

[alamb]

I think the EnforceSorting rule already pushes sorts down in the plan -- https://docs.rs/datafusion/latest/datafusion/physical_optimizer/enforce_sorting/struct.EnforceSorting.html

Do you think we will need more sort pushdown? Or will this always just be "pass down preferred sorts" to LogicalPlans?

[adriangb]

Nice. I was hoping another optimizer rule does the "hard work" so we can do just the simple thing here (only a subset of node types we need to support).

[alamb]

FWIW I think this table scan should have the preferred ordering passed to it, but I am not sure the current code will do so

[alamb]

Here is a PR that avoids some clones, which might improve performance

• Refactor PushdownSort to avoid some clones pydantic/datafusion#39

[alamb]

🤔 this seems to have caused a massive slowdown in the sql planner benchmark somehow:

Benchmarking physical_sorted_union_order_by_300: Warming up for 3.0000 s
Warning: Unable to complete 100 samples in 5.0s. You may wish to increase target time to 3942.3s, or reduce sample count to 10.
physical_sorted_union_order_by_300
                        time:   [38.914 s 38.997 s 39.079 s]

Benchmarking logical_plan_optimize: Warming up for 3.0000 s
Warning: Unable to complete 100 samples in 5.0s. You may wish to increase target time to 17189.6s, or reduce sample count to 10.

It is still running...

[adriangb]

I'm sure it's just a dumb mistake on my end. Let me do a round of looking at your comments and investigating, thank you for your patience 🙏🏻

[adriangb]

I also wonder if we should wait for the DataFusion 50 release before merging this

I think we should wait until after v50