[pingzh]

Which issue does this PR close?

Related to #6741. Part 3 of the split of #7117, following merged #7260 and #7261. The native Date32 overflow correction remains a separate follow-up.

Rationale for this change

With ANSI mode enabled, an input iterator can yield a valid date row followed by a row with month 13. Spark's LIMIT 1 stops before evaluating the invalid date; a native projection can evaluate both rows in a batch and throw. Conditional projections and other lazy consumers can expose the same difference.

What changes are included in this PR?

• Preserve Spark evaluation around affected make_date projections and consumers, including limits, filters, samples, joins, aggregates, ordered limits, window group limits, mapPartitions, and Dataset.rdd. Native scans and independent materialization boundaries remain available.
• Track deferred date outputs, argument producers, and sibling computations across projections while retaining native execution for provably eager consumers. Preserve logical-plan tags and restore existing native broadcasts when a consuming join falls back.
• Add supported spark.comet.exec.makeDate.operatorFallback.enabled (default true) and document its tradeoff. Setting it to false allows these native operators to evaluate dates on rows Spark would skip; expression-level safeguards and invalid-date handling remain in effect.
• Add focused regressions and two reproducible performance cases using CometDatetimeExpressionBenchmark --make-date-only.

Start in MakeDateEvaluation.scala: the main review questions are where row demand stops at materialization boundaries, which projection outputs Spark may defer, and whether falling back preserves broadcast reuse and logical-plan links. CometExecRule.scala applies that decision before join rewrites and native conversion.

How are these changes tested?

• Spark 4.1.3 / JDK 17: all 120 CometExecRuleSuite tests and both make_date tests in CometTemporalExpressionSuite pass, including skipped invalid-month errors, deferred sibling casts, AQE/reused broadcasts, native eager controls, and the opt-out.
• Local runtime validation uses the unchanged native library from upstream commit 967d09b52694d64c423012ed7dcc8f7ec6e20413 (CI run). This PR changes no native source. Its build is Cargo ci (release-derived, LTO off, debug assertions on), targeting x86-64-v3.
• Before merging, request run-spark-4.1-tests and run-all-spark-profiles for Spark SQL and version coverage.

Benchmark results below are from commit f07f57bc5, before the projection follow-ups; they have not been rerun for these follow-ups. Two fresh JVM runs on valid Parquet data (1,048,576 probe rows, 1,024 build rows), local[1], AQE off, one shuffle partition, whole-stage codegen on, ANSI off, 8 GiB heap. The Spark benchmark harness uses at least five iterations per case. Values are mean milliseconds (run 1 / run 2):

Query	Spark	Comet default	Operator fallback disabled
LIMIT 10	43 / 46	36 / 34	24 / 26
Broadcast join with probe-side date projection	123 / 127	93 / 106	93 / 92

These measure end-to-end planning and execution of the different physical plans. Default Comet was faster than Spark on this setup. Relative to disabling the guard, LIMIT means increased by 8–12 ms; join means increased by 0–14 ms, with default standard deviations of 17–18 ms. This does not establish a precise join penalty or release-build performance.

Captured plans confirm Spark date projection/limit/join with native scans by default, and native projection/limit/join with the opt-out. The join uses a noop writer to consume the date output; its probe-side projection is asserted before timing.

Reproduce with SPARK_GENERATE_BENCHMARK_FILES=1 make BENCH_HEAP=8g benchmark-org.apache.spark.sql.benchmark.CometDatetimeExpressionBenchmark -- --make-date-only, in two fresh JVMs. The measurements used the matching upstream native artifact built with Cargo’s ci profile.

[ajsquared]

1 review finding.

[sunchao]

Reviewed f07f57bc5089400c34ac6661f838c46e31fd1e36 with five independent passes covering correctness, Spark compatibility, performance, design/AQE, and tests.

I confirmed the existing P2 deferred-producer finding and added two exact-head runtime reproductions in that thread. Spark skips the invalid cast and returns rows, while Comet keeps the producer native and throws CAST_INVALID_INPUT. The second reproduction selects the date eagerly while leaving a sibling cast deferred. I found no additional independent issue, but am withholding approval while this gap remains.

