fix: prevent silent overflow when reading Parquet TIMESTAMP_MILLIS values

native/core/src/execution/planner.rs

@@ -1723,7 +1723,11 @@ impl PhysicalPlanner {
                     object_store_url,
                     file_groups,
                     Some(projection_vector),
-                    Some(data_filters?),
+                    if common.has_data_filters || !common.data_filters.is_empty() {
+                        Some(data_filters?)
+                    } else {
+                        None
+                    },
                     default_values,
                     common.session_timezone.as_str(),
                     common.case_sensitive,

native/core/src/parquet/parquet_exec.rs

@@ -55,8 +55,9 @@ use std::sync::Arc;
 ///
 ///   `projection_vector`: A vector of the indexes in the schema of the fields to be projected
 ///
-///   `data_filters`: Any predicate that must be applied to the data returned by the scan. If
-/// specified, then `data_schema` must also be specified.
+///   `data_filters`: Any predicate that must be applied to the data returned by the scan. An empty
+/// `Vec` means Spark supplied filters that Comet could not serialize. If specified, then
+/// `data_schema` must also be specified.
 #[allow(clippy::too_many_arguments)]
 pub(crate) fn init_datasource_exec(
     required_schema: SchemaRef,
@@ -94,6 +95,12 @@ 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 can discard filtered-out values before timestamp conversion using statistics,
+    // dictionary, and row-level filters. Comet cannot mirror every pruning path, so applying
+    // checked conversion in a filtered scan can fail on values Spark never reads. Preserve the
+    // existing safe cast for filtered scans and use checked conversion only when every value is
+    // necessarily read.
+    spark_parquet_options.checked_timestamp_overflow = data_filters.is_none();
 
     // 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

@@ -23,8 +23,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},
@@ -101,6 +101,12 @@ 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`. Filtered scans set this to false and retain the safe cast
+    /// (overflow -> NULL), because Spark may discard values through pruning paths that
+    /// DataFusion cannot fully mirror before conversion.
+    pub checked_timestamp_overflow: bool,
 }
 
 impl SparkParquetOptions {
@@ -117,6 +123,7 @@ impl SparkParquetOptions {
             ignore_missing_field_id: false,
             allow_type_promotion: false,
             allow_timestamp_ltz_to_ntz: false,
+            checked_timestamp_overflow: true,
         }
     }
 
@@ -133,6 +140,7 @@ impl SparkParquetOptions {
             ignore_missing_field_id: false,
             allow_type_promotion: false,
             allow_timestamp_ltz_to_ntz: false,
+            checked_timestamp_overflow: true,
         }
     }
 }
@@ -169,6 +177,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();
@@ -184,10 +201,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(
@@ -197,6 +215,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
@@ -328,10 +368,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 {
@@ -379,15 +420,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(
@@ -684,4 +727,71 @@ 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}"
+        );
+
+        // Filtered scans disable checked conversion because Spark may prune values before
+        // conversion through paths DataFusion cannot fully mirror.
+        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));
+    }
 }

native/core/src/parquet/schema_adapter.rs

@@ -667,12 +667,12 @@ impl SparkPhysicalExprAdapter {
                         }
 
                         let cast_expr: Arc<dyn PhysicalExpr> = Arc::new(
-                            CometCastColumnExpr::new(
+                            CometCastColumnExpr::try_new(
                                 remapped,
                                 Arc::clone(physical_field),
                                 Arc::clone(logical_field),
                                 None,
-                            )
+                            )?
                             .with_parquet_options(self.parquet_options.clone()),
                         );
                         return Ok(Transformed::yes(cast_expr));
@@ -956,12 +956,12 @@ impl SparkPhysicalExprAdapter {
                     | (DataType::Timestamp(_, _), DataType::Int64)
             ) {
                 let comet_cast: Arc<dyn PhysicalExpr> = Arc::new(
-                    CometCastColumnExpr::new(
+                    CometCastColumnExpr::try_new(
                         child,
                         input_field,
                         Arc::clone(cast.target_field()),
                         None,
-                    )
+                    )?
                     .with_parquet_options(self.parquet_options.clone()),
                 );
                 return Ok(Transformed::yes(comet_cast));

native/proto/src/proto/operator.proto

@@ -176,6 +176,9 @@ message NativeScanCommon {
   // SchemaColumnConvertNotSupportedException (Spark 3.x, SPARK-36182). Set
   // from Comet's per-Spark-version constant in ShimCometConf.
   bool allow_timestamp_ltz_to_ntz = 18;
+  // True when Spark supplied a Parquet data filter, including when Comet could
+  // not serialize any of those filters into data_filters.
+  bool has_data_filters = 19;
 }
 
 message NativeScan {

native/spark-expr/src/datetime_funcs/date_from_unix_date.rs

@@ -15,13 +15,12 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use arrow::array::{Array, Date32Array, Int32Array};
+use arrow::compute::cast_with_options;
 use arrow::datatypes::DataType;
-use datafusion::common::{utils::take_function_args, DataFusionError, Result, ScalarValue};
+use datafusion::common::{format::DEFAULT_CAST_OPTIONS, utils::take_function_args, Result};
 use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
 };
-use std::sync::Arc;
 
 /// Spark-compatible date_from_unix_date function.
 /// Converts an integer representing days since Unix epoch (1970-01-01) to a Date32 value.
@@ -61,33 +60,18 @@ impl ScalarUDFImpl for SparkDateFromUnixDate {
 
     fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
         let [unix_date] = take_function_args(self.name(), args.args)?;
+        // The input is guaranteed to be Int32 by `Signature::exact` and the Comet serde
+        // (Spark's DateFromUnixDate only accepts IntegerType), so the casts below take
+        // Arrow's zero-copy `Int32 -> Date32` reinterpret path.
         match unix_date {
-            ColumnarValue::Array(arr) => {
-                let int_array = arr.as_any().downcast_ref::<Int32Array>().ok_or_else(|| {
-                    DataFusionError::Execution(
-                        "date_from_unix_date expects Int32Array input".to_string(),
-                    )
-                })?;
-
-                // Date32 and Int32 both represent days since epoch, so we can directly
-                // reinterpret the values. The only operation needed is creating a Date32Array
-                // from the same underlying i32 values.
-                let date_array =
-                    Date32Array::new(int_array.values().clone(), int_array.nulls().cloned());
-
-                Ok(ColumnarValue::Array(Arc::new(date_array)))
+            ColumnarValue::Array(arr) => Ok(ColumnarValue::Array(cast_with_options(
+                arr.as_ref(),
+                &DataType::Date32,
+                &DEFAULT_CAST_OPTIONS,
+            )?)),
+            ColumnarValue::Scalar(scalar) => {
+                Ok(ColumnarValue::Scalar(scalar.cast_to(&DataType::Date32)?))
             }
-            ColumnarValue::Scalar(scalar) => match scalar {
-                ScalarValue::Int32(Some(days)) => {
-                    Ok(ColumnarValue::Scalar(ScalarValue::Date32(Some(days))))
-                }
-                ScalarValue::Int32(None) | ScalarValue::Null => {
-                    Ok(ColumnarValue::Scalar(ScalarValue::Date32(None)))
-                }
-                _ => Err(DataFusionError::Execution(
-                    "date_from_unix_date expects Int32 scalar input".to_string(),
-                )),
-            },
         }
     }
 

native/spark-expr/src/utils.rs

@@ -29,7 +29,6 @@ use std::sync::Arc;
 
 use arrow::array::timezone::Tz;
 use arrow::array::types::TimestampMillisecondType;
-use arrow::array::TimestampMicrosecondArray;
 use arrow::datatypes::{MAX_DECIMAL128_FOR_EACH_PRECISION, MIN_DECIMAL128_FOR_EACH_PRECISION};
 use arrow::error::ArrowError;
 use arrow::{
@@ -81,13 +80,6 @@ pub fn array_with_timezone(
                     // so the result has the exact annotation the caller expects.
                     timestamp_ntz_to_timestamp(array, timezone.as_str(), Some(target_tz.as_ref()))
                 }
-                Some(DataType::Timestamp(TimeUnit::Microsecond, None)) => {
-                    // Convert from Timestamp(Millisecond, None) to Timestamp(Microsecond, None)
-                    let millis_array = as_primitive_array::<TimestampMillisecondType>(&array);
-                    let micros_array: TimestampMicrosecondArray =
-                        arrow::compute::kernels::arity::unary(millis_array, |v| v * 1000);
-                    Ok(Arc::new(micros_array))
-                }
                 _ => {
                     // Not supported
                     Err(ArrowError::CastError(format!(
@@ -376,6 +368,7 @@ pub fn unlikely(b: bool) -> bool {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use arrow::array::TimestampMicrosecondArray;
 
     fn array_containing(local_datetime: &str) -> ArrayRef {
         let dt = NaiveDateTime::parse_from_str(local_datetime, "%Y-%m-%d %H:%M:%S").unwrap();

spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala

@@ -142,8 +142,10 @@ object CometNativeScan extends CometOperatorSerde[CometScanExec] with CometTypeS
       builder.clearChildren()
 
       if (scan.conf.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED)) {
+        val supportedDataFilters = scan.supportedDataFilters
+        commonBuilder.setHasDataFilters(supportedDataFilters.nonEmpty)
         val dataFilters = new ListBuffer[Expr]()
-        for (filter <- scan.supportedDataFilters) {
+        for (filter <- supportedDataFilters) {
           exprToProto(filter, scan.output) match {
             case Some(proto) => dataFilters += proto
             case _ =>

spark/src/main/scala/org/apache/spark/sql/comet/CometNativeScanExec.scala

@@ -200,6 +200,7 @@ case class CometNativeScanExec(
         })
       if (resolvedFilters.nonEmpty) {
         val commonBuilder = base.toBuilder
+        commonBuilder.setHasDataFilters(true)
         for (filter <- resolvedFilters) {
           exprToProto(filter, output) match {
             case Some(proto) => commonBuilder.addDataFilters(proto)