This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 5b3fe07bc7167c0ff01b67bf13f67d1ea9808ed7 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 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. (cherry picked from commit 120d73fcc4dba7ddfae0b9693f487190a8410d34) --- .../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 7dd00a9e430c..f9c99d1db022 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 a930eaa4b921..040c63bd0158 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 3bdf12d9059b..eb36a1f9a955 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 77b280cdd91f..f48df51c47f2 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.table.HoodieTableConfig; @@ -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 ddbb798a5711..c653d3f0f9e5 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 722833a7756c..7f245f65ef6d 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 @@ -28,6 +28,7 @@ import org.apache.hudi.common.schema.HoodieSchemaField; 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.exception.HoodieException; import org.apache.hudi.exception.HoodieNotSupportedException; import org.apache.hudi.io.memory.HoodieArrowAllocator; import org.apache.hudi.storage.HoodieStorage; @@ -639,4 +640,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
