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-12761-5ffbce52b185109aab525f01953d065506feaf7e in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 59775c6cb02ed4dbb0415dd15c468b3cafcad9ca Author: zhengxiang378928908-code <[email protected]> AuthorDate: Sat Oct 17 07:52:19 2026 +0000 [Fix][Connector-V2][Doris] Support IPV4 and IPV6 columns (#12761) Co-authored-by: Asthenia0412 <[email protected]> --- docs/en/connectors/sink/Doris.md | 1 + docs/en/connectors/source/Doris.md | 3 + docs/zh/connectors/sink/Doris.md | 1 + docs/zh/connectors/source/Doris.md | 3 + .../doris/datatype/AbstractDorisTypeConverter.java | 14 ++ .../doris/datatype/DorisTypeConverterV2.java | 10 ++ .../doris/source/reader/DorisValueReader.java | 59 ++++++++ .../connectors/doris/util/SchemaUtils.java | 14 +- .../doris/datatype/DorisTypeConvertorV2Test.java | 49 +++++++ .../source/reader/DorisValueReaderIpv4Test.java | 148 +++++++++++++++++++++ .../connectors/doris/util/SchemaUtilsTest.java} | 47 +++---- 11 files changed, 317 insertions(+), 32 deletions(-) diff --git a/docs/en/connectors/sink/Doris.md b/docs/en/connectors/sink/Doris.md index 4b3be2c2c4..9d44003242 100644 --- a/docs/en/connectors/sink/Doris.md +++ b/docs/en/connectors/sink/Doris.md @@ -180,6 +180,7 @@ You can use the following placeholders | MAP | MAP | | JSON | STRING | | VARIANT | STRING | +| IPV4<br/>IPV6 | STRING | | HLL | Not supported yet | | BITMAP | Not supported yet | | QUANTILE_STATE | Not supported yet | diff --git a/docs/en/connectors/source/Doris.md b/docs/en/connectors/source/Doris.md index 7e9aeae210..cc885a6fec 100644 --- a/docs/en/connectors/source/Doris.md +++ b/docs/en/connectors/source/Doris.md @@ -56,10 +56,13 @@ Used to read data from Apache Doris. | CHAR<br/>VARCHAR<br/>STRING<br/>TEXT | STRING | | JSON | STRING | | VARIANT | STRING | +| IPV4<br/>IPV6 | STRING | | DATE | DATE | | DATETIME<br/>DATETIME(p) | TIMESTAMP | | ARRAY | ARRAY | +`IPV4` and `IPV6` are read as their text form, for example `192.168.0.1` and `2001:db8::1`. They are recognized only when the Doris version reports them in `information_schema.columns` (verified on Doris 4.0.6; Doris 2.1.0 reports these columns as `unknown`). + ## Source Options Base configuration: diff --git a/docs/zh/connectors/sink/Doris.md b/docs/zh/connectors/sink/Doris.md index 36d6d62807..722e273d38 100644 --- a/docs/zh/connectors/sink/Doris.md +++ b/docs/zh/connectors/sink/Doris.md @@ -178,6 +178,7 @@ CREATE TABLE IF NOT EXISTS `${database}`.`${table_name}` | MAP | MAP | | JSON | STRING | | VARIANT | STRING | +| IPV4<br/>IPV6 | STRING | | HLL | 尚不支持 | | BITMAP | 尚不支持 | | QUANTILE_STATE | 尚不支持 | diff --git a/docs/zh/connectors/source/Doris.md b/docs/zh/connectors/source/Doris.md index 5bee0a2f44..a6186b7662 100644 --- a/docs/zh/connectors/source/Doris.md +++ b/docs/zh/connectors/source/Doris.md @@ -56,10 +56,13 @@ import ChangeLog from '../changelog/connector-doris.md'; | CHAR<br/>VARCHAR<br/>STRING<br/>TEXT | STRING | | JSON | STRING | | VARIANT | STRING | +| IPV4<br/>IPV6 | STRING | | DATE | DATE | | DATETIME<br/>DATETIME(p) | TIMESTAMP | | ARRAY | ARRAY | +`IPV4` 和 `IPV6` 以文本形式读取，例如 `192.168.0.1`、`2001:db8::1`。只有当 Doris 版本在 `information_schema.columns` 中正确报告这两种类型时才能识别（已在 Doris 4.0.6 上验证；Doris 2.1.0 会把这类列报告为 `unknown`）。 + ## 源选项 基础配置: diff --git a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/datatype/AbstractDorisTypeConverter.java b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/datatype/AbstractDorisTypeConverter.java index f56de2739c..8ef92c8dfd 100644 --- a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/datatype/AbstractDorisTypeConverter.java +++ b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/datatype/AbstractDorisTypeConverter.java @@ -76,6 +76,8 @@ public abstract class AbstractDorisTypeConverter implements TypeConverter<BasicT public static final String DORIS_JSON = "JSON"; public static final String DORIS_JSONB = "JSONB"; public static final String DORIS_VARIANT = "VARIANT"; + public static final String DORIS_IPV4 = "IPV4"; + public static final String DORIS_IPV6 = "IPV6"; public static final Long DEFAULT_PRECISION = 9L; public static final Long MAX_PRECISION = 38L; @@ -96,6 +98,10 @@ public abstract class AbstractDorisTypeConverter implements TypeConverter<BasicT 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; + // dotted-decimal IPv4 text, e.g. 255.255.255.255 + public static final long MAX_IPV4_STRING_LENGTH = 15; + // full-form IPv6 text, e.g. ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff + public static final long MAX_IPV6_STRING_LENGTH = 39; protected PhysicalColumn.PhysicalColumnBuilder getPhysicalColumnBuilder( BasicTypeDefine typeDefine, boolean caseSensitive) { @@ -192,6 +198,14 @@ public abstract class AbstractDorisTypeConverter implements TypeConverter<BasicT builder.dataType(BasicType.STRING_TYPE); builder.columnLength(MAX_STRING_LENGTH); break; + case DORIS_IPV4: + builder.dataType(BasicType.STRING_TYPE); + builder.columnLength(MAX_IPV4_STRING_LENGTH); + break; + case DORIS_IPV6: + builder.dataType(BasicType.STRING_TYPE); + builder.columnLength(MAX_IPV6_STRING_LENGTH); + break; default: throw CommonError.convertToSeaTunnelTypeError( IDENTIFIER, dorisColumnType, typeDefine.getName()); diff --git a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/datatype/DorisTypeConverterV2.java b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/datatype/DorisTypeConverterV2.java index 289d94f643..fe2b11639a 100644 --- a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/datatype/DorisTypeConverterV2.java +++ b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/datatype/DorisTypeConverterV2.java @@ -292,6 +292,16 @@ public class DorisTypeConverterV2 extends AbstractDorisTypeConverter { builder.dataType(DORIS_VARIANT); return; } + if (column.getSourceType() != null && column.getSourceType().equalsIgnoreCase(DORIS_IPV4)) { + builder.columnType(DORIS_IPV4); + builder.dataType(DORIS_IPV4); + return; + } + if (column.getSourceType() != null && column.getSourceType().equalsIgnoreCase(DORIS_IPV6)) { + builder.columnType(DORIS_IPV6); + builder.dataType(DORIS_IPV6); + return; + } super.sampleReconvertString(column, builder); } diff --git a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/source/reader/DorisValueReader.java b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/source/reader/DorisValueReader.java index ee369d94ad..9a5bf1572e 100644 --- a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/source/reader/DorisValueReader.java +++ b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/source/reader/DorisValueReader.java @@ -17,6 +17,10 @@ package org.apache.seatunnel.connectors.doris.source.reader; +import org.apache.seatunnel.shade.com.google.common.annotations.VisibleForTesting; + +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.catalog.Column; import org.apache.seatunnel.api.table.type.SeaTunnelRow; import org.apache.seatunnel.api.table.type.SeaTunnelRowType; import org.apache.seatunnel.connectors.doris.backend.BackendClient; @@ -39,13 +43,18 @@ import org.apache.doris.sdk.thrift.TScanOpenResult; import lombok.extern.slf4j.Slf4j; +import java.util.ArrayList; import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; +import static org.apache.seatunnel.connectors.doris.datatype.AbstractDorisTypeConverter.DORIS_IPV4; import static org.apache.seatunnel.connectors.doris.util.ErrorMessages.SHOULD_NOT_HAPPEN_MESSAGE; @Slf4j @@ -71,6 +80,8 @@ public class DorisValueReader { protected Schema schema; protected SeaTunnelRowType seaTunnelRowType; + // Doris BE sends IPV4 through Arrow as a 32-bit integer, while the column is a STRING + private final int[] ipv4FieldIndexes; protected boolean asyncThreadStarted; public DorisValueReader( @@ -83,6 +94,8 @@ public class DorisValueReader { this.client = backendClient(); this.deserializeArrowToRowBatchAsync = config.getDeserializeArrowAsync(); this.seaTunnelRowType = dorisSourceTable.getCatalogTable().getSeaTunnelRowType(); + this.ipv4FieldIndexes = + ipv4FieldIndexes(dorisSourceTable.getCatalogTable(), seaTunnelRowType); int blockingQueueSize = config.getDeserializeQueueSize(); if (this.deserializeArrowToRowBatchAsync) { this.rowBatchBlockingQueue = new ArrayBlockingQueue<>(blockingQueueSize); @@ -284,10 +297,56 @@ public class DorisValueReader { DorisConnectorErrorCode.SHOULD_NEVER_HAPPEN, "never happen error."); } SeaTunnelRow next = rowBatch.next(); + convertIpv4Fields(next, ipv4FieldIndexes); next.setTableId(dorisSourceTable.getTablePath().toString()); return next; } + /** + * Finds the IPV4 fields by their catalog source type, because the row type only says STRING and + * the Arrow reader cannot tell an IPV4 column from a plain integer. + */ + @VisibleForTesting + static int[] ipv4FieldIndexes(CatalogTable catalogTable, SeaTunnelRowType rowType) { + Map<String, String> sourceTypes = new HashMap<>(); + for (Column column : catalogTable.getTableSchema().getColumns()) { + sourceTypes.put(column.getName(), column.getSourceType()); + } + List<Integer> indexes = new ArrayList<>(); + String[] fieldNames = rowType.getFieldNames(); + for (int i = 0; i < fieldNames.length; i++) { + if (DORIS_IPV4.equalsIgnoreCase(sourceTypes.get(fieldNames[i]))) { + indexes.add(i); + } + } + return indexes.stream().mapToInt(Integer::intValue).toArray(); + } + + /** + * Runs after the Arrow-to-row step and turns the integer that Doris sends for IPV4 into + * dotted-decimal text, leaving NULL as is. + */ + @VisibleForTesting + static void convertIpv4Fields(SeaTunnelRow row, int[] ipv4FieldIndexes) { + for (int index : ipv4FieldIndexes) { + Object value = row.getField(index); + if (value instanceof Number) { + row.setField(index, ipv4ToString(((Number) value).longValue())); + } + } + } + + @VisibleForTesting + static String ipv4ToString(long value) { + return ((value >>> 24) & 0xFF) + + "." + + ((value >>> 16) & 0xFF) + + "." + + ((value >>> 8) & 0xFF) + + "." + + (value & 0xFF); + } + public void close() { // Stop the asynchronous fetch before taking the client lock: the fetch thread holds that // lock for the whole scan, so a reader that is still waiting for its first batch would diff --git a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/util/SchemaUtils.java b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/util/SchemaUtils.java index bf8f3b9a26..8725a4d7f7 100644 --- a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/util/SchemaUtils.java +++ b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/util/SchemaUtils.java @@ -38,13 +38,13 @@ public class SchemaUtils { .forEach( desc -> schema.put( - new Field( - desc.getName(), - desc.getType().name(), - "", - 0, - 0, - ""))); + new Field(desc.getName(), typeName(desc), "", 0, 0, ""))); return schema; } + + // The shaded thrift-service only knows the types of its Doris release. Newer ones such as + // IPV4, IPV6 or VARIANT deserialize to null; the name is only used for logging. + private static String typeName(TScanColumnDesc desc) { + return desc.getType() == null ? "UNKNOWN" : desc.getType().name(); + } } diff --git a/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/datatype/DorisTypeConvertorV2Test.java b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/datatype/DorisTypeConvertorV2Test.java index f48ff23f5b..dccf6d5615 100644 --- a/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/datatype/DorisTypeConvertorV2Test.java +++ b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/datatype/DorisTypeConvertorV2Test.java @@ -325,6 +325,30 @@ public class DorisTypeConvertorV2Test { Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType()); } + @Test + public void testConvertIpv4() { + BasicTypeDefine<Object> typeDefine = + BasicTypeDefine.builder().name("test").columnType("ipv4").dataType("ipv4").build(); + Column column = DorisTypeConverterV2.INSTANCE.convert(typeDefine); + Assertions.assertEquals(typeDefine.getName(), column.getName()); + Assertions.assertEquals(BasicType.STRING_TYPE, column.getDataType()); + Assertions.assertEquals( + DorisTypeConverterV2.MAX_IPV4_STRING_LENGTH, column.getColumnLength()); + Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType()); + } + + @Test + public void testConvertIpv6() { + BasicTypeDefine<Object> typeDefine = + BasicTypeDefine.builder().name("test").columnType("ipv6").dataType("ipv6").build(); + Column column = DorisTypeConverterV2.INSTANCE.convert(typeDefine); + Assertions.assertEquals(typeDefine.getName(), column.getName()); + Assertions.assertEquals(BasicType.STRING_TYPE, column.getDataType()); + Assertions.assertEquals( + DorisTypeConverterV2.MAX_IPV6_STRING_LENGTH, column.getColumnLength()); + Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType()); + } + @Test public void testConvertDate() { BasicTypeDefine<Object> typeDefine = @@ -954,6 +978,31 @@ public class DorisTypeConvertorV2Test { Assertions.assertEquals(DorisTypeConverterV2.DORIS_STRING, typeDefine.getDataType()); } + @Test + public void testReconvertIpFromDorisSource() { + Column column = + PhysicalColumn.builder() + .name("test") + .dataType(BasicType.STRING_TYPE) + .columnLength(DorisTypeConverterV2.MAX_IPV4_STRING_LENGTH) + .sourceType("ipv4") + .build(); + BasicTypeDefine typeDefine = DorisTypeConverterV2.INSTANCE.reconvert(column); + Assertions.assertEquals(DorisTypeConverterV2.DORIS_IPV4, typeDefine.getColumnType()); + Assertions.assertEquals(DorisTypeConverterV2.DORIS_IPV4, typeDefine.getDataType()); + + column = + PhysicalColumn.builder() + .name("test") + .dataType(BasicType.STRING_TYPE) + .columnLength(DorisTypeConverterV2.MAX_IPV6_STRING_LENGTH) + .sourceType("ipv6") + .build(); + typeDefine = DorisTypeConverterV2.INSTANCE.reconvert(column); + Assertions.assertEquals(DorisTypeConverterV2.DORIS_IPV6, typeDefine.getColumnType()); + Assertions.assertEquals(DorisTypeConverterV2.DORIS_IPV6, typeDefine.getDataType()); + } + @Test public void testReconvertDate() { Column column = diff --git a/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/source/reader/DorisValueReaderIpv4Test.java b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/source/reader/DorisValueReaderIpv4Test.java new file mode 100644 index 0000000000..24b80248a2 --- /dev/null +++ b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/source/reader/DorisValueReaderIpv4Test.java @@ -0,0 +1,148 @@ +/* + * 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.source.reader; + +import org.apache.seatunnel.shade.org.apache.arrow.memory.RootAllocator; +import org.apache.seatunnel.shade.org.apache.arrow.vector.IntVector; +import org.apache.seatunnel.shade.org.apache.arrow.vector.VarCharVector; +import org.apache.seatunnel.shade.org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.seatunnel.shade.org.apache.arrow.vector.ipc.ArrowStreamWriter; + +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.catalog.PhysicalColumn; +import org.apache.seatunnel.api.table.catalog.TableIdentifier; +import org.apache.seatunnel.api.table.catalog.TableSchema; +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.connectors.seatunnel.common.source.arrow.reader.ArrowToSeatunnelRowReader; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayOutputStream; +import java.nio.channels.Channels; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Collections; + +public class DorisValueReaderIpv4Test { + + @Test + public void testIpv4ToString() { + // Doris BE sends IPV4 as a signed 32-bit integer + Assertions.assertEquals("0.0.0.0", DorisValueReader.ipv4ToString(0)); + Assertions.assertEquals("255.255.255.255", DorisValueReader.ipv4ToString(-1)); + Assertions.assertEquals("192.168.0.1", DorisValueReader.ipv4ToString(-1062731775)); + Assertions.assertEquals("10.0.0.255", DorisValueReader.ipv4ToString(167772415)); + Assertions.assertEquals("255.255.255.255", DorisValueReader.ipv4ToString(4294967295L)); + } + + @Test + public void testConvertOnlyIpv4Fields() { + CatalogTable catalogTable = catalogTable(); + int[] indexes = + DorisValueReader.ipv4FieldIndexes(catalogTable, catalogTable.getSeaTunnelRowType()); + Assertions.assertArrayEquals(new int[] {1}, indexes); + + SeaTunnelRow row = new SeaTunnelRow(new Object[] {1, -1062731775, "2001:db8::1", "a"}); + DorisValueReader.convertIpv4Fields(row, indexes); + Assertions.assertArrayEquals( + new Object[] {1, "192.168.0.1", "2001:db8::1", "a"}, row.getFields()); + + SeaTunnelRow nullRow = new SeaTunnelRow(new Object[] {2, null, null, null}); + DorisValueReader.convertIpv4Fields(nullRow, indexes); + Assertions.assertNull(nullRow.getField(1)); + } + + @Test + public void testConvertIpv4ReadFromArrowInt32() throws Exception { + // the same Arrow shape Doris BE sends: IPV4 as Int(32, true), IPV6 as Utf8 + byte[] arrow; + try (RootAllocator allocator = new RootAllocator(Long.MAX_VALUE); + IntVector id = new IntVector("id", allocator); + IntVector v4 = new IntVector("v4", allocator); + VarCharVector v6 = new VarCharVector("v6", allocator); + VarCharVector name = new VarCharVector("name", allocator)) { + id.allocateNew(2); + v4.allocateNew(2); + v6.allocateNew(2); + name.allocateNew(2); + id.set(0, 1); + v4.set(0, -1062731775); + v6.setSafe(0, "2001:db8::1".getBytes(StandardCharsets.UTF_8)); + name.setSafe(0, "a".getBytes(StandardCharsets.UTF_8)); + id.set(1, 2); + v4.setNull(1); + v6.setNull(1); + name.setNull(1); + id.setValueCount(2); + v4.setValueCount(2); + v6.setValueCount(2); + name.setValueCount(2); + try (VectorSchemaRoot root = VectorSchemaRoot.of(id, v4, v6, name); + ByteArrayOutputStream output = new ByteArrayOutputStream(); + ArrowStreamWriter writer = + new ArrowStreamWriter(root, null, Channels.newChannel(output))) { + writer.start(); + writer.writeBatch(); + writer.end(); + arrow = output.toByteArray(); + } + } + + CatalogTable catalogTable = catalogTable(); + int[] indexes = + DorisValueReader.ipv4FieldIndexes(catalogTable, catalogTable.getSeaTunnelRowType()); + try (ArrowToSeatunnelRowReader reader = + new ArrowToSeatunnelRowReader(arrow, catalogTable.getSeaTunnelRowType()) + .readArrow()) { + SeaTunnelRow row = reader.next(); + // the Arrow reader keeps the integer even though the field type is STRING + Assertions.assertEquals(-1062731775, row.getField(1)); + DorisValueReader.convertIpv4Fields(row, indexes); + Assertions.assertArrayEquals( + new Object[] {1, "192.168.0.1", "2001:db8::1", "a"}, row.getFields()); + + SeaTunnelRow nullRow = reader.next(); + DorisValueReader.convertIpv4Fields(nullRow, indexes); + Assertions.assertArrayEquals(new Object[] {2, null, null, null}, nullRow.getFields()); + } + } + + private static CatalogTable catalogTable() { + return CatalogTable.of( + TableIdentifier.of("doris", "db", "t"), + TableSchema.builder() + .columns( + Arrays.asList( + column("id", BasicType.INT_TYPE, "int"), + column("v4", BasicType.STRING_TYPE, "ipv4"), + column("v6", BasicType.STRING_TYPE, "ipv6"), + column("name", BasicType.STRING_TYPE, "varchar(10)"))) + .build(), + Collections.emptyMap(), + Collections.emptyList(), + ""); + } + + private static PhysicalColumn column( + String name, SeaTunnelDataType<?> type, String sourceType) { + return PhysicalColumn.builder().name(name).dataType(type).sourceType(sourceType).build(); + } +} diff --git a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/util/SchemaUtils.java b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/util/SchemaUtilsTest.java similarity index 50% copy from seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/util/SchemaUtils.java copy to seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/util/SchemaUtilsTest.java index bf8f3b9a26..35de1e0901 100644 --- a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/util/SchemaUtils.java +++ b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/util/SchemaUtilsTest.java @@ -17,34 +17,31 @@ package org.apache.seatunnel.connectors.doris.util; -import org.apache.seatunnel.connectors.doris.rest.models.Field; import org.apache.seatunnel.connectors.doris.rest.models.Schema; +import org.apache.doris.sdk.thrift.TPrimitiveType; import org.apache.doris.sdk.thrift.TScanColumnDesc; -import java.util.List; - -public class SchemaUtils { - - /** - * convert Doris return schema to inner schema struct. - * - * @param tscanColumnDescs Doris BE return schema - * @return inner schema struct - */ - public static Schema convertToSchema(List<TScanColumnDesc> tscanColumnDescs) { - Schema schema = new Schema(tscanColumnDescs.size()); - tscanColumnDescs.stream() - .forEach( - desc -> - schema.put( - new Field( - desc.getName(), - desc.getType().name(), - "", - 0, - 0, - ""))); - return schema; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; + +public class SchemaUtilsTest { + + @Test + public void testConvertToSchemaWithUnknownType() { + TScanColumnDesc known = new TScanColumnDesc(); + known.setName("id"); + known.setType(TPrimitiveType.INT); + // a type the shaded thrift-service does not know, such as IPV4, deserializes to null + TScanColumnDesc unknown = new TScanColumnDesc(); + unknown.setName("v4"); + + Schema schema = SchemaUtils.convertToSchema(Arrays.asList(known, unknown)); + + Assertions.assertEquals(2, schema.size()); + Assertions.assertEquals("INT", schema.getProperties().get(0).getType()); + Assertions.assertEquals("UNKNOWN", schema.getProperties().get(1).getType()); } }
