fix: prevent silent overflow when reading Parquet TIMESTAMP_MILLIS values

native/core/src/parquet/parquet_exec.rs

@@ -33,11 +33,18 @@ use datafusion::physical_expr::PhysicalExpr;
 use datafusion::physical_expr_adapter::PhysicalExprAdapterFactory;
 use datafusion::prelude::SessionContext;
 use datafusion::scalar::ScalarValue;
-use datafusion_comet_spark_expr::EvalMode;
+use datafusion_comet_spark_expr::{EvalMode, GetStructField};
 use datafusion_datasource::TableSchema;
 use std::collections::HashMap;
 use std::sync::Arc;
 
+/// True when the expression reads a struct field (directly or in any child), i.e. the
+/// predicate references a nested column.
+fn references_nested_field(expr: &Arc<dyn PhysicalExpr>) -> bool {
+    expr.downcast_ref::<GetStructField>().is_some()
+        || expr.children().iter().any(|c| references_nested_field(c))
+}
+
 /// Initializes a DataSourceExec plan with a ParquetSource for Comet's native Parquet scan.
 ///
 ///   `required_schema`: Schema to be projected by the scan.
@@ -93,6 +100,16 @@ pub(crate) fn init_datasource_exec(
     );
     spark_parquet_options.use_field_id = use_field_id;
     spark_parquet_options.ignore_missing_field_id = ignore_missing_field_id;
+    // Spark only avoids the TIMESTAMP_MILLIS overflow error for filtered-out values through
+    // row-group statistics pruning. DataFusion can neither prune nested-field predicates
+    // (`PruningPredicate` has no nested-field support) nor evaluate them as Parquet row
+    // filters (struct columns are classified non-pushable), so a scan carrying such a
+    // predicate would decode row groups Spark prunes and fail on the checked conversion of
+    // any overflowing top-level column. Fall back to the safe cast (overflow -> NULL) for
+    // these scans; the filter above the scan then discards the rows like Spark's pruning.
+    spark_parquet_options.checked_timestamp_overflow = !data_filters
+        .as_ref()
+        .is_some_and(|filters| filters.iter().any(references_nested_field));
 
     // Determine the schema and projection to use for ParquetSource.
     // When data_schema is provided, use it as the base schema so DataFusion knows the full

native/core/src/parquet/parquet_support.rs

@@ -22,8 +22,8 @@ use arrow::compute::can_cast_types;
 use arrow::datatypes::{FieldRef, Fields};
 use arrow::{
     array::{
-        cast::AsArray, new_null_array, types::TimestampMicrosecondType, Array, ArrayRef,
-        StructArray,
+        cast::AsArray, new_null_array, types::TimestampMicrosecondType,
+        types::TimestampMillisecondType, Array, ArrayRef, ArrowNativeTypeOp, StructArray,
     },
     compute::{cast_with_options, CastOptions},
     datatypes::{DataType, TimeUnit},
@@ -99,6 +99,16 @@ pub struct SparkParquetOptions {
     /// (Spark 3.x, SPARK-36182). Mirrors Comet's per-Spark-version constant
     /// in ShimCometConf.
     pub allow_timestamp_ltz_to_ntz: bool,
+    /// When true (the default), a top-level TIMESTAMP_MILLIS column that overflows during
+    /// the millis->micros upscale raises an error, matching Spark's checked
+    /// `millisToMicros`. Scans whose data filters reference nested struct fields set this
+    /// to false and fall back to the safe cast (overflow -> NULL): Spark only avoids the
+    /// overflow error for filtered-out values through row-group statistics pruning, and
+    /// DataFusion can neither prune (`PruningPredicate` has no nested-field support) nor
+    /// row-filter (struct columns are classified non-pushable in
+    /// `can_expr_be_pushed_down_with_schemas`) such predicates, so a checked conversion
+    /// would fail queries Spark answers with zero rows.
+    pub checked_timestamp_overflow: bool,
 }
 
 impl SparkParquetOptions {
@@ -115,6 +125,7 @@ impl SparkParquetOptions {
             ignore_missing_field_id: false,
             allow_type_promotion: false,
             allow_timestamp_ltz_to_ntz: false,
+            checked_timestamp_overflow: true,
         }
     }
 
@@ -131,6 +142,7 @@ impl SparkParquetOptions {
             ignore_missing_field_id: false,
             allow_type_promotion: false,
             allow_timestamp_ltz_to_ntz: false,
+            checked_timestamp_overflow: true,
         }
     }
 }
