This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12962-1ca46905e28b08c460bac3d0e15ebcbec5181540 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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); + } }
