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 2021/03/27 04:35:31 UTC
[flink] branch master updated (5b33c79 -> 0736500)
This is an automated email from the ASF dual-hosted git repository.
jark pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 5b33c79 [FLINK-21917][docs] Add back missing kafka.md Chinese documentation (#15392)
new 1080a63 [FLINK-21463][table-api] Support to parse 'RESET key' command
new 0736500 [FLINK-21463][sql-client] Unify SQL Client parser and TableEnvironment parser
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/flink/table/client/cli/CliClient.java | 481 +++++------------
.../apache/flink/table/client/cli/CliStrings.java | 108 +---
.../flink/table/client/cli/SqlCommandParser.java | 414 ---------------
.../flink/table/client/cli/SqlCompleter.java | 48 --
.../flink/table/client/gateway/Executor.java | 17 +-
.../table/client/gateway/local/LocalExecutor.java | 111 ++--
.../flink/table/client/cli/CliClientTest.java | 61 +--
.../flink/table/client/cli/CliResultViewTest.java | 31 +-
.../table/client/cli/SqlCommandParserTest.java | 586 ---------------------
.../flink/table/client/cli/TestingExecutor.java | 15 +-
.../table/client/gateway/local/DependencyTest.java | 19 +-
.../client/gateway/local/LocalExecutorITCase.java | 37 +-
.../src/test/resources/sql/catalog_database.q | 80 +--
.../src/test/resources/sql/function.q | 50 +-
.../src/test/resources/sql/module.q | 18 +-
.../src/test/resources/sql/select.q | 2 +-
.../flink-sql-client/src/test/resources/sql/set.q | 2 +-
.../src/test/resources/sql/table.q | 18 +-
.../flink-sql-client/src/test/resources/sql/view.q | 18 +-
.../flink/table/api/internal/TableResultImpl.java | 2 +-
.../table/operations/command/ResetOperation.java | 26 +-
.../planner/parse/ResetOperationParseStrategy.java | 17 +-
.../table/planner/delegation/ParserImplTest.java | 2 +
.../table/parse/ResetOperationParseStrategy.java | 17 +-
24 files changed, 448 insertions(+), 1732 deletions(-)
delete mode 100644 flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCommandParser.java
delete mode 100644 flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/SqlCommandParserTest.java
[flink] 02/02: [FLINK-21463][sql-client] Unify SQL Client parser
and TableEnvironment parser
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 07365002df382dbdaf6f25f0eb2700925cabc9c6
Author: Shengkai <10...@qq.com>
AuthorDate: Fri Mar 26 00:25:12 2021 +0800
[FLINK-21463][sql-client] Unify SQL Client parser and TableEnvironment parser
This closes #15315
---
.../apache/flink/table/client/cli/CliClient.java | 481 +++++------------
.../apache/flink/table/client/cli/CliStrings.java | 108 +---
.../flink/table/client/cli/SqlCommandParser.java | 414 ---------------
.../flink/table/client/cli/SqlCompleter.java | 48 --
.../flink/table/client/gateway/Executor.java | 17 +-
.../table/client/gateway/local/LocalExecutor.java | 111 ++--
.../flink/table/client/cli/CliClientTest.java | 61 +--
.../flink/table/client/cli/CliResultViewTest.java | 31 +-
.../table/client/cli/SqlCommandParserTest.java | 586 ---------------------
.../flink/table/client/cli/TestingExecutor.java | 15 +-
.../table/client/gateway/local/DependencyTest.java | 19 +-
.../client/gateway/local/LocalExecutorITCase.java | 37 +-
.../src/test/resources/sql/catalog_database.q | 80 +--
.../src/test/resources/sql/function.q | 50 +-
.../src/test/resources/sql/module.q | 18 +-
.../src/test/resources/sql/select.q | 2 +-
.../flink-sql-client/src/test/resources/sql/set.q | 2 +-
.../src/test/resources/sql/table.q | 18 +-
.../flink-sql-client/src/test/resources/sql/view.q | 18 +-
.../flink/table/api/internal/TableResultImpl.java | 2 +-
20 files changed, 392 insertions(+), 1726 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 5215542..f2e037e 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
@@ -22,11 +22,17 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.client.SqlClient;
import org.apache.flink.table.client.SqlClientException;
-import org.apache.flink.table.client.cli.SqlCommandParser.SqlCommandCall;
-import org.apache.flink.table.client.config.YamlConfigUtils;
import org.apache.flink.table.client.gateway.Executor;
import org.apache.flink.table.client.gateway.ResultDescriptor;
import org.apache.flink.table.client.gateway.SqlExecutionException;
+import org.apache.flink.table.operations.CatalogSinkModifyOperation;
+import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.operations.QueryOperation;
+import org.apache.flink.table.operations.command.ClearOperation;
+import org.apache.flink.table.operations.command.HelpOperation;
+import org.apache.flink.table.operations.command.QuitOperation;
+import org.apache.flink.table.operations.command.ResetOperation;
+import org.apache.flink.table.operations.command.SetOperation;
import org.apache.flink.table.utils.PrintUtils;
import org.jline.reader.EndOfFileException;
@@ -47,18 +53,22 @@ import javax.annotation.Nullable;
import java.io.IOError;
import java.io.IOException;
-import java.nio.charset.Charset;
import java.nio.file.Files;
import java.nio.file.Path;
-import java.nio.file.Paths;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import static org.apache.flink.table.api.internal.TableResultImpl.TABLE_RESULT_OK;
import static org.apache.flink.table.client.cli.CliStrings.MESSAGE_DEPRECATED_KEY;
+import static org.apache.flink.table.client.cli.CliStrings.MESSAGE_EXECUTE_STATEMENT;
import static org.apache.flink.table.client.cli.CliStrings.MESSAGE_REMOVED_KEY;
import static org.apache.flink.table.client.cli.CliStrings.MESSAGE_RESET_KEY;
-import static org.apache.flink.table.client.cli.CliStrings.MESSAGE_SET;
+import static org.apache.flink.table.client.cli.CliStrings.MESSAGE_SET_KEY;
+import static org.apache.flink.table.client.config.YamlConfigUtils.getOptionNameWithDeprecatedKey;
+import static org.apache.flink.table.client.config.YamlConfigUtils.getPropertiesInPretty;
+import static org.apache.flink.table.client.config.YamlConfigUtils.isDeprecatedKey;
+import static org.apache.flink.table.client.config.YamlConfigUtils.isRemovedKey;
import static org.apache.flink.util.Preconditions.checkState;
/** SQL CLI client. */
@@ -84,8 +94,6 @@ public class CliClient implements AutoCloseable {
private static final int PLAIN_TERMINAL_HEIGHT = 30;
- private static final int SOURCE_MAX_SIZE = 50_000;
-
/**
* Creates a CLI instance with a custom terminal. Make sure to close the CLI instance afterwards
* using {@link #close()}.
@@ -222,8 +230,8 @@ public class CliClient implements AutoCloseable {
if (line == null) {
continue;
}
- final Optional<SqlCommandCall> cmdCall = parseCommand(line);
- cmdCall.ifPresent(this::callCommand);
+ final Optional<Operation> operation = parseCommand(line);
+ operation.ifPresent(this::callOperation);
}
}
@@ -247,18 +255,16 @@ public class CliClient implements AutoCloseable {
terminal.writer().println(new AttributedString(statement).toString());
terminal.flush();
- final Optional<SqlCommandCall> parsedStatement = parseCommand(statement);
+ final Optional<Operation> operation = parseCommand(statement);
// only support INSERT INTO/OVERWRITE
- return parsedStatement
+ return operation
.map(
- cmdCall -> {
- switch (cmdCall.command) {
- case INSERT_INTO:
- case INSERT_OVERWRITE:
- return callInsert(cmdCall);
- default:
- printError(CliStrings.MESSAGE_UNSUPPORTED_SQL);
- return false;
+ op -> {
+ if (op instanceof CatalogSinkModifyOperation) {
+ return callInsert((CatalogSinkModifyOperation) op);
+ } else {
+ printError(CliStrings.MESSAGE_UNSUPPORTED_SQL);
+ return false;
}
})
.orElse(false);
@@ -266,141 +272,47 @@ public class CliClient implements AutoCloseable {
// --------------------------------------------------------------------------------------------
- private Optional<SqlCommandCall> parseCommand(String line) {
- final SqlCommandCall parsedLine;
+ private Optional<Operation> parseCommand(String stmt) {
+ // normalize
+ stmt = stmt.trim();
+ // remove ';' at the end
+ if (stmt.endsWith(";")) {
+ stmt = stmt.substring(0, stmt.length() - 1).trim();
+ }
try {
- parsedLine = SqlCommandParser.parse(executor.getSqlParser(sessionId), line);
+ Operation operation = executor.parseStatement(sessionId, stmt);
+ return Optional.of(operation);
} catch (SqlExecutionException e) {
printExecutionException(e);
- return Optional.empty();
}
- return Optional.of(parsedLine);
- }
-
- private void callCommand(SqlCommandCall cmdCall) {
- switch (cmdCall.command) {
- case QUIT:
- callQuit();
- break;
- case CLEAR:
- callClear();
- break;
- case RESET:
- callReset(cmdCall);
- break;
- case SET:
- callSet(cmdCall);
- break;
- case HELP:
- callHelp();
- break;
- case SHOW_CATALOGS:
- case SHOW_DATABASES:
- case SHOW_TABLES:
- case SHOW_VIEWS:
- case SHOW_FUNCTIONS:
- case SHOW_MODULES:
- case SHOW_PARTITIONS:
- case SHOW_CURRENT_CATALOG:
- case SHOW_CURRENT_DATABASE:
- callShow(cmdCall);
- break;
- case USE_CATALOG:
- callUseCatalog(cmdCall);
- break;
- case USE:
- callUseDatabase(cmdCall);
- break;
- case DESCRIBE:
- callDescribe(cmdCall);
- break;
- case EXPLAIN:
- callExplain(cmdCall);
- break;
- case SELECT:
- callSelect(cmdCall);
- break;
- case INSERT_INTO:
- case INSERT_OVERWRITE:
- callInsert(cmdCall);
- break;
- case CREATE_TABLE:
- callDdl(cmdCall.operands[0], CliStrings.MESSAGE_TABLE_CREATED);
- break;
- case DROP_TABLE:
- callDdl(cmdCall.operands[0], CliStrings.MESSAGE_TABLE_REMOVED);
- break;
- case CREATE_VIEW:
- callDdl(cmdCall.operands[0], CliStrings.MESSAGE_VIEW_CREATED);
- break;
- case DROP_VIEW:
- callDdl(cmdCall.operands[0], CliStrings.MESSAGE_VIEW_REMOVED);
- break;
- case ALTER_VIEW:
- callDdl(
- cmdCall.operands[0],
- CliStrings.MESSAGE_ALTER_VIEW_SUCCEEDED,
- CliStrings.MESSAGE_ALTER_VIEW_FAILED);
- break;
- case CREATE_FUNCTION:
- callDdl(cmdCall.operands[0], CliStrings.MESSAGE_FUNCTION_CREATED);
- break;
- case DROP_FUNCTION:
- callDdl(cmdCall.operands[0], CliStrings.MESSAGE_FUNCTION_REMOVED);
- break;
- case ALTER_FUNCTION:
- callDdl(
- cmdCall.operands[0],
- CliStrings.MESSAGE_ALTER_FUNCTION_SUCCEEDED,
- CliStrings.MESSAGE_ALTER_FUNCTION_FAILED);
- break;
- case SOURCE:
- callSource(cmdCall);
- break;
- case CREATE_DATABASE:
- callDdl(cmdCall.operands[0], CliStrings.MESSAGE_DATABASE_CREATED);
- break;
- case DROP_DATABASE:
- callDdl(cmdCall.operands[0], CliStrings.MESSAGE_DATABASE_REMOVED);
- break;
- case ALTER_DATABASE:
- callDdl(
- cmdCall.operands[0],
- CliStrings.MESSAGE_ALTER_DATABASE_SUCCEEDED,
- CliStrings.MESSAGE_ALTER_DATABASE_FAILED);
- break;
- case ALTER_TABLE:
- callDdl(
- cmdCall.operands[0],
- CliStrings.MESSAGE_ALTER_TABLE_SUCCEEDED,
- CliStrings.MESSAGE_ALTER_TABLE_FAILED);
- break;
- case CREATE_CATALOG:
- callDdl(cmdCall.operands[0], CliStrings.MESSAGE_CATALOG_CREATED);
- break;
- case DROP_CATALOG:
- callDdl(cmdCall.operands[0], CliStrings.MESSAGE_CATALOG_REMOVED);
- break;
- case LOAD_MODULE:
- callDdl(
- cmdCall.operands[0],
- CliStrings.MESSAGE_LOAD_MODULE_SUCCEEDED,
- CliStrings.MESSAGE_LOAD_MODULE_FAILED);
- break;
- case UNLOAD_MODULE:
- callDdl(
- cmdCall.operands[0],
- CliStrings.MESSAGE_UNLOAD_MODULE_SUCCEEDED,
- CliStrings.MESSAGE_UNLOAD_MODULE_FAILED);
- break;
- case USE_MODULES:
- callDdl(
- cmdCall.operands[0],
- CliStrings.MESSAGE_USE_MODULES_SUCCEEDED,
- CliStrings.MESSAGE_USE_MODULES_FAILED);
- break;
- default:
- throw new SqlClientException("Unsupported command: " + cmdCall.command);
+ return Optional.empty();
+ }
+
+ private void callOperation(Operation operation) {
+ if (operation instanceof QuitOperation) {
+ // QUIT/EXIT
+ callQuit();
+ } else if (operation instanceof ClearOperation) {
+ // CLEAR
+ callClear();
+ } else if (operation instanceof HelpOperation) {
+ // HELP
+ callHelp();
+ } else if (operation instanceof SetOperation) {
+ // SET
+ callSet((SetOperation) operation);
+ } else if (operation instanceof ResetOperation) {
+ // RESET
+ callReset((ResetOperation) operation);
+ } else if (operation instanceof CatalogSinkModifyOperation) {
+ // INSERT INTO/OVERWRITE
+ callInsert((CatalogSinkModifyOperation) operation);
+ } else if (operation instanceof QueryOperation) {
+ // SELECT
+ callSelect((QueryOperation) operation);
+ } else {
+ // fallback to default implementation
+ executeOperation(operation);
}
}
@@ -413,152 +325,59 @@ public class CliClient implements AutoCloseable {
clearTerminal();
}
- private void callReset(SqlCommandCall cmdCall) {
+ private void callReset(ResetOperation resetOperation) {
try {
// reset all session properties
- if (cmdCall.operands.length == 0) {
+ if (!resetOperation.getKey().isPresent()) {
executor.resetSessionProperties(sessionId);
printInfo(CliStrings.MESSAGE_RESET);
}
// reset a session property
else {
- String key = cmdCall.operands[0].trim();
+ String key = resetOperation.getKey().get();
executor.resetSessionProperty(sessionId, key);
-
- boolean isRemovedKey = YamlConfigUtils.isRemovedKey(key);
- boolean isDeprecatedKey = YamlConfigUtils.isDeprecatedKey(key);
- if (isRemovedKey || isDeprecatedKey) {
- printRemovedOrDeprecatedKeyMessage(key);
- }
- // when it's not removedKey, need to print normal message
- if (!isRemovedKey) {
- terminal.writer()
- .println(
- CliStrings.messageInfo(String.format(MESSAGE_RESET_KEY, key))
- .toAnsi());
- terminal.flush();
- }
+ printSetResetConfigKeyMessage(key, MESSAGE_RESET_KEY);
}
} catch (SqlExecutionException e) {
printExecutionException(e);
- return;
}
}
- private void callSet(SqlCommandCall cmdCall) {
- // show all properties
- if (cmdCall.operands.length == 0) {
- final Map<String, String> properties;
- try {
- properties = executor.getSessionProperties(sessionId);
- } catch (SqlExecutionException e) {
- printExecutionException(e);
- return;
- }
- if (properties.isEmpty()) {
- terminal.writer()
- .println(CliStrings.messageInfo(CliStrings.MESSAGE_EMPTY).toAnsi());
- } else {
- List<String> prettyEntries = YamlConfigUtils.getPropertiesInPretty(properties);
- prettyEntries.forEach(entry -> terminal.writer().println(entry));
- }
- terminal.flush();
- }
- // set a property
- else {
- String key = cmdCall.operands[0].trim();
- String value = cmdCall.operands[1].trim();
- try {
+ private void callSet(SetOperation setOperation) {
+ try {
+ // set a property
+ if (setOperation.getKey().isPresent() && setOperation.getValue().isPresent()) {
+ String key = setOperation.getKey().get().trim();
+ String value = setOperation.getValue().get().trim();
executor.setSessionProperty(sessionId, key, value);
- } catch (SqlExecutionException e) {
- printExecutionException(e);
- return;
+ printSetResetConfigKeyMessage(key, MESSAGE_SET_KEY);
}
- boolean isRemovedKey = YamlConfigUtils.isRemovedKey(key);
- boolean isDeprecatedKey = YamlConfigUtils.isDeprecatedKey(key);
- if (isRemovedKey || isDeprecatedKey) {
- printRemovedOrDeprecatedKeyMessage(key);
- }
- // it's not removedKey, need to print info message
- if (!isRemovedKey) {
- terminal.writer().println(CliStrings.messageInfo(MESSAGE_SET).toAnsi());
+ // show all properties
+ else {
+ final Map<String, String> properties = executor.getSessionProperties(sessionId);
+ if (properties.isEmpty()) {
+ terminal.writer()
+ .println(CliStrings.messageInfo(CliStrings.MESSAGE_EMPTY).toAnsi());
+ } else {
+ List<String> prettyEntries = getPropertiesInPretty(properties);
+ prettyEntries.forEach(entry -> terminal.writer().println(entry));
+ }
terminal.flush();
}
- }
- }
-
- private void callHelp() {
- terminal.writer().println(CliStrings.MESSAGE_HELP);
- terminal.flush();
- }
-
- private void callShow(SqlCommandCall cmdCall) {
- getResultAsTableauForm(
- cmdCall.operands.length == 0 ? cmdCall.command.toString() : cmdCall.operands[0]);
- }
-
- private void callUseCatalog(SqlCommandCall cmdCall) {
- try {
- executor.executeSql(sessionId, "USE CATALOG `" + cmdCall.operands[0] + "`");
- } catch (SqlExecutionException e) {
- printExecutionException(e);
- return;
- }
- printInfo(CliStrings.MESSAGE_CATALOG_CHANGED);
- terminal.flush();
- }
-
- private void callUseDatabase(SqlCommandCall cmdCall) {
- try {
- executor.executeSql(sessionId, "USE `" + cmdCall.operands[0] + "`");
- } catch (SqlExecutionException e) {
- printExecutionException(e);
- return;
- }
- printInfo(CliStrings.MESSAGE_DATABASE_CHANGED);
- terminal.flush();
- }
-
- private void callDescribe(SqlCommandCall cmdCall) {
- getResultAsTableauForm("DESCRIBE " + cmdCall.operands[0]);
- }
-
- private void getResultAsTableauForm(String statement) {
- final TableResult result;
- try {
- result = executor.executeSql(sessionId, statement);
} catch (SqlExecutionException e) {
printExecutionException(e);
- return;
}
- PrintUtils.printAsTableauForm(
- result.getResolvedSchema(),
- result.collect(),
- terminal.writer(),
- Integer.MAX_VALUE,
- "",
- false,
- false);
- terminal.flush();
}
- private void callExplain(SqlCommandCall cmdCall) {
- final String explanation;
- try {
- TableResult tableResult = executor.executeSql(sessionId, cmdCall.operands[0]);
- explanation = tableResult.collect().next().getField(0).toString();
- } catch (SqlExecutionException e) {
- printExecutionException(e);
- return;
- }
- terminal.writer().println(explanation);
+ private void callHelp() {
+ terminal.writer().println(CliStrings.MESSAGE_HELP);
terminal.flush();
}
- private void callSelect(SqlCommandCall cmdCall) {
+ private void callSelect(QueryOperation operation) {
final ResultDescriptor resultDesc;
try {
- resultDesc = executor.executeQuery(sessionId, cmdCall.operands[0]);
+ resultDesc = executor.executeQuery(sessionId, operation);
} catch (SqlExecutionException e) {
printExecutionException(e);
return;
@@ -572,7 +391,7 @@ public class CliClient implements AutoCloseable {
printExecutionException(e);
}
} else {
- final CliResultView view;
+ final CliResultView<?> view;
if (resultDesc.isMaterialized()) {
view = new CliTableResultView(this, resultDesc);
} else {
@@ -591,12 +410,12 @@ public class CliClient implements AutoCloseable {
}
}
- private boolean callInsert(SqlCommandCall cmdCall) {
+ private boolean callInsert(CatalogSinkModifyOperation operation) {
printInfo(CliStrings.MESSAGE_SUBMITTING_STATEMENT);
try {
- final TableResult tableResult = executor.executeSql(sessionId, cmdCall.operands[0]);
- checkState(tableResult.getJobClient().isPresent());
+ TableResult result = executor.executeOperation(sessionId, operation);
+ checkState(result.getJobClient().isPresent());
terminal.writer()
.println(
CliStrings.messageInfo(CliStrings.MESSAGE_STATEMENT_SUBMITTED)
@@ -606,84 +425,44 @@ public class CliClient implements AutoCloseable {
.println(
String.format(
"Job ID: %s\n",
- tableResult.getJobClient().get().getJobID().toString()));
+ result.getJobClient().get().getJobID().toString()));
terminal.flush();
+ return true;
} catch (SqlExecutionException e) {
printExecutionException(e);
- return false;
- }
- return true;
- }
-
- private void callSource(SqlCommandCall cmdCall) {
- final String pathString = cmdCall.operands[0];
-
- // load file
- final String stmt;
- try {
- final Path path = Paths.get(pathString);
- byte[] encoded = Files.readAllBytes(path);
- stmt = new String(encoded, Charset.defaultCharset());
- } catch (IOException e) {
- printExecutionException(e);
- return;
- }
-
- // limit the output a bit
- if (stmt.length() > SOURCE_MAX_SIZE) {
- printExecutionError(CliStrings.MESSAGE_MAX_SIZE_EXCEEDED);
- return;
}
-
- terminal.writer().println(CliStrings.messageInfo(CliStrings.MESSAGE_WILL_EXECUTE).toAnsi());
- terminal.writer().println(new AttributedString(stmt).toString());
- terminal.flush();
-
- // try to run it
- final Optional<SqlCommandCall> call = parseCommand(stmt);
- call.ifPresent(this::callCommand);
- }
-
- private void callDdl(String ddl, String successMessage) {
- callDdl(ddl, successMessage, null);
+ return false;
}
- private void callDdl(String ddl, String successMessage, String errorMessage) {
+ private void executeOperation(Operation operation) {
try {
- executor.executeSql(sessionId, ddl);
- printInfo(successMessage);
+ TableResult result = executor.executeOperation(sessionId, operation);
+ if (TABLE_RESULT_OK == result) {
+ // print more meaningful message than tableau OK result
+ printInfo(MESSAGE_EXECUTE_STATEMENT);
+ } else {
+ // print tableau if result has content
+ PrintUtils.printAsTableauForm(
+ result.getResolvedSchema(),
+ result.collect(),
+ terminal.writer(),
+ Integer.MAX_VALUE,
+ "",
+ false,
+ false);
+ terminal.flush();
+ }
} catch (SqlExecutionException e) {
- printExecutionException(errorMessage, e);
+ printExecutionException(e);
}
}
// --------------------------------------------------------------------------------------------
private void printExecutionException(Throwable t) {
- printExecutionException(null, t);
- }
-
- private void printExecutionException(String message, Throwable t) {
- final String finalMessage;
- if (message == null) {
- finalMessage = CliStrings.MESSAGE_SQL_EXECUTION_ERROR;
- } else {
- finalMessage = CliStrings.MESSAGE_SQL_EXECUTION_ERROR + ' ' + message;
- }
- printException(finalMessage, t);
- }
-
- private void printExecutionError(String message) {
- terminal.writer()
- .println(
- CliStrings.messageError(CliStrings.MESSAGE_SQL_EXECUTION_ERROR, message)
- .toAnsi());
- terminal.flush();
- }
-
- private void printException(String message, Throwable t) {
- LOG.warn(message, t);
- terminal.writer().println(CliStrings.messageError(message, t).toAnsi());
+ final String errorMessage = CliStrings.MESSAGE_SQL_EXECUTION_ERROR;
+ LOG.warn(errorMessage, t);
+ terminal.writer().println(CliStrings.messageError(errorMessage, t).toAnsi());
terminal.flush();
}
@@ -702,15 +481,27 @@ public class CliClient implements AutoCloseable {
terminal.flush();
}
- private void printRemovedOrDeprecatedKeyMessage(String key) {
- String msg =
- YamlConfigUtils.isRemovedKey(key)
- ? MESSAGE_REMOVED_KEY
- : String.format(
- MESSAGE_DEPRECATED_KEY,
- key,
- YamlConfigUtils.getOptionNameWithDeprecatedKey(key));
- printWarning(msg);
+ private void printSetResetConfigKeyMessage(String key, String message) {
+ boolean isRemovedKey = isRemovedKey(key);
+ boolean isDeprecatedKey = isDeprecatedKey(key);
+
+ // print warning information if the given key is removed or deprecated
+ if (isRemovedKey || isDeprecatedKey) {
+ String warningMsg =
+ isRemovedKey
+ ? MESSAGE_REMOVED_KEY
+ : String.format(
+ MESSAGE_DEPRECATED_KEY,
+ key,
+ getOptionNameWithDeprecatedKey(key));
+ printWarning(warningMsg);
+ }
+
+ // when the key is not removed, need to print normal message
+ if (!isRemovedKey) {
+ terminal.writer().println(CliStrings.messageInfo(message).toAnsi());
+ terminal.flush();
+ }
}
// --------------------------------------------------------------------------------------------
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliStrings.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliStrings.java
index 3858b9c..e750438 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliStrings.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliStrings.java
@@ -18,8 +18,6 @@
package org.apache.flink.table.client.cli;
-import org.apache.flink.table.client.cli.SqlCommandParser.SqlCommand;
-
import org.jline.utils.AttributedString;
import org.jline.utils.AttributedStringBuilder;
import org.jline.utils.AttributedStyle;
@@ -39,82 +37,81 @@ public final class CliStrings {
public static final AttributedString MESSAGE_HELP =
new AttributedStringBuilder()
.append("The following commands are available:\n\n")
- .append(formatCommand(SqlCommand.CLEAR, "Clears the current terminal."))
+ .append(formatCommand("CLEAR", "Clears the current terminal."))
.append(
formatCommand(
- SqlCommand.CREATE_TABLE,
+ "CREATE TABLE",
"Create table under current catalog and database."))
.append(
formatCommand(
- SqlCommand.DROP_TABLE,
+ "DROP TABLE",
"Drop table with optional catalog and database. Syntax: 'DROP TABLE [IF EXISTS] <name>;'"))
.append(
formatCommand(
- SqlCommand.CREATE_VIEW,
+ "CREATE VIEW",
"Creates a virtual table from a SQL query. Syntax: 'CREATE VIEW <name> AS <query>;'"))
.append(
formatCommand(
- SqlCommand.DESCRIBE,
+ "DESCRIBE",
"Describes the schema of a table with the given name."))
.append(
formatCommand(
- SqlCommand.DROP_VIEW,
+ "DROP VIEW",
"Deletes a previously created virtual table. Syntax: 'DROP VIEW <name>;'"))
.append(
formatCommand(
- SqlCommand.EXPLAIN,
+ "EXPLAIN",
"Describes the execution plan of a query or table with the given name."))
- .append(formatCommand(SqlCommand.HELP, "Prints the available commands."))
+ .append(formatCommand("HELP", "Prints the available commands."))
.append(
formatCommand(
- SqlCommand.INSERT_INTO,
+ "INSERT INTO",
"Inserts the results of a SQL SELECT query into a declared table sink."))
.append(
formatCommand(
- SqlCommand.INSERT_OVERWRITE,
+ "INSERT OVERWRITE",
"Inserts the results of a SQL SELECT query into a declared table sink and overwrite existing data."))
- .append(formatCommand(SqlCommand.QUIT, "Quits the SQL CLI client."))
+ .append(formatCommand("QUIT", "Quits the SQL CLI client."))
.append(
formatCommand(
- SqlCommand.RESET,
+ "RESET",
"Resets a session configuration property. Syntax: 'RESET <key>;'. Use 'RESET;' for reset all session properties."))
.append(
formatCommand(
- SqlCommand.SELECT,
- "Executes a SQL SELECT query on the Flink cluster."))
+ "SELECT", "Executes a SQL SELECT query on the Flink cluster."))
.append(
formatCommand(
- SqlCommand.SET,
+ "SET",
"Sets a session configuration property. Syntax: 'SET <key>=<value>;'. Use 'SET;' for listing all properties."))
.append(
formatCommand(
- SqlCommand.SHOW_FUNCTIONS,
+ "SHOW FUNCTIONS",
"Shows all user-defined and built-in functions or only user-defined functions. Syntax: 'SHOW [USER] FUNCTIONS;'"))
- .append(formatCommand(SqlCommand.SHOW_TABLES, "Shows all registered tables."))
+ .append(formatCommand("SHOW TABLES", "Shows all registered tables."))
.append(
formatCommand(
- SqlCommand.SOURCE,
+ "SOURCE",
"Reads a SQL SELECT query from a file and executes it on the Flink cluster."))
.append(
formatCommand(
- SqlCommand.USE_CATALOG,
+ "USE CATALOG",
"Sets the current catalog. The current database is set to the catalog's default one. Experimental! Syntax: 'USE CATALOG <name>;'"))
.append(
formatCommand(
- SqlCommand.USE,
+ "USE",
"Sets the current default database. Experimental! Syntax: 'USE <name>;'"))
.append(
formatCommand(
- SqlCommand.LOAD_MODULE,
+ "LOAD MODULE",
"Load a module. Syntax: 'LOAD MODULE <name> [WITH ('<key1>' = "
+ "'<value1>' [, '<key2>' = '<value2>', ...])];'"))
.append(
formatCommand(
- SqlCommand.UNLOAD_MODULE,
+ "UNLOAD MODULE",
"Unload a module. Syntax: 'UNLOAD MODULE <name>;'"))
.append(
formatCommand(
- SqlCommand.USE_MODULES,
+ "USE MODULES",
"Enable loaded modules. Syntax: 'USE MODULES <name1> [, <name2>, ...];'"))
.style(AttributedStyle.DEFAULT.underline())
.append("\nHint")
@@ -180,7 +177,7 @@ public final class CliStrings {
public static final String MESSAGE_RESET_KEY = "Session property has been reset.";
- public static final String MESSAGE_SET = "Session property has been set.";
+ public static final String MESSAGE_SET_KEY = "Session property has been set.";
public static final String MESSAGE_REMOVED_KEY = "The specified key is not supported anymore.";
@@ -197,64 +194,11 @@ public final class CliStrings {
public static final String MESSAGE_STATEMENT_SUBMITTED =
"Table update statement has been successfully submitted to the cluster:";
- public static final String MESSAGE_MAX_SIZE_EXCEEDED =
- "The given file exceeds the maximum number of characters.";
-
public static final String MESSAGE_WILL_EXECUTE = "Executing the following statement:";
public static final String MESSAGE_UNSUPPORTED_SQL = "Unsupported SQL statement.";
- public static final String MESSAGE_TABLE_CREATED = "Table has been created.";
-
- public static final String MESSAGE_TABLE_REMOVED = "Table has been removed.";
-
- public static final String MESSAGE_ALTER_TABLE_SUCCEEDED = "Alter table succeeded!";
-
- public static final String MESSAGE_ALTER_TABLE_FAILED = "Alter table failed!";
-
- public static final String MESSAGE_VIEW_CREATED = "View has been created.";
-
- public static final String MESSAGE_VIEW_REMOVED = "View has been removed.";
-
- public static final String MESSAGE_ALTER_VIEW_SUCCEEDED = "Alter view succeeded!";
-
- public static final String MESSAGE_ALTER_VIEW_FAILED = "Alter view failed!";
-
- public static final String MESSAGE_FUNCTION_CREATED = "Function has been created.";
-
- public static final String MESSAGE_FUNCTION_REMOVED = "Function has been removed.";
-
- public static final String MESSAGE_ALTER_FUNCTION_SUCCEEDED = "Alter function succeeded!";
-
- public static final String MESSAGE_ALTER_FUNCTION_FAILED = "Alter function failed!";
-
- public static final String MESSAGE_DATABASE_CREATED = "Database has been created.";
-
- public static final String MESSAGE_DATABASE_REMOVED = "Database has been removed.";
-
- public static final String MESSAGE_DATABASE_CHANGED = "Database changed.";
-
- public static final String MESSAGE_ALTER_DATABASE_SUCCEEDED = "Alter database succeeded!";
-
- public static final String MESSAGE_ALTER_DATABASE_FAILED = "Alter database failed!";
-
- public static final String MESSAGE_CATALOG_CREATED = "Catalog has been created.";
-
- public static final String MESSAGE_CATALOG_REMOVED = "Catalog has been removed.";
-
- public static final String MESSAGE_CATALOG_CHANGED = "Catalog changed.";
-
- public static final String MESSAGE_LOAD_MODULE_SUCCEEDED = "Load module succeeded!";
-
- public static final String MESSAGE_UNLOAD_MODULE_SUCCEEDED = "Unload module succeeded!";
-
- public static final String MESSAGE_USE_MODULES_SUCCEEDED = "Use modules succeeded!";
-
- public static final String MESSAGE_LOAD_MODULE_FAILED = "Load module failed!";
-
- public static final String MESSAGE_UNLOAD_MODULE_FAILED = "Unload module failed!";
-
- public static final String MESSAGE_USE_MODULES_FAILED = "Use modules failed!";
+ public static final String MESSAGE_EXECUTE_STATEMENT = "Execute statement succeed.";
// --------------------------------------------------------------------------------------------
@@ -381,10 +325,10 @@ public final class CliStrings {
return builder.toAttributedString();
}
- private static AttributedString formatCommand(SqlCommand cmd, String description) {
+ private static AttributedString formatCommand(String cmd, String description) {
return new AttributedStringBuilder()
.style(AttributedStyle.DEFAULT.bold())
- .append(cmd.toString())
+ .append(cmd)
.append("\t\t")
.style(AttributedStyle.DEFAULT)
.append(description)
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
deleted file mode 100644
index 5fc5b91..0000000
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCommandParser.java
+++ /dev/null
@@ -1,414 +0,0 @@
-/*
- * 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;
-
-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.LoadModuleOperation;
-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.ShowCurrentCatalogOperation;
-import org.apache.flink.table.operations.ShowCurrentDatabaseOperation;
-import org.apache.flink.table.operations.ShowDatabasesOperation;
-import org.apache.flink.table.operations.ShowFunctionsOperation;
-import org.apache.flink.table.operations.ShowModulesOperation;
-import org.apache.flink.table.operations.ShowPartitionsOperation;
-import org.apache.flink.table.operations.ShowTablesOperation;
-import org.apache.flink.table.operations.ShowViewsOperation;
-import org.apache.flink.table.operations.UnloadModuleOperation;
-import org.apache.flink.table.operations.UseCatalogOperation;
-import org.apache.flink.table.operations.UseDatabaseOperation;
-import org.apache.flink.table.operations.UseModulesOperation;
-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.AlterViewOperation;
-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;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
-/** Simple parser for determining the type of command and its parameters. */
-public final class SqlCommandParser {
-
- private SqlCommandParser() {
- // private
- }
-
- /**
- * Parse a sql statement and return corresponding {@link SqlCommandCall}. If the statement is
- * invalid, a {@link SqlExecutionException} will be thrown.
- *
- * @param sqlParser The sql parser instance
- * @param stmt The statement to be parsed
- * @return the corresponding SqlCommandCall.
- */
- public static SqlCommandCall parse(Parser sqlParser, String stmt) {
- // normalize
- stmt = stmt.trim();
- // remove ';' at the end
- if (stmt.endsWith(";")) {
- stmt = stmt.substring(0, stmt.length() - 1).trim();
- }
-
- // parse statement via regex matching first
- Optional<SqlCommandCall> callOpt = parseByRegexMatching(stmt);
- if (callOpt.isPresent()) {
- return callOpt.get();
- } else {
- return parseBySqlParser(sqlParser, stmt);
- }
- }
-
- private static SqlCommandCall parseBySqlParser(Parser sqlParser, String stmt) {
- List<Operation> operations;
- try {
- operations = sqlParser.parse(stmt);
- } catch (Throwable e) {
- throw new SqlExecutionException("Invalidate SQL statement.", e);
- }
- 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;
- } else if (operation instanceof DropViewOperation) {
- cmd = SqlCommand.DROP_VIEW;
- } else if (operation instanceof AlterViewOperation) {
- cmd = SqlCommand.ALTER_VIEW;
- } 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[] {((UseCatalogOperation) operation).getCatalogName()};
- } else if (operation instanceof UseDatabaseOperation) {
- cmd = SqlCommand.USE;
- operands = new String[] {((UseDatabaseOperation) operation).getDatabaseName()};
- } else if (operation instanceof ShowCatalogsOperation) {
- cmd = SqlCommand.SHOW_CATALOGS;
- operands = new String[0];
- } else if (operation instanceof ShowCurrentCatalogOperation) {
- cmd = SqlCommand.SHOW_CURRENT_CATALOG;
- operands = new String[0];
- } else if (operation instanceof ShowDatabasesOperation) {
- cmd = SqlCommand.SHOW_DATABASES;
- operands = new String[0];
- } else if (operation instanceof ShowCurrentDatabaseOperation) {
- cmd = SqlCommand.SHOW_CURRENT_DATABASE;
- operands = new String[0];
- } else if (operation instanceof ShowModulesOperation) {
- cmd = SqlCommand.SHOW_MODULES;
- } else if (operation instanceof ShowTablesOperation) {
- cmd = SqlCommand.SHOW_TABLES;
- operands = new String[0];
- } else if (operation instanceof ShowViewsOperation) {
- cmd = SqlCommand.SHOW_VIEWS;
- operands = new String[0];
- } else if (operation instanceof ShowFunctionsOperation) {
- cmd = SqlCommand.SHOW_FUNCTIONS;
- } else if (operation instanceof ShowPartitionsOperation) {
- cmd = SqlCommand.SHOW_PARTITIONS;
- } 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 LoadModuleOperation) {
- cmd = SqlCommand.LOAD_MODULE;
- } else if (operation instanceof UnloadModuleOperation) {
- cmd = SqlCommand.UNLOAD_MODULE;
- } else if (operation instanceof UseModulesOperation) {
- cmd = SqlCommand.USE_MODULES;
- } else if (operation instanceof DescribeTableOperation) {
- cmd = SqlCommand.DESCRIBE;
- operands =
- new String[] {
- ((DescribeTableOperation) operation)
- .getSqlIdentifier()
- .asSerializableString()
- };
- } else if (operation instanceof QueryOperation) {
- cmd = SqlCommand.SELECT;
- } else {
- throw new SqlExecutionException("Unknown operation: " + operation.asSummaryString());
- }
-
- return new SqlCommandCall(cmd, operands);
- }
-
- private static Optional<SqlCommandCall> parseByRegexMatching(String stmt) {
- // parse statement via regex matching
- for (SqlCommand cmd : SqlCommand.values()) {
- 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]
- + " "
- + operands[1]
- };
- }
- return new SqlCommandCall(cmd, newOperands);
- });
- }
- }
- }
- return Optional.empty();
- }
-
- // --------------------------------------------------------------------------------------------
-
- private static final Function<String[], Optional<String[]>> NO_OPERANDS =
- (operands) -> Optional.of(new String[0]);
-
- private static final Function<String[], Optional<String[]>> SINGLE_OPERAND =
- (operands) -> Optional.of(new String[] {operands[0]});
-
- private static final int DEFAULT_PATTERN_FLAGS = Pattern.CASE_INSENSITIVE | Pattern.DOTALL;
-
- /** Supported SQL commands. */
- enum SqlCommand {
- QUIT("(QUIT|EXIT)", NO_OPERANDS),
-
- CLEAR("CLEAR", NO_OPERANDS),
-
- HELP("HELP", NO_OPERANDS),
-
- SHOW_CATALOGS,
-
- SHOW_CURRENT_CATALOG,
-
- SHOW_DATABASES,
-
- SHOW_CURRENT_DATABASE,
-
- SHOW_TABLES,
-
- SHOW_VIEWS,
-
- SHOW_FUNCTIONS,
-
- SHOW_MODULES,
-
- LOAD_MODULE,
-
- UNLOAD_MODULE,
-
- USE_MODULES,
-
- SHOW_PARTITIONS,
-
- USE_CATALOG,
-
- USE,
-
- CREATE_CATALOG,
-
- DROP_CATALOG,
-
- DESCRIBE,
-
- // supports both `explain xx` and `explain plan for xx` now
- // TODO should keep `explain xx` ?
- // only match "EXPLAIN SELECT xx" and "EXPLAIN INSERT xx" here
- // "EXPLAIN PLAN FOR xx" should be parsed via sql parser
- EXPLAIN(
- "EXPLAIN\\s+(SELECT|INSERT)\\s+(.*)",
- (operands) -> {
- return Optional.of(new String[] {operands[0], operands[1]});
- }),
-
- CREATE_DATABASE,
-
- DROP_DATABASE,
-
- ALTER_DATABASE,
-
- CREATE_TABLE,
-
- DROP_TABLE,
-
- ALTER_TABLE,
-
- CREATE_VIEW,
-
- DROP_VIEW,
-
- ALTER_VIEW,
-
- CREATE_FUNCTION,
-
- DROP_FUNCTION,
-
- ALTER_FUNCTION,
-
- SELECT,
-
- INSERT_INTO,
-
- INSERT_OVERWRITE,
-
- SET(
- "SET(\\s+(\\S+)\\s*=(.*))?", // whitespace is only ignored on the left side of '='
- (operands) -> {
- if (operands.length < 3) {
- return Optional.empty();
- } else if (operands[0] == null) {
- return Optional.of(new String[0]);
- }
- return Optional.of(new String[] {operands[1], operands[2]});
- }),
-
- RESET(
- "RESET(\\s+(\\S+)\\s*)?",
- (operands) -> {
- if (operands.length < 2) {
- return Optional.empty();
- } else if (operands[0] == null) {
- return Optional.of(new String[0]);
- }
- return Optional.of(new String[] {operands[1]});
- }),
-
- SOURCE("SOURCE\\s+(.*)", SINGLE_OPERAND);
-
- 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);
- this.operandConverter = operandConverter;
- }
-
- @Override
- public String toString() {
- return super.toString().replace('_', ' ');
- }
-
- public boolean hasOperands() {
- return operandConverter != NO_OPERANDS;
- }
-
- public boolean hasRegexPattern() {
- return pattern != null;
- }
- }
-
- /** Call of SQL command with operands and command type. */
- public static class SqlCommandCall {
- public final SqlCommand command;
- public final String[] operands;
-
- public SqlCommandCall(SqlCommand command, String[] operands) {
- this.command = command;
- this.operands = operands;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- SqlCommandCall that = (SqlCommandCall) o;
- return command == that.command && Arrays.equals(operands, that.operands);
- }
-
- @Override
- public int hashCode() {
- int result = Objects.hash(command);
- result = 31 * result + Arrays.hashCode(operands);
- return result;
- }
-
- @Override
- public String toString() {
- return command + "(" + Arrays.toString(operands) + ")";
- }
- }
-}
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 f919696..0ac9261 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,7 +29,6 @@ import org.jline.utils.AttributedString;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.Arrays;
import java.util.List;
/** SQL code completer. */
@@ -37,8 +36,6 @@ public class SqlCompleter implements Completer {
private static final Logger LOG = LoggerFactory.getLogger(SqlCompleter.class);
- public static final String[] COMMAND_HINTS = getCommandHints();
-
private String sessionId;
private Executor executor;
@@ -56,17 +53,6 @@ public class SqlCompleter implements Completer {
statement = statement.substring(0, statement.length() - 1);
}
- // handle SQL client specific commands
- final String statementNormalized = statement.toUpperCase().trim();
- for (String commandHint : COMMAND_HINTS) {
- if (commandHint.startsWith(statementNormalized)
- && line.cursor() < commandHint.length()) {
- candidates.add(
- createCandidate(getCompletionHint(statementNormalized, commandHint)));
- }
- }
-
- // fallback to Table API hinting
try {
executor.completeStatement(sessionId, statement, line.cursor())
.forEach(hint -> candidates.add(createCandidate(hint)));
@@ -75,41 +61,7 @@ public class SqlCompleter implements Completer {
}
}
- private String getCompletionHint(String statementNormalized, String commandHint) {
- if (statementNormalized.length() == 0) {
- return commandHint;
- }
- int cursorPos = statementNormalized.length() - 1;
- int returnStartPos;
- if (Character.isWhitespace(commandHint.charAt(cursorPos + 1))) {
- returnStartPos = Math.min(commandHint.length() - 1, cursorPos + 2);
- } else {
- returnStartPos = cursorPos;
- while (returnStartPos > 0
- && !Character.isWhitespace(commandHint.charAt(returnStartPos - 1))) {
- returnStartPos--;
- }
- }
-
- return commandHint.substring(returnStartPos);
- }
-
private Candidate createCandidate(String hint) {
return new Candidate(AttributedString.stripAnsi(hint), hint, null, null, null, null, true);
}
-
- private static String[] getCommandHints() {
- 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 f306155..f4fc20e 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
@@ -19,7 +19,8 @@
package org.apache.flink.table.client.gateway;
import org.apache.flink.table.api.TableResult;
-import org.apache.flink.table.delegation.Parser;
+import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.operations.QueryOperation;
import org.apache.flink.types.Row;
import javax.annotation.Nullable;
@@ -81,17 +82,19 @@ public interface Executor {
void setSessionProperty(String sessionId, String key, String value)
throws SqlExecutionException;
- /** Executes a SQL statement, and return {@link TableResult} as execution result. */
- TableResult executeSql(String sessionId, String statement) throws SqlExecutionException;
-
- /** Returns a sql parser instance. */
- Parser getSqlParser(String sessionId);
+ /** Parse a SQL statement to {@link Operation}. */
+ Operation parseStatement(String sessionId, String statement) throws SqlExecutionException;
/** Returns a list of completion hints for the given statement at the given position. */
List<String> completeStatement(String sessionId, String statement, int position);
+ /** Executes an operation, and return {@link TableResult} as execution result. */
+ TableResult executeOperation(String sessionId, Operation operation)
+ throws SqlExecutionException;
+
/** Submits a Flink SQL query job (detached) and returns the result descriptor. */
- ResultDescriptor executeQuery(String sessionId, String query) throws SqlExecutionException;
+ ResultDescriptor executeQuery(String sessionId, QueryOperation query)
+ throws SqlExecutionException;
/** Asks for the next changelog results (non-blocking). */
TypedResult<List<Row>> retrieveResultChanges(String sessionId, String resultId)
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 4ec6c7d..bb457ff 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
@@ -23,9 +23,7 @@ import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
-import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.api.internal.TableEnvironmentInternal;
-import org.apache.flink.table.catalog.UnresolvedIdentifier;
import org.apache.flink.table.client.config.ResultMode;
import org.apache.flink.table.client.gateway.Executor;
import org.apache.flink.table.client.gateway.ResultDescriptor;
@@ -38,8 +36,8 @@ 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.expressions.ResolvedExpression;
import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.operations.QueryOperation;
import org.apache.flink.types.Row;
import org.slf4j.Logger;
@@ -54,6 +52,7 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import static org.apache.flink.configuration.ExecutionOptions.RUNTIME_MODE;
+import static org.apache.flink.table.client.cli.CliStrings.MESSAGE_SQL_EXECUTION_ERROR;
import static org.apache.flink.table.client.config.SqlClientOptions.EXECUTION_RESULT_MODE;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -160,45 +159,22 @@ public class LocalExecutor implements Executor {
}
@Override
- public TableResult executeSql(String sessionId, String statement) throws SqlExecutionException {
+ public Operation parseStatement(String sessionId, String statement)
+ throws SqlExecutionException {
final ExecutionContext context = getExecutionContext(sessionId);
- final TableEnvironment tEnv = context.getTableEnvironment();
+ final TableEnvironment tableEnv = context.getTableEnvironment();
+ Parser parser = ((TableEnvironmentInternal) tableEnv).getParser();
+
+ List<Operation> operations;
try {
- return context.wrapClassLoader(() -> tEnv.executeSql(statement));
+ operations = context.wrapClassLoader(() -> parser.parse(statement));
} catch (Exception e) {
- throw new SqlExecutionException("Could not execute statement: " + statement, e);
+ throw new SqlExecutionException("Failed to parse statement: " + statement, e);
}
- }
-
- @Override
- public Parser getSqlParser(String sessionId) {
- final ExecutionContext context = getExecutionContext(sessionId);
- final TableEnvironment tableEnv = context.getTableEnvironment();
- final Parser parser = ((TableEnvironmentInternal) tableEnv).getParser();
- return new Parser() {
- @Override
- public List<Operation> parse(String statement) {
- return context.wrapClassLoader(() -> parser.parse(statement));
- }
-
- @Override
- public UnresolvedIdentifier parseIdentifier(String identifier) {
- return context.wrapClassLoader(() -> parser.parseIdentifier(identifier));
- }
-
- @Override
- public ResolvedExpression parseSqlExpression(
- String sqlExpression, TableSchema inputSchema) {
- return context.wrapClassLoader(
- () -> parser.parseSqlExpression(sqlExpression, inputSchema));
- }
-
- @Override
- public String[] getCompletionHints(String statement, int position) {
- return context.wrapClassLoader(
- () -> parser.getCompletionHints(statement, position));
- }
- };
+ if (operations.isEmpty()) {
+ throw new SqlExecutionException("Failed to parse statement: " + statement);
+ }
+ return operations.get(0);
}
@Override
@@ -222,11 +198,35 @@ public class LocalExecutor implements Executor {
}
@Override
- public ResultDescriptor executeQuery(String sessionId, String query)
+ public TableResult executeOperation(String sessionId, Operation operation)
+ throws SqlExecutionException {
+ final ExecutionContext context = getExecutionContext(sessionId);
+ final TableEnvironmentInternal tEnv =
+ (TableEnvironmentInternal) context.getTableEnvironment();
+ try {
+ return context.wrapClassLoader(() -> tEnv.executeInternal(operation));
+ } catch (Exception e) {
+ throw new SqlExecutionException(MESSAGE_SQL_EXECUTION_ERROR, e);
+ }
+ }
+
+ @Override
+ public ResultDescriptor executeQuery(String sessionId, QueryOperation query)
throws SqlExecutionException {
+ final TableResult tableResult = executeOperation(sessionId, query);
final SessionContext context = getSessionContext(sessionId);
- return executeQueryInternal(
- context.getExecutionContext(), query, context.getReadableConfig());
+ final ReadableConfig config = context.getReadableConfig();
+ final DynamicResult result = resultStore.createResult(config, tableResult);
+ checkArgument(tableResult.getJobClient().isPresent());
+ String jobId = tableResult.getJobClient().get().getJobID().toString();
+ // store the result under the JobID
+ resultStore.storeResult(jobId, result);
+ return new ResultDescriptor(
+ jobId,
+ tableResult.getResolvedSchema(),
+ result.isMaterialized(),
+ config.get(EXECUTION_RESULT_MODE).equals(ResultMode.TABLEAU),
+ config.get(RUNTIME_MODE).equals(RuntimeExecutionMode.STREAMING));
}
@Override
@@ -272,12 +272,6 @@ public class LocalExecutor implements Executor {
@Override
public void cancelQuery(String sessionId, String resultId) throws SqlExecutionException {
- cancelQueryInternal(resultId);
- }
-
- // --------------------------------------------------------------------------------------------
-
- private void cancelQueryInternal(String resultId) {
final DynamicResult result = resultStore.getResult(resultId);
if (result == null) {
throw new SqlExecutionException(
@@ -294,27 +288,4 @@ public class LocalExecutor implements Executor {
}
resultStore.removeResult(resultId);
}
-
- private ResultDescriptor executeQueryInternal(
- ExecutionContext context, String query, ReadableConfig config) {
- final TableResult tableResult;
- try {
- tableResult =
- context.wrapClassLoader(() -> context.getTableEnvironment().executeSql(query));
- } catch (Throwable t) {
- // catch everything such that the query does not crash the executor
- throw new SqlExecutionException("Invalid SQL statement.", t);
- }
- final DynamicResult result = resultStore.createResult(config, tableResult);
- checkArgument(tableResult.getJobClient().isPresent());
- String jobId = tableResult.getJobClient().get().getJobID().toString();
- // store the result under the JobID
- resultStore.storeResult(jobId, result);
- return new ResultDescriptor(
- jobId,
- tableResult.getResolvedSchema(),
- result.isMaterialized(),
- config.get(EXECUTION_RESULT_MODE).equals(ResultMode.TABLEAU),
- config.get(RUNTIME_MODE).equals(RuntimeExecutionMode.STREAMING));
- }
}
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 1655faf..f9f32b8 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
@@ -36,6 +36,9 @@ import org.apache.flink.table.client.gateway.SqlExecutionException;
import org.apache.flink.table.client.gateway.TypedResult;
import org.apache.flink.table.client.gateway.context.DefaultContext;
import org.apache.flink.table.client.gateway.context.SessionContext;
+import org.apache.flink.table.operations.ModifyOperation;
+import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.operations.QueryOperation;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;
import org.apache.flink.util.TestLogger;
@@ -96,27 +99,15 @@ public class CliClientTest extends TestLogger {
@Test
public void testSqlCompletion() throws IOException {
- 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(
- "show ", 4, Collections.singletonList("HintA"), Collections.singletonList("QUIT;"));
- verifySqlCompletion(
- "show modules", 13, Collections.emptyList(), Collections.singletonList("QUIT;"));
+ verifySqlCompletion("", 0, Arrays.asList("CLEAR", "HELP", "EXIT", "QUIT", "RESET", "SET"));
+ verifySqlCompletion("SELE", 4, Collections.emptyList());
+ verifySqlCompletion("QU", 2, Collections.singletonList("QUIT"));
+ verifySqlCompletion("qu", 2, Collections.singletonList("QUIT"));
+ verifySqlCompletion(" qu", 2, Collections.singletonList("QUIT"));
+ verifySqlCompletion("set ", 3, Collections.emptyList());
+ verifySqlCompletion("show t ", 6, Collections.emptyList());
+ verifySqlCompletion("show ", 4, Collections.emptyList());
+ verifySqlCompletion("show modules", 12, Collections.emptyList());
}
@Test
@@ -162,11 +153,7 @@ public class CliClientTest extends TestLogger {
}
}
- private void verifySqlCompletion(
- String statement,
- int position,
- List<String> expectedHints,
- List<String> notExpectedHints)
+ private void verifySqlCompletion(String statement, int position, List<String> expectedHints)
throws IOException {
final MockExecutor mockExecutor = new MockExecutor();
String sessionId = mockExecutor.openSession("test-session");
@@ -188,11 +175,6 @@ public class CliClientTest extends TestLogger {
assertEquals(statement, mockExecutor.receivedStatement);
assertEquals(position, mockExecutor.receivedPosition);
- assertTrue(results.contains("HintA"));
- assertTrue(results.contains("Hint B"));
-
- results.retainAll(notExpectedHints);
- assertEquals(0, results.size());
}
}
@@ -249,14 +231,12 @@ public class CliClientTest extends TestLogger {
throws SqlExecutionException {}
@Override
- public TableResult executeSql(String sessionId, String statement)
+ public TableResult executeOperation(String sessionId, Operation operation)
throws SqlExecutionException {
- receivedStatement = statement;
if (failExecution) {
throw new SqlExecutionException("Fail execution.");
}
- if (statement.toLowerCase().startsWith("insert ")
- || statement.toLowerCase().startsWith("select ")) {
+ if (operation instanceof ModifyOperation || operation instanceof QueryOperation) {
return new TestTableResult(
new TestingJobClient(),
ResultKind.SUCCESS_WITH_CONTENT,
@@ -268,19 +248,22 @@ public class CliClientTest extends TestLogger {
}
@Override
- public org.apache.flink.table.delegation.Parser getSqlParser(String sessionId) {
- return helper.getSqlParser();
+ public Operation parseStatement(String sessionId, String statement)
+ throws SqlExecutionException {
+ receivedStatement = statement;
+ List<Operation> ops = helper.getSqlParser().parse(statement);
+ return ops.get(0);
}
@Override
public List<String> completeStatement(String sessionId, String statement, int position) {
receivedStatement = statement;
receivedPosition = position;
- return Arrays.asList("HintA", "Hint B");
+ return Arrays.asList(helper.getSqlParser().getCompletionHints(statement, position));
}
@Override
- public ResultDescriptor executeQuery(String sessionId, String query)
+ public ResultDescriptor executeQuery(String sessionId, QueryOperation query)
throws SqlExecutionException {
return null;
}
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 be0a2b0..a4069aa 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
@@ -26,7 +26,8 @@ import org.apache.flink.table.client.gateway.Executor;
import org.apache.flink.table.client.gateway.ResultDescriptor;
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.table.operations.Operation;
+import org.apache.flink.table.operations.QueryOperation;
import org.apache.flink.types.Row;
import org.jline.utils.AttributedString;
@@ -93,24 +94,19 @@ public class CliResultViewTest {
true);
Thread resultViewRunner = null;
- CliClient cli = null;
- try {
- cli =
- new CliClient(
- TerminalUtils.createDummyTerminal(),
- sessionId,
- executor,
- File.createTempFile("history", "tmp").toPath(),
- null);
+ try (CliClient cli =
+ new CliClient(
+ TerminalUtils.createDummyTerminal(),
+ sessionId,
+ executor,
+ File.createTempFile("history", "tmp").toPath(),
+ null)) {
resultViewRunner = new Thread(new TestingCliResultView(cli, descriptor, isTableMode));
resultViewRunner.start();
} finally {
if (resultViewRunner != null && !resultViewRunner.isInterrupted()) {
resultViewRunner.interrupt();
}
- if (cli != null) {
- cli.close();
- }
}
assertTrue(
@@ -161,23 +157,24 @@ public class CliResultViewTest {
throws SqlExecutionException {}
@Override
- public TableResult executeSql(String sessionId, String statement)
+ public Operation parseStatement(String sessionId, String statement)
throws SqlExecutionException {
return null;
}
@Override
- public Parser getSqlParser(String sessionId) {
+ public List<String> completeStatement(String sessionId, String statement, int position) {
return null;
}
@Override
- public List<String> completeStatement(String sessionId, String statement, int position) {
+ public TableResult executeOperation(String sessionId, Operation operation)
+ throws SqlExecutionException {
return null;
}
@Override
- public ResultDescriptor executeQuery(String sessionId, String query)
+ public ResultDescriptor executeQuery(String sessionId, QueryOperation query)
throws SqlExecutionException {
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
deleted file mode 100644
index 7cb855d..0000000
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/SqlCommandParserTest.java
+++ /dev/null
@@ -1,586 +0,0 @@
-/*
- * 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;
-
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.table.api.SqlDialect;
-import org.apache.flink.table.api.TableEnvironment;
-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.Collections;
-import java.util.List;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-/** Tests for {@link SqlCommandParser}. */
-public class SqlCommandParserTest {
-
- private Parser parser;
- private TableEnvironment tableEnv;
-
- @Before
- public void setup() {
- SqlParserHelper helper = new SqlParserHelper();
- helper.registerTables();
- parser = helper.getSqlParser();
- tableEnv = helper.getTableEnv();
- }
-
- @Test
- 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.DESCRIBE,
- "`default_catalog`.`default_database`.`MyTable`")
- .cannotParseComment(),
- TestItem.validSql(
- "DESC MyTable ",
- SqlCommand.DESCRIBE,
- "`default_catalog`.`default_database`.`MyTable`")
- .cannotParseComment(),
- TestItem.invalidSql(
- "DESC ", // no table name
- SqlExecutionException.class,
- "Non-query expression encountered in illegal context"),
- // 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
- SqlExecutionException.class,
- "Encountered \"<EOF>\" "),
- // explain xx
- TestItem.validSql(
- "EXPLAIN SELECT a FROM MyTable",
- SqlCommand.EXPLAIN,
- "EXPLAIN PLAN FOR SELECT a FROM MyTable")
- .cannotParseComment(),
- TestItem.validSql(
- "EXPLAIN INSERT INTO MySink(a) SELECT a FROM MyTable",
- SqlCommand.EXPLAIN,
- "EXPLAIN PLAN FOR INSERT INTO MySink(a) SELECT a FROM MyTable")
- .cannotParseComment(),
- TestItem.invalidSql(
- "EXPLAIN ", // no query
- SqlExecutionException.class,
- "Encountered \"<EOF>\""),
- // 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.invalidSql(
- "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,
- "CREATE VIEW x AS SELECT 1+1"),
- TestItem.validSql(
- "CREATE VIEW x AS SELECT 1+1 FROM MyTable",
- SqlCommand.CREATE_VIEW,
- "CREATE VIEW x AS SELECT 1+1 FROM MyTable"),
- TestItem.invalidSql(
- "CREATE VIEW x SELECT 1+1 ", // missing AS
- SqlExecutionException.class,
- "Encountered \"SELECT\""),
- // drop view xx
- TestItem.validSql(
- "DROP VIEW TestView1", SqlCommand.DROP_VIEW, "DROP VIEW TestView1"),
- TestItem.invalidSql(
- "DROP VIEW ", // missing name
- SqlExecutionException.class,
- "Encountered \"<EOF>\""),
- // alter view
- TestItem.validSql(
- SqlDialect.HIVE,
- "ALTER VIEW MyView RENAME TO MyView1",
- SqlCommand.ALTER_VIEW,
- "ALTER VIEW MyView RENAME TO MyView1"),
- // 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(),
- // reset xx.yy
- TestItem.validSql("reset xx.yy;", SqlCommand.RESET, "xx.yy")
- .cannotParseComment(),
- // reset xx-yy
- TestItem.validSql("reset xx-yy;", SqlCommand.RESET, "xx-yy")
- .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"),
- TestItem.invalidSql(
- "use catalog ", // no catalog name
- SqlExecutionException.class,
- "Encountered \"<EOF>\""),
- // 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 table",
- SqlExecutionException.class,
- "Encountered \"<EOF>\""),
- TestItem.invalidSql(
- "CREATE tables",
- SqlExecutionException.class,
- "Encountered \"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", SqlExecutionException.class, "Encountered \"<EOF>\""),
- TestItem.invalidSql(
- "DROP tables",
- SqlExecutionException.class,
- "Encountered \"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 current catalog
- TestItem.validSql("show current catalog", SqlCommand.SHOW_CURRENT_CATALOG),
- TestItem.validSql(
- "show current catalog", SqlCommand.SHOW_CURRENT_CATALOG),
- // show databases
- TestItem.validSql("SHOW DATABASES;", SqlCommand.SHOW_DATABASES),
- TestItem.validSql(" SHOW DATABASES ;", SqlCommand.SHOW_DATABASES),
- // show current database
- TestItem.validSql(
- "show current database", SqlCommand.SHOW_CURRENT_DATABASE),
- // show views
- TestItem.validSql("SHOW VIEWS;", SqlCommand.SHOW_VIEWS),
- TestItem.validSql(" SHOW VIEWS ;", SqlCommand.SHOW_VIEWS),
- // load module with module name as identifier
- TestItem.validSql(
- "LOAD MODULE dummy", SqlCommand.LOAD_MODULE, "LOAD MODULE dummy"),
- // load module with module name as reversed keyword
- TestItem.validSql(
- "LOAD MODULE `MODULE`",
- SqlCommand.LOAD_MODULE,
- "LOAD MODULE `MODULE`"),
- // load module with module name as literal
- TestItem.invalidSql(
- "LOAD MODULE 'dummy'",
- SqlExecutionException.class,
- "Encountered \"\\'dummy\\'\""),
- TestItem.validSql(
- "LOAD MODULE dummy WITH ('dummy-version' = '1')",
- SqlCommand.LOAD_MODULE,
- "LOAD MODULE dummy WITH ('dummy-version' = '1')"),
- // unload module
- TestItem.validSql(
- "UNLOAD MODULE dummy",
- SqlCommand.UNLOAD_MODULE,
- "UNLOAD MODULE dummy"),
- // use one module
- TestItem.validSql(
- "USE MODULES dummy", SqlCommand.USE_MODULES, "USE MODULES dummy"),
- // use multiple modules
- TestItem.validSql(
- "USE MODULES x, y, z",
- SqlCommand.USE_MODULES,
- "USE MODULES x, y, z"),
- // use modules with module names as reserved keywords
- TestItem.validSql(
- "USE MODULES `MODULE`, `MODULES`",
- SqlCommand.USE_MODULES,
- "USE MODULES `MODULE`, `MODULES`"),
- // use modules with module names as literals
- TestItem.invalidSql(
- "USE MODULES 'dummy'",
- SqlExecutionException.class,
- "Encountered \"\\'dummy\\'\""),
- TestItem.invalidSql(
- "USE MODULES",
- SqlExecutionException.class,
- "Encountered \"<EOF>\""),
- // show modules
- TestItem.validSql("SHOW MODULES", SqlCommand.SHOW_MODULES, "SHOW MODULES"),
- // show full modules
- TestItem.validSql(
- "SHOW FULL MODULES", SqlCommand.SHOW_MODULES, "SHOW FULL MODULES"),
- // show functions
- TestItem.validSql(
- "SHOW FUNCTIONS;", SqlCommand.SHOW_FUNCTIONS, "SHOW FUNCTIONS"),
- TestItem.validSql(
- " SHOW FUNCTIONS ;",
- SqlCommand.SHOW_FUNCTIONS,
- "SHOW FUNCTIONS"),
- // show user functions
- TestItem.validSql(
- "SHOW USER FUNCTIONS;",
- SqlCommand.SHOW_FUNCTIONS,
- "SHOW USER FUNCTIONS"),
- TestItem.validSql(
- " SHOW USER FUNCTIONS ;",
- SqlCommand.SHOW_FUNCTIONS,
- "SHOW USER FUNCTIONS"),
- // Test create function.
- TestItem.invalidSql(
- "CREATE FUNCTION ",
- SqlExecutionException.class,
- "Encountered \"<EOF>\""),
- TestItem.invalidSql(
- "CREATE FUNCTIONS ",
- SqlExecutionException.class,
- "Encountered \"FUNCTIONS\""),
- TestItem.invalidSql(
- "CREATE FUNCTIONS ",
- SqlExecutionException.class,
- "Encountered \"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 func1 as 'class_name' LANGUAGE JAVA",
- SqlCommand.CREATE_FUNCTION,
- "CREATE TEMPORARY SYSTEM FUNCTION func1 as 'class_name' LANGUAGE JAVA"),
- // drop function xx
- TestItem.invalidSql(
- "DROP FUNCTION ",
- SqlExecutionException.class,
- "Encountered \"<EOF>\""),
- TestItem.invalidSql(
- "DROP FUNCTIONS ",
- SqlExecutionException.class,
- "Encountered \"FUNCTIONS\""),
- TestItem.invalidSql(
- "DROP FUNCTIONS ",
- SqlExecutionException.class,
- "Encountered \"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 ",
- SqlExecutionException.class,
- "Encountered \"<EOF>\""),
- TestItem.invalidSql(
- "ALTER FUNCTIONS ",
- SqlExecutionException.class,
- "Encountered \"FUNCTIONS\""),
- TestItem.invalidSql(
- "ALTER FUNCTIONS ",
- SqlExecutionException.class,
- "Encountered \"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.invalidSql(
- "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) {
- tableEnv.getConfig().setSqlDialect(item.sqlDialect);
- runTestItem(item);
- }
- }
-
- @Test
- public void testHiveCommands() throws Exception {
- List<TestItem> testItems =
- Collections.singletonList(
- // show partitions
- TestItem.validSql(
- SqlDialect.HIVE,
- "SHOW PARTITIONS t1",
- SqlCommand.SHOW_PARTITIONS,
- "SHOW PARTITIONS t1"));
- for (TestItem item : testItems) {
- tableEnv.getConfig().setSqlDialect(item.sqlDialect);
- runTestItem(item);
- }
- }
-
- private void runTestItem(TestItem item) {
- Tuple2<Boolean, SqlCommandCall> checkFlagAndActualCall = parseSqlAndCheckException(item);
- if (!checkFlagAndActualCall.f0) {
- return;
- }
- SqlCommandCall actualCall = checkFlagAndActualCall.f1;
- assertNotNull(item.expectedCmd);
- assertEquals(
- "test statement: " + item.sql,
- new SqlCommandCall(item.expectedCmd, item.expectedOperands),
- actualCall);
-
- String stmtWithComment = "-- comments \n " + item.sql;
- try {
- actualCall = SqlCommandParser.parse(parser, stmtWithComment);
- } catch (SqlExecutionException e) {
- if (!item.cannotParseComment) {
- fail("test statement: " + item.sql);
- }
- return;
- }
- assertEquals(item.expectedCmd, actualCall.command);
- }
-
- private Tuple2<Boolean, SqlCommandCall> parseSqlAndCheckException(TestItem item) {
- SqlCommandCall call = null;
- 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, null);
- }
-
- private static class TestItem {
- private final String sql;
- 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 SqlDialect sqlDialect = SqlDialect.DEFAULT;
-
- private TestItem(String sql) {
- this.sql = sql;
- }
-
- public static TestItem invalidSql(
- String sql, Class<? extends Throwable> expectedException, String exceptedMsg) {
- TestItem testItem = new TestItem(sql);
- testItem.expectedException = expectedException;
- testItem.expectedExceptionMsg = exceptedMsg;
- return testItem;
- }
-
- public static TestItem validSql(
- String sql, SqlCommand expectedCmd, String... expectedOperands) {
- TestItem testItem = new TestItem(sql);
- testItem.expectedCmd = expectedCmd;
- testItem.expectedOperands = expectedOperands;
- testItem.cannotParseComment = false; // default is false
- return testItem;
- }
-
- public static TestItem validSql(
- SqlDialect sqlDialect,
- String sql,
- SqlCommand expectedCmd,
- String... expectedOperands) {
- TestItem testItem = new TestItem(sql);
- testItem.expectedCmd = expectedCmd;
- testItem.expectedOperands = expectedOperands;
- testItem.cannotParseComment = false; // default is false
- testItem.sqlDialect = sqlDialect;
- return testItem;
- }
-
- public TestItem cannotParseComment() {
- cannotParseComment = true;
- return this;
- }
-
- @Override
- public String toString() {
- return this.sql;
- }
- }
-}
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 422c7e0..526ffe8 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
@@ -23,7 +23,8 @@ import org.apache.flink.table.client.gateway.Executor;
import org.apache.flink.table.client.gateway.ResultDescriptor;
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.table.operations.Operation;
+import org.apache.flink.table.operations.QueryOperation;
import org.apache.flink.types.Row;
import org.apache.flink.util.function.SupplierWithException;
@@ -115,22 +116,24 @@ class TestingExecutor implements Executor {
}
@Override
- public TableResult executeSql(String sessionId, String statement) throws SqlExecutionException {
+ public List<String> completeStatement(String sessionId, String statement, int position) {
throw new UnsupportedOperationException("Not implemented.");
}
@Override
- public Parser getSqlParser(String sessionId) {
- return helper.getSqlParser();
+ public TableResult executeOperation(String sessionId, Operation operation)
+ throws SqlExecutionException {
+ throw new UnsupportedOperationException("Not implemented.");
}
@Override
- public List<String> completeStatement(String sessionId, String statement, int position) {
+ public Operation parseStatement(String sessionId, String statement)
+ throws SqlExecutionException {
throw new UnsupportedOperationException("Not implemented.");
}
@Override
- public ResultDescriptor executeQuery(String sessionId, String query)
+ public ResultDescriptor executeQuery(String sessionId, QueryOperation query)
throws SqlExecutionException {
throw new UnsupportedOperationException("Not implemented.");
}
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/DependencyTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/DependencyTest.java
index df582d3..3e538f5 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/DependencyTest.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/DependencyTest.java
@@ -44,15 +44,16 @@ import org.apache.flink.table.catalog.hive.HiveTestUtils;
import org.apache.flink.table.catalog.hive.factories.HiveCatalogFactory;
import org.apache.flink.table.catalog.hive.factories.HiveCatalogFactoryOptions;
import org.apache.flink.table.client.config.Environment;
+import org.apache.flink.table.client.gateway.Executor;
import org.apache.flink.table.client.gateway.context.DefaultContext;
import org.apache.flink.table.client.gateway.utils.EnvironmentFileUtil;
import org.apache.flink.table.client.gateway.utils.TestTableSinkFactoryBase;
import org.apache.flink.table.client.gateway.utils.TestTableSourceFactoryBase;
-import org.apache.flink.table.delegation.Parser;
import org.apache.flink.table.factories.CatalogFactory;
import org.apache.flink.table.factories.ModuleFactory;
import org.apache.flink.table.module.Module;
import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.operations.QueryOperation;
import org.apache.flink.table.types.DataType;
import org.apache.flink.types.Row;
import org.apache.flink.util.CollectionUtil;
@@ -93,7 +94,8 @@ public class DependencyTest {
final LocalExecutor executor = createLocalExecutor();
String sessionId = executor.openSession("test-session");
try {
- final TableResult tableResult = executor.executeSql(sessionId, "DESCRIBE TableNumber1");
+ final TableResult tableResult =
+ executeSql(executor, sessionId, "DESCRIBE TableNumber1");
assertEquals(
tableResult.getResolvedSchema(),
ResolvedSchema.physical(
@@ -128,11 +130,11 @@ public class DependencyTest {
final LocalExecutor executor = createLocalExecutor();
String sessionId = executor.openSession("test-session");
try {
- final Parser sqlParser = executor.getSqlParser(sessionId);
- List<Operation> operations =
- sqlParser.parse("SELECT IntegerField1, StringField1 FROM TableNumber1");
+ Operation operation =
+ executor.parseStatement(
+ sessionId, "SELECT IntegerField1, StringField1 FROM TableNumber1");
- assertTrue(operations != null && operations.size() == 1);
+ assertTrue(operation instanceof QueryOperation);
} finally {
executor.closeSession(sessionId);
}
@@ -159,6 +161,11 @@ public class DependencyTest {
return new LocalExecutor(defaultContext);
}
+ private TableResult executeSql(Executor executor, String sessionId, String sql) {
+ Operation operation = executor.parseStatement(sessionId, sql);
+ return executor.executeOperation(sessionId, operation);
+ }
+
// --------------------------------------------------------------------------------------------
/** Table source that can be discovered if classloading is correct. */
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java
index 6826f70..fd9b4cb 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java
@@ -38,6 +38,8 @@ import org.apache.flink.table.client.gateway.utils.EnvironmentFileUtil;
import org.apache.flink.table.client.gateway.utils.SimpleCatalogFactory;
import org.apache.flink.table.client.gateway.utils.TestUserClassLoaderJar;
import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.operations.QueryOperation;
import org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory;
import org.apache.flink.test.util.MiniClusterWithClientResource;
import org.apache.flink.test.util.TestBaseUtils;
@@ -175,7 +177,8 @@ public class LocalExecutorITCase extends TestLogger {
try {
// start job and retrieval
final ResultDescriptor desc =
- executor.executeQuery(
+ executeQuery(
+ executor,
sessionId,
"SELECT scalarUDF(IntegerField1), StringField1, 'ABC' FROM TableNumber1");
@@ -228,7 +231,8 @@ public class LocalExecutorITCase extends TestLogger {
for (int i = 0; i < 3; i++) {
// start job and retrieval
final ResultDescriptor desc =
- executor.executeQuery(
+ executeQuery(
+ executor,
sessionId,
"SELECT scalarUDF(IntegerField1), StringField1 FROM TableNumber1");
@@ -350,7 +354,7 @@ public class LocalExecutorITCase extends TestLogger {
try {
final ResultDescriptor desc =
- executor.executeQuery(sessionId, "SELECT *, 'ABC' FROM TestView1");
+ executeQuery(executor, sessionId, "SELECT *, 'ABC' FROM TestView1");
assertTrue(desc.isMaterialized());
@@ -401,7 +405,7 @@ public class LocalExecutorITCase extends TestLogger {
try {
for (int i = 0; i < 3; i++) {
final ResultDescriptor desc =
- executor.executeQuery(sessionId, "SELECT * FROM TestView1");
+ executeQuery(executor, sessionId, "SELECT * FROM TestView1");
assertTrue(desc.isMaterialized());
@@ -439,7 +443,7 @@ public class LocalExecutorITCase extends TestLogger {
assertEquals("test-session", sessionId);
try {
- executor.executeSql(sessionId, "CREATE FUNCTION LowerUDF AS 'LowerUDF'");
+ executeSql(executor, sessionId, "CREATE FUNCTION LowerUDF AS 'LowerUDF'");
// Case 1: Registered sink
// Case 1.1: Registered sink with uppercase insert into keyword.
// FLINK-18302: wrong classloader when INSERT INTO with UDF
@@ -457,17 +461,18 @@ public class LocalExecutorITCase extends TestLogger {
executeAndVerifySinkResult(executor, sessionId, statement2, csvOutputPath);
// Case 2: Temporary sink
- executor.executeSql(sessionId, "use catalog `simple-catalog`");
- executor.executeSql(sessionId, "use default_database");
+ executeSql(executor, sessionId, "use catalog `simple-catalog`");
+ executeSql(executor, sessionId, "use default_database");
// create temporary sink
- executor.executeSql(
+ executeSql(
+ executor,
sessionId,
"CREATE TEMPORARY TABLE MySink (id int, str VARCHAR) WITH ('connector' = 'COLLECTION')");
final String statement3 = "INSERT INTO MySink select * from `test-table`";
// all queries are pipelined to an in-memory sink, check it is properly registered
final ResultDescriptor otherCatalogDesc =
- executor.executeQuery(sessionId, "SELECT * FROM `test-table`");
+ executeQuery(executor, sessionId, "SELECT * FROM `test-table`");
final List<String> otherCatalogResults =
retrieveTableResult(executor, sessionId, otherCatalogDesc.getResultId());
@@ -488,6 +493,16 @@ public class LocalExecutorITCase extends TestLogger {
// Helper method
// --------------------------------------------------------------------------------------------
+ private TableResult executeSql(Executor executor, String sessionId, String sql) {
+ Operation operation = executor.parseStatement(sessionId, sql);
+ return executor.executeOperation(sessionId, operation);
+ }
+
+ private ResultDescriptor executeQuery(Executor executor, String sessionId, String query) {
+ Operation operation = executor.parseStatement(sessionId, query);
+ return executor.executeQuery(sessionId, (QueryOperation) operation);
+ }
+
private LocalExecutor createLocalExecutor(Environment environment, List<URL> dependencies) {
DefaultContext defaultContext =
@@ -512,7 +527,7 @@ public class LocalExecutorITCase extends TestLogger {
try {
// start job and retrieval
- final ResultDescriptor desc = executor.executeQuery(sessionId, query);
+ final ResultDescriptor desc = executeQuery(executor, sessionId, query);
assertTrue(desc.isMaterialized());
@@ -543,7 +558,7 @@ public class LocalExecutorITCase extends TestLogger {
private void executeAndVerifySinkResult(
Executor executor, String sessionId, String statement, String resultPath)
throws Exception {
- final TableResult tableResult = executor.executeSql(sessionId, statement);
+ final TableResult tableResult = executeSql(executor, sessionId, statement);
checkState(tableResult.getJobClient().isPresent());
// wait for job completion
tableResult.await();
diff --git a/flink-table/flink-sql-client/src/test/resources/sql/catalog_database.q b/flink-table/flink-sql-client/src/test/resources/sql/catalog_database.q
index 1d291c5..6f31601 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/catalog_database.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/catalog_database.q
@@ -48,7 +48,7 @@ org.apache.flink.table.api.ValidationException: Catalog my does not exist
# ==========================================================================
create catalog c1 with ('type'='generic_in_memory');
-[INFO] Catalog has been created.
+[INFO] Execute statement succeed.
!info
show catalogs;
@@ -71,7 +71,7 @@ show current catalog;
!ok
use catalog c1;
-[INFO] Catalog changed.
+[INFO] Execute statement succeed.
!info
show current catalog;
@@ -84,7 +84,7 @@ show current catalog;
!ok
drop catalog default_catalog;
-[INFO] Catalog has been removed.
+[INFO] Execute statement succeed.
!info
# ==========================================================================
@@ -92,7 +92,7 @@ drop catalog default_catalog;
# ==========================================================================
create database db1;
-[INFO] Database has been created.
+[INFO] Execute statement succeed.
!info
show databases;
@@ -115,7 +115,7 @@ show current database;
!ok
use db1;
-[INFO] Database changed.
+[INFO] Execute statement succeed.
!info
show current database;
@@ -128,11 +128,11 @@ show current database;
!ok
create database db2 comment 'db2_comment' with ('k1' = 'v1');
-[INFO] Database has been created.
+[INFO] Execute statement succeed.
!info
alter database db2 set ('k1' = 'a', 'k2' = 'b');
-[INFO] Alter database succeeded!
+[INFO] Execute statement succeed.
!info
# TODO: show database properties when we support DESCRIBE DATABSE
@@ -149,7 +149,7 @@ show databases;
!ok
drop database if exists db2;
-[INFO] Database has been removed.
+[INFO] Execute statement succeed.
!info
show databases;
@@ -167,23 +167,23 @@ show databases;
# ==========================================================================
create catalog `mod` with ('type'='generic_in_memory');
-[INFO] Catalog has been created.
+[INFO] Execute statement succeed.
!info
use catalog `mod`;
-[INFO] Catalog changed.
+[INFO] Execute statement succeed.
!info
use `default`;
-[INFO] Database changed.
+[INFO] Execute statement succeed.
!info
drop database `default`;
-[INFO] Database has been removed.
+[INFO] Execute statement succeed.
!info
drop catalog `mod`;
-[INFO] Catalog has been removed.
+[INFO] Execute statement succeed.
!info
# ==========================================================================
@@ -194,11 +194,11 @@ create catalog hivecatalog with (
'type' = 'hive-test',
'hive-version' = '2.3.4'
);
-[INFO] Catalog has been created.
+[INFO] Execute statement succeed.
!info
use catalog hivecatalog;
-[INFO] Catalog changed.
+[INFO] Execute statement succeed.
!info
show current catalog;
@@ -230,7 +230,7 @@ show tables;
!ok
use additional_test_database;
-[INFO] Database changed.
+[INFO] Execute statement succeed.
!info
show tables;
@@ -287,15 +287,15 @@ Received a total of 0 row
# ==========================================================================
use catalog hivecatalog;
-[INFO] Catalog changed.
+[INFO] Execute statement succeed.
!info
create table MyTable1 (a int, b string);
-[INFO] Table has been created.
+[INFO] Execute statement succeed.
!info
create table MyTable2 (a int, b string);
-[INFO] Table has been created.
+[INFO] Execute statement succeed.
!info
# hive catalog is case-insensitive
@@ -315,11 +315,11 @@ Empty set
!ok
create view MyView1 as select 1 + 1;
-[INFO] View has been created.
+[INFO] Execute statement succeed.
!info
create view MyView2 as select 1 + 1;
-[INFO] View has been created.
+[INFO] Execute statement succeed.
!info
show views;
@@ -334,27 +334,27 @@ show views;
# test create with full qualified name
create table c1.db1.MyTable3 (a int, b string);
-[INFO] Table has been created.
+[INFO] Execute statement succeed.
!info
create table c1.db1.MyTable4 (a int, b string);
-[INFO] Table has been created.
+[INFO] Execute statement succeed.
!info
create view c1.db1.MyView3 as select 1 + 1;
-[INFO] View has been created.
+[INFO] Execute statement succeed.
!info
create view c1.db1.MyView4 as select 1 + 1;
-[INFO] View has been created.
+[INFO] Execute statement succeed.
!info
use catalog c1;
-[INFO] Catalog changed.
+[INFO] Execute statement succeed.
!info
use db1;
-[INFO] Database changed.
+[INFO] Execute statement succeed.
!info
show tables;
@@ -381,23 +381,23 @@ show views;
# test create with database name
create table `default`.MyTable5 (a int, b string);
-[INFO] Table has been created.
+[INFO] Execute statement succeed.
!info
create table `default`.MyTable6 (a int, b string);
-[INFO] Table has been created.
+[INFO] Execute statement succeed.
!info
create view `default`.MyView5 as select 1 + 1;
-[INFO] View has been created.
+[INFO] Execute statement succeed.
!info
create view `default`.MyView6 as select 1 + 1;
-[INFO] View has been created.
+[INFO] Execute statement succeed.
!info
use `default`;
-[INFO] Database changed.
+[INFO] Execute statement succeed.
!info
show tables;
@@ -423,15 +423,15 @@ show views;
!ok
drop table db1.MyTable3;
-[INFO] Table has been removed.
+[INFO] Execute statement succeed.
!info
drop view db1.MyView3;
-[INFO] View has been removed.
+[INFO] Execute statement succeed.
!info
use db1;
-[INFO] Database changed.
+[INFO] Execute statement succeed.
!info
show tables;
@@ -454,15 +454,15 @@ show views;
!ok
drop table c1.`default`.MyTable6;
-[INFO] Table has been removed.
+[INFO] Execute statement succeed.
!info
drop view c1.`default`.MyView6;
-[INFO] View has been removed.
+[INFO] Execute statement succeed.
!info
use `default`;
-[INFO] Database changed.
+[INFO] Execute statement succeed.
!info
show tables;
@@ -493,7 +493,7 @@ SET sql-client.execution.result-mode = changelog;
!info
create table MyTable7 (a int, b string);
-[INFO] Table has been created.
+[INFO] Execute statement succeed.
!info
show tables;
@@ -512,7 +512,7 @@ reset;
!info
drop table MyTable5;
-[INFO] Table has been removed.
+[INFO] Execute statement succeed.
!info
show tables;
diff --git a/flink-table/flink-sql-client/src/test/resources/sql/function.q b/flink-table/flink-sql-client/src/test/resources/sql/function.q
index ddbf1ba..783f8a5 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/function.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/function.q
@@ -17,7 +17,7 @@
# this also tests user classloader because the LowerUDF is in user jar
create function func1 as 'LowerUDF' LANGUAGE JAVA;
-[INFO] Function has been created.
+[INFO] Execute statement succeed.
!info
show user functions;
@@ -47,7 +47,7 @@ Received a total of 2 rows
# ====== test temporary function ======
create temporary function if not exists func2 as 'LowerUDF' LANGUAGE JAVA;
-[INFO] Function has been created.
+[INFO] Execute statement succeed.
!info
show user functions;
@@ -63,27 +63,27 @@ show user functions;
# ====== test function with full qualified name ======
create catalog c1 with ('type'='generic_in_memory');
-[INFO] Catalog has been created.
+[INFO] Execute statement succeed.
!info
use catalog c1;
-[INFO] Catalog changed.
+[INFO] Execute statement succeed.
!info
create database db;
-[INFO] Database has been created.
+[INFO] Execute statement succeed.
!info
use catalog default_catalog;
-[INFO] Catalog changed.
+[INFO] Execute statement succeed.
!info
create function c1.db.func3 as 'LowerUDF' LANGUAGE JAVA;
-[INFO] Function has been created.
+[INFO] Execute statement succeed.
!info
create temporary function if not exists c1.db.func4 as 'LowerUDF' LANGUAGE JAVA;
-[INFO] Function has been created.
+[INFO] Execute statement succeed.
!info
# no func3 and func4 because we are not under catalog c1
@@ -98,11 +98,11 @@ show user functions;
!ok
use catalog c1;
-[INFO] Catalog changed.
+[INFO] Execute statement succeed.
!info
use db;
-[INFO] Database changed.
+[INFO] Execute statement succeed.
!info
# should show func3 and func4 now
@@ -118,15 +118,15 @@ show user functions;
# test create function with database name
create function `default`.func5 as 'LowerUDF';
-[INFO] Function has been created.
+[INFO] Execute statement succeed.
!info
create function `default`.func6 as 'LowerUDF';
-[INFO] Function has been created.
+[INFO] Execute statement succeed.
!info
use `default`;
-[INFO] Database changed.
+[INFO] Execute statement succeed.
!info
# should show func5 and func6
@@ -145,27 +145,27 @@ show user functions;
# ==========================================================================
create function c1.db.func10 as 'LowerUDF';
-[INFO] Function has been created.
+[INFO] Execute statement succeed.
!info
create function c1.db.func11 as 'LowerUDF';
-[INFO] Function has been created.
+[INFO] Execute statement succeed.
!info
drop function if exists c1.db.func10;
-[INFO] Function has been removed.
+[INFO] Execute statement succeed.
!info
use catalog c1;
-[INFO] Catalog changed.
+[INFO] Execute statement succeed.
!info
use db;
-[INFO] Database changed.
+[INFO] Execute statement succeed.
!info
drop function if exists non_func;
-[INFO] Function has been removed.
+[INFO] Execute statement succeed.
!info
# should contain func11, not contain func10
@@ -185,18 +185,18 @@ show user functions;
# ==========================================================================
alter function func11 as 'org.apache.flink.table.client.gateway.local.LocalExecutorITCase$TestScalaFunction';
-[INFO] Alter function succeeded!
+[INFO] Execute statement succeed.
!info
# TODO: show func11 when we support DESCRIBE FUNCTION
create temporary function tmp_func as 'LowerUDF';
-[INFO] Function has been created.
+[INFO] Execute statement succeed.
!info
# should throw unsupported error
alter temporary function tmp_func as 'org.apache.flink.table.client.gateway.local.LocalExecutorITCase$TestScalaFunction';
-[ERROR] Could not execute SQL statement. Alter function failed! Reason:
+[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.table.api.ValidationException: Alter temporary catalog function is not supported
!error
@@ -206,15 +206,15 @@ org.apache.flink.table.api.ValidationException: Alter temporary catalog function
# ==========================================================================
create catalog hivecatalog with ('type'='hive-test', 'hive-version'='2.3.4');
-[INFO] Catalog has been created.
+[INFO] Execute statement succeed.
!info
use catalog hivecatalog;
-[INFO] Catalog changed.
+[INFO] Execute statement succeed.
!info
create function lowerudf AS 'LowerUDF';
-[INFO] Function has been created.
+[INFO] Execute statement succeed.
!info
show user functions;
diff --git a/flink-table/flink-sql-client/src/test/resources/sql/module.q b/flink-table/flink-sql-client/src/test/resources/sql/module.q
index b40b3b0..f466384 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/module.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/module.q
@@ -45,7 +45,7 @@ SHOW FULL MODULES;
# load core module twice
LOAD MODULE core;
-[ERROR] Could not execute SQL statement. Load module failed! Reason:
+[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.table.api.ValidationException: A module with name 'core' already exists
!error
@@ -71,7 +71,7 @@ Was expecting one of:
# load hive module with module name capitalized
LOAD MODULE Hive;
-[ERROR] Could not execute SQL statement. Load module failed! Reason:
+[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.table.api.NoMatchingTableFactoryException: Could not find a suitable table factory for 'org.apache.flink.table.factories.ModuleFactory' in
the classpath.
@@ -88,12 +88,12 @@ org.apache.flink.table.module.hive.HiveModuleFactory
# load hive module with specifying type
LOAD MODULE myhive WITH ('type' = 'hive');
-[ERROR] Could not execute SQL statement. Load module failed! Reason:
+[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.table.api.ValidationException: Property 'type' = 'hive' is not supported since module name is used to find module
!error
LOAD MODULE hive;
-[INFO] Load module succeeded!
+[INFO] Execute statement succeed.
!info
# show enabled modules
@@ -134,13 +134,13 @@ Received a total of 1 row
# use duplicate modules
USE MODULES hive, core, hive;
-[ERROR] Could not execute SQL statement. Use modules failed! Reason:
+[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.table.api.ValidationException: Module 'hive' appears more than once
!error
# change module resolution order
USE MODULES hive, core;
-[INFO] Use modules succeeded!
+[INFO] Execute statement succeed.
!info
SHOW MODULES;
@@ -165,7 +165,7 @@ SHOW FULL MODULES;
# disable hive module
USE MODULES core;
-[INFO] Use modules succeeded!
+[INFO] Execute statement succeed.
!info
SHOW MODULES;
@@ -198,7 +198,7 @@ org.apache.calcite.sql.validate.SqlValidatorException: No match found for functi
# ==========================================================================
UNLOAD MODULE core;
-[INFO] Unload module succeeded!
+[INFO] Execute statement succeed.
!info
SHOW MODULES;
@@ -216,6 +216,6 @@ SHOW FULL MODULES;
# unload core module twice
UNLOAD MODULE core;
-[ERROR] Could not execute SQL statement. Unload module failed! Reason:
+[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.table.api.ValidationException: No module with name 'core' exists
!error
diff --git a/flink-table/flink-sql-client/src/test/resources/sql/select.q b/flink-table/flink-sql-client/src/test/resources/sql/select.q
index 1f2b7ea..29a7fab 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/select.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/select.q
@@ -37,7 +37,7 @@ create table src (
'data-id' = 'non-exist',
'failing-source' = 'true'
);
-[INFO] Table has been created.
+[INFO] Execute statement succeed.
!info
SELECT UPPER(str), id FROM src;
diff --git a/flink-table/flink-sql-client/src/test/resources/sql/set.q b/flink-table/flink-sql-client/src/test/resources/sql/set.q
index ac1064e..12b4c86 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/set.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/set.q
@@ -50,7 +50,7 @@ CREATE TABLE hive_table (
) PARTITIONED BY (pt_year STRING, pt_month STRING, pt_day STRING) TBLPROPERTIES (
'streaming-source.enable' = 'true'
);
-[INFO] Table has been created.
+[INFO] Execute statement succeed.
!info
# list the configured configuration
diff --git a/flink-table/flink-sql-client/src/test/resources/sql/table.q b/flink-table/flink-sql-client/src/test/resources/sql/table.q
index bed6fa8..f1dbc13 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/table.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/table.q
@@ -60,7 +60,7 @@ CREATE TABLE IF NOT EXISTS orders (
) with (
'connector' = 'datagen'
);
-[INFO] Table has been created.
+[INFO] Execute statement succeed.
!info
# test SHOW TABLES
@@ -79,12 +79,12 @@ show tables;
# test alter table name
alter table orders rename to orders2;
-[INFO] Alter table succeeded!
+[INFO] Execute statement succeed.
!info
# test alter table properties
alter table orders2 set ('connector' = 'kafka');
-[INFO] Alter table succeeded!
+[INFO] Execute statement succeed.
!info
# TODO: verify properties using SHOW CREATE TABLE in the future
@@ -122,7 +122,7 @@ desc orders2;
# ==========================================================================
drop table orders2;
-[INFO] Table has been removed.
+[INFO] Execute statement succeed.
!info
# verify table is dropped
@@ -141,7 +141,7 @@ create temporary table tbl1 (
) with (
'connector' = 'datagen'
);
-[INFO] Table has been created.
+[INFO] Execute statement succeed.
!info
# TODO: warning users the table already exists
@@ -152,7 +152,7 @@ create temporary table if not exists tbl1 (
) with (
'connector' = 'datagen'
);
-[INFO] Table has been created.
+[INFO] Execute statement succeed.
!info
# list permanent and temporary tables together
@@ -166,7 +166,7 @@ show tables;
!ok
drop temporary table tbl1;
-[INFO] Table has been removed.
+[INFO] Execute statement succeed.
!info
# ==========================================================================
@@ -174,7 +174,7 @@ drop temporary table tbl1;
# ==========================================================================
create table `mod` (`table` string, `database` string);
-[INFO] Table has been created.
+[INFO] Execute statement succeed.
!info
describe `mod`;
@@ -198,7 +198,7 @@ desc `mod`;
!ok
drop table `mod`;
-[INFO] Table has been removed.
+[INFO] Execute statement succeed.
!info
show tables;
diff --git a/flink-table/flink-sql-client/src/test/resources/sql/view.q b/flink-table/flink-sql-client/src/test/resources/sql/view.q
index 6fd9859..4f587b4 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/view.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/view.q
@@ -27,13 +27,13 @@ CREATE TABLE orders (
) with (
'connector' = 'datagen'
);
-[INFO] Table has been created.
+[INFO] Execute statement succeed.
!info
# ==== test temporary view =====
create temporary view v1 as select * from orders;
-[INFO] View has been created.
+[INFO] Execute statement succeed.
!info
create temporary view v1 as select * from orders;
@@ -43,12 +43,12 @@ org.apache.flink.table.api.ValidationException: Temporary table '`default_catalo
# TODO: warning users the view already exists
create temporary view if not exists v1 as select * from orders;
-[INFO] View has been created.
+[INFO] Execute statement succeed.
!info
# test create a view reference another view
create temporary view if not exists v2 as select * from v1;
-[INFO] View has been created.
+[INFO] Execute statement succeed.
!info
show tables;
@@ -76,7 +76,7 @@ show views;
# register a permanent view with the duplicate name with temporary view
create view v1 as select * from orders;
-[INFO] View has been created.
+[INFO] Execute statement succeed.
!info
# test create duplicate view
@@ -118,12 +118,12 @@ org.apache.flink.table.api.ValidationException: Temporary view with identifier '
# although temporary v2 needs temporary v1, dropping v1 first does not throw exception
drop temporary view v1;
-[INFO] View has been removed.
+[INFO] Execute statement succeed.
!info
# now we can drop permanent view v1
drop view v1;
-[INFO] View has been removed.
+[INFO] Execute statement succeed.
!info
# test drop invalid table
@@ -135,7 +135,7 @@ org.apache.flink.table.api.ValidationException: View with identifier 'default_ca
# ===== test playing with keyword identifiers =====
create view `mod` as select * from orders;
-[INFO] View has been created.
+[INFO] Execute statement succeed.
!info
describe `mod`;
@@ -165,7 +165,7 @@ desc `mod`;
!ok
drop view `mod`;
-[INFO] View has been removed.
+[INFO] Execute statement succeed.
!info
show tables;
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultImpl.java
index e933126..36e5607 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultImpl.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultImpl.java
@@ -47,7 +47,7 @@ import java.util.concurrent.TimeoutException;
/** Implementation for {@link TableResult}. */
@Internal
-class TableResultImpl implements TableResult {
+public class TableResultImpl implements TableResult {
public static final TableResult TABLE_RESULT_OK =
TableResultImpl.builder()
.resultKind(ResultKind.SUCCESS)
[flink] 01/02: [FLINK-21463][table-api] Support to parse 'RESET
key' command
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1080a638382676b85b9e739d41775b4dcab7b439
Author: Shengkai <10...@qq.com>
AuthorDate: Fri Mar 26 00:15:33 2021 +0800
[FLINK-21463][table-api] Support to parse 'RESET key' command
---
.../table/operations/command/ResetOperation.java | 26 ++++++++++++++++++++--
.../planner/parse/ResetOperationParseStrategy.java | 17 ++++++++++++--
.../table/planner/delegation/ParserImplTest.java | 2 ++
.../table/parse/ResetOperationParseStrategy.java | 17 ++++++++++++--
4 files changed, 56 insertions(+), 6 deletions(-)
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/command/ResetOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/command/ResetOperation.java
index 8a33b57..3fc95af 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/command/ResetOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/command/ResetOperation.java
@@ -20,10 +20,32 @@ package org.apache.flink.table.operations.command;
import org.apache.flink.table.operations.Operation;
-/** Operation that represent RESET command. */
+import javax.annotation.Nullable;
+
+import java.util.Optional;
+
+/**
+ * Operation to represent RESET command. If {@link #getKey()} is empty, it means reset the session
+ * configuration.
+ */
public class ResetOperation implements Operation {
+
+ @Nullable private final String key;
+
+ public ResetOperation(@Nullable String key) {
+ this.key = key;
+ }
+
+ public Optional<String> getKey() {
+ return Optional.ofNullable(key);
+ }
+
@Override
public String asSummaryString() {
- return "RESET";
+ if (key == null) {
+ return "RESET";
+ } else {
+ return String.format("RESET %s", key);
+ }
}
}
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/parse/ResetOperationParseStrategy.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/parse/ResetOperationParseStrategy.java
index 8a7e751..708834d 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/parse/ResetOperationParseStrategy.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/parse/ResetOperationParseStrategy.java
@@ -18,9 +18,11 @@
package org.apache.flink.table.planner.parse;
+import org.apache.flink.table.api.TableException;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.command.ResetOperation;
+import java.util.regex.Matcher;
import java.util.regex.Pattern;
/** Strategy to parse statement to {@link ResetOperation}. */
@@ -29,12 +31,23 @@ public class ResetOperationParseStrategy extends AbstractRegexParseStrategy {
static final ResetOperationParseStrategy INSTANCE = new ResetOperationParseStrategy();
private ResetOperationParseStrategy() {
- super(Pattern.compile("RESET", DEFAULT_PATTERN_FLAGS));
+ super(Pattern.compile("RESET(\\s+(?<key>\\S+)\\s*)?", DEFAULT_PATTERN_FLAGS));
}
@Override
public Operation convert(String statement) {
- return new ResetOperation();
+ Matcher matcher = pattern.matcher(statement.trim());
+ String key;
+
+ if (matcher.find()) {
+ key = matcher.group("key");
+ } else {
+ throw new TableException(
+ String.format(
+ "Failed to convert the statement to RESET operation: %s.", statement));
+ }
+
+ return new ResetOperation(key);
}
@Override
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/delegation/ParserImplTest.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/delegation/ParserImplTest.java
index 961e7ed..fc70477 100644
--- a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/delegation/ParserImplTest.java
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/delegation/ParserImplTest.java
@@ -87,6 +87,8 @@ public class ParserImplTest {
forStatement("qUIT").summary("QUIT"),
forStatement("ExIT").summary("QUIT"),
forStatement("REsEt").summary("RESET"),
+ forStatement("REsEt execution.runtime-type")
+ .summary("RESET execution.runtime-type"),
forStatement(" SEt ").summary("SET"),
forStatement("SET execution.runtime-type=batch")
.summary("SET execution.runtime-type=batch"),
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/parse/ResetOperationParseStrategy.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/parse/ResetOperationParseStrategy.java
index 3602a4b..aa67021 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/parse/ResetOperationParseStrategy.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/parse/ResetOperationParseStrategy.java
@@ -18,9 +18,11 @@
package org.apache.flink.table.parse;
+import org.apache.flink.table.api.TableException;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.command.ResetOperation;
+import java.util.regex.Matcher;
import java.util.regex.Pattern;
/** Strategy to parse statement to {@link ResetOperation}. */
@@ -29,12 +31,23 @@ public class ResetOperationParseStrategy extends AbstractRegexParseStrategy {
static final ResetOperationParseStrategy INSTANCE = new ResetOperationParseStrategy();
private ResetOperationParseStrategy() {
- super(Pattern.compile("RESET", DEFAULT_PATTERN_FLAGS));
+ super(Pattern.compile("RESET(\\s+(?<key>\\S+)\\s*)?", DEFAULT_PATTERN_FLAGS));
}
@Override
public Operation convert(String statement) {
- return new ResetOperation();
+ Matcher matcher = pattern.matcher(statement.trim());
+ String key;
+
+ if (matcher.find()) {
+ key = matcher.group("key");
+ } else {
+ throw new TableException(
+ String.format(
+ "Failed to convert the statement to RESET operation: %s.", statement));
+ }
+
+ return new ResetOperation(key);
}
@Override