Flink SQL JDBC Upsert模式下Oracle BLOB字段插入失败求助

数据源为MySQL，包含BLOB类型字段；通过Flink SQL以Upsert方式下沉至Oracle（对应字段类型同样为BLOB）时，BLOB字段插入失败，报错信息：

ORA-01461: Only the LONG value to be inserted into the LONG column can be bound.

对应的Flink SQL语句 #

create table t_test_table
(
    wid  STRING ,
    name STRING ,
    photos BYTES,
    PRIMARY KEY (wid) NOT ENFORCED
) WITH  (
         'connector' = 'jdbc',
          'url' = '***',
          'table-name' = 't_test_table',
          'username' = '***',
          'password' = '***'
       );

create table t_test_table_bak
(
    wid  STRING ,
    name STRING ,
    photos BYTES,
    PRIMARY KEY (wid) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:oracle:thin:...',
    'table-name' = 't_test_table_bak',
    'username' = '***',
    'password' = '***'
);
INSERT INTO t_test_table_bak SELECT wid, name, photos FROM t_test_table;

Upsert模式对应的MERGE语句 #

MERGE INTO t_test_table_bak t  
USING (SELECT :wid wid, :name name, :photos photos FROM DUAL) s  
ON (t.wid=s.wid)  
WHEN MATCHED THEN UPDATE SET t.name=s.name, t.photos=s.photos 
WHEN NOT MATCHED THEN INSERT (wid, name, photos) VALUES (s.wid, s.name, s.photos)

解决方法 #

1. 禁用Oracle批量Upsert

在Oracle表的WITH配置中添加'oracle.use-batch-upsert' = 'false'，强制Flink使用单条语句处理Upsert，避免批量绑定逻辑对BLOB字段的错误映射：

create table t_test_table_bak
(
    wid  STRING ,
    name STRING ,
    photos BYTES,
    PRIMARY KEY (wid) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:oracle:thin:...',
    'table-name' = 't_test_table_bak',
    'username' = '***',
    'password' = '***',
    'oracle.use-batch-upsert' = 'false'
);

2. 显式指定字段的JDBC类型映射

启用新版JDBC属性并为BLOB字段标注对应的Oracle类型，确保Flink正确识别：

create table t_test_table_bak
(
    wid  STRING ,
    name STRING ,
    photos BYTES /*+ OPTIONS('jdbc-type'='BLOB') */,
    PRIMARY KEY (wid) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:oracle:thin:...',
    'table-name' = 't_test_table_bak',
    'username' = '***',
    'password' = '***',
    'connector.property-version' = '2'
);

3. 调整MERGE语句的参数绑定方式

避免在USING子句中通过DUAL绑定BLOB参数，直接引用源表数据生成MERGE逻辑：

MERGE INTO t_test_table_bak t
USING (SELECT wid, name, photos FROM t_test_table) s
ON (t.wid = s.wid)
WHEN MATCHED THEN UPDATE SET t.name = s.name, t.photos = s.photos
WHEN NOT MATCHED THEN INSERT (wid, name, photos) VALUES (s.wid, s.name, s.photos)

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