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

docs/content/docs/connectors/flink-sources/mysql-cdc.md

@@ -391,6 +391,18 @@ During a snapshot operation, the connector will query each included table to pro
           hex: The binary data type is converted to a hexadecimal string and transmitted.
       The default value is none. Depending on your requirements and data types, you can choose the appropriate processing mode. If your database contains a large number of binary data types, it is recommended to use base64 or hex mode to make it easier to handle during transmission.</td> 
     </tr>
+    <tr>
+      <td>use.legacy.json.format</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">true</td>
+      <td>Boolean</td>
+      <td>Whether to use legacy JSON format to cast JSON type data in binlog. <br>
+          It determines whether to use the legacy JSON format when retrieving JSON type data in binlog. 
+          If the user configures 'use.legacy.json.format' = 'true', whitespace before values and after commas in the JSON type data is removed. For example,
+          JSON type data {"key1": "value1", "key2": "value2"} in binlog would be converted to {"key1":"value1","key2":"value2"}.
+          When 'use.legacy.json.format' = 'false', the data would be converted to {"key1": "value1", "key2": "value2"}, with whitespace before values and after commas preserved.
+      </td>
+    </tr>
     </tbody>
 </table>
 </div>

docs/content/docs/connectors/pipeline-connectors/mysql.md

@@ -293,6 +293,18 @@ pipeline:
           scan.binlog.newly-added-table.enabled: only do binlog-reading for newly added table during binlog reading phase.
       </td>
     </tr>
+    <tr>
+      <td>use.legacy.json.format</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">true</td>
+      <td>Boolean</td>
+      <td>Whether to use legacy JSON format to cast JSON type data in binlog. <br>
+          It determines whether to use the legacy JSON format when retrieving JSON type data in binlog. 
+          If the user configures 'use.legacy.json.format' = 'true', whitespace before values and after commas in the JSON type data is removed. For example,
+          JSON type data {"key1": "value1", "key2": "value2"} in binlog would be converted to {"key1":"value1","key2":"value2"}.
+          When 'use.legacy.json.format' = 'false', the data would be converted to {"key1": "value1", "key2": "value2"}, with whitespace before values and after commas preserved.
+      </td>
+    </tr>
     </tbody>
 </table>
 </div>

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java

@@ -88,6 +88,7 @@
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.TABLES;
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.TABLES_EXCLUDE;
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.USERNAME;
+import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.USE_LEGACY_JSON_FORMAT;
 import static org.apache.flink.cdc.connectors.mysql.source.utils.ObjectUtils.doubleCompare;
 import static org.apache.flink.cdc.debezium.table.DebeziumOptions.DEBEZIUM_OPTIONS_PREFIX;
 import static org.apache.flink.cdc.debezium.table.DebeziumOptions.getDebeziumProperties;
@@ -139,6 +140,7 @@ public DataSource createDataSource(Context context) {
         boolean scanNewlyAddedTableEnabled = config.get(SCAN_NEWLY_ADDED_TABLE_ENABLED);
         boolean scanBinlogNewlyAddedTableEnabled =
                 config.get(SCAN_BINLOG_NEWLY_ADDED_TABLE_ENABLED);
+        boolean useLegacyJsonFormat = config.get(USE_LEGACY_JSON_FORMAT);
 
         validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1);
         validateIntegerOption(CHUNK_META_GROUP_SIZE, splitMetaGroupSize, 1);
@@ -175,7 +177,8 @@ public DataSource createDataSource(Context context) {
                         .includeSchemaChanges(includeSchemaChanges)
                         .debeziumProperties(getDebeziumProperties(configMap))
                         .jdbcProperties(getJdbcProperties(configMap))
-                        .scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled);
+                        .scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled)
+                        .useLegacyJsonFormat(useLegacyJsonFormat);
 
         List<TableId> tableIds = MySqlSchemaUtils.listTables(configFactory.createConfig(0), null);
 
@@ -306,6 +309,7 @@ public Set<ConfigOption<?>> optionalOptions() {
         options.add(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND);
         options.add(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND);
         options.add(SCAN_BINLOG_NEWLY_ADDED_TABLE_ENABLED);
+        options.add(USE_LEGACY_JSON_FORMAT);
         options.add(METADATA_LIST);
         return options;
     }

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java

@@ -273,6 +273,14 @@ public class MySqlDataSourceOptions {
                                     + "scan.newly-added-table.enabled: do re-snapshot & binlog-reading for newly added table when restored; \n"
                                     + "scan.binlog.newly-added-table.enabled: only do binlog-reading for newly added table during binlog reading phase.");
 
+    @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.");
+
     @Experimental
     public static final ConfigOption<String> METADATA_LIST =
             ConfigOptions.key("metadata.list")

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

 } 
 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); 
 } 
