davidradl commented on code in PR #79:
URL:
https://github.com/apache/flink-connector-jdbc/pull/79#discussion_r1425382471
##########
flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableSource.java:
##########
@@ -96,28 +98,70 @@ 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++) {
+
+ for (int i = 0; i < context.getKeys().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()];
+ for (int i = 0; i < this.resolvedPredicates.size(); i++) {
+ String resolvedPredicate = this.resolvedPredicates.get(i);
+ String param = this.pushdownParams[i].toString();
+ /*
+ * 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'= ?).
+ */
+ conditions[i] = resolvePredicateParam(resolvedPredicate,
param);
Review Comment:
@snuyanzin I am reading JdbcFilterPushdownPreparedStatementVisitor and it
looks like it only attempts to pushdown simple join expressions; in which case
this code is ok (as that was my assumption). I will test to check my assumption
is correct that a complex join will not be pushed down. If this code needs to
handle the complex joins , it would need to more sophisticated.
##########
flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableSource.java:
##########
@@ -96,28 +98,70 @@ 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++) {
+
+ for (int i = 0; i < context.getKeys().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()];
+ for (int i = 0; i < this.resolvedPredicates.size(); i++) {
+ String resolvedPredicate = this.resolvedPredicates.get(i);
+ String param = this.pushdownParams[i].toString();
+ /*
+ * 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'= ?).
+ */
+ conditions[i] = resolvePredicateParam(resolvedPredicate,
param);
Review Comment:
@snuyanzin I am reading JdbcFilterPushdownPreparedStatementVisitor and it
looks like it only attempts to pushdown simple join expressions; in which case
this code is ok (as that was my assumption). I will test to check my assumption
is correct that a complex join will not be pushed down. If this code needs to
handle the complex joins , it would need to be more sophisticated.
--
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
