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 3341f3b3cb [Fix][Connector-V2] Fix Paimon nested row write (#10888)
3341f3b3cb is described below
commit 3341f3b3cb079eee5b6dd6df7cd732e86d054d10
Author: QuakeWang <[email protected]>
AuthorDate: Wed May 27 22:09:45 2026 +0800
[Fix][Connector-V2] Fix Paimon nested row write (#10888)
Signed-off-by: QuakeWang <[email protected]>
---
.../seatunnel/paimon/utils/RowConverter.java | 31 ++--
.../seatunnel/paimon/utils/RowTypeConverter.java | 13 +-
.../seatunnel/paimon/utils/RowConverterTest.java | 186 ++++++++++++++++++++-
.../paimon/utils/RowTypeConverterTest.java | 16 ++
4 files changed, 226 insertions(+), 20 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowConverter.java
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowConverter.java
index c47e5e140b..e25de32f24 100644
---
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowConverter.java
+++
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowConverter.java
@@ -365,13 +365,17 @@ public class RowConverter {
SeaTunnelRow seaTunnelRow,
SeaTunnelRowType seaTunnelRowType,
TableSchema sinkTableSchema) {
- List<DataField> sinkTotalFields = sinkTableSchema.fields();
+ return reconvert(seaTunnelRow, seaTunnelRowType,
sinkTableSchema.fields());
+ }
+
+ private static InternalRow reconvert(
+ SeaTunnelRow seaTunnelRow,
+ SeaTunnelRowType seaTunnelRowType,
+ List<DataField> sinkFields) {
int sourceTotalFields = seaTunnelRowType.getTotalFields();
- if (sourceTotalFields != sinkTotalFields.size()) {
+ if (sourceTotalFields != sinkFields.size()) {
throw CommonError.writeRowErrorWithFieldsCountNotMatch(
- PaimonBaseOptions.CONNECTOR_IDENTITY,
- sourceTotalFields,
- sinkTotalFields.size());
+ PaimonBaseOptions.CONNECTOR_IDENTITY, sourceTotalFields,
sinkFields.size());
}
BinaryRow binaryRow = new BinaryRow(sourceTotalFields);
BinaryWriter binaryWriter = new BinaryRowWriter(binaryRow);
@@ -393,7 +397,7 @@ public class RowConverter {
binaryWriter.setNullAt(i);
continue;
}
- checkCanWriteWithSchema(i, seaTunnelRowType, sinkTotalFields,
fieldValue);
+ checkCanWriteWithSchema(i, seaTunnelRowType, sinkFields,
fieldValue);
String fieldName = seaTunnelRowType.getFieldName(i);
switch (fieldTypes[i].getSqlType()) {
case TINYINT:
@@ -415,8 +419,7 @@ public class RowConverter {
binaryWriter.writeDouble(i, (Double) fieldValue);
break;
case DECIMAL:
- DataField decimalDataField =
- SchemaUtil.getDataField(sinkTotalFields,
fieldName);
+ DataField decimalDataField =
SchemaUtil.getDataField(sinkFields, fieldName);
org.apache.paimon.types.DecimalType decimalType =
(org.apache.paimon.types.DecimalType)
decimalDataField.type();
binaryWriter.writeDecimal(
@@ -442,7 +445,7 @@ public class RowConverter {
.setValue(binaryWriter, i,
DateTimeUtils.toInternal(date));
break;
case TIMESTAMP:
- DataField dataField =
SchemaUtil.getDataField(sinkTotalFields, fieldName);
+ DataField dataField = SchemaUtil.getDataField(sinkFields,
fieldName);
int precision = ((TimestampType)
dataField.type()).getPrecision();
LocalDateTime datetime = (LocalDateTime) fieldValue;
binaryWriter.writeTimestamp(
@@ -483,11 +486,13 @@ public class RowConverter {
case ROW:
SeaTunnelDataType<?> rowType =
seaTunnelRowType.getFieldType(i);
Object row = fieldValue;
- TableSchema nestedSchema =
extractNestedSchema(sinkTotalFields.get(i).type());
- InternalRow paimonRow =
- reconvert((SeaTunnelRow) row, (SeaTunnelRowType)
rowType, nestedSchema);
RowType paimonRowType =
- RowTypeConverter.reconvert((SeaTunnelRowType)
rowType, nestedSchema);
+ (RowType) SchemaUtil.getDataField(sinkFields,
fieldName).type();
+ InternalRow paimonRow =
+ reconvert(
+ (SeaTunnelRow) row,
+ (SeaTunnelRowType) rowType,
+ paimonRowType.getFields());
binaryWriter.writeRow(i, paimonRow, new
InternalRowSerializer(paimonRowType));
break;
default:
diff --git
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowTypeConverter.java
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowTypeConverter.java
index ef9d24c3d4..df437f1ab5 100644
---
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowTypeConverter.java
+++
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowTypeConverter.java
@@ -422,13 +422,16 @@ public class RowTypeConverter {
SeaTunnelDataType<?>[] fieldTypes = row.getFieldTypes();
String[] fieldNames = row.getFieldNames();
int totalFields = row.getTotalFields();
- DataType[] dataTypes = new DataType[totalFields];
+ DataField[] dataFields = new DataField[totalFields];
for (int i = 0; i < totalFields; i++) {
- dataTypes[i] =
- SeaTunnelTypeToPaimonVisitor.INSTANCE.visit(
- fieldNames[i], fieldTypes[i]);
+ dataFields[i] =
+ new DataField(
+ i,
+ fieldNames[i],
+
SeaTunnelTypeToPaimonVisitor.INSTANCE.visit(
+ fieldNames[i], fieldTypes[i]));
}
- return DataTypes.ROW(dataTypes);
+ return DataTypes.ROW(dataFields);
default:
throw CommonError.unsupportedDataType(
PaimonBaseOptions.CONNECTOR_IDENTITY,
diff --git
a/seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowConverterTest.java
b/seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowConverterTest.java
index f62884c005..92d26e523f 100644
---
a/seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowConverterTest.java
+++
b/seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowConverterTest.java
@@ -369,6 +369,152 @@ public class RowConverterTest {
});
}
+ @Test
+ public void seaTunnelToPaimonWithNestedRow() {
+ SeaTunnelRowType sourceType = nestedRowType(BasicType.STRING_TYPE);
+ SeaTunnelRow data = nestedRowData("Alice");
+ RowType sinkRowType =
+ nestedPaimonRowType(
+ RowType.of(
+ new DataType[] {DataTypes.BIGINT(),
DataTypes.STRING()},
+ new String[] {"id", "name"}));
+
+ InternalRow actual = RowConverter.reconvert(data, sourceType,
tableSchema(sinkRowType));
+
+ Assertions.assertEquals(1L, actual.getLong(0));
+ InternalRow buyer = actual.getRow(1, 2);
+ Assertions.assertEquals(1001L, buyer.getLong(0));
+ Assertions.assertEquals("Alice", buyer.getString(1).toString());
+ Assertions.assertEquals(
+ new BigDecimal("20.50"), actual.getDecimal(2, 10,
2).toBigDecimal());
+ }
+
+ @Test
+ public void seaTunnelToPaimonWithNestedRowFieldCountMismatch() {
+ SeaTunnelRowType sourceType = nestedRowType(BasicType.STRING_TYPE);
+ SeaTunnelRow data = nestedRowData("Alice");
+ RowType sinkRowType =
+ nestedPaimonRowType(
+ RowType.of(new DataType[] {DataTypes.BIGINT()}, new
String[] {"id"}));
+
+ SeaTunnelRuntimeException actual =
+ Assertions.assertThrows(
+ SeaTunnelRuntimeException.class,
+ () -> RowConverter.reconvert(data, sourceType,
tableSchema(sinkRowType)));
+
+ Assertions.assertEquals(
+ CommonError.writeRowErrorWithFieldsCountNotMatch("Paimon", 2,
1).getMessage(),
+ actual.getMessage());
+ }
+
+ @Test
+ public void seaTunnelToPaimonRejectsIncompatibleNestedRowSchema() {
+ SeaTunnelRowType sourceType = nestedRowType(BasicType.STRING_TYPE);
+ SeaTunnelRow data = nestedRowData("Alice");
+ RowType sinkBuyerRowType =
+ RowType.of(
+ new DataType[] {DataTypes.BIGINT(), DataTypes.INT()},
+ new String[] {"id", "name"});
+ RowType sinkRowType = nestedPaimonRowType(sinkBuyerRowType);
+
+ SeaTunnelRuntimeException actual =
+ Assertions.assertThrows(
+ SeaTunnelRuntimeException.class,
+ () -> RowConverter.reconvert(data, sourceType,
tableSchema(sinkRowType)));
+
+ Assertions.assertEquals(
+ CommonError.writeRowErrorWithSchemaIncompatibleSchema(
+ "Paimon",
+ "name" + StringUtils.SPACE + "STRING",
+ new DataField(1, "name",
DataTypes.STRING()).asSQLString(),
+
sinkBuyerRowType.getFields().get(1).asSQLString())
+ .getMessage(),
+ actual.getMessage());
+ }
+
+ @Test
+ public void seaTunnelToPaimonRejectsNestedRowFieldNameMismatch() {
+ SeaTunnelRowType sourceType = nestedRowType(BasicType.STRING_TYPE);
+ SeaTunnelRow data = nestedRowData("Alice");
+ RowType sinkBuyerRowType =
+ RowType.of(
+ new DataType[] {DataTypes.BIGINT(),
DataTypes.STRING()},
+ new String[] {"id", "full_name"});
+ RowType sinkRowType = nestedPaimonRowType(sinkBuyerRowType);
+
+ SeaTunnelRuntimeException actual =
+ Assertions.assertThrows(
+ SeaTunnelRuntimeException.class,
+ () -> RowConverter.reconvert(data, sourceType,
tableSchema(sinkRowType)));
+
+ Assertions.assertEquals(
+ CommonError.writeRowErrorWithSchemaIncompatibleSchema(
+ "Paimon",
+ "name" + StringUtils.SPACE + "STRING",
+ new DataField(1, "name",
DataTypes.STRING()).asSQLString(),
+
sinkBuyerRowType.getFields().get(1).asSQLString())
+ .getMessage(),
+ actual.getMessage());
+ }
+
+ @Test
+ public void seaTunnelToPaimonWithNullNestedRow() {
+ SeaTunnelRowType sourceType = nestedRowType(BasicType.STRING_TYPE);
+ SeaTunnelRow data = new SeaTunnelRow(new Object[] {1L, null, new
BigDecimal("20.50")});
+ RowType sinkRowType =
+ nestedPaimonRowType(
+ RowType.of(
+ new DataType[] {DataTypes.BIGINT(),
DataTypes.STRING()},
+ new String[] {"id", "name"}));
+
+ InternalRow actual = RowConverter.reconvert(data, sourceType,
tableSchema(sinkRowType));
+
+ Assertions.assertTrue(actual.isNullAt(1));
+ }
+
+ @Test
+ public void seaTunnelToPaimonWithNestedRowPrecision() {
+ SeaTunnelRowType buyerType =
+ new SeaTunnelRowType(
+ new String[] {"created_at", "score"},
+ new SeaTunnelDataType<?>[] {
+ LocalTimeType.LOCAL_DATE_TIME_TYPE, new
DecimalType(6, 2)
+ });
+ SeaTunnelRowType sourceType =
+ new SeaTunnelRowType(
+ new String[] {"order_id", "buyer", "amount"},
+ new SeaTunnelDataType<?>[] {
+ BasicType.LONG_TYPE, buyerType, new
DecimalType(10, 2)
+ });
+ LocalDateTime createdAt = LocalDateTime.of(2024, 1, 2, 3, 4, 5,
123000000);
+ SeaTunnelRow data =
+ new SeaTunnelRow(
+ new Object[] {
+ 1L,
+ new SeaTunnelRow(new Object[] {createdAt, new
BigDecimal("12.34")}),
+ new BigDecimal("20.50")
+ });
+ RowType sinkRowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.BIGINT(),
+ RowType.of(
+ new DataType[] {
+ DataTypes.TIMESTAMP(3),
DataTypes.DECIMAL(8, 4)
+ },
+ new String[] {"created_at", "score"}),
+ DataTypes.DECIMAL(10, 2)
+ },
+ new String[] {"order_id", "buyer", "amount"});
+
+ InternalRow actual = RowConverter.reconvert(data, sourceType,
tableSchema(sinkRowType));
+
+ InternalRow buyer = actual.getRow(1, 2);
+ Assertions.assertEquals(createdAt, buyer.getTimestamp(0,
3).toLocalDateTime());
+ Assertions.assertEquals(
+ 0, new BigDecimal("12.34").compareTo(buyer.getDecimal(1, 8,
4).toBigDecimal()));
+ }
+
@Test
public void nestedRowPaimonToSeaTunnel() {
// Top-level: c_int INT, c_string STRING, c_nested ROW<a INT, b STRING>
@@ -417,7 +563,7 @@ public class RowConverterTest {
BasicType.INT_TYPE, BasicType.STRING_TYPE,
seaTunnelNestedType
});
- // Convert Paimon → SeaTunnel
+ // Convert Paimon to SeaTunnel
SeaTunnelRow result = RowConverter.convert(binaryRow,
seaTunnelTopLevelType, tableSchema);
Assertions.assertEquals(1, result.getField(0));
@@ -465,7 +611,7 @@ public class RowConverterTest {
SeaTunnelRow nestedRow = new SeaTunnelRow(new Object[] {42, "world"});
SeaTunnelRow topLevelRow = new SeaTunnelRow(new Object[] {1, "hello",
nestedRow});
- // Convert SeaTunnel → Paimon (would throw before fix due to field
count mismatch)
+ // Convert SeaTunnel to Paimon (would throw before fix due to field
count mismatch)
InternalRow result =
RowConverter.reconvert(topLevelRow, seaTunnelTopLevelType,
tableSchema);
@@ -475,4 +621,40 @@ public class RowConverterTest {
Assertions.assertEquals(42, nestedResult.getInt(0));
Assertions.assertEquals("world", nestedResult.getString(1).toString());
}
+
+ private static SeaTunnelRowType nestedRowType(SeaTunnelDataType<?>
buyerNameType) {
+ SeaTunnelRowType buyerType =
+ new SeaTunnelRowType(
+ new String[] {"id", "name"},
+ new SeaTunnelDataType<?>[] {BasicType.LONG_TYPE,
buyerNameType});
+ return new SeaTunnelRowType(
+ new String[] {"order_id", "buyer", "amount"},
+ new SeaTunnelDataType<?>[] {
+ BasicType.LONG_TYPE, buyerType, new DecimalType(10, 2)
+ });
+ }
+
+ private static SeaTunnelRow nestedRowData(String buyerName) {
+ return new SeaTunnelRow(
+ new Object[] {
+ 1L, new SeaTunnelRow(new Object[] {1001L, buyerName}), new
BigDecimal("20.50")
+ });
+ }
+
+ private static RowType nestedPaimonRowType(RowType buyerType) {
+ return RowType.of(
+ new DataType[] {DataTypes.BIGINT(), buyerType,
DataTypes.DECIMAL(10, 2)},
+ new String[] {"order_id", "buyer", "amount"});
+ }
+
+ private static TableSchema tableSchema(RowType rowType) {
+ return new TableSchema(
+ 0,
+ TableSchema.newFields(rowType),
+ rowType.getFieldCount(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyMap(),
+ "");
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowTypeConverterTest.java
b/seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowTypeConverterTest.java
index 0d29320768..1ae7fa1396 100644
---
a/seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowTypeConverterTest.java
+++
b/seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/RowTypeConverterTest.java
@@ -231,6 +231,22 @@ public class RowTypeConverterTest {
Assertions.assertEquals(convert, rowType);
}
+ @Test
+ public void seaTunnelNestedRowToPaimonKeepsFieldNames() {
+ SeaTunnelRowType buyerType =
+ new SeaTunnelRowType(
+ new String[] {"id", "name"},
+ new SeaTunnelDataType<?>[] {BasicType.LONG_TYPE,
BasicType.STRING_TYPE});
+
+ DataType convert = RowTypeConverter.reconvert("buyer", buyerType);
+
+ RowType expected =
+ RowType.of(
+ new DataType[] {DataTypes.BIGINT(),
DataTypes.STRING()},
+ new String[] {"id", "name"});
+ Assertions.assertEquals(expected, convert);
+ }
+
@Test
public void paimonDataTypeToSeaTunnelColumn() {
Column column = RowTypeConverter.convert(typeDefine);
