Kafka-connect-jdbc: MySQL identifier quoting doesn't work for qualified columns

[marioalvial]

I configured my connector with a JOIN query. This is my configuration:

   "name":"sync_jdbc_connect",
   "config":{
      "name":"sync_jdbc_connect",
      "connector.class":"io.confluent.connect.jdbc.JdbcSourceConnector",
      "tasks.max":"1",
      "connection.url":"jdbc:mysql://mysql/mock",
      "connection.user":"mock",
      "connection.password":"mock",
      "mode":"incrementing",
      "incrementing.column.name": "id",
      "query":"SELECT cards.issuer_reference, cards.external_id, cards.status, customers.mpa FROM 
                     cards JOIN customers ON (cards.customer_id = customers.id)",
      "topic.prefix":"mysql-joined-data"

Running the connector I got an error "Ambiguous column field 'id'"

Ok, it was a fair error. Both my tables have an 'id' column field. So I change my configuration:

"incrementing.column.name": "customers.id"

But I got another error: "Unknown column 'customers.id' in where clause"

So now I don't know what to do. How I identify the incrementing column name when the identifier (column) exists in both tables?

[rhauch]

@marioalvial can you use aliases on the columns in the SELECT clause, and then reference those aliases in the column names?

[marioalvial]

@rhauch sorry, I forgot to say that I already tried to use aliases... It didn't work either, I received the "Unknown column 'cus.id' in field list" error. Also, I tried to use table.types=VIEW and I created a VIEW in my MySQL database but IT didn't change the outcome, I still got the same error.

My configuration with aliases:

"incrementing.column.name": "cus.id",
"query":"SELECT c.issuer_reference, c.external_id, c.status, cus.mpa FROM cards c JOIN customers cus ON (c.customer_id = cus.id)"

[chucklin85]

Try this:

"name":"sync_jdbc_connect",
"config":{
"name":"sync_jdbc_connect",
"connector.class":"io.confluent.connect.jdbc.JdbcSourceConnector",
"tasks.max":"1",
"connection.url":"jdbc:mysql://mysql/mock",
"connection.user":"mock",
"connection.password":"mock",
"mode":"incrementing",
"incrementing.column.name": "customerid",
"query":"SELECT cards.customerid as customerid, cards.issuer_reference, cards.external_id, cards.status, customers.mpa FROM
cards JOIN customers ON (cards.customer_id = customers.id)",
"topic.prefix":"mysql-joined-data"

[marioalvial]

With this config:

"timestamp.column.name":"updatedat",

It didn't work.

ERROR Failed to run query for table TimestampIncrementingTableQuerier{name='null', query='SELECT c.a, c.b, c.c, c.updated_at as updatedat, cus.d FROM cards c JOIN customers cus ON (c.customer_id = cus.id)', java.sql.SQLSyntaxErrorException: Unknown column 'updatedat' in 'where clause'

[mihbor]

@rhauch , I believe the reason why using aliases doesn't work is that the ExpressionBuilder adds quotes around the incrementing/timestamp column names, hence "alias.column" is then interpreted as a whole as a column name and therefore fails.

[rmoff]

+1 just hit this too.

With this config:

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
        "name": "jdbc_source_mysql_10",
        "config": {
                "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
                "connection.url": "jdbc:mysql://mysql:3306/demo",
                "connection.user": "connect_user",
                "connection.password": "asgard",
                "topic.prefix": "mysql-10",
                "mode":"incrementing",
                "query":"SELECT t.id, t.customer_id, t.amount, t.currency, t.txn_timestamp, c.first_name, c.last_name, c.email, c.gender, c.comments FROM demo.transactions t LEFT OUTER JOIN demo.customers c on t.customer_id = c.id",
                "incrementing.column.name": "t.id",
                "validate.non.null": false
                }
        }'

connector fails:

java.sql.SQLSyntaxErrorException: Unknown column 't.id' in 'where clause'

Looking at the prepared statement the column is quoted:

DEBUG TimestampIncrementingTableQuerier{table=null, query='SELECT t.id, t.customer_id, t.amount, t.currency, t.txn_timestamp, c.first_name, c.last_name, c.email, c.gender, c.comments FROM demo.transactions t LEFT OUTER JOIN demo.customers c on t.customer_id = c.id', topicPrefix='mysql-10', incrementingColumn='t.id', timestampColumns=[]} prepared SQL query: SELECT t.id, t.customer_id, t.amount, t.currency, t.txn_timestamp, c.first_name, c.last_name, c.email, c.gender, c.comments FROM demo.transactions t LEFT OUTER JOIN demo.customers c on t.customer_id = c.id WHERE `t.id` > ? ORDER BY `t.id` ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier)

Running this in MySQL fails and reproduces the error seen in Kafka Connect:

mysql> SELECT t.id, t.customer_id, t.amount, t.currency, t.txn_timestamp, c.first_name, c.last_name, c.email, c.gender, c.comments FROM demo.transactions t LEFT OUTER JOIN demo.customers c on t.customer_id = c.id WHERE `t.id` > 999;
ERROR 1054 (42S22): Unknown column 't.id' in 'where clause'

Removing the quoting works:

mysql> SELECT t.id, t.customer_id, t.amount, t.currency, t.txn_timestamp, c.first_name, c.last_name, c.email, c.gender, c.comments FROM demo.transactions t LEFT OUTER JOIN demo.customers c on t.customer_id = c.id WHERE t.id > 999;
+------+-------------+--------+----------+----------------------+------------+-----------+----------------------+--------+-----------------------------------------+
| id   | customer_id | amount | currency | txn_timestamp        | first_name | last_name | email                | gender | comments                                |
+------+-------------+--------+----------+----------------------+------------+-----------+----------------------+--------+-----------------------------------------+
| 1000 |           5 |  28.15 | IRR      | 2018-01-12T14:53:49Z | Modestia   | Coltart   | [email protected] | Female | Reverse-engineered non-volatile success |
+------+-------------+--------+----------+----------------------+------------+-----------+----------------------+--------+-----------------------------------------+
1 row in set (0.00 sec)

The column _has_ to be qualified though, because both tables in the query have the id column:

mysql> SELECT t.id, t.customer_id, t.amount, t.currency, t.txn_timestamp, c.first_name, c.last_name, c.email, c.gender, c.comments FROM demo.transactions t LEFT OUTER JOIN demo.customers c on t.customer_id = c.id WHERE id > 999;
ERROR 1052 (23000): Column 'id' in where clause is ambiguous

[rmoff]

MySQL quoting is defined here: https://github.com/confluentinc/kafka-connect-jdbc/blob/master/src/main/java/io/confluent/connect/jdbc/dialect/MySqlDatabaseDialect.java#L61

This StackOverflow answer suggests the placeholders _shouldn't_ be backtick quoted in MySQL.

[edit]

This is incorrect:

WHERE `t.txn_id` > 999;

and this is correct:

WHERE `t`.`txn_id` > 999;

By wrapping the entire qualified column name string in quotes, MySQL treats it as a single column name (a column called t.txn_id, rather than a column called txn_id with a qualifier of t)

[marioalvial]

Hi @rmoff thanks for the explanation. There is a way to bypass these quotes?

[rmoff]

@marioalvial I don't know. My guess is it may be possible to change this through the MySQL dialect?

[rhauch]

@marioalvial, it is possible to bypass these quotes with #572, which was recently made and has not yet been released, though it will be in the 5.0.2, 5.1.1, and 5.2.0 releases whenever they occur. In the meantime, you can build the connector locally and try the new feature to disable quoting.

I'll mark this as closed, but feel free to reopen if you try the feature and it does not work for you.

[AlexVPopov]

Hello. I am using 5.2.1:

[2019-05-30 10:52:02,480] INFO Loading plugin from: /connectors/kafka-connect-jdbc-5.2.1 (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:222)

and this is the config of my source job:

connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
timestamp.column.name=authors_updated_at,posts_updated_at,comments_updated_at
incrementing.column.name=author_id,post_id,comment_id
connection.password=example
tasks.max=1
query=SELECT * FROM ( select posts.title, comments.body, authors.first_name, authors.last_name, authors.updated_at AS authors_updated_at, posts.updated_at AS posts_updated_at, comments.updated_at AS comments_updated_at, authors.id AS author_id, posts.id AS post_id, comments.id AS comment_id from authors inner join posts on posts.author_id = authors.id inner join comments on comments.post_id = posts.id ) AS authors_posts_comments
mode=timestamp+incrementing
quote.sql.identifier=never
topic.prefix=authors-posts-comments-1
connection.user=root
poll.interval.ms=1000
connection.url=jdbc:mysql://mysql:3306/sourcy_development

and I still get a quoting error:

2019-05-30 11:04:57,626] ERROR Failed to run query for table TimestampIncrementingTableQuerier{table=null, query='SELECT * FROM ( select posts.title, comments.body, authors.first_name, authors.last_name, authors.updated_at AS authors_updated_at, posts.updated_at AS posts_updated_at, comments.updated_at AS comments_updated_at, authors.id AS author_id, posts.id AS post_id, comments.id AS comment_id from authors inner join posts on posts.author_id = authors.id inner join comments on comments.post_id = posts.id ) AS authors_posts_comments', topicPrefix='authors-posts-comments-1', incrementingColumn='author_id,post_id,comment_id', timestampColumns=[authors_updated_at, posts_updated_at, comments_updated_at]}: {} (io.confluent.connect.jdbc.source.JdbcSourceTask:332)
java.sql.SQLSyntaxErrorException: Unknown column 'author_id,post_id,comment_id' in 'where clause'

