Skip to content

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

Description

@parthchandra

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.

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 (feat: build gate + inert wiring for contrib Delta scans [Delta contrib split, part 2] #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 (feat: build gate + inert wiring for contrib Delta scans [Delta contrib split, part 2] #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<Vec<ParquetAccessPlan>>) --
    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?

Metadata

Metadata

Assignees

No one assigned

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions