[FLINK-3226] implement getUniqueName method in TranslationContext

This closes #1600 and #1567
apache · Feb 11, 2016 · fff25df · fff25df
1 parent 509c4b9
commit fff25df
Show file tree

Hide file tree

Showing 5 changed files with 20 additions and 4 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();

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,6 +70,20 @@ class AggregationsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBa
     TestBaseUtils.compareResultAsText(results.asJava, expected)
   }
 
+  @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)
+  }
+
   @Ignore // it seems like the arithmetic expression is added to the field position
   @Test(expected = classOf[NotImplementedError])
   def testAggregationWithArithmetic(): Unit = {