You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "twalthr (via GitHub)" <gi...@apache.org> on 2023/09/06 07:49:57 UTC
[GitHub] [flink] twalthr opened a new pull request, #23365: [FLINK-33028][table] Make expanding behavior of virtual metadata columns configurable
twalthr opened a new pull request, #23365:
URL: https://github.com/apache/flink/pull/23365
## What is the purpose of the change
This introduces a new `table.column-expansion-strategy` option according to FLIP-348. It updates all code locations in SQL and Table API where column expansion takes place. In other words: where `*` is a valid column selector.
For SQL this includes `SELECT *`. For Table API, includes `$(*)`, `withColumns` and `withoutColumns`.
## Brief change log
Updated the resolution logic in SQL validator and Table API resolution rules.
## Verifying this change
This change added tests and can be verified as follows: `ColumnExpansionTest`
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: yes
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? docs
--
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
[GitHub] [flink] twalthr commented on a diff in pull request #23365: [FLINK-33028][table] Make expanding behavior of virtual metadata columns configurable
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #23365:
URL: https://github.com/apache/flink/pull/23365#discussion_r1316945655
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/TableConfigOptions.java:
##########
@@ -313,4 +325,32 @@ public InlineElement getDescription() {
return description;
}
}
+
+ /** Strategy to expand columns in {@code SELECT *} queries. */
+ @PublicEvolving
+ public enum ColumnExpansionStrategy implements DescribedEnum {
+ EXCLUDE_ALIASED_VIRTUAL_METADATA_COLUMNS(
+ text(
+ "Excludes virtual metadata columns that reference a metadata key via an alias. "
+ + "For example, a column declared as 'c METADATA VIRTUAL FROM k' "
+ + "is not selected by default if the strategy is applied.")),
+
+ EXCLUDE_DEFAULT_VIRTUAL_METADATA_COLUMNS(
+ text(
+ "Excludes virtual metadata columns that directly reference a metadata key. "
+ + "For example, a column declared as 'k METADATA VIRTUAL' "
+ + "is not selected by default if the strategy is applied."));
Review Comment:
This is to provide more flexibility. E.g. if a metadata virtual column should serve as a kind of system column, it might make sense to only set `EXCLUDE_DEFAULT_VIRTUAL_METADATA_COLUMNS` and give users the chance to define virtual metadata columns that are not excluded using the `METADATA FROM` clause. Anyway I wanted to give some options already. I was also thinking about supporting exclusions of computed columns in the first version, but this was actually not part of the FLIP.
--
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
[GitHub] [flink] snuyanzin commented on pull request #23365: [FLINK-33028][table] Make expanding behavior of virtual metadata columns configurable
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #23365:
URL: https://github.com/apache/flink/pull/23365#issuecomment-1708239255
Thanks for explanation, I also think it is not related to the PR
just wanted to note this
--
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
[GitHub] [flink] twalthr merged pull request #23365: [FLINK-33028][table] Make expanding behavior of virtual metadata columns configurable
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr merged PR #23365:
URL: https://github.com/apache/flink/pull/23365
--
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
[GitHub] [flink] snuyanzin commented on pull request #23365: [FLINK-33028][table] Make expanding behavior of virtual metadata columns configurable
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #23365:
URL: https://github.com/apache/flink/pull/23365#issuecomment-1708113247
One inconsistency I noticed between table and sql (probably minor):
e.g. there is a table with columns `a`, `b`, `c`
```sql
SELECT *, * FROM my_table;
```
results to
```
a b c a0 b0 c0
------- -------- -------- --------- --------- --------
```
in case of table api
```java
...
tableEnv.from("my_table").select(withAllColumns(), withAllColumns())
...
```
it just fails like
```
org.apache.flink.table.api.ValidationException: Ambiguous column name: a
at org.apache.flink.table.operations.utils.ProjectionOperationFactory.lambda$validateAndGetUniqueNames$4(ProjectionOperationFactory.java:109)
at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:485)
at org.apache.flink.table.operations.utils.ProjectionOperationFactory.validateAndGetUniqueNames(ProjectionOperationFactory.java:106)
at org.apache.flink.table.operations.utils.ProjectionOperationFactory.create(ProjectionOperationFactory.java:74)
at org.apache.flink.table.operations.utils.OperationTreeBuilder.projectInternal(OperationTreeBuilder.java:200)
at org.apache.flink.table.operations.utils.OperationTreeBuilder.project(OperationTreeBuilder.java:174)
```
--
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
[GitHub] [flink] twalthr commented on pull request #23365: [FLINK-33028][table] Make expanding behavior of virtual metadata columns configurable
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on PR #23365:
URL: https://github.com/apache/flink/pull/23365#issuecomment-1708121466
Thanks for taking a look @snuyanzin. There are more inconsistencies in Table API but those are not related to this PR. Also selecting columns from a join will create the ambiguous column exception.
--
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
[GitHub] [flink] flinkbot commented on pull request #23365: [FLINK-33028][table] Make expanding behavior of virtual metadata columns configurable
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #23365:
URL: https://github.com/apache/flink/pull/23365#issuecomment-1707854870
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "775a2922b557a12416c41a3d5ec5b8fd0f5c4a55",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "775a2922b557a12416c41a3d5ec5b8fd0f5c4a55",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 775a2922b557a12416c41a3d5ec5b8fd0f5c4a55 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
[GitHub] [flink] dawidwys commented on a diff in pull request #23365: [FLINK-33028][table] Make expanding behavior of virtual metadata columns configurable
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #23365:
URL: https://github.com/apache/flink/pull/23365#discussion_r1316930548
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/TableConfigOptions.java:
##########
@@ -313,4 +325,32 @@ public InlineElement getDescription() {
return description;
}
}
+
+ /** Strategy to expand columns in {@code SELECT *} queries. */
+ @PublicEvolving
+ public enum ColumnExpansionStrategy implements DescribedEnum {
+ EXCLUDE_ALIASED_VIRTUAL_METADATA_COLUMNS(
+ text(
+ "Excludes virtual metadata columns that reference a metadata key via an alias. "
+ + "For example, a column declared as 'c METADATA VIRTUAL FROM k' "
+ + "is not selected by default if the strategy is applied.")),
+
+ EXCLUDE_DEFAULT_VIRTUAL_METADATA_COLUMNS(
+ text(
+ "Excludes virtual metadata columns that directly reference a metadata key. "
+ + "For example, a column declared as 'k METADATA VIRTUAL' "
+ + "is not selected by default if the strategy is applied."));
Review Comment:
Why would we want to exclude one and not the other? I'm wondering if we need the two values. I'm fine the way it is, just wanted to understand what's the idea here.
--
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
[GitHub] [flink] dawidwys commented on a diff in pull request #23365: [FLINK-33028][table] Make expanding behavior of virtual metadata columns configurable
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #23365:
URL: https://github.com/apache/flink/pull/23365#discussion_r1316935241
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/lookups/FieldReferenceLookup.java:
##########
@@ -75,40 +95,89 @@ public Optional<FieldReferenceExpression> lookupField(String name) {
* @return concatenated list of fields of all inputs.
*/
public List<FieldReferenceExpression> getAllInputFields() {
- return fieldReferences.stream().flatMap(input -> input.values().stream()).collect(toList());
+ return getInputFields(Collections.emptyList());
}
- private static List<Map<String, FieldReferenceExpression>> prepareFieldReferences(
+ /**
+ * Gives matching fields of underlying inputs in order of those inputs and order of fields
+ * within input.
+ *
+ * @return concatenated list of matching fields of all inputs.
+ */
+ public List<FieldReferenceExpression> getInputFields(
+ List<ColumnExpansionStrategy> expansionStrategies) {
+ return fieldReferences.stream()
+ .flatMap(input -> input.values().stream())
+ .filter(fieldRef -> includeExpandedColumn(fieldRef.column, expansionStrategies))
+ .map(FieldReference::toExpr)
+ .collect(toList());
+ }
+
+ private static List<Map<String, FieldReference>> prepareFieldReferences(
List<QueryOperation> queryOperations) {
return IntStream.range(0, queryOperations.size())
.mapToObj(idx -> prepareFieldsInInput(queryOperations.get(idx), idx))
.collect(Collectors.toList());
}
- private static Map<String, FieldReferenceExpression> prepareFieldsInInput(
+ private static Map<String, FieldReference> prepareFieldsInInput(
QueryOperation input, int inputIdx) {
ResolvedSchema resolvedSchema = input.getResolvedSchema();
return IntStream.range(0, resolvedSchema.getColumnCount())
- .mapToObj(
- i ->
- new FieldReferenceExpression(
- resolvedSchema.getColumnNames().get(i),
- resolvedSchema.getColumnDataTypes().get(i),
- inputIdx,
- i))
+ .mapToObj(i -> new FieldReference(resolvedSchema.getColumns().get(i), inputIdx, i))
.collect(
Collectors.toMap(
- FieldReferenceExpression::getName,
+ fieldRef -> fieldRef.column.getName(),
Function.identity(),
(fieldRef1, fieldRef2) -> {
- throw failAmbiuguousColumn(fieldRef1.getName());
+ throw failAmbiguousColumn(fieldRef1.column.getName());
},
// we need to maintain order of fields within input for resolving
// e.g. '*' reference
LinkedHashMap::new));
}
- private static ValidationException failAmbiuguousColumn(String name) {
+ private static ValidationException failAmbiguousColumn(String name) {
return new ValidationException("Ambiguous column name: " + name);
}
+
+ // --------------------------------------------------------------------------------------------
+ // Shared code with SQL validator
+ // --------------------------------------------------------------------------------------------
+
+ public static boolean includeExpandedColumn(
+ Column column, List<ColumnExpansionStrategy> strategies) {
+ for (ColumnExpansionStrategy strategy : strategies) {
+ switch (strategy) {
+ case EXCLUDE_ALIASED_VIRTUAL_METADATA_COLUMNS:
+ {
+ if (!(column instanceof Column.MetadataColumn)) {
+ continue;
+ }
+ final Column.MetadataColumn metadataColumn = (Column.MetadataColumn) column;
+ if (metadataColumn.isVirtual()
+ && metadataColumn.getMetadataKey().isPresent()) {
+ return false;
+ }
+ }
+ break;
Review Comment:
Could we already extract this to a separate method?
--
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