Copilot commented on code in PR #69948: URL: https://github.com/apache/doris/pull/69948#discussion_r4438714349
########## fs_brokers/cdc_client/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/TableMapEventMetadataDeserializer.java: ########## @@ -0,0 +1,231 @@ +/* + * 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 com.github.shyiko.mysql.binlog.event.deserialization; + +import com.github.shyiko.mysql.binlog.event.TableMapEventMetadata; +import com.github.shyiko.mysql.binlog.event.TableMapEventMetadata.DefaultCharset; +import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.BitSet; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.logging.Level; +import java.util.logging.Logger; + +/** + * Copied from Flink CDC 3.6.0-1.20 (flink-connector-mysql-cdc). + * + * <p>Lines 109-111 and 204: decode VECTOR_DIMENSIONALITY (field 13) as packed integers and + * store the dimensions in TableMapEventMetadata; retain the signedness block boundary handling. + */ +public class TableMapEventMetadataDeserializer { + + private final Logger logger = Logger.getLogger(getClass().getName()); + + public TableMapEventMetadata deserialize( + ByteArrayInputStream inputStream, int nColumns, int nNumericColumns) + throws IOException { + int remainingBytes = inputStream.available(); + if (remainingBytes <= 0) { + return null; + } + + TableMapEventMetadata result = new TableMapEventMetadata(); + + for (; remainingBytes > 0; inputStream.enterBlock(remainingBytes)) { + int code = inputStream.readInteger(1); + + MetadataFieldType fieldType = MetadataFieldType.byCode(code); + if (fieldType == null) { + throw new IOException("Unsupported table metadata field type " + code); + } + if (MetadataFieldType.UNKNOWN_METADATA_FIELD_TYPE.equals(fieldType)) { + if (logger.isLoggable(Level.FINE)) { + logger.fine("Received metadata field of unknown type"); + } + continue; + } + + int fieldLength = inputStream.readPackedInteger(); + + remainingBytes = inputStream.available(); + inputStream.enterBlock(fieldLength); + + switch (fieldType) { + case SIGNEDNESS: + result.setSignedness(readBooleanList(inputStream, nNumericColumns)); + inputStream.skipToTheEndOfTheBlock(); + break; + case DEFAULT_CHARSET: + result.setDefaultCharset(readDefaultCharset(inputStream)); + break; + case COLUMN_CHARSET: + result.setColumnCharsets(readIntegers(inputStream)); + break; + case COLUMN_NAME: + result.setColumnNames(readColumnNames(inputStream)); + break; + case SET_STR_VALUE: + result.setSetStrValues(readTypeValues(inputStream)); + break; + case ENUM_STR_VALUE: + result.setEnumStrValues(readTypeValues(inputStream)); + break; + case GEOMETRY_TYPE: + result.setGeometryTypes(readIntegers(inputStream)); + break; + case SIMPLE_PRIMARY_KEY: + result.setSimplePrimaryKeys(readIntegers(inputStream)); + break; + case PRIMARY_KEY_WITH_PREFIX: + result.setPrimaryKeysWithPrefix(readIntegerPairs(inputStream)); + break; + case ENUM_AND_SET_DEFAULT_CHARSET: + result.setEnumAndSetDefaultCharset(readDefaultCharset(inputStream)); + break; + case ENUM_AND_SET_COLUMN_CHARSET: + result.setEnumAndSetColumnCharsets(readIntegers(inputStream)); + result.setVisibility(readBooleanList(inputStream, nColumns)); + break; Review Comment: For ENUM_AND_SET_COLUMN_CHARSET metadata (field 11), readIntegers() consumes the entire length-delimited block. The following readBooleanList() then tries to read at least one more byte for any nonempty table, so a valid TABLE_MAP event containing this field fails decoding and stops binlog ingestion. Visibility has its own field-12 branch below; remove this extra visibility read. ########## fs_brokers/cdc_client/src/main/java/io/debezium/connector/mysql/MySqlValueConverters.java: ########## @@ -0,0 +1,1187 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.mysql; + +import com.github.shyiko.mysql.binlog.event.deserialization.json.JsonBinary; +import io.debezium.DebeziumException; +import io.debezium.annotation.Immutable; +import io.debezium.config.CommonConnectorConfig.BinaryHandlingMode; +import io.debezium.connector.mysql.antlr.MySqlAntlrDdlParser; +import io.debezium.data.Json; +import io.debezium.jdbc.JdbcValueConverters; +import io.debezium.jdbc.TemporalPrecisionMode; +import io.debezium.relational.Column; +import io.debezium.relational.Table; +import io.debezium.relational.ValueConverter; +import io.debezium.time.Year; +import io.debezium.util.Strings; +import org.apache.kafka.connect.data.Decimal; +import org.apache.kafka.connect.data.Field; +import org.apache.kafka.connect.data.Schema; +import org.apache.kafka.connect.data.SchemaBuilder; +import org.apache.kafka.connect.errors.ConnectException; +import org.apache.kafka.connect.source.SourceRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.math.BigDecimal; +import java.nio.ByteOrder; +import java.nio.charset.Charset; +import java.nio.charset.IllegalCharsetNameException; +import java.nio.charset.StandardCharsets; +import java.sql.Timestamp; +import java.sql.Types; +import java.time.Duration; +import java.time.LocalDate; +import java.time.ZoneOffset; +import java.time.temporal.ChronoField; +import java.time.temporal.ChronoUnit; +import java.time.temporal.Temporal; +import java.time.temporal.TemporalAdjuster; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * Copied from Flink CDC 3.6.0-1.20 (flink-connector-mysql-cdc). + * + * <p>Lines 249-251: add the FloatVector schema. + * + * <p>Lines 261-263 and 419-472: port VECTOR conversions with method-level source notes. + * Retain the existing FLOAT conversions and SQL NULL handling. + */ +@Immutable +public class MySqlValueConverters extends JdbcValueConverters { + + @FunctionalInterface + public static interface ParsingErrorHandler { + void error(String message, Exception exception); + } + + private static final Logger LOGGER = LoggerFactory.getLogger(MySqlValueConverters.class); + + /** Used to parse values of TIME columns. Format: 000:00:00.000000. */ + private static final Pattern TIME_FIELD_PATTERN = + Pattern.compile("(\\-?[0-9]*):([0-9]*):([0-9]*)(\\.([0-9]*))?"); + + /** Used to parse values of DATE columns. Format: 000-00-00. */ + private static final Pattern DATE_FIELD_PATTERN = Pattern.compile("([0-9]*)-([0-9]*)-([0-9]*)"); + + /** Used to parse values of TIMESTAMP columns. Format: 000-00-00 00:00:00.000. */ + private static final Pattern TIMESTAMP_FIELD_PATTERN = + Pattern.compile("([0-9]*)-([0-9]*)-([0-9]*) .*"); + + /** + * A utility method that adjusts <a + * href="https://dev.mysql.com/doc/refman/5.7/en/two-digit-years.html";>ambiguous</a> 2-digit + * year values of DATETIME, DATE, and TIMESTAMP types using these MySQL-specific rules: + * + * <ul> + * <li>Year values in the range 00-69 are converted to 2000-2069. + * <li>Year values in the range 70-99 are converted to 1970-1999. + * </ul> + * + * @param temporal the temporal instance to adjust; may not be null + * @return the possibly adjusted temporal instance; never null + */ + public static Temporal adjustTemporal(Temporal temporal) { + if (temporal.isSupported(ChronoField.YEAR)) { + int year = temporal.get(ChronoField.YEAR); + if (0 <= year && year <= 69) { + temporal = temporal.plus(2000, ChronoUnit.YEARS); + } else if (70 <= year && year <= 99) { + temporal = temporal.plus(1900, ChronoUnit.YEARS); + } + } + return temporal; + } + + private final ParsingErrorHandler parsingErrorHandler; + + /** + * Create a new instance that always uses UTC for the default time zone when_needed converting + * values without timezone information to values that require timezones. + * + * <p> + * + * @param decimalMode how {@code DECIMAL} and {@code NUMERIC} values should be treated; may be + * null if {@link io.debezium.jdbc.JdbcValueConverters.DecimalMode#PRECISE} is to be used + * @param temporalPrecisionMode temporal precision mode based on {@link + * io.debezium.jdbc.TemporalPrecisionMode} + * @param bigIntUnsignedMode how {@code BIGINT UNSIGNED} values should be treated; may be null + * if {@link io.debezium.jdbc.JdbcValueConverters.BigIntUnsignedMode#PRECISE} is to be used + * @param binaryMode how binary columns should be represented + */ + public MySqlValueConverters( + DecimalMode decimalMode, + TemporalPrecisionMode temporalPrecisionMode, + BigIntUnsignedMode bigIntUnsignedMode, + BinaryHandlingMode binaryMode) { + this( + decimalMode, + temporalPrecisionMode, + bigIntUnsignedMode, + binaryMode, + x -> x, + MySqlValueConverters::defaultParsingErrorHandler); + } + + /** + * Create a new instance that always uses UTC for the default time zone when converting values + * without timezone information to values that require timezones. + * + * <p> + * + * @param decimalMode how {@code DECIMAL} and {@code NUMERIC} values should be treated; may be + * null if {@link io.debezium.jdbc.JdbcValueConverters.DecimalMode#PRECISE} is to be used + * @param temporalPrecisionMode temporal precision mode based on {@link + * io.debezium.jdbc.TemporalPrecisionMode} + * @param bigIntUnsignedMode how {@code BIGINT UNSIGNED} values should be treated; may be null + * if {@link io.debezium.jdbc.JdbcValueConverters.BigIntUnsignedMode#PRECISE} is to be used + * @param binaryMode how binary columns should be represented + * @param adjuster a temporal adjuster to make a database specific time modification before + * conversion + * @param handler for errors during postponed binlog parsing + */ + public MySqlValueConverters( + DecimalMode decimalMode, + TemporalPrecisionMode temporalPrecisionMode, + BigIntUnsignedMode bigIntUnsignedMode, + BinaryHandlingMode binaryMode, + TemporalAdjuster adjuster, + ParsingErrorHandler parsingErrorHandler) { + super( + decimalMode, + temporalPrecisionMode, + ZoneOffset.UTC, + adjuster, + bigIntUnsignedMode, + binaryMode); + this.parsingErrorHandler = parsingErrorHandler; + } + + @Override + protected ByteOrder byteOrderOfBitType() { + return ByteOrder.BIG_ENDIAN; + } + + @Override + public SchemaBuilder schemaBuilder(Column column) { + // Handle a few MySQL-specific types based upon how they are handled by the MySQL binlog + // client ... + String typeName = column.typeName().toUpperCase(); + if (matches(typeName, "JSON")) { + return Json.builder(); + } + if (matches(typeName, "POINT")) { + return io.debezium.data.geometry.Point.builder(); + } + if (matches(typeName, "GEOMETRY") + || matches(typeName, "LINESTRING") + || matches(typeName, "POLYGON") + || matches(typeName, "MULTIPOINT") + || matches(typeName, "MULTILINESTRING") + || matches(typeName, "MULTIPOLYGON") + || isGeometryCollection(typeName)) { + return io.debezium.data.geometry.Geometry.builder(); + } + if (matches(typeName, "YEAR")) { + return Year.builder(); + } + if (matches(typeName, "ENUM")) { + String commaSeparatedOptions = extractEnumAndSetOptionsAsString(column); + return io.debezium.data.Enum.builder(commaSeparatedOptions); + } + if (matches(typeName, "SET")) { + String commaSeparatedOptions = extractEnumAndSetOptionsAsString(column); + return io.debezium.data.EnumSet.builder(commaSeparatedOptions); + } + if (matches(typeName, "SMALLINT UNSIGNED") + || matches(typeName, "SMALLINT UNSIGNED ZEROFILL") + || matches(typeName, "INT2 UNSIGNED") + || matches(typeName, "INT2 UNSIGNED ZEROFILL")) { + // In order to capture unsigned SMALLINT 16-bit data source, INT32 will be required to + // safely capture all valid values + // Source: + // https://kafka.apache.org/0102/javadoc/org/apache/kafka/connect/data/Schema.Type.html + return SchemaBuilder.int32(); + } + if (matches(typeName, "INT UNSIGNED") + || matches(typeName, "INT UNSIGNED ZEROFILL") + || matches(typeName, "INT4 UNSIGNED") + || matches(typeName, "INT4 UNSIGNED ZEROFILL")) { + // In order to capture unsigned INT 32-bit data source, INT64 will be required to safely + // capture all valid values + // Source: + // https://kafka.apache.org/0102/javadoc/org/apache/kafka/connect/data/Schema.Type.html + return SchemaBuilder.int64(); + } + if (matches(typeName, "BIGINT UNSIGNED") + || matches(typeName, "BIGINT UNSIGNED ZEROFILL") + || matches(typeName, "INT8 UNSIGNED") + || matches(typeName, "INT8 UNSIGNED ZEROFILL")) { + switch (super.bigIntUnsignedMode) { + case LONG: + return SchemaBuilder.int64(); + case PRECISE: + // In order to capture unsigned INT 64-bit data source, + // org.apache.kafka.connect.data.Decimal:Byte will be required to safely capture + // all valid values with scale of 0 + // Source: + // https://kafka.apache.org/0102/javadoc/org/apache/kafka/connect/data/Schema.Type.html + return Decimal.builder(0); + } + } + if ((matches(typeName, "FLOAT") + || matches(typeName, "FLOAT UNSIGNED") + || matches(typeName, "FLOAT UNSIGNED ZEROFILL")) + && !column.scale().isPresent() + && column.length() <= 24) { + return SchemaBuilder.float32(); + } + if (matches(typeName, "VECTOR")) { + return floatVectorSchema(); + } + // Otherwise, let the base class handle it ... + return super.schemaBuilder(column); + } + + @Override + public ValueConverter converter(Column column, Field fieldDefn) { + // Handle a few MySQL-specific types based upon how they are handled by the MySQL binlog + // client ... + String typeName = column.typeName().toUpperCase(); + if (matches(typeName, "VECTOR")) { + return data -> convertVector(column, fieldDefn, data); + } + if (matches(typeName, "JSON")) { + return (data) -> convertJson(column, fieldDefn, data); + } + if (matches(typeName, "GEOMETRY") + || matches(typeName, "LINESTRING") + || matches(typeName, "POLYGON") + || matches(typeName, "MULTIPOINT") + || matches(typeName, "MULTILINESTRING") + || matches(typeName, "MULTIPOLYGON") + || isGeometryCollection(typeName)) { + return (data -> convertGeometry(column, fieldDefn, data)); + } + if (matches(typeName, "POINT")) { + // backwards compatibility + return (data -> convertPoint(column, fieldDefn, data)); + } + if (matches(typeName, "YEAR")) { + return (data) -> convertYearToInt(column, fieldDefn, data); + } + if (matches(typeName, "ENUM")) { + // Build up the character array based upon the column's type ... + List<String> options = extractEnumAndSetOptions(column); + return (data) -> convertEnumToString(options, column, fieldDefn, data); + } + if (matches(typeName, "SET")) { + // Build up the character array based upon the column's type ... + List<String> options = extractEnumAndSetOptions(column); + return (data) -> convertSetToString(options, column, fieldDefn, data); + } + if (matches(typeName, "TINYINT UNSIGNED") + || matches(typeName, "TINYINT UNSIGNED ZEROFILL") + || matches(typeName, "INT1 UNSIGNED") + || matches(typeName, "INT1 UNSIGNED ZEROFILL")) { + // Convert TINYINT UNSIGNED internally from SIGNED to UNSIGNED based on the boundary + // settings + return (data) -> convertUnsignedTinyint(column, fieldDefn, data); + } + if (matches(typeName, "SMALLINT UNSIGNED") + || matches(typeName, "SMALLINT UNSIGNED ZEROFILL") + || matches(typeName, "INT2 UNSIGNED") + || matches(typeName, "INT2 UNSIGNED ZEROFILL")) { + // Convert SMALLINT UNSIGNED internally from SIGNED to UNSIGNED based on the boundary + // settings + return (data) -> convertUnsignedSmallint(column, fieldDefn, data); + } + if (matches(typeName, "MEDIUMINT UNSIGNED") + || matches(typeName, "MEDIUMINT UNSIGNED ZEROFILL") + || matches(typeName, "INT3 UNSIGNED") + || matches(typeName, "INT3 UNSIGNED ZEROFILL") + || matches(typeName, "MIDDLEINT UNSIGNED") + || matches(typeName, "MIDDLEINT UNSIGNED ZEROFILL")) { + // Convert MEDIUMINT UNSIGNED internally from SIGNED to UNSIGNED based on the boundary + // settings + return (data) -> convertUnsignedMediumint(column, fieldDefn, data); + } + if (matches(typeName, "INT UNSIGNED") + || matches(typeName, "INT UNSIGNED ZEROFILL") + || matches(typeName, "INT4 UNSIGNED") + || matches(typeName, "INT4 UNSIGNED ZEROFILL")) { + // Convert INT UNSIGNED internally from SIGNED to UNSIGNED based on the boundary + // settings + return (data) -> convertUnsignedInt(column, fieldDefn, data); + } + if (matches(typeName, "BIGINT UNSIGNED") + || matches(typeName, "BIGINT UNSIGNED ZEROFILL") + || matches(typeName, "INT8 UNSIGNED") + || matches(typeName, "INT8 UNSIGNED ZEROFILL")) { + switch (super.bigIntUnsignedMode) { + case LONG: + return (data) -> convertBigInt(column, fieldDefn, data); + case PRECISE: + // Convert BIGINT UNSIGNED internally from SIGNED to UNSIGNED based on the + // boundary settings + return (data) -> convertUnsignedBigint(column, fieldDefn, data); + } + } + + // We have to convert bytes encoded in the column's character set ... + switch (column.jdbcType()) { + case Types.CHAR: // variable-length + case Types.VARCHAR: // variable-length + case Types.LONGVARCHAR: // variable-length + case Types.CLOB: // variable-length + case Types.NCHAR: // fixed-length + case Types.NVARCHAR: // fixed-length + case Types.LONGNVARCHAR: // fixed-length + case Types.NCLOB: // fixed-length + case Types.DATALINK: + case Types.SQLXML: + Charset charset = charsetFor(column); + if (charset != null) { + logger.debug("Using {} charset by default for column: {}", charset, column); + return (data) -> convertString(column, fieldDefn, charset, data); + } + logger.warn( + "Using UTF-8 charset by default for column without charset: {}", column); + return (data) -> convertString(column, fieldDefn, StandardCharsets.UTF_8, data); + case Types.TIME: + if (adaptiveTimeMicrosecondsPrecisionMode) { + return data -> convertDurationToMicroseconds(column, fieldDefn, data); + } + case Types.TIMESTAMP: + return ((ValueConverter) + (data -> convertTimestampToLocalDateTime(column, fieldDefn, data))) + .and(super.converter(column, fieldDefn)); + default: + break; + } + + // Otherwise, let the base class handle it ... + return super.converter(column, fieldDefn); + } + + /** + * Return the {@link Charset} instance with the MySQL-specific character set name used by the + * given column. + * + * @param column the column in which the character set is used; never null + * @return the Java {@link Charset}, or null if there is no mapping + */ + protected Charset charsetFor(Column column) { + String mySqlCharsetName = column.charsetName(); + if (mySqlCharsetName == null) { + logger.warn("Column is missing a character set: {}", column); + return null; + } + String encoding = MySqlConnection.getJavaEncodingForMysqlCharSet(mySqlCharsetName); + if (encoding == null) { + logger.debug( + "Column uses MySQL character set '{}', which has no mapping to a Java character set, will try it in lowercase", + mySqlCharsetName); + encoding = + MySqlConnection.getJavaEncodingForMysqlCharSet(mySqlCharsetName.toLowerCase()); + } + if (encoding == null) { + logger.warn( + "Column uses MySQL character set '{}', which has no mapping to a Java character set", + mySqlCharsetName); + } else { + try { + return Charset.forName(encoding); + } catch (IllegalCharsetNameException e) { + logger.error( + "Unable to load Java charset '{}' for column with MySQL character set '{}'", + encoding, + mySqlCharsetName); + } + } + return null; + } + + /** + * Copied from Debezium 3.0.8.Final SchemaFactory.datatypeFloatVectorSchema. + * Inlined the FloatVector constants because Debezium 1.9.8 does not provide that class. + */ + private SchemaBuilder floatVectorSchema() { + return SchemaBuilder.array(Schema.FLOAT32_SCHEMA) + .name("io.debezium.data.FloatVector") + .version(1); + } + + /** + * Copied from Debezium 3.0.8.Final BinlogValueConverters.convertVector. + * Calls local conversion methods and uses a List fallback to match the ARRAY schema. + */ + protected Object convertVector(Column column, Field fieldDefn, Object rawData) { + return convertValue(column, fieldDefn, rawData, Collections.emptyList(), (r) -> { + if (rawData instanceof float[] data) { + r.deliver(convertFloatVector(data)); + } + else if (rawData instanceof byte[] data) { + r.deliver(convertFloatVector(fieldDefn, data)); + } + }); + } + + /** + * Copied from Debezium 3.0.8.Final FloatVector.fromLogical(Schema, float[]). + * Renamed the method and removed the unused schema parameter. + */ + private List<Float> convertFloatVector(float[] value) { + final List<Float> ret = new ArrayList<>(value.length); + for (float v : value) { + ret.add(v); + } + return ret; + } + + /** + * Copied from Debezium 3.0.8.Final FloatVector.fromLogical(Field, byte[]). + * Renamed the method. + */ + private List<Float> convertFloatVector(Field fieldDfn, byte[] value) { + if (value.length % Float.BYTES != 0) { + LOGGER.warn("Cannot convert field '{}', the octet stream is not multiply of {}", fieldDfn.name(), Float.BYTES); + return Collections.emptyList(); Review Comment: A malformed VECTOR payload is logged but delivered as a successful empty array. For example, the new three-byte negative test makes a non-null source value become [] rather than failing conversion, allowing ingestion to advance past corrupted data. Throw a contextual conversion exception here and change truncatedFloat32ReturnsEmptyList to expect failure instead of silently fabricating a value. ########## fs_brokers/cdc_client/src/main/java/io/debezium/connector/mysql/MySqlValueConverters.java: ########## @@ -0,0 +1,1187 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.mysql; + +import com.github.shyiko.mysql.binlog.event.deserialization.json.JsonBinary; +import io.debezium.DebeziumException; +import io.debezium.annotation.Immutable; +import io.debezium.config.CommonConnectorConfig.BinaryHandlingMode; +import io.debezium.connector.mysql.antlr.MySqlAntlrDdlParser; +import io.debezium.data.Json; +import io.debezium.jdbc.JdbcValueConverters; +import io.debezium.jdbc.TemporalPrecisionMode; +import io.debezium.relational.Column; +import io.debezium.relational.Table; +import io.debezium.relational.ValueConverter; +import io.debezium.time.Year; +import io.debezium.util.Strings; +import org.apache.kafka.connect.data.Decimal; +import org.apache.kafka.connect.data.Field; +import org.apache.kafka.connect.data.Schema; +import org.apache.kafka.connect.data.SchemaBuilder; +import org.apache.kafka.connect.errors.ConnectException; +import org.apache.kafka.connect.source.SourceRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.math.BigDecimal; +import java.nio.ByteOrder; +import java.nio.charset.Charset; +import java.nio.charset.IllegalCharsetNameException; +import java.nio.charset.StandardCharsets; +import java.sql.Timestamp; +import java.sql.Types; +import java.time.Duration; +import java.time.LocalDate; +import java.time.ZoneOffset; +import java.time.temporal.ChronoField; +import java.time.temporal.ChronoUnit; +import java.time.temporal.Temporal; +import java.time.temporal.TemporalAdjuster; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * Copied from Flink CDC 3.6.0-1.20 (flink-connector-mysql-cdc). + * + * <p>Lines 249-251: add the FloatVector schema. + * + * <p>Lines 261-263 and 419-472: port VECTOR conversions with method-level source notes. + * Retain the existing FLOAT conversions and SQL NULL handling. + */ +@Immutable +public class MySqlValueConverters extends JdbcValueConverters { + + @FunctionalInterface + public static interface ParsingErrorHandler { + void error(String message, Exception exception); + } + + private static final Logger LOGGER = LoggerFactory.getLogger(MySqlValueConverters.class); + + /** Used to parse values of TIME columns. Format: 000:00:00.000000. */ + private static final Pattern TIME_FIELD_PATTERN = + Pattern.compile("(\\-?[0-9]*):([0-9]*):([0-9]*)(\\.([0-9]*))?"); + + /** Used to parse values of DATE columns. Format: 000-00-00. */ + private static final Pattern DATE_FIELD_PATTERN = Pattern.compile("([0-9]*)-([0-9]*)-([0-9]*)"); + + /** Used to parse values of TIMESTAMP columns. Format: 000-00-00 00:00:00.000. */ + private static final Pattern TIMESTAMP_FIELD_PATTERN = + Pattern.compile("([0-9]*)-([0-9]*)-([0-9]*) .*"); + + /** + * A utility method that adjusts <a + * href="https://dev.mysql.com/doc/refman/5.7/en/two-digit-years.html";>ambiguous</a> 2-digit + * year values of DATETIME, DATE, and TIMESTAMP types using these MySQL-specific rules: + * + * <ul> + * <li>Year values in the range 00-69 are converted to 2000-2069. + * <li>Year values in the range 70-99 are converted to 1970-1999. + * </ul> + * + * @param temporal the temporal instance to adjust; may not be null + * @return the possibly adjusted temporal instance; never null + */ + public static Temporal adjustTemporal(Temporal temporal) { + if (temporal.isSupported(ChronoField.YEAR)) { + int year = temporal.get(ChronoField.YEAR); + if (0 <= year && year <= 69) { + temporal = temporal.plus(2000, ChronoUnit.YEARS); + } else if (70 <= year && year <= 99) { + temporal = temporal.plus(1900, ChronoUnit.YEARS); + } + } + return temporal; + } + + private final ParsingErrorHandler parsingErrorHandler; + + /** + * Create a new instance that always uses UTC for the default time zone when_needed converting + * values without timezone information to values that require timezones. + * + * <p> + * + * @param decimalMode how {@code DECIMAL} and {@code NUMERIC} values should be treated; may be + * null if {@link io.debezium.jdbc.JdbcValueConverters.DecimalMode#PRECISE} is to be used + * @param temporalPrecisionMode temporal precision mode based on {@link + * io.debezium.jdbc.TemporalPrecisionMode} + * @param bigIntUnsignedMode how {@code BIGINT UNSIGNED} values should be treated; may be null + * if {@link io.debezium.jdbc.JdbcValueConverters.BigIntUnsignedMode#PRECISE} is to be used + * @param binaryMode how binary columns should be represented + */ + public MySqlValueConverters( + DecimalMode decimalMode, + TemporalPrecisionMode temporalPrecisionMode, + BigIntUnsignedMode bigIntUnsignedMode, + BinaryHandlingMode binaryMode) { + this( + decimalMode, + temporalPrecisionMode, + bigIntUnsignedMode, + binaryMode, + x -> x, + MySqlValueConverters::defaultParsingErrorHandler); + } + + /** + * Create a new instance that always uses UTC for the default time zone when converting values + * without timezone information to values that require timezones. + * + * <p> + * + * @param decimalMode how {@code DECIMAL} and {@code NUMERIC} values should be treated; may be + * null if {@link io.debezium.jdbc.JdbcValueConverters.DecimalMode#PRECISE} is to be used + * @param temporalPrecisionMode temporal precision mode based on {@link + * io.debezium.jdbc.TemporalPrecisionMode} + * @param bigIntUnsignedMode how {@code BIGINT UNSIGNED} values should be treated; may be null + * if {@link io.debezium.jdbc.JdbcValueConverters.BigIntUnsignedMode#PRECISE} is to be used + * @param binaryMode how binary columns should be represented + * @param adjuster a temporal adjuster to make a database specific time modification before + * conversion + * @param handler for errors during postponed binlog parsing + */ + public MySqlValueConverters( + DecimalMode decimalMode, + TemporalPrecisionMode temporalPrecisionMode, + BigIntUnsignedMode bigIntUnsignedMode, + BinaryHandlingMode binaryMode, + TemporalAdjuster adjuster, + ParsingErrorHandler parsingErrorHandler) { + super( + decimalMode, + temporalPrecisionMode, + ZoneOffset.UTC, + adjuster, + bigIntUnsignedMode, + binaryMode); + this.parsingErrorHandler = parsingErrorHandler; + } + + @Override + protected ByteOrder byteOrderOfBitType() { + return ByteOrder.BIG_ENDIAN; + } + + @Override + public SchemaBuilder schemaBuilder(Column column) { + // Handle a few MySQL-specific types based upon how they are handled by the MySQL binlog + // client ... + String typeName = column.typeName().toUpperCase(); + if (matches(typeName, "JSON")) { + return Json.builder(); + } + if (matches(typeName, "POINT")) { + return io.debezium.data.geometry.Point.builder(); + } + if (matches(typeName, "GEOMETRY") + || matches(typeName, "LINESTRING") + || matches(typeName, "POLYGON") + || matches(typeName, "MULTIPOINT") + || matches(typeName, "MULTILINESTRING") + || matches(typeName, "MULTIPOLYGON") + || isGeometryCollection(typeName)) { + return io.debezium.data.geometry.Geometry.builder(); + } + if (matches(typeName, "YEAR")) { + return Year.builder(); + } + if (matches(typeName, "ENUM")) { + String commaSeparatedOptions = extractEnumAndSetOptionsAsString(column); + return io.debezium.data.Enum.builder(commaSeparatedOptions); + } + if (matches(typeName, "SET")) { + String commaSeparatedOptions = extractEnumAndSetOptionsAsString(column); + return io.debezium.data.EnumSet.builder(commaSeparatedOptions); + } + if (matches(typeName, "SMALLINT UNSIGNED") + || matches(typeName, "SMALLINT UNSIGNED ZEROFILL") + || matches(typeName, "INT2 UNSIGNED") + || matches(typeName, "INT2 UNSIGNED ZEROFILL")) { + // In order to capture unsigned SMALLINT 16-bit data source, INT32 will be required to + // safely capture all valid values + // Source: + // https://kafka.apache.org/0102/javadoc/org/apache/kafka/connect/data/Schema.Type.html + return SchemaBuilder.int32(); + } + if (matches(typeName, "INT UNSIGNED") + || matches(typeName, "INT UNSIGNED ZEROFILL") + || matches(typeName, "INT4 UNSIGNED") + || matches(typeName, "INT4 UNSIGNED ZEROFILL")) { + // In order to capture unsigned INT 32-bit data source, INT64 will be required to safely + // capture all valid values + // Source: + // https://kafka.apache.org/0102/javadoc/org/apache/kafka/connect/data/Schema.Type.html + return SchemaBuilder.int64(); + } + if (matches(typeName, "BIGINT UNSIGNED") + || matches(typeName, "BIGINT UNSIGNED ZEROFILL") + || matches(typeName, "INT8 UNSIGNED") + || matches(typeName, "INT8 UNSIGNED ZEROFILL")) { + switch (super.bigIntUnsignedMode) { + case LONG: + return SchemaBuilder.int64(); + case PRECISE: + // In order to capture unsigned INT 64-bit data source, + // org.apache.kafka.connect.data.Decimal:Byte will be required to safely capture + // all valid values with scale of 0 + // Source: + // https://kafka.apache.org/0102/javadoc/org/apache/kafka/connect/data/Schema.Type.html + return Decimal.builder(0); + } + } + if ((matches(typeName, "FLOAT") + || matches(typeName, "FLOAT UNSIGNED") + || matches(typeName, "FLOAT UNSIGNED ZEROFILL")) + && !column.scale().isPresent() + && column.length() <= 24) { + return SchemaBuilder.float32(); + } + if (matches(typeName, "VECTOR")) { + return floatVectorSchema(); + } + // Otherwise, let the base class handle it ... + return super.schemaBuilder(column); + } + + @Override + public ValueConverter converter(Column column, Field fieldDefn) { + // Handle a few MySQL-specific types based upon how they are handled by the MySQL binlog + // client ... + String typeName = column.typeName().toUpperCase(); + if (matches(typeName, "VECTOR")) { + return data -> convertVector(column, fieldDefn, data); + } + if (matches(typeName, "JSON")) { + return (data) -> convertJson(column, fieldDefn, data); + } + if (matches(typeName, "GEOMETRY") + || matches(typeName, "LINESTRING") + || matches(typeName, "POLYGON") + || matches(typeName, "MULTIPOINT") + || matches(typeName, "MULTILINESTRING") + || matches(typeName, "MULTIPOLYGON") + || isGeometryCollection(typeName)) { + return (data -> convertGeometry(column, fieldDefn, data)); + } + if (matches(typeName, "POINT")) { + // backwards compatibility + return (data -> convertPoint(column, fieldDefn, data)); + } + if (matches(typeName, "YEAR")) { + return (data) -> convertYearToInt(column, fieldDefn, data); + } + if (matches(typeName, "ENUM")) { + // Build up the character array based upon the column's type ... + List<String> options = extractEnumAndSetOptions(column); + return (data) -> convertEnumToString(options, column, fieldDefn, data); + } + if (matches(typeName, "SET")) { + // Build up the character array based upon the column's type ... + List<String> options = extractEnumAndSetOptions(column); + return (data) -> convertSetToString(options, column, fieldDefn, data); + } + if (matches(typeName, "TINYINT UNSIGNED") + || matches(typeName, "TINYINT UNSIGNED ZEROFILL") + || matches(typeName, "INT1 UNSIGNED") + || matches(typeName, "INT1 UNSIGNED ZEROFILL")) { + // Convert TINYINT UNSIGNED internally from SIGNED to UNSIGNED based on the boundary + // settings + return (data) -> convertUnsignedTinyint(column, fieldDefn, data); + } + if (matches(typeName, "SMALLINT UNSIGNED") + || matches(typeName, "SMALLINT UNSIGNED ZEROFILL") + || matches(typeName, "INT2 UNSIGNED") + || matches(typeName, "INT2 UNSIGNED ZEROFILL")) { + // Convert SMALLINT UNSIGNED internally from SIGNED to UNSIGNED based on the boundary + // settings + return (data) -> convertUnsignedSmallint(column, fieldDefn, data); + } + if (matches(typeName, "MEDIUMINT UNSIGNED") + || matches(typeName, "MEDIUMINT UNSIGNED ZEROFILL") + || matches(typeName, "INT3 UNSIGNED") + || matches(typeName, "INT3 UNSIGNED ZEROFILL") + || matches(typeName, "MIDDLEINT UNSIGNED") + || matches(typeName, "MIDDLEINT UNSIGNED ZEROFILL")) { + // Convert MEDIUMINT UNSIGNED internally from SIGNED to UNSIGNED based on the boundary + // settings + return (data) -> convertUnsignedMediumint(column, fieldDefn, data); + } + if (matches(typeName, "INT UNSIGNED") + || matches(typeName, "INT UNSIGNED ZEROFILL") + || matches(typeName, "INT4 UNSIGNED") + || matches(typeName, "INT4 UNSIGNED ZEROFILL")) { + // Convert INT UNSIGNED internally from SIGNED to UNSIGNED based on the boundary + // settings + return (data) -> convertUnsignedInt(column, fieldDefn, data); + } + if (matches(typeName, "BIGINT UNSIGNED") + || matches(typeName, "BIGINT UNSIGNED ZEROFILL") + || matches(typeName, "INT8 UNSIGNED") + || matches(typeName, "INT8 UNSIGNED ZEROFILL")) { + switch (super.bigIntUnsignedMode) { + case LONG: + return (data) -> convertBigInt(column, fieldDefn, data); + case PRECISE: + // Convert BIGINT UNSIGNED internally from SIGNED to UNSIGNED based on the + // boundary settings + return (data) -> convertUnsignedBigint(column, fieldDefn, data); + } + } + + // We have to convert bytes encoded in the column's character set ... + switch (column.jdbcType()) { + case Types.CHAR: // variable-length + case Types.VARCHAR: // variable-length + case Types.LONGVARCHAR: // variable-length + case Types.CLOB: // variable-length + case Types.NCHAR: // fixed-length + case Types.NVARCHAR: // fixed-length + case Types.LONGNVARCHAR: // fixed-length + case Types.NCLOB: // fixed-length + case Types.DATALINK: + case Types.SQLXML: + Charset charset = charsetFor(column); + if (charset != null) { + logger.debug("Using {} charset by default for column: {}", charset, column); + return (data) -> convertString(column, fieldDefn, charset, data); + } + logger.warn( + "Using UTF-8 charset by default for column without charset: {}", column); + return (data) -> convertString(column, fieldDefn, StandardCharsets.UTF_8, data); + case Types.TIME: + if (adaptiveTimeMicrosecondsPrecisionMode) { + return data -> convertDurationToMicroseconds(column, fieldDefn, data); + } + case Types.TIMESTAMP: + return ((ValueConverter) + (data -> convertTimestampToLocalDateTime(column, fieldDefn, data))) + .and(super.converter(column, fieldDefn)); + default: + break; + } + + // Otherwise, let the base class handle it ... + return super.converter(column, fieldDefn); + } + + /** + * Return the {@link Charset} instance with the MySQL-specific character set name used by the + * given column. + * + * @param column the column in which the character set is used; never null + * @return the Java {@link Charset}, or null if there is no mapping + */ + protected Charset charsetFor(Column column) { + String mySqlCharsetName = column.charsetName(); + if (mySqlCharsetName == null) { + logger.warn("Column is missing a character set: {}", column); + return null; + } + String encoding = MySqlConnection.getJavaEncodingForMysqlCharSet(mySqlCharsetName); + if (encoding == null) { + logger.debug( + "Column uses MySQL character set '{}', which has no mapping to a Java character set, will try it in lowercase", + mySqlCharsetName); + encoding = + MySqlConnection.getJavaEncodingForMysqlCharSet(mySqlCharsetName.toLowerCase()); + } + if (encoding == null) { + logger.warn( + "Column uses MySQL character set '{}', which has no mapping to a Java character set", + mySqlCharsetName); + } else { + try { + return Charset.forName(encoding); + } catch (IllegalCharsetNameException e) { + logger.error( + "Unable to load Java charset '{}' for column with MySQL character set '{}'", + encoding, + mySqlCharsetName); + } + } + return null; + } + + /** + * Copied from Debezium 3.0.8.Final SchemaFactory.datatypeFloatVectorSchema. + * Inlined the FloatVector constants because Debezium 1.9.8 does not provide that class. + */ + private SchemaBuilder floatVectorSchema() { + return SchemaBuilder.array(Schema.FLOAT32_SCHEMA) + .name("io.debezium.data.FloatVector") + .version(1); + } + + /** + * Copied from Debezium 3.0.8.Final BinlogValueConverters.convertVector. + * Calls local conversion methods and uses a List fallback to match the ARRAY schema. + */ + protected Object convertVector(Column column, Field fieldDefn, Object rawData) { + return convertValue(column, fieldDefn, rawData, Collections.emptyList(), (r) -> { + if (rawData instanceof float[] data) { + r.deliver(convertFloatVector(data)); + } + else if (rawData instanceof byte[] data) { + r.deliver(convertFloatVector(fieldDefn, data)); + } + }); + } + + /** + * Copied from Debezium 3.0.8.Final FloatVector.fromLogical(Schema, float[]). + * Renamed the method and removed the unused schema parameter. + */ + private List<Float> convertFloatVector(float[] value) { + final List<Float> ret = new ArrayList<>(value.length); + for (float v : value) { + ret.add(v); + } + return ret; + } + + /** + * Copied from Debezium 3.0.8.Final FloatVector.fromLogical(Field, byte[]). + * Renamed the method. + */ + private List<Float> convertFloatVector(Field fieldDfn, byte[] value) { + if (value.length % Float.BYTES != 0) { + LOGGER.warn("Cannot convert field '{}', the octet stream is not multiply of {}", fieldDfn.name(), Float.BYTES); + return Collections.emptyList(); + } + + final List<Float> ret = new ArrayList<>(value.length); Review Comment: This overload receives bytes, but the list contains one Float per four bytes. Reserving value.length therefore allocates four times the required reference capacity for every snapshot/binlog vector, adding avoidable memory and GC overhead on the ingestion path. Size the list using the number of float elements instead. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
