You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "venkata91 (via GitHub)" <gi...@apache.org> on 2023/08/29 01:13:17 UTC
[GitHub] [flink] venkata91 opened a new pull request, #23313: [FLINK-20767][table planner] Support filter push down on nested fields
venkata91 opened a new pull request, #23313:
URL: https://github.com/apache/flink/pull/23313
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
Support filter push down on nested fields
For example:
`SELECT * FROM users WHERE user.age > 18`
In the above SQL, `user.age > 18` filter can be pushed to table source to avoid scanning records that don't match the predicate. This is especially useful in analytics with columnar formats as well as in JDBC datasources.
## Brief change log
- Introduce a new `ResolvedExpression` called `NestedFieldReferenceExpression` to handle filters on nested fields
- Changes to `RexNodeToExpressionConverter` and `ExpressionConverter` to convert `RexFieldAccess` <=> `NestedFieldReferenceExpression` and vice versa.
## Verifying this change
This change added tests and can be verified as follows:
Added unit tests in `PushFilterIntoTableSourceScanRuleTest`
## 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)`: no
- 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?
- If yes, how is the feature documented? not applicable
--
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] venkata91 commented on a diff in pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on code in PR #23313:
URL: https://github.com/apache/flink/pull/23313#discussion_r1334815951
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java:
##########
@@ -1183,7 +1194,9 @@ private Map<Map<String, String>, Collection<Row>> filterAllData(
for (Row row : allData.get(partition)) {
boolean isRetained =
FilterUtils.isRetainedAfterApplyingFilterPredicates(
- filterPredicates, getValueGetter(row));
+ filterPredicates,
+ getValueGetter(row),
+ Optional.of(getNestedValueGetter(row)));
Review Comment:
Sure. Changed it from `Optional` to `@Nullable`.
--
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] swuferhong commented on a diff in pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on code in PR #23313:
URL: https://github.com/apache/flink/pull/23313#discussion_r1323892173
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala:
##########
@@ -395,9 +397,19 @@ class RexNodeToExpressionConverter(
inputNames: Array[String],
functionCatalog: FunctionCatalog,
catalogManager: CatalogManager,
- timeZone: TimeZone)
+ timeZone: TimeZone,
+ relDataType: Option[RelDataType] = None)
extends RexVisitor[Option[ResolvedExpression]] {
+ def this(
+ rexBuilder: RexBuilder,
+ inputNames: Array[String],
+ functionCatalog: FunctionCatalog,
+ catalogManager: CatalogManager,
+ timeZone: TimeZone) = {
+ this(rexBuilder, inputNames, functionCatalog, catalogManager, timeZone, null)
Review Comment:
There is no need to add `null`. `this(rexBuilder, inputNames, functionCatalog, catalogManager, timeZone)` is ok.
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/NestedFieldReferenceExpression.java:
##########
@@ -0,0 +1,126 @@
+/*
+ * 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.expressions;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.types.DataType;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * A reference to a nested field in an input. The reference contains.
Review Comment:
`contains.` -> `contains:`
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/NestedFieldReferenceExpression.java:
##########
@@ -0,0 +1,126 @@
+/*
+ * 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.expressions;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.types.DataType;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * A reference to a nested field in an input. The reference contains.
+ *
+ * <ul>
+ * <li>nested field names to traverse from the top level column to the nested leaf column.
+ * <li>nested field indices to traverse from the top level column to the nested leaf column.
+ * <li>type
+ * </ul>
+ */
+@PublicEvolving
+public class NestedFieldReferenceExpression implements ResolvedExpression {
+
+ /** Nested field names to traverse from the top level column to the nested leaf column. */
+ private final String[] fieldNames;
+
+ /** Nested field index to traverse from the top level column to the nested leaf column. */
+ private final int[] fieldIndices;
+
+ private final DataType dataType;
+
+ public NestedFieldReferenceExpression(
+ String[] fieldNames, int[] fieldIndices, DataType dataType) {
+ this.fieldNames = fieldNames;
+ this.fieldIndices = fieldIndices;
+ this.dataType = dataType;
+ }
+
+ public String[] getFieldNames() {
+ return fieldNames;
+ }
+
+ public int[] getFieldIndices() {
+ return fieldIndices;
+ }
+
+ public String getName() {
+ return String.format(
+ "`%s`",
+ String.join(
+ ".",
+ Arrays.stream(fieldNames)
+ .map(this::quoteIdentifier)
+ .toArray(String[]::new)));
+ }
+
+ @Override
+ public DataType getOutputDataType() {
+ return dataType;
+ }
+
+ @Override
+ public List<ResolvedExpression> getResolvedChildren() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public String asSummaryString() {
+ return getName();
+ }
+
+ @Override
+ public List<Expression> getChildren() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public <R> R accept(ExpressionVisitor<R> visitor) {
+ return visitor.visit(this);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ NestedFieldReferenceExpression that = (NestedFieldReferenceExpression) o;
+ return Arrays.equals(fieldNames, that.fieldNames)
+ && Arrays.equals(fieldIndices, that.fieldIndices)
+ && dataType.equals(that.dataType);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(fieldNames, fieldIndices, dataType);
+ }
Review Comment:
Idea warning: Change `fieldName` to `Arrays.hashCode(fieldName)`, change `fieldIndices` to `Arrays.hashCode(fieldIndices)`
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTest.java:
##########
@@ -118,4 +151,34 @@ public void testWithInterval() {
util.tableEnv().executeSql(ddl);
super.testWithInterval();
}
+
+ @Test
+ public void testBasicNestedFilter() {
+ util.verifyRelPlan("SELECT * FROM NestedTable WHERE deepNested.nested1.`value` > 2");
+ }
+
+ @Test
+ public void testNestedFilterWithDotInTheName() {
+ util.verifyRelPlan(
+ "SELECT id FROM NestedTable WHERE `deepNestedWith.`.nested.`.value` > 5");
+ }
+
+ @Test
+ public void testNestedFilterWithBacktickInTheName() {
+ util.verifyRelPlan(
+ "SELECT id FROM NestedTable WHERE `deepNestedWith.`.nested.```name` = 'foo'");
Review Comment:
One small question, is this the standard way of adding "`"? I couldn't find the relevant documents.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTest.java:
##########
@@ -86,6 +86,39 @@ public void setup() {
+ ")";
Review Comment:
Better to add some ITCase To verify the correctness of the data after push down
--
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] venkata91 commented on a diff in pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on code in PR #23313:
URL: https://github.com/apache/flink/pull/23313#discussion_r1331904794
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java:
##########
@@ -1183,7 +1194,9 @@ private Map<Map<String, String>, Collection<Row>> filterAllData(
for (Row row : allData.get(partition)) {
boolean isRetained =
FilterUtils.isRetainedAfterApplyingFilterPredicates(
- filterPredicates, getValueGetter(row));
+ filterPredicates,
+ getValueGetter(row),
+ Optional.of(getNestedValueGetter(row)));
Review Comment:
I see the point around not passing `Optional` as method param. Let me try to fix it.
--
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] venkata91 commented on a diff in pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on code in PR #23313:
URL: https://github.com/apache/flink/pull/23313#discussion_r1324865286
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTest.java:
##########
@@ -118,4 +151,34 @@ public void testWithInterval() {
util.tableEnv().executeSql(ddl);
super.testWithInterval();
}
+
+ @Test
+ public void testBasicNestedFilter() {
+ util.verifyRelPlan("SELECT * FROM NestedTable WHERE deepNested.nested1.`value` > 2");
+ }
+
+ @Test
+ public void testNestedFilterWithDotInTheName() {
+ util.verifyRelPlan(
+ "SELECT id FROM NestedTable WHERE `deepNestedWith.`.nested.`.value` > 5");
+ }
+
+ @Test
+ public void testNestedFilterWithBacktickInTheName() {
+ util.verifyRelPlan(
+ "SELECT id FROM NestedTable WHERE `deepNestedWith.`.nested.```name` = 'foo'");
Review Comment:
This tests the case where the column name has backtick (`) in it and should be escaped as the whole nested field expression name itself has to be wrapped inside backticks. File formats like ORC etc requires the entire nested field to be with in backticks (`)
--
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] becketqin commented on a diff in pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "becketqin (via GitHub)" <gi...@apache.org>.
becketqin commented on code in PR #23313:
URL: https://github.com/apache/flink/pull/23313#discussion_r1324619744
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala:
##########
@@ -538,8 +550,34 @@ class RexNodeToExpressionConverter(
}
}
- override def visitFieldAccess(fieldAccess: RexFieldAccess): Option[ResolvedExpression] = None
+ override def visitFieldAccess(fieldAccess: RexFieldAccess): Option[ResolvedExpression] = {
+ fieldAccess.getReferenceExpr match {
+ // push down on nested field inside a composite type like map or array is not supported
+ case _: RexCall => return None
Review Comment:
Nit: The RexCall also includes aggregation. Maybe update the comment to make that clear as well.
--
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] venkata91 commented on a diff in pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on code in PR #23313:
URL: https://github.com/apache/flink/pull/23313#discussion_r1325495894
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTest.java:
##########
@@ -86,6 +86,39 @@ public void setup() {
+ ")";
Review Comment:
Added ITCases for the new feature.
--
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] venkata91 commented on a diff in pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on code in PR #23313:
URL: https://github.com/apache/flink/pull/23313#discussion_r1324917758
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala:
##########
@@ -395,9 +397,19 @@ class RexNodeToExpressionConverter(
inputNames: Array[String],
functionCatalog: FunctionCatalog,
catalogManager: CatalogManager,
- timeZone: TimeZone)
+ timeZone: TimeZone,
+ relDataType: Option[RelDataType] = None)
extends RexVisitor[Option[ResolvedExpression]] {
+ def this(
+ rexBuilder: RexBuilder,
+ inputNames: Array[String],
+ functionCatalog: FunctionCatalog,
+ catalogManager: CatalogManager,
+ timeZone: TimeZone) = {
+ this(rexBuilder, inputNames, functionCatalog, catalogManager, timeZone, null)
Review Comment:
Will change it to `None` instead of `null` to conform with Scala.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala:
##########
@@ -395,9 +397,19 @@ class RexNodeToExpressionConverter(
inputNames: Array[String],
functionCatalog: FunctionCatalog,
catalogManager: CatalogManager,
- timeZone: TimeZone)
+ timeZone: TimeZone,
+ relDataType: Option[RelDataType] = None)
extends RexVisitor[Option[ResolvedExpression]] {
+ def this(
+ rexBuilder: RexBuilder,
+ inputNames: Array[String],
+ functionCatalog: FunctionCatalog,
+ catalogManager: CatalogManager,
+ timeZone: TimeZone) = {
+ this(rexBuilder, inputNames, functionCatalog, catalogManager, timeZone, null)
Review Comment:
Will change it to `None` instead of `null` to conform with Scala rules/conventions.
--
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] venkata91 commented on pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on PR #23313:
URL: https://github.com/apache/flink/pull/23313#issuecomment-1710555055
@wuchong @swuferhong @JingsongLi @becketqin please review
--
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] becketqin commented on pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "becketqin (via GitHub)" <gi...@apache.org>.
becketqin commented on PR #23313:
URL: https://github.com/apache/flink/pull/23313#issuecomment-1736426613
Merged to master: 5be4688e8b8a055b9e8891c265f6f354083d630f
--
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] becketqin merged pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "becketqin (via GitHub)" <gi...@apache.org>.
becketqin merged PR #23313:
URL: https://github.com/apache/flink/pull/23313
--
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] venkata91 commented on a diff in pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on code in PR #23313:
URL: https://github.com/apache/flink/pull/23313#discussion_r1331902142
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java:
##########
@@ -1183,7 +1194,9 @@ private Map<Map<String, String>, Collection<Row>> filterAllData(
for (Row row : allData.get(partition)) {
boolean isRetained =
FilterUtils.isRetainedAfterApplyingFilterPredicates(
- filterPredicates, getValueGetter(row));
+ filterPredicates,
+ getValueGetter(row),
+ Optional.of(getNestedValueGetter(row)));
Review Comment:
`FilterUtils.isRetainedAfterApplyingFilterPredicates` is used in 2 places:
1. `TestValuesTableFactory` - For regular table read with projections and filter pushdowns.
2. `TestValuesCatalog` - Here it is used in the context of `partition` values and since we don't have support for nested fields with partitions, we pass `Optional.empty` in this case.
`getNestedValueGetter` itself returns a `Function` which is wrapped inside `Optional`. `getNestedValueGetter` function itself won't return null. Given that we are using it only in the `TestValuesTableFactory` but not in `TestValuesCatalog`, I wrap it with in `Optional` and pass it to `FilterUtils.isRetainedAfterApplyingFilterPredicates` which is used in both of the above classes.
--
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] venkata91 commented on a diff in pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on code in PR #23313:
URL: https://github.com/apache/flink/pull/23313#discussion_r1324893510
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala:
##########
@@ -538,8 +550,34 @@ class RexNodeToExpressionConverter(
}
}
- override def visitFieldAccess(fieldAccess: RexFieldAccess): Option[ResolvedExpression] = None
+ override def visitFieldAccess(fieldAccess: RexFieldAccess): Option[ResolvedExpression] = {
+ fieldAccess.getReferenceExpr match {
+ // push down on nested field inside a composite type like map or array is not supported
+ case _: RexCall => return None
Review Comment:
Do you mean make it clear in the comment that `lower(a.b.c) = "foo"` is not supported? The above addresses cases like:
```
"SELECT * FROM NestedItemTable WHERE `Result`.`Mid`.data_arr[2].`value` = 3"
```
where `data_arr` is an Array and `data_arr[2]` is expressed as `RexCall(ITEM, data_arr, 2)` which won't be supported.
--
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] venkata91 commented on a diff in pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on code in PR #23313:
URL: https://github.com/apache/flink/pull/23313#discussion_r1331902142
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java:
##########
@@ -1183,7 +1194,9 @@ private Map<Map<String, String>, Collection<Row>> filterAllData(
for (Row row : allData.get(partition)) {
boolean isRetained =
FilterUtils.isRetainedAfterApplyingFilterPredicates(
- filterPredicates, getValueGetter(row));
+ filterPredicates,
+ getValueGetter(row),
+ Optional.of(getNestedValueGetter(row)));
Review Comment:
`FilterUtils.isRetainedAfterApplyingFilterPredicates` is used in 2 places:
1. `TestValuesTableFactory` - For regular table read with projections and filter pushdowns.
2. `TestValuesCatalog` - Here it is used in the context of `partition` values and since we don't have support for nested fields with partitions, we pass `Optional.empty` in this case.
`getNestedValueGetter` itself returns a `Function` which is wrapped inside `Optional`. Applying the function itself won't return null. Given that we are using it only in the `TestValuesTableFactory` but not in `TestValuesCatalog`, I wrap it with in `Optional` and pass it to `FilterUtils.isRetainedAfterApplyingFilterPredicates` which is used in both of the above classes.
--
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] becketqin commented on a diff in pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "becketqin (via GitHub)" <gi...@apache.org>.
becketqin commented on code in PR #23313:
URL: https://github.com/apache/flink/pull/23313#discussion_r1331649554
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java:
##########
@@ -1183,7 +1194,9 @@ private Map<Map<String, String>, Collection<Row>> filterAllData(
for (Row row : allData.get(partition)) {
boolean isRetained =
FilterUtils.isRetainedAfterApplyingFilterPredicates(
- filterPredicates, getValueGetter(row));
+ filterPredicates,
+ getValueGetter(row),
+ Optional.of(getNestedValueGetter(row)));
Review Comment:
If `getNestedValueGetter(row)` can ever return null, `Optional.ofNullable()` should be used here.
BTW, usually we want to avoid using `Optional` as method parameter. Although this is in the test, would be good to keep good practice.
--
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] becketqin commented on a diff in pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "becketqin (via GitHub)" <gi...@apache.org>.
becketqin commented on code in PR #23313:
URL: https://github.com/apache/flink/pull/23313#discussion_r1332312205
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java:
##########
@@ -1183,7 +1194,9 @@ private Map<Map<String, String>, Collection<Row>> filterAllData(
for (Row row : allData.get(partition)) {
boolean isRetained =
FilterUtils.isRetainedAfterApplyingFilterPredicates(
- filterPredicates, getValueGetter(row));
+ filterPredicates,
+ getValueGetter(row),
+ Optional.of(getNestedValueGetter(row)));
Review Comment:
I think this is more of a programing convention for language. Optional is equivalent to `@Nullable` annotation in Java.
The Optional parameter is kind of a paradox. If a parameter is Optional, that means a null value is acceptable. So users can actually pass the nullable value without wrapping it with Optional. If a parameter is not Optional, it implies that the value cannot be null, so user need to make a null check on the value. If it is null, likely an exception needs to be thrown, which is pretty much what the invoked method will do when it sees null for the non-optional value. So in either case, having an Optional parameter does not bring much value, while introduce extra wrapping in the first case.
From a caller's perspective, a parameter value is known to be null or not, if the value is going to be passed to the method call anyways, wrapping the value with Optional is extra burden compared with passing the value directly. On the other handle, the return value of the method call is unknown to the caller. If the return value of a method call is an Optional, it explicitly tells the caller that the return value needs to be checked because it might return nothing. This is why Optional is useful for return value, but not the parameters.
--
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] venkata91 commented on pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on PR #23313:
URL: https://github.com/apache/flink/pull/23313#issuecomment-1700010200
@flinkbot run azure
--
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] venkata91 commented on a diff in pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on code in PR #23313:
URL: https://github.com/apache/flink/pull/23313#discussion_r1331924882
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java:
##########
@@ -1183,7 +1194,9 @@ private Map<Map<String, String>, Collection<Row>> filterAllData(
for (Row row : allData.get(partition)) {
boolean isRetained =
FilterUtils.isRetainedAfterApplyingFilterPredicates(
- filterPredicates, getValueGetter(row));
+ filterPredicates,
+ getValueGetter(row),
+ Optional.of(getNestedValueGetter(row)));
Review Comment:
Regarding the `Optional` use as a method parameter has great discussions [here](https://stackoverflow.com/questions/31922866/why-should-java-8s-optional-not-be-used-in-arguments).
Per [Brian Goetz](https://stackoverflow.com/a/26328555),
> [W]e did have a clear intention when adding [Optional], and it was not to be a general purpose Maybe type, as much as many people would have liked us to do so. Our intention was to provide a limited mechanism for library method return types where there needed to be a clear way to represent "no result", and using null for such was overwhelmingly likely to cause errors.
>
> For example, you probably should never use it for something that returns an array of results, or a list of results; instead return an empty array or list. You should almost never use it as a field of something or a method parameter.
But it is not clear why using the `Option` or `Maybe` type in other languages like Scala etc in method params is fine.
Having said that, I will try to remove the use of `Optional` if you feel strongly about it.
--
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] venkata91 commented on pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on PR #23313:
URL: https://github.com/apache/flink/pull/23313#issuecomment-1726236594
@swuferhong @becketqin @lsyldliu Gentle ping for reviews.
--
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] venkata91 commented on pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on PR #23313:
URL: https://github.com/apache/flink/pull/23313#issuecomment-1724620100
> @venkata91 Thanks for your contribution, please use git rebase instead of git merge. I can't get your all commits correctly when pull your pr to local.
@lsyldliu Do you want me to fix it?
--
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 #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #23313:
URL: https://github.com/apache/flink/pull/23313#issuecomment-1696630993
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5470431742bf63292142aaa7c96b4c894b3a7c82",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5470431742bf63292142aaa7c96b4c894b3a7c82",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5470431742bf63292142aaa7c96b4c894b3a7c82 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] venkata91 commented on a diff in pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on code in PR #23313:
URL: https://github.com/apache/flink/pull/23313#discussion_r1324916866
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala:
##########
@@ -395,9 +397,19 @@ class RexNodeToExpressionConverter(
inputNames: Array[String],
functionCatalog: FunctionCatalog,
catalogManager: CatalogManager,
- timeZone: TimeZone)
+ timeZone: TimeZone,
+ relDataType: Option[RelDataType] = None)
extends RexVisitor[Option[ResolvedExpression]] {
+ def this(
+ rexBuilder: RexBuilder,
+ inputNames: Array[String],
+ functionCatalog: FunctionCatalog,
+ catalogManager: CatalogManager,
+ timeZone: TimeZone) = {
+ this(rexBuilder, inputNames, functionCatalog, catalogManager, timeZone, null)
Review Comment:
I think `None` or `null` is required without which scala compiler fails with `constructor invokes itself`
--
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] venkata91 commented on pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on PR #23313:
URL: https://github.com/apache/flink/pull/23313#issuecomment-1724625067
> > @venkata91 Thanks for your contribution, please use git rebase instead of git merge. I can't get your all commits correctly when pull your pr to local.
>
> @lsyldliu Do you want me to fix it?
I used to use rebase before. But with `rebase` the commit ids changes and therefore it is hard to only see the requested changes which is why I changed to `merge` instead of `rebase`. See [here](https://github.com/apache/spark/pull/28287#issuecomment-662458128), started doing it after this comment.
--
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] venkata91 commented on a diff in pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on code in PR #23313:
URL: https://github.com/apache/flink/pull/23313#discussion_r1331904794
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java:
##########
@@ -1183,7 +1194,9 @@ private Map<Map<String, String>, Collection<Row>> filterAllData(
for (Row row : allData.get(partition)) {
boolean isRetained =
FilterUtils.isRetainedAfterApplyingFilterPredicates(
- filterPredicates, getValueGetter(row));
+ filterPredicates,
+ getValueGetter(row),
+ Optional.of(getNestedValueGetter(row)));
Review Comment:
I see the point around not passing `Optional` as method param. Let me try to fix it.
--
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] venkata91 commented on pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on PR #23313:
URL: https://github.com/apache/flink/pull/23313#issuecomment-1724629169
> > @venkata91 Thanks for your contribution, please use git rebase instead of git merge. I can't get your all commits correctly when pull your pr to local.
>
> @lsyldliu Do you want me to fix it?
nvm, fixed it as requested.
--
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] venkata91 commented on pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on PR #23313:
URL: https://github.com/apache/flink/pull/23313#issuecomment-1710555541
Also tagging @slinkydeveloper
--
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] venkata91 commented on pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on PR #23313:
URL: https://github.com/apache/flink/pull/23313#issuecomment-1718916688
@becketqin @swuferhong Addressed review comments mainly adding ITCases.
--
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] venkata91 commented on a diff in pull request #23313: [FLINK-20767][table planner] Support filter push down on nested fields
Posted by "venkata91 (via GitHub)" <gi...@apache.org>.
venkata91 commented on code in PR #23313:
URL: https://github.com/apache/flink/pull/23313#discussion_r1324012808
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTest.java:
##########
@@ -86,6 +86,39 @@ public void setup() {
+ ")";
Review Comment:
Sure. Could you please point me to where I can add ITcase for this fix?
--
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