hudi-agent commented on code in PR #19897:
URL: https://github.com/apache/hudi/pull/19897#discussion_r3989748519
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/lance/SparkLanceReaderBase.scala:
##########
@@ -78,141 +79,171 @@ 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
+ }
+ // Empty projections (e.g. COUNT(*), partition-only queries, or missing
columns under
+ // schema evolution) can use the Lance metadata row count without
reading data columns.
+ lanceIterator = if (iteratorSchema.isEmpty) {
+ new ClosableIterator[UnsafeRow] {
Review Comment:
🤖 nit: could you pull this ~30-line anonymous iterator out into a small
named class (e.g. `LanceRowCountIterator(lanceReader, allocator)`), next to
`LanceRecordIterator`? `read` is already long and the inline class makes the
three-way `if/else if/else` hard to scan; a name also documents that it only
yields `numRows()` empty rows.
<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]
