You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2019/06/07 08:53:09 UTC
[flink] 04/04: [FLINK-12690][table-api] Introduce a Planner
interface
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2ea0aba4bf4912c5672eb99f7e6aa5601e979bc0
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Wed May 29 15:50:48 2019 +0200
[FLINK-12690][table-api] Introduce a Planner interface
The Planner interface separates parsing, optimization & translating from
a relational QueryTableOperation representation into a runnable graph of
StreamTransformations. This enables switching between different
implementations of the planner in a TableEnvironment.
---
.../operations/CatalogSinkModifyOperation.java | 68 ++++++++++++++
.../{Operation.java => ModifyOperation.java} | 27 +++---
...{Operation.java => ModifyOperationVisitor.java} | 24 ++---
.../apache/flink/table/operations/Operation.java | 4 +-
.../OutputConversionModifyOperation.java | 88 ++++++++++++++++++
.../UnregisteredSinkModifyOperation.java | 62 +++++++++++++
.../org/apache/flink/table/planner/Planner.java | 103 +++++++++++++++++++++
7 files changed, 349 insertions(+), 27 deletions(-)
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/CatalogSinkModifyOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/CatalogSinkModifyOperation.java
new file mode 100644
index 0000000..4fee1c6
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/CatalogSinkModifyOperation.java
@@ -0,0 +1,68 @@
+/*
+ * 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.operations;
+
+import org.apache.flink.annotation.Internal;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * DML operation that tells to write to a sink. The sink has to be looked up in a
+ * {@link org.apache.flink.table.catalog.Catalog}.
+ */
+@Internal
+public class CatalogSinkModifyOperation implements ModifyOperation {
+
+ private final List<String> tablePath;
+ private final QueryOperation child;
+
+ public CatalogSinkModifyOperation(List<String> tablePath, QueryOperation child) {
+ this.tablePath = tablePath;
+ this.child = child;
+ }
+
+ public List<String> getTablePath() {
+ return tablePath;
+ }
+
+ @Override
+ public QueryOperation getChild() {
+ return child;
+ }
+
+ @Override
+ public <T> T accept(ModifyOperationVisitor<T> visitor) {
+ return visitor.visit(this);
+ }
+
+ @Override
+ public String asSummaryString() {
+ Map<String, Object> params = new LinkedHashMap<>();
+ params.put("tablePath", tablePath);
+
+ return OperationUtils.formatWithChildren(
+ "CatalogSink",
+ params,
+ Collections.singletonList(child),
+ Operation::asSummaryString);
+ }
+}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/Operation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ModifyOperation.java
similarity index 58%
copy from flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/Operation.java
copy to flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ModifyOperation.java
index 91a15a7..63f205a 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/Operation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ModifyOperation.java
@@ -18,21 +18,24 @@
package org.apache.flink.table.operations;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.Internal;
+
+import java.util.List;
/**
- * Covers all sort of Table operations such as queries(DQL), modifications(DML), definitions(DDL),
- * or control actions(DCL).
+ * A {@link Operation} that describes the DML queries such as e.g. INSERT or conversion to a
+ * DataStream.
+ *
+ * <p>A tree of {@link QueryOperation} with a {@link ModifyOperation} on top
+ * represents a runnable query that can be transformed into a graph of
+ * {@link org.apache.flink.streaming.api.transformations.StreamTransformation}
+ * via {@link org.apache.flink.table.planner.Planner#translate(List)}
*
* @see QueryOperation
*/
-@PublicEvolving
-public interface Operation {
- /**
- * Returns a string that summarizes this operation for printing to a console. An implementation might
- * skip very specific properties.
- *
- * @return summary string of this operation for debugging purposes
- */
- String asSummaryString();
+@Internal
+public interface ModifyOperation extends Operation {
+ QueryOperation getChild();
+
+ <T> T accept(ModifyOperationVisitor<T> visitor);
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/Operation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ModifyOperationVisitor.java
similarity index 63%
copy from flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/Operation.java
copy to flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ModifyOperationVisitor.java
index 91a15a7..63dd156 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/Operation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ModifyOperationVisitor.java
@@ -18,21 +18,17 @@
package org.apache.flink.table.operations;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.Internal;
/**
- * Covers all sort of Table operations such as queries(DQL), modifications(DML), definitions(DDL),
- * or control actions(DCL).
- *
- * @see QueryOperation
+ * Class that implements visitor pattern. It allows type safe logic on top of tree
+ * of {@link ModifyOperation}s.
*/
-@PublicEvolving
-public interface Operation {
- /**
- * Returns a string that summarizes this operation for printing to a console. An implementation might
- * skip very specific properties.
- *
- * @return summary string of this operation for debugging purposes
- */
- String asSummaryString();
+@Internal
+public interface ModifyOperationVisitor<T> {
+ T visit(CatalogSinkModifyOperation catalogSink);
+
+ T visit(OutputConversionModifyOperation outputConversion);
+
+ <U> T visit(UnregisteredSinkModifyOperation<U> unregisteredSink);
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/Operation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/Operation.java
index 91a15a7..2ce8bfe 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/Operation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/Operation.java
@@ -22,9 +22,11 @@ import org.apache.flink.annotation.PublicEvolving;
/**
* Covers all sort of Table operations such as queries(DQL), modifications(DML), definitions(DDL),
- * or control actions(DCL).
+ * or control actions(DCL). This is the output of
+ * {@link org.apache.flink.table.planner.Planner#parse(String)}.
*
* @see QueryOperation
+ * @see ModifyOperation
*/
@PublicEvolving
public interface Operation {
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/OutputConversionModifyOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/OutputConversionModifyOperation.java
new file mode 100644
index 0000000..bda9035
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/OutputConversionModifyOperation.java
@@ -0,0 +1,88 @@
+/*
+ * 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.operations;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.transformations.StreamTransformation;
+import org.apache.flink.table.types.DataType;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * Special, internal kind of {@link ModifyOperation} that allows converting a tree of
+ * {@link QueryOperation}s to a {@link StreamTransformation} of given type described with
+ * {@link TypeInformation}. This is used to convert a relational query to a datastream.
+ */
+@Internal
+public class OutputConversionModifyOperation implements ModifyOperation {
+
+ /**
+ * Should the output type contain the change flag, and what should the
+ * flag represent (retraction or deletion).
+ */
+ public enum UpdateMode {
+ APPEND,
+ RETRACT,
+ UPSERT
+ }
+
+ private final QueryOperation child;
+ private final DataType type;
+ private final UpdateMode updateMode;
+
+ public OutputConversionModifyOperation(QueryOperation child, DataType type, UpdateMode updateMode) {
+ this.child = child;
+ this.type = type;
+ this.updateMode = updateMode;
+ }
+
+ public UpdateMode getUpdateMode() {
+ return updateMode;
+ }
+
+ public DataType getType() {
+ return type;
+ }
+
+ @Override
+ public QueryOperation getChild() {
+ return child;
+ }
+
+ @Override
+ public String asSummaryString() {
+ Map<String, Object> params = new LinkedHashMap<>();
+ params.put("updateMode", updateMode);
+ params.put("type", type);
+
+ return OperationUtils.formatWithChildren(
+ "Output",
+ params,
+ Collections.singletonList(child),
+ Operation::asSummaryString);
+ }
+
+ @Override
+ public <R> R accept(ModifyOperationVisitor<R> visitor) {
+ return visitor.visit(this);
+ }
+}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/UnregisteredSinkModifyOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/UnregisteredSinkModifyOperation.java
new file mode 100644
index 0000000..1c42a3e
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/UnregisteredSinkModifyOperation.java
@@ -0,0 +1,62 @@
+/*
+ * 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.operations;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.sinks.TableSink;
+
+import java.util.Collections;
+
+/**
+ * DML operation that tells to write to the given sink.
+ */
+@Internal
+public class UnregisteredSinkModifyOperation<T> implements ModifyOperation {
+
+ private final TableSink<T> sink;
+ private final QueryOperation child;
+
+ public UnregisteredSinkModifyOperation(TableSink<T> sink, QueryOperation child) {
+ this.sink = sink;
+ this.child = child;
+ }
+
+ public TableSink<T> getSink() {
+ return sink;
+ }
+
+ @Override
+ public QueryOperation getChild() {
+ return child;
+ }
+
+ @Override
+ public String asSummaryString() {
+ return OperationUtils.formatWithChildren(
+ "UnregisteredSink",
+ Collections.emptyMap(),
+ Collections.singletonList(child),
+ Operation::asSummaryString);
+ }
+
+ @Override
+ public <R> R accept(ModifyOperationVisitor<R> visitor) {
+ return visitor.visit(this);
+ }
+}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/planner/Planner.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/planner/Planner.java
new file mode 100644
index 0000000..aedc159
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/planner/Planner.java
@@ -0,0 +1,103 @@
+/*
+ * 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.planner;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.streaming.api.transformations.StreamTransformation;
+import org.apache.flink.table.operations.ModifyOperation;
+import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.operations.QueryOperation;
+
+import java.util.List;
+
+/**
+ * This interface serves two purposes:
+ * <ul>
+ * <li>SQL parser - transforms a SQL string into a Table API specific tree of
+ * {@link Operation}s</li>
+ * <li>relational planner - provides a way to plan, optimize and transform tree of
+ * {@link ModifyOperation} into a runnable form ({@link StreamTransformation})</li>
+ * </ul>.
+ *
+ * <p>The Planner is execution agnostic. It is up to the
+ * {@link org.apache.flink.table.api.TableEnvironment} to ensure that if any of the
+ * {@link QueryOperation} pull any runtime configuration, all those configurations are
+ * equivalent. Example: If some of the {@link QueryOperation}s scan DataStreams, all
+ * those DataStreams must come from the same StreamExecutionEnvironment, because the result
+ * of {@link Planner#translate(List)} will strip any execution configuration from
+ * the DataStream information.
+ *
+ * <p>All Tables referenced in either {@link Planner#parse(String)} or
+ * {@link Planner#translate(List)} should be previously registered in a
+ * {@link org.apache.flink.table.catalog.CatalogManager}, which will be provided during
+ * instantiation of the {@link Planner}.
+ */
+@Internal
+public interface Planner {
+
+ /**
+ * Entry point for parsing sql queries expressed as a String.
+ *
+ * <p><b>Note:</b>If the created {@link Operation} is a {@link QueryOperation}
+ * it must be in a form that will be understood by the
+ * {@link Planner#translate(List)} method.
+ *
+ * <p>The produced Operation trees should already be validated.
+ *
+ * @param statement the sql statement to evaluate
+ * @return parsed queries as trees of relational {@link Operation}s
+ */
+ List<Operation> parse(String statement);
+
+ /**
+ * Converts a relational tree of {@link ModifyOperation}s into a set of runnable
+ * {@link StreamTransformation}s.
+ *
+ * <p>This method accepts a list of {@link ModifyOperation}s to allow reusing common
+ * subtrees of multiple relational queries. Each query's top node should be a {@link ModifyOperation}
+ * in order to pass the expected properties of the output {@link StreamTransformation} such as
+ * output mode (append, retract, upsert) or the expected output type.
+ *
+ * @param modifyOperations list of relational operations to plan, optimize and convert in a
+ * single run.
+ * @return list of corresponding {@link StreamTransformation}s.
+ */
+ List<StreamTransformation<?>> translate(List<ModifyOperation> modifyOperations);
+
+ /**
+ * Returns the AST of the specified Table API and SQL queries and the execution plan
+ * to compute the result of the given collection of {@link QueryOperation}s.
+ *
+ * @param queryOperations The collection of relational queries for which the AST
+ * and execution plan will be returned.
+ * @param extended if the plan should contain additional properties such as
+ * e.g. estimated cost, traits
+ */
+ String explain(List<QueryOperation> queryOperations, boolean extended);
+
+ /**
+ * 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
+ */
+ String[] getCompletionHints(String statement, int position);
+}