JingsongLi commented on code in PR #7865:
URL: https://github.com/apache/paimon/pull/7865#discussion_r3485066393
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RescaleBucketITCase.java:
##########
@@ -229,6 +233,234 @@ private void innerTest(String catalogName, String
tableName) {
.containsExactlyInAnyOrderElementsOf(expected);
}
+ @Test
+ public void testRescaleSinglePartitionViaInsertOverwrite() throws
Exception {
+ // Create a partitioned primary-key table with 2 buckets.
+ // Primary-key tables are sensitive to bucket routing: each row must
land in the
+ // bucket determined by hash(pk) % totalBuckets, otherwise merging
during reads
+ // produces duplicates (LSM finds the same key in different buckets
and returns both).
+ batchSql("USE CATALOG fs_catalog");
+ batchSql("DROP TABLE IF EXISTS `TP`");
+ batchSql(
+ "CREATE TABLE `TP` "
+ + "(dt STRING, f0 INT, PRIMARY KEY (dt, f0) NOT
ENFORCED) "
+ + "PARTITIONED BY (dt) "
+ + "WITH ('bucket' = '2')");
+
+ // Insert data into two partitions
+ batchSql("INSERT INTO TP VALUES ('p1', 1), ('p1', 2), ('p1', 3)");
+ batchSql("INSERT INTO TP VALUES ('p2', 4), ('p2', 5), ('p2', 6)");
+
+ // Alter bucket count to 4
+ batchSql(alterTableSql, "TP", 4);
+
+ // INSERT OVERWRITE only partition p1 via the SQL path.
+ // The SQL path goes through FlinkTableSinkBase which does NOT wrap
the table in
+ // OverwriteFileStoreTable. As a result, table.createRowKeyExtractor()
calls
+ // PartitionBucketMapping.loadFromTable() which reads the old
per-partition bucket
+ // count from the manifest (2 for p1) and uses it for row routing
(hashing).
+ // Rows land in buckets [0,1] but files are stamped totalBuckets=4.
+ // A subsequent upsert into p1 routes rows using the new 4-bucket
mapping — the
+ // same key may appear in two different files with different bucket
assignments,
+ // causing duplicate reads.
+ batchSql("INSERT OVERWRITE TP PARTITION (dt = 'p1') SELECT f0 FROM TP
WHERE dt = 'p1'");
+
+ // Verify the overwrite preserved data correctly
+ assertThat(batchSql("SELECT * FROM TP WHERE dt = 'p1'"))
+ .containsExactlyInAnyOrder(Row.of("p1", 1), Row.of("p1", 2),
Row.of("p1", 3));
+ assertThat(batchSql("SELECT * FROM TP WHERE dt = 'p2'"))
+ .containsExactlyInAnyOrder(Row.of("p2", 4), Row.of("p2", 5),
Row.of("p2", 6));
+
+ // Now upsert (update) existing rows in p1 — this uses the new
4-bucket mapping.
+ // If the overwrite used the old 2-bucket hashing, the original row
for key (p1,1)
+ // lands in bucket hash(1)%2, but the update lands in bucket hash(1)%4.
+ // The LSM reader finds two files for key (p1,1) in different buckets
and returns
+ // both — causing duplicate rows.
+ batchSql("INSERT INTO TP VALUES ('p1', 1), ('p1', 2), ('p1', 3)");
+ assertThat(batchSql("SELECT * FROM TP WHERE dt = 'p1'"))
+ .as(
+ "After rescale overwrite + upsert of same keys, each
key must appear "
+ + "exactly once. If INSERT OVERWRITE used the
old 2-bucket "
+ + "routing instead of the new 4-bucket routing
(via "
+ + "OverwriteFileStoreTable), the same key will
exist in two "
+ + "different buckets causing duplicate rows on
read.")
+ .containsExactlyInAnyOrder(Row.of("p1", 1), Row.of("p1", 2),
Row.of("p1", 3));
+
+ // Also verify the files written by INSERT OVERWRITE are stamped with
the new bucket count
+ FileStoreTable fileStoreTable = paimonTable("TP");
+ Iterator<ManifestEntry> it =
+ fileStoreTable
+ .newSnapshotReader()
+ .withPartitionFilter(Collections.singletonMap("dt",
"p1"))
+ .onlyReadRealBuckets()
+ .readFileIterator();
+ assertThat(it.hasNext()).isTrue();
+ while (it.hasNext()) {
+ ManifestEntry entry = it.next();
+ assertThat(entry.totalBuckets())
+ .as("Files in partition p1 must be stamped with the new
bucket count (4)")
+ .isEqualTo(4);
+ assertThat(entry.bucket()).as("Bucket index must be in range [0,
3]").isBetween(0, 3);
+ }
+
+ batchSql("USE CATALOG default_catalog");
+ }
+
+ @Test
+ public void testWriteToEmptyBucketAfterRescaleKeepsPartitionBucketCount()
throws Exception {
+ // End-to-end reproduction of the FileSystemWriteRestore empty-bucket
scenario.
+ //
+ // The scenario occurs when a partition has a bucket count that
differs from
+ // the table-level default, AND at least one of that partition's
buckets
+ // has NO existing data files. A subsequent write that lands in the
+ // empty bucket must keep the partition's bucket count, not silently
+ // adopt the table default.
+ //
+ // Final state we want before the bug-triggering write (Step 5):
+ // - table default bucket : 2
+ // - partition p1 bucket : 4 (held over from a prior
rescale+overwrite)
+ // - partition p1 bucket=X: has data files, totalBuckets = 4
+ // - partition p1 bucket=Y: NO data files
+ //
+ // The Step 5 write fills bucket=Y in p1. Post-fix, the new file MUST
+ // be stamped totalBuckets=4. Before the fix, FileSystemWriteRestore
+ // returned null for the empty bucket, the writer fell back to the
+ // table default (2), and the new file was stamped totalBuckets=2,
+ // corrupting p1's bucket layout.
+
+ // Step 1: create a partitioned PK table with bucket=2 and seed one row
+ // into p1.
+ batchSql(
+ "CREATE TABLE IF NOT EXISTS `fs_catalog`.`default`.`TPE` "
+ + "(dt STRING, f0 INT, PRIMARY KEY (dt, f0) NOT
ENFORCED) "
+ + "PARTITIONED BY (dt) "
+ + "WITH ('bucket' = '2')");
+ batchSql("USE CATALOG fs_catalog");
+ batchSql("INSERT INTO TPE VALUES ('p1', 1)");
+ // Also seed partition p2 — this acts as a positive control: p2 is
+ // never rescaled and must always be stamped with the current table
+ // default bucket count.
+ batchSql("INSERT INTO TPE VALUES ('p2', 1)");
+
+ // Step 2: rescale the table default to 4.
+ batchSql(alterTableSql, "TPE", 4);
+
+ // Step 3: INSERT OVERWRITE p1 — this rewrites p1's files using the
+ // current table default (4), so afterwards p1's files are stamped
+ // totalBuckets=4. With only a single row in p1, at least one of p1's
+ // four buckets is guaranteed to be empty.
+ batchSql("INSERT OVERWRITE TPE PARTITION (dt = 'p1') SELECT f0 FROM
TPE WHERE dt = 'p1'");
+
+ // Step 4 (NEW): rescale the table default BACK to 2, while p1 keeps
+ // its 4-bucket layout. Now the per-partition bucket count (4) diverges
+ // from the table default (2) — the precondition for the bug.
+ batchSql(alterTableSql, "TPE", 2);
+
+ // Step 5: insert more rows into p1. PartitionBucketMapping reads the
+ // manifest and routes these rows using p1's actual bucket count (4).
+ // Several rows are inserted so that — with high probability — at
+ // least one lands in the bucket that was previously empty, exercising
+ // the FileSystemWriteRestore empty-bucket code path.
+ // Also insert another row into p2 — p2 has never been rescaled and
+ // must remain stamped with the current table default bucket count (2).
+ batchSql(
+ "INSERT INTO TPE VALUES "
+ + "('p1', 2), ('p1', 3), ('p1', 4), ('p1', 5), ('p1',
6), "
+ + "('p2', 2), ('p2', 3), ('p2', 4), ('p2', 5), ('p2',
6)");
+
+ // Sanity check: each PK appears exactly once. Duplicates would
indicate
+ // that the new rows landed in an inconsistent bucket layout (e.g. some
+ // files stamped totalBuckets=2 and others stamped totalBuckets=4),
+ // causing the LSM reader to find the same PK in two different buckets.
+ assertThat(batchSql("SELECT * FROM TPE"))
+ .as(
+ "Each PK must appear exactly once across all
partitions. Duplicates "
+ + "indicate rows landed in inconsistent bucket
layouts.")
+ .containsExactlyInAnyOrder(
+ Row.of("p1", 1),
+ Row.of("p1", 2),
+ Row.of("p1", 3),
+ Row.of("p1", 4),
+ Row.of("p1", 5),
+ Row.of("p1", 6),
+ Row.of("p2", 1),
+ Row.of("p2", 2),
+ Row.of("p2", 3),
+ Row.of("p2", 4),
+ Row.of("p2", 5),
+ Row.of("p2", 6));
+
+ // Strong assertion (p1 — the rescaled partition): every file must be
+ // stamped totalBuckets=4 (the partition's actual bucket count), NOT 2
+ // (the new table default). This is the precise condition the fix
+ // guarantees for partitions whose bucket count differs from the table
+ // default.
+ FileStoreTable fileStoreTable = paimonTable("TPE");
+ Iterator<ManifestEntry> p1Iter =
+ fileStoreTable
+ .newSnapshotReader()
+ .withPartitionFilter(Collections.singletonMap("dt",
"p1"))
+ .onlyReadRealBuckets()
+ .readFileIterator();
+ assertThat(p1Iter.hasNext()).isTrue();
+ while (p1Iter.hasNext()) {
+ ManifestEntry entry = p1Iter.next();
+ assertThat(entry.totalBuckets())
+ .as(
+ "Files in partition p1 must keep the partition's
bucket count (4). "
+ + "If 2, FileSystemWriteRestore failed to
fall back to "
+ + "PartitionBucketMapping for buckets that
had no existing "
+ + "files at restore time, and stamped new
files with the "
+ + "table-level default bucket count
instead.")
+ .isEqualTo(4);
+ assertThat(entry.bucket()).as("Bucket index must be in range [0,
3]").isBetween(0, 3);
+ }
+
+ // Positive control (p2 — never rescaled): every file must be stamped
+ // with the current table default (2). This guards against an
+ // over-eager fix that would erroneously route ALL writes through
+ // PartitionBucketMapping even when no per-partition override exists.
+ Iterator<ManifestEntry> p2Iter =
+ fileStoreTable
+ .newSnapshotReader()
+ .withPartitionFilter(Collections.singletonMap("dt",
"p2"))
+ .onlyReadRealBuckets()
+ .readFileIterator();
+ assertThat(p2Iter.hasNext()).isTrue();
+ java.util.List<ManifestEntry> p2Entries = new java.util.ArrayList<>();
+ while (p2Iter.hasNext()) {
+ p2Entries.add(p2Iter.next());
+ }
+ // Diagnostic: dump every manifest entry in p2 so we can see what each
+ // file is stamped with (FileKind, bucket, totalBuckets, file name,
+ // snapshot/sequence info via the file meta).
+ System.out.println("=== p2 manifest entries (" + p2Entries.size() + "
total) ===");
Review Comment:
Please remove this diagnostic dump before merging. It prints multiple
manifest-entry lines during normal test runs, and the assertions below already
validate the p2 entries; if extra context is needed, prefer putting it into the
assertion descriptions.
##########
docs/docs/maintenance/rescale-bucket.md:
##########
@@ -45,14 +45,17 @@ Please note that
- `ALTER TABLE` only modifies the table's metadata and will **NOT** reorganize
or reformat existing data.
Reorganize existing data must be achieved by `INSERT OVERWRITE`.
- Rescale bucket number does not influence the read and running write jobs.
-- Once the bucket number is changed, any newly scheduled `INSERT INTO` jobs
which write to without-reorganized
- existing table/partition will throw a `TableException` with message like
+- **Partitioned tables** support per-partition bucket counts. Each partition
retains its own bucket
+ count from its data files, and the new bucket count only applies to newly
created partitions or partitions that
+ have been reorganized with `INSERT OVERWRITE`.
+- **Unpartitioned tables** require a full rescale before writing. If you
change the bucket number and attempt
+ to write without reorganizing the data first, a `RuntimeException` will be
thrown:
```text
- Try to write table/partition ... with a new bucket num ...,
+ Try to write table with a new bucket num ...,
Review Comment:
Please remove the trailing spaces introduced in this doc. `git diff --check
origin/master...HEAD` currently reports this line and lines 149, 170, and 171,
so the final verification will fail.
--
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]
