[lake/paimon] Support tiering multi partition paimon table (#1024)

MehulBatra · luoyuxia · web-flow · commit 20b243efad11 · 2025-06-10T11:40:27.000+08:00
---------

Co-authored-by: luoyuxia &lt;luoyuxia@alumni.sjtu.edu.cn&gt;
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/com/alibaba/fluss/lake/paimon/tiering/FlussRecordAsPaimonRow.java b/fluss-lake/fluss-lake-paimon/src/main/java/com/alibaba/fluss/lake/paimon/tiering/FlussRecordAsPaimonRow.java
@@ -50,7 +50,8 @@ public void setFlussRecord(LogRecord logRecord) {
     @Override
     public int getFieldCount() {
         return
-        // three system fields: bucket, offset, timestamp
+        //  business (including partitions) + system (three system fields: bucket, offset,
+        // timestamp)
         originRowFieldCount + 3;
     }
 
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/com/alibaba/fluss/lake/paimon/tiering/PaimonLakeWriter.java b/fluss-lake/fluss-lake-paimon/src/main/java/com/alibaba/fluss/lake/paimon/tiering/PaimonLakeWriter.java
@@ -28,6 +28,7 @@
 import org.apache.paimon.table.sink.CommitMessage;
 
 import java.io.IOException;
+import java.util.List;
 
 import static com.alibaba.fluss.lake.paimon.utils.PaimonConversions.toPaimon;
 
@@ -43,16 +44,20 @@ public PaimonLakeWriter(
         this.paimonCatalog = paimonCatalogProvider.get();
         FileStoreTable fileStoreTable = getTable(writerInitContext.tablePath());
 
+        List<String> partitionKeys = fileStoreTable.partitionKeys();
+
         this.recordWriter =
                 fileStoreTable.primaryKeys().isEmpty()
                         ? new AppendOnlyWriter(
                                 fileStoreTable,
                                 writerInitContext.tableBucket(),
-                                writerInitContext.partition())
+                                writerInitContext.partition(),
+                                partitionKeys)
                         : new MergeTreeWriter(
                                 fileStoreTable,
                                 writerInitContext.tableBucket(),
-                                writerInitContext.partition());
+                                writerInitContext.partition(),
+                                partitionKeys);
     }
 
     @Override
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/com/alibaba/fluss/lake/paimon/tiering/RecordWriter.java b/fluss-lake/fluss-lake-paimon/src/main/java/com/alibaba/fluss/lake/paimon/tiering/RecordWriter.java
@@ -27,7 +27,7 @@
 
 import java.util.List;
 
-import static com.alibaba.fluss.lake.paimon.utils.PaimonConversions.toPaimonBinaryRow;
+import static com.alibaba.fluss.lake.paimon.utils.PaimonConversions.toPaimonPartitionBinaryRow;
 import static com.alibaba.fluss.utils.Preconditions.checkState;
 
 /** A base interface to write {@link LogRecord} to Paimon. */
