Skip to content

[spark] Expose written columns for streaming micro-batches - #9023

Open
LsomeYeah wants to merge 3 commits into
apache:masterfrom
LsomeYeah:codex/spark-streaming-written-columns
Open

[spark] Expose written columns for streaming micro-batches#9023
LsomeYeah wants to merge 3 commits into
apache:masterfrom
LsomeYeah:codex/spark-streaming-written-columns

Conversation

@LsomeYeah

@LsomeYeah LsomeYeah commented Aug 4, 2026

Copy link
Copy Markdown
Contributor

Summary

Expose opt-in, plan-level written-column metadata for each Spark Structured Streaming micro-batch. This lets consumers of wide sparse change streams obtain stable field IDs from the files admitted to the current batch before materializing business rows, instead of scanning every column to infer updates from null values.

Changes

  • Add an explicit Core metadata model with KnownWrittenColumns(fieldIds) and the conservative AllColumns fallback.
  • Resolve file write columns through each file schema so field IDs remain stable across schema evolution and column renames. Legacy files without writeCols expand to their file schema, while unresolved metadata falls back to all columns.
  • Cache each schemaId -> TableSchema for the stream lifetime, in addition to caching (schemaId, writeCols) projections while collecting a batch summary.
  • Add the opt-in Spark option read.stream.batch-written-columns.enabled, disabled by default.
  • Build the summary from the exact splits admitted for the (startOffset, endOffset] micro-batch and attach it to the corresponding InputPartitions for driver-side discovery.
  • Keep batch metadata driver-only: it remains available while inspecting the planned InputPartitions but is not serialized with every Spark task.
  • Add the experimental PaimonSparkMicroBatchMetadata.writtenColumns(Dataset) Driver-side, zero-Action API for raw foreachBatch datasets.
  • Resolve the current query through Spark's query-id-scoped SharedState and require exactly one distinct Paimon source before returning metadata. This also handles sources that Spark replaces with LocalRelation because they have no new offset in the current batch.
  • Isolate RDD/InputPartition access behind Spark 3.2, 3.3, 3.4/3.5, and Spark 4 shims, including the Spark 4.0 implementation.
  • Return an empty result for disabled metadata, non-Paimon input, incomplete lineage, Spark-internal incompatibility, or ambiguous multiple sources so callers can conservatively fall back.
  • Document the option, result semantics, supported raw-batch usage, and the boundary that this does not provide dynamic physical reader pruning.

Testing

  • Core unit tests for multi-file/schema unions, field-ID stability, legacy metadata expansion, caching, and conservative fallback.
  • Spark integration tests for default-disabled behavior, raw foreachBatch access after a projection, and same-source self-union.
  • Verify metadata lookup does not start a Spark job.
  • Verify maxFilesPerTrigger does not mix columns from the next batch on a Data Evolution $row_tracking stream.
  • Verify an explicitly written all-null column is still reported as written.
  • Verify multiple Paimon sources remain ambiguous when one source has no data in the current batch.
  • Verify batch metadata is removed by InputPartition serialization while split data remains intact.
  • Run DataEvolutionUtilsTest and DataEvolutionFileStoreScanTest (17 tests).
  • Run PaimonSourceTest (24 tests).
  • Compile Spark 3.2, 3.3, 3.4, and 3.5 modules.
  • Package Spark 4.0 and 4.1 reactor modules with JDK 21.
  • Run formatting, validation, and diff checks.

Notes

The public Spark helper is marked @Experimental. The option is disabled by default and does not change source schema, offset/checkpoint semantics, split admission, or row reader behavior.

This PR exposes planning metadata for early narrowing inside foreachBatch; it does not implement per-micro-batch physical reader column pruning because the Structured Streaming source schema remains fixed for the query.

The query/source lookup uses Spark internals only inside paimon-spark and fails closed to an empty result if those internals cannot be resolved. It does not introduce a Paimon global registry or rely on completed source metrics.

@LsomeYeah
LsomeYeah force-pushed the codex/spark-streaming-written-columns branch from f7a16e4 to 7e33328 Compare August 5, 2026 10:34
@LsomeYeah
LsomeYeah marked this pull request as ready for review August 5, 2026 10:41
<td>Whether to read row in the form of changelog (add rowkind column in row to represent its change type).</td>
</tr>
<tr>
<td><h5>read.stream.batch-written-columns.enabled</h5></td>

@JingsongLi JingsongLi Aug 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants