apache / apache/datafusion-comet
Native Delta scan: close predicate-pushdown gaps vs Delta Scala (StartsWith, EqualNullSafe, materialised InSubquery, generated columns)
- Dominant language
- Scala
- Stars
- 1.3k
- Forks
- 373
- Avg merge
- 2d 4h
- Merged PRs (30d)
- 198
Description
## Context
Comet's native Delta scan (PR #3932) translates Catalyst predicates to `delta-kernel-rs` predicates for file-level data skipping, then lets DataFusion's `ParquetSource` handle row-group / page-index / row-level filtering within each file.
Comet's predicate translator (`native/core/src/delta/predicate.rs` → `catalyst_to_kernel_predicate_with_names`) currently handles the core comparisons (`=`, `!=`, `<`, `<=`, `>`, `>=`, `AND`, `OR`, `NOT`, `IS NULL`, `IS NOT NULL`, `IN`, `NOT IN`, `Cast`-unwrap). For any predicate shape it doesn't recognise, it falls through to `Predicate::unknown(...)`, which disables stats-based skipping for that leg (conservative, never incorrect).
Delta's Scala `DataSkippingReader` supports several predicate shapes that Comet does not. On workloads that rely on those shapes, Comet reads more files than Delta's Scala path does for the same query, even though both paths end up with the same Parquet-level filtering once the files are opened.
This is an umbrella tracking the four gaps I've identified while doing a side-by-side analysis of both paths. Each subtask is independently mergeable.
## Gaps
### 1. `StartsWith` file-level skipping
**What Delta Scala does** (prefix-bounds comparison on truncated string stats):
`spark/src/main/scala/org/apache/spark/sql/delta/stats/DataSkippingReader.scala:557`
```scala
case StartsWith(SkippingEligibleColumn(a, _), v @ Literal(s: UTF8String, dt: StringType)) =>
statsProvider.getPredicateWithStatTypes(a, dt, MIN, MAX) { (min, max) =>
val sLen = s.numChars()
substring(min, 0, sLen) <= v && substring(max, 0, sLen) >= v
}
```
A file can be skipped only if no row in it could begin with the literal prefix.
**What Comet should do:** In `native/core/src/delta/predicate.rs`, add a case for `StartsWith` that emits the equivalent kernel predicate (kernel supports substring / string comparison). Preserves file skipping for prefix-filtered string-key workloads (log tables keyed on `run_id`, `customer_prefix`, etc.).
### 2. `EqualNullSafe` (`<=>`) file-level skipping
**What Delta Scala does** (rewrite to handle null literal separately):
`spark/src/main/scala/org/apache/spark/sql/delta/stats/DataSkippingReader.scala:508`
```scala
case EqualNullSafe(a, v: Literal) =>
val rewrittenExpr = if (v.value != null) And(IsNotNull(a), EqualTo(a, v)) else IsNull(a)
constructDataFilters(rewrittenExpr)
case EqualNullSafe(v: Literal, a) =>
constructDataFilters(EqualNullSafe(a, v))
case Not(EqualNullSafe(a, v: Literal)) =>
val rewrittenExpr = if (v.value != null) And(IsNotNull(a), EqualTo(a, v)) else IsNull(a)
constructDataFilters(Not(rewrittenExpr))
```
Non-null literal → `IsNotNull AND EqualTo` (reuses min/max equality path). Null literal → `IsNull` (uses `nullCount` stats).
**What Comet should do:** In `predicate.rs`, recognise `EqualNullSafe` (Catalyst's `<=>`) and apply the same rewrite before translation. Cheap, local change.
### 3. Materialised `InSubqueryExec` file-level skipping
**What Delta Scala does** (treat the resolved subquery as a literal IN-list):
`spark/src/main/scala/org/apache/spark/sql/delta/stats/DataSkippingReader.scala:571`
```scala
case in: InSubqueryExec =>
// At this point the subquery has been materialized, but values() can return None if
// the subquery was bypassed at runtime.
in.values().flatMap(v => constructLiteralInListDataFilters(in.child, v.toSeq))
```
Once the subquery has run, its values are just a materialised `IN` list, so all the standard min/max-range logic applies.
**What Comet should do:** Currently Comet strips subquery expressions before handing predicates to the kernel (see `CometDeltaNativeScan.scala` predicate-build path). This is correct for DPP *partition* pruning (which happens later at `doExecuteColumnar`), but it means **file-level** stats skipping misses out. Two possibilities:
- (a) At planning time, if `InSubqueryExec.values()` is already `Some(...)`, unwrap it into a literal `IN` proto before kernel predicate build.
- (b) Let it fall through as `unknown` (current behaviour) -- option (a) is a net gain.
### 4. Generated column partition filter derivation
**What Delta Scala does** (derive partition filters from data filters for generated partition columns):
`spark/src/main/scala/org/apache/spark/sql/delta/GeneratedColumn.scala:485` and `:497`
```scala
def partitionFilterOptimizationEnabled(spark: SparkSession): Boolean =
spark.sessionState.conf
.getConf(DeltaSQLConf.GENERATED_COLUMN_PARTITION_FILTER_OPTIMIZATION_ENABLED)
def generatePartitionFilters(...): Seq[Expression] = {
...
val partitionFilters = dataFilters.flatMap { filter =>
preprocess(filter) match {
case LessThan(ExtractBaseColumn(nameParts, _), lit: Literal) =>
toPartitionFilter(nameParts, _.lessThan(lit))
case LessThanOrEqual(...) => toPartitionFilter(..., _.lessThanOrEqual(lit))
case EqualTo(...) => toPartitionFilter(..., _.equalTo(lit))
case GreaterThan(...) => toPartitionFilter(..., _.greaterThan(lit))
case GreaterThanOrEqual(...) => toPartitionFilter(..., _.greaterThanOrEqual(lit))
case IsNull(...) => toPartitionFilter(..., _.isNull())
case _ => Nil
}
}
```
Each registered `OptimizablePartitionExpression` knows how to turn a data-column predicate into a partition-column predicate (e.g. `event_time > 'X'` → `partition_date > date(X)`).
**What Comet should do:** Port the registry for the common generated-column forms (`year(col)`, `date(col)`, `hour(col)`, truncation expressions) and derive additional partition filters before handing them to `CometDeltaNativeScan.prunePartitions`. Probably lives in a new helper called from `CometScanRule` right before `applyRowTrackingRewrite`.
This is the heaviest of the four -- Delta's registry has a lot of special cases. Start with `date(timestamp)` and `year(timestamp)` since those cover the majority of real-world generated-partition-column tables.
## Priority / rough effort
| # | Gap | Effort | Impact |
|---|---|---|---|
| 1 | `StartsWith` | Small (pattern-match + kernel predicate) | Medium -- prefix-filtered string keys |
| 2 | `EqualNullSafe` | Small (local rewrite) | Low -- uncommon predicate |
| 3 | Materialised `InSubquery` | Small-Medium (plumbing pre-resolved values into predicate proto) | Medium -- joins feeding partition-level subqueries |
| 4 | Generated column derivation | Medium-Large (expression registry port) | High for tables using generated partition columns |
## Acceptance criteria
For each subtask: a correctness test in `spark/src/test/scala/org/apache/comet/` that verifies Comet reads the same set of files as Delta's Scala path for a representative query, and any regression tests in the Delta suite pass.
## Non-goals
- Row-level filtering *within* a row group: already covered by DataFusion's `RowFilter` / `ArrowPredicate` and actually stronger than Delta's Spark-vectorised reader, which only does row-group-level pushdown. Not a gap.
- DV + Parquet pushdown interaction: Comet is already stronger here (Delta disables Parquet pushdown when `useMetadataRowIndex=false`; Comet does not).
Contributor guide
Assessment
This issue has not been assessed yet.