@@ -39,10 +39,13 @@ public abstract class RecordWriter<T> implements AutoCloseable {
     protected final FlussRecordAsPaimonRow flussRecordAsPaimonRow;
 
     public RecordWriter(
-            TableWriteImpl<T> tableWrite, TableBucket tableBucket, @Nullable String partition) {
+            TableWriteImpl<T> tableWrite,
+            TableBucket tableBucket,
+            @Nullable String partition,
+            List<String> partitionKeys) {
         this.tableWrite = tableWrite;
         this.bucket = tableBucket.getBucket();
-        this.partition = toPaimonBinaryRow(partition);
+        this.partition = toPaimonPartitionBinaryRow(partitionKeys, partition);
         this.flussRecordAsPaimonRow = new FlussRecordAsPaimonRow(tableBucket.getBucket());
     }
 
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/com/alibaba/fluss/lake/paimon/tiering/append/AppendOnlyWriter.java b/fluss-lake/fluss-lake-paimon/src/main/java/com/alibaba/fluss/lake/paimon/tiering/append/AppendOnlyWriter.java
@@ -26,20 +26,26 @@
 
 import javax.annotation.Nullable;
 
+import java.util.List;
+
 import static com.alibaba.fluss.lake.paimon.tiering.PaimonLakeTieringFactory.FLUSS_LAKE_TIERING_COMMIT_USER;
 
 /** A {@link RecordWriter} to write to Paimon's append-only table. */
 public class AppendOnlyWriter extends RecordWriter<InternalRow> {
 
     public AppendOnlyWriter(
-            FileStoreTable fileStoreTable, TableBucket tableBucket, @Nullable String partition) {
+            FileStoreTable fileStoreTable,
+            TableBucket tableBucket,
+            @Nullable String partition,
+            List<String> partitionKeys) {
         //noinspection unchecked
         super(
                 (TableWriteImpl<InternalRow>)
                         // todo: set ioManager to support write-buffer-spillable
                         fileStoreTable.newWrite(FLUSS_LAKE_TIERING_COMMIT_USER),
                 tableBucket,
-                partition);
+                partition,
+                partitionKeys); // Pass to parent
     }
 
     @Override
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/com/alibaba/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java b/fluss-lake/fluss-lake-paimon/src/main/java/com/alibaba/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java
@@ -27,6 +27,8 @@
 
 import javax.annotation.Nullable;
 
+import java.util.List;
+
 import static com.alibaba.fluss.lake.paimon.tiering.PaimonLakeTieringFactory.FLUSS_LAKE_TIERING_COMMIT_USER;
 import static com.alibaba.fluss.lake.paimon.utils.PaimonConversions.toRowKind;
 
@@ -38,8 +40,11 @@ public class MergeTreeWriter extends RecordWriter<KeyValue> {
     private final RowKeyExtractor rowKeyExtractor;
 
     public MergeTreeWriter(
-            FileStoreTable fileStoreTable, TableBucket tableBucket, @Nullable String partition) {
-        super(createTableWrite(fileStoreTable), tableBucket, partition);
+            FileStoreTable fileStoreTable,
+            TableBucket tableBucket,
+            @Nullable String partition,
+            List<String> partitionKeys) {
+        super(createTableWrite(fileStoreTable), tableBucket, partition, partitionKeys);
         this.rowKeyExtractor = fileStoreTable.createRowKeyExtractor();
     }
 
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/com/alibaba/fluss/lake/paimon/utils/PaimonConversions.java b/fluss-lake/fluss-lake-paimon/src/main/java/com/alibaba/fluss/lake/paimon/utils/PaimonConversions.java
@@ -16,6 +16,7 @@
 
 package com.alibaba.fluss.lake.paimon.utils;
 
+import com.alibaba.fluss.metadata.ResolvedPartitionSpec;
 import com.alibaba.fluss.metadata.TablePath;
 import com.alibaba.fluss.record.ChangeType;
 
@@ -27,6 +28,8 @@
 
 import javax.annotation.Nullable;
 
+import java.util.List;
+
 /** Utils for conversion between Paimon and Fluss. */
 public class PaimonConversions {
 
@@ -50,14 +53,28 @@ public static Identifier toPaimon(TablePath tablePath) {
         return Identifier.create(tablePath.getDatabaseName(), tablePath.getTableName());
     }
 
-    public static BinaryRow toPaimonBinaryRow(@Nullable String value) {
-        if (value == null) {
+    public static BinaryRow toPaimonPartitionBinaryRow(
+            List<String> partitionKeys, @Nullable String partitionName) {
+        if (partitionName == null || partitionKeys.isEmpty()) {
             return BinaryRow.EMPTY_ROW;
         }
-        BinaryRow binaryRow = new BinaryRow(1);
-        BinaryRowWriter writer = new BinaryRowWriter(binaryRow);
-        writer.writeString(0, BinaryString.fromString(value));
+
+        //  Fluss's existing utility
+        ResolvedPartitionSpec resolvedPartitionSpec =
+                ResolvedPartitionSpec.fromPartitionName(partitionKeys, partitionName);
+
+        BinaryRow partitionBinaryRow = new BinaryRow(partitionKeys.size());
+        BinaryRowWriter writer = new BinaryRowWriter(partitionBinaryRow);
+
+        List<String> partitionValues = resolvedPartitionSpec.getPartitionValues();
+        for (int i = 0; i < partitionKeys.size(); i++) {
+            // Todo Currently, partition column must be String datatype, so we can always use
+            // `BinaryString.fromString` to convert to Paimon's data structure. Revisit here when
+            // #489 is finished.
+            writer.writeString(i, BinaryString.fromString(partitionValues.get(i)));
+        }
+
         writer.complete();
-        return binaryRow;
+        return partitionBinaryRow;
     }
 }
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/com/alibaba/fluss/lake/paimon/tiering/FlussRecordAsPaimonRowTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/com/alibaba/fluss/lake/paimon/tiering/FlussRecordAsPaimonRowTest.java
@@ -83,7 +83,8 @@ void testLogTableRecordAllTypes() {
                 .isEqualTo(1698235273182L);
         assertThat(flussRecordAsPaimonRow.getBinary(12)).isEqualTo(new byte[] {1, 2, 3, 4});
         assertThat(flussRecordAsPaimonRow.isNullAt(13)).isTrue();
-        // verify FlussRecordAsPaimonRow system columns
+
+        // verify FlussRecordAsPaimonRow system columns (no partition fields, so indices stay same)
         assertThat(flussRecordAsPaimonRow.getInt(14)).isEqualTo(bucket);
         assertThat(flussRecordAsPaimonRow.getLong(15)).isEqualTo(logOffset);
         assertThat(flussRecordAsPaimonRow.getLong(16)).isEqualTo(timeStamp);
@@ -92,11 +93,7 @@ void testLogTableRecordAllTypes() {
         assertThat(flussRecordAsPaimonRow.getRowKind()).isEqualTo(RowKind.INSERT);
 
         assertThat(flussRecordAsPaimonRow.getFieldCount())
-                .isEqualTo(
-                        14
-                                +
-                                // 3 is for system columns
-                                3);
+                .isEqualTo(14 + 3); // business  + system = 14 + 0 + 3 = 17
     }
 
     @Test
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/com/alibaba/fluss/lake/paimon/tiering/PaimonTieringTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/com/alibaba/fluss/lake/paimon/tiering/PaimonTieringTest.java

Original file line number	Diff line number	Diff line change
`@@ -50,7 +50,8 @@ public void setFlussRecord(LogRecord logRecord) {`
`50`	`50`	`@Override`
`51`	`51`	`public int getFieldCount() {`
`52`	`52`	`return`
`53`		`- // three system fields: bucket, offset, timestamp`
	`53`	`+ // business (including partitions) + system (three system fields: bucket, offset,`
	`54`	`+ // timestamp)`
`54`	`55`	`originRowFieldCount + 3;`
`55`	`56`	`}`
`56`	`57`