@@ -106,11 +106,39 @@ public void testMysql57CommonDataTypes() throws Throwable {
         testCommonDataTypes(fullTypesMySql57Database);
     }
 
+    @Test
+    public void testMysql57JsonDataTypes() throws Throwable {
+        // Set `useLegacyJsonFormat` as false, so the json string will have no whitespace
+        // before value and after comma in json format be formatted with legacy format.
+        testJsonDataType(fullTypesMySql57Database, false);
+    }
+
+    @Test
+    public void testMysql57JsonDataTypesWithUseLegacyJsonFormat() throws Throwable {
+        // Set `useLegacyJsonFormat` as true, so the json string will have whitespace before
+        // value and after comma in json format be formatted with legacy format.
+        testJsonDataType(fullTypesMySql57Database, true);
+    }
+
     @Test
     public void testMySql8CommonDataTypes() throws Throwable {
         testCommonDataTypes(fullTypesMySql8Database);
     }
 
+    @Test
+    public void testMySql8JsonDataTypes() throws Throwable {
+        // Set `useLegacyJsonFormat` as false, so the json string will have no whitespace
+        // before value and after comma in json format be formatted with legacy format.
+        testJsonDataType(fullTypesMySql8Database, false);
+    }
+
+    @Test
+    public void testMySql8JsonDataTypesWithUseLegacyJsonFormat() throws Throwable {
+        // Set `useLegacyJsonFormat` as true, so the json string will have whitespace before
+        // value and after comma in json format be formatted with legacy format.
+        testJsonDataType(fullTypesMySql8Database, true);
+    }
+
     @Test
     public void testMysql57TimeDataTypes() throws Throwable {
         RowType recordType =
@@ -321,9 +349,13 @@ public void testMysqlPrecisionTypes(UniqueDatabase database) throws Throwable {
                 };
 
         database.createAndInitialize();
+        Boolean useLegacyJsonFormat = true;
         CloseableIterator<Event> iterator =
                 env.fromSource(
-                                getFlinkSourceProvider(new String[] {"precision_types"}, database)
+                                getFlinkSourceProvider(
+                                                new String[] {"precision_types"},
+                                                database,
+                                                useLegacyJsonFormat)
                                         .getSource(),
                                 WatermarkStrategy.noWatermarks(),
                                 "Event-Source")
@@ -351,9 +383,15 @@ public void testMysqlPrecisionTypes(UniqueDatabase database) throws Throwable {
 
     private void testCommonDataTypes(UniqueDatabase database) throws Exception {
         database.createAndInitialize();
+        // Set useLegacyJsonFormat option as true, so the json string will have no whitespace before
+        // value and after comma in json format.be formatted with legacy format.
+        Boolean useLegacyJsonFormat = true;
         CloseableIterator<Event> iterator =
                 env.fromSource(
-                                getFlinkSourceProvider(new String[] {"common_types"}, database)
+                                getFlinkSourceProvider(
+                                                new String[] {"common_types"},
+                                                database,
+                                                useLegacyJsonFormat)
                                         .getSource(),
                                 WatermarkStrategy.noWatermarks(),
                                 "Event-Source")
@@ -446,7 +484,7 @@ private void testCommonDataTypes(UniqueDatabase database) throws Exception {
         }
 
         expectedSnapshot[30] = null;
-        // The json string from binlog will remove useless space
+        // Legacy format removes useless space in json string from binlog
         expectedSnapshot[44] = BinaryStringData.fromString("{\"key1\":\"value1\"}");
         Object[] expectedStreamRecord = expectedSnapshot;
 
@@ -457,6 +495,66 @@ private void testCommonDataTypes(UniqueDatabase database) throws Exception {
                 .isEqualTo(expectedStreamRecord);
     }
 
+    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;
+        List<Event> streamResults =
+                MySqSourceTestUtils.fetchResultsAndCreateTableEvent(iterator, 1).f0;
+        RecordData streamRecord = ((DataChangeEvent) streamResults.get(0)).after();
+
+        expectedSnapshot[4] = null;
+
+        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);
+        }
+    }
+
     private Instant toInstant(String ts) {
         return Timestamp.valueOf(ts).toLocalDateTime().atZone(ZoneId.of("UTC")).toInstant();
     }
