hudi-agent commented on code in PR #19897:
URL: https://github.com/apache/hudi/pull/19897#discussion_r3986025272
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/lance/SparkLanceReaderBase.scala:
##########
@@ -78,141 +79,165 @@ class SparkLanceReaderBase(enableVectorizedReader:
Boolean) extends SparkColumna
val filePath = file.filePath.toString
- if (requiredSchema.isEmpty && partitionSchema.isEmpty) {
- // No columns requested - return empty iterator
- Iterator.empty
- } else {
- // Track iterator for cleanup. Typed as ClosableIterator so we can swap
in the
- // DESCRIPTOR-mode iterator when the user opts into that blob read mode.
- var lanceIterator: ClosableIterator[UnsafeRow] = null
-
- // Create child allocator for reading
- val dataAllocatorSize = storageConf.unwrap().getLong(
- HoodieStorageConfig.LANCE_READ_ALLOCATOR_SIZE_BYTES.key(),
-
HoodieStorageConfig.LANCE_READ_ALLOCATOR_SIZE_BYTES.defaultValue().toLong)
- val allocator = HoodieArrowAllocator.newChildAllocator(
- getClass.getSimpleName + "-data-" + filePath, dataAllocatorSize)
-
- try {
- // Open Lance file reader
- val lanceReader = LanceFileReader.open(filePath, allocator)
-
- // Get schema from Lance file. lance-spark strips Hudi's VECTOR
descriptor during
- // Arrow→Spark conversion but keeps the fixed-size-list dimension on
the Spark
- // field metadata; rebuild the descriptor from that, using
requiredSchema
- // as the source of truth for which columns are Hudi VECTORs — so
non-Hudi fixed-size-lists aren't mis-tagged.
- val arrowSchema = lanceReader.schema()
- val vectorColumnNames: java.util.Set[String] = VectorConversionUtils
- .detectVectorColumnsFromMetadata(requiredSchema)
- .keySet()
- .asScala
- .map(i => requiredSchema.fields(i).name)
- .toSet
- .asJava
- val fileSchema = VectorConversionUtils.restoreVectorMetadata(
- LanceArrowUtils.fromArrowSchema(arrowSchema), vectorColumnNames)
-
- // Build type change info for schema evolution
- val (implicitTypeChangeInfo, sparkRequestSchema) =
- SparkSchemaTransformUtils.buildImplicitSchemaChangeInfo(fileSchema,
requiredSchema)
-
- // Filter schema to only fields that exist in file (Lance can only
read columns present in file).
- val requestSchema =
-
SparkSchemaTransformUtils.filterSchemaByFileSchema(sparkRequestSchema,
fileSchema)
-
- // Lance returns null BLOB sub-structs as non-null parents with null
children; widen
- // nullability inside BLOB subtrees so the codegen projection doesn't
NPE on them.
- val iteratorSchema = widenBlobSubtreeNullability(requestSchema)
-
- val columnNames = if (iteratorSchema.nonEmpty) {
- iteratorSchema.fieldNames.toList.asJava
- } else {
- // If only partition columns requested, read minimal data
- null
- }
+ // Track iterator for cleanup. Typed as ClosableIterator so we can swap in
the
+ // DESCRIPTOR-mode iterator when the user opts into that blob read mode.
+ var lanceIterator: ClosableIterator[UnsafeRow] = null
+ var lanceReader: LanceFileReader = null
+
+ // Create child allocator for reading
+ val dataAllocatorSize = storageConf.unwrap().getLong(
+ HoodieStorageConfig.LANCE_READ_ALLOCATOR_SIZE_BYTES.key(),
+
HoodieStorageConfig.LANCE_READ_ALLOCATOR_SIZE_BYTES.defaultValue().toLong)
+ val allocator = HoodieArrowAllocator.newChildAllocator(
+ getClass.getSimpleName + "-data-" + filePath, dataAllocatorSize)
+
+ try {
+ // Open Lance file reader
+ lanceReader = LanceFileReader.open(filePath, allocator)
+
+ // Get schema from Lance file. lance-spark strips Hudi's VECTOR
descriptor during
+ // Arrow→Spark conversion but keeps the fixed-size-list dimension on the
Spark
+ // field metadata; rebuild the descriptor from that, using requiredSchema
+ // as the source of truth for which columns are Hudi VECTORs — so
non-Hudi fixed-size-lists aren't mis-tagged.
+ val arrowSchema = lanceReader.schema()
+ val vectorColumnNames: java.util.Set[String] = VectorConversionUtils
+ .detectVectorColumnsFromMetadata(requiredSchema)
+ .keySet()
+ .asScala
+ .map(i => requiredSchema.fields(i).name)
+ .toSet
+ .asJava
+ val fileSchema = VectorConversionUtils.restoreVectorMetadata(
+ LanceArrowUtils.fromArrowSchema(arrowSchema), vectorColumnNames)
+
+ // Build type change info for schema evolution
+ val (implicitTypeChangeInfo, sparkRequestSchema) =
+ SparkSchemaTransformUtils.buildImplicitSchemaChangeInfo(fileSchema,
requiredSchema)
+
+ // Filter schema to only fields that exist in file (Lance can only read
columns present in file).
+ val requestSchema =
+ SparkSchemaTransformUtils.filterSchemaByFileSchema(sparkRequestSchema,
fileSchema)
+
+ // Lance returns null BLOB sub-structs as non-null parents with null
children; widen
+ // nullability inside BLOB subtrees so the codegen projection doesn't
NPE on them.
+ val iteratorSchema = widenBlobSubtreeNullability(requestSchema)
+
+ val columnNames = iteratorSchema.fieldNames.toList.asJava
+
+ // 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()
+
+ // 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: 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
+ }
+ // For empty projections (e.g. COUNT(*), partition-only queries, or
missing columns under schema
+ // evolution), use metadata-only row count without reading data columns.
For BLOB-containing reads,
+ // drain the file in <=512-row range chunks to avoid JNI aborts.
Otherwise, keep the single streamed reader.
+ lanceIterator = if (iteratorSchema.isEmpty) {
+ new ClosableIterator[UnsafeRow] {
+ private var remaining = lanceReader.numRows()
+ private val emptyRow = {
+ val r = new UnsafeRow(0)
+ r.pointTo(new Array[Byte](0), 0)
+ r
+ }
- // 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()
-
- // 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: 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.
- // 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)
- }
+ override def hasNext: Boolean = remaining > 0
- // Register cleanup listener
- Option(TaskContext.get()).foreach { ctx =>
- ctx.addTaskCompletionListener[Unit](_ => lanceIterator.close())
- }
+ override def next(): UnsafeRow = {
+ if (remaining <= 0) {
+ throw new NoSuchElementException("No more records available")
+ }
+ remaining -= 1
+ emptyRow
+ }
- val baseIter: Iterator[InternalRow] = lanceIterator.asScala
-
- // Create the following projections for schema evolution:
- // 1. Padding projection: add NULL for missing columns
- // 2. Casting projection: handle type conversions
- val schemaUtils = sparkAdapter.getSchemaUtils
- val paddingProj =
SparkSchemaTransformUtils.generateNullPaddingProjection(iteratorSchema,
requiredSchema)
- val castProj = SparkSchemaTransformUtils.generateUnsafeProjection(
- schemaUtils.toAttributes(requiredSchema),
- Some(SQLConf.get.sessionLocalTimeZone),
- implicitTypeChangeInfo,
- requiredSchema,
- new StructType(),
- schemaUtils
- )
-
- // Unify projections by applying padding and then casting for each row
- val projection: UnsafeProjection = new UnsafeProjection {
- def apply(row: InternalRow): UnsafeRow =
- castProj(paddingProj(row))
+ override def close(): Unit = {
Review Comment:
🤖 Could this `close()` get called twice? If anything throws after
`lanceIterator` is assigned and the task-completion listener is registered
(e.g. codegen failure in `generateUnsafeProjection`), the catch block closes it
and then the listener closes it again at task end. `LanceFileReader.close()` in
lance-core 4.0.0 calls `closeNative(handle)` unconditionally without zeroing
the handle, so the second call hits JNI with a freed handle.
`LanceRecordIterator` guards this with a `closed` flag — it might be worth
adding the same guard here (and reusing `LanceResourceCloser.closeAll(null,
null, lanceReader, allocator)` so the checked `Exception` from
`lanceReader.close()` is wrapped, matching the `ClosableIterator.close()`
contract of not throwing).
<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]
