This is an automated email from the ASF dual-hosted git repository.
davidzollo 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 0f48aef4ff [Fix][Connector-V2] Fix Hudi null struct field conversion
(#11196)
0f48aef4ff is described below
commit 0f48aef4ffd8f19a77a1071e6efade3124149e3f
Author: Jast <[email protected]>
AuthorDate: Sat Jun 27 20:50:08 2026 +0800
[Fix][Connector-V2] Fix Hudi null struct field conversion (#11196)
---
.../hudi/sink/convert/AvroSchemaConverter.java | 18 ++++--
.../hudi/sink/convert/HudiRecordConverter.java | 14 +----
.../connectors/seatunnel/hudi/HudiTest.java | 65 ++++++++++++++++++++++
3 files changed, 82 insertions(+), 15 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-hudi/src/main/java/org/apache/seatunnel/connectors/seatunnel/hudi/sink/convert/AvroSchemaConverter.java
b/seatunnel-connectors-v2/connector-hudi/src/main/java/org/apache/seatunnel/connectors/seatunnel/hudi/sink/convert/AvroSchemaConverter.java
index acb1021275..2f9ac4f2a7 100644
---
a/seatunnel-connectors-v2/connector-hudi/src/main/java/org/apache/seatunnel/connectors/seatunnel/hudi/sink/convert/AvroSchemaConverter.java
+++
b/seatunnel-connectors-v2/connector-hudi/src/main/java/org/apache/seatunnel/connectors/seatunnel/hudi/sink/convert/AvroSchemaConverter.java
@@ -64,6 +64,11 @@ public class AvroSchemaConverter implements Serializable {
* @return Avro's {@link Schema} matching this logical type.
*/
public static Schema convertToSchema(SeaTunnelDataType<?> dataType, String
rowName) {
+ return convertToSchema(dataType, rowName, false);
+ }
+
+ private static Schema convertToSchema(
+ SeaTunnelDataType<?> dataType, String rowName, boolean
nullableRow) {
switch (dataType.getSqlType()) {
case BOOLEAN:
Schema bool = SchemaBuilder.builder().booleanType();
@@ -126,25 +131,30 @@ public class AvroSchemaConverter implements Serializable {
SeaTunnelDataType<?> fieldType = rowType.getFieldType(i);
SchemaBuilder.GenericDefault<Schema> fieldBuilder =
builder.name(fieldName)
- .type(convertToSchema(fieldType, rowName +
"." + fieldName));
+ .type(
+ convertToSchema(
+ fieldType, rowName + "." +
fieldName, true));
builder = fieldBuilder.withDefault(null);
}
- return builder.endRecord();
+ Schema record = builder.endRecord();
+ return nullableRow ? nullableSchema(record) : record;
case MAP:
Schema map =
SchemaBuilder.builder()
.map()
.values(
convertToSchema(
-
extractValueTypeToAvroMap(dataType), rowName));
+
extractValueTypeToAvroMap(dataType),
+ rowName,
+ true));
return nullableSchema(map);
case ARRAY:
ArrayType<?, ?> arrayType = (ArrayType<?, ?>) dataType;
Schema array =
SchemaBuilder.builder()
.array()
-
.items(convertToSchema(arrayType.getElementType(), rowName));
+
.items(convertToSchema(arrayType.getElementType(), rowName, true));
return nullableSchema(array);
default:
throw new UnsupportedOperationException(
diff --git
a/seatunnel-connectors-v2/connector-hudi/src/main/java/org/apache/seatunnel/connectors/seatunnel/hudi/sink/convert/HudiRecordConverter.java
b/seatunnel-connectors-v2/connector-hudi/src/main/java/org/apache/seatunnel/connectors/seatunnel/hudi/sink/convert/HudiRecordConverter.java
index e8d23ab4ef..0cc8ba557a 100644
---
a/seatunnel-connectors-v2/connector-hudi/src/main/java/org/apache/seatunnel/connectors/seatunnel/hudi/sink/convert/HudiRecordConverter.java
+++
b/seatunnel-connectors-v2/connector-hudi/src/main/java/org/apache/seatunnel/connectors/seatunnel/hudi/sink/convert/HudiRecordConverter.java
@@ -24,7 +24,6 @@ import
org.apache.seatunnel.connectors.seatunnel.hudi.config.HudiTableConfig;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
-import org.apache.hudi.avro.AvroSchemaUtils;
import org.apache.hudi.common.model.HoodieAvroPayload;
import org.apache.hudi.common.model.HoodieAvroRecord;
import org.apache.hudi.common.model.HoodieKey;
@@ -39,7 +38,6 @@ import java.util.Arrays;
import java.util.UUID;
import java.util.stream.Collectors;
-import static
org.apache.seatunnel.connectors.seatunnel.hudi.sink.convert.AvroSchemaConverter.convertToSchema;
import static
org.apache.seatunnel.connectors.seatunnel.hudi.sink.convert.RowDataToAvroConverters.createConverter;
public class HudiRecordConverter implements Serializable {
@@ -59,17 +57,11 @@ public class HudiRecordConverter implements Serializable {
HudiTableConfig hudiTableConfig) {
GenericRecord rec = new GenericData.Record(schema);
for (int i = 0; i < seaTunnelRowType.getTotalFields(); i++) {
+ String fieldName = seaTunnelRowType.getFieldNames()[i];
rec.put(
- seaTunnelRowType.getFieldNames()[i],
+ fieldName,
createConverter(seaTunnelRowType.getFieldType(i))
- .convert(
- convertToSchema(
- seaTunnelRowType.getFieldType(i),
-
AvroSchemaUtils.getAvroRecordQualifiedName(
-
hudiTableConfig.getTableName())
- + "."
- +
seaTunnelRowType.getFieldNames()[i]),
- element.getField(i)));
+ .convert(schema.getField(fieldName).schema(),
element.getField(i)));
}
return new HoodieAvroRecord<>(
getHoodieKey(element, seaTunnelRowType, hudiTableConfig),
diff --git
a/seatunnel-connectors-v2/connector-hudi/src/test/java/org/apache/seatunnel/connectors/seatunnel/hudi/HudiTest.java
b/seatunnel-connectors-v2/connector-hudi/src/test/java/org/apache/seatunnel/connectors/seatunnel/hudi/HudiTest.java
index 7dbfc402b6..633b5e69aa 100644
---
a/seatunnel-connectors-v2/connector-hudi/src/test/java/org/apache/seatunnel/connectors/seatunnel/hudi/HudiTest.java
+++
b/seatunnel-connectors-v2/connector-hudi/src/test/java/org/apache/seatunnel/connectors/seatunnel/hudi/HudiTest.java
@@ -23,10 +23,15 @@ 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.hudi.config.HudiTableConfig;
+import
org.apache.seatunnel.connectors.seatunnel.hudi.sink.convert.HudiRecordConverter;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.EncoderFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hudi.avro.AvroSchemaUtils;
import org.apache.hudi.client.HoodieJavaWriteClient;
@@ -37,6 +42,7 @@ import org.apache.hudi.common.model.HoodieAvroRecord;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
@@ -53,6 +59,7 @@ import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
import org.junit.jupiter.api.io.TempDir;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.math.BigDecimal;
import java.sql.Timestamp;
@@ -127,6 +134,45 @@ public class HudiTest {
getSchema());
}
+ @Test
+ void testConvertNullableStructField() throws IOException {
+ SeaTunnelRowType structType =
+ new SeaTunnelRowType(
+ new String[] {"field1", "field2"},
+ new SeaTunnelDataType[] {STRING_TYPE, INT_TYPE});
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"id", "payload"},
+ new SeaTunnelDataType[] {INT_TYPE, structType});
+ Schema schema =
+ convertToSchema(rowType,
AvroSchemaUtils.getAvroRecordQualifiedName(tableName));
+
+ Assertions.assertEquals(Schema.Type.RECORD, schema.getType());
+ Assertions.assertEquals(Schema.Type.UNION,
schema.getField("payload").schema().getType());
+
+ SeaTunnelRow nullStructRow = new SeaTunnelRow(2);
+ nullStructRow.setField(0, 1);
+ nullStructRow.setField(1, null);
+
+ GenericRecord nullStructRecord = convertToAvroRecord(schema, rowType,
nullStructRow);
+ Assertions.assertNull(nullStructRecord.get("payload"));
+ writeAvroRecord(schema, nullStructRecord);
+
+ SeaTunnelRow struct = new SeaTunnelRow(2);
+ struct.setField(0, "value");
+ struct.setField(1, 100);
+
+ SeaTunnelRow structRow = new SeaTunnelRow(2);
+ structRow.setField(0, 2);
+ structRow.setField(1, struct);
+
+ GenericRecord structRecord = convertToAvroRecord(schema, rowType,
structRow);
+ GenericRecord payload = (GenericRecord) structRecord.get("payload");
+ Assertions.assertEquals("value", payload.get("field1").toString());
+ Assertions.assertEquals(100, payload.get("field2"));
+ writeAvroRecord(schema, structRecord);
+ }
+
@Test
@DisabledOnOs(OS.WINDOWS)
void testWriteData() throws IOException {
@@ -210,6 +256,25 @@ public class HudiTest {
getHoodieKey(element, seaTunnelRowType), new
HoodieAvroPayload(Option.of(rec)));
}
+ private GenericRecord convertToAvroRecord(
+ Schema schema, SeaTunnelRowType rowType, SeaTunnelRow element)
throws IOException {
+ HudiTableConfig tableConfig =
+ HudiTableConfig.builder()
+ .tableName(tableName)
+ .opType(WriteOperationType.INSERT)
+ .build();
+ HoodieRecord<HoodieAvroPayload> record =
+ new HudiRecordConverter().convertRow(schema, rowType, element,
tableConfig);
+ return (GenericRecord) record.getData().getInsertValue(schema).get();
+ }
+
+ private void writeAvroRecord(Schema schema, GenericRecord record) throws
IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(out, null);
+ new GenericDatumWriter<GenericRecord>(schema).write(record, encoder);
+ encoder.flush();
+ }
+
private HoodieKey getHoodieKey(SeaTunnelRow element, SeaTunnelRowType
seaTunnelRowType) {
String partitionPath = getRecordPartitionPath(element,
seaTunnelRowType);
String rowKey = getRecordKey(element, seaTunnelRowType);
