JingsongLi commented on code in PR #9048:
URL: https://github.com/apache/paimon/pull/9048#discussion_r3726218456
##########
paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java:
##########
@@ -251,24 +253,62 @@ 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));
+ Snapshot start,
+ Map<BinaryRow, Map<Integer, List<ManifestEntry>>> startFiles,
+ Snapshot end,
+ Map<BinaryRow, Map<Integer, List<ManifestEntry>>> endFiles) {
+
+ for (Map.Entry<BinaryRow, Map<Integer, List<ManifestEntry>>> entry :
+ startFiles.entrySet()) {
+ Map<Integer, List<ManifestEntry>> endPartitionFiles =
endFiles.get(entry.getKey());
+ if (endPartitionFiles == null) {
+ continue;
}
+
+ Integer startPartitionBucketNumber =
+ realBucketNumbers(
+ entry.getValue().values().stream()
+ .flatMap(List::stream)
+ .collect(Collectors.toList()));
+ Integer endPartitionBucketNumber =
+ realBucketNumbers(
+ endPartitionFiles.values().stream()
+ .flatMap(List::stream)
+ .collect(Collectors.toList()));
+
+ if (startPartitionBucketNumber != null
+ && endPartitionBucketNumber != null
+ &&
startPartitionBucketNumber.equals(endPartitionBucketNumber)) {
+ return;
Review Comment:
**[P1] Continue validating the remaining partitions**
When the first intersecting partition has the same actual bucket count, this
`return` exits the entire method, so a later partition whose bucket count
changed is never checked. With a `bucket = -2` partitioned table, I reproduced
this using unchanged common partitions plus one partition rescaled from 1 to 2
buckets between the tags: no `InconsistentTagBucketException` was thrown, and
the Spark `EXCEPT` fallback would therefore be skipped. Replacing this with
`continue` makes the regression pass. Please also add a multi-partition test
covering this ordering.
##########
paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java:
##########
@@ -581,6 +584,119 @@ 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");
+ 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);
+ 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, "
Review Comment:
**[P1] Keep the assertion in sync with the new message**
This targeted test currently fails because the assertion still expects `The
real bucket number...`, while `checkRescaleBucketForIncrementalDiffQuery` now
throws `The bucket number of two snapshots are different (1, 2)...`. Running
`mvn -pl paimon-core -Pfast-build -DwildcardSuites=none
-Dtest=IncrementalTableTest#testPostponeBucketNumberChangedInIncrementalDiff
test` fails at this assertion. Please update the expectation, or preserve the
previous exception message.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
