You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2018/10/16 11:55:55 UTC
[flink] branch master updated (b40b27a -> 2678a67)
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from b40b27a [FLINK-10474][table] Evaluate IN/NOT_IN with literals as local predicate.
new aa6ff5f [FLINK-8865] [sql-client] Add CLI query code completion in SQL Client
new 2678a67 [FLINK-8865] [sql-client] Finalize CLI query code completion in SQL Client
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:
flink-libraries/flink-sql-client/pom.xml | 4 +-
.../apache/flink/table/client/cli/CliClient.java | 5 ++
.../flink/table/client/cli/SqlCommandParser.java | 6 +-
.../flink/table/client/cli/SqlCompleter.java | 96 ++++++++++++++++++++++
.../flink/table/client/cli/SqlMultiLineParser.java | 61 +++++++++++---
.../flink/table/client/gateway/Executor.java | 5 ++
.../table/client/gateway/local/LocalExecutor.java | 17 ++++
.../flink/table/client/cli/CliClientTest.java | 55 +++++++++++++
.../client/gateway/local/LocalExecutorITCase.java | 19 +++++
.../apache/flink/table/api/TableEnvironment.scala | 19 ++++-
.../flink/table/calcite/FlinkPlannerImpl.scala | 37 ++++++++-
11 files changed, 306 insertions(+), 18 deletions(-)
create mode 100644 flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCompleter.java
[flink] 02/02: [FLINK-8865] [sql-client] Finalize CLI query code
completion in SQL Client
Posted by tw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2678a671b02ddea118adf8183ba9539fb379587f
Author: Timo Walther <tw...@apache.org>
AuthorDate: Fri Oct 12 20:58:16 2018 +0200
[FLINK-8865] [sql-client] Finalize CLI query code completion in SQL Client
- Allow code completion also for SQL Client specific statements
- Fix invalid escaping/quoting
- Remove invalid metadata table completion
---
flink-libraries/flink-sql-client/pom.xml | 4 +-
.../apache/flink/table/client/cli/CliClient.java | 5 +-
.../flink/table/client/cli/SqlCommandParser.java | 6 ++-
.../flink/table/client/cli/SqlCompleter.java | 58 ++++++++++++++++++----
.../flink/table/client/cli/SqlMultiLineParser.java | 54 ++++++++++++++++++--
.../flink/table/client/gateway/Executor.java | 5 ++
.../table/client/gateway/local/LocalExecutor.java | 15 +++---
.../flink/table/client/cli/CliClientTest.java | 55 ++++++++++++++++++++
.../client/gateway/local/LocalExecutorITCase.java | 54 +++++---------------
.../apache/flink/table/api/TableEnvironment.scala | 42 ++++++----------
.../flink/table/calcite/FlinkPlannerImpl.scala | 37 ++++++++++++--
11 files changed, 240 insertions(+), 95 deletions(-)
diff --git a/flink-libraries/flink-sql-client/pom.xml b/flink-libraries/flink-sql-client/pom.xml
index 28e0468..68ec718 100644
--- a/flink-libraries/flink-sql-client/pom.xml
+++ b/flink-libraries/flink-sql-client/pom.xml
@@ -88,13 +88,13 @@ under the License.
<dependency>
<groupId>org.jline</groupId>
<artifactId>jline-terminal</artifactId>
- <version>3.5.2</version>
+ <version>3.9.0</version>
</dependency>
<dependency>
<groupId>org.jline</groupId>
<artifactId>jline-reader</artifactId>
- <version>3.5.2</version>
+ <version>3.9.0</version>
</dependency>
<!-- configuration -->
diff --git a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java
index d43e82f..6da1d58 100644
--- a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java
+++ b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java
@@ -102,9 +102,10 @@ public class CliClient {
// this option is disabled for now for correct backslash escaping
// a "SELECT '\'" query should return a string with a backslash
lineReader.option(LineReader.Option.DISABLE_EVENT_EXPANSION, true);
- // this variable indicates the distance between the words in typoMatcher
- // when do complete
+ // set strict "typo" distance between words when doing code completion
lineReader.setVariable(LineReader.ERRORS, 1);
+ // perform code completion case insensitive
+ lineReader.option(LineReader.Option.CASE_INSENSITIVE, true);
// create prompt
prompt = new AttributedStringBuilder()
diff --git a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCommandParser.java b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCommandParser.java
index 83fbabc..db67138 100644
--- a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCommandParser.java
+++ b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCommandParser.java
@@ -37,7 +37,7 @@ public final class SqlCommandParser {
public static Optional<SqlCommandCall> parse(String stmt) {
// normalize
stmt = stmt.trim();
- // remove ';' at the end because many people type it intuitively
+ // remove ';' at the end
if (stmt.endsWith(";")) {
stmt = stmt.substring(0, stmt.length() - 1).trim();
}
@@ -151,6 +151,10 @@ public final class SqlCommandParser {
public String toString() {
return super.toString().replace('_', ' ');
}
+
+ public boolean hasOperands() {
+ return operandConverter != NO_OPERANDS;
+ }
}
/**
diff --git a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCompleter.java b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCompleter.java
index 0c11eba..861d921 100644
--- a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCompleter.java
+++ b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCompleter.java
@@ -20,21 +20,29 @@ package org.apache.flink.table.client.cli;
import org.apache.flink.table.client.gateway.Executor;
import org.apache.flink.table.client.gateway.SessionContext;
-import org.apache.flink.table.client.gateway.local.LocalExecutor;
+import org.apache.flink.table.client.gateway.SqlExecutionException;
import org.jline.reader.Candidate;
import org.jline.reader.Completer;
import org.jline.reader.LineReader;
import org.jline.reader.ParsedLine;
import org.jline.utils.AttributedString;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.List;
/**
- * SQL Completer.
+ * SQL code completer.
*/
public class SqlCompleter implements Completer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SqlCompleter.class);
+
+ public static final String[] COMMAND_HINTS = getCommandHints();
+
private SessionContext context;
+
private Executor executor;
public SqlCompleter(SessionContext context, Executor executor) {
@@ -43,16 +51,46 @@ public class SqlCompleter implements Completer {
}
public void complete(LineReader reader, ParsedLine line, List<Candidate> candidates) {
- if (executor instanceof LocalExecutor) {
- try {
- LocalExecutor localExecutor = (LocalExecutor) executor;
- List<String> hints = localExecutor.getCompletionHints(context, line.line(), line.cursor());
- hints.forEach(hint->
- candidates.add(new Candidate(AttributedString.stripAnsi(hint), hint , null, null, null, null, true)));
- } catch (Exception e) {
- //ignore completer exception;
+ String statement = line.line();
+
+ // remove ';' at the end
+ if (statement.endsWith(";")) {
+ 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(commandHint));
}
}
+
+ // fallback to Table API hinting
+ try {
+ executor.completeStatement(context, statement, line.cursor())
+ .forEach(hint -> candidates.add(createCandidate(hint)));
+ } catch (SqlExecutionException e) {
+ LOG.debug("Could not complete statement at " + line.cursor() + ":" + statement, e);
+ }
+ }
+
+ private Candidate createCandidate(String hint) {
+ return new Candidate(AttributedString.stripAnsi(hint), hint, null, null, null, null, true);
}
+ private static String[] getCommandHints() {
+ final SqlCommandParser.SqlCommand[] commands = SqlCommandParser.SqlCommand.values();
+ final String[] hints = new String[commands.length];
+ for (int i = 0; i < commands.length; i++) {
+ final SqlCommandParser.SqlCommand command = commands[i];
+ // add final ";" for convenience if no operands can follow
+ if (command.hasOperands()) {
+ hints[i] = command.toString();
+ } else {
+ hints[i] = command.toString() + ";";
+ }
+ }
+ return hints;
+ }
}
diff --git a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlMultiLineParser.java b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlMultiLineParser.java
index ee7e087..722ea62 100644
--- a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlMultiLineParser.java
+++ b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlMultiLineParser.java
@@ -22,24 +22,72 @@ import org.jline.reader.EOFError;
import org.jline.reader.ParsedLine;
import org.jline.reader.impl.DefaultParser;
+import java.util.List;
+
/**
* Multi-line parser for parsing an arbitrary number of SQL lines until a line ends with ';'.
+ *
+ * <p>Quoting and escaping are disabled for now.
*/
public class SqlMultiLineParser extends DefaultParser {
private static final String EOF_CHARACTER = ";";
private static final String NEW_LINE_PROMPT = ""; // results in simple '>' output
+ public SqlMultiLineParser() {
+ setEscapeChars(null);
+ setQuoteChars(null);
+ }
+
@Override
public ParsedLine parse(String line, int cursor, ParseContext context) {
- if (!line.trim().endsWith(EOF_CHARACTER)
- && context != ParseContext.COMPLETE) {
+ if (!line.trim().endsWith(EOF_CHARACTER) && context != ParseContext.COMPLETE) {
throw new EOFError(
-1,
-1,
"New line without EOF character.",
NEW_LINE_PROMPT);
}
- return super.parse(line, cursor, context);
+ final ArgumentList parsedLine = (ArgumentList) super.parse(line, cursor, context);
+ return new SqlArgumentList(
+ parsedLine.line(),
+ parsedLine.words(),
+ parsedLine.wordIndex(),
+ parsedLine.wordCursor(),
+ parsedLine.cursor(),
+ null,
+ parsedLine.rawWordCursor(),
+ parsedLine.rawWordLength());
+ }
+
+ private class SqlArgumentList extends DefaultParser.ArgumentList {
+
+ public SqlArgumentList(
+ String line,
+ List<String> words,
+ int wordIndex,
+ int wordCursor,
+ int cursor,
+ String openingQuote,
+ int rawWordCursor,
+ int rawWordLength) {
+
+ super(
+ line,
+ words,
+ wordIndex,
+ wordCursor,
+ cursor,
+ openingQuote,
+ rawWordCursor,
+ rawWordLength);
+ }
+
+ @Override
+ public CharSequence escape(CharSequence candidate, boolean complete) {
+ // escaping is skipped for now because it highly depends on the context in the SQL statement
+ // e.g. 'INS(ERT INTO)' should not be quoted but 'SELECT * FROM T(axi Rides)' should
+ return candidate;
+ }
}
}
diff --git a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java
index d641016..9aa99a7 100644
--- a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java
+++ b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java
@@ -62,6 +62,11 @@ public interface Executor {
String explainStatement(SessionContext session, String statement) throws SqlExecutionException;
/**
+ * Returns a list of completion hints for the given statement at the given position.
+ */
+ List<String> completeStatement(SessionContext session, String statement, int position);
+
+ /**
* Submits a Flink SQL query job (detached) and returns the result descriptor.
*/
ResultDescriptor executeQuery(SessionContext session, String query) throws SqlExecutionException;
diff --git a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
index be9958f..5c2cb3f 100644
--- a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
+++ b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
@@ -235,18 +235,21 @@ public class LocalExecutor implements Executor {
}
}
- public List<String> getCompletionHints(SessionContext session, String line, Integer pos) {
+ @Override
+ public List<String> completeStatement(SessionContext session, String statement, int position) {
final TableEnvironment tableEnv = getOrCreateExecutionContext(session)
.createEnvironmentInstance()
.getTableEnvironment();
- // complete
try {
- return Arrays.asList(tableEnv.getCompletionHints(line, pos));
- } catch (Exception e) {
- //ignore completer exception
+ return Arrays.asList(tableEnv.getCompletionHints(statement, position));
+ } catch (Throwable t) {
+ // catch everything such that the query does not crash the executor
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Could not complete statement at " + position + ":" + statement, t);
+ }
+ return Collections.emptyList();
}
- return Collections.emptyList();
}
@Override
diff --git a/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java b/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java
index 90c9195..4bf284b 100644
--- a/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java
+++ b/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java
@@ -30,8 +30,16 @@ import org.apache.flink.table.client.gateway.TypedResult;
import org.apache.flink.types.Row;
import org.apache.flink.util.TestLogger;
+import org.jline.reader.Candidate;
+import org.jline.reader.LineReader;
+import org.jline.reader.LineReaderBuilder;
+import org.jline.reader.ParsedLine;
+import org.jline.reader.Parser;
import org.junit.Test;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -61,6 +69,18 @@ public class CliClientTest extends TestLogger {
verifyUpdateSubmission(SELECT_STATEMENT, false, true);
}
+ @Test
+ public void testSqlCompletion() {
+ verifySqlCompletion("", 0, Arrays.asList("SELECT", "QUIT;", "RESET;"), Collections.emptyList());
+ verifySqlCompletion("SELEC", 5, Collections.singletonList("SELECT"), Collections.singletonList("QUIT;"));
+ verifySqlCompletion("SELE", 0, Collections.singletonList("SELECT"), Collections.singletonList("QUIT;"));
+ verifySqlCompletion("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"));
+ }
+
// --------------------------------------------------------------------------------------------
private void verifyUpdateSubmission(String statement, boolean failExecution, boolean testFailure) {
@@ -79,6 +99,32 @@ public class CliClientTest extends TestLogger {
}
}
+ private void verifySqlCompletion(String statement, int position, List<String> expectedHints, List<String> notExpectedHints) {
+ final SessionContext context = new SessionContext("test-session", new Environment());
+ final MockExecutor mockExecutor = new MockExecutor();
+
+ final SqlCompleter completer = new SqlCompleter(context, mockExecutor);
+ final SqlMultiLineParser parser = new SqlMultiLineParser();
+ final LineReader reader = LineReaderBuilder.builder().build();
+
+ final ParsedLine parsedLine = parser.parse(statement, position, Parser.ParseContext.COMPLETE);
+ final List<Candidate> candidates = new ArrayList<>();
+ final List<String> results = new ArrayList<>();
+ completer.complete(reader, parsedLine, candidates);
+ candidates.forEach(item -> results.add(item.value()));
+
+ assertTrue(results.containsAll(expectedHints));
+
+ assertEquals(statement, mockExecutor.receivedStatement);
+ assertEquals(context, mockExecutor.receivedContext);
+ assertEquals(position, mockExecutor.receivedPosition);
+ assertTrue(results.contains("HintA"));
+ assertTrue(results.contains("Hint B"));
+
+ results.retainAll(notExpectedHints);
+ assertEquals(0, results.size());
+ }
+
// --------------------------------------------------------------------------------------------
private static class MockExecutor implements Executor {
@@ -87,6 +133,7 @@ public class CliClientTest extends TestLogger {
public SessionContext receivedContext;
public String receivedStatement;
+ public int receivedPosition;
@Override
public void start() throws SqlExecutionException {
@@ -119,6 +166,14 @@ public class CliClientTest extends TestLogger {
}
@Override
+ public List<String> completeStatement(SessionContext session, String statement, int position) {
+ receivedContext = session;
+ receivedStatement = statement;
+ receivedPosition = position;
+ return Arrays.asList("HintA", "Hint B");
+ }
+
+ @Override
public ResultDescriptor executeQuery(SessionContext session, String query) throws SqlExecutionException {
return null;
}
diff --git a/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java b/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java
index 970f313..575af1c 100644
--- a/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java
+++ b/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java
@@ -31,8 +31,6 @@ import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.configuration.WebOptions;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.table.api.TableSchema;
-import org.apache.flink.table.client.cli.SqlCompleter;
-import org.apache.flink.table.client.cli.SqlMultiLineParser;
import org.apache.flink.table.client.config.Environment;
import org.apache.flink.table.client.gateway.Executor;
import org.apache.flink.table.client.gateway.ProgramTargetDescriptor;
@@ -47,11 +45,6 @@ import org.apache.flink.test.util.TestBaseUtils;
import org.apache.flink.types.Row;
import org.apache.flink.util.TestLogger;
-import org.jline.reader.Candidate;
-import org.jline.reader.LineReader;
-import org.jline.reader.LineReaderBuilder;
-import org.jline.reader.ParsedLine;
-import org.jline.reader.Parser.ParseContext;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
@@ -230,43 +223,22 @@ public class LocalExecutorITCase extends TestLogger {
}
@Test
- public void testSqlCompleter() throws Exception {
+ public void testCompleteStatement() throws Exception {
final Executor executor = createDefaultExecutor(clusterClient);
final SessionContext session = new SessionContext("test-session", new Environment());
- SqlCompleter completer = new SqlCompleter(session, executor);
- SqlMultiLineParser parser = new SqlMultiLineParser();
- LineReader reader = LineReaderBuilder.builder().build();
-
- String line = "SELECT * FROM Ta";
- ParsedLine parsedLine = parser.parse(line, 16, ParseContext.COMPLETE);
- List<Candidate> candidates = new ArrayList<Candidate>();
- List<String> results = new ArrayList<String>();
- completer.complete(reader, parsedLine, candidates);
- candidates.forEach(item->results.add(item.value()));
- List<String> expectedResults = Arrays.asList(
- "TableNumber1",
- "TableNumber2",
- "TableSourceSink");
- assertTrue(results.containsAll(expectedResults));
-
- line = "SELECT * FROM TableNumber2 WH";
- parsedLine = parser.parse(line, 29, ParseContext.COMPLETE);
- candidates.clear();
- results.clear();
- completer.complete(reader, parsedLine, candidates);
- candidates.forEach(item->results.add(item.value()));
- expectedResults = Arrays.asList("WHERE");
- assertEquals(expectedResults, results);
-
- line = "SELECT * FROM TableNumber1 WHERE Inte";
- parsedLine = parser.parse(line, 37, ParseContext.COMPLETE);
- candidates.clear();
- results.clear();
- completer.complete(reader, parsedLine, candidates);
- candidates.forEach(item->results.add(item.value()));
- expectedResults = Arrays.asList("IntegerField1");
- assertTrue(results.containsAll(expectedResults));
+ final List<String> expectedTableHints = Arrays.asList(
+ "TABLE",
+ "TableNumber1",
+ "TableNumber2",
+ "TableSourceSink");
+ assertEquals(expectedTableHints, executor.completeStatement(session, "SELECT * FROM Ta", 16));
+
+ final List<String> expectedClause = Collections.singletonList("WHERE");
+ assertEquals(expectedClause, executor.completeStatement(session, "SELECT * FROM TableNumber2 WH", 29));
+
+ final List<String> expectedField = Arrays.asList("INTERVAL", "IntegerField1");
+ assertEquals(expectedField, executor.completeStatement(session, "SELECT * FROM TableNumber1 WHERE Inte", 37));
}
@Test(timeout = 30_000L)
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
index 69d723a..7cfe249 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
@@ -19,7 +19,6 @@
package org.apache.flink.table.api
import _root_.java.lang.reflect.Modifier
-import _root_.java.util.Collections
import _root_.java.util.concurrent.atomic.AtomicInteger
import com.google.common.collect.ImmutableList
@@ -28,15 +27,12 @@ import org.apache.calcite.jdbc.CalciteSchema
import org.apache.calcite.plan.RelOptPlanner.CannotPlanException
import org.apache.calcite.plan.hep.{HepMatchOrder, HepPlanner, HepProgramBuilder}
import org.apache.calcite.plan.{Convention, RelOptPlanner, RelOptUtil, RelTraitSet}
-import org.apache.calcite.prepare.CalciteCatalogReader
import org.apache.calcite.rel.RelNode
import org.apache.calcite.schema.SchemaPlus
import org.apache.calcite.schema.impl.AbstractTable
import org.apache.calcite.sql._
-import org.apache.calcite.sql.advise.{SqlAdvisor, SqlAdvisorValidator}
import org.apache.calcite.sql.parser.SqlParser
import org.apache.calcite.sql.util.ChainedSqlOperatorTable
-import org.apache.calcite.sql.validate.SqlConformance
import org.apache.calcite.sql2rel.SqlToRelConverter
import org.apache.calcite.tools._
import org.apache.flink.api.common.functions.MapFunction
@@ -50,7 +46,7 @@ import org.apache.flink.streaming.api.environment.{StreamExecutionEnvironment =>
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment => ScalaStreamExecEnv}
import org.apache.flink.table.api.java.{BatchTableEnvironment => JavaBatchTableEnv, StreamTableEnvironment => JavaStreamTableEnv}
import org.apache.flink.table.api.scala.{BatchTableEnvironment => ScalaBatchTableEnv, StreamTableEnvironment => ScalaStreamTableEnv}
-import org.apache.flink.table.calcite._
+import org.apache.flink.table.calcite.{FlinkPlannerImpl, FlinkRelBuilder, FlinkTypeFactory, FlinkTypeSystem}
import org.apache.flink.table.catalog.{ExternalCatalog, ExternalCatalogSchema}
import org.apache.flink.table.codegen.{ExpressionReducer, FunctionCodeGenerator, GeneratedFunction}
import org.apache.flink.table.descriptors.{ConnectorDescriptor, TableDescriptor}
@@ -81,7 +77,7 @@ abstract class TableEnvironment(val config: TableConfig) {
// the catalog to hold all registered and translated tables
// we disable caching here to prevent side effects
- private val internalSchema: CalciteSchema = CalciteSchema.createRootSchema(true, false)
+ private val internalSchema: CalciteSchema = CalciteSchema.createRootSchema(false, false)
private val rootSchema: SchemaPlus = internalSchema.plus()
// Table API/SQL function catalog
@@ -235,7 +231,6 @@ abstract class TableEnvironment(val config: TableConfig) {
SqlParser
.configBuilder()
.setLex(Lex.JAVA)
- .setCaseSensitive(false)
.build()
case Some(sqlParserConfig) =>
@@ -683,25 +678,20 @@ abstract class TableEnvironment(val config: TableConfig) {
def explain(table: Table): String
/**
- * Gets completion hints for the sql query
- */
- def getCompletionHints(query: String, pos: Integer) : Array[String] = {
- val catalogReader = new CalciteCatalogReader(
- CalciteSchema.from(rootSchema),
- Collections.emptyList(),
- typeFactory,
- CalciteConfig.connectionConfig(getSqlParserConfig))
- val validator : SqlAdvisorValidator = new SqlAdvisorValidator(
- getSqlOperatorTable,
- catalogReader,
- typeFactory,
- SqlConformance.DEFAULT)
- val advisor: SqlAdvisor = new SqlAdvisor(validator)
- val replaced: Array[String] = Array(null)
- val hints = advisor.getCompletionHints(query, pos, replaced)
- .asScala
- .map(item => item.toIdentifier.toString)
- hints.toArray
+ * Returns completion hints for the given statement at the given cursor position.
+ * The completion happens case insensitively.
+ *
+ * @param statement Partial or slightly incorrect SQL statement
+ * @param position cursor position
+ * @return completion hints that fit at the current cursor position
+ */
+ def getCompletionHints(statement: String, position: Int): Array[String] = {
+ val planner = new FlinkPlannerImpl(
+ getFrameworkConfig,
+ getPlanner,
+ getTypeFactory,
+ sqlToRelConverterConfig)
+ planner.getCompletionHints(statement, position)
}
/**
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkPlannerImpl.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkPlannerImpl.scala
index b9e334a..2cd0c61 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkPlannerImpl.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkPlannerImpl.scala
@@ -29,6 +29,7 @@ import org.apache.calcite.rel.RelRoot
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rex.RexBuilder
import org.apache.calcite.schema.SchemaPlus
+import org.apache.calcite.sql.advise.{SqlAdvisor, SqlAdvisorValidator}
import org.apache.calcite.sql.parser.{SqlParser, SqlParseException => CSqlParseException}
import org.apache.calcite.sql.validate.SqlValidator
import org.apache.calcite.sql.{SqlNode, SqlOperatorTable}
@@ -69,6 +70,19 @@ class FlinkPlannerImpl(
}
}
+ def getCompletionHints(sql: String, cursor: Int): Array[String] = {
+ val advisorValidator = new SqlAdvisorValidator(
+ operatorTable,
+ createCatalogReader(true), // ignore cases for lenient completion
+ typeFactory,
+ config.getParserConfig.conformance())
+ val advisor = new SqlAdvisor(advisorValidator)
+ val replaced = Array[String](null)
+ val hints = advisor.getCompletionHints(sql, cursor, replaced)
+ .map(item => item.toIdentifier.toString)
+ hints.toArray
+ }
+
def parse(sql: String): SqlNode = {
try {
ready()
@@ -82,7 +96,10 @@ class FlinkPlannerImpl(
}
def validate(sqlNode: SqlNode): SqlNode = {
- validator = new FlinkCalciteSqlValidator(operatorTable, createCatalogReader, typeFactory)
+ validator = new FlinkCalciteSqlValidator(
+ operatorTable,
+ createCatalogReader(false),
+ typeFactory)
validator.setIdentifierExpansion(true)
try {
validator.validate(sqlNode)
@@ -101,7 +118,7 @@ class FlinkPlannerImpl(
val sqlToRelConverter: SqlToRelConverter = new SqlToRelConverter(
new ViewExpanderImpl,
validator,
- createCatalogReader,
+ createCatalogReader(false),
cluster,
convertletTable,
sqlToRelConverterConfig)
@@ -139,7 +156,8 @@ class FlinkPlannerImpl(
case e: CSqlParseException =>
throw SqlParserException(s"SQL parse failed. ${e.getMessage}", e)
}
- val catalogReader: CalciteCatalogReader = createCatalogReader.withSchemaPath(schemaPath)
+ val catalogReader: CalciteCatalogReader = createCatalogReader(false)
+ .withSchemaPath(schemaPath)
val validator: SqlValidator =
new FlinkCalciteSqlValidator(operatorTable, catalogReader, typeFactory)
validator.setIdentifierExpansion(true)
@@ -160,8 +178,19 @@ class FlinkPlannerImpl(
}
}
- private def createCatalogReader: CalciteCatalogReader = {
+ private def createCatalogReader(lenientCaseSensitivity: Boolean): CalciteCatalogReader = {
val rootSchema: SchemaPlus = FlinkPlannerImpl.rootSchema(defaultSchema)
+
+ val caseSensitive = if (lenientCaseSensitivity) {
+ false
+ } else {
+ this.parserConfig.caseSensitive()
+ }
+
+ val parserConfig = SqlParser.configBuilder(this.parserConfig)
+ .setCaseSensitive(caseSensitive)
+ .build()
+
new CalciteCatalogReader(
CalciteSchema.from(rootSchema),
CalciteSchema.from(defaultSchema).path(null),
[flink] 01/02: [FLINK-8865] [sql-client] Add CLI query code
completion in SQL Client
Posted by tw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit aa6ff5fbd0d233196b0bf3dfb0344d0a75caa01e
Author: xueyu <27...@qq.com>
AuthorDate: Thu Oct 4 12:34:42 2018 +0800
[FLINK-8865] [sql-client] Add CLI query code completion in SQL Client
This closes #6791.
---
.../apache/flink/table/client/cli/CliClient.java | 4 ++
.../flink/table/client/cli/SqlCompleter.java | 58 ++++++++++++++++++++++
.../flink/table/client/cli/SqlMultiLineParser.java | 15 ++----
.../table/client/gateway/local/LocalExecutor.java | 14 ++++++
.../client/gateway/local/LocalExecutorITCase.java | 47 ++++++++++++++++++
.../apache/flink/table/api/TableEnvironment.scala | 29 ++++++++++-
6 files changed, 155 insertions(+), 12 deletions(-)
diff --git a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java
index 7867532..d43e82f 100644
--- a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java
+++ b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java
@@ -97,10 +97,14 @@ public class CliClient {
.terminal(terminal)
.appName(CliStrings.CLI_NAME)
.parser(new SqlMultiLineParser())
+ .completer(new SqlCompleter(context, executor))
.build();
// this option is disabled for now for correct backslash escaping
// a "SELECT '\'" query should return a string with a backslash
lineReader.option(LineReader.Option.DISABLE_EVENT_EXPANSION, true);
+ // this variable indicates the distance between the words in typoMatcher
+ // when do complete
+ lineReader.setVariable(LineReader.ERRORS, 1);
// create prompt
prompt = new AttributedStringBuilder()
diff --git a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCompleter.java b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCompleter.java
new file mode 100644
index 0000000..0c11eba
--- /dev/null
+++ b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlCompleter.java
@@ -0,0 +1,58 @@
+/*
+ * 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.Executor;
+import org.apache.flink.table.client.gateway.SessionContext;
+import org.apache.flink.table.client.gateway.local.LocalExecutor;
+
+import org.jline.reader.Candidate;
+import org.jline.reader.Completer;
+import org.jline.reader.LineReader;
+import org.jline.reader.ParsedLine;
+import org.jline.utils.AttributedString;
+
+import java.util.List;
+
+/**
+ * SQL Completer.
+ */
+public class SqlCompleter implements Completer {
+ private SessionContext context;
+ private Executor executor;
+
+ public SqlCompleter(SessionContext context, Executor executor) {
+ this.context = context;
+ this.executor = executor;
+ }
+
+ public void complete(LineReader reader, ParsedLine line, List<Candidate> candidates) {
+ if (executor instanceof LocalExecutor) {
+ try {
+ LocalExecutor localExecutor = (LocalExecutor) executor;
+ List<String> hints = localExecutor.getCompletionHints(context, line.line(), line.cursor());
+ hints.forEach(hint->
+ candidates.add(new Candidate(AttributedString.stripAnsi(hint), hint , null, null, null, null, true)));
+ } catch (Exception e) {
+ //ignore completer exception;
+ }
+ }
+ }
+
+}
diff --git a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlMultiLineParser.java b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlMultiLineParser.java
index fb44a1e..ee7e087 100644
--- a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlMultiLineParser.java
+++ b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/SqlMultiLineParser.java
@@ -20,33 +20,26 @@ package org.apache.flink.table.client.cli;
import org.jline.reader.EOFError;
import org.jline.reader.ParsedLine;
-import org.jline.reader.Parser;
import org.jline.reader.impl.DefaultParser;
-import java.util.Collections;
-
/**
* Multi-line parser for parsing an arbitrary number of SQL lines until a line ends with ';'.
*/
-public class SqlMultiLineParser implements Parser {
+public class SqlMultiLineParser extends DefaultParser {
private static final String EOF_CHARACTER = ";";
private static final String NEW_LINE_PROMPT = ""; // results in simple '>' output
@Override
public ParsedLine parse(String line, int cursor, ParseContext context) {
- if (!line.trim().endsWith(EOF_CHARACTER)) {
+ if (!line.trim().endsWith(EOF_CHARACTER)
+ && context != ParseContext.COMPLETE) {
throw new EOFError(
-1,
-1,
"New line without EOF character.",
NEW_LINE_PROMPT);
}
- return new DefaultParser.ArgumentList(
- line,
- Collections.singletonList(line),
- 0,
- 0,
- cursor);
+ return super.parse(line, cursor, context);
}
}
diff --git a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
index 1318043..be9958f 100644
--- a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
+++ b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
@@ -235,6 +235,20 @@ public class LocalExecutor implements Executor {
}
}
+ public List<String> getCompletionHints(SessionContext session, String line, Integer pos) {
+ final TableEnvironment tableEnv = getOrCreateExecutionContext(session)
+ .createEnvironmentInstance()
+ .getTableEnvironment();
+
+ // complete
+ try {
+ return Arrays.asList(tableEnv.getCompletionHints(line, pos));
+ } catch (Exception e) {
+ //ignore completer exception
+ }
+ return Collections.emptyList();
+ }
+
@Override
public ResultDescriptor executeQuery(SessionContext session, String query) throws SqlExecutionException {
final ExecutionContext<?> context = getOrCreateExecutionContext(session);
diff --git a/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java b/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java
index 495f5e0..970f313 100644
--- a/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java
+++ b/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java
@@ -31,6 +31,8 @@ import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.configuration.WebOptions;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.client.cli.SqlCompleter;
+import org.apache.flink.table.client.cli.SqlMultiLineParser;
import org.apache.flink.table.client.config.Environment;
import org.apache.flink.table.client.gateway.Executor;
import org.apache.flink.table.client.gateway.ProgramTargetDescriptor;
@@ -45,6 +47,11 @@ import org.apache.flink.test.util.TestBaseUtils;
import org.apache.flink.types.Row;
import org.apache.flink.util.TestLogger;
+import org.jline.reader.Candidate;
+import org.jline.reader.LineReader;
+import org.jline.reader.LineReaderBuilder;
+import org.jline.reader.ParsedLine;
+import org.jline.reader.Parser.ParseContext;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
@@ -222,6 +229,46 @@ public class LocalExecutorITCase extends TestLogger {
assertEquals(expectedTableSchema, actualTableSchema);
}
+ @Test
+ public void testSqlCompleter() throws Exception {
+ final Executor executor = createDefaultExecutor(clusterClient);
+ final SessionContext session = new SessionContext("test-session", new Environment());
+
+ SqlCompleter completer = new SqlCompleter(session, executor);
+ SqlMultiLineParser parser = new SqlMultiLineParser();
+ LineReader reader = LineReaderBuilder.builder().build();
+
+ String line = "SELECT * FROM Ta";
+ ParsedLine parsedLine = parser.parse(line, 16, ParseContext.COMPLETE);
+ List<Candidate> candidates = new ArrayList<Candidate>();
+ List<String> results = new ArrayList<String>();
+ completer.complete(reader, parsedLine, candidates);
+ candidates.forEach(item->results.add(item.value()));
+ List<String> expectedResults = Arrays.asList(
+ "TableNumber1",
+ "TableNumber2",
+ "TableSourceSink");
+ assertTrue(results.containsAll(expectedResults));
+
+ line = "SELECT * FROM TableNumber2 WH";
+ parsedLine = parser.parse(line, 29, ParseContext.COMPLETE);
+ candidates.clear();
+ results.clear();
+ completer.complete(reader, parsedLine, candidates);
+ candidates.forEach(item->results.add(item.value()));
+ expectedResults = Arrays.asList("WHERE");
+ assertEquals(expectedResults, results);
+
+ line = "SELECT * FROM TableNumber1 WHERE Inte";
+ parsedLine = parser.parse(line, 37, ParseContext.COMPLETE);
+ candidates.clear();
+ results.clear();
+ completer.complete(reader, parsedLine, candidates);
+ candidates.forEach(item->results.add(item.value()));
+ expectedResults = Arrays.asList("IntegerField1");
+ assertTrue(results.containsAll(expectedResults));
+ }
+
@Test(timeout = 30_000L)
public void testStreamQueryExecutionChangelog() throws Exception {
final URL url = getClass().getClassLoader().getResource("test-data.csv");
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
index e395868..69d723a 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
@@ -19,6 +19,7 @@
package org.apache.flink.table.api
import _root_.java.lang.reflect.Modifier
+import _root_.java.util.Collections
import _root_.java.util.concurrent.atomic.AtomicInteger
import com.google.common.collect.ImmutableList
@@ -27,12 +28,15 @@ import org.apache.calcite.jdbc.CalciteSchema
import org.apache.calcite.plan.RelOptPlanner.CannotPlanException
import org.apache.calcite.plan.hep.{HepMatchOrder, HepPlanner, HepProgramBuilder}
import org.apache.calcite.plan.{Convention, RelOptPlanner, RelOptUtil, RelTraitSet}
+import org.apache.calcite.prepare.CalciteCatalogReader
import org.apache.calcite.rel.RelNode
import org.apache.calcite.schema.SchemaPlus
import org.apache.calcite.schema.impl.AbstractTable
import org.apache.calcite.sql._
+import org.apache.calcite.sql.advise.{SqlAdvisor, SqlAdvisorValidator}
import org.apache.calcite.sql.parser.SqlParser
import org.apache.calcite.sql.util.ChainedSqlOperatorTable
+import org.apache.calcite.sql.validate.SqlConformance
import org.apache.calcite.sql2rel.SqlToRelConverter
import org.apache.calcite.tools._
import org.apache.flink.api.common.functions.MapFunction
@@ -46,7 +50,7 @@ import org.apache.flink.streaming.api.environment.{StreamExecutionEnvironment =>
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment => ScalaStreamExecEnv}
import org.apache.flink.table.api.java.{BatchTableEnvironment => JavaBatchTableEnv, StreamTableEnvironment => JavaStreamTableEnv}
import org.apache.flink.table.api.scala.{BatchTableEnvironment => ScalaBatchTableEnv, StreamTableEnvironment => ScalaStreamTableEnv}
-import org.apache.flink.table.calcite.{FlinkPlannerImpl, FlinkRelBuilder, FlinkTypeFactory, FlinkTypeSystem}
+import org.apache.flink.table.calcite._
import org.apache.flink.table.catalog.{ExternalCatalog, ExternalCatalogSchema}
import org.apache.flink.table.codegen.{ExpressionReducer, FunctionCodeGenerator, GeneratedFunction}
import org.apache.flink.table.descriptors.{ConnectorDescriptor, TableDescriptor}
@@ -231,6 +235,7 @@ abstract class TableEnvironment(val config: TableConfig) {
SqlParser
.configBuilder()
.setLex(Lex.JAVA)
+ .setCaseSensitive(false)
.build()
case Some(sqlParserConfig) =>
@@ -678,6 +683,28 @@ abstract class TableEnvironment(val config: TableConfig) {
def explain(table: Table): String
/**
+ * Gets completion hints for the sql query
+ */
+ def getCompletionHints(query: String, pos: Integer) : Array[String] = {
+ val catalogReader = new CalciteCatalogReader(
+ CalciteSchema.from(rootSchema),
+ Collections.emptyList(),
+ typeFactory,
+ CalciteConfig.connectionConfig(getSqlParserConfig))
+ val validator : SqlAdvisorValidator = new SqlAdvisorValidator(
+ getSqlOperatorTable,
+ catalogReader,
+ typeFactory,
+ SqlConformance.DEFAULT)
+ val advisor: SqlAdvisor = new SqlAdvisor(validator)
+ val replaced: Array[String] = Array(null)
+ val hints = advisor.getCompletionHints(query, pos, replaced)
+ .asScala
+ .map(item => item.toIdentifier.toString)
+ hints.toArray
+ }
+
+ /**
* Evaluates a SQL query on registered tables and retrieves the result as a [[Table]].
*
* All tables referenced by the query must be registered in the TableEnvironment.