snuyanzin commented on code in PR #79: URL: https://github.com/apache/flink-connector-jdbc/pull/79#discussion_r1434166658
########## flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableSource.java: ########## @@ -96,28 +97,115 @@ public JdbcDynamicTableSource( public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) { // JDBC only support non-nested look up keys String[] keyNames = new String[context.getKeys().length]; + for (int i = 0; i < keyNames.length; i++) { int[] innerKeyArr = context.getKeys()[i]; Preconditions.checkArgument( innerKeyArr.length == 1, "JDBC only support non-nested look up keys"); keyNames[i] = DataType.getFieldNames(physicalRowDataType).get(innerKeyArr[0]); } + final RowType rowType = (RowType) physicalRowDataType.getLogicalType(); + + String[] conditions = null; + + if (this.resolvedPredicates != null) { + conditions = new String[this.resolvedPredicates.size()]; + int processedPushdownParamsIndex = 0; + for (int i = 0; i < this.resolvedPredicates.size(); i++) { + String resolvedPredicate = this.resolvedPredicates.get(i); + + /* + * This replace seems like it should be using a Flink class to resolve the parameter. It does not + * effect the dialects as the placeholder comes from JdbcFilterPushdownPreparedStatementVisitor. + * + * Here is what has been considered as alternatives. + * + * We cannot use the way this is done in getScanRuntimeProvider, as the index we have is the index + * into the filters, but it needs the index into the fields. For example one lookup key and one filter + * would both have an index of 0, which the subsequent code would incorrectly resolve to the first + * field. + * We cannot use the PreparedStatement as we have not got access to the statement here. + * We cannot use ParameterizedPredicate as it takes the filter expression as input (e.g EQUALS(...) + * not the form we have here an example would be ('field1'= ?). + * + * An entry in the resolvedPredicates list may have more than one associated pushdown parameter, for example + * a query like this : ... on e.type = 2 and (e.age = 50 OR height > 90) and a.ip = e.ip; + * will have 2 resolvedPredicates and 3 pushdownParams. The 2nd and 3rd pushdownParams will be for the second + * resolvedPredicate. + * + */ + ArrayList<String> paramsForThisPredicate = new ArrayList(); + char placeholderChar = + JdbcFilterPushdownPreparedStatementVisitor.PUSHDOWN_PREDICATE_PLACEHOLDER + .charAt(0); + + int count = + (int) resolvedPredicate.chars().filter(ch -> ch == placeholderChar).count(); + + for (int j = processedPushdownParamsIndex; + j < processedPushdownParamsIndex + count; + j++) { + paramsForThisPredicate.add(this.pushdownParams[j].toString()); + } + processedPushdownParamsIndex = processedPushdownParamsIndex + count; Review Comment: I'm not sure this is a reliable approach for instance if we take the same test `org.apache.flink.connector.jdbc.table.JdbcTablePlanTest#testLookupJoinWithORFilter` and rename `age` column with``?age` like ```java util.tableEnv() .executeSql( "CREATE TABLE d ( " + "ip varchar(20), type int, ```?age` int" + ") WITH (" + " 'connector'='jdbc'," + " 'url'='jdbc:derby:memory:test1'," + " 'table-name'='d'" + ")"); ``` then apply this rename to the query in test like ```java @Test public void testLookupJoinWithORFilter() { util.verifyExecPlan( "SELECT * FROM a left join d FOR SYSTEM_TIME AS OF a.proctime on (d.```?age` = 50 OR d.type = 1) and a.ip = d.ip"); } ``` and now if we run this test it starts failing with `java.lang.ArrayIndexOutOfBoundsException: 2` like ``` Caused by: java.lang.ArrayIndexOutOfBoundsException: 2 at org.apache.flink.connector.jdbc.table.JdbcDynamicTableSource.getLookupRuntimeProvider(JdbcDynamicTableSource.java:149) at org.apache.flink.table.planner.plan.utils.LookupJoinUtil.createLookupRuntimeProvider(LookupJoinUtil.java:626) at org.apache.flink.table.planner.plan.utils.LookupJoinUtil.isAsyncLookup(LookupJoinUtil.java:413) at org.apache.flink.table.planner.plan.nodes.physical.common.CommonPhysicalLookupJoin.isAsyncEnabled$lzycompute(CommonPhysicalLookupJoin.scala:120) at org.apache.flink.table.planner.plan.nodes.physical.common.CommonPhysicalLookupJoin.isAsyncEnabled(CommonPhysicalLookupJoin.scala:116) at org.apache.flink.table.planner.plan.nodes.physical.common.CommonPhysicalLookupJoin.asyncOptions$lzycompute(CommonPhysicalLookupJoin.scala:129) at org.apache.flink.table.planner.plan.nodes.physical.common.CommonPhysicalLookupJoin.asyncOptions(CommonPhysicalLookupJoin.scala:129) at org.apache.flink.table.planner.plan.nodes.physical.common.CommonPhysicalLookupJoin.explainTerms(CommonPhysicalLookupJoin.scala:206) at org.apache.calcite.rel.AbstractRelNode.getDigestItems(AbstractRelNode.java:409) at org.apache.calcite.rel.AbstractRelNode.deepHashCode(AbstractRelNode.java:391) at org.apache.calcite.rel.AbstractRelNode$InnerRelDigest.hashCode(AbstractRelNode.java:443) at java.util.HashMap.hash(HashMap.java:340) at java.util.HashMap.get(HashMap.java:558) at org.apache.calcite.plan.volcano.VolcanoPlanner.registerImpl(VolcanoPlanner.java:1150) at org.apache.calcite.plan.volcano.VolcanoPlanner.register(VolcanoPlanner.java:589) at org.apache.calcite.plan.volcano.VolcanoPlanner.ensureRegistered(VolcanoPlanner.java:604) at org.apache.calcite.plan.volcano.VolcanoRuleCall.transformTo(VolcanoRuleCall.java:148) ``` ########## flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableSource.java: ########## @@ -96,28 +97,115 @@ public JdbcDynamicTableSource( public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) { // JDBC only support non-nested look up keys String[] keyNames = new String[context.getKeys().length]; + for (int i = 0; i < keyNames.length; i++) { int[] innerKeyArr = context.getKeys()[i]; Preconditions.checkArgument( innerKeyArr.length == 1, "JDBC only support non-nested look up keys"); keyNames[i] = DataType.getFieldNames(physicalRowDataType).get(innerKeyArr[0]); } + final RowType rowType = (RowType) physicalRowDataType.getLogicalType(); + + String[] conditions = null; + + if (this.resolvedPredicates != null) { + conditions = new String[this.resolvedPredicates.size()]; + int processedPushdownParamsIndex = 0; + for (int i = 0; i < this.resolvedPredicates.size(); i++) { + String resolvedPredicate = this.resolvedPredicates.get(i); + + /* + * This replace seems like it should be using a Flink class to resolve the parameter. It does not + * effect the dialects as the placeholder comes from JdbcFilterPushdownPreparedStatementVisitor. + * + * Here is what has been considered as alternatives. + * + * We cannot use the way this is done in getScanRuntimeProvider, as the index we have is the index + * into the filters, but it needs the index into the fields. For example one lookup key and one filter + * would both have an index of 0, which the subsequent code would incorrectly resolve to the first + * field. + * We cannot use the PreparedStatement as we have not got access to the statement here. + * We cannot use ParameterizedPredicate as it takes the filter expression as input (e.g EQUALS(...) + * not the form we have here an example would be ('field1'= ?). + * + * An entry in the resolvedPredicates list may have more than one associated pushdown parameter, for example + * a query like this : ... on e.type = 2 and (e.age = 50 OR height > 90) and a.ip = e.ip; + * will have 2 resolvedPredicates and 3 pushdownParams. The 2nd and 3rd pushdownParams will be for the second + * resolvedPredicate. + * + */ + ArrayList<String> paramsForThisPredicate = new ArrayList(); + char placeholderChar = + JdbcFilterPushdownPreparedStatementVisitor.PUSHDOWN_PREDICATE_PLACEHOLDER + .charAt(0); + + int count = + (int) resolvedPredicate.chars().filter(ch -> ch == placeholderChar).count(); + + for (int j = processedPushdownParamsIndex; + j < processedPushdownParamsIndex + count; + j++) { + paramsForThisPredicate.add(this.pushdownParams[j].toString()); + } + processedPushdownParamsIndex = processedPushdownParamsIndex + count; + + conditions[i] = + resolvePredicateWithParams( + resolvedPredicate, paramsForThisPredicate, placeholderChar); + } + } + JdbcRowDataLookupFunction lookupFunction = new JdbcRowDataLookupFunction( options, lookupMaxRetryTimes, DataType.getFieldNames(physicalRowDataType).toArray(new String[0]), DataType.getFieldDataTypes(physicalRowDataType).toArray(new DataType[0]), keyNames, - rowType); + rowType, + conditions); if (cache != null) { return PartialCachingLookupProvider.of(lookupFunction, cache); } else { return LookupFunctionProvider.of(lookupFunction); } } + /** + * Replace all instances of with the supplied params in order, as long as the characters are not + * in quotes. + * + * @param string string containing the placeholder characters + * @param params parameters to replace the the placeholder characters with. + * @return resolved predicate with parameters replacing the placeholder character + */ + @VisibleForTesting + protected static String resolvePredicateWithParams( + String string, ArrayList<String> params, char placeholderChar) { + final char quoteIdentifier = '`'; + StringBuffer resolvedPredicateWithParams = new StringBuffer(); Review Comment: `StringBuffer` is thread safe meaning it has corresponding trade offs for that however we don't have multiple threads using this var, so I think `StringBuilder` should be enough -- 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-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org