@@ -167,6 +179,15 @@ fn parquet_convert_array(
     array: ArrayRef,
     to_type: &DataType,
     parquet_options: &SparkParquetOptions,
+) -> DataFusionResult<ArrayRef> {
+    parquet_convert_array_impl(array, to_type, parquet_options, true)
+}
+
+fn parquet_convert_array_impl(
+    array: ArrayRef,
+    to_type: &DataType,
+    parquet_options: &SparkParquetOptions,
+    top_level: bool,
 ) -> DataFusionResult<ArrayRef> {
     use DataType::*;
     let from_type = array.data_type();
@@ -182,10 +203,11 @@ fn parquet_convert_array(
         )?),
         (List(_), List(to_inner_type)) => {
             let list_arr: &ListArray = array.as_list();
-            let cast_field = parquet_convert_array(
+            let cast_field = parquet_convert_array_impl(
                 Arc::clone(list_arr.values()),
                 to_inner_type.data_type(),
                 parquet_options,
+                false,
             )?;
 
             Ok(Arc::new(ListArray::new(
@@ -195,6 +217,28 @@ fn parquet_convert_array(
                 list_arr.nulls().cloned(),
             )))
         }
+        (
+            Timestamp(TimeUnit::Millisecond, _),
+            Timestamp(TimeUnit::Microsecond, target_tz),
+        ) if top_level && parquet_options.checked_timestamp_overflow => {
+            // Spark's Parquet reader calls the checked `millisToMicros` conversion for both
+            // direct and dictionary values, independent of CAST evaluation mode:
+            // https://github.com/apache/spark/blob/v4.2.0/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java#L817-L833
+            // `millisToMicros` uses `Math.multiplyExact`:
+            // https://github.com/apache/spark/blob/v4.2.0/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkDateTimeUtils.scala#L103-L108
+            //
+            // The checked conversion is limited to TOP-LEVEL columns. Spark only avoids the
+            // error for filtered-out values through row-group statistics pruning, and
+            // DataFusion's PruningPredicate does not support nested fields yet, so a checked
+            // conversion on a nested field would fail queries whose predicates Spark prunes
+            // (e.g. `WHERE s.ts < X` over an all-overflowing file). Nested fields keep the
+            // pre-existing safe-cast behavior below (overflow -> NULL).
+            let micros = array
+                .as_primitive::<TimestampMillisecondType>()
+                .try_unary::<_, TimestampMicrosecondType, _>(|value| value.mul_checked(1_000))?
+                .with_timezone_opt(target_tz.clone());
+            Ok(Arc::new(micros))
+        }
         (Timestamp(TimeUnit::Microsecond, None), Timestamp(TimeUnit::Microsecond, Some(tz))) => {
             Ok(Arc::new(
                 array
@@ -300,10 +344,11 @@ fn parquet_convert_struct_to_struct(
                 };
 
                 if let Some(from_index) = from_index {
-                    cast_fields.push(parquet_convert_array(
+                    cast_fields.push(parquet_convert_array_impl(
                         Arc::clone(array.column(from_index)),
                         to_field.data_type(),
                         parquet_options,
+                        false,
                     )?);
                     field_overlap = true;
                 } else {
@@ -351,15 +396,17 @@ fn parquet_convert_map_to_map(
                 "map is missing value field".to_string(),
             ))?;
 
-            let key_array = parquet_convert_array(
+            let key_array = parquet_convert_array_impl(
                 Arc::clone(from.keys()),
                 key_field.data_type(),
                 parquet_options,
+                false,
             )?;
-            let value_array = parquet_convert_array(
+            let value_array = parquet_convert_array_impl(
                 Arc::clone(from.values()),
                 value_field.data_type(),
                 parquet_options,
+                false,
             )?;
 
             Ok(Arc::new(MapArray::new(
@@ -656,4 +703,72 @@ mod tests {
             }
         }
     }
+
+    #[test]
+    fn test_millis_to_micros_overflow_checked_only_at_top_level() {
+        use crate::parquet::parquet_support::{parquet_convert_array, SparkParquetOptions};
+        use arrow::array::{Array, ArrayRef, StructArray, TimestampMillisecondArray};
+        use arrow::datatypes::{DataType, Field, Fields, TimeUnit};
+        use datafusion_comet_spark_expr::EvalMode;
+        use std::sync::Arc;
+
+        let options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
+        let overflow_millis = 9_223_372_036_854_776_i64;
+        let millis: ArrayRef = Arc::new(TimestampMillisecondArray::from(vec![
+            Some(overflow_millis),
+            None,
+        ]));
+        let micros_type = DataType::Timestamp(TimeUnit::Microsecond, None);
+
+        // Top-level: checked, matching Spark's `millisToMicros` (`Math.multiplyExact`).
+        let err = parquet_convert_array(Arc::clone(&millis), &micros_type, &options)
+            .expect_err("top-level overflow must error");
+        assert!(
+            err.to_string().to_lowercase().contains("overflow"),
+            "unexpected error: {err}"
+        );
+
+        // Scans whose data filters reference nested fields disable the checked
+        // conversion (DataFusion cannot prune or row-filter those predicates the way
+        // Spark's statistics pruning protects them), falling back to overflow -> NULL.
+        let mut unchecked_options = options.clone();
+        unchecked_options.checked_timestamp_overflow = false;
+        let converted =
+            parquet_convert_array(Arc::clone(&millis), &micros_type, &unchecked_options)
+                .expect("unchecked overflow must not error");
+        assert!(converted.is_null(0), "overflow must become NULL");
+        assert!(converted.is_null(1));
+
+        // Nested: DataFusion's PruningPredicate cannot prune nested fields, so a
+        // checked conversion would fail queries whose predicates Spark satisfies via
+        // row-group statistics pruning. The nested field keeps the safe-cast behavior:
+        // overflow becomes NULL.
+        let child_field = Arc::new(Field::new(
+            "ts",
+            DataType::Timestamp(TimeUnit::Millisecond, None),
+            true,
+        ));
+        let strukt: ArrayRef = Arc::new(StructArray::new(
+            Fields::from(vec![Arc::clone(&child_field)]),
+            vec![millis],
+            None,
+        ));
+        let target = DataType::Struct(Fields::from(vec![Arc::new(Field::new(
+            "ts",
+            micros_type.clone(),
+            true,
+        ))]));
+        let converted = parquet_convert_array(strukt, &target, &options)
+            .expect("nested overflow must not error");
+        let converted_child = Arc::clone(
+            converted
+                .as_any()
+                .downcast_ref::<StructArray>()
+                .unwrap()
+                .column(0),
+        );
+        assert_eq!(converted_child.data_type(), &micros_type);
+        assert!(converted_child.is_null(0), "overflow must become NULL");
+        assert!(converted_child.is_null(1));
+    }
 }