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

  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(())
  }