[FLINK-3226] implement getUniqueName method in TranslationContext

This closes apache#1600 and apache#1567
robert-schmidtke · Mar 18, 2016 · 670ec6a · 670ec6a
1 parent 7972426
commit 670ec6a
Show file tree

Hide file tree

Showing 6 changed files with 22 additions and 9 deletions.
diff --git a/...raries/flink-table/src/main/scala/org/apache/flink/api/table/plan/RexNodeTranslator.scala b/...raries/flink-table/src/main/scala/org/apache/flink/api/table/plan/RexNodeTranslator.scala
@@ -35,7 +35,7 @@ object RexNodeTranslator {
 
     exp match {
       case agg: Aggregation =>
-        val name = "TMP_" + agg.hashCode().toHexString.toUpperCase
+        val name = TranslationContext.getUniqueName
         val aggCall = toAggCall(agg, name, relBuilder)
         val fieldExp = new UnresolvedFieldReference(name)
         (fieldExp, List(aggCall))

diff --git a/...aries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TranslationContext.scala b/...aries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TranslationContext.scala
@@ -70,6 +70,10 @@ object TranslationContext {
 
   }
 
+  def getUniqueName: String = {
+    "TMP_" + nameCntr.getAndIncrement()
+  }
+
   def getRelBuilder: RelBuilder = {
     relBuilder
   }

diff --git a/...ble/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala b/...ble/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala
@@ -67,7 +67,7 @@ class DataSetGroupReduce(
       config: TableConfig,
       expectedType: Option[TypeInformation[Any]]): DataSet[Any] = {
 
-    val inputDS = input.asInstanceOf[DataSetRel].translateToPlan(config)
+    val inputDS = input.asInstanceOf[DataSetRel].translateToPlan(config, expectedType)
 
     // get the output types
     val fieldsNames = rowType.getFieldNames

diff --git a/...es/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java b/...es/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java
@@ -62,7 +62,6 @@ public AggregationsITCase(TestExecutionMode mode){
 		super(mode);
 	}
 
-	@Ignore //DataSetMap needs to be implemented
 	@Test
 	public void testAggregationTypes() throws Exception {
 		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
@@ -116,8 +115,7 @@ public void testWorkingAggregationDataTypes() throws Exception {
 		compareResultAsText(results, expected);
 	}
 
-	@Ignore // it seems like the arithmetic expression is added to the field position
-	@Test(expected = NotImplementedError.class)
+	@Test
 	public void testAggregationWithArithmetic() throws Exception {
 		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
 		TableEnvironment tableEnv = new TableEnvironment();

diff --git a/...flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AggregationsITCase.scala b/...flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AggregationsITCase.scala
@@ -32,7 +32,6 @@ import scala.collection.JavaConverters._
 @RunWith(classOf[Parameterized])
 class AggregationsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
 
-  @Ignore //DataSetMap needs to be implemented
   @Test
   def testAggregationTypes(): Unit = {
 
@@ -71,8 +70,21 @@ class AggregationsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBa
     TestBaseUtils.compareResultAsText(results.asJava, expected)
   }
 
-  @Ignore // it seems like the arithmetic expression is added to the field position
-  @Test(expected = classOf[NotImplementedError])
+  @Test
+  def testProjection(): Unit = {
+
+    val env = ExecutionEnvironment.getExecutionEnvironment
+    val t = env.fromElements(
+      (1: Byte, 1: Short),
+      (2: Byte, 2: Short)).toTable
+      .select('_1.avg, '_1.sum, '_1.count, '_2.avg, '_2.sum)
+
+    val expected = "1,3,2,1,3"
+    val results = t.toDataSet[Row].collect()
+    TestBaseUtils.compareResultAsText(results.asJava, expected)
+  }
+
+  @Test
   def testAggregationWithArithmetic(): Unit = {
 
     val env = ExecutionEnvironment.getExecutionEnvironment

diff --git a/.../flink-table/src/test/scala/org/apache/flink/api/scala/table/test/ExpressionsITCase.scala b/.../flink-table/src/test/scala/org/apache/flink/api/scala/table/test/ExpressionsITCase.scala
@@ -77,7 +77,6 @@ class ExpressionsITCase(
     TestBaseUtils.compareResultAsText(results.asJava, expected)
   }
 
-  @Ignore
   @Test
   def testCaseInsensitiveForAs(): Unit = {
-Original file line number
+Diff line change
@@ Expand Up / @@ -77,7 +77,6 @@ class ExpressionsITCase( @@
         TestBaseUtils.compareResultAsText(results.asJava, expected)
       }
-      @Ignore
       @Test
       def testCaseInsensitiveForAs(): Unit = {
@@ Expand Down @@