From 1d5b8393d53eec44b25922536252d692b631f6ab Mon Sep 17 00:00:00 2001 From: yuzelin Date: Wed, 5 Aug 2026 16:18:42 +0800 Subject: [PATCH 1/8] [core] Tag diff query at postpone table should check bucket number --- .../apache/paimon/table/PostponeUtils.java | 17 +++-- .../IncrementalDiffStartingScanner.java | 8 +- .../table/source/snapshot/TimeTravelUtil.java | 45 ++++++++--- .../paimon/table/IncrementalTableTest.java | 74 +++++++++++++++++++ .../spark/sql/TableValuedFunctionsTest.scala | 40 ++++++++++ 5 files changed, 163 insertions(+), 21 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java b/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java index 0d2a77ad0b54..bb0720107a0d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java @@ -25,6 +25,7 @@ import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.InternalRow; import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.manifest.FileEntry; import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.manifest.SimpleFileEntry; import org.apache.paimon.operation.FileStoreScan; @@ -347,8 +348,7 @@ private static int roundUpToPowerOfTwo(BigInteger value, int upperBound) { } public static Map getKnownNumBuckets(FileStoreTable table) { - return getKnownNumBuckets( - table.store().newScan().onlyReadRealBuckets().readSimpleEntries()); + return getKnownNumBuckets(table.store().newScan().onlyReadRealBuckets().readFileIterator()); } public static Map getKnownNumBuckets( @@ -371,19 +371,24 @@ public static Map getKnownNumBuckets( .readSimpleEntries()); } + public static Map getKnownNumBuckets( + SnapshotReader reader, long snapshotId) { + return getKnownNumBuckets(reader.withSnapshot(snapshotId).readFileIterator()); + } + static Map getKnownNumBuckets( FileStoreTable table, long snapshotId, @Nullable PartitionPredicate partitionFilter) { FileStoreScan scan = table.store().newScan().withSnapshot(snapshotId).onlyReadRealBuckets(); if (partitionFilter != null) { scan.withPartitionFilter(partitionFilter); } - return getKnownNumBuckets(scan.readSimpleEntries()); + return getKnownNumBuckets(scan.readFileIterator()); } - private static Map getKnownNumBuckets( - List simpleFileEntries) { + private static Map getKnownNumBuckets(Iterator fileEntries) { Map knownNumBuckets = new HashMap<>(); - for (SimpleFileEntry entry : simpleFileEntries) { + while (fileEntries.hasNext()) { + FileEntry entry = fileEntries.next(); if (entry.totalBuckets() >= 0) { Integer oldTotalBuckets = knownNumBuckets.put(entry.partition(), entry.totalBuckets()); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/IncrementalDiffStartingScanner.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/IncrementalDiffStartingScanner.java index 3a42f16e2c74..678cf1b40e41 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/IncrementalDiffStartingScanner.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/IncrementalDiffStartingScanner.java @@ -53,18 +53,18 @@ public IncrementalDiffStartingScanner( this.start = start; this.end = end; this.startingSnapshotId = start.id(); + } + @Override + public Result scan(SnapshotReader reader) { TimeTravelUtil.checkRescaleBucketForIncrementalDiffQuery( new SchemaManager( snapshotManager.fileIO(), snapshotManager.tablePath(), snapshotManager.branch()), + reader, start, end); - } - - @Override - public Result scan(SnapshotReader reader) { return StartingScanner.fromPlan(reader.withSnapshot(end).readIncrementalDiff(start)); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java index 786fecd0cb9b..3e7fa5083ffa 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java @@ -20,10 +20,13 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.Snapshot; +import org.apache.paimon.data.BinaryRow; import org.apache.paimon.options.Options; import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.BucketMode; import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.PostponeUtils; import org.apache.paimon.utils.ChangelogManager; import org.apache.paimon.utils.FunctionWithException; import org.apache.paimon.utils.SnapshotManager; @@ -38,6 +41,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.Map; import java.util.Optional; import java.util.TimeZone; @@ -251,17 +255,36 @@ private static Snapshot changelogOrSnapshot( } public static void checkRescaleBucketForIncrementalDiffQuery( - SchemaManager schemaManager, Snapshot start, Snapshot end) { - if (start.schemaId() != end.schemaId()) { - int startBucketNumber = bucketNumber(schemaManager, start.schemaId()); - int endBucketNumber = bucketNumber(schemaManager, end.schemaId()); - if (startBucketNumber != endBucketNumber) { - throw new InconsistentTagBucketException( - start.id(), - end.id(), - String.format( - "The bucket number of two snapshots are different (%s, %s), which is not supported in incremental diff query.", - startBucketNumber, endBucketNumber)); + SchemaManager schemaManager, SnapshotReader reader, Snapshot start, Snapshot end) { + TableSchema schema = schemaManager.latest().get(); + if (!schema.primaryKeys().isEmpty() && schema.numBuckets() == BucketMode.POSTPONE_BUCKET) { + Map startBucketNumbers = + PostponeUtils.getKnownNumBuckets(reader, start.id()); + Map endBucketNumbers = + PostponeUtils.getKnownNumBuckets(reader, end.id()); + for (Map.Entry entry : startBucketNumbers.entrySet()) { + Integer endPartitionBucketNumber = endBucketNumbers.get(entry.getKey()); + if (endPartitionBucketNumber != null + && !entry.getValue().equals(endPartitionBucketNumber)) { + throw new InconsistentTagBucketException( + start.id(), + end.id(), + "The real bucket number of two snapshots in postpone-bucket mode are different, " + + "which is not supported in incremental diff query."); + } + } + } else { + if (start.schemaId() != end.schemaId()) { + int startBucketNumber = bucketNumber(schemaManager, start.schemaId()); + int endBucketNumber = bucketNumber(schemaManager, end.schemaId()); + if (startBucketNumber != endBucketNumber) { + throw new InconsistentTagBucketException( + start.id(), + end.id(), + String.format( + "The bucket number of two snapshots are different (%s, %s), which is not supported in incremental diff query.", + startBucketNumber, endBucketNumber)); + } } } } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java index 3a85110bec62..7e04898c1c48 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java @@ -31,8 +31,10 @@ import org.apache.paimon.table.sink.BatchTableWrite; import org.apache.paimon.table.sink.BatchWriteBuilder; import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.PostponeFixedBucketWriteBuilder; import org.apache.paimon.table.sink.TableCommitImpl; import org.apache.paimon.table.sink.TableWriteImpl; +import org.apache.paimon.table.source.snapshot.TimeTravelUtil.InconsistentTagBucketException; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowKind; import org.apache.paimon.utils.Pair; @@ -42,6 +44,7 @@ import org.junit.jupiter.api.Test; import java.time.LocalDateTime; +import java.util.Collections; import java.util.List; import static org.apache.paimon.CoreOptions.INCREMENTAL_BETWEEN; @@ -581,6 +584,77 @@ public void testIncrementalEmptyResult() throws Exception { .isEmpty(); } + @Test + public void testPostponeBucketNumberChangedInIncrementalDiff() throws Exception { + Identifier identifier = identifier("T"); + Schema schema = + Schema.newBuilder() + .column("pk", DataTypes.INT()) + .column("col1", DataTypes.INT()) + .primaryKey("pk") + .option("bucket", String.valueOf(BucketMode.POSTPONE_BUCKET)) + .build(); + catalog.createTable(identifier, schema, true); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + + PostponeFixedBucketWriteBuilder builder = table.newPostponeFixedBucketWriteBuilder(); + try (TableWriteImpl write = builder.newWrite(); + BatchTableCommit commit = builder.newCommit()) { + write.writeAndReturn(GenericRow.of(1, 1), 0, 1); + commit.commit(write.prepareCommit()); + } + table.createTag("TAG1", 1); + + builder = table.newPostponeFixedBucketWriteBuilder().withOverwrite(Collections.emptyMap()); + try (TableWriteImpl write = builder.newWrite(); + BatchTableCommit commit = builder.newCommit()) { + write.writeAndReturn(GenericRow.of(1, 2), 0, 2); + write.writeAndReturn(GenericRow.of(2, 1), 1, 2); + commit.commit(write.prepareCommit()); + } + table.createTag("TAG2", 2); + + assertThatThrownBy(() -> read(table, Pair.of(INCREMENTAL_BETWEEN, "TAG1,TAG2"))) + .isInstanceOf(InconsistentTagBucketException.class) + .hasMessageContaining( + "The real bucket number of two snapshots in postpone-bucket mode are different, " + + "which is not supported in incremental diff query."); + } + + @Test + public void testPostponeDifferentBucketNumbersForDifferentPartitions() throws Exception { + Identifier identifier = identifier("T"); + Schema schema = + Schema.newBuilder() + .column("pt", DataTypes.INT()) + .column("pk", DataTypes.INT()) + .column("col1", DataTypes.INT()) + .partitionKeys("pt") + .primaryKey("pk", "pt") + .option("bucket", String.valueOf(BucketMode.POSTPONE_BUCKET)) + .build(); + catalog.createTable(identifier, schema, true); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + + PostponeFixedBucketWriteBuilder builder = table.newPostponeFixedBucketWriteBuilder(); + try (TableWriteImpl write = builder.newWrite(); + BatchTableCommit commit = builder.newCommit()) { + write.writeAndReturn(GenericRow.of(1, 1, 1), 0, 1); + commit.commit(write.prepareCommit()); + } + table.createTag("TAG1", 1); + + try (TableWriteImpl write = builder.newWrite(); + BatchTableCommit commit = builder.newCommit()) { + write.writeAndReturn(GenericRow.of(2, 1, 1), 1, 2); + commit.commit(write.prepareCommit()); + } + table.createTag("TAG2", 2); + + assertThat(read(table, Pair.of(INCREMENTAL_BETWEEN, "TAG1,TAG2"))) + .containsExactly(GenericRow.of(2, 1, 1)); + } + private static long utcMills(String timestamp) { return Timestamp.fromLocalDateTime(LocalDateTime.parse(timestamp)).getMillisecond(); } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/TableValuedFunctionsTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/TableValuedFunctionsTest.scala index e46a00358d18..301a175611a2 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/TableValuedFunctionsTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/TableValuedFunctionsTest.scala @@ -754,6 +754,46 @@ class TableValuedFunctionsTest extends PaimonHiveTestBase with AdaptiveSparkPlan } } + test("Table Valued Functions: incremental query with inconsistent postpone bucket") { + withTable("t") { + sql(""" + |CREATE TABLE t (a INT, b INT) USING paimon + |TBLPROPERTIES ('primary-key'='a', 'bucket' = '-2') + |""".stripMargin) + + val table = loadTable("t") + var builder = table.newPostponeFixedBucketWriteBuilder() + var write = builder.newWrite() + var commit = builder.newCommit() + try { + write.writeAndReturn(GenericRow.of(1, 11), 0, 1) + commit.commit(write.prepareCommit()) + } finally { + write.close() + commit.close() + } + table.createTag("2024-01-01", 1) + + builder = table.newPostponeFixedBucketWriteBuilder().withOverwrite(Collections.emptyMap()) + write = builder.newWrite() + commit = builder.newCommit() + try { + write.writeAndReturn(GenericRow.of(1, 22), 0, 2) + write.writeAndReturn(GenericRow.of(2, 22), 1, 2) + commit.commit(write.prepareCommit()) + } finally { + write.close() + commit.close() + } + table.createTag("2024-01-02", 2) + + checkAnswer( + sql( + "SELECT * FROM paimon_incremental_query('t', '2024-01-01', '2024-01-02') ORDER BY a, b"), + Seq(Row(1, 22), Row(2, 22))) + } + } + test("Table Valued Functions: incremental query with delete after minor compact") { withTable("t") { sql(""" From 722cd9862a1d8b5fbbdf10995db3d7958b1ea72d Mon Sep 17 00:00:00 2001 From: yuzelin Date: Wed, 5 Aug 2026 19:37:39 +0800 Subject: [PATCH 2/8] fix --- .../src/main/java/org/apache/paimon/table/PostponeUtils.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java b/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java index bb0720107a0d..c4c1d6365178 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java @@ -368,7 +368,7 @@ public static Map getKnownNumBuckets( .withSnapshot(snapshotId) .withPartitionFilter(partitions) .onlyReadRealBuckets() - .readSimpleEntries()); + .readFileIterator()); } public static Map getKnownNumBuckets( From cc881b922786e662c85f5d6f47feddd86b6cb946 Mon Sep 17 00:00:00 2001 From: yuzelin Date: Wed, 5 Aug 2026 20:02:47 +0800 Subject: [PATCH 3/8] fix --- .../java/org/apache/paimon/table/PostponeUtilsTest.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java index 1d07359b6db9..bcac28de025f 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java @@ -55,12 +55,12 @@ public class PostponeUtilsTest { public void testGetKnownNumBucketsFromSnapshot() { BinaryRow partition = partition(1); PartitionPredicate partitionFilter = mock(PartitionPredicate.class); - SimpleFileEntry entry = mock(SimpleFileEntry.class); + ManifestEntry entry = mock(ManifestEntry.class); when(entry.partition()).thenReturn(partition); when(entry.totalBuckets()).thenReturn(4); FileStoreScan scan = mock(FileStoreScan.class, RETURNS_SELF); - when(scan.readSimpleEntries()).thenReturn(Collections.singletonList(entry)); + when(scan.readFileIterator()).thenReturn(Collections.singletonList(entry).iterator()); FileStore store = mock(FileStore.class); when(store.newScan()).thenReturn(scan); FileStoreTable table = mock(FileStoreTable.class); @@ -77,12 +77,12 @@ public void testGetKnownNumBucketsFromSnapshot() { public void testGetKnownNumBucketsByPartitions() { BinaryRow partition = partition(1); List partitions = Collections.singletonList(partition); - SimpleFileEntry entry = mock(SimpleFileEntry.class); + ManifestEntry entry = mock(ManifestEntry.class); when(entry.partition()).thenReturn(partition); when(entry.totalBuckets()).thenReturn(4); FileStoreScan scan = mock(FileStoreScan.class, RETURNS_SELF); - when(scan.readSimpleEntries()).thenReturn(Collections.singletonList(entry)); + when(scan.readFileIterator()).thenReturn(Collections.singletonList(entry).iterator()); FileStore store = mock(FileStore.class); when(store.newScan()).thenReturn(scan); FileStoreTable table = mock(FileStoreTable.class); From 0cf1f2b0eb3bc6fb183187998e4b3ab69830b46b Mon Sep 17 00:00:00 2001 From: yuzelin Date: Thu, 6 Aug 2026 12:24:52 +0800 Subject: [PATCH 4/8] fix comments --- .../apache/paimon/table/PostponeUtils.java | 19 +++--- .../IncrementalDiffStartingScanner.java | 9 --- .../source/snapshot/SnapshotReaderImpl.java | 2 + .../table/source/snapshot/TimeTravelUtil.java | 58 ++++++------------- .../paimon/table/IncrementalTableTest.java | 11 ++++ .../paimon/table/PostponeUtilsTest.java | 8 +-- 6 files changed, 43 insertions(+), 64 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java b/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java index c4c1d6365178..0d2a77ad0b54 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java @@ -25,7 +25,6 @@ import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.InternalRow; import org.apache.paimon.io.DataFileMeta; -import org.apache.paimon.manifest.FileEntry; import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.manifest.SimpleFileEntry; import org.apache.paimon.operation.FileStoreScan; @@ -348,7 +347,8 @@ private static int roundUpToPowerOfTwo(BigInteger value, int upperBound) { } public static Map getKnownNumBuckets(FileStoreTable table) { - return getKnownNumBuckets(table.store().newScan().onlyReadRealBuckets().readFileIterator()); + return getKnownNumBuckets( + table.store().newScan().onlyReadRealBuckets().readSimpleEntries()); } public static Map getKnownNumBuckets( @@ -368,12 +368,7 @@ public static Map getKnownNumBuckets( .withSnapshot(snapshotId) .withPartitionFilter(partitions) .onlyReadRealBuckets() - .readFileIterator()); - } - - public static Map getKnownNumBuckets( - SnapshotReader reader, long snapshotId) { - return getKnownNumBuckets(reader.withSnapshot(snapshotId).readFileIterator()); + .readSimpleEntries()); } static Map getKnownNumBuckets( @@ -382,13 +377,13 @@ static Map getKnownNumBuckets( if (partitionFilter != null) { scan.withPartitionFilter(partitionFilter); } - return getKnownNumBuckets(scan.readFileIterator()); + return getKnownNumBuckets(scan.readSimpleEntries()); } - private static Map getKnownNumBuckets(Iterator fileEntries) { + private static Map getKnownNumBuckets( + List simpleFileEntries) { Map knownNumBuckets = new HashMap<>(); - while (fileEntries.hasNext()) { - FileEntry entry = fileEntries.next(); + for (SimpleFileEntry entry : simpleFileEntries) { if (entry.totalBuckets() >= 0) { Integer oldTotalBuckets = knownNumBuckets.put(entry.partition(), entry.totalBuckets()); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/IncrementalDiffStartingScanner.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/IncrementalDiffStartingScanner.java index 678cf1b40e41..f7bb09e1127b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/IncrementalDiffStartingScanner.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/IncrementalDiffStartingScanner.java @@ -21,7 +21,6 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.Snapshot; import org.apache.paimon.manifest.PartitionEntry; -import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.tag.Tag; import org.apache.paimon.tag.TagPeriodHandler; import org.apache.paimon.utils.Pair; @@ -57,14 +56,6 @@ public IncrementalDiffStartingScanner( @Override public Result scan(SnapshotReader reader) { - TimeTravelUtil.checkRescaleBucketForIncrementalDiffQuery( - new SchemaManager( - snapshotManager.fileIO(), - snapshotManager.tablePath(), - snapshotManager.branch()), - reader, - start, - end); return StartingScanner.fromPlan(reader.withSnapshot(end).readIncrementalDiff(start)); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java index 9826965039b2..a0f90fe3caee 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java @@ -608,6 +608,8 @@ public Plan readIncrementalDiff(Snapshot before) { groupByPartFiles(plan.files(FileKind.ADD)); Map>> beforeFiles = groupByPartFiles(scan.withSnapshot(before).plan().files(FileKind.ADD)); + TimeTravelUtil.checkRescaleBucketForIncrementalDiffQuery( + before, beforeFiles, plan.snapshot(), afterFiles); return toIncrementalPlan( false, new LazyField<>(() -> before), diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java index 3e7fa5083ffa..7d6af8cd37bc 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java @@ -21,12 +21,9 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.Snapshot; import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.options.Options; -import org.apache.paimon.schema.SchemaManager; -import org.apache.paimon.schema.TableSchema; -import org.apache.paimon.table.BucketMode; import org.apache.paimon.table.FileStoreTable; -import org.apache.paimon.table.PostponeUtils; import org.apache.paimon.utils.ChangelogManager; import org.apache.paimon.utils.FunctionWithException; import org.apache.paimon.utils.SnapshotManager; @@ -255,45 +252,28 @@ private static Snapshot changelogOrSnapshot( } public static void checkRescaleBucketForIncrementalDiffQuery( - SchemaManager schemaManager, SnapshotReader reader, Snapshot start, Snapshot end) { - TableSchema schema = schemaManager.latest().get(); - if (!schema.primaryKeys().isEmpty() && schema.numBuckets() == BucketMode.POSTPONE_BUCKET) { - Map startBucketNumbers = - PostponeUtils.getKnownNumBuckets(reader, start.id()); - Map endBucketNumbers = - PostponeUtils.getKnownNumBuckets(reader, end.id()); - for (Map.Entry entry : startBucketNumbers.entrySet()) { - Integer endPartitionBucketNumber = endBucketNumbers.get(entry.getKey()); - if (endPartitionBucketNumber != null - && !entry.getValue().equals(endPartitionBucketNumber)) { - throw new InconsistentTagBucketException( - start.id(), - end.id(), - "The real bucket number of two snapshots in postpone-bucket mode are different, " - + "which is not supported in incremental diff query."); - } - } - } else { - if (start.schemaId() != end.schemaId()) { - int startBucketNumber = bucketNumber(schemaManager, start.schemaId()); - int endBucketNumber = bucketNumber(schemaManager, end.schemaId()); - if (startBucketNumber != endBucketNumber) { - throw new InconsistentTagBucketException( - start.id(), - end.id(), - String.format( - "The bucket number of two snapshots are different (%s, %s), which is not supported in incremental diff query.", - startBucketNumber, endBucketNumber)); - } + Snapshot start, + Map>> startFiles, + Snapshot end, + Map>> endFiles) { + for (Map.Entry>> entry : + startFiles.entrySet()) { + Map> endPartitionFiles = endFiles.get(entry.getKey()); + Integer endPartitionBucketNumber = + endPartitionFiles == null ? null : endPartitionFiles.size(); + int startPartitionBucketNumber = entry.getValue().size(); + if (endPartitionBucketNumber != null + && startPartitionBucketNumber != endPartitionBucketNumber) { + throw new InconsistentTagBucketException( + start.id(), + end.id(), + String.format( + "Found different real bucket number (%s, %s), which is not supported in incremental diff query.", + startPartitionBucketNumber, endPartitionBucketNumber)); } } } - private static int bucketNumber(SchemaManager schemaManager, long schemaId) { - TableSchema schema = schemaManager.schema(schemaId); - return CoreOptions.fromMap(schema.options()).bucket(); - } - /** * Exception thrown when the bucket number of two tags are different in incremental tag query. */ diff --git a/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java index 7e04898c1c48..43ce25603d3b 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java @@ -651,6 +651,17 @@ public void testPostponeDifferentBucketNumbersForDifferentPartitions() throws Ex } table.createTag("TAG2", 2); + // test snapshot expiration won't affect tag diff query + try (TableWriteImpl write = builder.newWrite(); + BatchTableCommit commit = builder.newCommit()) { + write.writeAndReturn(GenericRow.of(3, 1, 1), 0, 1); + commit.commit(write.prepareCommit()); + } + table.newExpireSnapshots() + .config(ExpireConfig.builder().snapshotRetainMax(1).snapshotRetainMin(1).build()) + .expire(); + assertThat(table.snapshotManager().snapshotCount()).isEqualTo(1); + assertThat(read(table, Pair.of(INCREMENTAL_BETWEEN, "TAG1,TAG2"))) .containsExactly(GenericRow.of(2, 1, 1)); } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java index bcac28de025f..1d07359b6db9 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java @@ -55,12 +55,12 @@ public class PostponeUtilsTest { public void testGetKnownNumBucketsFromSnapshot() { BinaryRow partition = partition(1); PartitionPredicate partitionFilter = mock(PartitionPredicate.class); - ManifestEntry entry = mock(ManifestEntry.class); + SimpleFileEntry entry = mock(SimpleFileEntry.class); when(entry.partition()).thenReturn(partition); when(entry.totalBuckets()).thenReturn(4); FileStoreScan scan = mock(FileStoreScan.class, RETURNS_SELF); - when(scan.readFileIterator()).thenReturn(Collections.singletonList(entry).iterator()); + when(scan.readSimpleEntries()).thenReturn(Collections.singletonList(entry)); FileStore store = mock(FileStore.class); when(store.newScan()).thenReturn(scan); FileStoreTable table = mock(FileStoreTable.class); @@ -77,12 +77,12 @@ public void testGetKnownNumBucketsFromSnapshot() { public void testGetKnownNumBucketsByPartitions() { BinaryRow partition = partition(1); List partitions = Collections.singletonList(partition); - ManifestEntry entry = mock(ManifestEntry.class); + SimpleFileEntry entry = mock(SimpleFileEntry.class); when(entry.partition()).thenReturn(partition); when(entry.totalBuckets()).thenReturn(4); FileStoreScan scan = mock(FileStoreScan.class, RETURNS_SELF); - when(scan.readFileIterator()).thenReturn(Collections.singletonList(entry).iterator()); + when(scan.readSimpleEntries()).thenReturn(Collections.singletonList(entry)); FileStore store = mock(FileStore.class); when(store.newScan()).thenReturn(scan); FileStoreTable table = mock(FileStoreTable.class); From c7935c1becaeac50a6fa4c9def3598194f99264a Mon Sep 17 00:00:00 2001 From: yuzelin Date: Thu, 6 Aug 2026 13:20:20 +0800 Subject: [PATCH 5/8] fix --- .../table/source/snapshot/TimeTravelUtil.java | 57 +++++++++++++++---- .../paimon/table/IncrementalTableTest.java | 33 ++++++++++- 2 files changed, 79 insertions(+), 11 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java index 7d6af8cd37bc..703338ae0831 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java @@ -41,6 +41,7 @@ import java.util.Map; import java.util.Optional; import java.util.TimeZone; +import java.util.stream.Collectors; import static org.apache.paimon.CoreOptions.SCAN_SNAPSHOT_ID; import static org.apache.paimon.CoreOptions.SCAN_TAG_NAME; @@ -256,22 +257,58 @@ public static void checkRescaleBucketForIncrementalDiffQuery( Map>> startFiles, Snapshot end, Map>> endFiles) { + for (Map.Entry>> entry : startFiles.entrySet()) { Map> endPartitionFiles = endFiles.get(entry.getKey()); + if (endPartitionFiles == null) { + continue; + } + + Integer startPartitionBucketNumber = + realBucketNumbers( + entry.getValue().values().stream() + .flatMap(List::stream) + .collect(Collectors.toList())); Integer endPartitionBucketNumber = - endPartitionFiles == null ? null : endPartitionFiles.size(); - int startPartitionBucketNumber = entry.getValue().size(); - if (endPartitionBucketNumber != null - && startPartitionBucketNumber != endPartitionBucketNumber) { - throw new InconsistentTagBucketException( - start.id(), - end.id(), - String.format( - "Found different real bucket number (%s, %s), which is not supported in incremental diff query.", - startPartitionBucketNumber, endPartitionBucketNumber)); + realBucketNumbers( + endPartitionFiles.values().stream() + .flatMap(List::stream) + .collect(Collectors.toList())); + + if (startPartitionBucketNumber != null + && endPartitionBucketNumber != null + && startPartitionBucketNumber.equals(endPartitionBucketNumber)) { + return; + } + + throw new InconsistentTagBucketException( + start.id(), + end.id(), + String.format( + "The bucket number of two snapshots are different (%s, %s), which is not supported in incremental diff query.", + startPartitionBucketNumber, endPartitionBucketNumber)); + } + } + + @Nullable + private static Integer realBucketNumbers(List entries) { + Integer totalBuckets = null; + for (ManifestEntry entry : entries) { + if (entry.totalBuckets() >= 0) { + if (totalBuckets != null && totalBuckets != entry.totalBuckets()) { + throw new IllegalStateException( + "Partition " + + entry.partition() + + " has different totalBuckets " + + totalBuckets + + " and " + + entry.totalBuckets()); + } + totalBuckets = entry.totalBuckets(); } } + return totalBuckets; } /** diff --git a/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java index 43ce25603d3b..cc90e4f4ea1a 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java @@ -584,6 +584,38 @@ public void testIncrementalEmptyResult() throws Exception { .isEmpty(); } + @Test + public void testPostponeSameBucketNumberWithDifferentActiveBuckets() throws Exception { + Identifier identifier = identifier("T"); + Schema schema = + Schema.newBuilder() + .column("pk", DataTypes.INT()) + .column("col1", DataTypes.INT()) + .primaryKey("pk") + .option("bucket", String.valueOf(BucketMode.POSTPONE_BUCKET)) + .build(); + catalog.createTable(identifier, schema, true); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + + PostponeFixedBucketWriteBuilder builder = table.newPostponeFixedBucketWriteBuilder(); + try (TableWriteImpl write = builder.newWrite(); + BatchTableCommit commit = builder.newCommit()) { + write.writeAndReturn(GenericRow.of(1, 1), 0, 2); + commit.commit(write.prepareCommit()); + } + table.createTag("TAG1", 1); + + try (TableWriteImpl write = builder.newWrite(); + BatchTableCommit commit = builder.newCommit()) { + write.writeAndReturn(GenericRow.of(2, 2), 1, 2); + commit.commit(write.prepareCommit()); + } + table.createTag("TAG2", 2); + + assertThat(read(table, Pair.of(INCREMENTAL_BETWEEN, "TAG1,TAG2"))) + .containsExactly(GenericRow.of(2, 2)); + } + @Test public void testPostponeBucketNumberChangedInIncrementalDiff() throws Exception { Identifier identifier = identifier("T"); @@ -609,7 +641,6 @@ public void testPostponeBucketNumberChangedInIncrementalDiff() throws Exception try (TableWriteImpl write = builder.newWrite(); BatchTableCommit commit = builder.newCommit()) { write.writeAndReturn(GenericRow.of(1, 2), 0, 2); - write.writeAndReturn(GenericRow.of(2, 1), 1, 2); commit.commit(write.prepareCommit()); } table.createTag("TAG2", 2); From 05da3049aa578e78d76ff8ce2c59808e721a4199 Mon Sep 17 00:00:00 2001 From: yuzelin Date: Thu, 6 Aug 2026 15:06:23 +0800 Subject: [PATCH 6/8] fix comments --- .../table/source/snapshot/TimeTravelUtil.java | 2 +- .../paimon/table/IncrementalTableTest.java | 42 ++++++++++++++++++- 2 files changed, 42 insertions(+), 2 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java index 703338ae0831..7c5a1a8d4692 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java @@ -279,7 +279,7 @@ public static void checkRescaleBucketForIncrementalDiffQuery( if (startPartitionBucketNumber != null && endPartitionBucketNumber != null && startPartitionBucketNumber.equals(endPartitionBucketNumber)) { - return; + continue; } throw new InconsistentTagBucketException( diff --git a/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java index cc90e4f4ea1a..51f245982e9b 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java @@ -648,7 +648,47 @@ public void testPostponeBucketNumberChangedInIncrementalDiff() throws Exception assertThatThrownBy(() -> read(table, Pair.of(INCREMENTAL_BETWEEN, "TAG1,TAG2"))) .isInstanceOf(InconsistentTagBucketException.class) .hasMessageContaining( - "The real bucket number of two snapshots in postpone-bucket mode are different, " + "The bucket number of two snapshots are different (1, 2), " + + "which is not supported in incremental diff query."); + } + + @Test + public void testPostponeBucketNumberChangedInLaterPartition() throws Exception { + Identifier identifier = identifier("T"); + Schema schema = + Schema.newBuilder() + .column("pt", DataTypes.INT()) + .column("pk", DataTypes.INT()) + .column("col1", DataTypes.INT()) + .partitionKeys("pt") + .primaryKey("pk", "pt") + .option("bucket", String.valueOf(BucketMode.POSTPONE_BUCKET)) + .build(); + catalog.createTable(identifier, schema, true); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier); + + PostponeFixedBucketWriteBuilder builder = table.newPostponeFixedBucketWriteBuilder(); + try (TableWriteImpl write = builder.newWrite(); + BatchTableCommit commit = builder.newCommit()) { + write.writeAndReturn(GenericRow.of(2, 2, 2), 0, 1); + write.writeAndReturn(GenericRow.of(1, 1, 1), 0, 1); + commit.commit(write.prepareCommit()); + } + table.createTag("TAG1", 1); + + builder = table.newPostponeFixedBucketWriteBuilder().withOverwrite(Collections.emptyMap()); + try (TableWriteImpl write = builder.newWrite(); + BatchTableCommit commit = builder.newCommit()) { + write.writeAndReturn(GenericRow.of(2, 2, 3), 0, 1); + write.writeAndReturn(GenericRow.of(1, 1, 2), 0, 2); + commit.commit(write.prepareCommit()); + } + table.createTag("TAG2", 2); + + assertThatThrownBy(() -> read(table, Pair.of(INCREMENTAL_BETWEEN, "TAG1,TAG2"))) + .isInstanceOf(InconsistentTagBucketException.class) + .hasMessageContaining( + "The bucket number of two snapshots are different (1, 2), " + "which is not supported in incremental diff query."); } From aee99f8fa7803f0d590adaba5d6ee89138e6bd42 Mon Sep 17 00:00:00 2001 From: yuzelin Date: Thu, 6 Aug 2026 16:13:47 +0800 Subject: [PATCH 7/8] fix --- .../paimon/table/source/snapshot/SnapshotReaderImpl.java | 2 +- .../apache/paimon/table/source/snapshot/TimeTravelUtil.java | 5 +++++ 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java index a0f90fe3caee..cb338c857089 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java @@ -609,7 +609,7 @@ public Plan readIncrementalDiff(Snapshot before) { Map>> beforeFiles = groupByPartFiles(scan.withSnapshot(before).plan().files(FileKind.ADD)); TimeTravelUtil.checkRescaleBucketForIncrementalDiffQuery( - before, beforeFiles, plan.snapshot(), afterFiles); + tableSchema, before, beforeFiles, plan.snapshot(), afterFiles); return toIncrementalPlan( false, new LazyField<>(() -> before), diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java index 7c5a1a8d4692..4299a89b2c72 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java @@ -23,6 +23,7 @@ import org.apache.paimon.data.BinaryRow; import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.options.Options; +import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.utils.ChangelogManager; import org.apache.paimon.utils.FunctionWithException; @@ -253,10 +254,14 @@ private static Snapshot changelogOrSnapshot( } public static void checkRescaleBucketForIncrementalDiffQuery( + TableSchema schema, Snapshot start, Map>> startFiles, Snapshot end, Map>> endFiles) { + if (schema.primaryKeys().isEmpty() && schema.numBuckets() == -1) { + return; + } for (Map.Entry>> entry : startFiles.entrySet()) { From f0a9f974ce37d0aeccf0913d0fd63147c018d009 Mon Sep 17 00:00:00 2001 From: yuzelin Date: Thu, 6 Aug 2026 17:19:49 +0800 Subject: [PATCH 8/8] fix --- .../org/apache/paimon/table/source/snapshot/TimeTravelUtil.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java index 4299a89b2c72..47d7bc6d4424 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java @@ -259,7 +259,7 @@ public static void checkRescaleBucketForIncrementalDiffQuery( Map>> startFiles, Snapshot end, Map>> endFiles) { - if (schema.primaryKeys().isEmpty() && schema.numBuckets() == -1) { + if (schema.numBuckets() == -1) { return; }