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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
