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 f7981ce93c [core] Show readable partitions in commit conflict entries
(#10566)
f7981ce93c is described below
commit f7981ce93c26363efc9d4a845c126aee9e90c442
Author: Ran Tao <[email protected]>
AuthorDate: Fri Oct 16 08:50:52 2026 +0800
[core] Show readable partitions in commit conflict entries (#10566)
---
.../apache/paimon/manifest/SimpleFileEntry.java | 10 +-
.../paimon/manifest/SimpleFileEntryWithDV.java | 4 +-
.../paimon/operation/commit/ConflictDetection.java | 28 ++-
.../commit/DataEvolutionConflictDetection.java | 10 +-
.../operation/commit/ConflictDetectionTest.java | 231 ++++++++++++++++++++-
5 files changed, 272 insertions(+), 11 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/SimpleFileEntry.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/SimpleFileEntry.java
index 77ae12b580..297e406414 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/SimpleFileEntry.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/SimpleFileEntry.java
@@ -238,11 +238,19 @@ public class SimpleFileEntry implements FileEntry {
@Override
public String toString() {
+ return toStringWithPartition(String.valueOf(partition));
+ }
+
+ /**
+ * Formats the entry with the supplied partition representation while
retaining its other
+ * diagnostic fields.
+ */
+ public String toStringWithPartition(String partitionString) {
return "{"
+ "kind="
+ kind
+ ", partition="
- + partition
+ + partitionString
+ ", bucket="
+ bucket
+ ", totalBuckets="
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/SimpleFileEntryWithDV.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/SimpleFileEntryWithDV.java
index 6b4c479157..ed03c9fcda 100644
---
a/paimon-core/src/main/java/org/apache/paimon/manifest/SimpleFileEntryWithDV.java
+++
b/paimon-core/src/main/java/org/apache/paimon/manifest/SimpleFileEntryWithDV.java
@@ -79,8 +79,8 @@ public class SimpleFileEntryWithDV extends SimpleFileEntry {
}
@Override
- public String toString() {
- return super.toString() + ", {dvFileName=" + dvFileName + '}';
+ public String toStringWithPartition(String partitionString) {
+ return super.toStringWithPartition(partitionString) + ", {dvFileName="
+ dvFileName + '}';
}
/**
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
index 212ce0962a..3685d96311 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
@@ -653,11 +653,13 @@ public abstract class ConflictDetection {
String baseEntriesString =
"Base entries are:\n"
+ baseEntries.stream()
- .map(Object::toString)
+ .map(this::fileEntryToString)
.collect(Collectors.joining("\n"));
String changesString =
"Changes are:\n"
- +
changes.stream().map(Object::toString).collect(Collectors.joining("\n"));
+ + changes.stream()
+ .map(this::fileEntryToString)
+ .collect(Collectors.joining("\n"));
RuntimeException fullException =
new RuntimeException(
@@ -679,12 +681,12 @@ public abstract class ConflictDetection {
"Base entries are:\n"
+ baseEntries.subList(0,
Math.min(baseEntries.size(), maxEntry))
.stream()
- .map(Object::toString)
+ .map(this::fileEntryToString)
.collect(Collectors.joining("\n"));
changesString =
"Changes are:\n"
+ changes.subList(0, Math.min(changes.size(),
maxEntry)).stream()
- .map(Object::toString)
+ .map(this::fileEntryToString)
.collect(Collectors.joining("\n"));
simplifiedException =
new RuntimeException(
@@ -706,6 +708,24 @@ public abstract class ConflictDetection {
}
}
+ /**
+ * Converts a file entry to a string with a readable partition, falling
back to its original
+ * representation if partition formatting fails.
+ */
+ protected final String fileEntryToString(SimpleFileEntry entry) {
+ if (partitionType.getFieldCount() == 0) {
+ return entry.toStringWithPartition("<non-partitioned>");
+ }
+ String partitionString;
+ try {
+ partitionString =
pathFactory.getPartitionString(entry.partition());
+ } catch (RuntimeException e) {
+ // fallback to original representation
+ return entry.toString();
+ }
+ return entry.toStringWithPartition(partitionString);
+ }
+
/** Factory to create {@link ConflictDetection}. */
public interface Factory {
ConflictDetection create(CommitScanner scanner);
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 c1a31e1de3..b3a7c82fae 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
@@ -410,7 +410,9 @@ public class DataEvolutionConflictDetection extends
ConflictDetection {
"For Data Evolution table, multiple 'MERGE
INTO' and 'COMPACT' "
+ "operations "
+ "have encountered conflicts, data
files: "
- + dataFileGroup));
+ + dataFileGroup.stream()
+ .map(this::fileEntryToString)
+
.collect(Collectors.toList())));
}
}
return Optional.empty();
@@ -448,10 +450,12 @@ public class DataEvolutionConflictDetection extends
ConflictDetection {
"For Data Evolution table, multiple 'MERGE
INTO' and 'COMPACT' "
+ "operations have encountered
conflicts, dedicated "
+ "file %s %s %s: %s",
- dedicatedFile,
+ fileEntryToString(dedicatedFile),
dedicatedRange,
conflictReason,
- intersectingDataFiles)));
+ intersectingDataFiles.stream()
+ .map(this::fileEntryToString)
+ .collect(Collectors.toList()))));
}
return Optional.empty();
}
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 41022f77b4..2a22fd58b3 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
@@ -18,8 +18,12 @@
package org.apache.paimon.operation.commit;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.fs.Path;
import org.apache.paimon.index.DeletionVectorMeta;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileHandler;
@@ -34,11 +38,19 @@ import org.apache.paimon.manifest.SimpleFileEntryWithDV;
import
org.apache.paimon.operation.commit.RetryCommitResult.CommitFailRetryResult;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.table.BucketMode;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.SnapshotManager;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import javax.annotation.Nullable;
@@ -68,6 +80,177 @@ import static org.mockito.Mockito.when;
class ConflictDetectionTest {
+ @Test
+ void testFileEntryWithDVToStringPreservesOriginalOutput() {
+ // Guard the inherited toString() path after moving DV formatting to
toStringWithPartition.
+ BinaryRow partition = BinaryRow.singleColumn(12);
+ SimpleFileEntry entry = createFileEntryWithRowId("base", ADD,
partition, 2, 10L, 3L);
+ SimpleFileEntryWithDV entryWithDV = new SimpleFileEntryWithDV(entry,
"dv-1");
+
+ assertThat(entryWithDV.toString())
+ .isEqualTo(entry + ", {dvFileName=dv-1}")
+ .contains("partition=" + partition);
+ }
+
+ @Test
+ void testConflictEntriesShowPartitionInFullAndSimplifiedExceptions() {
+ // Exceed the 50-entry limit in both lists while preserving DV
metadata and the cause.
+ FileStorePathFactory pathFactory = mock(FileStorePathFactory.class);
+ BinaryRow partition = BinaryRow.singleColumn(12);
+ BinaryRow changedPartition = BinaryRow.singleColumn(13);
+ when(pathFactory.getPartitionString(partition)).thenReturn("hr=12/");
+
when(pathFactory.getPartitionString(changedPartition)).thenReturn("hr=13/");
+ ConflictDetection detection =
+ new AppendConflictDetection(
+ "test-table",
+ "test-user",
+ RowType.of(new IntType()),
+ pathFactory,
+ BucketMode.HASH_FIXED,
+ false,
+ null,
+ null);
+ SimpleFileEntry entry = createFileEntryWithRowId("base", ADD,
partition, 2, 10L, 3L);
+ SimpleFileEntry change =
+ createFileEntryWithRowId("changed", DELETE, changedPartition,
2, 13L, 1L);
+ List<SimpleFileEntry> baseEntries = new
ArrayList<>(Collections.nCopies(49, entry));
+ baseEntries.add(createFileEntryWithRowId("base-50", ADD, partition, 2,
13L, 1L));
+ baseEntries.add(createFileEntryWithRowId("base-51", ADD, partition, 2,
14L, 1L));
+ List<SimpleFileEntry> changes =
+ new ArrayList<>(Collections.nCopies(49, new
SimpleFileEntryWithDV(change, "dv-1")));
+ changes.add(createFileEntryWithRowId("changed-50", DELETE,
changedPartition, 2, 14L, 1L));
+ changes.add(createFileEntryWithRowId("changed-51", DELETE,
changedPartition, 2, 15L, 1L));
+ RuntimeException cause = new RuntimeException("original conflict");
+
+ Pair<RuntimeException, RuntimeException> exceptions =
+ detection.createConflictException(
+ "commit conflict", "base-user", baseEntries, changes,
cause);
+
+ assertThat(exceptions.getLeft())
+ .hasMessageContaining("partition=hr=12/")
+ .hasMessageContaining("Changes are:\n{kind=DELETE,
partition=hr=13/")
+ .hasMessageContaining("totalBuckets=1")
+ .hasMessageContaining("rowCount=3")
+ .hasMessageContaining("fileName=base-51")
+ .hasMessageContaining("fileName=changed-51")
+ .hasCause(cause);
+ assertThat(exceptions.getRight())
+ .hasMessageContaining("Base entries are:")
+ .hasMessageContaining("Changes are:\n{kind=DELETE,
partition=hr=13/")
+ .hasMessageContaining("partition=hr=12/")
+ .hasMessageContaining("dvFileName=dv-1")
+ .hasMessageContaining("fileName=base-50")
+ .hasMessageContaining("fileName=changed-50")
+ .hasMessageNotContaining("fileName=base-51")
+ .hasMessageNotContaining("fileName=changed-51")
+ .hasCause(cause);
+ }
+
+ @Test
+ void testConflictEntriesUseEscapedPartitionPath(@TempDir
java.nio.file.Path tempDir) {
+ // Use the real formatter to cover escaping and null defaults in a
multi-column partition.
+ RowType partitionType =
+ RowType.of(
+ new DataType[] {new VarCharType(10), new IntType()},
+ new String[] {"dt", "hr"});
+ FileStorePathFactory pathFactory =
+ new FileStorePathFactory(
+ new Path(tempDir.toString()),
+ partitionType,
+ "default",
+ CoreOptions.FILE_FORMAT.defaultValue().toString(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+ CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue(),
+
CoreOptions.PARTITION_GENERATE_LEGACY_NAME.defaultValue(),
+
CoreOptions.FILE_SUFFIX_INCLUDE_COMPRESSION.defaultValue(),
+ CoreOptions.FILE_COMPRESSION.defaultValue(),
+ null,
+ null,
+ CoreOptions.ExternalPathStrategy.NONE,
+ null,
+ false,
+ null);
+ BinaryRow partition = new BinaryRow(2);
+ BinaryRowWriter writer = new BinaryRowWriter(partition);
+ writer.writeString(0, BinaryString.fromString("a/b=c"));
+ writer.setNullAt(1);
+ writer.complete();
+ ConflictDetection detection =
+ new AppendConflictDetection(
+ "test-table",
+ "test-user",
+ partitionType,
+ pathFactory,
+ BucketMode.HASH_FIXED,
+ false,
+ null,
+ null);
+
+ Pair<RuntimeException, RuntimeException> exceptions =
+ detection.createConflictException(
+ "commit conflict",
+ "base-user",
+ Collections.singletonList(
+ createFileEntryWithRowId("base", ADD,
partition, 2, 10L, 3L)),
+ Collections.emptyList(),
+ null);
+
+
assertThat(exceptions.getLeft()).hasMessageContaining("partition=dt=a%2Fb%3Dc/hr=default/");
+ }
+
+ @Test
+ void testConflictEntriesForNonPartitionedTable() {
+ // An empty partition schema should produce an explicit label without
a path factory.
+ ConflictDetection detection = createConflictDetection(null, false,
false);
+ SimpleFileEntry entry = createFileEntry("base", ADD);
+
+ Pair<RuntimeException, RuntimeException> exceptions =
+ detection.createConflictException(
+ "commit conflict",
+ "base-user",
+ Collections.singletonList(entry),
+ Collections.emptyList(),
+ null);
+
+ assertThat(exceptions.getLeft())
+ .isSameAs(exceptions.getRight())
+ .hasMessageContaining("partition=<non-partitioned>");
+ }
+
+ @Test
+ void testConflictEntryFormattingFailurePreservesOriginalOutputAndCause() {
+ // A formatting error must not hide the original conflict or discard
DV metadata.
+ FileStorePathFactory pathFactory = mock(FileStorePathFactory.class);
+ BinaryRow partition = BinaryRow.singleColumn(12);
+ when(pathFactory.getPartitionString(partition))
+ .thenThrow(new IllegalArgumentException("invalid partition"));
+ ConflictDetection detection =
+ new AppendConflictDetection(
+ "test-table",
+ "test-user",
+ RowType.of(new IntType()),
+ pathFactory,
+ BucketMode.HASH_FIXED,
+ false,
+ null,
+ null);
+ RuntimeException cause = new RuntimeException("original conflict");
+ SimpleFileEntry entry = createFileEntryWithRowId("base", ADD,
partition, 2, 10L, 3L);
+ SimpleFileEntryWithDV entryWithDV = new SimpleFileEntryWithDV(entry,
"dv-1");
+
+ Pair<RuntimeException, RuntimeException> exceptions =
+ detection.createConflictException(
+ "commit conflict",
+ "base-user",
+ Arrays.asList(entry, entryWithDV),
+ Collections.emptyList(),
+ cause);
+
+ assertThat(exceptions.getLeft())
+ .hasMessageContaining("Base entries are:\n" + entry + "\n" +
entryWithDV)
+ .hasCause(cause);
+ }
+
@Test
void testCreateConflictDetectionByTableType() {
ConflictDetection append = createConflictDetection(null, false, false);
@@ -1356,7 +1539,9 @@ class ConflictDetectionTest {
Snapshot.CommitKind.COMPACT);
assertThat(exception).isPresent();
-
assertThat(exception.get()).isInstanceOf(DataEvolutionRowRangeConflictException.class);
+ assertThat(exception.get())
+ .isInstanceOf(DataEvolutionRowRangeConflictException.class)
+ .hasMessageContaining("partition=<non-partitioned>");
}
@Test
@@ -1377,6 +1562,7 @@ class ConflictDetectionTest {
assertThat(exception).isPresent();
assertThat(exception.get())
.isNotInstanceOf(DataEvolutionRowRangeConflictException.class)
+ .hasMessageContaining("partition=<non-partitioned>")
.hasMessageContaining("dedicated file")
.hasMessageContaining("p1.blob")
.hasMessageContaining("spans multiple data file ranges")
@@ -1384,6 +1570,49 @@ class ConflictDetectionTest {
.hasMessageContaining("f2");
}
+ @ParameterizedTest
+ @ValueSource(strings = {"compacted", "p1.blob", "p1.vector.data"})
+ void testRowRangeConflictEntriesShowPartitions(String fileName) {
+ // Cover overlapping data files and dedicated blob/vector files
spanning adjacent ranges.
+ BinaryRow partition = BinaryRow.singleColumn(12);
+ FileStorePathFactory pathFactory = mock(FileStorePathFactory.class);
+ when(pathFactory.getPartitionString(partition)).thenReturn("hr=12/");
+ DataEvolutionConflictDetection detection =
+ new DataEvolutionConflictDetection(
+ "test-table",
+ "test-user",
+ RowType.of(new DataType[] {new IntType()}, new
String[] {"hr"}),
+ pathFactory,
+ BucketMode.HASH_FIXED,
+ false,
+ false,
+ null,
+ null,
+ null);
+
+ Optional<RuntimeException> exception =
+ detection.checkConflicts(
+ snapshot(1),
+ Arrays.asList(
+ createFileEntryWithRowId("f1", ADD, partition,
0, 0L, 2L),
+ createFileEntryWithRowId("f2", ADD, partition,
0, 2L, 2L)),
+ Collections.singletonList(
+ createFileEntryWithRowId(fileName, ADD,
partition, 0, 0L, 4L)),
+ Collections.emptyList(),
+ null,
+ Snapshot.CommitKind.COMPACT);
+
+ assertThat(exception).isPresent();
+ assertThat(exception.get())
+ .hasMessageContaining("partition=hr=12/")
+
.hasMessageNotContaining("partition=org.apache.paimon.data.BinaryRow@")
+ .hasMessageContaining("fileName=f1")
+ .hasMessageContaining("fileName=f2")
+ .hasMessageContaining("fileName=" + fileName)
+ .hasMessageContaining("rowCount=4")
+ .hasMessageContaining("firstRowId=0");
+ }
+
@Test
void testCheckRowIdRangeConflictsRejectsOverlappingNormalFiles() {
DataEvolutionConflictDetection detection = createConflictDetection();
