Hisoka-X commented on code in PR #6354: URL: https://github.com/apache/seatunnel/pull/6354#discussion_r1537023402
########## seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/datatype/AbstractDorisTypeConverter.java: ########## @@ -0,0 +1,423 @@ +/* + * 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.seatunnel.connectors.doris.datatype; + +import org.apache.seatunnel.api.table.catalog.Column; +import org.apache.seatunnel.api.table.catalog.PhysicalColumn; +import org.apache.seatunnel.api.table.converter.BasicTypeDefine; +import org.apache.seatunnel.api.table.converter.TypeConverter; +import org.apache.seatunnel.api.table.type.ArrayType; +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.DecimalType; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.common.exception.CommonError; +import org.apache.seatunnel.connectors.doris.config.DorisConfig; + +import lombok.extern.slf4j.Slf4j; + +import java.util.Locale; + +@Slf4j +public abstract class AbstractDorisTypeConverter implements TypeConverter<BasicTypeDefine> { + public static final String DORIS_NULL = "NULL"; + public static final String DORIS_BOOLEAN = "BOOLEAN"; + public static final String DORIS_TINYINT = "TINYINT"; + public static final String DORIS_SMALLINT = "SMALLINT"; + public static final String DORIS_INT = "INT"; + public static final String DORIS_BIGINT = "BIGINT"; + public static final String DORIS_LARGEINT = "LARGEINT"; + public static final String DORIS_FLOAT = "FLOAT"; + public static final String DORIS_DOUBLE = "DOUBLE"; + public static final String DORIS_DECIMAL = "DECIMAL"; + public static final String DORIS_DECIMALV3 = "DECIMALV3"; + public static final String DORIS_DATE = "DATE"; + public static final String DORIS_DATETIME = "DATETIME"; + public static final String DORIS_CHAR = "CHAR"; + public static final String DORIS_VARCHAR = "VARCHAR"; + public static final String DORIS_STRING = "STRING"; + + public static final String DORIS_BOOLEAN_ARRAY = "ARRAY<boolean>"; + public static final String DORIS_TINYINT_ARRAY = "ARRAY<tinyint>"; + public static final String DORIS_SMALLINT_ARRAY = "ARRAY<smallint>"; + public static final String DORIS_INT_ARRAY = "ARRAY<int(11)>"; + public static final String DORIS_BIGINT_ARRAY = "ARRAY<bigint>"; + public static final String DORIS_FLOAT_ARRAY = "ARRAY<float>"; + public static final String DORIS_DOUBLE_ARRAY = "ARRAY<double>"; + public static final String DORIS_DECIMALV3_ARRAY = "ARRAY<DECIMALV3>"; + public static final String DORIS_DECIMALV3_ARRAY_COLUMN_TYPE_TMP = "ARRAY<DECIMALV3(%, %)>"; + public static final String DORIS_DATEV2_ARRAY = "ARRAY<DATEV2>"; + public static final String DORIS_DATETIMEV2_ARRAY = "ARRAY<DATETIMEV2>"; + public static final String DORIS_STRING_ARRAY = "ARRAY<STRING>"; + + // Because can not get the column length from array, So the following types of arrays cannot be + // generated properly. + public static final String DORIS_LARGEINT_ARRAY = "ARRAY<largeint>"; + public static final String DORIS_CHAR_ARRAY = "ARRAY<CHAR>"; + public static final String DORIS_CHAR_ARRAY_COLUMN_TYPE_TMP = "ARRAY<CHAR(%s)>"; + public static final String DORIS_VARCHAR_ARRAY = "ARRAY<VARCHAR>"; + public static final String DORIS_VARCHAR_ARRAY_COLUMN_TYPE_TMP = "ARRAY<VARCHAR(%s)>"; + + public static final String DORIS_JSON = "JSON"; + public static final String DORIS_JSONB = "JSONB"; + + public static final Integer DEFAULT_PRECISION = 9; + public static final Integer MAX_PRECISION = 38; + + public static final Integer DEFAULT_SCALE = 0; + + public static final Integer MAX_DATETIME_SCALE = 6; + + // Min value of LARGEINT is -170141183460469231731687303715884105728, it will use 39 bytes in + // UTF-8. + // Add a bit to prevent overflow + public static final long MAX_DORIS_LARGEINT_TO_VARCHAR_LENGTH = 39L; + + public static final long POWER_2_8 = (long) Math.pow(2, 8); + public static final long POWER_2_16 = (long) Math.pow(2, 16); + public static final long MAX_STRING_LENGTH = 2147483643; + + protected PhysicalColumn.PhysicalColumnBuilder getPhysicalColumnBuilder( + BasicTypeDefine typeDefine) { + PhysicalColumn.PhysicalColumnBuilder builder = + PhysicalColumn.builder() + .name(typeDefine.getName()) + .sourceType(typeDefine.getColumnType()) + .nullable(typeDefine.isNullable()) + .defaultValue(typeDefine.getDefaultValue()) + .comment(typeDefine.getComment()); + return builder; + } + + protected BasicTypeDefine.BasicTypeDefineBuilder getBasicTypeDefineBuilder(Column column) { + BasicTypeDefine.BasicTypeDefineBuilder builder = + BasicTypeDefine.builder() + .name(column.getName()) + .nullable(column.isNullable()) + .comment(column.getComment()) + .defaultValue(column.getDefaultValue()); + return builder; + } + + protected String getDorisColumnName(BasicTypeDefine typeDefine) { + String dorisColumnType = typeDefine.getColumnType().toUpperCase(Locale.ROOT); + int idx = dorisColumnType.indexOf("("); + int idx2 = dorisColumnType.indexOf("<"); + if (idx != -1) { + dorisColumnType = dorisColumnType.substring(0, idx); + } + if (idx2 != -1) { + dorisColumnType = dorisColumnType.substring(0, idx2); + } + return dorisColumnType; + } + + public void sampleTypeConverter( + PhysicalColumn.PhysicalColumnBuilder builder, + BasicTypeDefine typeDefine, + String dorisColumnType) { + switch (dorisColumnType) { + case DORIS_NULL: + builder.dataType(BasicType.VOID_TYPE); + break; + case DORIS_BOOLEAN: + builder.dataType(BasicType.BOOLEAN_TYPE); + break; + case DORIS_TINYINT: + if (typeDefine.getColumnType().equalsIgnoreCase("tinyint(1)")) { + builder.dataType(BasicType.BOOLEAN_TYPE); + } else { + builder.dataType(BasicType.BYTE_TYPE); + } + break; + case DORIS_SMALLINT: + builder.dataType(BasicType.SHORT_TYPE); + break; + case DORIS_INT: + builder.dataType(BasicType.INT_TYPE); + break; + case DORIS_BIGINT: + builder.dataType(BasicType.LONG_TYPE); + break; + case DORIS_FLOAT: + builder.dataType(BasicType.FLOAT_TYPE); + break; + case DORIS_DOUBLE: + builder.dataType(BasicType.DOUBLE_TYPE); + break; + case DORIS_CHAR: + case DORIS_VARCHAR: + if (typeDefine.getLength() != null && typeDefine.getLength() > 0) { + builder.columnLength(typeDefine.getLength()); + } + builder.dataType(BasicType.STRING_TYPE); + break; + case DORIS_LARGEINT: + builder.dataType(BasicType.STRING_TYPE); + builder.columnLength(MAX_DORIS_LARGEINT_TO_VARCHAR_LENGTH); + break; + case DORIS_STRING: + case DORIS_JSON: + builder.dataType(BasicType.STRING_TYPE); + builder.columnLength(MAX_STRING_LENGTH); + break; + default: + throw CommonError.convertToSeaTunnelTypeError( + DorisConfig.IDENTIFIER, dorisColumnType, typeDefine.getName()); + } + } + + protected void sampleReconvertString( + Column column, BasicTypeDefine.BasicTypeDefineBuilder builder) { + // source is doris too. + if (column.getSourceType() != null + && column.getSourceType().equalsIgnoreCase(DORIS_LARGEINT)) { + builder.columnType(DORIS_LARGEINT); + builder.dataType(DORIS_LARGEINT); + return; + } + + if (column.getColumnLength() == null || column.getColumnLength() <= 0) { + builder.columnType(DORIS_STRING); + builder.dataType(DORIS_STRING); + return; + } + + if (column.getColumnLength() < POWER_2_8) { + if (column.getSourceType() != null + && column.getSourceType().toUpperCase(Locale.ROOT).startsWith(DORIS_VARCHAR)) { + builder.columnType( + String.format("%s(%s)", DORIS_VARCHAR, column.getColumnLength())); + builder.dataType(DORIS_VARCHAR); + } else { + builder.columnType(String.format("%s(%s)", DORIS_CHAR, column.getColumnLength())); + builder.dataType(DORIS_CHAR); + } + return; + } + + if (column.getColumnLength() < POWER_2_16) { Review Comment: `POWER_2_16` not equals `65533`. ```suggestion if (column.getColumnLength() < 65533) { ``` -- 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]
