apache / apache/datafusion-comet

Converge the two Delta read paths into one plugin (clean architecture + performance)

Open
#5,411 0 comments 1 reaction 0 assignees View on GitHub
area:scan enhancement
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.