Discussion: API for Join Access Path and Join Order Selection

[alamb]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.

This ticket is to coordinate discussion and ideas for improving DataFusion join access path (aka algorithm) and join order selection.

DataFusion does not currently have a sophisticated cost based optimizer, but has support for the mechanics of join reordering and statistics. There seems to be broad desire for improve DataFusion for more advanced use cases (e.g. TPC-DS / distributed joins / ASOF joins). However as described below there are many tradeoffs involved with join access path and join order selection (e.g. #17432 recently) , so it is not likely any one implementation will be good for all users

Instead, I would like to discuss adding an API that users can use to customize the behavior to their use case.

Current State of DataFusion Optimizer:

Quoting from "Optimizing SQL (and DataFrames) in DataFusion, Part 2: Optimizers in Apache DataFusion"

DataFusion purposely does not include a sophisticated cost based optimizer. Instead, keeping with its design goals it provides a reasonable default implementation along with extension points to customize behavior.

Specifically, DataFusion includes

1. “Syntactic Optimizer” (joins in the order they are listed in the query) with basic join re-ordering (source) to prevent join disasters.
2. Support for ColumnStatistics and Table Statistics
3. The framework for filter selectivity + join cardinality estimation.
4. APIs for easily rewriting plans, such as the TreeNode API and reordering joins

This combination of features along with custom optimizer passes lets users customize the behavior to their use case, such as custom indexes like uWheel and materialized views.

The rationale for including only a basic optimizer is that any one particular set of heuristics and cost model is unlikely to work well for the wide variety of DataFusion users because of the tradeoffs involved.

For example, some users may always have access to adequate resources, and want the fastest query execution, and are willing to tolerate runtime errors or a performance cliff when there is insufficient memory. Other users, however, may be willing to accept a slower maximum performance in return for more predictable performance when running in a resource constrained environment. This approach is not universally agreed.

Usecases

Here are some use cases that have been discussed in various issues and PRs:

• Choose different join algorithms based on tradeoffs, such as memory usage vs CPU usage Add configuration to choose specific join implementation #17432
• Distributed join planning (e.g choosing between resegment or broadcast joins) where the join order might be different based on the distribution of the source data
• Join optimization research, such as join reordering based on different statistics/ cost models (e.g. optd from CMUDB) or different join algorithms
• Add new join implementations such as ASOF / specialized temporal joins: ASOF join support / Specialize Range Joins #318

Describe the solution you'd like

I would like

1. An API for join access path, algorithm, and join order selection that is flexible enough to support the use cases above, but can be extended /customized by users
2. Documentation and examples of how to use the API
3. A default implementation that works reasonably well (e.g. prevents join disasters in TPCH with basic statistics), and respects the same basic knobs we have today

Example of existing configuration settings

• repartition_joins
• allow_symmetric_joins_without_pruning
• prefer_hash_join
• hash_join_single_partition_threshold
• hash_join_single_partition_threshold_rows

Describe alternatives you've considered

One idea (from #17467 (review) a discussion with @2010YOUY01):

Maybe we could make JoinPlanner trait that can be registered with the SessionContext or the Optimizer the same way as ExtensionPlanners?. Then we can provide a default JoinPlanner (what currently exists) that has its own config namespace, etc

Something like this

trait JoinPlanner {
  // plan the initial join when converting from Logical --> Physical join
  fn plan_initial_join(
    session_state: &SessionState,
    physical_left: Arc<dyn ExecutionPlan>,
    physical_right: Arc<dyn ExecutionPlan>,
    join_on: join_utils::JoinOn,
    join_filter: Option<join_utils::JoinFilter>,
    join_type: &JoinType,
    null_equality: &datafusion_common::NullEquality,) -> Arc<dyn ExecutionPlan>;
   )
   // TODO other APIs here
}

Another possibility is to introduce a JoinGraph structure with the relevant APIs for walking and building plans

• Consider adding a JoinGraph structure for representing joins #17719

trait JoinPlanner {
  // plan a specific  join graph when converting from Logical --> Physical join
  fn plan_join_graph(
    session_state: &SessionState,
    join_graph: JoinGraph,
   )-> Result<Arc<dyn ExecutionPlan>;
}

Additional context

Related discussions

• [EPIC] Improving cost calculations and cost based optimizations #3929
• [Epic] Remove Sort Merge Join Experimental status #9846 (comment)
• Add configuration to choose specific join implementation #17432

[alamb]

FYI @adriangb @jonathanc-n @2010YOUY01 @NGA-TRAN -- I think you have all expressed interest in this type of feature. Perhaps you have thoughts to share

[NGA-TRAN]

I’m planning to draft a write-up on high-level approaches and examples in October, and I’ll share it with you for feedback and discussion. Hopefully, you can help shape the API from there.

[comphead]

FYI CS Berkeley provides interesting selectivity defaults to deal with join orderings https://cs186berkeley.net/notes/note10/#appendix---selectivity-values

Postgres has hardcoded values for selectivity like (if there is no stats involved)
https://github.com/postgres/postgres/blob/master/src/include/utils/selfuncs.h#L34

I would be happy to join to investigate

• The framework for filter selectivity + join cardinality estimation.

[alamb]

Here is a note from @pepijnve in discord with a usecase I thought was interesting (basically wants to use a custom join type):

The ST_KNN join query implementation technique in Sedona is really interesting. I had a similar need in my project where I want to join on the longest prefix in another table efficiently. The actual implementation is a custom LogicalPlan and an ExecutionPlan that builds a trie based on the build side and then probes that.

I couldn't figure out a simple way to plug a custom join into the SQL parser though so we ended up punting on that and simply create the logical plan via the dataframe API instead. Being able to use SQL would be nice though.
There's first class support for ASOF JOIN in the SQL parser, but I didn't see a way to add a custom JOIN. Using a marker function and an optimizer rule is a neat solution, but feels a bit brittle. The comment in the docs about UnsupportedOperationException with Spark seems to kind of confirm that.

Is there any consensus in the community on how to add non standard join strategies? Is the Sedona ST_KNN approach the way to go or are there other ways to implement this?

[alamb]

More feedback from @zhangfengcdt in discord which I think is also really interesting:

Really nice discussion! When we implement the KNN join in SedonaDB, we had the similar concerns and did some experiments, and found the marker function + optimizer rule pattern works well for our case. There are mainly two challenges in our case: (1) asymmetric KNN execution, meaning we need to build a spatial index on the build side, and for each left side geometry find the k nearest neighbors (2) join order control, meaning we need to control the predicate evaluation order

For the first challenge, we register a stub scalar UDF and in query planner, we detect ST_KNN predicates in join filters and transforms them into a specialized SpatialJoinExec physical plan with KNN semantics. for the second challange, we add a barrier function to serve as an optimization barrier to prevent filter pushdown and control predicate evaluation order. This is critical for maintaining semantic correctness (KNN then filter vs. filter then KNN). Both work well for the purpose.

The marker function + optimizer rule pattern is indeed the most practical approach for adding custom join strategies to DataFusion/Arrow-based systems. It's more robust than it might seem because:

The optimizer rule has full control over when to apply the transformation
The stub function provides type checking and documentation
It integrates naturally with SQL without parser modifications

We have the similar approach on Apache SedonaSpark as well. I would recommend these steps for custom joins for reference:

Use marker functions for SQL integration
Implement robust pattern matching in optimizer rules
Provide optimization barriers when semantics are order-dependent
Document the transformation clearly for users
Consider providing both SQL and DataFrame APIs

[NGA-TRAN]

I’m drafting a high-level proposal focused on modular, customizable components:

• Join enumeration leverages join ranking, customized joins, and plan pruning
• Join ranking can be tailored to fit specific needs
• Plan pruning supports early elimination of plans using custom rules or a cost model
• Cost model is also customizable to suit different scenarios

I’ll include examples for each. It should be ready for discussion in a week or two.

Note: I do not propose the API. I just provide some examples for us to discuss the API

[alamb]

FYI @NGA-TRAN there is a lot of discussion from @JanKaul @LorrensP-2158466 and others about JoinGraphs here:

• Consider adding a JoinGraph structure for representing joins #17719 (comment)