[spark] Expose written columns for streaming micro-batches

[LsomeYeah]

Summary

Expose plan-level written-column IDs for each Spark Structured Streaming micro-batch. Consumers of wide sparse change streams can inspect the exact fields written by files admitted to the current batch before materializing business rows, avoiding a full-column NULL scan.

Changes

• Resolve written columns from the exact splits admitted for the current (startOffset, endOffset] batch, using each file's historical schema so field IDs remain stable across schema evolution and renames.
• Add the experimental, driver-side, zero-Action PaimonSparkMicroBatchMetadata.writtenColumnIds(Dataset) API for raw foreachBatch datasets.
• Return Optional[List[Integer]]: a present value is the exact sorted immutable field-ID list, including a known empty list; an empty Optional means unavailable, ambiguous, or conservatively unresolved.
• Carry private transient metadata on planned input partitions and recover it through the raw batch RDD lineage, without a global registry, public metadata model, Spark shim changes, or task-side metadata serialization.
• Resolve file and historical-schema metadata lazily and memoize the result. Queries that never call the API do not traverse file metadata; schema and projection caches avoid repeated metadata reads when it is called.
• Keep Spark-internal compatibility local to the helper, supporting Spark 3.2's singular input-partition accessor and later Spark versions' plural accessor with fail-closed reflection.
• Reuse Data Evolution field-ID resolution. Legacy files without writeCols expand to their historical file schema; missing schemas or unknown non-system write columns return an empty Optional for conservative all-column fallback.
• Document API semantics and the boundary that this enables early narrowing inside foreachBatch, but not per-batch physical reader pruning.

Testing

• Core DataEvolutionUtilsTest (10/10): multi-file and multi-schema unions, rename-stable field IDs, legacy writeCols, strict fallback, system fields, exact empty sets, schema caching, and projection caching.
• Spark 3 PaimonSourceTest (24/24): lazy and memoized evaluation, driver-only serialization, raw foreachBatch access after projection, zero Action, same-source self-union, multiple-source ambiguity, read limits, Data Evolution $row_tracking, and an explicitly written all-null column.
• Spark 4 runtime verification: all seven metadata tests passed. The full focused suite ran 23/24; the only failure was the existing timing-sensitive Trigger ProcessingTime 5s case, unrelated to metadata extraction.
• Spark 4 clean build with JDK 21.
• Spark 3 clean build plus Checkstyle, Spotless, Enforcer, compilation, and git diff --check.
• Independent correctness/performance review after simplification; no remaining findings.

Notes

The helper is @Experimental. Metadata planning does not change source schema, offsets or checkpoints, split admission, or row-reader behavior.

Written-column IDs are resolved only when the helper is called. If metadata cannot be proven complete, the API returns an empty Optional and callers must conservatively process all columns.

This supports early projection and narrow caching inside foreachBatch; it does not implement per-micro-batch physical reader pruning because the Structured Streaming source schema remains fixed for the query.

[JingsongLi]

Why introducing this option? Is there any problem with enabling it by default?

[LsomeYeah]

Good point. There is no known issue with enabling it by default, so I removed the option, the metadata is now planned automatically.

[JingsongLi]

Thanks for working on this. The core idea—attaching driver-only metadata to the exact InputPartitions admitted for the micro-batch and recovering it from the raw foreachBatch Dataset—looks sound to me. It avoids a global registry and correctly binds the result to the planned batch.

However, I think the implementation can be made substantially smaller:

1. Please avoid computing this metadata eagerly for every streaming query. planInputPartitions currently calls collectWrittenColumns unconditionally. That adds an extra O(number of admitted files) driver-side traversal to every micro-batch, and may also load historical schemas, even when the new API is never called. Could PaimonMicroBatchMetadata instead hold a lazy supplier over the admitted splits and compute the IDs only when writtenColumns(batch) actually finds and reads that metadata?

2. The public result model seems to have one state too many. Both Optional.empty() and AllColumns.INSTANCE require exactly the same consumer behavior: conservatively process all columns. I think the API could simply return Optional[java.util.List[Integer]] (ideally named writtenColumnIds):

   • Optional.of(emptyList()): the exact known set is empty;
   • Optional.of(ids): the exact known field IDs;
   • Optional.empty(): unavailable, ambiguous, or conservatively unknown.

   This would remove WrittenColumns, KnownWrittenColumns, and AllColumns, and callers would no longer need instanceof plus a cast.

3. The Spark-version compatibility code can probably stay local to this helper. The only difference here is Spark 3.2's inputPartition versus later versions' inputPartitions. Since this helper already relies on Spark internals, uses reflection for SharedState / StreamExecution, and fails closed, a small reflective accessor in PaimonSparkMicroBatchMetadata would avoid touching SparkShim and all of the 3.x/4.x shim implementations for this one private detail.

4. The field-ID resolution can share more of the existing Core logic. DataEvolutionUtils already has fileFieldIds; the new collector adds a second name-to-ID implementation. It may be worth extracting a strict resolver used by both paths, while preserving the new conservative fallback for an unresolved non-system field.

So my preferred shape would be: keep the transient InputPartition + RDD-lineage approach, but use a private lazy batch-metadata object, return Optional[List[Integer]], and isolate the two Spark accessor variants in the helper itself. This keeps the exact-batch, self-union, empty-source, and zero-Action semantics while reducing the public API and the number of version-specific files considerably.

One additional verification gap: because this feature depends on SharedState, StreamExecution, and DataSourceRDDPartition internals, it would be valuable to run the actual foreachBatch metadata test on at least one Spark 4 runtime, rather than relying only on compile/package coverage.

[LsomeYeah]

@JingsongLi Thanks for the thoughtful guidance! I’ve updated the PR accordingly: written-column resolution is now lazy and memoized, the API is simplified to Optional[List[Integer]], Spark-version handling is kept local to the helper, and Core field-ID resolution is shared. The Spark 4 foreachBatch metadata tests also pass, and the change is now down from 17 files to 7.