sunchao commented on code in PR #7343:
URL: https://github.com/apache/datafusion-comet/pull/7343#discussion_r4429687620
##########
spark/src/test/scala/org/apache/comet/rules/CometExecRuleSuite.scala:
##########
@@ -183,12 +183,27 @@ class CometExecRuleSuite extends CometTestBase {
ansi),
"date")()),
input)
- val native = applyCometExecRule(project)
- assert(countOperators(native, classOf[CometProjectExec]) == 1,
native.treeString)
+ val root = applyCometExecRule(project)
+ val requiresRows = ansi || !boundedYear
+ assert(
+ countOperators(root, classOf[CometProjectExec]) == (if
(requiresRows) 0 else 1),
Review Comment:
**[P2] Keep nonthrowing wrappers of a bounded date projection native**
With ANSI off and Parquet `INT` columns `m` and `d`, the direct
`make_date(2024,m,d)` projection stays native. Wrapping that same expression in
`unix_date(...)` or `... IS NULL` changes a `CometProject` on merge base
`0f90a7d` to a Spark `Project` above `CometColumnarToRow` on this head. I
collected both plans’ results, including invalid month 13, and they matched.
The year fits Date32 for every valid day, invalid fields yield NULL, and
neither wrapper can throw. Please recognize these wrappers when checking
whether eager projection is safe and cover a wrapped expression alongside the
bare one here.
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala:
##########
@@ -308,6 +308,10 @@ object CometNativeScan extends
CometOperatorSerde[CometScanExec] with CometTypeS
hadoopConf = hadoopConf,
conf = scan.conf) match {
case Some(commonBuilder) =>
+ // Original filters include unsupported expressions and unresolved
subqueries omitted
+ // from native predicates. A pushed sibling must not hide their
make_date errors.
+ commonBuilder.setDisableRowFilterPushdown(
+ scan.dataFilters.exists(_.exists(_.isInstanceOf[MakeDate])))
Review Comment:
**[P2] Detect date predicates that Spark leaves entirely above the file
scan**
`scan.dataFilters` already excludes a predicate that uses both a file column
and a partition column. I reproduced this on Spark 4.1.3 with ANSI off. Use the
1,024-row mask fixture from the new reader test, add `1 AS m`, and write it
with `partitionBy("m")`. Only the final row has `y=6000000` and an empty mask.
```sql
SELECT count(*) FROM t
WHERE unix_date(make_date(y,m,1)) > 0 AND length(mask) > 0
```
Spark and Comet with row-filter pushdown off raise `ARITHMETIC_OVERFLOW`.
Turning it on makes Comet return `1023`. The date is present in the Spark
filter above the scan, but absent from the scan’s data filters, so this flag
stays false and the reader discards the overflow row. A date conjunct
containing `rand(42)` exhibited the same omission. Please carry the guard from
the enclosing filter before Spark narrows its predicates to file columns.
##########
spark/src/main/scala/org/apache/comet/rules/MakeDateEvaluation.scala:
##########
@@ -393,7 +394,9 @@ private[rules] object MakeDateEvaluation {
case _ => false
}
val rowReason = if (evaluatesDate && !exchange && !safeProjection) {
- rowDemand.map(boundary => s"make_date requires Spark evaluation below
$boundary")
+ rowDemand
+ .filterNot(boundary => boundary == externalRowConsumer &&
localMaterialization)
Review Comment:
**[P2] Keep the native Parquet reader when the date is evaluated by the
filter above it**
`SELECT id FROM t WHERE unix_date(make_date(2024,m,d)) > 0` on Parquet `INT`
columns with ANSI off has a Spark `Filter` on both revisions. Its child changes
from `CometNativeScan` on `0f90a7d` to Spark `FileScan` on this head, with the
default row-filter settings. The scan lists the residual date in `dataFilters`,
so it receives the external-row fallback reason even though the date stays
above the reader. Please distinguish residual data filters from expressions the
scan actually evaluates, such as partition filters, when deciding whether the
reader needs to fall back.
##########
spark/src/main/scala/org/apache/comet/rules/MakeDateEvaluation.scala:
##########
@@ -464,14 +467,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:
**[P2] Stop external row demand at a file write**
On Spark 4.1.3, `EXPLAIN INSERT OVERWRITE TABLE out_dt SELECT
make_date(y,1,1) FROM t` with a Parquet `INT` year and ANSI off now moves the
projection to Spark. With `spark.comet.write.parquet.enabled=true` and
`spark.comet.operator.WriteFilesExec.allowIncompatible=true`, merge base
`0f90a7d` plans `CometWriteFiles → CometProject`. This head plans Spark
`WriteFiles → Project → CometColumnarToRow`. Disabling only the date safeguard
restores the native writer and projection.
A file write consumes all input rows before it returns its result, so the
caller cannot stop that input early. Please clear the external-iterator demand
at the write boundary while allowing limits, filters, and other consumers
inside the write query to establish their own safeguards.
##########
native/core/src/parquet/parquet_exec.rs:
##########
@@ -106,12 +109,27 @@ pub(crate) fn init_datasource_exec(
rebase_from_file_metadata: bool,
datetime_rebase_mode_in_read: &str,
int96_rebase_mode_in_read: &str,
+ disable_row_filter_pushdown: bool,
) -> Result<Arc<DataSourceExec>, ExecutionError> {
// Computed once and reused below for `try_pushdown_filters`.
`copied_config()` clones only
// `SessionConfig` (an `Arc<ConfigOptions>` plus a small extensions map);
`SessionContext::
// state()` would clone the whole `SessionState`, including the function
registries (~250+
// UDFs) and optimizer rules, on every scan init.
- let session_config = session_ctx.copied_config();
+ let mut session_config = session_ctx.copied_config();
+ if disable_row_filter_pushdown
+ || data_filters
+ .as_ref()
+ .is_some_and(|filters| filters.iter().any(contains_make_date))
+ {
+ // RowFilter can reorder predicates and flatten their Spark errors.
Even pushing only
+ // a later sibling could hide a make_date error that Spark evaluates
first. Keep all
+ // row filtering above this scan, while retaining metadata pruning for
other predicates.
Review Comment:
**[P2] Preserve Spark’s per-type settings when retaining sibling predicates
for metadata pruning**
I reproduced a suppressed overflow even with Comet row-filter pushdown
**off**. Use one Parquet row group with `d = DATE '2020-01-01'` throughout and
one row with `y=6000000`. Set ANSI off and
`spark.sql.parquet.filterPushdown.date=false`, leaving global Parquet pushdown
on.
```sql
SELECT count(*) FROM t
WHERE unix_date(make_date(y,1,1)) > 0 AND d < DATE '1900-01-01'
```
Spark reads the group and raises `ARITHMETIC_OVERFLOW`. This head returns
`0` with Comet row-filter pushdown either off or on. The retained date
comparison can still eliminate the whole group by statistics. I also reproduced
this with a decimal sibling and
`spark.sql.parquet.filterPushdown.decimal=false`. Please restrict the retained
metadata predicates to those Spark would push for this scan, or skip sibling
pruning when that equivalence cannot be established.
--
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]
