Merge branch 'apache:main' into lance-writer

Author: xx789633

.github/workflows/ci-template.yaml

@@ -48,7 +48,7 @@ jobs:
           distribution: 'temurin'
       - name: Build
         run: |
-          mvn -T 1C -B clean install -DskipTests
+          mvn -T 1C -B clean install -DskipTests ${{ inputs.maven-parameters }}
       - name: Test
         timeout-minutes: 60
         run: |

fluss-client/src/main/java/com/alibaba/fluss/client/metadata/LakeSnapshot.java

@@ -20,6 +20,7 @@
 import com.alibaba.fluss.annotation.PublicEvolving;
 import com.alibaba.fluss.metadata.TableBucket;
 
+import java.util.Collections;
 import java.util.Map;
 
 /**
@@ -36,17 +37,29 @@ public class LakeSnapshot {
     // the specific log offset of the snapshot
     private final Map<TableBucket, Long> tableBucketsOffset;
 
-    public LakeSnapshot(long snapshotId, Map<TableBucket, Long> tableBucketsOffset) {
+    // the partition name by partition id of this lake snapshot if
+    // is a partitioned table, empty if not a partitioned table
+    private final Map<Long, String> partitionNameById;
+
+    public LakeSnapshot(
+            long snapshotId,
+            Map<TableBucket, Long> tableBucketsOffset,
+            Map<Long, String> partitionNameById) {
         this.snapshotId = snapshotId;
         this.tableBucketsOffset = tableBucketsOffset;
+        this.partitionNameById = partitionNameById;
     }
 
     public long getSnapshotId() {
         return snapshotId;
     }
 
     public Map<TableBucket, Long> getTableBucketsOffset() {
-        return tableBucketsOffset;
+        return Collections.unmodifiableMap(tableBucketsOffset);
+    }
+
+    public Map<Long, String> getPartitionNameById() {
+        return Collections.unmodifiableMap(partitionNameById);
     }
 
     @Override
@@ -56,6 +69,8 @@ public String toString() {
                 + snapshotId
                 + ", tableBucketsOffset="
                 + tableBucketsOffset
+                + ", partitionNameById="
+                + partitionNameById
                 + '}';
     }
 }

fluss-client/src/main/java/com/alibaba/fluss/client/utils/ClientRpcMessageUtils.java

@@ -202,16 +202,20 @@ public static LakeSnapshot toLakeTableSnapshotInfo(GetLatestLakeSnapshotResponse
         long snapshotId = response.getSnapshotId();
         Map<TableBucket, Long> tableBucketsOffset =
                 new HashMap<>(response.getBucketSnapshotsCount());
+        Map<Long, String> partitionNameById = new HashMap<>();
         for (PbLakeSnapshotForBucket pbLakeSnapshotForBucket : response.getBucketSnapshotsList()) {
             Long partitionId =
                     pbLakeSnapshotForBucket.hasPartitionId()
                             ? pbLakeSnapshotForBucket.getPartitionId()
                             : null;
             TableBucket tableBucket =
                     new TableBucket(tableId, partitionId, pbLakeSnapshotForBucket.getBucketId());
+            if (partitionId != null && pbLakeSnapshotForBucket.hasPartitionName()) {
+                partitionNameById.put(partitionId, pbLakeSnapshotForBucket.getPartitionName());
+            }
             tableBucketsOffset.put(tableBucket, pbLakeSnapshotForBucket.getLogOffset());
         }
-        return new LakeSnapshot(snapshotId, tableBucketsOffset);
+        return new LakeSnapshot(snapshotId, tableBucketsOffset, partitionNameById);
     }
 
     public static List<FsPathAndFileName> toFsPathAndFileName(

fluss-common/src/main/java/com/alibaba/fluss/lake/committer/BucketOffset.java

@@ -31,17 +31,17 @@ public class BucketOffset implements Serializable {
     private final long logOffset;
     private final int bucket;
     private final @Nullable Long partitionId;
-    private final @Nullable String partitionName;
+    private final @Nullable String partitionQualifiedName;
 
     public BucketOffset(
             long logOffset,
             int bucket,
             @Nullable Long partitionId,
-            @Nullable String partitionName) {
+            @Nullable String partitionQualifiedName) {
         this.logOffset = logOffset;
         this.bucket = bucket;
         this.partitionId = partitionId;
-        this.partitionName = partitionName;
+        this.partitionQualifiedName = partitionQualifiedName;
     }
 
     public long getLogOffset() {
@@ -58,8 +58,8 @@ public Long getPartitionId() {
     }
 
     @Nullable
-    public String getPartitionName() {
-        return partitionName;
+    public String getPartitionQualifiedName() {
+        return partitionQualifiedName;
     }
 
     @Override
@@ -74,6 +74,6 @@ public boolean equals(Object o) {
         return bucket == that.bucket
                 && logOffset == that.logOffset
                 && Objects.equals(partitionId, that.partitionId)
-                && Objects.equals(partitionName, that.partitionName);
+                && Objects.equals(partitionQualifiedName, that.partitionQualifiedName);
     }
 }

fluss-common/src/main/java/com/alibaba/fluss/lake/committer/CommittedLakeSnapshot.java

@@ -34,6 +34,10 @@ public class CommittedLakeSnapshot {
     // partition bucket
     private final Map<Tuple2<Long, Integer>, Long> logEndOffsets = new HashMap<>();
 
+    // partition id -> partition name, will be empty if is not a partitioned table
+    // the partition name is a qualified name in the format: key1=value1/key2=value2
+    private final Map<Long, String> qualifiedPartitionNameById = new HashMap<>();
+
     public CommittedLakeSnapshot(long lakeSnapshotId) {
         this.lakeSnapshotId = lakeSnapshotId;
     }
@@ -46,27 +50,34 @@ public void addBucket(int bucketId, long offset) {
         logEndOffsets.put(Tuple2.of(null, bucketId), offset);
     }
 
-    public void addPartitionBucket(Long partitionId, int bucketId, long offset) {
+    public void addPartitionBucket(
+            Long partitionId, String partitionQualifiedName, int bucketId, long offset) {
         logEndOffsets.put(Tuple2.of(partitionId, bucketId), offset);
+        qualifiedPartitionNameById.put(partitionId, partitionQualifiedName);
     }
 
     public Map<Tuple2<Long, Integer>, Long> getLogEndOffsets() {
         return logEndOffsets;
     }
 
+    public Map<Long, String> getQualifiedPartitionNameById() {
+        return qualifiedPartitionNameById;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (o == null || getClass() != o.getClass()) {
             return false;
         }
         CommittedLakeSnapshot that = (CommittedLakeSnapshot) o;
         return lakeSnapshotId == that.lakeSnapshotId
-                && Objects.equals(logEndOffsets, that.logEndOffsets);
+                && Objects.equals(logEndOffsets, that.logEndOffsets)
+                && Objects.equals(qualifiedPartitionNameById, that.qualifiedPartitionNameById);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(lakeSnapshotId, logEndOffsets);
+        return Objects.hash(lakeSnapshotId, logEndOffsets, qualifiedPartitionNameById);
     }
 
     @Override
@@ -76,6 +87,8 @@ public String toString() {
                 + lakeSnapshotId
                 + ", logEndOffsets="
                 + logEndOffsets
+                + ", partitionNameById="
+                + qualifiedPartitionNameById
                 + '}';
     }
 }

fluss-common/src/main/java/com/alibaba/fluss/lake/writer/WriterInitContext.java

@@ -24,6 +24,8 @@
 
 import javax.annotation.Nullable;
 
+import java.util.Map;
+
 /**
  * The WriterInitContext interface provides the context needed to create a LakeWriter. It includes
  * methods to obtain the table path, table bucket, and an optional partition.
@@ -61,4 +63,11 @@ public interface WriterInitContext {
      * @return the table schema
      */
     Schema schema();
