This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 59775c6cb0 [Fix][Connector-V2][Doris] Support IPV4 and IPV6 columns
(#12761)
59775c6cb0 is described below
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());
}
}
