wombatu-kun opened a new pull request, #20318: URL: https://github.com/apache/hudi/pull/20318
### Describe the issue this Pull Request addresses Closes #20317. With `read.source-v2.enabled=true`, a Flink query that filters a bucket index table on all its bucket index key fields fails while Flink builds the job graph, with `java.io.NotSerializableException: org.apache.hudi.source.prune.PrimaryKeyPruners$$Lambda$...`. `PrimaryKeyPruners.getBucketIdFunc` returns a plain `java.util.function.Function` lambda, and the FLIP-27 `HoodieSource` carries it into the job graph twice: through its reader function supplier, which captures the `HoodieTableSource` and with it the `dataBucketFunc` field, and through the `PartitionBucketIdFunc` in its scan context. ### Summary and Changelog The bucket id function built for bucket pruning is serializable, so Source V2 queries that filter by the bucket keys can be submitted. - `PrimaryKeyPruners#getBucketIdFunc` returns Flink's `SerializableFunction`, which extends `java.util.function.Function`, so the returned lambda is serializable while its callers keep using it as a `Function`. - Tests: `ITTestHoodieDataSource#testBucketPruning` also runs with Source V2, on COPY_ON_WRITE and MERGE_ON_READ. `TestPrimaryKeyPruners#testBucketIdFunctionIsSerializable` checks that the function survives Java serialization and still computes the same bucket ids. ### Impact Source V2 batch and streaming queries with an equality filter on all the bucket index key fields of a bucket index table run instead of failing at job submission. 1.2.1 fails the same way, and its `PrimaryKeyPruners` is identical to master's. No storage format, configuration or user-facing API change. ### Risk Level Low. Only the target type of the lambda changes. It captures the same list of key values, so the bucket ids it computes do not change. `org.apache.flink.util.function.SerializableFunction` is `@Public` and available for every supported Flink version: in `flink-core` up to 1.19, and in `flink-core-api`, a dependency of `flink-core`, from 1.20. Verified: - The Source V2 cases of `testBucketPruning` fail on master with `NotSerializableException` and pass with the fix, under the `flink1.20` profile and CI's `flink2.2` profile; the legacy cases pass either way. A streaming Source V2 read with the same filter fails and passes the same way. - The new unit test fails without the fix. - The other objects the source carries for bucket pruning are serializable, including the calculator of the partition-level bucket index, whose Caffeine cache serializes its settings without its entries. - It merges cleanly with #20292. It conflicts with my #20314 on the `getBucketIdFunc` signature; whichever lands second resolves it by having both overloads return `SerializableFunction`, which the compiler enforces. ### Documentation Update None. ### Contributor's checklist - [x] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute) - [x] Enough context is provided in the sections above - [x] Adequate tests were added if applicable -- 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]
