[cdc-pipeline-connector][mysql] Support more mysql types
Jiabao Sun
1 year ago
committed by
Hang Ruan
parent
ec3d558485
commit
174282d3ef
@ -0,0 +1,43 @@
|
||||
/*
|
||||
* Copyright 2023 Ververica Inc.
|
||||
*
|
||||
* Licensed 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.ververica.cdc.connectors.mysql.source;
|
||||
|
||||
import com.ververica.cdc.common.annotation.Internal;
|
||||
import com.ververica.cdc.common.types.DataType;
|
||||
import com.ververica.cdc.common.types.DataTypes;
|
||||
import com.ververica.cdc.debezium.event.DebeziumSchemaDataTypeInference;
|
||||
import io.debezium.data.geometry.Geometry;
|
||||
import io.debezium.data.geometry.Point;
|
||||
import org.apache.kafka.connect.data.Schema;
|
||||
|
||||
/** {@link DataType} inference for MySQL debezium {@link Schema}. */
|
||||
@Internal
|
||||
public class MySqlSchemaDataTypeInference extends DebeziumSchemaDataTypeInference {
|
||||
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
protected DataType inferStruct(Object value, Schema schema) {
|
||||
// the Geometry datatype in MySQL will be converted to
|
||||
// a String with Json format
|
||||
if (Point.LOGICAL_NAME.equals(schema.name())
|
||||
|| Geometry.LOGICAL_NAME.equals(schema.name())) {
|
||||
return DataTypes.STRING();
|
||||
} else {
|
||||
return super.inferStruct(value, schema);
|
||||
}
|
||||
}
|
||||
}
|
||||
58
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/schema/MySqlCdcCommonTypeUtils.java → flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/com/ververica/cdc/connectors/mysql/utils/MySqlTypeUtils.java
58
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/schema/MySqlCdcCommonTypeUtils.java → flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/com/ververica/cdc/connectors/mysql/utils/MySqlTypeUtils.java
@ -1,67 +0,0 @@
|
||||
/*
|
||||
* Copyright 2023 Ververica Inc.
|
||||
*
|
||||
* Licensed 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.ververica.cdc.debezium.event;
|
||||
|
||||
import com.ververica.cdc.common.types.DataField;
|
||||
import com.ververica.cdc.common.types.DataType;
|
||||
import com.ververica.cdc.common.types.DataTypes;
|
||||
import org.apache.kafka.connect.data.Schema;
|
||||
|
||||
/** Utility class to convert {@link Schema} to {@link DataType}. */
|
||||
public class ConnectSchemaTypeInference {
|
||||
|
||||
public static DataType infer(Schema schema) {
|
||||
return schema.isOptional()
|
||||
? infer(schema, schema.type())
|
||||
: infer(schema, schema.type()).notNull();
|
||||
}
|
||||
|
||||
private static DataType infer(Schema schema, Schema.Type type) {
|
||||
switch (type) {
|
||||
case INT8:
|
||||
return DataTypes.TINYINT();
|
||||
case INT16:
|
||||
return DataTypes.SMALLINT();
|
||||
case INT32:
|
||||
return DataTypes.INT();
|
||||
case INT64:
|
||||
return DataTypes.BIGINT();
|
||||
case FLOAT32:
|
||||
return DataTypes.FLOAT();
|
||||
case FLOAT64:
|
||||
return DataTypes.DOUBLE();
|
||||
case BOOLEAN:
|
||||
return DataTypes.BOOLEAN();
|
||||
case STRING:
|
||||
return DataTypes.STRING();
|
||||
case BYTES:
|
||||
return DataTypes.BYTES();
|
||||
case ARRAY:
|
||||
return DataTypes.ARRAY(infer(schema.valueSchema()));
|
||||
case MAP:
|
||||
return DataTypes.MAP(infer(schema.keySchema()), infer(schema.valueSchema()));
|
||||
case STRUCT:
|
||||
return DataTypes.ROW(
|
||||
schema.fields().stream()
|
||||
.map(f -> DataTypes.FIELD(f.name(), infer(f.schema())))
|
||||
.toArray(DataField[]::new));
|
||||
default:
|
||||
throw new UnsupportedOperationException(
|
||||
"Unsupported type: " + schema.type().getName());
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,198 @@
|
||||
/*
|
||||
* Copyright 2023 Ververica Inc.
|
||||
*
|
||||
* Licensed 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.ververica.cdc.debezium.event;
|
||||
|
||||
import com.ververica.cdc.common.annotation.Internal;
|
||||
import com.ververica.cdc.common.types.DataField;
|
||||
import com.ververica.cdc.common.types.DataType;
|
||||
import com.ververica.cdc.common.types.DataTypes;
|
||||
import io.debezium.data.VariableScaleDecimal;
|
||||
import io.debezium.data.geometry.Geometry;
|
||||
import io.debezium.data.geometry.Point;
|
||||
import io.debezium.time.MicroTime;
|
||||
import io.debezium.time.MicroTimestamp;
|
||||
import io.debezium.time.NanoTime;
|
||||
import io.debezium.time.NanoTimestamp;
|
||||
import io.debezium.time.Time;
|
||||
import io.debezium.time.Timestamp;
|
||||
import io.debezium.time.ZonedTimestamp;
|
||||
import org.apache.kafka.connect.data.Date;
|
||||
import org.apache.kafka.connect.data.Decimal;
|
||||
import org.apache.kafka.connect.data.Schema;
|
||||
import org.apache.kafka.connect.data.Struct;
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.math.BigDecimal;
|
||||
import java.time.Instant;
|
||||
import java.util.Optional;
|
||||
|
||||
import static com.ververica.cdc.common.types.DecimalType.DEFAULT_PRECISION;
|
||||
|
||||
/** {@link DataType} inference for debezium {@link Schema}. */
|
||||
@Internal
|
||||
public class DebeziumSchemaDataTypeInference implements SchemaDataTypeInference, Serializable {
|
||||
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
@Override
|
||||
public DataType infer(Object value, Schema schema) {
|
||||
return schema.isOptional()
|
||||
? infer(value, schema, schema.type())
|
||||
: infer(value, schema, schema.type()).notNull();
|
||||
}
|
||||
|
||||
protected DataType infer(Object value, Schema schema, Schema.Type type) {
|
||||
switch (type) {
|
||||
case INT8:
|
||||
return inferInt8(value, schema);
|
||||
case INT16:
|
||||
return inferInt16(value, schema);
|
||||
case INT32:
|
||||
return inferInt32(value, schema);
|
||||
case INT64:
|
||||
return inferInt64(value, schema);
|
||||
case FLOAT32:
|
||||
return inferFloat32(value, schema);
|
||||
case FLOAT64:
|
||||
return inferFloat64(value, schema);
|
||||
case BOOLEAN:
|
||||
return inferBoolean(value, schema);
|
||||
case STRING:
|
||||
return inferString(value, schema);
|
||||
case BYTES:
|
||||
return inferBytes(value, schema);
|
||||
case STRUCT:
|
||||
return inferStruct(value, schema);
|
||||
case ARRAY:
|
||||
return inferArray(value, schema);
|
||||
case MAP:
|
||||
return inferMap(value, schema);
|
||||
default:
|
||||
throw new UnsupportedOperationException(
|
||||
"Unsupported type: " + schema.type().getName());
|
||||
}
|
||||
}
|
||||
|
||||
protected DataType inferBoolean(Object value, Schema schema) {
|
||||
return DataTypes.BOOLEAN();
|
||||
}
|
||||
|
||||
protected DataType inferInt8(Object value, Schema schema) {
|
||||
return DataTypes.TINYINT();
|
||||
}
|
||||
|
||||
protected DataType inferInt16(Object value, Schema schema) {
|
||||
return DataTypes.SMALLINT();
|
||||
}
|
||||
|
||||
protected DataType inferInt32(Object value, Schema schema) {
|
||||
if (Date.LOGICAL_NAME.equals(schema.name())) {
|
||||
return DataTypes.DATE();
|
||||
}
|
||||
if (Time.SCHEMA_NAME.equals(schema.name())) {
|
||||
return DataTypes.TIME(3);
|
||||
}
|
||||
return DataTypes.INT();
|
||||
}
|
||||
|
||||
protected DataType inferInt64(Object value, Schema schema) {
|
||||
if (MicroTime.SCHEMA_NAME.equals(schema.name())) {
|
||||
return DataTypes.TIME(6);
|
||||
}
|
||||
if (NanoTime.SCHEMA_NAME.equals(schema.name())) {
|
||||
return DataTypes.TIME(9);
|
||||
}
|
||||
if (Timestamp.SCHEMA_NAME.equals(schema.name())) {
|
||||
return DataTypes.TIMESTAMP(3);
|
||||
}
|
||||
if (MicroTimestamp.SCHEMA_NAME.equals(schema.name())) {
|
||||
return DataTypes.TIMESTAMP(6);
|
||||
}
|
||||
if (NanoTimestamp.SCHEMA_NAME.equals(schema.name())) {
|
||||
return DataTypes.TIMESTAMP(9);
|
||||
}
|
||||
return DataTypes.BIGINT();
|
||||
}
|
||||
|
||||
protected DataType inferFloat32(Object value, Schema schema) {
|
||||
return DataTypes.FLOAT();
|
||||
}
|
||||
|
||||
protected DataType inferFloat64(Object value, Schema schema) {
|
||||
return DataTypes.DOUBLE();
|
||||
}
|
||||
|
||||
protected DataType inferString(Object value, Schema schema) {
|
||||
if (ZonedTimestamp.SCHEMA_NAME.equals(schema.name())) {
|
||||
int nano =
|
||||
Optional.ofNullable((String) value)
|
||||
.map(Instant::parse)
|
||||
.map(Instant::getNano)
|
||||
.orElse(0);
|
||||
|
||||
int precision;
|
||||
if (nano == 0) {
|
||||
precision = 0;
|
||||
} else if (nano % 1000 > 0) {
|
||||
precision = 9;
|
||||
} else if (nano % 1000_000 > 0) {
|
||||
precision = 6;
|
||||
} else if (nano % 1000_000_000 > 0) {
|
||||
precision = 3;
|
||||
} else {
|
||||
precision = 0;
|
||||
}
|
||||
return DataTypes.TIMESTAMP(precision);
|
||||
}
|
||||
return DataTypes.STRING();
|
||||
}
|
||||
|
||||
protected DataType inferBytes(Object value, Schema schema) {
|
||||
if (Decimal.LOGICAL_NAME.equals(schema.name())
|
||||
|| VariableScaleDecimal.LOGICAL_NAME.equals(schema.name())) {
|
||||
if (value instanceof BigDecimal) {
|
||||
BigDecimal decimal = (BigDecimal) value;
|
||||
return DataTypes.DECIMAL(decimal.precision(), decimal.scale());
|
||||
}
|
||||
return DataTypes.DECIMAL(DEFAULT_PRECISION, 0);
|
||||
}
|
||||
return DataTypes.BYTES();
|
||||
}
|
||||
|
||||
protected DataType inferStruct(Object value, Schema schema) {
|
||||
Struct struct = (Struct) value;
|
||||
if (Geometry.LOGICAL_NAME.equals(schema.name())
|
||||
|| Point.LOGICAL_NAME.equals(schema.name())) {
|
||||
return DataTypes.STRING();
|
||||
}
|
||||
return DataTypes.ROW(
|
||||
schema.fields().stream()
|
||||
.map(
|
||||
f ->
|
||||
DataTypes.FIELD(
|
||||
f.name(), infer(struct.get(f.name()), f.schema())))
|
||||
.toArray(DataField[]::new));
|
||||
}
|
||||
|
||||
protected DataType inferArray(Object value, Schema schema) {
|
||||
throw new UnsupportedOperationException("Unsupported type ARRAY");
|
||||
}
|
||||
|
||||
protected DataType inferMap(Object value, Schema schema) {
|
||||
throw new UnsupportedOperationException("Unsupported type MAP");
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,35 @@
|
||||
/*
|
||||
* Copyright 2023 Ververica Inc.
|
||||
*
|
||||
* Licensed 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.ververica.cdc.debezium.event;
|
||||
|
||||
import com.ververica.cdc.common.annotation.Internal;
|
||||
import com.ververica.cdc.common.types.DataType;
|
||||
import org.apache.kafka.connect.data.Schema;
|
||||
|
||||
/** {@link DataType} inference for kafka connect {@link Schema}. */
|
||||
@Internal
|
||||
public interface SchemaDataTypeInference {
|
||||
|
||||
/**
|
||||
* Infer {@link DataType} from {@link Schema}.
|
||||
*
|
||||
* @param value the value corresponding value to SCHEMA
|
||||
* @param schema the kafka connect schema
|
||||
* @return the inferred data type
|
||||
*/
|
||||
DataType infer(Object value, Schema schema);
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
Loading…
Reference in New Issue