Merge branch 'master' into FLINK-36282-2

Author: leonardBang

.github/workflows/flink_cdc_base.yml

@@ -148,7 +148,7 @@ jobs:
           maven-version: 3.8.6
 
       - name: Compile and test
-        timeout-minutes: 60
+        timeout-minutes: 90
         run: |
           set -o pipefail
 

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

@@ -401,7 +401,19 @@ During a snapshot operation, the connector will query each included table to pro
         Schema change events are applied to a "shadow" table and then swapped with the original table later.
         <br>
         This is an experimental feature, and subject to change in the future.
-      </td> 
+      </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>

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

@@ -320,6 +320,18 @@ pipeline:
       <td>Boolean</td>
       <td>Whether treat TINYINT(1) as boolean, by default is true.</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

@@ -94,6 +94,7 @@
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.TABLES_EXCLUDE;
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.TREAT_TINYINT1_AS_BOOLEAN_ENABLED;
 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;
@@ -148,6 +149,7 @@ public DataSource createDataSource(Context context) {
         boolean scanBinlogNewlyAddedTableEnabled =
                 config.get(SCAN_BINLOG_NEWLY_ADDED_TABLE_ENABLED);
         boolean isParsingOnLineSchemaChanges = config.get(PARSE_ONLINE_SCHEMA_CHANGES);
+        boolean useLegacyJsonFormat = config.get(USE_LEGACY_JSON_FORMAT);
 
         validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1);
         validateIntegerOption(CHUNK_META_GROUP_SIZE, splitMetaGroupSize, 1);
@@ -198,7 +200,8 @@ public DataSource createDataSource(Context context) {
                         .jdbcProperties(getJdbcProperties(configMap))
                         .scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled)
                         .parseOnLineSchemaChanges(isParsingOnLineSchemaChanges)
-                        .treatTinyInt1AsBoolean(treatTinyInt1AsBoolean);
+                        .treatTinyInt1AsBoolean(treatTinyInt1AsBoolean)
+                        .useLegacyJsonFormat(useLegacyJsonFormat);
 
         List<TableId> tableIds = MySqlSchemaUtils.listTables(configFactory.createConfig(0), null);
 
@@ -332,6 +335,7 @@ public Set<ConfigOption<?>> optionalOptions() {
         options.add(METADATA_LIST);
         options.add(INCLUDE_COMMENTS_ENABLED);
         options.add(TREAT_TINYINT1_AS_BOOLEAN_ENABLED);
+        options.add(USE_LEGACY_JSON_FORMAT);
         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

@@ -305,4 +305,12 @@ public class MySqlDataSourceOptions {
                     .booleanType()
                     .defaultValue(true)
                     .withDescription("Whether treat TINYINT(1) as boolean, by default is true. ");
+
+    @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.");
 }

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

@@ -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

@@ -222,13 +222,15 @@ public void testMysql57PrecisionTypesSchema() {
         fullTypesMySql57Database.createAndInitialize();
 
         String[] tables = new String[] {"precision_types"};
+
         MySqlMetadataAccessor metadataAccessor =
                 getMetadataAccessor(tables, fullTypesMySql57Database, true);
 
         Schema actualSchema =
                 metadataAccessor.getTableSchema(
                         TableId.tableId(
                                 fullTypesMySql57Database.getDatabaseName(), "precision_types"));
+
         Schema expectedSchema =
                 Schema.newBuilder()
                         .primaryKey("id")
@@ -304,6 +306,7 @@ public void testMysql8PrecisionTypesSchema() {
                 metadataAccessor.getTableSchema(
                         TableId.tableId(
                                 fullTypesMySql8Database.getDatabaseName(), "precision_types"));
+
         Schema expectedSchema =
                 Schema.newBuilder()
                         .primaryKey("id")
@@ -370,7 +373,8 @@ public void testMysql8PrecisionTypesSchema() {
     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, true);
 
         assertThatThrownBy(metadataAccessor::listNamespaces)
@@ -528,6 +532,66 @@ private MySqlMetadataAccessor getMetadataAccessor(
         return new MySqlMetadataAccessor(sourceConfig);
     }
 
+    @Test
+    public void testMysql57AccessJsonTypesSchema() {
+        fullTypesMySql57Database.createAndInitialize();
+
+        String[] tables = new String[] {"json_types"};
+        MySqlMetadataAccessor metadataAccessor =
+                getMetadataAccessor(tables, fullTypesMySql57Database, true);
+
+        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, true);
+
+        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 MySqlSourceConfig getConfig(
             String[] captureTables, UniqueDatabase database, boolean tinyint1IsBit) {
         String[] captureTableIds =