Skip to content

fix: align CrossJoinExec schema metadata merge with the logical plan #23442

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
fangchenli wants to merge 1 commit into
base: main
Choose a base branch
from
+36 −19
Open

fix: align CrossJoinExec schema metadata merge with the logical plan #23442

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
26 changes: 7 additions & 19 deletions datafusion/physical-plan/src/joins/cross_join.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ use std::{sync::Arc, task::Poll};
use super::utils::{
BatchSplitter, BatchTransformer, BuildProbeJoinMetrics, NoopBatchTransformer,
OnceAsync, OnceFut, StatefulStreamResult, adjust_right_output_partitioning,
reorder_output_after_swap,
build_join_schema, reorder_output_after_swap,
};
use crate::execution_plan::{EmissionType, boundedness_from_children};
use crate::metrics::{ExecutionPlanMetricsSet, MetricsSet};
Expand All @@ -41,7 +41,7 @@ use crate::{

use arrow::array::{RecordBatch, RecordBatchOptions};
use arrow::compute::concat_batches;
use arrow::datatypes::{Fields, Schema, SchemaRef};
use arrow::datatypes::{Schema, SchemaRef};
use datafusion_common::stats::Precision;
use datafusion_common::{
JoinType, Result, ScalarValue, assert_eq_or_internal_err, internal_err,
Expand Down Expand Up @@ -102,23 +102,11 @@ pub struct CrossJoinExec {
impl CrossJoinExec {
/// Create a new [CrossJoinExec].
pub fn new(left: Arc<dyn ExecutionPlan>, right: Arc<dyn ExecutionPlan>) -> Self {
// left then right
let (all_columns, metadata) = {
let left_schema = left.schema();
let right_schema = right.schema();
let left_fields = left_schema.fields().iter();
let right_fields = right_schema.fields().iter();

let mut metadata = left_schema.metadata().clone();
metadata.extend(right_schema.metadata().clone());

(
left_fields.chain(right_fields).cloned().collect::<Fields>(),
metadata,
)
};

let schema = Arc::new(Schema::new(all_columns).with_metadata(metadata));
// Use the shared helper (inner join) so metadata merges the same way as
// the logical plan; merging it here independently let schemas diverge.
let (schema, _) =
build_join_schema(&left.schema(), &right.schema(), &JoinType::Inner);

@nathanb9 nathanb9 Jul 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

This aligns the initial plan, but the swap path still diverges: swap_inputs (line 188) rebuilds via CrossJoinExec::new(right, left), and with left-wins semantics the metadata winner flips under swap. OptimizationInvariantChecker then rejects the rule output. Running the issue's tables through SELECT * FROM t_left CROSS JOIN t_right on this branch:

Internal error: PhysicalOptimizer rule 'join_selection' failed. Schema mismatch. Expected original schema: Field { "a": Int32 }, Field { "b": Int32 }, got new schema: Field { "a": Int32 }, Field { "b": Int32 }.

(The printed schemas look identical because only fields are displayed; the difference is the schema-level metadata.) Setting datafusion.optimizer.join_reordering = false makes it pass, so it is the swap. The same flip exists for HashJoin/NLJ Inner and Full swaps on main, so it is pre-existing and arguably out of scope, but it would be good to note it in the PR description as a known remaining gap, or fix reorder_output_after_swap to re-impose the original schema metadata on the reverting projection. Happy to file a follow-up issue.

@nathanb9 nathanb9 Jul 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

nit: PR #16221 added test_join_metadata in both joins/utils.rs and the logical builder.rs to pin the left-wins/right-wins convention per join type. A matching unit test in cross_join.rs (construct a CrossJoinExec from two inputs with conflicting schema metadata, assert output metadata equals the left input's value) would be cheap to add and guards against future drift without needing the full planning pipeline.

let schema = Arc::new(schema);
let cache = Self::compute_properties(&left, &right, Arc::clone(&schema)).unwrap();

CrossJoinExec {
Expand Down
19 changes: 19 additions & 0 deletions datafusion/sqllogictest/src/test_context.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -521,6 +521,25 @@ pub async fn register_metadata_tables(ctx: &SessionContext) {
.unwrap();

ctx.register_batch("table_with_metadata", batch).unwrap();

// Same `metadata_key` as `table_with_metadata` but a different value, so a
// cross join of the two exercises the schema-metadata merge (#23434).
let alt_id =
Field::new("id", DataType::Int32, true).with_metadata(HashMap::from([(
String::from("metadata_key"),
String::from("the alt id field"),
)]));
let alt_schema = Schema::new(vec![alt_id]).with_metadata(HashMap::from([(
String::from("metadata_key"),
String::from("the other schema"),
)]));
let alt_batch = RecordBatch::try_new(
Arc::new(alt_schema),
vec![Arc::new(Int32Array::from(vec![Some(10), Some(20)])) as _],
)
.unwrap();
ctx.register_batch("table_with_metadata_alt", alt_batch)
.unwrap();
}

/// Create a UDF function named "example". See the `sample_udf.rs` example
Expand Down
10 changes: 10 additions & 0 deletions datafusion/sqllogictest/test_files/metadata.slt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,16 @@ FROM
----
6

# Regression test: cross join over two tables with conflicting schema metadata,
# feeding an aggregate, used to fail the physical planner's schema check.
# count(DISTINCT ...) keeps a real cross join (plain count folds from stats).
# See https://github.com/apache/datafusion/issues/23434
query I
SELECT count(DISTINCT "l"."id")

@nathanb9 nathanb9 Jul 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

The aggregate on top masks the swap path (see the comment on cross_join.rs), so this test cannot catch the join_selection failure mode. Consider also adding a SELECT * variant with reordering disabled to pin the fix directly:

statement ok
set datafusion.optimizer.join_reordering = false;

query II
SELECT "l"."id", "r"."id"
FROM "table_with_metadata" AS "l", "table_with_metadata_alt" AS "r"
ORDER BY "l"."id", "r"."id";

FROM "table_with_metadata" AS "l", "table_with_metadata_alt" AS "r";
----
2

# Regression test: missing field metadata, from the NULL field on the left side of the union
query ITT
(SELECT id, NULL::string as name, l_name FROM "table_with_metadata")
Expand Down