diff --git a/docs/docs/multimodal-table/data-evolution-maintenance.mdx b/docs/docs/multimodal-table/data-evolution-maintenance.mdx index f0fe00407750..bdb1310b7ce3 100644 --- a/docs/docs/multimodal-table/data-evolution-maintenance.mdx +++ b/docs/docs/multimodal-table/data-evolution-maintenance.mdx @@ -78,6 +78,12 @@ nor include more than one file smaller than `vector.target-file-size`. See [Dedicated Compaction](../maintenance/dedicated-compaction) for scheduling compaction jobs. +`data-evolution.compaction.discard-conflicting-tasks` (default `false`) lets a +data-only compaction commit skip complete tasks whose row ranges conflict with +concurrent updates. Their input files remain live for later compaction; other +conflicts still fail the commit. Unreferenced outputs can be reclaimed by +`remove_orphan_files` after its safety window. + Ordinary compaction **does not physically remove rows hidden by deletion vectors**. Removing their positions would change the alignment with untouched column files. diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 0fc8c42fd261..600a15fffb0f 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -524,6 +524,12 @@ Duration The TTL in local index for cross partition upsert (primary keys not contain all partition fields), this can avoid maintaining too many indexes and lead to worse and worse performance, but please note that this may also cause data duplication. + +
data-evolution.compaction.discard-conflicting-tasks
+ false + Boolean + Skip complete data-evolution compaction tasks with row-range conflicts. Skipped inputs remain live for later compaction. Applies only to data-only compaction messages without row-id validation baselines. +
data-evolution.compaction.large-file-ratio
2.0 diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index c3202188dc09..a375b961083f 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -2774,6 +2774,15 @@ public String toString() { + "Row IDs and logical deletions are preserved, and associated " + "BLOB and VECTOR files are not rewritten by this option."); + public static final ConfigOption DATA_EVOLUTION_COMPACTION_DISCARD_CONFLICTING_TASKS = + key("data-evolution.compaction.discard-conflicting-tasks") + .booleanType() + .defaultValue(false) + .withDescription( + "Skip complete data-evolution compaction tasks with row-range conflicts. " + + "Skipped inputs remain live for later compaction. Applies only to " + + "data-only compaction messages without row-id validation baselines."); + public static final ConfigOption DATA_EVOLUTION_COMPACTION_LARGE_FILE_RATIO = key("data-evolution.compaction.large-file-ratio") .doubleType() diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java b/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java index 8cc109b6a882..90be065f8af7 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java @@ -29,6 +29,7 @@ import javax.annotation.Nullable; +import java.io.Serializable; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -82,7 +83,9 @@ public interface FileEntry { * The same {@link Identifier} indicates that the {@link ManifestEntry} refers to the same data * file. */ - class Identifier { + class Identifier implements Serializable { + + private static final long serialVersionUID = 1L; public final BinaryRow partition; public final int bucket; diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index 3354d4ab1a3b..860835de53da 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -51,6 +51,8 @@ import org.apache.paimon.operation.commit.CommitScanner; import org.apache.paimon.operation.commit.ConflictDetection; import org.apache.paimon.operation.commit.DataEvolutionChangesProvider; +import org.apache.paimon.operation.commit.DataEvolutionRowRangeConflictException; +import org.apache.paimon.operation.commit.DiscardingCompactChangesProvider; import org.apache.paimon.operation.commit.ManifestEntryChanges; import org.apache.paimon.operation.commit.RetryCommitResult; import org.apache.paimon.operation.commit.RetryCommitResult.CommitFailRetryResult; @@ -347,6 +349,7 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) { List commitMessages = committable.fileCommittables(); configureRowIdCheckFromMessages(commitMessages); ManifestEntryChanges changes = collectChanges(commitMessages); + List committedCompactFiles = changes.compactTableFiles; Set> materializedBuckets = materializedBuckets(commitMessages); try { List appendSimpleEntries = @@ -401,15 +404,11 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) { if (!changes.compactTableFiles.isEmpty() || !changes.compactChangelog.isEmpty() || !changes.compactIndexFiles.isEmpty()) { + CommitChangesProvider compactProvider = + compactChangesProvider(changes, materializedBuckets, commitMessages); attempts += tryCommit( - dataEvolutionChangesProvider( - new CommitChanges( - changes.compactTableFiles, - changes.compactChangelog, - changes.compactIndexFiles), - CommitKind.COMPACT, - materializedBuckets), + compactProvider, committable.identifier(), committable.watermark(), committable.properties(), @@ -417,6 +416,11 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) { false, true, null); + if (compactProvider instanceof DiscardingCompactChangesProvider) { + committedCompactFiles = + ((DiscardingCompactChangesProvider) compactProvider) + .retainedOriginalFiles(); + } generatedSnapshot += 1; } } finally { @@ -429,7 +433,7 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) { reportCommit( changes.appendTableFiles, changes.appendChangelog, - changes.compactTableFiles, + committedCompactFiles, changes.compactChangelog, commitDuration, generatedSnapshot, @@ -862,6 +866,56 @@ private Set> materializedBuckets(List co return result; } + private CommitChangesProvider compactChangesProvider( + ManifestEntryChanges changes, + Set> materializedBuckets, + List messages) { + CommitChangesProvider provider = + dataEvolutionChangesProvider( + new CommitChanges( + changes.compactTableFiles, + changes.compactChangelog, + changes.compactIndexFiles), + CommitKind.COMPACT, + materializedBuckets); + if (!options.dataEvolutionEnabled() + || !options.toConfiguration() + .get(CoreOptions.DATA_EVOLUTION_COMPACTION_DISCARD_CONFLICTING_TASKS) + || !materializedBuckets.isEmpty() + || conflictDetection.shouldCheckRowIdFromSnapshot(CommitKind.COMPACT) + || !changes.appendTableFiles.isEmpty() + || !changes.appendChangelog.isEmpty() + || !changes.appendIndexFiles.isEmpty() + || !changes.compactChangelog.isEmpty() + || !changes.compactIndexFiles.isEmpty()) { + return provider; + } + List> tasks = new ArrayList<>(); + Set taskFiles = new HashSet<>(); + for (CommitMessage message : messages) { + ManifestEntryChanges task = new ManifestEntryChanges(options.bucket()); + task.collect(message); + if (task.compactTableFiles.stream().noneMatch(entry -> entry.kind() == FileKind.ADD) + || task.compactTableFiles.stream() + .noneMatch(entry -> entry.kind() == FileKind.DELETE) + || task.compactTableFiles.stream() + .anyMatch( + entry -> + entry.firstRowId() == null + || isBlobFile(entry.fileName()) + || isVectorStoreFile(entry.fileName()))) { + return provider; + } + for (ManifestEntry entry : task.compactTableFiles) { + if (!taskFiles.add(entry.identifier())) { + return provider; + } + } + tasks.add(task.compactTableFiles); + } + return new DiscardingCompactChangesProvider(provider, tasks); + } + @VisibleForTesting CommitChangesProvider dataEvolutionChangesProvider( CommitChanges changes, @@ -1251,6 +1305,27 @@ private CommitResult tryCommitOnceWithoutLock( indexFiles, rowIdConflictChecker, commitKind); + while (exception.isPresent() + && exception.get() instanceof DataEvolutionRowRangeConflictException + && changesProvider instanceof DiscardingCompactChangesProvider + && ((DiscardingCompactChangesProvider) changesProvider) + .discard( + ((DataEvolutionRowRangeConflictException) exception.get()) + .conflictingFiles())) { + LOG.warn( + "Discarding conflicting compaction tasks for table {}: {}", + tableName, + exception.get().getMessage()); + deltaFiles = changesProvider.provide(latestSnapshot).tableFiles; + exception = + conflictDetection.checkConflicts( + latestSnapshot, + baseDataFiles, + SimpleFileEntry.from(deltaFiles), + indexFiles, + rowIdConflictChecker, + commitKind); + } if (exception.isPresent()) { if (allowRollback && rollback != null diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java index b3a7c82faedb..ae14f7a9d82d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java @@ -412,7 +412,10 @@ private Optional checkDataFileRowIdRangeConflicts( + "have encountered conflicts, data files: " + dataFileGroup.stream() .map(this::fileEntryToString) - .collect(Collectors.toList()))); + .collect(Collectors.toList()), + dataFileGroup.stream() + .map(FileEntry::identifier) + .collect(Collectors.toList()))); } } return Optional.empty(); diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionRowRangeConflictException.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionRowRangeConflictException.java index f958ff0fef3b..ea294a381189 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionRowRangeConflictException.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionRowRangeConflictException.java @@ -18,10 +18,30 @@ package org.apache.paimon.operation.commit; +import org.apache.paimon.manifest.FileEntry; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + /** Conflict caused by incompatible row-range boundaries in a data-evolution table. */ public final class DataEvolutionRowRangeConflictException extends RuntimeException { + private static final long serialVersionUID = 1L; + + private final List conflictingFiles; + public DataEvolutionRowRangeConflictException(String message) { + this(message, Collections.emptyList()); + } + + public DataEvolutionRowRangeConflictException( + String message, List conflictingFiles) { super(message); + this.conflictingFiles = Collections.unmodifiableList(new ArrayList<>(conflictingFiles)); + } + + public List conflictingFiles() { + return conflictingFiles; } } diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DiscardingCompactChangesProvider.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DiscardingCompactChangesProvider.java new file mode 100644 index 000000000000..c36eb4a7bfb0 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DiscardingCompactChangesProvider.java @@ -0,0 +1,84 @@ +/* + * 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.operation.commit; + +import org.apache.paimon.Snapshot; +import org.apache.paimon.manifest.FileEntry; +import org.apache.paimon.manifest.FileKind; +import org.apache.paimon.manifest.ManifestEntry; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; + +/** Drops complete compaction messages, preserving their inputs for a later attempt. */ +public class DiscardingCompactChangesProvider implements CommitChangesProvider { + private final CommitChangesProvider delegate; + private final List> tasks; + private final Set discarded = new HashSet<>(); + + public DiscardingCompactChangesProvider( + CommitChangesProvider delegate, List> tasks) { + this.delegate = delegate; + this.tasks = tasks; + } + + public boolean discard(List conflicts) { + Set identifiers = new HashSet<>(conflicts); + int previous = discarded.size(); + for (List task : tasks) { + if (task.stream() + .anyMatch( + entry -> + entry.kind() == FileKind.ADD + && identifiers.contains(entry.identifier()))) { + task.forEach(entry -> discarded.add(entry.identifier())); + } + } + return discarded.size() > previous; + } + + @Override + public CommitChanges provide(@Nullable Snapshot latest) { + CommitChanges changes = delegate.provide(latest); + List retained = new ArrayList<>(); + for (ManifestEntry entry : changes.tableFiles) { + if (!discarded.contains(entry.identifier())) { + retained.add(entry); + } + } + return new CommitChanges(retained, changes.changelogFiles, changes.indexFiles); + } + + public List retainedOriginalFiles() { + return tasks.stream() + .flatMap(List::stream) + .filter(entry -> !discarded.contains(entry.identifier())) + .collect(Collectors.toList()); + } + + @Override + public Set rebasedReassignments() { + return delegate.rebasedReassignments(); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DiscardingCompactChangesTest.java b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DiscardingCompactChangesTest.java new file mode 100644 index 000000000000..5eb92a5159b5 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DiscardingCompactChangesTest.java @@ -0,0 +1,329 @@ +/* + * 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.append.dataevolution; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.Snapshot; +import org.apache.paimon.catalog.RenamingSnapshotCommit; +import org.apache.paimon.catalog.SnapshotCommit; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.manifest.ManifestCommittable; +import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.operation.FileStoreCommit; +import org.apache.paimon.operation.Lock; +import org.apache.paimon.partition.PartitionStatistics; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.CatalogEnvironment; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FileStoreTableFactory; +import org.apache.paimon.table.TableTestBase; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.sink.BatchTableWrite; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.CommitMessageImpl; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.utils.SnapshotManager; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Real data-evolution compaction commits with concurrent column updates. */ +public class DiscardingCompactChangesTest extends TableTestBase { + + @Override + protected Schema schemaDefault() { + return Schema.newBuilder() + .column("pt", DataTypes.STRING()) + .column("id", DataTypes.INT()) + .column("quality", DataTypes.INT()) + .partitionKeys("pt") + .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true") + .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true") + .build(); + } + + @Test + void testDiscardOnlyConflictingTask() throws Exception { + FileStoreTable table = prepare(); + CommitMessageImpl taskA = compact(table, "a"); + CommitMessageImpl taskB = compact(table, "b"); + updateQuality(table, "a", 0, 99); + + List messages = Arrays.asList(taskA, taskB); + assertThatThrownBy(() -> commit(table, messages, false)).hasMessageContaining("data files"); + commit(table, messages, true); + + List liveFiles = liveFiles(table); + assertThat(liveFiles) + .containsAll(fileNames(taskA.compactIncrement().compactBefore())) + .containsAll(fileNames(taskB.compactIncrement().compactAfter())) + .doesNotContainAnyElementsOf(fileNames(taskA.compactIncrement().compactAfter())) + .doesNotContainAnyElementsOf(fileNames(taskB.compactIncrement().compactBefore())); + assertRows(table, 99, 0, 0, 0); + } + + @Test + void testAllTasksConflictCreatesIdempotentEmptyCompact() throws Exception { + FileStoreTable table = prepare(); + CommitMessageImpl taskA = compact(table, "a"); + updateQuality(table, "a", 0, 99); + long before = table.snapshotManager().latestSnapshotId(); + + FileStoreTable enabled = + table.copy( + Collections.singletonMap( + CoreOptions.DATA_EVOLUTION_COMPACTION_DISCARD_CONFLICTING_TASKS + .key(), + "true")); + ManifestCommittable committable = new ManifestCommittable(42); + committable.addFileCommittable(taskA); + try (FileStoreCommit commit = enabled.store().newCommit("compact-task-filter", enabled)) { + commit.commit(committable, false); + assertThat(commit.filterCommitted(Collections.singletonList(committable))).isEmpty(); + } + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(before + 1); + assertThat(table.snapshotManager().latestSnapshot().commitKind()) + .isEqualTo(Snapshot.CommitKind.COMPACT); + assertThat(liveFiles(table)) + .containsAll(fileNames(taskA.compactIncrement().compactBefore())) + .doesNotContainAnyElementsOf(fileNames(taskA.compactIncrement().compactAfter())); + assertRows(table, 99, 0, 0, 0); + } + + @Test + void testSharedInputTasksFailClosed() throws Exception { + FileStoreTable table = prepare(); + CommitMessageImpl first = compact(table, "a"); + CommitMessageImpl second = compact(table, "a"); + CommitMessageImpl independent = compact(table, "b"); + updateQuality(table, "a", 0, 99); + + long before = table.snapshotManager().latestSnapshotId(); + assertThatThrownBy(() -> commit(table, Arrays.asList(first, second, independent), true)) + .hasStackTraceContaining("not previously added"); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(before); + assertRows(table, 99, 0, 0, 0); + } + + @Test + void testRetryChecksNewSnapshotAndDiscardsNewConflict() throws Exception { + FileStoreTable table = prepare(); + CommitMessageImpl taskA = compact(table, "a"); + CommitMessageImpl taskB = compact(table, "b"); + updateQuality(table, "a", 0, 99); + AtomicInteger attempts = new AtomicInteger(); + CatalogEnvironment environment = + new CatalogEnvironment(null, null, null, null, null, null, false, false) { + @Override + public SnapshotCommit snapshotCommit( + SnapshotManager manager, CoreOptions options, String commitUser) { + SnapshotCommit delegate = new RenamingSnapshotCommit(manager, Lock.empty()); + return new SnapshotCommit(Lock.empty()) { + @Override + protected boolean commitImpl( + String baseUuid, + Snapshot snapshot, + String branch, + List statistics) + throws Exception { + if (snapshot.commitKind() == Snapshot.CommitKind.COMPACT + && attempts.getAndIncrement() == 0) { + updateQuality(table, "b", 2, 77); + } + return delegate.commit(baseUuid, snapshot, branch, statistics); + } + + @Override + protected void closeImpl() throws Exception { + delegate.close(); + } + }; + } + }; + FileStoreTable committingTable = + FileStoreTableFactory.create( + table.fileIO(), table.location(), table.schema(), environment); + + commit(committingTable, Arrays.asList(taskA, taskB), true); + assertThat(attempts.get()).isEqualTo(2); + assertThat(liveFiles(table)) + .containsAll(fileNames(taskA.compactIncrement().compactBefore())) + .containsAll(fileNames(taskB.compactIncrement().compactBefore())) + .doesNotContainAnyElementsOf(fileNames(taskA.compactIncrement().compactAfter())) + .doesNotContainAnyElementsOf(fileNames(taskB.compactIncrement().compactAfter())); + assertRows(table, 99, 0, 77, 0); + } + + @Test + void testRowIdBaselineConflictStillFails() throws Exception { + FileStoreTable table = prepare(); + long baseline = table.snapshotManager().latestSnapshotId(); + CommitMessageImpl task = compact(table, "a"); + updateQuality(table, "a", 0, 99); + long before = table.snapshotManager().latestSnapshotId(); + + assertThatThrownBy( + () -> + commit( + table, + Collections.singletonList( + task.withCheckFromSnapshot(baseline)), + true)) + .hasStackTraceContaining("conflict"); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(before); + assertRows(table, 99, 0, 0, 0); + } + + @Test + void testOrdinaryWriteUnaffected() throws Exception { + FileStoreTable table = prepare(); + FileStoreTable enabled = + table.copy( + Collections.singletonMap( + CoreOptions.DATA_EVOLUTION_COMPACTION_DISCARD_CONFLICTING_TASKS + .key(), + "true")); + write(enabled, GenericRow.of(BinaryString.fromString("c"), 4, 7)); + assertThat(read(table)).hasSize(5); + } + + @Test + void testCompactOutputRebasedAfterRowIdReassignment() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + write(table, GenericRow.of(BinaryString.fromString("a"), 0, 0)); + write(table, GenericRow.of(BinaryString.fromString("a"), 1, 0)); + write(table, GenericRow.of(BinaryString.fromString("b"), 2, 0)); + write(table, GenericRow.of(BinaryString.fromString("b"), 3, 0)); + write(table, GenericRow.of(BinaryString.fromString("a"), 4, 0)); + CommitMessageImpl taskB = compact(table, "b"); + long planned = table.snapshotManager().latestSnapshotId(); + assertThat(new DataEvolutionRowIdReassigner(table).reassign().reassigned).isTrue(); + + Map options = new HashMap<>(); + options.put(CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(), Long.toString(planned)); + options.put(CoreOptions.DATA_EVOLUTION_COMPACTION_DISCARD_CONFLICTING_TASKS.key(), "true"); + try (BatchTableCommit commit = table.copy(options).newBatchWriteBuilder().newCommit()) { + commit.commit(Collections.singletonList(taskB)); + } + assertThat(liveFiles(table)) + .containsAll(fileNames(taskB.compactIncrement().compactAfter())) + .doesNotContainAnyElementsOf(fileNames(taskB.compactIncrement().compactBefore())); + assertThat(read(table)).hasSize(5); + } + + private FileStoreTable prepare() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + for (String partition : Arrays.asList("a", "b")) { + for (int id = 0; id < 2; id++) { + write(table, GenericRow.of(BinaryString.fromString(partition), id, 0)); + } + } + return table; + } + + private CommitMessageImpl compact(FileStoreTable table, String partition) throws Exception { + List inputs = + table.store().newScan().plan().files().stream() + .filter( + entry -> + entry.partition().getString(0).toString().equals(partition)) + .sorted(Comparator.comparingLong(ManifestEntry::firstRowId)) + .collect(Collectors.toList()); + assertThat(inputs).hasSize(2); + return (CommitMessageImpl) + new DataEvolutionNormalCompactTask( + inputs.get(0).partition(), + inputs.stream() + .map(ManifestEntry::file) + .collect(Collectors.toList())) + .doCompact(table, "compact-worker"); + } + + private void updateQuality(FileStoreTable table, String partition, long firstRowId, int quality) + throws Exception { + try (BatchTableWrite write = + table.newBatchWriteBuilder() + .newWrite() + .withWriteType(table.rowType().project("pt", "quality")); + BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { + write.write(GenericRow.of(BinaryString.fromString(partition), quality)); + List messages = write.prepareCommit(); + for (CommitMessage message : messages) { + List files = + ((CommitMessageImpl) message).newFilesIncrement().newFiles(); + List assigned = + files.stream() + .map(file -> file.assignFirstRowId(firstRowId)) + .collect(Collectors.toList()); + files.clear(); + files.addAll(assigned); + } + commit.commit(messages); + } + } + + private void commit(FileStoreTable table, List messages, boolean discard) + throws Exception { + Map options = new HashMap<>(); + options.put( + CoreOptions.DATA_EVOLUTION_COMPACTION_DISCARD_CONFLICTING_TASKS.key(), + Boolean.toString(discard)); + try (BatchTableCommit commit = table.copy(options).newBatchWriteBuilder().newCommit()) { + commit.commit(messages); + } + } + + private List liveFiles(FileStoreTable table) { + return table.store().newScan().plan().files().stream() + .map(ManifestEntry::fileName) + .collect(Collectors.toList()); + } + + private List fileNames(List files) { + return files.stream().map(DataFileMeta::fileName).collect(Collectors.toList()); + } + + private void assertRows(FileStoreTable table, int... expectedQuality) throws Exception { + List rows = new ArrayList<>(read(table)); + rows.sort( + Comparator.comparing((InternalRow row) -> row.getString(0).toString()) + .thenComparingInt(row -> row.getInt(1))); + assertThat(rows).hasSize(expectedQuality.length); + for (int i = 0; i < rows.size(); i++) { + assertThat(rows.get(i).getInt(2)).isEqualTo(expectedQuality[i]); + } + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java index 2a22fd58b38f..5eb4db4c799f 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java @@ -54,6 +54,10 @@ import javax.annotation.Nullable; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -1523,7 +1527,7 @@ void testCheckRowIdExistenceCompactDoesNotMergeAcrossPartitions() { } @Test - void testCheckRowIdRangeConflictsUsesRetryableExceptionForDataFiles() { + void testCheckRowIdRangeConflictsUsesRetryableExceptionForDataFiles() throws Exception { DataEvolutionConflictDetection detection = createConflictDetection(); Optional exception = @@ -1542,6 +1546,21 @@ void testCheckRowIdRangeConflictsUsesRetryableExceptionForDataFiles() { assertThat(exception.get()) .isInstanceOf(DataEvolutionRowRangeConflictException.class) .hasMessageContaining("partition="); + DataEvolutionRowRangeConflictException conflict = + (DataEvolutionRowRangeConflictException) exception.get(); + assertThat(conflict.conflictingFiles()) + .contains(createFileEntryWithRowId("compacted", ADD, 0L, 4L).identifier()); + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (ObjectOutputStream output = new ObjectOutputStream(bytes)) { + output.writeObject(conflict); + } + try (ObjectInputStream input = + new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) { + DataEvolutionRowRangeConflictException restored = + (DataEvolutionRowRangeConflictException) input.readObject(); + assertThat(restored.conflictingFiles()) + .containsExactlyElementsOf(conflict.conflictingFiles()); + } } @Test