This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 120d73fcc4db fix(spark): read INLINE blobs as CONTENT on internal
write-side Lance… (#19236)
120d73fcc4db is described below
commit 120d73fcc4dba7ddfae0b9693f487190a8410d34
Author: voonhous <[email protected]>
AuthorDate: Sat Jul 11 13:21:09 2026 +0800
fix(spark): read INLINE blobs as CONTENT on internal write-side Lance…
(#19236)
* fix(spark): read INLINE blobs as CONTENT on internal write-side Lance
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
* test(spark): assert INLINE blob bytes via plain projection before
read_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
#19232 look like an error by design. Also drop a stale comment crediting
HoodieSparkLanceReader's CONTENT pin for compaction correctness.
* test(spark): assert single file group in blob compaction round-trip test
The un-masking of #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.
* docs(spark): fix DESCRIPTOR-mode comment; transform keeps type=INLINE
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 #19232.
* test(spark): guard CONTENT pin in TestSparkReaderContextFactory
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.
* refactor(spark): hoist blob struct layout into shared BlobStructLayout
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.
* docs(spark): correct stale CONTENT-pin attributions for Lance readers
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.
* fix(spark): widen blob guard error to cover user query round-trips
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.
* test(spark): assert clustering completed and rewrote base files in blob
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 #19232).
* fix(test): define missing engineContext in latestBaseFileNames helper
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.
* docs(test): reword CONTENT-projection comments to the shape they actually
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.
* test(spark): cover CoW upsert merge path for INLINE blob preservation
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.
---
.../client/common/SparkReaderContextFactory.java | 5 +
.../hudi/io/storage/BlobDescriptorTransform.java | 24 +-
.../apache/hudi/io/storage/BlobStructLayout.java | 46 +++
.../hudi/io/storage/HoodieSparkLanceReader.java | 6 +-
.../hudi/io/storage/HoodieSparkLanceWriter.java | 58 +++-
.../common/TestSparkReaderContextFactory.java | 6 +
.../datasources/lance/SparkLanceReaderBase.scala | 8 +-
.../io/storage/TestHoodieSparkLanceWriter.java | 92 +++++
.../hudi/functional/TestLanceDataSource.scala | 379 ++++++++++++++++++++-
9 files changed, 589 insertions(+), 35 deletions(-)
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/SparkReaderContextFactory.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/SparkReaderContextFactory.java
index bb54f9cc1fc0..6229cde9fcca 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/SparkReaderContextFactory.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/SparkReaderContextFactory.java
@@ -23,6 +23,7 @@ import org.apache.hudi.HoodieSparkUtils;
import org.apache.hudi.SparkAdapterSupport$;
import org.apache.hudi.SparkFileFormatInternalRowReaderContext;
import org.apache.hudi.client.utils.SparkInternalSchemaConverter;
+import org.apache.hudi.common.config.HoodieReaderConfig;
import org.apache.hudi.common.engine.HoodieReaderContext;
import org.apache.hudi.common.engine.ReaderContextFactory;
import org.apache.hudi.common.model.HoodieFileFormat;
@@ -89,6 +90,10 @@ public class SparkReaderContextFactory implements
ReaderContextFactory<InternalR
// 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);
configs.set(SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE().key(),
sqlConf.getConfString(SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE().key()));
configs.set(SQLConf.PARQUET_WRITE_LEGACY_FORMAT().key(),
sqlConf.getConfString(SQLConf.PARQUET_WRITE_LEGACY_FORMAT().key()));
configurationBroadcast = jsc.broadcast(new
SerializableConfiguration(configs));
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/BlobDescriptorTransform.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/BlobDescriptorTransform.java
index b893811cf28e..1032c5d36806 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/BlobDescriptorTransform.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/BlobDescriptorTransform.java
@@ -18,7 +18,6 @@
package org.apache.hudi.io.storage;
-import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.exception.HoodieException;
import org.apache.spark.sql.catalyst.InternalRow;
@@ -35,6 +34,14 @@ import org.lance.spark.vectorized.LanceArrowColumnVector;
import java.util.HashSet;
import java.util.Set;
+import static org.apache.hudi.io.storage.BlobStructLayout.DATA_IDX;
+import static org.apache.hudi.io.storage.BlobStructLayout.FIELD_COUNT;
+import static org.apache.hudi.io.storage.BlobStructLayout.INLINE_UTF8;
+import static org.apache.hudi.io.storage.BlobStructLayout.OUT_OF_LINE_UTF8;
+import static org.apache.hudi.io.storage.BlobStructLayout.REF_FIELD_COUNT;
+import static org.apache.hudi.io.storage.BlobStructLayout.REF_IDX;
+import static org.apache.hudi.io.storage.BlobStructLayout.TYPE_IDX;
+
/**
* Per-row transform that rewrites BLOB columns from Lance's DESCRIPTOR shape
into the Hudi BLOB
* shape. Composed into {@link LanceRecordIterator} for DESCRIPTOR-mode reads.
@@ -52,16 +59,6 @@ import java.util.Set;
*/
public final class BlobDescriptorTransform {
- private static final UTF8String OUT_OF_LINE_UTF8 =
- UTF8String.fromString(HoodieSchema.Blob.OUT_OF_LINE);
- private static final UTF8String INLINE_UTF8 =
- UTF8String.fromString(HoodieSchema.Blob.INLINE);
-
- // Child field indices within the Hudi BLOB struct: {type(0), data(1),
reference(2)}.
- private static final int TYPE_IDX = 0;
- private static final int DATA_IDX = 1;
- private static final int REF_IDX = 2;
-
private final Set<String> blobFieldNames;
private final UTF8String lanceFilePathUtf8;
private final String lanceFilePath;
@@ -105,7 +102,7 @@ public final class BlobDescriptorTransform {
for (int i = 0; i < outputFields.length; i++) {
if (blobColumnIndices.contains(i)) {
rowBuffer[i] = row.isNullAt(i) ? null
- : buildBlobOutputRow(row.getStruct(i, 3), columnVectors[i], rowId);
+ : buildBlobOutputRow(row.getStruct(i, FIELD_COUNT),
columnVectors[i], rowId);
} else {
rowBuffer[i] = row.isNullAt(i) ? null : row.get(i,
outputFields[i].dataType());
}
@@ -131,7 +128,8 @@ public final class BlobDescriptorTransform {
UTF8String type = blobStruct.getUTF8String(TYPE_IDX);
if (type.equals(OUT_OF_LINE_UTF8)) {
- InternalRow refRow = blobStruct.isNullAt(REF_IDX) ? null :
blobStruct.getStruct(REF_IDX, 4);
+ InternalRow refRow = blobStruct.isNullAt(REF_IDX) ? null
+ : blobStruct.getStruct(REF_IDX, REF_FIELD_COUNT);
return new GenericInternalRow(new Object[] { OUT_OF_LINE_UTF8, null,
refRow });
}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/BlobStructLayout.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/BlobStructLayout.java
new file mode 100644
index 000000000000..e49e68e4b9c9
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/BlobStructLayout.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io.storage;
+
+import org.apache.hudi.common.schema.HoodieSchema;
+
+import org.apache.spark.unsafe.types.UTF8String;
+
+/**
+ * Layout of the Hudi BLOB struct {@code {type, data, reference}} as decoded
by the Spark-side
+ * Lance code. Shared by {@link BlobDescriptorTransform} (read) and {@link
HoodieSparkLanceWriter}
+ * (write) so the two cannot drift apart. Ordinals mirror the field order
defined by
+ * {@link HoodieSchema.Blob}.
+ */
+final class BlobStructLayout {
+
+ // Child field indices within the Hudi BLOB struct: {type(0), data(1),
reference(2)}.
+ static final int TYPE_IDX = 0;
+ static final int DATA_IDX = 1;
+ static final int REF_IDX = 2;
+ static final int FIELD_COUNT = HoodieSchema.Blob.getFieldCount();
+ static final int REF_FIELD_COUNT =
HoodieSchema.Blob.getReferenceFieldCount();
+
+ // Precomputed type tokens, compared against each row's type field without
per-row allocation.
+ static final UTF8String INLINE_UTF8 =
UTF8String.fromString(HoodieSchema.Blob.INLINE);
+ static final UTF8String OUT_OF_LINE_UTF8 =
UTF8String.fromString(HoodieSchema.Blob.OUT_OF_LINE);
+
+ private BlobStructLayout() {
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkLanceReader.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkLanceReader.java
index 436c2e69fef0..f3ea4620503e 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkLanceReader.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkLanceReader.java
@@ -205,8 +205,10 @@ public class HoodieSparkLanceReader implements
HoodieSparkFileReader {
columnNames.add(field.name());
}
- // Pinned to CONTENT: compaction/merge/log-replay need actual bytes to
rewrite.
- // The user-facing `hoodie.read.blob.inline.mode` is honored by
SparkLanceReaderBase.
+ // Pinned to CONTENT: callers (LanceUtils stats/key reads, bloom-index
lookups via
+ // HoodieReadHandle, old-base-file reads in the legacy
HoodieWriteMergeHandle merge path)
+ // need actual bytes. Default-path compaction/merge reads go through
SparkLanceReaderBase
+ // instead, which honors the user-facing `hoodie.read.blob.inline.mode`.
FileReadOptions readOpts =
FileReadOptions.builder().blobReadMode(BlobReadMode.CONTENT).build();
// BLOB reads must be chunked to dodge a lance-core FFI abort (see
LanceRecordIterator).
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkLanceWriter.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkLanceWriter.java
index cf8601dfc9ae..a94015744817 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkLanceWriter.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkLanceWriter.java
@@ -27,6 +27,7 @@ import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaField;
import org.apache.hudi.common.schema.HoodieSchemaType;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieNotSupportedException;
import org.apache.hudi.io.lance.HoodieBaseLanceWriter;
import org.apache.hudi.io.storage.row.HoodieBloomFilterRowWriteSupport;
@@ -67,6 +68,11 @@ import static
org.apache.hudi.common.model.HoodieRecord.HoodieMetadataField.FILE
import static
org.apache.hudi.common.model.HoodieRecord.HoodieMetadataField.PARTITION_PATH_METADATA_FIELD;
import static
org.apache.hudi.common.model.HoodieRecord.HoodieMetadataField.RECORD_KEY_METADATA_FIELD;
import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
+import static org.apache.hudi.io.storage.BlobStructLayout.DATA_IDX;
+import static org.apache.hudi.io.storage.BlobStructLayout.FIELD_COUNT;
+import static org.apache.hudi.io.storage.BlobStructLayout.INLINE_UTF8;
+import static org.apache.hudi.io.storage.BlobStructLayout.REF_IDX;
+import static org.apache.hudi.io.storage.BlobStructLayout.TYPE_IDX;
/**
* Spark Lance file writer implementing {@link HoodieSparkFileWriter} and
{@link HoodieInternalRowFileWriter}.
@@ -96,6 +102,10 @@ public class HoodieSparkLanceWriter extends
HoodieBaseLanceWriter<InternalRow, U
private final boolean populateMetaFields;
private final Function<Long, String> seqIdGenerator;
private final long maxFileSize;
+ // Top-level BLOB column ordinals and their names (parallel arrays), used by
the per-row
+ // descriptor-leak guard. Empty when the schema has no BLOB columns, making
the guard a no-op.
+ private final int[] blobFieldOrdinals;
+ private final String[] blobFieldNames;
private long recordCountForNextSizeCheck = MIN_RECORDS_FOR_SIZE_CHECK;
/**
@@ -159,6 +169,20 @@ public class HoodieSparkLanceWriter extends
HoodieBaseLanceWriter<InternalRow, U
super(file, DEFAULT_BATCH_SIZE, allocatorSize, flushByteWatermark,
bloomFilterOpt.map(HoodieBloomFilterRowWriteSupport::new));
this.sparkSchema = enrichSparkSchemaForLance(sparkSchema);
+ StructField[] topFields = this.sparkSchema.fields();
+ List<Integer> blobOrds = new ArrayList<>();
+ for (int i = 0; i < topFields.length; i++) {
+ if (isBlobField(topFields[i])) {
+ blobOrds.add(i);
+ }
+ }
+ this.blobFieldOrdinals = new int[blobOrds.size()];
+ this.blobFieldNames = new String[blobOrds.size()];
+ for (int i = 0; i < blobOrds.size(); i++) {
+ int ord = blobOrds.get(i);
+ this.blobFieldOrdinals[i] = ord;
+ this.blobFieldNames[i] = topFields[ord].name();
+ }
Schema baseArrow = LanceArrowUtils.toArrowSchema(this.sparkSchema,
DEFAULT_TIMEZONE, true);
// Force LargeBinary + `lance-encoding:blob=true` on each BLOB's nested
`data` Arrow leaf.
// Can't be expressed Spark-side: toArrowSchema drops nested-field
metadata, and tagging
@@ -367,7 +391,7 @@ public class HoodieSparkLanceWriter extends
HoodieBaseLanceWriter<InternalRow, U
@Override
protected ArrowWriter<InternalRow> createArrowWriter(VectorSchemaRoot root) {
- return SparkArrowWriter.of(LanceArrowWriter.create(root, sparkSchema));
+ return SparkArrowWriter.of(LanceArrowWriter.create(root, sparkSchema),
blobFieldOrdinals, blobFieldNames);
}
/**
@@ -446,12 +470,44 @@ public class HoodieSparkLanceWriter extends
HoodieBaseLanceWriter<InternalRow, U
@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, FIELD_COUNT);
+ if (blob.isNullAt(TYPE_IDX) ||
!INLINE_UTF8.equals(blob.getUTF8String(TYPE_IDX))) {
+ continue;
+ }
+ if (blob.isNullAt(DATA_IDX) && !blob.isNullAt(REF_IDX)) {
+ 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) and any query whose rows
are written back "
+ + "must read with hoodie.read.blob.inline.mode=CONTENT.");
+ }
+ }
+ }
+
@Override
public void reset() {
lanceArrowWriter.reset();
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/common/TestSparkReaderContextFactory.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/common/TestSparkReaderContextFactory.java
index 5c297846181c..80dc618c4992 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/common/TestSparkReaderContextFactory.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/common/TestSparkReaderContextFactory.java
@@ -20,6 +20,7 @@ package org.apache.hudi.client.common;
import org.apache.hudi.HoodieSparkUtils;
import org.apache.hudi.client.utils.SparkInternalSchemaConverter;
+import org.apache.hudi.common.config.HoodieReaderConfig;
import org.apache.hudi.common.engine.HoodieReaderContext;
import org.apache.hudi.common.model.ActionType;
import org.apache.hudi.common.schema.internal.InternalSchema;
@@ -113,6 +114,11 @@ class TestSparkReaderContextFactory extends
HoodieClientTestBase {
String inlineClassName = createdConfig.get("fs." + InLineFileSystem.SCHEME
+ ".impl");
assertEquals(InLineFileSystem.class.getName(), inlineClassName);
+ // Internal write-side reads must pin CONTENT; a DESCRIPTOR leak here
drops blob bytes (#19232).
+ assertEquals(
+ HoodieReaderConfig.BLOB_INLINE_READ_MODE_CONTENT,
+ createdConfig.get(HoodieReaderConfig.BLOB_INLINE_READ_MODE.key()));
+
assertEquals(
"0001_0005.deltacommit,0002_0006.deltacommit,0003_0007.commit",
createdConfig.get(SparkInternalSchemaConverter.HOODIE_VALID_COMMITS_LIST));
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/lance/SparkLanceReaderBase.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/lance/SparkLanceReaderBase.scala
index 6f9f7783c04c..d709a2085b10 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/lance/SparkLanceReaderBase.scala
+++
b/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 turns into a
synthesized `reference`
+ // while leaving `type = INLINE`; CONTENT is the opt-in mode that
materializes INLINE
+ // bytes in the `data` column. Non-blob Lance columns ignore the
option regardless.
val blobMode = resolveBlobReadMode(storageConf)
val readOpts = FileReadOptions.builder().blobReadMode(blobMode).build()
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/storage/TestHoodieSparkLanceWriter.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/storage/TestHoodieSparkLanceWriter.java
index 68328d158e10..7fb501c7ee31 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/storage/TestHoodieSparkLanceWriter.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/storage/TestHoodieSparkLanceWriter.java
@@ -30,6 +30,7 @@ import org.apache.hudi.common.schema.HoodieSchemaType;
import org.apache.hudi.common.testutils.HoodieTestUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.core.io.storage.HoodieIOFactory;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieNotSupportedException;
import org.apache.hudi.exception.MetadataNotFoundException;
import org.apache.hudi.io.memory.HoodieArrowAllocator;
@@ -696,4 +697,95 @@ public class TestHoodieSparkLanceWriter {
() -> HoodieSparkLanceWriter.validateNoVariantColumns(record));
assertTrue(ex.getMessage().contains("payload"), "Error should name the
field: " + ex.getMessage());
}
+
+ // ----- INLINE blob descriptor-leak guard tests -----
+
+ /**
+ * Builds a two-column schema (id INT + a canonical BLOB column) so the
writer recognizes the
+ * second column as a blob via the {@code hudi_type=BLOB} field metadata.
The struct layout is
+ * the canonical one produced by {@link
org.apache.spark.sql.types.BlobType}: type, data,
+ * reference.
+ */
+ private StructType createBlobSchema() {
+ StructType blobStruct = (StructType)
org.apache.spark.sql.types.BlobType.dataType();
+ Metadata blobMetadata = new MetadataBuilder()
+ .putString(HoodieSchema.TYPE_METADATA_FIELD,
HoodieSchemaType.BLOB.name())
+ .build();
+ return new StructType()
+ .add(new StructField("id", DataTypes.IntegerType, false,
Metadata.empty()))
+ .add(new StructField("payload", blobStruct, true, blobMetadata));
+ }
+
+ /**
+ * Builds the reference sub-struct {external_path, offset, length, managed}.
+ */
+ private InternalRow blobReference(String externalPath, Long offset, Long
length, boolean managed) {
+ return new GenericInternalRow(new Object[] {
+ externalPath == null ? null : UTF8String.fromString(externalPath),
+ offset,
+ length,
+ managed
+ });
+ }
+
+ /**
+ * Writing an INLINE blob with null {@code data} but a non-null {@code
reference} is the
+ * descriptor-leak shape: it means an internal write-side read handed the
writer a DESCRIPTOR row
+ * (reference populated, bytes dropped) instead of CONTENT. The writer must
reject it with a
+ * {@link HoodieException} that points at {@code
hoodie.read.blob.inline.mode}, rather than
+ * silently persisting a blob with no bytes (#19232).
+ */
+ @Test
+ public void testWriteInlineBlobWithNullDataAndReference_throws() {
+ StructType schema = createBlobSchema();
+ StoragePath path = new StoragePath(tempDir.getAbsolutePath() +
"/test_blob_descriptor_leak.lance");
+
+ InternalRow reference = blobReference("s3://bucket/blob.bin", 0L, 1024L,
false);
+ InternalRow blob = new GenericInternalRow(new Object[] {
+ UTF8String.fromString(HoodieSchema.Blob.INLINE), // type = INLINE
+ null, // data = null
(leaked)
+ reference // reference =
non-null (leaked)
+ });
+ InternalRow row = new GenericInternalRow(new Object[] {1, blob});
+
+ HoodieException ex = assertThrows(HoodieException.class, () -> {
+ try (HoodieSparkLanceWriter writer = HoodieSparkLanceWriter.builder()
+
.file(path).sparkSchema(schema).instantTime(instantTime).taskContextSupplier(taskContextSupplier)
+
.storage(storage).bloomFilterOpt(Option.of(simpleBloomFilter)).build()) {
+ writer.writeRow("key1", row);
+ }
+ });
+ assertTrue(ex.getMessage() != null &&
ex.getMessage().contains("hoodie.read.blob.inline.mode"),
+ "Descriptor-leak guard must reference hoodie.read.blob.inline.mode: "
+ ex.getMessage());
+ }
+
+ /**
+ * An INLINE blob with null {@code data} AND null {@code reference} is a
legitimate null inline
+ * payload, not a descriptor leak. The writer must accept it and close
cleanly.
+ */
+ @Test
+ public void testWriteInlineBlobWithNullDataAndNullReference_succeeds()
throws Exception {
+ StructType schema = createBlobSchema();
+ StoragePath path = new StoragePath(tempDir.getAbsolutePath() +
"/test_blob_null_inline.lance");
+
+ InternalRow blob = new GenericInternalRow(new Object[] {
+ UTF8String.fromString(HoodieSchema.Blob.INLINE), // type = INLINE
+ null, // data = null (legit
null payload)
+ null // reference = null
+ });
+ InternalRow row = new GenericInternalRow(new Object[] {1, blob});
+
+ try (HoodieSparkLanceWriter writer = HoodieSparkLanceWriter.builder()
+
.file(path).sparkSchema(schema).instantTime(instantTime).taskContextSupplier(taskContextSupplier)
+
.storage(storage).bloomFilterOpt(Option.of(simpleBloomFilter)).build()) {
+ writer.writeRow("key1", row);
+ }
+
+ assertTrue(storage.exists(path), "Lance file with a null INLINE payload
should be written");
+ try (BufferAllocator allocator = HoodieArrowAllocator.newChildAllocator(
+ "testWriteInlineBlobWithNullDataAndNullReference",
TEST_LANCE_DATA_ALLOCATOR_SIZE);
+ LanceFileReader reader = LanceFileReader.open(path.toString(),
allocator)) {
+ assertEquals(1, reader.numRows(), "The single null-payload row should be
written");
+ }
+ }
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLanceDataSource.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLanceDataSource.scala
index 419a7bcf6498..b281b83bf6c5 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLanceDataSource.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLanceDataSource.scala
@@ -1303,12 +1303,22 @@ 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 serves LanceUtils
stats/key reads,
+ * bloom-index key lookups, and legacy HoodieWriteMergeHandle merges, with
its own CONTENT pin).
+ * SparkLanceReaderBase honors {@code hoodie.read.blob.inline.mode}, whose
DESCRIPTOR default
+ * would read null {@code data} and rewrite a base file without bytes,
silently corrupting
+ * untouched rows. Correctness now relies on SparkReaderContextFactory
pinning
+ * {@code hoodie.read.blob.inline.mode=CONTENT} in the broadcast conf used
by all internal reads.
+ * User-facing queries are unaffected because they build their own conf.
+ *
+ * The test forces all rows into a single file group (coalesce(1) plus
bulk-insert/insert
+ * shuffle parallelism 1) so the untouched rows genuinely go through the
compaction rewrite.
+ * Without that, untouched rows land in log-free file groups that compaction
never rewrites,
+ * which is how the original bug (#19232) stayed masked. 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.
*/
@Test
def testBlobInlineCompactionRoundTrip(): Unit = {
@@ -1332,7 +1342,7 @@ 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)
}
// First commit: bulk_insert ids 0..5 with the initial pattern. Lands in a
base file.
@@ -1340,7 +1350,9 @@ class TestLanceDataSource extends
HoodieSparkClientTestBase {
asInlineDf(initialPayloads.zipWithIndex.map { case (b, i) => (i, b) }),
saveMode = SaveMode.Overwrite,
operation = Some("bulk_insert"),
- extraOptions = Map(PRECOMBINE_FIELD.key() -> "id"))
+ extraOptions = Map(PRECOMBINE_FIELD.key() -> "id",
+ "hoodie.bulkinsert.shuffle.parallelism" -> "1",
+ "hoodie.insert.shuffle.parallelism" -> "1"))
assertLanceBlobEncoding(tablePath)
@@ -1366,9 +1378,9 @@ class TestLanceDataSource extends
HoodieSparkClientTestBase {
.getInstants.asScala
val deltaCommits = completedInstants.filter(_.getAction == "deltacommit")
assertTrue(deltaCommits.nonEmpty,
- "Upsert must have written a deltacommit on MOR — without log files the
compaction " +
+ "Upsert must have written a deltacommit on MOR -- without log files the
compaction " +
"round-trip below would be a no-op and the test would silently pass
even if the " +
- "CONTENT-pin in HoodieSparkLanceReader were broken.")
+ "CONTENT-pin in SparkReaderContextFactory were broken.")
val compactionCommits = completedInstants.filter(_.getAction == "commit")
assertTrue(compactionCommits.nonEmpty, "Compaction commit should be
present after upsert")
@@ -1386,6 +1398,13 @@ class TestLanceDataSource extends
HoodieSparkClientTestBase {
val fsView = viewManager.getFileSystemView(metaClient)
try {
fsView.loadAllPartitions()
+ // Pin the single-file-group invariant the whole test rests on. If rows
ever spread across
+ // multiple file groups, the untouched ids could sit in log-free groups
that compaction never
+ // rewrites, and the round-trip below would pass without exercising the
regression (#19232).
+ assertEquals(1L, fsView.getAllFileGroups("").count(),
+ s"All rows must land in exactly one file group (coalesce(1) + shuffle
parallelism 1) " +
+ s"at $tablePath; otherwise untouched ids may sit in log-free file
groups that " +
+ s"compaction never rewrites and the regression is not exercised")
val anyHadLogs = fsView.getAllFileGroups("").iterator().asScala.exists {
fg =>
fg.getAllFileSlices.iterator().asScala.exists(_.hasLogFiles)
}
@@ -1422,11 +1441,34 @@ 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.
+ // Byte check via a plain projection under CONTENT. A broken rewrite could
produce two shapes:
+ // - {INLINE, null data, populated reference}: a DESCRIPTOR-mode read
leaked into the write
+ // path. HoodieSparkLanceWriter.validateBlobRow rejects this shape and
fails the compaction
+ // itself, so it can never reach the base file.
+ // - {INLINE, null data, null reference}: legitimate for an empty inline
blob, so the writer
+ // guard lets it through. If a rewrite dropped the bytes this way, only
the null-data
+ // assertion below would catch it.
+ // The CONTENT pin on internal reads is unit-tested in
TestSparkReaderContextFactory.
+ val contentRows = spark.read.format("hudi")
+ .option("hoodie.read.blob.inline.mode", "CONTENT")
+ .load(tablePath)
+ .select($"id", $"payload")
+ .orderBy($"id")
+ .collect()
+ assertEquals(numRows, contentRows.length)
+ contentRows.foreach { row =>
+ val id = row.getInt(row.fieldIndex("id"))
+ val payload = row.getStruct(row.fieldIndex("payload"))
+
assertFalse(payload.isNullAt(payload.fieldIndex(HoodieSchema.Blob.INLINE_DATA_FIELD)),
+ s"null data under CONTENT: the compaction rewrite dropped the bytes
(id=$id)")
+ assertArrayEquals(expected(id),
+
payload.getAs[Array[Byte]](payload.fieldIndex(HoodieSchema.Blob.INLINE_DATA_FIELD)),
+ s"INLINE data bytes must survive the compaction rewrite (id=$id)")
+ }
+
+ // read_blob() under CONTENT verifies the same bytes through the SQL
expression path
+ // (read_blob() on INLINE rows throws under the DESCRIPTOR default, so
CONTENT is
+ // required here).
val viewName = s"${tableName}_view"
spark.read.format("hudi")
.option("hoodie.read.blob.inline.mode", "CONTENT")
@@ -1443,6 +1485,313 @@ 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)
+
+ // Snapshot the first commit's base file(s); the disjointness check below
uses them to prove
+ // clustering rewrote (not skipped) them, else the byte checks pass on
stale bytes (#19232).
+ val metaClientAfterFirst = HoodieTableMetaClient.builder()
+ .setConf(HoodieTestUtils.getDefaultStorageConf)
+ .setBasePath(tablePath)
+ .build()
+ val preClusterBaseFiles = latestBaseFileNames(metaClientAfterFirst,
tablePath)
+ assertFalse(preClusterBaseFiles.isEmpty, "First commit should have written
at least one base file")
+
+ // 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"))
+
+ // Require a COMPLETED replacecommit. getLastClusteringInstant filters by
action only, so a
+ // REQUESTED/INFLIGHT instant satisfies isPresent; isCompleted confirms
the rewrite finished.
+ val metaClient = HoodieTableMetaClient.builder()
+ .setConf(HoodieTestUtils.getDefaultStorageConf)
+ .setBasePath(tablePath)
+ .build()
+ val lastClustering = metaClient.getActiveTimeline.getLastClusteringInstant
+ assertTrue(lastClustering.isPresent && lastClustering.get.isCompleted,
+ "A COMPLETED clustering (replacecommit) instant must exist after inline
clustering; without a " +
+ "completed rewrite the blob-loss regression below could not be
exercised")
+
+ // ...and that it rewrote the base file(s) into new ones. Disjoint sets
prove the rewrite ran
+ // instead of the byte checks reading untouched originals (#19232).
+ val postClusterBaseFiles = latestBaseFileNames(metaClient, tablePath)
+ assertTrue(preClusterBaseFiles.intersect(postClusterBaseFiles).isEmpty,
+ s"Post-clustering base files must be disjoint from the pre-clustering
base file(s), proving the " +
+ s"rewrite ran (pre=$preClusterBaseFiles, post=$postClusterBaseFiles)")
+
+ // Read back in CONTENT mode and assert every row's bytes survived the
clustering rewrite.
+ val allExpected: Map[Int, Array[Byte]] =
+ (expectedPayloads.zipWithIndex.map { case (b, i) => i -> b } ++
extraPayloads).toMap
+
+ // Byte check via a plain projection under CONTENT. As in the compaction
test: a DESCRIPTOR
+ // leak already fails the rewrite in
HoodieSparkLanceWriter.validateBlobRow, so what this
+ // catches is the guard-allowed empty-inline shape {INLINE, null data,
null reference},
+ // where dropped bytes would persist silently.
+ val contentRows = spark.read.format("hudi")
+ .option("hoodie.read.blob.inline.mode", "CONTENT")
+ .load(tablePath)
+ .select($"id", $"payload")
+ .orderBy($"id")
+ .collect()
+ assertEquals(allExpected.size, contentRows.length)
+ contentRows.foreach { row =>
+ val id = row.getInt(row.fieldIndex("id"))
+ val payload = row.getStruct(row.fieldIndex("payload"))
+
assertFalse(payload.isNullAt(payload.fieldIndex(HoodieSchema.Blob.INLINE_DATA_FIELD)),
+ s"null data under CONTENT: the clustering rewrite dropped the bytes
(id=$id)")
+ assertArrayEquals(allExpected(id),
+
payload.getAs[Array[Byte]](payload.fieldIndex(HoodieSchema.Blob.INLINE_DATA_FIELD)),
+ s"INLINE data bytes must survive the clustering rewrite (id=$id)")
+ }
+
+ val viewName = s"${tableName}_view"
+ spark.read.format("hudi")
+ .option("hoodie.read.blob.inline.mode", "CONTENT")
+ .load(tablePath)
+ .createOrReplaceTempView(viewName)
+ val materialized = spark.sql(
+ s"SELECT id, read_blob(payload) AS bytes FROM $viewName ORDER BY
id").collect()
+ assertEquals(allExpected.size, materialized.length)
+ materialized.foreach { row =>
+ val id = row.getInt(row.fieldIndex("id"))
+ val bytes = row.getAs[Array[Byte]]("bytes")
+ assertArrayEquals(allExpected(id), bytes,
+ s"read_blob() must return correct bytes post-clustering (id=$id)")
+ }
+ }
+
+ /**
+ * A CoW upsert merge must preserve INLINE blob bytes under the DESCRIPTOR
default.
+ *
+ * Compaction and clustering (covered above) obtain their reader through
+ * {@code HoodieEngineContext.getReaderContextFactory}. A CoW upsert takes a
different path: it
+ * rewrites the base file through FileGroupReaderBasedMergeHandle, which
resolves its reader
+ * through {@code getReaderContextFactoryForWrite}. That method branches on
the record merger
+ * type: AvroReaderContextFactory for AVRO, SparkReaderContextFactory for
SPARK (the datasource
+ * default, used here). A DESCRIPTOR leak on this branch would rewrite
untouched rows with null
+ * {@code data}, the same silent loss as #19232.
+ *
+ * The test bulk-inserts INLINE blobs into a single file group, upserts a
subset, proves the
+ * merge rewrote the base file, and verifies touched rows carry the new
bytes while untouched
+ * rows keep the originals.
+ */
+ @Test
+ def testBlobInlineCowUpsertMergeRoundTrip(): Unit = {
+ val tableType = HoodieTableType.COPY_ON_WRITE
+ val tableName = "test_lance_blob_inline_upsert_merge_cow"
+ val tablePath = s"$basePath/$tableName"
+
+ val payloadLen = 1024
+ val numRows = 6
+ val initialPayloads: 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).coalesce(1)
+ }
+
+ // First commit: bulk_insert ids 0..5 into a single base file. A single
file group is required
+ // so the untouched ids 3..5 genuinely pass through the merge rewrite; in
their own group the
+ // upsert would never touch them and the byte checks below would pass
vacuously.
+ writeDataframe(tableType, tableName, tablePath,
+ asInlineDf(initialPayloads.zipWithIndex.map { case (b, i) => (i, b) }),
+ saveMode = SaveMode.Overwrite,
+ operation = Some("bulk_insert"),
+ extraOptions = Map(PRECOMBINE_FIELD.key() -> "id",
+ "hoodie.bulkinsert.shuffle.parallelism" -> "1",
+ "hoodie.insert.shuffle.parallelism" -> "1"))
+
+ assertLanceBlobEncoding(tablePath)
+
+ val metaClientAfterFirst = HoodieTableMetaClient.builder()
+ .setConf(HoodieTestUtils.getDefaultStorageConf)
+ .setBasePath(tablePath)
+ .build()
+ val preUpsertBaseFiles = latestBaseFileNames(metaClientAfterFirst,
tablePath)
+ assertEquals(1, preUpsertBaseFiles.size,
+ s"All rows must land in exactly one base file (coalesce(1) + shuffle
parallelism 1) at " +
+ s"$tablePath, got $preUpsertBaseFiles; otherwise untouched ids sit in
file groups the " +
+ s"upsert never rewrites and the merge path is not exercised")
+
+ // Second commit: upsert ids 0..2 with all-0xEE payloads. On CoW this
routes every existing
+ // file-group record through the merge handle's CONTENT-pinned base-file
read and rewrite.
+ val updatedPayloadByte: Byte = 0xEE.toByte
+ val updatedIds = 0 until 3
+ val updatedPayloads = updatedIds.map(i => (i,
Array.fill[Byte](payloadLen)(updatedPayloadByte)))
+ writeDataframe(tableType, tableName, tablePath,
+ asInlineDf(updatedPayloads),
+ operation = Some("upsert"),
+ extraOptions = Map(PRECOMBINE_FIELD.key() -> "id"))
+
+ // The upsert must have stayed on the CoW commit path: two completed
commits, no deltacommits
+ // (a deltacommit would mean an append path that never rewrites the base
file).
+ val metaClient = HoodieTableMetaClient.builder()
+ .setConf(HoodieTestUtils.getDefaultStorageConf)
+ .setBasePath(tablePath)
+ .build()
+ val completedInstants =
metaClient.reloadActiveTimeline().filterCompletedInstants()
+ .getInstants.asScala
+ assertEquals(2, completedInstants.count(_.getAction == "commit"),
+ "Expected exactly two completed commits (bulk_insert + upsert) on CoW")
+ assertTrue(completedInstants.forall(_.getAction != "deltacommit"),
+ "CoW upsert must not write deltacommits; the merge rewrite would not be
exercised")
+
+ // The merge must also have replaced the base file: a single new name,
disjoint from the
+ // pre-upsert one. If the old name were still the latest, the untouched
ids were never
+ // merged and the byte checks below would read stale bytes.
+ val postUpsertBaseFiles = latestBaseFileNames(metaClient, tablePath)
+ assertEquals(1, postUpsertBaseFiles.size,
+ s"Upsert must keep all rows in one file group, got $postUpsertBaseFiles")
+ assertTrue(preUpsertBaseFiles.intersect(postUpsertBaseFiles).isEmpty,
+ s"Post-upsert base file must differ from the pre-upsert one, proving the
merge rewrote it " +
+ s"(pre=$preUpsertBaseFiles, post=$postUpsertBaseFiles)")
+
+ val expected: Map[Int, Array[Byte]] = (
+ updatedIds.map(i => i ->
Array.fill[Byte](payloadLen)(updatedPayloadByte)) ++
+ (updatedIds.length until numRows).map(i => i -> initialPayloads(i))
+ ).toMap
+
+ // Plain read yields the DESCRIPTOR shape, confirming the user-facing
default end-to-end.
+ val readRows = spark.read.format("hudi")
+ .load(tablePath)
+ .select($"id", $"payload")
+ .orderBy($"id")
+ .collect()
+ assertEquals(numRows, readRows.length)
+ readRows.foreach { row =>
+ val id = row.getInt(row.fieldIndex("id"))
+ val payload = row.getStruct(row.fieldIndex("payload"))
+ assertEquals(HoodieSchema.Blob.INLINE,
+ payload.getString(payload.fieldIndex(HoodieSchema.Blob.TYPE)),
+ s"Type must remain INLINE post-merge (id=$id)")
+
assertTrue(payload.isNullAt(payload.fieldIndex(HoodieSchema.Blob.INLINE_DATA_FIELD)),
+ s"DESCRIPTOR default should null `data` on plain read (id=$id)")
+
assertNotNull(payload.getStruct(payload.fieldIndex(HoodieSchema.Blob.EXTERNAL_REFERENCE)),
+ s"DESCRIPTOR default should populate reference on plain read (id=$id)")
+ }
+
+ // Byte check via a plain projection under CONTENT. As in the compaction
test: a DESCRIPTOR
+ // leak already fails the merge in HoodieSparkLanceWriter.validateBlobRow,
so what this
+ // catches is the guard-allowed empty-inline shape {INLINE, null data,
null reference},
+ // where dropped bytes would persist silently.
+ val contentRows = spark.read.format("hudi")
+ .option("hoodie.read.blob.inline.mode", "CONTENT")
+ .load(tablePath)
+ .select($"id", $"payload")
+ .orderBy($"id")
+ .collect()
+ assertEquals(numRows, contentRows.length)
+ contentRows.foreach { row =>
+ val id = row.getInt(row.fieldIndex("id"))
+ val payload = row.getStruct(row.fieldIndex("payload"))
+
assertFalse(payload.isNullAt(payload.fieldIndex(HoodieSchema.Blob.INLINE_DATA_FIELD)),
+ s"null data under CONTENT: the merge rewrite dropped the bytes
(id=$id)")
+ assertArrayEquals(expected(id),
+
payload.getAs[Array[Byte]](payload.fieldIndex(HoodieSchema.Blob.INLINE_DATA_FIELD)),
+ s"INLINE data bytes must survive the merge rewrite (id=$id)")
+ }
+
+ // read_blob() under CONTENT verifies the same bytes through the SQL
expression path.
+ val viewName = s"${tableName}_view"
+ spark.read.format("hudi")
+ .option("hoodie.read.blob.inline.mode", "CONTENT")
+ .load(tablePath)
+ .createOrReplaceTempView(viewName)
+ val materializedRows = spark.sql(
+ s"SELECT id, read_blob(payload) AS bytes FROM $viewName ORDER BY
id").collect()
+ assertEquals(numRows, materializedRows.length)
+ materializedRows.foreach { row =>
+ val id = row.getInt(row.fieldIndex("id"))
+ val bytes = row.getAs[Array[Byte]]("bytes")
+ assertArrayEquals(expected(id), bytes,
+ s"read_blob() must return correct bytes post-merge (id=$id)")
+ }
+ }
+
+ /**
+ * Latest base file name per file group under {@code tablePath}. Snapshots
taken before and
+ * after a rewriting table service (clustering, CoW upsert merge) are
compared for disjointness
+ * to prove the rewrite actually replaced the base file(s).
+ */
+ private def latestBaseFileNames(mc: HoodieTableMetaClient, tablePath:
String): Set[String] = {
+ val engineContext = new HoodieLocalEngineContext(mc.getStorageConf)
+ val metadataConfig = HoodieMetadataConfig.newBuilder.build
+ val viewManager = FileSystemViewManager.createViewManager(
+ engineContext, metadataConfig,
FileSystemViewStorageConfig.newBuilder.build,
+ HoodieCommonConfig.newBuilder.build,
+ (m: HoodieTableMetaClient) => mc.getTableFormat
+ .getMetadataFactory.create(engineContext, m.getStorage,
metadataConfig, tablePath))
+ val fsView = viewManager.getFileSystemView(mc)
+ try {
+ fsView.getLatestBaseFiles("")
+
.collect(Collectors.toList[org.apache.hudi.common.model.HoodieBaseFile])
+ .asScala.map(_.getFileName).toSet
+ } finally {
+ fsView.close()
+ }
+ }
+
/**
* Shared implementation for OUT_OF_LINE blob tests. Writes rows with
external references,
* reads them back (optionally with a specific read mode), and asserts the
reference survives
