[FLINK-37603] QueryOperation SQL Serialization customization

Author: dawidwys

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> getOperationSerializationContext() {
+        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,23 @@ 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 withOperationSerializationContext(
+                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);
         }
     }
 }

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

@@ -23,6 +23,7 @@
 import org.apache.flink.table.catalog.ResolvedSchema;
 import org.apache.flink.table.connector.sink.DynamicTableSink;
 import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.expressions.DefaultSerializationContext;
 import org.apache.flink.table.expressions.Expression;
 import org.apache.flink.table.functions.ProcessTableFunction;
 import org.apache.flink.table.functions.TableFunction;
@@ -102,7 +103,8 @@ public interface Table extends Explainable<Table>, Executable {
      */
     @Deprecated
     default TableSchema getSchema() {
-        return TableSchema.fromResolvedSchema(getResolvedSchema());
+        return TableSchema.fromResolvedSchema(
+                getResolvedSchema(), new DefaultSerializationContext());
     }
 
     /** Returns the resolved schema of this table. */