Skip to content

Commit

Permalink
[core] Support Table API in RESTCatalog (apache#4736)
Browse files Browse the repository at this point in the history
  • Loading branch information
@jerry-024
jerry-024 authored Dec 24, 2024
1 parent d6677fc commit 9ac36bb
Show file tree
Hide file tree
Showing 29 changed files with 1,936 additions and 206 deletions.
93 changes: 11 additions & 82 deletions paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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() {
Expand Down Expand Up @@ -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);
}
Expand All @@ -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)));
Expand Down Expand Up @@ -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() {
Expand Down Expand Up @@ -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())) {
Expand All @@ -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(
Expand Down
33 changes: 33 additions & 0 deletions paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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 {

Expand Down Expand Up @@ -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 {

Expand Down
101 changes: 101 additions & 0 deletions paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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}. */
Expand Down Expand Up @@ -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;
}
}
5 changes: 4 additions & 1 deletion paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalog.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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}. */
Expand Down Expand Up @@ -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));
}
Expand Down
Loading

0 comments on commit 9ac36bb

Please sign in to comment.