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 861d741f2d [Fix][Format] Reject incompatible JSON array and map shapes
(#12962)
861d741f2d is described below
commit 861d741f2d6488e8eff31760aa3c6aff553362fd
Author: hutiefang76 <[email protected]>
AuthorDate: Sat Oct 24 02:37:39 2026 +0000
[Fix][Format] Reject incompatible JSON array and map shapes (#12962)
---
.../introduction/concepts/incompatible-changes.md | 8 +
.../introduction/concepts/incompatible-changes.md | 8 +
.../seatunnel/format/json/JsonToRowConverters.java | 8 +
.../format/json/JsonRowDataSerDeSchemaTest.java | 269 +++++++++++++++++++++
4 files changed, 293 insertions(+)
diff --git a/docs/en/introduction/concepts/incompatible-changes.md
b/docs/en/introduction/concepts/incompatible-changes.md
index 856f75963a..0decce7cd3 100644
--- a/docs/en/introduction/concepts/incompatible-changes.md
+++ b/docs/en/introduction/concepts/incompatible-changes.md
@@ -5,6 +5,14 @@ You need to check this document before you upgrade to related
version.
## dev
+### JSON format: `ARRAY` and `MAP` columns reject values of the wrong
container type
+
+When a field is declared as `ARRAY`, a JSON value that is not an array (a
string, a number, a boolean or an object) previously produced an empty array
`[]`, or an array holding `null` for every member of the object, without any
warning. When a field is declared as `MAP`, a JSON value that is not an object
silently produced an empty map `{}`. Both happened with `ignore-parse-errors`
disabled as well as enabled.
+
+Such a value now causes a parse error that identifies the field and the
expected container type; with `ignore-parse-errors = true` only that container
value is set to `null`, and every other value of the row is kept. The same
holds for a wrong container type nested inside an array or a map: only the
offending nested value becomes `null` while its siblings and the enclosing
container are retained, and empty arrays and objects of the declared container
type, `null` and missing values, and [...]
+
+The same strictness applies wherever the shared JSON-to-row conversion is used
without ignoring parse errors: the `JsonPath` transform (column `dest_type`),
the `Python` transform (declared output columns and their declared types) and
the Assert sink's `equals_to` check for `ARRAY` and `MAP` values. These callers
do not consult the JSON format's `ignore-parse-errors` option, so that switch
does not apply to them: a value that does not match the declared `ARRAY` or
`MAP` type still fails [...]
+
### File sink: XML applies `date_format`, `datetime_format` and `time_format`
- **Behavior change: the XML writer of the file sinks formats DATE, TIME and
TIMESTAMP values with the configured formats**
diff --git a/docs/zh/introduction/concepts/incompatible-changes.md
b/docs/zh/introduction/concepts/incompatible-changes.md
index f1de528596..aca33c6a98 100644
--- a/docs/zh/introduction/concepts/incompatible-changes.md
+++ b/docs/zh/introduction/concepts/incompatible-changes.md
@@ -4,6 +4,14 @@
## dev
+### JSON 格式：`ARRAY` 与 `MAP` 字段拒绝容器类型不匹配的值
+
+当字段声明为 `ARRAY` 时，此前若 JSON 值不是数组（字符串、数字、布尔值或对象），会静默得到空数组 `[]`，若是对象则得到每个成员都为
`null` 的数组；当字段声明为 `MAP` 时，非对象的 JSON 值会静默得到空 Map `{}`。无论 `ignore-parse-errors`
关闭还是开启，行为都是如此。
+
+现在这类值会报告解析错误，指出字段名和预期的容器类型；配置 `ignore-parse-errors = true` 时只把该容器值置为
`null`，同一行的其余值仍然保留。嵌套在数组或 Map 中的类似值同样如此：只有出错的那个嵌套值变为
`null`，其兄弟值和外层容器仍然保留；与声明类型匹配的空数组、空对象，以及 `null` 与缺失值，以及元素和键的转换行为保持不变。如需迁移，请为
`ARRAY` 字段提供真正的 JSON 数组、为 `MAP` 字段提供真正的 JSON 对象，或开启 `ignore-parse-errors =
true`，使格式不合法的容器值变成 `null`而不是让作业失败。
+
+同样的严格校验也适用于未忽略解析错误而使用共享 JSON 转行转换的地方：`JsonPath` 转换（列的 `dest_type`）、`Python`
转换（声明的输出列及其声明类型），以及 Assert 的 `equals_to` 对 `ARRAY`、`MAP` 值的比较。这些调用方不会读取 JSON
格式的 `ignore-parse-errors` 选项，因此该开关对它们不生效：与声明的 `ARRAY` 或 `MAP`
类型不匹配的值在那里仍会失败，而这些转换插件自身的错误处理行为保持不变。
+
### File Sink：XML 应用 `date_format`、`datetime_format` 和 `time_format`
- **行为变更：文件类 Sink 的 XML 写入器按配置的格式输出 DATE、TIME 和 TIMESTAMP**
diff --git
a/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/JsonToRowConverters.java
b/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/JsonToRowConverters.java
index 040329a869..850705877e 100644
---
a/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/JsonToRowConverters.java
+++
b/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/JsonToRowConverters.java
@@ -455,6 +455,10 @@ public class JsonToRowConverters implements Serializable {
return new JsonToObjectConverter() {
@Override
public Object convert(JsonNode jsonNode, String fieldName) {
+ if (!jsonNode.isArray()) {
+ throw new JsonParseException(
+ String.format("Field '%s' expects an array for
ARRAY.", fieldName));
+ }
Object arr =
Array.newInstance(type.getElementType().getTypeClass(), jsonNode.size());
for (int i = 0; i < jsonNode.size(); i++) {
@@ -471,6 +475,10 @@ public class JsonToRowConverters implements Serializable {
return new JsonToObjectConverter() {
@Override
public Object convert(JsonNode jsonNode, String fieldName) {
+ if (!jsonNode.isObject()) {
+ throw new JsonParseException(
+ String.format("Field '%s' expects an object for
MAP.", fieldName));
+ }
Map<Object, Object> value = new HashMap<>();
jsonNode.fields()
.forEachRemaining(
diff --git
a/seatunnel-formats/seatunnel-format-json/src/test/java/org/apache/seatunnel/format/json/JsonRowDataSerDeSchemaTest.java
b/seatunnel-formats/seatunnel-format-json/src/test/java/org/apache/seatunnel/format/json/JsonRowDataSerDeSchemaTest.java
index 25d5053b8f..b719b5b814 100644
---
a/seatunnel-formats/seatunnel-format-json/src/test/java/org/apache/seatunnel/format/json/JsonRowDataSerDeSchemaTest.java
+++
b/seatunnel-formats/seatunnel-format-json/src/test/java/org/apache/seatunnel/format/json/JsonRowDataSerDeSchemaTest.java
@@ -26,6 +26,7 @@ import
org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ObjectNode
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.DecimalType;
import org.apache.seatunnel.api.table.type.LocalTimeType;
import org.apache.seatunnel.api.table.type.MapType;
@@ -76,6 +77,27 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class JsonRowDataSerDeSchemaTest {
+ private static final SeaTunnelRowType CONTAINER_ARRAY_ROW_TYPE =
+ new SeaTunnelRowType(
+ new String[] {"id", "arr"},
+ new SeaTunnelDataType<?>[] {INT_TYPE, INT_ARRAY_TYPE});
+
+ private static final SeaTunnelRowType CONTAINER_MAP_ROW_TYPE =
+ new SeaTunnelRowType(
+ new String[] {"id", "m"},
+ new SeaTunnelDataType<?>[] {INT_TYPE, new
MapType<>(STRING_TYPE, INT_TYPE)});
+
+ private static final ArrayType<Map<String, Integer>[], Map<String,
Integer>> MAP_ARRAY_TYPE =
+ ArrayType.of(new MapType<>(STRING_TYPE, INT_TYPE));
+
+ private static final MapType<String, Integer[]> ARRAY_MAP_TYPE =
+ new MapType<>(STRING_TYPE, INT_ARRAY_TYPE);
+
+ private static final SeaTunnelRowType NESTED_CONTAINER_ROW_TYPE =
+ new SeaTunnelRowType(
+ new String[] {"id", "outer", "nested"},
+ new SeaTunnelDataType<?>[] {INT_TYPE, MAP_ARRAY_TYPE,
ARRAY_MAP_TYPE});
+
@Test
public void testTimestampTzSerializedAsSessionWallClock() {
// Issue #10795: with serializeTimestampTzAsLocal=true (used by the
Doris
@@ -906,4 +928,251 @@ public class JsonRowDataSerDeSchemaTest {
LocalDateTime.of(2022, 9, 24, 22, 45, 0, 123_000_000),
fractionalPrecisionRow.getField(0));
}
+
+ @Test
+ public void testArrayFieldRejectsNonArrayNodeInStrictMode() {
+ JsonDeserializationSchema schema =
+ new JsonDeserializationSchema(false, false,
CONTAINER_ARRAY_ROW_TYPE);
+ final String[] payloads = {
+ "{\"id\":7,\"arr\":\"abc\"}",
+ "{\"id\":7,\"arr\":42}",
+ "{\"id\":7,\"arr\":true}",
+ "{\"id\":7,\"arr\":{\"k\":1}}"
+ };
+ Assertions.assertAll(
+ "ARRAY field must reject non-array JSON nodes in strict mode",
+ () ->
+ assertContainerTypeRejectedInStrictMode(
+ schema, payloads[0], "'arr'", "ARRAY"),
+ () ->
+ assertContainerTypeRejectedInStrictMode(
+ schema, payloads[1], "'arr'", "ARRAY"),
+ () ->
+ assertContainerTypeRejectedInStrictMode(
+ schema, payloads[2], "'arr'", "ARRAY"),
+ () ->
+ assertContainerTypeRejectedInStrictMode(
+ schema, payloads[3], "'arr'", "ARRAY"));
+ }
+
+ @Test
+ public void testArrayFieldRejectsNonArrayNodeInLenientMode() throws
Exception {
+ JsonDeserializationSchema schema =
+ new JsonDeserializationSchema(false, true,
CONTAINER_ARRAY_ROW_TYPE);
+ final String[] payloads = {
+ "{\"id\":7,\"arr\":\"abc\"}",
+ "{\"id\":7,\"arr\":42}",
+ "{\"id\":7,\"arr\":true}",
+ "{\"id\":7,\"arr\":{\"k\":1}}"
+ };
+ Assertions.assertAll(
+ "ARRAY field must be null for non-array JSON nodes in lenient
mode",
+ () -> assertContainerFieldNullInLenientMode(schema,
payloads[0]),
+ () -> assertContainerFieldNullInLenientMode(schema,
payloads[1]),
+ () -> assertContainerFieldNullInLenientMode(schema,
payloads[2]),
+ () -> assertContainerFieldNullInLenientMode(schema,
payloads[3]));
+ }
+
+ @Test
+ public void testArrayFieldAcceptsArrayNodeInBothModes() throws Exception {
+ String[] payloads = {
+ "{\"id\":7,\"arr\":[]}",
+ "{\"id\":\"7\",\"arr\":[1,2]}",
+ "{\"id\":7,\"arr\":[1,null]}",
+ "{\"id\":7,\"arr\":null}"
+ };
+ Object[] expected = {new Integer[0], new Integer[] {1, 2}, new
Integer[] {1, null}, null};
+ for (boolean ignoreErrors : new boolean[] {false, true}) {
+ JsonDeserializationSchema schema =
+ new JsonDeserializationSchema(false, ignoreErrors,
CONTAINER_ARRAY_ROW_TYPE);
+ for (int i = 0; i < payloads.length; i++) {
+ String payload = payloads[i];
+ String message = payload + " (ignoreErrors=" + ignoreErrors +
")";
+ SeaTunnelRow rowFromBytes =
+
schema.deserialize(payload.getBytes(StandardCharsets.UTF_8));
+ SeaTunnelRow rowFromString = schema.deserialize(payload);
+ Assertions.assertNotNull(rowFromBytes, message);
+ Assertions.assertNotNull(rowFromString, message);
+ assertEquals(7, rowFromBytes.getField(0), message);
+ assertEquals(rowFromBytes.getField(0),
rowFromString.getField(0), message);
+ if (expected[i] == null) {
+ assertNull(rowFromBytes.getField(1), message);
+ assertNull(rowFromString.getField(1), message);
+ } else {
+ assertArrayEquals(
+ (Integer[]) expected[i], (Integer[])
rowFromBytes.getField(1), message);
+ assertArrayEquals(
+ (Integer[]) expected[i],
+ (Integer[]) rowFromString.getField(1),
+ message);
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testMapFieldRejectsNonObjectNodeInStrictMode() {
+ JsonDeserializationSchema schema =
+ new JsonDeserializationSchema(false, false,
CONTAINER_MAP_ROW_TYPE);
+ final String[] payloads = {
+ "{\"id\":7,\"m\":\"abc\"}",
+ "{\"id\":7,\"m\":42}",
+ "{\"id\":7,\"m\":true}",
+ "{\"id\":7,\"m\":[1,2]}"
+ };
+ Assertions.assertAll(
+ "MAP field must reject non-object JSON nodes in strict mode",
+ () -> assertContainerTypeRejectedInStrictMode(schema,
payloads[0], "'m'", "MAP"),
+ () -> assertContainerTypeRejectedInStrictMode(schema,
payloads[1], "'m'", "MAP"),
+ () -> assertContainerTypeRejectedInStrictMode(schema,
payloads[2], "'m'", "MAP"),
+ () -> assertContainerTypeRejectedInStrictMode(schema,
payloads[3], "'m'", "MAP"));
+ }
+
+ @Test
+ public void testMapFieldRejectsNonObjectNodeInLenientMode() throws
Exception {
+ JsonDeserializationSchema schema =
+ new JsonDeserializationSchema(false, true,
CONTAINER_MAP_ROW_TYPE);
+ final String[] payloads = {
+ "{\"id\":7,\"m\":\"abc\"}",
+ "{\"id\":7,\"m\":42}",
+ "{\"id\":7,\"m\":true}",
+ "{\"id\":7,\"m\":[1,2]}"
+ };
+ Assertions.assertAll(
+ "MAP field must be null for non-object JSON nodes in lenient
mode",
+ () -> assertContainerFieldNullInLenientMode(schema,
payloads[0]),
+ () -> assertContainerFieldNullInLenientMode(schema,
payloads[1]),
+ () -> assertContainerFieldNullInLenientMode(schema,
payloads[2]),
+ () -> assertContainerFieldNullInLenientMode(schema,
payloads[3]));
+ }
+
+ @Test
+ public void testMapFieldAcceptsObjectNodeInBothModes() throws Exception {
+ Map<Object, Object> oneEntry = new HashMap<>();
+ oneEntry.put("a", 1);
+ oneEntry.put("b", null);
+ String[] payloads = {
+ "{\"id\":7,\"m\":{}}", "{\"id\":7,\"m\":{\"a\":1,\"b\":null}}",
"{\"id\":7,\"m\":null}"
+ };
+ Object[] expected = {new HashMap<Object, Object>(), oneEntry, null};
+ for (boolean ignoreErrors : new boolean[] {false, true}) {
+ JsonDeserializationSchema schema =
+ new JsonDeserializationSchema(false, ignoreErrors,
CONTAINER_MAP_ROW_TYPE);
+ for (int i = 0; i < payloads.length; i++) {
+ String payload = payloads[i];
+ String message = payload + " (ignoreErrors=" + ignoreErrors +
")";
+ SeaTunnelRow rowFromBytes =
+
schema.deserialize(payload.getBytes(StandardCharsets.UTF_8));
+ SeaTunnelRow rowFromString = schema.deserialize(payload);
+ Assertions.assertNotNull(rowFromBytes, message);
+ Assertions.assertNotNull(rowFromString, message);
+ assertEquals(7, rowFromBytes.getField(0), message);
+ assertEquals(rowFromBytes.getField(0),
rowFromString.getField(0), message);
+ if (expected[i] == null) {
+ assertNull(rowFromBytes.getField(1), message);
+ assertNull(rowFromString.getField(1), message);
+ } else {
+ assertEquals(expected[i], rowFromBytes.getField(1),
message);
+ assertEquals(rowFromBytes.getField(1),
rowFromString.getField(1), message);
+ }
+ }
+ }
+ }
+
+ @Test
+ public void
testNestedContainerMismatchKeepsEnclosingContainerInLenientMode() throws
Exception {
+ JsonDeserializationSchema lenientSchema =
+ new JsonDeserializationSchema(false, true,
NESTED_CONTAINER_ROW_TYPE);
+ JsonDeserializationSchema strictSchema =
+ new JsonDeserializationSchema(false, false,
NESTED_CONTAINER_ROW_TYPE);
+
+ Map<Object, Object> firstElement = new HashMap<>();
+ firstElement.put("a", 1);
+ Map<Object, Object> secondElement = new HashMap<>();
+ secondElement.put("b", 2);
+ String outerPayload = "{\"id\":7,\"outer\":[{\"a\":1},42,{\"b\":2}]}";
+ SeaTunnelRow[] outerRows = {
+
lenientSchema.deserialize(outerPayload.getBytes(StandardCharsets.UTF_8)),
+ lenientSchema.deserialize(outerPayload)
+ };
+ for (SeaTunnelRow row : outerRows) {
+ Assertions.assertNotNull(row, outerPayload);
+ assertEquals(7, row.getField(0), outerPayload);
+ Object[] outer = (Object[]) row.getField(1);
+ Assertions.assertNotNull(outer, outerPayload);
+ assertEquals(3, outer.length, outerPayload);
+ assertEquals(firstElement, outer[0], outerPayload);
+ assertNull(outer[1], outerPayload);
+ assertEquals(secondElement, outer[2], outerPayload);
+ }
+
+ String nestedPayload =
"{\"id\":7,\"nested\":{\"good\":[1,null],\"bad\":true}}";
+ SeaTunnelRow[] nestedRows = {
+
lenientSchema.deserialize(nestedPayload.getBytes(StandardCharsets.UTF_8)),
+ lenientSchema.deserialize(nestedPayload)
+ };
+ for (SeaTunnelRow row : nestedRows) {
+ Assertions.assertNotNull(row, nestedPayload);
+ assertEquals(7, row.getField(0), nestedPayload);
+ @SuppressWarnings("unchecked")
+ Map<Object, Object> nested = (Map<Object, Object>) row.getField(2);
+ Assertions.assertNotNull(nested, nestedPayload);
+ assertEquals(2, nested.size(), nestedPayload);
+ assertArrayEquals(
+ new Integer[] {1, null}, (Integer[]) nested.get("good"),
nestedPayload);
+ assertTrue(nested.containsKey("bad"), nestedPayload);
+ assertNull(nested.get("bad"), nestedPayload);
+ }
+
+ assertContainerTypeRejectedInStrictMode(strictSchema, outerPayload,
"'outer'", "MAP");
+ assertContainerTypeRejectedInStrictMode(strictSchema, nestedPayload,
"nested", "ARRAY");
+ }
+
+ private static void assertContainerFieldNullInLenientMode(
+ JsonDeserializationSchema schema, String payload) throws Exception
{
+ SeaTunnelRow rowFromBytes =
schema.deserialize(payload.getBytes(StandardCharsets.UTF_8));
+ SeaTunnelRow rowFromString = schema.deserialize(payload);
+ Assertions.assertNotNull(rowFromBytes, payload);
+ Assertions.assertNotNull(rowFromString, payload);
+ assertEquals(7, rowFromBytes.getField(0), payload);
+ assertEquals(7, rowFromString.getField(0), payload);
+ assertNull(rowFromBytes.getField(1), payload);
+ assertNull(rowFromString.getField(1), payload);
+ }
+
+ private static void assertContainerTypeRejectedInStrictMode(
+ JsonDeserializationSchema schema, String payload, String
fieldName, String typeToken) {
+ SeaTunnelRuntimeException fromBytes =
+ assertThrows(
+ SeaTunnelRuntimeException.class,
+ () ->
schema.deserialize(payload.getBytes(StandardCharsets.UTF_8)),
+ payload);
+ SeaTunnelRuntimeException fromString =
+ assertThrows(
+ SeaTunnelRuntimeException.class,
+ () -> schema.deserialize(payload),
+ payload);
+ assertCauseChainMentions(fromBytes, fieldName, typeToken, payload);
+ assertCauseChainMentions(fromString, fieldName, typeToken, payload);
+ }
+
+ private static void assertCauseChainMentions(
+ Throwable failure, String fieldName, String typeToken, String
payload) {
+ Throwable current = failure;
+ while (current != null) {
+ String message = current.getMessage();
+ if (message != null && message.contains(fieldName) &&
message.contains(typeToken)) {
+ return;
+ }
+ current = current.getCause();
+ }
+ throw new AssertionError(
+ payload
+ + " must fail with a cause mentioning "
+ + fieldName
+ + " and "
+ + typeToken
+ + ", but was: "
+ + failure);
+ }
}
