[FLINK-39010][transform] YAML Pipeline supports item subscription of Variant

Hisoka-X · Hisoka-X · commit 856eb7f9563b · 2026-02-05T11:13:27.000+08:00
diff --git a/flink-cdc-composer/src/test/resources/specs/nested.yaml b/flink-cdc-composer/src/test/resources/specs/nested.yaml
@@ -35,7 +35,7 @@
     id_
     PARSE_JSON(id_) AS variant_id_
     PARSE_JSON('[' || id_ || ',' || bool_ || ']') AS variant_array_
-    PARSE_JSON('{\"id\": ' || id_ || ', \"name\": \"' || string_ || '\"}') AS variant_doc_
+    PARSE_JSON('{"id": ' || id_ || ', "name": "' || string_ || '"}') AS variant_doc_
   primary-key: id_
   expect: |-
     CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT NULL 'Identifier',`variant_id_` VARIANT,`variant_array_` VARIANT,`variant_doc_` VARIANT}, primaryKeys=id_, options=()}
@@ -55,7 +55,7 @@
     id_
     TRY_PARSE_JSON(id_) AS variant_id_
     TRY_PARSE_JSON('[' || id_ || ',' || bool_ || ']') AS variant_array_
-    TRY_PARSE_JSON('{\"id\": ' || id_ || ', \"name\": \"' || string_ || '\"}') AS variant_doc_
+    TRY_PARSE_JSON('{"id": ' || id_ || ', "name": "' || string_ || '"}') AS variant_doc_
   primary-key: id_
   expect: |-
     CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT NULL 'Identifier',`variant_id_` VARIANT,`variant_array_` VARIANT,`variant_doc_` VARIANT}, primaryKeys=id_, options=()}
@@ -79,13 +79,13 @@
 - do: Parse Variant from JSON with duplicate key rejected
   projection: |-
     id_
-    PARSE_JSON('{\"id\": 1, \"id\": 2}') AS variant_
+    PARSE_JSON('{"id": 1, "id": 2}') AS variant_
   primary-key: id_
   expect-error: 'org.apache.flink.cdc.common.types.variant.VariantTypeException: VARIANT_DUPLICATE_KEY'
 - do: Parse Variant from JSON with duplicate key allowed
   projection: |-
     id_
-    PARSE_JSON('{\"id\": 1, \"id\": 2}', TRUE) AS variant_
+    PARSE_JSON('{"id": 1, "id": 2}', TRUE) AS variant_
   primary-key: id_
   expect: |-
     CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT NULL 'Identifier',`variant_` VARIANT}, primaryKeys=id_, options=()}
@@ -97,7 +97,7 @@
 - do: Try Parse Variant from JSON with duplicate key rejected
   projection: |-
     id_
-    TRY_PARSE_JSON('{\"id\": 1, \"id\": 2}') AS variant_
+    TRY_PARSE_JSON('{"id": 1, "id": 2}') AS variant_
   primary-key: id_
   expect: |-
     CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT NULL 'Identifier',`variant_` VARIANT}, primaryKeys=id_, options=()}
@@ -109,7 +109,7 @@
 - do: Try Parse Variant from JSON with duplicate key allowed
   projection: |-
     id_
-    TRY_PARSE_JSON('{\"id\": 1, \"id\": 2}', TRUE) AS variant_
+    TRY_PARSE_JSON('{"id": 1, "id": 2}', TRUE) AS variant_
   primary-key: id_
   expect: |-
     CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT NULL 'Identifier',`variant_` VARIANT}, primaryKeys=id_, options=()}
@@ -283,4 +283,83 @@
   expect: |-
     CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT NULL 'Identifier',`complex_row_` ROW<`name` STRING, `length` INT>}, primaryKeys=id_, options=()}
     DataChangeEvent{tableId=foo.bar.baz, before=[1, {name: STRING -> Alice, length: INT -> 5}], after=[-1, {name: STRING -> Derrida, length: INT -> 7}], op=UPDATE, meta=()}
