Skip to content

Commit

Permalink
[FLINK-21974][table] Support to match quoted values for SET option st…
Browse files Browse the repository at this point in the history 
…atement

This closes apache#15372
  • Loading branch information
@fsk119 @wuchong
fsk119 authored and wuchong committed Mar 26, 2021
1 parent 9c95cc1 commit 17a4cba
Show file tree
Hide file tree
Showing 5 changed files with 108 additions and 76 deletions.
36 changes: 28 additions & 8 deletions ...-table-api-java/src/main/java/org/apache/flink/table/operations/command/SetOperation.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,25 +20,45 @@

import org.apache.flink.table.operations.Operation;

/** Operation to represent SET command. */
import javax.annotation.Nullable;

import java.util.Optional;

import static org.apache.flink.util.Preconditions.checkNotNull;

/**
* Operation to represent SET command. If {@link #getKey()} and {@link #getValue()} are empty, it
* means show all the configurations. Otherwise, set value to the configuration key.
*/
public class SetOperation implements Operation {

private final String[] operands;
@Nullable private final String key;
@Nullable private final String value;

public SetOperation() {
this.key = null;
this.value = null;
}

public SetOperation(String key, String value) {
this.key = checkNotNull(key);
this.value = checkNotNull(value);
}

public SetOperation(String[] operands) {
this.operands = operands;
public Optional<String> getKey() {
return Optional.ofNullable(key);
}

public String[] getOperands() {
return operands;
public Optional<String> getValue() {
return Optional.ofNullable(value);
}

@Override
public String asSummaryString() {
if (operands.length == 0) {
if (key == null && value == null) {
return "SET";
} else {
return String.format("SET %s=%s", operands[0], operands[1]);
return String.format("SET %s=%s", key, value);
}
}
}
22 changes: 13 additions & 9 deletions ...r-blink/src/main/java/org/apache/flink/table/planner/parse/SetOperationParseStrategy.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,27 +33,31 @@ public class SetOperationParseStrategy extends AbstractRegexParseStrategy {
static final SetOperationParseStrategy INSTANCE = new SetOperationParseStrategy();

protected SetOperationParseStrategy() {
super(Pattern.compile("SET(\\s+(\\S+)\\s*=(.+))?", DEFAULT_PATTERN_FLAGS));
super(
Pattern.compile(
"SET(\\s+(?<key>\\S+)\\s*=\\s*('(?<quotedVal>[^']*)'|(?<val>\\S+)))?",
DEFAULT_PATTERN_FLAGS));
}

@Override
public Operation convert(String statement) {
Matcher matcher = pattern.matcher(statement.trim());
final List<String> operands = new ArrayList<>();
if (matcher.find()) {
for (int i = 0; i < matcher.groupCount(); i++) {
if (matcher.group(i + 1) != null) {
operands.add(matcher.group(i + 1));
}
if (matcher.group("key") != null) {
operands.add(matcher.group("key"));
operands.add(
matcher.group("quotedVal") != null
? matcher.group("quotedVal")
: matcher.group("val"));
}
}

// only capture SET
if (operands.isEmpty()) {
return new SetOperation(new String[0]);
} else if (operands.size() == 3) {
return new SetOperation(
operands.subList(1, 3).stream().map(String::trim).toArray(String[]::new));
return new SetOperation();
} else if (operands.size() == 2) {
return new SetOperation(operands.get(0), operands.get(1));
} else {
// impossible
throw new TableException(
Expand Down
98 changes: 49 additions & 49 deletions ...planner-blink/src/test/java/org/apache/flink/table/planner/delegation/ParserImplTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,27 +31,24 @@
import org.apache.flink.table.planner.catalog.CatalogManagerCalciteSchema;
import org.apache.flink.table.utils.CatalogManagerMocks;

import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;

import javax.annotation.Nullable;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.function.Supplier;

import static org.apache.calcite.jdbc.CalciteSchemaBuilder.asRootSchema;
import static org.apache.flink.core.testutils.CommonTestUtils.assertThrows;
import static org.apache.flink.table.planner.delegation.ParserImplTest.TestSpec.forStatement;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.fail;
import static org.junit.Assert.assertEquals;

/** Test for {@link ParserImpl}. */
public class ParserImplTest {

@Rule public ExpectedException thrown = ExpectedException.none();

private final boolean isStreamingMode = false;
private final TableConfig tableConfig = new TableConfig();
private final Catalog catalog = new GenericInMemoryCatalog("MockCatalog", "default");
Expand Down Expand Up @@ -83,46 +80,41 @@ public class ParserImplTest {
plannerContext.createSqlExprToRexConverter(
plannerContext.getTypeFactory().buildRelNodeRowType(t)));

private List<TestSpec> testLegalStatements;
private List<TestSpec> testIllegalStatements;

@Before
public void setup() {
testLegalStatements =
Arrays.asList(
TestSpec.forStatement("ClEaR").expectedSummary("CLEAR"),
TestSpec.forStatement("hElP").expectedSummary("HELP"),
TestSpec.forStatement("qUIT").expectedSummary("QUIT"),
TestSpec.forStatement("ExIT").expectedSummary("EXIT"),
TestSpec.forStatement("REsEt").expectedSummary("RESET"),
TestSpec.forStatement(" SEt ").expectedSummary("SET"),
TestSpec.forStatement("SET execution.runtime-type=batch")
.expectedSummary("SET execution.runtime-type=batch"),
TestSpec.forStatement("SET pipeline.jars = /path/to/test-_-jar.jar")
.expectedSummary("SET pipeline.jars=/path/to/test-_-jar.jar"),
TestSpec.forStatement("USE test.db1").expectedSummary("USE test.db1"),
TestSpec.forStatement("SHOW tables").expectedSummary("SHOW TABLES"),
TestSpec.forStatement("SET pipeline.name = ' '")
.expectedSummary("SET pipeline.name = ' '"));

testIllegalStatements =
Collections.singletonList(TestSpec.forStatement("SET execution.runtime-type="));
}
private static final List<TestSpec> TEST_SPECS =
Arrays.asList(
forStatement("ClEaR").summary("CLEAR"),
forStatement("hElP").summary("HELP"),
forStatement("qUIT").summary("QUIT"),
forStatement("ExIT").summary("QUIT"),
forStatement("REsEt").summary("RESET"),
forStatement(" SEt ").summary("SET"),
forStatement("SET execution.runtime-type=batch")
.summary("SET execution.runtime-type=batch"),
forStatement("SET pipeline.jars = /path/to/test-_-jar.jar")
.summary("SET pipeline.jars=/path/to/test-_-jar.jar"),
forStatement("USE test.db1").summary("USE test.db1"),
forStatement("SHOW tables").summary("SHOW TABLES"),
forStatement("SET pipeline.name = 'test name'")
.summary("SET pipeline.name=test name"),
forStatement("SET pipeline.name = ' '").summary("SET pipeline.name= "),
forStatement("SET execution.runtime-type=")
// TODO: the exception message should be "no value defined"
.error("SQL parse failed. Encountered \"-\" at line 1, column 22"));

@Test
public void testParseLegalStatements() {
for (TestSpec spec : testLegalStatements) {
Operation op = parser.parse(spec.statement).get(0);
Assert.assertEquals(op.asSummaryString(), op.asSummaryString());
}
}

@Test
public void testParseIllegalStatements() {
thrown.expect(SqlParserException.class);
for (TestSpec spec : testIllegalStatements) {
parser.parse(spec.statement);
fail("Should fail.");
for (TestSpec spec : TEST_SPECS) {
if (spec.expectedSummary != null) {
Operation op = parser.parse(spec.statement).get(0);
assertEquals(spec.expectedSummary, op.asSummaryString());
}

if (spec.expectedError != null) {
assertThrows(
spec.expectedError,
SqlParserException.class,
() -> parser.parse(spec.statement));
}
}
}

Expand All @@ -136,10 +128,13 @@ public void testCompletionTest() {

// ~ Tool Methods ----------------------------------------------------------

private static class TestSpec {
static class TestSpec {

private final String statement;

@Nullable private String expectedSummary;

private String statement;
private String expectedSummary;
@Nullable private String expectedError;

private TestSpec(String statement) {
this.statement = statement;
Expand All @@ -149,10 +144,15 @@ static TestSpec forStatement(String statement) {
return new TestSpec(statement);
}

TestSpec expectedSummary(String expectedSummary) {
TestSpec summary(String expectedSummary) {
this.expectedSummary = expectedSummary;
return this;
}

TestSpec error(String expectedError) {
this.expectedError = expectedError;
return this;
}
}

private void verifySqlCompletion(String statement, int position, String[] expectedHints) {
Expand Down
22 changes: 13 additions & 9 deletions ...k-table-planner/src/main/java/org/apache/flink/table/parse/SetOperationParseStrategy.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,27 +33,31 @@ public class SetOperationParseStrategy extends AbstractRegexParseStrategy {
static final SetOperationParseStrategy INSTANCE = new SetOperationParseStrategy();

protected SetOperationParseStrategy() {
super(Pattern.compile("SET(\\s+(\\S+)\\s*=(.+))?", DEFAULT_PATTERN_FLAGS));
super(
Pattern.compile(
"SET(\\s+(?<key>\\S+)\\s*=\\s*('(?<quotedVal>[^']*)'|(?<val>\\S+)))?",
DEFAULT_PATTERN_FLAGS));
}

@Override
public Operation convert(String statement) {
Matcher matcher = pattern.matcher(statement.trim());
final List<String> operands = new ArrayList<>();
if (matcher.find()) {
for (int i = 0; i < matcher.groupCount(); i++) {
if (matcher.group(i + 1) != null) {
operands.add(matcher.group(i + 1));
}
if (matcher.group("key") != null) {
operands.add(matcher.group("key"));
operands.add(
matcher.group("quotedVal") != null
? matcher.group("quotedVal")
: matcher.group("val"));
}
}

// only capture SET
if (operands.isEmpty()) {
return new SetOperation(new String[0]);
} else if (operands.size() == 3) {
return new SetOperation(
operands.subList(1, 3).stream().map(String::trim).toArray(String[]::new));
return new SetOperation();
} else if (operands.size() == 2) {
return new SetOperation(operands.get(0), operands.get(1));
} else {
// impossible
throw new TableException(
Expand Down
6 changes: 5 additions & 1 deletion ...ble-planner/src/test/java/org/apache/flink/table/sqlexec/SqlToOperationConverterTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,7 @@

import javax.annotation.Nullable;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
Expand Down Expand Up @@ -747,8 +748,11 @@ private void assertSimpleCommand(String statement, Matcher<? super Operation> ma

private void assertSetCommand(String statement, String... operands) {
SetOperation operation = (SetOperation) parser.parse(statement).get(0);
List<String> actualOperands = new ArrayList<>();
operation.getKey().ifPresent(actualOperands::add);
operation.getValue().ifPresent(actualOperands::add);

assertArrayEquals(operands, operation.getOperands());
assertArrayEquals(operands, actualOperands.toArray(new String[0]));
}

private void assertFailedSetCommand(String statement) {
Expand Down

0 comments on commit 17a4cba

Please sign in to comment.