more

Author: LiebingYu

fluss-common/src/main/java/org/apache/fluss/metadata/PartitionInfo.java

@@ -18,8 +18,12 @@
 package org.apache.fluss.metadata;
 
 import org.apache.fluss.annotation.PublicEvolving;
+import org.apache.fluss.fs.FsPath;
+
+import javax.annotation.Nullable;
 
 import java.util.Objects;
+import java.util.Optional;
 
 /**
  * Information of a partition metadata, includes the partition's name and the partition id that
@@ -31,10 +35,17 @@
 public class PartitionInfo {
     private final long partitionId;
     private final ResolvedPartitionSpec partitionSpec;
+    private final @Nullable FsPath remoteDataDir;
 
     public PartitionInfo(long partitionId, ResolvedPartitionSpec partitionSpec) {
+        this(partitionId, partitionSpec, null);
+    }
+
+    public PartitionInfo(
+            long partitionId, ResolvedPartitionSpec partitionSpec, @Nullable FsPath remoteDataDir) {
         this.partitionId = partitionId;
         this.partitionSpec = partitionSpec;
+        this.remoteDataDir = remoteDataDir;
     }
 
     /** Get the partition id. The id is globally unique in the Fluss cluster. */
@@ -58,6 +69,10 @@ public PartitionSpec getPartitionSpec() {
         return partitionSpec.toPartitionSpec();
     }
 
+    public Optional<FsPath> getRemoteDataDir() {
+        return Optional.ofNullable(remoteDataDir);
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -67,16 +82,25 @@ public boolean equals(Object o) {
             return false;
         }
         PartitionInfo that = (PartitionInfo) o;
-        return partitionId == that.partitionId && Objects.equals(partitionSpec, that.partitionSpec);
+        return partitionId == that.partitionId
+                && Objects.equals(partitionSpec, that.partitionSpec)
+                && Objects.equals(remoteDataDir, that.remoteDataDir);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(partitionId, partitionSpec);
+        return Objects.hash(partitionId, partitionSpec, remoteDataDir);
     }
 
     @Override
     public String toString() {
-        return "Partition{name='" + getPartitionName() + '\'' + ", id=" + partitionId + '}';
+        return "Partition{name='"
+                + getPartitionName()
+                + '\''
+                + ", id="
+                + partitionId
+                + ", remoteDataDir="
+                + remoteDataDir
+                + '}';
     }
 }

fluss-common/src/main/java/org/apache/fluss/metadata/TableInfo.java

@@ -20,6 +20,7 @@
 import org.apache.fluss.annotation.PublicEvolving;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.config.TableConfig;
+import org.apache.fluss.fs.FsPath;
 import org.apache.fluss.types.RowType;
 
 import javax.annotation.Nullable;
