[FLINK-20897][table-planner] Add FlinkContext.isBatchMode flag

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/delegation/PlannerContext.java

@@ -102,6 +102,7 @@ public class PlannerContext {
  private final FrameworkConfig frameworkConfig;
 
  public PlannerContext(
+ boolean isBatchMode,
  TableConfig tableConfig,
  FunctionCatalog functionCatalog,
  CatalogManager catalogManager,
@@ -111,7 +112,11 @@ public PlannerContext(
 
  this.context =
  new FlinkContextImpl(
- tableConfig, functionCatalog, catalogManager, rexConverterFactory);
+ isBatchMode,
+ tableConfig,
+ functionCatalog,
+ catalogManager,
+ rexConverterFactory);
 
  this.rootSchema = rootSchema;
  this.traitDefs = traitDefs;

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/source/SourceAbilityContext.java

@@ -51,6 +51,11 @@ public SourceAbilityContext(FlinkContext context, RowType sourceRowType) {
  this.sourceRowType = sourceRowType;
  }
 
+ @Override
+ public boolean isBatchMode() {
+ return context.isBatchMode();
+ }
+
  @Override
  public TableConfig getTableConfig() {
  return context.getTableConfig();

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkContext.scala

@@ -28,6 +28,11 @@ import org.apache.calcite.plan.Context
  */
 trait FlinkContext extends Context {
 
+ /**
+ * Returns whether the planner runs in batch mode.
+ */
+ def isBatchMode: Boolean
+
  /**
  * Gets [[TableConfig]] instance defined in [[org.apache.flink.table.api.TableEnvironment]].
  */

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkContextImpl.scala

@@ -22,12 +22,15 @@ import org.apache.flink.table.api.TableConfig
 import org.apache.flink.table.catalog.{CatalogManager, FunctionCatalog}
 
 class FlinkContextImpl(
+ inBatchMode: Boolean,
  tableConfig: TableConfig,
  functionCatalog: FunctionCatalog,
  catalogManager: CatalogManager,
  toRexFactory: SqlExprToRexConverterFactory)
  extends FlinkContext {
 
+ override def isBatchMode: Boolean = inBatchMode
+
  override def getTableConfig: TableConfig = tableConfig
 
  override def getFunctionCatalog: FunctionCatalog = functionCatalog

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala

@@ -101,6 +101,7 @@ abstract class PlannerBase(
  @VisibleForTesting
  private[flink] val plannerContext: PlannerContext =
  new PlannerContext(
+ !isStreamingMode,
  config,
  functionCatalog,
  catalogManager,

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/BatchCommonSubGraphBasedOptimizer.scala

@@ -85,6 +85,9 @@ class BatchCommonSubGraphBasedOptimizer(planner: BatchPlanner)
  val context = relNode.getCluster.getPlanner.getContext.unwrap(classOf[FlinkContext])
 
  programs.optimize(relNode, new BatchOptimizeContext {
+
+ override def isBatchMode: Boolean = true
+
  override def getTableConfig: TableConfig = config
 
  override def getFunctionCatalog: FunctionCatalog = planner.functionCatalog

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/StreamCommonSubGraphBasedOptimizer.scala

@@ -162,6 +162,8 @@ class StreamCommonSubGraphBasedOptimizer(planner: StreamPlanner)
 
  programs.optimize(relNode, new StreamOptimizeContext() {
 
+ override def isBatchMode: Boolean = false
+
  override def getTableConfig: TableConfig = config
 
  override def getFunctionCatalog: FunctionCatalog = planner.functionCatalog

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/delegation/ParserImplTest.java

@@ -59,6 +59,7 @@ public class ParserImplTest {
  new FunctionCatalog(tableConfig, catalogManager, moduleManager);
  private final PlannerContext plannerContext =
  new PlannerContext(
+ !isStreamingMode,
  tableConfig,
  functionCatalog,
  catalogManager,

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/expressions/converter/ExpressionConverterTest.java

@@ -62,6 +62,7 @@ public class ExpressionConverterTest {
  private final CatalogManager catalogManager = CatalogManagerMocks.createEmptyCatalogManager();
  private final PlannerContext plannerContext =
  new PlannerContext(
+ false,
  tableConfig,
  new FunctionCatalog(tableConfig, catalogManager, new ModuleManager()),
  catalogManager,

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlToOperationConverterTest.java

@@ -139,6 +139,7 @@ public class SqlToOperationConverterTest {
  catalogManager.getCurrentDatabase());
  private final PlannerContext plannerContext =
  new PlannerContext(
+ false,
  tableConfig,
  functionCatalog,
  catalogManager,

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSinkSpecSerdeTest.java

@@ -69,6 +69,7 @@ public void testDynamicTableSinkSpecSerde() throws IOException {
  SerdeContext serdeCtx =
  new SerdeContext(
  new FlinkContextImpl(
+ false,
  TableConfig.getDefault(),
  null,
  CatalogManagerMocks.createEmptyCatalogManager(),

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSourceSpecSerdeTest.java

@@ -89,6 +89,7 @@ public void testDynamicTableSourceSpecSerde() throws IOException {
  SerdeContext serdeCtx =
  new SerdeContext(
  new FlinkContextImpl(
+ false,
  TableConfig.getDefault(),
  null,
  CatalogManagerMocks.createEmptyCatalogManager(),

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalTypeSerdeTest.java

@@ -95,7 +95,7 @@ public class LogicalTypeSerdeTest {
  public void testLogicalTypeSerde() throws IOException {
  SerdeContext serdeCtx =
  new SerdeContext(
- new FlinkContextImpl(TableConfig.getDefault(), null, null, null),
+ new FlinkContextImpl(false, TableConfig.getDefault(), null, null, null),
  Thread.currentThread().getContextClassLoader(),
  FlinkTypeFactory.INSTANCE(),
  FlinkSqlOperatorTable.instance());

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LogicalWindowSerdeTest.java

@@ -115,7 +115,7 @@ public static List<LogicalWindow> testData() {
  public void testLogicalWindowSerde() throws JsonProcessingException {
  SerdeContext serdeCtx =
  new SerdeContext(
- new FlinkContextImpl(TableConfig.getDefault(), null, null, null),
+ new FlinkContextImpl(false, TableConfig.getDefault(), null, null, null),
  Thread.currentThread().getContextClassLoader(),
  FlinkTypeFactory.INSTANCE(),
  FlinkSqlOperatorTable.instance());

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LookupKeySerdeTest.java

@@ -56,6 +56,7 @@ public void testLookupKey() throws JsonProcessingException {
  .build();
  FlinkContext flinkContext =
  new FlinkContextImpl(
+ false,
  tableConfig,
  new FunctionCatalog(tableConfig, catalogManager, new ModuleManager()),
  catalogManager,

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RelDataTypeSerdeTest.java

@@ -259,7 +259,7 @@ public static Collection<RelDataType> parameters() {
  public void testTypeSerde() throws Exception {
  SerdeContext serdeCtx =
  new SerdeContext(
- new FlinkContextImpl(TableConfig.getDefault(), null, null, null),
+ new FlinkContextImpl(false, TableConfig.getDefault(), null, null, null),
  Thread.currentThread().getContextClassLoader(),
  FlinkTypeFactory.INSTANCE(),
  FlinkSqlOperatorTable.instance());

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RexNodeSerdeTest.java

@@ -86,6 +86,7 @@ public static Object[][] parameters() {
  .build();
  FlinkContext flinkContext =
  new FlinkContextImpl(
+ false,
  tableConfig,
  new FunctionCatalog(tableConfig, catalogManager, new ModuleManager()),
  catalogManager,

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RexWindowBoundSerdeTest.java

@@ -43,7 +43,7 @@ public class RexWindowBoundSerdeTest {
  public void testSerde() throws JsonProcessingException {
  SerdeContext serdeCtx =
  new SerdeContext(
- new FlinkContextImpl(TableConfig.getDefault(), null, null, null),
+ new FlinkContextImpl(false, TableConfig.getDefault(), null, null, null),
  Thread.currentThread().getContextClassLoader(),
  FlinkTypeFactory.INSTANCE(),
  FlinkSqlOperatorTable.instance());

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/TemporalTableSourceSpecSerdeTest.java

@@ -65,6 +65,7 @@ public void testTemporalTableSourceSpecSerde() throws IOException {
  SerdeContext serdeCtx =
  new SerdeContext(
  new FlinkContextImpl(
+ false,
  TableConfig.getDefault(),
  null,
  CatalogManagerMocks.createEmptyCatalogManager(),

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/PlannerMocks.java

@@ -37,19 +37,18 @@
 /** A utility class for creating an instance of {@link FlinkPlannerImpl} for testing. */
 public class PlannerMocks {
  public static FlinkPlannerImpl createDefaultPlanner() {
- final boolean isStreamingMode = false;
  TableConfig tableConfig = new TableConfig();
  CatalogManager catalogManager = CatalogManagerMocks.createEmptyCatalogManager();
  ModuleManager moduleManager = new ModuleManager();
  FunctionCatalog functionCatalog =
  new FunctionCatalog(tableConfig, catalogManager, moduleManager);
  PlannerContext plannerContext =
  new PlannerContext(
+ false,
  tableConfig,
  functionCatalog,
  catalogManager,
- asRootSchema(
- new CatalogManagerCalciteSchema(catalogManager, isStreamingMode)),
+ asRootSchema(new CatalogManagerCalciteSchema(catalogManager, true)),
  new ArrayList<>());
  FlinkPlannerImpl planner =
  plannerContext.createFlinkPlanner(
@@ -61,7 +60,7 @@ public static FlinkPlannerImpl createDefaultPlanner() {
  planner::parser,
  plannerContext.getSqlExprToRexConverterFactory());
  catalogManager.initSchemaResolver(
- isStreamingMode,
+ true,
  ExpressionResolver.resolverFor(
  tableConfig,
  name -> {

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/codegen/WatermarkGeneratorCodeGenTest.scala

@@ -61,6 +61,7 @@ class WatermarkGeneratorCodeGenTest(useDefinedConstructor: Boolean) {
  val catalogManager: CatalogManager = CatalogManagerMocks.createEmptyCatalogManager()
  val functionCatalog = new FunctionCatalog(config, catalogManager, new ModuleManager)
  val plannerContext = new PlannerContext(
+ false,
  config,
  functionCatalog,
  catalogManager,

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/AggCallSelectivityEstimatorTest.scala

@@ -84,6 +84,7 @@ class AggCallSelectivityEstimatorTest {
  rootSchema.add("test", table)
  val plannerContext: PlannerContext =
  new PlannerContext(
+ false,
  tableConfig,
  new FunctionCatalog(tableConfig, catalogManager, new ModuleManager),
  catalogManager,

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala

@@ -88,6 +88,7 @@ class FlinkRelMdHandlerTestBase {
  // and RelOptCluster due to they have different trait definitions.
  val plannerContext: PlannerContext =
  new PlannerContext(
+ false,
  tableConfig,
  new FunctionCatalog(tableConfig, catalogManager, moduleManager),
  catalogManager,

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/SelectivityEstimatorTest.scala

@@ -85,6 +85,7 @@ class SelectivityEstimatorTest {
  rootSchema.add("test", table)
  val plannerContext: PlannerContext =
  new PlannerContext(
+ false,
  tableConfig,
  new FunctionCatalog(tableConfig, catalogManager, new ModuleManager),
  catalogManager,

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/nodes/calcite/RelNodeTestBase.scala

@@ -51,6 +51,7 @@ class RelNodeTestBase {
  val moduleManager = new ModuleManager
 
  val plannerContext: PlannerContext = new PlannerContext(
+ false,
  tableConfig,
  new FunctionCatalog(tableConfig, catalogManager, moduleManager),
  catalogManager,