[FLINK-30668][table-api] Introduce ExplainFormat to Explainable and T…

…ableEnvironment This closes apache#21662
silenceland · Jan 15, 2023 · 91fc5ba · 91fc5ba
1 parent c26daff
commit 91fc5ba
Show file tree

Hide file tree

Showing 9 changed files with 91 additions and 13 deletions.
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/ExplainFormat.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/ExplainFormat.java
@@ -0,0 +1,29 @@
+/*
+ * 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.api;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+/** Explain format categorizes the output format of explain result. */
+@PublicEvolving
+public enum ExplainFormat {
+
+    /** Explain a {@link Explainable} with plain text format. */
+    TEXT
+}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Explainable.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Explainable.java
@@ -36,7 +36,20 @@ public interface Explainable<SELF extends Explainable<SELF>> {
      *     e.g. estimated cost, changelog mode for streaming
      * @return AST and the execution plan.
      */
-    String explain(ExplainDetail... extraDetails);
+    default String explain(ExplainDetail... extraDetails) {
+        return explain(ExplainFormat.TEXT, extraDetails);
+    }
+
+    /**
+     * Returns the AST of this object and the execution plan to compute the result of the given
+     * statement.
+     *
+     * @param format The output format of explained plan
+     * @param extraDetails The extra explain details which the result of this method should include,
+     *     e.g. estimated cost, changelog mode for streaming
+     * @return AST and the execution plan.
+     */
+    String explain(ExplainFormat format, ExplainDetail... extraDetails);
 
     /** Like {@link #explain(ExplainDetail...)}, but piping the result to {@link System#out}. */
     @SuppressWarnings("unchecked")

diff --git a/...table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java b/...table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java
@@ -970,7 +970,21 @@ void createTemporarySystemFunction(
      *     estimated cost, changelog mode for streaming, displaying execution plan in json format
      * @return AST and the execution plan.
      */
-    String explainSql(String statement, ExplainDetail... extraDetails);
+    default String explainSql(String statement, ExplainDetail... extraDetails) {
+        return explainSql(statement, ExplainFormat.TEXT, extraDetails);
+    }
+
+    /**
+     * Returns the AST of the specified statement and the execution plan to compute the result of
+     * the given statement.
+     *
+     * @param statement The statement for which the AST and execution plan will be returned.
+     * @param format The output format of explained plan.
+     * @param extraDetails The extra explain details which the explain result should include, e.g.
+     *     estimated cost, changelog mode for streaming, displaying execution plan in json format
+     * @return AST and the execution plan.
+     */
+    String explainSql(String statement, ExplainFormat format, ExplainDetail... extraDetails);
 
     /**
      * Returns completion hints for the given statement at the given cursor position. The completion

diff --git a/...nk-table-api-java/src/main/java/org/apache/flink/table/api/internal/CompiledPlanImpl.java b/...nk-table-api-java/src/main/java/org/apache/flink/table/api/internal/CompiledPlanImpl.java
@@ -22,6 +22,7 @@
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.table.api.CompiledPlan;
 import org.apache.flink.table.api.ExplainDetail;
+import org.apache.flink.table.api.ExplainFormat;
 import org.apache.flink.table.api.TableResult;
 import org.apache.flink.table.api.config.TableConfigOptions;
 import org.apache.flink.table.delegation.InternalPlan;
@@ -68,7 +69,7 @@ public TableResult execute() {
     }
 
     @Override
-    public String explain(ExplainDetail... extraDetails) {
+    public String explain(ExplainFormat format, ExplainDetail... extraDetails) {
         return tableEnvironment.explainPlan(internalPlan, extraDetails);
     }
 

diff --git a/...nk-table-api-java/src/main/java/org/apache/flink/table/api/internal/StatementSetImpl.java b/...nk-table-api-java/src/main/java/org/apache/flink/table/api/internal/StatementSetImpl.java
@@ -23,6 +23,7 @@
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.table.api.CompiledPlan;
 import org.apache.flink.table.api.ExplainDetail;
+import org.apache.flink.table.api.ExplainFormat;
 import org.apache.flink.table.api.StatementSet;
 import org.apache.flink.table.api.Table;
 import org.apache.flink.table.api.TableDescriptor;
@@ -96,10 +97,10 @@ public StatementSet addInsert(
     }
 
     @Override
-    public String explain(ExplainDetail... extraDetails) {
+    public String explain(ExplainFormat format, ExplainDetail... extraDetails) {
         List<Operation> operationList =
                 operations.stream().map(o -> (Operation) o).collect(Collectors.toList());
-        return tableEnvironment.explainInternal(operationList, extraDetails);
+        return tableEnvironment.explainInternal(operationList, format, extraDetails);
     }
 
     @Override

diff --git a/...able-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java b/...able-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
@@ -28,6 +28,7 @@
 import org.apache.flink.table.api.DataTypes;
 import org.apache.flink.table.api.EnvironmentSettings;
 import org.apache.flink.table.api.ExplainDetail;
+import org.apache.flink.table.api.ExplainFormat;
 import org.apache.flink.table.api.PlanReference;
 import org.apache.flink.table.api.ResultKind;
 import org.apache.flink.table.api.SqlParserException;
@@ -689,7 +690,8 @@ public String[] listFunctions() {
     }
 
     @Override
-    public String explainSql(String statement, ExplainDetail... extraDetails) {
+    public String explainSql(
+            String statement, ExplainFormat format, ExplainDetail... extraDetails) {
         List<Operation> operations = getParser().parse(statement);
 
         if (operations.size() != 1) {
@@ -701,11 +703,12 @@ public String explainSql(String statement, ExplainDetail... extraDetails) {
             operations =
                     new ArrayList<>(((StatementSetOperation) operations.get(0)).getOperations());
         }
-        return explainInternal(operations, extraDetails);
+        return explainInternal(operations, format, extraDetails);
     }
 
     @Override
-    public String explainInternal(List<Operation> operations, ExplainDetail... extraDetails) {
+    public String explainInternal(
+            List<Operation> operations, ExplainFormat format, ExplainDetail... extraDetails) {
         operations =
                 operations.stream()
                         .filter(o -> !(o instanceof NopOperation))

diff --git a/...-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentInternal.java b/...-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentInternal.java
@@ -22,6 +22,7 @@
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.table.api.CompiledPlan;
 import org.apache.flink.table.api.ExplainDetail;
+import org.apache.flink.table.api.ExplainFormat;
 import org.apache.flink.table.api.Table;
 import org.apache.flink.table.api.TableEnvironment;
 import org.apache.flink.table.catalog.CatalogManager;
@@ -91,7 +92,21 @@ public interface TableEnvironmentInternal extends TableEnvironment {
      *     estimated cost, changelog mode for streaming
      * @return AST and the execution plan.
      */
-    String explainInternal(List<Operation> operations, ExplainDetail... extraDetails);
+    default String explainInternal(List<Operation> operations, ExplainDetail... extraDetails) {
+        return explainInternal(operations, ExplainFormat.TEXT, extraDetails);
+    }
+
+    /**
+     * Returns the AST of this table and the execution plan to compute the result of this table.
+     *
+     * @param operations The operations to be explained.
+     * @param format The output format.
+     * @param extraDetails The extra explain details which the explain result should include, e.g.
+     *     estimated cost, changelog mode for streaming
+     * @return AST and the execution plan.
+     */
+    String explainInternal(
+            List<Operation> operations, ExplainFormat format, ExplainDetail... extraDetails);
 
     /**
      * Registers an external {@link TableSource} in this {@link TableEnvironment}'s catalog.

diff --git a/...ble/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableImpl.java b/...ble/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableImpl.java
@@ -21,6 +21,7 @@
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.table.api.AggregatedTable;
 import org.apache.flink.table.api.ExplainDetail;
+import org.apache.flink.table.api.ExplainFormat;
 import org.apache.flink.table.api.FlatAggregateTable;
 import org.apache.flink.table.api.GroupWindow;
 import org.apache.flink.table.api.GroupWindowedTable;
@@ -476,9 +477,9 @@ public TableResult execute() {
     }
 
     @Override
-    public String explain(ExplainDetail... extraDetails) {
+    public String explain(ExplainFormat format, ExplainDetail... extraDetails) {
         return tableEnvironment.explainInternal(
-                Collections.singletonList(getQueryOperation()), extraDetails);
+                Collections.singletonList(getQueryOperation()), format, extraDetails);
     }
 
     @Override

diff --git a/...k-table-api-java/src/main/java/org/apache/flink/table/api/internal/TablePipelineImpl.java b/...k-table-api-java/src/main/java/org/apache/flink/table/api/internal/TablePipelineImpl.java
@@ -20,6 +20,7 @@
 
 import org.apache.flink.table.api.CompiledPlan;
 import org.apache.flink.table.api.ExplainDetail;
+import org.apache.flink.table.api.ExplainFormat;
 import org.apache.flink.table.api.TableException;
 import org.apache.flink.table.api.TablePipeline;
 import org.apache.flink.table.api.TableResult;
@@ -57,8 +58,8 @@ public TableResult execute() {
     }
 
     @Override
-    public String explain(ExplainDetail... extraDetails) {
-        return tableEnvironment.explainInternal(singletonList(operation), extraDetails);
+    public String explain(ExplainFormat format, ExplainDetail... extraDetails) {
+        return tableEnvironment.explainInternal(singletonList(operation), format, extraDetails);
     }
 
     @Override