You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Jiabao-Sun (via GitHub)" <gi...@apache.org> on 2023/10/19 10:15:13 UTC
[PR] [FLINK-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Jiabao-Sun opened a new pull request, #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17
[FLINK-33257][connectors/mongodb] Support filter pushdown in MongoDB connector
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on code in PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#discussion_r1374448348
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/table/MongoDynamicTableSource.java:
##########
@@ -148,13 +163,16 @@ public ChangelogMode getChangelogMode() {
@Override
public DynamicTableSource copy() {
- return new MongoDynamicTableSource(
- connectionOptions,
- readOptions,
- lookupCache,
- lookupMaxRetries,
- lookupRetryIntervalMs,
- producedDataType);
+ MongoDynamicTableSource newSource =
+ new MongoDynamicTableSource(
+ connectionOptions,
+ readOptions,
+ lookupCache,
+ lookupMaxRetries,
+ lookupRetryIntervalMs,
+ producedDataType);
+ newSource.filter = BsonDocument.parse(filter.toJson());
Review Comment:
As the java doc of the copy method, the copy should be a deep copy of all.
```java
/**
* Creates a copy of this instance during planning. The copy should be a deep copy of all
* mutable members.
*/
DynamicTableSource copy();
```
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#discussion_r1374396462
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/table/MongoDynamicTableSource.java:
##########
@@ -178,6 +196,41 @@ public void applyProjection(int[][] projectedFields, DataType producedDataType)
this.producedDataType = producedDataType;
}
+ @Override
+ public Result applyFilters(List<ResolvedExpression> filters) {
+ List<ResolvedExpression> acceptedFilters = new ArrayList<>();
+ List<ResolvedExpression> remainingFilters = new ArrayList<>();
+
+ List<Bson> mongoFilters = new ArrayList<>();
+ for (ResolvedExpression filter : filters) {
+ BsonDocument simpleFilter = parseFilter(filter);
+ if (simpleFilter.isEmpty()) {
+ remainingFilters.add(filter);
+ } else {
+ acceptedFilters.add(filter);
+ mongoFilters.add(simpleFilter);
+ }
+ }
+
+ if (!mongoFilters.isEmpty()) {
+ Bson mergedFilter =
+ mongoFilters.size() == 1 ? mongoFilters.get(0) : Filters.and(mongoFilters);
+ this.filter = mergedFilter.toBsonDocument();
+ LOG.debug("Pushed down filters: {}", filter.toJson());
Review Comment:
I prefer `LOG.info` here, WDYT?
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/table/MongoDynamicTableSource.java:
##########
@@ -148,13 +163,16 @@ public ChangelogMode getChangelogMode() {
@Override
public DynamicTableSource copy() {
- return new MongoDynamicTableSource(
- connectionOptions,
- readOptions,
- lookupCache,
- lookupMaxRetries,
- lookupRetryIntervalMs,
- producedDataType);
+ MongoDynamicTableSource newSource =
+ new MongoDynamicTableSource(
+ connectionOptions,
+ readOptions,
+ lookupCache,
+ lookupMaxRetries,
+ lookupRetryIntervalMs,
+ producedDataType);
+ newSource.filter = BsonDocument.parse(filter.toJson());
Review Comment:
Why not use `newSource.filter = this.filter;` directly?
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/table/MongoDynamicTableSource.java:
##########
@@ -178,6 +196,41 @@ public void applyProjection(int[][] projectedFields, DataType producedDataType)
this.producedDataType = producedDataType;
}
+ @Override
+ public Result applyFilters(List<ResolvedExpression> filters) {
+ List<ResolvedExpression> acceptedFilters = new ArrayList<>();
Review Comment:
I'm not sure if this feature needs to be compatible with the behavior before while users upgrade their jobs. If it is, maybe adding a config for it is better.
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/table/MongoFilterPushDownVisitor.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.connector.mongodb.table;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.table.expressions.CallExpression;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.ExpressionDefaultVisitor;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.expressions.ResolvedExpression;
+import org.apache.flink.table.expressions.ValueLiteralExpression;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import com.mongodb.client.model.Filters;
+import org.bson.BsonBoolean;
+import org.bson.BsonDateTime;
+import org.bson.BsonDecimal128;
+import org.bson.BsonDocument;
+import org.bson.BsonDouble;
+import org.bson.BsonInt32;
+import org.bson.BsonInt64;
+import org.bson.BsonNull;
+import org.bson.BsonString;
+import org.bson.BsonType;
+import org.bson.BsonUndefined;
+import org.bson.BsonValue;
+import org.bson.conversions.Bson;
+import org.bson.types.Decimal128;
+
+import java.math.BigDecimal;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * Visitor that convert Expression to Bson filter. Return {@link Filters#empty()} if we cannot push
+ * down the filter.
+ */
+@Experimental
+public class MongoFilterPushDownVisitor extends ExpressionDefaultVisitor<BsonValue> {
+
+ public static final MongoFilterPushDownVisitor INSTANCE = new MongoFilterPushDownVisitor();
+
+ private MongoFilterPushDownVisitor() {
+ // Singleton instance.
+ }
+
+ @Override
+ public BsonDocument visit(CallExpression call) {
+ Bson filter = Filters.empty();
+ if (BuiltInFunctionDefinitions.EQUALS.equals(call.getFunctionDefinition())) {
Review Comment:
nit: using `switch` instead of too many `if`, and return directly if matched.
##########
flink-connector-mongodb/src/test/java/org/apache/flink/connector/mongodb/source/MongoSourceITCase.java:
##########
@@ -76,7 +76,7 @@
/** IT cases for using Mongo Sink. */
@Testcontainers
-public class MongoSourceITCase {
+class MongoSourceITCase {
Review Comment:
I wonder why you remove the `public` here and in the other functions?
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/table/MongoFilterPushDownVisitor.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.connector.mongodb.table;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.table.expressions.CallExpression;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.ExpressionDefaultVisitor;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.expressions.ResolvedExpression;
+import org.apache.flink.table.expressions.ValueLiteralExpression;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import com.mongodb.client.model.Filters;
+import org.bson.BsonBoolean;
+import org.bson.BsonDateTime;
+import org.bson.BsonDecimal128;
+import org.bson.BsonDocument;
+import org.bson.BsonDouble;
+import org.bson.BsonInt32;
+import org.bson.BsonInt64;
+import org.bson.BsonNull;
+import org.bson.BsonString;
+import org.bson.BsonType;
+import org.bson.BsonUndefined;
+import org.bson.BsonValue;
+import org.bson.conversions.Bson;
+import org.bson.types.Decimal128;
+
+import java.math.BigDecimal;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * Visitor that convert Expression to Bson filter. Return {@link Filters#empty()} if we cannot push
+ * down the filter.
+ */
+@Experimental
+public class MongoFilterPushDownVisitor extends ExpressionDefaultVisitor<BsonValue> {
+
+ public static final MongoFilterPushDownVisitor INSTANCE = new MongoFilterPushDownVisitor();
+
+ private MongoFilterPushDownVisitor() {
+ // Singleton instance.
+ }
+
+ @Override
+ public BsonDocument visit(CallExpression call) {
+ Bson filter = Filters.empty();
+ if (BuiltInFunctionDefinitions.EQUALS.equals(call.getFunctionDefinition())) {
+ filter = renderBinaryComparisonOperator("$eq", call.getResolvedChildren());
+ }
+ if (BuiltInFunctionDefinitions.LESS_THAN.equals(call.getFunctionDefinition())) {
+ filter = renderBinaryComparisonOperator("$lt", call.getResolvedChildren());
+ }
+ if (BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL.equals(call.getFunctionDefinition())) {
+ filter = renderBinaryComparisonOperator("$lte", call.getResolvedChildren());
+ }
+ if (BuiltInFunctionDefinitions.GREATER_THAN.equals(call.getFunctionDefinition())) {
+ filter = renderBinaryComparisonOperator("$gt", call.getResolvedChildren());
+ }
+ if (BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL.equals(call.getFunctionDefinition())) {
+ filter = renderBinaryComparisonOperator("$gte", call.getResolvedChildren());
+ }
+ if (BuiltInFunctionDefinitions.NOT_EQUALS.equals(call.getFunctionDefinition())) {
+ filter = renderBinaryComparisonOperator("$ne", call.getResolvedChildren());
+ }
+ if (BuiltInFunctionDefinitions.IS_NULL.equals(call.getFunctionDefinition())) {
+ filter =
+ renderUnaryComparisonOperator(
+ "$eq", call.getResolvedChildren().get(0), BsonNull.VALUE);
+ }
+ if (BuiltInFunctionDefinitions.IS_NOT_NULL.equals(call.getFunctionDefinition())) {
+ filter =
+ renderUnaryComparisonOperator(
+ "$ne", call.getResolvedChildren().get(0), BsonNull.VALUE);
+ }
+ if (BuiltInFunctionDefinitions.OR.equals(call.getFunctionDefinition())) {
+ filter = renderLogicalOperator("$or", call.getResolvedChildren());
+ }
+ if (BuiltInFunctionDefinitions.AND.equals(call.getFunctionDefinition())) {
+ filter = renderLogicalOperator("$and", call.getResolvedChildren());
+ }
+ return filter.toBsonDocument();
+ }
+
+ private Bson renderBinaryComparisonOperator(
+ String operator, List<ResolvedExpression> expressions) {
+ Optional<FieldReferenceExpression> fieldReferenceExpr =
Review Comment:
Just curious, currently `f1 = f2` is not supported while `f1` and `f2` are both source fields?
##########
flink-connector-mongodb/src/test/java/org/apache/flink/connector/mongodb/table/MongoDynamicTableSourceITCase.java:
##########
@@ -319,12 +361,13 @@ private static String createTestDDl(Map<String, String> extraOptions) {
"CREATE TABLE mongo_source",
"(",
" _id BIGINT,",
+ " f0 STRING,",
" f1 STRING,",
" f2 BOOLEAN,",
" f3 BINARY,",
" f4 INTEGER,",
- " f5 TIMESTAMP_LTZ(6),",
- " f6 TIMESTAMP_LTZ(3),",
+ " f5 TIMESTAMP_LTZ(3),",
Review Comment:
I'm not sure about these changes. These changes seem to be changed by refacting test data, right?
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#issuecomment-1794172578
> Thanks @leonardBang, shall we remove `flink-table-planner-loader` dependency and replace it with `flink-table-planner-2.12`?
`flink-table-planner-loader` is recommended from the context https://issues.apache.org/jira/browse/FLINK-24427, but it's wired using `flink-table-planner-loader` lead to above errors.
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#issuecomment-1770506248
Hi @leonardBang, could you help review this when you have time?
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#discussion_r1374480051
##########
flink-connector-mongodb/src/test/java/org/apache/flink/connector/mongodb/source/MongoSourceITCase.java:
##########
@@ -76,7 +76,7 @@
/** IT cases for using Mongo Sink. */
@Testcontainers
-public class MongoSourceITCase {
+class MongoSourceITCase {
Review Comment:
go 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
Re: [PR] [FLINK-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on code in PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#discussion_r1374455479
##########
flink-connector-mongodb/src/test/java/org/apache/flink/connector/mongodb/source/MongoSourceITCase.java:
##########
@@ -76,7 +76,7 @@
/** IT cases for using Mongo Sink. */
@Testcontainers
-public class MongoSourceITCase {
+class MongoSourceITCase {
Review Comment:
> Test Classes and Methods
In JUnit 5, test classes, test methods, and lifecycle methods are not required to be public, but they must not be private. And the [JUnit5 user guide ](https://junit.org/junit5/docs/snapshot/user-guide/#writing-tests-classes-and-methods)recommends omitting the public modifier for them unless there is a technical reason – for example, when a test class is extended by a test class in another package. So please follow this rule in the migration.
Refers to
- https://docs.google.com/document/d/1514Wa_aNB9bJUen4xm5uiuXOooOJTtXqS_Jqk9KJitU/edit#heading=h.4iilyfqpoiw2
- https://issues.apache.org/jira/browse/FLINK-25325
##########
flink-connector-mongodb/src/test/java/org/apache/flink/connector/mongodb/source/MongoSourceITCase.java:
##########
@@ -76,7 +76,7 @@
/** IT cases for using Mongo Sink. */
@Testcontainers
-public class MongoSourceITCase {
+class MongoSourceITCase {
Review Comment:
> Test Classes and Methods
In JUnit 5, test classes, test methods, and lifecycle methods are not required to be public, but they must not be private. And the [JUnit5 user guide ](https://junit.org/junit5/docs/snapshot/user-guide/#writing-tests-classes-and-methods)recommends omitting the public modifier for them unless there is a technical reason – for example, when a test class is extended by a test class in another package. So please follow this rule in the migration.
Refers to
- https://docs.google.com/document/d/1514Wa_aNB9bJUen4xm5uiuXOooOJTtXqS_Jqk9KJitU/edit#heading=h.4iilyfqpoiw2
- https://issues.apache.org/jira/browse/FLINK-25325
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#issuecomment-1860520670
Thanks @xuyangzhong for the review.
Hi @dannycranmer, @leonardBang.
Could you please help take a look when you have time?
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#issuecomment-1794153008
> Hi @xuyangzhong。 To implement the planner's test, the [flink-table-planner](https://issues.apache.org/jira/browse/FLINK-table-planner)_2.12 related dependencies need to be introduced, but it will result in the following error. It seems that the error is caused by the presence of an ExecutorFactory with the identifier "default" in the [flink-table-planner-loader](https://issues.apache.org/jira/browse/FLINK-table-planner-loader) as well.
>
> ```
> org.apache.flink.table.api.ValidationException: Multiple factories for identifier 'default' that implement 'org.apache.flink.table.delegation.ExecutorFactory' found in the classpath.
>
> Ambiguous factory classes are:
>
> org.apache.flink.table.planner.delegation.DefaultExecutorFactory
> org.apache.flink.table.planner.loader.DelegateExecutorFactory
> ```
>
> Do you have some suggestions? cc @leonardBang.
IIRC, jdbc connector also contains planner tests, maybe we can refer it to avoid this case
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#issuecomment-1798132958
Hi, @Jiabao-Sun, sorry for the late reply.
What about adding more cases while testing plan such as the following?
1. pushdown some expressions and retain other expressions
2. retain all expressions
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#issuecomment-1794389313
Hi @leonardBang, sorry for the CI failure.
Please help trigger the CI again.
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#issuecomment-1794157929
Thanks @leonardBang, shall we remove `flink-table-planner-loader` dependency and replace it with `flink-table-planner-2.12`?
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#issuecomment-1794209251
Tests for plan has been added.
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on code in PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#discussion_r1374475234
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/table/MongoFilterPushDownVisitor.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.connector.mongodb.table;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.table.expressions.CallExpression;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.ExpressionDefaultVisitor;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.expressions.ResolvedExpression;
+import org.apache.flink.table.expressions.ValueLiteralExpression;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import com.mongodb.client.model.Filters;
+import org.bson.BsonBoolean;
+import org.bson.BsonDateTime;
+import org.bson.BsonDecimal128;
+import org.bson.BsonDocument;
+import org.bson.BsonDouble;
+import org.bson.BsonInt32;
+import org.bson.BsonInt64;
+import org.bson.BsonNull;
+import org.bson.BsonString;
+import org.bson.BsonType;
+import org.bson.BsonUndefined;
+import org.bson.BsonValue;
+import org.bson.conversions.Bson;
+import org.bson.types.Decimal128;
+
+import java.math.BigDecimal;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * Visitor that convert Expression to Bson filter. Return {@link Filters#empty()} if we cannot push
+ * down the filter.
+ */
+@Experimental
+public class MongoFilterPushDownVisitor extends ExpressionDefaultVisitor<BsonValue> {
+
+ public static final MongoFilterPushDownVisitor INSTANCE = new MongoFilterPushDownVisitor();
+
+ private MongoFilterPushDownVisitor() {
+ // Singleton instance.
+ }
+
+ @Override
+ public BsonDocument visit(CallExpression call) {
+ Bson filter = Filters.empty();
+ if (BuiltInFunctionDefinitions.EQUALS.equals(call.getFunctionDefinition())) {
+ filter = renderBinaryComparisonOperator("$eq", call.getResolvedChildren());
+ }
+ if (BuiltInFunctionDefinitions.LESS_THAN.equals(call.getFunctionDefinition())) {
+ filter = renderBinaryComparisonOperator("$lt", call.getResolvedChildren());
+ }
+ if (BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL.equals(call.getFunctionDefinition())) {
+ filter = renderBinaryComparisonOperator("$lte", call.getResolvedChildren());
+ }
+ if (BuiltInFunctionDefinitions.GREATER_THAN.equals(call.getFunctionDefinition())) {
+ filter = renderBinaryComparisonOperator("$gt", call.getResolvedChildren());
+ }
+ if (BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL.equals(call.getFunctionDefinition())) {
+ filter = renderBinaryComparisonOperator("$gte", call.getResolvedChildren());
+ }
+ if (BuiltInFunctionDefinitions.NOT_EQUALS.equals(call.getFunctionDefinition())) {
+ filter = renderBinaryComparisonOperator("$ne", call.getResolvedChildren());
+ }
+ if (BuiltInFunctionDefinitions.IS_NULL.equals(call.getFunctionDefinition())) {
+ filter =
+ renderUnaryComparisonOperator(
+ "$eq", call.getResolvedChildren().get(0), BsonNull.VALUE);
+ }
+ if (BuiltInFunctionDefinitions.IS_NOT_NULL.equals(call.getFunctionDefinition())) {
+ filter =
+ renderUnaryComparisonOperator(
+ "$ne", call.getResolvedChildren().get(0), BsonNull.VALUE);
+ }
+ if (BuiltInFunctionDefinitions.OR.equals(call.getFunctionDefinition())) {
+ filter = renderLogicalOperator("$or", call.getResolvedChildren());
+ }
+ if (BuiltInFunctionDefinitions.AND.equals(call.getFunctionDefinition())) {
+ filter = renderLogicalOperator("$and", call.getResolvedChildren());
+ }
+ return filter.toBsonDocument();
+ }
+
+ private Bson renderBinaryComparisonOperator(
+ String operator, List<ResolvedExpression> expressions) {
+ Optional<FieldReferenceExpression> fieldReferenceExpr =
Review Comment:
MongoDB is more complicated in this situation, and it will cause full collection scanning, so it is temporarily regardless of the scene of Pushdown.
https://stackoverflow.com/questions/4442453/mongodb-query-condition-on-comparing-2-fields
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on code in PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#discussion_r1374478320
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/table/MongoDynamicTableSource.java:
##########
@@ -178,6 +196,41 @@ public void applyProjection(int[][] projectedFields, DataType producedDataType)
this.producedDataType = producedDataType;
}
+ @Override
+ public Result applyFilters(List<ResolvedExpression> filters) {
+ List<ResolvedExpression> acceptedFilters = new ArrayList<>();
+ List<ResolvedExpression> remainingFilters = new ArrayList<>();
+
+ List<Bson> mongoFilters = new ArrayList<>();
+ for (ResolvedExpression filter : filters) {
+ BsonDocument simpleFilter = parseFilter(filter);
+ if (simpleFilter.isEmpty()) {
+ remainingFilters.add(filter);
+ } else {
+ acceptedFilters.add(filter);
+ mongoFilters.add(simpleFilter);
+ }
+ }
+
+ if (!mongoFilters.isEmpty()) {
+ Bson mergedFilter =
+ mongoFilters.size() == 1 ? mongoFilters.get(0) : Filters.and(mongoFilters);
+ this.filter = mergedFilter.toBsonDocument();
+ LOG.debug("Pushed down filters: {}", filter.toJson());
Review Comment:
Fixed.
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang merged PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on code in PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#discussion_r1374460482
##########
flink-connector-mongodb/src/test/java/org/apache/flink/connector/mongodb/table/MongoDynamicTableSourceITCase.java:
##########
@@ -319,12 +361,13 @@ private static String createTestDDl(Map<String, String> extraOptions) {
"CREATE TABLE mongo_source",
"(",
" _id BIGINT,",
+ " f0 STRING,",
" f1 STRING,",
" f2 BOOLEAN,",
" f3 BINARY,",
" f4 INTEGER,",
- " f5 TIMESTAMP_LTZ(6),",
- " f6 TIMESTAMP_LTZ(3),",
+ " f5 TIMESTAMP_LTZ(3),",
Review Comment:
In fact, MongoDB supports the maximum time accuracy of 3. Although it does not have any impact on the test, this may cause misunderstandings, so I fix it by the way.
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#issuecomment-1784701738
Hi @xuyangzhong。
To implement the planner's test, the flink-table-planner_2.12 related dependencies need to be introduced, but it will result in the following error. It seems that the error is caused by the presence of an ExecutorFactory with the identifier "default" in the flink-table-planner-loader as well.
```
org.apache.flink.table.api.ValidationException: Multiple factories for identifier 'default' that implement 'org.apache.flink.table.delegation.ExecutorFactory' found in the classpath.
Ambiguous factory classes are:
org.apache.flink.table.planner.delegation.DefaultExecutorFactory
org.apache.flink.table.planner.loader.DelegateExecutorFactory
```
Do you have some suggestions? cc @leonardBang.
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "xuyangzhong (via GitHub)" <gi...@apache.org>.
xuyangzhong commented on code in PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#discussion_r1374480777
##########
flink-connector-mongodb/src/test/java/org/apache/flink/connector/mongodb/table/MongoDynamicTableSourceITCase.java:
##########
@@ -319,12 +361,13 @@ private static String createTestDDl(Map<String, String> extraOptions) {
"CREATE TABLE mongo_source",
"(",
" _id BIGINT,",
+ " f0 STRING,",
" f1 STRING,",
" f2 BOOLEAN,",
" f3 BINARY,",
" f4 INTEGER,",
- " f5 TIMESTAMP_LTZ(6),",
- " f6 TIMESTAMP_LTZ(3),",
+ " f5 TIMESTAMP_LTZ(3),",
Review Comment:
Got it. It is best to submit changes unrelated to this core issue to a separate commit in one pr.
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on code in PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#discussion_r1374496162
##########
flink-connector-mongodb/src/test/java/org/apache/flink/connector/mongodb/table/MongoDynamicTableSourceITCase.java:
##########
@@ -319,12 +361,13 @@ private static String createTestDDl(Map<String, String> extraOptions) {
"CREATE TABLE mongo_source",
"(",
" _id BIGINT,",
+ " f0 STRING,",
" f1 STRING,",
" f2 BOOLEAN,",
" f3 BINARY,",
" f4 INTEGER,",
- " f5 TIMESTAMP_LTZ(6),",
- " f6 TIMESTAMP_LTZ(3),",
+ " f5 TIMESTAMP_LTZ(3),",
Review Comment:
OK, thanks.
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#issuecomment-1801742466
Thanks @xuyangzhong and @leonardBang for the review.
Could you help review it again when you have time?
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#issuecomment-1794142093
Hi @leonardBang, could you help do the final check?
I think we may introduce it in the release 1.18 version.
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on code in PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#discussion_r1374465459
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/table/MongoFilterPushDownVisitor.java:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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.connector.mongodb.table;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.table.expressions.CallExpression;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.ExpressionDefaultVisitor;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.expressions.ResolvedExpression;
+import org.apache.flink.table.expressions.ValueLiteralExpression;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import com.mongodb.client.model.Filters;
+import org.bson.BsonBoolean;
+import org.bson.BsonDateTime;
+import org.bson.BsonDecimal128;
+import org.bson.BsonDocument;
+import org.bson.BsonDouble;
+import org.bson.BsonInt32;
+import org.bson.BsonInt64;
+import org.bson.BsonNull;
+import org.bson.BsonString;
+import org.bson.BsonType;
+import org.bson.BsonUndefined;
+import org.bson.BsonValue;
+import org.bson.conversions.Bson;
+import org.bson.types.Decimal128;
+
+import java.math.BigDecimal;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * Visitor that convert Expression to Bson filter. Return {@link Filters#empty()} if we cannot push
+ * down the filter.
+ */
+@Experimental
+public class MongoFilterPushDownVisitor extends ExpressionDefaultVisitor<BsonValue> {
+
+ public static final MongoFilterPushDownVisitor INSTANCE = new MongoFilterPushDownVisitor();
+
+ private MongoFilterPushDownVisitor() {
+ // Singleton instance.
+ }
+
+ @Override
+ public BsonDocument visit(CallExpression call) {
+ Bson filter = Filters.empty();
+ if (BuiltInFunctionDefinitions.EQUALS.equals(call.getFunctionDefinition())) {
Review Comment:
I'm afraid we can't use switch here.
> Incompatible types. Found: 'org.apache.flink.table.functions.FunctionDefinition', required: 'char, byte, short, int, Character, Byte, Short, Integer, String, or an enum'
--
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-33257][connectors/mongodb] Support filter pushdown in MongoDB connector [flink-connector-mongodb]
Posted by "Jiabao-Sun (via GitHub)" <gi...@apache.org>.
Jiabao-Sun commented on PR #17:
URL: https://github.com/apache/flink-connector-mongodb/pull/17#issuecomment-1782818570
> Thanks for your contribution. Overall LGTM. I left some minor comments here.
>
> BTW, I think it's better to add more tests for plan to verify these filters can be pushed down on the plan as expected.
Thanks @xuyangzhong for the hard review.
I'll add some add more tests for plan later.
--
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