This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch fix-lance-blob-batch-read in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 42a7e0d3c92f39c83b3b4a4ba0ab808102a8dfb7 Author: voon <[email protected]> AuthorDate: Thu Jul 9 14:19:31 2026 +0800 refactor(spark): rename ArrowReaderSupplier to ArrowReaderSequence, use nonEmpty --- .../hudi/io/storage/LanceRecordIterator.java | 22 +++++++++++----------- .../datasources/lance/SparkLanceReaderBase.scala | 10 +++++----- 2 files changed, 16 insertions(+), 16 deletions(-) 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 f587ce9c939c..f16d4dd551ac 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 @@ -61,18 +61,18 @@ public final class LanceRecordIterator implements ClosableIterator<UnsafeRow> { public static final int BLOB_READ_CHUNK_ROWS = 512; /** - * Supplies the sequence of {@link ArrowReader}s to drain, one after another. Single-reader + * 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 ArrowReaderSupplier { + 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 ArrowReaderSupplier readerSupplier; + private final ArrowReaderSequence readerSequence; private final StructType sparkSchema; private final UnsafeProjection projection; private final String path; @@ -116,18 +116,18 @@ public final class LanceRecordIterator implements ClosableIterator<UnsafeRow> { StructType schema, String path, BlobDescriptorTransform blobTransform) { - this(allocator, lanceReader, singleReaderSupplier(arrowReader), schema, path, blobTransform); + this(allocator, lanceReader, singleReaderSequence(arrowReader), schema, path, blobTransform); } private LanceRecordIterator(BufferAllocator allocator, LanceFileReader lanceReader, - ArrowReaderSupplier readerSupplier, + ArrowReaderSequence readerSequence, StructType schema, String path, BlobDescriptorTransform blobTransform) { this.allocator = allocator; this.lanceReader = lanceReader; - this.readerSupplier = readerSupplier; + this.readerSequence = readerSequence; this.sparkSchema = schema; this.projection = UnsafeProjection.create(schema); this.path = path; @@ -156,7 +156,7 @@ public final class LanceRecordIterator implements ClosableIterator<UnsafeRow> { StructType schema, String path, BlobDescriptorTransform blobTransform) { - ArrowReaderSupplier supplier = new ArrowReaderSupplier() { + ArrowReaderSequence sequence = new ArrowReaderSequence() { private long nextStart = 0; @Override @@ -172,11 +172,11 @@ public final class LanceRecordIterator implements ClosableIterator<UnsafeRow> { return lanceReader.readAll(columnNames, ranges, BLOB_READ_CHUNK_ROWS, readOpts); } }; - return new LanceRecordIterator(allocator, lanceReader, supplier, schema, path, blobTransform); + return new LanceRecordIterator(allocator, lanceReader, sequence, schema, path, blobTransform); } - private static ArrowReaderSupplier singleReaderSupplier(ArrowReader arrowReader) { - return new ArrowReaderSupplier() { + private static ArrowReaderSequence singleReaderSequence(ArrowReader arrowReader) { + return new ArrowReaderSequence() { private ArrowReader remaining = arrowReader; @Override @@ -202,7 +202,7 @@ public final class LanceRecordIterator implements ClosableIterator<UnsafeRow> { try { while (true) { if (currentReader == null) { - currentReader = readerSupplier.next(); + currentReader = readerSequence.next(); if (currentReader == null) { return false; } 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 d2b4e1c620d0..c3576bfe4f1d 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 @@ -140,17 +140,17 @@ class SparkLanceReaderBase(enableVectorizedReader: Boolean) extends SparkColumna // 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 } // 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. - lanceIterator = if (!blobFieldNames.isEmpty) { + lanceIterator = if (blobFieldNames.nonEmpty) { LanceRecordIterator.chunkedBlobReader( allocator, lanceReader, columnNames, readOpts, lanceReader.numRows(), iteratorSchema, filePath, blobTransform)
