[FLINK-21463][table-api] Support to parse 'RESET key' command

deepfield · Mar 27, 2021 · 1080a63 · 1080a63
1 parent 5b33c79
commit 1080a63
Show file tree

Hide file tree

Showing 4 changed files with 56 additions and 6 deletions.
diff --git a/...able-api-java/src/main/java/org/apache/flink/table/operations/command/ResetOperation.java b/...able-api-java/src/main/java/org/apache/flink/table/operations/command/ResetOperation.java
@@ -20,10 +20,32 @@
 
 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/...blink/src/main/java/org/apache/flink/table/planner/parse/ResetOperationParseStrategy.java b/...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/...planner-blink/src/test/java/org/apache/flink/table/planner/delegation/ParserImplTest.java b/...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/...table-planner/src/main/java/org/apache/flink/table/parse/ResetOperationParseStrategy.java b/...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