apache · VasShabu · Aug 10, 2026
diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
@@ -928,6 +928,23 @@ collection:
   - sql: MAP_FROM_ARRAYS(array_of_keys, array_of_values)
     table: mapFromArrays(array_of_keys, array_of_values)
     description: Returns a map created from an arrays of keys and values. Note that the lengths of two arrays should be the same.
+  - sql: MAP_FROM_ENTRIES(array_of_entries)
+    table: array.mapFromEntries()
+    description: |
+      Returns a map created from the given array of entries. Each entry must be a ROW with exactly
+      two fields, where the first field becomes the key and the second one the value. If there are
+      duplicate keys, the value of the last entry with that key wins. If the array itself or any
+      of its entries is null, null is returned.
+      eg.
+        -- {1=one, 2=two}
+        MAP_FROM_ENTRIES(ARRAY[ROW(1, 'one'), ROW(2, 'two')])
+
+        -- {1=uno, 2=two}
+        MAP_FROM_ENTRIES(ARRAY[ROW(1, 'one'), ROW(2, 'two'), ROW(1, 'uno')])
+
+        -- NULL
+        MAP_FROM_ENTRIES(ARRAY[ROW(1, 'one'), CAST(NULL AS ROW(k INT, v STRING))])
+
   - sql: SPLIT(string, delimiter)
     table: string.split(delimiter)
     description: Returns an array of substrings by splitting the input string based on the given delimiter. If the delimiter is not found in the string, the original string is returned as the only element in the array. If the delimiter is empty, every character in the string is split. If the string or delimiter is null, a null value is returned. If the delimiter is found at the beginning or end of the string, or there are contiguous delimiters, then an empty string is added to the array.

diff --git a/docs/data/sql_functions_zh.yml b/docs/data/sql_functions_zh.yml
@@ -1055,6 +1055,22 @@ collection:
   - sql: MAP_FROM_ARRAYS(array_of_keys, array_of_values)
     table: mapFromArrays(array_of_keys, array_of_values)
     description: 返回由 key 的数组 keys 和 value 的数组 values 创建的 map。请注意两个数组的长度应该相等。
+  - sql: MAP_FROM_ENTRIES(array_of_entries)
+    table: array.mapFromEntries()
+    description: |
+      Returns a map created from the given array of entries. Each entry must be a ROW with exactly
+      two fields, where the first field becomes the key and the second one the value. If there are
+      duplicate keys, the value of the last entry with that key wins. If the array itself or any
+      of its entries is null, null is returned.
+      eg.
+        -- {1=one, 2=two}
+        MAP_FROM_ENTRIES(ARRAY[ROW(1, 'one'), ROW(2, 'two')])
+
+        -- {1=uno, 2=two}
+        MAP_FROM_ENTRIES(ARRAY[ROW(1, 'one'), ROW(2, 'two'), ROW(1, 'uno')])
+
+        -- NULL
+        MAP_FROM_ENTRIES(ARRAY[ROW(1, 'one'), CAST(NULL AS ROW(k INT, v STRING))])
   - sql: MAP_UNION(map1, map2)
     table: map1.mapUnion(map2)
     description: 返回一个通过合并两个图 'map1' 和 'map2' 创建的图。这两个图应该具有共同的图类型。如果有重叠的键，'map2' 的值将覆盖 'map1' 的值。如果任一图为空，则返回 null。

diff --git a/flink-python/docs/reference/pyflink.table/expressions.rst b/flink-python/docs/reference/pyflink.table/expressions.rst
@@ -257,6 +257,7 @@ advanced type helper functions
     Expression.array_sort
     Expression.array_union
     Expression.map_entries
+    Expression.map_from_entries
     Expression.map_keys
     Expression.map_union
     Expression.map_values

