fix: report make_date Date32 overflow in both ANSI modes

contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/CometDeltaNativeScan.scala

@@ -353,11 +353,13 @@ object CometDeltaNativeScan
           case s: ExecScalarSubquery =>
             Literal.create(s.eval(null), s.dataType)
         }
-        val proto = exprToProto(resolved, output)
-        if (proto.isEmpty) {
-          logWarning(s"Could not serialize resolved scalar subquery filter: $resolved")
+        CometNativeScan.filterForParquetPushdown(resolved).flatMap { pushable =>
+          val proto = exprToProto(pushable, output)
+          if (proto.isEmpty) {
+            logWarning(s"Could not serialize resolved scalar subquery filter: $pushable")
+          }
+          proto
         }
-        proto
       }
     ResolvedSubqueryFilters(hasResolvedFilters = true, protos)
   }
@@ -432,6 +434,7 @@ object CometDeltaNativeScan
       val dataColIds = output.take(strippedLen).map(_.exprId).toSet
       val filterProtos = scanHelper.supportedDataFilters
         .filter(_.references.forall(r => dataColIds.contains(r.exprId)))
+        .flatMap(CometNativeScan.filterForParquetPushdown)
         .flatMap(f => exprToProto(f, output))
       commonBuilder.addAllDataFilters(filterProtos.asJava)
     }

contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/DeltaScanContribSuite.scala

@@ -32,6 +32,7 @@ import org.apache.hadoop.fs.{Path, RawLocalFileSystem}
 import org.apache.hadoop.fs.s3a.{S3AFileSystem, S3AUtils}
 import org.apache.hadoop.security.alias.CredentialProviderFactory
 import org.apache.spark.sql.comet.CometDeltaNativeScanExec
+import org.apache.spark.sql.delta.DeltaLog
 import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor
 import org.apache.spark.sql.execution.SparkPlan
 
@@ -92,6 +93,47 @@ class DeltaScanContribSuite extends CometDeltaTestBase {
     }
   }
 
