[FLINK-30828][table-planner] Let SortAggITCase#testLeadLag run in ALL…

…_EXCHANGES_PIPELINED mode This closes apache#21793.
meiqiao · Jan 31, 2023 · 9a5c2db · 9a5c2db
1 parent f6694ca
commit 9a5c2db
Showing 1 changed file with 6 additions and 0 deletions.
diff --git a/...test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/AggregateITCaseBase.scala b/...test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/AggregateITCaseBase.scala
@@ -17,10 +17,13 @@
  */
 package org.apache.flink.table.planner.runtime.batch.sql.agg
 
+import org.apache.flink.api.common.BatchShuffleMode
 import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.java.tuple.{Tuple2 => JTuple2}
 import org.apache.flink.api.java.typeutils.RowTypeInfo
 import org.apache.flink.api.scala._
+import org.apache.flink.configuration.{ExecutionOptions, JobManagerOptions}
+import org.apache.flink.configuration.JobManagerOptions.SchedulerType
 import org.apache.flink.table.api.{DataTypes, TableException, Types}
 import org.apache.flink.table.data.DecimalDataUtils
 import org.apache.flink.table.planner.factories.TestValuesTableFactory
@@ -904,6 +907,9 @@ abstract class AggregateITCaseBase(testName: String) extends BatchTestBase {
 
   @Test
   def testLeadLag(): Unit = {
+    tEnv.getConfig.set(JobManagerOptions.SCHEDULER, SchedulerType.Default)
+    tEnv.getConfig
+      .set(ExecutionOptions.BATCH_SHUFFLE_MODE, BatchShuffleMode.ALL_EXCHANGES_PIPELINED)
 
     val testAllDataTypeCardinality = tEnv.fromValues(
       DataTypes.ROW(