davidradl commented on code in PR #79:
URL:
https://github.com/apache/flink-connector-jdbc/pull/79#discussion_r1438684576
##########
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:
@snuyanzin Yes I had not thought of that. I have reviewed this. I suspect I
will need to be aware of what part of the statement might be a column name and
not attempt to replace ?'s for the column name. This would mean parsing the
statement for the operations used in
JdbcFilterPushdownPreparedStatementVisitor, so I can recognise the binary
operation left and right hand side of the statement accounting for the OR
operation which could link binary operations. Also the logic needs to work with
operation names in the column names. I need to check if a single quote is
allowed in a column name. Within the current design this seems the only
option. I will put up a change along these lines. I would appreciate your
thoughts. I think the existing logic could be applied to the right hand sides
of the binary 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
