sunchao commented on code in PR #7343:
URL: https://github.com/apache/datafusion-comet/pull/7343#discussion_r4434286070
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala:
##########
@@ -327,6 +333,62 @@ object CometNativeScan extends
CometOperatorSerde[CometScanExec] with CometTypeS
}
}
+ /**
+ * Apply Spark's Parquet pushdown settings before serializing scan
predicates. Unsupported
+ * conjuncts may be dropped, including inside OR, but never below NOT:
weakening a negated
+ * predicate could prune rows Spark would read and hide an error in the
enclosing filter.
+ *
+ * Public because execution-time scalar-subquery filters are serialized from
Spark's package.
+ */
+ def filterForParquetPushdown(filter: Expression, conf: SQLConf):
Option[Expression] = {
+ // Exclude the whole predicate before rewriting: removing make_date could
leave a metadata
+ // filter that skips its required overflow. Independent predicates remain
eligible.
+ if (filter.exists(_.isInstanceOf[MakeDate])) {
+ return None
+ }
+
+ val pushDownDate = conf.parquetFilterPushDownDate
+ val pushDownTimestamp = conf.parquetFilterPushDownTimestamp
+ val pushDownDecimal = conf.parquetFilterPushDownDecimal
+ val pushDownStringPredicate = conf.parquetFilterPushDownStringPredicate
+ val inFilterThreshold = conf.parquetFilterPushDownInFilterThreshold
+
+ def usesDisabledSetting(expression: Expression): Boolean =
expression.exists {
+ case _: StartsWith | _: EndsWith | _: Contains =>
!pushDownStringPredicate
+ case e =>
+ e.dataType match {
+ case DateType => !pushDownDate
+ case TimestampType | TimestampNTZType => !pushDownTimestamp
+ case _: DecimalType => !pushDownDecimal
+ case _ => false
+ }
+ }
+
+ def select(expression: Expression, canPartialPushDown: Boolean):
Option[Expression] =
+ expression match {
+ case And(left, right) =>
+ (select(left, canPartialPushDown), select(right,
canPartialPushDown)) match {
+ case (Some(l), Some(r)) => Some(And(l, r))
+ case (l, r) if canPartialPushDown => l.orElse(r)
Review Comment:
**[P2] Check whether Spark can translate the whole sibling predicate before
simplifying it**
Use one Parquet row group with `a=b=0`, `d=DATE '2020-01-01'`, and one
`y=6000000` among otherwise valid years. With
`spark.sql.parquet.filterPushdown.date=false` and Comet row filtering off, this
returns `0` on this head. Spark and the base production code both raise
`ARITHMETIC_OVERFLOW`.
```sql
SELECT count(*) FROM t
WHERE unix_date(make_date(y,1,1)) > 0
AND ((months_between(d, DATE '2020-01-01') > 0 AND a=1) OR b=1)
```
The date-producing conjunct is already excluded. Spark’s V1 translation
rejects the separate `OR` because it cannot translate `months_between`. Here
`l.orElse(r)` strips that expression and sends `a=1 OR b=1` to Parquet, which
discards the group. Restoring only the date pushdown setting makes this head
raise again. Please gate the sibling on Spark’s V1 translation before applying
Parquet’s partial rewrite. A simpler example with default Spark settings is `s
LIKE 'zzz_%'` on an all-`'aaa'` group. Spark leaves this complex pattern above
the scan and raises, while Comet prunes the group. That case also reproduces on
the base.
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala:
##########
@@ -327,6 +333,62 @@ object CometNativeScan extends
CometOperatorSerde[CometScanExec] with CometTypeS
}
}
+ /**
+ * Apply Spark's Parquet pushdown settings before serializing scan
predicates. Unsupported
+ * conjuncts may be dropped, including inside OR, but never below NOT:
weakening a negated
+ * predicate could prune rows Spark would read and hide an error in the
enclosing filter.
+ *
+ * Public because execution-time scalar-subquery filters are serialized from
Spark's package.
+ */
+ def filterForParquetPushdown(filter: Expression, conf: SQLConf):
Option[Expression] = {
+ // Exclude the whole predicate before rewriting: removing make_date could
leave a metadata
+ // filter that skips its required overflow. Independent predicates remain
eligible.
+ if (filter.exists(_.isInstanceOf[MakeDate])) {
+ return None
+ }
+
+ val pushDownDate = conf.parquetFilterPushDownDate
+ val pushDownTimestamp = conf.parquetFilterPushDownTimestamp
+ val pushDownDecimal = conf.parquetFilterPushDownDecimal
+ val pushDownStringPredicate = conf.parquetFilterPushDownStringPredicate
+ val inFilterThreshold = conf.parquetFilterPushDownInFilterThreshold
+
+ def usesDisabledSetting(expression: Expression): Boolean =
expression.exists {
+ case _: StartsWith | _: EndsWith | _: Contains =>
!pushDownStringPredicate
+ case e =>
+ e.dataType match {
+ case DateType => !pushDownDate
+ case TimestampType | TimestampNTZType => !pushDownTimestamp
Review Comment:
**[P2] Account for the Parquet timestamp encoding when retaining metadata
filters**
On Spark 4.1.3, write one row group with `nt=TIMESTAMP_NTZ '2020-01-01'`
throughout and one overflowing year. For `WHERE unix_date(make_date(y,1,1)) > 0
AND nt < TIMESTAMP_NTZ '2000-01-01'`, Spark raises `ARITHMETIC_OVERFLOW`. With
Comet row filtering off and default Spark pushdown settings, the native scan
returns count `0`. Disabling only `spark.sql.parquet.filterPushdown.timestamp`
restores the error.
Spark’s Parquet comparison accepts UTC-adjusted timestamp encodings. An NTZ
column has `isAdjustedToUTC=false`, so enabling timestamp pushdown does not
make this predicate eligible. This mismatch also exists on the base. Please
account for Spark’s physical-type eligibility before retaining the sibling for
a throwing date filter.
##########
spark/src/main/scala/org/apache/comet/rules/MakeDateEvaluation.scala:
##########
@@ -240,8 +244,17 @@ private[rules] object MakeDateEvaluation {
case broadcast: CometBroadcastExchangeExec => broadcast.originalPlan
case _ => node
}
- val evaluatesDate = containsMakeDate(original.expressions)
+ val evaluatedExpressions = original match {
+ // Residual data filters run in the enclosing FilterExec, not in the
reader.
+ case scan: FileSourceScanExec => scan.partitionFilters
Review Comment:
**[P2] Preserve Spark’s dictionary pruning when a residual date filter can
throw**
In one dictionary-encoded Parquet row group, alternate `k` between `1` and
`3`, and put `y=6000000` on the final row. For `WHERE
unix_date(make_date(y,1,1)) > 0 AND k=2`, Spark returns count `0`. The
dictionary rules out `2` even though it falls within the min/max range. With
the native scan, this head raises `ARITHMETIC_OVERFLOW` with Comet row
filtering both on and off. The base returned `0` with row filtering on.
Disabling only `spark.hadoop.parquet.filter.dictionary.enabled` makes Spark
raise too. As a control, `k=5` returns `0` in both readers through min/max
pruning. Please retain Spark’s reader for affected scans when native pruning
cannot preserve which rows reach the throwing filter, or provide equivalent
dictionary pruning.
##########
spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala:
##########
@@ -1568,6 +1571,310 @@ class CometNativeReaderSuite extends CometTestBase with
AdaptiveSparkPlanHelper
}
}
+ test("make_date predicates preserve evaluation order and errors with
row-level pushdown") {
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> "false",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.key -> "true") {
+ withTempPath { path =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ spark
+ .range(1024)
+ .coalesce(1)
+ .selectExpr(
+ "CAST(IF(id = 1023, 6000000, 2024) AS INT) AS y",
+ "IF(id = 1023, '', sha2(CAST(id AS STRING), 512)) AS mask",
+ "IF(id = 1023, 0, 1) AS keep")
+ .write
+ .parquet(path.toString)
+ }
+ withTempView("make_date_filters") {
+
spark.read.parquet(path.toString).createOrReplaceTempView("make_date_filters")
+ val datePredicate = "unix_date(make_date(y, 1, 1)) > 0"
+ val masked = sql(
+ s"SELECT count(*) FROM make_date_filters WHERE length(mask) > 0
AND $datePredicate")
+ assert(
+ find(masked.queryExecution.executedPlan)(
+ _.isInstanceOf[CometNativeScanExec]).isDefined)
+ checkSparkAnswer(masked)
+ assert(masked.collect().toSeq == Seq(Row(1023L)))
+
+ // A later mask must not hide an overflow that Spark evaluates
first. The unmasked
+ // predicate must also preserve Spark's structured error outside
Parquet's RowFilter.
+ Seq(
+ datePredicate,
+ s"$datePredicate AND length(mask) > 0",
+ // OR keeps the date predicate in Spark; its serializable sibling
must not
+ // become a row filter that hides an error before Spark evaluates
the OR.
+ s"(y = 0 OR $datePredicate) AND keep = 1").foreach { predicate =>
+ val df = sql(s"SELECT count(*) FROM make_date_filters WHERE
$predicate")
+ assert(
+
find(df.queryExecution.executedPlan)(_.isInstanceOf[CometNativeScanExec]).isDefined)
+ val (sparkError, cometError) = checkSparkAnswerMaybeThrows(df)
+ val failures = Seq(sparkError, cometError).map { error =>
+ val failure = error.getOrElse(fail(s"Expected overflow for
$predicate"))
+ val chain = causeChain(failure)
+ assert(!chain.exists(_.isInstanceOf[CometNativeException]),
failure.toString)
+ chain.collect { case e: SparkThrowable with Throwable => e
}.lastOption.getOrElse {
+ fail(s"Expected a SparkThrowable for $predicate: $failure")
+ }
+ }
+ val expected = failures.head
+ val actual = failures(1)
+ assert(expected.getErrorClass == "ARITHMETIC_OVERFLOW")
+ assert(actual.getClass == expected.getClass)
+ assert(actual.getErrorClass == expected.getErrorClass)
+ assert(actual.getSqlState == expected.getSqlState)
+ }
+ }
+ }
+ }
+ }
+
+ Seq(
+ "partition column" -> "unix_date(make_date(y, m, 1)) > 0",
+ "nondeterministic bound" -> "unix_date(make_date(y, 1, 1)) >
rand(42)").foreach {
+ case (shape, datePredicate) =>
+ test(s"make_date predicates outside scan data filters preserve errors:
$shape") {
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> "false",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.key -> "true") {
+ withTempPath { path =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ spark
+ .range(0L, 1024L, 1L, 1)
+ .selectExpr(
+ "CAST(IF(id = 1023, 6000000, 2024) AS INT) AS y",
+ "IF(id = 1023, '', sha2(CAST(id AS STRING), 512)) AS mask",
+ "1 AS m")
+ .write
+ .partitionBy("m")
+ .parquet(path.toString)
+ }
+ withTempView("make_date_residual_filters") {
+ spark.read
+ .parquet(path.toString)
+ .createOrReplaceTempView("make_date_residual_filters")
+ val df = sql(
+ "SELECT count(*) FROM make_date_residual_filters " +
+ s"WHERE $datePredicate AND length(mask) > 0")
+ val scan =
find(df.queryExecution.executedPlan)(_.isInstanceOf[CometNativeScanExec])
+ .getOrElse(fail("Expected a native Parquet scan"))
+ .asInstanceOf[CometNativeScanExec]
+ // Spark omitted the date predicate before Comet saw the scan's
data filters.
+
assert(!scan.dataFilters.exists(_.exists(_.isInstanceOf[MakeDate])))
+ val (sparkError, cometError) = checkSparkAnswerMaybeThrows(df)
+ val failures = Seq(sparkError, cometError).map { error =>
+ val failure = error.getOrElse(fail(s"Expected overflow for
$shape"))
+ val chain = causeChain(failure)
+ assert(!chain.exists(_.isInstanceOf[CometNativeException]),
failure.toString)
+ chain
+ .collect { case e: SparkThrowable with Throwable => e }
+ .lastOption
+ .getOrElse {
+ fail(s"Expected a SparkThrowable: $failure")
+ }
+ }
+ assert(failures.head.getErrorClass == "ARITHMETIC_OVERFLOW")
+ assert(failures(1).getClass == failures.head.getClass)
+ assert(failures(1).getErrorClass == failures.head.getErrorClass)
+ assert(failures(1).getSqlState == failures.head.getSqlState)
+ }
+ }
+ }
+ }
+ }
+
+ test("make_date sibling metadata pruning respects Parquet type settings") {
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> "false",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ withTempPath { path =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ spark
+ .range(0L, 1024L, 1L, 1)
+ .selectExpr(
+ "CAST(IF(id = 1023, 6000000, 2024) AS INT) AS y",
+ "DATE '2020-01-01' AS d",
+ "CAST(1 AS DECIMAL(10, 2)) AS amount")
+ .write
+ .parquet(path.toString)
+ }
+ withTempView("make_date_metadata") {
+
spark.read.parquet(path.toString).createOrReplaceTempView("make_date_metadata")
+ val literalPredicates = Seq(
+ SQLConf.PARQUET_FILTER_PUSHDOWN_DATE_ENABLED.key -> "d < DATE
'1900-01-01'",
+ SQLConf.PARQUET_FILTER_PUSHDOWN_DECIMAL_ENABLED.key ->
+ "amount < CAST(0 AS DECIMAL(10, 2))")
+ // Spark 4 also pushes scalar subqueries, which are serialized after
planning.
+ val predicates = literalPredicates ++ (if (isSpark40Plus) {
+ Seq(
+
SQLConf.PARQUET_FILTER_PUSHDOWN_DATE_ENABLED.key ->
+ "d < (SELECT DATE
'1900-01-01' + CAST(min(id) AS INT) FROM range(1))")
+ } else {
+ Seq.empty
+ })
+ for ((setting, predicate) <- predicates; rowPushdown <- Seq(false,
true)) {
+ withSQLConf(
+ setting -> "false",
+ CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.key ->
rowPushdown.toString) {
+ val query = s"""SELECT count(*) FROM make_date_metadata
+ |WHERE unix_date(make_date(y, 1, 1)) > 0 AND
$predicate""".stripMargin
+ val df = sql(query)
+ val scan =
find(df.queryExecution.executedPlan)(_.isInstanceOf[CometNativeScanExec])
+ .getOrElse(fail("Expected a native Parquet scan"))
+ .asInstanceOf[CometNativeScanExec]
+ if (predicate.contains("SELECT")) {
+
assert(scan.dataFilters.exists(_.exists(_.isInstanceOf[ScalarSubquery])))
+ }
+ val (sparkError, cometError) = checkSparkAnswerMaybeThrows(df)
+ val failures = Seq(sparkError, cometError).map { error =>
+ val failure = error.getOrElse(fail(s"Expected overflow with
$setting=false"))
+ val chain = causeChain(failure)
+ assert(!chain.exists(_.isInstanceOf[CometNativeException]),
failure.toString)
+ chain
+ .collect { case e: SparkThrowable with Throwable => e }
+ .lastOption
+ .getOrElse {
+ fail(s"Expected a SparkThrowable: $failure")
+ }
+ }
+ assert(failures.head.getErrorClass == "ARITHMETIC_OVERFLOW")
+ assert(failures(1).getClass == failures.head.getClass)
+ assert(failures(1).getErrorClass == failures.head.getErrorClass)
+ assert(failures(1).getSqlState == failures.head.getSqlState)
+
+ // When Spark does permit the sibling predicate, both readers
may skip this
+ // entire row group, including its overflowing year.
+ withSQLConf(setting -> "true") {
+ checkSparkAnswer(sql(query))
+ assert(sql(query).collect().toSeq == Seq(Row(0L)))
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("make_date disjunctions are not weakened before Parquet metadata
pruning") {
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> "false",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.PARQUET_FILTER_PUSHDOWN_DATE_ENABLED.key -> "false") {
+ withTempPath { path =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ spark
+ .range(0L, 1024L, 1L, 1)
+ .selectExpr(
+ "CAST(IF(id = 1023, 6000000, 2024) AS INT) AS y",
+ "0 AS a",
+ "0 AS b",
+ "0 AS keep")
+ .write
+ .parquet(path.toString)
+ }
+ withTempView("make_date_disjunction") {
+
spark.read.parquet(path.toString).createOrReplaceTempView("make_date_disjunction")
+ val query = """SELECT count(*) FROM make_date_disjunction
+ |WHERE ((unix_date(make_date(y, 1, 1)) > 0 AND a = 1) OR b =
1)""".stripMargin
+ for (rowPushdown <- Seq(false, true)) {
+ withSQLConf(
+ CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.key ->
rowPushdown.toString) {
+ val df = sql(query)
+ assert(
+ find(df.queryExecution.executedPlan)(
+ _.isInstanceOf[CometNativeScanExec]).isDefined)
+ // Removing make_date would leave a = 1 OR b = 1, whose
statistics discard the
+ // entire row group before Spark can observe its overflowing
year.
+ val (sparkError, cometError) = checkSparkAnswerMaybeThrows(df)
Review Comment:
**[P3] Share the structured-error assertion across the reader tests**
Five new blocks repeat the same checks for the exception chain, class, error
condition, and SQLSTATE. `CometTestBase.checkSparkError` already implements
these comparisons. Its full-native-plan assertion is the part these mixed
Spark-filter/native-scan tests cannot use. Could you extract the comparison
from that helper and call it after each test’s native-scan assertion? That
would keep the Spark-version-specific error checks in one place.
##########
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.
+ session_config
+ .options_mut()
+ .execution
+ .parquet
+ .pushdown_filters = false;
Review Comment:
**[P2] Use the effective scan setting when planning floating-point
comparisons**
With one Parquet row group where `f=42D` throughout and one `y=6000000`,
count rows satisfying `unix_date(make_date(y,1,1)) > 0 AND f < 0D`. Spark and
the base return `0` with Comet row filtering enabled. This head raises
`ARITHMETIC_OVERFLOW`. Switching only
`spark.comet.parquet.rowFilterPushdown.enabled` to `false` restores `0`.
The planner has already normalized the floating-point operand using the
session-wide setting by the time this line disables row filtering. Parquet can
no longer use the plain column comparison for statistics pruning. Please choose
the operand representation after accounting for the per-scan guard.
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala:
##########
@@ -308,6 +309,11 @@ 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.wrapped.getTagValue(CometScanRule.DISABLE_ROW_FILTER_PUSHDOWN_TAG).isDefined
||
+ scan.dataFilters.exists(_.exists(_.isInstanceOf[MakeDate])))
Review Comment:
**[P2] Keep sibling row filtering for dates that cannot throw**
With ANSI off, `make_date(2024,m,d)` over `INT` columns cannot overflow, and
an invalid month yields NULL. The blanket guard still disables row filtering
for `WHERE unix_date(make_date(2024,m,d)) > 0 AND y=6000000`.
I checked one 128-row group with `(y,m,d)=(2024,1,1)` on its first 127 rows
and `(6000000,13,1)` on its last. With Comet row filtering enabled, both
revisions return count `0`. The base filters all 128 rows inside the scan. This
head filters zero there and emits all 128 to the parent. Please reuse the
bounded-year and safe-argument check when deciding whether sibling row
filtering would change evaluation.
--
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]
