You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ku...@apache.org on 2019/06/20 10:56:11 UTC
[flink] 01/03: [FLINK-12815] [table-planner-blink] Do not register
the intermediate optimization result as a table in batch and stream
CommonSubGraphBasedOptimizer
This is an automated email from the ASF dual-hosted git repository.
kurt pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 9a292e228533520fc4be4da6cca99f6fb7dde80a
Author: godfreyhe <go...@163.com>
AuthorDate: Wed Jun 12 19:50:08 2019 +0800
[FLINK-12815] [table-planner-blink] Do not register the intermediate optimization result as a table in batch and stream CommonSubGraphBasedOptimizer
---
.../BatchCommonSubGraphBasedOptimizer.scala | 15 ++++-------
.../optimize/CommonSubGraphBasedOptimizer.scala | 29 +++++++++++++++++++--
.../flink/table/plan/optimize/RelNodeBlock.scala | 4 +--
.../StreamCommonSubGraphBasedOptimizer.scala | 30 +++++++++-------------
4 files changed, 46 insertions(+), 32 deletions(-)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/BatchCommonSubGraphBasedOptimizer.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/BatchCommonSubGraphBasedOptimizer.scala
index 6071cf1..073d673 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/BatchCommonSubGraphBasedOptimizer.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/BatchCommonSubGraphBasedOptimizer.scala
@@ -18,7 +18,7 @@
package org.apache.flink.table.plan.optimize
-import org.apache.flink.table.api.{BatchTableEnvironment, TableConfig, TableImpl}
+import org.apache.flink.table.api.{BatchTableEnvironment, TableConfig}
import org.apache.flink.table.plan.nodes.physical.batch.BatchExecSink
import org.apache.flink.table.plan.optimize.program.{BatchOptimizeContext, FlinkBatchProgram}
import org.apache.flink.table.plan.schema.IntermediateRelTable
@@ -53,20 +53,15 @@ class BatchCommonSubGraphBasedOptimizer(tEnv: BatchTableEnvironment)
optimizedTree match {
case _: BatchExecSink[_] => // ignore
case _ =>
- val name = tEnv.createUniqueTableName()
- registerIntermediateTable(name, optimizedTree)
- val newTable = tEnv.scan(name)
- block.setNewOutputNode(newTable.asInstanceOf[TableImpl].getRelNode)
+ val name = createUniqueIntermediateRelTableName
+ val intermediateRelTable = new IntermediateRelTable(optimizedTree)
+ val newTableScan = wrapIntermediateRelTableToTableScan(intermediateRelTable, name)
+ block.setNewOutputNode(newTableScan)
block.setOutputTableName(name)
}
block.setOptimizedPlan(optimizedTree)
}
- private def registerIntermediateTable(name: String, relNode: RelNode): Unit = {
- val table = new IntermediateRelTable(relNode)
- tEnv.registerTableInternal(name, table)
- }
-
/**
* Generates the optimized [[RelNode]] tree from the original relational node tree.
*
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/CommonSubGraphBasedOptimizer.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/CommonSubGraphBasedOptimizer.scala
index 8627d86..e7f720b 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/CommonSubGraphBasedOptimizer.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/CommonSubGraphBasedOptimizer.scala
@@ -18,12 +18,15 @@
package org.apache.flink.table.plan.optimize
-import org.apache.flink.table.plan.schema.IntermediateRelTable
+import org.apache.flink.table.plan.schema.{FlinkRelOptTable, IntermediateRelTable}
import org.apache.calcite.rel.core.TableScan
+import org.apache.calcite.rel.logical.LogicalTableScan
import org.apache.calcite.rel.{RelNode, RelShuttleImpl}
import java.util
+import java.util.Collections
+import java.util.concurrent.atomic.AtomicInteger
import scala.collection.JavaConversions._
@@ -38,7 +41,7 @@ import scala.collection.JavaConversions._
* 1. Decompose [[RelNode]] DAG into multiple [[RelNodeBlock]]s, and build [[RelNodeBlock]] DAG.
* Each [[RelNodeBlock]] has only one sink output, and represents a common sub-graph.
* 2. optimize recursively each [[RelNodeBlock]] from leaf block to root(sink) block,
- * and register the optimized result of non-root block as an [[IntermediateRelTable]].
+ * and wrap the optimized result of non-root block as an [[IntermediateRelTable]].
* 3. expand [[IntermediateRelTable]] into RelNode tree in each [[RelNodeBlock]].
*
* Currently, we choose this strategy based on the following considerations:
@@ -54,6 +57,16 @@ import scala.collection.JavaConversions._
abstract class CommonSubGraphBasedOptimizer extends Optimizer {
/**
+ * ID for IntermediateRelTable.
+ */
+ private val NEXT_ID = new AtomicInteger(0)
+
+ protected def createUniqueIntermediateRelTableName: String = {
+ val id = NEXT_ID.getAndIncrement()
+ s"IntermediateRelTable_$id"
+ }
+
+ /**
* Generates the optimized [[RelNode]] DAG from the original relational nodes.
* NOTES: the reused node in result DAG will be converted to the same RelNode,
* and the result doesn't contain [[IntermediateRelTable]].
@@ -80,6 +93,18 @@ abstract class CommonSubGraphBasedOptimizer extends Optimizer {
protected def doOptimize(roots: Seq[RelNode]): Seq[RelNodeBlock]
/**
+ * Returns a new table scan which wraps the given IntermediateRelTable.
+ */
+ protected def wrapIntermediateRelTableToTableScan(
+ relTable: IntermediateRelTable,
+ name: String): TableScan = {
+ val cluster = relTable.relNode.getCluster
+ val table = FlinkRelOptTable.create(
+ null, relTable.getRowType(cluster.getTypeFactory), Collections.singletonList(name), relTable)
+ new LogicalTableScan(cluster, cluster.traitSet, table)
+ }
+
+ /**
* Expand [[IntermediateRelTable]] in each RelNodeBlock.
*/
private def expandIntermediateTableScan(nodes: Seq[RelNode]): Seq[RelNode] = {
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/RelNodeBlock.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/RelNodeBlock.scala
index b25e3c1..6a464ff 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/RelNodeBlock.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/RelNodeBlock.scala
@@ -103,8 +103,8 @@ import scala.collection.mutable
* }}}
*
* The optimizing order is from child block to parent. The optimized result (RelNode)
- * will be registered into tables first, and then be converted to a new TableScan which is the
- * new output node of current block and is also the input of its parent blocks.
+ * will be wrapped as an IntermediateRelTable first, and then be converted to a new TableScan
+ * which is the new output node of current block and is also the input of its parent blocks.
*
* @param outputNode A RelNode of the output in the block, which could be a [[Sink]] or
* other RelNode which data outputs to multiple [[Sink]]s.
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/StreamCommonSubGraphBasedOptimizer.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/StreamCommonSubGraphBasedOptimizer.scala
index 1ae7151..a30f1d2 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/StreamCommonSubGraphBasedOptimizer.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/StreamCommonSubGraphBasedOptimizer.scala
@@ -18,7 +18,7 @@
package org.apache.flink.table.plan.optimize
-import org.apache.flink.table.api.{StreamTableEnvironment, TableConfig, TableImpl}
+import org.apache.flink.table.api.{StreamTableEnvironment, TableConfig}
import org.apache.flink.table.plan.`trait`.{AccMode, AccModeTraitDef, UpdateAsRetractionTraitDef}
import org.apache.flink.table.plan.metadata.FlinkRelMetadataQuery
import org.apache.flink.table.plan.nodes.calcite.Sink
@@ -111,10 +111,10 @@ class StreamCommonSubGraphBasedOptimizer(tEnv: StreamTableEnvironment)
isSinkBlock = isSinkBlock)
val isAccRetract = optimizedPlan.getTraitSet
.getTrait(AccModeTraitDef.INSTANCE).getAccMode == AccMode.AccRetract
- val name = tEnv.createUniqueTableName()
- registerIntermediateTable(tEnv, name, optimizedPlan, isAccRetract)
- val newTable = tEnv.scan(name)
- block.setNewOutputNode(newTable.asInstanceOf[TableImpl].getRelNode)
+ val name = createUniqueIntermediateRelTableName
+ val intermediateRelTable = createIntermediateRelTable(optimizedPlan, isAccRetract)
+ val newTableScan = wrapIntermediateRelTableToTableScan(intermediateRelTable, name)
+ block.setNewOutputNode(newTableScan)
block.setOutputTableName(name)
block.setOptimizedPlan(optimizedPlan)
}
@@ -181,10 +181,10 @@ class StreamCommonSubGraphBasedOptimizer(tEnv: StreamTableEnvironment)
case o =>
val optimizedPlan = optimizeTree(
o, retractionFromRoot, isSinkBlock = isSinkBlock)
- val name = tEnv.createUniqueTableName()
- registerIntermediateTable(tEnv, name, optimizedPlan, isAccRetract = false)
- val newTable = tEnv.scan(name)
- block.setNewOutputNode(newTable.asInstanceOf[TableImpl].getRelNode)
+ val name = createUniqueIntermediateRelTableName
+ val intermediateRelTable = createIntermediateRelTable(optimizedPlan, isAccRetract = false)
+ val newTableScan = wrapIntermediateRelTableToTableScan(intermediateRelTable, name)
+ block.setNewOutputNode(newTableScan)
block.setOutputTableName(name)
block.setOptimizedPlan(optimizedPlan)
}
@@ -242,11 +242,9 @@ class StreamCommonSubGraphBasedOptimizer(tEnv: StreamTableEnvironment)
}
}
- private def registerIntermediateTable(
- tEnv: StreamTableEnvironment,
- name: String,
+ private def createIntermediateRelTable(
relNode: RelNode,
- isAccRetract: Boolean): Unit = {
+ isAccRetract: Boolean): IntermediateRelTable = {
val uniqueKeys = getUniqueKeys(relNode)
val monotonicity = FlinkRelMetadataQuery
.reuseOrCreate(tEnv.getRelBuilder.getCluster.getMetadataQuery)
@@ -256,11 +254,7 @@ class StreamCommonSubGraphBasedOptimizer(tEnv: StreamTableEnvironment)
.relModifiedMonotonicity(monotonicity)
.build()
- val table = new IntermediateRelTable(
- relNode,
- isAccRetract,
- statistic)
- tEnv.registerTableInternal(name, table)
+ new IntermediateRelTable(relNode, isAccRetract, statistic)
}
private def getUniqueKeys(relNode: RelNode): util.Set[_ <: util.Set[String]] = {