Support native Iceberg scans in Spark Structured Streaming

### What is the problem the feature request solves?

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

https://github.com/apache/datafusion-comet/blob/8a8fe44753cd2fbacdc01fb4328398b966604823/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala#L561-L564

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](https://github.com/apache/iceberg/pull/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](https://github.com/apache/iceberg-rust/issues/1636), but is not required while Comet reuses Java planning. Related: [issue #16342](https://github.com/apache/iceberg/issues/16342) (open MOR CDC discussion) and [earlier PR #10935](https://github.com/apache/iceberg/pull/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](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/BaseIncrementalChangelogScan.java#L94).
- Upstream status rechecked on September 13, 2026. The [author's September 12 update to #14264](https://github.com/apache/iceberg/pull/14264#issuecomment-5649211018) 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](https://github.com/peterxcli/datafusion-comet/blob/666073fd9eed10fc82683cdfb82afb9596f9bc87/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergChangelogExec.scala#L77).
- [Iceberg batch CDC procedure](https://iceberg.apache.org/docs/latest/spark-procedures/#create_changelog_view) and [Spark streaming reads](https://iceberg.apache.org/docs/latest/spark-structured-streaming/#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.