[
https://issues.apache.org/jira/browse/FLINK-15728?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17021042#comment-17021042
]
Jark Wu commented on FLINK-15728:
---------------------------------
Hi [~dasbh], thanks for reporting this.
First of all, I think this is a bug. The default implemenation of
{{JDBCDialect#getUpdateStatement}} is problematic. If we add a
{{'connector.write.flush.max-rows'='1'}} property to
{{JDBCUpsertTableSinkITCase#testUpsert}}, the result is incorrect.
Regarding to the fixing, I agree with [~lzljs3620320], it's not a small change.
The root problem is that the current statement is based on index, not on name.
We have a similar discussion to improve this issue in this
[PR|https://github.com/apache/flink/pull/9335#discussion_r311845039]. An
solution would be
[named-statement|https://www.javaworld.com/article/2077706/named-parameters-for-preparedstatement.html]
and I think it's the time to do it now.
> JDBCUpsertOutputFormat does not set bind parameter keyFields in
> updateStatement
> -------------------------------------------------------------------------------
>
> Key: FLINK-15728
> URL: https://issues.apache.org/jira/browse/FLINK-15728
> Project: Flink
> Issue Type: Bug
> Components: Connectors / JDBC
> Affects Versions: 1.9.1
> Reporter: Bhagavan
> Assignee: Bhagavan
> Priority: Major
>
> When using JDBCUpsertOutputFormat custom dialect e.g. H2/Oracle which usesÂ
> UpsertWriterUsingInsertUpdateStatement, code fails with below error.
> {code:java}
> Caused by: org.h2.jdbc.JdbcSQLDataException: Parameter "#6" is not set
> [90012-200]
> at org.h2.message.DbException.getJdbcSQLException(DbException.java:590)
> at org.h2.message.DbException.getJdbcSQLException(DbException.java:429)
> at org.h2.message.DbException.get(DbException.java:205)
> at org.h2.message.DbException.get(DbException.java:181)
> at org.h2.expression.Parameter.checkSet(Parameter.java:83)
> at
> org.h2.jdbc.JdbcPreparedStatement.addBatch(JdbcPreparedStatement.java:1275)
> at
> org.apache.flink.api.java.io.jdbc.writer.UpsertWriter$UpsertWriterUsingInsertUpdateStatement.processOneRowInBatch(UpsertWriter.java:233)
>
> at
> org.apache.flink.api.java.io.jdbc.writer.UpsertWriter.executeBatch(UpsertWriter.java:111)
> {code}
> This is due to UpsertWriterUsingInsertUpdateStatement#processOneRowInBatch
> does not set all bind paramters in case of Update.
> This bug does get surfaced while using Derby DB.
> In JDBCUpsertOutputFormatTest if we replace Derby with H2 we can reproduce
> the bug.
> The fix is trivial. Happy to raise PR.
> {code:java}
> //for update case replace below
> setRecordToStatement(updateStatement, fieldTypes, row);
> //with
> setRecordToStatement(updateStatement, fieldTypes + pkTypes, row + pkRow);
> //NOTE: as prepared updateStatement contains additional where clause we need
> pass additional bind values and its sql Types
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
