twalthr commented on a change in pull request #16962: URL: https://github.com/apache/flink/pull/16962#discussion_r784918322
########## File path: flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/mysql/MySQLTypeMapper.java ########## @@ -0,0 +1,215 @@ +/* + * 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.connector.jdbc.dialect.mysql; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.connector.jdbc.dialect.JdbcDialectTypeMapper; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.DecimalType; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.sql.ResultSetMetaData; +import java.sql.SQLException; + +/** MySQLTypeMapper util class. */ +@Internal +public class MySQLTypeMapper implements JdbcDialectTypeMapper { + + private static final Logger LOG = LoggerFactory.getLogger(MySQLTypeMapper.class); + + // ============================data types===================== + + private static final String MYSQL_UNKNOWN = "UNKNOWN"; + private static final String MYSQL_BIT = "BIT"; + + // -------------------------number---------------------------- + private static final String MYSQL_TINYINT = "TINYINT"; + private static final String MYSQL_TINYINT_UNSIGNED = "TINYINT UNSIGNED"; + private static final String MYSQL_SMALLINT = "SMALLINT"; + private static final String MYSQL_SMALLINT_UNSIGNED = "SMALLINT UNSIGNED"; + private static final String MYSQL_MEDIUMINT = "MEDIUMINT"; + private static final String MYSQL_MEDIUMINT_UNSIGNED = "MEDIUMINT UNSIGNED"; + private static final String MYSQL_INT = "INT"; + private static final String MYSQL_INT_UNSIGNED = "INT UNSIGNED"; + private static final String MYSQL_INTEGER = "INTEGER"; + private static final String MYSQL_INTEGER_UNSIGNED = "INTEGER UNSIGNED"; + private static final String MYSQL_BIGINT = "BIGINT"; + private static final String MYSQL_BIGINT_UNSIGNED = "BIGINT UNSIGNED"; + private static final String MYSQL_DECIMAL = "DECIMAL"; + private static final String MYSQL_DECIMAL_UNSIGNED = "DECIMAL UNSIGNED"; + private static final String MYSQL_FLOAT = "FLOAT"; + private static final String MYSQL_FLOAT_UNSIGNED = "FLOAT UNSIGNED"; + private static final String MYSQL_DOUBLE = "DOUBLE"; + private static final String MYSQL_DOUBLE_UNSIGNED = "DOUBLE UNSIGNED"; + + // -------------------------string---------------------------- + private static final String MYSQL_CHAR = "CHAR"; + private static final String MYSQL_VARCHAR = "VARCHAR"; + private static final String MYSQL_TINYTEXT = "TINYTEXT"; + private static final String MYSQL_MEDIUMTEXT = "MEDIUMTEXT"; + private static final String MYSQL_TEXT = "TEXT"; + private static final String MYSQL_LONGTEXT = "LONGTEXT"; + private static final String MYSQL_JSON = "JSON"; + + // ------------------------------time------------------------- + private static final String MYSQL_DATE = "DATE"; + private static final String MYSQL_DATETIME = "DATETIME"; + private static final String MYSQL_TIME = "TIME"; + private static final String MYSQL_TIMESTAMP = "TIMESTAMP"; + private static final String MYSQL_YEAR = "YEAR"; + + // ------------------------------blob------------------------- + private static final String MYSQL_TINYBLOB = "TINYBLOB"; + private static final String MYSQL_MEDIUMBLOB = "MEDIUMBLOB"; + private static final String MYSQL_BLOB = "BLOB"; + private static final String MYSQL_LONGBLOB = "LONGBLOB"; + private static final String MYSQL_BINARY = "BINARY"; + private static final String MYSQL_VARBINARY = "VARBINARY"; + private static final String MYSQL_GEOMETRY = "GEOMETRY"; + + private static final int RAW_TIME_LENGTH = 10; + private static final int RAW_TIMESTAMP_LENGTH = 19; + + private final String databaseVersion; + private final String driverVersion; + + public MySQLTypeMapper(String databaseVersion, String driverVersion) { + this.databaseVersion = databaseVersion; + this.driverVersion = driverVersion; + } + + @Override + public DataType mapping(ObjectPath tablePath, ResultSetMetaData metadata, int colIndex) + throws SQLException { + String mysqlType = metadata.getColumnTypeName(colIndex).toUpperCase(); + String columnName = metadata.getColumnName(colIndex); + int precision = metadata.getPrecision(colIndex); + int scale = metadata.getScale(colIndex); + + switch (mysqlType) { + case MYSQL_BIT: + return DataTypes.BOOLEAN(); + case MYSQL_TINYBLOB: + case MYSQL_MEDIUMBLOB: + case MYSQL_BLOB: + case MYSQL_LONGBLOB: + case MYSQL_VARBINARY: Review comment: I wasn't aware that `LegacyTypeInfoDataTypeConverter` is still used somewhere. This utility must be removed soon. -- 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]
