This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 40d486e2b64 [FLINK-40264][format/avro] Propagate
legacyTimestampMapping into nested row converters
40d486e2b64 is described below
commit 40d486e2b6465680ed93305ebc51bcc87b26b995
Author: Pritam Kumar Mishra <[email protected]>
AuthorDate: Wed Jul 29 22:38:43 2026 +0530
[FLINK-40264][format/avro] Propagate legacyTimestampMapping into nested row
converters
AvroToRowDataConverters#createConverter dispatched nested ROW types to the
single-argument createRowConverter overload, which always assumes the legacy
timestamp mapping. As a result a nested TIMESTAMP_LTZ column failed with
'Unsupported type: TIMESTAMP_LTZ' even when the non-legacy mapping was
requested, because only the non-legacy mapping supports that type.
---
.../formats/avro/AvroToRowDataConverters.java | 2 +-
.../avro/AvroRowDataDeSerializationSchemaTest.java | 40 ++++++++++++++++++++++
2 files changed, 41 insertions(+), 1 deletion(-)
diff --git
a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroToRowDataConverters.java
b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroToRowDataConverters.java
index 9c63c56c4dc..97a57c1d9f0 100644
---
a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroToRowDataConverters.java
+++
b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroToRowDataConverters.java
@@ -146,7 +146,7 @@ public class AvroToRowDataConverters {
case ARRAY:
return createArrayConverter((ArrayType) type,
legacyTimestampMapping);
case ROW:
- return createRowConverter((RowType) type);
+ return createRowConverter((RowType) type,
legacyTimestampMapping);
case MAP:
case MULTISET:
return createMapConverter(type, legacyTimestampMapping);
diff --git
a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowDataDeSerializationSchemaTest.java
b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowDataDeSerializationSchemaTest.java
index 84f9994f666..b9253504034 100644
---
a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowDataDeSerializationSchemaTest.java
+++
b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowDataDeSerializationSchemaTest.java
@@ -27,6 +27,8 @@ import
org.apache.flink.formats.avro.typeutils.AvroSchemaConverter;
import org.apache.flink.formats.avro.utils.AvroTestUtils;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.data.util.DataFormatConverters;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.table.types.AtomicDataType;
@@ -80,6 +82,7 @@ import static org.apache.flink.table.api.DataTypes.SMALLINT;
import static org.apache.flink.table.api.DataTypes.STRING;
import static org.apache.flink.table.api.DataTypes.TIME;
import static org.apache.flink.table.api.DataTypes.TIMESTAMP;
+import static org.apache.flink.table.api.DataTypes.TIMESTAMP_LTZ;
import static org.apache.flink.table.api.DataTypes.TINYINT;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -402,6 +405,43 @@ class AvroRowDataDeSerializationSchemaTest {
.isEqualTo("1970-01-01T00:02:03.456");
}
+ @Test
+ void testTimestampTypeNewMappingInNestedRow() throws Exception {
+ // TIMESTAMP_LTZ is only supported by the non-legacy mapping, so a
nested row holding one
+ // only converts if the flag is propagated into the nested converters
as well.
+ final DataType dataType =
+ ROW(
+ FIELD("id", INT().notNull()),
+ FIELD(
+ "nested",
+ ROW(
+ FIELD("ltz3",
TIMESTAMP_LTZ(3).notNull()),
+ FIELD("name",
STRING().notNull()))
+ .notNull()))
+ .notNull();
+
+ AvroRowDataSerializationSchema serializationSchema =
+ createSerializationSchema(dataType, AvroEncoding.BINARY,
false);
+ AvroRowDataDeserializationSchema deserializationSchema =
+ createDeserializationSchema(dataType, AvroEncoding.BINARY,
false);
+
+ final GenericRowData nested = new GenericRowData(2);
+ nested.setField(0, TimestampData.fromEpochMillis(1589530213123L));
+ nested.setField(1, StringData.fromString("inner"));
+ final GenericRowData row = new GenericRowData(2);
+ row.setField(0, 7);
+ row.setField(1, nested);
+
+ RowData roundTripped =
+
deserializationSchema.deserialize(serializationSchema.serialize(row));
+
+ assertThat(roundTripped.getInt(0)).isEqualTo(7);
+ final RowData actualNested = roundTripped.getRow(1, 2);
+ assertThat(actualNested.getTimestamp(0, 3).toInstant())
+ .isEqualTo(Instant.ofEpochMilli(1589530213123L));
+ assertThat(actualNested.getString(1).toString()).isEqualTo("inner");
+ }
+
private AvroRowDataSerializationSchema createSerializationSchema(
DataType dataType, AvroEncoding encoding, boolean
legacyTimestampMapping)
throws Exception {
