You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "dawidwys (via GitHub)" <gi...@apache.org> on 2025/04/02 14:55:48 UTC
[PR] [FLINK-37603] QueryOperation SQL Serialization customization [flink]
dawidwys opened a new pull request, #26389:
URL: https://github.com/apache/flink/pull/26389
## What is the purpose of the change
Provide a way to customize SQL serialization of ResolvedExpressions
## Verifying this change
Added a test in `QueryOperationSqlSerializationTest`
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (**yes** / no)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (**yes** / no)
- If yes, how is the feature documented? (not applicable / docs / **JavaDocs** / not documented)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37603] QueryOperation SQL Serialization customization [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #26389:
URL: https://github.com/apache/flink/pull/26389#issuecomment-2772870602
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ef72e1a156a0ab82dd908aed02d41f9fd7eddeca",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "ef72e1a156a0ab82dd908aed02d41f9fd7eddeca",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ef72e1a156a0ab82dd908aed02d41f9fd7eddeca UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37603] QueryOperation SQL Serialization customization [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #26389:
URL: https://github.com/apache/flink/pull/26389#discussion_r2035131209
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java:
##########
@@ -561,12 +560,12 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL)
BuiltInFunctionDefinition.newBuilder()
.name("ifThenElse")
.callSyntax(
- (sqlName, operands) ->
+ (sqlName, operands, context) ->
Review Comment:
Seems this is actually not resolved? Search for `context` in all files of this PR to find the remaining locations. Mostly around CallSynax.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37603] QueryOperation SQL Serialization customization [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #26389:
URL: https://github.com/apache/flink/pull/26389#discussion_r2032767093
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java:
##########
@@ -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;
Review Comment:
```suggestion
private final @Nullable SqlSerializationContext serializationContext;
```
In the FLIP we agreed on providing a bit more context in the naming.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java:
##########
@@ -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
Review Comment:
```suggestion
* Provides a way to customize the process of serializing Table API to a SQL string. This
```
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/AggregateQueryOperation.java:
##########
@@ -77,23 +78,29 @@ public List<ResolvedExpression> getAggregateExpressions() {
}
@Override
- public String asSerializableString() {
- final String groupingExprs = getGroupingExprs();
+ public String asSerializableString(
+ org.apache.flink.table.operations.SerializationContext context) {
Review Comment:
Find better names that don't collide.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/QueryOperation.java:
##########
@@ -40,10 +41,12 @@ public interface QueryOperation extends Operation {
* Returns a string that fully serializes this instance. The serialized string can be used for
* storing the query in e.g. a {@link org.apache.flink.table.catalog.Catalog} as a view.
*
+ * @param context can be used to customize the serialization to a SQL string
* @return detailed string for persisting in a catalog
* @see Operation#asSummaryString()
+ * @see EnvironmentSettings.Builder#withSerializationContext(SerializationContext)
*/
- default String asSerializableString() {
+ default String asSerializableString(SerializationContext context) {
Review Comment:
This is a public evolving class. We need to keep the old asSerializableString around.
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/ResolvedCatalogTable.java:
##########
@@ -71,8 +72,8 @@ public ResolvedSchema getResolvedSchema() {
* symmetric. The framework will resolve functions and perform other validation tasks. A catalog
* implementation must not deal with this during a read operation.
*/
- public Map<String, String> toProperties() {
- return CatalogPropertiesUtil.serializeCatalogTable(this);
+ public Map<String, String> toProperties(SerializationContext context) {
Review Comment:
This is a public evolving class. We need to keep the old `toProperties` around.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/WindowAggregateQueryOperation.java:
##########
@@ -117,7 +126,10 @@ public String asSerializableString() {
OperationExpressionsUtils
.scopeReferencesWithAlias(
INPUT_ALIAS, expr))
- .map(ResolvedExpression::asSerializableString))
+ .map(
+ resolvedExpression1 ->
Review Comment:
```suggestion
e ->
```
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/DefaultSerializationContext.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.table.operations;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.functions.FunctionDefinition;
+
+/**
+ * Default implementation of {@link SerializationContext} that throws an exception when trying to
+ * serialize an inline function.
+ */
+@Internal
+public class DefaultSerializationContext implements SerializationContext {
+ @Override
+ public String serializeInlineFunction(FunctionDefinition functionDefinition) {
+ throw new TableException(
Review Comment:
```suggestion
throw new ValidationException(
```
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/SerializationContextAdapters.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.table.operations;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.expressions.SerializationContext;
+
+/** Class for adapting serialization contexts between operations and expressions specific ones. */
+@Internal
+public class SerializationContextAdapters {
Review Comment:
Just use one unified context. Let's keep it simple.
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/ResolvedCatalogModel.java:
##########
@@ -48,11 +49,11 @@ public interface ResolvedCatalogModel extends CatalogModel {
* <p>Compared to the pure table options in {@link #getOptions()}, the map includes input
* schema, output schema, comment and options.
*/
- Map<String, String> toProperties();
+ Map<String, String> toProperties(SerializationContext context);
Review Comment:
This is a public evolving class. We need to keep the old toProperties around.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java:
##########
@@ -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;
Review Comment:
But I'm wondering whether we can come up with a better name: How about `SqlFactory`, `SqlBuilder`, `SqlGenerator`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37603] QueryOperation SQL Serialization customization [flink]
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys merged PR #26389:
URL: https://github.com/apache/flink/pull/26389
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37603] QueryOperation SQL Serialization customization [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #26389:
URL: https://github.com/apache/flink/pull/26389#discussion_r2035013674
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/AggregateQueryOperation.java:
##########
@@ -77,23 +78,25 @@ public List<ResolvedExpression> getAggregateExpressions() {
}
@Override
- public String asSerializableString() {
- final String groupingExprs = getGroupingExprs();
+ public String asSerializableString(SqlFactory sqlFactory) {
+ final String groupingExprs = getGroupingExprs(sqlFactory);
return String.format(
"SELECT %s FROM (%s\n) %s\nGROUP BY %s",
Stream.concat(groupingExpressions.stream(), aggregateExpressions.stream())
.map(
expr ->
OperationExpressionsUtils.scopeReferencesWithAlias(
INPUT_ALIAS, expr))
- .map(ResolvedExpression::asSerializableString)
+ .map(
+ resolvedExpression ->
+ resolvedExpression.asSerializableString(sqlFactory))
.collect(Collectors.joining(", ")),
- OperationUtils.indent(child.asSerializableString()),
+ OperationUtils.indent(child.asSerializableString(sqlFactory)),
INPUT_ALIAS,
groupingExprs);
}
- private String getGroupingExprs() {
+ private String getGroupingExprs(SqlFactory context) {
Review Comment:
```suggestion
private String getGroupingExprs(SqlFactory sqlFactory) {
```
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogPropertiesUtil.java:
##########
@@ -72,11 +73,12 @@ public final class CatalogPropertiesUtil {
public static final String FLINK_PROPERTY_PREFIX = "flink.";
/** Serializes the given {@link ResolvedCatalogTable} into a map of string properties. */
- public static Map<String, String> serializeCatalogTable(ResolvedCatalogTable resolvedTable) {
+ public static Map<String, String> serializeCatalogTable(
+ ResolvedCatalogTable resolvedTable, SqlFactory context) {
Review Comment:
```suggestion
ResolvedCatalogTable resolvedTable, SqlFactory sqlFactory) {
```
Maybe perform a `SqlFactory context` search on the code base?
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java:
##########
@@ -561,12 +560,12 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL)
BuiltInFunctionDefinition.newBuilder()
.name("ifThenElse")
.callSyntax(
- (sqlName, operands) ->
+ (sqlName, operands, context) ->
Review Comment:
```suggestion
(sqlName, operands, sqlFactory) ->
```
in entire class
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/QueryOperation.java:
##########
@@ -44,6 +47,19 @@ public interface QueryOperation extends Operation {
* @see Operation#asSummaryString()
*/
default String asSerializableString() {
+ return asSerializableString(new DefaultSqlFactory());
+ }
+
+ /**
+ * Returns a string that fully serializes this instance. The serialized string can be used for
Review Comment:
```suggestion
* Returns a SQL string that fully serializes this instance. The serialized string can be used for
```
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/QueryOperation.java:
##########
@@ -44,6 +47,19 @@ public interface QueryOperation extends Operation {
* @see Operation#asSummaryString()
*/
default String asSerializableString() {
+ return asSerializableString(new DefaultSqlFactory());
+ }
+
+ /**
+ * Returns a string that fully serializes this instance. The serialized string can be used for
+ * storing the query in e.g. a {@link org.apache.flink.table.catalog.Catalog} as a view.
+ *
+ * @param sqlFactory can be used to customize the serialization to a SQL string
+ * @return detailed string for persisting in a catalog
Review Comment:
```suggestion
* @return Flink SQL string for persisting in a catalog
```
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/QueryOperation.java:
##########
@@ -44,6 +47,19 @@ public interface QueryOperation extends Operation {
* @see Operation#asSummaryString()
*/
default String asSerializableString() {
+ return asSerializableString(new DefaultSqlFactory());
Review Comment:
nit: `DefaultSqlFactory.INSTANCE`?
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/WindowAggregateQueryOperation.java:
##########
@@ -262,23 +266,23 @@ public String asSummaryString() {
}
}
- public String asSerializableString(String table) {
+ public String asSerializableString(String table, SqlFactory context) {
Review Comment:
```suggestion
public String asSerializableString(String table, SqlFactory sqlFactory) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37603] QueryOperation SQL Serialization customization [flink]
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #26389:
URL: https://github.com/apache/flink/pull/26389#discussion_r2032821467
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/QueryOperation.java:
##########
@@ -40,10 +41,12 @@ public interface QueryOperation extends Operation {
* Returns a string that fully serializes this instance. The serialized string can be used for
* storing the query in e.g. a {@link org.apache.flink.table.catalog.Catalog} as a view.
*
+ * @param context can be used to customize the serialization to a SQL string
* @return detailed string for persisting in a catalog
* @see Operation#asSummaryString()
+ * @see EnvironmentSettings.Builder#withSerializationContext(SerializationContext)
*/
- default String asSerializableString() {
+ default String asSerializableString(SerializationContext context) {
Review Comment:
Yes, I did it for `ResolvedExpression` but missed it here, sorry.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org