+
+    /**
+     * Returns the table custom properties.
+     *
+     * @return the table custom properties
+     */
+    Map<String, String> customProperties();
 }

fluss-common/src/main/java/com/alibaba/fluss/metadata/ResolvedPartitionSpec.java

@@ -125,6 +125,25 @@ public String getPartitionQualifiedName() {
         return sb.toString();
     }
 
+    public static ResolvedPartitionSpec fromPartitionQualifiedName(String qualifiedPartitionName) {
+        // convert from qualified name to ResolvedPartitionSpec
+        List<String> keys = new ArrayList<>();
+        List<String> values = new ArrayList<>();
+
+        String[] keyValuePairs = qualifiedPartitionName.split("/");
+
+        for (String pair : keyValuePairs) {
+            String[] keyValue = pair.split("=", 2);
+            if (keyValue.length != 2) {
+                throw new IllegalArgumentException(
+                        "Invalid partition name format. Expected key=value, got: " + pair);
+            }
+            keys.add(keyValue[0]);
+            values.add(keyValue[1]);
+        }
+        return new ResolvedPartitionSpec(keys, values);
+    }
+
     @Override
     public boolean equals(Object o) {
         if (o == null || getClass() != o.getClass()) {

fluss-common/src/main/java/com/alibaba/fluss/metrics/groups/AbstractMetricGroup.java

@@ -163,9 +163,13 @@ String getLogicalScope(CharacterFilter filter, char delimiter, int reporterIndex
 
     protected String createLogicalScope(CharacterFilter filter, char delimiter) {
         final String groupName = getGroupName(filter);
-        return parent == null
-                ? groupName
-                : parent.getLogicalScope(filter, delimiter) + delimiter + groupName;
+        if (parent == null) {
+            return groupName;
+        }
+        if (groupName == null || groupName.isEmpty()) {
+            return parent.getLogicalScope(filter, delimiter);
+        }
+        return parent.getLogicalScope(filter, delimiter) + delimiter + groupName;
     }
 
     /** Return the parent of the metric group. */

fluss-common/src/main/java/com/alibaba/fluss/utils/json/BucketOffsetJsonSerde.java

@@ -60,8 +60,8 @@ public void serialize(BucketOffset bucketOffset, JsonGenerator generator) throws
         generator.writeNumberField(BUCKET_ID, bucketOffset.getBucket());
 
         // serialize partition name
-        if (bucketOffset.getPartitionName() != null) {
-            generator.writeStringField(PARTITION_NAME, bucketOffset.getPartitionName());
+        if (bucketOffset.getPartitionQualifiedName() != null) {
+            generator.writeStringField(PARTITION_NAME, bucketOffset.getPartitionQualifiedName());
         }
 
         // serialize bucket offset

fluss-common/src/test/java/com/alibaba/fluss/metadata/ResolvedPartitionSpecTest.java

@@ -38,6 +38,11 @@ void testResolvedPartitionSpec() {
                         new PartitionSpec(Collections.singletonMap("a", "1")));
         assertThat(resolvedPartitionSpec.getPartitionName()).isEqualTo("1");
         assertThat(resolvedPartitionSpec.getPartitionQualifiedName()).isEqualTo("a=1");
+        assertThat(
+                        ResolvedPartitionSpec.fromPartitionQualifiedName(
+                                resolvedPartitionSpec.getPartitionQualifiedName()))
+                .isEqualTo(resolvedPartitionSpec);
+
         assertThat(resolvedPartitionSpec.getPartitionKeys())
                 .isEqualTo(Collections.singletonList("a"));
         assertThat(resolvedPartitionSpec.getPartitionValues())
@@ -52,5 +57,9 @@ void testResolvedPartitionSpec() {
         assertThat(resolvedPartitionSpec.getPartitionQualifiedName()).isEqualTo("a=1/b=2");
         assertThat(resolvedPartitionSpec.getPartitionKeys()).isEqualTo(Arrays.asList("a", "b"));
         assertThat(resolvedPartitionSpec.getPartitionValues()).isEqualTo(Arrays.asList("1", "2"));
+        assertThat(
+                        ResolvedPartitionSpec.fromPartitionQualifiedName(
+                                resolvedPartitionSpec.getPartitionQualifiedName()))
+                .isEqualTo(resolvedPartitionSpec);
     }
 }