[FLINK-37603] QueryOperation SQL Serialization customization

Author: dawidwys

flink-python/pyflink/table/tests/test_environment_settings_completeness.py

@@ -38,7 +38,7 @@ def java_class(cls):
     def excluded_methods(cls):
         # internal interfaces, no need to expose to users.
         return {'getPlanner', 'getExecutor', 'getUserClassLoader', 'getCatalogStore',
-                'toConfiguration', 'fromConfiguration'}
+                'toConfiguration', 'fromConfiguration', 'getSerializationContext'}
 
 
 class EnvironmentSettingsBuilderCompletenessTests(PythonAPICompletenessTestCase, PyFlinkTestCase):
@@ -58,7 +58,8 @@ def java_class(cls):
     @classmethod
     def excluded_methods(cls):
         # internal interfaces, no need to expose to users.
-        return {'withClassLoader', 'withCatalogStore'}
+        # withSerializationContext - needs to be implemented
+        return {'withClassLoader', 'withCatalogStore', 'withSerializationContext'}
 
 if __name__ == '__main__':
     import unittest

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java

@@ -24,9 +24,12 @@
 import org.apache.flink.table.api.config.TableConfigOptions;
 import org.apache.flink.table.catalog.CatalogStore;
 import org.apache.flink.table.functions.UserDefinedFunction;
+import org.apache.flink.table.operations.SerializationContext;
 
 import javax.annotation.Nullable;
 
+import java.util.Optional;
+
 import static org.apache.flink.api.common.RuntimeExecutionMode.BATCH;
 import static org.apache.flink.api.common.RuntimeExecutionMode.STREAMING;
 import static org.apache.flink.configuration.ExecutionOptions.RUNTIME_MODE;
@@ -62,16 +65,17 @@ public class EnvironmentSettings {
     private final ClassLoader classLoader;
 
     private final @Nullable CatalogStore catalogStore;
-
-    private EnvironmentSettings(Configuration configuration, ClassLoader classLoader) {
-        this(configuration, classLoader, null);
-    }
+    private final @Nullable SerializationContext serializationContext;
 
     private EnvironmentSettings(
-            Configuration configuration, ClassLoader classLoader, CatalogStore catalogStore) {
+            Configuration configuration,
+            ClassLoader classLoader,
+            CatalogStore catalogStore,
+            SerializationContext serializationContext) {
         this.configuration = configuration;
         this.classLoader = classLoader;
         this.catalogStore = catalogStore;
+        this.serializationContext = serializationContext;
     }
 
     /**
@@ -145,6 +149,11 @@ public CatalogStore getCatalogStore() {
         return catalogStore;
     }
 
+    @Internal
+    public Optional<SerializationContext> getSerializationContext() {
+        return Optional.ofNullable(serializationContext);
+    }
+
     /** A builder for {@link EnvironmentSettings}. */
     @PublicEvolving
     public static class Builder {
@@ -153,6 +162,7 @@ public static class Builder {
         private ClassLoader classLoader;
 
         private @Nullable CatalogStore catalogStore;
+        private @Nullable SerializationContext serializationContext;
 
         public Builder() {}
 
@@ -232,12 +242,22 @@ public Builder withCatalogStore(CatalogStore catalogStore) {
             return this;
         }
 
+        /**
+         * Provides a way to customize the process of serializing operations to an SQL string. This
+         * is useful, for example, for customizing the serialization of inline functions.
+         */
+        public Builder withSerializationContext(SerializationContext serializationContext) {
+            this.serializationContext = serializationContext;
+            return this;
+        }
+
         /** Returns an immutable instance of {@link EnvironmentSettings}. */
         public EnvironmentSettings build() {
             if (classLoader == null) {
                 classLoader = Thread.currentThread().getContextClassLoader();
             }
-            return new EnvironmentSettings(configuration, classLoader, catalogStore);
+            return new EnvironmentSettings(
+                    configuration, classLoader, catalogStore, serializationContext);
         }
     }
 }