-    DataChangeEvent{tableId=foo.bar.baz, before=[-1, {name: STRING -> Derrida, length: INT -> 7}], after=[], op=DELETE, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[-1, {name: STRING -> Derrida, length: INT -> 7}], after=[], op=DELETE, meta=()}
+- do: Variant Object Subscripting With String Key
+  projection: |-
+    id_
+    variant_
+    variant_['k'] AS variant_k
+  primary-key: id_
+  expect: |-
+    CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT NULL 'Identifier',`variant_` VARIANT,`variant_k` VARIANT}, primaryKeys=id_, options=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, {"array":[1,2,{"kk":1.123}],"k":1,"object":{"k":"hello"}}, 1], op=INSERT, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[1, {"array":[1,2,{"kk":1.123}],"k":1,"object":{"k":"hello"}}, 1], after=[-1, [{"k":1},"hello",{"k":2}], null], op=UPDATE, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[-1, [{"k":1},"hello",{"k":2}], null], after=[], op=DELETE, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, null], op=INSERT, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[0, null, null], after=[], op=DELETE, meta=()}
+- do: Variant Array Subscripting With Integer Index
+  projection: |-
+    id_
+    variant_
+    variant_[1] AS variant_first
+    variant_[2] AS variant_second
+  primary-key: id_
+  expect: |-
+    CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT NULL 'Identifier',`variant_` VARIANT,`variant_first` VARIANT,`variant_second` VARIANT}, primaryKeys=id_, options=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, {"array":[1,2,{"kk":1.123}],"k":1,"object":{"k":"hello"}}, null, null], op=INSERT, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[1, {"array":[1,2,{"kk":1.123}],"k":1,"object":{"k":"hello"}}, null, null], after=[-1, [{"k":1},"hello",{"k":2}], {"k":1}, "hello"], op=UPDATE, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[-1, [{"k":1},"hello",{"k":2}], {"k":1}, "hello"], after=[], op=DELETE, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, null, null], op=INSERT, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[0, null, null, null], after=[], op=DELETE, meta=()}
+- do: Variant Nested Subscripting
+  projection: |-
+    id_
+    variant_['object']['k'] AS nested_object_k
+    variant_['array'][1] AS nested_array_first
+  primary-key: id_
+  expect: |-
+    CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT NULL 'Identifier',`nested_object_k` VARIANT,`nested_array_first` VARIANT}, primaryKeys=id_, options=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, "hello", 1], op=INSERT, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[1, "hello", 1], after=[-1, null, null], op=UPDATE, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[-1, null, null], after=[], op=DELETE, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, null], op=INSERT, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[0, null, null], after=[], op=DELETE, meta=()}
+- do: Variant Subscripting With Absent Key
+  projection: |-
+    id_
+    variant_['nonexistent'] AS absent_key
+    variant_[1000] AS absent_index
+  primary-key: id_
+  expect: |-
+    CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT NULL 'Identifier',`absent_key` VARIANT,`absent_index` VARIANT}, primaryKeys=id_, options=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, null, null], op=INSERT, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[1, null, null], after=[-1, null, null], op=UPDATE, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[-1, null, null], after=[], op=DELETE, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, null], op=INSERT, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[0, null, null], after=[], op=DELETE, meta=()}
+- do: Variant Subscripting With Parsed JSON
+  projection: |-
+    id_
+    PARSE_JSON('[1, 2, 3]')[2] AS array_second
+    PARSE_JSON('[10, 20, 30]')[1] AS array_first
+  primary-key: id_
+  expect: |-
+    CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT NULL 'Identifier',`array_second` VARIANT,`array_first` VARIANT}, primaryKeys=id_, options=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 2, 10], op=INSERT, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[1, 2, 10], after=[-1, 2, 10], op=UPDATE, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[-1, 2, 10], after=[], op=DELETE, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, 2, 10], op=INSERT, meta=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[0, 2, 10], after=[], op=DELETE, meta=()}
+- do: Variant Subscripting With Parsed JSON Object
+  projection: |-
+    id_
+    PARSE_JSON('{"key": "value"}')['key'] AS json_key
+    PARSE_JSON('{"nested": {"inner": 42}}')['nested']['inner'] AS nested_value
+  primary-key: id_
+  expect: |-
+    CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT NULL 'Identifier',`json_key` VARIANT,`nested_value` VARIANT}, primaryKeys=id_, options=()}
+    DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, "value", 42], op=INSERT, meta=()}
+    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=()}
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
@@ -17,14 +17,16 @@
 
 package org.apache.flink.cdc.runtime.functions.impl;
 
