(flink) branch release-1.18 updated: [FLINK-34115][table-planner] Fix TableAggregateITCase unstable test

[jchan]

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

jchan pushed a commit to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.18 by this push:
     new d903990e1ed [FLINK-34115][table-planner] Fix TableAggregateITCase 
unstable test
d903990e1ed is described below

commit d903990e1ede423e92b6d6ec93876500519aab14
Author: Jane Chan <qingyue....@gmail.com>
AuthorDate: Tue Jan 30 15:56:02 2024 +0800

    [FLINK-34115][table-planner] Fix TableAggregateITCase unstable test
    
    This closes #24222
---
 .../stream/table/TableAggregateITCase.scala        | 28 +++++++++++-----------
 .../table/planner/runtime/utils/TestData.scala     |  8 +++++++
 2 files changed, 22 insertions(+), 14 deletions(-)

diff --git 
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/TableAggregateITCase.scala
 
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/TableAggregateITCase.scala
index d90bdfa9232..cfc828545ab 100644
--- 
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/TableAggregateITCase.scala
+++ 
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/TableAggregateITCase.scala
@@ -21,6 +21,8 @@ import org.apache.flink.api.common.time.Time
 import org.apache.flink.api.scala._
 import org.apache.flink.table.api._
 import org.apache.flink.table.api.bridge.scala._
+import org.apache.flink.table.planner.factories.TestValuesTableFactory
+import 
org.apache.flink.table.planner.runtime.utils.{JavaUserDefinedTableAggFunctions, 
StreamingWithStateTestBase, TestData, TestingRetractSink}
 import 
org.apache.flink.table.planner.runtime.utils.{JavaUserDefinedTableAggFunctions, 
StreamingWithStateTestBase, TestingRetractSink}
 import 
org.apache.flink.table.planner.runtime.utils.JavaUserDefinedAggFunctions.OverloadedDoubleMaxFunction
 import 
org.apache.flink.table.planner.runtime.utils.StreamingWithStateTestBase.StateBackendMode
@@ -37,23 +39,20 @@ import org.junit.runners.Parameterized
 @RunWith(classOf[Parameterized])
 class TableAggregateITCase(mode: StateBackendMode) extends 
StreamingWithStateTestBase(mode) {
 
-  var myTable: Table = _
-
   @Before
   override def before(): Unit = {
     super.before()
     tEnv.getConfig.setIdleStateRetentionTime(Time.hours(1), Time.hours(2))
-    myTable = tEnv.fromValues(
-      DataTypes.ROW(
-        DataTypes.FIELD("id", DataTypes.INT),
-        DataTypes.FIELD("name", DataTypes.STRING),
-        DataTypes.FIELD("price", DataTypes.INT)),
-      row(1, "Latte", 6: java.lang.Integer),
-      row(2, "Milk", 3: java.lang.Integer),
-      row(3, "Breve", 5: java.lang.Integer),
-      row(4, "Mocha", 8: java.lang.Integer),
-      row(5, "Tea", 4: java.lang.Integer)
-    )
+    tEnv.executeSql(s"""
+                       |CREATE TABLE myTable (
+                       |  `id` INT,
+                       |  `name` STRING,
+                       |  `price` INT
+                       |) WITH (
+                       |  'connector' = 'values',
+                       |  'data-id' = 
'${TestValuesTableFactory.registerData(TestData.tupleData4)}'
+                       |)
+                       |""".stripMargin)
   }
 
   @Test
@@ -115,7 +114,8 @@ class TableAggregateITCase(mode: StateBackendMode) extends 
StreamingWithStateTes
 
   def checkRank(func: String, expectedResult: List[String]): Unit = {
     val resultTable =
-      myTable
+      tEnv
+        .from("myTable")
         .flatAggregate(call(func, $("price")).as("top_price", "rank"))
         .select($("top_price"), $("rank"))
 
diff --git 
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/utils/TestData.scala
 
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/utils/TestData.scala
index 2ac34b05e05..0c23c276d16 100644
--- 
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/utils/TestData.scala
+++ 
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/utils/TestData.scala
@@ -283,6 +283,14 @@ object TestData {
 
   lazy val data3: Seq[Row] = tupleData3.map(d => row(d.productIterator.toList: 
_*))
 
+  lazy val tupleData4: Seq[Row] = Seq(
+    row(1, "Latte", 6),
+    row(2, "Milk", 3),
+    row(3, "Breve", 5),
+    row(4, "Mocha", 8),
+    row(5, "Tea", 4)
+  )
+
   val nullablesOfData3 = Array(true, true, true)
 
   val nullablesOfData4 = Array(true, true, true)