This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 9e33353713 [Fix][Format] Add opt-in nullable Avro ROW layout (#12900)
9e33353713 is described below
commit 9e33353713a8b216130abe89481bcf05ed7c0ae7
Author: hutiefang76 <[email protected]>
AuthorDate: Sat Oct 24 15:46:26 2026 +0000
[Fix][Format] Add opt-in nullable Avro ROW layout (#12900)
---
docs/en/connectors/formats/avro.md | 55 ++++++
docs/en/connectors/sink/Kafka.md | 1 +
docs/en/connectors/sink/Pulsar.md | 1 +
docs/en/connectors/source/Kafka.md | 1 +
docs/en/connectors/source/Pulsar.md | 3 +-
docs/zh/connectors/formats/avro.md | 55 ++++++
docs/zh/connectors/sink/Kafka.md | 1 +
docs/zh/connectors/sink/Pulsar.md | 1 +
docs/zh/connectors/source/Kafka.md | 1 +
docs/zh/connectors/source/Pulsar.md | 3 +-
.../serialize/DefaultSeaTunnelRowSerializer.java | 8 +-
.../seatunnel/kafka/sink/KafkaSinkFactory.java | 4 +-
.../seatunnel/kafka/source/KafkaSourceConfig.java | 9 +-
.../seatunnel/kafka/source/KafkaSourceFactory.java | 2 +
.../DefaultSeaTunnelRowSerializerTest.java | 127 +++++++++++++
.../kafka/source/KafkaSourceConfigTest.java | 111 ++++++++++++
.../seatunnel/pulsar/sink/PulsarSinkFactory.java | 2 +
.../seatunnel/pulsar/sink/PulsarSinkWriter.java | 9 +-
.../seatunnel/pulsar/source/PulsarSource.java | 9 +-
.../pulsar/source/PulsarSourceFactory.java | 4 +-
.../pulsar/sink/PulsarSinkWriterTest.java | 138 ++++++++++++++-
.../seatunnel/pulsar/source/PulsarSourceTest.java | 177 +++++++++++++++++++
.../seatunnel/e2e/connector/kafka/KafkaIT.java | 126 +++++++++++++
.../avro/kafka_avro_nullable_row_to_kafka.conf | 56 ++++++
.../e2e/connector/pulsar/PulsarFormatIT.java | 119 ++++++++++++-
.../avro/pulsar_avro_nullable_row_to_pulsar.conf | 67 +++++++
.../format/avro/AvroDeserializationSchema.java | 28 ++-
.../seatunnel/format/avro/AvroFormatOptions.java | 40 +++++
.../format/avro/AvroSerializationSchema.java | 13 +-
.../seatunnel/format/avro/AvroToRowConverter.java | 35 +++-
.../seatunnel/format/avro/RowToAvroConverter.java | 44 ++++-
.../SeaTunnelRowTypeToAvroSchemaConverter.java | 63 ++++++-
.../seatunnel/format/avro/AvroConverterTest.java | 113 ++++++++++++
.../format/avro/AvroSerializationSchemaTest.java | 196 +++++++++++++++++++++
34 files changed, 1585 insertions(+), 37 deletions(-)
diff --git a/docs/en/connectors/formats/avro.md
b/docs/en/connectors/formats/avro.md
index e80c45f7ad..811eb936e9 100644
--- a/docs/en/connectors/formats/avro.md
+++ b/docs/en/connectors/formats/avro.md
@@ -8,6 +8,61 @@ This format decodes plain Avro binary directly using the
`schema` you configure;
:::
+## Options
+
+| Name | Type | Required | Default | Description
|
+|-------------------|---------|----------|---------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| avro_row_nullable | Boolean | No | false | Whether every ROW field
is written as a nullable Avro union so a `null` ROW value can be serialized.
Supported by the Kafka and Pulsar source and sink connectors. Must be enabled
on the writer and the reader alike. |
+
+### avro_row_nullable
+
+By default a ROW field is a bare Avro record, which has no representation for
`null`, so writing a `null` ROW value fails. Set `avro_row_nullable = true` to
describe every ROW field as the union `[null, record]` instead. The flag
applies at any nesting depth and to a ROW that is nested in another ROW, in an
ARRAY or in a MAP, so a `null` ROW field, a `null` nested ROW field, a `null`
ARRAY element and a `null` MAP value can all be written.
+
+Because the option changes the binary layout of every schema that contains a
ROW field, enable it on the writer **and** on the reader: a job that enables
only one side cannot decode the other side's messages.
+
+Writer job configuration:
+
+```hocon
+sink {
+ Kafka {
+ bootstrap.servers = "localhost:9092"
+ topic = "test_avro_nullable_row"
+ format = avro
+ avro_row_nullable = true
+ }
+}
+
+```
+
+Reader job configuration:
+
+```hocon
+source {
+ Kafka {
+ bootstrap.servers = "localhost:9092"
+ topic = "test_avro_nullable_row"
+ format = avro
+ avro_row_nullable = true
+ schema = {
+ fields {
+ id = bigint
+ c_row = {
+ c_int = int
+ }
+ }
+ }
+ }
+}
+```
+
+A legacy reader path stays open for mixed deployments: when the Kafka source
is given the writer's `avro_schema`, that schema stays authoritative for the
wire layout and the messages are decoded exactly as they were written, whatever
`avro_row_nullable` says. See [Avro
Deserialization](../source/Kafka.md#avro-deserialization).
+
+:::note Raw Avro bytes are not self-describing
+
+SeaTunnel decodes raw Avro bytes with the `schema` you configure; there is no
schema autodetection and no per-message schema lookup, so a reader cannot tell
whether the messages were written with or without `avro_row_nullable`. The
option does not reconcile the two layouts and does not migrate messages that
were already written with the default layout: those still need the default
layout, or their exact writer schema via `avro_schema`, on the reader side.
+
+:::
+
# How To Use
## Kafka uses example
diff --git a/docs/en/connectors/sink/Kafka.md b/docs/en/connectors/sink/Kafka.md
index e2d75ece28..7105fee72d 100644
--- a/docs/en/connectors/sink/Kafka.md
+++ b/docs/en/connectors/sink/Kafka.md
@@ -69,6 +69,7 @@ They can be downloaded via install-plugin.sh or from the
Maven central repositor
| common-options | | No | - | Sink plugin common
parameters, please refer to [Sink Common
Options](../common-options/sink-common-options.md) for details
[...]
| protobuf_message_name | String | No | - | Effective when the
format is set to protobuf, specifies the Message name
[...]
| protobuf_schema | String | No | - | Effective when the
format is set to protobuf, specifies the Schema definition
[...]
+| avro_row_nullable | Boolean | No | false | Effective when the
format is set to avro. Set `true` when a ROW field (including a ROW nested in
another ROW, an ARRAY or a MAP) may be `null`, so the ROW is written as a
nullable Avro union instead of a bare record; the reader must enable the same
option. See [Avro format](../formats/avro.md#avro_row_nullable). |
## Parameter Interpretation
diff --git a/docs/en/connectors/sink/Pulsar.md
b/docs/en/connectors/sink/Pulsar.md
index 56a2e1a32e..c86b6b2f84 100644
--- a/docs/en/connectors/sink/Pulsar.md
+++ b/docs/en/connectors/sink/Pulsar.md
@@ -37,6 +37,7 @@ The Pulsar sink writes SeaTunnel rows to Apache Pulsar
topics. It can write to o
| auth.params | String | No | - |
Parameters for the authentication plugin. Configure it together with
`auth.plugin-class`. |
| format | String | No | json | Data
format. The default format is json. Optional text and avro format.
|
| field_delimiter | String | No | , | Field
delimiter used when `format = "text"`.
|
+| avro_row_nullable | Boolean | No | false |
Effective when `format = "avro"`. Set `true` when a ROW field (including a ROW
nested in another ROW, an ARRAY or a MAP) may be `null`, so the ROW is written
as a nullable Avro union instead of a bare record; the reader must enable the
same option. See [Avro format](../formats/avro.md#avro_row_nullable). |
| semantics | Enum | No | AT_LEAST_ONCE | Write
consistency. Valid values: `NON`, `AT_LEAST_ONCE`, `EXACTLY_ONCE`.
|
| transaction_timeout | Int | No | 600 | Pulsar
transaction timeout in seconds. Used by `EXACTLY_ONCE`.
|
| pulsar.config | Map | No | - | Extra
producer properties passed to the Pulsar producer client.
|
diff --git a/docs/en/connectors/source/Kafka.md
b/docs/en/connectors/source/Kafka.md
index 6157b88774..4d4e1183a2 100644
--- a/docs/en/connectors/source/Kafka.md
+++ b/docs/en/connectors/source/Kafka.md
@@ -71,6 +71,7 @@ They can be downloaded via install-plugin.sh or from the
Maven central repositor
| format | String
| No | json |
Data format. The default format is json. Optional text format, canal_json,
debezium_json, compatible_kafka_connect_json, maxwell_json, ogg_json, avro ,
protobuf and native. If you use json or text format. The default field
separator is ", ". If you customize the delimiter, add the "field_delimiter"
option.If you use canal format, pleas [...]
| avro_schema | String
| No | - |
Effective when `format` is `avro`. Provides the writer Avro schema used to
deserialize binary Avro messages whose record name, namespace, or union layout
differs from the SeaTunnel schema.
[...]
| value_converter_schema_enabled | Boolean
| No | true |
Effective only when `format = compatible_kafka_connect_json`. Whether the Kafka
Connect value converter carries its schema in the JSON payload.
[...]
+| avro_row_nullable | Boolean
| No | false |
Effective when `format` is `avro`. Set `true` when the producer wrote ROW
fields as nullable Avro unions (the writer had `avro_row_nullable = true`), so
a `null` ROW value can be decoded; the reader must match the writer. An
explicit `avro_schema` stays authoritative for the wire layout. See [Avro
format](../formats/avro.md#avro_row_nu [...]
| format_error_handle_way | String
| No | fail | The
processing method of data format error. The default value is fail, and the
optional value is (fail, skip). When fail is selected, data format error will
block and an exception will be thrown. When skip is selected, data format error
will skip this line data.
[...]
| debezium_record_include_schema | Boolean
| No | true |
Effective when `format` is `debezium_json`. Set whether Debezium records
include schema information.
[...]
| debezium_record_table_filter | Config
| No | - |
Used for filtering data in debezium format, only when the format is set to
`debezium_json`. Please refer `debezium_record_table_filter` below
[...]
diff --git a/docs/en/connectors/source/Pulsar.md
b/docs/en/connectors/source/Pulsar.md
index cbcc17af0a..a670d0fdd4 100644
--- a/docs/en/connectors/source/Pulsar.md
+++ b/docs/en/connectors/source/Pulsar.md
@@ -49,6 +49,7 @@ Source connector for Apache Pulsar.
| schema | Config | No | - | Data
structure including field names and types
|
| format | String | No | json | Data format.
Default is json. Supported formats: json, canal_json, avro and text. **Text is
supported only in single-table mode; multi-table mode supports JSON, CANAL_JSON
and AVRO** |
| field_delimiter | String | No | , | Field
delimiter for `text` format.
|
+| avro_row_nullable | Boolean | No | false | Effective
when `format` is `avro`. Set `true` when the producer wrote ROW fields as
nullable Avro unions; Pulsar has no writer-schema option, so the reader schema
is derived from `schema` and must follow the writer layout. See [Avro
format](../formats/avro.md#avro_row_nullable). |
| common-options | | No | - | Source
plugin common parameters. See [Source Common
Options](../common-options/source-common-options.md) for details |
### topic [String]
@@ -169,7 +170,7 @@ reference to
[Schema-Feature](../../introduction/concepts/schema-feature.md)
### format [String]
-Data format. The default format is json. Supported formats are json,
canal_json, avro and text. The `schema` option is required when using avro
format. Text format is supported only in single-table mode. See
[formats](../formats) for more details.
+Data format. The default format is json. Supported formats are json,
canal_json, avro and text. The `schema` option is required when using avro
format. Text format is supported only in single-table mode. See
[formats](../formats) for more details. When the producer writes ROW fields as
nullable Avro unions, set `avro_row_nullable = true` so a `null` ROW value can
be decoded; see [Avro format](../formats/avro.md#avro_row_nullable).
### field_delimiter [String]
diff --git a/docs/zh/connectors/formats/avro.md
b/docs/zh/connectors/formats/avro.md
index 42a35ecd7a..da98b68428 100644
--- a/docs/zh/connectors/formats/avro.md
+++ b/docs/zh/connectors/formats/avro.md
@@ -8,6 +8,61 @@ Avro 在流式数据处理管道中非常流行。现在seatunnel在kafka连接
:::
+## 选项
+
+| 名称 | 类型 | 是否必须 | 默认值 | 描述
|
+|-------------------|---------|----------|--------|-----------------------------------------------------------------------------------------------------------------------------------|
+| avro_row_nullable | Boolean | 否 | false | 是否将所有 ROW 字段写为可空的 Avro
union，以便序列化值为 `null` 的 ROW。Kafka 与 Pulsar 的 source 和 sink 连接器均支持。写入端与读取端必须同时开启。
|
+
+### avro_row_nullable
+
+默认情况下 ROW 字段是一个裸 Avro record，无法表示 `null`，因此写入值为 `null` 的 ROW 会失败。将
`avro_row_nullable` 设为 `true` 后，每个 ROW 字段都会改为 `[null, record]`
union。该开关适用于任意嵌套深度，也适用于嵌套在另一个 ROW、ARRAY 或 MAP 中的 ROW，因此值为 `null` 的 ROW 字段、嵌套
ROW 字段、ARRAY 元素和 MAP value 都可以写入。
+
+由于该选项会改变所有包含 ROW 字段的 schema 的二进制布局，必须在写入端**和**读取端同时开启：只开启一侧的作业无法解码另一侧写出的消息。
+
+写入作业配置：
+
+```hocon
+sink {
+ Kafka {
+ bootstrap.servers = "localhost:9092"
+ topic = "test_avro_nullable_row"
+ format = avro
+ avro_row_nullable = true
+ }
+}
+
+```
+
+读取作业配置：
+
+```hocon
+source {
+ Kafka {
+ bootstrap.servers = "localhost:9092"
+ topic = "test_avro_nullable_row"
+ format = avro
+ avro_row_nullable = true
+ schema = {
+ fields {
+ id = bigint
+ c_row = {
+ c_int = int
+ }
+ }
+ }
+ }
+}
+```
+
+混合部署时仍可使用旧版读取路径：Kafka source 配置了 `avro_schema` 后，该 schema 始终是线上布局的权威描述，无论
`avro_row_nullable` 如何配置，消息都会按写入时的布局解码。参见 [Avro
反序列化](../source/Kafka.md#avro-反序列化)。
+
+:::note 原始 Avro 字节不自描述
+
+SeaTunnel 使用你配置的 `schema` 解码原始 Avro 字节，没有 schema 自动探测，也不会逐条消息查找
schema，因此读取端无法判断消息是用何种 `avro_row_nullable`
配置写入的。该选项既不负责调和两种布局，也不会迁移已经按默认布局写入的历史消息：这些消息在读取端仍需使用默认布局，或通过 `avro_schema`
提供其精确的 writer schema。
+
+:::
+
# 怎样用
## Kafka 使用示例
diff --git a/docs/zh/connectors/sink/Kafka.md b/docs/zh/connectors/sink/Kafka.md
index d61ea10c4b..b86d93638f 100644
--- a/docs/zh/connectors/sink/Kafka.md
+++ b/docs/zh/connectors/sink/Kafka.md
@@ -66,6 +66,7 @@ Dry-run 诊断指出无效选项（包括 `topic` 中引用缺失字段的 `${fi
| common-options | | 否 | - | Sink插件常用参数，请参考 [Sink常用选项
](../common-options/sink-common-options.md) 了解详情
|
|protobuf_message_name|String|否|-| format配置为protobuf时生效，取Message名称
|
|protobuf_schema|String|否|-| format配置为protobuf时生效取Schema名称
|
+| avro_row_nullable | Boolean | 否 | false | 当格式设置为 avro 时生效。当 ROW
字段（含嵌套在另一个 ROW、ARRAY 或 MAP 中的 ROW）可能为 `null` 时设为 `true`，此时 ROW 以可空 Avro union
而非裸 record 写入；读取端必须开启同一选项。详见 [Avro 格式](../formats/avro.md#avro_row_nullable)。 |
## 参数解释
diff --git a/docs/zh/connectors/sink/Pulsar.md
b/docs/zh/connectors/sink/Pulsar.md
index f818dd0993..8077991d88 100644
--- a/docs/zh/connectors/sink/Pulsar.md
+++ b/docs/zh/connectors/sink/Pulsar.md
@@ -37,6 +37,7 @@ Pulsar Sink 用于将 SeaTunnel 数据写入 Apache Pulsar topic。它既可以
| auth.params | String | 否 | - |
认证插件参数。需要和 `auth.plugin-class` 一起配置。 |
| format | String | 否 | json |
数据格式。默认格式为 json。可选 text 和 avro 格式。 |
| field_delimiter | String | 否 | , | 当
`format = "text"` 时使用的字段分隔符。 |
+| avro_row_nullable | Boolean | 否 | false | 当
`format = "avro"` 时生效。当 ROW 字段（含嵌套在另一个 ROW、ARRAY 或 MAP 中的 ROW）可能为 `null` 时设为
`true`，此时 ROW 以可空 Avro union 而非裸 record 写入；读取端必须开启同一选项。详见 [Avro
格式](../formats/avro.md#avro_row_nullable)。 |
| semantics | Enum | 否 | AT_LEAST_ONCE |
写入一致性语义。可选值：`NON`、`AT_LEAST_ONCE`、`EXACTLY_ONCE`。 |
| transaction_timeout | Int | 否 | 600 | Pulsar
事务超时时间，单位为秒。用于 `EXACTLY_ONCE`。 |
| pulsar.config | Map | 否 | - | 传给
Pulsar 生产者客户端的额外参数。 |
diff --git a/docs/zh/connectors/source/Kafka.md
b/docs/zh/connectors/source/Kafka.md
index fec7504b46..baf14efa39 100644
--- a/docs/zh/connectors/source/Kafka.md
+++ b/docs/zh/connectors/source/Kafka.md
@@ -70,6 +70,7 @@ header 字段和事件时间元数据。
| format | String |
否 | json | 数据格式。默认格式为 json。可选格式包括 text, canal_json,
debezium_json, compatible_kafka_connect_json, ogg_json, maxwell_json, avro ,
protobuf和native。默认字段分隔符为 ", "。如果自定义分隔符，添加 "field_delimiter" 选项。如果使用 canal
格式，请参考 [canal-json](../formats/canal-json.md) 了解详细信息。如果使用 debezium 格式，请参考
[debezium-json](../formats/debezium-json.md)。一些Format的详细信息请参考
[formats](../formats) |
| avro_schema | String |
否 | - | 当 `format` 为 `avro` 时生效。用于提供二进制 Avro 消息的
writer schema，适用于消息的 record 名称、namespace 或 union 结构与 SeaTunnel schema 不完全一致的场景。
|
| value_converter_schema_enabled | Boolean |
否 | true | 仅在 `format =
compatible_kafka_connect_json` 时生效。Kafka Connect value 转换后的 JSON 载荷是否携带 schema。
|
+| avro_row_nullable | Boolean |
否 | false | 当 `format` 为 `avro` 时生效。当生产端按可空 Avro
union 写入 ROW 字段（写入端配置了 `avro_row_nullable = true`）时设为 `true`，否则无法解码值为 `null` 的
ROW；读取端必须与写入端保持一致。显式配置的 `avro_schema` 始终是线上布局的权威描述。详见 [Avro
格式](../formats/avro.md#avro_row_nullable)。 使用 `tables_configs` 或 `table_list`
时，请在每个表配置中设置此选项。 |
| format_error_handle_way | String |
否 | fail | 数据格式错误的处理方式。默认值为 fail，可选值为 fail 和
skip。当选择 fail 时，数据格式错误将阻塞并抛出异常。当选择 skip 时，数据格式错误将跳过此行数据。
|
| debezium_record_include_schema | Boolean |
否 | true | 当 `format` 为 `debezium_json` 时生效，用于说明
Debezium 记录中是否携带 schema 信息。
|
| debezium_record_table_filter | Config |
否 | - | 用于过滤 debezium 格式的数据，仅当格式设置为
`debezium_json` 时使用。请参阅下面的 `debezium_record_table_filter`
|
diff --git a/docs/zh/connectors/source/Pulsar.md
b/docs/zh/connectors/source/Pulsar.md
index 1badcd636a..be59880ab3 100644
--- a/docs/zh/connectors/source/Pulsar.md
+++ b/docs/zh/connectors/source/Pulsar.md
@@ -49,6 +49,7 @@ Apache Pulsar 的源连接器。
| schema | Config | 否 | - | 数据结构,包括字段名称和字段类型
|
| format | String | 否 | json | 数据格式。默认为 json。支持
json、canal_json、avro 和 text 格式。**text 仅支持单表模式；多表模式支持 JSON、CANAL_JSON 和 AVRO**
|
| field_delimiter | String | 否 | , | `text` 格式使用的字段分隔符。
|
+| avro_row_nullable | Boolean | 否 | false | 当 `format` 为 `avro`
时生效。当生产端按可空 Avro union 写入 ROW 字段时设为 `true`；Pulsar 没有 writer schema 选项，读取 schema
由 `schema` 推导，必须与写入端布局一致。详见 [Avro 格式](../formats/avro.md#avro_row_nullable)。 |
| common-options | | 否 | - | Source 插件通用参数,请参考
[Source Common Options](../common-options/source-common-options.md) 了解详情
|
### topic [String]
@@ -162,7 +163,7 @@ Pulsar 消费者的启动模式，有效值为 `'EARLIEST'`、`'LATEST'`、`'SUB
### format [String]
-数据格式。默认值为 `json`。支持 json、canal_json、avro 和 text 格式。使用 avro 格式时需要配置
`schema`。text 格式仅支持单表模式。更多格式说明参考 [formats](../formats)。
+数据格式。默认值为 `json`。支持 json、canal_json、avro 和 text 格式。使用 avro 格式时需要配置
`schema`。text 格式仅支持单表模式。更多格式说明参考 [formats](../formats)。当生产端按可空 Avro union 写入
ROW 字段时，需设置 `avro_row_nullable = true` 才能解码值为 `null` 的 ROW，参见 [Avro
格式](../formats/avro.md#avro_row_nullable)。
### field_delimiter [String]
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/serialize/DefaultSeaTunnelRowSerializer.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/serialize/DefaultSeaTunnelRowSerializer.java
index 397437feec..7ebb3ec43b 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/serialize/DefaultSeaTunnelRowSerializer.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/serialize/DefaultSeaTunnelRowSerializer.java
@@ -25,6 +25,7 @@ import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
import org.apache.seatunnel.connectors.seatunnel.kafka.config.MessageFormat;
import
org.apache.seatunnel.connectors.seatunnel.kafka.exception.KafkaConnectorException;
+import org.apache.seatunnel.format.avro.AvroFormatOptions;
import org.apache.seatunnel.format.avro.AvroSerializationSchema;
import
org.apache.seatunnel.format.compatible.debezium.json.CompatibleDebeziumJsonDeserializationSchema;
import
org.apache.seatunnel.format.compatible.debezium.json.CompatibleDebeziumJsonSerializationSchema;
@@ -443,7 +444,12 @@ public class DefaultSeaTunnelRowSerializer implements
SeaTunnelRowSerializer {
case COMPATIBLE_DEBEZIUM_JSON:
return new CompatibleDebeziumJsonSerializationSchema(rowType,
isKey);
case AVRO:
- return new AvroSerializationSchema(rowType);
+ // Opt-in: a null plugin config (legacy create() overloads)
keeps the default
+ // binary layout.
+ boolean avroRowNullable =
+ pluginConfig != null
+ &&
pluginConfig.get(AvroFormatOptions.AVRO_ROW_NULLABLE);
+ return new AvroSerializationSchema(rowType, avroRowNullable);
case PROTOBUF:
String protobufMessageName =
pluginConfig.get(PROTOBUF_MESSAGE_NAME);
String protobufSchema = pluginConfig.get(PROTOBUF_SCHEMA);
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkFactory.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkFactory.java
index 73e9bcba5d..10b38cdb93 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkFactory.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkFactory.java
@@ -24,6 +24,7 @@ import
org.apache.seatunnel.api.table.factory.SupportSinkDryRunValidation;
import org.apache.seatunnel.api.table.factory.TableSinkFactory;
import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
import org.apache.seatunnel.connectors.seatunnel.kafka.config.KafkaSinkOptions;
+import org.apache.seatunnel.format.avro.AvroFormatOptions;
import com.google.auto.service.AutoService;
@@ -45,7 +46,8 @@ public class KafkaSinkFactory implements TableSinkFactory,
SupportSinkDryRunVali
KafkaSinkOptions.TRANSACTION_PREFIX,
KafkaSinkOptions.SEMANTICS,
KafkaSinkOptions.PARTITION,
- KafkaSinkOptions.PARTITION_KEY_FIELDS)
+ KafkaSinkOptions.PARTITION_KEY_FIELDS,
+ AvroFormatOptions.AVRO_ROW_NULLABLE)
.build();
}
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java
index 3964f8a599..1dbe456b04 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java
@@ -44,6 +44,7 @@ import
org.apache.seatunnel.connectors.seatunnel.kafka.config.MessageFormat;
import
org.apache.seatunnel.connectors.seatunnel.kafka.config.MessageFormatErrorHandleWay;
import
org.apache.seatunnel.connectors.seatunnel.kafka.config.TableIdentifierConfig;
import org.apache.seatunnel.format.avro.AvroDeserializationSchema;
+import org.apache.seatunnel.format.avro.AvroFormatOptions;
import
org.apache.seatunnel.format.compatible.kafka.connect.json.CompatibleKafkaConnectDeserializationSchema;
import
org.apache.seatunnel.format.compatible.kafka.connect.json.KafkaConnectJsonFormatOptions;
import
org.apache.seatunnel.format.compatible.kafka.connect.json.NativeKafkaConnectDeserializationSchema;
@@ -431,11 +432,17 @@ public class KafkaSourceConfig implements Serializable {
CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT,
"avro_schema must be configured when
strip_schema_registry_header is enabled for avro format");
}
+ // An explicit avro_schema describes the actual wire
layout and stays
+ // authoritative; the nullable row flag only shapes the
schema derived from
+ // the configured schema block when avro_schema is absent.
+ boolean avroRowNullable =
+
readonlyConfig.get(AvroFormatOptions.AVRO_ROW_NULLABLE);
schema =
new AvroDeserializationSchema(
catalogTable,
avroSchema.orElse(null),
- stripAvroSchemaRegistryHeader);
+ stripAvroSchemaRegistryHeader,
+ avroRowNullable);
break;
case PROTOBUF:
boolean stripSchemaRegistryHeader =
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceFactory.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceFactory.java
index 8c820597a3..445158df0f 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceFactory.java
@@ -33,6 +33,7 @@ import
org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
import
org.apache.seatunnel.connectors.seatunnel.kafka.config.KafkaSourceOptions;
import org.apache.seatunnel.connectors.seatunnel.kafka.config.MessageFormat;
import org.apache.seatunnel.connectors.seatunnel.kafka.config.StartMode;
+import org.apache.seatunnel.format.avro.AvroFormatOptions;
import com.google.auto.service.AutoService;
@@ -65,6 +66,7 @@ public class KafkaSourceFactory implements
TableSourceFactory, SupportSourceDryR
KafkaSourceOptions.SCHEMA,
KafkaSourceOptions.FORMAT,
KafkaSourceOptions.AVRO_SCHEMA,
+ AvroFormatOptions.AVRO_ROW_NULLABLE,
KafkaSourceOptions.DEBEZIUM_RECORD_INCLUDE_SCHEMA,
KafkaSourceOptions.DEBEZIUM_RECORD_TABLE_FILTER,
KafkaSourceOptions.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS,
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/serialize/DefaultSeaTunnelRowSerializerTest.java
b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/serialize/DefaultSeaTunnelRowSerializerTest.java
index 374a18d9e7..344c49ca50 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/serialize/DefaultSeaTunnelRowSerializerTest.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/serialize/DefaultSeaTunnelRowSerializerTest.java
@@ -19,10 +19,16 @@
package org.apache.seatunnel.connectors.seatunnel.kafka.serialize;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
import org.apache.seatunnel.api.table.type.BasicType;
+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.kafka.config.MessageFormat;
+import org.apache.seatunnel.connectors.seatunnel.kafka.sink.KafkaSinkFactory;
+import org.apache.seatunnel.format.avro.AvroDeserializationSchema;
+import org.apache.seatunnel.format.avro.AvroFormatOptions;
+import org.apache.seatunnel.format.avro.AvroSerializationSchema;
import
org.apache.seatunnel.format.compatible.debezium.json.CompatibleDebeziumJsonDeserializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;
@@ -31,6 +37,7 @@ import org.apache.kafka.common.header.Header;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collections;
@@ -382,4 +389,124 @@ public class DefaultSeaTunnelRowSerializerTest {
Assertions.assertFalse(valueString.contains("\"id\""));
Assertions.assertFalse(valueString.contains("\"traceId\""));
}
+
+ // avro_row_nullable is opt-in, so the sink has to keep emitting the
legacy ROW layout when the
+ // option is absent: the bytes stay identical to the explicit false case
and stay readable by a
+ // default layout reader.
+ @Test
+ public void testAvroRowNullableKeepsLegacyLayoutByDefault() throws
IOException {
+ SeaTunnelRowType rowType = avroRowTypeWithNestedRow();
+ SeaTunnelRow row = avroRow(42);
+
+ byte[] flagAbsentBytes =
+ serializeAvroValue(rowType, row,
ReadonlyConfig.fromMap(Collections.emptyMap()));
+ byte[] flagFalseBytes = serializeAvroValue(rowType, row,
avroRowNullableConfig(false));
+ byte[] legacyBytes = new AvroSerializationSchema(rowType,
false).serialize(row);
+
+ Assertions.assertArrayEquals(legacyBytes, flagAbsentBytes);
+ Assertions.assertArrayEquals(legacyBytes, flagFalseBytes);
+
+ SeaTunnelRow readBack = avroDeserializer(rowType,
false).deserialize(flagAbsentBytes);
+ Assertions.assertEquals(1, readBack.getField(0));
+ Assertions.assertEquals(42, ((SeaTunnelRow)
readBack.getField(1)).getField(0));
+ }
+
+ // With the option enabled the sink writes a null ROW value, which the
enabled reader returns as
+ // a null field.
+ @Test
+ public void testAvroRowNullableSerializesNullRowField() throws IOException
{
+ SeaTunnelRowType rowType = avroRowTypeWithNestedRow();
+ SeaTunnelRow row = avroRow(null);
+
+ byte[] bytes = serializeAvroValue(rowType, row,
avroRowNullableConfig(true));
+
+ SeaTunnelRow readBack = avroDeserializer(rowType,
true).deserialize(bytes);
+ Assertions.assertEquals(1, readBack.getField(0));
+ Assertions.assertNull(readBack.getField(1));
+ }
+
+ // The enabled layout adds the union branch index of every ROW value, so a
present ROW produces
+ // different bytes in the two modes while still round tripping to the same
value.
+ @Test
+ public void testAvroRowNullableChangesLayoutOfPresentRowField() throws
IOException {
+ SeaTunnelRowType rowType = avroRowTypeWithNestedRow();
+ SeaTunnelRow row = avroRow(42);
+
+ byte[] legacyBytes =
+ serializeAvroValue(rowType, row,
ReadonlyConfig.fromMap(Collections.emptyMap()));
+ byte[] nullableBytes = serializeAvroValue(rowType, row,
avroRowNullableConfig(true));
+
+ Assertions.assertFalse(Arrays.equals(legacyBytes, nullableBytes));
+
+ SeaTunnelRow readBack = avroDeserializer(rowType,
true).deserialize(nullableBytes);
+ Assertions.assertEquals(42, ((SeaTunnelRow)
readBack.getField(1)).getField(0));
+ }
+
+ // The default layout gives a ROW field no NULL union branch, which is
exactly why a null ROW
+ // value cannot be written without the option. The exception type is
deliberately left open so
+ // the assertion does not depend on Avro internals.
+ @Test
+ public void testAvroNullRowFieldRejectedWithoutOption() {
+ SeaTunnelRowType rowType = avroRowTypeWithNestedRow();
+ SeaTunnelRow row = avroRow(null);
+
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () ->
+ serializeAvroValue(
+ rowType, row,
ReadonlyConfig.fromMap(Collections.emptyMap())));
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () -> serializeAvroValue(rowType, row,
avroRowNullableConfig(false)));
+ }
+
+ // The option has to be declared by the sink factory, otherwise the
unknown option validator
+ // rejects a job that sets avro_row_nullable.
+ @Test
+ public void testKafkaSinkFactoryAcceptsAvroRowNullable() {
+ Assertions.assertTrue(
+ new KafkaSinkFactory()
+ .optionRule()
+ .getOptionalOptions()
+ .contains(AvroFormatOptions.AVRO_ROW_NULLABLE));
+
Assertions.assertFalse(AvroFormatOptions.AVRO_ROW_NULLABLE.defaultValue());
+ }
+
+ private static byte[] serializeAvroValue(
+ SeaTunnelRowType rowType, SeaTunnelRow row, ReadonlyConfig
pluginConfig) {
+ DefaultSeaTunnelRowSerializer serializer =
+ DefaultSeaTunnelRowSerializer.create(
+ "test_topic", rowType, MessageFormat.AVRO, null,
pluginConfig);
+ ProducerRecord<byte[], byte[]> record = serializer.serializeRow(row);
+ return record.value();
+ }
+
+ private static SeaTunnelRowType avroRowTypeWithNestedRow() {
+ SeaTunnelRowType payloadType =
+ new SeaTunnelRowType(
+ new String[] {"InnerID"}, new SeaTunnelDataType<?>[]
{BasicType.INT_TYPE});
+ return new SeaTunnelRowType(
+ new String[] {"id", "payload"},
+ new SeaTunnelDataType<?>[] {BasicType.INT_TYPE, payloadType});
+ }
+
+ private static SeaTunnelRow avroRow(Integer innerId) {
+ SeaTunnelRow payload = innerId == null ? null : new SeaTunnelRow(new
Object[] {innerId});
+ return new SeaTunnelRow(new Object[] {1, payload});
+ }
+
+ private static ReadonlyConfig avroRowNullableConfig(boolean rowNullable) {
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put(AvroFormatOptions.AVRO_ROW_NULLABLE.key(), rowNullable);
+ return ReadonlyConfig.fromMap(configMap);
+ }
+
+ private static AvroDeserializationSchema avroDeserializer(
+ SeaTunnelRowType rowType, boolean rowNullable) {
+ return new AvroDeserializationSchema(
+ CatalogTableUtil.getCatalogTable("", "", "", "test_topic",
rowType),
+ null,
+ false,
+ rowNullable);
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfigTest.java
b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfigTest.java
index e4dd9dc6eb..202dd912a7 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfigTest.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfigTest.java
@@ -24,13 +24,18 @@ import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.type.BasicType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.format.avro.AvroDeserializationSchema;
+import org.apache.seatunnel.format.avro.AvroFormatOptions;
+import org.apache.seatunnel.format.avro.AvroSerializationSchema;
import
org.apache.seatunnel.format.json.debezium.DebeziumJsonDeserializationSchemaDispatcher;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -247,4 +252,110 @@ public class KafkaSourceConfigTest {
Assertions.assertNotNull(sourceConfig.getMapMetadata().get(TablePath.of(null,
table)));
}
+
+ // With the option enabled on the source, a message whose ROW value was
written by an enabled
+ // sink reads back with a null ROW field.
+ @Test
+ void testAvroRowNullableDeserializesNullRowField() throws IOException {
+ KafkaSourceConfig sourceConfig =
+ new
KafkaSourceConfig(ReadonlyConfig.fromMap(avroSourceConfig(true)));
+ ConsumerMetadata metadata =
sourceConfig.getMapMetadata().get(TablePath.of("test"));
+ SeaTunnelRowType rowType =
metadata.getCatalogTable().getSeaTunnelRowType();
+ Assertions.assertEquals(2, rowType.getTotalFields());
+
+ DeserializationSchema<SeaTunnelRow> deserializationSchema =
+ metadata.getDeserializationSchema();
+ // The AVRO schema is always the delegate of the event time wrapper.
+ Assertions.assertInstanceOf(
+ KafkaEventTimeDeserializationSchema.class,
deserializationSchema);
+ Assertions.assertInstanceOf(
+ AvroDeserializationSchema.class,
+ ((KafkaEventTimeDeserializationSchema)
deserializationSchema).getDelegate());
+
+ SeaTunnelRow sourceRow = new SeaTunnelRow(rowType.getTotalFields());
+ sourceRow.setField(rowType.indexOf("id"), 1);
+ sourceRow.setField(rowType.indexOf("payload"), null);
+ byte[] message = new AvroSerializationSchema(rowType,
true).serialize(sourceRow);
+
+ SeaTunnelRow row = deserializationSchema.deserialize(message);
+
+ Assertions.assertEquals(1, row.getField(rowType.indexOf("id")));
+ Assertions.assertNull(row.getField(rowType.indexOf("payload")));
+ }
+
+ // The option is opt-in: when it is absent the source keeps the legacy ROW
layout, so messages
+ // written with the default layout still read back unchanged.
+ @Test
+ void testAvroRowNullableAbsentKeepsLegacyLayout() throws IOException {
+ assertLegacyAvroLayoutRoundTrip(null);
+ }
+
+ @Test
+ void testAvroRowNullableFalseKeepsLegacyLayout() throws IOException {
+ assertLegacyAvroLayoutRoundTrip(false);
+ }
+
+ // The option has to be declared by the source factory, otherwise the
unknown option validator
+ // rejects a job that sets avro_row_nullable.
+ @Test
+ void testKafkaSourceFactoryAcceptsAvroRowNullable() {
+ Assertions.assertTrue(
+ new KafkaSourceFactory()
+ .optionRule()
+ .getOptionalOptions()
+ .contains(AvroFormatOptions.AVRO_ROW_NULLABLE));
+
Assertions.assertFalse(AvroFormatOptions.AVRO_ROW_NULLABLE.defaultValue());
+ }
+
+ /**
+ * Builds an AVRO source config whose schema contains a nested ROW field,
as a nested map is the
+ * idiomatic way to declare a ROW in the schema block.
+ *
+ * @param rowNullable value for the avro_row_nullable option, or null to
leave it unset
+ * @return config map accepted by {@link KafkaSourceConfig}
+ */
+ private static Map<String, Object> avroSourceConfig(Boolean rowNullable) {
+ Map<String, Object> payloadFields = new LinkedHashMap<>();
+ payloadFields.put("InnerID", "int");
+ Map<String, Object> schemaFields = new LinkedHashMap<>();
+ schemaFields.put("id", "int");
+ schemaFields.put("payload", payloadFields);
+ Map<String, Object> schema = new HashMap<>();
+ schema.put("fields", schemaFields);
+
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put("bootstrap.servers", "localhost:9092");
+ configMap.put("group.id", "test");
+ configMap.put("topic", "test");
+ configMap.put("schema", schema);
+ configMap.put("format", "avro");
+ if (rowNullable != null) {
+ configMap.put(AvroFormatOptions.AVRO_ROW_NULLABLE.key(),
rowNullable);
+ }
+ return configMap;
+ }
+
+ /**
+ * Serializes a row with a present ROW value in the legacy layout and
asserts that the source
+ * schema built from the given config reads it back unchanged.
+ *
+ * @param rowNullable value for the avro_row_nullable option, or null to
leave it unset
+ */
+ private static void assertLegacyAvroLayoutRoundTrip(Boolean rowNullable)
throws IOException {
+ KafkaSourceConfig sourceConfig =
+ new
KafkaSourceConfig(ReadonlyConfig.fromMap(avroSourceConfig(rowNullable)));
+ ConsumerMetadata metadata =
sourceConfig.getMapMetadata().get(TablePath.of("test"));
+ SeaTunnelRowType rowType =
metadata.getCatalogTable().getSeaTunnelRowType();
+
+ SeaTunnelRow sourceRow = new SeaTunnelRow(rowType.getTotalFields());
+ sourceRow.setField(rowType.indexOf("id"), 1);
+ sourceRow.setField(rowType.indexOf("payload"), new SeaTunnelRow(new
Object[] {42}));
+ byte[] message = new AvroSerializationSchema(rowType,
false).serialize(sourceRow);
+
+ SeaTunnelRow row =
metadata.getDeserializationSchema().deserialize(message);
+
+ Assertions.assertEquals(1, row.getField(rowType.indexOf("id")));
+ Assertions.assertEquals(
+ 42, ((SeaTunnelRow)
row.getField(rowType.indexOf("payload"))).getField(0));
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkFactory.java
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkFactory.java
index a1fb885db6..e2a204dee7 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkFactory.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkFactory.java
@@ -28,6 +28,7 @@ import
org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
import org.apache.seatunnel.common.exception.CommonErrorCode;
import
org.apache.seatunnel.connectors.seatunnel.pulsar.config.PulsarSinkOptions;
import
org.apache.seatunnel.connectors.seatunnel.pulsar.exception.PulsarConnectorException;
+import org.apache.seatunnel.format.avro.AvroFormatOptions;
import com.google.auto.service.AutoService;
@@ -56,6 +57,7 @@ public class PulsarSinkFactory implements TableSinkFactory {
PulsarSinkOptions.TRANSACTION_TIMEOUT,
PulsarSinkOptions.PULSAR_CONFIG,
PulsarSinkOptions.PARTITION_KEY_FIELDS,
+ AvroFormatOptions.AVRO_ROW_NULLABLE,
SinkConnectorCommonOptions.MULTI_TABLE_SINK_REPLICA)
.conditional(
PulsarSinkOptions.FORMAT,
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java
index 677ea795cb..8c39e35b7b 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java
@@ -37,6 +37,7 @@ import
org.apache.seatunnel.connectors.seatunnel.pulsar.exception.PulsarConnecto
import
org.apache.seatunnel.connectors.seatunnel.pulsar.exception.PulsarConnectorException;
import org.apache.seatunnel.connectors.seatunnel.pulsar.state.PulsarCommitInfo;
import org.apache.seatunnel.connectors.seatunnel.pulsar.state.PulsarSinkState;
+import org.apache.seatunnel.format.avro.AvroFormatOptions;
import org.apache.seatunnel.format.avro.AvroSerializationSchema;
import org.apache.seatunnel.format.json.JsonSerializationSchema;
import org.apache.seatunnel.format.json.exception.SeaTunnelJsonFormatException;
@@ -110,10 +111,12 @@ public class PulsarSinkWriter
ProducerCreator producerCreator) {
String format = pluginConfig.get(PulsarSinkOptions.FORMAT);
String delimiter = pluginConfig.get(PulsarSinkOptions.FIELD_DELIMITER);
+ boolean avroRowNullable =
pluginConfig.get(AvroFormatOptions.AVRO_ROW_NULLABLE);
this.transactionTimeout =
pluginConfig.get(PulsarSinkOptions.TRANSACTION_TIMEOUT);
this.pulsarSemantics = pluginConfig.get(PulsarSinkOptions.SEMANTICS);
this.messageRoutingMode =
pluginConfig.get(PulsarSinkOptions.MESSAGE_ROUTING_MODE);
- this.serializationSchema = createSerializationSchema(seaTunnelRowType,
format, delimiter);
+ this.serializationSchema =
+ createSerializationSchema(seaTunnelRowType, format, delimiter,
avroRowNullable);
List<String> partitionKeyList = getPartitionKeyFields(pluginConfig,
seaTunnelRowType);
this.keySerializationSchema =
createKeySerializationSchema(partitionKeyList,
seaTunnelRowType);
@@ -274,7 +277,7 @@ public class PulsarSinkWriter
}
private SerializationSchema createSerializationSchema(
- SeaTunnelRowType rowType, String format, String delimiter) {
+ SeaTunnelRowType rowType, String format, String delimiter, boolean
avroRowNullable) {
if (PulsarSinkOptions.DEFAULT_FORMAT.equals(format)) {
return new JsonSerializationSchema(rowType);
} else if (PulsarSinkOptions.TEXT_FORMAT.equals(format)) {
@@ -283,7 +286,7 @@ public class PulsarSinkWriter
.delimiter(delimiter)
.build();
} else if (PulsarSinkOptions.AVRO_FORMAT.equals(format)) {
- return new AvroSerializationSchema(rowType);
+ return new AvroSerializationSchema(rowType, avroRowNullable);
} else {
throw new SeaTunnelJsonFormatException(
CommonErrorCode.UNSUPPORTED_DATA_TYPE, "Unsupported
format: " + format);
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSource.java
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSource.java
index 5de647d83d..8bf260621f 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSource.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSource.java
@@ -53,6 +53,7 @@ import
org.apache.seatunnel.connectors.seatunnel.pulsar.source.format.PulsarCana
import
org.apache.seatunnel.connectors.seatunnel.pulsar.source.reader.PulsarSourceReader;
import
org.apache.seatunnel.connectors.seatunnel.pulsar.source.split.PulsarPartitionSplit;
import org.apache.seatunnel.format.avro.AvroDeserializationSchema;
+import org.apache.seatunnel.format.avro.AvroFormatOptions;
import org.apache.seatunnel.format.json.JsonDeserializationSchema;
import org.apache.seatunnel.format.json.canal.CanalJsonDeserializationSchema;
import org.apache.seatunnel.format.json.exception.SeaTunnelJsonFormatException;
@@ -275,7 +276,13 @@ public class PulsarSource
.setIgnoreParseErrors(true)
.build());
case "AVRO":
- return new AvroDeserializationSchema(catalogTable);
+ // Pulsar has no writer schema option, so the reader schema is
always derived from
+ // the row type and must follow the writer's ROW layout.
+ return new AvroDeserializationSchema(
+ catalogTable,
+ null,
+ false,
+
tableConfig.getSchemaConfig().get(AvroFormatOptions.AVRO_ROW_NULLABLE));
case "TEXT":
return TextDeserializationSchema.builder()
.seaTunnelRowType(catalogTable.getSeaTunnelRowType())
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSourceFactory.java
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSourceFactory.java
index a26dba8cd5..3b00623cd1 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSourceFactory.java
@@ -32,6 +32,7 @@ import
org.apache.seatunnel.api.table.factory.TableSourceFactory;
import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
import
org.apache.seatunnel.connectors.seatunnel.pulsar.config.PulsarSourceOptions;
import
org.apache.seatunnel.connectors.seatunnel.pulsar.exception.PulsarConnectorException;
+import org.apache.seatunnel.format.avro.AvroFormatOptions;
import com.google.auto.service.AutoService;
@@ -64,7 +65,8 @@ public class PulsarSourceFactory implements
TableSourceFactory {
PulsarSourceOptions.POLL_INTERVAL,
PulsarSourceOptions.POLL_BATCH_SIZE,
PulsarSourceOptions.FORMAT,
- PulsarSourceOptions.SCHEMA)
+ PulsarSourceOptions.SCHEMA,
+ AvroFormatOptions.AVRO_ROW_NULLABLE)
.exclusive(
PulsarSourceOptions.TOPIC,
PulsarSourceOptions.TOPIC_PATTERN,
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriterTest.java
b/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriterTest.java
index 021836b926..78caaafa06 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriterTest.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriterTest.java
@@ -18,11 +18,18 @@
package org.apache.seatunnel.connectors.seatunnel.pulsar.sink;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.serialization.SerializationSchema;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+import org.apache.seatunnel.api.table.type.BasicType;
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.pulsar.config.PulsarConfigUtil;
import
org.apache.seatunnel.connectors.seatunnel.pulsar.exception.PulsarConnectorException;
+import org.apache.seatunnel.format.avro.AvroDeserializationSchema;
+import org.apache.seatunnel.format.avro.AvroFormatOptions;
+import org.apache.seatunnel.format.avro.AvroSerializationSchema;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
@@ -37,7 +44,9 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -45,8 +54,13 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class PulsarSinkWriterTest {
private PulsarSinkWriter createWriter(ReadonlyConfig config) {
+ return createWriter(
+ new SeaTunnelRowType(new String[] {}, new SeaTunnelDataType[]
{}), config);
+ }
+
+ private PulsarSinkWriter createWriter(SeaTunnelRowType rowType,
ReadonlyConfig config) {
return new PulsarSinkWriter(
- new SeaTunnelRowType(new String[] {}, new SeaTunnelDataType[]
{}),
+ rowType,
config,
java.util.Collections.emptyList(),
null,
@@ -215,6 +229,128 @@ public class PulsarSinkWriterTest {
assertSame(PulsarConfigUtil.class.getClassLoader(),
clientCloseClassLoader.get());
}
+ // avro_row_nullable is a sink-level option: it must reach the Avro format
object that the
+ // writer builds, otherwise a null nested ROW value can never be written.
These tests drive the
+ // real serialization schema obtained from the writer, so they fail if the
option is dropped
+ // anywhere between the plugin config and the format layer.
+ @Test
+ public void testAvroSinkSerializesNullNestedRowWhenOptionEnabled() throws
Exception {
+ SeaTunnelRowType rowType = createRowTypeWithNestedRow();
+ Map<String, Object> configMap = createAvroSinkConfigMap();
+ configMap.put("avro_row_nullable", true);
+ PulsarSinkWriter writer = createWriter(rowType,
ReadonlyConfig.fromMap(configMap));
+
+ SeaTunnelRow sourceRow = new SeaTunnelRow(new Object[] {1, null});
+ byte[] message = getSerializationSchema(writer).serialize(sourceRow);
+
+ SeaTunnelRow deserialized =
+ new AvroDeserializationSchema(createCatalogTable(rowType),
null, false, true)
+ .deserialize(message);
+ assertEquals(1, deserialized.getField(0));
+ assertNull(deserialized.getField(1));
+ }
+
+ @Test
+ public void testAvroSinkEnabledOptionKeepsPresentNestedRowValue() throws
Exception {
+ SeaTunnelRowType rowType = createRowTypeWithNestedRow();
+ Map<String, Object> configMap = createAvroSinkConfigMap();
+ configMap.put("avro_row_nullable", true);
+ PulsarSinkWriter writer = createWriter(rowType,
ReadonlyConfig.fromMap(configMap));
+
+ SeaTunnelRow sourceRow = new SeaTunnelRow(new Object[] {1,
createInnerRow(42)});
+ byte[] message = getSerializationSchema(writer).serialize(sourceRow);
+
+ SeaTunnelRow deserialized =
+ new AvroDeserializationSchema(createCatalogTable(rowType),
null, false, true)
+ .deserialize(message);
+ assertEquals(1, deserialized.getField(0));
+ assertEquals(42, ((SeaTunnelRow)
deserialized.getField(1)).getField(0));
+ }
+
+ @Test
+ public void testAvroSinkDefaultLayoutCannotSerializeNullNestedRow() throws
Exception {
+ SeaTunnelRowType rowType = createRowTypeWithNestedRow();
+ PulsarSinkWriter writer =
+ createWriter(rowType,
ReadonlyConfig.fromMap(createAvroSinkConfigMap()));
+
+ SeaTunnelRow sourceRow = new SeaTunnelRow(new Object[] {1, null});
+
+ assertThrows(
+ RuntimeException.class, () ->
getSerializationSchema(writer).serialize(sourceRow));
+ }
+
+ @Test
+ public void
testAvroSinkAbsentOptionMatchesExplicitFalseForPresentNestedRow() throws
Exception {
+ SeaTunnelRowType rowType = createRowTypeWithNestedRow();
+ SeaTunnelRow sourceRow = new SeaTunnelRow(new Object[] {1,
createInnerRow(42)});
+
+ Map<String, Object> explicitFalseConfig = createAvroSinkConfigMap();
+ explicitFalseConfig.put("avro_row_nullable", false);
+
+ byte[] absentOptionBytes =
+ getSerializationSchema(
+ createWriter(
+ rowType,
ReadonlyConfig.fromMap(createAvroSinkConfigMap())))
+ .serialize(sourceRow);
+ byte[] explicitFalseBytes =
+ getSerializationSchema(
+ createWriter(rowType,
ReadonlyConfig.fromMap(explicitFalseConfig)))
+ .serialize(sourceRow);
+ byte[] formatDefaultBytes = new
AvroSerializationSchema(rowType).serialize(sourceRow);
+
+ assertArrayEquals(formatDefaultBytes, absentOptionBytes);
+ assertArrayEquals(formatDefaultBytes, explicitFalseBytes);
+ }
+
+ @Test
+ public void testSinkFactoryDeclaresAvroRowNullableOption() {
+ assertTrue(
+ new PulsarSinkFactory()
+ .optionRule()
+ .getOptionalOptions()
+ .contains(AvroFormatOptions.AVRO_ROW_NULLABLE));
+ }
+
+ private Map<String, Object> createAvroSinkConfigMap() {
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put("topic", "persistent://tenant/ns/topic-avro");
+ configMap.put("format", "avro");
+ configMap.put("field_delimiter", ",");
+ configMap.put("transaction_timeout", 1000);
+ configMap.put("semantics", "NON");
+ configMap.put("message.routing.mode", "RoundRobinPartition");
+ return configMap;
+ }
+
+ private SeaTunnelRowType createRowTypeWithNestedRow() {
+ return new SeaTunnelRowType(
+ new String[] {"id", "payload"},
+ new SeaTunnelDataType[] {BasicType.INT_TYPE,
createInnerRowType()});
+ }
+
+ private SeaTunnelRowType createInnerRowType() {
+ return new SeaTunnelRowType(
+ new String[] {"InnerID"}, new SeaTunnelDataType[]
{BasicType.INT_TYPE});
+ }
+
+ private SeaTunnelRow createInnerRow(int innerId) {
+ SeaTunnelRow innerRow = new SeaTunnelRow(1);
+ innerRow.setField(0, innerId);
+ return innerRow;
+ }
+
+ private CatalogTable createCatalogTable(SeaTunnelRowType rowType) {
+ return CatalogTableUtil.getCatalogTable("", "", "",
"pulsar_avro_sink", rowType);
+ }
+
+ @SuppressWarnings("unchecked")
+ private SerializationSchema getSerializationSchema(PulsarSinkWriter writer)
+ throws ReflectiveOperationException {
+ Field field =
PulsarSinkWriter.class.getDeclaredField("serializationSchema");
+ field.setAccessible(true);
+ return (SerializationSchema) field.get(writer);
+ }
+
@SuppressWarnings("unchecked")
private static Producer<byte[]> createProducerProxy() {
return createProducerProxy(new AtomicBoolean(false));
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSourceTest.java
b/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSourceTest.java
index c3d5b6fcaf..2f4c468838 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSourceTest.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSourceTest.java
@@ -30,10 +30,15 @@ import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
import org.apache.seatunnel.api.table.type.BasicType;
+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.common.constants.JobMode;
import org.apache.seatunnel.common.utils.SerializationUtils;
import
org.apache.seatunnel.connectors.seatunnel.pulsar.exception.PulsarConnectorException;
+import org.apache.seatunnel.format.avro.AvroDeserializationSchema;
+import org.apache.seatunnel.format.avro.AvroFormatOptions;
+import org.apache.seatunnel.format.avro.AvroSerializationSchema;
import org.apache.seatunnel.format.text.TextDeserializationSchema;
import org.junit.jupiter.api.Assertions;
@@ -44,6 +49,7 @@ import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -223,6 +229,177 @@ class PulsarSourceTest {
Assertions.assertDoesNotThrow(() ->
SerializationUtils.serialize(source));
}
+ // avro_row_nullable is a source-level option: the Pulsar reader has no
writer-schema option, so
+ // the reader schema is always derived from the row type and must follow
the layout the upstream
+ // writer used. The reader is driven through the real deserialization
schema built for the
+ // consumer metadata, so these tests fail if the option is dropped between
config and format.
+ @Test
+ void shouldEnableAvroRowNullableInSingleTableSource() throws Exception {
+ Map<String, Object> config = createAvroSourceConfig("LATEST");
+ config.put("topic", "persistent://public/default/avro-events");
+ config.put("avro_row_nullable", true);
+
+ PulsarSource source =
+ new PulsarSource(ReadonlyConfig.fromMap(config),
createAvroCatalogTable());
+ DeserializationSchema<SeaTunnelRow> deserializationSchema =
+ getDeserializationSchema(source);
+
+ Assertions.assertTrue(deserializationSchema instanceof
AvroDeserializationSchema);
+ SeaTunnelRowType rowType = getProducedRowType(source);
+ byte[] message = new AvroSerializationSchema(rowType,
true).serialize(rowWithoutPayload());
+
+ SeaTunnelRow row = deserializationSchema.deserialize(message);
+
+ Assertions.assertEquals(1, row.getField(0));
+ Assertions.assertNull(row.getField(1));
+ }
+
+ @Test
+ void shouldKeepLegacyAvroRowLayoutByDefaultInSingleTableSource() throws
Exception {
+ Map<String, Object> config = createAvroSourceConfig("LATEST");
+ config.put("topic", "persistent://public/default/avro-events");
+
+ PulsarSource source =
+ new PulsarSource(ReadonlyConfig.fromMap(config),
createAvroCatalogTable());
+ DeserializationSchema<SeaTunnelRow> deserializationSchema =
+ getDeserializationSchema(source);
+
+ SeaTunnelRowType rowType = getProducedRowType(source);
+ byte[] message = new
AvroSerializationSchema(rowType).serialize(rowWithPayload(42));
+
+ SeaTunnelRow row = deserializationSchema.deserialize(message);
+
+ Assertions.assertEquals(1, row.getField(0));
+ Assertions.assertEquals(42, ((SeaTunnelRow)
row.getField(1)).getField(0));
+ }
+
+ @Test
+ void shouldKeepLegacyAvroRowLayoutWhenOptionDisabledInSingleTableSource()
throws Exception {
+ Map<String, Object> config = createAvroSourceConfig("LATEST");
+ config.put("topic", "persistent://public/default/avro-events");
+ config.put("avro_row_nullable", false);
+
+ PulsarSource source =
+ new PulsarSource(ReadonlyConfig.fromMap(config),
createAvroCatalogTable());
+ DeserializationSchema<SeaTunnelRow> deserializationSchema =
+ getDeserializationSchema(source);
+
+ SeaTunnelRowType rowType = getProducedRowType(source);
+ byte[] message = new
AvroSerializationSchema(rowType).serialize(rowWithPayload(42));
+
+ SeaTunnelRow row = deserializationSchema.deserialize(message);
+
+ Assertions.assertEquals(42, ((SeaTunnelRow)
row.getField(1)).getField(0));
+ }
+
+ // The per-table merged config is the value the reader actually sees, so a
per-table flag must
+ // be able to enable the nullable layout even when the global config does
not set it.
+ @Test
+ void shouldEnableAvroRowNullablePerTableInTablesConfigs() throws Exception
{
+ Map<String, Object> config = createAvroSourceConfig("LATEST");
+ Map<String, Object> tableConfig =
+ createTableConfig("db.orders",
"persistent://public/default/orders");
+ tableConfig.put("avro_row_nullable", true);
+ config.put("tables_configs", Arrays.asList(tableConfig));
+
+ PulsarSource source =
+ new PulsarSource(
+ ReadonlyConfig.fromMap(config),
CatalogTableUtil.buildSimpleTextTable());
+ DeserializationSchema<SeaTunnelRow> deserializationSchema =
+ getDeserializationSchema(source);
+
+ SeaTunnelRowType rowType = getProducedRowType(source);
+ byte[] message = new AvroSerializationSchema(rowType,
true).serialize(rowWithoutPayload());
+
+ SeaTunnelRow row = deserializationSchema.deserialize(message);
+
+ Assertions.assertEquals(1, row.getField(0));
+ Assertions.assertNull(row.getField(1));
+ }
+
+ // Precedence: the per-table value overrides the global one, so a
per-table false must keep the
+ // legacy layout even when the global option is enabled.
+ @Test
+ void shouldGivePerTableAvroRowNullablePrecedenceOverGlobalValue() throws
Exception {
+ Map<String, Object> config = createAvroSourceConfig("LATEST");
+ config.put("avro_row_nullable", true);
+ Map<String, Object> tableConfig =
+ createTableConfig("db.orders",
"persistent://public/default/orders");
+ tableConfig.put("avro_row_nullable", false);
+ config.put("tables_configs", Arrays.asList(tableConfig));
+
+ PulsarSource source =
+ new PulsarSource(
+ ReadonlyConfig.fromMap(config),
CatalogTableUtil.buildSimpleTextTable());
+ DeserializationSchema<SeaTunnelRow> deserializationSchema =
+ getDeserializationSchema(source);
+
+ SeaTunnelRowType rowType = getProducedRowType(source);
+ byte[] message = new
AvroSerializationSchema(rowType).serialize(rowWithPayload(42));
+
+ SeaTunnelRow row = deserializationSchema.deserialize(message);
+
+ Assertions.assertEquals(1, row.getField(0));
+ Assertions.assertEquals(42, ((SeaTunnelRow)
row.getField(1)).getField(0));
+ }
+
+ @Test
+ void shouldDeclareAvroRowNullableOptionInSourceFactory() {
+ Assertions.assertTrue(
+ new PulsarSourceFactory()
+ .optionRule()
+ .getOptionalOptions()
+ .contains(AvroFormatOptions.AVRO_ROW_NULLABLE));
+ }
+
+ private Map<String, Object> createAvroSourceConfig(String stopMode) {
+ Map<String, Object> config = createBaseConfig(stopMode);
+ config.put("format", "AVRO");
+ config.put("schema", createNestedRowSchema());
+ return config;
+ }
+
+ /**
+ * Builds the schema block for a row type of {@code {id INT, payload
ROW<InnerID INT>}}. The
+ * nested ROW is expressed as a nested map, which is the idiomatic {@code
schema.fields} form
+ * for a complex type.
+ */
+ private Map<String, Object> createNestedRowSchema() {
+ Map<String, Object> payloadFields = new LinkedHashMap<>();
+ payloadFields.put("InnerID", "INT");
+ Map<String, Object> fields = new LinkedHashMap<>();
+ fields.put("id", "INT");
+ fields.put("payload", payloadFields);
+ Map<String, Object> schema = new HashMap<>();
+ schema.put("fields", fields);
+ return schema;
+ }
+
+ private CatalogTable createAvroCatalogTable() {
+ SeaTunnelRowType innerRowType =
+ new SeaTunnelRowType(
+ new String[] {"InnerID"}, new SeaTunnelDataType[]
{BasicType.INT_TYPE});
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"id", "payload"},
+ new SeaTunnelDataType[] {BasicType.INT_TYPE,
innerRowType});
+ return CatalogTableUtil.getCatalogTable("", "", "",
"pulsar_avro_source", rowType);
+ }
+
+ private SeaTunnelRowType getProducedRowType(PulsarSource source) {
+ return source.getProducedCatalogTables().get(0).getSeaTunnelRowType();
+ }
+
+ private SeaTunnelRow rowWithoutPayload() {
+ return new SeaTunnelRow(new Object[] {1, null});
+ }
+
+ private SeaTunnelRow rowWithPayload(int innerId) {
+ SeaTunnelRow payload = new SeaTunnelRow(1);
+ payload.setField(0, innerId);
+ return new SeaTunnelRow(new Object[] {1, payload});
+ }
+
private Map<String, Object> createBaseConfig(String stopMode) {
Map<String, Object> config = new HashMap<>();
config.put("subscription.name", "seatunnel-sub");
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
index 27b7eb7aa3..3eb38f5482 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
@@ -46,6 +46,7 @@ import
org.apache.seatunnel.e2e.common.container.TestContainer;
import org.apache.seatunnel.e2e.common.container.TestContainerId;
import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
import org.apache.seatunnel.format.avro.AvroDeserializationSchema;
+import org.apache.seatunnel.format.avro.AvroSerializationSchema;
import org.apache.seatunnel.format.protobuf.ProtobufDeserializationSchema;
import org.apache.seatunnel.format.text.TextSerializationSchema;
@@ -102,6 +103,7 @@ import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -1761,6 +1763,130 @@ public class KafkaIT extends AbstractKafkaIT {
});
}
+ /**
+ * Row type of the opt-in nullable ROW round trip: a scalar id, a scalar
string, and a ROW field
+ * that itself holds a nested ROW. The nesting lets the test cover a null
ROW field and a null
+ * ROW two levels deep, both of which the default Avro layout cannot
represent.
+ */
+ private static final SeaTunnelRowType AVRO_NULLABLE_ROW_TYPE =
+ new SeaTunnelRowType(
+ new String[] {"id", "c_string", "c_row"},
+ new SeaTunnelDataType<?>[] {
+ BasicType.LONG_TYPE,
+ BasicType.STRING_TYPE,
+ new SeaTunnelRowType(
+ new String[] {"c_int", "c_sub_row"},
+ new SeaTunnelDataType<?>[] {
+ BasicType.INT_TYPE,
+ new SeaTunnelRowType(
+ new String[] {"c_string"},
+ new SeaTunnelDataType<?>[]
{BasicType.STRING_TYPE})
+ })
+ });
+
+ /**
+ * Round-trips the opt-in nullable ROW layout through the Kafka source and
the Kafka sink.
+ *
+ * <p>The messages on the source topic are written by the test itself with
{@code
+ * avro_row_nullable = true}, one of them with a null ROW field and one
with a null ROW nested
+ * two levels deep. Both connectors have to honour the option for the job
to succeed: the source
+ * cannot decode the nullable layout without it, and the sink cannot
re-encode a null ROW
+ * without it, so the exit code plus the decoded values are the assertion.
+ *
+ * <p>Both topics carry the engine container name because the engine
containers of one run share
+ * the broker: a topic reused across engines would already hold the rows
of the previous engine
+ * when the next one starts.
+ */
+ @TestTemplate
+ @DisabledOnContainer(value = {TestContainerId.SPARK_2_4})
+ public void testKafkaAvroNullableRowRoundTrip(TestContainer container)
+ throws IOException, InterruptedException {
+ String sourceTopic = avroNullableRowTopic(container, "source");
+ String sinkTopic = avroNullableRowTopic(container, "sink");
+ createKafkaTopic(sourceTopic);
+ createKafkaTopic(sinkTopic);
+ produceNullableRowAvroData(sourceTopic);
+
+ Container.ExecResult execResult =
+ container.executeJob(
+ "/avro/kafka_avro_nullable_row_to_kafka.conf",
+ Arrays.asList("source_topic=" + sourceTopic,
"sink_topic=" + sinkTopic));
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+
+ CatalogTable catalogTable =
+ CatalogTableUtil.getCatalogTable("", "", "", "test",
AVRO_NULLABLE_ROW_TYPE);
+ AvroDeserializationSchema avroDeserializationSchema =
+ new AvroDeserializationSchema(catalogTable, null, false, true);
+ List<SeaTunnelRow> rows =
+ getKafkaSTRow(
+ sinkTopic,
+ value -> {
+ try {
+ return
avroDeserializationSchema.deserialize(value);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ Assertions.assertEquals(3, rows.size());
+ // Rows are compared by id: the assertions below must not depend on
delivery order.
+ rows.sort(Comparator.comparingLong(row -> (Long) row.getField(0)));
+
+ SeaTunnelRow rowWithNullRow = rows.get(0);
+ Assertions.assertEquals(1L, rowWithNullRow.getField(0));
+ Assertions.assertEquals("row-1", rowWithNullRow.getField(1));
+ Assertions.assertNull(rowWithNullRow.getField(2));
+
+ SeaTunnelRow rowWithNullNestedRow = rows.get(1);
+ Assertions.assertEquals(2L, rowWithNullNestedRow.getField(0));
+ SeaTunnelRow nestedRow = (SeaTunnelRow)
rowWithNullNestedRow.getField(2);
+ Assertions.assertEquals(2, nestedRow.getField(0));
+ Assertions.assertNull(nestedRow.getField(1));
+
+ SeaTunnelRow rowWithPresentRow = rows.get(2);
+ Assertions.assertEquals(3L, rowWithPresentRow.getField(0));
+ SeaTunnelRow presentRow = (SeaTunnelRow) rowWithPresentRow.getField(2);
+ Assertions.assertEquals(3, presentRow.getField(0));
+ SeaTunnelRow presentNestedRow = (SeaTunnelRow) presentRow.getField(1);
+ Assertions.assertEquals("nested", presentNestedRow.getField(0));
+ }
+
+ private String avroNullableRowTopic(TestContainer container, String role) {
+ return "test_avro_nullable_row_"
+ + role
+ + "_"
+ + container.identifier().name().toLowerCase(Locale.ROOT);
+ }
+
+ private void produceNullableRowAvroData(String sourceTopic) {
+ AvroSerializationSchema serializationSchema =
+ new AvroSerializationSchema(AVRO_NULLABLE_ROW_TYPE, true);
+ List<SeaTunnelRow> rows = new ArrayList<>();
+ rows.add(new SeaTunnelRow(new Object[] {1L, "row-1", null}));
+ rows.add(
+ new SeaTunnelRow(
+ new Object[] {2L, "row-2", new SeaTunnelRow(new
Object[] {2, null})}));
+ rows.add(
+ new SeaTunnelRow(
+ new Object[] {
+ 3L,
+ "row-3",
+ new SeaTunnelRow(
+ new Object[] {3, new SeaTunnelRow(new
Object[] {"nested"})})
+ }));
+ for (SeaTunnelRow row : rows) {
+ try {
+ producer.send(new ProducerRecord<>(sourceTopic,
serializationSchema.serialize(row)))
+ .get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("Interrupted while producing
Avro test data", e);
+ } catch (ExecutionException e) {
+ throw new IllegalStateException("Failed to produce Avro test
data", e);
+ }
+ }
+ producer.flush();
+ }
+
@TestTemplate
@DisabledOnContainer(value = {TestContainerId.SPARK_2_4})
public void testKafkaAvroTombstoneToAssert(TestContainer container)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/avro/kafka_avro_nullable_row_to_kafka.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/avro/kafka_avro_nullable_row_to_kafka.conf
new file mode 100644
index 0000000000..cc793b5f40
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/avro/kafka_avro_nullable_row_to_kafka.conf
@@ -0,0 +1,56 @@
+#
+# 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.
+#
+
+# Both ends of the pipeline enable the nullable ROW layout: the messages on the
+# source topic were written with avro_row_nullable = true, and the sink writes
+# the same layout again so a null ROW survives the round trip.
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Kafka {
+ bootstrap.servers = "kafkaCluster:9092"
+ topic = "${source_topic}"
+ consumer.group = "e2e-avro-nullable-row-group"
+ start_mode = "earliest"
+ format = avro
+ avro_row_nullable = true
+ schema = {
+ fields {
+ id = bigint
+ c_string = string
+ c_row = {
+ c_int = int
+ c_sub_row = {
+ c_string = string
+ }
+ }
+ }
+ }
+ }
+}
+
+sink {
+ Kafka {
+ bootstrap.servers = "kafkaCluster:9092"
+ topic = "${sink_topic}"
+ format = avro
+ avro_row_nullable = true
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/PulsarFormatIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/PulsarFormatIT.java
index b6f2ce92bc..380cbf5527 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/PulsarFormatIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/PulsarFormatIT.java
@@ -59,8 +59,11 @@ import java.time.Duration;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.Comparator;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.TimeUnit;
@@ -75,6 +78,27 @@ public class PulsarFormatIT extends TestSuiteBase implements
TestResource {
private static final String FAKE_SOURCE_TOPIC =
"test_avro_topic_fake_source";
private static final String SOURCE_TOPIC = "test_avro_topic";
+ /**
+ * Row type of the opt-in nullable ROW round trip: a scalar id, a scalar
string, and a ROW field
+ * that itself holds a nested ROW. The nesting lets the test cover a null
ROW field and a null
+ * ROW two levels deep, both of which the default Avro layout cannot
represent.
+ */
+ private static final SeaTunnelRowType NULLABLE_ROW_AVRO_ROW_TYPE =
+ new SeaTunnelRowType(
+ new String[] {"id", "c_string", "c_row"},
+ new SeaTunnelDataType<?>[] {
+ BasicType.LONG_TYPE,
+ BasicType.STRING_TYPE,
+ new SeaTunnelRowType(
+ new String[] {"c_int", "c_sub_row"},
+ new SeaTunnelDataType<?>[] {
+ BasicType.INT_TYPE,
+ new SeaTunnelRowType(
+ new String[] {"c_string"},
+ new SeaTunnelDataType<?>[]
{BasicType.STRING_TYPE})
+ })
+ });
+
private static final SeaTunnelRowType FAKE_SOURCE_ROW_TYPE =
new SeaTunnelRowType(
new String[] {
@@ -227,6 +251,94 @@ public class PulsarFormatIT extends TestSuiteBase
implements TestResource {
Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
}
+ /**
+ * Round-trips the opt-in nullable ROW layout through the Pulsar source
and the Pulsar sink.
+ *
+ * <p>The messages on the source topic are written by the test itself with
{@code
+ * avro_row_nullable = true}, one of them with a null ROW field and one
with a null ROW nested
+ * two levels deep. Both connectors have to honour the option for the job
to succeed: the source
+ * derives its reader schema from the configured {@code schema} block, so
it cannot decode the
+ * nullable layout without the option, and the sink cannot re-encode a
null ROW without it.
+ *
+ * <p>Both topics carry the engine container name because the engine
containers of one run share
+ * the Pulsar cluster: a topic reused across engines would already hold
the rows of the previous
+ * engine when the next one starts.
+ */
+ @TestTemplate
+ @DisabledOnContainer(value = {TestContainerId.SPARK_2_4})
+ public void testPulsarAvroNullableRowRoundTrip(TestContainer container)
throws Exception {
+ String sourceTopic = nullableRowTopic(container, "source");
+ String sinkTopic = nullableRowTopic(container, "sink");
+ produceNullableRowAvroTestData(sourceTopic);
+ Container.ExecResult execResult =
+ container.executeJob(
+ "/avro/pulsar_avro_nullable_row_to_pulsar.conf",
+ Arrays.asList("source_topic=" + sourceTopic,
"sink_topic=" + sinkTopic));
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+
+ CatalogTable catalogTable =
+ CatalogTableUtil.getCatalogTable("", "", "", "test",
NULLABLE_ROW_AVRO_ROW_TYPE);
+ AvroDeserializationSchema avroDeserializationSchema =
+ new AvroDeserializationSchema(catalogTable, null, false, true);
+ List<SeaTunnelRow> rows = consumeAvroRows(sinkTopic,
avroDeserializationSchema, 3);
+ Assertions.assertEquals(3, rows.size());
+ // Rows are compared by id: the assertions below must not depend on
delivery order.
+ rows.sort(Comparator.comparingLong(row -> (Long) row.getField(0)));
+
+ SeaTunnelRow rowWithNullRow = rows.get(0);
+ Assertions.assertEquals(1L, rowWithNullRow.getField(0));
+ Assertions.assertEquals("row-1", rowWithNullRow.getField(1));
+ Assertions.assertNull(rowWithNullRow.getField(2));
+
+ SeaTunnelRow rowWithNullNestedRow = rows.get(1);
+ Assertions.assertEquals(2L, rowWithNullNestedRow.getField(0));
+ SeaTunnelRow nestedRow = (SeaTunnelRow)
rowWithNullNestedRow.getField(2);
+ Assertions.assertEquals(2, nestedRow.getField(0));
+ Assertions.assertNull(nestedRow.getField(1));
+
+ SeaTunnelRow rowWithPresentRow = rows.get(2);
+ Assertions.assertEquals(3L, rowWithPresentRow.getField(0));
+ SeaTunnelRow presentRow = (SeaTunnelRow) rowWithPresentRow.getField(2);
+ Assertions.assertEquals(3, presentRow.getField(0));
+ SeaTunnelRow presentNestedRow = (SeaTunnelRow) presentRow.getField(1);
+ Assertions.assertEquals("nested", presentNestedRow.getField(0));
+ }
+
+ private String nullableRowTopic(TestContainer container, String role) {
+ return "test_avro_nullable_row_"
+ + role
+ + "_"
+ + container.identifier().name().toLowerCase(Locale.ROOT);
+ }
+
+ private void produceNullableRowAvroTestData(String sourceTopic) throws
Exception {
+ AvroSerializationSchema serializationSchema =
+ new AvroSerializationSchema(NULLABLE_ROW_AVRO_ROW_TYPE, true);
+ try (Producer<byte[]> producer =
+ client.newProducer(Schema.BYTES).topic(sourceTopic).create()) {
+ producer.send(
+ serializationSchema.serialize(
+ new SeaTunnelRow(new Object[] {1L, "row-1",
null})));
+ producer.send(
+ serializationSchema.serialize(
+ new SeaTunnelRow(
+ new Object[] {
+ 2L, "row-2", new SeaTunnelRow(new
Object[] {2, null})
+ })));
+ producer.send(
+ serializationSchema.serialize(
+ new SeaTunnelRow(
+ new Object[] {
+ 3L,
+ "row-3",
+ new SeaTunnelRow(
+ new Object[] {
+ 3, new SeaTunnelRow(new
Object[] {"nested"})
+ })
+ })));
+ }
+ }
+
private void produceAvroTestData(int start, int end) throws Exception {
AvroSerializationSchema serializationSchema =
new AvroSerializationSchema(SEATUNNEL_ROW_TYPE);
@@ -259,6 +371,11 @@ public class PulsarFormatIT extends TestSuiteBase
implements TestResource {
private List<SeaTunnelRow> consumeAvroRows(
String topic, AvroDeserializationSchema deserializationSchema) {
+ return consumeAvroRows(topic, deserializationSchema, 90);
+ }
+
+ private List<SeaTunnelRow> consumeAvroRows(
+ String topic, AvroDeserializationSchema deserializationSchema, int
expectedRows) {
List<SeaTunnelRow> rows = new ArrayList<>();
try (Consumer<byte[]> consumer =
client.newConsumer(Schema.BYTES)
@@ -267,7 +384,7 @@ public class PulsarFormatIT extends TestSuiteBase
implements TestResource {
.subscriptionType(SubscriptionType.Exclusive)
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
.subscribe()) {
- for (int i = 0; i < 90; i++) {
+ for (int i = 0; i < expectedRows; i++) {
Message<byte[]> message = consumer.receive(30,
TimeUnit.SECONDS);
if (message == null) {
break;
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/resources/avro/pulsar_avro_nullable_row_to_pulsar.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/resources/avro/pulsar_avro_nullable_row_to_pulsar.conf
new file mode 100644
index 0000000000..8557c68745
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/resources/avro/pulsar_avro_nullable_row_to_pulsar.conf
@@ -0,0 +1,67 @@
+#
+# 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.
+#
+
+# Both ends of the pipeline enable the nullable ROW layout: the messages on the
+# source topic were written with avro_row_nullable = true, and the sink writes
+# the same layout again so a null ROW survives the round trip. Pulsar has no
+# writer-schema option, so the reader schema is derived from the configured
+# schema block and has to follow the writer layout.
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Pulsar {
+ client.service-url = "pulsar://pulsar.e2e.format:6650"
+ admin.service-url = "http://pulsar.e2e.format:8080";
+ cursor.startup.mode = "EARLIEST"
+ cursor.stop.mode = "LATEST"
+ # Pulsar sinks route by row tableId, so map this input to the target topic.
+ tables_configs = [
+ {
+ topic = "${source_topic}"
+ table_path = "${sink_topic}"
+ subscription.name = "e2e-avro-nullable-row"
+ format = avro
+ avro_row_nullable = true
+ schema = {
+ fields {
+ id = bigint
+ c_string = string
+ c_row = {
+ c_int = int
+ c_sub_row = {
+ c_string = string
+ }
+ }
+ }
+ }
+ }
+ ]
+ }
+}
+
+sink {
+ Pulsar {
+ topic = "${sink_topic}"
+ client.service-url = "pulsar://pulsar.e2e.format:6650"
+ admin.service-url = "http://pulsar.e2e.format:8080";
+ format = avro
+ avro_row_nullable = true
+ }
+}
diff --git
a/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroDeserializationSchema.java
b/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroDeserializationSchema.java
index b34663259f..e97695ef49 100644
---
a/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroDeserializationSchema.java
+++
b/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroDeserializationSchema.java
@@ -50,12 +50,32 @@ public class AvroDeserializationSchema implements
DeserializationSchema<SeaTunne
public AvroDeserializationSchema(
CatalogTable catalogTable, String writerSchema, boolean
stripSchemaRegistryHeader) {
+ this(catalogTable, writerSchema, stripSchemaRegistryHeader, false);
+ }
+
+ /**
+ * Creates a deserialization schema with the opt-in nullable ROW layout.
+ *
+ * <p>When {@code writerSchema} is supplied it stays authoritative for the
wire layout and
+ * {@code rowNullable} has no effect, because the reader schema must match
the schema the
+ * message was written with. Otherwise the reader schema is derived from
the row type, and
+ * {@code rowNullable} must match the writer side for ROW fields to be
read back correctly.
+ *
+ * @param catalogTable catalog table of the messages
+ * @param writerSchema schema the message was written with, or null to
derive it from the row
+ * type
+ * @param stripSchemaRegistryHeader whether a Confluent Schema Registry
header is stripped
+ * @param rowNullable whether a ROW value was written as a nullable Avro
union, see {@link
+ * AvroFormatOptions#AVRO_ROW_NULLABLE}
+ */
+ public AvroDeserializationSchema(
+ CatalogTable catalogTable,
+ String writerSchema,
+ boolean stripSchemaRegistryHeader,
+ boolean rowNullable) {
this.catalogTable = catalogTable;
this.rowType = catalogTable.getSeaTunnelRowType();
- this.converter =
- writerSchema == null
- ? new AvroToRowConverter(rowType)
- : new AvroToRowConverter(rowType, writerSchema);
+ this.converter = new AvroToRowConverter(rowType, writerSchema,
rowNullable);
this.stripSchemaRegistryHeader = stripSchemaRegistryHeader;
}
diff --git
a/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroFormatOptions.java
b/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroFormatOptions.java
new file mode 100644
index 0000000000..5ec3bd7752
--- /dev/null
+++
b/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroFormatOptions.java
@@ -0,0 +1,40 @@
+/*
+ * 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.format.avro;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+
+/** Options that control how the Avro format serializes and deserializes
SeaTunnel rows. */
+public class AvroFormatOptions {
+
+ public static final Option<Boolean> AVRO_ROW_NULLABLE =
+ Options.key("avro_row_nullable")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether a ROW field, including ROW values nested
in ARRAY and MAP, is "
+ + "written as a nullable Avro union of
[null, record] instead of "
+ + "a bare Avro record. Enabling it lets a
null ROW value be "
+ + "serialized. The option is disabled by
default because it "
+ + "changes the binary layout of every
schema that contains a ROW "
+ + "field, so the writer and the reader
must be enabled and "
+ + "disabled together; messages written
with the default layout "
+ + "cannot be read with the option enabled
unless the original "
+ + "writer schema is supplied explicitly.");
+}
diff --git
a/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroSerializationSchema.java
b/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroSerializationSchema.java
index 16afe4b693..020a6be9b8 100644
---
a/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroSerializationSchema.java
+++
b/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroSerializationSchema.java
@@ -42,9 +42,20 @@ public class AvroSerializationSchema implements
SerializationSchema {
private final DatumWriter<GenericRecord> writer;
public AvroSerializationSchema(SeaTunnelRowType rowType) {
+ this(rowType, false);
+ }
+
+ /**
+ * Creates a serialization schema with the opt-in nullable ROW layout.
+ *
+ * @param rowType row type to serialize
+ * @param rowNullable whether a ROW value is written as a nullable Avro
union, see {@link
+ * AvroFormatOptions#AVRO_ROW_NULLABLE}
+ */
+ public AvroSerializationSchema(SeaTunnelRowType rowType, boolean
rowNullable) {
this.out = new ByteArrayOutputStream();
this.encoder = EncoderFactory.get().binaryEncoder(out, null);
- this.converter = new RowToAvroConverter(rowType);
+ this.converter = new RowToAvroConverter(rowType, rowNullable);
this.writer = this.converter.getWriter();
}
diff --git
a/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroToRowConverter.java
b/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroToRowConverter.java
index af7912d702..ff746d0486 100644
---
a/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroToRowConverter.java
+++
b/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroToRowConverter.java
@@ -48,13 +48,44 @@ public class AvroToRowConverter implements Serializable {
private Schema schema;
public AvroToRowConverter(SeaTunnelRowType rowType) {
- schema =
SeaTunnelRowTypeToAvroSchemaConverter.buildAvroSchemaWithRowType(rowType);
+ this(rowType, false);
+ }
+
+ /**
+ * Creates a converter that derives the reader schema from the row type.
+ *
+ * @param rowType row type to read into
+ * @param rowNullable whether a ROW value was written as a nullable Avro
union, see {@link
+ * AvroFormatOptions#AVRO_ROW_NULLABLE}
+ */
+ public AvroToRowConverter(SeaTunnelRowType rowType, boolean rowNullable) {
+ schema =
+
SeaTunnelRowTypeToAvroSchemaConverter.buildAvroSchemaWithRowType(
+ rowType, rowNullable);
}
public AvroToRowConverter(SeaTunnelRowType rowType, String writerSchema) {
+ this(rowType, writerSchema, false);
+ }
+
+ /**
+ * Creates a converter for a message written with the given schema.
+ *
+ * <p>A supplied writer schema is authoritative: it describes the actual
wire layout, so {@code
+ * rowNullable} only shapes the schema derived from the row type when no
writer schema is
+ * supplied.
+ *
+ * @param rowType row type to read into
+ * @param writerSchema schema the message was written with, or null to
derive it from the row
+ * type
+ * @param rowNullable whether a ROW value is written as a nullable Avro
union, see {@link
+ * AvroFormatOptions#AVRO_ROW_NULLABLE}
+ */
+ public AvroToRowConverter(SeaTunnelRowType rowType, String writerSchema,
boolean rowNullable) {
schema =
writerSchema == null
- ?
SeaTunnelRowTypeToAvroSchemaConverter.buildAvroSchemaWithRowType(rowType)
+ ?
SeaTunnelRowTypeToAvroSchemaConverter.buildAvroSchemaWithRowType(
+ rowType, rowNullable)
: new Schema.Parser().parse(writerSchema);
}
diff --git
a/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/RowToAvroConverter.java
b/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/RowToAvroConverter.java
index 967d66b454..9f958d224f 100644
---
a/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/RowToAvroConverter.java
+++
b/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/RowToAvroConverter.java
@@ -48,9 +48,25 @@ public class RowToAvroConverter implements Serializable {
private final Schema schema;
private final SeaTunnelRowType rowType;
private final DatumWriter<GenericRecord> writer;
+ private final boolean rowNullable;
public RowToAvroConverter(SeaTunnelRowType rowType) {
- this.schema =
SeaTunnelRowTypeToAvroSchemaConverter.buildAvroSchemaWithRowType(rowType);
+ this(rowType, false);
+ }
+
+ /**
+ * Creates a converter that serializes rows with either the legacy ROW
layout or the opt-in
+ * nullable ROW layout.
+ *
+ * @param rowType row type to serialize
+ * @param rowNullable whether a ROW value is written as a nullable Avro
union, see {@link
+ * AvroFormatOptions#AVRO_ROW_NULLABLE}
+ */
+ public RowToAvroConverter(SeaTunnelRowType rowType, boolean rowNullable) {
+ this.rowNullable = rowNullable;
+ this.schema =
+
SeaTunnelRowTypeToAvroSchemaConverter.buildAvroSchemaWithRowType(
+ rowType, rowNullable);
this.rowType = rowType;
this.writer = createWriter();
}
@@ -79,12 +95,13 @@ public class RowToAvroConverter implements Serializable {
for (int i = 0; i < fieldNames.length; i++) {
String fieldName = rowType.getFieldName(i);
Object value = element.getField(i);
- builder.set(fieldName, resolveObject(value,
rowType.getFieldType(i)));
+ builder.set(fieldName, resolveObject(value,
rowType.getFieldType(i), fieldName));
}
return builder.build();
}
- private Object resolveObject(Object data, SeaTunnelDataType<?>
seaTunnelDataType) {
+ private Object resolveObject(
+ Object data, SeaTunnelDataType<?> seaTunnelDataType, String
recordName) {
if (data == null) {
return null;
}
@@ -121,7 +138,7 @@ public class RowToAvroConverter implements Serializable {
int length = Array.getLength(data);
ArrayList<Object> records = new ArrayList<>(length);
for (int i = 0; i < length; i++) {
- records.add(resolveObject(Array.get(data, i), basicType));
+ records.add(resolveObject(Array.get(data, i), basicType,
recordName));
}
return records;
case MAP:
@@ -131,8 +148,8 @@ public class RowToAvroConverter implements Serializable {
Map<Object, Object> mapData = new HashMap<>();
for (Map.Entry<?, ?> entry : ((Map<Object, Object>)
data).entrySet()) {
mapData.put(
- resolveObject(entry.getKey(), keyType),
- resolveObject(entry.getValue(), valueType));
+ resolveObject(entry.getKey(), keyType, recordName),
+ resolveObject(entry.getValue(), valueType,
recordName));
}
return mapData;
@@ -141,13 +158,22 @@ public class RowToAvroConverter implements Serializable {
SeaTunnelDataType<?>[] fieldTypes =
((SeaTunnelRowType) seaTunnelDataType).getFieldTypes();
String[] fieldNames = ((SeaTunnelRowType)
seaTunnelDataType).getFieldNames();
+ // With the opt-in nullable layout the nested record must
reuse the name the
+ // enclosing schema gave the ROW, otherwise a present ROW
cannot be resolved
+ // against its [null, record] union branch. The legacy layout
keeps building the
+ // nested record exactly as before, so the default path yields
an unchanged
+ // GenericRecord shape (the historic top level record name)
and unchanged bytes.
Schema recordSchema =
-
SeaTunnelRowTypeToAvroSchemaConverter.buildAvroSchemaWithRowType(
- (SeaTunnelRowType) seaTunnelDataType);
+ rowNullable
+ ?
SeaTunnelRowTypeToAvroSchemaConverter.buildRecordSchema(
+ recordName, (SeaTunnelRowType)
seaTunnelDataType, true)
+ :
SeaTunnelRowTypeToAvroSchemaConverter.buildAvroSchemaWithRowType(
+ (SeaTunnelRowType) seaTunnelDataType);
GenericRecordBuilder recordBuilder = new
GenericRecordBuilder(recordSchema);
for (int i = 0; i < fieldNames.length; i++) {
recordBuilder.set(
- fieldNames[i],
resolveObject(seaTunnelRow.getField(i), fieldTypes[i]));
+ fieldNames[i],
+ resolveObject(seaTunnelRow.getField(i),
fieldTypes[i], fieldNames[i]));
}
return recordBuilder.build();
default:
diff --git
a/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/SeaTunnelRowTypeToAvroSchemaConverter.java
b/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/SeaTunnelRowTypeToAvroSchemaConverter.java
index 56a65d3e53..0955d56b09 100644
---
a/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/SeaTunnelRowTypeToAvroSchemaConverter.java
+++
b/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/SeaTunnelRowTypeToAvroSchemaConverter.java
@@ -34,27 +34,66 @@ import java.util.List;
public class SeaTunnelRowTypeToAvroSchemaConverter {
+ /**
+ * Builds the Avro record that describes the given row type with the
legacy layout, where a ROW
+ * field is a bare Avro record and therefore cannot hold a null value.
+ *
+ * @param seaTunnelRowType row type to describe
+ * @return the Avro record schema
+ */
public static Schema buildAvroSchemaWithRowType(SeaTunnelRowType
seaTunnelRowType) {
+ return buildAvroSchemaWithRowType(seaTunnelRowType, false);
+ }
+
+ /**
+ * Builds the Avro record that describes the given row type.
+ *
+ * <p>The {@code rowNullable} flag decides how every ROW field is
described, at any nesting
+ * depth and no matter whether the ROW is a field, an ARRAY element or a
MAP value. With {@code
+ * false} a ROW keeps the legacy layout of a bare Avro record; with {@code
true} each ROW is
+ * wrapped in the union {@code [null, record]} so that a null ROW value
can be written. The flag
+ * is applied recursively, so a ROW nested several levels deep is
described the same way as a
+ * top level one.
+ *
+ * @param seaTunnelRowType row type to describe
+ * @param rowNullable whether ROW fields are described as nullable unions
+ * @return the Avro record schema
+ */
+ public static Schema buildAvroSchemaWithRowType(
+ SeaTunnelRowType seaTunnelRowType, boolean rowNullable) {
+ return buildRecordSchema("SeaTunnelRecord", seaTunnelRowType,
rowNullable);
+ }
+
+ /**
+ * Builds the Avro record that describes the given row type under an
explicit record name.
+ *
+ * <p>The name is the one the enclosing schema assigned to the ROW: the
field name for a ROW
+ * field, and the enclosing field name for a ROW nested in an ARRAY or a
MAP. A serialized
+ * record must be built with the same name, otherwise Avro cannot resolve
it against the union
+ * branch of a nullable ROW.
+ */
+ static Schema buildRecordSchema(
+ String recordName, SeaTunnelRowType seaTunnelRowType, boolean
rowNullable) {
List<Schema.Field> fields = new ArrayList<>();
SeaTunnelDataType<?>[] fieldTypes = seaTunnelRowType.getFieldTypes();
String[] fieldNames = seaTunnelRowType.getFieldNames();
for (int i = 0; i < fieldNames.length; i++) {
- fields.add(generateField(fieldNames[i], fieldTypes[i]));
+ fields.add(generateField(fieldNames[i], fieldTypes[i],
rowNullable));
}
- return Schema.createRecord("SeaTunnelRecord", null, null, false,
fields);
+ return Schema.createRecord(recordName, null, null, false, fields);
}
private static Schema.Field generateField(
- String fieldName, SeaTunnelDataType<?> seaTunnelDataType) {
+ String fieldName, SeaTunnelDataType<?> seaTunnelDataType, boolean
rowNullable) {
return new Schema.Field(
fieldName,
- seaTunnelDataType2AvroDataType(fieldName, seaTunnelDataType),
+ seaTunnelDataType2AvroDataType(fieldName, seaTunnelDataType,
rowNullable),
null,
null);
}
private static Schema seaTunnelDataType2AvroDataType(
- String fieldName, SeaTunnelDataType<?> seaTunnelDataType) {
+ String fieldName, SeaTunnelDataType<?> seaTunnelDataType, boolean
rowNullable) {
switch (seaTunnelDataType.getSqlType()) {
case STRING:
@@ -84,22 +123,28 @@ public class SeaTunnelRowTypeToAvroSchemaConverter {
SeaTunnelDataType<?> valueType = ((MapType<?, ?>)
seaTunnelDataType).getValueType();
return Schema.createUnion(
Schema.create(Schema.Type.NULL),
-
Schema.createMap(seaTunnelDataType2AvroDataType(fieldName, valueType)));
+ Schema.createMap(
+ seaTunnelDataType2AvroDataType(fieldName,
valueType, rowNullable)));
case ARRAY:
SeaTunnelDataType<?> elementType =
((ArrayType<?, ?>) seaTunnelDataType).getElementType();
return Schema.createUnion(
Schema.create(Schema.Type.NULL),
-
Schema.createArray(seaTunnelDataType2AvroDataType(fieldName, elementType)));
+ Schema.createArray(
+ seaTunnelDataType2AvroDataType(
+ fieldName, elementType, rowNullable)));
case ROW:
SeaTunnelDataType<?>[] fieldTypes =
((SeaTunnelRowType) seaTunnelDataType).getFieldTypes();
String[] fieldNames = ((SeaTunnelRowType)
seaTunnelDataType).getFieldNames();
List<Schema.Field> subField = new ArrayList<>();
for (int i = 0; i < fieldNames.length; i++) {
- subField.add(generateField(fieldNames[i], fieldTypes[i]));
+ subField.add(generateField(fieldNames[i], fieldTypes[i],
rowNullable));
}
- return Schema.createRecord(fieldName, null, null, false,
subField);
+ Schema recordSchema = Schema.createRecord(fieldName, null,
null, false, subField);
+ return rowNullable
+ ? Schema.createUnion(Schema.create(Schema.Type.NULL),
recordSchema)
+ : recordSchema;
case DECIMAL:
int precision = ((DecimalType)
seaTunnelDataType).getPrecision();
int scale = ((DecimalType) seaTunnelDataType).getScale();
diff --git
a/seatunnel-formats/seatunnel-format-avro/src/test/java/org/apache/seatunnel/format/avro/AvroConverterTest.java
b/seatunnel-formats/seatunnel-format-avro/src/test/java/org/apache/seatunnel/format/avro/AvroConverterTest.java
index 60f7281331..32e35d756c 100644
---
a/seatunnel-formats/seatunnel-format-avro/src/test/java/org/apache/seatunnel/format/avro/AvroConverterTest.java
+++
b/seatunnel-formats/seatunnel-format-avro/src/test/java/org/apache/seatunnel/format/avro/AvroConverterTest.java
@@ -17,6 +17,8 @@
package org.apache.seatunnel.format.avro;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
import org.apache.seatunnel.api.table.type.ArrayType;
import org.apache.seatunnel.api.table.type.BasicType;
import org.apache.seatunnel.api.table.type.DecimalType;
@@ -27,10 +29,16 @@ import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.EncoderFactory;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.LocalDateTime;
@@ -174,4 +182,109 @@ class AvroConverterTest {
Assertions.assertEquals(converterRow, seaTunnelRow);
}
+
+ // The default mode is the legacy mode, so the GenericRecord shape built
for a nested ROW must
+ // stay exactly the shape that existed before the nullable ROW option was
introduced: the nested
+ // record still carries the historic top level record name and the same
values. The explicit
+ // false mode is the same public mode, so it has to produce that identical
shape as well.
+ @Test
+ public void testLegacyNestedRecordShapeAndValuesUnchanged() {
+ SeaTunnelRowType rowType = buildSeaTunnelRowType();
+ SeaTunnelRow seaTunnelRow = buildSeaTunnelRow();
+ SeaTunnelRowType nestedRowType = (SeaTunnelRowType)
rowType.getFieldType(14);
+
+ GenericRecord defaultRecord =
+ new
RowToAvroConverter(rowType).convertRowToGenericRecord(seaTunnelRow);
+ GenericRecord explicitFalseRecord =
+ new RowToAvroConverter(rowType,
false).convertRowToGenericRecord(seaTunnelRow);
+
+ Assertions.assertEquals(
+
SeaTunnelRowTypeToAvroSchemaConverter.buildAvroSchemaWithRowType(rowType)
+ .toString(),
+ defaultRecord.getSchema().toString());
+
+ GenericRecord defaultNested = (GenericRecord)
defaultRecord.get("c_row");
+ GenericRecord explicitFalseNested = (GenericRecord)
explicitFalseRecord.get("c_row");
+ String legacyNestedSchema =
+
SeaTunnelRowTypeToAvroSchemaConverter.buildAvroSchemaWithRowType(nestedRowType)
+ .toString();
+ Assertions.assertEquals("SeaTunnelRecord",
defaultNested.getSchema().getName());
+ Assertions.assertEquals(legacyNestedSchema,
defaultNested.getSchema().toString());
+ Assertions.assertEquals("SeaTunnelRecord",
explicitFalseNested.getSchema().getName());
+ Assertions.assertEquals(legacyNestedSchema,
explicitFalseNested.getSchema().toString());
+ Assertions.assertEquals(defaultNested.toString(),
explicitFalseNested.toString());
+ Assertions.assertEquals("strVal", defaultNested.get("c_string"));
+ Assertions.assertEquals(Integer.valueOf(3),
defaultNested.get("c_int"));
+ }
+
+ // With the opt-in nullable layout a ROW field is a [null, record] union,
so a present nested
+ // row must be built under the name the enclosing schema gave the field,
otherwise the write
+ // fails with UnresolvedUnionException.
+ @Test
+ public void testPresentNestedRowSerializesWithNullableRow() throws
IOException {
+ SeaTunnelRowType rowType = buildSeaTunnelRowType();
+ SeaTunnelRow seaTunnelRow = buildSeaTunnelRow();
+
+ RowToAvroConverter converter = new RowToAvroConverter(rowType, true);
+ GenericRecord record =
converter.convertRowToGenericRecord(seaTunnelRow);
+ Assertions.assertEquals(
+ "c_row", ((GenericRecord)
record.get("c_row")).getSchema().getName());
+
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(out, null);
+ converter.getWriter().write(record, encoder);
+ encoder.flush();
+ Assertions.assertTrue(out.size() > 0);
+ }
+
+ // Two ROW fields sit under different parents but share the same leaf ROW
field name, and the
+ // two nested records differ structurally: one carries a STRING and one an
INT. With the opt-in
+ // nullable layout the nested record name is derived from the field name,
so both nested
+ // records are named "detail" while describing different fields. The
public binary round trip
+ // still has to read every value back through its own parent branch, both
for a present and for
+ // a null nested row.
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void
testSameLeafRowNameUnderDifferentParentsRoundTripsNullableRow(boolean present)
+ throws IOException {
+ SeaTunnelRowType leftDetailType =
+ new SeaTunnelRowType(
+ new String[] {"code"}, new SeaTunnelDataType<?>[]
{BasicType.STRING_TYPE});
+ SeaTunnelRowType rightDetailType =
+ new SeaTunnelRowType(
+ new String[] {"amount"}, new SeaTunnelDataType<?>[]
{BasicType.INT_TYPE});
+ SeaTunnelRowType leftParentType =
+ new SeaTunnelRowType(
+ new String[] {"detail"}, new SeaTunnelDataType<?>[]
{leftDetailType});
+ SeaTunnelRowType rightParentType =
+ new SeaTunnelRowType(
+ new String[] {"detail"}, new SeaTunnelDataType<?>[]
{rightDetailType});
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"left_parent", "right_parent"},
+ new SeaTunnelDataType<?>[] {leftParentType,
rightParentType});
+ CatalogTable catalogTable = CatalogTableUtil.getCatalogTable("", "",
"", "test", rowType);
+
+ SeaTunnelRow leftDetail = new SeaTunnelRow(new Object[] {"left-code"});
+ SeaTunnelRow rightDetail = new SeaTunnelRow(new Object[] {7});
+ SeaTunnelRow leftParent = new SeaTunnelRow(new Object[] {present ?
leftDetail : null});
+ SeaTunnelRow rightParent = new SeaTunnelRow(new Object[] {present ?
rightDetail : null});
+ SeaTunnelRow sourceRow = new SeaTunnelRow(new Object[] {leftParent,
rightParent});
+
+ byte[] bytes = new AvroSerializationSchema(rowType,
true).serialize(sourceRow);
+ SeaTunnelRow result =
+ new AvroDeserializationSchema(catalogTable, null, false,
true).deserialize(bytes);
+
+ SeaTunnelRow leftResult = (SeaTunnelRow) result.getField(0);
+ SeaTunnelRow rightResult = (SeaTunnelRow) result.getField(1);
+ if (present) {
+ SeaTunnelRow leftDetailResult = (SeaTunnelRow)
leftResult.getField(0);
+ SeaTunnelRow rightDetailResult = (SeaTunnelRow)
rightResult.getField(0);
+ Assertions.assertEquals("left-code", leftDetailResult.getField(0));
+ Assertions.assertEquals(7, rightDetailResult.getField(0));
+ } else {
+ Assertions.assertNull(leftResult.getField(0));
+ Assertions.assertNull(rightResult.getField(0));
+ }
+ }
}
diff --git
a/seatunnel-formats/seatunnel-format-avro/src/test/java/org/apache/seatunnel/format/avro/AvroSerializationSchemaTest.java
b/seatunnel-formats/seatunnel-format-avro/src/test/java/org/apache/seatunnel/format/avro/AvroSerializationSchemaTest.java
index 5579e0b1b3..b9211a0390 100644
---
a/seatunnel-formats/seatunnel-format-avro/src/test/java/org/apache/seatunnel/format/avro/AvroSerializationSchemaTest.java
+++
b/seatunnel-formats/seatunnel-format-avro/src/test/java/org/apache/seatunnel/format/avro/AvroSerializationSchemaTest.java
@@ -43,6 +43,7 @@ import java.io.IOException;
import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.LocalDateTime;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
@@ -244,6 +245,201 @@ class AvroSerializationSchemaTest {
Assertions.assertEquals(subRow.getField(13), null);
}
+ // A null ROW value is legal for the write converter
(RowToAvroConverter#resolveObject returns
+ // null before the type switch for every SeaTunnelDataType), so it must
survive the public
+ // binary round trip once the opt-in nullable ROW layout is enabled on
both sides. The three
+ // tests below cover the single root cause "a ROW schema carries no NULL
union branch" through
+ // the three containers that reuse that schema: a top-level ROW field,
ARRAY<ROW> elements and
+ // MAP<STRING, ROW> values. They check round-trip values independently of
how the schema is
+ // constructed.
+ @Test
+ public void testSerializationNestedRowFieldValueNullWithNullableRow()
throws IOException {
+ SeaTunnelRowType nestedType =
+ new SeaTunnelRowType(
+ new String[] {"InnerID"}, new SeaTunnelDataType<?>[]
{BasicType.INT_TYPE});
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"payload"}, new SeaTunnelDataType<?>[]
{nestedType});
+ CatalogTable catalogTable = CatalogTableUtil.getCatalogTable("", "",
"", "test", rowType);
+ SeaTunnelRow sourceRow = new SeaTunnelRow(1);
+ sourceRow.setField(0, null);
+
+ byte[] bytes = new AvroSerializationSchema(rowType,
true).serialize(sourceRow);
+ SeaTunnelRow result =
nullableRowDeserializer(catalogTable).deserialize(bytes);
+
+ Assertions.assertNull(result.getField(0));
+ }
+
+ @Test
+ public void testSerializationArrayOfRowNullElementWithNullableRow() throws
IOException {
+ SeaTunnelRowType nestedType =
+ new SeaTunnelRowType(
+ new String[] {"InnerID"}, new SeaTunnelDataType<?>[]
{BasicType.INT_TYPE});
+ ArrayType<SeaTunnelRow[], SeaTunnelRow> arrayType =
ArrayType.of(nestedType);
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"items"}, new SeaTunnelDataType<?>[]
{arrayType});
+ CatalogTable catalogTable = CatalogTableUtil.getCatalogTable("", "",
"", "test", rowType);
+ SeaTunnelRow present = new SeaTunnelRow(1);
+ present.setField(0, 42);
+ SeaTunnelRow sourceRow = new SeaTunnelRow(1);
+ sourceRow.setField(0, new SeaTunnelRow[] {present, null});
+
+ byte[] bytes = new AvroSerializationSchema(rowType,
true).serialize(sourceRow);
+ SeaTunnelRow result =
nullableRowDeserializer(catalogTable).deserialize(bytes);
+
+ SeaTunnelRow[] items = (SeaTunnelRow[]) result.getField(0);
+ Assertions.assertEquals(42, items[0].getField(0));
+ Assertions.assertNull(items[1]);
+ }
+
+ @Test
+ public void testSerializationMapOfRowNullValueWithNullableRow() throws
IOException {
+ SeaTunnelRowType nestedType =
+ new SeaTunnelRowType(
+ new String[] {"InnerID"}, new SeaTunnelDataType<?>[]
{BasicType.INT_TYPE});
+ MapType<String, SeaTunnelRow> mapType = new
MapType<>(BasicType.STRING_TYPE, nestedType);
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"payload_map"}, new
SeaTunnelDataType<?>[] {mapType});
+ CatalogTable catalogTable = CatalogTableUtil.getCatalogTable("", "",
"", "test", rowType);
+ SeaTunnelRow present = new SeaTunnelRow(1);
+ present.setField(0, 42);
+ Map<String, SeaTunnelRow> mapValue = new HashMap<>();
+ mapValue.put("present", present);
+ mapValue.put("absent", null);
+ SeaTunnelRow sourceRow = new SeaTunnelRow(1);
+ sourceRow.setField(0, mapValue);
+
+ byte[] bytes = new AvroSerializationSchema(rowType,
true).serialize(sourceRow);
+ SeaTunnelRow result =
nullableRowDeserializer(catalogTable).deserialize(bytes);
+
+ Map<?, ?> roundTripped = (Map<?, ?>) result.getField(0);
+ Assertions.assertEquals(2, roundTripped.size());
+ Assertions.assertTrue(roundTripped.containsKey("absent"));
+ Assertions.assertEquals(42, ((SeaTunnelRow)
roundTripped.get("present")).getField(0));
+ Assertions.assertNull(roundTripped.get("absent"));
+ }
+
+ private static AvroDeserializationSchema
nullableRowDeserializer(CatalogTable catalogTable) {
+ return new AvroDeserializationSchema(catalogTable, null, false, true);
+ }
+
+ private static SeaTunnelRow buildNestedRowTypeValue(int innerId) {
+ SeaTunnelRow nestedRow = new SeaTunnelRow(1);
+ nestedRow.setField(0, innerId);
+ return nestedRow;
+ }
+
+ private static SeaTunnelRowType buildNestedRowType() {
+ return new SeaTunnelRowType(
+ new String[] {"InnerID"}, new SeaTunnelDataType<?>[]
{BasicType.INT_TYPE});
+ }
+
+ // Both modes must stay byte identical for a schema without any ROW field:
the option only
+ // describes the ROW layout, so the binary layout of every other schema is
untouched.
+ @Test
+ public void testRowFreeSchemaBytesUnaffectedByNullableRow() {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"id", "name", "amount"},
+ new SeaTunnelDataType<?>[] {
+ BasicType.INT_TYPE, BasicType.STRING_TYPE,
BasicType.LONG_TYPE
+ });
+ SeaTunnelRow row = new SeaTunnelRow(new Object[] {1, "seaTunnel",
42L});
+
+ byte[] defaultBytes = new
AvroSerializationSchema(rowType).serialize(row);
+ byte[] explicitFalseBytes = new AvroSerializationSchema(rowType,
false).serialize(row);
+ byte[] nullableRowBytes = new AvroSerializationSchema(rowType,
true).serialize(row);
+
+ Assertions.assertArrayEquals(defaultBytes, explicitFalseBytes);
+ Assertions.assertArrayEquals(explicitFalseBytes, nullableRowBytes);
+ }
+
+ // The default path is what every existing job uses, so it has to stay
byte identical to the
+ // explicit false path even when the schema contains a ROW field.
+ @Test
+ public void testDefaultBytesMatchExplicitFalseForRowBearingSchema() {
+ SeaTunnelRowType rowType = buildSeaTunnelRowType();
+ SeaTunnelRow row = buildSeaTunnelRow();
+
+ byte[] defaultBytes = new
AvroSerializationSchema(rowType).serialize(row);
+ byte[] explicitFalseBytes = new AvroSerializationSchema(rowType,
false).serialize(row);
+
+ Assertions.assertArrayEquals(defaultBytes, explicitFalseBytes);
+ }
+
+ // Enabling the option inserts the union branch index of every ROW value,
so a schema that
+ // contains a present ROW field cannot produce the same bytes in both
modes.
+ @Test
+ public void testNonNullRowBytesDifferWhenNullableRowEnabled() {
+ SeaTunnelRowType rowType = buildSeaTunnelRowType();
+ SeaTunnelRow row = buildSeaTunnelRow();
+
+ byte[] defaultBytes = new
AvroSerializationSchema(rowType).serialize(row);
+ byte[] nullableRowBytes = new AvroSerializationSchema(rowType,
true).serialize(row);
+
+ Assertions.assertFalse(Arrays.equals(defaultBytes, nullableRowBytes));
+ }
+
+ // The flag has to be applied recursively: a ROW that is nested inside
another ROW, and a ROW
+ // nested in an ARRAY of a nested ROW, must follow the same nullable
layout as a top level ROW.
+ @Test
+ public void testNullableRowAppliesBeyondOneNestingLevel() throws
IOException {
+ SeaTunnelRowType innermostType = buildNestedRowType();
+ SeaTunnelRowType middleType =
+ new SeaTunnelRowType(
+ new String[] {"inner", "inner_items"},
+ new SeaTunnelDataType<?>[] {innermostType,
ArrayType.of(innermostType)});
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"outer"}, new SeaTunnelDataType<?>[]
{middleType});
+ CatalogTable catalogTable = CatalogTableUtil.getCatalogTable("", "",
"", "test", rowType);
+
+ SeaTunnelRow middleRow = new SeaTunnelRow(2);
+ middleRow.setField(0, null);
+ middleRow.setField(1, new SeaTunnelRow[] {buildNestedRowTypeValue(7),
null});
+ SeaTunnelRow sourceRow = new SeaTunnelRow(1);
+ sourceRow.setField(0, middleRow);
+
+ byte[] bytes = new AvroSerializationSchema(rowType,
true).serialize(sourceRow);
+ SeaTunnelRow result =
nullableRowDeserializer(catalogTable).deserialize(bytes);
+
+ SeaTunnelRow middleResult = (SeaTunnelRow) result.getField(0);
+ Assertions.assertNull(middleResult.getField(0));
+ SeaTunnelRow[] innerItems = (SeaTunnelRow[]) middleResult.getField(1);
+ Assertions.assertEquals(7, innerItems[0].getField(0));
+ Assertions.assertNull(innerItems[1]);
+ }
+
+ // Messages written before the option existed keep the legacy layout. They
stay readable only
+ // through the supplied writer schema, which keeps describing the actual
wire layout even when
+ // the nullable flag is enabled on the reader.
+ @Test
+ public void
testSuppliedWriterSchemaStaysAuthoritativeForNullableRowReader()
+ throws IOException {
+ SeaTunnelRowType rowType = buildNestedRowType();
+ SeaTunnelRowType tableRowType =
+ new SeaTunnelRowType(
+ new String[] {"payload"}, new SeaTunnelDataType<?>[]
{rowType});
+ CatalogTable catalogTable =
+ CatalogTableUtil.getCatalogTable("", "", "", "test",
tableRowType);
+ SeaTunnelRow sourceRow = new SeaTunnelRow(1);
+ sourceRow.setField(0, buildNestedRowTypeValue(42));
+
+ byte[] legacyBytes = new
AvroSerializationSchema(tableRowType).serialize(sourceRow);
+ String legacyWriterSchema =
+
SeaTunnelRowTypeToAvroSchemaConverter.buildAvroSchemaWithRowType(
+ tableRowType, false)
+ .toString();
+
+ SeaTunnelRow result =
+ new AvroDeserializationSchema(catalogTable,
legacyWriterSchema, false, true)
+ .deserialize(legacyBytes);
+
+ Assertions.assertEquals(42, ((SeaTunnelRow)
result.getField(0)).getField(0));
+ }
+
@Test
public void testDeserializeWithProvidedWriterSchema() throws IOException {
SeaTunnelRowType rowType =
