fix: preserve make_date evaluation across lazy operators

docs/source/user-guide/latest/expressions.md

@@ -281,7 +281,7 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci
 | `hour` | ✅ | Native |  |
 | `last_day` | ✅ | Native |  |
 | `localtimestamp` | ✅ | — |  |
-| `make_date` | ✅ | Native | Computed arguments preserve year → month → day NULL checks. Expressions fall back when shared subexpressions or a parent could change which arguments Spark evaluates. IN probes stay native for literal/column candidates and array/struct constructors of those values; other computed candidates require Spark evaluation. |
+| `make_date` | ✅ | Native | Computed arguments preserve year → month → day NULL checks. Expressions fall back when shared subexpressions or a parent could change which arguments Spark evaluates. IN probes stay native for literal/column candidates and array/struct constructors of those values; other computed candidates require Spark evaluation. Operator-level safeguards and their opt-out are described in [make_date evaluation](#make_date-evaluation). |
 | `make_dt_interval` | ✅ | Codegen dispatch |  |
 | `make_interval` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default; intervals outside Arrow's nanosecond range are tracked by [#5279](https://github.com/apache/datafusion-comet/issues/5279); the native path is opt-in via allowIncompatible ([details](compatibility/expressions/datetime.md)) |
 | `make_time` | ✅ | — | Spark 4.1+; requires `spark.sql.timeType.enabled=true`, which Spark leaves off by default. Runs natively; remaining TIME type work is tracked by [#4288](https://github.com/apache/datafusion-comet/issues/4288) |
@@ -713,6 +713,45 @@ Comet also accelerates a number of Catalyst expressions that have no Spark SQL f
 
 This list is illustrative, not exhaustive: the per-function tables are not the complete set of expressions Comet can accelerate.
 
+## make_date evaluation
+
+With ANSI mode enabled, `make_date(2024, 13, 1)` raises an invalid-date error. If an
+input iterator yields a valid row followed by that invalid row, a `LIMIT 1` can stop
+before evaluating the second date. A native projection can evaluate a whole batch
+before the consuming operator decides which rows it needs.
+
+By default, Comet keeps affected operators in Spark to preserve skipped date evaluation.
+This applies to date projections below `LIMIT`, `SAMPLE`, `FILTER`, joins,
+`mapPartitions`, and `Dataset.rdd` consumers; date expressions inside joins, aggregates,
+and ordered limits; and projections below a window group limit. A parent projection can
+also defer a date output when an `IF`, `COALESCE`, or NULL-skipping operand does not need
+it. Comet preserves that evaluation across projections while allowing eager date consumers,
+such as `YEAR(dt)`, to remain native when their operands are safe to evaluate. An enclosing
+conditional can still defer the extractor's date input. These safeguards also preserve
+NULL short-circuiting when a date argument is computed in an earlier projection, including
+across the streamed side of a broadcast join. The materialized build side remains independent.
+These safeguards also cover required arguments, such as a computed year, on input rows that
+the join rejects before producing an output row.
+Selecting the date eagerly does not force conditionally used sibling expressions; those
+computations also retain Spark evaluation when needed.
+
+An unfiltered `COUNT(make_date(...))` can remain native when its sole argument is nullable,
+because Spark checks that argument on every input row. Filtered counts and counts with
+multiple arguments retain the safeguard when they can skip date evaluation.
+
+Only affected parts of the plan fall back. An ordinary native scan can remain native.
+Exchanges materialize their inputs and partition keys independently, so an outer limit does
+not force an eager date partition key to Spark. Local sorts and aggregates can also separate
+date evaluation from an outer limit.
+
+Users who accept eagerly evaluating dates, projected arguments, and sibling expressions can set
+`spark.comet.exec.makeDate.operatorFallback.enabled=false` to allow native operators for
+these shapes. The default is `true`. Disabling it can expose invalid-date or argument errors
+that Spark would skip under ANSI mode. This setting does not change
+`make_date`'s invalid-date handling or the expression-level safeguards for skipped
+arguments and operands. For example, an empty `IN` predicate containing `make_date`
+still falls back when needed.
+
 ## See also
 
 - [Comet Compatibility Guide](compatibility/index.md) - known incompatibilities and edge cases for supported expressions.

spark/src/main/scala/org/apache/comet/CometConf.scala

@@ -341,6 +341,17 @@ object CometConf extends ShimCometConf {
     .booleanConf
     .createWithDefault(true)
 
+  val COMET_EXEC_MAKE_DATE_OPERATOR_FALLBACK_ENABLED: ConfigEntry[Boolean] =
+    conf(s"$COMET_EXEC_CONFIG_PREFIX.makeDate.operatorFallback.enabled")
+      .category(CATEGORY_EXEC)
+      .doc(
+        "Keep affected make_date operators in Spark when native batching can evaluate rows " +
+          "that Spark skips. When false, allow native operators even if make_date can raise " +
+          "an error on a row Spark would not evaluate. This does not disable expression-level " +
+          "evaluation safeguards or change make_date's invalid-date handling.")
+      .booleanConf
+      .createWithDefault(true)
+
   val COMET_EXEC_PROJECT_ENABLED: ConfigEntry[Boolean] =
     createExecEnabledConfig("project", defaultValue = true)
   val COMET_EXEC_FILTER_ENABLED: ConfigEntry[Boolean] =

spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala

@@ -445,6 +445,9 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false)
     }
 
     def convertNode(op: SparkPlan): SparkPlan = op match {
+      case op if MakeDateEvaluation.reason(op).isDefined =>
+        withFallbackReason(op, MakeDateEvaluation.reason(op).get)
+
       // Scan marker produced by an optional, out-of-tree scan contrib (e.g. contrib/delta).
       // Matched by trait (no compile-time dependency on the contrib) and present only when that
       // contrib is on the classpath. The marker carries its own serde handler and typically wraps
@@ -1041,15 +1044,16 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false)
         plan
       }
     } else {
+      val planWithDateEvaluation = MakeDateEvaluation.preserve(plan)
       // The rewrite removes the sorts of each join it converts, so put back any sort a kept
       // operator above still needs. This runs before transform(), which converts the sorts it
       // adds like any other SortExec.
       val planWithJoinRewritten = if (CometConf.COMET_FORCE_SHJ.get()) {
-        RewriteJoin.restoreRequiredOrdering(plan.transformUp { case p =>
-          RewriteJoin.rewrite(p, conf)
+        RewriteJoin.restoreRequiredOrdering(planWithDateEvaluation.transformUp { case p =>
+          if (MakeDateEvaluation.reason(p).isDefined) p else RewriteJoin.rewrite(p, conf)
         })
       } else {
-        plan
+        planWithDateEvaluation
       }
 
       // Tag Partial aggregates that must not be converted to Comet because a