comphead commented on code in PR #7343:
URL: https://github.com/apache/datafusion-comet/pull/7343#discussion_r4485785207
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -127,6 +130,35 @@ case class CometScanRule(session: SparkSession)
}
}
+ /**
+ * Spark omits partition-column and nondeterministic predicates from a V1
scan's data filters.
+ * Preserve the enclosing filter's make_date guard before core or contrib
scans are converted.
+ */
+ private def tagScansWithMakeDateFilters(plan: SparkPlan): Unit = {
+ def tagScan(scan: FileSourceScanExec, hasMakeDateFilter: Boolean): Unit = {
+ // Parent context can change when Spark replans: do not retain a
previous walk's guard.
+ scan.unsetTagValue(CometScanRule.DISABLE_ROW_FILTER_PUSHDOWN_TAG)
+ if (hasMakeDateFilter) {
+ scan.setTagValue(CometScanRule.DISABLE_ROW_FILTER_PUSHDOWN_TAG, ())
+ }
+ }
+
+ def visit(node: SparkPlan, hasMakeDateFilter: Boolean): Unit = node match {
+ case scan: FileSourceScanExec => tagScan(scan, hasMakeDateFilter)
+ case scan: CometScanExec => tagScan(scan.wrapped, hasMakeDateFilter)
+ case filter: FilterExec =>
+ visit(
+ filter.child,
+ hasMakeDateFilter ||
filter.condition.exists(_.isInstanceOf[MakeDate]))
+ case project: ProjectExec => visit(project.child, hasMakeDateFilter)
+ // Only the same filter/project pipeline shares predicate evaluation
order. Joins and
+ // materializing operators must not propagate this guard into unrelated
scans.
+ case _ => node.children.foreach(visit(_, hasMakeDateFilter = false))
+ }
+
+ visit(plan, hasMakeDateFilter = false)
+ }
Review Comment:
This tag only feeds `disable_row_filter_pushdown`, which matters only with
`spark.comet.parquet.rowFilterPushdown.enabled=true` (default `false`). The
open findings on this head (dictionary pruning, NTZ, float operands, V1
translation) all come from matching Spark's per-file pruning, which Spark
builds from the footer schema. Would it be simpler to keep Spark's reader for a
scan under a filter whose `make_date` can throw, reusing the bounded-year check
so safe dates stay native? Then the proto field, the `init_datasource_exec`
argument, the native checks and the Delta setters would not be needed. I expect
that shape to be rare, but I have not measured it.
##########
spark/src/main/scala/org/apache/comet/rules/MakeDateEvaluation.scala:
##########
@@ -464,14 +479,19 @@ private[rules] object MakeDateEvaluation {
(rebuilt, fallbackReason, dateDerivedOutputs)
}
- // Dataset.rdd adds an object-producing root, and its caller may read only
a few objects.
- // EliminateSerialization can leave a typed operator, filter, or project
at that root.
- def producesObjects(op: SparkPlan): Boolean = op match {
- case _: ObjectProducerExec => true
- case _: FilterExec | _: ProjectExec => producesObjects(op.children.head)
+ // QueryExecution.toRdd (also used by PySpark's DataFrame.rdd) exposes
this plan's row
+ // iterator without adding Dataset.rdd's object-producing root. Its caller
can stop early,
+ // even if explain() or another action has already prepared the physical
plan.
+ // A one-row collect limit can retain its native input cap: producing its
only row cannot
+ // skip another row inside that cap. Larger limits and OFFSET do not
provide that guarantee.
+ def isSingleRowCollect(op: SparkPlan): Boolean = op match {
+ case limit: CollectLimitExec => limit.limit == 1 && limit.offset == 0
+ case native: CometExec => isSingleRowCollect(native.originalPlan)
+ case _: ColumnarToRowExec | _: CometColumnarToRowExec | _:
CometNativeColumnarToRowExec =>
+ isSingleRowCollect(op.children.head)
case _ => false
}
- val rootDemand = if (producesObjects(plan)) Some("code reading
Dataset.rdd") else None
+ val rootDemand = if (isSingleRowCollect(plan)) None else
Some(externalRowConsumer)
Review Comment:
With this default every plan except a `LIMIT 1` collect starts with
external-row demand, so a root projection with a `make_date` that can throw now
runs in Spark in both ANSI modes. On `main` that applied only to the
`Dataset.rdd` root. The fixtures show the reach: about 50 queries gained
`REPARTITION(1)` to stay native, so plain `SELECT make_date(y, m, d) FROM t` is
no longer tested as a native shape. It also drops #7283's native cap for `LIMIT
n` with `n > 1`, including `df.show()` (`CollectLimitExec(21)`). Note that
Spark's `CollectLimitExec.doExecute` evaluates `take(limit)` in every partition
when the child has more than one, so only a single-partition child streams.
Since the `DataFrame.rdd` gap already exists on `main` with ANSI on, could
this routing be a separate follow-up to #7283 with a benchmark of the default
path, plus one `expect_fallback` fixture for a plain root projection?
##########
native/core/src/parquet/parquet_exec.rs:
##########
@@ -351,6 +370,11 @@ fn calls_jvm(expr: &Arc<dyn PhysicalExpr>) -> bool {
.unwrap_or(true)
}
+fn contains_make_date(expr: &Arc<dyn PhysicalExpr>) -> bool {
+ expr.exists(|e| Ok(e.downcast_ref::<MakeDateExpr>().is_some()))
+ .unwrap_or(true)
+}
Review Comment:
I could not find a path that sends a `MakeDate` predicate here.
`buildNativeScanCommon`, the scalar-subquery path in `CometNativeScanExec` and
both Delta paths run `filterForParquetPushdown` first, which returns `None` for
any predicate containing `MakeDate`. If so, these checks and the ones at lines
119 to 122 and 317 cannot match, and no native test exercises them. Could they
go, so the decision lives only on the JVM side?
##########
spark/src/main/scala/org/apache/comet/rules/MakeDateEvaluation.scala:
##########
@@ -464,14 +479,19 @@ private[rules] object MakeDateEvaluation {
(rebuilt, fallbackReason, dateDerivedOutputs)
}
- // Dataset.rdd adds an object-producing root, and its caller may read only
a few objects.
- // EliminateSerialization can leave a typed operator, filter, or project
at that root.
- def producesObjects(op: SparkPlan): Boolean = op match {
- case _: ObjectProducerExec => true
- case _: FilterExec | _: ProjectExec => producesObjects(op.children.head)
+ // QueryExecution.toRdd (also used by PySpark's DataFrame.rdd) exposes
this plan's row
+ // iterator without adding Dataset.rdd's object-producing root. Its caller
can stop early,
+ // even if explain() or another action has already prepared the physical
plan.
+ // A one-row collect limit can retain its native input cap: producing its
only row cannot
+ // skip another row inside that cap. Larger limits and OFFSET do not
provide that guarantee.
+ def isSingleRowCollect(op: SparkPlan): Boolean = op match {
+ case limit: CollectLimitExec => limit.limit == 1 && limit.offset == 0
+ case native: CometExec => isSingleRowCollect(native.originalPlan)
+ case _: ColumnarToRowExec | _: CometColumnarToRowExec | _:
CometNativeColumnarToRowExec =>
Review Comment:
#7505 (approved) removes `CometNativeColumnarToRowExec`, including the
existing match on it in this file. The two PRs merge without a textual
conflict, but the merged tree still references the class on this line, so
whichever lands second will not compile.
##########
spark/src/test/resources/sql-tests/expressions/datetime/make_date_ansi.sql:
##########
@@ -196,10 +199,49 @@ INSERT INTO make_date_shared VALUES (NULL, 0), (2024, 0)
-- Spark computes the shared day expression outside the constructor's NULL
checks.
query expect_fallback(make_date requires Spark evaluation when a computed
argument)
-SELECT make_date(y, 1, d + 1), d + 1 FROM make_date_shared
+SELECT /*+ REPARTITION(1) */ make_date(y, 1, d + 1), d + 1 FROM
make_date_shared
statement
SET spark.sql.subexpressionElimination.enabled=false
query expect_native(make_date)
-SELECT make_date(y, 1, d + 1), d + 1 FROM make_date_shared
+SELECT /*+ REPARTITION(1) */ make_date(y, 1, d + 1), d + 1 FROM
make_date_shared
+
+-- The Date32 endpoints remain valid, and NULL arguments skip overflow
validation.
+statement
+CREATE TABLE make_date_bounds(y INT, m INT, d INT) USING parquet
+
+statement
+INSERT INTO make_date_bounds VALUES
+ (5881580, 7, 11), (-5877641, 6, 23),
+ (NULL, 13, 32), (6000000, NULL, 1), (6000000, 1, NULL)
+
+query expect_native(make_date)
+SELECT /*+ REPARTITION(1) */ unix_date(make_date(y, m, d)) FROM
make_date_bounds
+
+-- Valid calendar dates just outside Date32 must overflow, regardless of ANSI
mode.
+statement
+CREATE TABLE make_date_overflow(y INT, m INT, d INT) USING parquet
+
+statement
+INSERT INTO make_date_overflow VALUES (5881580, 7, 12), (-5877641, 6, 22)
+
+query expect_error_class(ARITHMETIC_OVERFLOW)
+SELECT /*+ REPARTITION(1) */ make_date(y, m, d) FROM make_date_overflow WHERE
y > 0
+
+query expect_error_class(ARITHMETIC_OVERFLOW)
+SELECT /*+ REPARTITION(1) */ make_date(y, m, d) FROM make_date_overflow WHERE
y < 0
+
+-- An empty input must not evaluate overflowing scalar arguments.
+query expect_native(make_date)
+SELECT /*+ REPARTITION(1) */ make_date(6000000, 1, 1) FROM make_date_overflow
WHERE y = 0
Review Comment:
This block repeats the new block in `make_date.sql` and only the ANSI
setting differs. Could it be one file with `-- ConfigMatrix:
spark.sql.ansi.enabled=false,true`?
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala:
##########
@@ -366,7 +428,7 @@ object CometNativeScan extends
CometOperatorSerde[CometScanExec] with CometTypeS
if (conf.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED)) {
commonBuilder.setHasDataFilters(dataFilters.nonEmpty)
val filterProtos = new ListBuffer[Expr]()
- for (filter <- dataFilters) {
+ for (filter <- dataFilters.flatMap(filterForParquetPushdown(_, conf))) {
Review Comment:
`filterForParquetPushdown` now runs on the data filters of every native
scan, so with a non-default `spark.sql.parquet.filterPushdown.date`,
`.timestamp`, `.decimal` or `.string.predicate`, queries without `make_date`
lose native pruning they have today. Honoring those settings may be right, but
it is a scan change of its own. Could it move to a separate PR with a test that
has no `make_date`, leaving only the whole-predicate `make_date` exclusion here?
--
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]
