apache / apache/datafusion-comet
Converge the two Delta read paths into one plugin (clean architecture + performance)
- Dominant language
- Scala
- Stars
- 1.3k
- Forks
- 373
- Avg merge
- 2d 4h
- Merged PRs (30d)
- 198
Description
# Proposal: converge the two Delta read paths into one plugin (clean architecture + performance)
## TL;DR
We currently have two paths, under development, for reading Delta files.
- **native-parquet-reads** -- delta-spark plans, DataFusion's `ParquetSource` reads the bytes.
This is the path in **PR #5365**.
- **kernel-read** -- delta-kernel-rs reads the bytes and does the transforms + CDF.
This is the path in **PR #4366**.
These are not competitors at the architecture level -- they are two *read strategies* that can
live behind **one** Delta plugin. The native-parquet-reads path's last rebase already renamed its
proto to `DeltaSpark*` and moved its native handler to `planner/delta_spark_scan.rs` as a "sibling
of the kernel path's `delta_scan.rs`, dispatched by type_url." That is exactly what is needed for
both to coexist. This proposal tries to make that explicit:
- **One plugin, two read strategies**, selected per-scan.
- **native-parquet-reads is the default** -> page-index/row-group pruning + filter pushdown +
in-scan DV. This is the hot path.
- **kernel-read is the capability fallback** -> CDF (`readChangeFeed`) and any transform the
native path can't yet express. Declines route *here*, not to vanilla Spark.
- **Zero default surface** -- both paths gated out of the default `libcomet`, claimed through the
already-merged `CometScanContrib` SPI (#4952).
Result: kernel-read's full feature coverage AND native-parquet-reads' performance, with a true
plugin boundary.
## Why this works (and the one hard constraint)
The only real divergence is *who reads the parquet bytes*. Page-index pruning, row-group pruning,
and filter pushdown live in DataFusion's parquet reader. Kernel's reader does not have them.
- native-parquet-reads' performance is therefore **not** retrofittable onto the kernel-read data
path -- getting it means routing bytes through `ParquetSource`, which *is* the native-parquet
design. So native-parquet-reads must be the performance path.
- Conversely, most of kernel-read's extra features (column-mapping id mode, generated columns,
row_index) are achievable on the native path with more work. **CDF is the exception** -- it wants
kernel's `TableChanges`; reimplementing it natively is a large, error-prone effort. So keep
kernel-read for CDF.
## Shared module layout
```
contrib/delta/ (single plugin module, -Pcontrib-delta / --features contrib-delta)
scala/
DeltaScanContrib.scala implements CometScanContrib (the ONE claim seam)
strategy/NativeParquetStrategy native-parquet planning: emit ContribScan{type_url=...delta_spark.*}
strategy/KernelStrategy kernel-read planning: emit ContribScan{type_url=...delta_kernel.*}
DeltaCdfScanExec.scala kernel-read CDF exec
native/ (comet-contrib-delta crate, linked ONLY under the feature)
delta_spark_scan.rs native-parquet handler (calls core::build_parquet_scan)
delta_dv.rs native-parquet roaring DV decode -> ParquetAccessPlan (MOVED out of core)
delta_scan.rs / kernel_scan.rs kernel-read read path
dv_reader.rs kernel-read DV masking
```
Both native handlers register under the merged `ContribScan contrib_scan = 200` envelope and are
dispatched by `type_url`. The injector already supports multiple injectors per kind.
## Which scan strategy routes where
The `DeltaScanContrib.tryTransformV1` (and CDF's V2/row-source hook) inspects the plan and picks:
| Scan shape | Path | Why |
|---|---|---|
| Plain / partitioned read, no column mapping | **native-parquet** | full pruning + pushdown |
| Deletion vectors (inline or on-disk) | **native-parquet** | DV -> `ParquetAccessPlan`, intersects page pruning |
| Column mapping **name mode** (incl. nested) | **native-parquet** | SchemaMapper by name |
| Column mapping **id mode** | **native-parquet** (target) / kernel-read (interim) | SchemaMapper by field id; land on native, kernel until then |
| Time travel, checkpoints, OPTIMIZE'd, schema evolution, INT96, special-char paths | **native-parquet** | all inherited from `ParquetSource` |
| DPP | **native-parquet** | `CometScanWithPlanData` derived-scan helper (#4700) |
| `_metadata.row_index` consumed by plan | **kernel-read** (interim) / native-parquet (target) | native row-index emit is doable but not yet wired |
| Row tracking (`row_id`, `row_commit_version`) | **kernel-read** | needs Delta row-tracking metadata logic |
| **CDF (`readChangeFeed`)** | **kernel-read** | `TableChanges`; native reimpl not worth it |
| Generated-column partition filters | **kernel-read** (interim) / native-parquet (target) | doable on native, defer |
| Encryption, `input_file_name()`, unknown reader features | **Decline** (vanilla Spark) | neither path supports |
Rule of thumb: **native-parquet by default; kernel-read only for what native can't do; Spark only
for what neither can do.** Every decline carries a `withFallbackReason` so EXPLAIN shows why.
## Core-surface budget (the clean-plugin part)
The default `libcomet` and core modules must carry **zero** Delta surface. Concretely:
1. **Reuse the merged `CometScanContrib`** (#4952). Delete native-parquet-reads' parallel
`CometScanRuleExtension` + its hook + suite. One SPI, not two.
2. **Ride the merged `ContribScan` type_url envelope.** Drop native-parquet-reads' `DeltaSpark*`
messages from `operator.proto` core; the plugin defines its own messages packed into the
envelope's `value`.
3. **Move `delta_dv.rs` + `roaring`/`crc32fast` into the contrib crate.** Remove `delta` from the
default cargo feature set. Default build -> no Delta symbols (assert via
`verify-contrib-delta-gate.sh`, which the kernel-read path already ships).
4. **Keep native-parquet-reads' `planner.rs` shared-builder extraction, but expose it as
format-neutral** `pub fn build_parquet_scan(common, files, Option>)` --
no Delta arm, no `#[cfg(feature="delta")]` Delta dispatch inside core. The contrib crate's
handler calls it.
5. The one *new* core concept -- a **generic per-file row-selection / access plan** on the native
scan -- is justified because it is format-neutral (Iceberg positional/equality deletes want the
same thing). Frame it as "core parquet scan supports row skipping," not "core has a Delta hook."
Net core delta after this: the already-merged SPI + one reusable pub builder fn + an optional
generic access-plan field. That matches the kernel-read path's zero-default-surface contract while
preserving native-parquet-reads' full `ParquetSource` performance.
## Migration steps
**For native-parquet-reads (do first -- it becomes the default path):**
1. Replace `CometScanRuleExtension` usage with `CometScanContrib.tryTransformV1`.
2. Remove `delta` from default cargo features; move `delta_dv.rs` + deps into the contrib crate.
3. Turn the `planner.rs` Delta arm into a format-neutral `build_parquet_scan` and have the contrib
crate's `delta_spark_scan.rs` call it.
4. Relocate module to `contrib/delta/` (shared with kernel-read) instead of `contrib/delta-spark/`.
**For kernel-read (becomes the fallback layer):**
5. Land its Rust driver/executor + Scala claim/serde + CDF units as the kernel strategy inside the
same plugin, registered under a distinct `type_url` (`...delta_kernel.*`).
6. Restrict the kernel-read claim to the shapes in the table above (CDF, row tracking, id mode
interim) -- it no longer claims plain reads (native-parquet owns those).
**Joint:**
7. `DeltaScanContrib` is the single claim point; it dispatches to the native-parquet vs kernel-read
strategy.
8. One test battery, run against both strategies for the shapes each owns; one CI workflow.
## Open questions
- Is "generic per-file access plan on the native scan" acceptable core surface, or should it be
produced entirely inside the contrib crate's `build_parquet_scan` call (leaving core with only
the pub builder fn)?
- Is native CDF a near-term requirement? If **no** then fallback to Spark until CDF is implemented.
- Do the two paths agree on the layout above?
Contributor guide
Research direction
Start by reading the merged CometScanContrib SPI (#4952), then inspect planner/delta_spark_scan.rs, delta_scan.rs, the contrib/delta/ layout, and DeltaScanContrib.tryTransformV1. Review the existing gate script and test batteries mentioned in the proposal. Done means one feature-gated Delta plugin selects native-parquet or kernel-read by scan shape, preserves fallbacks, and leaves the default core build without Delta surface.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust, scala, spark
- Domain
- backend, build-system, data-engineering
- Issue type
- Refactor
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100