fix: report make_date Date32 overflow in both ANSI modes

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

@@ -23,7 +23,7 @@ import scala.jdk.CollectionConverters._
 
 import org.apache.hadoop.fs.Path
 import org.apache.spark.internal.Logging
-import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.{Literal, MakeDate}
 import org.apache.spark.sql.comet.{CometScanExec, DeltaPlanDataInjector}
 import org.apache.spark.sql.delta.DeltaParquetFileFormat
 import org.apache.spark.sql.delta.RowIndexFilterType
@@ -166,6 +166,8 @@ object CometDeltaNativeScan
     }
 
     commonOpt.map { commonBuilder =>
+      commonBuilder.setDisableRowFilterPushdown(
+        scanExec.dataFilters.exists(_.exists(_.isInstanceOf[MakeDate])))
       // Already forced by declineReason on this claim; reused rather than deserialized again.
       val dvDescriptors = memo.dvDescriptors
       // Union object-store options over every authority a partition of this scan may need a

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

@@ -715,14 +715,20 @@ 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
 before the consuming operator decides which rows it needs.
 
 By default, Comet keeps affected operators in Spark to preserve skipped date evaluation.
 This applies to date projections below `LIMIT`, `SAMPLE`, `FILTER`, joins,
-`mapPartitions`, and `Dataset.rdd` consumers; date expressions inside joins, aggregates,
+`mapPartitions`, and external row iterators, including Scala's `Dataset.rdd` and PySpark's
+`DataFrame.rdd`; date expressions inside joins, aggregates,
 and ordered limits; and projections below a window group limit. A parent projection can
 also defer a date output when an `IF`, `COALESCE`, or NULL-skipping operand does not need
 it. Comet preserves that evaluation across projections while allowing eager date consumers,
@@ -735,6 +741,10 @@ the join rejects before producing an output row.
 Selecting the date eagerly does not force conditionally used sibling expressions; those
 computations also retain Spark evaluation when needed.
 
+The physical plan can be reused by a later RDD consumer, so uncapped root date projections
+also retain these safeguards, even when the current action collects the complete result.
+Materializing exchanges, sorts, and aggregates can still allow native date evaluation below them.
+
 With ANSI disabled, a direct date projection can remain native below these consumers when
 all its computations are safe to evaluate early. This includes `make_date(2024, month, day)`
 over input columns and years widened from `TINYINT` or `SMALLINT`. An unrestricted integer
@@ -750,18 +760,27 @@ because Spark checks that argument on every input row. Filtered counts and count
 multiple arguments retain the safeguard when they can skip date evaluation.
 
 Only affected parts of the plan fall back. An ordinary native scan can remain native.
-An adjacent collect limit and date projection can also stay native: Comet caps the projection's
-input before evaluating dates, so unread rows cannot raise errors. This requires a supported
+An adjacent `LIMIT 1` without `OFFSET` and date projection can also stay native: Comet caps
+the projection's input before evaluating dates, so unread rows cannot raise errors. Larger
+root limits retain the safeguard because an RDD caller can stop before reaching the limit.
+The cap requires a supported
 native local limit; otherwise the Spark safeguard remains. The cap does not bypass safeguards
 for a lower argument-producing projection or intervening filters and joins.
 Exchanges materialize their inputs and partition keys independently, so an outer limit does
 not force an eager date partition key to Spark. Local sorts and aggregates can also separate
 date evaluation from an outer limit.
 
+Predicates containing `make_date` stay above the Parquet scan. Comet disables row-level filter
+pushdown for that scan so predicates cannot be reordered or hide a required date error, and
+Spark's structured error is preserved. This safeguard uses Spark's original predicates, including
+those that cannot be serialized for native execution. Other predicates can still use metadata to
+skip row groups and pages when `spark.sql.parquet.filterPushdown` is enabled. That setting also
+applies to filters resolved from scalar subqueries at execution time.
+
 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
+that Spark would skip, including Date32 overflow with ANSI disabled. 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`
 still falls back when needed.

native/core/src/execution/delta_dv.rs

@@ -1592,6 +1592,7 @@ mod tests {
                 false,
                 "",
                 "",
+                false,
             )
             .unwrap();
             let batches: Vec<_> = scan

native/core/src/execution/operators/dynamic_filter/join/tests.rs

@@ -673,6 +673,7 @@ fn parquet_probe(
         false,
         "CORRECTED",
         "CORRECTED",
+        false,
     )
     .unwrap();
     (file, scan)
@@ -827,6 +828,7 @@ async fn reader_filter_crosses_null_check_conjunction_and_retains_residual() {
             false,
             "CORRECTED",
             "CORRECTED",
+            false,
         )
         .unwrap();
         let checks = [("key", 0), ("payload", 1), ("other", 2)].map(|(name, index)| {

native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors.rs

@@ -89,6 +89,7 @@ fn scan(
         false,
         "CORRECTED",
         "CORRECTED",
+        false,
     )
     .unwrap()
 }

native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors/partition_columns.rs

@@ -74,6 +74,7 @@ fn partitioned_scan(
         false,
         "CORRECTED",
         "CORRECTED",
+        false,
     )
     .unwrap()
 }

native/core/src/execution/operators/dynamic_filter/join/tests/timestamp_errors.rs

@@ -108,6 +108,7 @@ async fn assert_timestamp_overflow_preserved(nested: bool) {
             false,
             "CORRECTED",
             "CORRECTED",
+            false,
         )
         .unwrap();
         let join = single_key_join_plans(

native/core/src/execution/operators/dynamic_filter/topk/tests.rs

@@ -199,6 +199,7 @@ fn parquet_scan(
         false,
         "CORRECTED",
         "CORRECTED",
+        false,
     )
     .unwrap()
 }

native/core/src/execution/operators/dynamic_filter/topk/tests/timestamp.rs

@@ -98,6 +98,7 @@ fn timestamp_input(
         false,
         "CORRECTED",
         "CORRECTED",
+        false,
     )
     .unwrap();
     (file, scan)

native/core/src/execution/planner.rs

@@ -494,6 +494,7 @@ impl PhysicalPlanner {
             rebase_from_file_metadata,
             datetime_rebase_mode_in_read,
             int96_rebase_mode_in_read,
+            common.disable_row_filter_pushdown,
         )?;
         Ok(Arc::new(SparkPlan::new(plan_id, scan, vec![])))
     }

native/core/src/parquet/parquet_exec.rs

@@ -39,7 +39,7 @@ use datafusion::physical_expr_adapter::PhysicalExprAdapterFactory;
 use datafusion::prelude::SessionContext;
 use datafusion::scalar::ScalarValue;
 use datafusion_comet_spark_expr::jvm_udf::JvmScalarUdfExpr;
-use datafusion_comet_spark_expr::EvalMode;
+use datafusion_comet_spark_expr::{EvalMode, MakeDateExpr};
 use datafusion_datasource::TableSchema;
 use parquet::variant::VariantType;
 use std::collections::HashMap;
@@ -83,6 +83,9 @@ pub(crate) struct CometParquetScanOptions {
 ///   `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.
+///
+///   `disable_row_filter_pushdown`: Keep row filtering above the scan, including when the
+/// throwing Spark predicate is absent from `data_filters` because it could not be serialized.
 #[allow(clippy::too_many_arguments)]
 pub(crate) fn init_datasource_exec(
     required_schema: SchemaRef,
@@ -106,12 +109,27 @@ pub(crate) fn init_datasource_exec(
     rebase_from_file_metadata: bool,
     datetime_rebase_mode_in_read: &str,
     int96_rebase_mode_in_read: &str,
+    disable_row_filter_pushdown: bool,
 ) -> Result<Arc<DataSourceExec>, ExecutionError> {
     // Computed once and reused below for `try_pushdown_filters`. `copied_config()` clones only
     // `SessionConfig` (an `Arc<ConfigOptions>` plus a small extensions map); `SessionContext::
     // state()` would clone the whole `SessionState`, including the function registries (~250+
     // UDFs) and optimizer rules, on every scan init.
-    let session_config = session_ctx.copied_config();
+    let mut session_config = session_ctx.copied_config();
+    if disable_row_filter_pushdown
+        || data_filters
+            .as_ref()
+            .is_some_and(|filters| filters.iter().any(contains_make_date))
+    {
+        // RowFilter can reorder predicates and flatten their Spark errors. Even pushing only
+        // a later sibling could hide a make_date error that Spark evaluates first. Keep all
+        // row filtering above this scan, while retaining metadata pruning for other predicates.
+        session_config
+            .options_mut()
+            .execution
+            .parquet
+            .pushdown_filters = false;
+    }
     let (table_parquet_options, mut spark_parquet_options) = get_options(
         session_timezone,
         case_sensitive,
@@ -291,11 +309,12 @@ pub(crate) fn init_datasource_exec(
     // `propagation.parent_pushdown_result` because Spark's Filter above the
     // scan re-evaluates every dataFilter, so No-classified filters stay
     // correct without us inserting a FilterExec here. That also makes it safe
-    // to drop the filters that call into the JVM first; see `calls_jvm`.
+    // to drop the filters that call into the JVM first; see `calls_jvm`. A make_date predicate
+    // also stays above the scan so its evaluation and structured errors remain with the Filter.
     let data_filters = data_filters.map(|filters| {
         filters
             .into_iter()
-            .filter(|filter| !calls_jvm(filter))
+            .filter(|filter| !calls_jvm(filter) && !contains_make_date(filter))
             .collect::<Vec<_>>()
     });
     let file_source: Arc<dyn FileSource> = match data_filters {
@@ -351,6 +370,11 @@ fn calls_jvm(expr: &Arc<dyn PhysicalExpr>) -> bool {
         .unwrap_or(true)
 }
 
+fn contains_make_date(expr: &Arc<dyn PhysicalExpr>) -> bool {
+    expr.exists(|e| Ok(e.downcast_ref::<MakeDateExpr>().is_some()))
+        .unwrap_or(true)
+}
+
 // Registration URLs use a reserved suffix to distinguish backend/configuration
 // identities. Encryption must use the physical URI that Spark registered. Match
 // the complete suffix, from the right, so custom Hadoop schemes are preserved.
@@ -578,6 +602,7 @@ mod tests {
             false,
             "",
             "",
+            false,
         )
         .unwrap()
     }
@@ -715,6 +740,7 @@ mod tests {
             rebase_from_file_metadata,
             "",
             "",
+            false,
         )
         .unwrap();
 
@@ -791,6 +817,7 @@ mod tests {
             true,
             datetime_rebase_mode,
             datetime_rebase_mode,
+            false,
         )
         .unwrap();
 
@@ -900,6 +927,7 @@ mod tests {
                 true,
                 "CORRECTED",
                 "CORRECTED",
+                false,
             )
             .unwrap();
 
@@ -996,6 +1024,7 @@ mod tests {
             true,
             datetime_rebase_mode,
             int96_rebase_mode,
+            false,
         )
         .unwrap();
 
@@ -1143,6 +1172,7 @@ mod tests {
             true,
             datetime_rebase_mode,
             int96_rebase_mode,
+            false,
         )
         .unwrap();
 
@@ -1412,6 +1442,7 @@ mod tests {
             false,
             "",
             "",
+            false,
         )
         .unwrap();
 

native/core/src/parquet/parquet_exec/variant_tests.rs

@@ -166,6 +166,7 @@ async fn scan_variant_file(filename: PathBuf) -> VariantArray {
         false,
         "",
         "",
+        false,
     )
     .unwrap();
     let mut stream = scan.execute(0, session_ctx.task_ctx()).unwrap();
@@ -234,6 +235,7 @@ async fn unread_variant_does_not_override_arrow_schema_hint() {
         false,
         "",
         "",
+        false,
     )
     .unwrap();
     let mut stream = scan.execute(0, session.task_ctx()).unwrap();
@@ -267,6 +269,7 @@ fn encrypted_projected_variant_is_rejected_before_reader_creation() {
         false,
         "",
         "",
+        false,
     );
     assert!(result
         .unwrap_err()

native/proto/src/proto/operator.proto

@@ -185,6 +185,9 @@ message NativeScanCommon {
   // 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;
+  // Keep row filtering above this scan when an original Spark data filter can
+  // throw, even if that filter could not be serialized into data_filters.
+  bool disable_row_filter_pushdown = 20;
 }
 
 message NativeScan {