chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix parser failed when binlog_row_value_options=partial_json (#5018)

Browse Source 
* 修改json部分更新的问题

* 保证after-image的情况下位点回退完整
This commit is contained in:
Wang Minan
2024-01-02 10:44:40 +08:00
committed by GitHub
parent fb35dd4254
commit 54145ced5e
2 changed files with 57 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonDiffConversion.java
View File

@@ -3,6 +3,7 @@ package com.taobao.tddl.dbsync.binlog;
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;
@@ -71,12 +72,20 @@ public class JsonDiffConversion {
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 class JsonDiffConversion {
// 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 class JsonDiffConversion {
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 class JsonDiffConversion {
diff_i++;
}
if (buffer.position - position != len) {
throw new IllegalArgumentException("reading json diff");
}
return builder;
}

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

@@ -1077,6 +1077,7 @@ public final class RowsLogBuffer {
if (partialBits.get(1)) {
// print_json_diff
int position = buffer.position();
try {
StringBuilder builder = JsonDiffConversion.print_json_diff(buffer,
len,
columnName,
@@ -1084,23 +1085,13 @@ public final class RowsLogBuffer {
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,
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 @@ public final class RowsLogBuffer {
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;
}