[FLINK-15026][sql clientSupport create/drop/alter database in sql client

This closes apache#10419

flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java

@@ -305,6 +305,15 @@ private void callCommand(SqlCommandCall cmdCall) {
  case SOURCE:
  callSource(cmdCall);
  break;
+ case CREATE_DATABASE:
+ callCreateDatabase(cmdCall);
+ break;
+ case DROP_DATABASE:
+ callDropDatabase(cmdCall);
+ break;
+ case ALTER_DATABASE:
+ callAlterDatabase(cmdCall);
+ break;
  default:
  throw new SqlClientException("Unsupported command: " + cmdCall.command);
  }
@@ -594,6 +603,21 @@ private void callSource(SqlCommandCall cmdCall) {
  call.ifPresent(this::callCommand);
  }
 
+ private void callCreateDatabase(SqlCommandCall cmdCall) {
+ final String createDatabaseStmt = cmdCall.operands[0];
+ executor.executeUpdate(sessionId, createDatabaseStmt);
+ }
+
+ private void callDropDatabase(SqlCommandCall cmdCall) {
+ final String dropDatabaseStmt = cmdCall.operands[0];
+ executor.executeUpdate(sessionId, dropDatabaseStmt);
+ }
+
+ private void callAlterDatabase(SqlCommandCall cmdCall) {
+ final String alterDatabaseStmt = cmdCall.operands[0];
+ executor.executeUpdate(sessionId, alterDatabaseStmt);
+ }
+
  // --------------------------------------------------------------------------------------------
 
  private void printExecutionException(Throwable t) {

flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCommandParser.java

@@ -136,10 +136,22 @@ enum SqlCommand {
  return Optional.of(new String[]{operands[0], operands[1]});
  }),
 
+ CREATE_DATABASE(
+ "(CREATE\\s+DATABASE\\s+.*)",
+ SINGLE_OPERAND),
+
+ DROP_DATABASE(
+ "(DROP\\s+DATABASE\\s+.*)",
+ SINGLE_OPERAND),
+
  DROP_VIEW(
  "DROP\\s+VIEW\\s+(.*)",
  SINGLE_OPERAND),
 
+ ALTER_DATABASE(
+ "(ALTER\\s+DATABASE\\s+.*)",
+ SINGLE_OPERAND),
+
  SET(
  "SET(\\s+(\\S+)\\s*=(.*))?", // whitespace is only ignored on the left side of '='
  (operands) -> {

flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java

@@ -544,6 +544,11 @@ private <C> ProgramTargetDescriptor executeUpdateInternal(
  String statement) {
  applyUpdate(context, context.getTableEnvironment(), context.getQueryConfig(), statement);
 
+ //Todo: we should refactor following condition after TableEnvironment has support submit job directly.
+ if (!statement.trim().matches("(INSERT\\s+INTO.*)")) {
+ return null;
+ }
+
  // create job graph with dependencies
  final String jobName = sessionId + ": " + statement;
  final JobGraph jobGraph;

flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/SqlCommandParserTest.java

@@ -77,6 +77,12 @@ public void testCommands() {
  testValidSqlCommand("USE CATALOG default", new SqlCommandCall(SqlCommand.USE_CATALOG, new String[]{"default"}));
  testValidSqlCommand("use default", new SqlCommandCall(SqlCommand.USE, new String[] {"default"}));
  testInvalidSqlCommand("use catalog");
+ testValidSqlCommand("create database db1",
+ new SqlCommandCall(SqlCommand.CREATE_DATABASE, new String[] {"create database db1"}));
+ testValidSqlCommand("drop database db1",
+ new SqlCommandCall(SqlCommand.DROP_DATABASE, new String[] {"drop database db1"}));
+ testValidSqlCommand("alter database db1 set ('k1' = 'a')",
+ new SqlCommandCall(SqlCommand.ALTER_DATABASE, new String[] {"alter database db1 set ('k1' = 'a')"}));
  }
 
  private void testInvalidSqlCommand(String stmt) {

flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java

@@ -211,6 +211,63 @@ public void testListDatabases() throws Exception {
  executor.closeSession(sessionId);
  }
 
+ @Test
+ public void testCreateDatabase() throws Exception {
+ final Executor executor = createDefaultExecutor(clusterClient);
+ final SessionContext session = new SessionContext("test-session", new Environment());
+ String sessionId = executor.openSession(session);
+ assertEquals("test-session", sessionId);
+
+ executor.executeUpdate(sessionId, "create database db1");
+
+ final List<String> actualDatabases = executor.listDatabases(sessionId);
+ final List<String> expectedDatabases = Arrays.asList("default_database", "db1");
+ assertEquals(expectedDatabases, actualDatabases);
+
+ executor.closeSession(sessionId);
+ }
+
+ @Test
+ public void testDropDatabase() throws Exception {
+ final Executor executor = createDefaultExecutor(clusterClient);
+ final SessionContext session = new SessionContext("test-session", new Environment());
+ String sessionId = executor.openSession(session);
+ assertEquals("test-session", sessionId);
+
+ executor.executeUpdate(sessionId, "create database db1");
+
+ List<String> actualDatabases = executor.listDatabases(sessionId);
+ List<String> expectedDatabases = Arrays.asList("default_database", "db1");
+ assertEquals(expectedDatabases, actualDatabases);
+
+ executor.executeUpdate(sessionId, "drop database if exists db1");
+
+ actualDatabases = executor.listDatabases(sessionId);
+ expectedDatabases = Arrays.asList("default_database");
+ assertEquals(expectedDatabases, actualDatabases);
+
+ executor.closeSession(sessionId);
+ }
+
+ @Test
+ public void testAlterDatabase() throws Exception {
+ final Executor executor = createDefaultExecutor(clusterClient);
+ final SessionContext session = new SessionContext("test-session", new Environment());
+ String sessionId = executor.openSession(session);
+ assertEquals("test-session", sessionId);
+
+ executor.executeUpdate(sessionId, "create database db1 comment 'db1_comment' with ('k1' = 'v1')");
+
+ executor.executeUpdate(sessionId, "alter database db1 set ('k1' = 'a', 'k2' = 'b')");
+
+ final List<String> actualDatabases = executor.listDatabases(sessionId);
+ final List<String> expectedDatabases = Arrays.asList("default_database", "db1");
+ assertEquals(expectedDatabases, actualDatabases);
+ //todo: we should compare the new db1 properties after we support describe database in LocalExecutor.
+
+ executor.closeSession(sessionId);
+ }
+
  @Test
  public void testListTables() throws Exception {
  final Executor executor = createDefaultExecutor(clusterClient);