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/07/05 16:21:57 UTC
[flink] branch master updated:
[FLINK-13081][table-planner][table-api-java] Supports explain DAG plan
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
The following commit(s) were added to refs/heads/master by this push:
new 14cda6c [FLINK-13081][table-planner][table-api-java] Supports explain DAG plan
14cda6c is described below
commit 14cda6ce2b0baf88bf228294fff3c9ee878bb559
Author: godfreyhe <go...@163.com>
AuthorDate: Thu Jul 4 19:25:14 2019 +0800
[FLINK-13081][table-planner][table-api-java] Supports explain DAG plan
This closes #8973
---
.../org/apache/flink/table/api/TableEnvironment.java | 19 +++++++++++++++++++
.../table/api/internal/TableEnvironmentImpl.java | 15 ++++++++++++++-
.../org/apache/flink/table/delegation/Planner.java | 4 ++--
.../flink/table/api/internal/BatchTableEnvImpl.scala | 6 +++++-
.../apache/flink/table/planner/StreamPlanner.scala | 13 +++++++------
.../flink/table/utils/MockTableEnvironment.scala | 4 ++++
6 files changed, 51 insertions(+), 10 deletions(-)
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java
index 64ae899..10505ac 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java
@@ -290,6 +290,25 @@ public interface TableEnvironment {
String explain(Table table);
/**
+ * Returns the AST of the specified Table API and SQL queries and the execution plan to compute
+ * the result of the given {@link Table}.
+ *
+ * @param table The table 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(Table table, boolean extended);
+
+ /**
+ * Returns the AST of the specified Table API and SQL queries and the execution plan to compute
+ * the result of multiple-sinks plan.
+ *
+ * @param extended if the plan should contain additional properties such as
+ * e.g. estimated cost, traits
+ */
+ String explain(boolean extended);
+
+ /**
* Returns completion hints for the given statement at the given cursor position.
* The completion happens case insensitively.
*
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
index a7c9377..7d509b8 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
@@ -68,6 +68,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.stream.Collectors;
/**
* Implementation of {@link TableEnvironment} that works exclusively with Table API interfaces.
@@ -269,7 +270,19 @@ public class TableEnvironmentImpl implements TableEnvironment {
@Override
public String explain(Table table) {
- return planner.explain(Collections.singletonList(table.getQueryOperation()), false);
+ return explain(table, false);
+ }
+
+ @Override
+ public String explain(Table table, boolean extended) {
+ return planner.explain(Collections.singletonList(table.getQueryOperation()), extended);
+ }
+
+ @Override
+ public String explain(boolean extended) {
+ List<Operation> operations = bufferedModifyOperations.stream()
+ .map(o -> (Operation) o).collect(Collectors.toList());
+ return planner.explain(operations, extended);
}
@Override
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/Planner.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/Planner.java
index 727c79b..a6d0e8e 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/Planner.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/delegation/Planner.java
@@ -84,12 +84,12 @@ public interface Planner {
* 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
+ * @param operations 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);
+ String explain(List<Operation> operations, boolean extended);
/**
* Returns completion hints for the given statement at the given cursor position.
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/internal/BatchTableEnvImpl.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/internal/BatchTableEnvImpl.scala
index eb0c543..675f4ed 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/internal/BatchTableEnvImpl.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/internal/BatchTableEnvImpl.scala
@@ -198,10 +198,14 @@ abstract class BatchTableEnvImpl(
s"$sqlPlan"
}
- def explain(table: Table): String = explain(table: Table, extended = false)
+ override def explain(table: Table): String = explain(table: Table, extended = false)
override def execute(jobName: String): JobExecutionResult = execEnv.execute(jobName)
+ override def explain(extended: Boolean): String = {
+ throw new TableException("This method is unsupported in old planner.")
+ }
+
protected def asQueryOperation[T](dataSet: DataSet[T], fields: Option[Array[Expression]])
: DataSetQueryOperation[T] = {
val inputType = dataSet.getType
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/StreamPlanner.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/StreamPlanner.scala
index 8cd11fc..fd5a5c7 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/StreamPlanner.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/StreamPlanner.scala
@@ -124,12 +124,13 @@ class StreamPlanner(
tableOperations.asScala.map(translate).filter(Objects.nonNull).asJava
}
- override def explain(
- tableOperations: util.List[QueryOperation],
- extended: Boolean)
- : String = {
- tableOperations.asScala.map(explain(_, unwrapQueryConfig))
- .mkString(s"${System.lineSeparator}${System.lineSeparator}")
+ override def explain(operations: util.List[Operation], extended: Boolean): String = {
+ operations.asScala.map {
+ case queryOperation: QueryOperation =>
+ explain(queryOperation, unwrapQueryConfig)
+ case operation =>
+ throw new TableException(s"${operation.getClass.getCanonicalName} is not supported")
+ }.mkString(s"${System.lineSeparator}${System.lineSeparator}")
}
override def getCompletionHints(
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/MockTableEnvironment.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/MockTableEnvironment.scala
index 8f138de..cb4567f 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/MockTableEnvironment.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/MockTableEnvironment.scala
@@ -64,6 +64,10 @@ class MockTableEnvironment extends TableEnvironment {
override def explain(table: Table): String = ???
+ override def explain(table: Table, extended: Boolean): String = ???
+
+ override def explain(extended: Boolean): String = ???
+
override def getCompletionHints(statement: String, position: Int): Array[String] = ???
override def sqlQuery(query: String): Table = ???