@@ -68,6 +68,12 @@ Please note that
   After these operations, partition `dt=2022-01-01` uses 4 buckets, `dt=2022-01-02` uses 8 buckets, and any
   new partitions will use the latest table-level default (8 buckets in this case).
 - During overwrite period, make sure there are no other jobs writing the same table/partition.
+- **Streaming jobs must be restarted after rescaling a partition.** The per-partition bucket mapping
+  is loaded once when the streaming job starts (from the manifest files at that point in time). If a
+  partition is rescaled while the streaming job is running, the job will continue routing rows using
+  the old bucket count for that partition, which can cause rows to land in wrong buckets and lead to
+  data correctness issues. The recommended workflow is: suspend the streaming job with a savepoint →
+  perform the rescale overwrite → restart from the savepoint.
 
 ## Use Case
 
 
@@ -83,20 +83,26 @@ public int totalBuckets() {
     }
 
     public PartitionEntry merge(PartitionEntry entry) {
+        PartitionEntry newer = entry.lastFileCreationTime >= lastFileCreationTime ? entry : this;
+        PartitionEntry older = newer == entry ? this : entry;
+
         // Use the totalBuckets from the most recently created file. This correctly handles
         // the case where a partition has been overwritten with a different bucket count: the
         // newer files carry the new totalBuckets, and their creation time is always later.
-        long newLastCreationTime = Math.max(lastFileCreationTime, entry.lastFileCreationTime);
+        // When timestamps are equal (e.g., two files written in the same millisecond with
+        // different bucket counts), we take the larger totalBuckets value. This makes merge
+        // commutative and associative — a.merge(b) == b.merge(a)
         int newTotalBuckets =
-                lastFileCreationTime >= entry.lastFileCreationTime
-                        ? totalBuckets
-                        : entry.totalBuckets;
+                newer.lastFileCreationTime == older.lastFileCreationTime
+                        ? Math.max(newer.totalBuckets, older.totalBuckets)
+                        : newer.totalBuckets;
+
         return new PartitionEntry(
                 partition,
                 recordCount + entry.recordCount,
                 fileSizeInBytes + entry.fileSizeInBytes,
                 fileCount + entry.fileCount,
-                newLastCreationTime,
+                newer.lastFileCreationTime,
                 newTotalBuckets);
     }
 
 
@@ -65,6 +65,14 @@ public TableWriteImpl<?> newWrite(String commitUser, @Nullable Integer writeId)
         return wrapped().newWrite(commitUser, writeId, createRowKeyExtractor());
     }
 
