fix

Author: JingsongLi

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowDataHashPartitionChannelComputer.java

@@ -20,9 +20,8 @@
 
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.schema.TableSchema;
-import org.apache.paimon.table.sink.AppendTableRowKeyExtractor;
 import org.apache.paimon.table.sink.ChannelComputer;
-import org.apache.paimon.table.sink.KeyAndBucketExtractor;
+import org.apache.paimon.table.sink.RowPartitionKeyExtractor;
 
 /** This is only for partitioned unaware-buckets Append only table. */
 public class RowDataHashPartitionChannelComputer implements ChannelComputer<InternalRow> {
@@ -32,7 +31,7 @@ public class RowDataHashPartitionChannelComputer implements ChannelComputer<Inte
     private final TableSchema schema;
 
     private transient int numChannels;
-    private transient KeyAndBucketExtractor<InternalRow> extractor;
+    private transient RowPartitionKeyExtractor extractor;
 
     public RowDataHashPartitionChannelComputer(TableSchema schema) {
         this.schema = schema;
@@ -41,13 +40,12 @@ public RowDataHashPartitionChannelComputer(TableSchema schema) {
     @Override
     public void setup(int numChannels) {
         this.numChannels = numChannels;
-        this.extractor = new AppendTableRowKeyExtractor(schema);
+        this.extractor = new RowPartitionKeyExtractor(schema);
     }
 
     @Override
     public int channel(InternalRow record) {
-        extractor.setRecord(record);
-        return ChannelComputer.select(extractor.partition(), 0, numChannels);
+        return ChannelComputer.select(extractor.partition(record), 0, numChannels);
     }
 
     @Override