[FLINK-20706][table-planner-blink] Introduce BatchPhysicalUnion, and make...

[FLINK-20706][table-planner-blink] Introduce BatchPhysicalUnion, and make BatchExecUnion only extended from ExecNode

This closes #14455

flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecUnion.java

+/*
+ * 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.plan.nodes.exec.batch;
+
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecUnion;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.List;
+
+/**
+ * Batch {@link ExecNode} that is not a physical node and just union the inputs' records into one node.
+ */
+public class BatchExecUnion extends CommonExecUnion implements BatchExecNode<RowData> {
+
+	public BatchExecUnion(
+			List<ExecEdge> inputEdges,
+			RowType outputType,
+			String description) {
+		super(inputEdges, outputType, description);
+	}
+}

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchExecUnion.scala → flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalUnion.scala

@@ -18,13 +18,10 @@
 
 package org.apache.flink.table.planner.plan.nodes.physical.batch
 
-import org.apache.flink.api.dag.Transformation
-import org.apache.flink.streaming.api.transformations.UnionTransformation
-import org.apache.flink.table.data.RowData
-import org.apache.flink.table.planner.delegation.BatchPlanner
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory
 import org.apache.flink.table.planner.plan.`trait`.{FlinkRelDistribution, FlinkRelDistributionTraitDef}
-import org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecUnion
-import org.apache.flink.table.planner.plan.nodes.exec.{ExecEdge, LegacyBatchExecNode}
+import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecUnion
+import org.apache.flink.table.planner.plan.nodes.exec.{ExecEdge, ExecNode}
 
 import org.apache.calcite.plan.{RelOptCluster, RelOptRule, RelTraitSet}
 import org.apache.calcite.rel.RelDistribution.Type.{ANY, BROADCAST_DISTRIBUTED, HASH_DISTRIBUTED, RANDOM_DISTRIBUTED, RANGE_DISTRIBUTED, ROUND_ROBIN_DISTRIBUTED, SINGLETON}
@@ -39,23 +36,21 @@ import scala.collection.JavaConversions._
 /**
   * Batch physical RelNode for [[Union]].
   */
-class BatchExecUnion(
+class BatchPhysicalUnion(
     cluster: RelOptCluster,
     traitSet: RelTraitSet,
     inputRels: util.List[RelNode],
     all: Boolean,
     outputRowType: RelDataType)
   extends Union(cluster, traitSet, inputRels, all)
-  with BatchPhysicalRel
-  with LegacyBatchExecNode[RowData]
-  with CommonExecUnion {
+  with BatchPhysicalRel {
 
   require(all, "Only support union all now")
 
   override def deriveRowType(): RelDataType = outputRowType
 
   override def copy(traitSet: RelTraitSet, inputs: util.List[RelNode], all: Boolean): SetOp = {
-    new BatchExecUnion(cluster, traitSet, inputs, all, outputRowType)
+    new BatchPhysicalUnion(cluster, traitSet, inputs, all, outputRowType)
   }
 
   override def explainTerms(pw: RelWriter): RelWriter = {
@@ -98,16 +93,11 @@ class BatchExecUnion(
     Some(copy(providedTraitSet, newInputs))
   }
 
-  //~ ExecNode methods -----------------------------------------------------------
-
-  override def getInputEdges: util.List[ExecEdge] =
-    inputRels.map(_ => ExecEdge.DEFAULT)
-
-  override protected def translateToPlanInternal(
-      planner: BatchPlanner): Transformation[RowData] = {
-    val transformations = getInputNodes.map {
-      input => input.translateToPlan(planner).asInstanceOf[Transformation[RowData]]
-    }
-    new UnionTransformation(transformations)
+  override def translateToExecNode(): ExecNode[_] = {
+    new BatchExecUnion(
+      getInputs.map(_ => ExecEdge.DEFAULT),
+      FlinkTypeFactory.toLogicalRowType(getRowType),
+      getRelDetailedDescription
+    )
   }
 }

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala

@@ -402,7 +402,7 @@ object FlinkBatchRuleSets {
     BatchPhysicalCalcRule.INSTANCE,
     BatchPhysicalPythonCalcRule.INSTANCE,
     // union
-    BatchExecUnionRule.INSTANCE,
+    BatchPhysicalUnionRule.INSTANCE,
     // sort
     BatchExecSortRule.INSTANCE,
     BatchExecLimitRule.INSTANCE,

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchExecUnionRule.scala → flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalUnionRule.scala

@@ -20,7 +20,7 @@ package org.apache.flink.table.planner.plan.rules.physical.batch
 
 import org.apache.flink.table.planner.plan.nodes.FlinkConventions
 import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalUnion
-import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecUnion
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalUnion
 
 import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
 import org.apache.calcite.rel.RelNode
@@ -29,14 +29,14 @@ import org.apache.calcite.rel.convert.ConverterRule
 import scala.collection.JavaConversions._
 
 /**
-  * Rule that converts [[FlinkLogicalUnion]] to [[BatchExecUnion]].
+  * Rule that converts [[FlinkLogicalUnion]] to [[BatchPhysicalUnion]].
   */
-class BatchExecUnionRule
+class BatchPhysicalUnionRule
   extends ConverterRule(
     classOf[FlinkLogicalUnion],
     FlinkConventions.LOGICAL,
     FlinkConventions.BATCH_PHYSICAL,
-    "BatchExecUnionRule") {
+    "BatchPhysicalUnionRule") {
 
   override def matches(call: RelOptRuleCall): Boolean = {
     call.rel(0).asInstanceOf[FlinkLogicalUnion].all
@@ -47,7 +47,7 @@ class BatchExecUnionRule
     val traitSet = rel.getTraitSet.replace(FlinkConventions.BATCH_PHYSICAL)
     val newInputs = union.getInputs.map(RelOptRule.convert(_, FlinkConventions.BATCH_PHYSICAL))
 
-    new BatchExecUnion(
+    new BatchPhysicalUnion(
       rel.getCluster,
       traitSet,
       newInputs,
@@ -56,6 +56,6 @@ class BatchExecUnionRule
   }
 }
 
-object BatchExecUnionRule {
-  val INSTANCE: RelOptRule = new BatchExecUnionRule
+object BatchPhysicalUnionRule {
+  val INSTANCE: RelOptRule = new BatchPhysicalUnionRule
 }