Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 51 additions & 3 deletions datafusion/physical-plan/src/joins/cross_join.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This rebuilds the schema in the same way as CrossJoinExec::new does and preserves the 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(),
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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()
Expand Down
Loading