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-12770-705f7944caae403540eab8a0a40d0ad2258b6265 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 2f3bcd276a8a2ae4fc177e25c6c635d3eb527ee7 Author: Jast <[email protected]> AuthorDate: Sat Oct 17 06:55:15 2026 +0000 [Fix][Connector-V2] Fix Hudi catalog schema conversion for MAP and ARRAY columns (#12770) --- .../connectors/seatunnel/hudi/util/SchemaUtil.java | 7 +- .../seatunnel/hudi/util/SchemaUtilTest.java | 110 +++++++++++++++++++++ 2 files changed, 114 insertions(+), 3 deletions(-) diff --git a/seatunnel-connectors-v2/connector-hudi/src/main/java/org/apache/seatunnel/connectors/seatunnel/hudi/util/SchemaUtil.java b/seatunnel-connectors-v2/connector-hudi/src/main/java/org/apache/seatunnel/connectors/seatunnel/hudi/util/SchemaUtil.java index cd8eb620e8..b857f0c696 100644 --- a/seatunnel-connectors-v2/connector-hudi/src/main/java/org/apache/seatunnel/connectors/seatunnel/hudi/util/SchemaUtil.java +++ b/seatunnel-connectors-v2/connector-hudi/src/main/java/org/apache/seatunnel/connectors/seatunnel/hudi/util/SchemaUtil.java @@ -44,7 +44,7 @@ public class SchemaUtil { case NULL: return BasicType.STRING_TYPE; case ARRAY: - return convertListType(field, schema.getElementType()); + return convertListType(field, schema); case MAP: return convertMapType(field, schema); case BYTES: @@ -109,9 +109,9 @@ public class SchemaUtil { } private static MapType convertMapType(String field, Schema schema) { + // Avro map keys are always strings return new MapType( - convertSeaTunnelType(field, schema.getElementType()), - convertSeaTunnelType(field, schema.getValueType())); + BasicType.STRING_TYPE, convertSeaTunnelType(field, schema.getValueType())); } private static SeaTunnelRowType convertStructType(Schema schema) { @@ -127,6 +127,7 @@ public class SchemaUtil { } private static ArrayType convertListType(String field, Schema schema) { + // schema is the array schema; unwrap the element type here switch (schema.getElementType().getType()) { case BOOLEAN: return ArrayType.BOOLEAN_ARRAY_TYPE; diff --git a/seatunnel-connectors-v2/connector-hudi/src/test/java/org/apache/seatunnel/connectors/seatunnel/hudi/util/SchemaUtilTest.java b/seatunnel-connectors-v2/connector-hudi/src/test/java/org/apache/seatunnel/connectors/seatunnel/hudi/util/SchemaUtilTest.java new file mode 100644 index 0000000000..df3089c55f --- /dev/null +++ b/seatunnel-connectors-v2/connector-hudi/src/test/java/org/apache/seatunnel/connectors/seatunnel/hudi/util/SchemaUtilTest.java @@ -0,0 +1,110 @@ +/* + * 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.hudi.util; + +import org.apache.seatunnel.api.table.type.ArrayType; +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.DecimalType; +import org.apache.seatunnel.api.table.type.MapType; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; + +import org.apache.avro.LogicalTypes; +import org.apache.avro.Schema; +import org.apache.avro.SchemaBuilder; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class SchemaUtilTest { + + @Test + public void shouldConvertMapWithStringValues() { + Schema schema = SchemaBuilder.map().values().stringType(); + SeaTunnelDataType<?> dataType = SchemaUtil.convertSeaTunnelType("map_col", schema); + Assertions.assertEquals( + new MapType<>(BasicType.STRING_TYPE, BasicType.STRING_TYPE), dataType); + } + + @Test + public void shouldConvertMapWithIntValues() { + Schema schema = SchemaBuilder.map().values().intType(); + SeaTunnelDataType<?> dataType = SchemaUtil.convertSeaTunnelType("map_col", schema); + Assertions.assertEquals(new MapType<>(BasicType.STRING_TYPE, BasicType.INT_TYPE), dataType); + } + + @Test + public void shouldConvertMapWithNullableValues() { + Schema schema = + SchemaBuilder.map().values().unionOf().nullType().and().longType().endUnion(); + SeaTunnelDataType<?> dataType = SchemaUtil.convertSeaTunnelType("map_col", schema); + Assertions.assertEquals( + new MapType<>(BasicType.STRING_TYPE, BasicType.LONG_TYPE), dataType); + } + + @Test + public void shouldConvertMapWithArrayValues() { + Schema schema = SchemaBuilder.map().values().array().items().doubleType(); + SeaTunnelDataType<?> dataType = SchemaUtil.convertSeaTunnelType("map_col", schema); + Assertions.assertEquals( + new MapType<>(BasicType.STRING_TYPE, ArrayType.DOUBLE_ARRAY_TYPE), dataType); + } + + @Test + public void shouldConvertMapWithDecimalValues() { + Schema decimal = LogicalTypes.decimal(10, 2).addToSchema(Schema.create(Schema.Type.BYTES)); + Schema schema = Schema.createMap(decimal); + SeaTunnelDataType<?> dataType = SchemaUtil.convertSeaTunnelType("map_col", schema); + Assertions.assertEquals( + new MapType<>(BasicType.STRING_TYPE, new DecimalType(10, 2)), dataType); + } + + @Test + public void shouldConvertRecordContainingMapColumn() { + Schema schema = + SchemaBuilder.record("test_record") + .fields() + .name("id") + .type() + .longType() + .noDefault() + .name("attributes") + .type() + .map() + .values() + .stringType() + .noDefault() + .endRecord(); + SeaTunnelRowType rowType = + (SeaTunnelRowType) SchemaUtil.convertSeaTunnelType("record", schema); + Assertions.assertEquals(2, rowType.getTotalFields()); + Assertions.assertEquals("id", rowType.getFieldName(0)); + Assertions.assertEquals(BasicType.LONG_TYPE, rowType.getFieldType(0)); + Assertions.assertEquals("attributes", rowType.getFieldName(1)); + Assertions.assertEquals( + new MapType<>(BasicType.STRING_TYPE, BasicType.STRING_TYPE), + rowType.getFieldType(1)); + } + + @Test + public void shouldStillConvertArrays() { + Schema schema = SchemaBuilder.array().items().stringType(); + SeaTunnelDataType<?> dataType = SchemaUtil.convertSeaTunnelType("array_col", schema); + Assertions.assertEquals(ArrayType.STRING_ARRAY_TYPE, dataType); + } +}
