[core] Introduce Partition to Catalog.listPartitions

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

@@ -24,11 +24,11 @@
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.FileStatus;
 import org.apache.paimon.fs.Path;
-import org.apache.paimon.manifest.PartitionEntry;
 import org.apache.paimon.metastore.MetastoreClient;
 import org.apache.paimon.operation.FileStoreCommit;
 import org.apache.paimon.operation.Lock;
 import org.apache.paimon.options.Options;
+import org.apache.paimon.partition.Partition;
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.schema.SchemaChange;
 import org.apache.paimon.schema.SchemaManager;
@@ -65,6 +65,7 @@
 import static org.apache.paimon.catalog.CatalogUtils.checkNotSystemDatabase;
 import static org.apache.paimon.catalog.CatalogUtils.checkNotSystemTable;
 import static org.apache.paimon.catalog.CatalogUtils.isSystemDatabase;
+import static org.apache.paimon.catalog.CatalogUtils.listPartitionsFromFileSystem;
 import static org.apache.paimon.options.CatalogOptions.LOCK_ENABLED;
 import static org.apache.paimon.options.CatalogOptions.LOCK_TYPE;
 import static org.apache.paimon.table.system.AllTableOptionsTable.ALL_TABLE_OPTIONS;
@@ -200,9 +201,8 @@ public void dropPartition(Identifier identifier, Map<String, String> partitionSp
     }
 
     @Override
