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 fd3f51053b7783d9e608ab44395fb91eb446bf7a Author: vinoth chandar <[email protected]> AuthorDate: Fri Jul 10 01:53:42 2026 -0700 fix(spark): read Lance BLOB columns in <=512-row chunks to avoid lance-core FFI abort (#19181) * fix(spark): read Lance BLOB columns in <=512-row chunks to avoid lance-core FFI abort lance-core 4.0.0 aborts the JVM in its Arrow C-stream export (arrow_array::ffi_stream::get_next: "range end index N out of range for slice of length 0") whenever a single readAll stream crosses Lance's internal BLOB page boundary (512 rows); the requested batchSize does not help because Lance re-chunks BLOB columns at 512 internally. As a result CoW Lance tables with BLOB columns could not be read past 512 rows: OUT_OF_LINE reads threw an Arrow "should have as many children as in the schema" error and INLINE CONTENT reads crashed the JVM (SIGABRT). Read BLOB-containing Lance files in <=512-row row-range chunks, issuing a fresh readAll per chunk so each FFI stream stays within a single BLOB page and the buggy second-page export is never reached. LanceRecordIterator gains a chunkedBlobReader(...) factory driven by an ArrowReaderSupplier; SparkLanceReaderBase (SQL/DataFrame read) and HoodieSparkLanceReader (internal CONTENT path) route to it when the projection contains a BLOB field. Non-BLOB reads keep the single streamed reader and are unchanged; the per-row hot path is untouched, with only small fixed per-chunk allocations. Also document hoodie.read.blob.inline.mode option placement (read option for both DataFrame and SQL) and add batch-scale regression tests: INLINE CONTENT, VECTOR + OUT_OF_LINE BLOB at n=100/1000 (crossing the 512-row boundary), with vector top-k, projection/filter, and read_blob byte + SHA-256 checks. * refactor(spark): rename ArrowReaderSupplier to ArrowReaderSequence, use nonEmpty * fix(spark): recurse Lance BLOB chunking detection, pin chunk size to lance-core 4.0.0 Route reads through the chunked path when a BLOB appears at any nesting depth (HoodieSchema.containsBlobType() made public for the internal reader; recursive StructType walk in SparkLanceReaderBase), so a future nested-BLOB writer change cannot silently skip chunking and re-introduce the lance-core FFI abort. Document BLOB_READ_CHUNK_ROWS as pinned to lance-core 4.0.0's internal 512-row BLOB page size, revalidated on lance upgrades. --------- Co-authored-by: voon <[email protected]> (cherry picked from commit 14fe7393a42fc1f6ac79533fa5cac18d0812330a) --- .../hudi/io/storage/HoodieSparkLanceReader.java | 10 +- .../hudi/io/storage/LanceRecordIterator.java | 172 ++++++++++--- .../hudi/common/config/HoodieReaderConfig.java | 9 +- .../apache/hudi/common/schema/HoodieSchema.java | 7 +- .../datasources/lance/SparkLanceReaderBase.scala | 32 ++- .../hudi/functional/TestLanceDataSource.scala | 265 ++++++++++++++++++++- 6 files changed, 444 insertions(+), 51 deletions(-) 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 bd651a9e9bef..a930eaa4b921 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 @@ -208,8 +208,16 @@ public class HoodieSparkLanceReader implements HoodieSparkFileReader { // Pinned to CONTENT: compaction/merge/log-replay need actual bytes to rewrite. // The user-facing `hoodie.read.blob.inline.mode` is honored by SparkLanceReaderBase. FileReadOptions readOpts = FileReadOptions.builder().blobReadMode(BlobReadMode.CONTENT).build(); - ArrowReader arrowReader = lanceReader.readAll(columnNames, null, DEFAULT_BATCH_SIZE, readOpts); + // BLOB reads must be chunked to dodge a lance-core FFI abort (see LanceRecordIterator). + // containsBlobType() recurses through nested records/arrays/maps/unions, so a BLOB at any + // depth routes through the chunked path; a top-level-only check would silently skip + // chunking (and re-introduce the abort) if the writer ever gains nested-BLOB support. + if (requestedSchema.containsBlobType()) { + return LanceRecordIterator.chunkedBlobReader(allocator, lanceReader, columnNames, readOpts, + lanceReader.numRows(), requestedSparkSchema, path.toString(), null); + } + ArrowReader arrowReader = lanceReader.readAll(columnNames, null, DEFAULT_BATCH_SIZE, readOpts); return new LanceRecordIterator(allocator, lanceReader, arrowReader, requestedSparkSchema, path.toString()); } catch (Exception e) { allocator.close(); diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/LanceRecordIterator.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/LanceRecordIterator.java index 91a0421d0118..2cc6b5c314c7 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/LanceRecordIterator.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/LanceRecordIterator.java @@ -32,39 +32,58 @@ import org.apache.spark.sql.types.StructField; import org.apache.spark.sql.types.StructType; import org.apache.spark.sql.vectorized.ColumnVector; import org.apache.spark.sql.vectorized.ColumnarBatch; +import org.lance.file.FileReadOptions; import org.lance.file.LanceFileReader; import org.lance.spark.vectorized.LanceArrowColumnVector; +import org.lance.util.Range; import java.io.IOException; +import java.util.Collections; import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; /** - * Iterator for reading Lance files and converting Arrow batches to Spark {@link UnsafeRow}s. - * Used by both Hudi's internal Lance reader and Spark datasource integration. + * Iterator over a Lance file that converts Arrow batches to Spark {@link UnsafeRow}s, owning the + * {@link BufferAllocator}, {@link LanceFileReader}, {@link ArrowReader}(s) and current + * {@link ColumnarBatch}. Used by both Hudi's internal Lance reader and the Spark datasource. An + * optional {@link BlobDescriptorTransform} rewrites BLOB columns for DESCRIPTOR-mode reads. * - * <p>The iterator manages the lifecycle of: - * <ul> - * <li>BufferAllocator - Arrow memory management</li> - * <li>LanceFileReader - Lance file handle</li> - * <li>ArrowReader - Arrow batch reader</li> - * <li>ColumnarBatch - Current batch being iterated</li> - * </ul> - * - * <p>An optional {@link BlobDescriptorTransform} can be composed in to rewrite BLOB columns - * in DESCRIPTOR mode. + * <p>BLOB chunked reading: lance-core 4.0.0 aborts the JVM in its Arrow C-stream export whenever a + * single {@code readAll} stream crosses Lance's internal BLOB page boundary (512 rows), and the + * requested {@code batchSize} does not change that. So BLOB reads issue one {@code readAll} per + * row-range chunk of {@link #BLOB_READ_CHUNK_ROWS} rows; non-BLOB reads use one streamed reader. */ public final class LanceRecordIterator implements ClosableIterator<UnsafeRow> { + + /** + * Rows per {@code readAll} for BLOB reads. Must not exceed Lance's internal BLOB page size, + * which is 512 rows in lance-core 4.0.0 but is NOT exposed through any lance API; revalidate + * this constant against the batch-scale BLOB tests whenever {@code lance.version} is bumped. + */ + public static final int BLOB_READ_CHUNK_ROWS = 512; + + /** + * The sequence of {@link ArrowReader}s to drain, one after another. Single-reader + * mode yields one reader; BLOB chunked mode yields one fresh reader per row-range chunk. Each + * returned reader is owned (and closed) by {@link LanceRecordIterator}. + */ + private interface ArrowReaderSequence { + /** @return the next reader to drain, or {@code null} when the sequence is exhausted. */ + ArrowReader next() throws IOException; + } + private final BufferAllocator allocator; private final LanceFileReader lanceReader; - private final ArrowReader arrowReader; + private final ArrowReaderSequence readerSequence; private final StructType sparkSchema; private final UnsafeProjection projection; private final String path; private final BlobDescriptorTransform blobTransform; + /** The reader currently being drained; replaced as chunks are exhausted. */ + private ArrowReader currentReader; private ColumnarBatch currentBatch; private Iterator<InternalRow> rowIterator; private ColumnVector[] columnVectors; @@ -84,7 +103,8 @@ public final class LanceRecordIterator implements ClosableIterator<UnsafeRow> { } /** - * Creates a new Lance record iterator. + * Creates a new Lance record iterator that drains a single pre-built {@link ArrowReader}. + * Suitable for non-BLOB reads, where Lance's multi-batch FFI export is well-behaved. * * @param allocator Arrow buffer allocator for memory management * @param lanceReader Lance file reader @@ -100,15 +120,78 @@ public final class LanceRecordIterator implements ClosableIterator<UnsafeRow> { StructType schema, String path, BlobDescriptorTransform blobTransform) { + this(allocator, lanceReader, singleReaderSequence(arrowReader), schema, path, blobTransform); + } + + private LanceRecordIterator(BufferAllocator allocator, + LanceFileReader lanceReader, + ArrowReaderSequence readerSequence, + StructType schema, + String path, + BlobDescriptorTransform blobTransform) { this.allocator = allocator; this.lanceReader = lanceReader; - this.arrowReader = arrowReader; + this.readerSequence = readerSequence; this.sparkSchema = schema; this.projection = UnsafeProjection.create(schema); this.path = path; this.blobTransform = blobTransform; } + /** + * Creates a Lance record iterator that reads a BLOB-containing file in fixed-size row-range + * chunks, issuing a fresh {@code readAll} per chunk to dodge the lance-core multi-page BLOB + * FFI-export panic (see class javadoc). + * + * @param allocator Arrow buffer allocator for memory management + * @param lanceReader open Lance file reader (the iterator takes ownership and closes it) + * @param columnNames columns to project, or {@code null} for all columns + * @param readOpts Lance read options (e.g. blob read mode) + * @param totalRows total rows in the file ({@code lanceReader.numRows()}) + * @param schema Spark schema for the records + * @param path File path (for error messages) + * @param blobTransform optional blob descriptor transform for DESCRIPTOR-mode reads + */ + public static LanceRecordIterator chunkedBlobReader(BufferAllocator allocator, + LanceFileReader lanceReader, + List<String> columnNames, + FileReadOptions readOpts, + long totalRows, + StructType schema, + String path, + BlobDescriptorTransform blobTransform) { + ArrowReaderSequence sequence = new ArrowReaderSequence() { + private long nextStart = 0; + + @Override + public ArrowReader next() throws IOException { + if (nextStart >= totalRows) { + return null; + } + int start = Math.toIntExact(nextStart); + int end = Math.toIntExact(Math.min(nextStart + BLOB_READ_CHUNK_ROWS, totalRows)); + nextStart = end; + // A single range per readAll keeps the FFI stream within one BLOB page (<= 512 rows). + List<Range> ranges = Collections.singletonList(new Range(start, end)); + return lanceReader.readAll(columnNames, ranges, BLOB_READ_CHUNK_ROWS, readOpts); + } + }; + return new LanceRecordIterator(allocator, lanceReader, sequence, schema, path, blobTransform); + } + + private static ArrowReaderSequence singleReaderSequence(ArrowReader arrowReader) { + return new ArrowReaderSequence() { + private ArrowReader remaining = arrowReader; + + @Override + public ArrowReader next() { + ArrowReader r = remaining; + remaining = null; + return r; + } + }; + } + @Override public boolean hasNext() { if (rowIterator != null && rowIterator.hasNext()) { @@ -120,33 +203,50 @@ public final class LanceRecordIterator implements ClosableIterator<UnsafeRow> { currentBatch = null; } - // Try to load next batch. Loop so zero-row batches (legitimately returned e.g. after - // filter pushdown) don't silently terminate iteration and drop subsequent non-empty batches. try { - while (arrowReader.loadNextBatch()) { - VectorSchemaRoot root = arrowReader.getVectorSchemaRoot(); - - // Build ColumnVector[] in Spark-schema order by looking each field up by name; - // lance-spark 0.4.0's VectorSchemaRoot may return the file's on-disk order, which - // would misalign the UnsafeProjection. Cached on the first batch and reused thereafter. - if (columnVectors == null) { - buildColumnVectors(root); + while (true) { + if (currentReader == null) { + currentReader = readerSequence.next(); + if (currentReader == null) { + return false; + } + // Each reader (range chunk) returns a distinct VectorSchemaRoot, so the cached + // column vectors must be rebuilt against the new reader's vectors. + columnVectors = null; } - currentBatch = new ColumnarBatch(columnVectors, root.getRowCount()); - rowIterator = currentBatch.rowIterator(); - rowIdInBatch = 0; - if (rowIterator.hasNext()) { - return true; + // Try to load next batch from the current reader. Loop so zero-row batches + // (legitimately returned e.g. after filter pushdown) don't silently terminate. + while (currentReader.loadNextBatch()) { + VectorSchemaRoot root = currentReader.getVectorSchemaRoot(); + + // Build ColumnVector[] in Spark-schema order by looking each field up by name; + // lance-spark 0.4.0's VectorSchemaRoot may return the file's on-disk order, which + // would misalign the UnsafeProjection. Cached per reader and reused thereafter. + if (columnVectors == null) { + buildColumnVectors(root); + } + + currentBatch = new ColumnarBatch(columnVectors, root.getRowCount()); + rowIterator = currentBatch.rowIterator(); + rowIdInBatch = 0; + if (rowIterator.hasNext()) { + return true; + } + currentBatch.close(); + currentBatch = null; } - currentBatch.close(); - currentBatch = null; + + // Current reader exhausted; close it and advance to the next chunk (if any). + currentReader.close(); + currentReader = null; + columnVectors = null; } } catch (IOException e) { throw new HoodieException("Failed to read next batch from Lance file: " + path, e); + } catch (Exception e) { + throw new HoodieException("Failed to advance Lance reader for file: " + path, e); } - - return false; } @Override @@ -197,6 +297,8 @@ public final class LanceRecordIterator implements ClosableIterator<UnsafeRow> { closed = true; ColumnarBatch batch = currentBatch; currentBatch = null; - LanceResourceCloser.closeAll(batch, arrowReader, lanceReader, allocator); + ArrowReader reader = currentReader; + currentReader = null; + LanceResourceCloser.closeAll(batch, reader, lanceReader, allocator); } } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieReaderConfig.java b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieReaderConfig.java index 9cbab8f4468c..a1ecf64dc419 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieReaderConfig.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieReaderConfig.java @@ -112,8 +112,9 @@ public class HoodieReaderConfig extends HoodieConfig { .sinceVersion("1.2.0") .withValidValues(BLOB_INLINE_READ_MODE_CONTENT, BLOB_INLINE_READ_MODE_DESCRIPTOR) .withDocumentation("How Hudi interprets INLINE BLOB values on read. " - + "DESCRIPTOR (default) returns an OUT_OF_LINE-shaped reference pointing at the " - + "backing Lance file with the INLINE payload's position and size, so callers can " - + "skip the byte content read. " - + "CONTENT returns the raw inline bytes directly in the data field on every read."); + + "DESCRIPTOR (default) returns an OUT_OF_LINE-shaped reference (position and size) into " + + "the backing file, skipping the byte read. CONTENT returns the raw inline bytes in the " + + "data field. Materializing INLINE bytes via read_blob() requires CONTENT; under " + + "DESCRIPTOR it fails fast asking for CONTENT. Pass this as a read option, not a session " + + "config. OUT_OF_LINE blobs ignore this option."); } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java index f38b9e2c93ed..8c151af35a47 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java @@ -1390,7 +1390,12 @@ public class HoodieSchema implements Serializable { return HoodieSchema.createUnion(nonNullTypes); } - boolean containsBlobType() { + /** + * Recursively checks whether this schema is or contains a BLOB type at any nesting depth, + * descending through arrays, maps, unions and record fields. Unlike {@link #isBlobField()}, + * this also finds BLOBs nested inside records. + */ + public boolean containsBlobType() { if (getType() == HoodieSchemaType.BLOB) { return true; } else if (getType() == HoodieSchemaType.ARRAY) { 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 65dec06cd3e0..ddbb798a5711 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 @@ -137,19 +137,29 @@ class SparkLanceReaderBase(enableVectorizedReader: Boolean) extends SparkColumna // the option regardless. val blobMode = resolveBlobReadMode(storageConf) val readOpts = FileReadOptions.builder().blobReadMode(blobMode).build() - val arrowReader = lanceReader.readAll(columnNames, null, DEFAULT_BATCH_SIZE, readOpts) // Compose the DESCRIPTOR-aware blob transform only when the user opted into that mode // AND the request actually has BLOB columns (otherwise the rewrite has nothing to do). - val blobFieldNames: java.util.Set[String] = - iteratorSchema.fields.collect { case f if isBlobField(f) => f.name }.toSet.asJava - val blobTransform = if (blobMode == BlobReadMode.DESCRIPTOR && !blobFieldNames.isEmpty) { - new BlobDescriptorTransform(blobFieldNames, filePath) + val blobFieldNames: Set[String] = + iteratorSchema.fields.collect { case f if isBlobField(f) => f.name }.toSet + val blobTransform = if (blobMode == BlobReadMode.DESCRIPTOR && blobFieldNames.nonEmpty) { + new BlobDescriptorTransform(blobFieldNames.asJava, filePath) } else { null } - lanceIterator = new LanceRecordIterator( - allocator, lanceReader, arrowReader, iteratorSchema, filePath, blobTransform) + // lance-core 4.0.0 aborts the JVM when a single readAll stream crosses Lance's internal + // BLOB page boundary (512 rows). For BLOB-containing reads, drain the file in <=512-row + // range chunks (one fresh readAll each); non-BLOB reads keep the single streamed reader. + // The detection recurses so a nested BLOB (unsupported by the writer today) still chunks. + lanceIterator = if (containsBlobField(iteratorSchema)) { + LanceRecordIterator.chunkedBlobReader( + allocator, lanceReader, columnNames, readOpts, lanceReader.numRows(), + iteratorSchema, filePath, blobTransform) + } else { + val arrowReader = lanceReader.readAll(columnNames, null, DEFAULT_BATCH_SIZE, readOpts) + new LanceRecordIterator( + allocator, lanceReader, arrowReader, iteratorSchema, filePath, blobTransform) + } // Register cleanup listener Option(TaskContext.get()).foreach { ctx => @@ -250,6 +260,14 @@ class SparkLanceReaderBase(enableVectorizedReader: Boolean) extends SparkColumna .getType == HoodieSchemaType.BLOB } + /** Recursively checks for a BLOB field (see [[isBlobField]]) at any nesting depth. */ + private def containsBlobField(dt: DataType): Boolean = dt match { + case s: StructType => s.fields.exists(f => isBlobField(f) || containsBlobField(f.dataType)) + case a: ArrayType => containsBlobField(a.elementType) + case m: MapType => containsBlobField(m.valueType) + case _ => false + } + private def forceFieldNullable(field: StructField): StructField = field.copy(nullable = true, dataType = forceTypeNullable(field.dataType)) 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 abce02568633..419a7bcf6498 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 @@ -991,7 +991,7 @@ class TestLanceDataSource extends HoodieSparkClientTestBase { // read_blob() on INLINE rows under DESCRIPTOR mode is unsupported by design: DESCRIPTOR // is metadata-only and the synthesized reference is an internal pointer into the .lance // file's storage layout, not user-facing metadata. BatchedBlobReader must throw with a - // message that names both INLINE and DESCRIPTOR so the failure is actionable. + // message that names INLINE, DESCRIPTOR and the CONTENT fix so the failure is actionable val viewName = s"${tableName}_view" spark.read.format("hudi") .option(modeKey, "DESCRIPTOR") @@ -1004,8 +1004,8 @@ class TestLanceDataSource extends HoodieSparkClientTestBase { }) val msgChain = Iterator.iterate[Throwable](ex)(_.getCause).takeWhile(_ != null) .flatMap(t => Option(t.getMessage)).mkString(" | ") - assertTrue(msgChain.contains("INLINE") && msgChain.contains("DESCRIPTOR"), - s"error must mention INLINE and DESCRIPTOR; got: $msgChain") + assertTrue(Seq("INLINE", "DESCRIPTOR", "CONTENT").forall(msgChain.contains), + s"error must name INLINE, DESCRIPTOR and the CONTENT fix; got: $msgChain") } /** @@ -1981,6 +1981,253 @@ class TestLanceDataSource extends HoodieSparkClientTestBase { writer.mode(saveMode).save(tablePath) } + + // The small-n blob tests above all read fewer than 512 rows, so they never cross Lance's + // internal BLOB page boundary (512 rows). BLOB reads regress specifically once a single base + // file holds more than 512 rows: a single Lance readAll stream then exports a second BLOB page + // through the Arrow C FFI, which panics in lance-core 4.0.0. These tests size the base file + // above 512 rows (single coalesced file) to exercise the chunked-read work-around. + private val BATCH_SCALE_ROWS = 1000 + private val BATCH_SCALE_PARALLELISM_OPTS = Map( + "hoodie.bulkinsert.shuffle.parallelism" -> "1", + "hoodie.insert.shuffle.parallelism" -> "1") + + /** + * Batch-scale INLINE BLOB read regression (HUDI-UNSTRUCTURED-001). Writes + * {@code BATCH_SCALE_ROWS} inline-blob rows into a single Lance base file and reads them back + * under CONTENT mode, materializing each via {@code read_blob()}. Reproduces the native Lance + * BLOB decoder failure that only surfaces once the read crosses the 512-row batch boundary. + */ + @ParameterizedTest + @EnumSource(value = classOf[HoodieTableType]) + def testBlobInlineContentBatchScale(tableType: HoodieTableType): Unit = { + val tableName = s"test_lance_blob_inline_batch_${tableType.name().toLowerCase}" + val tablePath = s"$basePath/$tableName" + + val payloadLen = 256 + val n = BATCH_SCALE_ROWS + val sparkSess = spark + import sparkSess.implicits._ + // Deterministic per-row payload: row i -> bytes (i + j) % 256, so a row/byte misalignment + // across the batch boundary surfaces as a byte mismatch rather than silently passing. + def payloadFor(i: Int): Array[Byte] = + (0 until payloadLen).map(j => ((i + j) % 256).toByte).toArray + val baseDf = (0 until n).map(i => (i, payloadFor(i))) + .toDF("id", "bytes") + .coalesce(1) + val rawDf = baseDf.select($"id", BlobTestHelpers.inlineBlobStructCol("payload", $"bytes")) + val canonicalSchema = StructType(Seq( + StructField("id", IntegerType, nullable = false), + StructField("payload", BlobType().asInstanceOf[StructType], nullable = true, + BlobTestHelpers.blobMetadata) + )) + val df = spark.createDataFrame(rawDf.rdd, canonicalSchema).coalesce(1) + + writeDataframe(tableType, tableName, tablePath, df, saveMode = SaveMode.Overwrite, + operation = Some("bulk_insert"), + extraOptions = Map(PRECOMBINE_FIELD.key() -> "id") ++ BATCH_SCALE_PARALLELISM_OPTS) + + assertLanceBlobEncoding(tablePath) + assertSingleLanceBaseFileSpansMultipleBatches(tablePath) + + 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(n, materialized.length, "row count mismatch after read_blob at batch scale") + materialized.foreach { row => + val id = row.getInt(row.fieldIndex("id")) + val bytes = row.getAs[Array[Byte]]("bytes") + assertArrayEquals(payloadFor(id), bytes, s"inline read_blob() bytes mismatch for id=$id") + } + } + + /** + * Comprehensive batch-scale VECTOR + OUT_OF_LINE BLOB regression (HUDI-UNSTRUCTURED-002 and 003). + * Parameterized over table type and {@code n} in {100, 1000}; n=1000 crosses the 512-row BLOB page + * boundary that trips the lance-core FFI panic without the chunked-read fix. Writes one Lance base + * file with a VECTOR(32) and an OUT_OF_LINE BLOB column, then validates across DataFrame and SQL + * reads: row count; exact vector values (incl. rows straddling the boundary); top-k IDs via + * {@code hudi_vector_search}; column projection and id-range filtering; {@code read_blob()} bytes + + * SHA-256 (full and filtered); and DESCRIPTOR-mode reference pass-through. Default Lance read + * allocator (256MB) suffices — no non-default settings required. + */ + @ParameterizedTest + @MethodSource(Array("vectorBlobBatchParams")) + def testVectorAndBlobBatchScale(tableType: HoodieTableType, n: Int): Unit = { + val tableName = s"test_lance_vec_blob_batch_${n}_${tableType.name().toLowerCase}" + val tablePath = s"$basePath/$tableName" + + val dim = 32 + val payloadLen = 512 + val externalDir = Files.createDirectories( + Paths.get(s"$basePath/_vec_blob_ext_${n}_${tableType.name().toLowerCase}")) + val extPath = BlobTestHelpers.createTestFile(externalDir, "vec_blob.bin", n * payloadLen) + + val sparkSess = spark + import sparkSess.implicits._ + // Deterministic, strictly monotonic-by-id vector: row i -> [(i*dim+j)/1000f]. Monotonicity + // makes nearest-neighbor ordering predictable; per-row distinctness makes a row/value + // misalignment across the batch boundary surface as a value (or top-k) mismatch. + def vectorFor(i: Int): Array[Float] = (0 until dim).map(j => (i * dim + j) / 1000.0f).toArray + // Deterministic blob bytes for row i: (i*payloadLen + k) % 256, matching assertBytesContent. + def expectedBlob(i: Int): Array[Byte] = + (0 until payloadLen).map(k => ((i * payloadLen + k) % 256).toByte).toArray + def sha256(bytes: Array[Byte]): Seq[Byte] = + java.security.MessageDigest.getInstance("SHA-256").digest(bytes).toSeq + + val baseDf = (0 until n) + .map(i => (i, vectorFor(i), extPath, (i.toLong * payloadLen), payloadLen.toLong)) + .toDF("id", "embedding", "path", "offset", "length") + .coalesce(1) + val rawDf = baseDf.select($"id", $"embedding", + BlobTestHelpers.blobStructCol("payload", $"path", $"offset", $"length")) + val vectorMeta = new MetadataBuilder() + .putString(HoodieSchema.TYPE_METADATA_FIELD, s"VECTOR($dim)").build() + val canonicalSchema = StructType(Seq( + StructField("id", IntegerType, nullable = false), + StructField("embedding", ArrayType(FloatType, containsNull = false), nullable = false, + vectorMeta), + StructField("payload", BlobType().asInstanceOf[StructType], nullable = true, + BlobTestHelpers.blobMetadata) + )) + val df = spark.createDataFrame(rawDf.rdd, canonicalSchema).coalesce(1) + + writeDataframe(tableType, tableName, tablePath, df, saveMode = SaveMode.Overwrite, + operation = Some("bulk_insert"), + extraOptions = Map(PRECOMBINE_FIELD.key() -> "id") ++ BATCH_SCALE_PARALLELISM_OPTS) + + if (n > 512) { + assertSingleLanceBaseFileSpansMultipleBatches(tablePath) + } + + // --- Vectors (DataFrame path): row count + exact values on a sample straddling the boundary. + val vecRows = spark.read.format("hudi").load(tablePath) + .select($"id", $"embedding").orderBy($"id").collect() + assertEquals(n, vecRows.length, "vector row count mismatch at batch scale") + val sampleIds = (Seq(0, 1, n / 2, n - 1) ++ (if (n > 512) Seq(511, 512, 513) else Seq.empty)) + .filter(i => i >= 0 && i < n).distinct + val vecById = vecRows.map(r => r.getInt(r.fieldIndex("id")) -> r).toMap + sampleIds.foreach { id => + val emb = vecById(id).getSeq[Float](vecById(id).fieldIndex("embedding")).toArray + assertEquals(dim, emb.length, s"vector dim mismatch for id=$id") + val expected = vectorFor(id) + (0 until dim).foreach { j => + assertEquals(expected(j), emb(j), 1e-6f, s"vector value mismatch id=$id j=$j") + } + } + + // --- Top-k vector search IDs: query near row q nudged toward higher ids by a small delta so + // the L2 ordering is strict (q, q+1, q-1). hudi_vector_search must return those exact IDs. + val tkView = s"${tableName}_tk" + spark.read.format("hudi").load(tablePath) + .select("id", "embedding").createOrReplaceTempView(tkView) + val q = n / 2 + val delta = 0.005f + val queryLiteral = vectorFor(q).map(v => (v + delta).toDouble).mkString(", ") + val topk = spark.sql( + s"""SELECT id, _hudi_distance + |FROM hudi_vector_search('$tkView', 'embedding', ARRAY($queryLiteral), 3, 'l2') + |ORDER BY _hudi_distance""".stripMargin).collect() + val topkIds = topk.map(_.getInt(0)).toSeq + assertEquals(Seq(q, q + 1, q - 1), topkIds, + s"top-3 vector-search IDs mismatch for query near id=$q") + + // --- Projection (id only) and predicate filter (id range) correctness. + val idOnly = spark.read.format("hudi").load(tablePath).select("id") + assertEquals(1, idOnly.schema.fields.length, "projection should yield a single column") + // collect() (not count()) so the id column is actually projected/read; a count() would push an + // empty required schema down the scan, a separate code path not under test here. + val idOnlyVals = idOnly.collect().map(_.getInt(0)).toSet + assertEquals((0 until n).toSet, idOnlyVals, "projected id set mismatch") + // For n=1000 choose a range that straddles the 512-row batch boundary (400..620). + val lo = if (n > 512) 400 else n / 4 + val hi = math.min(n, lo + (if (n > 512) 220 else 30)) + val filteredIds = spark.read.format("hudi").load(tablePath) + .where(s"id >= $lo AND id < $hi").select("id").orderBy("id") + .collect().map(_.getInt(0)).toSeq + assertEquals((lo until hi).toSeq, filteredIds, "filtered id range mismatch") + + // --- Blobs (SQL path, CONTENT): full-scan read_blob byte content + SHA-256 round-trip. + val viewName = s"${tableName}_view" + spark.read.format("hudi") + .option("hoodie.read.blob.inline.mode", "CONTENT") + .load(tablePath) + .createOrReplaceTempView(viewName) + val blobRows = spark.sql( + s"SELECT id, read_blob(payload) AS bytes FROM $viewName ORDER BY id").collect() + assertEquals(n, blobRows.length, "blob row count mismatch at batch scale") + blobRows.foreach { row => + val id = row.getInt(row.fieldIndex("id")) + val bytes = row.getAs[Array[Byte]]("bytes") + assertEquals(payloadLen, bytes.length, s"blob length mismatch for id=$id") + BlobTestHelpers.assertBytesContent(bytes, expectedOffset = id * payloadLen) + } + // SHA-256 round-trip on a sample (faithful to the AC's SHA256 requirement). + val blobById = blobRows.map(r => r.getInt(r.fieldIndex("id")) -> r.getAs[Array[Byte]]("bytes")).toMap + sampleIds.foreach { id => + assertEquals(sha256(expectedBlob(id)), sha256(blobById(id)), + s"blob SHA-256 mismatch for id=$id") + } + + // --- Blobs under a filter (SQL path, CONTENT): read_blob restricted to an id range. + val filteredBlobs = spark.sql( + s"SELECT id, read_blob(payload) AS bytes FROM $viewName WHERE id >= $lo AND id < $hi ORDER BY id") + .collect() + assertEquals(hi - lo, filteredBlobs.length, "filtered read_blob count mismatch") + filteredBlobs.foreach { row => + val id = row.getInt(row.fieldIndex("id")) + assertEquals(sha256(expectedBlob(id)), sha256(row.getAs[Array[Byte]]("bytes")), + s"filtered blob SHA-256 mismatch for id=$id") + } + + // --- DESCRIPTOR-mode read at scale: exercises the chunked reader WITH the blob transform + // (re-initialized per chunk). OUT_OF_LINE references must pass through intact on both sides of + // the batch boundary. + val descRows = spark.read.format("hudi").option("hoodie.read.blob.inline.mode", "DESCRIPTOR") + .load(tablePath).select($"id", $"payload").orderBy($"id").collect() + val descById = descRows.map(r => r.getInt(r.fieldIndex("id")) -> r).toMap + sampleIds.foreach { id => + val payload = descById(id).getStruct(descById(id).fieldIndex("payload")) + assertEquals(HoodieSchema.Blob.OUT_OF_LINE, + payload.getString(payload.fieldIndex(HoodieSchema.Blob.TYPE)), s"type mismatch id=$id") + val ref = payload.getStruct(payload.fieldIndex(HoodieSchema.Blob.EXTERNAL_REFERENCE)) + assertTrue(ref.getString(ref.fieldIndex(HoodieSchema.Blob.EXTERNAL_REFERENCE_PATH)) + .endsWith(".bin"), s"external_path mismatch id=$id") + assertEquals(id.toLong * payloadLen, + ref.getLong(ref.fieldIndex(HoodieSchema.Blob.EXTERNAL_REFERENCE_OFFSET)), + s"reference offset mismatch id=$id") + } + } + + /** + * Guards the batch-scale tests' core assumption: exactly one Lance base file holds all rows and + * that file has more rows than a single Arrow read batch (512). If a future change splits the + * write across files or shrinks the row count below the batch size, the cross-batch drain path + * would no longer be exercised and the regression would silently stop reproducing. + */ + private def assertSingleLanceBaseFileSpansMultipleBatches(tablePath: String): Unit = { + val lanceFiles = Files.walk(Paths.get(tablePath)) + .filter(p => p.toString.endsWith(".lance")) + .collect(Collectors.toList[java.nio.file.Path]).asScala + assertEquals(1, lanceFiles.length, + s"expected exactly one Lance base file, found: ${lanceFiles.mkString(", ")}") + val allocator = new RootAllocator(64L * 1024 * 1024) + try { + val reader = LanceFileReader.open(lanceFiles.head.toString, allocator) + try { + assertTrue(reader.numRows() > 512, + s"base file must span >512 rows to cross a batch boundary, got ${reader.numRows()}") + } finally { + reader.close() + } + } finally { + allocator.close() + } + } } object TestLanceDataSource { @@ -1992,4 +2239,16 @@ object TestLanceDataSource { } yield Arguments.of(tableType, readMode: java.lang.String) java.util.stream.Stream.of(params: _*) } + + /** + * Cross-product of table types and row counts for the VECTOR+BLOB batch-scale suite. n=100 stays + * within one Arrow batch; n=1000 crosses the 512-row Lance BLOB page boundary. + */ + def vectorBlobBatchParams(): java.util.stream.Stream[Arguments] = { + val params = for { + tableType <- HoodieTableType.values() + n <- Array(100, 1000) + } yield Arguments.of(tableType, n: java.lang.Integer) + java.util.stream.Stream.of(params: _*) + } }
