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 14fe7393a42f fix(spark): read Lance BLOB columns in <=512-row chunks
to avoid lance-core FFI abort (#19181)
14fe7393a42f is described below
commit 14fe7393a42fc1f6ac79533fa5cac18d0812330a
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]>
---
.../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 be2209917f8c..436c2e69fef0 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 577863d48856..3e831173fb82 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
@@ -119,8 +119,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 b04b494bbaee..d8cd84e85f8a 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
@@ -1374,7 +1374,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 5f9cc6346942..6f9f7783c04c 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: _*)
+ }
}
