fhan688 opened a new pull request, #3541: URL: https://github.com/apache/fluss/pull/3541
Linked issue: https://github.com/apache/fluss/issues/3284 Extend the Hudi lake integration so Flink union read works for primary key tables. Building on the union-read test base added in #3522 and the log-table IT in #3528, this PR adds the missing piece for PK tables: a sorted record reader for Hudi merge-on-read splits (so remote Hudi rows can be merged with Fluss log records by primary key) plus end-to-end IT coverage. ### Brief change log - `fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSortedRecordReader.java` — new `SortedRecordReader` implementation for Hudi. Resolves record-key columns from `FlinkOptions.RECORD_KEY_FIELD`, drains the underlying `HudiR by the projected key positions, and exposes `order()` so the union-readmerger can interleave remote Hudi rows with Fluss log records. Includes a `canSortByRecordKey(...)` predicate that rejects projections dropping any record-key column. - `fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiLakeSource.java` — in `createRecordReader`, route to `HudiSortedRecordReader` when the Hudi table is `MERGE_ON_READ` and the projection retains every record-key field; otherwise keep the existing `HudiRecordReader` path. Mirrors the contract already used by `PaimonLakeSource` / `PaimonSortedRecordReader`. ### Tests - New IT `fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hryKeyTableITCase.java` (extends `FlinkUnionReadTestBase`): - `testUnionReadFullType` — batch union read over a full-type PK schema, projection (`select c3, c4`, `select c3`) and predicate pushdown (`c4 = 30`); validates state both before and after upsert. - `testUnionReadInStreamMode` — streaming changelog correctness on partitioned and non-partitioned PK tables, including `-U`/`+U` emission after upsert. - `testUnionReadPrimaryKeyTableFailover` — `INSERT ... SELECT` pipeline point and resumed; verifies changelog continuity across restart. - New UT `fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiSortedRecordReaderTest.java` — covers `canSortByRecordKey` projection rules for single-column and composite primary keys. ### API and Format No public API or storage format changes. `HudiSortedRecordReader` is `@Internal` to `fluss-lake-hudi` and reuses the existing `org.apache.fluss.lake.source.SortedRecordReader` contract defined in `fluss-common`. ### Documentation No new user-facing feature; no documentation changes required. -- 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]
