Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[core] Support Table API in RESTCatalog #4736

Draft
wants to merge 22 commits into
base: master
Choose a base branch
from
Draft

[core] Support Table API in RESTCatalog #4736

Show file tree
Hide file tree
Changes from 18 commits
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
1865f8e
change RESTCatalog extends AbstractCatalog
jerry-024 Dec 17, 2024
e140867
change RESTCatalog extend abstract catalog
jerry-024 Dec 18, 2024
e9d9a67
support get FileIO
jerry-024 Dec 18, 2024
5c11254
add api support list tables and get table
jerry-024 Dec 18, 2024
5434187
support create table
jerry-024 Dec 18, 2024
078c46a
add update table and drop table
jerry-024 Dec 18, 2024
4b9177d
fix test fail
jerry-024 Dec 18, 2024
8bc42c4
change RESTCatalog implements Catalog
jerry-024 Dec 18, 2024
d67735a
remove change in AbstractCatalog
jerry-024 Dec 18, 2024
67067d3
remove change in AbstractCatalog
jerry-024 Dec 18, 2024
a682c0f
support get table follow Abstract Catalog
jerry-024 Dec 18, 2024
a90c527
fix test fail in RESTCatalogTest
jerry-024 Dec 19, 2024
1d785b8
move some methods to CatalogUtils from AbstractCatalog
jerry-024 Dec 19, 2024
e997351
Merge branch 'apache:master' into rest-catalog
jerry-024 Dec 19, 2024
28c0a65
move getSystemTable to CatalogUtils
jerry-024 Dec 19, 2024
95103ae
handle exception when alter or rename table and add check system data…
jerry-024 Dec 19, 2024
711c43c
add exception TableNoPermissionException
jerry-024 Dec 19, 2024
09b4fbe
fix compile error
jerry-024 Dec 19, 2024
da23974
fix compile error
jerry-024 Dec 19, 2024
ba690d6
fix checkstyle error
jerry-024 Dec 19, 2024
5464dfa
fix compile error
jerry-024 Dec 19, 2024
6eb87f5
fix compile error
jerry-024 Dec 19, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 11 additions & 82 deletions paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
View file
Open in desktop
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
8 changes: 5 additions & 3 deletions paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -210,21 +210,23 @@ public void dropTable(Identifier identifier, boolean ignoreIfNotExists)

@Override
public void renameTable(Identifier fromTable, Identifier toTable, boolean ignoreIfNotExists)
throws TableNotExistException, TableAlreadyExistException {
throws TableNotExistException, TableAlreadyExistException, TableNoPermissionException {
super.renameTable(fromTable, toTable, ignoreIfNotExists);
invalidateTable(fromTable);
}

@Override
public void alterTable(
Identifier identifier, List<SchemaChange> changes, boolean ignoreIfNotExists)
throws TableNotExistException, ColumnAlreadyExistException, ColumnNotExistException {
throws TableNotExistException, ColumnAlreadyExistException, ColumnNotExistException,
TableNoPermissionException {
super.alterTable(identifier, changes, ignoreIfNotExists);
invalidateTable(identifier);
}

@Override
public Table getTable(Identifier identifier) throws TableNotExistException {
public Table getTable(Identifier identifier)
throws TableNotExistException, TableNoPermissionException {
Table table = tableCache.getIfPresent(identifier);
if (table != null) {
return table;
Expand Down
31 changes: 27 additions & 4 deletions paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,7 @@ void alterDatabase(String name, List<PropertyChange> changes, boolean ignoreIfNo
* @return The requested table
* @throws TableNotExistException if the target does not exist
*/
Table getTable(Identifier identifier) throws TableNotExistException;
Table getTable(Identifier identifier) throws TableNotExistException, TableNoPermissionException;

/**
* Get names of all tables under this database. An empty list is returned if none exists.
Expand Down Expand Up @@ -204,7 +204,7 @@ void createTable(Identifier identifier, Schema schema, boolean ignoreIfExists)
* @throws TableAlreadyExistException if the toTable already exists
*/
void renameTable(Identifier fromTable, Identifier toTable, boolean ignoreIfNotExists)
throws TableNotExistException, TableAlreadyExistException;
throws TableNotExistException, TableAlreadyExistException, TableNoPermissionException;

/**
* Modify an existing table from {@link SchemaChange}s.
Expand All @@ -218,7 +218,8 @@ void renameTable(Identifier fromTable, Identifier toTable, boolean ignoreIfNotEx
* @throws TableNotExistException if the table does not exist
*/
void alterTable(Identifier identifier, List<SchemaChange> changes, boolean ignoreIfNotExists)
throws TableNotExistException, ColumnAlreadyExistException, ColumnNotExistException;
throws TableNotExistException, ColumnAlreadyExistException, ColumnNotExistException,
TableNoPermissionException;

/**
* Invalidate cached table metadata for an {@link Identifier identifier}.
Expand Down Expand Up @@ -274,7 +275,8 @@ void dropPartition(Identifier identifier, Map<String, String> partitions)
* @throws TableNotExistException if the table does not exist
*/
default void alterTable(Identifier identifier, SchemaChange change, boolean ignoreIfNotExists)
throws TableNotExistException, ColumnAlreadyExistException, ColumnNotExistException {
throws TableNotExistException, ColumnAlreadyExistException, ColumnNotExistException,
TableNoPermissionException {
alterTable(identifier, Collections.singletonList(change), ignoreIfNotExists);
}

Expand Down Expand Up @@ -433,6 +435,27 @@ public ProcessSystemDatabaseException() {
}
}

/** Exception for trying to operate on a table that doesn't have permission. */
class TableNoPermissionException extends Exception {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe it is better to be a RuntimeException? It is better to not bother method interfaces.


private static final String MSG = "No permission for Table %s.";

private final Identifier identifier;

public TableNoPermissionException(Identifier identifier) {
this(identifier, null);
}

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 create a table that already exists. */
class TableAlreadyExistException extends Exception {

Expand Down
101 changes: 101 additions & 0 deletions paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java
View file
Open in desktop
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;
}
}
8 changes: 5 additions & 3 deletions paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,19 +108,21 @@ public void createTable(Identifier identifier, Schema schema, boolean ignoreIfEx

@Override
public void renameTable(Identifier fromTable, Identifier toTable, boolean ignoreIfNotExists)
throws TableNotExistException, TableAlreadyExistException {
throws TableNotExistException, TableAlreadyExistException, TableNoPermissionException {
wrapped.renameTable(fromTable, toTable, ignoreIfNotExists);
}

@Override
public void alterTable(
Identifier identifier, List<SchemaChange> changes, boolean ignoreIfNotExists)
throws TableNotExistException, ColumnAlreadyExistException, ColumnNotExistException {
throws TableNotExistException, ColumnAlreadyExistException, ColumnNotExistException,
TableNoPermissionException {
wrapped.alterTable(identifier, changes, ignoreIfNotExists);
}

@Override
public Table getTable(Identifier identifier) throws TableNotExistException {
public Table getTable(Identifier identifier)
throws TableNotExistException, TableNoPermissionException {
return wrapped.getTable(identifier);
}

Expand Down
Loading
Loading