codingapi · xlorne · Sep 22, 2026 · Sep 22, 2026 · Sep 22, 2026
diff --git a/pom.xml b/pom.xml
@@ -4,7 +4,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.codingapi.dbstream</groupId>
     <artifactId>dbstream-driver</artifactId>
-    <version>1.0.19</version>
+    <version>1.0.20</version>
 
     <url>https://github.com/codingapi/dbstream-driver</url>
     <name>dbstream-driver</name>

diff --git a/src/main/java/com/codingapi/dbstream/parser/DeleteDBEventParser.java b/src/main/java/com/codingapi/dbstream/parser/DeleteDBEventParser.java
@@ -40,7 +40,12 @@ private void updateRows() throws SQLException {
         String query = this.loadUpdateRowSQL();
         List<Object> params = this.loadUpdateRowParamList();
         prepareList.clear();
-        prepareList.addAll(this.executeState.query(query, params));
+        try {
+            prepareList.addAll(this.executeState.query(query, params));
+        } catch (SQLException e) {
+            // dbstream 内部 SQL 不经过业务侧的 SQL 日志，失败时必须带上原文，否则报错会归因到业务 SQL
+            throw new SQLException("dbstream 前镜像查询失败, sql=" + query, e);
+        }
     }
 
     /**

diff --git a/src/main/java/com/codingapi/dbstream/parser/UpdateDBEventParser.java b/src/main/java/com/codingapi/dbstream/parser/UpdateDBEventParser.java
@@ -40,7 +40,12 @@ public void prepare() throws SQLException {
     private void updateRows() throws SQLException {
         String query = this.loadUpdateRowSQL();
         List<Object> params = this.loadUpdateRowParamList();
-        prepareList = this.executeState.query(query, params);
+        try {
+            prepareList = this.executeState.query(query, params);
+        } catch (SQLException e) {
+            // dbstream 内部 SQL 不经过业务侧的 SQL 日志，失败时必须带上原文，否则报错会归因到业务 SQL
+            throw new SQLException("dbstream 前镜像查询失败, sql=" + query, e);
+        }
     }
 
     /**
@@ -101,7 +106,11 @@ private List<Object> loadUpdateRowParamList() {
 
     private List<Map<String, Object>> queryLatestData() throws SQLException {
         String sql = this.latestSQL();
-        return this.executeState.query(sql, new ArrayList<>());
+        try {
+            return this.executeState.query(sql, new ArrayList<>());
+        } catch (SQLException e) {
+            throw new SQLException("dbstream 后镜像查询失败, sql=" + sql, e);
+        }
     }
 
 
@@ -117,15 +126,20 @@ private String latestSQL() {
         querySQL.append(String.join(",", columns));
         querySQL.append(" FROM ").append(this.dbTable.getName());
         querySQL.append(" WHERE ");
+        List<String> conditions = new ArrayList<>();
         for (String primaryKey : this.dbTable.getPrimaryKeys()) {
-            querySQL.append(" ").append(primaryKey);
-            querySQL.append(" IN (");
             List<String> params = this.getPrimaryKeyStringValue(primaryKey);
-            querySQL.append(String.join(",", params));
-            querySQL.append(")");
-            querySQL.append(" AND ");
+            // 主键值列表为空时拼出的 "IN ()" 是非法 SQL，直接跳过该条件
+            if (params.isEmpty()) {
+                continue;
+            }
+            conditions.add(primaryKey + " IN (" + String.join(",", params) + ")");
+        }
+        if (conditions.isEmpty()) {
+            // 没有任何可用的主键值时返回空结果集，避免退化为全表扫描
+            conditions.add("1=0");
         }
-        querySQL.append(" 1=1 ");
+        querySQL.append(String.join(" AND ", conditions));
         return querySQL.toString();
     }
 
@@ -139,10 +153,15 @@ private List<String> getPrimaryKeyStringValue(String primaryKey) {
             for (String key : data.keySet()) {
                 if (key.equalsIgnoreCase(primaryKey)) {
                     Object value = data.get(key);
+                    // 主键理论上非空，取到 null 说明该行数据异常，跳过而不是拼出 "IN (null)"
+                    if (value == null) {
+                        continue;
+                    }
                     if (value instanceof String) {
-                        params.add(String.format("'%s'", value));
+                        // 字符串值需转义单引号，否则值本身含引号时会拼出非法 SQL
+                        params.add("'" + ((String) value).replace("'", "''") + "'");
                     } else {
-                        params.add(data.get(key).toString());
+                        params.add(String.valueOf(value));
                     }
                 }
             }
@@ -160,6 +179,10 @@ public List<DBEvent> loadEvents(Object result) throws SQLException {
         }
         String jdbcUrl = this.executeState.getJdbcUrl();
         String jdbcKey = this.executeState.getJdbcKey();
+        // 前镜像为空说明没有可按主键关联的行，无法定位后镜像，直接返回空事件
+        if (this.prepareList.isEmpty()) {
+            return eventList;
+        }
         // 根据id查询最新的数据
         List<Map<String, Object>> latestData = this.queryLatestData();
         for (Map<String, Object> params : latestData) {

diff --git a/src/main/java/com/codingapi/dbstream/parser/UpdateSQLParser.java b/src/main/java/com/codingapi/dbstream/parser/UpdateSQLParser.java
@@ -64,6 +64,9 @@ public String getTableAlias() {
     /**
      * 提取 SET 后面的字段名
      * 示例: SET name=?, age=?, updated_at=NOW()
+     * <p>
+     * SET 项之间才用逗号分隔，函数调用的参数逗号（如 {@code replace(col, ?, ?)}）不参与分割，
+     * 因此按括号深度与字符串状态切分；解析不出 {@code 列名=表达式} 结构的片段直接跳过。
      */
     public List<String> getColumnValues() {
         List<String> columns = new ArrayList<>();
@@ -74,23 +77,24 @@ public List<String> getColumnValues() {
         if (matcher.find()) {
             String setPart = matcher.group(1).trim();
 
-            // 分割字段赋值（按逗号分割）
-            String[] parts = setPart.split(",");
-            for (String part : parts) {
-                String[] kv = part.split("=", 2);
-                if (kv.length > 0) {
-                    String col = kv[0].trim();
-
-                    // 如果字段是带表别名的，如 "u.name"
-                    int dotIndex = col.lastIndexOf('.');
-                    if (dotIndex > 0) {
-                        col = col.substring(dotIndex + 1);
-                    }
-
-                    col = SQLUtils.stripQuotes(col);
-                    if (!col.isEmpty()) {
-                        columns.add(col);
-                    }
+            // 按顶层逗号切分字段赋值
+            for (String part : SQLUtils.splitTopLevel(setPart, ',')) {
+                // SET 项形如 "列名 = 表达式"，取第一个等号左侧为列名
+                int eqIndex = part.indexOf('=');
+                if (eqIndex < 0) {
+                    continue;
+                }
+                String col = part.substring(0, eqIndex).trim();
+
+                // 如果字段是带表别名的，如 "u.name"
+                int dotIndex = col.lastIndexOf('.');
+                if (dotIndex > 0) {
+                    col = col.substring(dotIndex + 1);
+                }
+
+                col = SQLUtils.stripQuotes(col);
+                if (!col.isEmpty()) {
+                    columns.add(col);
                 }
             }
         }

diff --git a/src/main/java/com/codingapi/dbstream/utils/SQLUtils.java b/src/main/java/com/codingapi/dbstream/utils/SQLUtils.java
@@ -81,27 +81,36 @@ public static String getWhereSQL(String sql) {
     }
 
     /**
-     * 提取insert语句中values的内容
+     * 按分隔符切分 SQL 片段，仅在「括号深度为 0 且不在字符串字面量内」时切分。
+     * <p>
+     * 用于处理函数调用的参数逗号，例如 {@code replace(col, ?, ?)} 中的逗号不应被视为
+     * 列分隔符。支持 {@code ''} 转义的单引号，片段内容原样保留（不做转义还原）。
      */
-    public static List<String> parseInsertSQLValues(String sqlValues) {
+    public static List<String> splitTopLevel(String sql, char delimiter) {
         List<String> result = new ArrayList<>();
-        StringBuilder current = new StringBuilder();
+        if (sql == null || sql.isEmpty()) {
+            return result;
+        }
 
+        StringBuilder current = new StringBuilder();
         boolean inString = false;
         int parenDepth = 0;
 
-        for (int i = 0; i < sqlValues.length(); i++) {
-            char c = sqlValues.charAt(i);
+        for (int i = 0; i < sql.length(); i++) {
+            char c = sql.charAt(i);
 
             if (c == '\'') {
-                // 切换字符串状态（需处理转义的单引号）
-                if (inString && i + 1 < sqlValues.length() && sqlValues.charAt(i + 1) == '\'') {
-                    // SQL转义 '' -> '
-                    current.append('\'');
-                    i++; // 跳过下一个引号
+                current.append(c);
+                if (inString) {
+                    // 字符串内的 '' 为转义的单引号，仍处于字符串内
+                    if (i + 1 < sql.length() && sql.charAt(i + 1) == '\'') {
+                        current.append('\'');
+                        i++; // 跳过下一个引号
+                    } else {
+                        inString = false;
+                    }
                 } else {
-                    inString = !inString;
-                    current.append(c);
+                    inString = true;
                 }
             } else if (!inString) {
                 if (c == '(') {
@@ -110,7 +119,7 @@ public static List<String> parseInsertSQLValues(String sqlValues) {
                 } else if (c == ')') {
                     parenDepth--;
                     current.append(c);
-                } else if (c == ',' && parenDepth == 0) {
+                } else if (c == delimiter && parenDepth == 0) {
                     // 只有在括号层为0时才分割
                     result.add(current.toString().trim());
                     current.setLength(0);
@@ -129,6 +138,13 @@ public static List<String> parseInsertSQLValues(String sqlValues) {
         return result;
     }
 
+    /**
+     * 提取insert语句中values的内容
+     */
+    public static List<String> parseInsertSQLValues(String sqlValues) {
+        return splitTopLevel(sqlValues, ',');
+    }
+
     /**
      * 判断是否为常见 SQL 关键字
      */

diff --git a/src/test/java/com/codingapi/dbstream/sqlparser/UpdateSQLParserTest.java b/src/test/java/com/codingapi/dbstream/sqlparser/UpdateSQLParserTest.java
@@ -1,10 +1,12 @@
 package com.codingapi.dbstream.sqlparser;
 
 import com.codingapi.dbstream.parser.UpdateSQLParser;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.CsvFileSource;
 
 import java.util.Arrays;
+import java.util.Collections;
 
 import static org.junit.jupiter.api.Assertions.*;
 
@@ -20,4 +22,38 @@ void update(String sql, String table, String alias, String columns, String where
         assertEquals(sqlParser.getWhereSQL(), where);
 
     }
+
+    /**
+     * issue #9：SET 子句含函数调用时，函数参数里的逗号被误判为列分隔符，
+     * 解析出 "?"、"?)" 等伪列名，最终拼出非法 SQL。
+     */
+    @Test
+    void setClauseWithFunctionCall() {
+        UpdateSQLParser sqlParser = new UpdateSQLParser(
+                "update biz_pbm_organization set short_name_tair=replace(short_name_tair,?,?) " +
+                        "where system_code like (?||'%') and short_name_tair like (?||'%') and id<>? and sys_deleted=0");
+
+        assertEquals(sqlParser.getTableName(), "biz_pbm_organization");
+        assertEquals(sqlParser.getColumnValues(), Collections.singletonList("short_name_tair"));
+        assertEquals(sqlParser.getWhereSQL(),
+                "system_code like (?||'%') and short_name_tair like (?||'%') and id<>? and sys_deleted=0");
+    }
+
+    /**
+     * 无等号的片段（SET 项解析异常）不应被当成列名
+     */
+    @Test
+    void setClauseWithoutAssignShouldBeIgnored() {
+        UpdateSQLParser sqlParser = new UpdateSQLParser("update user set name=?, age");
+        assertEquals(sqlParser.getColumnValues(), Collections.singletonList("name"));
+    }
+
+    /**
+     * 字符串字面量里的逗号与等号不应干扰列名提取
+     */
+    @Test
+    void setClauseWithCommaInStringLiteral() {
+        UpdateSQLParser sqlParser = new UpdateSQLParser("update user set info='a,b=c', name=?");
+        assertEquals(sqlParser.getColumnValues(), Arrays.asList("info", "name"));
+    }
 }
diff --git a/src/test/resources/com/codingapi/dbstream/sqlparser/update.sql.csv b/src/test/resources/com/codingapi/dbstream/sqlparser/update.sql.csv
@@ -3,4 +3,10 @@ UPDATE user SET name='Tom' WHERE id=1|user||name|id=1
 UPDATE user u SET name='Tom', age=? WHERE id=?|user|u|name,age|id=?
 UPDATE [user] AS t SET t.name=?, t.age=?|user|t|name,age|
 UPDATE schema.user AS u SET u.age=u.age+1|schema.user|u|age|
-UPDATE `user_info` SET nickname=?,email=?|user_info||nickname,email|
+UPDATE `user_info` SET nickname=?,email=?|user_info||nickname,email|
+UPDATE biz_pbm_organization SET short_name_tair=replace(short_name_tair,?,?) WHERE id<>? AND sys_deleted=0|biz_pbm_organization||short_name_tair|id<>? AND sys_deleted=0
+UPDATE user SET name=replace(name,?,?), age=? WHERE id=?|user||name,age|id=?
+UPDATE user SET name=concat(first_name, ',', last_name), age=? WHERE id=?|user||name,age|id=?
+UPDATE user SET amount=coalesce(amount, 0), note=?, updated_at=NOW()|user||amount,note,updated_at|
+UPDATE user AS u SET u.name=upper(u.name), u.age=substr(u.age, 1, 2) WHERE u.id=?|user|u|name,age|u.id=?
+UPDATE user SET info='a,b=c', name=?|user||info,name|