Debezium Kafka Connect从Oracle获取Schema时元数据查询失败

问题详情 #

创建Debezium Oracle Kafka连接器时，执行内置元数据查询报错ORA-01427: single-row subquery returns more than one row，但手动在Oracle数据库中执行相同查询却能正常运行，怀疑问题出在查询关联ALL_TYPES表的子查询上。

连接器配置 #

{
"name": "ORACLE_DEBEZIUM_CONNECTOR_TRANS_V4_SIT",  
"config": {
"connector.class"               : "io.debezium.connector.oracle.OracleConnector",
"database.hostname"             : "xx.xxx.xx.xx",
"database.port"                 : "1521",
"database.user"                 : "dummyuser",
"database.password"             : "dummypwd",
"database.dbname"               : "UATDB",
"database.server.name"          : "UAT_SCHEMA",
"key.converter"                 : "org.apache.kafka.connect.storage.StringConverter",
"key.converter.schema.registry.url" : "http://localhost:8081",
"snapshot.mode"                 : "schema_only",
"log.mining.archive.destination.name" : "LOG_ARCHIVE_DEST_1",
"heartbeat.interval.ms"         : "1000",
"tasks.max"                     : "1",
"time.precision.mode"           : "connect",
"decimal.handling.mode"         : "double",
"database.history.skip.unparseable.ddl" : "true",
"table.include.list"            : "UAT_SCHEMA.Tablename",
"column.include.list"           : "UAT_SCHEMA.Tablename.Col1, UAT_SCHEMA.Tablename.Col2",
"database.history.kafka.bootstrap.servers" : "localhost:9092",
"database.history.kafka.topic"  : "kaf_topic_history_sit_v4",
"transforms"                    : "route",
"transforms.route.type"         : "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.route.regex"        : "(\\[^.\\]+)\\.(\\[^.\\]+)\\.(\\[^.\\]+)",
"transforms.route.replacement"  : "KAF_TOPIC_SIT_V4\\_$3"
}
}

已尝试操作 #

• 提取连接器执行的元数据查询语句，手动在Oracle数据库中执行，运行正常，未返回多行结果。

报错关联的元数据查询 #

SELECT 
    NULL AS table_cat,
    t.owner AS table_schem,
    t.table_name AS table_name,
    t.column_name AS column_name,
    DECODE(
          (SELECT a.typecode FROM ALL_TYPES a WHERE a.type_name = t.data_type),
           'OBJECT', 2002, 
           'COLLECTION', 2003,
           DECODE(
                  SUBSTR(t.data_type, 1, 9), 
                  'TIMESTAMP', 
                  DECODE(SUBSTR(t.data_type, 10, 1), 
                         '(', 
                         DECODE(SUBSTR(t.data_type, 19, 5), 
                                'LOCAL', -102, 'TIME ', -101, 93), 
                         DECODE(SUBSTR(t.data_type, 16, 5), 
                                'LOCAL', -102, 'TIME ', -101, 93)),
                  'INTERVAL ', 
                  DECODE(SUBSTR(t.data_type, 10, 3), 
                         'DAY', -104, 'YEA', -103),
                  DECODE(t.data_type, 
                         'BINARY_DOUBLE', 101, 
                         'BINARY_FLOAT', 100, 
                         'BFILE', -13, 
                         'BLOB', 2004, 
                         'CHAR', 1, 
                         'CLOB', 2005, 
                         'COLLECTION', 2003, 
                         'DATE', 93, 
                         'FLOAT', 6, 
                         'LONG', -1, 
                         'LONG RAW', -4, 
                         'NCHAR', -15, 
                         'NCLOB', 2011, 
                         'NUMBER', 2, 
                         'NVARCHAR', -9, 
                         'NVARCHAR2', -9, 
                         'OBJECT', 2002, 
                         'OPAQUE/XMLTYPE', 2009, 
                         'RAW', -3, 
                         'REF', 2006, 
                         'ROWID', -8, 
                         'SQLXML', 2009, 
                         'UROWID', -8, 
                         'VARCHAR2', 12, 
                         'VARRAY', 2003, 
                         'XMLTYPE', 2009, 
                         1111)
                 )
               ) AS data_type,
    t.data_type AS type_name,
    DECODE(t.data_precision, 
           NULL, DECODE(t.data_type, 
                        'NUMBER', DECODE(t.data_scale, NULL, 0, 38), 
                        DECODE(t.data_type, 
                               'CHAR', t.char_length, 
                               'VARCHAR', t.char_length, 
                               'VARCHAR2', t.char_length, 
                               'NVARCHAR2', t.char_length, 
                               'NCHAR', t.char_length, 
                               'NUMBER', 0, 
                               t.data_length)), 
           t.data_precision) AS column_size,
    0 AS buffer_length,
    DECODE(t.data_type, 
           'NUMBER', DECODE(t.data_precision, 
                            NULL, DECODE(t.data_scale, NULL, -127, t.data_scale), 
                            t.data_scale), 
           t.data_scale) AS decimal_digits,
    10 AS num_prec_radix,
    DECODE(t.nullable, 'N', 0, 1) AS nullable,
    NULL AS remarks,
    t.data_default AS column_def,
    0 AS sql_data_type,
    0 AS sql_datetime_sub,
    t.data_length AS char_octet_length,
    t.column_id AS ordinal_position,
    DECODE(t.nullable, 'N', 'NO', 'YES') AS is_nullable,
    NULL AS scope_catalog,
    NULL AS scope_schema,
    NULL AS scope_table,
    NULL AS source_data_type,
    'NO' AS is_autoincrement,
    t.virtual_column AS is_generatedcolumn
FROM all_tab_cols t
WHERE t.owner LIKE 'UAT_SCHEMA'
  AND t.table_name LIKE 'UAT_SCHEMA.Tablename'
  AND t.user_generated = 'YES'
ORDER BY table_schem, table_name, ordinal_position;

解决方案 #

1. 修复子查询的多行匹配问题 #

报错根源是子查询(SELECT a.typecode FROM ALL_TYPES a WHERE a.type_name = t.data_type)返回了多条记录——ALL_TYPES中可能存在不同schema下同名的自定义类型，导致匹配到多条数据。

修改子查询，增加schema过滤，确保只返回当前表所属schema的类型：

(SELECT a.typecode FROM ALL_TYPES a WHERE a.type_name = t.data_type AND a.owner = t.owner)

2. 升级连接器或调整权限 #

• 升级连接器：将Debezium Oracle连接器升级到最新稳定版本，该问题可能已在新版本中被修复（部分旧版本存在此查询逻辑缺陷）。
• 调整数据库权限：给连接器使用的数据库账号赋予DBA权限，避免因权限差异导致ALL_TYPES查询结果范围与手动执行不一致。

3. 检查自定义类型冲突 #

确认目标表UAT_SCHEMA.Tablename中使用的自定义类型是否在多个schema下重名。如果存在重名类型，建议重命名类型以避免冲突，或者确保查询时明确指定schema。

内容的提问来源于stack exchange，提问作者Gaus Shaik