[flink] branch master updated: [FLINK-28536][table-planner] Adds an internal postOptimize method for physical dag processing

[godfrey]

This is an automated email from the ASF dual-hosted git repository.

godfrey 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 2073bd81d2b [FLINK-28536][table-planner] Adds an internal postOptimize 
method for physical dag processing
2073bd81d2b is described below

commit 2073bd81d2ba5b17b5ab214d69ff022f177fe057
Author: lincoln.lil <lincoln.8...@gmail.com>
AuthorDate: Wed Jul 13 17:46:39 2022 +0800

    [FLINK-28536][table-planner] Adds an internal postOptimize method for 
physical dag processing
    
    This closes #20262
---
 .../table/planner/delegation/PlannerBase.scala      | 12 ++----------
 .../optimize/CommonSubGraphBasedOptimizer.scala     | 21 ++++++++++++++++++++-
 2 files changed, 22 insertions(+), 11 deletions(-)

diff --git 
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala
 
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala
index c5cc1525a4d..1c8c6b7b2ff 100644
--- 
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala
+++ 
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala
@@ -46,8 +46,6 @@ import 
org.apache.flink.table.planner.plan.nodes.exec.processor.{ExecNodeGraphPr
 import org.apache.flink.table.planner.plan.nodes.exec.serde.SerdeContext
 import org.apache.flink.table.planner.plan.nodes.physical.FlinkPhysicalRel
 import org.apache.flink.table.planner.plan.optimize.Optimizer
-import org.apache.flink.table.planner.plan.reuse.SubplanReuser
-import org.apache.flink.table.planner.plan.utils.SameRelObjectShuttle
 import org.apache.flink.table.planner.sinks.DataStreamTableSink
 import 
org.apache.flink.table.planner.sinks.TableSinkUtils.{inferSinkPhysicalSchema, 
validateLogicalPhysicalTypesCompatible, validateTableSink}
 import 
org.apache.flink.table.planner.utils.InternalConfigOptions.{TABLE_QUERY_CURRENT_DATABASE,
 TABLE_QUERY_START_EPOCH_TIME, TABLE_QUERY_START_LOCAL_TIME}
@@ -63,7 +61,6 @@ import org.apache.calcite.plan.{RelTrait, RelTraitDef}
 import org.apache.calcite.rel.RelNode
 import org.apache.calcite.rel.hint.RelHint
 import org.apache.calcite.rel.logical.LogicalTableModify
-import org.apache.calcite.tools.FrameworkConfig
 
 import java.lang.{Long => JLong}
 import java.util
@@ -323,15 +320,10 @@ abstract class PlannerBase(
     }
 
     require(optimizedRelNodes.forall(_.isInstanceOf[FlinkPhysicalRel]))
-    // Rewrite same rel object to different rel objects
-    // in order to get the correct dag (dag reuse is based on object not 
digest)
-    val shuttle = new SameRelObjectShuttle()
-    val relsWithoutSameObj = optimizedRelNodes.map(_.accept(shuttle))
-    // reuse subplan
-    val reusedPlan = SubplanReuser.reuseDuplicatedSubplan(relsWithoutSameObj, 
tableConfig)
+
     // convert FlinkPhysicalRel DAG to ExecNodeGraph
     val generator = new ExecNodeGraphGenerator()
-    val execGraph = 
generator.generate(reusedPlan.map(_.asInstanceOf[FlinkPhysicalRel]))
+    val execGraph = 
generator.generate(optimizedRelNodes.map(_.asInstanceOf[FlinkPhysicalRel]))
 
     // process the graph
     val context = new ProcessorContext(this)
diff --git 
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/CommonSubGraphBasedOptimizer.scala
 
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/CommonSubGraphBasedOptimizer.scala
index c33a4408049..99f29090a64 100644
--- 
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/CommonSubGraphBasedOptimizer.scala
+++ 
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/CommonSubGraphBasedOptimizer.scala
@@ -17,7 +17,10 @@
  */
 package org.apache.flink.table.planner.plan.optimize
 
+import org.apache.flink.table.planner.plan.reuse.SubplanReuser
 import org.apache.flink.table.planner.plan.schema.IntermediateRelTable
+import org.apache.flink.table.planner.plan.utils.SameRelObjectShuttle
+import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
 
 import org.apache.calcite.rel.{RelNode, RelShuttleImpl}
 import org.apache.calcite.rel.core.TableScan
@@ -79,9 +82,25 @@ abstract class CommonSubGraphBasedOptimizer extends 
Optimizer {
         require(plan != null)
         plan
     }
-    expandIntermediateTableScan(optimizedPlan)
+    val expanded = expandIntermediateTableScan(optimizedPlan)
+
+    val postOptimizedPlan = postOptimize(expanded)
+
+    // Rewrite same rel object to different rel objects
+    // in order to get the correct dag (dag reuse is based on object not 
digest)
+    val shuttle = new SameRelObjectShuttle()
+    val relsWithoutSameObj = postOptimizedPlan.map(_.accept(shuttle))
+
+    // reuse subplan
+    SubplanReuser.reuseDuplicatedSubplan(relsWithoutSameObj, 
unwrapTableConfig(roots.head))
   }
 
+  /**
+   * Post process for the physical [[RelNode]] dag, e.g., can be overloaded 
for validation or
+   * rewriting purpose.
+   */
+  protected def postOptimize(expanded: Seq[RelNode]): Seq[RelNode] = expanded
+
   /**
    * Decompose RelNode trees into multiple [[RelNodeBlock]]s, optimize 
recursively each
    * [[RelNodeBlock]], return optimized [[RelNodeBlock]]s.