@@ -59,6 +60,7 @@ public final class TableInfo {
     private final Configuration properties;
     private final TableConfig tableConfig;
     private final Configuration customProperties;
+    private final @Nullable FsPath remoteDataDir;
     private final @Nullable String comment;
 
     private final long createdTime;
@@ -74,6 +76,7 @@ public TableInfo(
             int numBuckets,
             Configuration properties,
             Configuration customProperties,
+            @Nullable FsPath remoteDataDir,
             @Nullable String comment,
             long createdTime,
             long modifiedTime) {
@@ -90,6 +93,7 @@ public TableInfo(
         this.properties = properties;
         this.tableConfig = new TableConfig(properties);
         this.customProperties = customProperties;
+        this.remoteDataDir = remoteDataDir;
         this.comment = comment;
         this.createdTime = createdTime;
         this.modifiedTime = modifiedTime;
@@ -263,6 +267,11 @@ public Configuration getCustomProperties() {
         return customProperties;
     }
 
+    /** Returns the remote data directory of the table. */
+    public Optional<FsPath> getRemoteDataDir() {
+        return Optional.ofNullable(remoteDataDir);
+    }
+
     /** Returns the comment/description of the table. */
     public Optional<String> getComment() {
         return Optional.ofNullable(comment);
@@ -335,6 +344,7 @@ public static TableInfo of(
                 numBuckets,
                 Configuration.fromMap(tableDescriptor.getProperties()),
                 Configuration.fromMap(tableDescriptor.getCustomProperties()),
+                null,
                 tableDescriptor.getComment().orElse(null),
                 createdTime,
                 modifiedTime);
@@ -358,6 +368,7 @@ public boolean equals(Object o) {
                 && Objects.equals(partitionKeys, that.partitionKeys)
                 && Objects.equals(properties, that.properties)
                 && Objects.equals(customProperties, that.customProperties)
+                && Objects.equals(remoteDataDir, that.remoteDataDir)
                 && Objects.equals(comment, that.comment);
     }
 
@@ -376,6 +387,7 @@ public int hashCode() {
                 numBuckets,
                 properties,
                 customProperties,
+                remoteDataDir,
                 comment);
     }
 
@@ -402,6 +414,8 @@ public String toString() {
                 + properties
                 + ", customProperties="
                 + customProperties
+                + ", remoteDataDir="
+                + remoteDataDir
                 + ", comment='"
                 + comment
                 + '\''

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java

@@ -19,6 +19,7 @@
 
 import org.apache.fluss.annotation.VisibleForTesting;
 import org.apache.fluss.cluster.rebalance.ServerTag;
+import org.apache.fluss.metadata.PartitionInfo;
 import org.apache.fluss.metadata.PhysicalTablePath;
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TableBucketReplica;
@@ -84,6 +85,7 @@ public class CoordinatorContext {
     // a map from partition_id -> physicalTablePath
     private final Map<Long, PhysicalTablePath> pathByPartitionId = new HashMap<>();
     private final Map<PhysicalTablePath, Long> partitionIdByPath = new HashMap<>();
+    private final Map<Long, PartitionInfo> partitionInfoById = new HashMap<>();
 
     // a map from table_id to the table path
     private final Map<Long, TablePath> tablePathById = new HashMap<>();
@@ -252,9 +254,11 @@ public void putTableInfo(TableInfo tableInfo) {
         this.tableInfoById.put(tableInfo.getTableId(), tableInfo);
     }
 
-    public void putPartition(long partitionId, PhysicalTablePath physicalTablePath) {
+    public void putPartition(
+            long partitionId, PhysicalTablePath physicalTablePath, PartitionInfo partitionInfo) {
         this.pathByPartitionId.put(partitionId, physicalTablePath);
         this.partitionIdByPath.put(physicalTablePath, partitionId);
+        this.partitionInfoById.put(partitionId, partitionInfo);
     }
 
     public TableInfo getTableInfoById(long tableId) {
@@ -294,6 +298,10 @@ public Optional<Long> getPartitionId(PhysicalTablePath physicalTablePath) {
         return Optional.ofNullable(partitionIdByPath.get(physicalTablePath));
     }
 
+    public PartitionInfo getPartitionInfoById(long partitionId) {
+        return partitionInfoById.get(partitionId);
+    }
+
     public Map<Integer, List<Integer>> getTableAssignment(long tableId) {
         return tableAssignments.getOrDefault(tableId, Collections.emptyMap());
     }
@@ -427,6 +435,11 @@ public Set<TableBucketReplica> getAllReplicasForPartition(long tableId, long par
         return allReplicas;
     }
 
+    public boolean hasPartitionsToDelete(long tableId) {
+        return partitionsToBeDeleted.stream()
+                .anyMatch(partition -> partition.getTableId() == tableId);
+    }
+
     /**
      * Pick up the replicas that should retry delete and replicas that considered as success delete.
      *

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java

@@ -40,7 +40,9 @@
 import org.apache.fluss.exception.TabletServerNotAvailableException;
 import org.apache.fluss.exception.UnknownServerException;
 import org.apache.fluss.exception.UnknownTableOrBucketException;
+import org.apache.fluss.metadata.PartitionInfo;
 import org.apache.fluss.metadata.PhysicalTablePath;
+import org.apache.fluss.metadata.ResolvedPartitionSpec;
 import org.apache.fluss.metadata.SchemaInfo;
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TableBucketReplica;
@@ -108,6 +110,7 @@
 import org.apache.fluss.server.zk.data.BucketAssignment;
 import org.apache.fluss.server.zk.data.LeaderAndIsr;
 import org.apache.fluss.server.zk.data.PartitionAssignment;
+import org.apache.fluss.server.zk.data.PartitionRegistration;
 import org.apache.fluss.server.zk.data.RebalanceTask;
 import org.apache.fluss.server.zk.data.RemoteLogManifestHandle;
 import org.apache.fluss.server.zk.data.ServerTags;
@@ -393,8 +396,8 @@ private void initCoordinatorContext() throws Exception {
                         .filter(entry -> entry.getValue().isPartitioned())
                         .map(Map.Entry::getKey)
                         .collect(Collectors.toList());
-        Map<TablePath, Map<String, Long>> tablePathMap =
-                zooKeeperClient.getPartitionNameAndIdsForTables(partitionedTablePathList);
+        Map<TablePath, Map<String, PartitionRegistration>> tablePathMap =
+                zooKeeperClient.getPartitionNameAndRegistrationsForTables(partitionedTablePathList);
         for (TablePath tablePath : tablePathSet) {
             TableInfo tableInfo = tablePath2TableInfoMap.get(tablePath);
             coordinatorContext.putTablePath(tableInfo.getTableId(), tablePath);
@@ -405,13 +408,22 @@ private void initCoordinatorContext() throws Exception {
                 lakeTables.add(Tuple2.of(tableInfo, System.currentTimeMillis()));
             }
             if (tableInfo.isPartitioned()) {
-                Map<String, Long> partitions = tablePathMap.get(tablePath);
+                Map<String, PartitionRegistration> partitions = tablePathMap.get(tablePath);
                 if (partitions != null) {
-                    for (Map.Entry<String, Long> partition : partitions.entrySet()) {
+                    for (Map.Entry<String, PartitionRegistration> partition :
+                            partitions.entrySet()) {
+                        long partitionId = partition.getValue().getPartitionId();
                         // put partition info to coordinator context
+                        PartitionInfo partitionInfo =
+                                new PartitionInfo(
+                                        partitionId,
+                                        ResolvedPartitionSpec.fromPartitionName(
+                                                tableInfo.getPartitionKeys(), partition.getKey()),
+                                        partition.getValue().getRemoteDataDir());
                         coordinatorContext.putPartition(
-                                partition.getValue(),
-                                PhysicalTablePath.of(tableInfo.getTablePath(), partition.getKey()));
+                                partitionId,
+                                PhysicalTablePath.of(tableInfo.getTablePath(), partition.getKey()),
+                                partitionInfo);
                     }
                 }
                 // if the table is auto partition, put the partitions info
@@ -712,6 +724,7 @@ private void processSchemaChange(SchemaChangeEvent schemaChangeEvent) {
                         oldTableInfo.getNumBuckets(),
                         oldTableInfo.getProperties(),
                         oldTableInfo.getCustomProperties(),
+                        oldTableInfo.getRemoteDataDir().orElse(null),
                         oldTableInfo.getComment().orElse(null),
                         oldTableInfo.getCreatedTime(),
                         System.currentTimeMillis()));
@@ -792,12 +805,14 @@ private void processCreatePartition(CreatePartitionEvent createPartitionEvent) {
         TablePath tablePath = createPartitionEvent.getTablePath();
         String partitionName = createPartitionEvent.getPartitionName();
         PartitionAssignment partitionAssignment = createPartitionEvent.getPartitionAssignment();
-        tableManager.onCreateNewPartition(
-                tablePath,
-                tableId,
-                createPartitionEvent.getPartitionId(),
-                partitionName,
-                partitionAssignment);
+        TableInfo tableInfo = coordinatorContext.getTableInfoById(tableId);
+        PartitionInfo partitionInfo =
+                new PartitionInfo(
+                        partitionId,
+                        ResolvedPartitionSpec.fromPartitionName(
+                                tableInfo.getPartitionKeys(), partitionName),
+                        createPartitionEvent.getRemoteDataDir());
+        tableManager.onCreateNewPartition(tablePath, tableId, partitionInfo, partitionAssignment);
         autoPartitionManager.addPartition(tableId, partitionName);
 
         Set<TableBucket> tableBuckets = new HashSet<>();