apache / apache/datafusion-comet
Track nested-field statistics pruning and TIMESTAMP_MILLIS overflow parity in filtered scans
- Dominant language
- Scala
- Stars
- 1.3k
- Forks
- 373
- Avg merge
- 2d 4h
- Merged PRs (30d)
- 198
Description
## Problem
Track the remaining filtered-scan correctness gap from #5553 and the nested-field statistics pruning needed before removing Comet's checked-timestamp safeguard.
Comet currently sets `SparkParquetOptions.checked_timestamp_overflow = false` for **all scans with Spark data filters**, including filters Comet cannot serialize. An overflowing Parquet `TIMESTAMP_MILLIS` therefore becomes `NULL`, even when vanilla Spark reads the value and throws `ArithmeticException: long overflow`.
The unfiltered nested-field fix can land independently. This issue covers the filtered-scan work that remains afterward.
## Upstream dependency
https://github.com/apache/datafusion/issues/20871 tracks row-group statistics pruning for struct-field predicates. DataFusion 55 supports row filtering on primitive struct leaves, but does not yet use nested leaf statistics to prune row groups. These are distinct capabilities.
Nested statistics pruning is a prerequisite, not a sufficient fix by itself: Comet must also preserve predicate semantics across millisecond-to-microsecond conversion and account for Spark's other pruning/filtering paths before enabling checked conversion for filtered scans.
## Verified reproduction
Verified on Spark 4.1 with Comet's native DataFusion scan and DataFusion 55.0.0, with Comet row-filter pushdown both off and on.
Write one Parquet row group with plain encoding:
| s.k | s.ts (raw TIMESTAMP_MILLIS integer) |
| --- | --- |
| 0 | 0 |
| 1 | 9223372036854776 |
```sql
SELECT s.k, s.ts FROM parquet_file WHERE s.k >= 0;
```
Both rows satisfy the predicate, so statistics pruning cannot remove the overflowing row.
- Vanilla Spark: throws `java.lang.ArithmeticException: long overflow`.
- Comet: succeeds with `(0, 1970-01-01 00:00:00)` and `(1, NULL)`.
The second timestamp fits in a signed 64-bit millisecond value, but multiplying it by 1,000 does not fit in a signed 64-bit microsecond value.
### Runnable Comet test
Paste this test into `spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala`. It uses the suite's existing Parquet writer and Spark/Comet comparison helpers. It intentionally asserts the current divergence, so a passing test reproduces the bug.
```scala
test("reproduce #5553 filtered nested TIMESTAMP_MILLIS mismatch") {
withTempDir { dir =>
val path = new Path(dir.toURI.toString, "part-r-0.parquet")
val schema = MessageTypeParser.parseMessageType("""
|message root {
| optional group s {
| optional int32 k;
| optional int64 ts(TIMESTAMP_MILLIS);
| }
|}
|""".stripMargin)
val writer = createParquetWriter(schema, path, dictionaryEnabled = false)
Seq(0L, 9223372036854776L).zipWithIndex.foreach { case (millis, k) =>
val record = new SimpleGroup(schema)
val nested = record.addGroup(0)
nested.add(0, k)
nested.add(1, millis)
writer.write(record)
}
writer.close()
// Both rows pass the nested predicate, so Spark cannot prune the overflow away.
// This deliberately asserts the remaining mismatch, rather than parity with Spark.
Seq(false, true).foreach { rowFilterPushdown =>
withSQLConf(
SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true",
CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.key ->
rowFilterPushdown.toString) {
readParquetFile(path.toString) { df =>
val query = df.where("s.k >= 0").select("s.k", "s.ts")
assert(collect(query.queryExecution.executedPlan) { case _: CometNativeScanExec =>
true
}.nonEmpty)
val (sparkError, cometError) = checkSparkAnswerMaybeThrows(query)
assert(sparkError.exists { error =>
Iterator.iterate(error)(_.getCause).takeWhile(_ != null).exists {
case _: ArithmeticException => true
case _ => false
}
})
assert(cometError.isEmpty)
assert(query.collect().toSeq == Seq(Row(0, new java.sql.Timestamp(0)), Row(1, null)))
}
}
}
}
}
```
From the repository root, after building the native library:
```sh
SPARK_LOCAL_IP=127.0.0.1 ./mvnw test -Dtest=none \
-Dsuites="org.apache.comet.parquet.ParquetReadV1Suite reproduce #5553"
```
## Expected behavior and completion criteria
- If Spark reads an overflowing value and throws, Comet must also throw, including for nested values.
- If Spark prunes overflowing data and succeeds, Comet must also succeed.
- Cover mixed/unprunable row groups and prunable row groups, nested timestamp predicates, and conversion-aware predicate rewriting.
- Remove or narrow the filtered-scan safe-cast fallback only when those behaviors are verified; do not simply enable checked conversion globally.
Contributor guide
Assessment
This issue has not been assessed yet.