+import org.apache.flink.cdc.common.types.variant.Variant;
+
 import java.util.List;
 import java.util.Map;
 
 /**
  * Built-in functions for collection and struct data types.
  *
- * <p>These functions support accessing elements from collections (ARRAY, MAP) and structured data
- * types (ROW).
+ * <p>These functions support accessing elements from collections (ARRAY, MAP), structured data
+ * types (ROW), and semi-structured data types (VARIANT).
  */
 public class StructFunctions {
 
@@ -67,4 +69,49 @@ public static <K, V> V itemAccess(Map<K, V> map, K key) {
         }
         return map.get(key);
     }
+
+    /**
+     * Accesses an element from a VARIANT array by index (1-based, SQL standard).
+     *
+     * <p>variant[1] returns the first element.
+     *
+     * @param variant the variant (must be an array) to access
+     * @param index the 1-based index
+     * @return the element at the specified index as a Variant, or null if the variant is not an
+     *     array or index is out of bounds
+     */
+    public static Variant itemAccess(Variant variant, Integer index) {
+        if (variant == null || index == null) {
+            return null;
+        }
+        if (!variant.isArray()) {
+            return null;
+        }
+        // Convert 1-based index to 0-based (SQL standard uses 1-based indexing)
+        int zeroBasedIndex = index - 1;
+        if (zeroBasedIndex < 0 || zeroBasedIndex >= variant.arraySize()) {
+            return null;
+        }
+        return variant.getElement(zeroBasedIndex);
+    }
+
+    /**
+     * Accesses a field from a VARIANT object by field name.
+     *
+     * <p>variant['fieldName'] returns the value of the specified field.
+     *
+     * @param variant the variant (must be an object) to access
+     * @param fieldName the name of the field to look up
+     * @return the field value as a Variant, or null if the variant is not an object or field is not
+     *     found
+     */
+    public static Variant itemAccess(Variant variant, String fieldName) {
+        if (variant == null || fieldName == null) {
+            return null;
+        }
+        if (!variant.isObject()) {
+            return null;
+        }
+        return variant.getField(fieldName);
+    }
 }
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
@@ -174,7 +174,10 @@ private static Java.Rvalue translateSqlSqlLiteral(Context context, SqlLiteral sq
         Object value = sqlLiteral.getValue();
         if (sqlLiteral instanceof SqlCharStringLiteral) {
             // Double quotation marks represent strings in Janino.
-            value = "\"" + sqlLiteral.getValueAs(NlsString.class).getValue() + "\"";
+            // Escape backslashes first, then double quotes for proper Janino string literals.
+            String stringValue = sqlLiteral.getValueAs(NlsString.class).getValue();
+            stringValue = stringValue.replace("\\", "\\\\").replace("\"", "\\\"");
+            value = "\"" + stringValue + "\"";
         } else if (sqlLiteral instanceof SqlNumericLiteral) {
             if (((SqlNumericLiteral) sqlLiteral).isInteger()) {
                 long longValue = sqlLiteral.longValue(true);
@@ -478,6 +481,7 @@ private static Java.Rvalue generateItemAccessOperation(
                 "Expecting item accessing call %s to have 2 operands, got %s actually",
                 sqlBasicCall,
                 List.of(atoms));
+
         Java.Rvalue methodInvocation =
                 new Java.MethodInvocation(Location.NOWHERE, null, "itemAccess", atoms);
 
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
@@ -401,8 +401,8 @@ public SqlSyntax getSyntax() {
     // ---------------------
     // Struct Functions
     // ---------------------
-    // Supports accessing elements of ARRAY[index], ROW[index] and MAP[key]
-    public static final SqlOperator ITEM = SqlStdOperatorTable.ITEM;
+    // 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 AI_CHAT_PREDICT =
             new SqlFunction(
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/VariantAwareItemOperator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/VariantAwareItemOperator.java
@@ -0,0 +1,113 @@
+/*
+ * 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.flink.cdc.runtime.parser.metadata;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.sql.SqlCallBinding;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperandCountRange;
+import org.apache.calcite.sql.SqlOperatorBinding;
+import org.apache.calcite.sql.SqlSpecialOperator;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.sql.type.SqlTypeUtil;
+import org.apache.calcite.sql.validate.SqlValidator;
+
+/**
+ * A variant-aware implementation of the ITEM operator.
+ *
+ * <p>This extends the standard Calcite ITEM operator to support VARIANT type. For ARRAY, MAP, and
+ * ROW types, it delegates to the standard Calcite ITEM operator.
+ *
+ * <p>When accessing elements of a VARIANT:
+ *
+ * <ul>
+ *   <li>VARIANT[INTEGER] - accesses array elements (1-based index)
+ *   <li>VARIANT[STRING] - accesses object fields by name
+ * </ul>
+ *
+ * <p>The return type is always VARIANT when the source is VARIANT.
+ */
+public class VariantAwareItemOperator extends SqlSpecialOperator {
+
+    public VariantAwareItemOperator() {
+        super("ITEM", SqlKind.ITEM, 100, true, null, null, null);
+    }
+
+    @Override
+    public SqlOperandCountRange getOperandCountRange() {
+        return SqlStdOperatorTable.ITEM.getOperandCountRange();
+    }
+
+    @Override
+    public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFailure) {
+        final SqlValidator validator = callBinding.getValidator();
+        final RelDataType operandType =
+                validator.deriveType(callBinding.getScope(), callBinding.operand(0));
+        final SqlTypeName typeName = operandType.getSqlTypeName();
+
+        // Only handle VARIANT specially, delegate others to standard ITEM
+        if (typeName == SqlTypeName.VARIANT) {
+            final RelDataType keyType =
+                    validator.deriveType(callBinding.getScope(), callBinding.operand(1));
+            // VARIANT accepts integer or string keys
+            boolean valid =
+                    SqlTypeUtil.isIntType(keyType)
+                            || SqlTypeUtil.isCharacter(keyType)
+                            || keyType.getSqlTypeName() == SqlTypeName.ANY;
+            if (!valid && throwOnFailure) {
+                throw callBinding.newValidationSignatureError();
+            }
+            return valid;
+        }
+
+        // Delegate to standard ITEM for ARRAY, MAP, ROW, etc.
+        return SqlStdOperatorTable.ITEM.checkOperandTypes(callBinding, throwOnFailure);
+    }
+
+    @Override
+    public RelDataType inferReturnType(SqlOperatorBinding opBinding) {
+        final RelDataType operandType = opBinding.getOperandType(0);
+        final SqlTypeName typeName = operandType.getSqlTypeName();
+
+        // Only handle VARIANT specially
+        if (typeName == SqlTypeName.VARIANT) {
+            final RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
+            return typeFactory.createTypeWithNullability(
+                    typeFactory.createSqlType(SqlTypeName.VARIANT), true);
+        }
+
+        // Delegate to standard ITEM for other types
+        return SqlStdOperatorTable.ITEM.inferReturnType(opBinding);
+    }
+
+    @Override
+    public String getAllowedSignatures(String opName) {
+        // Include VARIANT in addition to standard signatures
+        return SqlStdOperatorTable.ITEM.getAllowedSignatures(opName)
+                + "\n<VARIANT>[<INTEGER>|<CHARACTER>]";
+    }
+
+    @Override
+    public void unparse(
+            SqlWriter writer, org.apache.calcite.sql.SqlCall call, int leftPrec, int rightPrec) {
+        SqlStdOperatorTable.ITEM.unparse(writer, call, leftPrec, rightPrec);
+    }
+}
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
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
