This is an automated email from the ASF dual-hosted git repository.
voonhous 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 1f5a4fe53aaa fix(spark): merge identical blob descriptors in the
batched read (#19913)
1f5a4fe53aaa is described below
commit 1f5a4fe53aaadd69d91a208a833ace990ae06122
Author: voonhous <[email protected]>
AuthorDate: Sat Sep 12 13:09:10 2026 +0800
fix(spark): merge identical blob descriptors in the batched read (#19913)
fix(spark): merge identical blob descriptors in batched read (#19913)
BatchedBlobReader plans reads one Spark task at a time: group the
task's rows by file, sort by offset, merge nearby ranges into one
read, slice each row's bytes out of the buffer. mergeRanges threw
"Overlapping blob ranges detected" whenever a row started before the
previous range ended.
A blob is a distinct entity and two blobs never share bytes (#18098),
so a real overlap is corruption and still throws. The check was
stricter than that invariant: it only compares a start offset to an
end offset, so two rows carrying the identical descriptor -- one blob
referenced by more than one row -- threw as well. A join with fan-out
produces exactly that, and ReadBlobRule wraps the join output without
adding a shuffle, so the repeated rows land in the same task.
Sort rows by (offset, length) so identical descriptors are adjacent,
and let a row whose descriptor matches the previous one join the
current range without growing it; it is served from the same read.
Partial overlaps and containment still throw, message unchanged.
Tests: TestBatchedBlobReaderMerge gains identical-merge,
nested-throws and duplicate-then-overlap-throws cases;
TestBatchedBlobReader gains an end-to-end identical-descriptor read;
TestReadBlobSQL gains a fan-out join that fails on master.
Fixes #19911
---
.../spark/sql/hudi/blob/BatchedBlobReader.scala | 27 +++++++++---
.../apache/hudi/blob/TestBatchedBlobReader.scala | 24 ++++++++++
.../org/apache/hudi/blob/TestReadBlobSQL.scala | 51 ++++++++++++++++++++++
.../sql/hudi/blob/TestBatchedBlobReaderMerge.scala | 41 +++++++++++++++++
4 files changed, 137 insertions(+), 6 deletions(-)
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
index 2074dee48d01..2506befddc8a 100644
---
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
@@ -285,8 +285,10 @@ class BatchedBlobReader(
/**
* 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.
+ * This method groups rows by file path, sorts them by (offset, length), and
merges
+ * ranges that are adjacent or within maxGapBytes of each other. Rows
carrying the identical
+ * descriptor (same file path, offset and length) share one read.
Overlapping ranges throw,
+ * because a blob is a distinct entity and two blobs never share bytes.
*
* @param rows Sequence of row information
* @return Sequence of merged ranges
@@ -298,8 +300,8 @@ class BatchedBlobReader(
val allRanges = ArrayBuffer[MergedRange[R]]()
byFile.foreach { case (filePath, fileRows) =>
- // Sort by offset
- val sorted = fileRows.sortBy(_.offset)
+ // Sort by offset, then by length, so rows carrying the identical
descriptor are adjacent
+ val sorted = fileRows.sortBy(r => (r.offset, r.length))
// Merge consecutive ranges
val merged = mergeRanges(sorted, maxGapBytes)
@@ -312,7 +314,13 @@ class BatchedBlobReader(
/**
* Merge consecutive ranges within the gap threshold.
*
- * @param rows Sorted rows from the same file
+ * Rows are grouped by file and sorted by (offset, length) before they reach
this method.
+ * Adjacent ranges and ranges within the gap threshold are merged into a
single read, and rows
+ * carrying the identical descriptor (same file path, offset and length)
share one read: that
+ * is one blob referenced by more than one row. Overlapping ranges throw,
because a blob is a
+ * distinct entity and two blobs never share bytes.
+ *
+ * @param rows Rows from the same file, sorted by (offset, length)
* @param maxGap Maximum gap to consider for merging
* @return Sequence of merged ranges
*/
@@ -331,9 +339,16 @@ class BatchedBlobReader(
currentStartOffset = row.offset
currentEndOffset = row.offset + row.length
currentRows = ArrayBuffer(row)
+ } else if (row.offset == currentRows.last.offset && row.length ==
currentRows.last.length) {
+ // Same descriptor as the previous row: one blob referenced by more
than one row (join
+ // fan-out, duplicate records). It is served from the current read and
the range does
+ // not grow.
+ currentRows += row
} else {
val gap = row.offset - currentEndOffset
- // Check for overlap
+ // A blob is a distinct entity, so two blobs never share bytes. Rows
are sorted by
+ // (offset, length) and identical descriptors were handled above, so a
start inside the
+ // current range means two different blobs overlap, which indicates
corruption.
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}"
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
index e9da416f5b76..c52f7db49218 100644
---
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
@@ -462,6 +462,30 @@ class TestBatchedBlobReader extends HoodieClientTestBase {
assertTrue(thrown.getCause.getMessage.contains("Overlapping blob ranges
detected"))
}
+ @Test
+ def testIdenticalRangesAreServedFromOneRead(): Unit = {
+ // A join fan-out or a duplicate record puts the same descriptor in one
task twice. That is one
+ // blob referenced by two rows, not two blobs sharing bytes, so the reader
must serve both from
+ // the single read of that range.
+ val filePath = createTestFile(tempDir, "identical.bin", 1000)
+ val inputDF = sparkSession.createDataFrame(Seq(
+ (filePath, 0L, 100L),
+ (filePath, 0L, 100L)
+ )).toDF("external_path", "offset", "length")
+ .withColumn("data", blobStructCol("data", col("external_path"),
col("offset"), col("length")))
+ .select("offset", "data")
+ .coalesce(1)
+
+ val results = BatchedBlobReader.readBatched(inputDF, storageConf).collect()
+
+ assertEquals(2, results.length)
+ results.foreach { row =>
+ val data = row.getAs[Array[Byte]]("data")
+ assertEquals(100, data.length)
+ assertBytesContent(data, expectedOffset = 0)
+ }
+ }
+
/**
* Blob references are absolute paths carried in row data, so the filesystem
a partition must read
* is not known until the rows arrive. These tests put the referenced files
behind an object-store
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
index e247e9758f2f..990ffbe11bf9 100644
---
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
@@ -253,6 +253,57 @@ class TestReadBlobSQL extends HoodieClientTestBase {
assertBytesContent(data1)
}
+ /**
+ * One blob row matching several rows on the other side of a join puts the
same descriptor
+ * into one task several times, and there is no shuffle between the join and
the batched
+ * read to collapse them. The reader must serve every row, not just the
first.
+ */
+ @Test
+ def testReadBlobWithJoinFanOut(): Unit = {
+ val filePath = createTestFile(tempDir, "join_fanout.bin", 10000)
+
+ // Blob side: one descriptor per id
+ 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_fanout")
+
+ // Event side: several rows per id, so each descriptor fans out across the
join
+ val eventsDF = sparkSession.createDataFrame(Seq(
+ (1, "e1"),
+ (1, "e2"),
+ (1, "e3"),
+ (2, "e4"),
+ (2, "e5")
+ )).toDF("id", "name")
+
+ eventsDF.createOrReplaceTempView("events_fanout")
+
+ val result = sparkSession.sql("""
+ SELECT e.id, e.name, read_blob(b.file_info) AS data
+ FROM events_fanout e
+ JOIN blob_table_fanout b ON e.id = b.id
+ ORDER BY e.id, e.name
+ """)
+
+ val rows = result.collect()
+ assertEquals(5, rows.length)
+
+ val expectedNames = Seq("e1", "e2", "e3", "e4", "e5")
+ rows.zipWithIndex.foreach { case (row, idx) =>
+ assertEquals(expectedNames(idx), row.getAs[String]("name"))
+ val data = row.getAs[Array[Byte]]("data")
+ assertEquals(100, data.length)
+ val expectedOffset = if (row.getAs[Int]("id") == 1) 0 else 100
+ assertBytesContent(data, expectedOffset = expectedOffset)
+ }
+ }
+
@Test
def testReadBlobInSubquery(): Unit = {
val filePath = createTestFile(tempDir, "subquery.bin", 10000)
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/blob/TestBatchedBlobReaderMerge.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/blob/TestBatchedBlobReaderMerge.scala
index 29aa75c89dbe..7d97b8a023f4 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/blob/TestBatchedBlobReaderMerge.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/blob/TestBatchedBlobReaderMerge.scala
@@ -180,4 +180,45 @@ class TestBatchedBlobReaderMerge {
() => reader().mergeRanges(rows, maxGap = 4096))
assertTrue(ex.getMessage.contains("Overlapping blob ranges detected"))
}
+
+ @Test
+ def testIdenticalRangesMergeIntoOne(): Unit = {
+ // One blob referenced by two rows (join fan-out, duplicate records): a
single read serves both
+ // and the range does not grow.
+ val rows = Seq(
+ row("/f", 0, 100, index = 0),
+ row("/f", 0, 100, index = 1))
+ val merged = reader().mergeRanges(rows, maxGap = 4096)
+ assertEquals(1, merged.size)
+ assertEquals(0L, merged.head.startOffset)
+ assertEquals(100L, merged.head.endOffset)
+ assertEquals(Seq(0L, 1L), merged.head.rows.map(_.index))
+ }
+
+ @Test
+ def testNestedRangeThrows(): Unit = {
+ // Containment is still an overlap: [0,512) then [0,1024) are two
different blobs sharing
+ // bytes. This is the shape that failed in TestLanceDataSource. Rows are
given in the order
+ // mergeRanges expects, sorted by (offset, length).
+ val rows = Seq(
+ row("/f", 0, 512, index = 0),
+ row("/f", 0, 1024, index = 1))
+ val ex = assertThrows(
+ classOf[IllegalArgumentException],
+ () => reader().mergeRanges(rows, maxGap = 4096))
+ assertTrue(ex.getMessage.contains("Overlapping blob ranges detected"))
+ }
+
+ @Test
+ def testIdenticalRangesDoNotWeakenOverlapCheck(): Unit = {
+ // A duplicate descriptor followed by a genuinely overlapping row still
throws
+ val rows = Seq(
+ row("/f", 0, 100, 0),
+ row("/f", 0, 100, 1),
+ row("/f", 50, 100, 2))
+ val ex = assertThrows(
+ classOf[IllegalArgumentException],
+ () => reader().mergeRanges(rows, maxGap = 4096))
+ assertTrue(ex.getMessage.contains("Overlapping blob ranges detected"))
+ }
}
