jdbc - 如何在 Kafka Connect JDBC Source 连接器中添加显式 WHERE 子句

Question

我正在使用 kafka 连接到从 DB2 到 kafka 主题的源数据，并且我正在配置 sql 查询以从 DB2 读取数据，下面是查询

SELECT SEQ_I AS error_id, TRIM(SEND_I) AS sca , to_char(CREATE_TS,'YYYY-MM-DD HH24:MI:SS.FF3') AS create_timestamp, CREATE_TS, TRIM(ERR_MSG) AS error_message , CASE substr(ERR_MSG,1,locate('-',ERR_MSG)-1) WHEN 'WARNING' THEN 'W' WHEN 'SUSPENDED' THEN 'F' END ERROR_TYPE FROM INTCHG_ERROR_DIR WHERE TRAN_I ='503' AND PRCS_N = 'GLOBAL'

我在这里使用设置"timestamp.column.name": "CREATE_TS"问题是在查询中他们已经是WHERE子句，并且 kafka connect 尝试添加另一个带有时间戳列的 where 子句，它正在创建问题，另一个问题是如果我从 sql 子句中删除 where 子句，如下所示

SELECT SEQ_I AS error_id, TRIM(SEND_I) AS sca , to_char(CREATE_TS,'YYYY-MM-DD HH24:MI:SS.FF3') AS create_timestamp, CREATE_TS, TRIM(ERR_MSG) AS error_message , CASE substr(ERR_MSG,1,locate('-',ERR_MSG)-1) WHEN 'WARNING' THEN 'W' WHEN 'SUSPENDED' THEN 'F' END ERROR_TYPE FROM INTCHG_ERROR_DIR

然后我得到了 substr 错误，如下所示

SQL Error [22011]: THE SECOND OR THIRD ARGUMENT OF THE SUBSTR OR SUBSTRING FUNCTION IS OUT OF RANGE. SQLCODE=-138, SQLSTATE=22011, DRIVER=4.19.26

任何人都可以就这两个问题提出建议，我被困在这一点上。

标签： jdbcapache-kafkadb2apache-kafka-connect

Answer 1

发生这种情况是因为您尝试同时使用"mode": "timestamp"和query。TimestampIncrementingTableQuerier将一个子句附加到与 . 中现有子句WHERE冲突的查询。WHEREquery

JDBC 源连接器文档对此很清楚：

query

如果指定，则执行查询以选择新的或更新的行。如果要连接表、选择表中的列子集或过滤数据，请使用此设置。如果使用，此连接器将仅使用此查询复制数据——全表复制将被禁用。增量更新仍然可以使用不同的查询模式，但是为了正确构造增量查询，必须可以在该查询中附加 WHERE 子句（即不能使用 WHERE 子句）。如果使用 WHERE 子句，它必须自己处理增量查询。

---

作为一种解决方法，您可以将查询修改为（取决于您使用的 SQL 风格）

SELECT * FROM ( SELECT * FROM table WHERE ...)

或者

WITH a AS
   SELECT * FROM b
    WHERE ...
SELECT * FROM a

例如，在您的情况下，查询应该是

"query":"SELECT * FROM (SELECT SEQ_I AS error_id, TRIM(SEND_I) AS sca , to_char(CREATE_TS,'YYYY-MM-DD HH24:MI:SS.FF3') AS create_timestamp, CREATE_TS, TRIM(ERR_MSG) AS error_message , CASE substr(ERR_MSG,1,locate('-',ERR_MSG)-1) WHEN 'WARNING' THEN 'W' WHEN 'SUSPENDED' THEN 'F' END ERROR_TYPE FROM INTCHG_ERROR_DIR WHERE TRAN_I ='503' AND PRCS_N = 'GLOBAL') o"