如何使用 JDBC 接收器连接器将表名转换为大写

Question

我正在 Postgres 和 Oracle 数据库之间实现基于 CDC 的复制。对于源连接器，我使用 Debezium 的 Postgres 连接器，而对于接收器连接器，我使用 Confluent 的 JDBC Sink 连接器。

在 Postgres 数据库中，所有的表都是小写的，但在 Oracle 数据库中，表都是大写的。如何在 JDBC 接收器连接器中动态设置配置，这样我就不需要在 transforms.route.replacement 中设置硬编码的大写表名？目前，如果我没有在配置中手动将目标表名称设置为大写，则连接器会添加双引号。然后，我收到以下错误：

原因：org.apache.kafka.connect.errors.ConnectException：表“TEST”.“dummy_table”丢失并且自动创建被禁用

我已禁用自动创建，因为我不需要它。

以下是我的 JDBC 接收器连接器配置示例。我省略了一些字段，如连接 url 等。如果我像这样离开配置，每次我需要添加一个新表时，我都需要创建一个新的 JDBC 接收器连接器。

{
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "topics": "test.public.dummy_table",
    "transforms": "route,unwrap",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
    "transforms.route.replacement": "TEST.DUMMY_TABLE", // I want to use TEST.$3 here, but apply uppercase. Is it possible?
    "transforms.unwrap.drop.tombstones": "false",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "pk.mode": "record_key",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "pk.fields": "id",
    "quote.sql.identifiers": "never"
}

是否可以使用 transforms.route.replacement 实现表名大写转换或告诉 sink 连接器不要添加双引号？如果没有，您能否建议如何实施？提前谢谢你。

Answer 1

我对你所描述的几乎一无所知。

然而：就 Oracle 而言，表名以大写形式存储在数据字典中，但 Oracle 允许您使用任何您想要的字母大小写（大写、小写、混合）来访问它们：

SQL> select count(*) from emp union all       --> lower
  2  select count(*) from EMP union all       --> upper
  3  select count(*) from EmP;                --> mixed

  COUNT(*)
----------
        14
        14
        14

SQL>

只要您在创建表格时不使用双引号即可；在这种情况下，每次访问时都必须使用相同的字母大小写并将表名括在双引号中：

SQL> create table "teST" (id number);

Table created.

SQL> select * from test;
select * from test
              *
ERROR at line 1:
ORA-00942: table or view does not exist


SQL> select * from TEST;
select * from TEST
              *
ERROR at line 1:
ORA-00942: table or view does not exist


SQL> select * from teST;
select * from teST
              *
ERROR at line 1:
ORA-00942: table or view does not exist


SQL> select * from "teST";

no rows selected

SQL>

所以：如果您不那么不幸地使用双引号，也许您不必费心（正如我的第一个示例所示）。只是在引用表格时不要使用双引号就可以了。

Answer 2

为避免 kafka 连接接收器中的引号问题，请尝试将 quote_sql_identifiers 设置为从不。

https://docs.confluent.io/kafka-connect-jdbc/current/sink-connector/index.html#jdbc-sink-identifier-quoting