[core] Support Table API in RESTCatalog

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

@@ -20,7 +20,6 @@
 
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.TableType;
-import org.apache.paimon.factories.FactoryUtil;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.FileStatus;
 import org.apache.paimon.fs.Path;
@@ -59,8 +58,11 @@
 
 import static org.apache.paimon.CoreOptions.TYPE;
 import static org.apache.paimon.CoreOptions.createCommitUser;
-import static org.apache.paimon.options.CatalogOptions.LOCK_ENABLED;
-import static org.apache.paimon.options.CatalogOptions.LOCK_TYPE;
+import static org.apache.paimon.catalog.CatalogUtils.checkNotBranch;
+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.lockFactory;
 import static org.apache.paimon.utils.BranchManager.DEFAULT_MAIN_BRANCH;
 import static org.apache.paimon.utils.Preconditions.checkArgument;
 import static org.apache.paimon.utils.Preconditions.checkNotNull;
@@ -94,31 +96,16 @@ public FileIO fileIO() {
         return fileIO;
     }
 
-    public Optional<CatalogLockFactory> lockFactory() {
-        if (!lockEnabled()) {
-            return Optional.empty();
-        }
-
-        String lock = catalogOptions.get(LOCK_TYPE);
-        if (lock == null) {
-            return defaultLockFactory();
-        }
-
-        return Optional.of(
-                FactoryUtil.discoverFactory(
-                        AbstractCatalog.class.getClassLoader(), CatalogLockFactory.class, lock));
-    }
-
     public Optional<CatalogLockFactory> defaultLockFactory() {
         return Optional.empty();
     }
 
     public Optional<CatalogLockContext> lockContext() {
-        return Optional.of(CatalogLockContext.fromOptions(catalogOptions));
+        return CatalogUtils.lockContext(catalogOptions);
     }
 
     protected boolean lockEnabled() {
-        return catalogOptions.getOptional(LOCK_ENABLED).orElse(fileIO.isObjectStore());
+        return CatalogUtils.lockEnabled(catalogOptions, fileIO);
     }
 
     protected boolean allowCustomTablePath() {
@@ -397,20 +384,7 @@ public Table getTable(Identifier identifier) throws TableNotExistException {
                                     identifier.getTableName(),
                                     identifier.getBranchName(),
                                     null));
-            if (!(originTable instanceof FileStoreTable)) {
-                throw new UnsupportedOperationException(
-                        String.format(
-                                "Only data table support system tables, but this table %s is %s.",
-                                identifier, originTable.getClass()));
-            }
-            Table table =
-                    SystemTableLoader.load(
-                            Preconditions.checkNotNull(identifier.getSystemTableName()),
-                            (FileStoreTable) originTable);
-            if (table == null) {
-                throw new TableNotExistException(identifier);
-            }
-            return table;
+            return CatalogUtils.getSystemTable(identifier, originTable);
         } else {
             return getDataOrFormatTable(identifier);
         }
@@ -428,7 +402,8 @@ protected Table getDataOrFormatTable(Identifier identifier) throws TableNotExist
                                 identifier,
                                 tableMeta.uuid,
                                 Lock.factory(
-                                        lockFactory().orElse(null),
+                                        lockFactory(catalogOptions, fileIO(), defaultLockFactory())
+                                                .orElse(null),
                                         lockContext().orElse(null),
                                         identifier),
                                 metastoreClientFactory(identifier).orElse(null)));
