[core] Refactor MetastoreClient methods to simplify catalog

paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java

@@ -417,7 +417,7 @@ protected Table getDataOrFormatTable(Identifier identifier) throws TableNotExist
                                         lockFactory().orElse(null),
                                         lockContext().orElse(null),
                                         identifier),
-                                metastoreClientFactory(identifier, tableMeta.schema).orElse(null)));
+                                metastoreClientFactory(identifier).orElse(null)));
         CoreOptions options = table.coreOptions();
         if (options.type() == TableType.OBJECT_TABLE) {
             String objectLocation = options.objectLocation();
@@ -485,8 +485,7 @@ protected abstract TableSchema getDataTableSchema(Identifier identifier)
             throws TableNotExistException;
 
     /** Get metastore client factory for the table specified by {@code identifier}. */
-    public Optional<MetastoreClient.Factory> metastoreClientFactory(
-            Identifier identifier, TableSchema schema) {
+    public Optional<MetastoreClient.Factory> metastoreClientFactory(Identifier identifier) {
         return Optional.empty();
     }
 

paimon-core/src/main/java/org/apache/paimon/metastore/AddPartitionCommitCallback.java

@@ -25,6 +25,7 @@
 import org.apache.paimon.manifest.ManifestEntry;
 import org.apache.paimon.table.sink.CommitCallback;
 import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.utils.InternalRowPartitionComputer;
 
 import org.apache.paimon.shade.guava30.com.google.common.cache.Cache;
 import org.apache.paimon.shade.guava30.com.google.common.cache.CacheBuilder;
@@ -48,9 +49,12 @@ public class AddPartitionCommitCallback implements CommitCallback {
                     .build();
 
     private final MetastoreClient client;
+    private final InternalRowPartitionComputer partitionComputer;
 
-    public AddPartitionCommitCallback(MetastoreClient client) {
+    public AddPartitionCommitCallback(
+            MetastoreClient client, InternalRowPartitionComputer partitionComputer) {
         this.client = client;
+        this.partitionComputer = partitionComputer;
     }
 
     @Override
@@ -81,7 +85,10 @@ private void addPartitions(Set<BinaryRow> partitions) {
                 }
             }
             if (!newPartitions.isEmpty()) {
-                client.addPartitions(newPartitions);
+                client.addPartitions(
+                        newPartitions.stream()
+                                .map(partitionComputer::generatePartValues)
+                                .collect(Collectors.toList()));
                 newPartitions.forEach(partition -> cache.put(partition, true));
             }
         } catch (Exception e) {

paimon-core/src/main/java/org/apache/paimon/metastore/AddPartitionTagCallback.java

@@ -49,7 +49,7 @@ public void notifyDeletion(String tagName) {
         LinkedHashMap<String, String> partitionSpec = new LinkedHashMap<>();
         partitionSpec.put(partitionField, tagName);
         try {
-            client.deletePartition(partitionSpec);
+            client.dropPartition(partitionSpec);
         } catch (Exception e) {
             throw new RuntimeException(e);
         }

paimon-core/src/main/java/org/apache/paimon/metastore/MetastoreClient.java

@@ -18,45 +18,28 @@
 
 package org.apache.paimon.metastore;
 
-import org.apache.paimon.data.BinaryRow;
-
 import java.io.Serializable;
 import java.util.LinkedHashMap;
 import java.util.List;
-import java.util.Map;
 
 /**
  * A metastore client related to a table. All methods of this interface operate on the same specific
  * table.
  */
 public interface MetastoreClient extends AutoCloseable {
 
-    void addPartition(BinaryRow partition) throws Exception;
-
-    default void addPartitions(List<BinaryRow> partitions) throws Exception {
-        for (BinaryRow partition : partitions) {
-            addPartition(partition);
-        }
-    }
+    void addPartition(LinkedHashMap<String, String> partition) throws Exception;
 
-    void addPartition(LinkedHashMap<String, String> partitionSpec) throws Exception;
+    void addPartitions(List<LinkedHashMap<String, String>> partitions) throws Exception;
 
-    default void addPartitionsSpec(List<LinkedHashMap<String, String>> partitionSpecsList)
-            throws Exception {
-        for (LinkedHashMap<String, String> partitionSpecs : partitionSpecsList) {
-            addPartition(partitionSpecs);
-        }
-    }
+    void dropPartition(LinkedHashMap<String, String> partition) throws Exception;
 
-    void deletePartition(LinkedHashMap<String, String> partitionSpec) throws Exception;
+    void dropPartitions(List<LinkedHashMap<String, String>> partitions) throws Exception;
 
-    void markDone(LinkedHashMap<String, String> partitionSpec) throws Exception;
+    void markPartitionDone(LinkedHashMap<String, String> partition) throws Exception;
 
     default void alterPartition(
-            LinkedHashMap<String, String> partitionSpec,
-            Map<String, String> parameters,
-            long modifyTime,
-            boolean ignoreIfNotExist)
+            LinkedHashMap<String, String> partition, PartitionStats partitionStats)
             throws Exception {
         throw new UnsupportedOperationException();
     }

paimon-core/src/main/java/org/apache/paimon/metastore/PartitionStats.java

@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.metastore;
+
+/** Statistic for partition. */
+public interface PartitionStats {
+
+    long numFiles();
+
+    long totalSize();
+
+    long numRows();
+
+    long lastUpdateTimeMillis();
+
+    static PartitionStats create(
+            long numFiles, long totalSize, long numRows, long lastUpdateTimeMillis) {
+        return new PartitionStats() {
+
+            @Override
+            public long numFiles() {
+                return numFiles;
+            }
+
+            @Override
+            public long totalSize() {
+                return totalSize;
+            }
+
+            @Override
+            public long numRows() {
+                return numRows;
+            }
+
+            @Override
+            public long lastUpdateTimeMillis() {
+                return lastUpdateTimeMillis;
+            }
+
+            @Override
+            public String toString() {
+                return String.format(
+                        "numFiles: %s, totalSize: %s, numRows: %s, lastUpdateTimeMillis: %s",
+                        numFiles, totalSize, numRows, lastUpdateTimeMillis);
+            }
+        };
+    }
+}

paimon-core/src/main/java/org/apache/paimon/operation/PartitionExpire.java

@@ -167,15 +167,13 @@ private List<Map<String, String>> doExpire(
     }
 
     private void deleteMetastorePartitions(List<Map<String, String>> partitions) {
-        if (metastoreClient != null) {
-            partitions.forEach(
-                    partition -> {
-                        try {
-                            metastoreClient.deletePartition(new LinkedHashMap<>(partition));
-                        } catch (Exception e) {
-                            throw new RuntimeException(e);
-                        }
-                    });
+        if (metastoreClient != null && partitions.size() > 0) {
+            try {
+                metastoreClient.addPartitions(
+                        partitions.stream().map(LinkedHashMap::new).collect(Collectors.toList()));
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
         }
     }
 

paimon-core/src/main/java/org/apache/paimon/partition/actions/MarkPartitionDoneEventAction.java

@@ -39,7 +39,7 @@ public MarkPartitionDoneEventAction(MetastoreClient metastoreClient) {
     public void markDone(String partition) throws Exception {
         LinkedHashMap<String, String> partitionSpec =
                 extractPartitionSpecFromPath(new Path(partition));
-        metastoreClient.markDone(partitionSpec);
+        metastoreClient.markPartitionDone(partitionSpec);
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java

@@ -61,6 +61,7 @@
 import org.apache.paimon.table.source.snapshot.TimeTravelUtil;
 import org.apache.paimon.tag.TagPreview;
 import org.apache.paimon.utils.BranchManager;
+import org.apache.paimon.utils.InternalRowPartitionComputer;
 import org.apache.paimon.utils.Preconditions;
 import org.apache.paimon.utils.SegmentsCache;
 import org.apache.paimon.utils.SimpleFileReader;
@@ -469,7 +470,15 @@ protected List<CommitCallback> createCommitCallbacks(String commitUser) {
         if (options.partitionedTableInMetastore()
                 && metastoreClientFactory != null
                 && !tableSchema.partitionKeys().isEmpty()) {
-            callbacks.add(new AddPartitionCommitCallback(metastoreClientFactory.create()));
+            InternalRowPartitionComputer partitionComputer =
+                    new InternalRowPartitionComputer(
+                            options.partitionDefaultName(),
+                            tableSchema.logicalPartitionType(),
+                            tableSchema.partitionKeys().toArray(new String[0]),
+                            options.legacyPartitionName());
+            callbacks.add(
+                    new AddPartitionCommitCallback(
+                            metastoreClientFactory.create(), partitionComputer));
         }
 
         TagPreview tagPreview = TagPreview.create(options);

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/PartitionStatisticsReporter.java

@@ -22,6 +22,7 @@
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.metastore.MetastoreClient;
+import org.apache.paimon.metastore.PartitionStats;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.source.DataSplit;
 import org.apache.paimon.table.source.ScanMode;
@@ -35,24 +36,16 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
-import java.util.Map;
 
-import static org.apache.paimon.catalog.Catalog.LAST_UPDATE_TIME_PROP;
-import static org.apache.paimon.catalog.Catalog.NUM_FILES_PROP;
-import static org.apache.paimon.catalog.Catalog.NUM_ROWS_PROP;
-import static org.apache.paimon.catalog.Catalog.TOTAL_SIZE_PROP;
 import static org.apache.paimon.utils.PartitionPathUtils.extractPartitionSpecFromPath;
 
 /** Action to report the table statistic from the latest snapshot to HMS. */
 public class PartitionStatisticsReporter implements Closeable {
 
     private static final Logger LOG = LoggerFactory.getLogger(PartitionStatisticsReporter.class);
 
-    private static final String HIVE_LAST_UPDATE_TIME_PROP = "transient_lastDdlTime";
-
     private final MetastoreClient metastoreClient;
     private final SnapshotReader snapshotReader;
     private final SnapshotManager snapshotManager;
@@ -64,7 +57,7 @@ public PartitionStatisticsReporter(FileStoreTable table, MetastoreClient client)
         this.snapshotManager = table.snapshotManager();
     }
 
-    public void report(String partition, long modifyTime) throws Exception {
+    public void report(String partition, long modifyTimeMillis) throws Exception {
         Snapshot snapshot = snapshotManager.latestSnapshot();
         if (snapshot != null) {
             LinkedHashMap<String, String> partitionSpec =
@@ -88,19 +81,11 @@ public void report(String partition, long modifyTime) throws Exception {
                     totalSize += fileMeta.fileSize();
                 }
             }
-            Map<String, String> statistic = new HashMap<>();
-            statistic.put(NUM_FILES_PROP, String.valueOf(fileCount));
-            statistic.put(TOTAL_SIZE_PROP, String.valueOf(totalSize));
-            statistic.put(NUM_ROWS_PROP, String.valueOf(rowCount));
-
-            String modifyTimeSeconds = String.valueOf(modifyTime / 1000);
-            statistic.put(LAST_UPDATE_TIME_PROP, modifyTimeSeconds);
-
-            // just for being compatible with hive metastore
-            statistic.put(HIVE_LAST_UPDATE_TIME_PROP, modifyTimeSeconds);
 
-            LOG.info("alter partition {} with statistic {}.", partitionSpec, statistic);
-            metastoreClient.alterPartition(partitionSpec, statistic, modifyTime, true);
+            PartitionStats partitionStats =
+                    PartitionStats.create(fileCount, totalSize, rowCount, modifyTimeMillis);
+            LOG.info("alter partition {} with statistic {}.", partitionSpec, partitionStats);
+            metastoreClient.alterPartition(partitionSpec, partitionStats);
         }
     }
 

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/AddDonePartitionActionTest.java

@@ -18,15 +18,15 @@
 
 package org.apache.paimon.flink.sink.partition;
 
-import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.metastore.MetastoreClient;
+import org.apache.paimon.metastore.PartitionStats;
 import org.apache.paimon.partition.actions.AddDonePartitionAction;
 
 import org.junit.jupiter.api.Test;
 
 import java.util.HashSet;
 import java.util.LinkedHashMap;
-import java.util.Map;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -41,39 +41,41 @@ public void test() throws Exception {
         Set<String> donePartitions = new HashSet<>();
         MetastoreClient metastoreClient =
                 new MetastoreClient() {
+
                     @Override
-                    public void addPartition(BinaryRow partition) {
-                        throw new UnsupportedOperationException();
+                    public void addPartition(LinkedHashMap<String, String> partition) {
+                        donePartitions.add(generatePartitionPath(partition));
+                    }
+
+                    @Override
+                    public void addPartitions(List<LinkedHashMap<String, String>> partitions) {
+                        partitions.forEach(this::addPartition);
                     }
 
                     @Override
-                    public void addPartition(LinkedHashMap<String, String> partitionSpec) {
-                        donePartitions.add(generatePartitionPath(partitionSpec));
+                    public void dropPartition(LinkedHashMap<String, String> partition) {
+                        throw new UnsupportedOperationException();
                     }
 
                     @Override
-                    public void deletePartition(LinkedHashMap<String, String> partitionSpec) {
+                    public void dropPartitions(List<LinkedHashMap<String, String>> partitions) {
                         throw new UnsupportedOperationException();
                     }
 
                     @Override
-                    public void markDone(LinkedHashMap<String, String> partitionSpec)
-                            throws Exception {
+                    public void markPartitionDone(LinkedHashMap<String, String> partitions) {
                         throw new UnsupportedOperationException();
                     }
 
                     @Override
                     public void alterPartition(
                             LinkedHashMap<String, String> partitionSpec,
-                            Map<String, String> parameters,
-                            long modifyTime,
-                            boolean ignoreIfNotExist)
-                            throws Exception {
+                            PartitionStats partitionStats) {
                         throw new UnsupportedOperationException();
                     }
 
                     @Override
-                    public void close() throws Exception {
+                    public void close() {
                         closed.set(true);
                     }
                 };