[spark] supports updating blobs through DataEvolution MergeInto

paimon-common/src/main/java/org/apache/paimon/utils/RowRangeIndex.java

@@ -43,8 +43,12 @@ private RowRangeIndex(List<Range> ranges) {
     }
 
     public static RowRangeIndex create(List<Range> ranges) {
+        return create(ranges, true);
+    }
+
+    public static RowRangeIndex create(List<Range> ranges, boolean mergeAdjacent) {
         checkArgument(ranges != null, "Ranges cannot be null");
-        return new RowRangeIndex(Range.sortAndMergeOverlap(ranges, true));
+        return new RowRangeIndex(Range.sortAndMergeOverlap(ranges, mergeAdjacent));
     }
 
     public List<Range> ranges() {
@@ -63,6 +67,13 @@ public boolean contains(Range range) {
                 && ends[candidate] >= range.to;
     }
 
+    public boolean containsExactly(Range range) {
+        int candidate = lowerBound(starts, range.from);
+        return candidate < starts.length
+                && starts[candidate] == range.from
+                && ends[candidate] == range.to;
+    }
+
     public List<Range> intersectedRanges(long start, long end) {
         int left = lowerBound(ends, start);
         if (left >= ranges.size()) {

paimon-core/src/main/java/org/apache/paimon/operation/BlobFallbackRecordReader.java

@@ -56,6 +56,7 @@ public class BlobFallbackRecordReader implements RecordReader<InternalRow> {
 
     private final List<RecordReader<InternalRow>> groupReaders = new ArrayList<>();
     private final int blobIndex;
+    private final int fieldCount;
     private boolean returned;
 
     BlobFallbackRecordReader(
@@ -65,6 +66,7 @@ public class BlobFallbackRecordReader implements RecordReader<InternalRow> {
             RowType readRowType,
             int blobIndex) {
         this.blobIndex = blobIndex;
+        this.fieldCount = readRowType.getFieldCount();
 
         checkArgument(!files.isEmpty(), "Blob bunch should not be empty.");
         long firstRowId = Long.MAX_VALUE;
@@ -172,8 +174,7 @@ public InternalRow next() throws IOException {
                     }
                 }
                 if (result == null) {
-                    throw new IllegalStateException(
-                            "Invalid state: all blob files at the same row id store a placeholder, it's a bug.");
+                    result = nullBlobRow();
                 }
                 return result;
             }
@@ -187,6 +188,10 @@ public void releaseBatch() {
         };
     }
 
+    private InternalRow nullBlobRow() {
+        return new GenericRow(fieldCount);
+    }
+
     private boolean isPlaceHolder(InternalRow row) {
         return !row.isNullAt(blobIndex) && row.getBlob(blobIndex) == BlobPlaceholder.INSTANCE;
     }

paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java

@@ -633,26 +633,23 @@ Optional<RuntimeException> checkRowIdExistence(
             return Optional.empty();
         }
 
-        Set<FileRowIdKey> existingIndex = new HashSet<>();
-        for (SimpleFileEntry base : baseEntries) {
-            if (base.firstRowId() != null) {
-                existingIndex.add(
-                        new FileRowIdKey(
-                                base.partition(),
-                                base.bucket(),
-                                base.firstRowId(),
-                                base.rowCount()));
-            }
-        }
+        List<Range> existingRanges =
+                baseEntries.stream()
+                        .filter(
+                                base ->
+                                        base.firstRowId() != null
+                                                && !dedicatedStorageFile(base.fileName()))
+                        .map(SimpleFileEntry::nonNullRowIdRange)
+                        .collect(Collectors.toList());
+        RowRangeIndex existingIndex = RowRangeIndex.create(existingRanges, false);
 
         for (SimpleFileEntry entry : filesToCheck) {
-            FileRowIdKey key =
-                    new FileRowIdKey(
-                            entry.partition(),
-                            entry.bucket(),
-                            entry.firstRowId(),
-                            entry.rowCount());
-            if (!existingIndex.contains(key)) {
+            Range rowRange = entry.nonNullRowIdRange();
+            boolean exists =
+                    dedicatedStorageFile(entry.fileName())
+                            ? existingIndex.contains(rowRange)
+                            : existingIndex.containsExactly(rowRange);
+            if (!exists) {
                 return Optional.of(
                         new RuntimeException(
                                 String.format(
@@ -670,40 +667,6 @@ Optional<RuntimeException> checkRowIdExistence(
         return Optional.empty();
     }
 
-    private static class FileRowIdKey {
-        private final BinaryRow partition;
-        private final int bucket;
-        private final long firstRowId;
-        private final long rowCount;
-
-        FileRowIdKey(BinaryRow partition, int bucket, long firstRowId, long rowCount) {
-            this.partition = partition;
-            this.bucket = bucket;
-            this.firstRowId = firstRowId;
-            this.rowCount = rowCount;
-        }
-
-        @Override
-        public boolean equals(Object o) {
-            if (this == o) {
-                return true;
-            }
-            if (o == null || getClass() != o.getClass()) {
-                return false;
-            }
-            FileRowIdKey that = (FileRowIdKey) o;
-            return bucket == that.bucket
-                    && firstRowId == that.firstRowId
-                    && rowCount == that.rowCount
-                    && Objects.equals(partition, that.partition);
-        }
-
-        @Override
-        public int hashCode() {
-            return Objects.hash(partition, bucket, firstRowId, rowCount);
-        }
-    }
-
     private static boolean dedicatedStorageFile(String fileName) {
         return isBlobFile(fileName) || isVectorStoreFile(fileName);
     }

paimon-core/src/test/java/org/apache/paimon/operation/BlobFallbackRecordReaderTest.java

@@ -45,7 +45,6 @@
 import java.util.Set;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Tests for {@link BlobFallbackRecordReader}. */
 public class BlobFallbackRecordReaderTest {
@@ -134,18 +133,19 @@ public void testBlobFallbackRecordReaderDerivesRowIdBoundsFromFiles() throws Exc
     }
 
     @Test
-    public void testBlobFallbackRecordReaderThrowsIfAllRowsArePlaceholders() {
+    public void testBlobFallbackRecordReaderReturnsNullIfAllRowsArePlaceholders() throws Exception {
         DataFileMeta newFile = blobFile("new-placeholder-file", 0, 1, 2);
         DataFileMeta oldFile = blobFile("old-placeholder-file", 0, 1, 1);
 
-        assertThatThrownBy(
-                        () ->
-                                readFallback(
-                                        Arrays.asList(newFile, oldFile),
-                                        null,
-                                        placeholderRows(newFile, 0, oldFile, 0)))
-                .isInstanceOf(IllegalStateException.class)
-                .hasMessageContaining("all blob files at the same row id store a placeholder");
+        ReadResult rows =
+                readFallback(
+                        Arrays.asList(newFile, oldFile),
+                        null,
+                        placeholderRows(newFile, 0, oldFile, 0));
+
+        assertThat(rows.rowIds).isEmpty();
+        assertThat(rows.nullBlobRowCount).isEqualTo(1);
+        assertThat(rows.placeholderRowCount).isEqualTo(0);
     }
 
     @Test
@@ -363,6 +363,7 @@ private static class ReadResult {
         final List<Long> sequenceNumbers = new ArrayList<>();
         final List<Integer> batchSizes = new ArrayList<>();
         int placeholderRowCount;
+        int nullBlobRowCount;
 
         static ReadResult read(RecordReader<InternalRow> reader) throws Exception {
             try {
@@ -385,7 +386,9 @@ static ReadResult read(RecordReader<InternalRow> reader) throws Exception {
         }
 
         private void add(InternalRow row) {
-            if (row.getBlob(BLOB_INDEX) == BlobPlaceholder.INSTANCE) {
+            if (row.isNullAt(BLOB_INDEX)) {
+                nullBlobRowCount++;
+            } else if (row.getBlob(BLOB_INDEX) == BlobPlaceholder.INSTANCE) {
                 placeholderRowCount++;
             } else {
                 rowIds.add(row.getLong(1));

paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java

@@ -454,6 +454,85 @@ void testCheckRowIdExistenceBaseFileRewritten() {
         assertThat(result.get().getMessage()).contains("Row ID existence conflict");
     }
 
+    @Test
+    void testCheckRowIdExistenceNormalFileRejectsAdjacentDataFiles() {
+        ConflictDetection detection = createConflictDetection();
+
+        List<SimpleFileEntry> baseEntries = new ArrayList<>();
+        baseEntries.add(createFileEntryWithRowId("f1", ADD, 0L, 2L));
+        baseEntries.add(createFileEntryWithRowId("f2", ADD, 2L, 2L));
+
+        List<SimpleFileEntry> deltaEntries = new ArrayList<>();
+        deltaEntries.add(createFileEntryWithRowId("p1", ADD, 0L, 4L));
+
+        Optional<RuntimeException> result =
+                detection.checkRowIdExistence(baseEntries, deltaEntries, 4L);
+        assertThat(result).isPresent();
+        assertThat(result.get().getMessage()).contains("Row ID existence conflict");
+    }
+
+    @Test
+    void testCheckRowIdExistenceDedicatedFileCoveredByDataFiles() {
+        ConflictDetection detection = createConflictDetection();
+
+        List<SimpleFileEntry> baseEntries = new ArrayList<>();
+        baseEntries.add(createFileEntryWithRowId("f1", ADD, 0L, 4L));
+
+        List<SimpleFileEntry> deltaEntries = new ArrayList<>();
+        deltaEntries.add(createFileEntryWithRowId("p1.blob", ADD, 0L, 2L));
+
+        assertThat(detection.checkRowIdExistence(baseEntries, deltaEntries, 4L)).isEmpty();
+    }
+
+    @Test
+    void testCheckRowIdExistenceDedicatedFileRejectsAdjacentDataFiles() {
+        ConflictDetection detection = createConflictDetection();
+
+        List<SimpleFileEntry> baseEntries = new ArrayList<>();
+        baseEntries.add(createFileEntryWithRowId("f1", ADD, 0L, 2L));
+        baseEntries.add(createFileEntryWithRowId("f2", ADD, 2L, 2L));
+
+        List<SimpleFileEntry> deltaEntries = new ArrayList<>();
+        deltaEntries.add(createFileEntryWithRowId("p1.blob", ADD, 0L, 4L));
+
+        Optional<RuntimeException> result =
+                detection.checkRowIdExistence(baseEntries, deltaEntries, 4L);
+        assertThat(result).isPresent();
+        assertThat(result.get().getMessage()).contains("Row ID existence conflict");
+    }
+
+    @Test
+    void testCheckRowIdExistenceDedicatedFileRejectsRangeNotCoveredByOneDataFile() {
+        ConflictDetection detection = createConflictDetection();
+
+        List<SimpleFileEntry> baseEntries = new ArrayList<>();
+        baseEntries.add(createFileEntryWithRowId("f1", ADD, 0L, 2L));
+
+        List<SimpleFileEntry> deltaEntries = new ArrayList<>();
+        deltaEntries.add(createFileEntryWithRowId("p1.blob", ADD, 0L, 3L));
+
+        Optional<RuntimeException> result =
+                detection.checkRowIdExistence(baseEntries, deltaEntries, 3L);
+        assertThat(result).isPresent();
+        assertThat(result.get().getMessage()).contains("Row ID existence conflict");
+    }
+
+    @Test
+    void testCheckRowIdExistenceDedicatedFileIgnoresBaseDedicatedFiles() {
+        ConflictDetection detection = createConflictDetection();
+
+        List<SimpleFileEntry> baseEntries = new ArrayList<>();
+        baseEntries.add(createFileEntryWithRowId("old.blob", ADD, 0L, 2L));
+
+        List<SimpleFileEntry> deltaEntries = new ArrayList<>();
+        deltaEntries.add(createFileEntryWithRowId("p1.blob", ADD, 0L, 2L));
+
+        Optional<RuntimeException> result =
+                detection.checkRowIdExistence(baseEntries, deltaEntries, 2L);
+        assertThat(result).isPresent();
+        assertThat(result.get().getMessage()).contains("Row ID existence conflict");
+    }
+
     @Test
     void testCheckRowIdExistenceSkipsNewlyAppendedFiles() {
         ConflictDetection detection = createConflictDetection();

paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/BlobUpdateTest.scala

@@ -0,0 +1,21 @@
+/*
+ * 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.paimon.spark.sql
+
+class BlobUpdateTest extends BlobUpdateTestBase {}

paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/BlobUpdateTest.scala

@@ -0,0 +1,21 @@
+/*
+ * 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.paimon.spark.sql
+
+class BlobUpdateTest extends BlobUpdateTestBase {}

paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/BlobUpdateTest.scala

@@ -0,0 +1,21 @@
+/*
+ * 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.paimon.spark.sql
+
+class BlobUpdateTest extends BlobUpdateTestBase {}

paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/BlobUpdateTest.scala

@@ -0,0 +1,21 @@
+/*
+ * 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.paimon.spark.sql
+
+class BlobUpdateTest extends BlobUpdateTestBase {}