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