Skip to content

Commit c280807

Browse files
steFaizsteFaiz
authored
[spark] supports updating blobs through DataEvolution MergeInto (#8175)
Supports Spark: ```sql MERGE INTO t USING s ON ... WHEN MATCHED THEN UPDATE SET t.raw_blob = s.raw_blob ``` where raw_blob means blobs stored in BlobFormat Files ### Implementation Introduce several marker columns during data evolution: ```text update columns..., _ROW_ID, _FIRST_ROW_ID, marker columns... ``` This is because spark only allow literal columns for basic types. i.e. BlobPlaceholder is not allowed. Each blob column have one marker column, representing whether write blob values or `BlobPlaceholder.INSTANCE`
1 parent 40c57b9 commit c280807

17 files changed

Lines changed: 1057 additions & 199 deletions

File tree

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

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,8 +43,12 @@ private RowRangeIndex(List<Range> ranges) {
4343
}
4444

4545
public static RowRangeIndex create(List<Range> ranges) {
46+
return create(ranges, true);
47+
}
48+
49+
public static RowRangeIndex create(List<Range> ranges, boolean mergeAdjacent) {
4650
checkArgument(ranges != null, "Ranges cannot be null");
47-
return new RowRangeIndex(Range.sortAndMergeOverlap(ranges, true));
51+
return new RowRangeIndex(Range.sortAndMergeOverlap(ranges, mergeAdjacent));
4852
}
4953

5054
public List<Range> ranges() {
@@ -63,6 +67,13 @@ public boolean contains(Range range) {
6367
&& ends[candidate] >= range.to;
6468
}
6569

70+
public boolean containsExactly(Range range) {
71+
int candidate = lowerBound(starts, range.from);
72+
return candidate < starts.length
73+
&& starts[candidate] == range.from
74+
&& ends[candidate] == range.to;
75+
}
76+
6677
public List<Range> intersectedRanges(long start, long end) {
6778
int left = lowerBound(ends, start);
6879
if (left >= ranges.size()) {

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

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@ public class BlobFallbackRecordReader implements RecordReader<InternalRow> {
5656

5757
private final List<RecordReader<InternalRow>> groupReaders = new ArrayList<>();
5858
private final int blobIndex;
59+
private final int fieldCount;
5960
private boolean returned;
6061

6162
BlobFallbackRecordReader(
@@ -65,6 +66,7 @@ public class BlobFallbackRecordReader implements RecordReader<InternalRow> {
6566
RowType readRowType,
6667
int blobIndex) {
6768
this.blobIndex = blobIndex;
69+
this.fieldCount = readRowType.getFieldCount();
6870

6971
checkArgument(!files.isEmpty(), "Blob bunch should not be empty.");
7072
long firstRowId = Long.MAX_VALUE;
@@ -172,8 +174,7 @@ public InternalRow next() throws IOException {
172174
}
173175
}
174176
if (result == null) {
175-
throw new IllegalStateException(
176-
"Invalid state: all blob files at the same row id store a placeholder, it's a bug.");
177+
result = nullBlobRow();
177178
}
178179
return result;
179180
}
@@ -187,6 +188,10 @@ public void releaseBatch() {
187188
};
188189
}
189190

191+
private InternalRow nullBlobRow() {
192+
return new GenericRow(fieldCount);
193+
}
194+
190195
private boolean isPlaceHolder(InternalRow row) {
191196
return !row.isNullAt(blobIndex) && row.getBlob(blobIndex) == BlobPlaceholder.INSTANCE;
192197
}

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

Lines changed: 15 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -633,26 +633,23 @@ Optional<RuntimeException> checkRowIdExistence(
633633
return Optional.empty();
634634
}
635635

636-
Set<FileRowIdKey> existingIndex = new HashSet<>();
637-
for (SimpleFileEntry base : baseEntries) {
638-
if (base.firstRowId() != null) {
639-
existingIndex.add(
640-
new FileRowIdKey(
641-
base.partition(),
642-
base.bucket(),
643-
base.firstRowId(),
644-
base.rowCount()));
645-
}
646-
}
636+
List<Range> existingRanges =
637+
baseEntries.stream()
638+
.filter(
639+
base ->
640+
base.firstRowId() != null
641+
&& !dedicatedStorageFile(base.fileName()))
642+
.map(SimpleFileEntry::nonNullRowIdRange)
643+
.collect(Collectors.toList());
644+
RowRangeIndex existingIndex = RowRangeIndex.create(existingRanges, false);
647645

648646
for (SimpleFileEntry entry : filesToCheck) {
649-
FileRowIdKey key =
650-
new FileRowIdKey(
651-
entry.partition(),
652-
entry.bucket(),
653-
entry.firstRowId(),
654-
entry.rowCount());
655-
if (!existingIndex.contains(key)) {
647+
Range rowRange = entry.nonNullRowIdRange();
648+
boolean exists =
649+
dedicatedStorageFile(entry.fileName())
650+
? existingIndex.contains(rowRange)
651+
: existingIndex.containsExactly(rowRange);
652+
if (!exists) {
656653
return Optional.of(
657654
new RuntimeException(
658655
String.format(
@@ -670,40 +667,6 @@ Optional<RuntimeException> checkRowIdExistence(
670667
return Optional.empty();
671668
}
672669

673-
private static class FileRowIdKey {
674-
private final BinaryRow partition;
675-
private final int bucket;
676-
private final long firstRowId;
677-
private final long rowCount;
678-
679-
FileRowIdKey(BinaryRow partition, int bucket, long firstRowId, long rowCount) {
680-
this.partition = partition;
681-
this.bucket = bucket;
682-
this.firstRowId = firstRowId;
683-
this.rowCount = rowCount;
684-
}
685-
686-
@Override
687-
public boolean equals(Object o) {
688-
if (this == o) {
689-
return true;
690-
}
691-
if (o == null || getClass() != o.getClass()) {
692-
return false;
693-
}
694-
FileRowIdKey that = (FileRowIdKey) o;
695-
return bucket == that.bucket
696-
&& firstRowId == that.firstRowId
697-
&& rowCount == that.rowCount
698-
&& Objects.equals(partition, that.partition);
699-
}
700-
701-
@Override
702-
public int hashCode() {
703-
return Objects.hash(partition, bucket, firstRowId, rowCount);
704-
}
705-
}
706-
707670
private static boolean dedicatedStorageFile(String fileName) {
708671
return isBlobFile(fileName) || isVectorStoreFile(fileName);
709672
}

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

Lines changed: 14 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,6 @@
4545
import java.util.Set;
4646

4747
import static org.assertj.core.api.Assertions.assertThat;
48-
import static org.assertj.core.api.Assertions.assertThatThrownBy;
4948

5049
/** Tests for {@link BlobFallbackRecordReader}. */
5150
public class BlobFallbackRecordReaderTest {
@@ -134,18 +133,19 @@ public void testBlobFallbackRecordReaderDerivesRowIdBoundsFromFiles() throws Exc
134133
}
135134

136135
@Test
137-
public void testBlobFallbackRecordReaderThrowsIfAllRowsArePlaceholders() {
136+
public void testBlobFallbackRecordReaderReturnsNullIfAllRowsArePlaceholders() throws Exception {
138137
DataFileMeta newFile = blobFile("new-placeholder-file", 0, 1, 2);
139138
DataFileMeta oldFile = blobFile("old-placeholder-file", 0, 1, 1);
140139

141-
assertThatThrownBy(
142-
() ->
143-
readFallback(
144-
Arrays.asList(newFile, oldFile),
145-
null,
146-
placeholderRows(newFile, 0, oldFile, 0)))
147-
.isInstanceOf(IllegalStateException.class)
148-
.hasMessageContaining("all blob files at the same row id store a placeholder");
140+
ReadResult rows =
141+
readFallback(
142+
Arrays.asList(newFile, oldFile),
143+
null,
144+
placeholderRows(newFile, 0, oldFile, 0));
145+
146+
assertThat(rows.rowIds).isEmpty();
147+
assertThat(rows.nullBlobRowCount).isEqualTo(1);
148+
assertThat(rows.placeholderRowCount).isEqualTo(0);
149149
}
150150

151151
@Test
@@ -363,6 +363,7 @@ private static class ReadResult {
363363
final List<Long> sequenceNumbers = new ArrayList<>();
364364
final List<Integer> batchSizes = new ArrayList<>();
365365
int placeholderRowCount;
366+
int nullBlobRowCount;
366367

367368
static ReadResult read(RecordReader<InternalRow> reader) throws Exception {
368369
try {
@@ -385,7 +386,9 @@ static ReadResult read(RecordReader<InternalRow> reader) throws Exception {
385386
}
386387

387388
private void add(InternalRow row) {
388-
if (row.getBlob(BLOB_INDEX) == BlobPlaceholder.INSTANCE) {
389+
if (row.isNullAt(BLOB_INDEX)) {
390+
nullBlobRowCount++;
391+
} else if (row.getBlob(BLOB_INDEX) == BlobPlaceholder.INSTANCE) {
389392
placeholderRowCount++;
390393
} else {
391394
rowIds.add(row.getLong(1));

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

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -454,6 +454,85 @@ void testCheckRowIdExistenceBaseFileRewritten() {
454454
assertThat(result.get().getMessage()).contains("Row ID existence conflict");
455455
}
456456

457+
@Test
458+
void testCheckRowIdExistenceNormalFileRejectsAdjacentDataFiles() {
459+
ConflictDetection detection = createConflictDetection();
460+
461+
List<SimpleFileEntry> baseEntries = new ArrayList<>();
462+
baseEntries.add(createFileEntryWithRowId("f1", ADD, 0L, 2L));
463+
baseEntries.add(createFileEntryWithRowId("f2", ADD, 2L, 2L));
464+
465+
List<SimpleFileEntry> deltaEntries = new ArrayList<>();
466+
deltaEntries.add(createFileEntryWithRowId("p1", ADD, 0L, 4L));
467+
468+
Optional<RuntimeException> result =
469+
detection.checkRowIdExistence(baseEntries, deltaEntries, 4L);
470+
assertThat(result).isPresent();
471+
assertThat(result.get().getMessage()).contains("Row ID existence conflict");
472+
}
473+
474+
@Test
475+
void testCheckRowIdExistenceDedicatedFileCoveredByDataFiles() {
476+
ConflictDetection detection = createConflictDetection();
477+
478+
List<SimpleFileEntry> baseEntries = new ArrayList<>();
479+
baseEntries.add(createFileEntryWithRowId("f1", ADD, 0L, 4L));
480+
481+
List<SimpleFileEntry> deltaEntries = new ArrayList<>();
482+
deltaEntries.add(createFileEntryWithRowId("p1.blob", ADD, 0L, 2L));
483+
484+
assertThat(detection.checkRowIdExistence(baseEntries, deltaEntries, 4L)).isEmpty();
485+
}
486+
487+
@Test
488+
void testCheckRowIdExistenceDedicatedFileRejectsAdjacentDataFiles() {
489+
ConflictDetection detection = createConflictDetection();
490+
491+
List<SimpleFileEntry> baseEntries = new ArrayList<>();
492+
baseEntries.add(createFileEntryWithRowId("f1", ADD, 0L, 2L));
493+
baseEntries.add(createFileEntryWithRowId("f2", ADD, 2L, 2L));
494+
495+
List<SimpleFileEntry> deltaEntries = new ArrayList<>();
496+
deltaEntries.add(createFileEntryWithRowId("p1.blob", ADD, 0L, 4L));
497+
498+
Optional<RuntimeException> result =
499+
detection.checkRowIdExistence(baseEntries, deltaEntries, 4L);
500+
assertThat(result).isPresent();
501+
assertThat(result.get().getMessage()).contains("Row ID existence conflict");
502+
}
503+
504+
@Test
505+
void testCheckRowIdExistenceDedicatedFileRejectsRangeNotCoveredByOneDataFile() {
506+
ConflictDetection detection = createConflictDetection();
507+
508+
List<SimpleFileEntry> baseEntries = new ArrayList<>();
509+
baseEntries.add(createFileEntryWithRowId("f1", ADD, 0L, 2L));
510+
511+
List<SimpleFileEntry> deltaEntries = new ArrayList<>();
512+
deltaEntries.add(createFileEntryWithRowId("p1.blob", ADD, 0L, 3L));
513+
514+
Optional<RuntimeException> result =
515+
detection.checkRowIdExistence(baseEntries, deltaEntries, 3L);
516+
assertThat(result).isPresent();
517+
assertThat(result.get().getMessage()).contains("Row ID existence conflict");
518+
}
519+
520+
@Test
521+
void testCheckRowIdExistenceDedicatedFileIgnoresBaseDedicatedFiles() {
522+
ConflictDetection detection = createConflictDetection();
523+
524+
List<SimpleFileEntry> baseEntries = new ArrayList<>();
525+
baseEntries.add(createFileEntryWithRowId("old.blob", ADD, 0L, 2L));
526+
527+
List<SimpleFileEntry> deltaEntries = new ArrayList<>();
528+
deltaEntries.add(createFileEntryWithRowId("p1.blob", ADD, 0L, 2L));
529+
530+
Optional<RuntimeException> result =
531+
detection.checkRowIdExistence(baseEntries, deltaEntries, 2L);
532+
assertThat(result).isPresent();
533+
assertThat(result.get().getMessage()).contains("Row ID existence conflict");
534+
}
535+
457536
@Test
458537
void testCheckRowIdExistenceSkipsNewlyAppendedFiles() {
459538
ConflictDetection detection = createConflictDetection();

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

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.paimon.spark.sql
20+
21+
class BlobUpdateTest extends BlobUpdateTestBase {}

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

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.paimon.spark.sql
20+
21+
class BlobUpdateTest extends BlobUpdateTestBase {}

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

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.paimon.spark.sql
20+
21+
class BlobUpdateTest extends BlobUpdateTestBase {}

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

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.paimon.spark.sql
20+
21+
class BlobUpdateTest extends BlobUpdateTestBase {}

0 commit comments

Comments
 (0)