You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "voonhous (via GitHub)" <gi...@apache.org> on 2026/07/09 12:03:30 UTC
[PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
voonhous opened a new pull request, #19236:
URL: https://github.com/apache/hudi/pull/19236
… 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 #19232
### 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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on code in PR #19236:
URL: https://github.com/apache/hudi/pull/19236#discussion_r3560444596
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLanceDataSource.scala:
##########
@@ -1443,6 +1481,150 @@ class TestLanceDataSource extends HoodieSparkClientTestBase {
}
}
+ /**
+ * Clustering must preserve INLINE blob bytes under the DESCRIPTOR default. Clustering rewrites
+ * ALL rows through MultipleSparkJobExecutionStrategy.readRecordsForGroupAsRow, which reads base
+ * files through the same internal write-side reader context as compaction
+ * (SparkReaderContextFactory -> SparkFileFormatInternalRowReaderContext -> SparkLanceReaderBase).
+ * Before SparkReaderContextFactory pinned {@code hoodie.read.blob.inline.mode=CONTENT} for
+ * internal reads, the first clustering of a Lance table with INLINE blobs read null {@code data}
+ * (the DESCRIPTOR default) and rewrote every row's blob with null bytes, silently losing all
+ * blob content (#19232). This test bulk-inserts INLINE blobs, triggers inline clustering,
+ * asserts a replacecommit actually completed, and verifies every row's bytes survived.
+ */
+ @Test
+ def testBlobInlineClusteringRoundTrip(): Unit = {
Review Comment:
Added testBlobInlineCowUpsertMergeRoundTrip: CoW bulk_insert of INLINE blobs into a single file group, then an upsert of a subset. It pins the merge actually ran (two commits, no deltacommits, post-upsert base file disjoint from pre-upsert) and checks touched rows carry the new bytes while untouched rows keep the originals, with the same DESCRIPTOR-shape and CONTENT-mode checks as the compaction/clustering tests. The class's write helper uses DefaultSparkRecordMerger, so this exercises the SPARK branch of getReaderContextFactoryForWrite (SparkReaderContextFactory), i.e. the default datasource path.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on code in PR #19236:
URL: https://github.com/apache/hudi/pull/19236#discussion_r3557432868
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkLanceWriter.java:
##########
@@ -446,12 +467,43 @@ protected void updateRecordMetadata(InternalRow row,
@AllArgsConstructor(staticName = "of")
private static class SparkArrowWriter implements ArrowWriter<InternalRow> {
private final LanceArrowWriter lanceArrowWriter;
+ // Parallel arrays of top-level BLOB column ordinals and names for the per-row guard.
+ private final int[] blobFieldOrdinals;
+ private final String[] blobFieldNames;
@Override
public void write(InternalRow row) {
+ validateBlobRow(row);
lanceArrowWriter.write(row);
}
+ /**
+ * Reject descriptor-shaped BLOB rows before they are persisted. A row is descriptor-shaped when
+ * a top-level BLOB struct is non-null, its type is INLINE, its data is null, and its reference
+ * is non-null; this only arises when a DESCRIPTOR-mode read leaks onto a write path, and writing
+ * it would silently drop the inline bytes. The legitimate empty-inline shape {INLINE, null, null}
+ * stays writable.
+ */
+ private void validateBlobRow(InternalRow row) {
+ for (int i = 0; i < blobFieldOrdinals.length; i++) {
+ int ordinal = blobFieldOrdinals[i];
+ if (row.isNullAt(ordinal)) {
+ continue;
+ }
+ InternalRow blob = row.getStruct(ordinal, 3);
Review Comment:
Hoisted them into a package-private BlobStructLayout holder (ordinals, arities sourced from HoodieSchema.Blob, and the precomputed type tokens); both classes read the layout from there now. The literal 4 for the reference arity is gone too.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLanceDataSource.scala:
##########
@@ -1303,12 +1303,21 @@ class TestLanceDataSource extends HoodieSparkClientTestBase {
/**
* Compaction must preserve INLINE blob bytes under the DESCRIPTOR default. MOR compaction reads
- * the base file via {@link HoodieSparkLanceReader}, which hard-pins CONTENT regardless of the
- * user-facing {@code hoodie.read.blob.inline.mode}. If that pin were to honor the default
- * (DESCRIPTOR), compaction would read null {@code data} and rewrite a base file without bytes,
- * silently corrupting untouched rows. This test inserts INLINE blobs, upserts a subset to force
- * compaction, and asserts that touched rows carry the new bytes while untouched rows retain the
- * originals.
+ * the base file through the internal write-side reader stack
+ * SparkReaderContextFactory -> SparkFileFormatInternalRowReaderContext -> SparkLanceReaderBase,
+ * not through HoodieSparkLanceReader (that reader only serves LanceUtils stats/key reads).
Review Comment:
Fixed all three spots: the deltaCommits message now points at SparkReaderContextFactory, the docstring parenthetical lists the reader's actual callers (stats/key reads, bloom-index lookups, legacy HoodieWriteMergeHandle merges), and the pin comment in HoodieSparkLanceReader credits those paths instead of compaction/merge/log-replay.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19236:
URL: https://github.com/apache/hudi/pull/19236#issuecomment-4926362290
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5722d39821a9d5717fa2025c1f4ff5b715be5dcc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15408",
"triggerID" : "5722d39821a9d5717fa2025c1f4ff5b715be5dcc",
"triggerType" : "PUSH"
}, {
"hash" : "78a15e131e1f0da182410559b56d3023ff50a69e",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15411",
"triggerID" : "78a15e131e1f0da182410559b56d3023ff50a69e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 78a15e131e1f0da182410559b56d3023ff50a69e Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15411)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on code in PR #19236:
URL: https://github.com/apache/hudi/pull/19236#discussion_r3557354320
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/lance/SparkLanceReaderBase.scala:
##########
@@ -131,10 +131,10 @@ class SparkLanceReaderBase(enableVectorizedReader: Boolean) extends SparkColumna
null
}
- // Honor `hoodie.read.blob.inline.mode`. CONTENT (default) materializes INLINE bytes in
- // the `data` column; DESCRIPTOR surfaces per-row {position, size} which the descriptor
- // iterator rewrites into Hudi OUT_OF_LINE references. Non-blob Lance columns ignore
- // the option regardless.
+ // Honor `hoodie.read.blob.inline.mode`. DESCRIPTOR (default) surfaces per-row
+ // {position, size} which the descriptor iterator rewrites into Hudi OUT_OF_LINE
Review Comment:
Right, the transform keeps type=INLINE and only synthesizes the reference; never emits OUT_OF_LINE. Reworded as suggested.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19236:
URL: https://github.com/apache/hudi/pull/19236#discussion_r3560189938
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLanceDataSource.scala:
##########
@@ -1422,11 +1441,30 @@ class TestLanceDataSource extends HoodieSparkClientTestBase {
s"DESCRIPTOR default should populate reference on plain read (id=$id)")
}
- // read_blob() under CONTENT mode is what we use to verify the post-compaction bytes
- // because read_blob() on INLINE rows throws under the DESCRIPTOR default. The bytes can
- // 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
Review 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`.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLanceDataSource.scala:
##########
@@ -1443,6 +1481,150 @@ class TestLanceDataSource extends HoodieSparkClientTestBase {
}
}
+ /**
+ * Clustering must preserve INLINE blob bytes under the DESCRIPTOR default. Clustering rewrites
+ * ALL rows through MultipleSparkJobExecutionStrategy.readRecordsForGroupAsRow, which reads base
+ * files through the same internal write-side reader context as compaction
+ * (SparkReaderContextFactory -> SparkFileFormatInternalRowReaderContext -> SparkLanceReaderBase).
+ * Before SparkReaderContextFactory pinned {@code hoodie.read.blob.inline.mode=CONTENT} for
+ * internal reads, the first clustering of a Lance table with INLINE blobs read null {@code data}
+ * (the DESCRIPTOR default) and rewrote every row's blob with null bytes, silently losing all
+ * blob content (#19232). This test bulk-inserts INLINE blobs, triggers inline clustering,
+ * asserts a replacecommit actually completed, and verifies every row's bytes survived.
+ */
+ @Test
+ def testBlobInlineClusteringRoundTrip(): Unit = {
Review 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.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on PR #19236:
URL: https://github.com/apache/hudi/pull/19236#issuecomment-4925349173
Our original test did not capture this error because:
1. The topology never forced the dangerous path. With default shuffle parallelism, the bulk_insert scattered ids `0..5` across multiple file groups. The upsert of ids `0..2` only produced log files in their file groups. Compaction only rewrites file groups that have log files, so the untouched ids `3..5` sat in log-free file groups that compaction never touched. Their bytes "survived" because their base file was never rewritten at all under compaction. The test even asserted that a deltacommit existed and that some file slice had log files, but never that an untouched row and a log file shared a filegroup, which is the only combination that exercises the carry-over rewrite.
2. Even the rewritten rows didn't test the base-file read. Ids `0..2` got their new bytes from the log records during the merge, not from reading the base file. So whatever blob mode the base-file read used (the actual bug) was invisible: the buggy read's output for those rows was discarded in favor of the log records anyway.
TLDR: New tests forces `3..5` into the same filegroup that needed to be re-read and rewritten into new base file under compaction but were never present in the log files, triggering the data loss error which we are fixing.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19236:
URL: https://github.com/apache/hudi/pull/19236#issuecomment-4924956337
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5722d39821a9d5717fa2025c1f4ff5b715be5dcc",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15408",
"triggerID" : "5722d39821a9d5717fa2025c1f4ff5b715be5dcc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5722d39821a9d5717fa2025c1f4ff5b715be5dcc Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15408)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19236:
URL: https://github.com/apache/hudi/pull/19236#issuecomment-4924901407
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5722d39821a9d5717fa2025c1f4ff5b715be5dcc",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5722d39821a9d5717fa2025c1f4ff5b715be5dcc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5722d39821a9d5717fa2025c1f4ff5b715be5dcc UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on code in PR #19236:
URL: https://github.com/apache/hudi/pull/19236#discussion_r3560399130
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLanceDataSource.scala:
##########
@@ -1422,11 +1441,30 @@ class TestLanceDataSource extends HoodieSparkClientTestBase {
s"DESCRIPTOR default should populate reference on plain read (id=$id)")
}
- // read_blob() under CONTENT mode is what we use to verify the post-compaction bytes
- // because read_blob() on INLINE rows throws under the DESCRIPTOR default. The bytes can
- // 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
Review 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.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on PR #19236:
URL: https://github.com/apache/hudi/pull/19236#issuecomment-4933774727
@wombatu-kun Comments addressed, can you PTAL?
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on code in PR #19236:
URL: https://github.com/apache/hudi/pull/19236#discussion_r3560399130
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLanceDataSource.scala:
##########
@@ -1422,11 +1441,30 @@ class TestLanceDataSource extends HoodieSparkClientTestBase {
s"DESCRIPTOR default should populate reference on plain read (id=$id)")
}
- // read_blob() under CONTENT mode is what we use to verify the post-compaction bytes
- // because read_blob() on INLINE rows throws under the DESCRIPTOR default. The bytes can
- // 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
Review 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.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLanceDataSource.scala:
##########
@@ -1443,6 +1481,150 @@ class TestLanceDataSource extends HoodieSparkClientTestBase {
}
}
+ /**
+ * Clustering must preserve INLINE blob bytes under the DESCRIPTOR default. Clustering rewrites
+ * ALL rows through MultipleSparkJobExecutionStrategy.readRecordsForGroupAsRow, which reads base
+ * files through the same internal write-side reader context as compaction
+ * (SparkReaderContextFactory -> SparkFileFormatInternalRowReaderContext -> SparkLanceReaderBase).
+ * Before SparkReaderContextFactory pinned {@code hoodie.read.blob.inline.mode=CONTENT} for
+ * internal reads, the first clustering of a Lance table with INLINE blobs read null {@code data}
+ * (the DESCRIPTOR default) and rewrote every row's blob with null bytes, silently losing all
+ * blob content (#19232). This test bulk-inserts INLINE blobs, triggers inline clustering,
+ * asserts a replacecommit actually completed, and verifies every row's bytes survived.
+ */
+ @Test
+ def testBlobInlineClusteringRoundTrip(): Unit = {
Review 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.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on code in PR #19236:
URL: https://github.com/apache/hudi/pull/19236#discussion_r3557526947
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkLanceWriter.java:
##########
@@ -446,12 +467,43 @@ protected void updateRecordMetadata(InternalRow row,
@AllArgsConstructor(staticName = "of")
private static class SparkArrowWriter implements ArrowWriter<InternalRow> {
private final LanceArrowWriter lanceArrowWriter;
+ // Parallel arrays of top-level BLOB column ordinals and names for the per-row guard.
+ private final int[] blobFieldOrdinals;
+ private final String[] blobFieldNames;
@Override
public void write(InternalRow row) {
+ validateBlobRow(row);
lanceArrowWriter.write(row);
}
+ /**
+ * Reject descriptor-shaped BLOB rows before they are persisted. A row is descriptor-shaped when
+ * a top-level BLOB struct is non-null, its type is INLINE, its data is null, and its reference
+ * is non-null; this only arises when a DESCRIPTOR-mode read leaks onto a write path, and writing
+ * it would silently drop the inline bytes. The legitimate empty-inline shape {INLINE, null, null}
+ * stays writable.
+ */
+ private void validateBlobRow(InternalRow row) {
+ for (int i = 0; i < blobFieldOrdinals.length; i++) {
+ int ordinal = blobFieldOrdinals[i];
+ if (row.isNullAt(ordinal)) {
+ continue;
+ }
+ InternalRow blob = row.getStruct(ordinal, 3);
+ if (blob.isNullAt(0) || !INLINE_TYPE_TOKEN.equals(blob.getUTF8String(0))) {
+ continue;
+ }
+ if (blob.isNullAt(1) && !blob.isNullAt(2)) {
+ throw new HoodieException(
+ "BLOB column '" + blobFieldNames[i] + "' has an INLINE row with null data but a "
+ + "populated reference: a DESCRIPTOR-mode read leaked into the write path, and "
+ + "persisting it would silently drop the blob bytes. Internal rewrites "
+ + "(compaction, clustering, merge) must read with hoodie.read.blob.inline.mode=CONTENT.");
Review Comment:
Widened the parenthetical as suggested, and called the INSERT INTO ... SELECT hard failure out under Impact in the description.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on code in PR #19236:
URL: https://github.com/apache/hudi/pull/19236#discussion_r3559543949
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLanceDataSource.scala:
##########
@@ -1443,6 +1481,114 @@ class TestLanceDataSource extends HoodieSparkClientTestBase {
}
}
+ /**
+ * Clustering must preserve INLINE blob bytes under the DESCRIPTOR default. Clustering rewrites
+ * ALL rows through MultipleSparkJobExecutionStrategy.readRecordsForGroupAsRow, which reads base
+ * files through the same internal write-side reader context as compaction
+ * (SparkReaderContextFactory -> SparkFileFormatInternalRowReaderContext -> SparkLanceReaderBase).
+ * Before SparkReaderContextFactory pinned {@code hoodie.read.blob.inline.mode=CONTENT} for
+ * internal reads, the first clustering of a Lance table with INLINE blobs read null {@code data}
+ * (the DESCRIPTOR default) and rewrote every row's blob with null bytes, silently losing all
+ * blob content (#19232). This test bulk-inserts INLINE blobs, triggers inline clustering,
+ * asserts a replacecommit actually completed, and verifies every row's bytes survived.
+ */
+ @Test
+ def testBlobInlineClusteringRoundTrip(): Unit = {
+ val tableType = HoodieTableType.COPY_ON_WRITE
+ val tableName = "test_lance_blob_inline_cluster_cow"
+ val tablePath = s"$basePath/$tableName"
+
+ val payloadLen = 1024
+ val numRows = 5
+ val expectedPayloads: Seq[Array[Byte]] = (0 until numRows).map { i =>
+ (0 until payloadLen).map(j => ((i + j) % 256).toByte).toArray
+ }
+ val sparkSess = spark
+ import sparkSess.implicits._
+
+ val canonicalSchema = StructType(Seq(
+ StructField("id", IntegerType, nullable = false),
+ StructField("payload", BlobType().asInstanceOf[StructType], nullable = true,
+ BlobTestHelpers.blobMetadata)
+ ))
+ def asInlineDf(idToBytes: Seq[(Int, Array[Byte])]): DataFrame = {
+ val rawDf = idToBytes.toDF("id", "bytes")
+ .select($"id", BlobTestHelpers.inlineBlobStructCol("payload", $"bytes"))
+ spark.createDataFrame(rawDf.rdd, canonicalSchema)
+ }
+
+ // First commit: bulk_insert ids 0..4 with the initial pattern into a base file.
+ writeDataframe(tableType, tableName, tablePath,
+ asInlineDf(expectedPayloads.zipWithIndex.map { case (b, i) => (i, b) }),
+ saveMode = SaveMode.Overwrite,
+ operation = Some("bulk_insert"),
+ extraOptions = Map(PRECOMBINE_FIELD.key() -> "id"))
+
+ assertLanceBlobEncoding(tablePath)
+
+ // Second commit: a small bulk_insert that trips inline clustering (max.commits=1). Clustering
+ // rewrites the existing base file's rows through readRecordsForGroupAsRow, which must read the
+ // INLINE bytes as CONTENT, not the DESCRIPTOR default, or every rewritten row loses its bytes.
+ val extraPayloads = (numRows until numRows + 2).map { i =>
+ (i, (0 until payloadLen).map(j => ((i + j) % 256).toByte).toArray)
+ }
+ writeDataframe(tableType, tableName, tablePath,
+ asInlineDf(extraPayloads),
+ operation = Some("bulk_insert"),
+ extraOptions = Map(PRECOMBINE_FIELD.key() -> "id",
+ "hoodie.clustering.inline" -> "true",
+ "hoodie.clustering.inline.max.commits" -> "1"))
+
+ // Assert a clustering (replacecommit) instant actually completed, so the test cannot silently
+ // pass without a rewrite.
+ val metaClient = HoodieTableMetaClient.builder()
+ .setConf(HoodieTestUtils.getDefaultStorageConf)
+ .setBasePath(tablePath)
+ .build()
+ assertTrue(metaClient.getActiveTimeline.getLastClusteringInstant.isPresent,
Review Comment:
Both fixed: now asserts `getLastClusteringInstant.get.isCompleted` (isPresent alone accepts a REQUESTED/INFLIGHT replacecommit), and snapshots the base files after the first commit then asserts the post-clustering set is disjoint, so a skipped rewrite fails instead of the byte checks reading the untouched originals.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19236:
URL: https://github.com/apache/hudi/pull/19236#discussion_r3558244240
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkLanceWriter.java:
##########
@@ -446,12 +467,43 @@ protected void updateRecordMetadata(InternalRow row,
@AllArgsConstructor(staticName = "of")
private static class SparkArrowWriter implements ArrowWriter<InternalRow> {
private final LanceArrowWriter lanceArrowWriter;
+ // Parallel arrays of top-level BLOB column ordinals and names for the per-row guard.
+ private final int[] blobFieldOrdinals;
+ private final String[] blobFieldNames;
@Override
public void write(InternalRow row) {
+ validateBlobRow(row);
lanceArrowWriter.write(row);
}
+ /**
+ * Reject descriptor-shaped BLOB rows before they are persisted. A row is descriptor-shaped when
+ * a top-level BLOB struct is non-null, its type is INLINE, its data is null, and its reference
+ * is non-null; this only arises when a DESCRIPTOR-mode read leaks onto a write path, and writing
+ * it would silently drop the inline bytes. The legitimate empty-inline shape {INLINE, null, null}
+ * stays writable.
+ */
+ private void validateBlobRow(InternalRow row) {
+ for (int i = 0; i < blobFieldOrdinals.length; i++) {
+ int ordinal = blobFieldOrdinals[i];
+ if (row.isNullAt(ordinal)) {
+ continue;
+ }
+ InternalRow blob = row.getStruct(ordinal, 3);
+ if (blob.isNullAt(0) || !INLINE_TYPE_TOKEN.equals(blob.getUTF8String(0))) {
+ continue;
+ }
+ if (blob.isNullAt(1) && !blob.isNullAt(2)) {
+ throw new HoodieException(
+ "BLOB column '" + blobFieldNames[i] + "' has an INLINE row with null data but a "
+ + "populated reference: a DESCRIPTOR-mode read leaked into the write path, and "
+ + "persisting it would silently drop the blob bytes. Internal rewrites "
+ + "(compaction, clustering, merge) must read with hoodie.read.blob.inline.mode=CONTENT.");
Review Comment:
The Impact section still reads "Fixes silent data loss on internal rewrites of Lance INLINE blobs. No public API or config default changes." The behavior change is not there: with the guard, `INSERT INTO t2 SELECT * FROM t1` over a Lance blob table at the DESCRIPTOR default now throws instead of writing a table whose INLINE rows carry a dangling reference. `read_blob()` already refuses those rows (`BatchedBlobReader` throws on INLINE with null data), so the guard is right, but it turns a silent-and-later-unreadable write into a hard failure at write time, and that belongs under Impact.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous merged PR #19236:
URL: https://github.com/apache/hudi/pull/19236
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19236:
URL: https://github.com/apache/hudi/pull/19236#issuecomment-4925355808
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5722d39821a9d5717fa2025c1f4ff5b715be5dcc",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15408",
"triggerID" : "5722d39821a9d5717fa2025c1f4ff5b715be5dcc",
"triggerType" : "PUSH"
}, {
"hash" : "78a15e131e1f0da182410559b56d3023ff50a69e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "78a15e131e1f0da182410559b56d3023ff50a69e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5722d39821a9d5717fa2025c1f4ff5b715be5dcc Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15408)
* 78a15e131e1f0da182410559b56d3023ff50a69e UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19236:
URL: https://github.com/apache/hudi/pull/19236#issuecomment-4933290389
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5722d39821a9d5717fa2025c1f4ff5b715be5dcc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15408",
"triggerID" : "5722d39821a9d5717fa2025c1f4ff5b715be5dcc",
"triggerType" : "PUSH"
}, {
"hash" : "78a15e131e1f0da182410559b56d3023ff50a69e",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15411",
"triggerID" : "78a15e131e1f0da182410559b56d3023ff50a69e",
"triggerType" : "PUSH"
}, {
"hash" : "ab8ea83cebed3f79d527c8a22b467b7fa9a11160",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "ab8ea83cebed3f79d527c8a22b467b7fa9a11160",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 78a15e131e1f0da182410559b56d3023ff50a69e Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15411)
* ab8ea83cebed3f79d527c8a22b467b7fa9a11160 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on code in PR #19236:
URL: https://github.com/apache/hudi/pull/19236#discussion_r3557383070
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/SparkReaderContextFactory.java:
##########
@@ -89,6 +90,10 @@ public SparkReaderContextFactory(HoodieSparkEngineContext hoodieSparkEngineConte
// Broadcast: Configuration.
Configuration configs = getHadoopConfiguration(jsc.hadoopConfiguration());
schemaEvolutionConfigs.forEach(configs::set);
+ // Internal write-side reads (compaction, clustering, merge) must materialize INLINE blob bytes
+ // so they can be rewritten into the new base file. DESCRIPTOR is a query-only optimization; if
+ // it leaked onto this path the rewrite would persist null blob data and silently drop the bytes.
+ configs.set(HoodieReaderConfig.BLOB_INLINE_READ_MODE.key(), HoodieReaderConfig.BLOB_INLINE_READ_MODE_CONTENT);
Review Comment:
Added the assert on the captured Configuration in testGetSchemaEvolutionConfigurations.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on code in PR #19236:
URL: https://github.com/apache/hudi/pull/19236#discussion_r3557288896
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLanceDataSource.scala:
##########
@@ -1332,15 +1341,17 @@ class TestLanceDataSource extends HoodieSparkClientTestBase {
def asInlineDf(idToBytes: Seq[(Int, Array[Byte])]): DataFrame = {
val rawDf = idToBytes.toDF("id", "bytes")
.select($"id", BlobTestHelpers.inlineBlobStructCol("payload", $"bytes"))
- spark.createDataFrame(rawDf.rdd, canonicalSchema)
+ spark.createDataFrame(rawDf.rdd, canonicalSchema).coalesce(1)
Review Comment:
Good catch, that invariant was load-bearing but never checked. Added the single-file-group assert right before the anyHadLogs scan, with a fresh getAllFileGroups call since the stream is single-use.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19236:
URL: https://github.com/apache/hudi/pull/19236#discussion_r3558242255
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLanceDataSource.scala:
##########
@@ -1443,6 +1481,114 @@ class TestLanceDataSource extends HoodieSparkClientTestBase {
}
}
+ /**
+ * Clustering must preserve INLINE blob bytes under the DESCRIPTOR default. Clustering rewrites
+ * ALL rows through MultipleSparkJobExecutionStrategy.readRecordsForGroupAsRow, which reads base
+ * files through the same internal write-side reader context as compaction
+ * (SparkReaderContextFactory -> SparkFileFormatInternalRowReaderContext -> SparkLanceReaderBase).
+ * Before SparkReaderContextFactory pinned {@code hoodie.read.blob.inline.mode=CONTENT} for
+ * internal reads, the first clustering of a Lance table with INLINE blobs read null {@code data}
+ * (the DESCRIPTOR default) and rewrote every row's blob with null bytes, silently losing all
+ * blob content (#19232). This test bulk-inserts INLINE blobs, triggers inline clustering,
+ * asserts a replacecommit actually completed, and verifies every row's bytes survived.
+ */
+ @Test
+ def testBlobInlineClusteringRoundTrip(): Unit = {
+ val tableType = HoodieTableType.COPY_ON_WRITE
+ val tableName = "test_lance_blob_inline_cluster_cow"
+ val tablePath = s"$basePath/$tableName"
+
+ val payloadLen = 1024
+ val numRows = 5
+ val expectedPayloads: Seq[Array[Byte]] = (0 until numRows).map { i =>
+ (0 until payloadLen).map(j => ((i + j) % 256).toByte).toArray
+ }
+ val sparkSess = spark
+ import sparkSess.implicits._
+
+ val canonicalSchema = StructType(Seq(
+ StructField("id", IntegerType, nullable = false),
+ StructField("payload", BlobType().asInstanceOf[StructType], nullable = true,
+ BlobTestHelpers.blobMetadata)
+ ))
+ def asInlineDf(idToBytes: Seq[(Int, Array[Byte])]): DataFrame = {
+ val rawDf = idToBytes.toDF("id", "bytes")
+ .select($"id", BlobTestHelpers.inlineBlobStructCol("payload", $"bytes"))
+ spark.createDataFrame(rawDf.rdd, canonicalSchema)
+ }
+
+ // First commit: bulk_insert ids 0..4 with the initial pattern into a base file.
+ writeDataframe(tableType, tableName, tablePath,
+ asInlineDf(expectedPayloads.zipWithIndex.map { case (b, i) => (i, b) }),
+ saveMode = SaveMode.Overwrite,
+ operation = Some("bulk_insert"),
+ extraOptions = Map(PRECOMBINE_FIELD.key() -> "id"))
+
+ assertLanceBlobEncoding(tablePath)
+
+ // Second commit: a small bulk_insert that trips inline clustering (max.commits=1). Clustering
+ // rewrites the existing base file's rows through readRecordsForGroupAsRow, which must read the
+ // INLINE bytes as CONTENT, not the DESCRIPTOR default, or every rewritten row loses its bytes.
+ val extraPayloads = (numRows until numRows + 2).map { i =>
+ (i, (0 until payloadLen).map(j => ((i + j) % 256).toByte).toArray)
+ }
+ writeDataframe(tableType, tableName, tablePath,
+ asInlineDf(extraPayloads),
+ operation = Some("bulk_insert"),
+ extraOptions = Map(PRECOMBINE_FIELD.key() -> "id",
+ "hoodie.clustering.inline" -> "true",
+ "hoodie.clustering.inline.max.commits" -> "1"))
+
+ // Assert a clustering (replacecommit) instant actually completed, so the test cannot silently
+ // pass without a rewrite.
+ val metaClient = HoodieTableMetaClient.builder()
+ .setConf(HoodieTestUtils.getDefaultStorageConf)
+ .setBasePath(tablePath)
+ .build()
+ assertTrue(metaClient.getActiveTimeline.getLastClusteringInstant.isPresent,
Review Comment:
`getLastClusteringInstant` does not filter by instant state: it reads `getCommitsTimeline()`, and `BaseHoodieTimeline.getTimelineOfActions` filters `getInstantsAsStream()` by action only, so a REQUESTED or INFLIGHT replacecommit satisfies `isPresent`. Nothing here checks completion; `getLastClusteringInstant.get.isCompleted` would.
The second half of the claim needs more. A completed clustering instant still does not prove ids 0..4 went through the rewrite: the first bulk-insert's base file qualifies for the plan only via the `hoodie.clustering.plan.strategy.small.file.limit` and single-group-clustering defaults, and if either drifts the byte assertions below read the untouched original bytes and pass. That is the masking mode of #19232, the same one `testBlobInlineCompactionRoundTrip` now pins with its single-file-group assert. Suggest capturing `fsView.getLatestBaseFiles("")` after the first commit and asserting the post-clustering base files are disjoint from it.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19236:
URL: https://github.com/apache/hudi/pull/19236#issuecomment-4925439577
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5722d39821a9d5717fa2025c1f4ff5b715be5dcc",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15408",
"triggerID" : "5722d39821a9d5717fa2025c1f4ff5b715be5dcc",
"triggerType" : "PUSH"
}, {
"hash" : "78a15e131e1f0da182410559b56d3023ff50a69e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15411",
"triggerID" : "78a15e131e1f0da182410559b56d3023ff50a69e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5722d39821a9d5717fa2025c1f4ff5b715be5dcc Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15408)
* 78a15e131e1f0da182410559b56d3023ff50a69e Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15411)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19236:
URL: https://github.com/apache/hudi/pull/19236#discussion_r3552385471
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/SparkReaderContextFactory.java:
##########
@@ -89,6 +90,10 @@ public SparkReaderContextFactory(HoodieSparkEngineContext hoodieSparkEngineConte
// Broadcast: Configuration.
Configuration configs = getHadoopConfiguration(jsc.hadoopConfiguration());
schemaEvolutionConfigs.forEach(configs::set);
+ // Internal write-side reads (compaction, clustering, merge) must materialize INLINE blob bytes
+ // so they can be rewritten into the new base file. DESCRIPTOR is a query-only optimization; if
+ // it leaked onto this path the rewrite would persist null blob data and silently drop the bytes.
+ configs.set(HoodieReaderConfig.BLOB_INLINE_READ_MODE.key(), HoodieReaderConfig.BLOB_INLINE_READ_MODE_CONTENT);
Review Comment:
`TestSparkReaderContextFactory#testGetSchemaEvolutionConfigurations` already captures this exact `Configuration` and asserts a handful of keys on it, so a one-line `assertEquals(BLOB_INLINE_READ_MODE_CONTENT, createdConfig.get(BLOB_INLINE_READ_MODE.key()))` would guard this pin directly. Today only the heavyweight Lance functional tests cover it, and they would not run if someone dropped this line while touching the factory for an unrelated reason.
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkLanceWriter.java:
##########
@@ -446,12 +467,43 @@ protected void updateRecordMetadata(InternalRow row,
@AllArgsConstructor(staticName = "of")
private static class SparkArrowWriter implements ArrowWriter<InternalRow> {
private final LanceArrowWriter lanceArrowWriter;
+ // Parallel arrays of top-level BLOB column ordinals and names for the per-row guard.
+ private final int[] blobFieldOrdinals;
+ private final String[] blobFieldNames;
@Override
public void write(InternalRow row) {
+ validateBlobRow(row);
lanceArrowWriter.write(row);
}
+ /**
+ * Reject descriptor-shaped BLOB rows before they are persisted. A row is descriptor-shaped when
+ * a top-level BLOB struct is non-null, its type is INLINE, its data is null, and its reference
+ * is non-null; this only arises when a DESCRIPTOR-mode read leaks onto a write path, and writing
+ * it would silently drop the inline bytes. The legitimate empty-inline shape {INLINE, null, null}
+ * stays writable.
+ */
+ private void validateBlobRow(InternalRow row) {
+ for (int i = 0; i < blobFieldOrdinals.length; i++) {
+ int ordinal = blobFieldOrdinals[i];
+ if (row.isNullAt(ordinal)) {
+ continue;
+ }
+ InternalRow blob = row.getStruct(ordinal, 3);
Review Comment:
`BlobDescriptorTransform`, in this package, already names these: `TYPE_IDX`/`DATA_IDX`/`REF_IDX` for the child ordinals, and `INLINE_UTF8` for the precomputed token that `INLINE_TYPE_TOKEN` duplicates. The struct arity is available as `HoodieSchema.Blob.getFieldCount()`, which `BatchedBlobReader` already uses for exactly this.
Suggest reusing those (or hoisting them into a small shared blob-layout holder) rather than re-deriving `0`/`1`/`2`/`3` and a second INLINE token, so the two classes that decode this struct cannot drift apart.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/lance/SparkLanceReaderBase.scala:
##########
@@ -131,10 +131,10 @@ class SparkLanceReaderBase(enableVectorizedReader: Boolean) extends SparkColumna
null
}
- // Honor `hoodie.read.blob.inline.mode`. CONTENT (default) materializes INLINE bytes in
- // the `data` column; DESCRIPTOR surfaces per-row {position, size} which the descriptor
- // iterator rewrites into Hudi OUT_OF_LINE references. Non-blob Lance columns ignore
- // the option regardless.
+ // Honor `hoodie.read.blob.inline.mode`. DESCRIPTOR (default) surfaces per-row
+ // {position, size} which the descriptor iterator rewrites into Hudi OUT_OF_LINE
Review Comment:
`BlobDescriptorTransform` keeps `type = INLINE` on the rewritten row and only synthesizes the `reference` sub-struct pointing at the current `.lance` file; it never emits `OUT_OF_LINE`. Wording it as "rewrites into Hudi OUT_OF_LINE references" is the mental model that made #19232 hard to spot, since the row that actually reaches the writer is `{INLINE, data=null, reference}`.
Suggest "...which the descriptor iterator turns into a synthesized `reference` while leaving `type = INLINE`".
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19236:
URL: https://github.com/apache/hudi/pull/19236#issuecomment-4925689562
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5722d39821a9d5717fa2025c1f4ff5b715be5dcc",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15408",
"triggerID" : "5722d39821a9d5717fa2025c1f4ff5b715be5dcc",
"triggerType" : "PUSH"
}, {
"hash" : "78a15e131e1f0da182410559b56d3023ff50a69e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15411",
"triggerID" : "78a15e131e1f0da182410559b56d3023ff50a69e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5722d39821a9d5717fa2025c1f4ff5b715be5dcc Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15408)
* 78a15e131e1f0da182410559b56d3023ff50a69e Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15411)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(spark): read INLINE blobs as CONTENT on internal write-side Lance… [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19236:
URL: https://github.com/apache/hudi/pull/19236#discussion_r3552349159
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkLanceWriter.java:
##########
@@ -446,12 +467,43 @@ protected void updateRecordMetadata(InternalRow row,
@AllArgsConstructor(staticName = "of")
private static class SparkArrowWriter implements ArrowWriter<InternalRow> {
private final LanceArrowWriter lanceArrowWriter;
+ // Parallel arrays of top-level BLOB column ordinals and names for the per-row guard.
+ private final int[] blobFieldOrdinals;
+ private final String[] blobFieldNames;
@Override
public void write(InternalRow row) {
+ validateBlobRow(row);
lanceArrowWriter.write(row);
}
+ /**
+ * Reject descriptor-shaped BLOB rows before they are persisted. A row is descriptor-shaped when
+ * a top-level BLOB struct is non-null, its type is INLINE, its data is null, and its reference
+ * is non-null; this only arises when a DESCRIPTOR-mode read leaks onto a write path, and writing
+ * it would silently drop the inline bytes. The legitimate empty-inline shape {INLINE, null, null}
+ * stays writable.
+ */
+ private void validateBlobRow(InternalRow row) {
+ for (int i = 0; i < blobFieldOrdinals.length; i++) {
+ int ordinal = blobFieldOrdinals[i];
+ if (row.isNullAt(ordinal)) {
+ continue;
+ }
+ InternalRow blob = row.getStruct(ordinal, 3);
+ if (blob.isNullAt(0) || !INLINE_TYPE_TOKEN.equals(blob.getUTF8String(0))) {
+ continue;
+ }
+ if (blob.isNullAt(1) && !blob.isNullAt(2)) {
+ throw new HoodieException(
+ "BLOB column '" + blobFieldNames[i] + "' has an INLINE row with null data but a "
+ + "populated reference: a DESCRIPTOR-mode read leaked into the write path, and "
+ + "persisting it would silently drop the blob bytes. Internal rewrites "
+ + "(compaction, clustering, merge) must read with hoodie.read.blob.inline.mode=CONTENT.");
Review Comment:
After the pin, no internal rewrite can produce this shape: `SparkLanceReaderBase` installs `BlobDescriptorTransform` only when the resolved mode is DESCRIPTOR, and every read through `SparkReaderContextFactory` now resolves CONTENT. The realistic remaining trigger is a user round-trip - read a Lance blob table at the DESCRIPTOR default and write those rows into a Lance table (`INSERT INTO t2 SELECT * FROM t1`; `alignFieldsNullability` preserves the BLOB metadata, so the guard does see the column). The message names the right config key, but attributes the leak to "internal rewrites", which that user is not running.
Suggest widening the parenthetical to cover the source read, e.g. "compaction, clustering, merge - and any query whose rows are written back - must read with hoodie.read.blob.inline.mode=CONTENT". Worth noting under Impact too: this converts a previously silent (but lossy) `INSERT INTO ... SELECT` over a blob table into a hard failure.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLanceDataSource.scala:
##########
@@ -1303,12 +1303,21 @@ class TestLanceDataSource extends HoodieSparkClientTestBase {
/**
* Compaction must preserve INLINE blob bytes under the DESCRIPTOR default. MOR compaction reads
- * the base file via {@link HoodieSparkLanceReader}, which hard-pins CONTENT regardless of the
- * user-facing {@code hoodie.read.blob.inline.mode}. If that pin were to honor the default
- * (DESCRIPTOR), compaction would read null {@code data} and rewrite a base file without bytes,
- * silently corrupting untouched rows. This test inserts INLINE blobs, upserts a subset to force
- * compaction, and asserts that touched rows carry the new bytes while untouched rows retain the
- * originals.
+ * the base file through the internal write-side reader stack
+ * SparkReaderContextFactory -> SparkFileFormatInternalRowReaderContext -> SparkLanceReaderBase,
+ * not through HoodieSparkLanceReader (that reader only serves LanceUtils stats/key reads).
Review Comment:
The `deltaCommits` assertion further down this method still says the round-trip "would silently pass even if the CONTENT-pin in HoodieSparkLanceReader were broken", which is the same misattribution this comment corrects and which the second commit's message claims to have dropped. Worth updating that message too.
The parenthetical is also narrower than reality: `HoodieSparkLanceReader` is opened by `HoodieReadHandle#createNewFileReader` for bloom-index lookups and by `HoodieMergeHelper#runMerge` for the old base file whenever `hoodie.write.merge.handle.class` is the legacy `HoodieWriteMergeHandle` instead of the default `FileGroupReaderBasedMergeHandle`, so its CONTENT pin is still load-bearing. The comment on that pin inside `HoodieSparkLanceReader` still credits it for "compaction/merge/log-replay" and now contradicts this one.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLanceDataSource.scala:
##########
@@ -1332,15 +1341,17 @@ class TestLanceDataSource extends HoodieSparkClientTestBase {
def asInlineDf(idToBytes: Seq[(Int, Array[Byte])]): DataFrame = {
val rawDf = idToBytes.toDF("id", "bytes")
.select($"id", BlobTestHelpers.inlineBlobStructCol("payload", $"bytes"))
- spark.createDataFrame(rawDf.rdd, canonicalSchema)
+ spark.createDataFrame(rawDf.rdd, canonicalSchema).coalesce(1)
Review Comment:
The un-masking now rests on `coalesce(1)` plus `hoodie.bulkinsert.shuffle.parallelism=1` yielding exactly one file group, but nothing asserts that. The `anyHadLogs` check below only requires that some file slice carried log files, which stays true even if the untouched ids drift back into a log-free file group - and then compaction never rewrites them, the regression is not exercised, and the test passes. That is how #19232 stayed hidden in the first place.
Suggest pinning the invariant directly, e.g. `assertEquals(1, fsView.getAllFileGroups("").count())` placed before the `anyHadLogs` scan, since the returned Stream is single-use.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org