fixed issue #5094 , support function unique index

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/TableMeta.java

@@ -74,6 +74,22 @@ public class TableMeta {
         throw new RuntimeException("unknow column : " + name);
     }
 
+    /**
+     * 尝试基于列名查找一下Filed信息,找不到时返回为null
+     * 
+     * @param name
+     * @return
+     */
+    public FieldMeta tryGetFieldMetaByName(String name) {
+        for (FieldMeta meta : fields) {
+            if (meta.getColumnName().equalsIgnoreCase(name)) {
+                return meta;
+            }
+        }
+
+        return null;
+    }
+
     public List<FieldMeta> getPrimaryFields() {
         List<FieldMeta> primarys = new ArrayList<>();
         for (FieldMeta meta : fields) {

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java

@@ -1,43 +1,28 @@
 package com.alibaba.otter.canal.parse.inbound.mysql.tsdb;
 
-import java.util.Arrays;
+import java.util.*;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
-import com.alibaba.druid.DbType;
-import com.alibaba.druid.sql.SQLUtils;
-import com.alibaba.druid.sql.visitor.SQLASTOutputVisitor;
-import com.alibaba.druid.sql.visitor.VisitorFeature;
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.alibaba.druid.DbType;
+import com.alibaba.druid.sql.SQLUtils;
 import com.alibaba.druid.sql.ast.SQLDataType;
 import com.alibaba.druid.sql.ast.SQLDataTypeImpl;
 import com.alibaba.druid.sql.ast.SQLExpr;
 import com.alibaba.druid.sql.ast.SQLStatement;
-import com.alibaba.druid.sql.ast.expr.SQLCharExpr;
+import com.alibaba.druid.sql.ast.expr.*;
-import com.alibaba.druid.sql.ast.expr.SQLIdentifierExpr;
+import com.alibaba.druid.sql.ast.statement.*;
-import com.alibaba.druid.sql.ast.expr.SQLMethodInvokeExpr;
-import com.alibaba.druid.sql.ast.expr.SQLNullExpr;
-import com.alibaba.druid.sql.ast.expr.SQLPropertyExpr;
-import com.alibaba.druid.sql.ast.statement.SQLColumnConstraint;
-import com.alibaba.druid.sql.ast.statement.SQLColumnDefinition;
-import com.alibaba.druid.sql.ast.statement.SQLColumnPrimaryKey;
-import com.alibaba.druid.sql.ast.statement.SQLColumnUniqueKey;
-import com.alibaba.druid.sql.ast.statement.SQLCreateTableStatement;
-import com.alibaba.druid.sql.ast.statement.SQLNotNullConstraint;
-import com.alibaba.druid.sql.ast.statement.SQLNullConstraint;
-import com.alibaba.druid.sql.ast.statement.SQLSelectOrderByItem;
-import com.alibaba.druid.sql.ast.statement.SQLTableElement;
 import com.alibaba.druid.sql.dialect.mysql.ast.MySqlPrimaryKey;
 import com.alibaba.druid.sql.dialect.mysql.ast.MySqlUnique;
 import com.alibaba.druid.sql.dialect.mysql.ast.expr.MySqlOrderingExpr;
 import com.alibaba.druid.sql.repository.Schema;
 import com.alibaba.druid.sql.repository.SchemaObject;
 import com.alibaba.druid.sql.repository.SchemaRepository;
+import com.alibaba.druid.sql.visitor.SQLASTOutputVisitor;
+import com.alibaba.druid.sql.visitor.VisitorFeature;
 import com.alibaba.druid.util.JdbcConstants;
 import com.alibaba.otter.canal.parse.inbound.TableMeta;
 import com.alibaba.otter.canal.parse.inbound.TableMeta.FieldMeta;
@@ -271,13 +256,52 @@ public class MemoryTableMeta implements TableMetaTSDB {
         } else if (element instanceof MySqlUnique) {
             MySqlUnique column = (MySqlUnique) element;
             List<SQLSelectOrderByItem> uks = column.getColumns();
+            // https://github.com/alibaba/canal/issues/5094
+            // 处理一下函数索引
+            List<String> columnNames = new ArrayList<String>();
             for (SQLSelectOrderByItem uk : uks) {
-                String name = getSqlName(uk.getExpr());
+                SQLExpr sqlName = uk.getExpr();
-                FieldMeta field = tableMeta.getFieldMetaByName(name);
+                columnNames.addAll(getIndexColumnNames(sqlName));
+            }
+            // uniqe打标
+            for (String name : columnNames) {
+                FieldMeta field = tableMeta.tryGetFieldMetaByName(name);
+                if (field != null) {
                     field.setUnique(true);
                 }
             }
         }
+    }
+
+    private List<String> getIndexColumnNames(SQLExpr expr) {
+        if (expr instanceof SQLMethodInvokeExpr) {
+            // 需要递归处理下函数索引, 尽可能收集一下列名
+            // 常见的case:
+            // 1. upper(col)
+            // 2. left(upper(col) , 10)
+            // 3. col(10)
+            List<SQLExpr> indexExpres = ((SQLMethodInvokeExpr) expr).getArguments();
+            List<String> columnNames = new ArrayList<String>();
+            // 加上当前列
+            columnNames.add(getSqlName(expr));
+            // 处理函数索引列
+            for (SQLExpr exArgs : indexExpres) {
+                if (exArgs instanceof SQLMethodInvokeExpr) {
+                    // 加上当前列
+                    columnNames.add(getSqlName(exArgs));
+                    columnNames.addAll(getIndexColumnNames(exArgs));
+                } else {
+                    String columnName = getSqlName(exArgs);
+                    columnNames.add(columnName);
+                }
+            }
+
+            return columnNames;
+        } else {
+            String columnName = getSqlName(expr);
+            return Arrays.asList(columnName);
+        }
+    }
 
     private String getSqlName(SQLExpr sqlName) {
         if (sqlName == null) {

parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/FastsqlSchemaTest.java

@@ -327,4 +327,17 @@ public class FastsqlSchemaTest {
         System.out.println(data.toString());
     }
 
+
+    @Test
+    public void test_function_index () throws Throwable {
+        SchemaRepository repository = new SchemaRepository(JdbcConstants.MYSQL);
+        repository.setDefaultSchema("test");
+        String sql = "CREATE TABLE test1 (\n" + "    id INT AUTO_INCREMENT PRIMARY KEY,\n"
+                     + "    owner_id INT NOT NULL,\n" + "    code VARCHAR(100) NOT NULL,\n"
+                     + "    UNIQUE KEY uk_owner_id_upper_code (owner_id, (upper(code)))\n" + ");";
+        repository.console(sql);
+        SchemaObject table = repository.findTable("test1");
+        Assert.assertTrue(table != null);
+    }
+
 }

parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta_DDL_Test.java

@@ -6,6 +6,9 @@ import java.net.URL;
 import java.util.ArrayList;
 import java.util.List;
 
+import com.alibaba.druid.sql.repository.SchemaObject;
+import com.alibaba.druid.sql.repository.SchemaRepository;
+import com.alibaba.druid.util.JdbcConstants;
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang.StringUtils;
 import org.junit.Assert;
@@ -109,4 +112,26 @@ public class MemoryTableMeta_DDL_Test {
             System.out.println(sourceMeta.toString());
         }
     }
+
+    @Test
+    public void test_function_index () throws Throwable {
+        MemoryTableMeta memoryTableMeta = new MemoryTableMeta();
+        URL url = Thread.currentThread().getContextClassLoader().getResource("dummy.txt");
+        File dummyFile = new File(url.getFile());
+        File create = new File(dummyFile.getParent() + "/ddl", "ddl_create_function_index.sql");
+        String sql = StringUtils.join(IOUtils.readLines(new FileInputStream(create)), "\n");
+        memoryTableMeta.apply(null, "test", sql, null);
+
+        List<String> tableNames = new ArrayList<>();
+        for (Schema schema : memoryTableMeta.getRepository().getSchemas()) {
+            tableNames.addAll(schema.showTables());
+        }
+
+        for (String table : tableNames) {
+            TableMeta sourceMeta = memoryTableMeta.find("test", table);
+            TableMeta.FieldMeta field = sourceMeta.getFieldMetaByName("code");
+            System.out.println(sourceMeta.toString());
+            Assert.assertTrue(field.isUnique());
+        }
+    }
 }

parse/src/test/resources/ddl/ddl_create_function_index.sql

@@ -0,0 +1 @@
+CREATE TABLE function_index_test (