diff --git a/flink-python/pyflink/table/expression.py b/flink-python/pyflink/table/expression.py
@@ -1966,6 +1966,22 @@ def map_entries(self) -> 'Expression':
         """
         return _unary_op("mapEntries")(self)
 
+    @property
+    def map_from_entries(self) -> 'Expression':
+        """
+        Returns a map created from the given array of entries. Each entry must be a row with
+        exactly two fields, where the first field becomes the key and the second one the value.
+        If there are duplicate keys, the value of the last entry with that key wins. If the array
+        itself or any of its entries is None, None is returned.
+
+        Examples:
+        ::
+
+            >>> array(row(1, "one"), row(2, "two")).map_from_entries # {1=one, 2=two}
+            >>> array(row(1, "one"), row(2, "two"), row(1, "uno")).map_from_entries # {1=uno, 2=two}
+        """
+        return _unary_op("mapFromEntries")(self)
+
     # ---------------------------- time definition functions -----------------------------
 
     @property

diff --git a/...ink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java b/...ink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
@@ -161,6 +161,7 @@
 import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.LTRIM;
 import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.MAKE_VALID_UTF8;
 import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.MAP_ENTRIES;
+import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.MAP_FROM_ENTRIES;
 import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.MAP_KEYS;
 import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.MAP_UNION;
 import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.MAP_VALUES;
@@ -1967,6 +1968,23 @@ public OutType mapEntries() {
         return toApiSpecificExpression(unresolvedCall(MAP_ENTRIES, toExpr()));
     }
 
+    /**
+     * Returns a map created from the given array of entries. Each entry must be a row with exactly
+     * two fields, where the first field becomes the key and the second one the value. If there are
+     * duplicate keys, the value of the last entry with that key wins. If the array itself or any of
+     * its entries is null, null is returned.
+     *
+     * <p>Examples:
+     *
+     * <pre>{@code
+     * array(row(1, "one"), row(2, "two")).mapFromEntries() // {1=one, 2=two}
+     * array(row(1, "one"), row(2, "two"), row(1, "uno")).mapFromEntries() // {1=uno, 2=two}
+     * }</pre>
+     */
+    public OutType mapFromEntries() {
+        return toApiSpecificExpression(unresolvedCall(MAP_FROM_ENTRIES, toExpr()));
+    }
+
     /**
      * Returns a map created by merging at least one map. These maps should have a common map type.
      * If there are overlapping keys, the value from 'map2' will overwrite the value from 'map1',

diff --git a/...ble-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java b/...ble-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -109,6 +109,7 @@
 import static org.apache.flink.table.types.inference.TypeStrategies.varyingString;
 import static org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.ARRAY_ELEMENT_ARG;
 import static org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.ARRAY_FULLY_COMPARABLE;
+import static org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.ARRAY_OF_ENTRIES_ARG;
 import static org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.FROM_CHANGELOG_INPUT_TYPE_STRATEGY;
 import static org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.INDEX;
 import static org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.JSON_ARGUMENT;
@@ -227,6 +228,19 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL)
                             "org.apache.flink.table.runtime.functions.scalar.MapFromArraysFunction")
                     .build();
 
+    public static final BuiltInFunctionDefinition MAP_FROM_ENTRIES =
+            BuiltInFunctionDefinition.newBuilder()
+                    .name("MAP_FROM_ENTRIES")
+                    .kind(SCALAR)
+                    .inputTypeStrategy(
+                            sequence(
+                                    new String[] {"input"},
+                                    new ArgumentTypeStrategy[] {ARRAY_OF_ENTRIES_ARG}))
+                    .outputTypeStrategy(SpecificTypeStrategies.MAP_FROM_ENTRIES)
+                    .runtimeClass(
+                            "org.apache.flink.table.runtime.functions.scalar.MapFromEntriesFunction")
+                    .build();
+
     public static final BuiltInFunctionDefinition SOURCE_WATERMARK =
             BuiltInFunctionDefinition.newBuilder()
                     .name("SOURCE_WATERMARK")

diff --git a/...org/apache/flink/table/types/inference/strategies/ArrayOfEntriesArgumentTypeStrategy.java b/...org/apache/flink/table/types/inference/strategies/ArrayOfEntriesArgumentTypeStrategy.java
@@ -0,0 +1,90 @@
+/*
+ * 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.table.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.types.CollectionDataType;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentTypeStrategy;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.StructuredType.StructuredComparison;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+
+import java.util.Optional;
+
+/**
+ * Strategy for an argument that must be an array of map entries, i.e. an {@code ARRAY} whose
+ * element is a {@code ROW} with exactly two fields. The first field becomes the map key, the second
+ * one the map value.
+ */
+@Internal
+public final class ArrayOfEntriesArgumentTypeStrategy implements ArgumentTypeStrategy {
+
+    @Override
+    public Optional<DataType> inferArgumentType(
+            CallContext callContext, int argumentPos, boolean throwOnFailure) {
+        final DataType actualType = callContext.getArgumentDataTypes().get(argumentPos);
+        if (!actualType.getLogicalType().is(LogicalTypeRoot.ARRAY)) {
+            return callContext.fail(
+                    throwOnFailure, "The input argument should be ARRAY<ROW<key, value>>");
+        }
+
+        final LogicalType elementType =
+                ((CollectionDataType) actualType).getElementDataType().getLogicalType();
+        if (!elementType.is(LogicalTypeRoot.ROW)
+                || LogicalTypeChecks.getFieldCount(elementType) != 2) {
+            return callContext.fail(
+                    throwOnFailure,
+                    "The input argument should be ARRAY<ROW<key, value>>, but the array element "
+                            + "type is '%s'. The element must be a ROW with exactly two fields.",
+                    elementType.asSummaryString());
+        }
+
+        // the key field must support equality, otherwise duplicate keys cannot be detected
+        final LogicalType keyType = LogicalTypeChecks.getFieldTypes(elementType).get(0);
+        if (!LogicalTypeChecks.areComparable(keyType, keyType, StructuredComparison.EQUALS)) {
+            return callContext.fail(
+                    throwOnFailure,
+                    "The map key type '%s' does not support equality comparison and therefore "
+                            + "cannot be used as the first field of a map entry.",
+                    keyType.asSummaryString());
+        }
+
+        return Optional.of(actualType);
+    }
+
+    @Override
+    public Argument getExpectedArgument(FunctionDefinition functionDefinition, int argumentPos) {
+        return Argument.of("ARRAY<ROW<key, value>>");
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        return this == o || o instanceof ArrayOfEntriesArgumentTypeStrategy;
+    }
+
+    @Override
+    public int hashCode() {
+        return ArrayOfEntriesArgumentTypeStrategy.class.hashCode();
+    }
+}
diff --git a/...n/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java b/...n/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
@@ -104,6 +104,10 @@ public static InputTypeStrategy windowTimeIndicator() {
     public static final ArgumentTypeStrategy ARRAY_FULLY_COMPARABLE =
             new ArrayComparableElementArgumentTypeStrategy(StructuredComparison.FULL);
 
+    /** See {@link ArrayOfEntriesArgumentTypeStrategy}. */
+    public static final ArgumentTypeStrategy ARRAY_OF_ENTRIES_ARG =
+            new ArrayOfEntriesArgumentTypeStrategy();
+
     /**
      * Input strategy for {@link BuiltInFunctionDefinitions#JSON_OBJECT}.
      *

diff --git a/...c/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java b/...c/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java
@@ -171,6 +171,27 @@ public final class SpecificTypeStrategies {
                                     ((CollectionDataType) callContext.getArgumentDataTypes().get(1))
                                             .getElementDataType()));
 
+    /**
+     * Type strategy specific for {@link BuiltInFunctionDefinitions#MAP_FROM_ENTRIES}.
+     *
+     * <p>Derives {@code MAP<key, value>} from the {@code ROW} element of the {@code ARRAY}
+     * argument. The result is nullable if the array itself is nullable or if its elements are,
+     * since a {@code NULL} entry makes the whole map {@code NULL}.
+     */
+    public static final TypeStrategy MAP_FROM_ENTRIES =
+            callContext -> {
+                final DataType arrayDataType = callContext.getArgumentDataTypes().get(0);
+                final DataType entryDataType =
+                        ((CollectionDataType) arrayDataType).getElementDataType();
+                final List<DataType> fieldDataTypes = entryDataType.getChildren();
+                final DataType mapDataType =
+                        DataTypes.MAP(fieldDataTypes.get(0), fieldDataTypes.get(1));
+                final boolean nullable =
+                        arrayDataType.getLogicalType().isNullable()
+                                || entryDataType.getLogicalType().isNullable();
+                return Optional.of(nullable ? mapDataType.nullable() : mapDataType.notNull());
+            };
+
     /**
      * Strategy for {@link org.apache.flink.table.functions.BuiltInFunctionDefinitions#LAG} and
      * {@link org.apache.flink.table.functions.BuiltInFunctionDefinitions#LEAD}. Returns a nullable