Skip to content

DF 55: physical-optimizer enforce_sorting panics on RightMark joins #24717

Description

@Tpt

Describe the bug

The build_join_column_index function does not support RightMark joins and panic on them even if HashJoinExec::maintains_input_order returns true for RightMark joins, leading the handle_hash_join to try propagating the order requirements and failing to do so.

This was working well with DF 54

To Reproduce

  use std::sync::Arc;

  use datafusion::{
      arrow::datatypes::{DataType, Field, Schema},
      common::JoinType,
      datasource::memory::MemorySourceConfig,
      error::Result,
      physical_expr::{LexOrdering, PhysicalSortExpr, expressions::Column},
      physical_optimizer::PhysicalOptimizerRule,
      physical_plan::{ExecutionPlan, joins::HashJoinExecBuilder, sorts::sort::SortExec},
      prelude::SessionContext,
  };

  fn source(name: &str) -> Result<Arc<dyn ExecutionPlan>> {                                                                                                                                            
      let schema = Arc::new(Schema::new(vec![Field::new(name, DataType::Utf8, false)]));
      Ok(MemorySourceConfig::try_new_exec(&[vec![]], schema, None)?)
  }

  #[tokio::test]
  async fn right_mark_hash_join_sort_pushdown_panics() -> Result<()> {
      let reference = source("ref_id")?;   // build side                                                                                                                                               
      let local = source("local_id")?;     // probe side

      let join_on = vec![(
          Arc::new(Column::new_with_schema("ref_id", &reference.schema())?) as _,
          Arc::new(Column::new_with_schema("local_id", &local.schema())?) as _,
      )];
      let join = HashJoinExecBuilder::new(reference, local, join_on, JoinType::RightMark)
          .build_exec()?;

      // Any sort above the join on a probe-side column.
      let sort = Arc::new(SortExec::new(                                                                                                                                                               
          LexOrdering::new(vec![PhysicalSortExpr::new_default(Arc::new(
              Column::new_with_schema("local_id", &join.schema())?,
          ))])
          .expect("non-empty ordering"),
          join,
      ));
  
      let config = SessionContext::new().copied_config().options().clone();
      let optimized = datafusion::physical_optimizer::ensure_requirements::EnsureRequirements {}
          .optimize(sort, &config)?;   // <-- panics here

      println!("{}", datafusion::physical_plan::displayable(optimized.as_ref()).indent(false));
      Ok(())
  }

Expected behavior

No response

Additional context

No response

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    bugSomething isn't workingregressionSomething that used to work no longer does

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions