pingzh opened a new pull request, #7283: URL: https://github.com/apache/datafusion-comet/pull/7283
## 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 outputs 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](https://github.com/apache/datafusion-comet/blob/f07f57bc5089400c34ac6661f838c46e31fd1e36/spark/src/main/scala/org/apache/comet/rules/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](https://github.com/apache/datafusion-comet/blob/f07f57bc5089400c34ac6661f838c46e31fd1e36/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala) applies that decision before join rewrites and native conversion. ## How are these changes tested? - Spark 4.1.3 / JDK 17: all 119 `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](https://github.com/apache/datafusion-comet/actions/runs/40607694862)). 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. 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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
