fix(spark): read INLINE blobs as CONTENT on internal write-side Lance…#19236
Conversation
… reads
MOR compaction, clustering, and upsert merge read Lance base files through
SparkReaderContextFactory -> SparkFileFormatInternalRowReaderContext ->
SparkLanceReaderBase, which resolves hoodie.read.blob.inline.mode from the
broadcast Hadoop conf. The factory never set it, so the DESCRIPTOR default
applied and rewrites persisted INLINE blobs with null data, silently losing
the bytes of every carried-over row (all rows, in the clustering case).
- Pin hoodie.read.blob.inline.mode=CONTENT in the conf that
SparkReaderContextFactory broadcasts; the factory only serves internal
write-side and index reads, never user-facing queries, which build their
own conf from session options.
- Add a per-row guard in HoodieSparkLanceWriter that rejects the
descriptor-leak shape {type=INLINE, data=null, reference!=null} so any
future leak fails loudly instead of dropping bytes; {INLINE, null, null}
stays writable.
- Fix the stale comment in SparkLanceReaderBase claiming CONTENT is the
config default (it is DESCRIPTOR).
- Un-mask testBlobInlineCompactionRoundTrip by forcing all rows into one
file group so untouched rows actually go through the compaction rewrite,
and correct its doc about which reader compaction uses.
- Add testBlobInlineClusteringRoundTrip and writer-guard tests.
Fixes apache#19232
…d_blob Assert payload.data under CONTENT mode before calling read_blob() in the compaction and clustering round-trip tests. On a regression the failure now reads as explicit data loss (data is null after the rewrite) instead of read_blob()'s misleading DESCRIPTOR-mode IllegalStateException, which made apache#19232 look like an error by design. Also drop a stale comment crediting HoodieSparkLanceReader's CONTENT pin for compaction correctness.
|
Our original test did not capture this error because:
TLDR: New tests forces |
The un-masking of apache#19232 relies on coalesce(1) plus shuffle parallelism 1 forcing all rows into one file group, but nothing asserted it. Pin the invariant before the anyHadLogs scan so drifting back to multiple file groups (log-free groups compaction never rewrites) fails loudly instead of silently re-masking the regression.
BlobDescriptorTransform never emits OUT_OF_LINE; it synthesizes the reference sub-struct while preserving type=INLINE. The old wording described the exact wrong mental model that hid apache#19232.
Assert hoodie.read.blob.inline.mode=CONTENT on the captured broadcast Configuration so dropping the pin in SparkReaderContextFactory fails a fast unit test instead of only the Lance functional suite.
BlobDescriptorTransform and the HoodieSparkLanceWriter guard each re-derived the BLOB struct ordinals, arities and INLINE token. Move them into a package-private BlobStructLayout holder (arities sourced from HoodieSchema.Blob) so the two decoders cannot drift apart.
The deltaCommits assertion message still credited the pin to HoodieSparkLanceReader; it lives in SparkReaderContextFactory. The test docstring understated HoodieSparkLanceReader's callers (bloom-index lookups and legacy HoodieWriteMergeHandle merges also use it), and the pin comment inside that reader still claimed compaction/merge/log-replay which now go through SparkLanceReaderBase.
After the SparkReaderContextFactory pin, internal rewrites cannot produce the descriptor shape; the realistic trigger is a user reading a blob table at the DESCRIPTOR default and writing the rows back (INSERT INTO ... SELECT). Stop attributing the leak solely to internal rewrites so that user knows to set hoodie.read.blob.inline.mode=CONTENT on their read.
|
@wombatu-kun Comments addressed, can you PTAL? |
…ob inline clustering test Address review on testBlobInlineClusteringRoundTrip. getLastClusteringInstant.isPresent is satisfied by a REQUESTED/INFLIGHT replacecommit (getTimelineOfActions filters by action only), so assert the instant isCompleted. Also snapshot the first commit's base files and assert the post-clustering set is disjoint, proving the rewrite ran rather than the byte assertions reading the untouched originals (the masking mode of apache#19232).
The latestBaseFileNames helper in testBlobInlineClusteringRoundTrip referenced engineContext without binding it, causing a Scala compile error. Define it locally via HoodieLocalEngineContext(mc.getStorageConf), matching the three other FileSystemViewManager call sites in this file.
github-actions
Bot
added
size:L
| // only come back if HoodieSparkLanceReader's CONTENT pin held during the compactor's | ||
| // base-file read — otherwise untouched ids 3..5 would have been rewritten with null | ||
| // `data` and CONTENT-mode read would surface that. | ||
| // Plain projection under CONTENT first: on a regression this fails with an explicit |
There was a problem hiding this comment.
This describes a failure mode the PR itself makes unreachable. A DESCRIPTOR-leaked row is {INLINE, data=null, reference!=null}, and every Lance write goes through HoodieSparkLanceWriter, whose new validateBlobRow rejects exactly that shape. If the CONTENT pin regresses, the inline compaction aborts inside the rewrite with a HoodieException and writeDataframe fails before this CONTENT projection ever runs.
The projection still earns its place, but against a different shape: {INLINE, null, null}, which the guard deliberately allows. Suggest rewording to that, and pointing at TestSparkReaderContextFactory as what actually pins the config. The same comment is duplicated on testBlobInlineClusteringRoundTrip.
There was a problem hiding this comment.
Right -- with validateBlobRow in the writer, a DESCRIPTOR leak now aborts the rewrite itself, so this projection can never see {INLINE, null, populated reference}. Reworded both comments (compaction + clustering) to say what the projection actually backstops -- the {INLINE, null, null} shape the guard deliberately allows -- and pointed at TestSparkReaderContextFactory for the CONTENT pin.
| * asserts a replacecommit actually completed, and verifies every row's bytes survived. | ||
| */ | ||
| @Test | ||
| def testBlobInlineClusteringRoundTrip(): Unit = { |
There was a problem hiding this comment.
SparkReaderContextFactory and HoodieSparkLanceReader both name "merge" as a path the CONTENT pin protects, but no blob test exercises it. Every blob test writes via bulk_insert or insert; the only upsert is the MOR one in testBlobInlineCompactionRoundTrip, which reaches the reader through HoodieCompactor. A CoW upsert that rewrites an existing base file through FileGroupReaderBasedMergeHandle is never covered.
That path is not just a third caller of the same code: it resolves the factory via getReaderContextFactoryForWrite, which branches to AvroReaderContextFactory for AVRO record types, unlike the getReaderContextFactory used by compaction and clustering. A COPY_ON_WRITE round-trip that bulk-inserts INLINE blobs and then upserts a subset would close it.
There was a problem hiding this comment.
Added testBlobInlineCowUpsertMergeRoundTrip: CoW bulk_insert of INLINE blobs into a single file group, then upsert of a subset.
- Pins that the merge actually ran: two commits, no deltacommits, post-upsert base file disjoint from the pre-upsert one.
- Checks touched rows carry the new bytes and untouched rows keep the originals, with the same DESCRIPTOR/CONTENT checks as the compaction and clustering tests.
The class's write helper uses DefaultSparkRecordMerger, so this exercises the SPARK branch of getReaderContextFactoryForWrite (SparkReaderContextFactory), i.e. the default datasource path.
…ally backstop
The old comments described catching a persisted DESCRIPTOR leak
({INLINE, null data, populated reference}), but validateBlobRow in
HoodieSparkLanceWriter now rejects that shape inside the rewrite, so the
test would fail at the compaction/clustering write before the projection
runs. Reword both sites (compaction + clustering) to the shape the guard
deliberately allows, {INLINE, null, null}, and point at
TestSparkReaderContextFactory as what pins the CONTENT config.
Compaction and clustering resolve their reader via getReaderContextFactory, but a CoW upsert rewrites the base file through FileGroupReaderBasedMergeHandle, which resolves it via getReaderContextFactoryForWrite -- a separate path that branches on the record merger type. No blob test exercised it. Add a CoW round-trip that bulk-inserts INLINE blobs into a single file group, upserts a subset, proves the merge rewrote the base file (disjoint base file names, no deltacommits), and verifies touched rows carry new bytes while untouched rows retain the originals. Hoist latestBaseFileNames out of the clustering test for reuse.
fa45454 to
37ae5f5
Compare
3 participants
Describe the issue this Pull Request addresses
Closes #19232.
On Lance tables with INLINE blobs, compaction/clustering/upsert-merge read base files through
SparkReaderContextFactory -> SparkLanceReaderBase, which honorshoodie.read.blob.inline.mode. The factory never set it, so the DESCRIPTOR default applied and rewrites persisted INLINE blobs withdata=null, silently losing the bytes of every carried-over row (all rows, in the clustering case).Summary and Changelog
hoodie.read.blob.inline.mode=CONTENTin the confSparkReaderContextFactorybroadcasts. The factory only serves internal write-side/index reads; user queries build their own conf, so their DESCRIPTOR/CONTENT choice is unaffected.HoodieSparkLanceWriterrejecting the descriptor-leak shape{type=INLINE, data=null, reference!=null}so any future leak fails loudly instead of dropping bytes.{INLINE, null, null}stays writable.SparkLanceReaderBaseclaiming CONTENT is the config default (it is DESCRIPTOR).testBlobInlineCompactionRoundTrip(force one file group so untouched rows actually go through the rewrite); addtestBlobInlineClusteringRoundTripand writer-guard tests.Known follow-up (not in this PR): Flink's
HoodieRowDataLanceReadercallsreadAllwithoutFileReadOptions, so it neither pins CONTENT nor honors the config.Impact
Fixes silent data loss on internal rewrites of Lance INLINE blobs. No public API or config default changes.
Risk Level
low. Regression tests cover compaction, clustering, and the writer guard.
Documentation Update
none
Contributor's checklist