Skip to content

Commit

Permalink
fix parser failed when binlog_row_value_options=partial_json (#5018)
Browse files Browse the repository at this point in the history 
* 修改json部分更新的问题

* 保证after-image的情况下位点回退完整
  • Loading branch information
@WangMinan
WangMinan authored Jan 2, 2024
1 parent fb35dd4 commit 54145ce
Show file tree
Hide file tree
Showing 2 changed files with 57 additions and 29 deletions.
26 changes: 22 additions & 4 deletions dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonDiffConversion.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
*/
Expand Down Expand Up @@ -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("(");
}
}
Expand All @@ -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;
}

Expand Down Expand Up @@ -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(")");
}

Expand All @@ -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;
}

Expand Down
60 changes: 35 additions & 25 deletions dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
*/
Expand Down Expand Up @@ -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
*/
Expand All @@ -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
*/
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
}
Expand Down

0 comments on commit 54145ce

Please sign in to comment.