This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-13023-dea766726249e72e8246b2abc6b89d98306f01b2 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 4d74c89d1341dda681119f77250c472ea2d64e5a Author: Sepuri Sai Krishna <[email protected]> AuthorDate: Wed Oct 28 16:59:40 2026 +0000 [Fix][Connector-V2] Map Databend DECIMAL and BINARY columns to the right types (#13023) --- .../databend/catalog/DatabendCatalog.java | 46 ++++++-- .../databend/catalog/DatabendCatalogTest.java | 124 +++++++++++++++++++++ .../e2e/connector/databend/DatabendIT.java | 25 ++++- .../src/test/resources/databend/databend_init.conf | 4 +- ...tabend_init.conf => databend_type_mapping.conf} | 28 ++++- 5 files changed, 213 insertions(+), 14 deletions(-) diff --git a/seatunnel-connectors-v2/connector-databend/src/main/java/org/apache/seatunnel/connectors/seatunnel/databend/catalog/DatabendCatalog.java b/seatunnel-connectors-v2/connector-databend/src/main/java/org/apache/seatunnel/connectors/seatunnel/databend/catalog/DatabendCatalog.java index 56010f1d69..edc07fa40a 100644 --- a/seatunnel-connectors-v2/connector-databend/src/main/java/org/apache/seatunnel/connectors/seatunnel/databend/catalog/DatabendCatalog.java +++ b/seatunnel-connectors-v2/connector-databend/src/main/java/org/apache/seatunnel/connectors/seatunnel/databend/catalog/DatabendCatalog.java @@ -33,6 +33,7 @@ import org.apache.seatunnel.api.table.catalog.exception.TableNotExistException; import org.apache.seatunnel.api.table.type.BasicType; import org.apache.seatunnel.api.table.type.DecimalType; import org.apache.seatunnel.api.table.type.LocalTimeType; +import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType; import org.apache.seatunnel.api.table.type.SeaTunnelDataType; import org.apache.seatunnel.connectors.seatunnel.databend.config.DatabendOptions; import org.apache.seatunnel.connectors.seatunnel.databend.exception.DatabendConnectorErrorCode; @@ -50,6 +51,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.List; +import java.util.Locale; import java.util.Map; @Slf4j @@ -457,14 +459,26 @@ public class DatabendCatalog implements Catalog { } } - private SeaTunnelDataType<?> convertDatabendType( + /** + * Converts a Databend column type, as reported by the JDBC driver, to the matching SeaTunnel + * type. Package private so the mapping can be asserted directly in unit tests. + * + * @param typeName the Databend type name held in information_schema.columns.data_type + * @param sqlType the java.sql.Types code the driver derives from that name + * @param columnSize precision for a DECIMAL column + * @param decimalDigits scale for a DECIMAL column + * @return the SeaTunnel type for the column + */ + static SeaTunnelDataType<?> convertDatabendType( String typeName, int sqlType, int columnSize, int decimalDigits) { - // This method should convert Databend data types to SeaTunnel data types - // This is a simplified version, you'll need to adjust based on Databend's actual type - // system - typeName = typeName.toUpperCase(); - - switch (typeName) { + // information_schema.columns reports data_type lower cased and keeps the arguments of the + // parameterised types, so a DECIMAL(10, 2) column arrives here as "decimal(10, 2)". Match + // on the bare upper case name, and uppercase with Locale.ROOT: the default locale turns + // the i of int, bigint, timestamp and binary into a dotted capital I under tr-TR and + // az-AZ, and each of those columns would then miss its branch and fall through to STRING. + String databendType = normalizeTypeName(typeName); + + switch (databendType) { case "BOOLEAN": return BasicType.BOOLEAN_TYPE; case "TINYINT": @@ -499,13 +513,29 @@ public class DatabendCatalog implements Catalog { return LocalTimeType.LOCAL_DATE_TIME_TYPE; case "VARBINARY": case "BINARY": - return BasicType.BYTE_TYPE; + // A Databend BINARY column holds a byte string, so it maps to BYTES. BYTE_TYPE is + // a single java.lang.Byte (SqlType.TINYINT), which contradicts both the documented + // mapping and toDatabendTypeString, where SeaTunnel BYTES is written as VARBINARY. + return PrimitiveByteArrayType.INSTANCE; default: log.warn("Unsupported Databend type: {}, fallback to STRING type", typeName); return BasicType.STRING_TYPE; } } + /** + * Reduces a Databend type name to the bare upper case name the type switch is keyed on, + * dropping any parameter list such as the precision and scale of a DECIMAL. + */ + private static String normalizeTypeName(String typeName) { + String bareName = typeName.trim(); + int parameterStart = bareName.indexOf('('); + if (parameterStart >= 0) { + bareName = bareName.substring(0, parameterStart).trim(); + } + return bareName.toUpperCase(Locale.ROOT); + } + private Connection getConnection() throws SQLException { return DatabendUtil.createConnection(this.readonlyConfig); } diff --git a/seatunnel-connectors-v2/connector-databend/src/test/java/org/apache/seatunnel/connectors/seatunnel/databend/catalog/DatabendCatalogTest.java b/seatunnel-connectors-v2/connector-databend/src/test/java/org/apache/seatunnel/connectors/seatunnel/databend/catalog/DatabendCatalogTest.java new file mode 100644 index 0000000000..8e6e53d402 --- /dev/null +++ b/seatunnel-connectors-v2/connector-databend/src/test/java/org/apache/seatunnel/connectors/seatunnel/databend/catalog/DatabendCatalogTest.java @@ -0,0 +1,124 @@ +/* + * 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.seatunnel.databend.catalog; + +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.DecimalType; +import org.apache.seatunnel.api.table.type.LocalTimeType; +import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.sql.Types; +import java.util.Locale; + +/** + * Covers the Databend to SeaTunnel type mapping the catalog applies to every column of every table + * the source reads. The type names used here are the ones a Databend server actually reports: + * information_schema.columns exposes data_type lower cased, and the JDBC driver passes that string + * through as TYPE_NAME, keeping the arguments of a parameterised type. + */ +public class DatabendCatalogTest { + + @Test + public void decimalKeepsItsPrecisionAndScale() { + // A Databend DECIMAL arrives with its arguments attached, so an exact match against the + // bare name "DECIMAL" never fires and the column used to degrade to STRING. + Assertions.assertEquals( + new DecimalType(10, 2), + DatabendCatalog.convertDatabendType("decimal(10, 2)", Types.DECIMAL, 10, 2)); + Assertions.assertEquals( + new DecimalType(38, 0), + DatabendCatalog.convertDatabendType("decimal(38, 0)", Types.DECIMAL, 38, 0)); + // The unparameterised spelling has to keep working too. + Assertions.assertEquals( + new DecimalType(20, 4), + DatabendCatalog.convertDatabendType("decimal", Types.DECIMAL, 20, 4)); + } + + @Test + public void binaryMapsToBytesRatherThanASingleByte() { + // BYTE_TYPE is one java.lang.Byte (SqlType.TINYINT). A Databend BINARY column holds a byte + // string, and toDatabendTypeString writes SeaTunnel BYTES back out as VARBINARY, so the + // reverse direction has to produce BYTES for the round trip to hold. + Assertions.assertEquals( + PrimitiveByteArrayType.INSTANCE, + DatabendCatalog.convertDatabendType("binary", Types.BINARY, 0, 0)); + Assertions.assertEquals( + PrimitiveByteArrayType.INSTANCE, + DatabendCatalog.convertDatabendType("varbinary", Types.VARBINARY, 0, 0)); + } + + @Test + public void theDocumentedMappingHolds() { + assertMaps("boolean", BasicType.BOOLEAN_TYPE); + assertMaps("tinyint", BasicType.BYTE_TYPE); + assertMaps("smallint", BasicType.SHORT_TYPE); + assertMaps("int", BasicType.INT_TYPE); + assertMaps("bigint", BasicType.LONG_TYPE); + assertMaps("float", BasicType.FLOAT_TYPE); + assertMaps("double", BasicType.DOUBLE_TYPE); + assertMaps("varchar", BasicType.STRING_TYPE); + assertMaps("date", LocalTimeType.LOCAL_DATE_TYPE); + assertMaps("timestamp", LocalTimeType.LOCAL_DATE_TIME_TYPE); + assertMaps("binary", PrimitiveByteArrayType.INSTANCE); + } + + @Test + public void theMappingIsIndependentOfTheDefaultLocale() { + // Under tr-TR and az-AZ an uppercase with the default locale turns the i of "int", + // "bigint", "timestamp", "binary" and "decimal" into a dotted capital I, so none of those + // names matches its branch any more and every one of those columns becomes a STRING. + Locale previous = Locale.getDefault(); + try { + for (Locale locale : new Locale[] {new Locale("tr", "TR"), new Locale("az", "AZ")}) { + Locale.setDefault(locale); + assertMaps("tinyint", BasicType.BYTE_TYPE); + assertMaps("smallint", BasicType.SHORT_TYPE); + assertMaps("int", BasicType.INT_TYPE); + assertMaps("bigint", BasicType.LONG_TYPE); + assertMaps("timestamp", LocalTimeType.LOCAL_DATE_TIME_TYPE); + assertMaps("binary", PrimitiveByteArrayType.INSTANCE); + Assertions.assertEquals( + new DecimalType(10, 2), + DatabendCatalog.convertDatabendType("decimal(10, 2)", Types.DECIMAL, 10, 2), + "a decimal column must not depend on the default locale"); + } + } finally { + Locale.setDefault(previous); + } + } + + @Test + public void anUnknownTypeStillFallsBackToString() { + // The fallback is deliberate: a type the connector has no column type for is carried as + // text rather than failing the job. + assertMaps("variant", BasicType.STRING_TYPE); + assertMaps("geometry", BasicType.STRING_TYPE); + assertMaps("array(int32)", BasicType.STRING_TYPE); + } + + private static void assertMaps(String databendTypeName, SeaTunnelDataType<?> expected) { + Assertions.assertEquals( + expected, + DatabendCatalog.convertDatabendType(databendTypeName, Types.OTHER, 0, 0), + "unexpected SeaTunnel type for Databend type " + databendTypeName); + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-databend-e2e/src/test/java/org/apache/seatunnel/e2e/connector/databend/DatabendIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-databend-e2e/src/test/java/org/apache/seatunnel/e2e/connector/databend/DatabendIT.java index af149c537b..92e437cfeb 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-databend-e2e/src/test/java/org/apache/seatunnel/e2e/connector/databend/DatabendIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-databend-e2e/src/test/java/org/apache/seatunnel/e2e/connector/databend/DatabendIT.java @@ -71,6 +71,8 @@ public class DatabendIT extends TestSuiteBase implements TestResource { private static final String SOURCE_TABLE = "source_table"; private static final String SINK_TABLE = "sink_table"; private static final String INSERT_SQL = "insert_sql"; + private static final String TYPE_MAPPING_TABLE = "type_mapping_table"; + private static final String TYPE_MAPPING_INSERT_SQL = "type_mapping_insert_sql"; private static final Config CONFIG = getInitDatabendConfig(); private DatabendContainer container; private GenericContainer<?> minioContainer; @@ -127,6 +129,20 @@ public class DatabendIT extends TestSuiteBase implements TestResource { Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr()); } + /** + * Reads a table holding a DECIMAL and a BINARY column with no schema declared in the job, so + * the run depends on the types DatabendCatalog resolves for them. information_schema.columns + * reports those two as "decimal(10, 2)" and "binary", the shapes the catalog used to resolve to + * STRING and to a single byte. + */ + @TestTemplate + public void testDatabendTypeMapping(TestContainer container) + throws IOException, InterruptedException { + Container.ExecResult execResult = + container.executeJob("/databend/databend_type_mapping.conf"); + Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr()); + } + @TestTemplate public void testSchemaEvolution(TestContainer container) throws IOException, InterruptedException, SQLException { @@ -346,7 +362,9 @@ public class DatabendIT extends TestSuiteBase implements TestResource { "INSERT INTO source_table (name, age, score) VALUES ('Charlie', 35, 92.5)"; stmt.execute(sql3); - LOG.info("Successfully inserted 3 test records"); + stmt.execute(CONFIG.getString(TYPE_MAPPING_INSERT_SQL)); + + LOG.info("Successfully inserted 3 test records and 1 type mapping record"); } catch (SQLException e) { LOG.error("Failed to insert test data", e); throw new RuntimeException("Failed to insert test data", e); @@ -371,6 +389,7 @@ public class DatabendIT extends TestSuiteBase implements TestResource { Statement statement = connection.createStatement(); ) { statement.execute(CONFIG.getString(SOURCE_TABLE)); statement.execute(CONFIG.getString(SINK_TABLE)); + statement.execute(CONFIG.getString(TYPE_MAPPING_TABLE)); } catch (SQLException e) { throw new RuntimeException("Initializing Databend tables failed!", e); } @@ -398,7 +417,9 @@ public class DatabendIT extends TestSuiteBase implements TestResource { Config config = ConfigFactory.parseFile(file); assert config.hasPath(SOURCE_TABLE) && config.hasPath(SINK_TABLE) - && config.hasPath(INSERT_SQL); + && config.hasPath(INSERT_SQL) + && config.hasPath(TYPE_MAPPING_TABLE) + && config.hasPath(TYPE_MAPPING_INSERT_SQL); return config; } diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-databend-e2e/src/test/resources/databend/databend_init.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-databend-e2e/src/test/resources/databend/databend_init.conf index 0cc7ec854b..127c6d2706 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-databend-e2e/src/test/resources/databend/databend_init.conf +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-databend-e2e/src/test/resources/databend/databend_init.conf @@ -17,4 +17,6 @@ source_table = "CREATE TABLE IF NOT EXISTS source_table (name STRING, age INT, score DOUBLE)" sink_table = "CREATE TABLE IF NOT EXISTS sink_table (name STRING, age INT, score DOUBLE)" -insert_sql = "INSERT INTO source_table (name, age, score) VALUES (?, ?, ?)" \ No newline at end of file +insert_sql = "INSERT INTO source_table (name, age, score) VALUES (?, ?, ?)" +type_mapping_table = "CREATE TABLE IF NOT EXISTS type_mapping_table (id INT, price DECIMAL(10, 2), payload BINARY)" +type_mapping_insert_sql = "INSERT INTO type_mapping_table (id, price, payload) VALUES (1, 1234.56, UNHEX('68656c6c6f'))" diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-databend-e2e/src/test/resources/databend/databend_init.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-databend-e2e/src/test/resources/databend/databend_type_mapping.conf similarity index 52% copy from seatunnel-e2e/seatunnel-connector-v2-e2e/connector-databend-e2e/src/test/resources/databend/databend_init.conf copy to seatunnel-e2e/seatunnel-connector-v2-e2e/connector-databend-e2e/src/test/resources/databend/databend_type_mapping.conf index 0cc7ec854b..3d369633bf 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-databend-e2e/src/test/resources/databend/databend_init.conf +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-databend-e2e/src/test/resources/databend/databend_type_mapping.conf @@ -15,6 +15,28 @@ # limitations under the License. # -source_table = "CREATE TABLE IF NOT EXISTS source_table (name STRING, age INT, score DOUBLE)" -sink_table = "CREATE TABLE IF NOT EXISTS sink_table (name STRING, age INT, score DOUBLE)" -insert_sql = "INSERT INTO source_table (name, age, score) VALUES (?, ?, ?)" \ No newline at end of file +# Reads a table whose columns cover the Databend types the catalog resolves through a parameterised +# or a byte string type name: DECIMAL(10, 2) arrives from information_schema.columns as +# "decimal(10, 2)" and BINARY as "binary". No schema is declared here on purpose, so the job runs +# on the schema DatabendCatalog infers for the table. +env { + execution.parallelism = 1 + job.mode = "BATCH" +} + +source { + Databend { + url = "jdbc:databend://databend:8000/default" + username = "root" + password = "" + database = "default" + table = "type_mapping_table" + query = "SELECT id, price, payload FROM type_mapping_table ORDER BY id" + fetch_size = 1000 + } +} + +sink { + Console { + } +}
