JingsongLi commented on code in PR #10297:
URL: https://github.com/apache/paimon/pull/10297#discussion_r4245180377
##########
paimon-core/src/main/java/org/apache/paimon/operation/FileSystemWriteRestore.java:
##########
@@ -74,6 +78,22 @@ private FileSystemWriteRestore(
this.scan.dropStats();
}
}
+ this.partitionBucketMapping = null;
+ }
+
+ private PartitionBucketMapping partitionBucketMapping(Snapshot snapshot) {
+ if (!options.bucketPerPartitionCountEnabled()) {
+ return PartitionBucketMapping.defaultBuckets(options.bucket());
+ }
+ if (partitionBucketMapping == null
+ || partitionBucketMappingSnapshotId == null
+ || partitionBucketMappingSnapshotId != snapshot.id()) {
+ partitionBucketMapping =
+ PartitionBucketMapping.loadFromScan(
+ scan.withSnapshot(snapshot), options.bucket(),
true);
Review Comment:
[P1] Rebuild the restore layout from an unfiltered scan
`restoreFiles` leaves this shared scan narrowed by
`withPartitionBucket(partition, bucket)`. When a later commit changes the
snapshot ID, this new mapping rebuild only changes the snapshot and retains the
previous partition and bucket filters, so it no longer obtains the
authoritative layout of other partitions. I reproduced a real PK partition P2
overwritten from 2 to 4 buckets (table default 4, mod key 3 in bucket 3), first
restored P1/bucket 0, committed another P1 row, then reused this restorer for
empty P2/bucket 1. The rebuilt mapping omits P2 and returns
`totalBuckets=null`; an old count-2 write passes the strict guard and an
ordinary commit/read returns both key-3 values 300 and 999. A fresh restorer
rejects the same write. Please use a separate unfiltered layout scan, or clear
both retained filters before rebuilding; cover one restorer serving multiple
partitions across successive snapshots, not only its initial restore.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinator.java:
##########
@@ -223,6 +218,28 @@ public void checkpoint() {
latestCommittedIdentifiers.clear();
}
+ private PartitionBucketMapping singlePartitionBucketMapping(BinaryRow
partition) {
+ int defaultBuckets = table.schema().numBuckets();
+ if (!table.coreOptions().bucketPerPartitionCountEnabled()) {
+ return PartitionBucketMapping.defaultBuckets(defaultBuckets);
+ }
+
+ FileStoreScan partitionScan =
+ newScan()
+ .withSnapshot(snapshot)
+
.withPartitionFilter(Collections.singletonList(partition));
+
+ return PartitionBucketMapping.loadFromScan(partitionScan,
defaultBuckets);
Review Comment:
[P1] Retain known default-count partitions in coordinator restores
This uses the overload that drops partitions whose current count equals
`defaultBuckets`, whereas the repaired FileSystemWriteRestore passes
`includeDefaultBucketCount=true`. With the supported
`sink.writer-coordinator.enabled=true` path, a stale job mapping P1 to 2
buckets after P1 is physically overwritten to the table default of 4 therefore
receives `totalBuckets=null` when opening an empty old-hash bucket.
CoordinatedWriteRestore forwards that null, so the writer skips its count check
and stamps the old count 2. The actual coordinator/restore/ordinary-commit
reproduction returned `(1,3,300)` and `(1,3,999)` for one primary key, with
manifest counts `{2,4}`; the identical extractor through a fresh filesystem
restorer correctly fails on new count 2 versus previous count 4. Please retain
default-count partitions here as well and add coordinator-mode coverage for a
rescale back to the default count with an empty target bucket.
--
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]
