Skip to content

Commit b43f513

Browse files
steFaizsteFaiz
committed
optim
1 parent ff62c57 commit b43f513

10 files changed

Lines changed: 656 additions & 162 deletions

File tree

‎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: 24 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -54,9 +54,12 @@
5454
import java.util.LinkedHashMap;
5555
import java.util.List;
5656
import java.util.Map;
57+
import java.util.Map.Entry;
58+
import java.util.NavigableMap;
5759
import java.util.Objects;
5860
import java.util.Optional;
5961
import java.util.Set;
62+
import java.util.TreeMap;
6063
import java.util.function.Function;
6164
import java.util.stream.Collectors;
6265

@@ -566,26 +569,21 @@ Optional<RuntimeException> checkRowIdExistence(
566569
return Optional.empty();
567570
}
568571

569-
Set<FileRowIdKey> existingIndex = new HashSet<>();
572+
NavigableMap<Long, Long> existingRanges = new TreeMap<>();
570573
for (SimpleFileEntry base : baseEntries) {
571-
if (base.firstRowId() != null) {
572-
existingIndex.add(
573-
new FileRowIdKey(
574-
base.partition(),
575-
base.bucket(),
576-
base.firstRowId(),
577-
base.rowCount()));
574+
if (base.firstRowId() != null && !dedicatedStorageFile(base.fileName())) {
575+
existingRanges.put(base.firstRowId(), base.rowCount());
578576
}
579577
}
580578

581579
for (SimpleFileEntry entry : filesToCheck) {
582-
FileRowIdKey key =
583-
new FileRowIdKey(
584-
entry.partition(),
585-
entry.bucket(),
586-
entry.firstRowId(),
587-
entry.rowCount());
588-
if (!existingIndex.contains(key)) {
580+
boolean exists =
581+
dedicatedStorageFile(entry.fileName())
582+
? rowIdRangeCovered(
583+
existingRanges, entry.firstRowId(), entry.rowCount())
584+
: rowIdRangeExists(
585+
existingRanges, entry.firstRowId(), entry.rowCount());
586+
if (!exists) {
589587
return Optional.of(
590588
new RuntimeException(
591589
String.format(
@@ -603,38 +601,18 @@ Optional<RuntimeException> checkRowIdExistence(
603601
return Optional.empty();
604602
}
605603

606-
private static class FileRowIdKey {
607-
private final BinaryRow partition;
608-
private final int bucket;
609-
private final long firstRowId;
610-
private final long rowCount;
611-
612-
FileRowIdKey(BinaryRow partition, int bucket, long firstRowId, long rowCount) {
613-
this.partition = partition;
614-
this.bucket = bucket;
615-
this.firstRowId = firstRowId;
616-
this.rowCount = rowCount;
617-
}
618-
619-
@Override
620-
public boolean equals(Object o) {
621-
if (this == o) {
622-
return true;
623-
}
624-
if (o == null || getClass() != o.getClass()) {
625-
return false;
626-
}
627-
FileRowIdKey that = (FileRowIdKey) o;
628-
return bucket == that.bucket
629-
&& firstRowId == that.firstRowId
630-
&& rowCount == that.rowCount
631-
&& Objects.equals(partition, that.partition);
632-
}
604+
private static boolean rowIdRangeCovered(
605+
NavigableMap<Long, Long> ranges, long firstRowId, long rowCount) {
606+
Entry<Long, Long> range = ranges.floorEntry(firstRowId);
607+
return range != null
608+
&& range.getKey() <= firstRowId
609+
&& range.getKey() + range.getValue() >= firstRowId + rowCount;
610+
}
633611

634-
@Override
635-
public int hashCode() {
636-
return Objects.hash(partition, bucket, firstRowId, rowCount);
637-
}
612+
private static boolean rowIdRangeExists(
613+
NavigableMap<Long, Long> ranges, long firstRowId, long rowCount) {
614+
Long existingRowCount = ranges.get(firstRowId);
615+
return existingRowCount != null && existingRowCount == rowCount;
638616
}
639617

640618
private static boolean dedicatedStorageFile(String fileName) {

‎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: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -419,6 +419,51 @@ void testCheckRowIdExistenceBaseFileRewritten() {
419419
assertThat(result.get().getMessage()).contains("Row ID existence conflict");
420420
}
421421

422+
@Test
423+
void testCheckRowIdExistenceDedicatedFileCoveredByDataFiles() {
424+
ConflictDetection detection = createConflictDetection();
425+
426+
List<SimpleFileEntry> baseEntries = new ArrayList<>();
427+
baseEntries.add(createFileEntryWithRowId("f1", ADD, 0L, 4L));
428+
429+
List<SimpleFileEntry> deltaEntries = new ArrayList<>();
430+
deltaEntries.add(createFileEntryWithRowId("p1.blob", ADD, 0L, 2L));
431+
432+
assertThat(detection.checkRowIdExistence(baseEntries, deltaEntries, 4L)).isEmpty();
433+
}
434+
435+
@Test
436+
void testCheckRowIdExistenceDedicatedFileRejectsRangeNotCoveredByOneDataFile() {
437+
ConflictDetection detection = createConflictDetection();
438+
439+
List<SimpleFileEntry> baseEntries = new ArrayList<>();
440+
baseEntries.add(createFileEntryWithRowId("f1", ADD, 0L, 2L));
441+
442+
List<SimpleFileEntry> deltaEntries = new ArrayList<>();
443+
deltaEntries.add(createFileEntryWithRowId("p1.blob", ADD, 0L, 3L));
444+
445+
Optional<RuntimeException> result =
446+
detection.checkRowIdExistence(baseEntries, deltaEntries, 3L);
447+
assertThat(result).isPresent();
448+
assertThat(result.get().getMessage()).contains("Row ID existence conflict");
449+
}
450+
451+
@Test
452+
void testCheckRowIdExistenceDedicatedFileIgnoresBaseDedicatedFiles() {
453+
ConflictDetection detection = createConflictDetection();
454+
455+
List<SimpleFileEntry> baseEntries = new ArrayList<>();
456+
baseEntries.add(createFileEntryWithRowId("old.blob", ADD, 0L, 2L));
457+
458+
List<SimpleFileEntry> deltaEntries = new ArrayList<>();
459+
deltaEntries.add(createFileEntryWithRowId("p1.blob", ADD, 0L, 2L));
460+
461+
Optional<RuntimeException> result =
462+
detection.checkRowIdExistence(baseEntries, deltaEntries, 2L);
463+
assertThat(result).isPresent();
464+
assertThat(result.get().getMessage()).contains("Row ID existence conflict");
465+
}
466+
422467
@Test
423468
void testCheckRowIdExistenceSkipsNewlyAppendedFiles() {
424469
ConflictDetection detection = createConflictDetection();

‎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 {}

‎paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPaimonWriter.scala‎

Lines changed: 34 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ import org.apache.paimon.types.VectorType.isVectorStoreFile
3030
import org.apache.paimon.utils.SerializationUtils
3131

3232
import org.apache.spark.sql._
33+
import org.apache.spark.sql.catalyst.analysis.SimpleAnalyzer.resolver
3334

3435
import java.util.Collections
3536

@@ -43,23 +44,42 @@ case class DataEvolutionPaimonWriter(paimonTable: FileStoreTable, dataSplits: Se
4344
override val table: FileStoreTable =
4445
paimonTable.copy(Collections.singletonMap(CoreOptions.TARGET_FILE_SIZE.key(), "99999 G"))
4546

46-
def writePartialFields(data: DataFrame, columnNames: Seq[String]): Seq[CommitMessage] = {
47+
def writePartialFields(
48+
data: DataFrame,
49+
columnNames: Seq[String],
50+
rawBlobPlaceholderMarkerColumns: Map[String, String] = Map.empty): Seq[CommitMessage] = {
4751
val sparkSession = data.sparkSession
4852
import sparkSession.implicits._
49-
assert(data.columns.length == columnNames.size + 2)
53+
assert(data.columns.length == columnNames.size + 2 + rawBlobPlaceholderMarkerColumns.size)
5054
val writeType = table.rowType().project(columnNames.asJava)
5155

5256
val options = new CoreOptions(table.schema().options())
53-
val updatableBlobFields = options.updatableBlobFields()
54-
val hasRawDataBlob = writeType.getFields.asScala.exists(
55-
f => f.`type`().is(BLOB) && !updatableBlobFields.contains(f.name()))
56-
if (hasRawDataBlob) {
57-
throw new UnsupportedOperationException(
58-
"DataEvolution does not support writing partial columns with raw-data BLOB type. " +
59-
"Only descriptor-based BLOB columns (configured via '" +
60-
CoreOptions.BLOB_DESCRIPTOR_FIELD.key() + "' or '" +
61-
CoreOptions.BLOB_VIEW_FIELD.key() + "' or '" +
62-
CoreOptions.BLOB_EXTERNAL_STORAGE_FIELD.key() + "') can be updated.")
57+
val blobInlineFields = options.blobInlineField().asScala.toSeq
58+
// Maps from blob field index to corresponding marker column index
59+
val rawBlobPlaceholderMarkerIndexes = writeType.getFields.asScala.flatMap {
60+
field =>
61+
if (
62+
field.`type`().is(BLOB) &&
63+
!blobInlineFields.exists(inlineField => resolver(inlineField, field.name()))
64+
) {
65+
val markerColumn = rawBlobPlaceholderMarkerColumns.getOrElse(
66+
field.name(),
67+
throw new UnsupportedOperationException(
68+
"DataEvolution raw-data BLOB partial writes require an internal placeholder marker " +
69+
s"for column ${field.name()}.")
70+
)
71+
Some(writeType.getFieldIndex(field.name()) -> data.schema.fieldIndex(markerColumn))
72+
} else {
73+
None
74+
}
75+
}.toMap
76+
val unusedMarkerColumns =
77+
rawBlobPlaceholderMarkerColumns.keySet -- rawBlobPlaceholderMarkerIndexes.keys.map(
78+
index => writeType.getFields.get(index).name())
79+
if (unusedMarkerColumns.nonEmpty) {
80+
throw new IllegalArgumentException(
81+
"Raw BLOB placeholder markers do not match partial write columns: " +
82+
unusedMarkerColumns.toSeq.sorted.mkString(", "))
6383
}
6484

6585
val firstRowIdToPartitionMap = new mutable.HashMap[Long, (Array[Byte], Long)]
@@ -91,7 +111,8 @@ case class DataEvolutionPaimonWriter(paimonTable: FileStoreTable, dataSplits: Se
91111
writeBuilder,
92112
writeType,
93113
firstRowIdToPartitionMapBroadcast.value,
94-
catalogContextForBlobDescriptor)
114+
catalogContextForBlobDescriptor,
115+
rawBlobPlaceholderMarkerIndexes)
95116
try {
96117
iter.foreach(row => write.write(row))
97118
Iterator.apply(write.commit)

0 commit comments

Comments
 (0)