This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23605-840da055be3809dd994b6942d4295223264ce81f in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit d75742edbfc4fb10fea20145b36ab776287276de Author: Michael Kleen <[email protected]> AuthorDate: Mon Jul 20 23:17:02 2026 +0200 fix: Preserve metadata when a cross-join is swapped (#23605) ## Which issue does this PR close? - Relates to #23461 but resolves it only for cross-joins. The same problem exisists for Hash/NL joins. There will be follow-ups. ## Rationale for this change When a `cross-join` is swapped, the metadata is rebuilt based on the inverted inputs. This can cause a change of metadata when the metadata has conflicting values e.g.: ```rust // left: 1000 rows, schema metadata {"key": "left"} // right: 2 rows, schema metadata {"key": "right"} // Cross join will swap the sides ctx.sql("select * from left cross join right").await?.collect().await?; // -> Internal error: ... 'join_selection' failed. Schema mismatch. // Metadata should be {"key" : "right" } but changed to {"key" : "left" } // because of the swap of sides and the rebuilt of the metadata starting from the other direction. ``` The cross-join should preserve the original metdata when it's swapped. ## What changes are included in this PR? - This pr avoids rebuilding the metadata based on the swapped inputs and uses the original metadata instead - Test ## Are these changes tested? Yes ## Are there any user-facing changes? No --- datafusion/physical-plan/src/joins/cross_join.rs | 54 ++++++++++++++++++++++-- 1 file changed, 51 insertions(+), 3 deletions(-) diff --git a/datafusion/physical-plan/src/joins/cross_join.rs b/datafusion/physical-plan/src/joins/cross_join.rs index 99c6800659..2e60e53681 100644 --- a/datafusion/physical-plan/src/joins/cross_join.rs +++ b/datafusion/physical-plan/src/joins/cross_join.rs @@ -186,8 +186,32 @@ impl CrossJoinExec { /// operators on the join's children. Check [`super::HashJoinExec::swap_inputs`] /// for more details. pub fn swap_inputs(&self) -> Result<Arc<dyn ExecutionPlan>> { - let new_join = - CrossJoinExec::new(Arc::clone(&self.right), Arc::clone(&self.left)); + // Rebuild schema with columns from right to left, preserve existing metadata + let new_columns = self + .right + .schema() + .fields + .iter() + .chain(self.left.schema().fields.iter()) + .cloned() + .collect::<Fields>(); + + let new_schema = Arc::new( + Schema::new(new_columns).with_metadata(self.schema.metadata.clone()), + ); + + let new_cache = + Self::compute_properties(&self.right, &self.left, Arc::clone(&new_schema))?; + + let new_join = CrossJoinExec { + left: Arc::clone(&self.right), + right: Arc::clone(&self.left), + schema: new_schema, + left_fut: Default::default(), + metrics: ExecutionPlanMetricsSet::default(), + cache: Arc::new(new_cache), + }; + reorder_output_after_swap( Arc::new(new_join), &self.left.schema(), @@ -701,7 +725,9 @@ impl<T: BatchTransformer> CrossJoinStream<T> { mod tests { use super::*; use crate::common; - use crate::test::{assert_join_metrics, build_table_scan_i32}; + use crate::test::{TestMemoryExec, assert_join_metrics, build_table_scan_i32}; + use arrow_schema::{DataType, Field}; + use std::collections::HashMap; use datafusion_common::{assert_contains, test_util::batches_to_sort_string}; use datafusion_execution::runtime_env::RuntimeEnvBuilder; @@ -994,6 +1020,28 @@ mod tests { Ok(()) } + #[test] + fn test_swapped_cross_join_schema_on_conflicting_metadata() { + let input = |field: &str, meta_value: &str| { + let schema = Arc::new( + Schema::new(vec![Field::new(field, DataType::Int32, false)]) + .with_metadata(HashMap::from([( + String::from("metadata_key"), + String::from(meta_value), + )])), + ); + TestMemoryExec::try_new_exec(&[vec![]], schema, None).unwrap() + }; + // Conflicting metadata on left and right input, right side wins "metadata_key" -> "right value" + let join = + CrossJoinExec::new(input("a", "left value"), input("b", "right value")); + + let swapped_join = join.swap_inputs().unwrap(); + + // The metadata of the cross-join and the swapped cross-join (with projection on top) must be the same + assert_eq!(join.schema().metadata(), swapped_join.schema().metadata()); + } + /// Returns the column names on the schema fn columns(schema: &Schema) -> Vec<String> { schema.fields().iter().map(|f| f.name().clone()).collect() --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
