JingsongLi commented on PR #9023:
URL: https://github.com/apache/paimon/pull/9023#issuecomment-5264322255
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