@@ -468,9 +566,13 @@ private void testTimeDataTypes(
             Object[] expectedStreamRecord)
             throws Exception {
         database.createAndInitialize();
+        Boolean useLegacyJsonFormat = true;
         CloseableIterator<Event> iterator =
                 env.fromSource(
-                                getFlinkSourceProvider(new String[] {"time_types"}, database)
+                                getFlinkSourceProvider(
+                                                new String[] {"time_types"},
+                                                database,
+                                                useLegacyJsonFormat)
                                         .getSource(),
                                 WatermarkStrategy.noWatermarks(),
                                 "Event-Source")
@@ -498,7 +600,7 @@ private void testTimeDataTypes(
     }
 
     private FlinkSourceProvider getFlinkSourceProvider(
-            String[] captureTables, UniqueDatabase database) {
+            String[] captureTables, UniqueDatabase database, Boolean useLegacyJsonFormat) {
         String[] captureTableIds =
                 Arrays.stream(captureTables)
                         .map(tableName -> database.getDatabaseName() + "." + tableName)
@@ -517,7 +619,8 @@ private FlinkSourceProvider getFlinkSourceProvider(
                         .username(database.getUsername())
                         .password(database.getPassword())
                         .serverTimeZone(ZoneId.of("UTC").toString())
-                        .serverId(MySqSourceTestUtils.getServerId(env.getParallelism()));
+                        .serverId(MySqSourceTestUtils.getServerId(env.getParallelism()))
+                        .useLegacyJsonFormat(useLegacyJsonFormat);
         return (FlinkSourceProvider) new MySqlDataSource(configFactory).getEventSourceProvider();
     }
 
@@ -577,4 +680,12 @@ private FlinkSourceProvider getFlinkSourceProvider(
                     DataTypes.STRING(),
                     DataTypes.STRING(),
                     DataTypes.STRING());
+
+    private static final RowType JSON_TYPES =
+            RowType.of(
+                    DataTypes.DECIMAL(20, 0).notNull(),
+                    DataTypes.STRING(),
+                    DataTypes.STRING(),
+                    DataTypes.STRING(),
+                    DataTypes.INT());
 }

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

@@ -357,10 +357,71 @@ public void testMysql8PrecisionTypesSchema() {
         assertThat(actualSchema).isEqualTo(expectedSchema);
     }
 
+    @Test
+    public void testMysql57AccessJsonTypesSchema() {
+        fullTypesMySql57Database.createAndInitialize();
+
+        String[] tables = new String[] {"json_types"};
+        MySqlMetadataAccessor metadataAccessor =
+                getMetadataAccessor(tables, fullTypesMySql57Database);
+
+        Schema actualSchema =
+                metadataAccessor.getTableSchema(
+                        TableId.tableId(fullTypesMySql57Database.getDatabaseName(), "json_types"));
+        Schema expectedSchema =
+                Schema.newBuilder()
+                        .primaryKey("id")
+                        .fromRowDataType(
+                                RowType.of(
+                                        new DataType[] {
+                                            DataTypes.DECIMAL(20, 0).notNull(),
+                                            DataTypes.STRING(),
+                                            DataTypes.STRING(),
+                                            DataTypes.STRING(),
+                                            DataTypes.INT()
+                                        },
+                                        new String[] {
+                                            "id", "json_c0", "json_c1", "json_c2", "int_c",
+                                        }))
+                        .build();
+        assertThat(actualSchema).isEqualTo(expectedSchema);
+    }
+
+    @Test
+    public void testMysql8AccessJsonTypesSchema() {
+        fullTypesMySql57Database.createAndInitialize();
+
+        String[] tables = new String[] {"json_types"};
+        MySqlMetadataAccessor metadataAccessor =
+                getMetadataAccessor(tables, fullTypesMySql57Database);
+
+        Schema actualSchema =
+                metadataAccessor.getTableSchema(
+                        TableId.tableId(fullTypesMySql57Database.getDatabaseName(), "json_types"));
+        Schema expectedSchema =
+                Schema.newBuilder()
+                        .primaryKey("id")
+                        .fromRowDataType(
+                                RowType.of(
+                                        new DataType[] {
+                                            DataTypes.DECIMAL(20, 0).notNull(),
+                                            DataTypes.STRING(),
+                                            DataTypes.STRING(),
+                                            DataTypes.STRING(),
+                                            DataTypes.INT()
+                                        },
+                                        new String[] {
+                                            "id", "json_c0", "json_c1", "json_c2", "int_c",
+                                        }))
+                        .build();
+        assertThat(actualSchema).isEqualTo(expectedSchema);
+    }
+
     private void testAccessDatabaseAndTable(UniqueDatabase database) {
         database.createAndInitialize();
 
-        String[] tables = new String[] {"common_types", "time_types", "precision_types"};
+        String[] tables =
+                new String[] {"common_types", "time_types", "precision_types", "json_types"};
         MySqlMetadataAccessor metadataAccessor = getMetadataAccessor(tables, database);
 
         assertThatThrownBy(metadataAccessor::listNamespaces)

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

@@ -180,4 +180,22 @@ VALUES (DEFAULT,
         23,
         29,
         31,
-        37);
+        37);
+
+CREATE TABLE json_types
+(
+    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
+       );