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-12846-1e9373184cf0f50f2cfbb7f846901940ef67e128 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 51c3c5fc527570904da4a8c6c4fbe74eb9a17a47 Author: Steven Zhao <[email protected]> AuthorDate: Mon Oct 19 12:47:51 2026 +0000 [Fix][Connector-V2] Reject unsupported Map columns in the Lance sink (#12846) --- docs/en/connectors/sink/Lance.md | 3 ++- docs/zh/connectors/sink/Lance.md | 3 ++- .../lance/exception/LanceConnectorErrorCode.java | 4 +++- .../seatunnel/lance/utils/SchemaUtils.java | 26 +++++++++++++++++++--- .../lance/utils/FragmentConverterTest.java | 26 ++++++++++++++++++++++ 5 files changed, 56 insertions(+), 6 deletions(-) diff --git a/docs/en/connectors/sink/Lance.md b/docs/en/connectors/sink/Lance.md index 9dfb8a484e..eac7f6d701 100644 --- a/docs/en/connectors/sink/Lance.md +++ b/docs/en/connectors/sink/Lance.md @@ -149,7 +149,8 @@ Lance uses the Apache Arrow type system. The sink creates the Lance schema from | TIMESTAMP | timestamp (microsecond, Asia/Shanghai timezone) | | STRING | utf8 | | ARRAY | list | -| MAP | map | + +Map columns are not supported. Lance's native format rejects Arrow Map values, so the sink fails during schema creation. Serialize the Map to a string or flatten it into columns before writing to Lance. :::tip diff --git a/docs/zh/connectors/sink/Lance.md b/docs/zh/connectors/sink/Lance.md index 0c1aec791d..7115fb3ead 100644 --- a/docs/zh/connectors/sink/Lance.md +++ b/docs/zh/connectors/sink/Lance.md @@ -141,7 +141,8 @@ Lance 使用 Apache Arrow 类型系统。sink 会根据上游 SeaTunnel 表结 | TIMESTAMP | timestamp（微秒精度，Asia/Shanghai 时区） | | STRING | utf8 | | ARRAY | list | -| MAP | map | + +暂不支持 Map 列。Lance 原生格式会拒绝 Arrow Map 值，因此 sink 在构建 schema 时会直接报错。请在写入 Lance 之前将 Map 序列化为字符串，或展开成多个列。 :::tip diff --git a/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/exception/LanceConnectorErrorCode.java b/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/exception/LanceConnectorErrorCode.java index 88523d3bfb..70bc981266 100644 --- a/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/exception/LanceConnectorErrorCode.java +++ b/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/exception/LanceConnectorErrorCode.java @@ -26,7 +26,9 @@ public enum LanceConnectorErrorCode implements SeaTunnelErrorCode { TABLE_DATASET_PATH_OPEN_EXCEPTION("LANCE-03", "DataSet path open exception"), - TABLE_DATASET_WRITE_ST_ROW_EXCEPTION("LANCE-04", "Dataset write seatunnelRow exception"); + TABLE_DATASET_WRITE_ST_ROW_EXCEPTION("LANCE-04", "Dataset write seatunnelRow exception"), + + UNSUPPORTED_MAP_TYPE("LANCE-05", "Lance does not support Map columns"); private final String code; private final String description; diff --git a/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/utils/SchemaUtils.java b/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/utils/SchemaUtils.java index 4884402f85..ffe9eee501 100644 --- a/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/utils/SchemaUtils.java +++ b/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/utils/SchemaUtils.java @@ -25,9 +25,12 @@ import org.apache.seatunnel.api.table.type.DecimalType; import org.apache.seatunnel.api.table.type.SeaTunnelDataType; import org.apache.seatunnel.api.table.type.SeaTunnelRow; import org.apache.seatunnel.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.api.table.type.SqlType; import org.apache.seatunnel.common.exception.CommonError; import org.apache.seatunnel.connectors.seatunnel.lance.config.LanceCommonConfig; import org.apache.seatunnel.connectors.seatunnel.lance.data.LanceTypeMapper; +import org.apache.seatunnel.connectors.seatunnel.lance.exception.LanceConnectorErrorCode; +import org.apache.seatunnel.connectors.seatunnel.lance.exception.LanceConnectorException; import org.apache.arrow.vector.types.DateUnit; import org.apache.arrow.vector.types.FloatingPointPrecision; @@ -54,6 +57,7 @@ public class SchemaUtils { } public static Schema convertSchema(SeaTunnelRow element, SeaTunnelRowType seaTunnelRowType) { + rejectUnsupportedTypes(seaTunnelRowType); SeaTunnelDataType<?>[] fieldTypes = seaTunnelRowType.getFieldTypes(); List<Field> fieldList = Lists.newArrayList(); for (int i = 0; i < fieldTypes.length; i++) { @@ -127,9 +131,7 @@ public class SchemaUtils { fieldList.add(field); break; case MAP: - field = Field.nullable(fieldName, new ArrowType.Map(true)); - fieldList.add(field); - break; + throw unsupportedMapException(fieldName); case ARRAY: field = Field.nullable(fieldName, new ArrowType.List()); fieldList.add(field); @@ -146,6 +148,24 @@ public class SchemaUtils { return new Schema(fieldList); } + private static void rejectUnsupportedTypes(SeaTunnelRowType seaTunnelRowType) { + SeaTunnelDataType<?>[] fieldTypes = seaTunnelRowType.getFieldTypes(); + for (int i = 0; i < fieldTypes.length; i++) { + if (fieldTypes[i].getSqlType() == SqlType.MAP) { + throw unsupportedMapException(seaTunnelRowType.getFieldName(i)); + } + } + } + + private static LanceConnectorException unsupportedMapException(String fieldName) { + return new LanceConnectorException( + LanceConnectorErrorCode.UNSUPPORTED_MAP_TYPE, + String.format( + "Field '%s' has type MAP. Serialize the Map to a string or flatten it " + + "into columns before writing to Lance.", + fieldName)); + } + public static JsonArrowSchema convertJsonArrowSchema(TableSchema schema) { List<JsonArrowField> fields = new ArrayList<>(); for (Column column : schema.getColumns()) { diff --git a/seatunnel-connectors-v2/connector-lance/src/test/java/org/apache/seatunnel/connectors/seatunnel/lance/utils/FragmentConverterTest.java b/seatunnel-connectors-v2/connector-lance/src/test/java/org/apache/seatunnel/connectors/seatunnel/lance/utils/FragmentConverterTest.java index 3575f094de..67083313cb 100644 --- a/seatunnel-connectors-v2/connector-lance/src/test/java/org/apache/seatunnel/connectors/seatunnel/lance/utils/FragmentConverterTest.java +++ b/seatunnel-connectors-v2/connector-lance/src/test/java/org/apache/seatunnel/connectors/seatunnel/lance/utils/FragmentConverterTest.java @@ -19,9 +19,11 @@ package org.apache.seatunnel.connectors.seatunnel.lance.utils; import org.apache.seatunnel.api.table.type.ArrayType; import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.MapType; import org.apache.seatunnel.api.table.type.SeaTunnelDataType; import org.apache.seatunnel.api.table.type.SeaTunnelRow; import org.apache.seatunnel.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.connectors.seatunnel.lance.exception.LanceConnectorException; import org.apache.arrow.memory.RootAllocator; import org.apache.arrow.vector.VarCharVector; @@ -179,6 +181,30 @@ class FragmentConverterTest { } } + @Test + void shouldRejectMapColumnsWhenBuildingSchema() { + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"id", "attributes"}, + new SeaTunnelDataType<?>[] { + BasicType.INT_TYPE, + new MapType<>(BasicType.STRING_TYPE, BasicType.INT_TYPE) + }); + + LanceConnectorException exception = + Assertions.assertThrows( + LanceConnectorException.class, + () -> + SchemaUtils.convertSchema( + new SeaTunnelRow(new Object[] {1, null}), rowType)); + + Assertions.assertEquals( + "ErrorCode:[LANCE-05], ErrorDescription:[Lance does not support Map columns] - " + + "Field 'attributes' has type MAP. Serialize the Map to a string or " + + "flatten it into columns before writing to Lance.", + exception.getMessage()); + } + private static void createEmptyDataset( RootAllocator allocator, String datasetPath, Schema schema) { try (Dataset ignored =
