[pingzh]

Which issue does this PR close?

Closes #6741. Final part of the split of #7117, following #7260, #7261, and #7283.

Rationale for this change

make_date(6000000, 1, 1) is a valid Java calendar date, but its epoch day exceeds Date32. Spark raises ARITHMETIC_OVERFLOW in both ANSI modes. Comet currently loses the overflow distinction and returns NULL with ANSI disabled or reports an invalid calendar date with ANSI enabled.

What changes are included in this PR?

Separate calendar validation from checked Date32 conversion and propagate overflow through the existing typed Spark error path. Invalid calendar fields retain their ANSI-dependent behavior, including Java's year limits; NULL arguments still skip validation.

Preserve partial consumption through external row iterators, including PySpark DataFrame.rdd and preplanned DataFrames. Potentially throwing root projections use Spark; safe ANSI-off projections, materialized inputs, and a direct LIMIT 1 without OFFSET can remain native. Larger root limits keep the safeguard because an RDD consumer can stop before the limit. The existing supported operator-fallback opt-out remains available.

Keep make_date predicates above Parquet scans and disable sibling row filters for affected scans, preserving predicate order and structured Spark errors. Other predicates retain metadata pruning.

Add coverage for Date32 endpoints and adjacent overflow, calendar-error precedence, NULL and empty input, and overflowing rows skipped by LIMIT, RDD.take, and mapPartitions. Document the overflow behavior.

How are these changes tested?

• The new native overflow regression fails before the fix; all 10 focused make_date Rust tests pass after it.
• Rust formatting and focused Clippy checks (-D warnings) pass.
• Full JVM reactor compilation and format/style checks pass with Spark 4.1.3 / JDK 17.
• The Python partial-consumption and Parquet predicate regressions fail before their fixes and pass afterward. Across the focused runs, all 133 JVM tests pass: the complete planner suite, both SQL files, temporal/Python-bridge checks, the empty-IN safeguard, and Parquet predicate/control tests. Potentially throwing expression fixtures materialize their results so they continue to verify native execution.
• All 24 public PySpark 4.1.3 DataFrame.rdd cases pass with ANSI and AQE on/off, comparing vanilla Spark and Comet for preplanned unlimited, LIMIT 4096, and LIMIT 1 DataFrames.
• The JNI build, JVM package build, and native Parquet Clippy check pass.
• Before merging, request run-spark-4.1-tests and run-all-spark-profiles for Spark SQL and version coverage.

[ajsquared]

Reviewed the current changes and feedback; no independently confirmed unresolved P1 remains.

[sunchao]

Reviewed the current head, including calendar validation, Date32 boundaries and the error path. A Java comparison covered 105,304 calendar cases and focused Rust CI passed; JVM CI was still running, Spark SQL suites were skipped, and local end-to-end Comet testing was unavailable.

[sunchao]

I reproduced two cases on Spark 4.1.3 where this head raises an overflow for a row that vanilla Spark skips. Both queries returned the Spark result with the merge-base native library (15305af). The inline comments include the fixtures and execution paths.

[ajsquared]

2 review findings.

[ajsquared]

[P2] With row-filter pushdown enabled, a date predicate that stays in Spark can still lose a required overflow. For example, use WHERE (y = 0 OR unix_date(make_date(y,1,1)) > 0) AND keep = 1 on a Parquet page containing both keep values, with y = 6000000 on a keep = 0 row. Spark evaluates the date on that row. unsupportedParentReason rejects the OR, so CometNativeScan.buildNativeScanCommon omits that conjunct but still serializes keep = 1. The check here sees no MakeDateExpr, and the reader drops the row before Spark can raise ARITHMETIC_OVERFLOW. Detect MakeDate in the original Catalyst data filters before serialization and carry that decision into the scan.

[ajsquared]

[P2] With spark.sql.parquet.filterPushdown=false, a resolved scalar-subquery predicate can still prune an entire row group and suppress a required overflow. For example, use WHERE unix_date(make_date(y,1,1)) > (SELECT max(v) FROM bounds) AND keep > (SELECT max(v) FROM bounds), with a bound of zero and a row group where every keep is zero and one year is 6000000. Spark reads the group and raises on the first conjunct. CometNativeScanExec.serializedPartitionData appends both resolved filters without checking the pushdown setting. This code removes the date predicate but still passes keep > 0 to Parquet for statistics pruning, so Comet can return an empty result. Apply the same PARQUET_FILTER_PUSHDOWN_ENABLED check to execution-time scalar-subquery filters that buildNativeScanCommon applies at planning time.

[ajsquared]

Reviewed the current changes and feedback; no independently confirmed unresolved P1 remains.