fix parser failed when binlog_row_value_options=partial_json (alibaba#5018)

* 修改json部分更新的问题

* 保证after-image的情况下位点回退完整

Author: WangMinan

dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonDiffConversion.java

@@ -3,13 +3,14 @@
 import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Objects;
 
 import com.taobao.tddl.dbsync.binlog.JsonConversion.Json_Value;
 import com.taobao.tddl.dbsync.binlog.JsonConversion.Json_enum_type;
 
 /**
  * 处理mysql8.0 parital json diff解析
- * 
+ *
  * @author agapple 2018年11月4日 下午3:53:46
  * @since 1.1.2
  */
@@ -71,12 +72,20 @@ public static StringBuilder print_json_diff(LogBuffer buffer, long len, String c
 
                 buffer.forward((int) value_length);
             }
+
+            if (buffer.position - position >= len) {
+                break;
+            }
+        }
+
+        if (buffer.position - position != len) {
+            throw new IllegalArgumentException("reading json diff");
         }
 
         // Print function names in reverse order.
         StringBuilder builder = new StringBuilder();
         for (int i = operation_names.size() - 1; i >= 0; i--) {
-            if (i == 0 || operation_names.get(i - 1) != operation_names.get(i)) {
+            if (i == 0 || !Objects.equals(operation_names.get(i - 1), operation_names.get(i))) {
                 builder.append(operation_names.get(i)).append("(");
             }
         }
@@ -90,7 +99,7 @@ public static StringBuilder print_json_diff(LogBuffer buffer, long len, String c
 
         // In case this vector is empty (a no-op), make an early return
         // after printing only the column name
-        if (operation_names.size() == 0) {
+        if (operation_names.isEmpty()) {
             return builder;
         }
 
@@ -128,8 +137,13 @@ public static StringBuilder print_json_diff(LogBuffer buffer, long len, String c
                 builder.append(jsonBuilder);
             }
 
+            if (buffer.position - position >= len) {
+                builder.append(")");
+                break;
+            }
+
             // Print closing parenthesis
-            if (!buffer.hasRemaining() || operation_names.get(diff_i + 1) != operation_names.get(diff_i)) {
+            if (!buffer.hasRemaining() || !Objects.equals(operation_names.get(diff_i + 1), operation_names.get(diff_i))) {
                 builder.append(")");
             }
 
@@ -139,6 +153,10 @@ public static StringBuilder print_json_diff(LogBuffer buffer, long len, String c
             diff_i++;
         }
 
+        if (buffer.position - position != len) {
+            throw new IllegalArgumentException("reading json diff");
+        }
+
         return builder;
     }
 

dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java

@@ -19,7 +19,7 @@
 
 /**
  * Extracting JDBC type & value information from packed rows-buffer.
- * 
+ *
  * @see mysql-5.1.60/sql/log_event.cc - Rows_log_event::print_verbose_one_row
  * @author <a href="mailto:[email protected]">Changyuan.lh</a>
  * @version 1.0
@@ -70,7 +70,7 @@ public final boolean nextOneRow(BitSet columns) {
 
     /**
      * Extracting next row from packed buffer.
-     * 
+     *
      * @see mysql-5.1.60/sql/log_event.cc -
      * Rows_log_event::print_verbose_one_row
      */
@@ -104,7 +104,7 @@ public final boolean nextOneRow(BitSet columns, boolean after) {
 
     /**
      * Extracting next field value from packed buffer.
-     * 
+     *
      * @see mysql-5.1.60/sql/log_event.cc -
      * Rows_log_event::print_verbose_one_row
      */
@@ -114,7 +114,7 @@ public final Serializable nextValue(final String columName, final int columnInde
 
     /**
      * Extracting next field value from packed buffer.
-     * 
+     *
      * @see mysql-5.1.60/sql/log_event.cc -
      * Rows_log_event::print_verbose_one_row
      */
@@ -277,7 +277,7 @@ static int mysqlToJavaType(int type, final int meta, boolean isBinary) {
 
     /**
      * Extracting next field value from packed buffer.
-     * 
+     *
      * @see mysql-5.1.60/sql/log_event.cc - log_event_print_value
      */
     final Serializable fetchValue(String columnName, int columnIndex, int type, final int meta, boolean isBinary) {
@@ -1077,30 +1077,21 @@ final Serializable fetchValue(String columnName, int columnIndex, int type, fina
                 if (partialBits.get(1)) {
                     // print_json_diff
                     int position = buffer.position();
-                    StringBuilder builder = JsonDiffConversion.print_json_diff(buffer,
-                        len,
-                        columnName,
-                        columnIndex,
-                            charset);
-                    value = builder.toString();
-                    buffer.position(position + len);
-                } else {
-                    if (0 == len) {
-                        // fixed issue #1 by lava, json column of zero length
-                        // has no
-                        // value, value parsing should be skipped
-                        value = "";
-                    } else {
-                        int position = buffer.position();
-                        Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(),
-                            buffer,
-                            len - 1,
+                    try {
+                        StringBuilder builder = JsonDiffConversion.print_json_diff(buffer,
+                                len,
+                                columnName,
+                                columnIndex,
                                 charset);
-                        StringBuilder builder = new StringBuilder();
-                        jsonValue.toJsonString(builder, charset);
                         value = builder.toString();
                         buffer.position(position + len);
+                    } catch (IllegalArgumentException e) {
+                        buffer.position(position);
+                        // print_json_diff failed, fallback to parse_value
+                        parseJsonFromFullValue(len);
                     }
+                } else {
+                    parseJsonFromFullValue(len);
                 }
                 javaType = Types.VARCHAR;
                 length = len;
@@ -1155,6 +1146,25 @@ final Serializable fetchValue(String columnName, int columnIndex, int type, fina
         return value;
     }
 
+    private void parseJsonFromFullValue(int len) {
+        if (0 == len) {
+            // fixed issue #1 by lava, json column of zero length
+            // has no
+            // value, value parsing should be skipped
+            value = "";
+        } else {
+            int position = buffer.position();
+            Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(),
+                buffer,
+                len - 1,
+                    charset);
+            StringBuilder builder = new StringBuilder();
+            jsonValue.toJsonString(builder, charset);
+            value = builder.toString();
+            buffer.position(position + len);
+        }
+    }
+
     public final boolean isNull() {
         return fNull;
     }