@@ -472,7 +447,7 @@ public void createFormatTable(Identifier identifier, Schema schema) {
      * @return The warehouse path for the database
      */
     public Path newDatabasePath(String database) {
-        return newDatabasePath(warehouse(), database);
+        return CatalogUtils.newDatabasePath(warehouse(), database);
     }
 
     public Map<String, Map<String, Path>> allTablePaths() {
@@ -507,16 +482,6 @@ public Path getTableLocation(Identifier identifier) {
         return new Path(newDatabasePath(identifier.getDatabaseName()), identifier.getTableName());
     }
 
-    protected static void checkNotBranch(Identifier identifier, String method) {
-        if (identifier.getBranchName() != null) {
-            throw new IllegalArgumentException(
-                    String.format(
-                            "Cannot '%s' for branch table '%s', "
-                                    + "please modify the table with the default branch.",
-                            method, identifier));
-        }
-    }
-
     protected void assertMainBranch(Identifier identifier) {
         if (identifier.getBranchName() != null
                 && !DEFAULT_MAIN_BRANCH.equals(identifier.getBranchName())) {
@@ -525,46 +490,10 @@ protected void assertMainBranch(Identifier identifier) {
         }
     }
 
-    protected static boolean isTableInSystemDatabase(Identifier identifier) {
-        return isSystemDatabase(identifier.getDatabaseName()) || identifier.isSystemTable();
-    }
-
-    protected static void checkNotSystemTable(Identifier identifier, String method) {
-        if (isTableInSystemDatabase(identifier)) {
-            throw new IllegalArgumentException(
-                    String.format(
-                            "Cannot '%s' for system table '%s', please use data table.",
-                            method, identifier));
-        }
-    }
-
     private void copyTableDefaultOptions(Map<String, String> options) {
         tableDefaultOptions.forEach(options::putIfAbsent);
     }
 
-    public static Path newTableLocation(String warehouse, Identifier identifier) {
-        checkNotBranch(identifier, "newTableLocation");
-        checkNotSystemTable(identifier, "newTableLocation");
-        return new Path(
-                newDatabasePath(warehouse, identifier.getDatabaseName()),
-                identifier.getTableName());
-    }
-
-    public static Path newDatabasePath(String warehouse, String database) {
-        return new Path(warehouse, database + DB_SUFFIX);
-    }
-
-    public static boolean isSystemDatabase(String database) {
-        return SYSTEM_DATABASE_NAME.equals(database);
-    }
-
-    /** Validate database cannot be a system database. */
-    protected void checkNotSystemDatabase(String database) {
-        if (isSystemDatabase(database)) {
-            throw new ProcessSystemDatabaseException();
-        }
-    }
-
     private void validateAutoCreateClose(Map<String, String> options) {
         checkArgument(
                 !Boolean.parseBoolean(

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

@@ -433,6 +433,22 @@ public ProcessSystemDatabaseException() {
         }
     }
 
+    /** Exception for trying to operate on the database that doesn't have permission. */
+    class DatabaseNoPermissionException extends RuntimeException {
+        private static final String MSG = "Database %s has no permission.";
+
+        private final String database;
+
+        public DatabaseNoPermissionException(String database, Throwable cause) {
+            super(String.format(MSG, database), cause);
+            this.database = database;
+        }
+
+        public String database() {
+            return database;
+        }
+    }
+
     /** Exception for trying to create a table that already exists. */
     class TableAlreadyExistException extends Exception {
 
@@ -475,6 +491,23 @@ public Identifier identifier() {
         }
     }
 
+    /** Exception for trying to operate on the table that doesn't have permission. */
+    class TableNoPermissionException extends RuntimeException {
+
+        private static final String MSG = "Table %s has no permission.";
+
+        private final Identifier identifier;
+
+        public TableNoPermissionException(Identifier identifier, Throwable cause) {
+            super(String.format(MSG, identifier.getFullName()), cause);
+            this.identifier = identifier;
+        }
+
+        public Identifier identifier() {
+            return identifier;
+        }
+    }
+
     /** Exception for trying to operate on a partition that doesn't exist. */
     class PartitionNotExistException extends Exception {
 

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

@@ -18,12 +18,24 @@
 
 package org.apache.paimon.catalog;
 
+import org.apache.paimon.factories.FactoryUtil;
+import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.Options;
 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.Preconditions;
 
 import java.util.Map;
+import java.util.Optional;
 
+import static org.apache.paimon.catalog.Catalog.DB_SUFFIX;
+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.CatalogOptions.LOCK_ENABLED;
+import static org.apache.paimon.options.CatalogOptions.LOCK_TYPE;
 import static org.apache.paimon.options.OptionsUtils.convertToPropertiesPrefixKey;
 
 /** Utils for {@link Catalog}. */
@@ -60,4 +72,93 @@ public static String table(String path) {
     public static Map<String, String> tableDefaultOptions(Map<String, String> options) {
         return convertToPropertiesPrefixKey(options, TABLE_DEFAULT_OPTION_PREFIX);
     }
+
+    public static boolean isSystemDatabase(String database) {
+        return SYSTEM_DATABASE_NAME.equals(database);
+    }
+
+    /** Validate database cannot be a system database. */
+    public static void checkNotSystemDatabase(String database) {
+        if (isSystemDatabase(database)) {
+            throw new Catalog.ProcessSystemDatabaseException();
+        }
+    }
+
+    public static boolean isTableInSystemDatabase(Identifier identifier) {
+        return isSystemDatabase(identifier.getDatabaseName()) || identifier.isSystemTable();
+    }
+
+    public static void checkNotSystemTable(Identifier identifier, String method) {
+        if (isTableInSystemDatabase(identifier)) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Cannot '%s' for system table '%s', please use data table.",
+                            method, identifier));
+        }
+    }
+
+    public static Path newDatabasePath(String warehouse, String database) {
+        return new Path(warehouse, database + DB_SUFFIX);
+    }
+
+    public static Path newTableLocation(String warehouse, Identifier identifier) {
+        checkNotBranch(identifier, "newTableLocation");
+        checkNotSystemTable(identifier, "newTableLocation");
+        return new Path(
+                newDatabasePath(warehouse, identifier.getDatabaseName()),
+                identifier.getTableName());
+    }
+
+    public static void checkNotBranch(Identifier identifier, String method) {
+        if (identifier.getBranchName() != null) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Cannot '%s' for branch table '%s', "
+                                    + "please modify the table with the default branch.",
+                            method, identifier));
+        }
+    }
+
+    public static Optional<CatalogLockFactory> lockFactory(
+            Options options, FileIO fileIO, Optional<CatalogLockFactory> defaultLockFactoryOpt) {
+        boolean lockEnabled = lockEnabled(options, fileIO);
+        if (!lockEnabled) {
+            return Optional.empty();
+        }
+
+        String lock = options.get(LOCK_TYPE);
+        if (lock == null) {
+            return defaultLockFactoryOpt;
+        }
+
+        return Optional.of(
+                FactoryUtil.discoverFactory(
+                        AbstractCatalog.class.getClassLoader(), CatalogLockFactory.class, lock));
+    }
+
+    public static Optional<CatalogLockContext> lockContext(Options options) {
+        return Optional.of(CatalogLockContext.fromOptions(options));
+    }
+
+    public static boolean lockEnabled(Options options, FileIO fileIO) {
+        return options.getOptional(LOCK_ENABLED).orElse(fileIO != null && fileIO.isObjectStore());
+    }
+
+    public static Table getSystemTable(Identifier identifier, Table originTable)
+            throws Catalog.TableNotExistException {
+        if (!(originTable instanceof FileStoreTable)) {
+            throw new UnsupportedOperationException(
+                    String.format(
+                            "Only data table support system tables, but this table %s is %s.",
+                            identifier, originTable.getClass()));
+        }
+        Table table =
+                SystemTableLoader.load(
+                        Preconditions.checkNotNull(identifier.getSystemTableName()),
+                        (FileStoreTable) originTable);
+        if (table == null) {
+            throw new Catalog.TableNotExistException(identifier);
+        }
+        return table;
+    }
 }

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

@@ -34,6 +34,7 @@
 import java.util.Map;
 import java.util.concurrent.Callable;
 
+import static org.apache.paimon.catalog.CatalogUtils.lockFactory;
 import static org.apache.paimon.options.CatalogOptions.CASE_SENSITIVE;
 
 /** A catalog implementation for {@link FileIO}. */
@@ -123,7 +124,9 @@ public void createTableImpl(Identifier identifier, Schema schema) {
     private SchemaManager schemaManager(Identifier identifier) {
         Path path = getTableLocation(identifier);
         CatalogLock catalogLock =
-                lockFactory().map(fac -> fac.createLock(assertGetLockContext())).orElse(null);
+                lockFactory(catalogOptions, fileIO(), defaultLockFactory())
+                        .map(fac -> fac.createLock(assertGetLockContext()))
+                        .orElse(null);
         return new SchemaManager(fileIO, path, identifier.getBranchNameOrDefault())
                 .withLock(catalogLock == null ? null : Lock.fromCatalog(catalogLock, identifier));
     }