This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 77c9f4e864 [core][spark] Fix reading newly added BLOB columns after
MERGE (#9413)
77c9f4e864 is described below
commit 77c9f4e864f8270ee6f207f0e0c64fa0981030f1
Author: zhoulii <[email protected]>
AuthorDate: Wed Aug 26 22:44:11 2026 +0800
[core][spark] Fix reading newly added BLOB columns after MERGE (#9413)
---
.../paimon/operation/DataEvolutionSplitRead.java | 31 ----------------------
.../operation/BlobFallbackRecordReaderTest.java | 15 +++++++++++
.../paimon/operation/DataEvolutionReadTest.java | 1 -
.../paimon/spark/sql/BlobUpdateTestBase.scala | 28 +++++++++++++++++++
4 files changed, 43 insertions(+), 32 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
index 2a606ee677..9e9959a3c6 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java
@@ -435,18 +435,7 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
deletionVector);
} else if (bunch instanceof BlobFileBunch) {
// for blob bunch, fallback on placeholders
-
- // fast path: only contains one max_seq group
BlobFileBunch blobBunch = (BlobFileBunch) bunch;
- if (blobBunch.sequentialReadOptimize()) {
- return sequentialReadFiles(
- bunch.files(),
- partition,
- dataFilePathFactory,
- formatReaderMapping,
- rowRanges,
- deletionVector);
- }
int blobIndex = findBlobFieldIndex(readRowType);
checkArgument(blobIndex >= 0, "Blob bunch read type should contain
a blob field.");
return new BlobFallbackRecordReader(
@@ -1145,26 +1134,6 @@ public class DataEvolutionSplitRead implements
SplitRead<InternalRow> {
return new Range(merged.get(0).from, merged.get(merged.size() -
1).to);
}
- private boolean fullyCoversLogicalRange() {
- List<Range> merged = Range.sortAndMergeOverlap(ranges, true);
- return merged.size() == 1 && merged.get(0).equals(logicalRange());
- }
-
- public boolean sequentialReadOptimize() {
- Preconditions.checkState(!files.isEmpty(), "Blob file bunch should
not be empty.");
-
- // If blob files share the same max_seq_num, we could sequentially
read them.
- // Files have already been sorted by first_row_id
- long maxSeq = files.get(0).maxSequenceNumber();
- for (int i = 1; i < files.size(); i++) {
- if (files.get(i).maxSequenceNumber() != maxSeq) {
- return false;
- }
- }
-
- return fullyCoversLogicalRange();
- }
-
@Override
public List<DataFileMeta> files() {
return files;
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/BlobFallbackRecordReaderTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/BlobFallbackRecordReaderTest.java
index 328aaa3061..7d58ac8517 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/BlobFallbackRecordReaderTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/BlobFallbackRecordReaderTest.java
@@ -357,6 +357,21 @@ public class BlobFallbackRecordReaderTest {
assertThat(rows.nullBlobRowIds).isEmpty();
}
+ @Test
+ public void
testBlobFallbackRecordReaderResolvesPlaceholderInSingleFullRangeGroup()
+ throws Exception {
+ DataFileMeta file = blobFile("single-full-range-file", 0, 3, 1);
+
+ ReadResult rows =
+ readFallback(Collections.singletonList(file), null,
placeholderRows(file, 1));
+
+ assertThat(rows.rowIds).containsExactly(0L, 2L);
+ assertThat(rows.sequenceNumbers).containsExactly(1L, 1L);
+ assertThat(rows.nullBlobRowIds).containsExactly(1L);
+ assertThat(rows.nullBlobSequenceNumbers).containsExactly(-1L);
+ assertThat(rows.placeholderRowCount).isZero();
+ }
+
@Test
public void
testBlobFallbackRecordReaderReturnsNullIfAllRowsArePlaceholders() throws
Exception {
DataFileMeta newFile = blobFile("new-placeholder-file", 0, 1, 2);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionReadTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionReadTest.java
index bdb9fcc935..748dbe7bdf 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionReadTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionReadTest.java
@@ -485,7 +485,6 @@ public class DataEvolutionReadTest {
assertThat(blobBunch.rowCount()).isEqualTo(10);
assertThat(blobBunch.logicalRange()).isEqualTo(new Range(100, 109));
- assertThat(blobBunch.sequentialReadOptimize()).isFalse();
}
@Test
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/BlobUpdateTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/BlobUpdateTestBase.scala
index f9d6597c94..9e76ca8bf5 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/BlobUpdateTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/BlobUpdateTestBase.scala
@@ -60,6 +60,34 @@ class BlobUpdateTestBase extends PaimonSparkTestBase {
}
}
+ test("Blob: merge-into updates newly added BLOB column") {
+ withTable("s", "t") {
+ sql(
+ "CREATE TABLE t (id INT, name STRING) TBLPROPERTIES " +
+ "('row-tracking.enabled'='true', 'data-evolution.enabled'='true')")
+ sql(
+ "INSERT INTO t SELECT /*+ REPARTITION(1) */ * FROM VALUES " +
+ "(1, 'name1'), (2, 'name2'), (3, 'name3') AS v(id, name)")
+
+ sql("ALTER TABLE t ADD COLUMN picture BINARY COMMENT '__BLOB_FIELD'")
+ sql("INSERT INTO t VALUES (4, 'name4', CAST(NULL AS BINARY))")
+
+ sql("CREATE TABLE s (id INT, picture BINARY)")
+ sql("INSERT INTO s VALUES (1, X'4E4557')")
+ sql("""
+ |MERGE INTO t
+ |USING s
+ |ON t.id = s.id
+ |WHEN MATCHED THEN UPDATE SET t.picture = s.picture
+ |""".stripMargin)
+
+ checkAnswer(
+ sql("SELECT id, picture FROM t WHERE id <= 3 ORDER BY id"),
+ Seq(Row(1, Array[Byte](78, 69, 87)), Row(2, null), Row(3, null))
+ )
+ }
+ }
+
test("Blob: merge-into updates raw-data array blob column") {
withTable("s", "t") {
sql("CREATE TABLE t (id INT, name STRING, pictures ARRAY<BINARY>)
TBLPROPERTIES " +
