Kafka-connect-jdbc: Support to use UUID as incrementing.column.name for incrementing mode

[BaggaShivanshu]

Hi,
I'm using JDBC source connector to connect to our PostgreSQL based database(Aurora).I convert UUID to varchar because there is no support yet for UUID.
"query":"select m_id::varchar,m_created from member_ds.member", "mode":"timestamp+incrementing", "incrementing.column.name":"m_id", "timestamp.column.name": "m_created",
With above configuration system failed to execute JDBC query with following error.

org.postgresql.util.PSQLException: ERROR: operator does not exist: uuid > bigint
  Hint: No operator matches the given name and argument type(s). You might need to add explicit type casts.
  Position: 253
    at org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2182)
    at org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:1911)

Is there any way to use UUID as incrementing.column.name for incrementing mode?If No, can we provide support for that?

[mariusneo]

As described here: https://docs.confluent.io/3.0.0/connect/connect-jdbc/docs/jdbc_connector.html#incremental-query-modes

the two modes:

• Incrementing Column
• Timestamp and Incrementing Columns

both depend on an incrementing column.
UUID is not an incrementing columnt (e.g. : AUTOINCREMENT is an incrementing column).

I don't see a solution with kafka-connect-jdbc for your usecase.

[gharris1727]

The incrementing nature of the row is necessary for the incrementing and timestamp+incrementing modes, because it allows the connector to exclude rows that have already been read. In order to use a uuid or any other type as a suitable replacement for the incrementing column, we would need an order on that type, so that we can read through . This happens to exist for UUIDs in PostgreSQL, but may not exist in general for other types and other databases.

I think this could be implemented in multiple different ways:

1. Have the database naively cast the long to the target type (probably won't work even for this case)
2. Add a DatabaseDialect method to cast the long to the target type, with a no-op in GenericDatabaseDialect, and some sort of cast long -> uuid in PostgreSqlDatabaseDialect. (inherently limited to 64-bit precision of a long, which is less than the UUIDv4 precision of 128 bits)
3. Make the incrementing offset more generic (Object), and add DatabaseDialect methods to start and advance the offset for the specific database and type. GenericDatabaseDialect would only support long, and the PostgreSQLDatabaseDialect would increment custom logic for uuid.

I think option 3 is the "right" solution, but requires a decent refactor of the existing code. We don't have any plans of implementing this right now, but we would accept and review a contribution for this feature.

[Schmidt96u]

Somebody is still on this Issue ?

i would be really happy to work on it !

[sunnikumar]

Somebody is still on this Issue ?

i would be really happy to work on it !

Any Update on this issue