[lake] The bucket of PaimonSplit should be -1 for bucket unwared paimon table (#1732)

Author: caozhen1937

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonSplit.java

@@ -32,12 +32,19 @@ public class PaimonSplit implements LakeSplit {
 
     private final DataSplit dataSplit;
 
-    public PaimonSplit(DataSplit dataSplit) {
+    private final boolean isBucketUnAware;
+
+    public PaimonSplit(DataSplit dataSplit, boolean isBucketUnAware) {
         this.dataSplit = dataSplit;
+        this.isBucketUnAware = isBucketUnAware;
     }
 
     @Override
     public int bucket() {
+        if (isBucketUnAware) {
+            // bucket-unaware table returns -1
+            return -1;
+        }
         return dataSplit.bucket();
     }
 
@@ -61,4 +68,8 @@ public List<String> partition() {
     public DataSplit dataSplit() {
         return dataSplit;
     }
+
+    public boolean isBucketUnAware() {
+        return isBucketUnAware;
+    }
 }

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonSplitPlanner.java

@@ -28,6 +28,7 @@
 import org.apache.paimon.catalog.CatalogFactory;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.table.BucketMode;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.source.DataSplit;
 import org.apache.paimon.table.source.InnerTableScan;
@@ -67,12 +68,14 @@ public List<PaimonSplit> plan() {
             try (Catalog catalog = getCatalog()) {
                 FileStoreTable fileStoreTable = getTable(catalog, tablePath, snapshotId);
                 InnerTableScan tableScan = fileStoreTable.newScan();
+                boolean isBucketUnAware = fileStoreTable.bucketMode() == BucketMode.BUCKET_UNAWARE;
+
                 if (predicate != null) {
                     tableScan = tableScan.withFilter(predicate);
                 }
                 for (Split split : tableScan.plan().splits()) {
                     DataSplit dataSplit = (DataSplit) split;
-                    splits.add(new PaimonSplit(dataSplit));
+                    splits.add(new PaimonSplit(dataSplit, isBucketUnAware));
                 }
             }
             return splits;

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonSplitSerializer.java

@@ -26,14 +26,17 @@
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
 import java.io.IOException;
 
 /** Serializer for paimon split. */
 public class PaimonSplitSerializer implements SimpleVersionedSerializer<PaimonSplit> {
 
+    private static final int VERSION_1 = 1;
+
     @Override
     public int getVersion() {
-        return 1;
+        return VERSION_1;
     }
 
     @Override
@@ -42,6 +45,7 @@ public byte[] serialize(PaimonSplit paimonSplit) throws IOException {
         DataOutputViewStreamWrapper view = new DataOutputViewStreamWrapper(out);
         DataSplit dataSplit = paimonSplit.dataSplit();
         InstantiationUtil.serializeObject(view, dataSplit);
+        view.writeBoolean(paimonSplit.isBucketUnAware());
         return out.toByteArray();
     }
 
@@ -51,9 +55,16 @@ public PaimonSplit deserialize(int version, byte[] serialized) throws IOExceptio
         DataSplit dataSplit;
         try {
             dataSplit = InstantiationUtil.deserializeObject(in, getClass().getClassLoader());
+
+            if (version == VERSION_1) {
+                DataInputStream dis = new DataInputStream(in);
+                boolean isBucketUnAware = dis.readBoolean();
+                return new PaimonSplit(dataSplit, isBucketUnAware);
+            } else {
+                throw new IOException("Unsupported PaimonSplit serialization version: " + version);
+            }
         } catch (ClassNotFoundException e) {
-            throw new IOException(e);
+            throw new IOException("Failed to deserialize PaimonSplit", e);
         }
-        return new PaimonSplit(dataSplit);
     }
 }

fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/PaimonSplitPlannerTest.java

@@ -27,6 +27,7 @@
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.data.BinaryString;
 import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.Table;
@@ -78,4 +79,35 @@ void testPlan() throws Exception {
                                 .map(PaimonSplit::dataSplit)
                                 .collect(Collectors.toList()));
     }
+
+    @Test
+    void testPlannerCreatesCorrectSplitsForLogTableWithoutBucketKey() throws Exception {
+        // test log table without bucket key
+        String tableName = "planner_log_table_without_bucket_key";
+        TablePath tablePath = TablePath.of(DEFAULT_DB, tableName);
+
+        Schema.Builder builder =
+                Schema.newBuilder()
+                        .column("id", DataTypes.INT())
+                        .column("message", DataTypes.STRING());
+        createTable(tablePath, builder.build());
+        Table table = getTable(tablePath);
+
+        List<InternalRow> records =
+                Arrays.asList(
+                        GenericRow.of(1, BinaryString.fromString("msg1")),
+                        GenericRow.of(2, BinaryString.fromString("msg2")));
+        writeRecord(tablePath, records);
+
+        Snapshot snapshot = table.latestSnapshot().get();
+
+        LakeSource<PaimonSplit> lakeSource = lakeStorage.createLakeSource(tablePath);
+        List<PaimonSplit> paimonSplits = lakeSource.createPlanner(snapshot::id).plan();
+
+        assertThat(paimonSplits).isNotEmpty();
+        for (PaimonSplit split : paimonSplits) {
+            assertThat(split.isBucketUnAware()).isTrue();
+            assertThat(split.bucket()).isEqualTo(-1);
+        }
+    }
 }

fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/PaimonSplitSerializerTest.java

@@ -72,6 +72,7 @@ void testSerializeAndDeserialize() throws Exception {
         PaimonSplit deserialized = serializer.deserialize(serializer.getVersion(), serialized);
 
         assertThat(deserialized.dataSplit()).isEqualTo(originalPaimonSplit.dataSplit());
+        assertThat(deserialized.isBucketUnAware()).isEqualTo(originalPaimonSplit.isBucketUnAware());
     }
 
     @Test