+ * -- NOTE: use '' as placeholder for optional arguments
+ *
+ * -- compact all databases
+ * CALL sys.compact_database()
+ *
+ * -- compact some databases (accept regular expression)
+ * CALL sys.compact_database('includingDatabases')
+ *
+ * -- set compact mode
+ * CALL sys.compact_database('includingDatabases', 'mode')
+ *
+ * -- compact some tables (accept regular expression)
+ * CALL sys.compact_database('includingDatabases', 'mode', 'includingTables')
+ *
+ * -- exclude some tables (accept regular expression)
+ * CALL sys.compact_database('includingDatabases', 'mode', 'includingTables', 'excludingTables')
+ *
+ * -- set table options ('k=v,...')
+ * CALL sys.compact_database('includingDatabases', 'mode', 'includingTables', 'excludingTables', 'tableOptions')
+ *
+ * CALL sys.create_branch('tableId', 'branchName', 'tagName')
+ *
+ * CALL sys.create_tag('tableId', 'tagName', snapshotId, 'timeRetained')
+ *
+ * CALL sys.mark_partition_done('tableId', 'partition1', 'partition2', ...)
+ *
+ * -- NOTE: use '' as placeholder for optional arguments
+ *
+ * -- when matched then upsert
+ * CALL sys.merge_into(
+ * 'targetTableId',
+ * 'targetAlias',
+ * 'sourceSqls', -- separate with ';'
+ * 'sourceTable',
+ * 'mergeCondition',
+ * 'matchedUpsertCondition',
+ * 'matchedUpsertSetting'
+ * )
+ *
+ * -- when matched then upsert + when not matched then insert
+ * CALL sys.merge_into(
+ * 'targetTableId'
+ * 'targetAlias',
+ * 'sourceSqls',
+ * 'sourceTable',
+ * 'mergeCondition',
+ * 'matchedUpsertCondition',
+ * 'matchedUpsertSetting',
+ * 'notMatchedInsertCondition',
+ * 'notMatchedInsertValues'
+ * )
+ *
+ * -- above + when matched then delete
+ * -- IMPORTANT: Use 'TRUE' if you want to delete data without filter condition.
+ * -- If matchedDeleteCondition='', it will ignore matched-delete action!
+ * CALL sys.merge_into(
+ * 'targetTableId',
+ * 'targetAlias',
+ * 'sourceSqls',
+ * 'sourceTable',
+ * 'mergeCondition',
+ * 'matchedUpsertCondition',
+ * 'matchedUpsertSetting',
+ * 'notMatchedInsertCondition',
+ * 'notMatchedInsertValues',
+ * 'matchedDeleteCondition'
+ * )
+ *
+ * -- when matched then delete (short form)
+ * CALL sys.merge_into(
+ * 'targetTableId'
+ * 'targetAlias',
+ * 'sourceSqls',
+ * 'sourceTable',
+ * 'mergeCondition',
+ * 'matchedDeleteCondition'
+ * )
+ * This procedure will be forced to use batch environments. Compared to {@link MergeIntoAction},
+ * this procedure doesn't provide arguments to control not-matched-by-source behavior because they
+ * are not commonly used and will make the methods too complex to use.
+ */
+public class MergeIntoProcedure extends ProcedureBase {
+
+ public static final String IDENTIFIER = "merge_into";
+
+ public String[] call(
+ ProcedureContext procedureContext,
+ String targetTableId,
+ String targetAlias,
+ String sourceSqls,
+ String sourceTable,
+ String mergeCondition,
+ String matchedUpsertCondition,
+ String matchedUpsertSetting) {
+ return call(
+ procedureContext,
+ targetTableId,
+ targetAlias,
+ sourceSqls,
+ sourceTable,
+ mergeCondition,
+ matchedUpsertCondition,
+ matchedUpsertSetting,
+ "",
+ "",
+ "");
+ }
+
+ public String[] call(
+ ProcedureContext procedureContext,
+ String targetTableId,
+ String targetAlias,
+ String sourceSqls,
+ String sourceTable,
+ String mergeCondition,
+ String matchedUpsertCondition,
+ String matchedUpsertSetting,
+ String notMatchedInsertCondition,
+ String notMatchedInsertValues) {
+ return call(
+ procedureContext,
+ targetTableId,
+ targetAlias,
+ sourceSqls,
+ sourceTable,
+ mergeCondition,
+ matchedUpsertCondition,
+ matchedUpsertSetting,
+ notMatchedInsertCondition,
+ notMatchedInsertValues,
+ "");
+ }
+
+ public String[] call(
+ ProcedureContext procedureContext,
+ String targetTableId,
+ String targetAlias,
+ String sourceSqls,
+ String sourceTable,
+ String mergeCondition,
+ String matchedDeleteCondition) {
+ return call(
+ procedureContext,
+ targetTableId,
+ targetAlias,
+ sourceSqls,
+ sourceTable,
+ mergeCondition,
+ "",
+ "",
+ "",
+ "",
+ matchedDeleteCondition);
+ }
+
+ public String[] call(
+ ProcedureContext procedureContext,
+ String targetTableId,
+ String targetAlias,
+ String sourceSqls,
+ String sourceTable,
+ String mergeCondition,
+ String matchedUpsertCondition,
+ String matchedUpsertSetting,
+ String notMatchedInsertCondition,
+ String notMatchedInsertValues,
+ String matchedDeleteCondition) {
+ String warehouse = catalog.warehouse();
+ Map
+ */
+public class RemoveOrphanFilesProcedure extends ProcedureBase {
+
+ public static final String IDENTIFIER = "remove_orphan_files";
+
+ public String[] call(ProcedureContext procedureContext, String tableId) throws Exception {
+ return call(procedureContext, tableId, "");
+ }
+
+ public String[] call(ProcedureContext procedureContext, String tableId, String olderThan)
+ throws Exception {
+ return call(procedureContext, tableId, olderThan, false);
+ }
+
+ public String[] call(
+ ProcedureContext procedureContext, String tableId, String olderThan, boolean dryRun)
+ throws Exception {
+ Identifier identifier = Identifier.fromString(tableId);
+ String databaseName = identifier.getDatabaseName();
+ String tableName = identifier.getObjectName();
+
+ List
+ * -- use the default file delete interval
+ * CALL sys.remove_orphan_files('tableId')
+ *
+ * -- use custom file delete interval
+ * CALL sys.remove_orphan_files('tableId', '2023-12-31 23:59:59')
+ *
+ * -- remove all tables' orphan files in db
+ * CALL sys.remove_orphan_files('databaseName.*', '2023-12-31 23:59:59')
+ *
+ */
+public class ResetConsumerProcedure extends ProcedureBase {
+
+ public static final String IDENTIFIER = "reset_consumer";
+
+ public String[] call(
+ ProcedureContext procedureContext,
+ String tableId,
+ String consumerId,
+ long nextSnapshotId)
+ throws Catalog.TableNotExistException {
+ FileStoreTable fileStoreTable =
+ (FileStoreTable) catalog.getTable(Identifier.fromString(tableId));
+ ConsumerManager consumerManager =
+ new ConsumerManager(
+ fileStoreTable.fileIO(),
+ fileStoreTable.location(),
+ fileStoreTable.snapshotManager().branch());
+ consumerManager.resetConsumer(consumerId, new Consumer(nextSnapshotId));
+
+ return new String[] {"Success"};
+ }
+
+ public String[] call(ProcedureContext procedureContext, String tableId, String consumerId)
+ throws Catalog.TableNotExistException {
+ FileStoreTable fileStoreTable =
+ (FileStoreTable) catalog.getTable(Identifier.fromString(tableId));
+ ConsumerManager consumerManager =
+ new ConsumerManager(
+ fileStoreTable.fileIO(),
+ fileStoreTable.location(),
+ fileStoreTable.snapshotManager().branch());
+ consumerManager.deleteConsumer(consumerId);
+
+ return new String[] {"Success"};
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+}
diff --git a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/RewriteFileIndexProcedure.java b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/RewriteFileIndexProcedure.java
new file mode 100644
index 0000000000000..29ae1b25b57a1
--- /dev/null
+++ b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/RewriteFileIndexProcedure.java
@@ -0,0 +1,138 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.sink.NoneCopyVersionedSerializerTypeSerializerProxy;
+import org.apache.paimon.flink.sink.RewriteFileIndexSink;
+import org.apache.paimon.flink.source.RewriteFileIndexSource;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestEntrySerializer;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.utils.StringUtils;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.GenericTypeInfo;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.procedure.ProcedureContext;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.ParameterUtils.getPartitions;
+
+/** Rewrite file index procedure to re-generated all file index. */
+public class RewriteFileIndexProcedure extends ProcedureBase {
+
+ @Override
+ public String identifier() {
+ return "rewrite_file_index";
+ }
+
+ public String[] call(ProcedureContext procedureContext, String sourceTablePath)
+ throws Exception {
+ return call(procedureContext, sourceTablePath, "");
+ }
+
+ public String[] call(
+ ProcedureContext procedureContext, String sourceTablePath, String partitions)
+ throws Exception {
+
+ StreamExecutionEnvironment env = procedureContext.getExecutionEnvironment();
+ Table table = catalog.getTable(Identifier.fromString(sourceTablePath));
+
+ List
+ * -- reset the new next snapshot id in the consumer
+ * CALL sys.reset_consumer('tableId', 'consumerId', nextSnapshotId)
+ *
+ * -- delete consumer
+ * CALL sys.reset_consumer('tableId', 'consumerId')
+ *
+ */
+public class RollbackToProcedure extends ProcedureBase {
+
+ public static final String IDENTIFIER = "rollback_to";
+
+ public String[] call(ProcedureContext procedureContext, String tableId, long snapshotId)
+ throws Catalog.TableNotExistException {
+ Table table = catalog.getTable(Identifier.fromString(tableId));
+ table.rollbackTo(snapshotId);
+
+ return new String[] {"Success"};
+ }
+
+ public String[] call(ProcedureContext procedureContext, String tableId, String tagName)
+ throws Catalog.TableNotExistException {
+ Table table = catalog.getTable(Identifier.fromString(tableId));
+ table.rollbackTo(tagName);
+
+ return new String[] {"Success"};
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+}
diff --git a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/privilege/GrantPrivilegeToUserProcedure.java b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/privilege/GrantPrivilegeToUserProcedure.java
new file mode 100644
index 0000000000000..e57b364a2f5d1
--- /dev/null
+++ b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/privilege/GrantPrivilegeToUserProcedure.java
@@ -0,0 +1,79 @@
+/*
+ * 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.flink.procedure.privilege;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.privilege.PrivilegeType;
+
+import org.apache.flink.table.procedure.ProcedureContext;
+
+/**
+ * Procedure to grant privilege to a user. Privilege can be granted on the whole catalog, a database
+ * or a table. Only users with {@link PrivilegeType#ADMIN} privilege can perform this operation.
+ * Usage:
+ *
+ *
+ * -- rollback to a snapshot
+ * CALL sys.rollback_to('tableId', snapshotId)
+ *
+ * -- rollback to a tag
+ * CALL sys.rollback_to('tableId', 'tagName')
+ *
+ */
+public class GrantPrivilegeToUserProcedure extends PrivilegeProcedureBase {
+
+ public static final String IDENTIFIER = "grant_privilege_to_user";
+
+ public String[] call(ProcedureContext procedureContext, String user, String privilege) {
+ getPrivilegedCatalog().grantPrivilegeOnCatalog(user, PrivilegeType.valueOf(privilege));
+ return new String[] {
+ String.format("User %s is granted with privilege %s on the catalog.", user, privilege)
+ };
+ }
+
+ public String[] call(
+ ProcedureContext procedureContext, String user, String privilege, String database) {
+ getPrivilegedCatalog()
+ .grantPrivilegeOnDatabase(user, database, PrivilegeType.valueOf(privilege));
+ return new String[] {
+ String.format(
+ "User %s is granted with privilege %s on database %s.",
+ user, privilege, database)
+ };
+ }
+
+ public String[] call(
+ ProcedureContext procedureContext,
+ String user,
+ String privilege,
+ String database,
+ String table) {
+ Identifier identifier = Identifier.create(database, table);
+ getPrivilegedCatalog()
+ .grantPrivilegeOnTable(user, identifier, PrivilegeType.valueOf(privilege));
+ return new String[] {
+ String.format(
+ "User %s is granted with privilege %s on table %s.",
+ user, privilege, identifier)
+ };
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+}
diff --git a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/privilege/RevokePrivilegeFromUserProcedure.java b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/privilege/RevokePrivilegeFromUserProcedure.java
new file mode 100644
index 0000000000000..5f511eaa61b98
--- /dev/null
+++ b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/privilege/RevokePrivilegeFromUserProcedure.java
@@ -0,0 +1,87 @@
+/*
+ * 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.flink.procedure.privilege;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.privilege.PrivilegeType;
+
+import org.apache.flink.table.procedure.ProcedureContext;
+
+/**
+ * Procedure to revoke privilege from a user. Privilege can be revoked from the whole catalog, a
+ * database or a table. Only users with {@link PrivilegeType#ADMIN} privilege can perform this
+ * operation. Usage:
+ *
+ *
+ * CALL sys.grant_privilege_to_user('username', 'privilege')
+ * CALL sys.grant_privilege_to_user('username', 'privilege', 'database')
+ * CALL sys.grant_privilege_to_user('username', 'privilege', 'database', 'table')
+ *
+ */
+public class RevokePrivilegeFromUserProcedure extends PrivilegeProcedureBase {
+
+ public static final String IDENTIFIER = "revoke_privilege_from_user";
+
+ public String[] call(ProcedureContext procedureContext, String user, String privilege) {
+ int count =
+ getPrivilegedCatalog()
+ .revokePrivilegeOnCatalog(user, PrivilegeType.valueOf(privilege));
+ return new String[] {
+ String.format("User %s is revoked with privilege %s on the catalog.", user, privilege),
+ "Number of privileges revoked: " + count
+ };
+ }
+
+ public String[] call(
+ ProcedureContext procedureContext, String user, String privilege, String database) {
+ int count =
+ getPrivilegedCatalog()
+ .revokePrivilegeOnDatabase(
+ user, database, PrivilegeType.valueOf(privilege));
+ return new String[] {
+ String.format(
+ "User %s is revoked with privilege %s on database %s.",
+ user, privilege, database),
+ "Number of privileges revoked: " + count
+ };
+ }
+
+ public String[] call(
+ ProcedureContext procedureContext,
+ String user,
+ String privilege,
+ String database,
+ String table) {
+ Identifier identifier = Identifier.create(database, table);
+ int count =
+ getPrivilegedCatalog()
+ .revokePrivilegeOnTable(user, identifier, PrivilegeType.valueOf(privilege));
+ return new String[] {
+ String.format(
+ "User %s is revoked with privilege %s on table %s.",
+ user, privilege, identifier),
+ "Number of privileges revoked: " + count
+ };
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+}
diff --git a/paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/procedure/ProcedurePositionalArgumentsITCase.java b/paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/procedure/ProcedurePositionalArgumentsITCase.java
index cd9a8b184f3e7..967eb9d65639b 100644
--- a/paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/procedure/ProcedurePositionalArgumentsITCase.java
+++ b/paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/procedure/ProcedurePositionalArgumentsITCase.java
@@ -19,22 +19,28 @@
package org.apache.paimon.flink.procedure;
import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.privilege.NoPrivilegeException;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.function.Executable;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.assertThrows;
/** Ensure that the legacy multiply overloaded CALL with positional arguments can be invoked. */
public class ProcedurePositionalArgumentsITCase extends CatalogITCaseBase {
-
@Test
- public void testCallCompact() {
+ public void testCompactDatabaseAndTable() {
sql(
"CREATE TABLE T ("
+ " k INT,"
@@ -61,6 +67,166 @@ public void testCallCompact() {
assertThatCode(() -> sql("CALL sys.compact('default.T', '', 'zorder', 'k', '','','5s')"))
.message()
.contains("sort compact do not support 'partition_idle_time'.");
+
+ assertThatCode(() -> sql("CALL sys.compact_database('default')"))
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ public void testUserPrivileges() throws Exception {
+ sql(
+ String.format(
+ "CREATE CATALOG mycat WITH (\n"
+ + " 'type' = 'paimon',\n"
+ + " 'warehouse' = '%s'\n"
+ + ")",
+ path));
+ sql("USE CATALOG mycat");
+ sql("CREATE DATABASE mydb");
+ sql("CREATE DATABASE mydb2");
+ sql(
+ "CREATE TABLE mydb.T1 (\n"
+ + " k INT,\n"
+ + " v INT,\n"
+ + " PRIMARY KEY (k) NOT ENFORCED\n"
+ + ")");
+ sql("INSERT INTO mydb.T1 VALUES (1, 10), (2, 20), (3, 30)");
+ sql("CALL sys.init_file_based_privilege('root-passwd')");
+
+ sql(
+ String.format(
+ "CREATE CATALOG anonymouscat WITH (\n"
+ + " 'type' = 'paimon',\n"
+ + " 'warehouse' = '%s'\n"
+ + ")",
+ path));
+
+ sql("USE CATALOG anonymouscat");
+ assertNoPrivilege(() -> sql("INSERT INTO mydb.T1 VALUES (1, 11), (2, 21)"));
+ assertNoPrivilege(() -> collect("SELECT * FROM mydb.T1 ORDER BY k"));
+
+ sql(
+ String.format(
+ "CREATE CATALOG rootcat WITH (\n"
+ + " 'type' = 'paimon',\n"
+ + " 'warehouse' = '%s',\n"
+ + " 'user' = 'root',\n"
+ + " 'password' = 'root-passwd'\n"
+ + ")",
+ path));
+ sql("USE CATALOG rootcat");
+ sql(
+ "CREATE TABLE mydb2.T2 (\n"
+ + " k INT,\n"
+ + " v INT,\n"
+ + " PRIMARY KEY (k) NOT ENFORCED\n"
+ + ")");
+ sql("INSERT INTO mydb2.T2 VALUES (100, 1000), (200, 2000), (300, 3000)");
+ sql("CALL sys.create_privileged_user('test', 'test-passwd')");
+ sql("CALL sys.grant_privilege_to_user('test', 'CREATE_TABLE', 'mydb')");
+ sql("CALL sys.grant_privilege_to_user('test', 'SELECT', 'mydb')");
+ sql("CALL sys.grant_privilege_to_user('test', 'INSERT', 'mydb')");
+
+ sql(
+ String.format(
+ "CREATE CATALOG testcat WITH (\n"
+ + " 'type' = 'paimon',\n"
+ + " 'warehouse' = '%s',\n"
+ + " 'user' = 'test',\n"
+ + " 'password' = 'test-passwd'\n"
+ + ")",
+ path));
+ sql("USE CATALOG testcat");
+ sql("INSERT INTO mydb.T1 VALUES (1, 12), (2, 22)");
+ assertThat(collect("SELECT * FROM mydb.T1 ORDER BY k"))
+ .isEqualTo(Arrays.asList(Row.of(1, 12), Row.of(2, 22), Row.of(3, 30)));
+ sql("CREATE TABLE mydb.S1 ( a INT, b INT )");
+ sql("INSERT INTO mydb.S1 VALUES (1, 100), (2, 200), (3, 300)");
+ assertThat(collect("SELECT * FROM mydb.S1 ORDER BY a"))
+ .isEqualTo(Arrays.asList(Row.of(1, 100), Row.of(2, 200), Row.of(3, 300)));
+ assertNoPrivilege(() -> sql("DROP TABLE mydb.T1"));
+ assertNoPrivilege(() -> sql("ALTER TABLE mydb.T1 RENAME TO mydb.T2"));
+ assertNoPrivilege(() -> sql("DROP TABLE mydb.S1"));
+ assertNoPrivilege(() -> sql("ALTER TABLE mydb.S1 RENAME TO mydb.S2"));
+ assertNoPrivilege(() -> sql("CREATE DATABASE anotherdb"));
+ assertNoPrivilege(() -> sql("DROP DATABASE mydb CASCADE"));
+ assertNoPrivilege(() -> sql("CALL sys.create_privileged_user('test2', 'test2-passwd')"));
+
+ sql("USE CATALOG rootcat");
+ sql("CALL sys.create_privileged_user('test2', 'test2-passwd')");
+ sql("CALL sys.grant_privilege_to_user('test2', 'SELECT', 'mydb2')");
+ sql("CALL sys.grant_privilege_to_user('test2', 'INSERT', 'mydb', 'T1')");
+ sql("CALL sys.grant_privilege_to_user('test2', 'SELECT', 'mydb', 'S1')");
+ sql("CALL sys.grant_privilege_to_user('test2', 'CREATE_DATABASE')");
+
+ sql(
+ String.format(
+ "CREATE CATALOG test2cat WITH (\n"
+ + " 'type' = 'paimon',\n"
+ + " 'warehouse' = '%s',\n"
+ + " 'user' = 'test2',\n"
+ + " 'password' = 'test2-passwd'\n"
+ + ")",
+ path));
+ sql("USE CATALOG test2cat");
+ sql("INSERT INTO mydb.T1 VALUES (1, 13), (2, 23)");
+ assertNoPrivilege(() -> collect("SELECT * FROM mydb.T1 ORDER BY k"));
+ assertNoPrivilege(() -> sql("CREATE TABLE mydb.S2 ( a INT, b INT )"));
+ assertNoPrivilege(() -> sql("INSERT INTO mydb.S1 VALUES (1, 100), (2, 200), (3, 300)"));
+ assertThat(collect("SELECT * FROM mydb.S1 ORDER BY a"))
+ .isEqualTo(Arrays.asList(Row.of(1, 100), Row.of(2, 200), Row.of(3, 300)));
+ assertNoPrivilege(
+ () -> sql("INSERT INTO mydb2.T2 VALUES (100, 1001), (200, 2001), (300, 3001)"));
+ assertThat(collect("SELECT * FROM mydb2.T2 ORDER BY k"))
+ .isEqualTo(Arrays.asList(Row.of(100, 1000), Row.of(200, 2000), Row.of(300, 3000)));
+ sql("CREATE DATABASE anotherdb");
+ assertNoPrivilege(() -> sql("DROP TABLE mydb.T1"));
+ assertNoPrivilege(() -> sql("ALTER TABLE mydb.T1 RENAME TO mydb.T2"));
+ assertNoPrivilege(() -> sql("DROP TABLE mydb.S1"));
+ assertNoPrivilege(() -> sql("ALTER TABLE mydb.S1 RENAME TO mydb.S2"));
+ assertNoPrivilege(() -> sql("DROP DATABASE mydb CASCADE"));
+ assertNoPrivilege(() -> sql("CALL sys.create_privileged_user('test3', 'test3-passwd')"));
+
+ sql("USE CATALOG rootcat");
+ assertThat(collect("SELECT * FROM mydb.T1 ORDER BY k"))
+ .isEqualTo(Arrays.asList(Row.of(1, 13), Row.of(2, 23), Row.of(3, 30)));
+ sql("CALL sys.revoke_privilege_from_user('test2', 'SELECT')");
+ sql("CALL sys.drop_privileged_user('test')");
+
+ sql("USE CATALOG testcat");
+ Exception e =
+ assertThrows(Exception.class, () -> collect("SELECT * FROM mydb.T1 ORDER BY k"));
+ assertThat(e).hasRootCauseMessage("User test not found, or password incorrect.");
+
+ sql("USE CATALOG test2cat");
+ assertNoPrivilege(() -> collect("SELECT * FROM mydb.S1 ORDER BY a"));
+ assertNoPrivilege(() -> collect("SELECT * FROM mydb2.T2 ORDER BY k"));
+ sql("INSERT INTO mydb.T1 VALUES (1, 14), (2, 24)");
+
+ sql("USE CATALOG rootcat");
+ assertThat(collect("SELECT * FROM mydb.T1 ORDER BY k"))
+ .isEqualTo(Arrays.asList(Row.of(1, 14), Row.of(2, 24), Row.of(3, 30)));
+ sql("DROP DATABASE mydb CASCADE");
+ sql("DROP DATABASE mydb2 CASCADE");
+ }
+
+ private List
+ * CALL sys.revoke_privilege_from_user('username', 'privilege')
+ * CALL sys.revoke_privilege_from_user('username', 'privilege', 'database')
+ * CALL sys.revoke_privilege_from_user('username', 'privilege', 'database', 'table')
+ *