[AlexVPopov]

This is the query kafka-connect-jdbc is trying to issue:

[2019-05-30 11:04:30,503] INFO Begin using SQL query: SELECT * FROM ( select posts.title, comments.body, authors.first_name, authors.last_name, authors.updated_at AS authors_updated_at, posts.updated_at AS posts_updated_at, comments.updated_at AS comments_updated_at, authors.id AS author_id, posts.id AS post_id, comments.id AS comment_id from authors inner join posts on posts.author_id = authors.id inner join comments on comments.post_id = posts.id ) AS authors_posts_comments WHERE COALESCE(`authors_updated_at`,`posts_updated_at`,`comments_updated_at`) < ? AND ((COALESCE(`authors_updated_at`,`posts_updated_at`,`comments_updated_at`) = ? AND `author_id,post_id,comment_id` > ?) OR COALESCE(`authors_updated_at`,`posts_updated_at`,`comments_updated_at`) > ?) ORDER BY COALESCE(`authors_updated_at`,`posts_updated_at`,`comments_updated_at`),`author_id,post_id,comment_id` ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:140)
[2019-05-30 11:04:30,510] ERROR Failed to run query for table TimestampIncrementingTableQuerier{table=null, query='SELECT * FROM ( select posts.title, comments.body, authors.first_name, authors.last_name, authors.updated_at AS authors_updated_at, posts.updated_at AS posts_updated_at, comments.updated_at AS comments_updated_at, authors.id AS author_id, posts.id AS post_id, comments.id AS comment_id from authors inner join posts on posts.author_id = authors.id inner join comments on comments.post_id = posts.id ) AS authors_posts_comments', topicPrefix='authors-posts-comments-1', incrementingColumn='author_id,post_id,comment_id', timestampColumns=[authors_updated_at, posts_updated_at, comments_updated_at]}: {} (io.confluent.connect.jdbc.source.JdbcSourceTask:332)
java.sql.SQLSyntaxErrorException: Unknown column 'author_id,post_id,comment_id' in 'where clause'

[AlexVPopov]

O.K., it appears, that it my fault, as incrementing.column.name, as opposed to timestamp.column.name does not support multiple values:

https://docs.confluent.io/current/connect/kafka-connect-jdbc/source-connector/source_config_options.html

[AlexVPopov]

For anyone coming here: the proper name of the setting is quote.sql.identifiers=never with an s at the end of indentifiers.

[sandycomp]

@marioalvial I don't know. My guess is it may be possible to change this through the MySQL dialect?

@rmoff I have similar issue on MariaDB. I have raised the issue https://github.com/confluentinc/kafka-connect-jdbc/issues/833. Please let me know what and where I'm going wrong. My observation is, error is thrown after source connector is working for 5-10 mins. I'm using version 5.4.1