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

    Labels

    area:scanParquet scan / data readingenhancementNew feature or request

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions