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 340363ef27 [Feature][Transform-V2] Add GET_JSON_OBJECT SQL function
(#12666)
340363ef27 is described below
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")));
+ }
+}
