From 62c81dad40ba7c25d55577dee63c7e6e13918bef Mon Sep 17 00:00:00 2001
From: vasia <vasia@apache.org>
Date: Tue, 12 Apr 2016 15:11:54 +0200
Subject: [PATCH] [FLINK-3735] Make DataSetUnionRule match only for union-all

---
 .../plan/rules/dataSet/DataSetUnionRule.scala | 34 ++++++++++++-------
 1 file changed, 21 insertions(+), 13 deletions(-)

diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetUnionRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetUnionRule.scala
index 32400d0ca2b75..cd1de1e71a83f 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetUnionRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataSet/DataSetUnionRule.scala
@@ -18,7 +18,7 @@
 
 package org.apache.flink.api.table.plan.rules.dataSet
 
-import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.plan.{RelOptRuleCall, Convention, RelOptRule, RelTraitSet}
 import org.apache.calcite.rel.RelNode
 import org.apache.calcite.rel.convert.ConverterRule
 import org.apache.calcite.rel.logical.LogicalUnion
@@ -32,21 +32,29 @@ class DataSetUnionRule
       "FlinkUnionRule")
   {
 
-    def convert(rel: RelNode): RelNode = {
+  /**
+   * Only translate UNION ALL
+   */
+  override def matches(call: RelOptRuleCall): Boolean = {
+    val union: LogicalUnion = call.rel(0).asInstanceOf[LogicalUnion]
+    union.all
+  }
+
+  def convert(rel: RelNode): RelNode = {
 
-      val union: LogicalUnion = rel.asInstanceOf[LogicalUnion]
-      val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
-      val convLeft: RelNode = RelOptRule.convert(union.getInput(0), DataSetConvention.INSTANCE)
-      val convRight: RelNode = RelOptRule.convert(union.getInput(1), DataSetConvention.INSTANCE)
+    val union: LogicalUnion = rel.asInstanceOf[LogicalUnion]
+    val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+    val convLeft: RelNode = RelOptRule.convert(union.getInput(0), DataSetConvention.INSTANCE)
+    val convRight: RelNode = RelOptRule.convert(union.getInput(1), DataSetConvention.INSTANCE)
 
-      new DataSetUnion(
-        rel.getCluster,
-        traitSet,
-        convLeft,
-        convRight,
-        rel.getRowType)
-    }
+    new DataSetUnion(
+      rel.getCluster,
+      traitSet,
+      convLeft,
+      convRight,
+      rel.getRowType)
   }
+}
 
 object DataSetUnionRule {
   val INSTANCE: RelOptRule = new DataSetUnionRule