This is an automated email from the ASF dual-hosted git repository.
yuxiqian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new cc384a0e0 [FLINK-40313][runtime] Support nested collection functions
in Transforms (#4496)
cc384a0e0 is described below
commit cc384a0e01628d9be49cc2b9417931d238a996fc
Author: haruki <[email protected]>
AuthorDate: Fri Aug 14 16:03:56 2026 +0800
[FLINK-40313][runtime] Support nested collection functions in Transforms
(#4496)
---
docs/content.zh/docs/core-concept/transform.md | 7 +
docs/content/docs/core-concept/transform.md | 7 +
.../src/test/resources/specs/nested.yaml | 48 ++++++-
.../runtime/functions/impl/StructFunctions.java | 155 +++++++++++++++++++++
.../flink/cdc/runtime/parser/JaninoCompiler.java | 140 ++++++++++++++++++-
.../flink/cdc/runtime/parser/TransformParser.java | 47 ++++++-
.../parser/metadata/TransformSqlOperatorTable.java | 22 +++
.../typeutils/CalciteDataTypeConverter.java | 19 +++
.../functions/impl/StructFunctionsTest.java | 65 +++++++++
.../cdc/runtime/parser/TransformParserTest.java | 110 +++++++++++++++
10 files changed, 616 insertions(+), 4 deletions(-)
diff --git a/docs/content.zh/docs/core-concept/transform.md
b/docs/content.zh/docs/core-concept/transform.md
index 545f069de..8636567cf 100644
--- a/docs/content.zh/docs/core-concept/transform.md
+++ b/docs/content.zh/docs/core-concept/transform.md
@@ -281,6 +281,13 @@ Flink CDC 使用 [Calcite](https://calcite.apache.org/)
来解析表达式并且
| 函数 | Janino 代码 | 描述 |
| -------- | ----------- | ----------- |
+| ARRAY[value, ...] | array(value, ...) | 使用给定值构造数组。所有值必须具有公共类型。 |
+| MAP[key, value, ...] | map(key, value, ...) | 使用键值对构造 Map。所有键、所有值必须分别具有公共类型。
|
+| ROW(value, ...) | row(value, ...) | 使用给定值构造 Row。字段名依次生成为 `f0`、`f1` 等。 |
+| CARDINALITY(arrayOrMap) | cardinality(arrayOrMap) | 返回数组或 Map 的元素数量。输入为 NULL
时返回 NULL。 |
+| ARRAY_CONTAINS(array, value) | arrayContains(array, value) | 返回数组是否包含给定值。 |
+| ARRAY_POSITION(array, value) | arrayPosition(array, value) | 返回首个匹配值的位置，位置从
1 开始；未找到时返回 0。 |
+| ELEMENT(array) | element(array) | 返回单元素数组中的元素。空数组返回 NULL；数组包含多个元素时抛出异常。 |
| array[index] | itemAccess(array, index) | 返回数组中位置 `index` 的元素。索引从 1 开始（SQL
标准）。如果索引超出范围或数组为 NULL，则返回 NULL。 |
| map[key] | itemAccess(map, key) | 返回 map 中与 `key` 关联的值。如果 key 不存在或 map 为
NULL，则返回 NULL。 |
| row[index] | itemAccess(row, index) | 返回 row 中位置 `index` 的字段。索引从 1
开始。索引必须是常量（不能是计算表达式），因为返回类型必须在静态阶段确定。 |
diff --git a/docs/content/docs/core-concept/transform.md
b/docs/content/docs/core-concept/transform.md
index 5e623b277..a9d217ae7 100644
--- a/docs/content/docs/core-concept/transform.md
+++ b/docs/content/docs/core-concept/transform.md
@@ -282,6 +282,13 @@ Struct functions are used to access elements in ARRAY,
MAP, ROW, and VARIANT typ
| Function | Janino Code | Description |
| -------- | ----------- | ----------- |
+| ARRAY[value, ...] | array(value, ...) | Creates an array from the given
values. The values must have a common type. |
+| MAP[key, value, ...] | map(key, value, ...) | Creates a map from key-value
pairs. Keys and values must have common types respectively. |
+| ROW(value, ...) | row(value, ...) | Creates a row from the given values.
Field names are generated as `f0`, `f1`, and so on. |
+| CARDINALITY(arrayOrMap) | cardinality(arrayOrMap) | Returns the number of
elements in an array or map. Returns NULL if the input is NULL. |
+| ARRAY_CONTAINS(array, value) | arrayContains(array, value) | Returns whether
the array contains the given value. |
+| ARRAY_POSITION(array, value) | arrayPosition(array, value) | Returns the
1-based position of the first matching value, or 0 if it is not found. |
+| ELEMENT(array) | element(array) | Returns the element of a single-element
array. Returns NULL for an empty array and throws an exception if the array has
more than one element. |
| array[index] | itemAccess(array, index) | Returns the element at position
`index` in the array. Index is 1-based (SQL standard). Returns NULL if the
index is out of bounds or if the array is NULL. |
| map[key] | itemAccess(map, key) | Returns the value associated with `key` in
the map. Returns NULL if the key does not exist or if the map is NULL. |
| row[index] | itemAccess(row, index) | Returns the field at position `index`
in the row. Index is 1-based. The index must be a constant (not a computed
expression) since the return type must be statically determined. |
diff --git a/flink-cdc-composer/src/test/resources/specs/nested.yaml
b/flink-cdc-composer/src/test/resources/specs/nested.yaml
index 3ac67facd..08bbaecb7 100644
--- a/flink-cdc-composer/src/test/resources/specs/nested.yaml
+++ b/flink-cdc-composer/src/test/resources/specs/nested.yaml
@@ -30,6 +30,52 @@
DataChangeEvent{tableId=foo.bar.baz, before=[-1, [2, 3, 5, 7, 11, 13, 17,
19], [二, san, 五, qi, 十一], {1 -> yi, 2 -> er, 3 -> san}, {二 -> [E, R], 三 -> [S,
A, N], 一 -> [Y, I]}, {name: STRING -> Derrida, length: INT -> 7},
[{"k":1},"hello",{"k":2}]], after=[], op=DELETE, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, null,
null, null, null, null], op=INSERT, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[0, null, null, null, null,
null, null], after=[], op=DELETE, meta=()}
+- do: Construct Nested Objects
+ projection: |-
+ id_
+ ARRAY[1, id_] AS constructed_array
+ MAP['id', id_, 'size', CARDINALITY(array_int_)] AS constructed_map
+ ROW(id_, ARRAY[string_, varchar_]) AS constructed_row
+ MAP['nested', ARRAY[id_, bigint_]] AS nested_map
+ ARRAY[ARRAY[1], ARRAY[id_]] AS nested_array
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`constructed_array` ARRAY<BIGINT>,`constructed_map`
MAP<STRING, BIGINT>,`constructed_row` ROW<`f0` BIGINT, `f1`
ARRAY<STRING>>,`nested_map` MAP<STRING, ARRAY<BIGINT>>,`nested_array`
ARRAY<ARRAY<BIGINT>>}, primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, [1, 1], {id ->
1, size -> 7}, {f0: BIGINT -> 1, f1: ARRAY<STRING> -> [From A to Z is Lie,
Zorro]}, {nested -> [1, 5]}, [[1], [1]]], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, [1, 1], {id -> 1, size ->
7}, {f0: BIGINT -> 1, f1: ARRAY<STRING> -> [From A to Z is Lie, Zorro]},
{nested -> [1, 5]}, [[1], [1]]], after=[-1, [1, -1], {id -> -1, size -> 8},
{f0: BIGINT -> -1, f1: ARRAY<STRING> -> [天地玄黄宇宙洪荒, 疯帽子]}, {nested -> [-1, -5]},
[[1], [-1]]], op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, [1, -1], {id -> -1, size
-> 8}, {f0: BIGINT -> -1, f1: ARRAY<STRING> -> [天地玄黄宇宙洪荒, 疯帽子]}, {nested ->
[-1, -5]}, [[1], [-1]]], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, [1, 0], {id ->
0, size -> null}, {f0: BIGINT -> 0, f1: ARRAY<STRING> -> [null, null]}, {nested
-> [0, null]}, [[1], [0]]], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, [1, 0], {id -> 0, size ->
null}, {f0: BIGINT -> 0, f1: ARRAY<STRING> -> [null, null]}, {nested -> [0,
null]}, [[1], [0]]], after=[], op=DELETE, meta=()}
+- do: Collection Functions
+ projection: |-
+ id_
+ CARDINALITY(array_int_) AS array_size
+ CARDINALITY(map_int_string_) AS map_size
+ ARRAY_CONTAINS(array_int_, 5) AS contains_five
+ ARRAY_POSITION(array_int_, 5) AS position_five
+ ELEMENT(ARRAY[id_]) AS only_element
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`array_size` INT,`map_size` INT,`contains_five`
BOOLEAN,`position_five` INT,`only_element` BIGINT}, primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 7, 3, true, 5,
1], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, 7, 3, true, 5, 1],
after=[-1, 8, 3, true, 3, -1], op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, 8, 3, true, 3, -1],
after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, null,
null, null, 0], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, null, null, null, null,
0], after=[], op=DELETE, meta=()}
+- do: Filter by Collection Functions
+ projection: id_, array_int_, array_string_
+ filter: ARRAY_CONTAINS(array_string_, '五') AND ARRAY_POSITION(array_int_, 5)
= 3
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`array_int_` ARRAY<INT>,`array_string_` ARRAY<STRING>},
primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[-1, [2, 3, 5, 7,
11, 13, 17, 19], [二, san, 五, qi, 十一]], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, [2, 3, 5, 7, 11, 13, 17,
19], [二, san, 五, qi, 十一]], after=[], op=DELETE, meta=()}
+- do: Element with Multiple Values
+ projection: |-
+ id_
+ ELEMENT(array_int_) AS invalid_element
+ primary-key: id_
+ expect-error: 'Array has more than one element.'
- do: Parse Variant from JSON
projection: |-
id_
@@ -363,4 +409,4 @@
DataChangeEvent{tableId=foo.bar.baz, before=[1, "value", 42], after=[-1,
"value", 42], op=UPDATE, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[-1, "value", 42], after=[],
op=DELETE, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, "value", 42],
op=INSERT, meta=()}
- DataChangeEvent{tableId=foo.bar.baz, before=[0, "value", 42], after=[],
op=DELETE, meta=()}
\ No newline at end of file
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, "value", 42], after=[],
op=DELETE, meta=()}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/StructFunctions.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/StructFunctions.java
index 9ce343782..32da35d38 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/StructFunctions.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/StructFunctions.java
@@ -19,6 +19,10 @@ package org.apache.flink.cdc.runtime.functions.impl;
import org.apache.flink.cdc.common.types.variant.Variant;
+import java.lang.reflect.Array;
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -30,6 +34,157 @@ import java.util.Map;
*/
public class StructFunctions {
+ /** Creates an ARRAY value. */
+ public static List<Object> array(Object... elements) {
+ List<Object> result = new ArrayList<>(elements.length);
+ for (Object element : elements) {
+ result.add(element);
+ }
+ return result;
+ }
+
+ /** Creates a MAP value from alternating keys and values. */
+ public static Map<Object, Object> map(Object... keyValues) {
+ if (keyValues.length == 0 || keyValues.length % 2 != 0) {
+ throw new IllegalArgumentException(
+ "MAP requires at least one key-value pair and an even
number of arguments.");
+ }
+ Map<Object, Object> result = new LinkedHashMap<>();
+ for (int i = 0; i < keyValues.length; i += 2) {
+ result.put(keyValues[i], keyValues[i + 1]);
+ }
+ return result;
+ }
+
+ /** Creates a ROW value. */
+ public static List<Object> row(Object... fields) {
+ List<Object> result = new ArrayList<>(fields.length);
+ for (Object field : fields) {
+ result.add(field);
+ }
+ return result;
+ }
+
+ /** Returns the number of elements in an ARRAY. */
+ public static Integer cardinality(List<?> array) {
+ return array == null ? null : array.size();
+ }
+
+ /** Returns the number of entries in a MAP. */
+ public static Integer cardinality(Map<?, ?> map) {
+ return map == null ? null : map.size();
+ }
+
+ /** Returns whether an ARRAY contains the given value. */
+ public static Boolean arrayContains(List<?> array, Object value) {
+ if (array == null) {
+ return null;
+ }
+ for (Object element : array) {
+ if (valueEquals(element, value)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /** Returns the 1-based position of a value in an ARRAY, or 0 if it is not
found. */
+ public static Integer arrayPosition(List<?> array, Object value) {
+ if (array == null || value == null) {
+ return null;
+ }
+ for (int i = 0; i < array.size(); i++) {
+ if (valueEquals(array.get(i), value)) {
+ return i + 1;
+ }
+ }
+ return 0;
+ }
+
+ /** Returns the only element of an ARRAY. */
+ public static <T> T element(List<T> array) {
+ if (array == null || array.isEmpty()) {
+ return null;
+ }
+ if (array.size() > 1) {
+ throw new IllegalArgumentException("Array has more than one
element.");
+ }
+ return array.get(0);
+ }
+
+ private static boolean valueEquals(Object left, Object right) {
+ if (left == right) {
+ return true;
+ }
+ if (left == null || right == null) {
+ return false;
+ }
+ if (left instanceof Number && right instanceof Number) {
+ return numberEquals((Number) left, (Number) right);
+ }
+ if (left instanceof List<?> && right instanceof List<?>) {
+ List<?> leftList = (List<?>) left;
+ List<?> rightList = (List<?>) right;
+ if (leftList.size() != rightList.size()) {
+ return false;
+ }
+ for (int i = 0; i < leftList.size(); i++) {
+ if (!valueEquals(leftList.get(i), rightList.get(i))) {
+ return false;
+ }
+ }
+ return true;
+ }
+ if (left instanceof Map<?, ?> && right instanceof Map<?, ?>) {
+ return mapEquals((Map<?, ?>) left, (Map<?, ?>) right);
+ }
+ if (left.getClass().isArray() && right.getClass().isArray()) {
+ int length = Array.getLength(left);
+ if (length != Array.getLength(right)) {
+ return false;
+ }
+ for (int i = 0; i < length; i++) {
+ if (!valueEquals(Array.get(left, i), Array.get(right, i))) {
+ return false;
+ }
+ }
+ return true;
+ }
+ return left.equals(right);
+ }
+
+ private static boolean numberEquals(Number left, Number right) {
+ if (left instanceof Float
+ || left instanceof Double
+ || right instanceof Float
+ || right instanceof Double) {
+ return Double.compare(left.doubleValue(), right.doubleValue()) ==
0;
+ }
+ return new BigDecimal(left.toString()).compareTo(new
BigDecimal(right.toString())) == 0;
+ }
+
+ private static boolean mapEquals(Map<?, ?> left, Map<?, ?> right) {
+ if (left.size() != right.size()) {
+ return false;
+ }
+ for (Map.Entry<?, ?> leftEntry : left.entrySet()) {
+ boolean found = false;
+ for (Map.Entry<?, ?> rightEntry : right.entrySet()) {
+ if (valueEquals(leftEntry.getKey(), rightEntry.getKey())) {
+ if (!valueEquals(leftEntry.getValue(),
rightEntry.getValue())) {
+ return false;
+ }
+ found = true;
+ break;
+ }
+ }
+ if (!found) {
+ return false;
+ }
+ }
+ return true;
+ }
+
/**
* Accesses an element from an ARRAY by index (1-based, SQL standard).
*
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
index abe2fb8c2..4caef2768 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
@@ -234,6 +234,10 @@ public class JaninoCompiler {
udfFunction.get(),
translateOperands(context, sqlBasicCall).toArray(new
Java.Rvalue[0]));
}
+ if (isCollectionConstructor(sqlBasicCall)) {
+ return generateCollectionConstructorOperation(
+ context, sqlBasicCall,
getCollectionConstructorFunctionName(sqlBasicCall));
+ }
if (isIntervalArithmetic(sqlBasicCall)) {
return generateIntervalArithmeticOperation(context, sqlBasicCall);
}
@@ -491,6 +495,131 @@ public class JaninoCompiler {
return new Java.MethodInvocation(Location.NOWHERE, null, functionName,
atoms);
}
+ private static Java.Rvalue generateCollectionConstructorOperation(
+ Context context, SqlBasicCall sqlBasicCall, String functionName) {
+ DataType resultType =
+ TransformParser.deduceSubExpressionType(
+ context.columns,
+ sqlBasicCall,
+ context.udfDescriptors,
+ context.supportedMetadataColumns);
+ return generateCollectionConstructorOperation(
+ context, sqlBasicCall, functionName, resultType);
+ }
+
+ private static Java.Rvalue generateCollectionConstructorOperation(
+ Context context, SqlBasicCall sqlBasicCall, String functionName,
DataType resultType) {
+ List<DataType> targetTypes = new ArrayList<>();
+ switch (sqlBasicCall.getKind()) {
+ case ARRAY_VALUE_CONSTRUCTOR:
+ for (int i = 0; i < sqlBasicCall.operandCount(); i++) {
+ targetTypes.add(resultType.getChildren().get(0));
+ }
+ break;
+ case MAP_VALUE_CONSTRUCTOR:
+ for (int i = 0; i < sqlBasicCall.operandCount(); i++) {
+ targetTypes.add(resultType.getChildren().get(i % 2));
+ }
+ break;
+ case ROW:
+ targetTypes.addAll(resultType.getChildren());
+ break;
+ default:
+ throw new ParseException("Unrecognized collection constructor:
" + sqlBasicCall);
+ }
+
+ List<SqlNode> operands = sqlBasicCall.getOperandList();
+ Java.Rvalue[] atoms = new Java.Rvalue[operands.size()];
+ for (int i = 0; i < operands.size(); i++) {
+ SqlNode operand = operands.get(i);
+ DataType targetType = targetTypes.get(i);
+ if (operand instanceof SqlBasicCall
+ && isCollectionConstructor((SqlBasicCall) operand)) {
+ SqlBasicCall nestedConstructor = (SqlBasicCall) operand;
+ atoms[i] =
+ generateCollectionConstructorOperation(
+ context,
+ nestedConstructor,
+
getCollectionConstructorFunctionName(nestedConstructor),
+ targetType);
+ } else {
+ atoms[i] = translateSqlNodeToJaninoRvalue(context, operand);
+ if (!(operand instanceof SqlLiteral) || ((SqlLiteral)
operand).getValue() != null) {
+ DataType operandType =
+ TransformParser.deduceSubExpressionType(
+ context.columns,
+ operand,
+ context.udfDescriptors,
+ context.supportedMetadataColumns);
+ atoms[i] = generateImplicitTypeConvertMethod(operandType,
targetType, atoms[i]);
+ }
+ }
+ }
+ return generateFunctionOperation(functionName, atoms);
+ }
+
+ private static boolean isCollectionConstructor(SqlBasicCall sqlBasicCall) {
+ return sqlBasicCall.getKind() == SqlKind.ARRAY_VALUE_CONSTRUCTOR
+ || sqlBasicCall.getKind() == SqlKind.MAP_VALUE_CONSTRUCTOR
+ || sqlBasicCall.getKind() == SqlKind.ROW;
+ }
+
+ private static String getCollectionConstructorFunctionName(SqlBasicCall
sqlBasicCall) {
+ switch (sqlBasicCall.getKind()) {
+ case ARRAY_VALUE_CONSTRUCTOR:
+ return "array";
+ case MAP_VALUE_CONSTRUCTOR:
+ return "map";
+ case ROW:
+ return "row";
+ default:
+ throw new ParseException("Unrecognized collection constructor:
" + sqlBasicCall);
+ }
+ }
+
+ private static Java.Rvalue generateImplicitTypeConvertMethod(
+ DataType sourceType, DataType targetType, Java.Rvalue atom) {
+ if (sourceType.getTypeRoot() == targetType.getTypeRoot()
+ && (!(sourceType instanceof DecimalType)
+ ||
sourceType.copy(true).equals(targetType.copy(true)))) {
+ return atom;
+ }
+ switch (targetType.getTypeRoot()) {
+ case BOOLEAN:
+ return generateFunctionOperation("castToBoolean", new
Java.Rvalue[] {atom});
+ case TINYINT:
+ return generateFunctionOperation("castToByte", new
Java.Rvalue[] {atom});
+ case SMALLINT:
+ return generateFunctionOperation("castToShort", new
Java.Rvalue[] {atom});
+ case INTEGER:
+ return generateFunctionOperation("castToInteger", new
Java.Rvalue[] {atom});
+ case BIGINT:
+ return generateFunctionOperation("castToLong", new
Java.Rvalue[] {atom});
+ case FLOAT:
+ return generateFunctionOperation("castToFloat", new
Java.Rvalue[] {atom});
+ case DOUBLE:
+ return generateFunctionOperation("castToDouble", new
Java.Rvalue[] {atom});
+ case DECIMAL:
+ DecimalType decimalType = (DecimalType) targetType;
+ return generateFunctionOperation(
+ "castToBigDecimal",
+ new Java.Rvalue[] {
+ atom,
+ new Java.AmbiguousName(
+ Location.NOWHERE,
+ new String[]
{String.valueOf(decimalType.getPrecision())}),
+ new Java.AmbiguousName(
+ Location.NOWHERE,
+ new String[]
{String.valueOf(decimalType.getScale())})
+ });
+ case CHAR:
+ case VARCHAR:
+ return generateFunctionOperation("castToString", new
Java.Rvalue[] {atom});
+ default:
+ return atom;
+ }
+ }
+
private static Java.Rvalue generateLazyBinaryFunctionOperation(
Context context, SqlBasicCall sqlBasicCall, String functionName,
Java.Rvalue[] atoms) {
if (atoms.length != 2) {
@@ -811,7 +940,11 @@ public class JaninoCompiler {
Java.Rvalue methodInvocation =
new Java.MethodInvocation(Location.NOWHERE, null,
"itemAccess", atoms);
- // Deduce the return type and add a cast to ensure proper type
conversion
+ return castExpressionToInferredType(context, sqlBasicCall,
methodInvocation);
+ }
+
+ private static Java.Rvalue castExpressionToInferredType(
+ Context context, SqlBasicCall sqlBasicCall, Java.Rvalue
expression) {
DataType resultType =
TransformParser.deduceSubExpressionType(
context.columns,
@@ -821,7 +954,7 @@ public class JaninoCompiler {
// Get the Java class for the result type and add a cast
// Use getCanonicalName() to correctly handle array types (e.g.,
byte[] instead of "[B")
- return castToJavaType(resultType, methodInvocation);
+ return castToJavaType(resultType, expression);
}
private static Java.Rvalue castToJavaType(DataType resultType, Java.Rvalue
expression) {
@@ -859,6 +992,9 @@ public class JaninoCompiler {
return generateIfNullOperation(context, sqlBasicCall, atoms);
} else if (operationName.equals("NULLIF")) {
return generateNullIfOperation(context, sqlBasicCall, atoms);
+ } else if (operationName.equals("ELEMENT")) {
+ return castExpressionToInferredType(
+ context, sqlBasicCall,
generateFunctionOperation("element", atoms));
} else {
if (isAiFunction(operationName) && atoms.length >= 1) {
if (!(sqlBasicCall.operand(0) instanceof
SqlCharStringLiteral)) {
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
index 8276d323c..5fcf1b608 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
@@ -66,7 +66,9 @@ import org.apache.calcite.sql.type.SqlReturnTypeInference;
import org.apache.calcite.sql.type.SqlTypeFactoryImpl;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.calcite.sql.util.SqlOperatorTables;
+import org.apache.calcite.sql.validate.SqlConformance;
import org.apache.calcite.sql.validate.SqlConformanceEnum;
+import org.apache.calcite.sql.validate.SqlDelegatingConformance;
import org.apache.calcite.sql.validate.SqlValidator;
import org.apache.calcite.sql.validate.SqlValidatorUtil;
import org.apache.calcite.sql2rel.SqlToRelConverter;
@@ -99,12 +101,55 @@ public class TransformParser {
private static final String DEFAULT_TABLE = "TB";
private static final String MAPPED_COLUMN_NAME_PREFIX = "$";
private static final String MAPPED_SINGLE_COLUMN_NAME =
MAPPED_COLUMN_NAME_PREFIX + "0";
+ private static final SqlConformance TRANSFORM_SQL_CONFORMANCE = new
TransformSqlConformance();
+
+ private static class TransformSqlConformance extends
SqlDelegatingConformance {
+
+ private TransformSqlConformance() {
+ super(SqlConformanceEnum.MYSQL_5);
+ }
+
+ @Override
+ public boolean allowCharLiteralAlias() {
+ return SqlConformanceEnum.MYSQL_5.allowCharLiteralAlias();
+ }
+
+ @Override
+ public boolean allowExplicitRowValueConstructor() {
+ return true;
+ }
+
+ @Override
+ public boolean isLimitStartCountAllowed() {
+ return SqlConformanceEnum.MYSQL_5.isLimitStartCountAllowed();
+ }
+
+ @Override
+ public boolean isPercentRemainderAllowed() {
+ return SqlConformanceEnum.MYSQL_5.isPercentRemainderAllowed();
+ }
+
+ @Override
+ public boolean allowGeometry() {
+ return SqlConformanceEnum.MYSQL_5.allowGeometry();
+ }
+
+ @Override
+ public boolean shouldConvertRaggedUnionTypesToVarying() {
+ return
SqlConformanceEnum.MYSQL_5.shouldConvertRaggedUnionTypesToVarying();
+ }
+
+ @Override
+ public boolean allowExtendedTrim() {
+ return SqlConformanceEnum.MYSQL_5.allowExtendedTrim();
+ }
+ }
private static SqlParser getCalciteParser(String sql) {
return SqlParser.create(
TransformSqlSyntaxRewriter.rewriteTryCast(sql),
SqlParser.Config.DEFAULT
- .withConformance(SqlConformanceEnum.MYSQL_5)
+ .withConformance(TRANSFORM_SQL_CONFORMANCE)
.withCaseSensitive(true)
.withLex(Lex.JAVA));
}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java
index ff6ca6803..0c37a3d0d 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java
@@ -685,8 +685,30 @@ public class TransformSqlOperatorTable extends
ReflectiveSqlOperatorTable {
// ---------------------
// Struct Functions
// ---------------------
+ public static final SqlOperator ARRAY =
SqlStdOperatorTable.ARRAY_VALUE_CONSTRUCTOR;
+ public static final SqlOperator MAP =
SqlStdOperatorTable.MAP_VALUE_CONSTRUCTOR;
+ public static final SqlOperator ROW = SqlStdOperatorTable.ROW;
+
// Supports accessing elements of ARRAY[index], ROW[index], MAP[key], and
VARIANT[index/key]
public static final SqlOperator ITEM = new VariantAwareItemOperator();
+ public static final SqlFunction CARDINALITY =
SqlStdOperatorTable.CARDINALITY;
+ public static final SqlFunction ELEMENT = SqlStdOperatorTable.ELEMENT;
+ public static final SqlFunction ARRAY_CONTAINS =
+ new SqlFunction(
+ "ARRAY_CONTAINS",
+ SqlKind.OTHER_FUNCTION,
+ ReturnTypes.BOOLEAN_NULLABLE,
+ null,
+ OperandTypes.family(SqlTypeFamily.ARRAY,
SqlTypeFamily.ANY),
+ SqlFunctionCategory.SYSTEM);
+ public static final SqlFunction ARRAY_POSITION =
+ new SqlFunction(
+ "ARRAY_POSITION",
+ SqlKind.OTHER_FUNCTION,
+ ReturnTypes.INTEGER_NULLABLE,
+ null,
+ OperandTypes.family(SqlTypeFamily.ARRAY,
SqlTypeFamily.ANY),
+ SqlFunctionCategory.SYSTEM);
public static final SqlFunction AI_CHAT_PREDICT =
new SqlFunction(
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/typeutils/CalciteDataTypeConverter.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/typeutils/CalciteDataTypeConverter.java
index 012cfa704..4c2dde3f5 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/typeutils/CalciteDataTypeConverter.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/typeutils/CalciteDataTypeConverter.java
@@ -23,6 +23,7 @@ import org.apache.flink.cdc.common.types.BigIntType;
import org.apache.flink.cdc.common.types.BinaryType;
import org.apache.flink.cdc.common.types.BooleanType;
import org.apache.flink.cdc.common.types.CharType;
+import org.apache.flink.cdc.common.types.DataField;
import org.apache.flink.cdc.common.types.DataType;
import org.apache.flink.cdc.common.types.DataTypes;
import org.apache.flink.cdc.common.types.DateType;
@@ -44,6 +45,8 @@ import org.apache.flink.cdc.common.types.ZonedTimestampType;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.sql.SqlUtil;
import org.apache.calcite.sql.type.SqlTypeName;
import java.util.List;
@@ -339,9 +342,25 @@ public class CalciteDataTypeConverter {
case VARIANT:
return DataTypes.VARIANT();
case ROW:
+ return DataTypes.ROW(
+ relDataType.getFieldList().stream()
+ .map(
+ field ->
+ DataTypes.FIELD(
+
normalizeRowFieldName(field),
+
convertCalciteRelDataTypeToDataType(
+
field.getType())))
+ .toArray(DataField[]::new));
default:
throw new UnsupportedOperationException(
"Unsupported type: " + relDataType.getSqlTypeName());
}
}
+
+ private static String normalizeRowFieldName(RelDataTypeField field) {
+ if
(field.getName().equals(SqlUtil.deriveAliasFromOrdinal(field.getIndex()))) {
+ return "f" + field.getIndex();
+ }
+ return field.getName();
+ }
}
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/StructFunctionsTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/StructFunctionsTest.java
index 4d9510451..84ec9e3e9 100644
---
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/StructFunctionsTest.java
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/StructFunctionsTest.java
@@ -31,10 +31,75 @@ import java.util.List;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Unit tests for {@link StructFunctions}. */
class StructFunctionsTest {
+ @Nested
+ class CollectionFunctionTests {
+
+ @Test
+ void testCollectionConstructors() {
+ assertThat(StructFunctions.array(1, "two",
null)).containsExactly(1, "two", null);
+ assertThat(StructFunctions.row(1, "two", null)).containsExactly(1,
"two", null);
+ assertThat(StructFunctions.map("one", 1, "two", 2, "one", 3))
+ .containsExactlyInAnyOrderEntriesOf(Map.of("one", 3,
"two", 2));
+ }
+
+ @Test
+ void testInvalidMapConstructor() {
+ assertThatThrownBy(() -> StructFunctions.map())
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> StructFunctions.map("one", 1, "two"))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ void testCardinality() {
+ assertThat(StructFunctions.cardinality(Arrays.asList(1, 2,
3))).isEqualTo(3);
+ assertThat(StructFunctions.cardinality(Map.of("one", 1, "two",
2))).isEqualTo(2);
+ assertThat(StructFunctions.cardinality((List<?>) null)).isNull();
+ assertThat(StructFunctions.cardinality((Map<?, ?>) null)).isNull();
+ }
+
+ @Test
+ void testArrayContains() {
+ List<Object> array =
+ Arrays.asList(
+ 1, null, new byte[] {1, 2}, Arrays.asList(2,
Map.of("three", 3L)));
+
+ assertThat(StructFunctions.arrayContains(array, 1L)).isTrue();
+ assertThat(StructFunctions.arrayContains(array, null)).isTrue();
+ assertThat(StructFunctions.arrayContains(array, new byte[] {1,
2})).isTrue();
+ assertThat(StructFunctions.arrayContains(array, Arrays.asList(2L,
Map.of("three", 3))))
+ .isTrue();
+ assertThat(StructFunctions.arrayContains(array, 4)).isFalse();
+ assertThat(StructFunctions.arrayContains(null, 1)).isNull();
+ }
+
+ @Test
+ void testArrayPosition() {
+ List<Object> array = Arrays.asList(1, null, 2L, 1);
+
+ assertThat(StructFunctions.arrayPosition(array, 1L)).isEqualTo(1);
+ assertThat(StructFunctions.arrayPosition(array, 2)).isEqualTo(3);
+ assertThat(StructFunctions.arrayPosition(array, 3)).isZero();
+ assertThat(StructFunctions.arrayPosition(array, null)).isNull();
+ assertThat(StructFunctions.arrayPosition(null, 1)).isNull();
+ }
+
+ @Test
+ void testElement() {
+ assertThat((Object) StructFunctions.element(null)).isNull();
+ assertThat((Object)
StructFunctions.element(Collections.emptyList())).isNull();
+
assertThat(StructFunctions.element(Collections.singletonList("one"))).isEqualTo("one");
+ assertThatThrownBy(() ->
StructFunctions.element(Arrays.asList("one", "two")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Array has more than one element.");
+ }
+ }
+
// ========================================
// List (ARRAY) Access Tests
// ========================================
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
index 8efde2e38..621188bbb 100644
---
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
@@ -136,6 +136,116 @@ class TransformParserTest {
Assertions.assertThat(parse.getWhere()).hasToString("`uniq_id` > 10
AND `id` IS NOT NULL");
}
+ @Test
+ void testCollectionConstructorsAndFunctions() {
+ List<Column> columns =
+ List.of(
+ Column.physicalColumn("id", DataTypes.BIGINT()),
+ Column.physicalColumn("arr",
DataTypes.ARRAY(DataTypes.INT())),
+ Column.physicalColumn(
+ "m", DataTypes.MAP(DataTypes.STRING(),
DataTypes.INT())));
+
+ List<ProjectionColumn> projectionColumns =
+ TransformParser.generateProjectionColumns(
+ "ARRAY[1, id] AS array_col, "
+ + "MAP['one', 1, 'id', id] AS map_col, "
+ + "ROW(id, arr) AS row_col, "
+ + "CARDINALITY(arr) AS array_size, "
+ + "CARDINALITY(m) AS map_size, "
+ + "ARRAY_CONTAINS(arr, 2) AS contains_two, "
+ + "ARRAY_POSITION(arr, 2) AS position_two, "
+ + "ELEMENT(ARRAY[id]) AS only_element, "
+ + "ARRAY[ARRAY[1], ARRAY[id]] AS nested_array",
+ columns,
+ Collections.emptyList(),
+ new SupportedMetadataColumn[0]);
+
+ Assertions.assertThat(projectionColumns)
+ .extracting(ProjectionColumn::getDataType)
+ .containsExactly(
+ DataTypes.ARRAY(DataTypes.BIGINT()),
+ DataTypes.MAP(DataTypes.STRING(), DataTypes.BIGINT()),
+ DataTypes.ROW(DataTypes.BIGINT(),
DataTypes.ARRAY(DataTypes.INT())),
+ DataTypes.INT(),
+ DataTypes.INT(),
+ DataTypes.BOOLEAN(),
+ DataTypes.INT(),
+ DataTypes.BIGINT(),
+ DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.BIGINT())));
+ Assertions.assertThat(projectionColumns)
+ .extracting(ProjectionColumn::getScriptExpression)
+ .containsExactly(
+ "array(castToLong(1), $0)",
+ "map(\"one\", castToLong(1), \"id\", $0)",
+ "row($0, $1)",
+ "cardinality($0)",
+ "cardinality($0)",
+ "arrayContains($0, 2)",
+ "arrayPosition($0, 2)",
+ "(java.lang.Long) element(array($0))",
+ "array(array(castToLong(1)), array($0))");
+ }
+
+ @Test
+ void testCollectionConstructorWithIncompatibleElementTypes() {
+ Assertions.assertThatThrownBy(
+ () ->
+ TransformParser.generateProjectionColumns(
+ "ARRAY[1, TRUE] AS invalid_array",
+ Collections.emptyList(),
+ Collections.emptyList(),
+ new SupportedMetadataColumn[0]))
+ .isExactlyInstanceOf(CalciteContextException.class)
+ .hasMessageContaining("Parameters must be of the same type");
+ }
+
+ @Test
+ void testPreservesNamedRowFieldsInCollectionTypes() {
+ DataType namedRowType =
+ DataTypes.ROW(
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("length", DataTypes.INT()));
+ List<Column> columns = List.of(Column.physicalColumn("complex_row_",
namedRowType));
+
+ List<ProjectionColumn> projectionColumns =
+ TransformParser.generateProjectionColumns(
+ "ELEMENT(ARRAY[complex_row_]) AS row_element, "
+ + "ARRAY[complex_row_] AS row_array, "
+ + "MAP['row', complex_row_] AS row_map",
+ columns,
+ Collections.emptyList(),
+ new SupportedMetadataColumn[0]);
+
+ Assertions.assertThat(projectionColumns)
+ .extracting(ProjectionColumn::getDataType)
+ .containsExactly(
+ namedRowType,
+ DataTypes.ARRAY(namedRowType),
+ DataTypes.MAP(DataTypes.STRING(), namedRowType));
+ }
+
+ @Test
+ void testTranslateCollectionFunctionsToJaninoExpression() {
+ List<Column> columns =
+ List.of(
+ Column.physicalColumn("id", DataTypes.BIGINT()),
+ Column.physicalColumn("arr",
DataTypes.ARRAY(DataTypes.INT())),
+ Column.physicalColumn(
+ "m", DataTypes.MAP(DataTypes.STRING(),
DataTypes.INT())));
+
+ testFilterExpressionWithColumns("CARDINALITY(arr)",
"cardinality(arr)", columns);
+ testFilterExpressionWithColumns("ARRAY_CONTAINS(arr, 2)",
"arrayContains(arr, 2)", columns);
+ testFilterExpressionWithColumns("ARRAY_POSITION(arr, 2)",
"arrayPosition(arr, 2)", columns);
+ testFilterExpressionWithColumns(
+ "ELEMENT(ARRAY[id])", "(java.lang.Long) element(array(id))",
columns);
+ testFilterExpressionWithColumns(
+ "MAP['one', 1]['one']",
+ "(java.lang.Integer) itemAccess(map(\"one\", 1), \"one\")",
+ columns);
+ testFilterExpressionWithColumns(
+ "ROW(id, arr)[2]", "(java.util.List) itemAccess(row(id, arr),
2)", columns);
+ }
+
@Test
void testTransformCalciteValidate() {
SqlSelect parse =
