add polardb-x parser test

* 修复升级2.0.4导致兼容的问题

* update druid & fastjson version

common/src/main/java/com/alibaba/otter/canal/common/utils/JsonUtils.java

@@ -1,17 +1,13 @@
 package com.alibaba.otter.canal.common.utils;
 
 import java.lang.reflect.Type;
-import java.net.Inet4Address;
-import java.net.Inet6Address;
 import java.net.InetAddress;
 import java.util.Arrays;
 import java.util.List;
 
-import com.alibaba.fastjson2.JSON;
-import com.alibaba.fastjson2.JSONFactory;
-import com.alibaba.fastjson2.TypeReference;
+import com.alibaba.fastjson2.*;
+import com.alibaba.fastjson2.filter.Filter;
 import com.alibaba.fastjson2.filter.PropertyFilter;
-import com.alibaba.fastjson2.JSONWriter;
 import com.alibaba.fastjson2.writer.ObjectWriter;
 
 
@@ -21,22 +17,18 @@ import com.alibaba.fastjson2.writer.ObjectWriter;
  * @author jianghang
  */
 public class JsonUtils {
-
-    static {
-        JSON.register(InetAddress.class, InetAddressWriter.instance);
-        JSON.register(Inet4Address.class, InetAddressWriter.instance);
-        JSON.register(Inet6Address.class, InetAddressWriter.instance);
-
-        JSONFactory.getDefaultObjectReaderProvider().addAutoTypeAccept("com.alibaba.otter.");
-        JSONFactory.getDefaultObjectReaderProvider().addAutoTypeAccept("com.taobao.tddl.dbsync.");
-    }
+    static final Filter AUTO_TYPE_FILTER = JSONReader.autoTypeFilter(
+            "com.alibaba.otter.",
+            "com.taobao.tddl.dbsync."
+    );
 
     public static <T> T unmarshalFromByte(byte[] bytes, Class<T> targetClass) {
-        return (T) JSON.parseObject(bytes, targetClass);// 默认为UTF-8
+        return (T) JSON.parseObject(bytes, targetClass, AUTO_TYPE_FILTER);// 默认为UTF-8
     }
 
     public static <T> T unmarshalFromByte(byte[] bytes, TypeReference<T> type) {
-        return (T) JSON.parseObject(bytes, type.getType());
+
+        return (T) JSON.parseObject(bytes, type.getType(), AUTO_TYPE_FILTER);
     }
 
     public static byte[] marshalToByte(Object obj) {
@@ -48,11 +40,11 @@ public class JsonUtils {
     }
 
     public static <T> T unmarshalFromString(String json, Class<T> targetClass) {
-        return (T) JSON.parseObject(json, targetClass);// 默认为UTF-8
+        return (T) JSON.parseObject(json, targetClass, AUTO_TYPE_FILTER);// 默认为UTF-8
     }
 
     public static <T> T unmarshalFromString(String json, TypeReference<T> type) {
-        return (T) JSON.parseObject(json, type);// 默认为UTF-8
+        return (T) JSON.parseObject(json, type.getType(), AUTO_TYPE_FILTER);// 默认为UTF-8
     }
 
     public static String marshalToString(Object obj) {
@@ -71,7 +63,7 @@ public class JsonUtils {
 
         return JSON.toJSONString(obj, new PropertyFilter() {
             @Override
-            public boolean process(Object object, String name, Object value) {
+            public boolean apply(Object object, String name, Object value) {
                 return !propertyFliters.contains(name);
             }
         });

common/src/test/java/com/alibaba/otter/canal/common/JsonUtilsTest.java

@@ -0,0 +1,21 @@
+package com.alibaba.otter.canal.common;
+
+import com.alibaba.otter.canal.common.utils.JsonUtils;
+import org.junit.Test;
+
+import java.net.InetAddress;
+
+import static org.junit.Assert.assertEquals;
+
+public class JsonUtilsTest {
+    @Test
+    public void marshalToString() throws Exception {
+        InetAddress address = InetAddress.getByName("localhost");
+
+        String json = JsonUtils.marshalToString(address);
+        assertEquals("\"localhost\"", json);
+
+        InetAddress address1 = JsonUtils.unmarshalFromString(json, InetAddress.class);
+        assertEquals(address, address1);
+    }
+}

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

@@ -14,6 +14,7 @@ import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.regex.Pattern;
 
+import com.alibaba.otter.canal.parse.driver.mysql.packets.server.FieldPacket;
 import org.apache.commons.beanutils.BeanUtils;
 import org.apache.commons.lang.ObjectUtils;
 import org.apache.commons.lang.StringUtils;
@@ -190,17 +191,28 @@ public class DatabaseTableMeta implements TableMetaTSDB {
     private boolean dumpTableMeta(MysqlConnection connection, final CanalEventFilter filter) {
         try {
             ResultSetPacket packet = connection.query("show databases");
+            int columnSize = packet.getFieldDescriptors().size();
+            int columnIndex = 0;
+            for (; columnIndex < columnSize; columnIndex++) {
+                FieldPacket value = packet.getFieldDescriptors().get(columnIndex);
+                if (StringUtils.equalsIgnoreCase(value.getName(), "Database")) {
+                    break;
+                }
+            }
+
             List<String> schemas = new ArrayList<>();
-            schemas.addAll(packet.getFieldValues());
+            for (int line = 0; line < packet.getFieldValues().size() / columnSize; line++) {
+                String schema = packet.getFieldValues().get(line * columnSize + columnIndex);
+                schemas.add(schema);
+            }
 
             for (String schema : schemas) {
                 // filter views
                 packet = connection.query("show full tables from `" + schema + "` where Table_type = 'BASE TABLE'");
+                int tableNameColumnIndex = 0; // default index is 0
                 List<String> tables = new ArrayList<>();
-                for (String table : packet.getFieldValues()) {
-                    if ("BASE TABLE".equalsIgnoreCase(table)) {
-                        continue;
-                    }
+                for (int line = 0; line < packet.getFieldValues().size() / columnSize; line++) {
+                    String table = packet.getFieldValues().get(line * columnSize + tableNameColumnIndex);
                     String fullName = schema + "." + table;
                     if (blackFilter == null || !blackFilter.filter(fullName)) {
                         if (filter == null || filter.filter(fullName)) {

parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java

@@ -13,6 +13,7 @@ import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.lang.StringUtils;
 import org.junit.Assert;
+import org.junit.Ignore;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -45,6 +46,7 @@ import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.XidLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.mariadb.AnnotateRowsEvent;
 
+@Ignore
 public class DirectLogFetcherTest {
 
     protected final Logger logger         = LoggerFactory.getLogger(this.getClass());

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

@@ -137,4 +137,52 @@ public class FastsqlSchemaTest {
         SchemaObject table = repository.findTable("test");
         Assert.assertTrue(table != null);
     }
+
+    @Test
+    public void test_polardb_x() throws Throwable {
+        SchemaRepository repository = new SchemaRepository(JdbcConstants.MYSQL);
+        repository.setDefaultSchema("test");
+
+        String sql1 = "CREATE TABLE `test1` (\n" + "  `id` int(11) UNSIGNED NOT NULL AUTO_INCREMENT,\n"
+                      + "  `serialNo` varchar(64) CHARACTER SET utf8mb4 NOT NULL DEFAULT '',\n"
+                      + "  `user_id` int(11) DEFAULT NULL COMMENT '用户id',\n" + "  PRIMARY KEY (`id`)\n"
+                      + ") ENGINE = InnoDB  PARTITION BY KEY(`tenant_id`,`id`)\n" + "PARTITIONS 21 tablegroup = `tg_p_msg`";
+        repository.console(sql1);
+        SchemaObject table = repository.findTable("test1");
+        Assert.assertTrue(table != null);
+
+
+        String sql2 = "CREATE TABLE `test2` (\n" + "  `id` int(11) UNSIGNED NOT NULL AUTO_INCREMENT,\n"
+                      + "  `serialNo` varchar(64) CHARACTER SET utf8mb4 NOT NULL DEFAULT '',\n"
+                      + "  `user_id` int(11) DEFAULT NULL COMMENT '用户id',\n" + "  PRIMARY KEY (`id`)\n"
+                      + ") ENGINE = InnoDB single";
+        repository.console(sql2);
+        table = repository.findTable("test2");
+        Assert.assertTrue(table != null);
+
+
+        String sql3 = "CREATE TABLE `test3` (\n" + "  `id` int(11) UNSIGNED NOT NULL AUTO_INCREMENT,\n"
+                      + "  `serialNo` varchar(64) CHARACTER SET utf8mb4 NOT NULL DEFAULT '',\n"
+                      + "  `user_id` int(11) DEFAULT NULL COMMENT '用户id',\n" + "  PRIMARY KEY (`id`)\n"
+                      + ") ENGINE = InnoDB locality = 'dn=polardbx-ng28-dn-1,polardbx-ng28-dn-2'";
+        repository.console(sql3);
+        table = repository.findTable("test3");
+        Assert.assertTrue(table != null);
+
+        String sql4 = "CREATE TABLE test4(\n" + " order_id int AUTO_INCREMENT primary key,\n"
+                      + " customer_id int,\n" + " country varchar(64),\n" + " city varchar(64),\n"
+                      + " order_time datetime not null)\n" + "PARTITION BY LIST COLUMNS(country,city)\n" + "(\n"
+                      + "  PARTITION p1 VALUES IN (('China','Shanghai')) LOCALITY = 'dn=polardbx-ng28-dn-2',\n"
+                      + "  PARTITION p2 VALUES IN (('China','Beijing')) LOCALITY = 'dn=polardbx-ng28-dn-2',\n"
+                      + "  PARTITION p3 VALUES IN (('China','Hangzhou')) ,\n"
+                      + "  PARTITION p4 VALUES IN (('China','Nanjing')) ,\n"
+                      + "  PARTITION p5 VALUES IN (('China','Guangzhou')) ,\n"
+                      + "  PARTITION p6 VALUES IN (('China','Shenzhen')) ,\n"
+                      + "  PARTITION p7 VALUES IN (('China','Wuhan')) ,\n"
+                      + "  PARTITION p8 VALUES IN (('America','New York'))\n"
+                      + ") LOCALITY = 'dn=polardbx-ng28-dn-0,polardbx-ng28-dn-1';";
+        repository.console(sql4);
+        table = repository.findTable("test4");
+        Assert.assertTrue(table != null);
+    }
 }

pom.xml

@@ -222,7 +222,7 @@
             <dependency>
                 <groupId>com.alibaba.fastjson2</groupId>
                 <artifactId>fastjson2</artifactId>
-                <version>2.0.4</version>
+                <version>2.0.7</version>
             </dependency>
             <dependency>
                 <groupId>com.google.guava</groupId>
@@ -262,7 +262,7 @@
             <dependency>
                 <groupId>com.alibaba</groupId>
                 <artifactId>druid</artifactId>
-                <version>1.2.10</version>
+                <version>1.2.11</version>
             </dependency>
             <dependency>
                 <groupId>com.lmax</groupId>