+  test("make_date filters preserve Spark's reader before the Delta scan claim") {
+    withSQLConf(
+      "spark.sql.ansi.enabled" -> "false",
+      "spark.sql.adaptive.enabled" -> "false",
+      CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.key -> "true") {
+      withTempPath { dir =>
+        val path = dir.getAbsolutePath
+        spark
+          .range(0, 3, 1, 1)
+          .selectExpr("id", "CAST(id + 2024 AS INT) AS y", "IF(id = 1, 13, 1) AS m")
+          .write
+          .format("delta")
+          .save(path)
+
+        for (deletionVectors <- Seq(false, true)) {
+          if (deletionVectors) {
+            spark.sql(
+              s"ALTER TABLE delta.`$path` SET TBLPROPERTIES ('delta.enableDeletionVectors' = 'true')")
+            spark.sql(s"DELETE FROM delta.`$path` WHERE id = 2")
+            assert(
+              DeltaLog
+                .forTable(spark, path)
+                .update()
+                .allFiles
+                .collect()
+                .exists(_.deletionVector != null))
+          }
+          val input = spark.read.format("delta").load(path)
+          val guarded = input.filter("unix_date(make_date(y, m, 1)) > 0")
+          checkSparkAnswer(guarded)
+          assert(deltaNativeScans(guarded).isEmpty)
+          assert(
+            collectByName(guarded.queryExecution.executedPlan, "FileSourceScanExec").nonEmpty)
+
+          // A bounded ANSI-off year is safe, including the invalid month that returns NULL.
+          checkDeltaNativeScanAnswer(input.filter("unix_date(make_date(2024, m, 1)) > 0"))
+        }
+      }
+    }
+  }
+
   test("declined scan carries the contrib's fallback reason, not core's generic one") {
     // Disabling Spark's vectorized Parquet reader is a scan the contrib recognizes
     // (DeltaScanSupport.isDeltaScan) but explicitly declines (DeltaScanSupport.declineReason,

docs/source/user-guide/latest/expressions.md

@@ -715,6 +715,11 @@ This list is illustrative, not exhaustive: the per-function tables are not the c
 
 ## make_date evaluation
 
+A valid calendar date outside Date32's range raises `ARITHMETIC_OVERFLOW` in both ANSI
+modes. For example, `make_date(6000000, 1, 1)` raises an error even with ANSI disabled.
+Invalid calendar fields still return NULL with ANSI disabled and raise a date error with
+ANSI enabled; a NULL argument returns NULL before date validation.
+
 With ANSI mode enabled, `make_date(2024, 13, 1)` raises an invalid-date error. If an
 input iterator yields a valid row followed by that invalid row, a `LIMIT 1` can stop
 before evaluating the second date. A native projection can evaluate a whole batch
@@ -758,12 +763,20 @@ Exchanges materialize their inputs and partition keys independently, so an outer
 not force an eager date partition key to Spark. Local sorts and aggregates can also separate
 date evaluation from an outer limit.
 
+Parquet scans below a filter whose `make_date` can throw use Spark's reader. This preserves
+Spark's dictionary and metadata pruning, which determine whether an overflowing date is read,
+as well as its predicate order and structured errors. The safeguard inspects complete filters,
+including projected date arguments and predicates absent from the native scan's data filters.
+ANSI-off dates with bounded years and safe arguments can retain the native reader and sibling
+row filtering. Predicates containing `make_date` remain above a native scan without weakening
+their Boolean structure. Other native scans retain their existing pruning behavior.
+
 Users who accept eagerly evaluating dates, projected arguments, and sibling expressions can set
 `spark.comet.exec.makeDate.operatorFallback.enabled=false` to allow native operators for
 these shapes. The default is `true`. Disabling it can expose invalid-date or argument errors
-that Spark would skip under ANSI mode. This setting does not change
-`make_date`'s invalid-date handling or the expression-level safeguards for skipped
-arguments and operands. For example, an empty `IN` predicate containing `make_date`
+that Spark would skip, including Date32 overflow with ANSI disabled. This setting does not change
+`make_date`'s invalid-date handling, the Spark-reader safeguard, or the expression-level safeguards
+for skipped arguments and operands. For example, an empty `IN` predicate containing `make_date`
 still falls back when needed.
 
 ## See also

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

@@ -28,7 +28,7 @@ use std::fmt::{Display, Formatter};
 use std::hash::{Hash, Hasher};
 use std::sync::Arc;
 
-use crate::{conversion_funcs::ymd_to_epoch_day, SparkError};
+use crate::{arithmetic_overflow_error, conversion_funcs::ymd_to_epoch_day, SparkError};
 
 /// Skips the entire date expression on empty input, before any argument can raise an error.
 /// `inner` retains the planner's argument evaluation, including NULL short-circuiting when enabled.
@@ -107,6 +107,7 @@ impl PhysicalExpr for MakeDateExpr {
 /// Creates a date from year, month, and day columns.
 /// For an invalid `(year, month, day)` triple Spark returns NULL when `spark.sql.ansi.enabled` is
 /// false, and throws otherwise. The ANSI flag is carried here as `fail_on_error`.
+/// Valid calendar dates outside Date32 raise arithmetic overflow in either mode.
 #[derive(Debug, PartialEq, Eq, Hash)]
 pub struct SparkMakeDate {
     signature: Signature,
@@ -174,10 +175,17 @@ fn cast_to_int32(arr: &Arc<dyn Array>) -> Result<Arc<dyn Array>> {
 }
 
 /// Convert year, month, day to days since Unix epoch (1970-01-01).
-/// Returns None if the date is invalid or its epoch day does not fit Date32.
-fn make_date(year: i32, month: i32, day: i32) -> Option<i32> {
-    ymd_to_epoch_day(year.into(), month.into(), day.into())
-        .and_then(|days| i32::try_from(days).ok())
+/// Invalid Java calendar dates return None; valid dates outside Date32 raise arithmetic overflow.
+fn make_date(year: i32, month: i32, day: i32) -> Result<Option<i32>> {
+    if !(-999_999_999..=999_999_999).contains(&year) {
+        return Ok(None);
+    }
+    let Some(days) = ymd_to_epoch_day(year.into(), month.into(), day.into()) else {
+        return Ok(None);
+    };
+    i32::try_from(days)
+        .map(Some)
+        .map_err(|_| arithmetic_overflow_error("integer").into())
 }
 
 impl ScalarUDFImpl for SparkMakeDate {
@@ -247,7 +255,7 @@ impl ScalarUDFImpl for SparkMakeDate {
                 let m = month_array.value(i);
                 let d = day_array.value(i);
 
-                match make_date(y, m, d) {
+                match make_date(y, m, d)? {
                     Some(days) => builder.append_value(days),
                     None => {
                         if self.fail_on_error {
@@ -278,6 +286,92 @@ mod tests {
     use datafusion::physical_expr::expressions::{col, lit, Literal};
     use datafusion::physical_expr::{PhysicalExpr, ScalarFunctionExpr};
 
+    fn invoke_make_date(values: [Option<i32>; 3], fail_on_error: bool) -> Result<ColumnarValue> {
+        SparkMakeDate::new(fail_on_error).invoke_with_args(ScalarFunctionArgs {
+            args: values
+                .into_iter()
+                .map(|value| ColumnarValue::Array(Arc::new(Int32Array::from(vec![value]))))
+                .collect(),
+            arg_fields: vec![],
+            number_rows: 1,
+            return_field: Arc::new(Field::new("make_date", DataType::Date32, true)),
+            config_options: Arc::new(ConfigOptions::default()),
+        })
+    }
+
+    #[test]
+    fn valid_dates_outside_date32_overflow_in_both_modes() {
+        for values in [
+            [5_881_580, 7, 12],
+            [-5_877_641, 6, 22],
+            [999_999_999, 1, 1],
+            [-999_999_999, 1, 1],
+        ] {
+            for fail_on_error in [false, true] {
+                let error = invoke_make_date(values.map(Some), fail_on_error).unwrap_err();
+                let DataFusionError::External(error) = error else {
+                    panic!("Expected typed SparkError, got {error:?}");
+                };
+                assert!(matches!(
+                    error.downcast_ref::<SparkError>(),
+                    Some(SparkError::ArithmeticOverflow { from_type, function_name })
+                        if from_type == "integer" && function_name.is_empty()
+                ));
+            }
+        }
+    }
+
+    #[test]
+    fn invalid_calendar_fields_take_precedence_over_date32_overflow() {
+        for (values, expected) in [
+            (
+                [1_000_000_000, 13, 32],
+                "Invalid value for Year (valid values -999999999 - 999999999): 1000000000",
+            ),
+            (
+                [-1_000_000_000, 1, 1],
+                "Invalid value for Year (valid values -999999999 - 999999999): -1000000000",
+            ),
+            (
+                [6_000_000, 13, 32],
+                "Invalid value for MonthOfYear (valid values 1 - 12): 13",
+            ),
+            ([6_000_000, 2, 30], "Invalid date 'FEBRUARY 30'"),
+        ] {
+            let result = invoke_make_date(values.map(Some), false)
+                .unwrap()
+                .into_array(1)
+                .unwrap();
+            assert!(result.is_null(0));
+            let error = invoke_make_date(values.map(Some), true).unwrap_err();
+            let DataFusionError::External(error) = error else {
+                panic!("Expected typed SparkError, got {error:?}");
+            };
+            assert!(matches!(
+                error.downcast_ref::<SparkError>(),
+                Some(SparkError::DatetimeFieldOutOfBounds { range_message })
+                    if range_message == expected
+            ));
+        }
+    }
+
+    #[test]
+    fn null_arguments_skip_date32_overflow() {
+        for values in [
+            [None, Some(13), Some(32)],
+            [Some(6_000_000), None, Some(1)],
+            [Some(6_000_000), Some(1), None],
+        ] {
+            for fail_on_error in [false, true] {
+                let result = invoke_make_date(values, fail_on_error)
+                    .unwrap()
+                    .into_array(1)
+                    .unwrap();
+                assert!(result.is_null(0));
+            }
+        }
+    }
+
     #[test]
     fn scalar_arguments_do_not_validate_dates_on_empty_batches() {
         let expression = MakeDateExpr::new(Arc::new(ScalarFunctionExpr::new(
@@ -359,54 +453,55 @@ mod tests {
     #[test]
     fn test_make_date_valid() {
         // Unix epoch
-        assert_eq!(make_date(1970, 1, 1), Some(0));
+        assert_eq!(make_date(1970, 1, 1).unwrap(), Some(0));
         // Day after epoch
-        assert_eq!(make_date(1970, 1, 2), Some(1));
+        assert_eq!(make_date(1970, 1, 2).unwrap(), Some(1));
         // Day before epoch
-        assert_eq!(make_date(1969, 12, 31), Some(-1));
+        assert_eq!(make_date(1969, 12, 31).unwrap(), Some(-1));
         // Leap years - just verify they return Some (valid dates)
-        assert!(make_date(2000, 2, 29).is_some()); // 2000 is a leap year
-        assert!(make_date(2004, 2, 29).is_some()); // 2004 is a leap year
-                                                   // Regular date
-        assert!(make_date(2023, 6, 15).is_some());
+        assert!(make_date(2000, 2, 29).unwrap().is_some()); // 2000 is a leap year
+        assert!(make_date(2004, 2, 29).unwrap().is_some()); // 2004 is a leap year
+
+        // Regular date
+        assert!(make_date(2023, 6, 15).unwrap().is_some());
     }
 
     #[test]
     fn test_make_date_invalid_month() {
-        assert_eq!(make_date(2023, 0, 15), None);
-        assert_eq!(make_date(2023, 13, 15), None);
-        assert_eq!(make_date(2023, -1, 15), None);
+        assert_eq!(make_date(2023, 0, 15).unwrap(), None);
+        assert_eq!(make_date(2023, 13, 15).unwrap(), None);
+        assert_eq!(make_date(2023, -1, 15).unwrap(), None);
     }
 
     #[test]
     fn test_make_date_invalid_day() {
-        assert_eq!(make_date(2023, 6, 0), None);
-        assert_eq!(make_date(2023, 6, 32), None);
-        assert_eq!(make_date(2023, 6, -1), None);
+        assert_eq!(make_date(2023, 6, 0).unwrap(), None);
+        assert_eq!(make_date(2023, 6, 32).unwrap(), None);
+        assert_eq!(make_date(2023, 6, -1).unwrap(), None);
     }
 
     #[test]
     fn test_make_date_invalid_dates() {
         // Feb 30 never exists
-        assert_eq!(make_date(2023, 2, 30), None);
+        assert_eq!(make_date(2023, 2, 30).unwrap(), None);
         // Feb 29 on non-leap year
-        assert_eq!(make_date(2023, 2, 29), None);
+        assert_eq!(make_date(2023, 2, 29).unwrap(), None);
         // 1900 is not a leap year (divisible by 100 but not 400)
-        assert_eq!(make_date(1900, 2, 29), None);
+        assert_eq!(make_date(1900, 2, 29).unwrap(), None);
         // 2100 will not be a leap year
-        assert_eq!(make_date(2100, 2, 29), None);
+        assert_eq!(make_date(2100, 2, 29).unwrap(), None);
         // April has 30 days
-        assert_eq!(make_date(2023, 4, 31), None);
+        assert_eq!(make_date(2023, 4, 31).unwrap(), None);
     }
 
     #[test]
     fn test_make_date_wide_year_range() {
-        assert_eq!(make_date(0, 1, 1), Some(-719_528));
-        assert_eq!(make_date(-1, 1, 1), Some(-719_893));
-        assert_eq!(make_date(300_000, 6, 15), Some(108_853_388));
-        assert_eq!(make_date(300_000, 2, 29), Some(108_853_281));
+        assert_eq!(make_date(0, 1, 1).unwrap(), Some(-719_528));
+        assert_eq!(make_date(-1, 1, 1).unwrap(), Some(-719_893));
+        assert_eq!(make_date(300_000, 6, 15).unwrap(), Some(108_853_388));
+        assert_eq!(make_date(300_000, 2, 29).unwrap(), Some(108_853_281));
 
-        assert_eq!(make_date(5_881_580, 7, 11), Some(i32::MAX));
-        assert_eq!(make_date(-5_877_641, 6, 23), Some(i32::MIN));
+        assert_eq!(make_date(5_881_580, 7, 11).unwrap(), Some(i32::MAX));
+        assert_eq!(make_date(-5_877_641, 6, 23).unwrap(), Some(i32::MIN));
     }
 }