From 5214257cd42801a8ec08155820d6cadf8c7ae4c3 Mon Sep 17 00:00:00 2001 From: rewerma Date: Mon, 16 Sep 2019 09:47:02 +0800 Subject: [PATCH] =?UTF-8?q?ES=E5=AD=90=E8=A1=A8=E5=90=ABgroup=E7=9A=84wher?= =?UTF-8?q?e=E6=9D=A1=E4=BB=B6=E6=8B=BC=E6=8E=A5=20(#2190)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * ES子表含group的where条件拼接 * modify --- .../client/adapter/es/config/SqlParser.java | 55 ++++++++++++++++--- .../adapter/es/service/ESSyncService.java | 34 +++++++++--- 2 files changed, 75 insertions(+), 14 deletions(-) diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java index b5f12b39..6f81f532 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java @@ -10,11 +10,7 @@ import java.util.stream.Collectors; import com.alibaba.fastsql.sql.SQLUtils; import com.alibaba.fastsql.sql.ast.SQLExpr; import com.alibaba.fastsql.sql.ast.expr.*; -import com.alibaba.fastsql.sql.ast.statement.SQLExprTableSource; -import com.alibaba.fastsql.sql.ast.statement.SQLJoinTableSource; -import com.alibaba.fastsql.sql.ast.statement.SQLSelectStatement; -import com.alibaba.fastsql.sql.ast.statement.SQLSubqueryTableSource; -import com.alibaba.fastsql.sql.ast.statement.SQLTableSource; +import com.alibaba.fastsql.sql.ast.statement.*; import com.alibaba.fastsql.sql.dialect.mysql.ast.statement.MySqlSelectQueryBlock; import com.alibaba.fastsql.sql.dialect.mysql.parser.MySqlStatementParser; import com.alibaba.fastsql.sql.parser.ParserException; @@ -125,8 +121,8 @@ public class SqlParser { } else if (expr instanceof SQLCaseExpr) { SQLCaseExpr sqlCaseExpr = (SQLCaseExpr) expr; fieldItem.setMethod(true); - sqlCaseExpr.getItems().forEach(item-> visitColumn(item.getConditionExpr(), fieldItem)); - }else if(expr instanceof SQLCharExpr) { + sqlCaseExpr.getItems().forEach(item -> visitColumn(item.getConditionExpr(), fieldItem)); + } else if (expr instanceof SQLCharExpr) { SQLCharExpr sqlCharExpr = (SQLCharExpr) expr; String owner = null; String columnName = null; @@ -232,4 +228,49 @@ public class SqlParser { throw new UnsupportedOperationException("Unsupported for complex of on-condition"); } } + + public static MySqlSelectQueryBlock parseSQLSelectQueryBlock(String sql) { + if (sql == null || "".equals(sql)) { + return null; + } + SQLStatementParser parser = new MySqlStatementParser(sql); + SQLSelectStatement statement = (SQLSelectStatement) parser.parseStatement(); + return (MySqlSelectQueryBlock) statement.getSelect().getQuery(); + } + + public static String parse4SQLSelectItem(MySqlSelectQueryBlock sqlSelectQueryBlock) { + List selectItems = sqlSelectQueryBlock.getSelectList(); + StringBuilder subSql = new StringBuilder(); + int i = 0; + for (SQLSelectItem sqlSelectItem : selectItems) { + if (i != 0) { + subSql.append(","); + } else { + i++; + } + subSql.append(SQLUtils.toMySqlString(sqlSelectItem)); + } + return subSql.toString(); + } + + public static String parse4FromTableSource(MySqlSelectQueryBlock sqlSelectQueryBlock) { + SQLTableSource sqlTableSource = sqlSelectQueryBlock.getFrom(); + return SQLUtils.toMySqlString(sqlTableSource); + } + + public static String parse4WhereItem(MySqlSelectQueryBlock sqlSelectQueryBlock) { + SQLExpr sqlExpr = sqlSelectQueryBlock.getWhere(); + if (sqlExpr != null) { + return SQLUtils.toMySqlString(sqlExpr); + } + return null; + } + + public static String parse4GroupBy(MySqlSelectQueryBlock sqlSelectQueryBlock) { + SQLSelectGroupByClause expr = sqlSelectQueryBlock.getGroupBy(); + if (expr != null) { + return SQLUtils.toMySqlString(expr); + } + return null; + } } diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java index 85a05316..4735086c 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java @@ -4,6 +4,8 @@ import java.util.*; import javax.sql.DataSource; +import com.alibaba.fastsql.sql.dialect.mysql.ast.statement.MySqlSelectQueryBlock; +import com.alibaba.otter.canal.client.adapter.es.config.SqlParser; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -259,7 +261,7 @@ public class ESSyncService { // 判断主键和所更新的字段是否全为简单字段 if (idFieldSimple && allUpdateFieldSimple && !fkChanged) { - singleTableSimpleFiledUpdate(config,schemaItem.getMainTable().getAlias(), dml, data, old); + singleTableSimpleFiledUpdate(config, schemaItem.getMainTable().getAlias(), dml, data, old); } else { mainTableUpdate(config, dml, data, old); } @@ -583,16 +585,34 @@ public class ESSyncService { private void subTableSimpleFieldOperation(ESSyncConfig config, Dml dml, Map data, Map old, TableItem tableItem) { ESMapping mapping = config.getEsMapping(); - StringBuilder sql = new StringBuilder( - "SELECT * FROM (" + tableItem.getSubQuerySql() + ") " + tableItem.getAlias() + " WHERE "); + MySqlSelectQueryBlock queryBlock = SqlParser.parseSQLSelectQueryBlock(tableItem.getSubQuerySql()); + StringBuilder sql = new StringBuilder(); + sql.append("SELECT ") + .append(SqlParser.parse4SQLSelectItem(queryBlock)) + .append(" FROM ") + .append(SqlParser.parse4FromTableSource(queryBlock)); + + String whereSql = SqlParser.parse4WhereItem(queryBlock); + if (whereSql != null) { + sql.append(" WHERE ").append(whereSql); + } else { + sql.append(" WHERE 1=1 "); + } + + List values = new ArrayList<>(); for (FieldItem fkFieldItem : tableItem.getRelationTableFields().keySet()) { String columnName = fkFieldItem.getColumn().getColumnName(); Object value = esTemplate.getValFromData(mapping, data, fkFieldItem.getFieldName(), columnName); - ESSyncUtil.appendCondition(sql, value, tableItem.getAlias(), columnName); + sql.append(" AND ").append(columnName).append("=? "); + values.add(value); } - int len = sql.length(); - sql.delete(len - 5, len); + + String groupSql = SqlParser.parse4GroupBy(queryBlock); + if (groupSql != null) { + sql.append(groupSql); + } + DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); if (logger.isTraceEnabled()) { logger.trace("Join table update es index by query sql, destination:{}, table: {}, index: {}, sql: {}", @@ -601,7 +621,7 @@ public class ESSyncService { mapping.get_index(), sql.toString().replace("\n", " ")); } - Util.sqlRS(ds, sql.toString(), rs -> { + Util.sqlRS(ds, sql.toString(), values, rs -> { try { while (rs.next()) { Map esFieldData = new LinkedHashMap<>();