+    @Override
+    public TableWriteImpl<?> newWrite(
+            String commitUser, @Nullable Integer writeId, RowKeyExtractor rowKeyExtractor) {
+        // Always use the schema-bucket-based extractor; ignore the caller-supplied extractor
+        // to ensure consistent per-partition bucket routing even when called via the 3-arg form.
+        return wrapped().newWrite(commitUser, writeId, createRowKeyExtractor());
+    }
+
     @Override
     public FileStoreTable copy(Map<String, String> dynamicOptions) {
         return new SchemaBucketFileStoreTable(wrapped().copy(dynamicOptions));
 
@@ -83,8 +83,7 @@ public PartitionBucketMapping(
      * <p>For partitioned tables, the method reads {@link
      * org.apache.paimon.manifest.PartitionEntry}s, which aggregate manifest entries per partition
      * during the scan and therefore have a much smaller memory footprint than loading all data file
-     * entries. If the scan fails for any reason, a fallback mapping with only the default bucket
-     * count is returned.
+     * entries. Any scan failure is propagated to the caller.
      *
      * @param table the {@link FileStoreTable} to load the mapping from
      * @return a {@link PartitionBucketMapping} reflecting the current bucket layout of the table
@@ -98,23 +97,21 @@ public static PartitionBucketMapping loadFromTable(FileStoreTable table) {
     }
 
     public static PartitionBucketMapping loadFromScan(FileStoreScan scan, int defaultBuckets) {
-        try {
-            List<PartitionEntry> partitionEntries = scan.readPartitionEntries();
-            Map<BinaryRow, Integer> partitionBucketMap = new HashMap<>();
-            for (PartitionEntry entry : partitionEntries) {
-                int totalBuckets = entry.totalBuckets();
-                // Only store partitions whose bucket count differs from the default.
-                // This keeps the map empty for partitions that have never been rescaled,
-                // avoiding per-partition BinaryRow copies and Integer allocations entirely.
-                if (totalBuckets > 0 && totalBuckets != defaultBuckets) {
-                    partitionBucketMap.put(entry.partition().copy(), totalBuckets);
-                }
-            }
-
-            return new PartitionBucketMapping(defaultBuckets, partitionBucketMap);
-        } catch (Exception e) {
+        if (scan == null) {
             return new PartitionBucketMapping(defaultBuckets, Collections.emptyMap());
         }
+        List<PartitionEntry> partitionEntries = scan.readPartitionEntries();
+        Map<BinaryRow, Integer> partitionBucketMap = new HashMap<>();
+        for (PartitionEntry entry : partitionEntries) {
+            int totalBuckets = entry.totalBuckets();
+            // Only store partitions whose bucket count differs from the default.
+            // This keeps the map empty for partitions that have never been rescaled,
+            // avoiding per-partition BinaryRow copies and Integer allocations entirely.
+            if (totalBuckets > 0 && totalBuckets != defaultBuckets) {
+                partitionBucketMap.put(entry.partition().copy(), totalBuckets);
+            }
+        }
+        return new PartitionBucketMapping(defaultBuckets, partitionBucketMap);
     }
 
     /**
 
@@ -103,16 +103,21 @@ public void testMergeWithDeleteEntryPreservesNewerTotalBuckets() {
     }
 
     @Test
-    public void testMergeWithEqualCreationTimeTakesFirstTotalBuckets() {
-        // When creation times are equal, the receiver's (this) totalBuckets is used.
-        // This is the tie-breaking behavior: the entry already accumulated takes precedence.
+    public void testMergeWithEqualCreationTimeIsCommutative() {
+        // When creation times are equal, merge must be commutative: a.merge(b) == b.merge(a).
+        // The tie-break takes the larger totalBuckets so that the parallel, non-deterministic
+        // aggregation in readPartitionEntries() always produces the same result regardless of
+        // manifest processing order.
         PartitionEntry a = entry(1, 2, 1000L);
         PartitionEntry b = entry(1, 4, 1000L);
 
-        PartitionEntry result = a.merge(b);
-        assertThat(result.totalBuckets())
-                .isEqualTo(2); // 'a' wins tie (lastCreationTime >= entry's)
-        assertThat(result.fileCount()).isEqualTo(2);
+        PartitionEntry ab = a.merge(b);
+        PartitionEntry ba = b.merge(a);
+
+        assertThat(ab.totalBuckets()).isEqualTo(4); // max(2, 4) = 4
+        assertThat(ba.totalBuckets()).isEqualTo(4); // max(4, 2) = 4, commutative
+        assertThat(ab.fileCount()).isEqualTo(2);
+        assertThat(ba.fileCount()).isEqualTo(2);
     }
 
     @Test
 
@@ -19,13 +19,16 @@
 package org.apache.paimon.table.sink;
 
 import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.operation.FileStoreScan;
 
 import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
 
 import java.util.HashMap;
 import java.util.Map;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Tests for {@link PartitionBucketMapping}. */
 public class PartitionBucketMappingTest {
@@ -60,6 +63,19 @@ public void testExplicitPartitionMapping() {
         assertThat(mapping.resolveNumBuckets(partC)).isEqualTo(16);
     }
 
+    @Test
+    public void testLoadFromScanPropagatesException() {
+        // Simulate a scan that throws (e.g. corrupted manifest, transient I/O error).
+        // loadFromScan must fail fast so the job does not silently write to wrong buckets.
+        FileStoreScan failingScan = Mockito.mock(FileStoreScan.class);
+        Mockito.when(failingScan.readPartitionEntries())
+                .thenThrow(new RuntimeException("simulated manifest scan failure"));
+
+        assertThatThrownBy(() -> PartitionBucketMapping.loadFromScan(failingScan, 8))
+                .isInstanceOf(RuntimeException.class)
+                .hasMessageContaining("simulated manifest scan failure");
+    }
+
     private static BinaryRow partition(int value) {
         return BinaryRow.singleColumn(value);
     }
 
@@ -199,6 +199,11 @@ public void checkpoint() {
 
     private void loadPartitionBucketMapping() {
         int defaultNumBuckets = table.schema().numBuckets();
+        // Note: `scan` is shared between this method (called during refresh/checkpoint) and the
+        // `scan(ScanCoordinationRequest)` method (which calls scan.withPartitionBucket(...)).
+        // Both callers always invoke scan.withSnapshot(...) before using the scan, so the shared
+        // state is safe. The partition-bucket filter set by withPartitionBucket is applied per
+        // plan() call and does not bleed across invocations.
         this.partitionBucketMapping = PartitionBucketMapping.loadFromScan(scan, defaultNumBuckets);
     }