[GitHub] asfgit closed pull request #6857: [hotfix] [table] Refactor SqlToConverter configuration

[GitBox <gi...@apache.org>]

asfgit closed pull request #6857: [hotfix] [table] Refactor SqlToConverter configuration
URL: https://github.com/apache/flink/pull/6857
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
index 7cfe249173c..e5c60ddeb04 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
@@ -91,6 +91,7 @@ abstract class TableEnvironment(val config: TableConfig) {
     .costFactory(new DataSetCostFactory)
     .typeSystem(new FlinkTypeSystem)
     .operatorTable(getSqlOperatorTable)
+    .sqlToRelConverterConfig(getSqlToRelConverterConfig)
     // set the executor to evaluate constant expressions
     .executor(new ExpressionReducer(config))
     .build
@@ -109,15 +110,6 @@ abstract class TableEnvironment(val config: TableConfig) {
   // registered external catalog names -> catalog
   private val externalCatalogs = new mutable.HashMap[String, ExternalCatalog]
 
-  // configuration for SqlToRelConverter
-  private[flink] lazy val sqlToRelConverterConfig: SqlToRelConverter.Config = {
-    val calciteConfig = config.getCalciteConfig
-    calciteConfig.getSqlToRelConverterConfig match {
-      case Some(c) => c
-      case None => getSqlToRelConverterConfig
-    }
-  }
-
   /** Returns the table config to define the runtime behavior of the Table API. */
   def getConfig: TableConfig = config
 
@@ -132,11 +124,18 @@ abstract class TableEnvironment(val config: TableConfig) {
     * Returns the SqlToRelConverter config.
     */
   protected def getSqlToRelConverterConfig: SqlToRelConverter.Config = {
-    SqlToRelConverter.configBuilder()
-      .withTrimUnusedFields(false)
-      .withConvertTableAccess(false)
-      .withInSubQueryThreshold(Integer.MAX_VALUE)
-      .build()
+    val calciteConfig = config.getCalciteConfig
+    calciteConfig.getSqlToRelConverterConfig match {
+
+      case None =>
+        SqlToRelConverter.configBuilder()
+          .withTrimUnusedFields(false)
+          .withConvertTableAccess(false)
+          .withInSubQueryThreshold(Integer.MAX_VALUE)
+          .build()
+
+      case Some(c) => c
+    }
   }
 
   /**
@@ -689,8 +688,7 @@ abstract class TableEnvironment(val config: TableConfig) {
     val planner = new FlinkPlannerImpl(
       getFrameworkConfig,
       getPlanner,
-      getTypeFactory,
-      sqlToRelConverterConfig)
+      getTypeFactory)
     planner.getCompletionHints(statement, position)
   }
 
@@ -712,8 +710,7 @@ abstract class TableEnvironment(val config: TableConfig) {
     * @return The result of the query as Table
     */
   def sqlQuery(query: String): Table = {
-    val planner = new FlinkPlannerImpl(
-      getFrameworkConfig, getPlanner, getTypeFactory, sqlToRelConverterConfig)
+    val planner = new FlinkPlannerImpl(getFrameworkConfig, getPlanner, getTypeFactory)
     // parse the sql query
     val parsed = planner.parse(query)
     if (null != parsed && parsed.getKind.belongsTo(SqlKind.QUERY)) {
@@ -773,8 +770,7 @@ abstract class TableEnvironment(val config: TableConfig) {
     * @param config The [[QueryConfig]] to use.
     */
   def sqlUpdate(stmt: String, config: QueryConfig): Unit = {
-    val planner = new FlinkPlannerImpl(
-      getFrameworkConfig, getPlanner, getTypeFactory, sqlToRelConverterConfig)
+    val planner = new FlinkPlannerImpl(getFrameworkConfig, getPlanner, getTypeFactory)
     // parse the sql query
     val parsed = planner.parse(stmt)
     parsed match {
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkPlannerImpl.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkPlannerImpl.scala
index 2cd0c61a0c3..ffab85933e2 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkPlannerImpl.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkPlannerImpl.scala
@@ -48,8 +48,7 @@ import scala.collection.JavaConversions._
 class FlinkPlannerImpl(
     config: FrameworkConfig,
     planner: RelOptPlanner,
-    typeFactory: FlinkTypeFactory,
-    sqlToRelConverterConfig: SqlToRelConverter.Config) {
+    typeFactory: FlinkTypeFactory) {
 
   val operatorTable: SqlOperatorTable = config.getOperatorTable
   /** Holds the trait definitions to be registered with planner. May be null. */
@@ -57,6 +56,7 @@ class FlinkPlannerImpl(
   val parserConfig: SqlParser.Config = config.getParserConfig
   val convertletTable: SqlRexConvertletTable = config.getConvertletTable
   val defaultSchema: SchemaPlus = config.getDefaultSchema
+  val sqlToRelConverterConfig: SqlToRelConverter.Config = config.getSqlToRelConverterConfig
 
   var validator: FlinkCalciteSqlValidator = _
   var root: RelRoot = _
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/utils/ExpressionTestBase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/utils/ExpressionTestBase.scala
index 8c0e5e3d16c..dd6e00e73bc 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/utils/ExpressionTestBase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/utils/ExpressionTestBase.scala
@@ -66,8 +66,7 @@ abstract class ExpressionTestBase {
   private val planner = new FlinkPlannerImpl(
     context._2.getFrameworkConfig,
     context._2.getPlanner,
-    context._2.getTypeFactory,
-    context._2.sqlToRelConverterConfig)
+    context._2.getTypeFactory)
   private val logicalOptProgram = Programs.ofRules(FlinkRuleSets.LOGICAL_OPT_RULES)
   private val dataSetOptProgram = Programs.ofRules(FlinkRuleSets.DATASET_OPT_RULES)
 


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services