Repository navigation
Conversation
ajsquared
left a comment
There was a problem hiding this comment.
Reviewed the current changes and feedback; no independently confirmed unresolved P1 remains.
sunchao
left a comment
There was a problem hiding this comment.
Reviewed the current head, including calendar validation, Date32 boundaries and the error path. A Java comparison covered 105,304 calendar cases and focused Rust CI passed; JVM CI was still running, Spark SQL suites were skipped, and local end-to-end Comet testing was unavailable.
sunchao
left a comment
There was a problem hiding this comment.
I reproduced two cases on Spark 4.1.3 where this head raises an overflow for a row that vanilla Spark skips. Both queries returned the Spark result with the merge-base native library (15305af). The inline comments include the fixtures and execution paths.
ajsquared
left a comment
There was a problem hiding this comment.
Reviewed the current changes and feedback; no independently confirmed unresolved P1 remains.
e518717 to
fb4d828
Compare
sunchao
left a comment
There was a problem hiding this comment.
Both earlier repros now pass on Spark 4.1.3 at c5c6793. PySpark df.rdd.take(1) returns 19723, and the original Parquet mask query returns 1023 with row-filter pushdown enabled. I found two other scan cases where Comet suppresses an overflow that Spark raises. I also compared physical plans with merge base 0f90a7d and found three cases where the new safeguard drops native execution unnecessarily.
ajsquared
left a comment
There was a problem hiding this comment.
Reviewed the current changes and feedback; no independently confirmed unresolved P1 remains.
sunchao
left a comment
There was a problem hiding this comment.
I reproduced Parquet queries that suppress a required ARITHMETIC_OVERFLOW and others that raise it after Spark would have skipped the row. The inline comments include the SQL and controls. The new PySpark RDD bridge test and partial-consumption test both passed locally on Spark 4.1.3.
I also rechecked the bounded-projection discussion. unix_date is covered now. month(make_date(2024,m,d)) over INT columns still moves from CometProject on the base to a Spark Project on this head, although every input is safe to evaluate with ANSI disabled.
The fork’s Rust tests and four default Spark 4.1 Comet shards passed. The expanded Spark SQL and Delta runtime jobs were skipped.
| 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) |
There was a problem hiding this 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.
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.
| case e => | ||
| e.dataType match { | ||
| case DateType => !pushDownDate | ||
| case TimestampType | TimestampNTZType => !pushDownTimestamp |
There was a problem hiding this 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.
| .options_mut() | ||
| .execution | ||
| .parquet | ||
| .pushdown_filters = false; |
There was a problem hiding this 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.
| 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 |
There was a problem hiding this 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.
| // 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]))) |
There was a problem hiding this 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.
| _.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) |
There was a problem hiding this 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.
ajsquared
left a comment
There was a problem hiding this comment.
Reviewed the current changes and feedback; no independently confirmed unresolved P1 remains.
comphead
left a comment
There was a problem hiding this comment.
Thanks @pingzh for working through all of this. The kernel change looks right: Spark's localDateToDays is MathUtils.toIntExact(localDate.toEpochDay) in 3.4.3, 3.5.8, 4.0.1 and 4.1.1, so ARITHMETIC_OVERFLOW in both ANSI modes matches every supported version. My comments are about the planner and scan parts, which reach further than the overflow.
One coordination point without an anchor: #6246 and #5932 also add NativeScanCommon field 20, and #6246 also edits init_datasource_exec.
| 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) |
There was a problem hiding this 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?
| 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) | ||
| } |
There was a problem hiding this 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.
| commonBuilder.setHasDataFilters(dataFilters.nonEmpty) | ||
| val filterProtos = new ListBuffer[Expr]() | ||
| for (filter <- dataFilters) { | ||
| for (filter <- dataFilters.flatMap(filterForParquetPushdown(_, conf))) { |
There was a problem hiding this 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?
| fn contains_make_date(expr: &Arc<dyn PhysicalExpr>) -> bool { | ||
| expr.exists(|e| Ok(e.downcast_ref::<MakeDateExpr>().is_some())) | ||
| .unwrap_or(true) | ||
| } |
There was a problem hiding this 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?
| 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 => |
There was a problem hiding this 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.
| -- 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 |
There was a problem hiding this 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?
Which issue does this PR close?
Closes #6741. Final part of the split of #7117, following #7260, #7261, and #7283.
Rationale for this change
make_date(6000000, 1, 1)is a valid Java calendar date, but its epoch day exceeds Date32. Spark raisesARITHMETIC_OVERFLOWin both ANSI modes. Comet currently loses the overflow distinction and returns NULL with ANSI disabled or reports an invalid calendar date with ANSI enabled.What changes are included in this PR?
Separate calendar validation from checked Date32 conversion and propagate overflow through the existing typed Spark error path. Invalid calendar fields retain their ANSI-dependent behavior, including Java's year limits; NULL arguments still skip validation.
Preserve partial consumption through external row iterators, including PySpark
DataFrame.rddand preplanned DataFrames. Potentially throwing root projections use Spark; safe ANSI-off projections and date wrappers, materialized inputs, complete file writes, and a directLIMIT 1without OFFSET can remain native. Limits or filters inside a write retain their safeguards, and residual date filters retain the native Parquet reader. Larger root limits keep the safeguard because an RDD consumer can stop before the limit. The existing supported operator-fallback opt-out remains available.Keep
make_datepredicates above Parquet scans and disable sibling row filters for affected scans, preserving predicate order and structured Spark errors. Derive this guard from enclosing Spark filters and the original Catalyst predicates so it survives partition-column, nondeterministic, unsupported-expression, and unresolved-subquery filtering, including in Delta scans. Exclude each original predicate containingmake_datebefore partially rewriting metadata predicates, so rewriting cannot hide an overflow. Independent predicates retain metadata pruning according to Spark's Parquet settings, including type-specific opt-outs, safe AND/OR/NOT composition, and Spark's IN threshold. Apply the same settings when scalar-subquery filters are resolved at execution time.Add coverage for Date32 endpoints and adjacent overflow, calendar-error precedence, NULL and empty input, and overflowing rows skipped by LIMIT,
RDD.take, andmapPartitions. Document the overflow behavior.How are these changes tested?
Focused validation used JDK 17:
Validation before the latest Scala changes also passed 10 native
make_datetests, 28 native Parquet tests, 133 broader JVM tests, and 24 public PySparkDataFrame.rddcases with ANSI and AQE on/off. The native overflow regression, unsupported-OR predicate regression, and disabled-pushdown scalar-subquery regression each failed before their respective fixes.Expanded CI coverage still needs
run-spark-4.1-tests,run-all-spark-profiles, andrun-delta-testsbefore merging.