Skip to content

[FLINK-40038][pipeline-connector/mysql] Cache inferred row DataType in MySQL deserialization #4458

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
monologuist wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[FLINK-40038][pipeline-connector/mysql] Cache inferred row DataType in MySQL deserialization #4458

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
105 changes: 103 additions & 2 deletions ...ql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -38,15 +41,20 @@
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;
import java.util.ArrayList;
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;
Expand All @@ -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";

Expand All @@ -72,6 +82,7 @@ public class MySqlEventDeserializer extends DebeziumEventDeserializationSchema {

private final List<MySqlReadableMetadata> readableMetadataList;
private final boolean isTableIdCaseInsensitive;
private transient ConcurrentMap<TableId, DataType> dataTypeCache;

public MySqlEventDeserializer(
DebeziumChangelogMode changelogMode,
Expand Down Expand Up @@ -102,6 +113,36 @@ public MySqlEventDeserializer(
this.isTableIdCaseInsensitive = isTableIdCaseInsensitive;
}

@Override
public List<DataChangeEvent> 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<String, String> 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<SchemaChangeEvent> deserializeSchemaChangeRecord(SourceRecord record) {
if (includeSchemaChanges) {
Expand All @@ -121,14 +162,65 @@ protected List<SchemaChangeEvent> deserializeSchemaChangeRecord(SourceRecord rec
historyRecord.document().getString(HistoryRecord.Fields.DDL_STATEMENTS);
customParser.setCurrentDatabase(databaseName);
customParser.parse(ddl, tables);
return customParser.getAndClearParsedEvents();
List<SchemaChangeEvent> 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<TableId, DataType> 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<SchemaChangeEvent> schemaChangeEvents) {
ConcurrentMap<TableId, DataType> 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<TableId, DataType> getOrCreateDataTypeCache() {
if (dataTypeCache == null) {
dataTypeCache = new ConcurrentHashMap<>();
}
return dataTypeCache;
}

@Override
protected boolean isDataChangeRecord(SourceRecord record) {
Schema valueSchema = record.valueSchema();
Expand All @@ -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
Expand Down Expand Up @@ -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;
}
}
208 changes: 208 additions & 0 deletions ...rc/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlEventDeserializerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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<? extends Event> 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<TableId, ?> 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<TableId, ?>) cache;
}
}
Loading
Loading