-    public List<PartitionEntry> listPartitions(Identifier identifier)
-            throws TableNotExistException {
-        return getTable(identifier).newReadBuilder().newScan().listPartitionEntries();
+    public List<Partition> listPartitions(Identifier identifier) throws TableNotExistException {
+        return listPartitionsFromFileSystem(getTable(identifier));
     }
 
     protected abstract void createDatabaseImpl(String name, Map<String, String> properties);

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

@@ -19,9 +19,9 @@
 package org.apache.paimon.catalog;
 
 import org.apache.paimon.fs.Path;
-import org.apache.paimon.manifest.PartitionEntry;
 import org.apache.paimon.options.MemorySize;
 import org.apache.paimon.options.Options;
+import org.apache.paimon.partition.Partition;
 import org.apache.paimon.schema.SchemaChange;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.Table;
@@ -61,7 +61,7 @@ public class CachingCatalog extends DelegateCatalog {
     @Nullable protected final SegmentsCache<Path> manifestCache;
 
     // partition cache will affect data latency
-    @Nullable protected final Cache<Identifier, List<PartitionEntry>> partitionCache;
+    @Nullable protected final Cache<Identifier, List<Partition>> partitionCache;
 
     public CachingCatalog(Catalog wrapped) {
         this(
@@ -130,7 +130,7 @@ public CachingCatalog(
                                 .executor(Runnable::run)
                                 .expireAfterAccess(expirationInterval)
                                 .weigher(
-                                        (Weigher<Identifier, List<PartitionEntry>>)
+                                        (Weigher<Identifier, List<Partition>>)
                                                 (identifier, v) -> v.size())
                                 .maximumWeight(cachedPartitionMaxNum)
                                 .ticker(ticker)
@@ -281,13 +281,12 @@ private void putTableCache(Identifier identifier, Table table) {
     }
 
     @Override
-    public List<PartitionEntry> listPartitions(Identifier identifier)
-            throws TableNotExistException {
+    public List<Partition> listPartitions(Identifier identifier) throws TableNotExistException {
         if (partitionCache == null) {
             return wrapped.listPartitions(identifier);
         }
 
-        List<PartitionEntry> result = partitionCache.getIfPresent(identifier);
+        List<Partition> result = partitionCache.getIfPresent(identifier);
         if (result == null) {
             result = wrapped.listPartitions(identifier);
             partitionCache.put(identifier, result);
@@ -321,7 +320,7 @@ public void invalidateTable(Identifier identifier) {
      */
     public void refreshPartitions(Identifier identifier) throws TableNotExistException {
         if (partitionCache != null) {
-            List<PartitionEntry> result = wrapped.listPartitions(identifier);
+            List<Partition> result = wrapped.listPartitions(identifier);
             partitionCache.put(identifier, result);
         }
     }
@@ -341,8 +340,7 @@ public CacheSizes estimatedCacheSizes() {
         }
         long partitionCacheSize = 0L;
         if (partitionCache != null) {
-            for (Map.Entry<Identifier, List<PartitionEntry>> entry :
-                    partitionCache.asMap().entrySet()) {
+            for (Map.Entry<Identifier, List<Partition>> entry : partitionCache.asMap().entrySet()) {
                 partitionCacheSize += entry.getValue().size();
             }
         }

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

@@ -20,7 +20,7 @@
 
 import org.apache.paimon.annotation.Public;
 import org.apache.paimon.fs.FileIO;
-import org.apache.paimon.manifest.PartitionEntry;
+import org.apache.paimon.partition.Partition;
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.schema.SchemaChange;
 import org.apache.paimon.table.Table;
@@ -255,12 +255,12 @@ void dropPartition(Identifier identifier, Map<String, String> partition)
             throws TableNotExistException, PartitionNotExistException;
 
     /**
-     * Get PartitionEntry of all partitions of the table.
+     * Get Partition of all partitions of the table.
      *
      * @param identifier path of the table to list partitions
      * @throws TableNotExistException if the table does not exist
      */
-    List<PartitionEntry> listPartitions(Identifier identifier) throws TableNotExistException;
+    List<Partition> listPartitions(Identifier identifier) throws TableNotExistException;
 
     /**
      * Modify an existing table from a {@link SchemaChange}.

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

@@ -19,14 +19,22 @@
 package org.apache.paimon.catalog;
 
 import org.apache.paimon.fs.Path;
+import org.apache.paimon.manifest.PartitionEntry;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.partition.Partition;
 import org.apache.paimon.schema.SchemaManager;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.Table;
 import org.apache.paimon.table.system.SystemTableLoader;
+import org.apache.paimon.utils.InternalRowPartitionComputer;
 import org.apache.paimon.utils.Preconditions;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 
+import static org.apache.paimon.CoreOptions.PARTITION_DEFAULT_NAME;
+import static org.apache.paimon.CoreOptions.PARTITION_GENERATE_LEGCY_NAME;
 import static org.apache.paimon.catalog.Catalog.SYSTEM_DATABASE_NAME;
 import static org.apache.paimon.catalog.Catalog.TABLE_DEFAULT_OPTION_PREFIX;
 import static org.apache.paimon.options.OptionsUtils.convertToPropertiesPrefixKey;
@@ -117,4 +125,27 @@ public static Table createSystemTable(Identifier identifier, Table originTable)
         }
         return table;
     }
+
+    public static List<Partition> listPartitionsFromFileSystem(Table table) {
+        Options options = Options.fromMap(table.options());
+        InternalRowPartitionComputer computer =
+                new InternalRowPartitionComputer(
+                        options.get(PARTITION_DEFAULT_NAME),
+                        table.rowType(),
+                        table.partitionKeys().toArray(new String[0]),
+                        options.get(PARTITION_GENERATE_LEGCY_NAME));
+        List<PartitionEntry> partitionEntries =
+                table.newReadBuilder().newScan().listPartitionEntries();
+        List<Partition> partitions = new ArrayList<>(partitionEntries.size());
+        for (PartitionEntry entry : partitionEntries) {
+            partitions.add(
+                    new Partition(
+                            computer.generatePartValues(entry.partition()),
+                            entry.recordCount(),
+                            entry.fileSizeInBytes(),
+                            entry.fileCount(),
+                            entry.lastFileCreationTime()));
+        }
+        return partitions;
+    }
 }

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

@@ -19,7 +19,7 @@
 package org.apache.paimon.catalog;
 
 import org.apache.paimon.fs.FileIO;
-import org.apache.paimon.manifest.PartitionEntry;
+import org.apache.paimon.partition.Partition;
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.schema.SchemaChange;
 import org.apache.paimon.table.Table;
@@ -165,8 +165,7 @@ public void dropPartition(Identifier identifier, Map<String, String> partitions)
     }
 
     @Override
-    public List<PartitionEntry> listPartitions(Identifier identifier)
-            throws TableNotExistException {
+    public List<Partition> listPartitions(Identifier identifier) throws TableNotExistException {
         return wrapped.listPartitions(identifier);
     }
 

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

@@ -18,6 +18,8 @@
 
 package org.apache.paimon.metastore;
 
+import org.apache.paimon.partition.Partition;
+
 import java.io.Serializable;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -38,9 +40,7 @@ public interface MetastoreClient extends AutoCloseable {
 
     void markPartitionDone(LinkedHashMap<String, String> partition) throws Exception;
 
-    default void alterPartition(
-            LinkedHashMap<String, String> partition, PartitionStats partitionStats)
-            throws Exception {
+    default void alterPartition(Partition partition) throws Exception {
         throw new UnsupportedOperationException();
     }