Fix Bug from comments

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java

@@ -53,7 +53,6 @@
 public class KeyValueFileStore extends AbstractFileStore<KeyValue> {
 
     private final boolean crossPartitionUpdate;
-    private final RowType bucketKeyType;
     private final RowType keyType;
     private final RowType valueType;
     private final KeyValueFieldsExtractor keyValueFieldsExtractor;
@@ -68,7 +67,6 @@ public KeyValueFileStore(
             boolean crossPartitionUpdate,
             CoreOptions options,
             RowType partitionType,
-            RowType bucketKeyType,
             RowType keyType,
             RowType valueType,
             KeyValueFieldsExtractor keyValueFieldsExtractor,
@@ -77,7 +75,6 @@ public KeyValueFileStore(
             CatalogEnvironment catalogEnvironment) {
         super(fileIO, schemaManager, schema, tableName, options, partitionType, catalogEnvironment);
         this.crossPartitionUpdate = crossPartitionUpdate;
-        this.bucketKeyType = bucketKeyType;
         this.keyType = keyType;
         this.valueType = valueType;
         this.keyValueFieldsExtractor = keyValueFieldsExtractor;
@@ -203,7 +200,7 @@ public KeyValueFileStoreScan newScan() {
                         options.bucketFunctionType(),
                         schema.logicalRowType(),
                         partitionType,
-                        bucketKeyType);
+                        schema.logicalBucketKeyType());
 
         return new KeyValueFileStoreScan(
                 newManifestsReader(),

paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java

@@ -93,8 +93,6 @@ public KeyValueFileStore store() {
                             tableSchema.crossPartitionUpdate(),
                             options,
                             tableSchema.logicalPartitionType(),
-                            PrimaryKeyTableUtils.addKeyNamePrefix(
-                                    tableSchema.logicalBucketKeyType()),
                             keyType,
                             rowType,
                             extractor,

paimon-core/src/test/java/org/apache/paimon/TestFileStore.java

@@ -134,7 +134,6 @@ private TestFileStore(
                 options,
                 partitionType,
                 keyType,
-                keyType,
                 valueType,
                 keyValueFieldsExtractor,
                 mfFactory,

paimon-core/src/test/java/org/apache/paimon/table/BucketFilterScanTest.java

@@ -18,7 +18,6 @@
 
 package org.apache.paimon.table;
 
-import org.apache.paimon.CoreOptions;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.predicate.Predicate;
@@ -32,6 +31,8 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import static org.apache.paimon.CoreOptions.BUCKET;
+import static org.apache.paimon.CoreOptions.BUCKET_KEY;
 import static org.assertj.core.api.Assertions.assertThat;
 
 /**
@@ -40,6 +41,26 @@
  */
 public class BucketFilterScanTest extends TableTestBase {
 
+    @Test
+    public void testBucketFilterWithCompoundPredicateOnAppendTable() throws Exception {
+        testBucketFilterWithCompoundPredicate(false);
+    }
+
+    @Test
+    public void testBucketFilterWithCompoundPredicateOnPkTable() throws Exception {
+        testBucketFilterWithCompoundPredicate(true);
+    }
+
+    @Test
+    public void testCompositeBucketFilterWithCompoundPredicateOnAppendTable() throws Exception {
+        testCompositeBucketFilterWithCompoundPredicate(false);
+    }
+
+    @Test
+    public void testCompositeBucketFilterWithCompoundPredicateOnPkTable() throws Exception {
+        testCompositeBucketFilterWithCompoundPredicate(true);
+    }
+
     /**
      * Tests bucket filtering with compound predicates on a single-field bucket key.
      *
@@ -67,18 +88,21 @@ public class BucketFilterScanTest extends TableTestBase {
      *       partition-b combination.
      * </ol>
      */
-    @Test
-    public void testBucketFilterWithCompoundPredicate() throws Exception {
+    private void testBucketFilterWithCompoundPredicate(boolean pk) throws Exception {
         // ---- schema & table ----
-        Schema schema =
+        Schema.Builder builder =
                 Schema.newBuilder()
                         .column("a", DataTypes.INT())
                         .column("b", DataTypes.INT())
                         .column("c", DataTypes.INT())
                         .partitionKeys("a")
-                        .option(CoreOptions.BUCKET.key(), "10")
-                        .option(CoreOptions.BUCKET_KEY.key(), "b")
-                        .build();
+                        .option(BUCKET.key(), "10");
+        if (pk) {
+            builder.primaryKey("a", "b");
+        } else {
+            builder.option(BUCKET_KEY.key(), "b");
+        }
+        Schema schema = builder.build();
 
         Identifier tableId = identifier("test_bucket_filter");
         catalog.createTable(tableId, schema, false);
@@ -148,18 +172,21 @@ public void testBucketFilterWithCompoundPredicate() throws Exception {
      *       targeting.
      * </ol>
      */
-    @Test
-    public void testCompositeBucketFilterWithCompoundPredicate() throws Exception {
+    private void testCompositeBucketFilterWithCompoundPredicate(boolean pk) throws Exception {
         // ---- schema & table ----
-        Schema schema =
+        Schema.Builder builder =
                 Schema.newBuilder()
                         .column("a", DataTypes.INT())
                         .column("b", DataTypes.INT())
                         .column("c", DataTypes.INT())
                         .partitionKeys("a")
-                        .option(CoreOptions.BUCKET.key(), "10")
-                        .option(CoreOptions.BUCKET_KEY.key(), "b,c")
-                        .build();
+                        .option(BUCKET.key(), "10");
+        if (pk) {
+            builder.primaryKey("a", "b", "c");
+        } else {
+            builder.option(BUCKET_KEY.key(), "b,c");
+        }
+        Schema schema = builder.build();
 
         Identifier tableId = identifier("test_composite_bucket_filter");
         catalog.createTable(tableId, schema, false);