[FLINK-36578][MySQL] Added modified JsonStringFormatter and option use.legacy.json.format #3658

SML0127 · 2024-10-22T02:40:16Z

The function to convert json to string is different in snapshot step and binlog step.
This causes the following differences.

original data : {"key": "value", "key1": "value1"}
snapshot : {"key": "value", "key1": "value1"}
binlog : {"key":"value","key1":"value1"} // no whitespace

It is already known issue, but I think it is necessary to process json in the same way in snapshot step and binlog step.
So I modified and added JsonStringFormatter to flink-cdc to handle json in the same way as the result of snapshot step.

modified code line
- line 105: Added whitespace before value
- line 207: Added whitespace after comma

jira issue: https://issues.apache.org/jira/browse/FLINK-36578

SML0127 · 2024-10-22T02:41:34Z

@leonardBang @PatrickRen @lvyanquan PTAL

all test passed (https://github.com/SML0127/flink-cdc-connectors/actions/runs/11455082745)

yuxiqian

Thanks for @SML0127's work! Just left some minor comments.

yuxiqian · 2024-11-11T12:04:29Z

...r-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlFullTypesITCase.java

        // The json string from binlog will remove useless space
-        expectedSnapshot[44] = BinaryStringData.fromString("{\"key1\":\"value1\"}");
+        expectedSnapshot[44] = BinaryStringData.fromString("{\"key1\": \"value1\"}");


This additional assignment was meant to fix the format discrepancy and could be removed now.

yuxiqian · 2024-11-11T12:07:27Z

...main/java/com/github/shyiko/mysql/binlog/event/deserialization/json/JsonStringFormatter.java

Can we make this enhancement an optional configuration and defaults to the legacy format, considering some users may have relied on this specific JSON format?

I added useLegacyJsonFormat option as a static variable in MySqlSourceConfig, and shared it with legcay MySqlSource for legacy MySqlSource user.

Also the default value of this option is true, so users who want to continue using current format(legacy format) don't need to worry about this new option.

yuxiqian · 2024-11-11T12:45:54Z

...r-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlFullTypesITCase.java

Can we add some tests to verify if we have the same JSON format between snapshot and binlog stage?

I added tests for this new option. When useLegacyJsonFormat = flase, it verifies that the JSON type data format is the same between snapshot stage and binlog stage.

flink-cdc/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlFullTypesITCase.java

Lines 498 to 551 in 337b5de

private void testJsonDataType(UniqueDatabase database, Boolean useLegacyJsonFormat)

throws Exception {

database.createAndInitialize();

CloseableIterator<Event> iterator =

env.fromSource(

getFlinkSourceProvider(

new String[] {"json_types"},

database,

useLegacyJsonFormat)

.getSource(),

WatermarkStrategy.noWatermarks(),

"Event-Source")

.executeAndCollect();

Object[] expectedSnapshot =

new Object[] {

DecimalData.fromBigDecimal(new BigDecimal("1"), 20, 0),

BinaryStringData.fromString("{\"key1\": \"value1\"}"),

BinaryStringData.fromString("{\"key1\": \"value1\", \"key2\": \"value2\"}"),

BinaryStringData.fromString(

"[{\"key1\": \"value1\", \"key2\": {\"key2_1\": \"value2_1\", \"key2_2\": \"value2_2\"}, \"key3\": [\"value3\"], \"key4\": [\"value4_1\", \"value4_2\"]}, {\"key5\": \"value5\"}]"),

1

};

// skip CreateTableEvent

List<Event> snapshotResults =

MySqSourceTestUtils.fetchResultsAndCreateTableEvent(iterator, 1).f0;

RecordData snapshotRecord = ((DataChangeEvent) snapshotResults.get(0)).after();

Assertions.assertThat(RecordDataTestUtils.recordFields(snapshotRecord, JSON_TYPES))

.isEqualTo(expectedSnapshot);

try (Connection connection = database.getJdbcConnection();

Statement statement = connection.createStatement()) {

statement.execute("UPDATE json_types SET int_c = null WHERE id = 1;");

}

Object[] expectedStreamRecord = expectedSnapshot;

if (useLegacyJsonFormat) {

expectedSnapshot[1] = BinaryStringData.fromString("{\"key1\":\"value1\"}");

expectedSnapshot[2] =

BinaryStringData.fromString("{\"key1\":\"value1\",\"key2\":\"value2\"}");

expectedSnapshot[3] =

BinaryStringData.fromString(

"[{\"key1\":\"value1\",\"key2\":{\"key2_1\":\"value2_1\",\"key2_2\":\"value2_2\"},\"key3\":[\"value3\"],\"key4\":[\"value4_1\",\"value4_2\"]},{\"key5\":\"value5\"}]");

}

expectedSnapshot[4] = null;

List<Event> streamResults =

MySqSourceTestUtils.fetchResultsAndCreateTableEvent(iterator, 1).f0;

RecordData streamRecord = ((DataChangeEvent) streamResults.get(0)).after();

Assertions.assertThat(RecordDataTestUtils.recordFields(streamRecord, JSON_TYPES))

.isEqualTo(expectedStreamRecord);

}

…eLegacyJsonFromat default is true

…seLegacyJsonFormat is true

…able

…mysql.md

…-formatter

…hema, testMysql8PrecisionTypesSchema

SML0127

@yuxiqian PTAL😄
CI 👉 https://github.com/SML0127/flink-cdc-connectors/actions/runs/11877667979

SML0127 · 2024-11-17T09:23:50Z

...main/java/com/github/shyiko/mysql/binlog/event/deserialization/json/JsonStringFormatter.java

+        if (useLegacyJsonFormat) {
+            sb.append("\":");
+        } else {
+            sb.append("\": ");
+        }


Now we can choose JSON type data format (whitespace in before value)

SML0127 · 2024-11-17T09:24:10Z

...main/java/com/github/shyiko/mysql/binlog/event/deserialization/json/JsonStringFormatter.java

+        if (useLegacyJsonFormat) {
+            sb.append(",");
+        } else {
+            sb.append(", ");
+        }


Now we can choose JSON type data format (whitespace in after comma)

SML0127 · 2024-11-17T09:25:07Z

...ne-connectors/flink-cdc-pipeline-connector-mysql/src/test/resources/ddl/column_type_test.sql

+(
+    id                   SERIAL,
+    json_c0               JSON,
+    json_c1               JSON,
+    json_c2               JSON,
+    int_c  INTEGER,
+    PRIMARY KEY (id)
+) DEFAULT CHARSET=utf8;
+
+INSERT INTO json_types
+VALUES (DEFAULT,
+        '{"key1":"value1"}',
+        '{"key1":"value1","key2":"value2"}',
+        '[{"key1":"value1","key2":{"key2_1":"value2_1","key2_2":"value2_2"},"key3":["value3"],"key4":["value4_1","value4_2"]},{"key5":"value5"}]',
+        1
+       );


test table for various JSON format data

SML0127 · 2024-11-17T09:29:13Z

.../main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java

+    public MySqlSourceConfigFactory useLegacyJsonFormat(boolean useLegacyJsonFormat) {
+        this.useLegacyJsonFormat = useLegacyJsonFormat;
+        return this;
+    }
+


setter for useLegacyJsonFormat option

SML0127 · 2024-11-17T09:29:44Z

...mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java

+    @Experimental
+    public static final ConfigOption<Boolean> USE_LEGACY_JSON_FORMAT =
+            ConfigOptions.key("use.legacy.json.format")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription(
+                            "Whether to use legacy json format. The default value is true, which means there is no whitespace before value and after comma in json format.");


use.legacy.json.format option for pipeline connector. default is true.

SML0127 · 2024-11-17T09:30:00Z

...dc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceOptions.java

+    @Experimental
+    public static final ConfigOption<Boolean> USE_LEGACY_JSON_FORMAT =
+            ConfigOptions.key("use.legacy.json.format")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription(
+                            "Whether to use legacy json format. The default value is true, which means there is no whitespace before value and after comma in json format.");


use.legacy.json.format option for source connector. default is true.

yuxiqian

Thanks for @SML0127's nice work, looks good overall (I left some suggestions on test cases). Still need hearing from @ruanhang1993 on the changed configs.

yuxiqian · 2024-11-27T08:45:17Z

...main/java/com/github/shyiko/mysql/binlog/event/deserialization/json/JsonStringFormatter.java

+/**
+ * Copied from mysql-binlog-connector-java 0.27.2 to add whitespace before value and after comma.
+ *
+ * <p>Line 105: Added whitespace before value, Line 207: Added whitespace after comma


Seems line numbers have been differed from actual changes.

yuxiqian · 2024-11-27T08:46:18Z

...urce-connectors/flink-connector-mysql-cdc/src/test/resources/ddl/column_type_test_mysql8.sql

Accidental change?

it was my mistake...

yuxiqian · 2024-11-27T08:55:05Z

...r-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlFullTypesITCase.java

+        if (useLegacyJsonFormat) {
+            expectedSnapshot[1] = BinaryStringData.fromString("{\"key1\":\"value1\"}");
+            expectedSnapshot[2] =
+                    BinaryStringData.fromString("{\"key1\":\"value1\",\"key2\":\"value2\"}");
+            expectedSnapshot[3] =
+                    BinaryStringData.fromString(
+                            "[{\"key1\":\"value1\",\"key2\":{\"key2_1\":\"value2_1\",\"key2_2\":\"value2_2\"},\"key3\":[\"value3\"],\"key4\":[\"value4_1\",\"value4_2\"]},{\"key5\":\"value5\"}]");
+        }
+        expectedSnapshot[4] = null;


Modifying expected data is a little confusing. Will something like

if (useLegacyJsonFormat) { assertThat(RecordDataTestUtils.recordFields(...)) .containsExactly( BinaryStringData.fromString("{\"key1\":\"value1\"}"), BinaryStringData.fromString("{\"key2\":\"value2\"}"), ... ); } else { assertThat(RecordDataTestUtils.recordFields(...)) .containsExactly(...); }

be clearer?

I revised it more clearly as shown below.

if (useLegacyJsonFormat) { // removed whitespace before value and after comma in json format string value Assertions.assertThat(RecordDataTestUtils.recordFields(streamRecord, JSON_TYPES)) .containsExactly( DecimalData.fromBigDecimal(new BigDecimal("1"), 20, 0), BinaryStringData.fromString("{\"key1\":\"value1\"}"), BinaryStringData.fromString( "{\"key1\":\"value1\",\"key2\":\"value2\"}"), BinaryStringData.fromString( "[{\"key1\":\"value1\",\"key2\":{\"key2_1\":\"value2_1\",\"key2_2\":\"value2_2\"},\"key3\":[\"value3\"],\"key4\":[\"value4_1\",\"value4_2\"]},{\"key5\":\"value5\"}]"), null); } else { Assertions.assertThat(RecordDataTestUtils.recordFields(streamRecord, JSON_TYPES)) .containsExactly(expectedStreamRecord); }

yuxiqian · 2024-11-27T08:57:36Z

...l-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactory.java

Since DataStream / Table source connectors' behavior also changed, corresponding integrated test cases might be necessary.

I will add integrated test cases if necessary.

…-formatter

SML0127 · 2025-01-03T06:26:20Z

@ruanhang1993 @leonardBang PTAL

ADD: added modified JsonStringFormatter

542d856

github-actions bot added the mysql-cdc-connector label Oct 22, 2024

SML0127 changed the title ~~[FLINK-36578][mysql] Add table information of binlog offsets when checkpointing added modified JsonStringFormatter~~ [FLINK-36578][mysql][BUG] Add table information of binlog offsets when checkpointing added modified JsonStringFormatter Oct 22, 2024

Seung-Min Lee added 3 commits October 22, 2024 12:13

EDIT: modified license style

51ddb26

EDIT: changed test data for json column

52c808a

EDIT: changed test data for json

1117342

github-actions bot added mysql-pipeline-connector tidb-cdc-connector labels Oct 22, 2024

ROLLBACK: rollback TiDBConnectorITCase.java

bac5584

github-actions bot removed the tidb-cdc-connector label Oct 22, 2024

yuxiqian reviewed Nov 11, 2024

View reviewed changes

Seung-Min Lee added 13 commits November 12, 2024 22:14

ADD: added useLegacyJsonFromat option (default true)

bebeaa4

FIX: mvn spotless:apply

5050a8e

FIX: rollbacked white space for json string in test codes. becuase us…

1ca8990

…eLegacyJsonFromat default is true

FIX: mvn spotless:apply

29872a6

ROLLBACK: rollbacked somes test resources. because default value of u…

426d3e7

…seLegacyJsonFormat is true

FIX: added whitespace

64a4a57

ADD: useLegacyJsonFormat for legacy MySqlSource

638cd21

ADD: added test codes, resources for useLegacyJsonType option

9ed0778

ADD: added test codes in MySqlMetadataAccessorITCase for json_types t…

ffcf1f8

…able

FIX: mvn spotless:apply

396fab1

ADD: added USE_LEGACY_JSON_FORMAT for flink-cdc-pipeline-connector-mysql

37b52cf

UPDATE: added se.legacy.json.format option to mysql-cdc.md, pipeline/…

401d403

…mysql.md

FIX: removed json_types table in legacy test code resources

1265453

github-actions bot added the docs Improvements or additions to documentation label Nov 17, 2024

Merge branch 'master' into feature/added-whitespace-in-json-to-string…

337b5de

…-formatter

SML0127 changed the title ~~[FLINK-36578][mysql][BUG] Add table information of binlog offsets when checkpointing added modified JsonStringFormatter~~ [FLINK-36578][mysql] Add table information of binlog offsets when checkpointing added modified JsonStringFormatter Nov 17, 2024

SML0127 changed the title ~~[FLINK-36578][mysql] Add table information of binlog offsets when checkpointing added modified JsonStringFormatter~~ [FLINK-36578][MySQL] Add table information of binlog offsets when checkpointing added modified JsonStringFormatter Nov 17, 2024

ROLLBACK: recovered accidientally deleted testMysql57PrecisionTypesSc…

c46890d

…hema, testMysql8PrecisionTypesSchema

SML0127 commented Nov 17, 2024

View reviewed changes

SML0127 changed the title ~~[FLINK-36578][MySQL] Add table information of binlog offsets when checkpointing added modified JsonStringFormatter~~ [FLINK-36578][MySQL] Added modified JsonStringFormatter and option use.legacy.json.format Nov 22, 2024

SML0127 requested a review from yuxiqian November 27, 2024 08:42

yuxiqian approved these changes Nov 27, 2024

View reviewed changes

github-actions bot added the reviewed label Nov 27, 2024

Seung-Min Lee and others added 4 commits November 28, 2024 16:56

FIX: fixed wrong line number in annotation

945f398

REFACTOR: changed codes for modifying expected data to be more clear

5bf33c9

EDIT: added annotations for modifying expected data

5dd1f51

Merge branch 'master' into feature/added-whitespace-in-json-to-string…

285280a

…-formatter

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-36578][MySQL] Added modified JsonStringFormatter and option use.legacy.json.format #3658

[FLINK-36578][MySQL] Added modified JsonStringFormatter and option use.legacy.json.format #3658

SML0127 commented Oct 22, 2024 •

edited

Loading

SML0127 commented Oct 22, 2024 •

edited

Loading

yuxiqian left a comment

yuxiqian Nov 11, 2024

SML0127 Nov 17, 2024

yuxiqian Nov 11, 2024

SML0127 Nov 17, 2024

yuxiqian Nov 11, 2024

SML0127 Nov 17, 2024 •

edited

Loading

SML0127 left a comment •

edited

Loading

SML0127 Nov 17, 2024 •

edited

Loading

SML0127 Nov 17, 2024 •

edited

Loading

SML0127 Nov 17, 2024 •

edited

Loading

SML0127 Nov 17, 2024

SML0127 Nov 17, 2024 •

edited

Loading

SML0127 Nov 17, 2024 •

edited

Loading

yuxiqian left a comment

yuxiqian Nov 27, 2024

SML0127 Nov 28, 2024

yuxiqian Nov 27, 2024

SML0127 Nov 28, 2024

yuxiqian Nov 27, 2024

SML0127 Nov 28, 2024 •

edited

Loading

yuxiqian Nov 27, 2024

SML0127 Nov 28, 2024

SML0127 commented Jan 3, 2025

	private void testJsonDataType(UniqueDatabase database, Boolean useLegacyJsonFormat)
	throws Exception {
	database.createAndInitialize();
	CloseableIterator<Event> iterator =
	env.fromSource(
	getFlinkSourceProvider(
	new String[] {"json_types"},
	database,
	useLegacyJsonFormat)
	.getSource(),
	WatermarkStrategy.noWatermarks(),
	"Event-Source")
	.executeAndCollect();

	Object[] expectedSnapshot =
	new Object[] {
	DecimalData.fromBigDecimal(new BigDecimal("1"), 20, 0),
	BinaryStringData.fromString("{\"key1\": \"value1\"}"),
	BinaryStringData.fromString("{\"key1\": \"value1\", \"key2\": \"value2\"}"),
	BinaryStringData.fromString(
	"[{\"key1\": \"value1\", \"key2\": {\"key2_1\": \"value2_1\", \"key2_2\": \"value2_2\"}, \"key3\": [\"value3\"], \"key4\": [\"value4_1\", \"value4_2\"]}, {\"key5\": \"value5\"}]"),
	1
	};

	// skip CreateTableEvent
	List<Event> snapshotResults =
	MySqSourceTestUtils.fetchResultsAndCreateTableEvent(iterator, 1).f0;
	RecordData snapshotRecord = ((DataChangeEvent) snapshotResults.get(0)).after();
	Assertions.assertThat(RecordDataTestUtils.recordFields(snapshotRecord, JSON_TYPES))
	.isEqualTo(expectedSnapshot);

	try (Connection connection = database.getJdbcConnection();
	Statement statement = connection.createStatement()) {
	statement.execute("UPDATE json_types SET int_c = null WHERE id = 1;");
	}

	Object[] expectedStreamRecord = expectedSnapshot;

	if (useLegacyJsonFormat) {
	expectedSnapshot[1] = BinaryStringData.fromString("{\"key1\":\"value1\"}");
	expectedSnapshot[2] =
	BinaryStringData.fromString("{\"key1\":\"value1\",\"key2\":\"value2\"}");
	expectedSnapshot[3] =
	BinaryStringData.fromString(
	"[{\"key1\":\"value1\",\"key2\":{\"key2_1\":\"value2_1\",\"key2_2\":\"value2_2\"},\"key3\":[\"value3\"],\"key4\":[\"value4_1\",\"value4_2\"]},{\"key5\":\"value5\"}]");
	}
	expectedSnapshot[4] = null;

	List<Event> streamResults =
	MySqSourceTestUtils.fetchResultsAndCreateTableEvent(iterator, 1).f0;
	RecordData streamRecord = ((DataChangeEvent) streamResults.get(0)).after();
	Assertions.assertThat(RecordDataTestUtils.recordFields(streamRecord, JSON_TYPES))
	.isEqualTo(expectedStreamRecord);
	}

[FLINK-36578][MySQL] Added modified JsonStringFormatter and option use.legacy.json.format #3658

Are you sure you want to change the base?

[FLINK-36578][MySQL] Added modified JsonStringFormatter and option use.legacy.json.format #3658

Conversation

SML0127 commented Oct 22, 2024 • edited Loading

SML0127 commented Oct 22, 2024 • edited Loading

yuxiqian left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SML0127 Nov 17, 2024 • edited Loading

Choose a reason for hiding this comment

SML0127 left a comment • edited Loading

Choose a reason for hiding this comment

SML0127 Nov 17, 2024 • edited Loading

Choose a reason for hiding this comment

SML0127 Nov 17, 2024 • edited Loading

Choose a reason for hiding this comment

SML0127 Nov 17, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SML0127 Nov 17, 2024 • edited Loading

Choose a reason for hiding this comment

SML0127 Nov 17, 2024 • edited Loading

Choose a reason for hiding this comment

yuxiqian left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SML0127 Nov 28, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SML0127 commented Jan 3, 2025

SML0127 commented Oct 22, 2024 •

edited

Loading

SML0127 commented Oct 22, 2024 •

edited

Loading

SML0127 Nov 17, 2024 •

edited

Loading

SML0127 left a comment •

edited

Loading

SML0127 Nov 17, 2024 •

edited

Loading

SML0127 Nov 17, 2024 •

edited

Loading

SML0127 Nov 17, 2024 •

edited

Loading

SML0127 Nov 17, 2024 •

edited

Loading

SML0127 Nov 17, 2024 •

edited

Loading

SML0127 Nov 28, 2024 •

edited

Loading