Local validation used Spark 4.1.3, JDK 17, and the native artifact from this head's CI run 40613167896. The full reactor test compilation succeeded, and all 10 make_date tests selected from CometExecRuleSuite and CometTemporalExpressionSuite passed. The two additional SQL probes reproduced the gap despite those passing tests.

The materialization boundaries, repeated-plan handling, and broadcast restoration received source review. Compared with the make_date path on branch-1.2 (9bcce3ac), restricting native execution for affected operators is an intended, documented change with a supported opt-out. I did not rerun the benchmarks or the cross-version suites. At publication, 22 checks had passed, the scan/expression/exec CI jobs were still running, and the expanded Spark SQL and other-profile runtime coverage remained skipped.

[sunchao]

Reviewed the current changes around deferred projections, lazy consumers, materialization boundaries, and AQE/broadcast fallback. No additional P1/P2 issues found; the prior deferred-producer thread is resolved.

The current-head CI workflow passed. I did not run local Spark/Comet runtime tests or benchmarks, and the expanded Spark SQL suites were skipped in CI, so cross-version runtime coverage remains unverified.

[sunchao]

Reviewed 0db23962399be74e629409a9993388133ad17a22 with five independent scopes covering correctness, resources, design/reuse, tests, and release/documentation compatibility.

The two earlier direct-projection reproductions are fixed. One remaining correctness gap needs addressing: a broadcast join can forward deferred streamed-side expressions, but the new rule discards their demand at every join. The inline comment includes a current-head reproduction and an isolated control. I am requesting changes for that gap.

Validation: compiled the current JVM sources and passed all 120 CometExecRuleSuite tests plus both make_date temporal tests on Spark 4.1.3 / JDK 17. Suite registration passed. The additional Spark/Comet probe fails only in the joined-argument case. Removing the join reset in a separate diagnostic classpath makes that case pass. The native library came from the earlier reviewed commit's CI artifact; its native source tree is identical to this head. No PR source was changed.

Current-head CI reports 26 successful checks and 18 skips. Spark SQL 4.1 and the other Spark runtime profiles have not run, so please obtain the requested run-spark-4.1-tests / run-all-spark-profiles verdicts after the fix. Compared with branch-1.2, the guarded fallback is an intended, documented change. The join finding is an incomplete correction, not a demonstrated new regression from that release. Historical benchmark numbers remain correctly labeled as predating the projection follow-ups; I did not rerun them.

[ajsquared]

1 review finding.

[ajsquared]

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

[sunchao]

Three previously reported performance cases from the split parent remain present in this PR's source. The linked measurements were made on the parent PR; I have not rerun them on this head.

[ajsquared]

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

[sunchao]

[P2] Make the appended tuple explicit for Scala 2.12

The exact-head Spark 3.5 strict-warnings CI job fails at this line: Scala 2.12 adapts the three arguments to a tuple and emits the argument-adaptation warning, which the job treats as an error. This prevents the Spark 3.5 test sources from compiling under the required strict configuration. Wrap the tuple in another pair of parentheses (:+ ((2024, if (invalid) 13 else 1, 1))) so it is passed as a single argument without adaptation.

[ajsquared]

1 review finding.

[ajsquared]

[P2] With ANSI enabled, COUNT(date_add(make_date(y, 1, 1), CAST(s AS INT))) can throw CAST_INVALID_INPUT where Spark returns 1. On Parquet rows (NULL, 'bad') and (2024, '1'), Spark skips the days operand on the first row because the date is NULL. The COUNT exemption accepts the entire nullable argument, and CometDateAdd serializes both operands without null short-circuiting. The same expression is eligible as a native hash partition key below a limit. Guard DateAdd when a nullable left date can hide a throwing right operand, while preserving native eligibility for bare make_date.

[ajsquared]

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

[sunchao]

Reviewed the exact head against 967d09b, including the previous projection, join, shuffle, COUNT, YEAR, and Scala 2.12 findings. No additional introduced or materially worsened P1/P2 was established. The strict Scala 3.5 and Spark 4.1 exec jobs passed on this head.

The open DateAdd/COUNT concern appears present on the specified base as well, based on source tracing; I did not execute a base-versus-head reproduction. Expanded Spark SQL and cross-profile runtime jobs were skipped, and I did not rerun performance benchmarks.