use openContext name

flink-core/src/main/java/org/apache/flink/api/common/functions/WithConfigurationOpenContext.java

@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.common.functions;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.Configuration;
+
+/** A special {@link OpenContext} for passing configuration to udf. */
+@PublicEvolving
+public class WithConfigurationOpenContext implements OpenContext {
+    private final Configuration configuration;
+
+    public WithConfigurationOpenContext(Configuration configuration) {
+        this.configuration = configuration;
+    }
+
+    public Configuration getConfiguration() {
+        return configuration;
+    }
+}

flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/FunctionContext.java

@@ -20,7 +20,9 @@
 
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.externalresource.ExternalResourceInfo;
+import org.apache.flink.api.common.functions.OpenContext;
 import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.functions.WithConfigurationOpenContext;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.metrics.MetricGroup;
 import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
@@ -62,10 +64,16 @@ public class FunctionContext {
     public FunctionContext(
             @Nullable RuntimeContext context,
             @Nullable ClassLoader userClassLoader,
-            @Nullable Configuration jobParameters) {
+            @Nullable OpenContext openContext) {
         this.context = context;
         this.userClassLoader = userClassLoader;
-        this.jobParameters = jobParameters != null ? jobParameters.toMap() : null;
+        if (openContext instanceof WithConfigurationOpenContext) {
+            Configuration configuration =
+                    ((WithConfigurationOpenContext) openContext).getConfiguration();
+            this.jobParameters = configuration.toMap();
+        } else {
+            this.jobParameters = null;
+        }
     }
 
     public FunctionContext(RuntimeContext context) {

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CollectorCodeGenerator.scala

@@ -75,7 +75,7 @@ object CollectorCodeGenerator {
         }
 
         @Override
-        public void open(${className[OpenContext]} context) throws Exception {
+        public void open(${className[OpenContext]} openContext) throws Exception {
           ${ctx.reuseOpenCode()}
         }
 
@@ -143,7 +143,7 @@ object CollectorCodeGenerator {
         }
 
         @Override
-        public void open(${className[OpenContext]} context) throws Exception {
+        public void open(${className[OpenContext]} openContext) throws Exception {
           ${ctx.reuseOpenCode()}
         }
 

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala

@@ -17,7 +17,7 @@
  */
 package org.apache.flink.table.planner.codegen
 
-import org.apache.flink.api.common.functions.{DefaultOpenContext, MapFunction, OpenContext, RichMapFunction}
+import org.apache.flink.api.common.functions.{DefaultOpenContext, MapFunction, OpenContext, RichMapFunction, WithConfigurationOpenContext}
 import org.apache.flink.configuration.{Configuration, PipelineOptions, ReadableConfig}
 import org.apache.flink.table.api.{TableConfig, TableException}
 import org.apache.flink.table.data.{DecimalData, GenericRowData, TimestampData}
@@ -102,9 +102,12 @@ class ExpressionReducer(
         throw new TableException("RichMapFunction[GenericRowData, GenericRowData] required here")
     }
 
+    val parameters = toScala(tableConfig.getOptional(PipelineOptions.GLOBAL_JOB_PARAMETERS))
+      .map(Configuration.fromMap)
+      .getOrElse(new Configuration)
     val reduced =
       try {
-        richMapFunction.open(DefaultOpenContext.INSTANCE)
+        richMapFunction.open(new WithConfigurationOpenContext(parameters))
         // execute
         richMapFunction.map(EMPTY_ROW)
       } catch {
@@ -312,7 +315,7 @@ class ConstantCodeGeneratorContext(tableConfig: ReadableConfig, classLoader: Cla
     super.addReusableFunction(
       function,
       classOf[FunctionContext],
-      Seq("null", "this.getClass().getClassLoader()", "context"))
+      Seq("null", "this.getClass().getClassLoader()", "openContext"))
   }
 
   override def addReusableConverter(dataType: DataType, classLoaderTerm: String = null): String = {

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/FunctionCodeGenerator.scala

@@ -154,7 +154,7 @@ object FunctionCodeGenerator {
         ${ctx.reuseConstructorCode(funcName)}
 
         @Override
-        public void open(${classOf[OpenContext].getCanonicalName} context) throws Exception {
+        public void open(${classOf[OpenContext].getCanonicalName} openContext) throws Exception {
           ${ctx.reuseOpenCode()}
         }
 
@@ -230,7 +230,7 @@ object FunctionCodeGenerator {
         ${ctx.reuseConstructorCode(funcName)}
 
         @Override
-        public void open(${className[OpenContext]} context) throws Exception {
+        public void open(${className[OpenContext]} openContext) throws Exception {
           ${ctx.reuseOpenCode()}
         }
 

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/LookupJoinCodeGenerator.scala

@@ -381,7 +381,7 @@ object LookupJoinCodeGenerator {
         }
 
         @Override
-        public void open(${className[OpenContext]} context) throws Exception {
+        public void open(${className[OpenContext]} openContext) throws Exception {
           ${ctx.reuseOpenCode()}
         }
 
@@ -492,7 +492,7 @@ object LookupJoinCodeGenerator {
         }
 
         @Override
-        public void open(${className[OpenContext]} context) throws Exception {
+        public void open(${className[OpenContext]} openContext) throws Exception {
           ${ctx.reuseOpenCode()}
         }
 

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/WatermarkGeneratorCodeGenerator.scala

@@ -88,7 +88,7 @@ object WatermarkGeneratorCodeGenerator {
         }
 
         @Override
-        public void open(${classOf[OpenContext].getCanonicalName} context) throws Exception {
+        public void open(${classOf[OpenContext].getCanonicalName} openContext) throws Exception {
           ${ctx.reuseOpenCode()}
         }
 

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingFunctionGenUtil.scala

@@ -630,7 +630,7 @@ object BridgingFunctionGenUtil {
            |    ${ctx.reuseInitCode()}
            |  }
            |
-           |  public void open(${className[OpenContext]} context) throws Exception {
+           |  public void open(${className[OpenContext]} openContext) throws Exception {
            |    ${ctx.reuseOpenCode()}
            |  }
            |