You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2020/05/19 10:41:28 UTC
[flink] branch release-1.11 updated: [FLINK-17728] [sql-client] sql
client supports parser statements via sql parser
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.11 by this push:
new 6178496 [FLINK-17728] [sql-client] sql client supports parser statements via sql parser
6178496 is described below
commit 6178496be82c1b5aac6647693d8a1acd28fee6e2
Author: godfrey he <go...@163.com>
AuthorDate: Tue May 19 18:39:39 2020 +0800
[FLINK-17728] [sql-client] sql client supports parser statements via sql parser
This closes #12188
---
.../apache/flink/table/client/cli/CliClient.java | 6 +-
.../flink/table/client/cli/SqlCommandParser.java | 263 ++++++++----
.../flink/table/client/cli/SqlCompleter.java | 23 +-
.../flink/table/client/gateway/Executor.java | 5 +-
.../table/client/gateway/local/LocalExecutor.java | 13 +-
.../flink/table/client/cli/CliClientTest.java | 22 +-
.../flink/table/client/cli/CliResultViewTest.java | 3 +-
.../table/client/cli/SqlCommandParserTest.java | 467 +++++++++++++++------
.../flink/table/client/cli/TestingExecutor.java | 10 +-
.../table/client/cli/utils/SqlParserHelper.java | 55 +++
10 files changed, 611 insertions(+), 256 deletions(-)
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java
index 0a1111f..ffb2d0e 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.client.cli;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.client.SqlClientException;
import org.apache.flink.table.client.cli.SqlCommandParser.SqlCommandCall;
@@ -252,7 +253,7 @@ public class CliClient {
// --------------------------------------------------------------------------------------------
private Optional<SqlCommandCall> parseCommand(String line) {
- final Optional<SqlCommandCall> parsedLine = SqlCommandParser.parse(line);
+ final Optional<SqlCommandCall> parsedLine = SqlCommandParser.parse(executor.getSqlParser(sessionId), line);
if (!parsedLine.isPresent()) {
printError(CliStrings.MESSAGE_UNKNOWN_SQL);
}
@@ -517,7 +518,8 @@ public class CliClient {
private void callExplain(SqlCommandCall cmdCall) {
final String explanation;
try {
- explanation = executor.explainStatement(sessionId, cmdCall.operands[0]);
+ TableResult tableResult = executor.executeSql(sessionId, cmdCall.operands[0]);
+ explanation = tableResult.collect().next().getField(0).toString();
} catch (SqlExecutionException e) {
printExecutionException(e);
return;
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCommandParser.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCommandParser.java
index 5a9f1be..5b0a88c 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCommandParser.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCommandParser.java
@@ -18,7 +18,40 @@
package org.apache.flink.table.client.cli;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.delegation.Parser;
+import org.apache.flink.table.operations.CatalogSinkModifyOperation;
+import org.apache.flink.table.operations.DescribeTableOperation;
+import org.apache.flink.table.operations.ExplainOperation;
+import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.operations.QueryOperation;
+import org.apache.flink.table.operations.ShowCatalogsOperation;
+import org.apache.flink.table.operations.ShowDatabasesOperation;
+import org.apache.flink.table.operations.ShowFunctionsOperation;
+import org.apache.flink.table.operations.ShowTablesOperation;
+import org.apache.flink.table.operations.UseCatalogOperation;
+import org.apache.flink.table.operations.UseDatabaseOperation;
+import org.apache.flink.table.operations.ddl.AlterCatalogFunctionOperation;
+import org.apache.flink.table.operations.ddl.AlterDatabaseOperation;
+import org.apache.flink.table.operations.ddl.AlterTableOperation;
+import org.apache.flink.table.operations.ddl.CreateCatalogFunctionOperation;
+import org.apache.flink.table.operations.ddl.CreateCatalogOperation;
+import org.apache.flink.table.operations.ddl.CreateDatabaseOperation;
+import org.apache.flink.table.operations.ddl.CreateTableOperation;
+import org.apache.flink.table.operations.ddl.CreateTempSystemFunctionOperation;
+import org.apache.flink.table.operations.ddl.CreateViewOperation;
+import org.apache.flink.table.operations.ddl.DropCatalogFunctionOperation;
+import org.apache.flink.table.operations.ddl.DropCatalogOperation;
+import org.apache.flink.table.operations.ddl.DropDatabaseOperation;
+import org.apache.flink.table.operations.ddl.DropTableOperation;
+import org.apache.flink.table.operations.ddl.DropTempSystemFunctionOperation;
+import org.apache.flink.table.operations.ddl.DropViewOperation;
+
+import javax.annotation.Nullable;
+
import java.util.Arrays;
+import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.function.Function;
@@ -34,7 +67,7 @@ public final class SqlCommandParser {
// private
}
- public static Optional<SqlCommandCall> parse(String stmt) {
+ public static Optional<SqlCommandCall> parse(Parser sqlParser, String stmt) {
// normalize
stmt = stmt.trim();
// remove ';' at the end
@@ -42,16 +75,123 @@ public final class SqlCommandParser {
stmt = stmt.substring(0, stmt.length() - 1).trim();
}
- // parse
+ // parse statement via sql parser first
+ Optional<SqlCommandCall> callOpt = parseBySqlParser(sqlParser, stmt);
+ if (callOpt.isPresent()) {
+ return callOpt;
+ } else {
+ return parseByRegexMatching(stmt);
+ }
+ }
+
+ private static Optional<SqlCommandCall> parseBySqlParser(Parser sqlParser, String stmt) {
+ List<Operation> operations;
+ try {
+ operations = sqlParser.parse(stmt);
+ } catch (Throwable e) {
+ if (e instanceof ValidationException) {
+ // can be parsed via sql parser, but is not validated.
+ // throw exception directly
+ throw new SqlExecutionException("Invalidate SQL statement.", e);
+ }
+ return Optional.empty();
+ }
+ if (operations.size() != 1) {
+ throw new SqlExecutionException("Only single statement is supported now.");
+ }
+
+ final SqlCommand cmd;
+ String[] operands = new String[] { stmt };
+ Operation operation = operations.get(0);
+ if (operation instanceof CatalogSinkModifyOperation) {
+ boolean overwrite = ((CatalogSinkModifyOperation) operation).isOverwrite();
+ cmd = overwrite ? SqlCommand.INSERT_OVERWRITE : SqlCommand.INSERT_INTO;
+ } else if (operation instanceof CreateTableOperation) {
+ cmd = SqlCommand.CREATE_TABLE;
+ } else if (operation instanceof DropTableOperation) {
+ cmd = SqlCommand.DROP_TABLE;
+ } else if (operation instanceof AlterTableOperation) {
+ cmd = SqlCommand.ALTER_TABLE;
+ } else if (operation instanceof CreateViewOperation) {
+ cmd = SqlCommand.CREATE_VIEW;
+ CreateViewOperation op = (CreateViewOperation) operation;
+ operands = new String[] { op.getViewIdentifier().asSerializableString(),
+ op.getCatalogView().getOriginalQuery() };
+ } else if (operation instanceof DropViewOperation) {
+ cmd = SqlCommand.DROP_VIEW;
+ operands = new String[] { ((DropViewOperation) operation).getViewIdentifier().asSerializableString() };
+ } else if (operation instanceof CreateDatabaseOperation) {
+ cmd = SqlCommand.CREATE_DATABASE;
+ } else if (operation instanceof DropDatabaseOperation) {
+ cmd = SqlCommand.DROP_DATABASE;
+ } else if (operation instanceof AlterDatabaseOperation) {
+ cmd = SqlCommand.ALTER_DATABASE;
+ } else if (operation instanceof CreateCatalogOperation) {
+ cmd = SqlCommand.CREATE_CATALOG;
+ } else if (operation instanceof DropCatalogOperation) {
+ cmd = SqlCommand.DROP_CATALOG;
+ } else if (operation instanceof UseCatalogOperation) {
+ cmd = SqlCommand.USE_CATALOG;
+ operands = new String[] { String.format("`%s`", ((UseCatalogOperation) operation).getCatalogName()) };
+ } else if (operation instanceof UseDatabaseOperation) {
+ cmd = SqlCommand.USE;
+ UseDatabaseOperation op = ((UseDatabaseOperation) operation);
+ operands = new String[] { String.format("`%s`.`%s`", op.getCatalogName(), op.getDatabaseName()) };
+ } else if (operation instanceof ShowCatalogsOperation) {
+ cmd = SqlCommand.SHOW_CATALOGS;
+ operands = new String[0];
+ } else if (operation instanceof ShowDatabasesOperation) {
+ cmd = SqlCommand.SHOW_DATABASES;
+ operands = new String[0];
+ } else if (operation instanceof ShowTablesOperation) {
+ cmd = SqlCommand.SHOW_TABLES;
+ operands = new String[0];
+ } else if (operation instanceof ShowFunctionsOperation) {
+ cmd = SqlCommand.SHOW_FUNCTIONS;
+ operands = new String[0];
+ } else if (operation instanceof CreateCatalogFunctionOperation ||
+ operation instanceof CreateTempSystemFunctionOperation) {
+ cmd = SqlCommand.CREATE_FUNCTION;
+ } else if (operation instanceof DropCatalogFunctionOperation ||
+ operation instanceof DropTempSystemFunctionOperation) {
+ cmd = SqlCommand.DROP_FUNCTION;
+ } else if (operation instanceof AlterCatalogFunctionOperation) {
+ cmd = SqlCommand.ALTER_FUNCTION;
+ } else if (operation instanceof ExplainOperation) {
+ cmd = SqlCommand.EXPLAIN;
+ } else if (operation instanceof DescribeTableOperation) {
+ cmd = SqlCommand.DESCRIBE;
+ operands = new String[] { ((DescribeTableOperation) operation).getSqlIdentifier().asSerializableString() };
+ } else if (operation instanceof QueryOperation) {
+ cmd = SqlCommand.SELECT;
+ } else {
+ cmd = null;
+ }
+
+ return cmd == null ? Optional.empty() : Optional.of(new SqlCommandCall(cmd, operands));
+ }
+
+ private static Optional<SqlCommandCall> parseByRegexMatching(String stmt) {
+ // parse statement via regex matching
for (SqlCommand cmd : SqlCommand.values()) {
- final Matcher matcher = cmd.pattern.matcher(stmt);
- if (matcher.matches()) {
- final String[] groups = new String[matcher.groupCount()];
- for (int i = 0; i < groups.length; i++) {
- groups[i] = matcher.group(i + 1);
+ if (cmd.hasRegexPattern()) {
+ final Matcher matcher = cmd.pattern.matcher(stmt);
+ if (matcher.matches()) {
+ final String[] groups = new String[matcher.groupCount()];
+ for (int i = 0; i < groups.length; i++) {
+ groups[i] = matcher.group(i + 1);
+ }
+ return cmd.operandConverter.apply(groups)
+ .map((operands) -> {
+ String[] newOperands = operands;
+ if (cmd == SqlCommand.EXPLAIN) {
+ // convert `explain xx` to `explain plan for xx`
+ // which can execute through executeSql method
+ newOperands = new String[] { "EXPLAIN PLAN FOR " + operands[0] };
+ }
+ return new SqlCommandCall(cmd, newOperands);
+ });
}
- return cmd.operandConverter.apply(groups)
- .map((operands) -> new SqlCommandCall(cmd, operands));
}
}
return Optional.empty();
@@ -83,102 +223,66 @@ public final class SqlCommandParser {
"HELP",
NO_OPERANDS),
- SHOW_CATALOGS(
- "SHOW\\s+CATALOGS",
- NO_OPERANDS),
+ SHOW_CATALOGS,
- SHOW_DATABASES(
- "SHOW\\s+DATABASES",
- NO_OPERANDS),
+ SHOW_DATABASES,
- SHOW_TABLES(
- "SHOW\\s+TABLES",
- NO_OPERANDS),
+ SHOW_TABLES,
- SHOW_FUNCTIONS(
- "SHOW\\s+FUNCTIONS",
- NO_OPERANDS),
+ SHOW_FUNCTIONS,
+ // FLINK-17396
SHOW_MODULES(
"SHOW\\s+MODULES",
NO_OPERANDS),
- USE_CATALOG(
- "USE\\s+CATALOG\\s+(.*)",
- SINGLE_OPERAND),
+ USE_CATALOG,
- USE(
- "USE\\s+(?!CATALOG)(.*)",
- SINGLE_OPERAND),
+ USE,
- CREATE_CATALOG(
- "(CREATE\\s+CATALOG\\s+.*)",
- SINGLE_OPERAND),
+ CREATE_CATALOG,
- DROP_CATALOG(
- "(DROP\\s+CATALOG\\s+.*)",
- SINGLE_OPERAND),
+ DROP_CATALOG,
DESC(
"DESC\\s+(.*)",
SINGLE_OPERAND),
- DESCRIBE(
- "DESCRIBE\\s+(.*)",
- SINGLE_OPERAND),
+ DESCRIBE,
+ // supports both `explain xx` and `explain plan for xx` now
+ // TODO should keep `explain xx` ?
EXPLAIN(
"EXPLAIN\\s+(.*)",
SINGLE_OPERAND),
- SELECT(
- "(WITH.*SELECT.*|SELECT.*)",
- SINGLE_OPERAND),
+ CREATE_DATABASE,
- INSERT_INTO(
- "(INSERT\\s+INTO.*)",
- SINGLE_OPERAND),
+ DROP_DATABASE,
- INSERT_OVERWRITE(
- "(INSERT\\s+OVERWRITE.*)",
- SINGLE_OPERAND),
+ ALTER_DATABASE,
- CREATE_TABLE("(CREATE\\s+TABLE\\s+.*)", SINGLE_OPERAND),
+ CREATE_TABLE,
- DROP_TABLE("(DROP\\s+TABLE\\s+.*)", SINGLE_OPERAND),
+ DROP_TABLE,
- ALTER_TABLE(
- "(ALTER\\s+TABLE\\s+.*)",
- SINGLE_OPERAND),
+ ALTER_TABLE,
- CREATE_VIEW(
- "CREATE\\s+VIEW\\s+(\\S+)\\s+AS\\s+(.*)",
- (operands) -> {
- if (operands.length < 2) {
- return Optional.empty();
- }
- return Optional.of(new String[]{operands[0], operands[1]});
- }),
+ CREATE_VIEW,
- DROP_VIEW("DROP\\s+VIEW\\s+(.*)", SINGLE_OPERAND),
+ DROP_VIEW,
- CREATE_FUNCTION("(CREATE.+FUNCTION\\s+(\\S+)\\s+AS.+)", SINGLE_OPERAND),
+ CREATE_FUNCTION,
- DROP_FUNCTION("(DROP.+FUNCTION\\s+.+)", SINGLE_OPERAND),
+ DROP_FUNCTION,
- ALTER_FUNCTION("(ALTER.+FUNCTION\\s+.+)", SINGLE_OPERAND),
+ ALTER_FUNCTION,
- CREATE_DATABASE(
- "(CREATE\\s+DATABASE\\s+.*)",
- SINGLE_OPERAND),
+ SELECT,
- DROP_DATABASE(
- "(DROP\\s+DATABASE\\s+.*)",
- SINGLE_OPERAND),
+ INSERT_INTO,
- ALTER_DATABASE(
- "(ALTER\\s+DATABASE\\s+.*)",
- SINGLE_OPERAND),
+ INSERT_OVERWRITE,
SET(
"SET(\\s+(\\S+)\\s*=(.*))?", // whitespace is only ignored on the left side of '='
@@ -199,8 +303,13 @@ public final class SqlCommandParser {
"SOURCE\\s+(.*)",
SINGLE_OPERAND);
- public final Pattern pattern;
- public final Function<String[], Optional<String[]>> operandConverter;
+ public final @Nullable Pattern pattern;
+ public final @Nullable Function<String[], Optional<String[]>> operandConverter;
+
+ SqlCommand() {
+ this.pattern = null;
+ this.operandConverter = null;
+ }
SqlCommand(String matchingRegex, Function<String[], Optional<String[]>> operandConverter) {
this.pattern = Pattern.compile(matchingRegex, DEFAULT_PATTERN_FLAGS);
@@ -215,6 +324,10 @@ public final class SqlCommandParser {
public boolean hasOperands() {
return operandConverter != NO_OPERANDS;
}
+
+ public boolean hasRegexPattern() {
+ return pattern != null;
+ }
}
/**
@@ -229,10 +342,6 @@ public final class SqlCommandParser {
this.operands = operands;
}
- public SqlCommandCall(SqlCommand command) {
- this(command, new String[0]);
- }
-
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCompleter.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCompleter.java
index 5dd2bc7..c9edd69 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCompleter.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCompleter.java
@@ -29,6 +29,7 @@ import org.jline.utils.AttributedString;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Arrays;
import java.util.List;
/**
@@ -97,17 +98,15 @@ public class SqlCompleter implements Completer {
}
private static String[] getCommandHints() {
- final SqlCommandParser.SqlCommand[] commands = SqlCommandParser.SqlCommand.values();
- final String[] hints = new String[commands.length];
- for (int i = 0; i < commands.length; i++) {
- final SqlCommandParser.SqlCommand command = commands[i];
- // add final ";" for convenience if no operands can follow
- if (command.hasOperands()) {
- hints[i] = command.toString();
- } else {
- hints[i] = command.toString() + ";";
- }
- }
- return hints;
+ return Arrays.stream(SqlCommandParser.SqlCommand.values())
+ .filter(SqlCommandParser.SqlCommand::hasRegexPattern)
+ .map(cmd -> {
+ // add final ";" for convenience if no operands can follow
+ if (cmd.hasOperands()) {
+ return cmd.toString();
+ } else {
+ return cmd.toString() + ";";
+ }
+ }).toArray(String[]::new);
}
}
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java
index 365e9f5..f35a77a 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.client.config.entries.ViewEntry;
+import org.apache.flink.table.delegation.Parser;
import org.apache.flink.types.Row;
import java.util.List;
@@ -166,9 +167,9 @@ public interface Executor {
TableSchema getTableSchema(String sessionId, String name) throws SqlExecutionException;
/**
- * Returns a string-based explanation about AST and execution plan of the given statement.
+ * Returns a sql parser instance.
*/
- String explainStatement(String sessionId, String statement) throws SqlExecutionException;
+ Parser getSqlParser(String sessionId);
/**
* Returns a list of completion hints for the given statement at the given position.
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
index ff608ce..fd78712 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
@@ -55,6 +55,7 @@ import org.apache.flink.table.client.gateway.TypedResult;
import org.apache.flink.table.client.gateway.local.result.ChangelogResult;
import org.apache.flink.table.client.gateway.local.result.DynamicResult;
import org.apache.flink.table.client.gateway.local.result.MaterializedResult;
+import org.apache.flink.table.delegation.Parser;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.utils.LogicalTypeUtils;
import org.apache.flink.table.types.utils.DataTypeUtils;
@@ -79,7 +80,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.function.Supplier;
import java.util.regex.Pattern;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -457,17 +457,10 @@ public class LocalExecutor implements Executor {
}
@Override
- public String explainStatement(String sessionId, String statement) throws SqlExecutionException {
+ public Parser getSqlParser(String sessionId) {
final ExecutionContext<?> context = getExecutionContext(sessionId);
final TableEnvironment tableEnv = context.getTableEnvironment();
- // translate
- try {
- final Table table = createTable(context, tableEnv, statement);
- return context.wrapClassLoader((Supplier<String>) table::explain);
- } catch (Throwable t) {
- // catch everything such that the query does not crash the executor
- throw new SqlExecutionException("Invalid SQL statement.", t);
- }
+ return ((TableEnvironmentInternal) tableEnv).getParser();
}
@Override
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java
index c785cab..8a7ca75 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.client.cli.utils.SqlParserHelper;
import org.apache.flink.table.client.cli.utils.TerminalUtils;
import org.apache.flink.table.client.config.Environment;
import org.apache.flink.table.client.config.entries.ViewEntry;
@@ -91,18 +92,18 @@ public class CliClientTest extends TestLogger {
@Test
public void testSqlCompletion() throws IOException {
- verifySqlCompletion("", 0, Arrays.asList("SELECT", "QUIT;", "RESET;"), Collections.emptyList());
- verifySqlCompletion("SELEC", 5, Collections.singletonList("SELECT"), Collections.singletonList("QUIT;"));
- verifySqlCompletion("SELE", 0, Collections.singletonList("SELECT"), Collections.singletonList("QUIT;"));
+ verifySqlCompletion("", 0, Arrays.asList("SOURCE", "QUIT;", "RESET;"), Collections.emptyList());
+ verifySqlCompletion("SELE", 5, Collections.singletonList("HintA"), Collections.singletonList("QUIT;"));
+ verifySqlCompletion("SOUR", 5, Collections.singletonList("SOURCE"), Collections.singletonList("QUIT;"));
+ verifySqlCompletion("SOUR", 0, Collections.singletonList("SOURCE"), Collections.singletonList("QUIT;"));
verifySqlCompletion("QU", 2, Collections.singletonList("QUIT;"), Collections.singletonList("SELECT"));
verifySqlCompletion("qu", 2, Collections.singletonList("QUIT;"), Collections.singletonList("SELECT"));
verifySqlCompletion(" qu", 2, Collections.singletonList("QUIT;"), Collections.singletonList("SELECT"));
verifySqlCompletion("set ", 3, Collections.emptyList(), Collections.singletonList("SET"));
verifySqlCompletion("show t ", 6, Collections.emptyList(), Collections.singletonList("SET"));
- verifySqlCompletion("use cat", 7, Collections.singletonList("CATALOG"), Collections.singletonList("USE CATALOG"));
- verifySqlCompletion("use ", 4, Collections.singletonList("CATALOG"), Collections.singletonList("USE CATALOG"));
- verifySqlCompletion("use catalog", 11, Collections.emptyList(), Collections.singletonList("CATALOG"));
-
+ verifySqlCompletion("show", 7, Collections.singletonList("MODULES;"), Collections.singletonList("QUIT;"));
+ verifySqlCompletion("show ", 4, Collections.singletonList("MODULES;"), Collections.singletonList("QUIT;"));
+ verifySqlCompletion("show modules", 13, Collections.emptyList(), Collections.singletonList("QUIT;"));
}
@Test
@@ -265,16 +266,17 @@ public class CliClientTest extends TestLogger {
public String receivedStatement;
public int receivedPosition;
private final Map<String, SessionContext> sessionMap = new HashMap<>();
+ private final SqlParserHelper helper = new SqlParserHelper();
@Override
public void start() throws SqlExecutionException {
- // nothing to do
}
@Override
public String openSession(SessionContext session) throws SqlExecutionException {
String sessionId = UUID.randomUUID().toString();
sessionMap.put(sessionId, session);
+ helper.registerTables();
return sessionId;
}
@@ -374,8 +376,8 @@ public class CliClientTest extends TestLogger {
}
@Override
- public String explainStatement(String sessionId, String statement) throws SqlExecutionException {
- return null;
+ public org.apache.flink.table.delegation.Parser getSqlParser(String sessionId) {
+ return helper.getSqlParser();
}
@Override
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliResultViewTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliResultViewTest.java
index 11decbd..b2e4627 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliResultViewTest.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliResultViewTest.java
@@ -30,6 +30,7 @@ import org.apache.flink.table.client.gateway.ResultDescriptor;
import org.apache.flink.table.client.gateway.SessionContext;
import org.apache.flink.table.client.gateway.SqlExecutionException;
import org.apache.flink.table.client.gateway.TypedResult;
+import org.apache.flink.table.delegation.Parser;
import org.apache.flink.types.Row;
import org.jline.utils.AttributedString;
@@ -231,7 +232,7 @@ public class CliResultViewTest {
}
@Override
- public String explainStatement(String sessionId, String statement) throws SqlExecutionException {
+ public Parser getSqlParser(String sessionId) {
return null;
}
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/SqlCommandParserTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/SqlCommandParserTest.java
index 189ed72..c3e9250 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/SqlCommandParserTest.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/SqlCommandParserTest.java
@@ -18,14 +18,26 @@
package org.apache.flink.table.client.cli;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.client.cli.SqlCommandParser.SqlCommand;
import org.apache.flink.table.client.cli.SqlCommandParser.SqlCommandCall;
+import org.apache.flink.table.client.cli.utils.SqlParserHelper;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.delegation.Parser;
+import org.junit.Before;
import org.junit.Test;
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+import java.util.List;
import java.util.Optional;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
/**
@@ -33,153 +45,328 @@ import static org.junit.Assert.fail;
*/
public class SqlCommandParserTest {
+ private Parser parser;
+
+ @Before
+ public void setup() {
+ SqlParserHelper helper = new SqlParserHelper();
+ helper.registerTables();
+ parser = helper.getSqlParser();
+ }
+
@Test
- public void testCommands() {
- testValidSqlCommand("QUIT;", new SqlCommandCall(SqlCommand.QUIT));
- testValidSqlCommand("eXiT", new SqlCommandCall(SqlCommand.QUIT));
- testValidSqlCommand("CLEAR", new SqlCommandCall(SqlCommand.CLEAR));
- testValidSqlCommand("SHOW TABLES", new SqlCommandCall(SqlCommand.SHOW_TABLES));
- testValidSqlCommand(" SHOW TABLES ", new SqlCommandCall(SqlCommand.SHOW_TABLES));
- testValidSqlCommand("SHOW FUNCTIONS", new SqlCommandCall(SqlCommand.SHOW_FUNCTIONS));
- testValidSqlCommand(" SHOW FUNCTIONS ", new SqlCommandCall(SqlCommand.SHOW_FUNCTIONS));
- testValidSqlCommand("DESC MyTable", new SqlCommandCall(SqlCommand.DESC, new String[]{"MyTable"}));
- testValidSqlCommand("DESC MyTable ", new SqlCommandCall(SqlCommand.DESC, new String[]{"MyTable"}));
- testInvalidSqlCommand("DESC "); // no table name
- testValidSqlCommand("DESCRIBE MyTable", new SqlCommandCall(SqlCommand.DESCRIBE, new String[]{"MyTable"}));
- testValidSqlCommand("DESCRIBE MyTable ", new SqlCommandCall(SqlCommand.DESCRIBE, new String[]{"MyTable"}));
- testInvalidSqlCommand("DESCRIBE "); // no table name
- testValidSqlCommand(
- "EXPLAIN SELECT complicated FROM table",
- new SqlCommandCall(SqlCommand.EXPLAIN, new String[]{"SELECT complicated FROM table"}));
- testInvalidSqlCommand("EXPLAIN "); // no query
- testValidSqlCommand(
- "SELECT complicated FROM table",
- new SqlCommandCall(SqlCommand.SELECT, new String[]{"SELECT complicated FROM table"}));
- testValidSqlCommand(
- " SELECT complicated FROM table ",
- new SqlCommandCall(SqlCommand.SELECT, new String[]{"SELECT complicated FROM table"}));
- testValidSqlCommand(
- "WITH t as (select complicated from table) select complicated from t",
- new SqlCommandCall(SqlCommand.SELECT, new String[]{"WITH t as (select complicated from table) select complicated from t"}));
- testValidSqlCommand(
- " WITH t as (select complicated from table) select complicated from t ",
- new SqlCommandCall(SqlCommand.SELECT, new String[]{"WITH t as (select complicated from table) select complicated from t"}));
- testValidSqlCommand(
- "INSERT INTO other SELECT 1+1",
- new SqlCommandCall(SqlCommand.INSERT_INTO, new String[]{"INSERT INTO other SELECT 1+1"}));
- testValidSqlCommand(
- "INSERT OVERWRITE other SELECT 1+1",
- new SqlCommandCall(SqlCommand.INSERT_OVERWRITE, new String[]{"INSERT OVERWRITE other SELECT 1+1"}));
- testValidSqlCommand(
- "CREATE VIEW x AS SELECT 1+1",
- new SqlCommandCall(SqlCommand.CREATE_VIEW, new String[]{"x", "SELECT 1+1"}));
- testValidSqlCommand(
- "CREATE VIEW MyTable AS SELECT 1+1 FROM y",
- new SqlCommandCall(SqlCommand.CREATE_VIEW, new String[]{"MyTable", "SELECT 1+1 FROM y"}));
- testInvalidSqlCommand("CREATE VIEW x SELECT 1+1"); // missing AS
- testValidSqlCommand("DROP VIEW MyTable", new SqlCommandCall(SqlCommand.DROP_VIEW, new String[]{"MyTable"}));
- testValidSqlCommand("DROP VIEW MyTable", new SqlCommandCall(SqlCommand.DROP_VIEW, new String[]{"MyTable"}));
- testInvalidSqlCommand("DROP VIEW");
- testValidSqlCommand("SET", new SqlCommandCall(SqlCommand.SET));
- testValidSqlCommand("SET x=y", new SqlCommandCall(SqlCommand.SET, new String[] {"x", "y"}));
- testValidSqlCommand("SET x = y", new SqlCommandCall(SqlCommand.SET, new String[] {"x", " y"}));
- testValidSqlCommand("reset;", new SqlCommandCall(SqlCommand.RESET));
- testValidSqlCommand("source /my/file", new SqlCommandCall(SqlCommand.SOURCE, new String[] {"/my/file"}));
- testInvalidSqlCommand("source"); // missing path
- testValidSqlCommand("create CATALOG c1",
- new SqlCommandCall(SqlCommand.CREATE_CATALOG, new String[]{"create CATALOG c1"}));
- testValidSqlCommand("create CATALOG c1 WITH ('k'='v')",
- new SqlCommandCall(SqlCommand.CREATE_CATALOG, new String[]{"create CATALOG c1 WITH ('k'='v')"}));
- testValidSqlCommand("drop CATALOG c1",
- new SqlCommandCall(SqlCommand.DROP_CATALOG, new String[]{"drop CATALOG c1"}));
- 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')"}));
- testValidSqlCommand("alter table cat1.db1.tb1 rename to tb2",
- new SqlCommandCall(SqlCommand.ALTER_TABLE, new String[]{"alter table cat1.db1.tb1 rename to tb2"}));
- testValidSqlCommand("alter table cat1.db1.tb1 set ('k1'='v1', 'k2'='v2')",
- new SqlCommandCall(SqlCommand.ALTER_TABLE,
- new String[]{"alter table cat1.db1.tb1 set ('k1'='v1', 'k2'='v2')"}));
- // Test create table.
- testInvalidSqlCommand("CREATE tables");
- testInvalidSqlCommand("CREATE tables");
- testValidSqlCommand("create Table hello", new SqlCommandCall(SqlCommand.CREATE_TABLE, new String[]{"create Table hello"}));
- testValidSqlCommand("create Table hello(a int)", new SqlCommandCall(SqlCommand.CREATE_TABLE, new String[]{"create Table hello(a int)"}));
- testValidSqlCommand(" CREATE TABLE hello(a int)", new SqlCommandCall(SqlCommand.CREATE_TABLE, new String[]{"CREATE TABLE hello(a int)"}));
- testValidSqlCommand("CREATE TABLE T(\n"
- + " a int,\n"
- + " b varchar(20),\n"
- + " c as my_udf(b),\n"
- + " watermark for b as my_udf(b, 1) - INTERVAL '5' second\n"
- + ") WITH (\n"
- + " 'k1' = 'v1',\n"
- + " 'k2' = 'v2')\n",
- new SqlCommandCall(SqlCommand.CREATE_TABLE, new String[] {"CREATE TABLE T(\n"
- + " a int,\n"
- + " b varchar(20),\n"
- + " c as my_udf(b),\n"
- + " watermark for b as my_udf(b, 1) - INTERVAL '5' second\n"
- + ") WITH (\n"
- + " 'k1' = 'v1',\n"
- + " 'k2' = 'v2')"}));
- // Test drop table.
- testInvalidSqlCommand("DROP table");
- testInvalidSqlCommand("DROP tables");
- testInvalidSqlCommand("DROP tables");
- testValidSqlCommand("DROP TABLE t1", new SqlCommandCall(SqlCommand.DROP_TABLE, new String[]{"DROP TABLE t1"}));
- testValidSqlCommand("DROP TABLE IF EXISTS t1", new SqlCommandCall(SqlCommand.DROP_TABLE, new String[]{"DROP TABLE IF EXISTS t1"}));
- testValidSqlCommand("DROP TABLE IF EXISTS catalog1.db1.t1", new SqlCommandCall(SqlCommand.DROP_TABLE, new String[]{"DROP TABLE IF EXISTS catalog1.db1.t1"}));
- testValidSqlCommand("DROP TABLE IF EXISTS db1.t1", new SqlCommandCall(SqlCommand.DROP_TABLE, new String[]{"DROP TABLE IF EXISTS db1.t1"}));
- // Test create function.
- testInvalidSqlCommand("CREATE FUNCTION");
- testInvalidSqlCommand("CREATE FUNCTIONS");
- testInvalidSqlCommand("CREATE FUNCTIONS");
- testValidSqlCommand("CREATE FUNCTION catalog1.db1.func1 as 'class_name'",
- new SqlCommandCall(SqlCommand.CREATE_FUNCTION, new String[] {"CREATE FUNCTION catalog1.db1.func1 as 'class_name'"}));
- testValidSqlCommand("CREATE TEMPORARY FUNCTION catalog1.db1.func1 as 'class_name' LANGUAGE JAVA",
- new SqlCommandCall(SqlCommand.CREATE_FUNCTION, new String[] {"CREATE TEMPORARY FUNCTION catalog1.db1.func1 as 'class_name' LANGUAGE JAVA"}));
- testValidSqlCommand("CREATE TEMPORARY SYSTEM FUNCTION catalog1.db1.func1 as 'class_name' LANGUAGE JAVA",
- new SqlCommandCall(SqlCommand.CREATE_FUNCTION, new String[] {"CREATE TEMPORARY SYSTEM FUNCTION catalog1.db1.func1 as 'class_name' LANGUAGE JAVA"}));
- // Test drop function.
- testInvalidSqlCommand("DROP FUNCTION");
- testInvalidSqlCommand("DROP FUNCTIONS");
- testInvalidSqlCommand("DROP FUNCTIONS");
- testValidSqlCommand("DROP FUNCTION catalog1.db1.func1",
- new SqlCommandCall(SqlCommand.DROP_FUNCTION, new String[] {"DROP FUNCTION catalog1.db1.func1"}));
- testValidSqlCommand("DROP TEMPORARY FUNCTION catalog1.db1.func1",
- new SqlCommandCall(SqlCommand.DROP_FUNCTION, new String[] {"DROP TEMPORARY FUNCTION catalog1.db1.func1"}));
- testValidSqlCommand("DROP TEMPORARY SYSTEM FUNCTION IF EXISTS catalog1.db1.func1",
- new SqlCommandCall(SqlCommand.DROP_FUNCTION, new String[] {"DROP TEMPORARY SYSTEM FUNCTION IF EXISTS catalog1.db1.func1"}));
- // Test alter function.
- testInvalidSqlCommand("ALTER FUNCTION");
- testInvalidSqlCommand("ALTER FUNCTIONS");
- testInvalidSqlCommand("ALTER FUNCTIONS");
- testValidSqlCommand("ALTER FUNCTION catalog1.db1.func1 as 'a.b.c.func2'",
- new SqlCommandCall(SqlCommand.ALTER_FUNCTION, new String[] {"ALTER FUNCTION catalog1.db1.func1 as 'a.b.c.func2'"}));
- testValidSqlCommand("ALTER TEMPORARY FUNCTION IF EXISTS catalog1.db1.func1 as 'a.b.c.func2'",
- new SqlCommandCall(SqlCommand.ALTER_FUNCTION, new String[] {"ALTER TEMPORARY FUNCTION IF EXISTS catalog1.db1.func1 as 'a.b.c.func2'"}));
- testValidSqlCommand("ALTER TEMPORARY SYSTEM FUNCTION IF EXISTS catalog1.db1.func1 as 'a.b.c.func2'",
- new SqlCommandCall(SqlCommand.ALTER_FUNCTION, new String[] {"ALTER TEMPORARY SYSTEM FUNCTION IF EXISTS catalog1.db1.func1 as 'a.b.c.func2'"}));
+ public void testCommands() throws Exception {
+ List<TestItem> testItems = Arrays.asList(
+ TestItem.validSql("QUIT;", SqlCommand.QUIT).cannotParseComment(),
+ TestItem.validSql("eXiT;", SqlCommand.QUIT).cannotParseComment(),
+ TestItem.validSql("CLEAR;", SqlCommand.CLEAR).cannotParseComment(),
+ // desc xx
+ TestItem.validSql("DESC MyTable", SqlCommand.DESC, "MyTable").cannotParseComment(),
+ TestItem.validSql("DESC MyTable ", SqlCommand.DESC, "MyTable").cannotParseComment(),
+ TestItem.invalidSql("DESC "), // no table name
+ // describe xx
+ TestItem.validSql("DESCRIBE MyTable",
+ SqlCommand.DESCRIBE,
+ "`default_catalog`.`default_database`.`MyTable`"),
+ TestItem.validSql("DESCRIBE MyTable ",
+ SqlCommand.DESCRIBE,
+ "`default_catalog`.`default_database`.`MyTable`"),
+ TestItem.invalidSql("DESCRIBE "), // no table name
+ // explain xx
+ TestItem.validSql("EXPLAIN SELECT a FROM MyTable",
+ SqlCommand.EXPLAIN,
+ "EXPLAIN PLAN FOR SELECT a FROM MyTable").cannotParseComment(),
+ TestItem.invalidSql("EXPLAIN "), // no query
+ // explain plan for xx
+ TestItem.validSql("EXPLAIN PLAN FOR SELECT a FROM MyTable",
+ SqlCommand.EXPLAIN,
+ "EXPLAIN PLAN FOR SELECT a FROM MyTable"),
+ TestItem.validSql("EXPLAIN PLAN FOR INSERT INTO MySink(c) SELECT c FROM MyTable",
+ SqlCommand.EXPLAIN,
+ "EXPLAIN PLAN FOR INSERT INTO MySink(c) SELECT c FROM MyTable"),
+ TestItem.validSql("EXPLAIN PLAN FOR INSERT INTO MySink(c) SELECT c FROM MyTable",
+ SqlCommand.EXPLAIN,
+ "EXPLAIN PLAN FOR INSERT INTO MySink(c) SELECT c FROM MyTable"),
+ TestItem.sqlWithException("EXPLAIN PLAN FOR INSERT INTO MySink(c) SELECT xxx FROM MyTable",
+ SqlExecutionException.class,
+ "Column 'xxx' not found in any table"),
+ // select xx
+ TestItem.validSql("SELECT a FROM MyTable", SqlCommand.SELECT, "SELECT a FROM MyTable"),
+ TestItem.validSql("WITH t as (select a from MyTable) select a from t",
+ SqlCommand.SELECT,
+ "WITH t as (select a from MyTable) select a from t"),
+ // insert xx
+ TestItem.validSql("INSERT INTO other SELECT 1+1",
+ SqlCommand.INSERT_INTO,
+ "INSERT INTO other SELECT 1+1"),
+ TestItem.validSql("INSERT OVERWRITE other SELECT 1+1",
+ SqlCommand.INSERT_OVERWRITE,
+ "INSERT OVERWRITE other SELECT 1+1"),
+ // create view xx
+ TestItem.validSql("CREATE VIEW x AS SELECT 1+1",
+ SqlCommand.CREATE_VIEW,
+ "`default_catalog`.`default_database`.`x`", "SELECT 1 + 1"),
+ TestItem.validSql("CREATE VIEW x AS SELECT 1+1 FROM MyTable",
+ SqlCommand.CREATE_VIEW,
+ "`default_catalog`.`default_database`.`x`",
+ "SELECT 1 + 1\nFROM `default_catalog`.`default_database`.`MyTable` AS `MyTable`"),
+ TestItem.invalidSql("CREATE VIEW x SELECT 1+1 "), // missing AS
+ // drop view xx
+ TestItem.validSql("DROP VIEW TestView1",
+ SqlCommand.DROP_VIEW,
+ "`default_catalog`.`default_database`.`TestView1`"),
+ TestItem.invalidSql("DROP VIEW "), // missing name
+ // set
+ TestItem.validSql("SET", SqlCommand.SET).cannotParseComment(),
+ TestItem.validSql("SET x=y", SqlCommand.SET, "x", "y").cannotParseComment(),
+ TestItem.validSql("SET x = y", SqlCommand.SET, "x", " y").cannotParseComment(),
+ // reset
+ TestItem.validSql("reset;", SqlCommand.RESET).cannotParseComment(),
+ // source xx
+ TestItem.validSql("source /my/file;", SqlCommand.SOURCE, "/my/file").cannotParseComment(),
+ // create catalog xx
+ TestItem.validSql("create CATALOG c1 with('type'='generic_in_memory')",
+ SqlCommand.CREATE_CATALOG,
+ "create CATALOG c1 with('type'='generic_in_memory')"),
+ TestItem.validSql("create CATALOG c1 WITH ('type'='simple-catalog', 'default-database'='db1')",
+ SqlCommand.CREATE_CATALOG,
+ "create CATALOG c1 WITH ('type'='simple-catalog', 'default-database'='db1')"),
+ // drop catalog xx
+ TestItem.validSql("drop CATALOG c1", SqlCommand.DROP_CATALOG, "drop CATALOG c1"),
+ // use xx
+ TestItem.validSql("USE CATALOG catalog1;", SqlCommand.USE_CATALOG, "`catalog1`"),
+ TestItem.validSql("use `default`;", SqlCommand.USE, "`default_catalog`.`default`"),
+ TestItem.invalidSql("use catalog "), // no catalog name
+ // create database xx
+ TestItem.validSql("create database db1;", SqlCommand.CREATE_DATABASE, "create database db1"),
+ // drop database xx
+ TestItem.validSql("drop database db1;", SqlCommand.DROP_DATABASE, "drop database db1"),
+ // alter database xx
+ TestItem.validSql("alter database default_database set ('k1' = 'a')",
+ SqlCommand.ALTER_DATABASE,
+ "alter database default_database set ('k1' = 'a')"),
+ // alter table xx
+ TestItem.validSql("alter table default_catalog.default_database.MyTable rename to tb2",
+ SqlCommand.ALTER_TABLE, "alter table default_catalog.default_database.MyTable rename to tb2"),
+ TestItem.validSql("alter table MyTable set ('k1'='v1', 'k2'='v2')",
+ SqlCommand.ALTER_TABLE,
+ "alter table MyTable set ('k1'='v1', 'k2'='v2')"),
+ // create table xx
+ TestItem.invalidSql("CREATE tables"),
+ TestItem.invalidSql("CREATE tables"),
+ TestItem.validSql("create Table hello", SqlCommand.CREATE_TABLE, "create Table hello"),
+ TestItem.validSql("create Table hello(a int)", SqlCommand.CREATE_TABLE, "create Table hello(a int)"),
+ TestItem.validSql("CREATE TABLE T(\n"
+ + " a int,\n"
+ + " b varchar(20),\n"
+ + " c as to_timestamp(b),\n"
+ + " watermark for c as c - INTERVAL '5' second\n"
+ + ") WITH (\n"
+ + " 'k1' = 'v1',\n"
+ + " 'k2' = 'v2')\n",
+ SqlCommand.CREATE_TABLE,
+ "CREATE TABLE T(\n"
+ + " a int,\n"
+ + " b varchar(20),\n"
+ + " c as to_timestamp(b),\n"
+ + " watermark for c as c - INTERVAL '5' second\n"
+ + ") WITH (\n"
+ + " 'k1' = 'v1',\n"
+ + " 'k2' = 'v2')"),
+ // drop table xx
+ TestItem.invalidSql("DROP table"),
+ TestItem.invalidSql("DROP tables"),
+ TestItem.validSql("DROP TABLE t1", SqlCommand.DROP_TABLE, "DROP TABLE t1"),
+ TestItem.validSql("DROP TABLE IF EXISTS t1", SqlCommand.DROP_TABLE, "DROP TABLE IF EXISTS t1"),
+ TestItem.validSql("DROP TABLE IF EXISTS catalog1.db1.t1", SqlCommand.DROP_TABLE,
+ "DROP TABLE IF EXISTS catalog1.db1.t1"),
+ TestItem.validSql("DROP TABLE IF EXISTS db1.t1", SqlCommand.DROP_TABLE, "DROP TABLE IF EXISTS db1.t1"),
+ // show catalogs
+ TestItem.validSql("SHOW CATALOGS;", SqlCommand.SHOW_CATALOGS),
+ TestItem.validSql(" SHOW CATALOGS ;", SqlCommand.SHOW_CATALOGS),
+ // show databases
+ TestItem.validSql("SHOW DATABASES;", SqlCommand.SHOW_DATABASES),
+ TestItem.validSql(" SHOW DATABASES ;", SqlCommand.SHOW_DATABASES),
+ // show tables
+ TestItem.validSql("SHOW TABLES;", SqlCommand.SHOW_TABLES),
+ TestItem.validSql(" SHOW TABLES ;", SqlCommand.SHOW_TABLES),
+ // show functions
+ TestItem.validSql("SHOW FUNCTIONS;", SqlCommand.SHOW_FUNCTIONS),
+ TestItem.validSql(" SHOW FUNCTIONS ", SqlCommand.SHOW_FUNCTIONS),
+ // show modules
+ TestItem.validSql("SHOW MODULES", SqlCommand.SHOW_MODULES).cannotParseComment(),
+ TestItem.validSql(" SHOW MODULES ", SqlCommand.SHOW_MODULES).cannotParseComment(),
+ // Test create function.
+ TestItem.invalidSql("CREATE FUNCTION "),
+ TestItem.invalidSql("CREATE FUNCTIONS "),
+ TestItem.invalidSql("CREATE FUNCTIONS "),
+ TestItem.validSql("CREATE FUNCTION catalog1.db1.func1 as 'class_name'",
+ SqlCommand.CREATE_FUNCTION,
+ "CREATE FUNCTION catalog1.db1.func1 as 'class_name'"),
+ TestItem.validSql("CREATE TEMPORARY FUNCTION catalog1.db1.func1 as 'class_name' LANGUAGE JAVA",
+ SqlCommand.CREATE_FUNCTION,
+ "CREATE TEMPORARY FUNCTION catalog1.db1.func1 as 'class_name' LANGUAGE JAVA"),
+ TestItem.validSql("CREATE TEMPORARY SYSTEM FUNCTION catalog1.db1.func1 as 'class_name' LANGUAGE JAVA",
+ SqlCommand.CREATE_FUNCTION,
+ "CREATE TEMPORARY SYSTEM FUNCTION catalog1.db1.func1 as 'class_name' LANGUAGE JAVA"),
+ // drop function xx
+ TestItem.invalidSql("DROP FUNCTION "),
+ TestItem.invalidSql("DROP FUNCTIONS "),
+ TestItem.invalidSql("DROP FUNCTIONS "),
+ TestItem.validSql("DROP FUNCTION catalog1.db1.func1",
+ SqlCommand.DROP_FUNCTION,
+ "DROP FUNCTION catalog1.db1.func1"),
+ TestItem.validSql("DROP TEMPORARY FUNCTION catalog1.db1.func1",
+ SqlCommand.DROP_FUNCTION,
+ "DROP TEMPORARY FUNCTION catalog1.db1.func1"),
+ TestItem.validSql("DROP TEMPORARY SYSTEM FUNCTION IF EXISTS catalog1.db1.func1",
+ SqlCommand.DROP_FUNCTION,
+ "DROP TEMPORARY SYSTEM FUNCTION IF EXISTS catalog1.db1.func1"),
+ // alter function xx
+ TestItem.invalidSql("ALTER FUNCTION "),
+ TestItem.invalidSql("ALTER FUNCTIONS "),
+ TestItem.invalidSql("ALTER FUNCTIONS "),
+ TestItem.validSql("ALTER FUNCTION catalog1.db1.func1 as 'a.b.c.func2'",
+ SqlCommand.ALTER_FUNCTION,
+ "ALTER FUNCTION catalog1.db1.func1 as 'a.b.c.func2'"),
+ TestItem.validSql("ALTER TEMPORARY FUNCTION IF EXISTS catalog1.db1.func1 as 'a.b.c.func2'",
+ SqlCommand.ALTER_FUNCTION,
+ "ALTER TEMPORARY FUNCTION IF EXISTS catalog1.db1.func1 as 'a.b.c.func2'"),
+ TestItem.validSql("ALTER TEMPORARY FUNCTION IF EXISTS catalog1.db1.func1 as 'a.b.c.func2'",
+ SqlCommand.ALTER_FUNCTION,
+ "ALTER TEMPORARY FUNCTION IF EXISTS catalog1.db1.func1 as 'a.b.c.func2'"),
+ TestItem.sqlWithException(
+ "ALTER TEMPORARY SYSTEM FUNCTION IF EXISTS catalog1.db1.func1 as 'a.b.c.func2'",
+ SqlExecutionException.class,
+ "Alter temporary system function is not supported")
+ );
+ for (TestItem item : testItems) {
+ runTestItem(item);
+ }
}
- private void testInvalidSqlCommand(String stmt) {
- final Optional<SqlCommandCall> actualCall = SqlCommandParser.parse(stmt);
- if (actualCall.isPresent()) {
- fail();
+ private void runTestItem(TestItem item) {
+ if (item.isValidSqlCmd) {
+ runValidSql(item);
+ } else {
+ runInvalidSql(item);
}
}
- private void testValidSqlCommand(String stmt, SqlCommandCall expectedCall) {
- final Optional<SqlCommandCall> actualCall = SqlCommandParser.parse(stmt);
+ private void runValidSql(TestItem item) {
+ Tuple2<Boolean, Optional<SqlCommandCall>> checkFlagAndActualCall = parseSqlAndCheckException(item);
+ if (!checkFlagAndActualCall.f0) {
+ return;
+ }
+ Optional<SqlCommandCall> actualCall = checkFlagAndActualCall.f1;
if (!actualCall.isPresent()) {
- fail();
+ fail("test statement: " + item.sql);
+ }
+ assertNotNull(item.expectedCmd);
+ assertEquals("test statement: " + item.sql,
+ new SqlCommandCall(item.expectedCmd, item.expectedOperands), actualCall.get());
+
+ String stmtWithComment = "-- comments \n " + item.sql;
+ actualCall = SqlCommandParser.parse(parser, stmtWithComment);
+ if (item.cannotParseComment) {
+ assertFalse(actualCall.isPresent());
+ } else {
+ if (!actualCall.isPresent()) {
+ fail("test statement: " + item.sql);
+ }
+ assertEquals(item.expectedCmd, actualCall.get().command);
+ }
+ }
+
+ public void runInvalidSql(TestItem item) {
+ Tuple2<Boolean, Optional<SqlCommandCall>> checkFlagAndActualCall = parseSqlAndCheckException(item);
+ if (!checkFlagAndActualCall.f0) {
+ return;
+ }
+ if (checkFlagAndActualCall.f1.isPresent()) {
+ fail("test statement: " + item.sql);
+ }
+ }
+
+ private Tuple2<Boolean, Optional<SqlCommandCall>> parseSqlAndCheckException(TestItem item) {
+ Optional<SqlCommandCall> call = Optional.empty();
+ Throwable actualException = null;
+ try {
+ call = SqlCommandParser.parse(parser, item.sql);
+ } catch (Throwable e) {
+ actualException = e;
+ }
+
+ if (item.expectedException == null && actualException == null) {
+ return Tuple2.of(true, call);
+ } else if (item.expectedException == null) {
+ actualException.printStackTrace();
+ fail("Failed to run sql: " + item.sql);
+ } else if (actualException == null) {
+ fail("the excepted exception: '" + item.expectedException + "' does not occur.\n" +
+ "test statement: " + item.sql);
+ } else {
+ assertTrue(actualException.getClass().isAssignableFrom(item.expectedException));
+ boolean hasExpectedExceptionMsg = false;
+ while (actualException != null) {
+ if (actualException.getMessage().contains(item.expectedExceptionMsg)) {
+ hasExpectedExceptionMsg = true;
+ break;
+ }
+ actualException = actualException.getCause();
+ }
+ if (!hasExpectedExceptionMsg) {
+ fail("the excepted exception message: '" + item.expectedExceptionMsg + "' does not occur.\n" +
+ "test statement: " + item.sql);
+ }
+ }
+ return Tuple2.of(false, Optional.empty());
+ }
+
+ private static class TestItem {
+ private final String sql;
+ private final boolean isValidSqlCmd;
+ private boolean cannotParseComment = true;
+ private @Nullable
+ SqlCommand expectedCmd = null;
+ private String[] expectedOperands = new String[0];
+ private Class<? extends Throwable> expectedException = null;
+ private String expectedExceptionMsg = null;
+
+ private TestItem(String sql, boolean isValidSqlCmd) {
+ this.sql = sql;
+ this.isValidSqlCmd = isValidSqlCmd;
+ }
+
+ public static TestItem invalidSql(String sql) {
+ return new TestItem(sql, false);
+ }
+
+ public static TestItem validSql(
+ String sql, SqlCommand expectedCmd, String... expectedOperands) {
+ TestItem testItem = new TestItem(sql, true);
+ testItem.expectedCmd = expectedCmd;
+ testItem.expectedOperands = expectedOperands;
+ testItem.cannotParseComment = false; // default is false
+ return testItem;
+ }
+
+ public static TestItem sqlWithException(String sql, Class<? extends Throwable> expectedException,
+ String exceptedMsg) {
+ TestItem testItem = new TestItem(sql, true);
+ testItem.expectedException = expectedException;
+ testItem.expectedExceptionMsg = exceptedMsg;
+ return testItem;
+ }
+
+ public TestItem cannotParseComment() {
+ cannotParseComment = true;
+ return this;
+ }
+
+ @Override
+ public String toString() {
+ return this.sql;
}
- assertEquals(expectedCall, actualCall.get());
}
}
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/TestingExecutor.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/TestingExecutor.java
index 8e73e0a..b22f8d7 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/TestingExecutor.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/TestingExecutor.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.client.cli;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.client.cli.utils.SqlParserHelper;
import org.apache.flink.table.client.config.entries.ViewEntry;
import org.apache.flink.table.client.gateway.Executor;
import org.apache.flink.table.client.gateway.ProgramTargetDescriptor;
@@ -27,6 +28,7 @@ import org.apache.flink.table.client.gateway.ResultDescriptor;
import org.apache.flink.table.client.gateway.SessionContext;
import org.apache.flink.table.client.gateway.SqlExecutionException;
import org.apache.flink.table.client.gateway.TypedResult;
+import org.apache.flink.table.delegation.Parser;
import org.apache.flink.types.Row;
import org.apache.flink.util.function.BiConsumerWithException;
import org.apache.flink.util.function.SupplierWithException;
@@ -56,6 +58,8 @@ class TestingExecutor implements Executor {
private int numUseDatabaseCalls = 0;
private BiConsumerWithException<String, String, SqlExecutionException> useDatabaseConsumer;
+ private final SqlParserHelper helper;
+
TestingExecutor(
List<SupplierWithException<TypedResult<List<Tuple2<Boolean, Row>>>, SqlExecutionException>> resultChanges,
List<SupplierWithException<TypedResult<Integer>, SqlExecutionException>> snapshotResults,
@@ -67,6 +71,8 @@ class TestingExecutor implements Executor {
this.resultPages = resultPages;
this.useCatalogConsumer = useCatalogConsumer;
this.useDatabaseConsumer = useDatabaseConsumer;
+ helper = new SqlParserHelper();
+ helper.registerTables();
}
@Override
@@ -189,8 +195,8 @@ class TestingExecutor implements Executor {
}
@Override
- public String explainStatement(String sessionId, String statement) throws SqlExecutionException {
- throw new UnsupportedOperationException("Not implemented.");
+ public Parser getSqlParser(String sessionId) {
+ return helper.getSqlParser();
}
@Override
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/utils/SqlParserHelper.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/utils/SqlParserHelper.java
new file mode 100644
index 0000000..ad6bc9b
--- /dev/null
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/utils/SqlParserHelper.java
@@ -0,0 +1,55 @@
+/*
+ * 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.flink.table.client.cli.utils;
+
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.internal.TableEnvironmentInternal;
+import org.apache.flink.table.delegation.Parser;
+
+/**
+ * An utility class that provides pre-prepared tables and sql parser.
+ */
+public class SqlParserHelper {
+ // return the sql parser instance hold by this table evn.
+ private final TableEnvironment tableEnv;
+
+ public SqlParserHelper() {
+ tableEnv = TableEnvironment.create(EnvironmentSettings.newInstance().build());
+ }
+
+ /**
+ * prepare some tables for testing.
+ */
+ public void registerTables() {
+ registerTable("create table MyTable (a int, b bigint, c varchar(32)) " +
+ "with ('connector' = 'filesystem', 'path' = '/non')");
+ registerTable("create table MyOtherTable (a int, b bigint) " +
+ "with ('connector' = 'filesystem', 'path' = '/non')");
+ registerTable("create table MySink (a int, c varchar(32)) with ('connector' = 'COLLECTION' )");
+ }
+
+ public void registerTable(String createTableStmt) {
+ tableEnv.executeSql(createTableStmt);
+ }
+
+ public Parser getSqlParser() {
+ return ((TableEnvironmentInternal) tableEnv).getParser();
+ }
+}