[lake] Use DataLake's key encoding and bucket assigner for tables created in DataLake configured cluster (#362)

Author: luoyuxia

fluss-client/pom.xml

@@ -48,6 +48,7 @@
         </dependency>
 
         <!-- paimon bundle -->
+        <!-- remove paimon bundle in #408 -->
         <dependency>
             <groupId>org.apache.paimon</groupId>
             <artifactId>paimon-bundle</artifactId>

fluss-client/src/main/java/com/alibaba/fluss/client/lakehouse/LakeTableBucketAssigner.java

@@ -16,13 +16,17 @@
 
 package com.alibaba.fluss.client.lookup;
 
-import com.alibaba.fluss.client.lakehouse.LakeTableBucketAssigner;
 import com.alibaba.fluss.client.metadata.MetadataUpdater;
 import com.alibaba.fluss.client.table.getter.PartitionGetter;
+import com.alibaba.fluss.lakehouse.DataLakeFormat;
+import com.alibaba.fluss.lakehouse.LakeBucketAssigner;
+import com.alibaba.fluss.lakehouse.LakeBucketAssignerFactory;
+import com.alibaba.fluss.lakehouse.LakeKeyEncoderFactory;
 import com.alibaba.fluss.metadata.TableBucket;
 import com.alibaba.fluss.metadata.TableInfo;
 import com.alibaba.fluss.row.InternalRow;
 import com.alibaba.fluss.row.decode.RowDecoder;
+import com.alibaba.fluss.row.encode.CompactedKeyEncoder;
 import com.alibaba.fluss.row.encode.KeyEncoder;
 import com.alibaba.fluss.row.encode.ValueDecoder;
 import com.alibaba.fluss.types.DataType;
@@ -33,6 +37,7 @@
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 
@@ -56,7 +61,8 @@ class PrefixKeyLookuper implements Lookuper {
 
     private final int numBuckets;
 
-    private final LakeTableBucketAssigner lakeTableBucketAssigner;
+    // won't be null if the datalake format is set in the table
+    private @Nullable final LakeBucketAssigner lakeBucketAssigner;
 
     /**
      * a getter to extract partition from prefix lookup key row, null when it's not a partitioned.
@@ -80,15 +86,19 @@ public PrefixKeyLookuper(
         this.lookupClient = lookupClient;
         // the row type of the input lookup row
         RowType lookupRowType = tableInfo.getRowType().project(lookupColumnNames);
-        this.bucketKeyEncoder =
-                KeyEncoder.createKeyEncoder(lookupRowType, tableInfo.getBucketKeys());
-
-        if (tableInfo.getTableConfig().isDataLakeEnabled()) {
-            this.lakeTableBucketAssigner =
-                    new LakeTableBucketAssigner(
-                            lookupRowType, tableInfo.getBucketKeys(), numBuckets);
+        Optional<DataLakeFormat> optDataLakeFormat = tableInfo.getTableConfig().getDataLakeFormat();
+
+        if (optDataLakeFormat.isPresent()) {
+            DataLakeFormat dataLakeFormat = optDataLakeFormat.get();
+            this.bucketKeyEncoder =
+                    LakeKeyEncoderFactory.createKeyEncoder(
+                            dataLakeFormat, lookupRowType, tableInfo.getBucketKeys());
+            this.lakeBucketAssigner =
+                    LakeBucketAssignerFactory.createLakeBucketAssigner(dataLakeFormat, numBuckets);
         } else {
-            this.lakeTableBucketAssigner = null;
+            this.bucketKeyEncoder =
+                    CompactedKeyEncoder.createKeyEncoder(lookupRowType, tableInfo.getBucketKeys());
+            this.lakeBucketAssigner = null;
         }
 
         this.partitionGetter =
@@ -152,14 +162,8 @@ private void validatePrefixLookup(TableInfo tableInfo, List<String> lookupColumn
 
     @Override
     public CompletableFuture<LookupResult> lookup(InternalRow prefixKey) {
-        byte[] bucketKeyBytes = bucketKeyEncoder.encode(prefixKey);
-        int bucketId =
-                getBucketId(
-                        bucketKeyBytes,
-                        prefixKey,
-                        lakeTableBucketAssigner,
-                        numBuckets,
-                        metadataUpdater);
+        byte[] bucketKeyBytes = bucketKeyEncoder.encodeKey(prefixKey);
+        int bucketId = getBucketId(bucketKeyBytes, lakeBucketAssigner, numBuckets);
 
         Long partitionId = null;
         if (partitionGetter != null) {