[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 and date wrappers, materialized inputs, complete file writes, and a direct LIMIT 1 without OFFSET can remain native. Limits or filters inside a write retain their safeguards, and residual date filters retain the native Parquet reader. 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. Derive this guard from enclosing Spark filters and the original Catalyst predicates so it survives partition-column, nondeterministic, unsupported-expression, and unresolved-subquery filtering, including in Delta scans. Exclude each original predicate containing make_date before partially rewriting metadata predicates, so rewriting cannot hide an overflow. Independent predicates retain metadata pruning according to Spark's Parquet settings, including type-specific opt-outs, safe AND/OR/NOT composition, and Spark's IN threshold. Apply the same settings when scalar-subquery filters are resolved at execution time.

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?

Focused validation used JDK 17:

• Six regression cases reproduced the earlier five review findings before their fixes. Two additional regressions fail before the partial-rewriting fix and pass afterward; the end-to-end case verifies independent row-group pruning through native metrics.
• Spark 4.1.3: all 12 focused scan and serialization tests pass, including overflow parity, independent metadata pruning, and scalar-filter controls. The same tests also pass with the Delta profile. Prior validation passed 27 focused JVM tests including lazy-evaluation safety controls.
• Spark 3.5: strict compiler checks and all 11 focused scan and serialization tests pass. Prior validation also covered the separate native-write path.
• The JNI rebuild and full Maven reactor build, including Delta test compilation, pass.

Validation before the latest Scala changes also passed 10 native make_date tests, 28 native Parquet tests, 133 broader JVM tests, and 24 public PySpark DataFrame.rdd cases with ANSI and AQE on/off. The native overflow regression, unsupported-OR predicate regression, and disabled-pushdown scalar-subquery regression each failed before their respective fixes.

Expanded CI coverage still needs run-spark-4.1-tests, run-all-spark-profiles, and run-delta-tests before merging.

[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]

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

[sunchao]

Both earlier repros now pass on Spark 4.1.3 at c5c6793. PySpark df.rdd.take(1) returns 19723, and the original Parquet mask query returns 1023 with row-filter pushdown enabled. I found two other scan cases where Comet suppresses an overflow that Spark raises. I also compared physical plans with merge base 0f90a7d and found three cases where the new safeguard drops native execution unnecessarily.

[ajsquared]

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

[ajsquared]

1 review finding.

[sunchao]

I reproduced Parquet queries that suppress a required ARITHMETIC_OVERFLOW and others that raise it after Spark would have skipped the row. The inline comments include the SQL and controls. The new PySpark RDD bridge test and partial-consumption test both passed locally on Spark 4.1.3.

I also rechecked the bounded-projection discussion. unix_date is covered now. month(make_date(2024,m,d)) over INT columns still moves from CometProject on the base to a Spark Project on this head, although every input is safe to evaluate with ANSI disabled.

The fork’s Rust tests and four default Spark 4.1 Comet shards passed. The expanded Spark SQL and Delta runtime jobs were skipped.

[sunchao]

[P2] Check whether Spark can translate the whole sibling predicate before simplifying it

Use one Parquet row group with a=b=0, d=DATE '2020-01-01', and one y=6000000 among otherwise valid years. With spark.sql.parquet.filterPushdown.date=false and Comet row filtering off, this returns 0 on this head. Spark and the base production code both raise ARITHMETIC_OVERFLOW.

SELECT count(*) FROM t
WHERE unix_date(make_date(y,1,1)) > 0
  AND ((months_between(d, DATE '2020-01-01') > 0 AND a=1) OR b=1)

The date-producing conjunct is already excluded. Spark’s V1 translation rejects the separate OR because it cannot translate months_between. Here l.orElse(r) strips that expression and sends a=1 OR b=1 to Parquet, which discards the group. Restoring only the date pushdown setting makes this head raise again. Please gate the sibling on Spark’s V1 translation before applying Parquet’s partial rewrite. A simpler example with default Spark settings is s LIKE 'zzz_%' on an all-'aaa' group. Spark leaves this complex pattern above the scan and raises, while Comet prunes the group. That case also reproduces on the base.

[sunchao]

[P2] Account for the Parquet timestamp encoding when retaining metadata filters

On Spark 4.1.3, write one row group with nt=TIMESTAMP_NTZ '2020-01-01' throughout and one overflowing year. For WHERE unix_date(make_date(y,1,1)) > 0 AND nt < TIMESTAMP_NTZ '2000-01-01', Spark raises ARITHMETIC_OVERFLOW. With Comet row filtering off and default Spark pushdown settings, the native scan returns count 0. Disabling only spark.sql.parquet.filterPushdown.timestamp restores the error.

Spark’s Parquet comparison accepts UTC-adjusted timestamp encodings. An NTZ column has isAdjustedToUTC=false, so enabling timestamp pushdown does not make this predicate eligible. This mismatch also exists on the base. Please account for Spark’s physical-type eligibility before retaining the sibling for a throwing date filter.

[sunchao]

[P2] Use the effective scan setting when planning floating-point comparisons

With one Parquet row group where f=42D throughout and one y=6000000, count rows satisfying unix_date(make_date(y,1,1)) > 0 AND f < 0D. Spark and the base return 0 with Comet row filtering enabled. This head raises ARITHMETIC_OVERFLOW. Switching only spark.comet.parquet.rowFilterPushdown.enabled to false restores 0.

The planner has already normalized the floating-point operand using the session-wide setting by the time this line disables row filtering. Parquet can no longer use the plain column comparison for statistics pruning. Please choose the operand representation after accounting for the per-scan guard.

[sunchao]

[P2] Preserve Spark’s dictionary pruning when a residual date filter can throw

In one dictionary-encoded Parquet row group, alternate k between 1 and 3, and put y=6000000 on the final row. For WHERE unix_date(make_date(y,1,1)) > 0 AND k=2, Spark returns count 0. The dictionary rules out 2 even though it falls within the min/max range. With the native scan, this head raises ARITHMETIC_OVERFLOW with Comet row filtering both on and off. The base returned 0 with row filtering on.

Disabling only spark.hadoop.parquet.filter.dictionary.enabled makes Spark raise too. As a control, k=5 returns 0 in both readers through min/max pruning. Please retain Spark’s reader for affected scans when native pruning cannot preserve which rows reach the throwing filter, or provide equivalent dictionary pruning.

[sunchao]

[P2] Keep sibling row filtering for dates that cannot throw

With ANSI off, make_date(2024,m,d) over INT columns cannot overflow, and an invalid month yields NULL. The blanket guard still disables row filtering for WHERE unix_date(make_date(2024,m,d)) > 0 AND y=6000000.

I checked one 128-row group with (y,m,d)=(2024,1,1) on its first 127 rows and (6000000,13,1) on its last. With Comet row filtering enabled, both revisions return count 0. The base filters all 128 rows inside the scan. This head filters zero there and emits all 128 to the parent. Please reuse the bounded-year and safe-argument check when deciding whether sibling row filtering would change evaluation.

[sunchao]

[P3] Share the structured-error assertion across the reader tests

Five new blocks repeat the same checks for the exception chain, class, error condition, and SQLSTATE. CometTestBase.checkSparkError already implements these comparisons. Its full-native-plan assertion is the part these mixed Spark-filter/native-scan tests cannot use. Could you extract the comparison from that helper and call it after each test’s native-scan assertion? That would keep the Spark-version-specific error checks in one place.