You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by yi...@apache.org on 2026/04/21 00:48:32 UTC
(hudi) branch master updated: feat(blob): Read Blobs in Spark SQL (#18098)
This is an automated email from the ASF dual-hosted git repository.
yihua 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 f35b69cf028e feat(blob): Read Blobs in Spark SQL (#18098)
f35b69cf028e is described below
commit f35b69cf028ec50dd13183e4272d1e28f16356de
Author: Tim Brown <ti...@generalintuition.com>
AuthorDate: Mon Apr 20 20:48:25 2026 -0400
feat(blob): Read Blobs in Spark SQL (#18098)
---
.../org/apache/spark/sql/hudi/SparkAdapter.scala | 11 +
.../spark/sql/hudi/blob/BatchedBlobRead.scala | 52 ++
.../spark/sql/hudi/blob/BatchedBlobReadExec.scala | 69 ++
.../spark/sql/hudi/blob/BatchedBlobReader.scala | 773 +++++++++++++++++++++
.../sql/hudi/blob/BatchedBlobReaderStrategy.scala | 61 ++
.../spark/sql/hudi/blob/ReadBlobExpression.scala | 48 ++
.../apache/spark/sql/hudi/blob/ReadBlobRule.scala | 173 +++++
.../spark/sql/hudi/blob/ScalarFunctions.scala | 95 +++
.../sql/hudi/HoodieSparkSessionExtension.scala | 2 +
.../spark/sql/hudi/analysis/HoodieAnalysis.scala | 3 +
.../org/apache/hudi/blob/BlobTestHelpers.scala | 6 +-
.../apache/hudi/blob/TestBatchedBlobReader.scala | 457 ++++++++++++
.../org/apache/hudi/blob/TestBlobSupport.scala | 7 +-
.../org/apache/hudi/blob/TestReadBlobSQL.scala | 439 ++++++++++++
.../apache/spark/sql/adapter/Spark3_3Adapter.scala | 11 +
.../apache/spark/sql/adapter/Spark3_4Adapter.scala | 11 +
.../apache/spark/sql/adapter/Spark3_5Adapter.scala | 12 +-
.../apache/spark/sql/adapter/Spark4_0Adapter.scala | 11 +
18 files changed, 2236 insertions(+), 5 deletions(-)
diff --git a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
index 0bf6611a911d..85db88c93506 100644
--- a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
+++ b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
@@ -76,6 +76,17 @@ trait SparkAdapter extends Serializable {
*/
def injectTableFunctions(extensions: SparkSessionExtensions): Unit = {}
+ /**
+ * Inject scalar functions into Spark SQL function registry.
+ * These functions can be used in SQL SELECT clauses.
+ */
+ def injectScalarFunctions(extensions: SparkSessionExtensions): Unit
+
+ /**
+ * Inject planner strategies to SparkSessionExtensions for converting custom logical plans into physical plans.
+ */
+ def injectPlannerStrategies(extensions: SparkSessionExtensions): Unit
+
/**
* Returns an instance of [[HoodieCatalystExpressionUtils]] providing for common utils operating
* on Catalyst [[Expression]]s
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/BatchedBlobRead.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/BatchedBlobRead.scala
new file mode 100644
index 000000000000..2d1b4f3da58a
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/BatchedBlobRead.scala
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.hudi.blob
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, AttributeSet, Expression}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, UnaryNode}
+import org.apache.spark.sql.hudi.blob.BatchedBlobReader.DATA_COL
+import org.apache.spark.sql.types.BinaryType
+
+/**
+ * Logical plan node for batched blob reading.
+ *
+ * Created by [[ReadBlobRule]] when `read_blob()` is detected in queries.
+ * Converted to [[BatchedBlobReadExec]] by [[BatchedBlobReaderStrategy]] during physical planning.
+ *
+ * @param child Child logical plan
+ * @param blobAttr The blob column attribute to read from
+ * @param dataAttr The output attribute for resolved blob data
+ */
+case class BatchedBlobRead(
+ child: LogicalPlan,
+ blobAttr: Attribute,
+ dataAttr: AttributeReference =
+ AttributeReference(DATA_COL, BinaryType, nullable = true)())
+ extends UnaryNode {
+
+ override def output: Seq[Attribute] = child.output :+ dataAttr
+
+ override protected def withNewChildInternal(newChild: LogicalPlan): LogicalPlan = {
+ copy(child = newChild)
+ }
+
+ override def producedAttributes: AttributeSet = AttributeSet(dataAttr)
+
+}
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/BatchedBlobReadExec.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/BatchedBlobReadExec.scala
new file mode 100644
index 000000000000..8967e151de3b
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/BatchedBlobReadExec.scala
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.hudi.blob
+
+import org.apache.hudi.storage.StorageConfiguration
+
+import org.apache.spark.broadcast.Broadcast
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.execution.{SparkPlan, UnaryExecNode}
+
+/**
+ * Physical plan node that executes batched blob reads.
+ *
+ * Reads blob data from storage using [[BatchedBlobReader]] to batch
+ * reads efficiently when data is sorted by file and position.
+ *
+ * @param child Child physical plan
+ * @param maxGapBytes Maximum gap between reads to batch (from config)
+ * @param storageConf Storage configuration for file access
+ * @param lookaheadSize Number of rows to buffer for batch detection
+ * @param logical The logical plan node this was created from
+ */
+case class BatchedBlobReadExec(child: SparkPlan,
+ maxGapBytes: Int,
+ storageConf: StorageConfiguration[_],
+ lookaheadSize: Int,
+ logical: BatchedBlobRead)
+ extends UnaryExecNode {
+
+ override def output: Seq[Attribute] = logical.output
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ val childRDD = child.execute()
+ // Broadcast storageConf to avoid per-task serialization
+ val broadcastConf: Broadcast[StorageConfiguration[_]] = childRDD.sparkContext.broadcast(storageConf)
+ // Use direct RDD processing - no DataFrame conversion!
+ BatchedBlobReader.processRDD(
+ childRDD,
+ child.schema,
+ broadcastConf,
+ maxGapBytes,
+ lookaheadSize,
+ logical.blobAttr.name
+ )
+ }
+
+ override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan = {
+ copy(child = newChild)
+ }
+}
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/BatchedBlobReader.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/BatchedBlobReader.scala
new file mode 100644
index 000000000000..356323ea3fae
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/BatchedBlobReader.scala
@@ -0,0 +1,773 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.hudi.blob
+
+import org.apache.hudi.HoodieSparkUtils.sparkAdapter
+import org.apache.hudi.common.schema.{HoodieSchema, HoodieSchemaType}
+import org.apache.hudi.io.SeekableDataInputStream
+import org.apache.hudi.storage.{HoodieStorage, HoodieStorageUtils, StorageConfiguration, StoragePath}
+
+import org.apache.spark.TaskContext
+import org.apache.spark.broadcast.Broadcast
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.{Dataset, Row}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{GenericRowWithSchema, SpecificInternalRow}
+import org.apache.spark.sql.types.{BinaryType, BlobType, DataType, StructField, StructType}
+import org.slf4j.LoggerFactory
+
+import java.io.InputStream
+
+import scala.collection.mutable.ArrayBuffer
+
+/**
+ * Batched byte range reader that optimizes I/O by combining consecutive reads for out-of-line data.
+ *
+ * This reader analyzes sequences of read requests within a partition and merges
+ * consecutive or nearby reads into single I/O operations. This significantly reduces
+ * the number of seeks and reads when processing sorted data.
+ *
+ * <h3>Schema Requirement:</h3>
+ * The blob column must match the schema defined in {@link HoodieSchema.Blob}:
+ * <pre>
+ * struct {
+ * type: string // "inline" or "out_of_line"
+ * data: binary (nullable) // inline data (null for out_of_line)
+ * reference: struct (nullable) { // file reference (null for inline)
+ * external_path: string
+ * offset: long
+ * length: long
+ * managed: boolean
+ * }
+ * }
+ * </pre>
+ *
+ * <h3>Key Features:</h3>
+ * <ul>
+ * <li>Batches consecutive reads from the same file</li>
+ * <li>Configurable gap threshold for merging nearby reads</li>
+ * <li>Lookahead buffer to identify batch opportunities</li>
+ * <li>Preserves input row order in output</li>
+ * </ul>
+ *
+ * <h3>Usage Example:</h3>
+ * {{{
+ * import org.apache.hudi.udf.BatchedByteRangeReader
+ * import org.apache.spark.sql.functions._
+ * // Read a table with a blob column (e.g. image_data)
+ * val df = spark.read.format("hudi").load("/my_path").select("image_data", "record_id")
+ *
+ * // Read with batching (best when data is sorted by external_path, offset)
+ * val result = BatchedByteRangeReader.readBatched(df, structColName = "file_info")
+ *
+ * // Result has: image_data, record_id, data
+ * result.show()
+ * }}}
+ *
+ * <h3>Performance Tips:</h3>
+ * <ul>
+ * <li>Sort input by (blob.reference.external_path, blob.reference.offset) for maximum batching effectiveness</li>
+ * <li>Increase lookaheadSize for better batch detection (at cost of memory)</li>
+ * <li>Tune maxGapBytes based on your data access patterns</li>
+ * </ul>
+ *
+ * @param storage HoodieStorage instance for file I/O
+ * @param maxGapBytes Maximum gap between ranges to consider for batching (default: 4KB)
+ * @param lookaheadRows Number of rows to buffer for batch detection (default: 50)
+ */
+class BatchedBlobReader(
+ storage: HoodieStorage,
+ maxGapBytes: Int = 4096,
+ lookaheadRows: Int = 50) {
+
+ private val logger = LoggerFactory.getLogger(classOf[BatchedBlobReader])
+
+ /**
+ * Process a partition iterator, batching consecutive reads.
+ *
+ * This method consumes the input iterator and produces an output iterator
+ * with each row containing the original data plus a "data" column with the
+ * bytes read from the file.
+ *
+ * @param rows Iterator of input rows with struct column
+ * @param structColIdx Index of the struct column in the row
+ * @param outputSchema Schema for output rows
+ * @param accessor Type class for accessing row fields
+ * @param builder Type class for building output rows
+ * @tparam R Row type (Row or InternalRow)
+ * @return Iterator of output rows with data column added
+ */
+ def processPartition[R](
+ rows: Iterator[R],
+ structColIdx: Int,
+ outputSchema: StructType)
+ (implicit accessor: RowAccessor[R], builder: RowBuilder[R]): Iterator[R] = {
+
+ // Create buffered iterator for lookahead
+ val bufferedRows = rows.buffered
+
+ // Result buffer to maintain order
+ val resultIterator = new Iterator[R] {
+ private var currentBatch: Iterator[R] = Iterator.empty
+ private var rowIndex = 0L
+
+ override def hasNext: Boolean = {
+ if (currentBatch.hasNext) {
+ true
+ } else if (bufferedRows.hasNext) {
+ // Process next batch
+ currentBatch = processNextBatch()
+ currentBatch.hasNext
+ } else {
+ false
+ }
+ }
+
+ override def next(): R = {
+ if (!hasNext) {
+ throw new NoSuchElementException("No more rows")
+ }
+ currentBatch.next()
+ }
+
+ /**
+ * Collect and process the next batch of rows.
+ */
+ private def processNextBatch(): Iterator[R] = {
+ // Collect up to lookaheadSize rows with their original indices
+ val batch = collectBatch()
+
+ if (batch.isEmpty) {
+ Iterator.empty
+ } else {
+ // Partition the batch into three groups
+ val (inlineRows, outOfLineRows) = batch.partition(_.inlineBytes.isDefined)
+ val (wholeFileRows, rangeRows) = outOfLineRows.partition(_.length < 0)
+
+ // Case 1: Inline — return bytes directly without I/O
+ val inlineResults = inlineRows.map { ri =>
+ RowResult(builder.buildRow(ri.originalRow, ri.inlineBytes.get, outputSchema), ri.index)
+ }
+
+ // Case 2: Whole-file reads
+ val wholeFileResults = wholeFileRows.map(readWholeFile(_, outputSchema))
+
+ // Case 3: Regular range reads — merge consecutive ranges and batch
+ val mergedRanges = identifyConsecutiveRanges(rangeRows)
+ val rangeResults = mergedRanges.flatMap(readAndSplitRange(_, outputSchema))
+
+ // Sort by original index to preserve input order
+ (inlineResults ++ wholeFileResults ++ rangeResults).sortBy(_.index).map(_.row).iterator
+ }
+ }
+
+ /**
+ * Collect up to lookaheadSize rows from the input iterator.
+ */
+ private def collectBatch(): Seq[RowInfo[R]] = {
+ val batch = ArrayBuffer[RowInfo[R]]()
+ var collected = 0
+
+ while (bufferedRows.hasNext && collected < lookaheadRows) {
+ val row = bufferedRows.next()
+ // Handle null struct column (null blob)
+ if (accessor.isNullAt(row, structColIdx)) {
+ batch += RowInfo[R](
+ originalRow = row,
+ filePath = "",
+ offset = -1,
+ length = -1,
+ index = rowIndex,
+ inlineBytes = Some(null)
+ )
+ rowIndex += 1
+ collected += 1
+ } else {
+ val blobStruct = accessor.getStruct(row, structColIdx, HoodieSchema.Blob.getFieldCount)
+ // Dispatch based on storage_type (field 0)
+ val storageType = accessor.getString(blobStruct, 0)
+ if (storageType == HoodieSchema.Blob.INLINE) {
+ // Case 1: Inline — bytes are in field 1
+ val bytes = accessor.getBytes(blobStruct, 1)
+ batch += RowInfo[R](
+ originalRow = row,
+ filePath = "",
+ offset = -1,
+ length = -1,
+ index = rowIndex,
+ inlineBytes = Some(bytes)
+ )
+ } else if (storageType == HoodieSchema.Blob.OUT_OF_LINE) {
+ // Case 2 or 3: Out-of-line — get reference struct (field 2)
+ require(!accessor.isNullAt(blobStruct, 2), s"Out-of-line blob at row $rowIndex must set reference")
+ val referenceStruct = accessor.getStruct(blobStruct, 2, HoodieSchema.Blob.getReferenceFieldCount)
+ val filePath = accessor.getString(referenceStruct, 0)
+ require(filePath != null && filePath.nonEmpty, s"Blob reference must have non-empty external_path at row $rowIndex")
+ val offsetIsNull = accessor.isNullAt(referenceStruct, 1)
+ val lengthIsNull = accessor.isNullAt(referenceStruct, 2)
+ if (offsetIsNull && lengthIsNull) {
+ // Case 2: Whole-file read — no offset/length specified; sentinel length = -1
+ batch += RowInfo[R](
+ originalRow = row,
+ filePath = filePath,
+ offset = 0,
+ length = -1,
+ index = rowIndex
+ )
+ } else if (offsetIsNull || lengthIsNull) {
+ throw new IllegalArgumentException(s"Blob reference for '$filePath' must set both offset and length, or neither")
+ } else {
+ // Case 3: Regular range read
+ val offset = accessor.getLong(referenceStruct, 1)
+ val length = accessor.getLong(referenceStruct, 2)
+ require(offset >= 0, s"Blob offset must be non-negative for '$filePath': $offset")
+ require(length >= 0, s"Blob length must be non-negative for '$filePath': $length")
+ batch += RowInfo[R](
+ originalRow = row,
+ filePath = filePath,
+ offset = offset,
+ length = length,
+ index = rowIndex
+ )
+ }
+ } else {
+ throw new IllegalArgumentException(s"Unsupported blob storage_type at row $rowIndex: $storageType")
+ }
+ rowIndex += 1
+ collected += 1
+ }
+ }
+ batch.toSeq
+ }
+ }
+
+ resultIterator
+ }
+
+ /**
+ * Identify consecutive ranges that can be batched together.
+ *
+ * This method groups rows by file path, sorts by offset, and merges
+ * ranges that are consecutive or within maxGapBytes of each other.
+ *
+ * @param rows Sequence of row information
+ * @return Sequence of merged ranges
+ */
+ private def identifyConsecutiveRanges[R](rows: Seq[RowInfo[R]]): Seq[MergedRange[R]] = {
+ // Group by file path
+ val byFile = rows.groupBy(_.filePath)
+
+ val allRanges = ArrayBuffer[MergedRange[R]]()
+
+ byFile.foreach { case (filePath, fileRows) =>
+ // Sort by offset
+ val sorted = fileRows.sortBy(_.offset)
+
+ // Merge consecutive ranges
+ val merged = mergeRanges(sorted, maxGapBytes)
+ allRanges ++= merged
+ }
+
+ allRanges.toSeq
+ }
+
+ /**
+ * Merge consecutive ranges within the gap threshold.
+ *
+ * @param rows Sorted rows from the same file
+ * @param maxGap Maximum gap to consider for merging
+ * @return Sequence of merged ranges
+ */
+ private def mergeRanges[R](rows: Seq[RowInfo[R]], maxGap: Int): Seq[MergedRange[R]] = {
+
+ val result = ArrayBuffer[MergedRange[R]]()
+ var currentFilePath: String = null
+ var currentStartOffset: Long = 0L
+ var currentEndOffset: Long = 0L
+ var currentRows: ArrayBuffer[RowInfo[R]] = null
+
+ rows.foreach { row =>
+ if (currentRows == null) {
+ // Start first range
+ currentFilePath = row.filePath
+ currentStartOffset = row.offset
+ currentEndOffset = row.offset + row.length
+ currentRows = ArrayBuffer(row)
+ } else {
+ val gap = row.offset - currentEndOffset
+ // Check for overlap
+ if (row.offset < currentEndOffset) {
+ throw new IllegalArgumentException(
+ s"Overlapping blob ranges detected: previous range [${currentStartOffset}, ${currentEndOffset}) and current row [${row.offset}, ${row.offset + row.length}) in file ${row.filePath}"
+ )
+ }
+ if (gap >= 0 && gap <= maxGap) {
+ // Merge into current range
+ currentEndOffset = math.max(currentEndOffset, row.offset + row.length)
+ currentRows += row
+ } else {
+ // Save current range and start new one
+ result += MergedRange[R](
+ filePath = currentFilePath,
+ startOffset = currentStartOffset,
+ endOffset = currentEndOffset,
+ rows = currentRows.toSeq
+ )
+ currentFilePath = row.filePath
+ currentStartOffset = row.offset
+ currentEndOffset = row.offset + row.length
+ currentRows = ArrayBuffer(row)
+ }
+ }
+ }
+
+ // Add final range
+ if (currentRows != null) {
+ result += MergedRange[R](
+ filePath = currentFilePath,
+ startOffset = currentStartOffset,
+ endOffset = currentEndOffset,
+ rows = currentRows.toSeq
+ )
+ }
+
+ result.toSeq
+ }
+
+ /**
+ * Read an entire file and return it as a single row result.
+ *
+ * Used for whole-file out-of-line blobs where no offset or length is specified.
+ *
+ * @param rowInfo Row information with the file path
+ * @param outputSchema Schema for output rows
+ * @param builder Type class for building output rows
+ * @tparam R Row type (Row or InternalRow)
+ * @return Sequence containing a single row result
+ */
+ private def readWholeFile[R](
+ rowInfo: RowInfo[R],
+ outputSchema: StructType)
+ (implicit builder: RowBuilder[R]): RowResult[R] = {
+
+ var inputStream: InputStream = null
+ try {
+ val path = new StoragePath(rowInfo.filePath)
+ inputStream = storage.open(path)
+ val buffer = inputStream.readAllBytes()
+
+ logger.debug(s"Read entire file ${rowInfo.filePath} (${buffer.length} bytes)")
+
+ RowResult(builder.buildRow(rowInfo.originalRow, buffer, outputSchema), rowInfo.index)
+ } finally {
+ if (inputStream != null) {
+ try {
+ inputStream.close()
+ } catch {
+ case e: Exception =>
+ logger.warn(s"Error closing stream for ${rowInfo.filePath}", e)
+ }
+ }
+ }
+ }
+
+ /**
+ * Read a merged range and split it back into individual row results.
+ *
+ * This method performs a single I/O operation to read the entire merged
+ * range, then splits the buffer into individual results for each original
+ * row.
+ *
+ * @param range The merged range to read
+ * @param outputSchema Schema for output rows
+ * @param builder Type class for building output rows
+ * @tparam R Row type (Row or InternalRow)
+ * @return Sequence of row results with original indices
+ */
+ private def readAndSplitRange[R](
+ range: MergedRange[R],
+ outputSchema: StructType)
+ (implicit builder: RowBuilder[R]): Seq[RowResult[R]] = {
+
+ var inputStream: SeekableDataInputStream = null
+ try {
+ // Get or open file handle
+ inputStream = storage.openSeekable(new StoragePath(range.filePath), false)
+
+ // Seek to start offset
+ inputStream.seek(range.startOffset)
+
+ // Read the entire merged range
+ val totalLengthLong = range.endOffset - range.startOffset
+ require(totalLengthLong >= 0 && totalLengthLong <= Int.MaxValue, s"Range too large: $totalLengthLong bytes exceeds Int.MaxValue")
+ val totalLength = totalLengthLong.toInt
+ val buffer = new Array[Byte](totalLength)
+ inputStream.readFully(buffer, 0, totalLength)
+
+ logger.debug(
+ s"Read ${totalLength} bytes from ${range.filePath} at offset ${range.startOffset} " +
+ s"for ${range.rows.length} rows"
+ )
+
+ // Split buffer into individual results
+ range.rows.map { rowInfo =>
+ val relativeOffset = (rowInfo.offset - range.startOffset).toInt
+ val data = buffer.slice(relativeOffset, relativeOffset + rowInfo.length.toInt)
+
+ // Build output row using type class
+ val outputRow = builder.buildRow(rowInfo.originalRow, data, outputSchema)
+
+ RowResult[R](
+ row = outputRow,
+ index = rowInfo.index
+ )
+ }
+
+ } catch {
+ case e: Exception =>
+ logger.error(
+ s"Failed to read range from ${range.filePath} " +
+ s"at offset ${range.startOffset}, length ${range.endOffset - range.startOffset}",
+ e
+ )
+ throw e
+ } finally {
+ if (inputStream != null) {
+ try {
+ inputStream.close()
+ } catch {
+ case e: Exception =>
+ logger.warn(s"Error closing input stream for ${range.filePath}", e)
+ }
+ }
+ }
+ }
+}
+
+/**
+ * Type class for accessing row fields.
+ * Abstracts over Row and InternalRow API differences.
+ *
+ * @tparam R Row type (Row or InternalRow)
+ */
+private[blob] trait RowAccessor[R] {
+ def getStruct(row: R, structColIdx: Int, numFields: Int): R
+ def getString(struct: R, fieldIdx: Int): String
+ def getLong(struct: R, fieldIdx: Int): Long
+ def getBytes(row: R, fieldIdx: Int): Array[Byte]
+ def isNullAt(row: R, fieldIdx: Int): Boolean
+}
+
+/**
+ * Type class for building output rows.
+ * Abstracts over Row and InternalRow construction.
+ *
+ * @tparam R Row type (Row or InternalRow)
+ */
+private[blob] trait RowBuilder[R] {
+ def buildRow(originalRow: R, data: Array[Byte], outputSchema: StructType): R
+}
+
+/**
+ * Type class instances for Row.
+ */
+private[blob] object RowAccessor {
+ implicit val rowAccessor: RowAccessor[Row] = new RowAccessor[Row] {
+ override def getStruct(row: Row, structColIdx: Int, numFields: Int): Row = row.getStruct(structColIdx)
+ override def getString(struct: Row, fieldIdx: Int): String = struct.getString(fieldIdx)
+ override def getLong(struct: Row, fieldIdx: Int): Long = struct.getLong(fieldIdx)
+ override def getBytes(row: Row, fieldIdx: Int): Array[Byte] = row.getAs[Array[Byte]](fieldIdx)
+ override def isNullAt(row: Row, fieldIdx: Int): Boolean = row.isNullAt(fieldIdx)
+ }
+
+ implicit val internalRowAccessor: RowAccessor[InternalRow] = new RowAccessor[InternalRow] {
+ override def getStruct(row: InternalRow, structColIdx: Int, numFields: Int): InternalRow = row.getStruct(structColIdx, numFields)
+ override def getString(struct: InternalRow, fieldIdx: Int): String = {
+ val utf8String = struct.getUTF8String(fieldIdx)
+ if (utf8String == null) null else utf8String.toString
+ }
+ override def getLong(struct: InternalRow, fieldIdx: Int): Long = struct.getLong(fieldIdx)
+ override def getBytes(row: InternalRow, fieldIdx: Int): Array[Byte] = row.getBinary(fieldIdx)
+ override def isNullAt(row: InternalRow, fieldIdx: Int): Boolean = row.isNullAt(fieldIdx)
+ }
+}
+
+/**
+ * Type class instances for Row builders.
+ */
+private[blob] object RowBuilder {
+ implicit val rowBuilder: RowBuilder[Row] = new RowBuilder[Row] {
+ override def buildRow(originalRow: Row, data: Array[Byte], outputSchema: StructType): Row = {
+ val outputValues = new Array[Any](originalRow.length + 1)
+ var i = 0
+ while (i < originalRow.length) {
+ outputValues(i) = originalRow.get(i)
+ i += 1
+ }
+ outputValues(originalRow.length) = data
+ new GenericRowWithSchema(outputValues, outputSchema)
+ }
+ }
+
+ implicit val internalRowBuilder: RowBuilder[InternalRow] = new RowBuilder[InternalRow] {
+ override def buildRow(originalRow: InternalRow, data: Array[Byte], outputSchema: StructType): InternalRow = {
+ val outputRow = new SpecificInternalRow(outputSchema.fields.map(_.dataType))
+ var i = 0
+ while (i < originalRow.numFields) {
+ if (originalRow.isNullAt(i)) {
+ outputRow.setNullAt(i)
+ } else {
+ val dataType = outputSchema.fields(i).dataType
+ // Copy field using generic get/update for compatibility
+ outputRow.update(i, InternalRow.copyValue(originalRow.get(i, dataType)))
+ }
+ i += 1
+ }
+ // Set the data field (last position)
+ outputRow.update(originalRow.numFields, data)
+ outputRow
+ }
+ }
+}
+
+/**
+ * Information about a single row to be read.
+ *
+ * @param originalRow Original input row
+ * @param filePath Path to the file
+ * @param offset Byte offset in file
+ * @param length Number of bytes to read
+ * @param index Original position in input (for ordering)
+ * @tparam R Row type (Row or InternalRow)
+ */
+private case class RowInfo[R](
+ originalRow: R,
+ filePath: String,
+ offset: Long,
+ length: Long,
+ index: Long,
+ inlineBytes: Option[Array[Byte]] = None)
+
+/**
+ * A merged range combining multiple consecutive reads.
+ *
+ * @param filePath Path to the file
+ * @param startOffset Start byte offset of merged range
+ * @param endOffset End byte offset of merged range (exclusive)
+ * @param rows Individual rows included in this range
+ * @tparam R Row type (Row or InternalRow)
+ */
+private case class MergedRange[R](
+ filePath: String,
+ startOffset: Long,
+ endOffset: Long,
+ rows: Seq[RowInfo[R]]) {
+}
+
+/**
+ * Result row with its original index for ordering.
+ *
+ * @param row Output row with data
+ * @param index Original position in input
+ * @tparam R Row type (Row or InternalRow)
+ */
+private case class RowResult[R](
+ row: R,
+ index: Long)
+
+/**
+ * Companion object providing the main API for batched byte range reading.
+ */
+object BatchedBlobReader {
+
+ val MAX_GAP_BYTES_CONF = "hoodie.blob.batching.max.gap.bytes"
+ val LOOKAHEAD_SIZE_CONF = "hoodie.blob.batching.lookahead.size"
+
+ /** Default maximum gap to consider for batching */
+ val DEFAULT_MAX_GAP_BYTES = 4096
+
+ /** Default lookahead buffer size */
+ val DEFAULT_LOOKAHEAD_SIZE = 50
+
+ val DATA_COL = "__temp__data"
+
+ /**
+ * Read byte ranges from a DataFrame with a Blob column.
+ *
+ * The struct column must contain the HoodieSchema.Blob structure:
+ * - storage_type (String): "out_of_line" or "inline"
+ * - bytes (Binary, nullable): inline blob data or null
+ * - reference (Struct, nullable): out-of-line reference with external_path, offset, length, and managed fields
+ * Returns a DataFrame with all original columns plus a "data" column containing byte arrays.
+ *
+ * For best performance, sort the input DataFrame by the struct fields (file_path, offset)
+ * before calling this method.
+ *
+ * @param df Input DataFrame with struct column
+ * @param storageConf Storage configuration for file access
+ * @param maxGapBytes Max gap to consider consecutive (default: 4096)
+ * @param lookaheadSize Rows to buffer for batching (default: 50)
+ * @param columnName Optional column name to resolve. If not provided, searches for column with hudi_blob=true metadata
+ * @param keepTempColumn If true, keeps __temp__data column; if false (default), renames it to original column name
+ * @return DataFrame with struct column + data column
+ * @throws IllegalArgumentException if struct column is missing or has wrong schema
+ */
+ def readBatched(
+ df: Dataset[Row],
+ storageConf: StorageConfiguration[_],
+ maxGapBytes: Int = DEFAULT_MAX_GAP_BYTES,
+ lookaheadSize: Int = DEFAULT_LOOKAHEAD_SIZE,
+ columnName: Option[String] = None,
+ keepTempColumn: Boolean = false): Dataset[Row] = {
+
+ require(maxGapBytes >= 0, "maxGapBytes must be non-negative")
+ require(lookaheadSize > 0, "lookaheadSize must be positive")
+
+ val spark = df.sparkSession
+
+ // Get struct column index - use provided column name or fallback to metadata search
+ val (structColIdx, structColName) = columnName match {
+ case Some(name) =>
+ // Use provided column name directly
+ val idx = df.schema.fieldIndex(name)
+ (idx, name)
+ case None =>
+ // Fallback to metadata-based inference
+ getBlobColumn(df.schema)
+ }
+
+ val structField = df.schema(structColIdx)
+ require(isCompatibleBlobType(structField.dataType),
+ s"Blob column '$structColName' must be compatible with BlobType (type, data, reference struct), found: ${structField.dataType}")
+
+ // Create output schema (input + data column)
+ val outputSchema = df.schema.add(StructField(DATA_COL, BinaryType, nullable = true))
+
+ // Broadcast storage configuration
+ val broadcastConf = spark.sparkContext.broadcast(storageConf)
+
+ // Apply mapPartitions
+ val result = df.mapPartitions { partition =>
+ // Create storage and reader for this partition
+ val storage = HoodieStorageUtils.getStorage(broadcastConf.value)
+ val reader = new BatchedBlobReader(storage, maxGapBytes, lookaheadSize)
+
+ // Import implicit instances for Row
+ import RowAccessor.rowAccessor
+ import RowBuilder.rowBuilder
+
+ // Process partition
+ val iter = reader.processPartition[Row](partition, structColIdx, outputSchema)
+ TaskContext.get().addTaskCompletionListener[Unit](_ => storage.close())
+ iter
+ } (sparkAdapter.getCatalystExpressionUtils.getEncoder(outputSchema))
+
+ if (keepTempColumn) {
+ // Keep both columns for ReadBlobRule
+ result
+ } else {
+ // Backwards compatible behavior: rename __temp__data to original column name
+ result.drop(structColName).withColumnRenamed(DATA_COL, structColName)
+ }
+ }
+
+ /**
+ * Process RDD[InternalRow] directly without DataFrame conversion.
+ *
+ * This method provides optimized processing for physical plan execution,
+ * avoiding the overhead of RDD → DataFrame → RDD conversions.
+ *
+ * @param rdd Input RDD of InternalRows
+ * @param schema Schema of the input RDD
+ * @param storageConf Storage configuration for file access
+ * @param maxGapBytes Max gap to consider consecutive (default: 4096)
+ * @param lookaheadSize Rows to buffer for batching (default: 50)
+ * @param columnName Optional column name to resolve. If not provided, searches for column with hudi_blob=true metadata
+ * @return RDD with struct column + data column
+ * @throws IllegalArgumentException if struct column is missing or has wrong schema
+ */
+ def processRDD(
+ rdd: RDD[InternalRow],
+ schema: StructType,
+ broadcastConf: Broadcast[StorageConfiguration[_]],
+ maxGapBytes: Int = DEFAULT_MAX_GAP_BYTES,
+ lookaheadSize: Int = DEFAULT_LOOKAHEAD_SIZE,
+ columnName: String): RDD[InternalRow] = {
+
+ require(maxGapBytes >= 0, "maxGapBytes must be non-negative")
+ require(lookaheadSize > 0, "lookaheadSize must be positive")
+
+ // Get struct column index
+ val structColIdx = schema.fieldIndex(columnName)
+ val structField = schema(structColIdx)
+ require(isCompatibleBlobType(structField.dataType),
+ s"Blob column '$columnName' must be compatible with BlobType (type, data, reference struct), found: ${structField.dataType}")
+
+ // Create output schema (input + __temp__data column)
+ val outputSchema = schema.add(StructField(DATA_COL, BinaryType, nullable = true))
+
+ // Process partitions using InternalRow type classes
+ rdd.mapPartitions { partition =>
+ val storage = HoodieStorageUtils.getStorage(broadcastConf.value)
+
+ val reader = new BatchedBlobReader(storage, maxGapBytes, lookaheadSize)
+
+ // Import implicit instances for InternalRow
+ import RowAccessor.internalRowAccessor
+ import RowBuilder.internalRowBuilder
+
+ val iter = reader.processPartition[InternalRow](partition, structColIdx, outputSchema)
+ TaskContext.get().addTaskCompletionListener[Unit](_ => storage.close())
+ iter
+ }
+ }
+
+ /**
+ * Find the blob column by searching for metadata hudi_blob=true.
+ *
+ * This is a fallback method used when no explicit column name is provided.
+ * Searches through all fields in the schema to find the first field
+ * that has metadata with key "hudi_blob" set to boolean true.
+ *
+ * @param structType The schema to search
+ * @return The index of the blob column and the name of the column
+ * @throws IllegalArgumentException if no blob column is found
+ */
+ private def getBlobColumn(structType: StructType): (Int, String) = {
+ // Find field with metadata key hudi_blob=true
+ val blobFieldIndex = structType.fields.zipWithIndex.find { case (field, _) =>
+ field.metadata.contains(HoodieSchema.TYPE_METADATA_FIELD) &&
+ field.metadata.getString(HoodieSchema.TYPE_METADATA_FIELD).equalsIgnoreCase(HoodieSchemaType.BLOB.name)
+ }.map(fieldAndIndex => (fieldAndIndex._2, fieldAndIndex._1.name))
+
+ blobFieldIndex.getOrElse {
+ throw new IllegalArgumentException(
+ s"No blob column found with metadata hudi_blob=true. " +
+ s"Available columns: ${structType.fieldNames.mkString(", ")}"
+ )
+ }
+ }
+
+ // Validate that the struct column is compatible with BlobType
+ private def isCompatibleBlobType(dt: DataType): Boolean = dt match {
+ case struct: StructType => DataType.equalsIgnoreCaseAndNullability(struct, BlobType.dataType)
+ case _ => false
+ }
+}
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/BatchedBlobReaderStrategy.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/BatchedBlobReaderStrategy.scala
new file mode 100644
index 000000000000..cda7cca57832
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/BatchedBlobReaderStrategy.scala
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.hudi.blob
+
+import org.apache.hudi.HoodieSparkConfUtils
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy}
+
+/**
+ * Spark strategy that converts [[BatchedBlobRead]] logical nodes to [[BatchedBlobReadExec]] physical nodes.
+ *
+ * Reads configuration for batching parameters and creates physical plan with appropriate settings.
+ *
+ * @param sparkSession SparkSession for accessing configuration
+ */
+case class BatchedBlobReaderStrategy(sparkSession: SparkSession) extends SparkStrategy {
+ def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case read @ BatchedBlobRead(child, _, _) =>
+ // TODO find proper way to access these configs
+ val maxGapBytes = HoodieSparkConfUtils.getConfigValue(
+ Map.empty, sparkSession.sessionState.conf,
+ BatchedBlobReader.MAX_GAP_BYTES_CONF,
+ String.valueOf(BatchedBlobReader.DEFAULT_MAX_GAP_BYTES)).toInt
+
+ val lookaheadSize = HoodieSparkConfUtils.getConfigValue(
+ Map.empty, sparkSession.sessionState.conf,
+ BatchedBlobReader.LOOKAHEAD_SIZE_CONF,
+ String.valueOf(BatchedBlobReader.DEFAULT_LOOKAHEAD_SIZE)).toInt
+
+ val storageConf = new HadoopStorageConfiguration(sparkSession.sparkContext.hadoopConfiguration)
+ BatchedBlobReadExec(
+ planLater(child),
+ maxGapBytes,
+ storageConf,
+ lookaheadSize,
+ read
+ ) :: Nil
+
+ case _ => Nil
+ }
+}
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/ReadBlobExpression.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/ReadBlobExpression.scala
new file mode 100644
index 000000000000..ffd00ff51313
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/ReadBlobExpression.scala
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.hudi.blob
+
+import org.apache.spark.sql.catalyst.expressions.{Expression, UnaryExpression, Unevaluable}
+import org.apache.spark.sql.types.{BinaryType, DataType}
+
+/**
+ * Marker expression for lazy blob data reading.
+ *
+ * This expression is detected by [[ReadBlobRule]] and transformed to use
+ * batched I/O for efficient blob reading during physical execution.
+ *
+ * Example: `SELECT id, read_blob(image_data) FROM table`
+ *
+ * @param child Expression representing the blob column (matching the definition in {@link org.apache.hudi.common.schema.HoodieSchema.Blob})
+ */
+case class ReadBlobExpression(child: Expression)
+ extends UnaryExpression
+ with Unevaluable {
+
+ override def dataType: DataType = BinaryType
+
+ override def nullable: Boolean = true
+
+ override protected def withNewChildInternal(newChild: Expression): ReadBlobExpression = {
+ copy(child = newChild)
+ }
+
+ override def toString: String = s"read_blob($child)"
+}
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/ReadBlobRule.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/ReadBlobRule.scala
new file mode 100644
index 000000000000..b91d08674cb9
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/ReadBlobRule.scala
@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.hudi.blob
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference, Expression, ExprId, NamedExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.types.{DataType, StructType}
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+
+/**
+ * Transforms queries with `read_blob()` to use lazy batched I/O.
+ *
+ * Replaces [[ReadBlobExpression]] markers with [[BatchedBlobRead]] nodes
+ * that read blob data during physical execution.
+ *
+ * Example: `SELECT id, read_blob(image_data) FROM table`
+ *
+ * @param spark SparkSession for accessing configuration
+ */
+case class ReadBlobRule(spark: SparkSession) extends Rule[LogicalPlan] {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperatorsUp {
+ case Project(projectList, Filter(condition, child))
+ if containsReadBlobExpression(projectList)
+ && containsReadBlobInExpression(condition)
+ && !child.isInstanceOf[BatchedBlobRead] =>
+ val projectBlobCols = extractAllBlobColumns(projectList)
+ val filterBlobCols = extractBlobColumnsFromExpression(condition)
+ val blobColumns = (projectBlobCols ++ filterBlobCols)
+ .foldLeft((mutable.LinkedHashSet.empty[ExprId], ArrayBuffer.empty[AttributeReference])) {
+ case ((seen, acc), a) if seen.add(a.exprId) => (seen, acc += a)
+ case ((seen, acc), _) => (seen, acc)
+ }._2.toSeq
+ val (wrappedPlan, blobToDataAttr) = wrapWithBlobReads(blobColumns, child)
+ val newCondition = replaceReadBlobExpression(condition, blobToDataAttr)
+ val newProjectList = transformNamedExpressions(projectList, blobToDataAttr)
+ Project(newProjectList, Filter(newCondition, wrappedPlan))
+
+ case Filter(condition, child)
+ if containsReadBlobInExpression(condition)
+ && !child.isInstanceOf[BatchedBlobRead] =>
+
+ val blobColumns = extractBlobColumnsFromExpression(condition)
+ val (wrappedPlan, blobToDataAttr) = wrapWithBlobReads(blobColumns, child)
+ val newCondition = replaceReadBlobExpression(condition, blobToDataAttr)
+ Project(child.output, Filter(newCondition, wrappedPlan))
+
+ case Project(projectList, child)
+ if containsReadBlobExpression(projectList)
+ && !child.isInstanceOf[BatchedBlobRead] =>
+
+ val blobColumns = extractAllBlobColumns(projectList)
+ val (wrappedPlan, blobToDataAttr) = wrapWithBlobReads(blobColumns, child)
+ val newProjectList = transformNamedExpressions(projectList, blobToDataAttr)
+ Project(newProjectList, wrappedPlan)
+
+ case node if containsReadBlobInAnyExpression(node) =>
+ throw new IllegalArgumentException(
+ s"read_blob() may only appear in SELECT or WHERE clauses. Found in unsupported logical plan node: ${node.nodeName}. " +
+ s"Move read_blob() to a SELECT or WHERE clause. Full plan: ${node.simpleStringWithNodeId()}")
+ }
+
+ private def containsReadBlobInAnyExpression(plan: LogicalPlan): Boolean = {
+ plan.expressions.exists(containsReadBlobInExpression)
+ }
+
+ private def wrapWithBlobReads(
+ blobColumns: Seq[AttributeReference],
+ child: LogicalPlan): (LogicalPlan, Map[ExprId, Attribute]) = {
+ if (blobColumns.isEmpty) {
+ throw new IllegalStateException("read_blob() found but no valid blob column reference extracted.")
+ }
+ blobColumns.foldLeft((child: LogicalPlan, Map.empty[ExprId, Attribute])) {
+ case ((currentPlan, mapping), blobAttr) =>
+ // Type compatibility check (early fail for non-struct columns)
+ blobAttr.dataType match {
+ case struct: StructType if DataType.equalsIgnoreCaseAndNullability(struct, org.apache.spark.sql.types.BlobType.dataType) =>
+ // Valid blob column
+ case _ =>
+ throw new IllegalArgumentException(
+ s"Blob column '${blobAttr.name}' must be compatible with BlobType (type, data, reference struct), found: ${blobAttr.dataType}")
+ }
+ val blobRead = BatchedBlobRead(currentPlan, blobAttr)
+ (blobRead, mapping + (blobAttr.exprId -> blobRead.dataAttr))
+ }
+ }
+
+ private def extractBlobColumnsFromExpression(expr: Expression): Seq[AttributeReference] = {
+ val seen = mutable.LinkedHashSet.empty[ExprId]
+ val result = ArrayBuffer.empty[AttributeReference]
+ collectBlobColumns(expr, seen, result)
+ result.toSeq
+ }
+
+ /**
+ * Check if any expression in the project list contains a ReadBlobExpression.
+ */
+ private def containsReadBlobExpression(projectList: Seq[Expression]): Boolean = {
+ projectList.exists(expr => containsReadBlobInExpression(expr))
+ }
+
+ private def containsReadBlobInExpression(expr: Expression): Boolean = {
+ expr match {
+ case _: ReadBlobExpression => true
+ case other => other.children.exists(containsReadBlobInExpression)
+ }
+ }
+
+ private def extractAllBlobColumns(expressions: Seq[Expression]): Seq[AttributeReference] = {
+ val seen = mutable.LinkedHashSet.empty[ExprId]
+ val result = ArrayBuffer.empty[AttributeReference]
+ expressions.foreach(collectBlobColumns(_, seen, result))
+ result.toSeq
+ }
+
+ private def collectBlobColumns(
+ expr: Expression,
+ seen: mutable.Set[ExprId],
+ result: ArrayBuffer[AttributeReference]): Unit = expr match {
+ case ReadBlobExpression(attr: AttributeReference) =>
+ if (seen.add(attr.exprId)) result += attr
+ case other =>
+ other.children.foreach(collectBlobColumns(_, seen, result))
+ }
+
+ private def transformNamedExpressions(
+ expressions: Seq[NamedExpression],
+ blobToDataAttr: Map[ExprId, Attribute]): Seq[NamedExpression] = {
+ expressions.map {
+ case alias @ Alias(childExpr, name) =>
+ val rewritten = replaceReadBlobExpression(childExpr, blobToDataAttr)
+ Alias(rewritten, name)(alias.exprId, alias.qualifier, alias.explicitMetadata)
+ case attr: AttributeReference => attr
+ case other =>
+ replaceReadBlobExpression(other, blobToDataAttr).asInstanceOf[NamedExpression]
+ }
+ }
+
+ private def replaceReadBlobExpression(
+ expr: Expression,
+ blobToDataAttr: Map[ExprId, Attribute]): Expression = expr match {
+ case ReadBlobExpression(attr: AttributeReference) =>
+ blobToDataAttr.getOrElse(attr.exprId, throw new IllegalArgumentException(
+ s"read_blob() called on column '${attr.name}' (exprId=${attr.exprId}) which was not registered for blob reading. " +
+ s"Available blob columns: ${blobToDataAttr.keys.mkString(", ")}"))
+ case ReadBlobExpression(_) =>
+ throw new IllegalStateException("read_blob() must be called on a direct column reference")
+ case other =>
+ other.mapChildren(replaceReadBlobExpression(_, blobToDataAttr))
+ }
+}
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/ScalarFunctions.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/ScalarFunctions.scala
new file mode 100644
index 000000000000..bf94b4519c33
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/blob/ScalarFunctions.scala
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.hudi.blob
+
+import org.apache.spark.sql.catalyst.FunctionIdentifier
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionInfo}
+
+/**
+ * Registry of scalar functions for Hudi SQL integration.
+ *
+ * These functions are registered as built-in functions that can be used
+ * in SQL queries. They integrate with Spark's function registry and are
+ * available in both SQL and DataFrame API contexts.
+ *
+ * <h3>Function Registration:</h3>
+ * Functions are registered via [[SparkAdapter.injectScalarFunctions]] which is
+ * called during [[HoodieSparkSessionExtension]] initialization.
+ *
+ * <h3>Adding New Functions:</h3>
+ * To add a new scalar function:
+ * <ol>
+ * <li>Create a marker expression class (extends Unevaluable)</li>
+ * <li>Add function definition tuple to [[funcs]] below</li>
+ * <li>Create a logical plan rule to handle the expression</li>
+ * <li>Register the rule in [[HoodieAnalysis.customPostHocResolutionRules]]</li>
+ * </ol>
+ */
+object ScalarFunctions {
+
+ private val READ_BLOB_FUNC_NAME = "read_blob"
+
+ /**
+ * Function definitions as tuples of:
+ * <ul>
+ * <li>FunctionIdentifier - function name</li>
+ * <li>ExpressionInfo - metadata for DESCRIBE FUNCTION</li>
+ * <li>Builder function - (Seq[Expression] => Expression)</li>
+ * </ul>
+ */
+ val funcs: Seq[(FunctionIdentifier, ExpressionInfo, Seq[Expression] => Expression)] = Seq(
+ (
+ FunctionIdentifier(READ_BLOB_FUNC_NAME),
+ new ExpressionInfo(
+ classOf[ReadBlobExpression].getCanonicalName,
+ READ_BLOB_FUNC_NAME,
+ """
+ |Usage: read_blob(blob_column) - Reads blob data from storage
+ |
+ |Reads byte ranges from files referenced in a blob-compatible column.
+ |The column must be structurally compatible with HoodieSchema.Blob.
+ |
+ |This function uses batched I/O operations for optimal performance.
+ |For best results, ensure data is sorted by (reference.file, reference.position).
+ |
+ |Example:
+ | SELECT id, name, read_blob(file_ref) as data FROM table
+ |
+ |Arguments:
+ | blob_column - Struct column with HoodieSchema.Blob structure:
+ | - type (string): "out_of_line" or "inline"
+ | - data (binary, nullable): inline blob data or null
+ | - reference (struct, nullable): {file, offset, length, managed}
+ |
+ |Returns:
+ | Binary data read from the file
+ |
+ |Performance:
+ | - Configure batching: hoodie.blob.batching.max.gap.bytes (default 4096)
+ | - Configure lookahead: hoodie.blob.batching.lookahead.size (default 50)
+ """.stripMargin
+ ),
+ (args: Seq[Expression]) => {
+ require(args.length == 1, s"read_blob expects exactly 1 argument, got ${args.length}")
+ ReadBlobExpression(args.head)
+ }
+ )
+ )
+}
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/HoodieSparkSessionExtension.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/HoodieSparkSessionExtension.scala
index 2cbf479bdead..f01ad896c657 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/HoodieSparkSessionExtension.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/HoodieSparkSessionExtension.scala
@@ -53,5 +53,7 @@ class HoodieSparkSessionExtension extends (SparkSessionExtensions => Unit)
*/
sparkAdapter.injectTableFunctions(extensions)
+ sparkAdapter.injectScalarFunctions(extensions)
+ sparkAdapter.injectPlannerStrategies(extensions)
}
}
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala
index e7576359c2a4..0d6d4efe304c 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala
@@ -33,6 +33,7 @@ import org.apache.spark.sql.execution.command._
import org.apache.spark.sql.execution.datasources.{CreateTable, LogicalRelation}
import org.apache.spark.sql.hudi.HoodieSqlCommonUtils.{isMetaField, removeMetaFields}
import org.apache.spark.sql.hudi.analysis.HoodieAnalysis.{sparkAdapter, MatchCreateIndex, MatchCreateTableLike, MatchDropIndex, MatchInsertIntoStatement, MatchMergeIntoTable, MatchRefreshIndex, MatchShowIndexes, ResolvesToHudiTable}
+import org.apache.spark.sql.hudi.blob.ReadBlobRule
import org.apache.spark.sql.hudi.command._
import org.apache.spark.sql.hudi.command.HoodieLeafRunnableCommand.stripMetaFieldAttributes
import org.apache.spark.sql.hudi.command.InsertIntoHoodieTableCommand.alignQueryOutput
@@ -169,6 +170,8 @@ object HoodieAnalysis extends SparkAdapterSupport {
}
rules += (spark => instantiateKlass(pruneFileSourcePartitionsClass, spark))
+ rules += (session => ReadBlobRule(session))
+
rules.toSeq
}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/BlobTestHelpers.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/BlobTestHelpers.scala
index 8783422c51b4..b94229aa478e 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/BlobTestHelpers.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/BlobTestHelpers.scala
@@ -39,8 +39,8 @@ object BlobTestHelpers {
def inlineBlobStructCol(name: String, bytesCol: Column): Column = {
struct(
lit(HoodieSchema.Blob.INLINE).as(HoodieSchema.Blob.TYPE),
- bytesCol.as(HoodieSchema.Blob.INLINE_DATA_FIELD),
- lit(null).cast("struct<externalPath:string,offset:bigint,length:bigint,managed:boolean>")
+ bytesCol.cast("binary").as(HoodieSchema.Blob.INLINE_DATA_FIELD),
+ lit(null).cast("struct<external_path:string,offset:bigint,length:bigint,managed:boolean>")
.as(HoodieSchema.Blob.EXTERNAL_REFERENCE)
).as(name, blobMetadata)
}
@@ -65,7 +65,7 @@ object BlobTestHelpers {
struct(
filePathCol.as(HoodieSchema.Blob.EXTERNAL_REFERENCE_PATH),
offsetCol.as(HoodieSchema.Blob.EXTERNAL_REFERENCE_OFFSET),
- lengthCol.as(HoodieSchema.Blob.EXTERNAL_REFERENCE_LENGTH),
+ lengthCol.cast("bigint").as(HoodieSchema.Blob.EXTERNAL_REFERENCE_LENGTH),
lit(false).as(HoodieSchema.Blob.EXTERNAL_REFERENCE_IS_MANAGED)
).as(HoodieSchema.Blob.EXTERNAL_REFERENCE)
).as(name, blobMetadata)
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/TestBatchedBlobReader.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/TestBatchedBlobReader.scala
new file mode 100644
index 000000000000..12f4496cc377
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/TestBatchedBlobReader.scala
@@ -0,0 +1,457 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.blob
+
+import org.apache.hudi.blob.BlobTestHelpers._
+import org.apache.hudi.common.schema.HoodieSchema
+import org.apache.hudi.testutils.HoodieClientTestBase
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.functions._
+import org.apache.spark.sql.hudi.blob.BatchedBlobReader
+import org.apache.spark.sql.types._
+import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.Test
+
+import java.util.Collections
+
+/**
+ * Tests for BatchedByteRangeReader.
+ *
+ * These tests verify the batching behavior and effectiveness of the
+ * BatchedByteRangeReader compared to non-batched approaches.
+ */
+class TestBatchedBlobReader extends HoodieClientTestBase {
+
+ @Test
+ def testBasicBatchedRead(): Unit = {
+ val filePath = createTestFile(tempDir, "basic.bin", 10000)
+
+ // Create input with struct column
+ val inputDF = sparkSession.createDataFrame(Seq(
+ (filePath, 0L, 100L),
+ (filePath, 100L, 100L),
+ (filePath, 200L, 100L)
+ )).toDF("external_path", "offset", "length")
+ .withColumn("data", blobStructCol("data", col("external_path"), col("offset"), col("length")))
+ .select("offset", "data")
+
+ // Read with batching
+ val resultDF = BatchedBlobReader.readBatched(inputDF, storageConf)
+
+ // Verify schema
+ assertTrue(resultDF.columns.contains("data"))
+ assertEquals(2, resultDF.columns.length) // offset, data
+
+ // Verify results
+ val results = resultDF.orderBy("offset").collect()
+ assertEquals(3, results.length)
+
+ // Check data content
+ results.zipWithIndex.foreach { case (row, i) =>
+ val data = row.getAs[Array[Byte]]("data")
+ assertEquals(100, data.length)
+
+ // Verify content matches expected pattern
+ assertBytesContent(data, expectedOffset = i * 100)
+ }
+ }
+
+ @Test
+ def testNoBatchingDifferentFiles(): Unit = {
+ // Create different files
+ val file1 = createTestFile(tempDir, "file1.bin", 5000)
+ val file2 = createTestFile(tempDir, "file2.bin", 5000)
+ val file3 = createTestFile(tempDir, "file3.bin", 5000)
+
+ // Reads from different files (no batching possible)
+ val inputDF = sparkSession.createDataFrame(Seq(
+ (file1, 0L, 100L),
+ (file2, 0L, 100L),
+ (file3, 0L, 100L)
+ )).toDF("external_path", "offset", "length")
+ .withColumn("data", blobStructCol("data", col("external_path"), col("offset"), col("length")))
+ .select("data")
+
+ val resultDF = BatchedBlobReader.readBatched(inputDF, storageConf)
+
+ val results = resultDF.collect()
+ assertEquals(3, results.length)
+
+ // Verify all reads succeeded
+ results.foreach { row =>
+ val data = row.getAs[Array[Byte]]("data")
+ assertEquals(100, data.length)
+ }
+ }
+
+ @Test
+ def testGapThresholdSmallGaps(): Unit = {
+
+ val filePath = createTestFile(tempDir, "small-gaps.bin", 10000)
+
+ // Reads with small gaps (should batch with default threshold of 4KB)
+ val inputDF = sparkSession.createDataFrame(Seq(
+ (filePath, 0L, 100L),
+ (filePath, 120L, 100L), // 20 byte gap
+ (filePath, 240L, 100L), // 20 byte gap
+ (filePath, 360L, 100L) // 20 byte gap
+ )).toDF("external_path", "offset", "length")
+ .withColumn("data", blobStructCol("data", col("external_path"), col("offset"), col("length")))
+ .select("data")
+
+ // Use default maxGapBytes=4096 which should batch these
+ val resultDF = BatchedBlobReader.readBatched(inputDF, storageConf, maxGapBytes = 4096)
+
+ val results = resultDF.collect()
+ assertEquals(4, results.length)
+
+ results.foreach { row =>
+ val data = row.getAs[Array[Byte]]("data")
+ assertEquals(100, data.length)
+ }
+ }
+
+ @Test
+ def testGapThresholdLargeGaps(): Unit = {
+
+ val filePath = createTestFile(tempDir, "large-gaps.bin", 50000)
+
+ // Reads with large gaps (should NOT batch with small threshold)
+ val inputDF = sparkSession.createDataFrame(Seq(
+ (filePath, 0L, 100L),
+ (filePath, 10000L, 100L), // 9.9KB gap
+ (filePath, 20000L, 100L), // 9.9KB gap
+ (filePath, 30000L, 100L) // 9.9KB gap
+ )).toDF("external_path", "offset", "length")
+ .withColumn("data", blobStructCol("data", col("external_path"), col("offset"), col("length")))
+ .select("data")
+
+ // Use small maxGapBytes that won't batch these
+ val resultDF = BatchedBlobReader.readBatched(inputDF, storageConf, maxGapBytes = 1000)
+
+ val results = resultDF.collect()
+ assertEquals(4, results.length)
+
+ // Verify data correctness
+ results.zipWithIndex.foreach { case (row, i) =>
+ val data = row.getAs[Array[Byte]]("data")
+ assertEquals(100, data.length)
+ }
+ }
+
+ @Test
+ def testPreserveInputOrder(): Unit = {
+
+ val filePath = createTestFile(tempDir, "order.bin", 10000)
+
+ // Create input in specific order with record IDs
+ val inputDF = sparkSession.createDataFrame(Seq(
+ (filePath, 0L, 100L, "rec1"),
+ (filePath, 100L, 100L, "rec2"),
+ (filePath, 200L, 100L, "rec3"),
+ (filePath, 300L, 100L, "rec4")
+ )).toDF("external_path", "offset", "length", "record_id")
+ .withColumn("data", blobStructCol("data", col("external_path"), col("offset"), col("length")))
+ .select("data", "record_id")
+
+ val resultDF = BatchedBlobReader.readBatched(inputDF, storageConf)
+
+ val results = resultDF.collect()
+ assertEquals(4, results.length)
+
+ // Verify order is preserved
+ assertEquals("rec1", results(0).getAs[String]("record_id"))
+ assertEquals("rec2", results(1).getAs[String]("record_id"))
+ assertEquals("rec3", results(2).getAs[String]("record_id"))
+ assertEquals("rec4", results(3).getAs[String]("record_id"))
+ }
+
+ @Test
+ def testMixedScenario(): Unit = {
+
+ val file1 = createTestFile(tempDir, "mixed1.bin", 10000)
+ val file2 = createTestFile(tempDir, "mixed2.bin", 10000)
+
+ // Mix of batchable and non-batchable reads
+ val inputDF = sparkSession.createDataFrame(Seq(
+ // Batchable group from file1
+ (file1, 0L, 100L),
+ (file1, 100L, 100L),
+ (file1, 200L, 100L),
+ // Single read from file2
+ (file2, 0L, 100L),
+ // Another batchable group from file1
+ (file1, 300L, 100L),
+ (file1, 400L, 100L),
+ // Large gap in file1 (may not batch depending on threshold)
+ (file1, 5000L, 100L)
+ )).toDF("external_path", "offset", "length")
+ .withColumn("data", blobStructCol("data", col("external_path"), col("offset"), col("length")))
+ .select("data")
+
+ val resultDF = BatchedBlobReader.readBatched(inputDF, storageConf)
+
+ val results = resultDF.collect()
+ assertEquals(7, results.length)
+
+ // Verify all reads succeeded
+ results.foreach { row =>
+ val data = row.getAs[Array[Byte]]("data")
+ assertEquals(100, data.length)
+ }
+ }
+
+ @Test
+ def testEmptyDataset(): Unit = {
+
+ val inputDF = sparkSession.createDataFrame(Seq.empty[(String, Long, Int)])
+ .toDF("external_path", "offset", "length")
+ .withColumn("data", blobStructCol("data", col("external_path"), col("offset"), col("length")))
+ .select("data")
+
+ val resultDF = BatchedBlobReader.readBatched(inputDF, storageConf)
+
+ assertEquals(0, resultDF.count())
+ }
+
+ @Test
+ def testPreserveAdditionalColumns(): Unit = {
+
+ val filePath = createTestFile(tempDir, "preserve-cols.bin", 5000)
+
+ // Input with multiple additional columns
+ val inputDF = sparkSession.createDataFrame(Seq(
+ (filePath, 0L, 100L, "rec1", 42, true, 3.14),
+ (filePath, 100L, 100L, "rec2", 43, false, 2.71)
+ )).toDF("external_path", "offset", "length", "record_id", "sequence", "flag", "value")
+ .withColumn("data", blobStructCol("data", col("external_path"), col("offset"), col("length")))
+ .select("data", "record_id", "sequence", "flag", "value")
+
+ val resultDF = BatchedBlobReader.readBatched(inputDF, storageConf)
+
+ // Verify all columns are preserved
+ assertColumnsExist(resultDF, "data", "record_id", "sequence", "flag", "value")
+
+ val results = resultDF.collect()
+ assertEquals(2, results.length)
+
+ // Verify data integrity
+ assertEquals("rec1", results(0).getAs[String]("record_id"))
+ assertEquals(42, results(0).getAs[Int]("sequence"))
+ assertEquals(true, results(0).getAs[Boolean]("flag"))
+ assertEquals(3.14, results(0).getAs[Double]("value"), 0.001)
+
+ assertEquals("rec2", results(1).getAs[String]("record_id"))
+ assertEquals(43, results(1).getAs[Int]("sequence"))
+ assertEquals(false, results(1).getAs[Boolean]("flag"))
+ assertEquals(2.71, results(1).getAs[Double]("value"), 0.001)
+ }
+
+ @Test
+ def testExplicitColumnNameWithMultipleBlobColumns(): Unit = {
+ // Test that explicit column name resolves the correct column
+ // when multiple blob columns exist in the schema
+
+ val file1 = createTestFile(tempDir, "blob1.bin", 5000)
+ val file2 = createTestFile(tempDir, "blob2.bin", 5000)
+
+ // Create DataFrame with two blob columns
+ val inputDF = sparkSession.createDataFrame(Seq(
+ (1, file1, 0L, 100L, file2, 0L, 50L),
+ (2, file1, 100L, 100L, file2, 50L, 50L)
+ )).toDF("id", "path1", "offset1", "len1", "path2", "offset2", "len2")
+ .withColumn("blob1", blobStructCol("blob1", col("path1"), col("offset1"), col("len1")))
+ .withColumn("blob2", blobStructCol("blob2", col("path2"), col("offset2"), col("len2")))
+ .select("id", "blob1", "blob2")
+
+ // Resolve blob1 explicitly
+ val result1 = BatchedBlobReader.readBatched(
+ inputDF,
+ storageConf,
+ columnName = Some("blob1")
+ )
+
+ val rows1 = result1.collect()
+ assertEquals(2, rows1.length)
+
+ // Verify blob1 data was read (100 bytes)
+ rows1.foreach { row =>
+ val data = row.getAs[Array[Byte]]("blob1")
+ assertEquals(100, data.length)
+ }
+
+ // Resolve blob2 explicitly
+ val result2 = BatchedBlobReader.readBatched(
+ inputDF,
+ storageConf,
+ columnName = Some("blob2")
+ )
+
+ val rows2 = result2.collect()
+ assertEquals(2, rows2.length)
+
+ // Verify blob2 data was read (50 bytes)
+ rows2.foreach { row =>
+ val data = row.getAs[Array[Byte]]("blob2")
+ assertEquals(50, data.length)
+ }
+ }
+
+ @Test
+ def testFallbackToMetadataWhenNoColumnNameProvided(): Unit = {
+ // Test that when no explicit column name is provided,
+ // it falls back to searching for hudi_blob=true metadata
+
+ val filePath = createTestFile(tempDir, "fallback.bin", 5000)
+
+ // Create DataFrame with blob metadata (the traditional way)
+ val inputDF = sparkSession.createDataFrame(Seq(
+ (filePath, 0L, 100L),
+ (filePath, 100L, 100L)
+ )).toDF("external_path", "offset", "length")
+ .withColumn("data", blobStructCol("data", col("external_path"), col("offset"), col("length")))
+ .select("data")
+
+ // Should work without explicit column name (uses metadata)
+ val resultDF = BatchedBlobReader.readBatched(
+ inputDF,
+ storageConf
+ // Note: columnName parameter is NOT provided
+ )
+
+ val results = resultDF.collect()
+ assertEquals(2, results.length)
+
+ results.foreach { row =>
+ val data = row.getAs[Array[Byte]]("data")
+ assertEquals(100, data.length)
+ }
+ }
+
+ @Test
+ def testInlineBlobRead(): Unit = {
+ val inlineData = Array[Byte](10, 20, 30, 40, 50)
+ val inputDF = sparkSession.createDataFrame(Seq(Tuple1("rec1")))
+ .toDF("record_id")
+ .withColumn("data", inlineBlobStructCol("data", lit(inlineData)))
+ .select("data", "record_id")
+
+ val resultDF = BatchedBlobReader.readBatched(inputDF, storageConf)
+ val results = resultDF.collect()
+ assertEquals(1, results.length)
+ assertArrayEquals(inlineData, results(0).getAs[Array[Byte]]("data"))
+ assertEquals("rec1", results(0).getAs[String]("record_id"))
+ }
+
+ @Test
+ def testWholeFileBlobRead(): Unit = {
+ val size = 500
+ val filePath = createTestFile(tempDir, "whole.bin", size)
+
+ val inputDF = sparkSession.createDataFrame(Seq(Tuple1("rec1")))
+ .toDF("record_id")
+ .withColumn("data", wholeFileBlobStructCol("data", lit(filePath)))
+ .select("data", "record_id")
+
+ val resultDF = BatchedBlobReader.readBatched(inputDF, storageConf)
+ val results = resultDF.collect()
+ assertEquals(1, results.length)
+ val data = results(0).getAs[Array[Byte]]("data")
+ assertEquals(size, data.length)
+ assertBytesContent(data, expectedOffset = 0)
+ }
+
+ @Test
+ def testMixedBlobTypes(): Unit = {
+ val fileSize = 1000
+ val filePath = createTestFile(tempDir, "mixed-types.bin", fileSize)
+ val inlineData = Array[Byte](7, 8, 9)
+
+ val outerSchema = StructType(Seq(
+ StructField("record_id", StringType, nullable = false),
+ StructField("data", BlobType(), nullable = false, metadata = blobMetadata)
+ ))
+
+ val rows = Seq(
+ Row("inline_row", Row(HoodieSchema.Blob.INLINE, inlineData, null)),
+ Row("wholefile_row", Row(HoodieSchema.Blob.OUT_OF_LINE, null, Row(filePath, null, null, false))),
+ Row("range_row_1", Row(HoodieSchema.Blob.OUT_OF_LINE, null, Row(filePath, 0L, 100L, false))),
+ Row("range_row_2", Row(HoodieSchema.Blob.OUT_OF_LINE, null, Row(filePath, 100L, 100L, false)))
+ )
+
+ val inputDF = sparkSession.createDataFrame(
+ sparkSession.sparkContext.parallelize(rows), outerSchema)
+
+ val resultDF = BatchedBlobReader.readBatched(inputDF, storageConf, columnName = Some("data"))
+ val results = resultDF.orderBy("record_id").collect()
+ assertEquals(4, results.length)
+
+ val inlineRow = results.find(_.getAs[String]("record_id") == "inline_row").get
+ assertArrayEquals(inlineData, inlineRow.getAs[Array[Byte]]("data"))
+
+ val wholeFileRow = results.find(_.getAs[String]("record_id") == "wholefile_row").get
+ val wholeFileData = wholeFileRow.getAs[Array[Byte]]("data")
+ assertEquals(fileSize, wholeFileData.length)
+
+ val range1 = results.find(_.getAs[String]("record_id") == "range_row_1").get
+ assertEquals(100, range1.getAs[Array[Byte]]("data").length)
+ assertBytesContent(range1.getAs[Array[Byte]]("data"), expectedOffset = 0)
+
+ val range2 = results.find(_.getAs[String]("record_id") == "range_row_2").get
+ assertEquals(100, range2.getAs[Array[Byte]]("data").length)
+ assertBytesContent(range2.getAs[Array[Byte]]("data"), expectedOffset = 100)
+ }
+
+ @Test
+ def testNullBlobStructColumnReturnsNull(): Unit = {
+ // Create a DataFrame with a null struct column
+ val schema = StructType(Seq(StructField("data", BlobType.dataType, nullable = true, metadata = blobMetadata)))
+ val rows = Collections.singletonList(Row(null))
+ val inputDF = sparkSession.createDataFrame(rows, schema)
+
+ val resultDF = BatchedBlobReader.readBatched(inputDF, storageConf)
+ val results = resultDF.collect()
+ assertEquals(1, results.length)
+ assertTrue(results(0).isNullAt(0))
+ }
+
+ @Test
+ def testOverlappingRangesThrowsException(): Unit = {
+ val filePath = createTestFile(tempDir, "overlap.bin", 1000)
+ // Overlapping: [0, 100) and [50, 100)
+ val inputDF = sparkSession.createDataFrame(Seq(
+ (filePath, 0L, 100L),
+ (filePath, 50L, 100L)
+ )).toDF("external_path", "offset", "length")
+ .withColumn("data", blobStructCol("data", col("external_path"), col("offset"), col("length")))
+ .select("offset", "data")
+ .coalesce(1)
+
+ val thrown = assertThrows(classOf[SparkException], () => {
+ val rows = BatchedBlobReader.readBatched(inputDF, storageConf).collect()
+ // Force access to the data column to trigger the batch read logic
+ rows.foreach(row => row.getAs[Array[Byte]]("data"))
+ })
+ assertTrue(thrown.getCause.isInstanceOf[IllegalArgumentException])
+ assertTrue(thrown.getCause.getMessage.contains("Overlapping blob ranges detected"))
+ }
+}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/TestBlobSupport.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/TestBlobSupport.scala
index 22e5eee77773..73913bdc0fcd 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/TestBlobSupport.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/TestBlobSupport.scala
@@ -29,7 +29,7 @@ import org.apache.hudi.config.HoodieIndexConfig
import org.apache.hudi.index.HoodieIndex
import org.apache.hudi.testutils.HoodieClientTestBase
-import org.apache.avro.generic.{GenericData, GenericRecord, IndexedRecord}
+import org.apache.avro.generic.{GenericData, IndexedRecord}
import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.EnumSource
@@ -111,6 +111,11 @@ class TestBlobSupport extends HoodieClientTestBase with SparkDatasetMixin {
val filePath = reference.getString(reference.fieldIndex(HoodieSchema.Blob.EXTERNAL_REFERENCE_PATH))
assertTrue(filePath.endsWith("file2.bin"))
}
+
+ // Verify SQL read_blob() function
+ table.createOrReplaceTempView("hudi_table_view")
+ val sqlRows = sparkSession.sql("SELECT id, value, read_blob(data) as full_bytes from hudi_table_view").collectAsList()
+ assertEquals(10, sqlRows.size())
}
private def createTestRecords(filePath: String): Seq[HoodieRecord[IndexedRecord]] = {
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/TestReadBlobSQL.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/TestReadBlobSQL.scala
new file mode 100644
index 000000000000..8d49d7017a8a
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/blob/TestReadBlobSQL.scala
@@ -0,0 +1,439 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.blob
+
+import org.apache.hudi.blob.BlobTestHelpers._
+import org.apache.hudi.exception.HoodieIOException
+import org.apache.hudi.testutils.HoodieClientTestBase
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.functions._
+import org.apache.spark.sql.types._
+import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.Test
+
+import java.util.Collections
+
+/**
+ * Tests for the read_blob() SQL function.
+ *
+ * This test suite verifies:
+ * <ul>
+ * <li>Basic SQL integration with read_blob()</li>
+ * <li>Integration with WHERE clauses, JOINs</li>
+ * <li>Configuration parameter handling</li>
+ * <li>Error handling for invalid inputs</li>
+ * </ul>
+ */
+class TestReadBlobSQL extends HoodieClientTestBase {
+
+ @Test
+ def testBasicReadBlobSQL(): Unit = {
+ val filePath = createTestFile(tempDir, "basic.bin", 10000)
+
+ // Main DataFrame with blobStructCol
+ val df = sparkSession.createDataFrame(Seq(
+ (1, "record1", filePath, 0L, 100L),
+ (3, "record3", filePath, 100L, 100L),
+ (4, "record4", filePath, 200L, 100L)
+ )).toDF("id", "name", "external_path", "offset", "length")
+ .withColumn("file_info", blobStructCol("file_info", col("external_path"), col("offset"), col("length")))
+ .select("id", "name", "file_info")
+
+ // Ensure file_info is nullable in the schema
+ val schema = StructType(df.schema.map {
+ case StructField("file_info", dt, _, md) => StructField("file_info", dt, nullable = true, md)
+ case other => other
+ })
+ val dfWithNullable = sparkSession.createDataFrame(df.rdd, schema)
+
+ // DataFrame with a null blob value
+ val nullRow = Row(2, "record2", null)
+ val nullDf = sparkSession.createDataFrame(Collections.singletonList(nullRow), schema)
+
+ // Union the null row
+ val fullDf = dfWithNullable.unionByName(nullDf)
+ fullDf.createOrReplaceTempView("test_table")
+
+ // Use SQL with read_blob
+ val result = sparkSession.sql("""
+ SELECT id, name, read_blob(file_info) as data
+ FROM test_table
+ WHERE id <= 3
+ ORDER BY id
+ """)
+
+ val rows = result.collect()
+ assertEquals(3, rows.length)
+
+ // Verify data is binary for non-null rows
+ val data1 = rows(0).getAs[Array[Byte]]("data")
+ assertEquals(100, data1.length)
+ assertBytesContent(data1)
+
+ // The null_blob row should have null data
+ assertTrue(rows(1).isNullAt(2))
+
+ val data3 = rows(2).getAs[Array[Byte]]("data")
+ assertEquals(100, data3.length)
+ assertBytesContent(data3, expectedOffset = 100)
+ }
+
+ @Test
+ def testReadBlobWithJoin(): Unit = {
+ val filePath = createTestFile(tempDir, "join.bin", 10000)
+
+ // Create blob table
+ val blobDF = sparkSession.createDataFrame(Seq(
+ (1, filePath, 0L, 100L),
+ (2, filePath, 100L, 100L)
+ )).toDF("id", "external_path", "offset", "length")
+ .withColumn("file_info",
+ blobStructCol("file_info", col("external_path"), col("offset"), col("length")))
+ .select("id", "file_info")
+
+ blobDF.createOrReplaceTempView("blob_table")
+
+ // Create metadata table
+ val metaDF = sparkSession.createDataFrame(Seq(
+ (1, "Alice"),
+ (2, "Bob")
+ )).toDF("id", "name")
+
+ metaDF.createOrReplaceTempView("meta_table")
+
+ // SQL with JOIN
+ val result = sparkSession.sql("""
+ SELECT m.id, m.name, read_blob(b.file_info) as data
+ FROM meta_table m
+ JOIN blob_table b ON m.id = b.id
+ ORDER BY m.id
+ """)
+
+ val rows = result.collect()
+ assertEquals(2, rows.length)
+ assertEquals("Alice", rows(0).getAs[String]("name"))
+ assertEquals(100, rows(0).getAs[Array[Byte]]("data").length)
+ assertEquals("Bob", rows(1).getAs[String]("name"))
+ assertEquals(100, rows(1).getAs[Array[Byte]]("data").length)
+
+ // Verify data content
+ val data1 = rows(0).getAs[Array[Byte]]("data")
+ assertBytesContent(data1)
+ }
+
+ @Test
+ def testReadBlobInSubquery(): Unit = {
+ val filePath = createTestFile(tempDir, "subquery.bin", 10000)
+
+ val df = sparkSession.createDataFrame(Seq(
+ (1, "A", filePath, 0L, 100L),
+ (2, "A", filePath, 100L, 100L),
+ (3, "B", filePath, 200L, 100L)
+ )).toDF("id", "category", "external_path", "offset", "length")
+ .withColumn("file_info",
+ blobStructCol("file_info", col("external_path"), col("offset"), col("length")))
+ .select("id", "category", "file_info")
+
+ df.createOrReplaceTempView("subquery_table")
+
+ // SQL with subquery
+ val result = sparkSession.sql("""
+ SELECT * FROM (
+ SELECT id, category, read_blob(file_info) as data
+ FROM subquery_table
+ ) WHERE category = 'A'
+ """)
+
+ val rows = result.collect()
+ assertEquals(2, rows.length)
+ rows.foreach { row =>
+ assertEquals("A", row.getAs[String]("category"))
+ assertEquals(100, row.getAs[Array[Byte]]("data").length)
+ }
+ }
+
+ @Test
+ def testConfigurationParameters(): Unit = {
+ val filePath = createTestFile(tempDir, "config.bin", 50000)
+
+ val df = sparkSession.createDataFrame(Seq(
+ (1, filePath, 0L, 100L),
+ (2, filePath, 5000L, 100L), // 4.9KB gap
+ (3, filePath, 10000L, 100L)
+ )).toDF("id", "external_path", "offset", "length")
+ .withColumn("file_info",
+ blobStructCol("file_info", col("external_path"), col("offset"), col("length")))
+ .select("id", "file_info")
+
+ df.createOrReplaceTempView("config_table")
+
+ // Use withSparkConfig to automatically manage configuration
+ withSparkConfig(sparkSession, Map(
+ "hoodie.blob.batching.max.gap.bytes" -> "10000",
+ "hoodie.blob.batching.lookahead.size" -> "100"
+ )) {
+ val result = sparkSession.sql("""
+ SELECT id, read_blob(file_info) as data
+ FROM config_table
+ """)
+
+ val rows = result.collect()
+ assertEquals(3, rows.length)
+
+ // Verify all reads completed successfully
+ rows.foreach { row =>
+ assertEquals(100, row.getAs[Array[Byte]]("data").length)
+ }
+ }
+ }
+
+ @Test
+ def testMultipleReadBlobInSameQuery(): Unit = {
+ val filePath1 = createTestFile(tempDir, "multi1.bin", 10000)
+ val filePath2 = createTestFile(tempDir, "multi2.bin", 10000)
+
+ val df = sparkSession.createDataFrame(Seq(
+ (1, filePath1, 0L, 50L, filePath2, 500L, 50L),
+ (2, filePath1, 100L, 50L, filePath2, 600L, 50L)
+ )).toDF("id", "external_path1", "offset1", "length1", "external_path2", "offset2", "length2")
+ .withColumn("file_info1",
+ blobStructCol("file_info1", col("external_path1"), col("offset1"), col("length1")))
+ .withColumn("file_info2",
+ blobStructCol("file_info2", col("external_path2"), col("offset2"), col("length2")))
+ .select("id", "file_info1", "file_info2")
+
+ df.createOrReplaceTempView("multi_table")
+
+ // SQL with multiple read_blob calls
+ val result = sparkSession.sql("""
+ SELECT
+ id,
+ read_blob(file_info1) as data1,
+ read_blob(file_info2) as data2
+ FROM multi_table
+ """)
+
+ val rows = result.collect()
+ assertEquals(2, rows.length)
+
+ // Row 1: data1 = file1 at offset 0, data2 = file2 at offset 500
+ val data1_row1 = rows(0).getAs[Array[Byte]]("data1")
+ val data2_row1 = rows(0).getAs[Array[Byte]]("data2")
+ assertEquals(50, data1_row1.length)
+ assertEquals(50, data2_row1.length)
+ assertBytesContent(data1_row1, expectedOffset = 0)
+ assertBytesContent(data2_row1, expectedOffset = 500)
+
+ // Row 2: data1 = file1 at offset 100, data2 = file2 at offset 600
+ val data1_row2 = rows(1).getAs[Array[Byte]]("data1")
+ val data2_row2 = rows(1).getAs[Array[Byte]]("data2")
+ assertBytesContent(data1_row2, expectedOffset = 100)
+ assertBytesContent(data2_row2, expectedOffset = 600)
+ }
+
+ @Test
+ def testReadBlobWithEmptyResult(): Unit = {
+ val filePath = createTestFile(tempDir, "empty.bin", 10000)
+
+ val df = sparkSession.createDataFrame(Seq(
+ (1, filePath, 0L, 100L),
+ (2, filePath, 100L, 100L)
+ )).toDF("id", "external_path", "offset", "length")
+ .withColumn("file_info",
+ blobStructCol("file_info", col("external_path"), col("offset"), col("length")))
+ .select("id", "file_info")
+
+ df.createOrReplaceTempView("empty_table")
+
+ // SQL that returns no rows
+ val result = sparkSession.sql("""
+ SELECT id, read_blob(file_info) as data
+ FROM empty_table
+ WHERE id > 100
+ """)
+
+ val rows = result.collect()
+ assertEquals(0, rows.length)
+ }
+
+ @Test
+ def testReadBlobMultipleFiles(): Unit = {
+ val filePath1 = createTestFile(tempDir, "file1.bin", 10000)
+ val filePath2 = createTestFile(tempDir, "file2.bin", 10000)
+
+ val df = sparkSession.createDataFrame(Seq(
+ (1, filePath1, 0L, 100L),
+ (2, filePath2, 0L, 100L),
+ (3, filePath1, 100L, 100L),
+ (4, filePath2, 100L, 100L)
+ )).toDF("id", "external_path", "offset", "length")
+ .withColumn("file_info",
+ blobStructCol("file_info", col("external_path"), col("offset"), col("length")))
+ .select("id", "file_info")
+
+ df.createOrReplaceTempView("multi_file_table")
+
+ // SQL reading from multiple files
+ val result = sparkSession.sql("""
+ SELECT id, read_blob(file_info) as data
+ FROM multi_file_table
+ ORDER BY id
+ """)
+
+ val rows = result.collect()
+ assertEquals(4, rows.length)
+
+ // Verify all data was read correctly
+ rows.foreach { row =>
+ assertEquals(100, row.getAs[Array[Byte]]("data").length)
+ }
+ }
+
+ @Test
+ def testReadBlobInWhereClause(): Unit = {
+ val filePath = createTestFile(tempDir, "where.bin", 10000)
+ val df = sparkSession.createDataFrame(Seq(
+ (1, filePath, 0L, 50L), // 50 bytes — filtered out
+ (2, filePath, 100L, 100L), // 100 bytes — passes filter
+ (3, filePath, 200L, 200L) // 200 bytes — passes filter
+ )).toDF("id", "external_path", "offset", "length")
+ .withColumn("file_info", blobStructCol("file_info", col("external_path"), col("offset"), col("length")))
+ .select("id", "file_info")
+ df.createOrReplaceTempView("where_table")
+
+ val result = sparkSession.sql("""
+ SELECT id, read_blob(file_info) AS data
+ FROM where_table
+ WHERE length(read_blob(file_info)) > 50
+ ORDER BY id
+ """)
+
+ val rows = result.collect()
+ assertEquals(2, rows.length)
+
+ // validate that rows with IDs 2 and 3 are returned
+ assertEquals(2, rows(0).getInt(0))
+ assertEquals(3, rows(1).getInt(0))
+ }
+
+ @Test
+ def testReadBlobWithCaseWhen(): Unit = {
+ val filePath = createTestFile(tempDir, "case.bin", 10000)
+
+ val df = sparkSession.createDataFrame(Seq(
+ (1, true, filePath, 0L, 100L),
+ (2, false, filePath, 100L, 100L),
+ (3, true, filePath, 200L, 100L)
+ )).toDF("id", "should_resolve", "external_path", "offset", "length")
+ .withColumn("file_info",
+ blobStructCol("file_info", col("external_path"), col("offset"), col("length")))
+ .select("id", "should_resolve", "file_info")
+
+ df.createOrReplaceTempView("case_table")
+
+ // SQL with CASE WHEN - note: this tests that the expression is handled
+ // even in conditional contexts
+ val result = sparkSession.sql("""
+ SELECT
+ id,
+ should_resolve,
+ CASE
+ WHEN should_resolve THEN read_blob(file_info)
+ ELSE NULL
+ END as data
+ FROM case_table
+ """)
+
+ val rows = result.collect()
+ assertEquals(3, rows.length)
+
+ // Row 1 should have data
+ assertTrue(rows(0).getAs[Boolean]("should_resolve"))
+ assertNotNull(rows(0).get(2))
+
+ // Row 2 should have null
+ assertFalse(rows(1).getAs[Boolean]("should_resolve"))
+ assertTrue(rows(1).isNullAt(2))
+
+ // Row 3 should have data
+ assertTrue(rows(2).getAs[Boolean]("should_resolve"))
+ assertNotNull(rows(2).get(2))
+ }
+
+ @Test
+ def testReadBlobWithMissingFile(): Unit = {
+ val missingPath = tempDir.resolve("does_not_exist.bin").toString
+ val df = sparkSession.createDataFrame(Seq(
+ (1, missingPath, 0L, 10L)
+ )).toDF("id", "external_path", "offset", "length")
+ .withColumn("file_info", blobStructCol("file_info", col("external_path"), col("offset"), col("length")))
+ .select("id", "file_info")
+ df.createOrReplaceTempView("missing_file_table")
+ val thrown = assertThrows(classOf[Exception], () => {
+ sparkSession.sql("SELECT id, read_blob(file_info) as data FROM missing_file_table").collect()
+ })
+ assertTrue(thrown.getCause.isInstanceOf[HoodieIOException])
+ }
+
+ @Test
+ def testReadBlobOnNonBlobColumn(): Unit = {
+ val df = sparkSession.createDataFrame(Seq(
+ (1, "not_a_blob")
+ )).toDF("id", "not_blob")
+ df.createOrReplaceTempView("non_blob_table")
+ val thrown = assertThrows(classOf[Exception], () => {
+ sparkSession.sql("SELECT id, read_blob(not_blob) as data FROM non_blob_table").collect()
+ })
+ assertTrue(thrown.isInstanceOf[IllegalArgumentException])
+ assertTrue(thrown.getMessage.contains("must be compatible with BlobType"))
+ }
+
+ @Test
+ def testReadBlobInProjectAndFilter(): Unit = {
+ val filePath = createTestFile(tempDir, "project_and_filter.bin", 10000)
+
+ // DataFrame with blobStructCol
+ val df = sparkSession.createDataFrame(Seq(
+ (1, "record1", filePath, 0L, 100L),
+ (2, "record2", filePath, 100L, 100L),
+ (3, "record3", filePath, 200L, 100L)
+ )).toDF("id", "name", "external_path", "offset", "length")
+ .withColumn("file_info", blobStructCol("file_info", col("external_path"), col("offset"), col("length")))
+ .select("id", "name", "file_info")
+
+ df.createOrReplaceTempView("project_and_filter_table")
+
+ // Query with read_blob in both SELECT and WHERE
+ val result = sparkSession.sql("""
+ SELECT id, name, read_blob(file_info) as data
+ FROM project_and_filter_table
+ WHERE length(read_blob(file_info)) = 100
+ ORDER BY id
+ """)
+
+ val rows = result.collect()
+ assertEquals(3, rows.length)
+ rows.zipWithIndex.foreach { case (row, idx) =>
+ assertEquals(100, row.getAs[Array[Byte]]("data").length)
+ assertBytesContent(row.getAs[Array[Byte]]("data"), expectedOffset = idx * 100)
+ }
+ }
+}
diff --git a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_3Adapter.scala b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_3Adapter.scala
index a180ab3483dc..06af2d316371 100644
--- a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_3Adapter.scala
+++ b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_3Adapter.scala
@@ -41,6 +41,7 @@ import org.apache.spark.sql.execution.datasources.orc.Spark33OrcReader
import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat, ParquetFilters, Spark33LegacyHoodieParquetFileFormat, Spark33ParquetReader}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.hudi.analysis.TableValuedFunctions
+import org.apache.spark.sql.hudi.blob.{BatchedBlobReaderStrategy, ScalarFunctions}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy
import org.apache.spark.sql.parser.{HoodieExtendedParserInterface, HoodieSpark3_3ExtendedSqlParser}
@@ -114,6 +115,16 @@ class Spark3_3Adapter extends BaseSpark3Adapter {
TableValuedFunctions.funcs.foreach(extensions.injectTableFunction)
}
+ override def injectScalarFunctions(extensions: SparkSessionExtensions): Unit = {
+ ScalarFunctions.funcs.foreach(extensions.injectFunction)
+ }
+
+ override def injectPlannerStrategies(extensions: SparkSessionExtensions): Unit = {
+ extensions.injectPlannerStrategy { session =>
+ BatchedBlobReaderStrategy(session)
+ }
+ }
+
/**
* Converts instance of [[StorageLevel]] to a corresponding string
*/
diff --git a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_4Adapter.scala b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_4Adapter.scala
index 32b21b936f09..21a2294e134d 100644
--- a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_4Adapter.scala
+++ b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_4Adapter.scala
@@ -42,6 +42,7 @@ import org.apache.spark.sql.execution.datasources.orc.Spark34OrcReader
import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat, ParquetFilters, Spark34LegacyHoodieParquetFileFormat, Spark34ParquetReader}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.hudi.analysis.TableValuedFunctions
+import org.apache.spark.sql.hudi.blob.{BatchedBlobReaderStrategy, ScalarFunctions}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy
import org.apache.spark.sql.parser.{HoodieExtendedParserInterface, HoodieSpark3_4ExtendedSqlParser}
@@ -115,6 +116,16 @@ class Spark3_4Adapter extends BaseSpark3Adapter {
TableValuedFunctions.funcs.foreach(extensions.injectTableFunction)
}
+ override def injectScalarFunctions(extensions: SparkSessionExtensions): Unit = {
+ ScalarFunctions.funcs.foreach(extensions.injectFunction)
+ }
+
+ override def injectPlannerStrategies(extensions: SparkSessionExtensions): Unit = {
+ extensions.injectPlannerStrategy { session =>
+ BatchedBlobReaderStrategy(session)
+ }
+ }
+
/**
* Converts instance of [[StorageLevel]] to a corresponding string
*/
diff --git a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_5Adapter.scala b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_5Adapter.scala
index 24492fdc358b..ab5d18024e42 100644
--- a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_5Adapter.scala
+++ b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_5Adapter.scala
@@ -21,7 +21,6 @@ import org.apache.hudi.Spark35HoodieFileScanRDD
import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.storage.StorageConfiguration
-import org.apache.avro.Schema
import org.apache.hadoop.conf.Configuration
import org.apache.parquet.schema.MessageType
import org.apache.spark.api.java.JavaSparkContext
@@ -42,6 +41,7 @@ import org.apache.spark.sql.execution.datasources.orc.Spark35OrcReader
import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat, ParquetFilters, Spark35LegacyHoodieParquetFileFormat, Spark35ParquetReader}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.hudi.analysis.TableValuedFunctions
+import org.apache.spark.sql.hudi.blob.{BatchedBlobReaderStrategy, ScalarFunctions}
import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
import org.apache.spark.sql.parser.{HoodieExtendedParserInterface, HoodieSpark3_5ExtendedSqlParser}
import org.apache.spark.sql.types.{DataType, DataTypes, Metadata, MetadataBuilder, StructType}
@@ -118,6 +118,16 @@ class Spark3_5Adapter extends BaseSpark3Adapter {
TableValuedFunctions.funcs.foreach(extensions.injectTableFunction)
}
+ override def injectScalarFunctions(extensions: SparkSessionExtensions): Unit = {
+ ScalarFunctions.funcs.foreach(extensions.injectFunction)
+ }
+
+ override def injectPlannerStrategies(extensions: SparkSessionExtensions): Unit = {
+ extensions.injectPlannerStrategy { session =>
+ BatchedBlobReaderStrategy(session)
+ }
+ }
+
/**
* Converts instance of [[StorageLevel]] to a corresponding string
*/
diff --git a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_0Adapter.scala b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_0Adapter.scala
index 7cb449809c2b..174f25b08b3d 100644
--- a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_0Adapter.scala
+++ b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_0Adapter.scala
@@ -39,6 +39,7 @@ import org.apache.spark.sql.execution.datasources.orc.Spark40OrcReader
import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat, Spark40LegacyHoodieParquetFileFormat, Spark40ParquetReader}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.hudi.analysis.TableValuedFunctions
+import org.apache.spark.sql.hudi.blob.{BatchedBlobReaderStrategy, ScalarFunctions}
import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
import org.apache.spark.sql.parser.{HoodieExtendedParserInterface, HoodieSpark4_0ExtendedSqlParser}
import org.apache.spark.sql.types.{DataType, DataTypes, Metadata, MetadataBuilder, StructType}
@@ -115,6 +116,16 @@ class Spark4_0Adapter extends BaseSpark4Adapter {
TableValuedFunctions.funcs.foreach(extensions.injectTableFunction)
}
+ override def injectScalarFunctions(extensions: SparkSessionExtensions): Unit = {
+ ScalarFunctions.funcs.foreach(extensions.injectFunction)
+ }
+
+ override def injectPlannerStrategies(extensions: SparkSessionExtensions): Unit = {
+ extensions.injectPlannerStrategy { session =>
+ BatchedBlobReaderStrategy(session)
+ }
+ }
+
/**
* Converts instance of [[StorageLevel]] to a corresponding string
*/