[FLINK-17339][table-planner-blink] Simplify test cases due to default…

… planner changing.

flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/api/EnvironmentTest.java

@@ -40,9 +40,7 @@ public class EnvironmentTest {
  @Test
  public void testPassingExecutionParameters() {
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- StreamTableEnvironment tEnv = StreamTableEnvironment.create(
- env,
- EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build());
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
 
  tEnv.getConfig().addConfiguration(
  new Configuration()

flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/api/TableUtilsStreamingITCase.java

@@ -48,7 +48,7 @@ public void before() {
  env.setParallelism(4);
  env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
 
- EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
+ EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
  tEnv = StreamTableEnvironment.create(env, settings);
  }
 

flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/catalog/CatalogConstraintTest.java

@@ -57,7 +57,7 @@ public class CatalogConstraintTest {
 
  @Before
  public void setup() {
- EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
+ EnvironmentSettings settings = EnvironmentSettings.newInstance().inBatchMode().build();
  tEnv = TableEnvironment.create(settings);
  catalog = tEnv.getCatalog(tEnv.getCurrentCatalog()).orElse(null);
  assertNotNull(catalog);

flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/catalog/CatalogITCase.java

@@ -47,7 +47,7 @@ public void testCreateCatalog() {
  }
 
  private TableEnvironment getTableEnvironment() {
- EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
+ EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  return StreamTableEnvironment.create(env, settings);
  }

flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/catalog/CatalogStatisticsTest.java

@@ -82,7 +82,7 @@ public class CatalogStatisticsTest {
 
  @Before
  public void setup() {
- EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
+ EnvironmentSettings settings = EnvironmentSettings.newInstance().inBatchMode().build();
  tEnv = TableEnvironment.create(settings);
  catalog = tEnv.getCatalog(tEnv.getCurrentCatalog()).orElse(null);
  assertNotNull(catalog);

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/api/TableEnvironmentITCase.scala

@@ -57,9 +57,9 @@ class TableEnvironmentITCase(tableEnvName: String, isStreaming: Boolean) extends
  var tEnv: TableEnvironment = _
 
  private val settings = if (isStreaming) {
- EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
+ EnvironmentSettings.newInstance().inStreamingMode().build()
  } else {
- EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
+ EnvironmentSettings.newInstance().inBatchMode().build()
  }
 
  @Before

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/api/TableEnvironmentTest.scala

@@ -91,7 +91,7 @@ class TableEnvironmentTest {
  @Test
  def testStreamTableEnvironmentExplain(): Unit = {
  val execEnv = StreamExecutionEnvironment.getExecutionEnvironment
- val settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
+ val settings = EnvironmentSettings.newInstance().inStreamingMode().build()
  val tEnv = StreamTableEnvironment.create(execEnv, settings)
 
  tEnv.registerTableSource("MyTable", TestTableSourceSinks.getPersonCsvTableSource)

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/catalog/CatalogTableITCase.scala

@@ -47,9 +47,9 @@ class CatalogTableITCase(isStreamingMode: Boolean) extends AbstractTestBase {
  //~ Instance fields --------------------------------------------------------
 
  private val settings = if (isStreamingMode) {
- EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
+ EnvironmentSettings.newInstance().inStreamingMode().build()
  } else {
- EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
+ EnvironmentSettings.newInstance().inBatchMode().build()
  }
 
  private val tableEnv: TableEnvironment = TableEnvironmentImpl.create(settings)

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/catalog/CatalogTableTest.scala

@@ -29,7 +29,7 @@ import org.junit.Assert.assertEquals
 class CatalogTableTest {
 
  val tEnv: TableEnvironment = TableEnvironmentImpl.create(
- EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build())
+ EnvironmentSettings.newInstance().inStreamingMode().build())
 
  @Test
  def testDDLSchema(): Unit = {

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/catalog/CatalogViewITCase.scala

@@ -40,9 +40,9 @@ class CatalogViewITCase(isStreamingMode: Boolean) extends AbstractTestBase {
  //~ Instance fields --------------------------------------------------------
 
  private val settings = if (isStreamingMode) {
- EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
+ EnvironmentSettings.newInstance().inStreamingMode().build()
  } else {
- EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
+ EnvironmentSettings.newInstance().inBatchMode().build()
  }
 
  private val tableEnv: TableEnvironment = TableEnvironmentImpl.create(settings)

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/codegen/agg/AggTestBase.scala

@@ -45,12 +45,12 @@ abstract class AggTestBase(isBatchMode: Boolean) {
  val typeFactory: FlinkTypeFactory = new FlinkTypeFactory(new FlinkTypeSystem())
  val env = new ScalaStreamExecEnv(new LocalStreamEnvironment)
  private val tEnv = if (isBatchMode) {
- val settings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
+ val settings = EnvironmentSettings.newInstance().inBatchMode().build()
  // use impl class instead of interface class to avoid
  // "Static methods in interface require -target:jvm-1.8"
  TableEnvironmentImpl.create(settings)
  } else {
- val settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
+ val settings = EnvironmentSettings.newInstance().inStreamingMode().build()
  StreamTableEnvironment.create(env, settings)
  }
  private val planner = tEnv.asInstanceOf[TableEnvironmentImpl].getPlanner.asInstanceOf[PlannerBase]

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/expressions/utils/ExpressionTestBase.scala

@@ -61,8 +61,7 @@ abstract class ExpressionTestBase {
  // (originalExpr, optimizedExpr, expectedResult)
  private val testExprs = mutable.ArrayBuffer[(String, RexNode, String)]()
  private val env = StreamExecutionEnvironment.createLocalEnvironment(4)
- private val setting = EnvironmentSettings.newInstance()
- .useBlinkPlanner().inStreamingMode().build()
+ private val setting = EnvironmentSettings.newInstance().inStreamingMode().build()
  // use impl class instead of interface class to avoid
  // "Static methods in interface require -target:jvm-1.8"
  private val tEnv = StreamTableEnvironmentImpl.create(env, setting, config)

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/batch/table/validation/JoinValidationTest.scala

@@ -96,7 +96,7 @@ class JoinValidationTest extends TableTestBase {
 
  @Test(expected = classOf[ValidationException])
  def testJoinTablesFromDifferentEnvs(): Unit = {
- val settings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
+ val settings = EnvironmentSettings.newInstance().inBatchMode().build()
  val tEnv1 = TableEnvironmentImpl.create(settings)
  val tEnv2 = TableEnvironmentImpl.create(settings)
  val ds1 = CollectionBatchExecTable.getSmall3TupleDataSet(tEnv1, "a, b, c")
@@ -108,7 +108,7 @@ class JoinValidationTest extends TableTestBase {
 
  @Test(expected = classOf[ValidationException])
  def testJoinTablesFromDifferentEnvsJava() {
- val settings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
+ val settings = EnvironmentSettings.newInstance().inBatchMode().build()
  val tEnv1 = TableEnvironmentImpl.create(settings)
  val tEnv2 = TableEnvironmentImpl.create(settings)
  val ds1 = CollectionBatchExecTable.getSmall3TupleDataSet(tEnv1, "a, b, c")

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/batch/table/validation/SetOperatorsValidationTest.scala

@@ -52,7 +52,7 @@ class SetOperatorsValidationTest extends TableTestBase {
 
  @Test(expected = classOf[ValidationException])
  def testUnionTablesFromDifferentEnvs(): Unit = {
- val settings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
+ val settings = EnvironmentSettings.newInstance().inBatchMode().build()
  val tEnv1 = TableEnvironmentImpl.create(settings)
  val tEnv2 = TableEnvironmentImpl.create(settings)
 
@@ -76,7 +76,7 @@ class SetOperatorsValidationTest extends TableTestBase {
 
  @Test(expected = classOf[ValidationException])
  def testMinusAllTablesFromDifferentEnvs(): Unit = {
- val settings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
+ val settings = EnvironmentSettings.newInstance().inBatchMode().build()
  val tEnv1 = TableEnvironmentImpl.create(settings)
  val tEnv2 = TableEnvironmentImpl.create(settings)
 
@@ -100,7 +100,7 @@ class SetOperatorsValidationTest extends TableTestBase {
 
  @Test(expected = classOf[ValidationException])
  def testIntersectTablesFromDifferentEnvs(): Unit = {
- val settings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
+ val settings = EnvironmentSettings.newInstance().inBatchMode().build()
  val tEnv1 = TableEnvironmentImpl.create(settings)
  val tEnv2 = TableEnvironmentImpl.create(settings)
 

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/utils/FlinkRelOptUtilTest.scala

@@ -41,7 +41,7 @@ class FlinkRelOptUtilTest {
 
  @Before
  def before(): Unit = {
- val settings = EnvironmentSettings.newInstance().useBlinkPlanner().build()
+ val settings = EnvironmentSettings.newInstance().build()
  val tEnv = TableEnvironmentImpl.create(settings)
  BatchTableEnvUtil.registerCollection(
  tEnv,

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/harness/GroupAggregateHarnessTest.scala

@@ -43,7 +43,7 @@ class GroupAggregateHarnessTest(mode: StateBackendMode) extends HarnessTestBase(
  @Before
  override def before(): Unit = {
  super.before()
- val setting = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
+ val setting = EnvironmentSettings.newInstance().inStreamingMode().build()
  val config = new TestTableConfig
  this.tEnv = StreamTableEnvironmentImpl.create(env, setting, config)
  }

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/harness/OverWindowHarnessTest.scala

@@ -45,7 +45,7 @@ class OverWindowHarnessTest(mode: StateBackendMode) extends HarnessTestBase(mode
  @Before
  override def before(): Unit = {
  super.before()
- val setting = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
+ val setting = EnvironmentSettings.newInstance().inStreamingMode().build()
  val config = new TestTableConfig
  this.tEnv = StreamTableEnvironmentImpl.create(env, setting, config)
  }

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/harness/TableAggregateHarnessTest.scala

@@ -43,7 +43,7 @@ class TableAggregateHarnessTest(mode: StateBackendMode) extends HarnessTestBase(
  @Before
  override def before(): Unit = {
  super.before()
- val setting = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
+ val setting = EnvironmentSettings.newInstance().inStreamingMode().build()
  val config = new TestTableConfig
  this.tEnv = StreamTableEnvironmentImpl.create(env, setting, config)
  }

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/utils/BatchTestBase.scala

@@ -56,7 +56,7 @@ import scala.util.Sorting
 
 class BatchTestBase extends BatchAbstractTestBase {
 
- private val settings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
+ private val settings = EnvironmentSettings.newInstance().inBatchMode().build()
  private val testingTableEnv: TestingTableEnvironment = TestingTableEnvironment
  .create(settings, catalogManager = None, new TableConfig)
  val tEnv: TableEnvironment = testingTableEnv

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/utils/StreamingTestBase.scala

@@ -51,7 +51,7 @@ class StreamingTestBase extends AbstractTestBase {
  if (enableObjectReuse) {
  this.env.getConfig.enableObjectReuse()
  }
- val setting = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
+ val setting = EnvironmentSettings.newInstance().inStreamingMode().build()
  this.tEnv = StreamTableEnvironment.create(env, setting)
  }
 }

flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala

@@ -127,9 +127,9 @@ abstract class TableTestUtilBase(test: TableTestBase, isStreamingMode: Boolean)
  protected lazy val diffRepository: DiffRepository = DiffRepository.lookup(test.getClass)
 
  protected val setting: EnvironmentSettings = if (isStreamingMode) {
- EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
+ EnvironmentSettings.newInstance().inStreamingMode().build()
  } else {
- EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
+ EnvironmentSettings.newInstance().inBatchMode().build()
  }
 
  // a counter for unique table names
@@ -1117,10 +1117,9 @@ object TestingTableEnvironment {
 
 object TableTestUtil {
 
- val STREAM_SETTING: EnvironmentSettings = EnvironmentSettings.newInstance()
- .useBlinkPlanner().inStreamingMode().build()
- val BATCH_SETTING: EnvironmentSettings = EnvironmentSettings.newInstance()
- .useBlinkPlanner().inBatchMode().build()
+ val STREAM_SETTING: EnvironmentSettings =
+ EnvironmentSettings.newInstance().inStreamingMode().build()
+ val BATCH_SETTING: EnvironmentSettings = EnvironmentSettings.newInstance().inBatchMode().build()
 
  /**
  * Converts operation tree in the given table to a RelNode tree.