From ef1298ce026924bbd8f80cf279eedfed06718635 Mon Sep 17 00:00:00 2001 From: dennis zhuang Date: Thu, 15 Oct 2026 03:28:25 +0000 Subject: [PATCH] fix(query): accept narrowed nullability in MergeScan remote schema check (#9592) On append-mode tables a region reports an exact row count. When it is zero, DataFusion's AggregateStatistics replaces the region-side __min_state/__max_state aggregate with a non-null struct literal, so the remote stream advertises the state column as non-nullable. The MergeScan schema check required equal nullability and failed the query. Accept a remote field that is non-nullable where the expected field is nullable, matching DataFusion's own physical-plan schema invariant. Batches are already rebuilt with the expected schema. Widening is still rejected, and nested field nullability is still compared exactly as part of the data type. Signed-off-by: Dennis Zhuang --- src/query/src/dist_plan/merge_scan.rs | 87 ++++++++++-- .../aggregate/empty_region_stats.result | 126 ++++++++++++++++++ .../common/aggregate/empty_region_stats.sql | 42 ++++++ 3 files changed, 242 insertions(+), 13 deletions(-) create mode 100644 tests/cases/standalone/common/aggregate/empty_region_stats.result create mode 100644 tests/cases/standalone/common/aggregate/empty_region_stats.sql diff --git a/src/query/src/dist_plan/merge_scan.rs b/src/query/src/dist_plan/merge_scan.rs index 041fea160cf..9717c436e0b 100644 --- a/src/query/src/dist_plan/merge_scan.rs +++ b/src/query/src/dist_plan/merge_scan.rs @@ -167,8 +167,20 @@ fn is_json_field(field: &Field) -> bool { == Some("Json") } +/// Returns true when the remote field's nullability fits the expected field. +/// +/// The remote physical optimizer may narrow a nullable column to non-nullable, +/// as DataFusion's own schema invariant permits: `AggregateStatistics` +/// replaces an aggregate over an empty region with a non-null literal. Batches +/// are rebuilt with the expected schema, so the narrowing does not leak. +/// Widening is rejected. Nested field nullability is part of the data type and +/// still has to match exactly. +fn nullability_compatible(expected_field: &Field, actual_field: &Field) -> bool { + expected_field.is_nullable() || !actual_field.is_nullable() +} + /// Returns true when two fields describe the same column semantics: same -/// name, same data type, and same nullability. +/// name, same data type, and compatible nullability. /// /// Arrow `Field` metadata is auxiliary information (indexing, encoding, /// compression hints, ...) and does not participate in the comparison. This @@ -179,7 +191,7 @@ fn is_json_field(field: &Field) -> bool { fn fields_semantically_equal(expected_field: &Field, actual_field: &Field) -> bool { expected_field.name() == actual_field.name() && expected_field.data_type() == actual_field.data_type() - && expected_field.is_nullable() == actual_field.is_nullable() + && nullability_compatible(expected_field, actual_field) } /// Returns true when two fields are semantically equivalent JSON columns. @@ -196,7 +208,7 @@ fn json_fields_compatible(expected_field: &Field, actual_field: &Field) -> bool is_json_field(expected_field) && is_json_field(actual_field) && expected_field.name() == actual_field.name() - && expected_field.is_nullable() == actual_field.is_nullable() + && nullability_compatible(expected_field, actual_field) // Both must carry the same JSON marker. && expected_field.metadata().get(datatypes::schema::TYPE_KEY) == actual_field.metadata().get(datatypes::schema::TYPE_KEY) @@ -212,14 +224,15 @@ fn json_fields_compatible(expected_field: &Field, actual_field: &Field) -> bool /// Validates the remote schema before positional column handling. /// /// Field metadata (indexing/encoding/compression hints) is non-semantic at -/// this boundary and is ignored: two fields are equal when their name, data -/// type, and nullability match (see [`fields_semantically_equal`]). A -/// timestamp timezone difference is the only intentional exception, accepted -/// when the fields share the same timestamp unit, distinct timezones, and -/// equal name and nullability. Top-level Arrow schema metadata is -/// non-semantic at this boundary as well. JSON columns are compared -/// semantically (see [`json_fields_compatible`]) because their wire and -/// decoded representations use different physical arrow types. +/// this boundary and is ignored: two fields are equal when their name and +/// data type match and their nullability is compatible (see +/// [`fields_semantically_equal`]). A timestamp timezone difference is the only +/// intentional exception, accepted when the fields share the same timestamp +/// unit, distinct timezones, equal name, and compatible nullability. +/// Top-level Arrow schema metadata is non-semantic at this boundary as well. +/// JSON columns are compared semantically (see [`json_fields_compatible`]) +/// because their wire and decoded representations use different physical +/// arrow types. fn validate_remote_schema( expected: &ArrowSchema, actual: &ArrowSchema, @@ -260,7 +273,7 @@ fn validate_remote_schema( ) if expected_unit == actual_unit && expected_timezone != actual_timezone && expected_field.name() == actual_field.name() - && expected_field.is_nullable() == actual_field.is_nullable() + && nullability_compatible(expected_field, actual_field) ); if !timezone_only_difference { return Err(remote_schema_mismatch(format!( @@ -2691,6 +2704,30 @@ mod tests { assert_eq!(batches[0].schema().as_ref(), &expected_schema); } + #[tokio::test] + async fn qbs_merge_scan_remote_schema_identity_emits_expected_schema_on_nullability_narrowing() + { + let batch = record_batch( + int64_schema(&["a", "b"]), + vec![ + Arc::new(Int64Vector::from_slice([11])) as _, + Arc::new(Int64Vector::from_slice([12])) as _, + ], + ); + let expected_schema = ArrowSchema::new(vec![ + Field::new("a", TestArrowDataType::Int64, true), + Field::new("b", TestArrowDataType::Int64, false), + ]); + let batches = collect_merge_scan(merge_scan_exec( + vec![(RegionId::new(1024, 1), batch)], + expected_schema.clone(), + 1, + )) + .await + .unwrap(); + assert_eq!(batches[0].schema().as_ref(), &expected_schema); + } + #[tokio::test] async fn qbs_merge_scan_remote_schema_identity_rejects_incompatible_type() { let batch = record_batch( @@ -2897,7 +2934,7 @@ mod tests { } #[test] - fn merge_scan_remote_schema_identity_ignores_field_metadata_but_rejects_nullability_mismatch() { + fn merge_scan_remote_schema_identity_ignores_field_metadata_but_rejects_nullability_widening() { let expected = expected_int64_schema(); let metadata_mismatch = ArrowSchema::new_with_metadata( vec![ @@ -2926,6 +2963,30 @@ mod tests { assert!(validate_remote_schema(&expected, &nullability_mismatch, "test").is_err()); } + #[test] + fn merge_scan_remote_schema_identity_accepts_top_level_nullability_narrowing_only() { + let state_type = |child_nullable| { + TestArrowDataType::Struct(arrow_schema::Fields::from(vec![Field::new( + "min[value]", + TestArrowDataType::Int64, + child_nullable, + )])) + }; + let schema = |nullable, child_nullable| { + ArrowSchema::new(vec![Field::new( + "__min_state(t.v)", + state_type(child_nullable), + nullable, + )]) + }; + + // `AggregateStatistics` turns the state aggregate over an empty region + // into a non-null struct literal. + assert!(validate_remote_schema(&schema(true, true), &schema(false, true), "test").is_ok()); + assert!(validate_remote_schema(&schema(false, true), &schema(true, true), "test").is_err()); + assert!(validate_remote_schema(&schema(true, true), &schema(true, false), "test").is_err()); + } + #[test] fn merge_scan_remote_schema_identity_ignores_skipping_index_field_metadata() { // Regression: a remote stream may advertise auxiliary field metadata diff --git a/tests/cases/standalone/common/aggregate/empty_region_stats.result b/tests/cases/standalone/common/aggregate/empty_region_stats.result new file mode 100644 index 00000000000..5f4a1bc59b0 --- /dev/null +++ b/tests/cases/standalone/common/aggregate/empty_region_stats.result @@ -0,0 +1,126 @@ +-- Append-mode regions report exact row counts, so the datanode can answer +-- count/min/max over a region from statistics. This only happens when every +-- aggregate in the query can be answered that way. +CREATE TABLE a1 (ts TIMESTAMP(9) TIME INDEX, k STRING, v BIGINT) +PARTITION ON COLUMNS (k) (k < 'b', k >= 'b' AND k < 'c', k >= 'c') +WITH (append_mode = 'true'); + +Affected Rows: 0 + +-- All regions are empty. +SELECT min(v), max(v) FROM a1; + ++-----------+-----------+ +| min(a1.v) | max(a1.v) | ++-----------+-----------+ +| | | ++-----------+-----------+ + +SELECT count(*), count(v) FROM a1; + ++----------+-------------+ +| count(*) | count(a1.v) | ++----------+-------------+ +| 0 | 0 | ++----------+-------------+ + +INSERT INTO a1 VALUES (1000, 'a', 1), (2000, 'a', 2); + +Affected Rows: 2 + +-- Two of three regions are empty. +SELECT min(v) FROM a1; + ++-----------+ +| min(a1.v) | ++-----------+ +| 1 | ++-----------+ + +SELECT max(v) FROM a1; + ++-----------+ +| max(a1.v) | ++-----------+ +| 2 | ++-----------+ + +SELECT min(ts), max(k) FROM a1; + ++----------------------------+-----------+ +| min(a1.ts) | max(a1.k) | ++----------------------------+-----------+ +| 1970-01-01T00:00:00.000001 | a | ++----------------------------+-----------+ + +SELECT count(*), count(v) FROM a1; + ++----------+-------------+ +| count(*) | count(a1.v) | ++----------+-------------+ +| 2 | 2 | ++----------+-------------+ + +SELECT min(v), count(*) FROM a1; + ++-----------+----------+ +| min(a1.v) | count(*) | ++-----------+----------+ +| 1 | 2 | ++-----------+----------+ + +SELECT sum(v), avg(v) FROM a1; + ++-----------+-----------+ +| sum(a1.v) | avg(a1.v) | ++-----------+-----------+ +| 3 | 1.5 | ++-----------+-----------+ + +SELECT k, min(v), max(v), count(*) FROM a1 GROUP BY k ORDER BY k; + ++---+-----------+-----------+----------+ +| k | min(a1.v) | max(a1.v) | count(*) | ++---+-----------+-----------+----------+ +| a | 1 | 2 | 2 | ++---+-----------+-----------+----------+ + +INSERT INTO a1 VALUES (3000, 'b', 3); + +Affected Rows: 1 + +-- One of three regions is empty. +SELECT min(v), max(v) FROM a1; + ++-----------+-----------+ +| min(a1.v) | max(a1.v) | ++-----------+-----------+ +| 1 | 3 | ++-----------+-----------+ + +SELECT k, min(v), max(v), count(*) FROM a1 GROUP BY k ORDER BY k; + ++---+-----------+-----------+----------+ +| k | min(a1.v) | max(a1.v) | count(*) | ++---+-----------+-----------+----------+ +| a | 1 | 2 | 2 | +| b | 3 | 3 | 1 | ++---+-----------+-----------+----------+ + +INSERT INTO a1 VALUES (4000, 'c', 4); + +Affected Rows: 1 + +-- No region is empty. +SELECT min(v), max(v), count(*) FROM a1; + ++-----------+-----------+----------+ +| min(a1.v) | max(a1.v) | count(*) | ++-----------+-----------+----------+ +| 1 | 4 | 4 | ++-----------+-----------+----------+ + +DROP TABLE a1; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/aggregate/empty_region_stats.sql b/tests/cases/standalone/common/aggregate/empty_region_stats.sql new file mode 100644 index 00000000000..a2d03e6b0e0 --- /dev/null +++ b/tests/cases/standalone/common/aggregate/empty_region_stats.sql @@ -0,0 +1,42 @@ +-- Append-mode regions report exact row counts, so the datanode can answer +-- count/min/max over a region from statistics. This only happens when every +-- aggregate in the query can be answered that way. +CREATE TABLE a1 (ts TIMESTAMP(9) TIME INDEX, k STRING, v BIGINT) +PARTITION ON COLUMNS (k) (k < 'b', k >= 'b' AND k < 'c', k >= 'c') +WITH (append_mode = 'true'); + +-- All regions are empty. +SELECT min(v), max(v) FROM a1; + +SELECT count(*), count(v) FROM a1; + +INSERT INTO a1 VALUES (1000, 'a', 1), (2000, 'a', 2); + +-- Two of three regions are empty. +SELECT min(v) FROM a1; + +SELECT max(v) FROM a1; + +SELECT min(ts), max(k) FROM a1; + +SELECT count(*), count(v) FROM a1; + +SELECT min(v), count(*) FROM a1; + +SELECT sum(v), avg(v) FROM a1; + +SELECT k, min(v), max(v), count(*) FROM a1 GROUP BY k ORDER BY k; + +INSERT INTO a1 VALUES (3000, 'b', 3); + +-- One of three regions is empty. +SELECT min(v), max(v) FROM a1; + +SELECT k, min(v), max(v), count(*) FROM a1 GROUP BY k ORDER BY k; + +INSERT INTO a1 VALUES (4000, 'c', 4); + +-- No region is empty. +SELECT min(v), max(v), count(*) FROM a1; + +DROP TABLE a1;