From 9998b217fa230349b269edc79ae4802a03637452 Mon Sep 17 00:00:00 2001
From: Au_Miner <358671982@qq.com>
Date: Thu, 11 Jun 2026 11:26:40 +0800
Subject: [PATCH] [FLINK-35275][table] Fix ArrayContainsFunction using
 incorrect element type from needle argument

---
 .../planner/functions/CollectionFunctionsITCase.java     | 9 +++++++++
 .../runtime/functions/scalar/ArrayContainsFunction.java  | 8 ++++++--
 2 files changed, 15 insertions(+), 2 deletions(-)

diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CollectionFunctionsITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CollectionFunctionsITCase.java
index f67d422c994a6..360ec6370cabb 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CollectionFunctionsITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CollectionFunctionsITCase.java
@@ -174,6 +174,11 @@ private Stream<TestSetSpec> arrayContainsTestCases() {
                                 "ARRAY_CONTAINS(f4, NULL)",
                                 true,
                                 DataTypes.BOOLEAN().nullable())
+                        .testResult(
+                                $("f4").arrayContains(lit(0, DataTypes.INT().notNull())),
+                                "ARRAY_CONTAINS(f4, 0)",
+                                false,
+                                DataTypes.BOOLEAN().nullable())
                         .testResult(
                                 $("f5").arrayContains(lit(null, DataTypes.INT())),
                                 "ARRAY_CONTAINS(f5, CAST(NULL AS INT))",
@@ -189,6 +194,10 @@ private Stream<TestSetSpec> arrayContainsTestCases() {
                                 "ARRAY_CONTAINS(f5, 3)",
                                 true,
                                 DataTypes.BOOLEAN().notNull())
+                        .testSqlResult(
+                                "ARRAY_CONTAINS(ARRAY[1, NULL], 0)",
+                                false,
+                                DataTypes.BOOLEAN().notNull())
                         // invalid signatures
                         .testSqlValidationError(
                                 "ARRAY_CONTAINS(f0, TRUE)",
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayContainsFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayContainsFunction.java
index 03669ce93ac27..7e263803e3d87 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayContainsFunction.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ArrayContainsFunction.java
@@ -25,6 +25,7 @@
 import org.apache.flink.table.functions.FunctionContext;
 import org.apache.flink.table.functions.SpecializedFunction.ExpressionEvaluator;
 import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.types.CollectionDataType;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.util.FlinkRuntimeException;
 
@@ -44,13 +45,16 @@ public class ArrayContainsFunction extends BuiltInScalarFunction {
 
     public ArrayContainsFunction(SpecializedContext context) {
         super(BuiltInFunctionDefinitions.ARRAY_CONTAINS, context);
+        final DataType elementDataType =
+                ((CollectionDataType) context.getCallContext().getArgumentDataTypes().get(0))
+                        .getElementDataType();
         final DataType needleDataType = context.getCallContext().getArgumentDataTypes().get(1);
-        elementGetter = ArrayData.createElementGetter(needleDataType.getLogicalType());
+        elementGetter = ArrayData.createElementGetter(elementDataType.getLogicalType());
         equalityEvaluator =
                 context.createEvaluator(
                         $("element").isEqual($("needle")),
                         DataTypes.BOOLEAN(),
-                        DataTypes.FIELD("element", needleDataType.notNull().toInternal()),
+                        DataTypes.FIELD("element", elementDataType.notNull().toInternal()),
                         DataTypes.FIELD("needle", needleDataType.notNull().toInternal()));
     }