Support native Iceberg scans in Spark Structured Streaming

[peterxcli]

What is the problem the feature request solves?

Comet currently skips native plan rewriting for all Spark Structured Streaming queries:

datafusion-comet/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala

Lines 561 to 564 in 8a8fe44

	// Comet does not support structured streaming. Fall back to Spark for any plan that
	// belongs to a streaming query (detected via StreamSourceAwareSparkPlan.getStream).
	if (ShimCometStreaming.isStreamingPlan(plan)) return plan

As a result, Iceberg streaming reads cannot use Comet’s existing native Iceberg scan.

This would benefit event-driven pipelines where a main Iceberg event table collects user activity and Spark Structured Streaming continuously updates downstream silver or gold tables.

Reference: https://iceberg.apache.org/docs/latest/spark-structured-streaming/

Describe the potential solution

A reasonable initial scope is native Iceberg micro-batch reads:

• Recognize supported Iceberg MicroBatchScanExec plans.
• Reuse Iceberg’s already-planned, offset-bounded file tasks for each micro-batch.
• Rewrite only the supported scan leaf to Comet’s existing native Iceberg scan.
• Preserve Spark fallback for unsupported formats or Iceberg features.
• Keep checkpoints, offsets, stateful operators, and streaming sink commits managed by Spark and Iceberg.
• Add tests covering multiple snapshots, checkpoint restart, progress metrics, and native-plan assertions.

Native stateless operators could be enabled around the scan separately once the scan and restart semantics are validated.

Additional context

This is an initial proposal. More investigation is needed around Spark-version-specific streaming plan and progress-reporting APIs.

Implementation status and remaining gaps (2026-09-13)

Draft PR #5888 implements experimental native Iceberg append micro-batch scans, compatible streaming aggregate computation around Spark state-store boundaries, and batch CDC through .changes / create_changelog_view. The batch CDC path covers carry-over removal, update before/after images, and net changes using Iceberg's existing snapshot/timestamp range planning. This is draft work, not merged support.

Gap ownership

Remaining gap	Where the limitation belongs	Work needed
CDC over MOR changes involving delete files	Iceberg 1.11's Java changelog planner, with subsequent Comet integration	The planner rejects non-replacement snapshots containing delete manifests, including with Comet disabled. Track apache/iceberg#14264 (open, unmerged): its September 12 revision includes core position/equality-delete and V3 deletion-vector changelog planning. Engine execution remains separate; Comet still needs to carry the new tasks into Rust and emit newly deleted rows while excluding prior deletes. A full Rust changelog planner is tracked by apache/iceberg-rust#1636, but is not required while Comet reuses Java planning. Related: issue #16342 (open MOR CDC discussion) and earlier PR #10935 (open, unmerged).
Binary values in native CDC processing	Comet	The JVM CDC path exists. The native operator does not yet safely reproduce its external-row comparison semantics, so processing retains the JVM iterator. This is not a general restriction on reading binary columns.
Nested floating-point values in native CDC processing	Comet	Complete and validate equivalent comparison behavior for floats/doubles inside structs, arrays, and maps. The current guard retains JVM CDC processing.
Unrecognized Iceberg procedure layouts	Comet's Spark/Iceberg compatibility adapter	Add and test recognition for additional procedure implementations, or a more stable integration boundary. Unknown layouts retain the original JVM iterator; this is not a missing Iceberg CDC operation.
Native stream-stream joins	Comet	Unfinished; there is no established external blocker. Implement persistent state for both inputs, cross-batch matching, watermark eviction, outer-join emission, and commit/abort/recovery semantics. Existing batch joins do not supply this contract.
Native streaming deduplication	Comet	Unfinished; there is no established external blocker. Implement membership state across batches, watermark-based expiration where applicable, and correct replay after failed attempts. Existing batch distinct is insufficient.
Comet-owned checkpoint persistence	Comet/Spark state and recovery integration	Spark retains checkpoint persistence; replacing that requires separate state/recovery integration. This includes a native state backend, offset/commit logs, schema compatibility, task retries, and sink completion. Retaining Spark checkpoints reuses its existing recovery protocol.

None of these gaps is a Polaris limitation. Polaris provides the Iceberg REST catalog; switching to Lakekeeper would not supply the missing changelog or native execution functionality.

Batch CDC versus streaming CDC

Iceberg's batch CDC API has landed. Its standard Spark readStream source still consumes append snapshots; overwrite/delete snapshots fail by default, and the skip options ignore those snapshots rather than produce change events. PR #5888 does not add a mutable-table streaming CDC source. Reading the current contents of an MOR table is also distinct from extracting its change history.

Evidence and validation

• Iceberg 1.11 changelog planner rejection.
• Upstream status rechecked on September 13, 2026. The author's September 12 update to #14264 describes reworked planning and DV handling; the current diff includes DV addition, replacement, and rewrite tests. This supersedes the earlier note that V3 DV planning was future work. The PR remains unmerged; engine execution is still separate.
• CDC type guards and procedure matching in PR #5888.
• Iceberg batch CDC procedure and Spark streaming reads.

The draft passed 17 integration tests against Polaris/RustFS with Spark 4.1.3 and Iceberg 1.11.0, including native-plan assertions, COW V1/V2/V3 CDC, and streaming checkpoint/retry checks. The same 17 tests passed with a local catalog after rebasing. These results do not establish support for the unfinished features above.

[peterxcli]

take