kbuci opened a new pull request, #19211:
URL: https://github.com/apache/hudi/pull/19211
…olumns
### Describe the issue this Pull Request addresses
Closes #19032.
This change adds opt-in OOL BLOB materialization directly in the Flink
source, matching the batched read behavior that Spark achieves via
`BatchedBlobReader` / `ReadBlobRule`.
### Summary and Changelog
**User-facing outcome:** When `read.blob.materialize=true` is set in the
Flink DDL `WITH` options, the source iterator materializes every OOL BLOB field
before emitting each row: the blob struct is replaced with an INLINE blob
(`type=INLINE, data=<bytes>, reference=null`). Nearby byte ranges from the same
external file are coalesced into a single seek+readFully, controlled by
`read.blob.batching.max.gap.bytes` and `read.blob.batching.lookahead.size`.
#### Changes
- **`BlobMaterializingIterator`** (new)
- A `ClosableIterator<RowData>` decorator that ports Spark's
`BatchedBlobReader` algorithm to Flink's `RowData` API.
- Buffers `lookaheadSize` rows per batch (shallow-copied to avoid
aliasing), classifies each BLOB field as INLINE (pass-through) or OOL (collect
`external_path`, `offset`, `length`), groups OOL refs by file path, sorts by
offset, merges ranges within `maxGapBytes` into a single `seek + readFully`,
then re-emits rows in original order with blob structs replaced by `(INLINE,
<bytes>, null)`.
- Whole-file OOL refs (no offset/length) are handled via
`HoodieStorage#open` per file.
- **`FlinkOptions`** (3 new config keys)
- `read.blob.materialize` (boolean, default `false`): enable OOL
materialization at the source.
- `read.blob.batching.max.gap.bytes` (int, default `4096`): max byte gap
between two OOL ranges in the same file to coalesce into one read.
- `read.blob.batching.lookahead.size` (int, default `50`): row lookahead
buffer size per batch.
- **`HoodieSplitReaderFunction`** (FLIP-27 / new source path)
- After `fileGroupReader.getClosableIterator()`, wraps the raw iterator
with `BlobMaterializingIterator` when `read.blob.materialize=true` and the
required schema contains at least one BLOB field (detected via
`HoodieSchema#isBlobField()`).
- **`MergeOnReadInputFormat`** (legacy streaming path)
- Same `maybeMaterialize()` injection at the end of `initIterator()`,
using `tableState.getRequiredRowType()` (already available) for field getter
construction.
#### Tests
- **`TestBlobMaterializingIterator`** (new, unit)
- 9 focused unit tests using a mock `HoodieStorage` and in-memory
`SeekableDataInputStream`:
- INLINE blobs pass through unchanged.
- Null blob fields pass through unchanged.
- Single OOL range is materialized with correct bytes.
- Two nearby ranges (gap ≤ `maxGapBytes`) are merged → single
`openSeekable` call (verified with Mockito `verify`).
- Two far-apart ranges are not merged.
- Original row order is preserved across multiple files.
- Mixed INLINE + OOL + null in the same batch.
- Non-blob fields (id, name, ts) are preserved in the output row.
- Missing OOL reference struct throws `IllegalStateException`.
- **`ITTestBlobWrite`** (extended)
- `testReadMaterializesOutOfLineBlob` (COW + MOR, `@EnumSource`): writes a
real temp binary file, inserts two OOL references, reads back with
`read.blob.materialize=true`, asserts `type=INLINE`, `data=<expected bytes>`,
`reference=null`. For MOR, also upserts so that merge + materialization is
exercised end-to-end.
- `testReadWithMaterializeDisabledReturnsOolStructAsIs`: same write, but
without `read.blob.materialize`; asserts the struct is still `OUT_OF_LINE` —
ensures the feature is strictly opt-in.
- `testBatchingCoalescesNearbyRanges` (COW + MOR): two OOL refs into the
same file with a 5-byte gap, `max.gap.bytes=128`; verifies both rows receive
correct byte arrays.
### Impact
- **Read path:** OOL BLOB fields can now be materialized at the Flink source
level with batched I/O, matching Spark's `BatchedBlobReader` behavior.
- **Write path:** unchanged.
- **On-disk layout:** unchanged.
- **No public API change.** Managed-blob cleaning, dynamic inline/OOL
placement, and a Flink `read_blob` UDF remain out of scope.
### Risk Level
**Low**
- Materialization is entirely opt-in (`read.blob.materialize=false` by
default). The existing read path is unaffected when the flag is unset.
- The `detectBlobFieldPositions` check short-circuits to a no-op if the
required schema has no BLOB fields, so non-BLOB tables are unaffected.
- `BlobMaterializingIterator` is a pure decorator: it does not modify the
underlying iterator, table state, or schema.
### Test plan
- [ ] `-Pflink2.1` CI: `TestBlobMaterializingIterator`
- [ ] `-Pflink2.1` CI: `ITTestBlobWrite` (COW + MOR + materialize variants)
### 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
### Describe the issue this Pull Request addresses
<!-- Either describe the issue inline here with motivation behind the
changes
(or) link to an issue by including `Closes #<issue-number>` for
context.
If this PR includes changes to the storage format, public APIs,
or has breaking changes, use `!` (e.g., feat!: ...) -->
### Summary and Changelog
<!-- Short, plain-English summary of what users gain or what changed in
behavior.
Followed by a detailed log of all the changes. Highlight if any code
was copied. -->
### Impact
<!-- Describe any public API or user-facing feature change or any
performance impact. -->
### Risk Level
<!-- Accepted values: none, low, medium or high. Other than `none`, explain
the risk.
If medium or high, explain what verification was done to mitigate the
risks. -->
### Documentation Update
<!-- Describe any necessary documentation update if there is any new
feature, config, or user-facing change. If not, put "none".
- The config description must be updated if new configs are added or the
default value of the configs are changed.
- Any new feature or user-facing change requires updating the Hudi website.
Please follow the
[instruction](https://hudi.apache.org/contribute/developer-setup#website)
to make changes to the website. -->
### Contributor's checklist
- [ ] Read through [contributor's
guide](https://hudi.apache.org/contribute/how-to-contribute)
- [ ] Enough context is provided in the sections above
- [ ] 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]
