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 <killme2008@gmail.com>

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

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
+

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;