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-12666-0c20c86ef8d84ba2b65bbda8784d8ec8e17b7e20 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 340363ef27b346c64519d9234107fd84a4288d9a Author: chovygo <[email protected]> AuthorDate: Fri Oct 9 05:26:21 2026 +0000 [Feature][Transform-V2] Add GET_JSON_OBJECT SQL function (#12666) Co-authored-by: zhaowei <[email protected]> --- docs/en/transforms/sql-functions.md | 37 +++ docs/zh/transforms/sql-functions.md | 37 +++ .../calcite/udf/GetJsonObjectFunction.java | 46 ++++ .../transform/sql/zeta/ZetaSQLFunction.java | 6 + .../seatunnel/transform/sql/zeta/ZetaSQLType.java | 1 + .../transform/sql/zeta/functions/JsonFunction.java | 200 +++++++++++++++++ .../calcite/udf/GetJsonObjectFunctionTest.java | 163 ++++++++++++++ .../transform/sql/SQLJsonFunctionsTest.java | 250 +++++++++++++++++++++ 8 files changed, 740 insertions(+) diff --git a/docs/en/transforms/sql-functions.md b/docs/en/transforms/sql-functions.md index 44966f6820..c59d122ed5 100644 --- a/docs/en/transforms/sql-functions.md +++ b/docs/en/transforms/sql-functions.md @@ -1165,6 +1165,43 @@ AES_DECRYPT(AES_ENCRYPT(name, 'mySecretPass'), 'mySecretPass') AES_DECRYPT(cipher, 'mySecretPass', '1234567890123456') +## JSON Functions + +### GET_JSON_OBJECT + +```GET_JSON_OBJECT(json, path) -> STRING``` + +Extracts a value from a JSON string using a JSON path and returns it as a string. Available in both the Zeta and Calcite SQL transforms. + +Supported path syntax: + +- `$` — the root document +- `.field` — object field access +- `[n]` — array index (non-negative integer) +- `['field']` / `["field"]` — bracket field access (for keys containing `.` or `[`) + +Wildcards (`[*]`, `.*`), recursive descent (`..`) and filter expressions (`[?(...)]`) are not supported. + +Return semantics: + +- Any null input (`json` or `path`) → `null` +- Invalid JSON → `null` +- Path does not match / array index out of bounds → `null` +- A JSON `null` value → `null` +- A string value → the unquoted string content +- A number or boolean → the value rendered as text +- An object or array → the raw (compact) JSON text of that node + +Example: + +GET_JSON_OBJECT('{"a":"hello","b":[10,20]}', '$.a') + +GET_JSON_OBJECT('{"a":{"b":1}}', '$.a') + +GET_JSON_OBJECT('{"b":[10,20]}', '$.b[1]') + +SELECT CASE WHEN GET_JSON_OBJECT(name, '$.key1') = 'value1' THEN 'A' ELSE 'B' END AS r FROM dual + ## System Functions ### CAST diff --git a/docs/zh/transforms/sql-functions.md b/docs/zh/transforms/sql-functions.md index ebdb15f6a8..3ac62d2a88 100644 --- a/docs/zh/transforms/sql-functions.md +++ b/docs/zh/transforms/sql-functions.md @@ -1170,6 +1170,43 @@ AES_DECRYPT(AES_ENCRYPT(name, 'mySecretPass'), 'mySecretPass') AES_DECRYPT(cipher, 'mySecretPass', '1234567890123456') +## JSON 函数 + +### GET_JSON_OBJECT + +```GET_JSON_OBJECT(json, path) -> STRING``` + +从 JSON 字符串中提取值，并以字符串形式返回。在 Zeta 与 Calcite 两种 SQL 引擎中均可用。 + +支持的 path 语法： + +- `$` — 根文档 +- `.field` — 对象字段访问 +- `[n]` — 数组下标（非负整数） +- `['field']` / `["field"]` — 括号字段访问（用于包含 `.` 或 `[` 的键） + +不支持通配符（`[*]`、`.*`）、递归下降（`..`）和过滤表达式（`[?(...)]`）。 + +返回语义： + +- 任一输入为 null（`json` 或 `path`）→ `null` +- 非法 JSON → `null` +- path 未命中 / 数组下标越界 → `null` +- JSON 的 `null` 值 → `null` +- 字符串值 → 去掉外层引号后的字符串内容 +- 数字或布尔值 → 转为文本后的值 +- 对象或数组 → 该节点的原始（紧凑）JSON 文本 + +示例： + +GET_JSON_OBJECT('{"a":"hello","b":[10,20]}', '$.a') + +GET_JSON_OBJECT('{"a":{"b":1}}', '$.a') + +GET_JSON_OBJECT('{"b":[10,20]}', '$.b[1]') + +SELECT CASE WHEN GET_JSON_OBJECT(name, '$.key1') = 'value1' THEN 'A' ELSE 'B' END AS r FROM dual + ## System Functions ### CAST diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/GetJsonObjectFunction.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/GetJsonObjectFunction.java new file mode 100644 index 0000000000..78f626bfa5 --- /dev/null +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/calcite/udf/GetJsonObjectFunction.java @@ -0,0 +1,46 @@ +/* + * 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.transform.calcite.udf; + +import org.apache.seatunnel.transform.sql.zeta.functions.JsonFunction; + +import com.google.auto.service.AutoService; + +/** + * {@code GET_JSON_OBJECT(json, path)} UDF for the Calcite SQL transform. + * + * <p>Delegates to {@link JsonFunction#getJsonObject(String, String)} so the Zeta and Calcite + * engines share one implementation of the path and return semantics. + * + * <p>Usage: {@code GET_JSON_OBJECT(json, '$.field.subfield[0]')} + */ +@AutoService(CalciteUdf.class) +public class GetJsonObjectFunction implements CalciteUdf { + + @Override + public String functionName() { + return "GET_JSON_OBJECT"; + } + + public static String eval(String json, String path) { + if (json == null || path == null) { + return null; + } + return JsonFunction.getJsonObject(json, path); + } +} diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFunction.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFunction.java index 233245d980..5a69ffd86b 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFunction.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFunction.java @@ -34,6 +34,7 @@ import org.apache.seatunnel.transform.exception.TransformException; import org.apache.seatunnel.transform.sql.zeta.functions.ArrayFunction; import org.apache.seatunnel.transform.sql.zeta.functions.CryptoFunction; import org.apache.seatunnel.transform.sql.zeta.functions.DateTimeFunction; +import org.apache.seatunnel.transform.sql.zeta.functions.JsonFunction; import org.apache.seatunnel.transform.sql.zeta.functions.MapFunction; import org.apache.seatunnel.transform.sql.zeta.functions.NumericFunction; import org.apache.seatunnel.transform.sql.zeta.functions.StringFunction; @@ -135,6 +136,9 @@ public class ZetaSQLFunction { public static final String AES_ENCRYPT = "AES_ENCRYPT"; public static final String AES_DECRYPT = "AES_DECRYPT"; + // -------------------------json functions---------------------------- + public static final String GET_JSON_OBJECT = "GET_JSON_OBJECT"; + // -------------------------numeric functions---------------------------- public static final String ABS = "ABS"; public static final String ACOS = "ACOS"; @@ -559,6 +563,8 @@ public class ZetaSQLFunction { return CryptoFunction.aesEncrypt(args); case AES_DECRYPT: return CryptoFunction.aesDecrypt(args); + case GET_JSON_OBJECT: + return JsonFunction.getJsonObject(args); case ABS: return NumericFunction.abs(args); case ACOS: diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java index b536a0746d..4d7adedf5a 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java @@ -360,6 +360,7 @@ public class ZetaSQLType { case ZetaSQLFunction.TRIM_SCALE: case ZetaSQLFunction.AES_ENCRYPT: case ZetaSQLFunction.AES_DECRYPT: + case ZetaSQLFunction.GET_JSON_OBJECT: return BasicType.STRING_TYPE; case ZetaSQLFunction.ASCII: case ZetaSQLFunction.LOCATE: diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/JsonFunction.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/JsonFunction.java new file mode 100644 index 0000000000..a070131cca --- /dev/null +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/JsonFunction.java @@ -0,0 +1,200 @@ +/* + * 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.transform.sql.zeta.functions; + +import org.apache.seatunnel.shade.com.fasterxml.jackson.core.JsonProcessingException; +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode; + +import org.apache.seatunnel.common.exception.CommonError; +import org.apache.seatunnel.common.utils.JsonUtils; +import org.apache.seatunnel.transform.sql.zeta.ZetaSQLFunction; + +import java.util.List; + +/** + * Built-in JSON functions for the Zeta SQL transform. + * + * <p>{@link #getJsonObject(List)} extracts a value from a JSON string using a JSON path. The + * supported path grammar is: + * + * <ul> + * <li>{@code $} — the root document + * <li>{@code .field} — object field access + * <li>{@code [n]} — array index (non-negative integer) + * <li>{@code ['field']} / {@code ["field"]} — bracket field access (allows keys that + * contain {@code .} or {@code [}) + * </ul> + * + * <p>Wildcards ({@code [*]}, {@code .*}), recursive descent ({@code ..}) and filter expressions + * ({@code [?(...)]}) are not supported. + * + * <p>Return semantics: + * + * <ul> + * <li>any null input (json or path) → {@code null} + * <li>invalid JSON → {@code null} + * <li>path does not match / indexes out of bounds → {@code null} + * <li>a JSON {@code null} value → {@code null} + * <li>a string value → the unquoted string content + * <li>a number or boolean → the value rendered as text + * <li>an object or array → the raw (compact) JSON text of that node + * </ul> + * + * <p>JSON is parsed via the shaded Jackson runtime shared across SeaTunnel; the path walker is a + * minimal hand-rolled tokenizer so the grammar above is matched exactly without pulling in a full + * JSONPath engine. + */ +public class JsonFunction { + + private JsonFunction() {} + + /** + * Extracts a value via {@code GET_JSON_OBJECT(json, path)}. + * + * @param args [json, path] + * @return the extracted value as a string, or null on null input / invalid JSON / missing path + */ + public static String getJsonObject(List<Object> args) { + String operation = ZetaSQLFunction.GET_JSON_OBJECT; + if (args.size() != 2) { + throw CommonError.illegalArgument( + String.valueOf(args.size()), operation + " expects 2 arguments: (json, path)"); + } + Object jsonArg = args.get(0); + Object pathArg = args.get(1); + if (jsonArg == null || pathArg == null) { + return null; + } + return getJsonObject(jsonArg.toString(), pathArg.toString()); + } + + /** Core implementation, shared with the Calcite UDF so both engines behave identically. */ + public static String getJsonObject(String json, String path) { + if (json == null || path == null) { + return null; + } + JsonNode node; + try { + node = JsonUtils.stringToJsonNode(json); + } catch (JsonProcessingException e) { + // Return null on invalid JSON rather than raising. + return null; + } + if (node == null) { + return null; + } + return render(resolvePath(node, path)); + } + + /** + * Walks the path against {@code root}, returning the matched node or null when the path is + * malformed or does not resolve. + */ + private static JsonNode resolvePath(JsonNode root, String path) { + if (path.isEmpty() || path.charAt(0) != '$') { + // The path must start with the root marker. + return null; + } + JsonNode current = root; + int i = 1; + int len = path.length(); + while (i < len && current != null) { + char c = path.charAt(i); + if (c == '.') { + int start = ++i; + while (i < len && path.charAt(i) != '.' && path.charAt(i) != '[') { + i++; + } + String field = path.substring(start, i); + if (field.isEmpty() || !current.isObject()) { + return null; + } + current = current.get(field); + } else if (c == '[') { + // Scan to the matching ']', skipping any ']' that appears inside a quoted field + // name so keys like ['a]b'] resolve correctly. + int close = -1; + char quoteChar = 0; + for (int j = i + 1; j < len; j++) { + char cj = path.charAt(j); + if (quoteChar != 0) { + if (cj == quoteChar) { + quoteChar = 0; + } + } else if (cj == '\'' || cj == '"') { + quoteChar = cj; + } else if (cj == ']') { + close = j; + break; + } + } + if (close < 0) { + return null; + } + String inside = path.substring(i + 1, close).trim(); + i = close + 1; + if (inside.isEmpty()) { + return null; + } + char quote = inside.charAt(0); + if (quote == '\'' || quote == '"') { + if (inside.length() < 2 || inside.charAt(inside.length() - 1) != quote) { + return null; + } + if (!current.isObject()) { + return null; + } + current = current.get(inside.substring(1, inside.length() - 1)); + } else { + int index; + try { + index = Integer.parseInt(inside); + } catch (NumberFormatException e) { + return null; + } + if (index < 0 || !current.isArray() || index >= current.size()) { + return null; + } + current = current.get(index); + } + } else { + // Unexpected char right after '$' or a segment: not a valid path. + return null; + } + } + return current; + } + + /** Renders the matched node per the return semantics (see class Javadoc). */ + private static String render(JsonNode node) { + if (node == null || node.isNull()) { + return null; + } + switch (node.getNodeType()) { + case OBJECT: + case ARRAY: + return node.toString(); + case STRING: + case NUMBER: + case BOOLEAN: + return node.asText(); + default: + return node.asText(); + } + } +} diff --git a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/calcite/udf/GetJsonObjectFunctionTest.java b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/calcite/udf/GetJsonObjectFunctionTest.java new file mode 100644 index 0000000000..f88cf40e1b --- /dev/null +++ b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/calcite/udf/GetJsonObjectFunctionTest.java @@ -0,0 +1,163 @@ +/* + * 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.transform.calcite.udf; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class GetJsonObjectFunctionTest { + + @Test + void testExtractsStringValue() { + Assertions.assertEquals("hello", GetJsonObjectFunction.eval("{\"a\":\"hello\"}", "$.a")); + } + + @Test + void testStringValueIsUnquoted() { + // Returns the decoded string content, without surrounding quotes. + Assertions.assertEquals( + "a \"b\" c", GetJsonObjectFunction.eval("{\"k\":\"a \\\"b\\\" c\"}", "$.k")); + } + + @Test + void testExtractsNumberValue() { + Assertions.assertEquals("42", GetJsonObjectFunction.eval("{\"a\":42}", "$.a")); + } + + @Test + void testExtractsDecimalNumber() { + Assertions.assertEquals("1.5", GetJsonObjectFunction.eval("{\"a\":1.5}", "$.a")); + } + + @Test + void testExtractsBooleanValue() { + Assertions.assertEquals("true", GetJsonObjectFunction.eval("{\"a\":true}", "$.a")); + Assertions.assertEquals("false", GetJsonObjectFunction.eval("{\"a\":false}", "$.a")); + } + + @Test + void testExtractsObjectAsRawJson() { + Assertions.assertEquals( + "{\"b\":1}", GetJsonObjectFunction.eval("{\"a\":{\"b\":1}}", "$.a")); + } + + @Test + void testExtractsArrayAsRawJson() { + Assertions.assertEquals("[1,2,3]", GetJsonObjectFunction.eval("{\"a\":[1,2,3]}", "$.a")); + } + + @Test + void testExtractsArrayElement() { + Assertions.assertEquals("20", GetJsonObjectFunction.eval("{\"a\":[10,20,30]}", "$.a[1]")); + } + + @Test + void testNestedPath() { + Assertions.assertEquals( + "deep", GetJsonObjectFunction.eval("{\"a\":{\"b\":{\"c\":\"deep\"}}}", "$.a.b.c")); + } + + @Test + void testRootPathReturnsWholeDocument() { + Assertions.assertEquals("{\"a\":1}", GetJsonObjectFunction.eval("{\"a\":1}", "$")); + } + + @Test + void testNullJsonReturnsNull() { + Assertions.assertNull(GetJsonObjectFunction.eval(null, "$.a")); + } + + @Test + void testNullPathReturnsNull() { + Assertions.assertNull(GetJsonObjectFunction.eval("{\"a\":1}", null)); + } + + @Test + void testInvalidJsonReturnsNull() { + Assertions.assertNull(GetJsonObjectFunction.eval("not a json", "$.a")); + } + + @Test + void testMissingPathReturnsNull() { + Assertions.assertNull(GetJsonObjectFunction.eval("{\"a\":1}", "$.missing")); + } + + @Test + void testJsonNullValueReturnsNull() { + Assertions.assertNull(GetJsonObjectFunction.eval("{\"a\":null}", "$.a")); + } + + @Test + void testArrayIndexOutOfBoundsReturnsNull() { + Assertions.assertNull(GetJsonObjectFunction.eval("{\"a\":[1,2]}", "$.a[5]")); + } + + @Test + void testBracketQuotedKeyWithDot() { + // Bracket-quoted field access lets keys contain '.' or '['. + Assertions.assertEquals("v", GetJsonObjectFunction.eval("{\"a.b\":\"v\"}", "$['a.b']")); + } + + @Test + void testBracketQuotedKeyWithClosingBracket() { + // The bracket scanner is quote-aware, so ']' inside a quoted key does not close early. + Assertions.assertEquals("v", GetJsonObjectFunction.eval("{\"a]b\":\"v\"}", "$['a]b']")); + } + + @Test + void testFieldAccessOnNonObjectReturnsNull() { + // .field on an array or scalar does not resolve. + Assertions.assertNull(GetJsonObjectFunction.eval("{\"a\":[1,2]}", "$.a.b")); + Assertions.assertNull(GetJsonObjectFunction.eval("{\"a\":1}", "$.a.b")); + } + + @Test + void testArrayIndexOnNonArrayReturnsNull() { + // [n] on an object or scalar does not resolve. + Assertions.assertNull(GetJsonObjectFunction.eval("{\"a\":{\"b\":1}}", "$.a[0]")); + Assertions.assertNull(GetJsonObjectFunction.eval("{\"a\":\"str\"}", "$.a[0]")); + } + + @Test + void testNegativeArrayIndexReturnsNull() { + Assertions.assertNull(GetJsonObjectFunction.eval("[1,2]", "$[-1]")); + } + + @Test + void testConsecutiveArrayIndexes() { + Assertions.assertEquals("20", GetJsonObjectFunction.eval("[[10,20],[30,40]]", "$[0][1]")); + } + + @Test + void testRootArrayAccess() { + Assertions.assertEquals("10", GetJsonObjectFunction.eval("[10,20]", "$[0]")); + } + + @Test + void testScalarRootReturnsScalar() { + // A scalar root document is rendered as text, not wrapped. + Assertions.assertEquals("42", GetJsonObjectFunction.eval("42", "$")); + Assertions.assertEquals("hello", GetJsonObjectFunction.eval("\"hello\"", "$")); + } + + @Test + void testFunctionName() { + GetJsonObjectFunction fn = new GetJsonObjectFunction(); + Assertions.assertEquals("GET_JSON_OBJECT", fn.functionName()); + } +} diff --git a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLJsonFunctionsTest.java b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLJsonFunctionsTest.java new file mode 100644 index 0000000000..7b101790aa --- /dev/null +++ b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLJsonFunctionsTest.java @@ -0,0 +1,250 @@ +/* + * 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.transform.sql; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +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.common.exception.SeaTunnelRuntimeException; +import org.apache.seatunnel.transform.sql.zeta.functions.JsonFunction; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.List; + +public class SQLJsonFunctionsTest { + + private SeaTunnelRow runSql(String query, SeaTunnelRowType rowType, Object... values) { + CatalogTable table = CatalogTableUtil.getCatalogTable("test", rowType); + ReadonlyConfig config = ReadonlyConfig.fromMap(Collections.singletonMap("query", query)); + SQLTransform transform = new SQLTransform(config, table); + List<SeaTunnelRow> out = transform.transformRow(new SeaTunnelRow(values)); + Assertions.assertNotNull(out); + Assertions.assertFalse(out.isEmpty()); + return out.get(0); + } + + private SeaTunnelRowType stringRowType(String field) { + return new SeaTunnelRowType( + new String[] {field}, new SeaTunnelDataType[] {BasicType.STRING_TYPE}); + } + + @Test + public void testExtractsStringValue() { + SeaTunnelRow out = + runSql( + "select GET_JSON_OBJECT(name, '$.a') as r from dual", + stringRowType("name"), + "{\"a\":\"hello\"}"); + Assertions.assertEquals("hello", out.getField(0)); + } + + @Test + public void testExtractsNumberValue() { + SeaTunnelRow out = + runSql( + "select GET_JSON_OBJECT(name, '$.a') as r from dual", + stringRowType("name"), + "{\"a\":42}"); + Assertions.assertEquals("42", out.getField(0)); + } + + @Test + public void testExtractsDecimalNumber() { + SeaTunnelRow out = + runSql( + "select GET_JSON_OBJECT(name, '$.a') as r from dual", + stringRowType("name"), + "{\"a\":1.5}"); + Assertions.assertEquals("1.5", out.getField(0)); + } + + @Test + public void testExtractsBooleanValue() { + SeaTunnelRow out = + runSql( + "select GET_JSON_OBJECT(name, '$.a') as r from dual", + stringRowType("name"), + "{\"a\":true}"); + Assertions.assertEquals("true", out.getField(0)); + } + + @Test + public void testExtractsObjectAsRawJson() { + SeaTunnelRow out = + runSql( + "select GET_JSON_OBJECT(name, '$.a') as r from dual", + stringRowType("name"), + "{\"a\":{\"b\":1}}"); + Assertions.assertEquals("{\"b\":1}", out.getField(0)); + } + + @Test + public void testExtractsArrayAsRawJson() { + SeaTunnelRow out = + runSql( + "select GET_JSON_OBJECT(name, '$.a') as r from dual", + stringRowType("name"), + "{\"a\":[1,2,3]}"); + Assertions.assertEquals("[1,2,3]", out.getField(0)); + } + + @Test + public void testExtractsArrayElement() { + SeaTunnelRow out = + runSql( + "select GET_JSON_OBJECT(name, '$.a[1]') as r from dual", + stringRowType("name"), + "{\"a\":[10,20,30]}"); + Assertions.assertEquals("20", out.getField(0)); + } + + @Test + public void testNestedPath() { + SeaTunnelRow out = + runSql( + "select GET_JSON_OBJECT(name, '$.a.b.c') as r from dual", + stringRowType("name"), + "{\"a\":{\"b\":{\"c\":\"deep\"}}}"); + Assertions.assertEquals("deep", out.getField(0)); + } + + @Test + public void testNestedArrayAndObject() { + SeaTunnelRow out = + runSql( + "select GET_JSON_OBJECT(name, '$.a[0].b') as r from dual", + stringRowType("name"), + "{\"a\":[{\"b\":\"first\"},{\"b\":\"second\"}]}"); + Assertions.assertEquals("first", out.getField(0)); + } + + @Test + public void testRootPathReturnsWholeDocument() { + SeaTunnelRow out = + runSql( + "select GET_JSON_OBJECT(name, '$') as r from dual", + stringRowType("name"), + "{\"a\":1}"); + Assertions.assertEquals("{\"a\":1}", out.getField(0)); + } + + @Test + public void testNullJsonReturnsNull() { + SeaTunnelRow out = + runSql( + "select GET_JSON_OBJECT(name, '$.a') as r from dual", + stringRowType("name"), + (Object) null); + Assertions.assertNull(out.getField(0)); + } + + @Test + public void testInvalidJsonReturnsNull() { + SeaTunnelRow out = + runSql( + "select GET_JSON_OBJECT(name, '$.a') as r from dual", + stringRowType("name"), + "not a json"); + Assertions.assertNull(out.getField(0)); + } + + @Test + public void testMissingPathReturnsNull() { + SeaTunnelRow out = + runSql( + "select GET_JSON_OBJECT(name, '$.missing') as r from dual", + stringRowType("name"), + "{\"a\":1}"); + Assertions.assertNull(out.getField(0)); + } + + @Test + public void testJsonNullValueReturnsNull() { + SeaTunnelRow out = + runSql( + "select GET_JSON_OBJECT(name, '$.a') as r from dual", + stringRowType("name"), + "{\"a\":null}"); + Assertions.assertNull(out.getField(0)); + } + + @Test + public void testArrayIndexOutOfBoundsReturnsNull() { + SeaTunnelRow out = + runSql( + "select GET_JSON_OBJECT(name, '$.a[5]') as r from dual", + stringRowType("name"), + "{\"a\":[1,2]}"); + Assertions.assertNull(out.getField(0)); + } + + @Test + public void testFunctionNameIsCaseInsensitive() { + // Function names are case-insensitive; the dispatcher upper-cases them. + SeaTunnelRow out = + runSql( + "select get_json_object(name, '$.a') as r from dual", + stringRowType("name"), + "{\"a\":\"ok\"}"); + Assertions.assertEquals("ok", out.getField(0)); + } + + @Test + public void testUsedInCaseWhen() { + // Reproduces the migration use case from the feature request. + SeaTunnelRow out = + runSql( + "select case when GET_JSON_OBJECT(name, '$.key1') = 'value1' then 'A'" + + " else 'B' end as r from dual", + stringRowType("name"), + "{\"key1\":\"value1\"}"); + Assertions.assertEquals("A", out.getField(0)); + } + + @Test + public void testMalformedPathReturnsNull() { + // Recursive descent, a trailing dot, and a path without the leading '$' are all rejected. + SeaTunnelRowType rowType = stringRowType("name"); + Assertions.assertNull( + runSql("select GET_JSON_OBJECT(name, '$..a') as r from dual", rowType, "{\"a\":1}") + .getField(0)); + Assertions.assertNull( + runSql("select GET_JSON_OBJECT(name, '$.a.') as r from dual", rowType, "{\"a\":1}") + .getField(0)); + Assertions.assertNull( + runSql( + "select GET_JSON_OBJECT(name, 'a.b') as r from dual", + rowType, + "{\"a\":{\"b\":1}}") + .getField(0)); + } + + @Test + public void testArgumentCountGuard() { + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> JsonFunction.getJsonObject(Collections.<Object>singletonList("only-one"))); + } +}
