hudi-agent commented on code in PR #19181:
URL: https://github.com/apache/hudi/pull/19181#discussion_r3522623363
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/LanceRecordIterator.java:
##########
@@ -32,39 +32,54 @@
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 512-row
BLOB page size. */
+ public static final int BLOB_READ_CHUNK_ROWS = 512;
+
+ /**
+ * Supplies 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}.
+ */
Review Comment:
🤖 nit: `ArrowReaderSupplier` borrows the `Supplier<T>` name from
`java.util.function`, which conventionally has a `get()` method and is usable
as a lambda — this interface throws `IOException` and uses `next()`, so it
doesn't fit that mental model. Something like `ArrowReaderSequence` or
`ReaderChain` might avoid the false association for future readers of this
class.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/lance/SparkLanceReaderBase.scala:
##########
@@ -148,8 +147,18 @@ class SparkLanceReaderBase(enableVectorizedReader:
Boolean) extends SparkColumna
} 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.
+ lanceIterator = if (!blobFieldNames.isEmpty) {
Review Comment:
🤖 nit: `blobFieldNames.nonEmpty` is the idiomatic Scala spelling — reads as
a positive predicate without the leading negation to parse.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
