[VL][DELTA] Support reading Delta Lake Change Data Feed (CDF) without falling back to vanilla Spark

[felipepessoto]

Description

Description

Gluten currently does not offload reads of Delta tables' Change Data Feed (spark.read.format("delta").option("readChangeFeed", "true")... or the table_changes() SQL function). These queries run entirely on vanilla Spark instead of the Velox backend.

Why it falls back today

A normal Delta scan is a FileSourceScanExec whose relation.fileFormat is a DeltaParquetFileFormat. Gluten's OffloadDeltaScan only matches that exact case and rewrites it into a DeltaScanTransformer:

case scan: FileSourceScanExec
    if scan.relation.fileFormat.getClass == classOf[DeltaParquetFileFormat] =>
  DeltaScanTransformer(scan)

CDF reads do not produce that plan. Delta builds them through CDCReader.DeltaCDFRelation, a generic BaseRelation whose buildScan returns RDD[Row]

Because the resulting plan is not a FileSourceScanExec over DeltaParquetFileFormat, OffloadDeltaScan never matches it, so the entire query (scan + projections building the metadata columns) stays on vanilla Spark.

Proposed work

• Recognize the CDF scan path (DeltaCDFRelation / the CDC file indexes) and offload the underlying parquet reads to Velox.
• Materialize the synthesized _change_type / _commit_version / _commit_timestamp columns (literals + projections) so they can be produced natively rather than forcing a fallback.
• Add gluten-ut coverage for batch CDF reads (readChangeFeed and table_changes()), including add/remove/cdc-file combinations and column mapping.

Gluten version

main branch

[malinjawi]

Thanks @felipepessoto for raising this issue!

I took a closer look at this, and I think this is a valid Gluten/Velox Delta gap.

The fallback seems to happen before we reach the normal Delta scan offload path. Delta CDF reads are exposed to Spark as CDCReader.DeltaCDFRelation, so Gluten does not initially see the usual Delta file scan shape that OffloadDeltaScan handles.

The useful detail is that Delta itself eventually expands the CDF read through CDCReader.changesToBatchDF(...), producing normal file scans over CDF-aware indexes such as TahoeChangeFileIndex, CdcAddFileIndex, and TahoeRemoveFileIndex, with DeltaParquetFileFormat(isCDCRead = true). The CDF metadata columns like _commit_version, _commit_timestamp, and _change_type are mostly represented through Delta’s generated/partition-style columns, while CDC parquet files carry the CDC row data.

So I do not think we need to start by adding a completely new native CDF reader in Velox. A cleaner approach may be to add a Gluten Delta planner strategy that recognizes CDCReader.DeltaCDFRelation, expands it using Delta’s internal CDF batch plan, and then lets the existing Gluten/Velox Delta scan offload rules apply normally.

A good fix should probably include:

• A planner strategy for DeltaCDFRelation.
• Version-specific Delta helper shims, since the CDF APIs differ between Delta versions.
• A regression test covering table_changes(...) with insert, update preimage/postimage, and delete rows.
• A plan assertion that the resulting CDF query contains DeltaScanTransformer instead of falling back fully to Spark.

I have tinkered a prototype in this direction and it compiles for the Spark 3.5 / Delta 3.3 profile. I have not yet validated the full Velox runtime test because that needs the native Gluten/Velox library setup, but the issue itself looks real and actionable.

cc: @zhztheplayer @zhouyuan Any other ideas?