diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializer.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializer.java index cce2258a38f..118656c7d7e 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializer.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializer.java @@ -18,9 +18,12 @@ package org.apache.flink.cdc.connectors.mysql.source; import org.apache.flink.cdc.common.annotation.Internal; +import org.apache.flink.cdc.common.data.RecordData; import org.apache.flink.cdc.common.data.binary.BinaryStringData; +import org.apache.flink.cdc.common.event.DataChangeEvent; import org.apache.flink.cdc.common.event.SchemaChangeEvent; import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.types.DataType; import org.apache.flink.cdc.connectors.mysql.source.parser.CustomMySqlAntlrDdlParser; import org.apache.flink.cdc.connectors.mysql.table.MySqlReadableMetadata; import org.apache.flink.cdc.debezium.event.DebeziumEventDeserializationSchema; @@ -38,6 +41,8 @@ import org.apache.kafka.connect.data.Schema; import org.apache.kafka.connect.data.Struct; import org.apache.kafka.connect.source.SourceRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.IOException; import java.nio.ByteBuffer; @@ -45,8 +50,11 @@ import java.util.Collections; import java.util.HashMap; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import static io.debezium.connector.AbstractSourceInfo.DATABASE_NAME_KEY; import static io.debezium.connector.AbstractSourceInfo.TABLE_NAME_KEY; @@ -58,6 +66,8 @@ public class MySqlEventDeserializer extends DebeziumEventDeserializationSchema { private static final long serialVersionUID = 1L; + private static final Logger LOG = LoggerFactory.getLogger(MySqlEventDeserializer.class); + public static final String SCHEMA_CHANGE_EVENT_KEY_NAME = "io.debezium.connector.mysql.SchemaChangeKey"; @@ -72,6 +82,7 @@ public class MySqlEventDeserializer extends DebeziumEventDeserializationSchema { private final List readableMetadataList; private final boolean isTableIdCaseInsensitive; + private transient ConcurrentMap dataTypeCache; public MySqlEventDeserializer( DebeziumChangelogMode changelogMode, @@ -102,6 +113,36 @@ public MySqlEventDeserializer( this.isTableIdCaseInsensitive = isTableIdCaseInsensitive; } + @Override + public List deserializeDataChangeRecord(SourceRecord record) throws Exception { + Envelope.Operation op = Envelope.operationFor(record); + TableId tableId = getTableId(record); + + Struct value = (Struct) record.value(); + Schema valueSchema = record.valueSchema(); + Map meta = getMetadata(record); + + if (op == Envelope.Operation.CREATE || op == Envelope.Operation.READ) { + RecordData after = extractAfterDataRecord(tableId, value, valueSchema); + return Collections.singletonList(DataChangeEvent.insertEvent(tableId, after, meta)); + } else if (op == Envelope.Operation.DELETE) { + RecordData before = extractBeforeDataRecord(tableId, value, valueSchema); + return Collections.singletonList(DataChangeEvent.deleteEvent(tableId, before, meta)); + } else if (op == Envelope.Operation.UPDATE) { + RecordData after = extractAfterDataRecord(tableId, value, valueSchema); + if (changelogMode == DebeziumChangelogMode.ALL) { + RecordData before = extractBeforeDataRecord(tableId, value, valueSchema); + return Collections.singletonList( + DataChangeEvent.updateEvent(tableId, before, after, meta)); + } + return Collections.singletonList( + DataChangeEvent.updateEvent(tableId, null, after, meta)); + } else { + LOG.trace("Received {} operation, skip", op); + return Collections.emptyList(); + } + } + @Override protected List deserializeSchemaChangeRecord(SourceRecord record) { if (includeSchemaChanges) { @@ -121,14 +162,65 @@ protected List deserializeSchemaChangeRecord(SourceRecord rec historyRecord.document().getString(HistoryRecord.Fields.DDL_STATEMENTS); customParser.setCurrentDatabase(databaseName); customParser.parse(ddl, tables); - return customParser.getAndClearParsedEvents(); + List schemaChangeEvents = customParser.getAndClearParsedEvents(); + clearDataTypeCache(schemaChangeEvents); + return schemaChangeEvents; } catch (IOException e) { throw new IllegalStateException("Failed to parse the schema change : " + record, e); } } + getOrCreateDataTypeCache().clear(); return Collections.emptyList(); } + private RecordData extractBeforeDataRecord(TableId tableId, Struct value, Schema valueSchema) + throws Exception { + Schema beforeSchema = fieldSchema(valueSchema, Envelope.FieldName.BEFORE); + Struct beforeValue = fieldStruct(value, Envelope.FieldName.BEFORE); + return extractDataRecord(tableId, beforeValue, beforeSchema); + } + + private RecordData extractAfterDataRecord(TableId tableId, Struct value, Schema valueSchema) + throws Exception { + Schema afterSchema = fieldSchema(valueSchema, Envelope.FieldName.AFTER); + Struct afterValue = fieldStruct(value, Envelope.FieldName.AFTER); + return extractDataRecord(tableId, afterValue, afterSchema); + } + + private RecordData extractDataRecord(TableId tableId, Struct value, Schema valueSchema) + throws Exception { + DataType dataType = getOrInferDataType(tableId, value, valueSchema); + return convertDataRecord(value, valueSchema, dataType); + } + + private DataType getOrInferDataType(TableId tableId, Struct value, Schema valueSchema) { + ConcurrentMap cache = getOrCreateDataTypeCache(); + DataType dataType = cache.get(tableId); + if (dataType == null) { + // Rows from the same table share the same schema until a schema change event arrives. + dataType = + cache.computeIfAbsent( + tableId, key -> schemaDataTypeInference.infer(value, valueSchema)); + } + return dataType; + } + + private void clearDataTypeCache(List schemaChangeEvents) { + ConcurrentMap cache = getOrCreateDataTypeCache(); + for (SchemaChangeEvent schemaChangeEvent : schemaChangeEvents) { + // Keep cache invalidation aligned with the normalized cache key in case-insensitive + // mode. + cache.remove(normalizeTableId(schemaChangeEvent.tableId())); + } + } + + private ConcurrentMap getOrCreateDataTypeCache() { + if (dataTypeCache == null) { + dataTypeCache = new ConcurrentHashMap<>(); + } + return dataTypeCache; + } + @Override protected boolean isDataChangeRecord(SourceRecord record) { Schema valueSchema = record.valueSchema(); @@ -151,7 +243,7 @@ protected TableId getTableId(SourceRecord record) { Struct source = value.getStruct(Envelope.FieldName.SOURCE); String dbName = source.getString(DATABASE_NAME_KEY); String tableName = source.getString(TABLE_NAME_KEY); - return TableId.tableId(dbName, tableName); + return normalizeTableId(TableId.tableId(dbName, tableName)); } @Override @@ -202,4 +294,13 @@ protected Object convertToString(Object dbzObj, Schema schema) { return BinaryStringData.fromString(dbzObj.toString()); } } + + private TableId normalizeTableId(TableId tableId) { + if (isTableIdCaseInsensitive) { + return TableId.tableId( + tableId.getSchemaName().toLowerCase(Locale.ROOT), + tableId.getTableName().toLowerCase(Locale.ROOT)); + } + return tableId; + } } diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializerTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializerTest.java new file mode 100644 index 00000000000..611a80b21ed --- /dev/null +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializerTest.java @@ -0,0 +1,208 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.mysql.source; + +import org.apache.flink.cdc.common.data.RecordData; +import org.apache.flink.cdc.common.data.binary.BinaryStringData; +import org.apache.flink.cdc.common.event.DataChangeEvent; +import org.apache.flink.cdc.common.event.Event; +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.debezium.table.DebeziumChangelogMode; + +import io.debezium.data.Envelope; +import org.apache.kafka.connect.data.Schema; +import org.apache.kafka.connect.data.SchemaBuilder; +import org.apache.kafka.connect.data.Struct; +import org.apache.kafka.connect.source.SourceRecord; +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Field; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Unit tests for {@link MySqlEventDeserializer}. */ +class MySqlEventDeserializerTest { + + private static final TableId PRODUCTS = TableId.tableId("inventory", "products"); + + @Test + void testDataTypeCacheLifecycle() throws Exception { + MySqlEventDeserializer deserializer = createDeserializer(); + Schema rowSchema = rowSchema(); + + DataChangeEvent first = deserializeDataRecord(deserializer, rowSchema, 1, "scooter"); + assertRecord(first.after(), 1, "scooter"); + assertThat(getDataTypeCache(deserializer)).hasSize(1).containsKey(PRODUCTS); + + DataChangeEvent second = deserializeDataRecord(deserializer, rowSchema, 2, "car battery"); + assertRecord(second.after(), 2, "car battery"); + assertThat(getDataTypeCache(deserializer)).hasSize(1).containsKey(PRODUCTS); + + List schemaChangeEvents = deserializer.deserialize(schemaChangeRecord()); + assertThat(schemaChangeEvents).isEmpty(); + assertThat(getDataTypeCache(deserializer)).isEmpty(); + + DataChangeEvent third = deserializeDataRecord(deserializer, rowSchema, 3, "12-pack drill"); + assertRecord(third.after(), 3, "12-pack drill"); + assertThat(getDataTypeCache(deserializer)).hasSize(1).containsKey(PRODUCTS); + } + + @Test + void testDataTypeCacheIsSeparatedByTableId() throws Exception { + MySqlEventDeserializer deserializer = createDeserializer(); + Schema rowSchema = rowSchema(); + TableId customers = TableId.tableId("inventory", "customers"); + + deserializeDataRecord(deserializer, rowSchema, PRODUCTS, 1, "scooter"); + assertThat(getDataTypeCache(deserializer)).hasSize(1).containsKey(PRODUCTS); + + deserializeDataRecord(deserializer, rowSchema, customers, 2, "alice"); + assertThat(getDataTypeCache(deserializer)).hasSize(2).containsKeys(PRODUCTS, customers); + + deserializeDataRecord(deserializer, rowSchema, PRODUCTS, 3, "hammer"); + assertThat(getDataTypeCache(deserializer)).hasSize(2).containsKeys(PRODUCTS, customers); + } + + @Test + void testDataTypeCacheNormalizesCaseInsensitiveTableId() throws Exception { + MySqlEventDeserializer deserializer = createDeserializer(true); + Schema rowSchema = rowSchema(); + TableId upperCaseProducts = TableId.tableId("Inventory", "Products"); + + deserializeDataRecord(deserializer, rowSchema, upperCaseProducts, 1, "scooter"); + assertThat(getDataTypeCache(deserializer)).hasSize(1).containsKey(PRODUCTS); + + deserializeDataRecord(deserializer, rowSchema, PRODUCTS, 2, "car battery"); + assertThat(getDataTypeCache(deserializer)).hasSize(1).containsKey(PRODUCTS); + } + + private static MySqlEventDeserializer createDeserializer() { + return createDeserializer(false); + } + + private static MySqlEventDeserializer createDeserializer(boolean isTableIdCaseInsensitive) { + return new MySqlEventDeserializer( + DebeziumChangelogMode.ALL, + false, + Collections.emptyList(), + false, + false, + isTableIdCaseInsensitive); + } + + private static DataChangeEvent deserializeDataRecord( + MySqlEventDeserializer deserializer, Schema rowSchema, int id, String name) + throws Exception { + return deserializeDataRecord(deserializer, rowSchema, PRODUCTS, id, name); + } + + private static DataChangeEvent deserializeDataRecord( + MySqlEventDeserializer deserializer, + Schema rowSchema, + TableId tableId, + int id, + String name) + throws Exception { + return deserializer + .deserializeDataChangeRecord(dataRecord(tableId, rowSchema, id, name)) + .get(0); + } + + private static SourceRecord dataRecord(TableId tableId, Schema rowSchema, int id, String name) { + Schema sourceSchema = + SchemaBuilder.struct() + .field("db", Schema.STRING_SCHEMA) + .field("table", Schema.STRING_SCHEMA) + .build(); + Schema valueSchema = + SchemaBuilder.struct() + .name("mysql-envelope") + .field(Envelope.FieldName.BEFORE, rowSchema) + .field(Envelope.FieldName.AFTER, rowSchema) + .field(Envelope.FieldName.SOURCE, sourceSchema) + .field(Envelope.FieldName.OPERATION, Schema.STRING_SCHEMA) + .build(); + Struct value = + new Struct(valueSchema) + .put(Envelope.FieldName.BEFORE, null) + .put(Envelope.FieldName.AFTER, row(rowSchema, id, name)) + .put(Envelope.FieldName.SOURCE, source(tableId, sourceSchema)) + .put(Envelope.FieldName.OPERATION, "c"); + return new SourceRecord( + Collections.emptyMap(), + Collections.emptyMap(), + "mysql_binlog_source." + tableId.identifier(), + null, + null, + null, + valueSchema, + value); + } + + private static Struct source(TableId tableId, Schema sourceSchema) { + return new Struct(sourceSchema) + .put("db", tableId.getSchemaName()) + .put("table", tableId.getTableName()); + } + + private static SourceRecord schemaChangeRecord() { + Schema keySchema = + SchemaBuilder.struct() + .name(MySqlEventDeserializer.SCHEMA_CHANGE_EVENT_KEY_NAME) + .build(); + return new SourceRecord( + Collections.emptyMap(), + Collections.emptyMap(), + "mysql_binlog_source", + null, + keySchema, + new Struct(keySchema), + null, + null); + } + + private static Schema rowSchema() { + return SchemaBuilder.struct() + .name("inventory.products.Value") + .optional() + .field("id", Schema.INT32_SCHEMA) + .field("name", Schema.STRING_SCHEMA) + .build(); + } + + private static Struct row(Schema schema, int id, String name) { + return new Struct(schema).put("id", id).put("name", name); + } + + private static void assertRecord(RecordData recordData, int id, String name) { + assertThat(recordData.getInt(0)).isEqualTo(id); + assertThat(recordData.getString(1)).isEqualTo(BinaryStringData.fromString(name)); + } + + @SuppressWarnings("unchecked") + private static Map getDataTypeCache(MySqlEventDeserializer deserializer) + throws Exception { + Field field = MySqlEventDeserializer.class.getDeclaredField("dataTypeCache"); + field.setAccessible(true); + Object cache = field.get(deserializer); + return cache == null ? Collections.emptyMap() : (Map) cache; + } +} diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java index 77897efb067..e2905feb12a 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java @@ -155,13 +155,24 @@ private RecordData extractAfterDataRecord(Struct value, Schema valueSchema) thro private RecordData extractDataRecord(Struct value, Schema valueSchema) throws Exception { DataType dataType = schemaDataTypeInference.infer(value, valueSchema); - return (RecordData) getOrCreateConverter(dataType).convert(value, valueSchema); + return convertDataRecord(value, valueSchema, dataType); } private DeserializationRuntimeConverter getOrCreateConverter(DataType type) { return CONVERTERS.computeIfAbsent(type, this::createConverter); } + /** + * Converts a Debezium row struct with a precomputed {@link DataType}. + * + *

Subclasses can reuse this helper when they cache or otherwise derive the row type outside + * of the default per-record inference path. + */ + protected final RecordData convertDataRecord( + Struct value, Schema valueSchema, DataType dataType) throws Exception { + return (RecordData) getOrCreateConverter(dataType).convert(value, valueSchema); + } + // ------------------------------------------------------------------------------------- // Runtime Converters // -------------------------------------------------------------------------------------