@@ -75,13 +75,8 @@ Please note that
 :::
 
 :::warning
-  Per-partition bucket counts are currently supported by the **Flink** engine only. The **Spark** writer
-  still derives the bucket from the single table-level bucket count, so when partitions have different
-  bucket counts (for example, after changing the table-level `bucket` while existing partitions keep
-  their previous count), Spark may route rows to buckets that do not belong to the partition and corrupt
-  the per-partition layout. Until Spark support is added, use Flink to write to tables that have
-  per-partition bucket counts, or perform a full-table rescale so every partition shares the same bucket
-  count before writing with Spark.
+  Per-partition bucket counts are currently supported by the **Flink** engine only. Spark rejects writes
+  to a table with `'bucket.per-partition-count-enabled' = 'true'`; use Flink to write such a table.
 :::
 - **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:
@@ -190,4 +185,4 @@ and the job's latency keeps increasing. To improve the data freshness, users can
        DATE_FORMAT(gmt_create, 'yyyy-MM-dd') AS dt
   FROM raw_orders
   WHERE order_status = 'verified';
-  ```
+  ```
@@ -88,14 +88,21 @@ public PartitionEntry merge(PartitionEntry 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.
-        // 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 =
-                newer.lastFileCreationTime == older.lastFileCreationTime
-                        ? Math.max(newer.totalBuckets, older.totalBuckets)
-                        : newer.totalBuckets;
+        // newer files carry the new totalBuckets. A creation timestamp does not provide an
+        // ordering when the files were created in the same millisecond. In that case, choosing
+        // either bucket count (for example, the larger one) can silently discard a rescale.
+        if (newer.lastFileCreationTime == older.lastFileCreationTime
+                && newer.totalBuckets != older.totalBuckets) {
+            throw new IllegalStateException(
+                    String.format(
+                            "Cannot determine the bucket layout for partition %s: files created at %s "
+                                    + "have conflicting bucket counts (%s and %s).",
+                            partition,
+                            newer.lastFileCreationTime,
+                            newer.totalBuckets,
+                            older.totalBuckets));
+        }
+        int newTotalBuckets = newer.totalBuckets;
 
         return new PartitionEntry(
                 partition,
 
@@ -23,6 +23,7 @@
 import org.junit.jupiter.api.Test;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Tests for {@link PartitionEntry#merge(PartitionEntry)}. */
 public class PartitionEntryTest {
@@ -103,21 +104,19 @@ public void testMergeWithDeleteEntryPreservesNewerTotalBuckets() {
     }
 
     @Test
-    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.
+    public void testMergeWithEqualCreationTimeAndConflictingBucketsFails() {
+        // Creation time alone cannot order layouts written in the same millisecond. Choosing one
+        // of them (for example, the larger bucket count) can silently retain the old layout when
+        // a partition is downscaled, so reject the ambiguous metadata instead.
         PartitionEntry a = entry(1, 2, 1000L);
         PartitionEntry b = entry(1, 4, 1000L);
 
-        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);
+        assertThatThrownBy(() -> a.merge(b))
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessageContaining("Cannot determine the bucket layout");
+        assertThatThrownBy(() -> b.merge(a))
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessageContaining("Cannot determine the bucket layout");
     }
 
     @Test
 
@@ -124,6 +124,12 @@ case class PaimonSparkWriter(
   }
 
   def write(data: DataFrame): Seq[CommitMessage] = {
+    if (coreOptions.bucketPerPartitionCountEnabled()) {
+      throw new UnsupportedOperationException(
+        "Spark does not support writing tables with per-partition bucket counts. " +
+          "Use Flink to write this table or disable 'bucket.per-partition-count-enabled'.")
+    }
+
     val sparkSession = data.sparkSession
     val uriReaderFactory = uriReaderFactoryForBlobDescriptor
     import sparkSession.implicits._
 
@@ -36,6 +36,22 @@ class PaimonSinkTest extends PaimonSparkTestBase with StreamTest {
 
   import testImplicits._
 
+  test("Paimon Sink: reject per-partition bucket counts") {
+    spark.sql("""CREATE TABLE per_partition_buckets (id INT, pt STRING)
+                |PARTITIONED BY (pt)
+                |TBLPROPERTIES (
+                |  'bucket' = '2',
+                |  'bucket.per-partition-count-enabled' = 'true')
+                |""".stripMargin)
+
+    val exception = intercept[UnsupportedOperationException] {
+      spark.sql("INSERT INTO per_partition_buckets VALUES (1, 'p1')")
+    }
+    assert(
+      exception.getMessage.contains(
+        "Spark does not support writing tables with per-partition bucket counts"))
+  }
+
   test("Paimon Sink: forEachBatch") {
     failAfter(streamingTimeout) {
       withTempDir {