This is an automated email from the ASF dual-hosted git repository.
gustavodemorais pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new cea216a8ee8 [FLINK-40356][table] Add new MapFromEntries built-in
function
cea216a8ee8 is described below
commit cea216a8ee8545bcba7e07e18ab02608c3b1a5e1
Author: Vasudev Kelappassery <[email protected]>
AuthorDate: Mon Aug 31 17:21:57 2026 +0100
[FLINK-40356][table] Add new MapFromEntries built-in function
This closes #28948.
---
docs/data/sql_functions.yml | 24 +++
docs/data/sql_functions_zh.yml | 23 +++
.../docs/reference/pyflink.table/expressions.rst | 1 +
flink-python/pyflink/table/expression.py | 18 ++
.../flink/table/api/internal/BaseExpressions.java | 20 +++
.../functions/BuiltInFunctionDefinitions.java | 14 ++
.../ArrayOfEntriesArgumentTypeStrategy.java | 92 ++++++++++
.../strategies/SpecificInputTypeStrategies.java | 4 +
.../strategies/SpecificTypeStrategies.java | 21 +++
.../ArrayOfEntriesArgumentTypeStrategyTest.java | 136 +++++++++++++++
.../strategies/MapFromEntriesTypeStrategyTest.java | 93 +++++++++++
.../table/planner/functions/MapFunctionITCase.java | 185 +++++++++++++++++++++
.../functions/scalar/MapFromArraysFunction.java | 28 +---
.../functions/scalar/MapFromEntriesFunction.java | 123 ++++++++++++++
.../runtime/functions/scalar/MapUnionFunction.java | 98 +++++------
.../flink/table/runtime/util/MapDataContainer.java | 51 ++++++
16 files changed, 847 insertions(+), 84 deletions(-)
diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
index 992f77f6234..e6ce79e47f7 100644
--- a/docs/data/sql_functions.yml
+++ b/docs/data/sql_functions.yml
@@ -928,6 +928,30 @@ 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; NULL keys are
+ treated as equal and collapse into a single entry. If the array itself
or any of its entries
+ is NULL, NULL is returned.
+
+ ```sql
+ -- Returns {1=one, 2=two}
+ MAP_FROM_ENTRIES(ARRAY[ROW(1, 'one'), ROW(2, 'two')])
+
+ -- Returns {1=uno, 2=two}
+ MAP_FROM_ENTRIES(ARRAY[ROW(1, 'one'), ROW(2, 'two'), ROW(1, 'uno')])
+
+ -- Returns {NULL=b}
+ MAP_FROM_ENTRIES(ARRAY[ROW(CAST(NULL AS INT), 'a'), ROW(CAST(NULL AS
INT), 'b')])
+
+ -- Returns 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
index cec2237d5af..b5819bbede2 100644
--- a/docs/data/sql_functions_zh.yml
+++ b/docs/data/sql_functions_zh.yml
@@ -1055,6 +1055,29 @@ 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; NULL keys are
+ treated as equal and collapse into a single entry. If the array itself
or any of its entries
+ is NULL, NULL is returned.
+
+ ```sql
+ -- Returns {1=one, 2=two}
+ MAP_FROM_ENTRIES(ARRAY[ROW(1, 'one'), ROW(2, 'two')])
+
+ -- Returns {1=uno, 2=two}
+ MAP_FROM_ENTRIES(ARRAY[ROW(1, 'one'), ROW(2, 'two'), ROW(1, 'uno')])
+
+ -- Returns {NULL=b}
+ MAP_FROM_ENTRIES(ARRAY[ROW(CAST(NULL AS INT), 'a'), ROW(CAST(NULL AS
INT), 'b')])
+
+ -- Returns 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
index 10614e996b1..2e2cee6cb93 100644
--- 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
index 33fd03122f8..e82e3bc02d2 100644
--- a/flink-python/pyflink/table/expression.py
+++ b/flink-python/pyflink/table/expression.py
@@ -1966,6 +1966,24 @@ class Expression(Generic[T]):
"""
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; None keys are
+ treated as equal and collapse into a single entry. 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/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
index 1dfc2687cbf..bd95cb9a73f 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
@@ -162,6 +162,7 @@ import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.LPAD;
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;
@@ -1968,6 +1969,25 @@ public abstract class BaseExpressions<InType, OutType> {
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.
+ *
+ * <p>If there are duplicate keys, the value of the last entry with that
key wins; null keys are
+ * treated as equal and collapse into a single entry. 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/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index 46e0583bb8c..a2a0cb5547f 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -107,6 +107,7 @@ import static
org.apache.flink.table.types.inference.TypeStrategies.nullableIfAr
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;
@@ -226,6 +227,19 @@ public final class BuiltInFunctionDefinitions {
"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/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ArrayOfEntriesArgumentTypeStrategy.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ArrayOfEntriesArgumentTypeStrategy.java
new file mode 100644
index 00000000000..6d78273eda9
--- /dev/null
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ArrayOfEntriesArgumentTypeStrategy.java
@@ -0,0 +1,92 @@
+/*
+ * 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 must be ARRAY<ROW<key, value>>, but
actual type was '%s'.",
+ actualType.getLogicalType().asSummaryString());
+ }
+
+ final LogicalType elementType =
+ ((CollectionDataType)
actualType).getElementDataType().getLogicalType();
+ if (!elementType.is(LogicalTypeRoot.ROW)
+ || LogicalTypeChecks.getFieldCount(elementType) != 2) {
+ return callContext.fail(
+ throwOnFailure,
+ "The 'input' argument must be ARRAY<ROW<key, value>>, but
the array element "
+ + "type was '%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/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
index 9aa75e02463..5f64808afa2 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificInputTypeStrategies.java
@@ -109,6 +109,10 @@ public final class SpecificInputTypeStrategies {
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/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java
index 77712114442..e3f8aa35327 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/SpecificTypeStrategies.java
+++
b/flink-table/flink-table-common/src/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 =
DataType.getFieldDataTypes(entryDataType);
+ 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
diff --git
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ArrayOfEntriesArgumentTypeStrategyTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ArrayOfEntriesArgumentTypeStrategyTest.java
new file mode 100644
index 00000000000..8a4d4e2f821
--- /dev/null
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ArrayOfEntriesArgumentTypeStrategyTest.java
@@ -0,0 +1,136 @@
+/*
+ * 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.api.common.serialization.SerializerConfigImpl;
+import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.types.inference.ArgumentTypeStrategy;
+import org.apache.flink.table.types.inference.InputTypeStrategiesTestBase;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.utils.DataTypeFactoryMock;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link ArrayOfEntriesArgumentTypeStrategy}. */
+class ArrayOfEntriesArgumentTypeStrategyTest extends
InputTypeStrategiesTestBase {
+
+ private static final InputTypeStrategy MAP_FROM_ENTRIES_INPUT_STRATEGY =
+ BuiltInFunctionDefinitions.MAP_FROM_ENTRIES
+ .getTypeInference(new DataTypeFactoryMock())
+ .getInputTypeStrategy();
+
+ @Override
+ protected Stream<TestSpec> testData() {
+ return Stream.of(
+ TestSpec.forStrategy(
+ "Array of two-field rows is accepted",
+ MAP_FROM_ENTRIES_INPUT_STRATEGY)
+ .calledWithArgumentTypes(
+ DataTypes.ARRAY(
+ DataTypes.ROW(
+ DataTypes.FIELD("key",
DataTypes.INT()),
+ DataTypes.FIELD("value",
DataTypes.STRING()))))
+ .expectSignature("f(input ARRAY<ROW<key, value>>)")
+ .expectArgumentTypes(
+ DataTypes.ARRAY(
+ DataTypes.ROW(
+ DataTypes.FIELD("key",
DataTypes.INT()),
+ DataTypes.FIELD("value",
DataTypes.STRING())))),
+ TestSpec.forStrategy(
+ "Nested and NOT NULL element types are
preserved",
+ MAP_FROM_ENTRIES_INPUT_STRATEGY)
+ .calledWithArgumentTypes(
+ DataTypes.ARRAY(
+ DataTypes.ROW(
+
DataTypes.FIELD(
+ "key",
DataTypes.STRING()),
+
DataTypes.FIELD(
+
"value",
+
DataTypes.ARRAY(
+
DataTypes.INT())))
+ .notNull())
+ .notNull())
+ .expectArgumentTypes(
+ DataTypes.ARRAY(
+ DataTypes.ROW(
+
DataTypes.FIELD(
+ "key",
DataTypes.STRING()),
+
DataTypes.FIELD(
+
"value",
+
DataTypes.ARRAY(
+
DataTypes.INT())))
+ .notNull())
+ .notNull()),
+ TestSpec.forStrategy(
+ "Non-array argument is rejected",
MAP_FROM_ENTRIES_INPUT_STRATEGY)
+ .calledWithArgumentTypes(DataTypes.STRING())
+ .expectErrorMessage("The 'input' argument must be
ARRAY<ROW<key, value>>"),
+ TestSpec.forStrategy(
+ "Array of non-row elements is rejected",
+ MAP_FROM_ENTRIES_INPUT_STRATEGY)
+
.calledWithArgumentTypes(DataTypes.ARRAY(DataTypes.INT()))
+ .expectErrorMessage(
+ "The 'input' argument must be ARRAY<ROW<key,
value>>, but the array "
+ + "element type was 'INT'. The element
must be a ROW with "
+ + "exactly two fields."),
+ TestSpec.forStrategy(
+ "Array of rows without exactly two fields is
rejected",
+ MAP_FROM_ENTRIES_INPUT_STRATEGY)
+ .calledWithArgumentTypes(
+ DataTypes.ARRAY(
+ DataTypes.ROW(
+ DataTypes.FIELD("key",
DataTypes.INT()),
+ DataTypes.FIELD("value",
DataTypes.STRING()),
+ DataTypes.FIELD("extra",
DataTypes.BOOLEAN()))))
+ .expectErrorMessage("The element must be a ROW with
exactly two fields."),
+ TestSpec.forStrategy(
+ "Key type without equality support is
rejected",
+ MAP_FROM_ENTRIES_INPUT_STRATEGY)
+ .calledWithArgumentTypes(
+ DataTypes.ARRAY(
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ "key",
+ DataTypes.RAW(
+ Object.class,
+ new
KryoSerializer<>(
+
Object.class,
+ new
SerializerConfigImpl()))),
+ DataTypes.FIELD("value",
DataTypes.STRING()))))
+ .expectErrorMessage(
+ "does not support equality comparison and
therefore cannot be "
+ + "used as the first field of a map
entry."));
+ }
+
+ @Test
+ void testEqualsAndHashCode() {
+ final ArgumentTypeStrategy strategy = new
ArrayOfEntriesArgumentTypeStrategy();
+ assertThat(strategy)
+ .isEqualTo(strategy)
+ .isEqualTo(new ArrayOfEntriesArgumentTypeStrategy())
+ .hasSameHashCodeAs(new ArrayOfEntriesArgumentTypeStrategy())
+ .isNotEqualTo("not a strategy");
+ }
+}
diff --git
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/MapFromEntriesTypeStrategyTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/MapFromEntriesTypeStrategyTest.java
new file mode 100644
index 00000000000..8689a788f9e
--- /dev/null
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/MapFromEntriesTypeStrategyTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.table.api.DataTypes;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.TypeStrategiesTestBase;
+
+import java.util.stream.Stream;
+
+/** Tests for {@link SpecificTypeStrategies#MAP_FROM_ENTRIES}. */
+class MapFromEntriesTypeStrategyTest extends TypeStrategiesTestBase {
+
+ private static DataType entry(boolean elementNullable) {
+ final DataType entryType =
+ DataTypes.ROW(
+ DataTypes.FIELD("key", DataTypes.INT()),
+ DataTypes.FIELD("value", DataTypes.STRING()));
+ return elementNullable ? entryType.nullable() : entryType.notNull();
+ }
+
+ private static DataType array(boolean arrayNullable, boolean
elementNullable) {
+ final DataType arrayType = DataTypes.ARRAY(entry(elementNullable));
+ return arrayNullable ? arrayType.nullable() : arrayType.notNull();
+ }
+
+ private static final DataType MAP_TYPE = DataTypes.MAP(DataTypes.INT(),
DataTypes.STRING());
+
+ @Override
+ protected Stream<TestSpec> testData() {
+ return Stream.of(
+ TestSpec.forStrategy(
+ "A nullable array yields a nullable map",
+ SpecificTypeStrategies.MAP_FROM_ENTRIES)
+ .inputTypes(array(true, false))
+ .expectDataType(MAP_TYPE.nullable()),
+ TestSpec.forStrategy(
+ "A nullable entry yields a nullable map",
+ SpecificTypeStrategies.MAP_FROM_ENTRIES)
+ .inputTypes(array(false, true))
+ .expectDataType(MAP_TYPE.nullable()),
+ TestSpec.forStrategy(
+ "NOT NULL array and entries yield a NOT NULL
map, "
+ + "carrying field nullability into the
map",
+ SpecificTypeStrategies.MAP_FROM_ENTRIES)
+ .inputTypes(
+ DataTypes.ARRAY(
+ DataTypes.ROW(
+
DataTypes.FIELD(
+ "key",
+
DataTypes.INT().notNull()),
+
DataTypes.FIELD(
+
"value",
+
DataTypes.STRING()
+
.notNull()))
+ .notNull())
+ .notNull())
+ .expectDataType(
+ DataTypes.MAP(
+ DataTypes.INT().notNull(),
+ DataTypes.STRING().notNull())
+ .notNull()),
+ TestSpec.forStrategy(
+ "Nested value type is preserved",
+ SpecificTypeStrategies.MAP_FROM_ENTRIES)
+ .inputTypes(
+ DataTypes.ARRAY(
+ DataTypes.ROW(
+ DataTypes.FIELD("key",
DataTypes.STRING()),
+ DataTypes.FIELD(
+ "value",
+
DataTypes.ARRAY(DataTypes.INT())))))
+ .expectDataType(
+ DataTypes.MAP(DataTypes.STRING(),
DataTypes.ARRAY(DataTypes.INT()))
+ .nullable()));
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MapFunctionITCase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MapFunctionITCase.java
index 0f3e305d16e..54b3b8cec05 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MapFunctionITCase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MapFunctionITCase.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.planner.functions;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.types.DataType;
import org.apache.flink.types.Row;
import org.apache.flink.util.CollectionUtil;
@@ -46,10 +47,13 @@ import static org.apache.flink.table.api.DataTypes.STRING;
import static org.apache.flink.table.api.DataTypes.TIME;
import static org.apache.flink.table.api.DataTypes.TIMESTAMP;
import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.array;
import static org.apache.flink.table.api.Expressions.call;
import static org.apache.flink.table.api.Expressions.lit;
import static org.apache.flink.table.api.Expressions.map;
import static org.apache.flink.table.api.Expressions.mapFromArrays;
+import static org.apache.flink.table.api.Expressions.nullOf;
+import static org.apache.flink.table.api.Expressions.row;
import static org.apache.flink.util.CollectionUtil.entry;
/** Test {@link BuiltInFunctionDefinitions#MAP} and its return type. */
@@ -73,6 +77,7 @@ public class MapFunctionITCase extends
BuiltInFunctionTestBase {
mapValuesTestCases(),
mapEntriesTestCases(),
mapFromArraysTestCases(),
+ mapFromEntriesTestCases(),
mapUnionTestCases())
.flatMap(s -> s);
}
@@ -406,6 +411,186 @@ public class MapFunctionITCase extends
BuiltInFunctionTestBase {
DataTypes.STRING(),
DataTypes.ARRAY(DataTypes.INT()))));
}
+ private Stream<TestSetSpec> mapFromEntriesTestCases() {
+ final DataType entryType =
+ DataTypes.ROW(
+ DataTypes.FIELD("key", DataTypes.INT()),
+ DataTypes.FIELD("value", DataTypes.STRING()));
+ final DataType nestedEntryType =
+ DataTypes.ROW(
+ DataTypes.FIELD("key", DataTypes.STRING()),
+ DataTypes.FIELD("value",
DataTypes.ARRAY(DataTypes.INT())));
+ final DataType rowKeyType =
+ DataTypes.ROW(
+ DataTypes.FIELD("a", DataTypes.INT()),
+ DataTypes.FIELD("b", DataTypes.STRING()));
+ return Stream.of(
+ TestSetSpec.forFunction(
+ BuiltInFunctionDefinitions.MAP_FROM_ENTRIES,
"Invalid input")
+ .onFieldsWithData("item", new Integer[] {1, 2})
+ .andDataTypes(DataTypes.STRING(),
DataTypes.ARRAY(DataTypes.INT()))
+ .testTableApiValidationError(
+ $("f0").mapFromEntries(),
+ "The 'input' argument must be ARRAY<ROW<key,
value>>")
+ .testSqlValidationError(
+ "MAP_FROM_ENTRIES(ARRAY[ROW(1, 'a', true)])",
+ "The element must be a ROW with exactly two
fields."),
+
TestSetSpec.forFunction(BuiltInFunctionDefinitions.MAP_FROM_ENTRIES)
+ .onFieldsWithData(
+ new Row[] {Row.of(1, "one"), Row.of(2, "two")},
+ new Row[] {Row.of(1, "one"), Row.of(1, "uno")},
+ null,
+ new Row[] {
+ Row.of("one", new Integer[] {1, 2}),
+ Row.of("two", new Integer[] {3, 4})
+ },
+ new Row[] {Row.of(null, "a"), Row.of(null,
"b")})
+ .andDataTypes(
+ DataTypes.ARRAY(entryType),
+ DataTypes.ARRAY(entryType),
+ DataTypes.ARRAY(entryType),
+ DataTypes.ARRAY(nestedEntryType),
+ DataTypes.ARRAY(entryType))
+ // duplicate keys: the last value wins
+ .testResult(
+ $("f1").mapFromEntries(),
+ "MAP_FROM_ENTRIES(f1)",
+ CollectionUtil.map(entry(1, "uno")),
+ DataTypes.MAP(DataTypes.INT(),
DataTypes.STRING()))
+ // an empty array yields an empty map
+ .testResult(
+ $("f0").arraySlice(20, 30).mapFromEntries(),
+ "MAP_FROM_ENTRIES(ARRAY_SLICE(f0, 20, 30))",
+ Collections.emptyMap(),
+ DataTypes.MAP(DataTypes.INT(),
DataTypes.STRING()))
+ // NULL array yields NULL
+ .testResult(
+ $("f2").mapFromEntries(),
+ "MAP_FROM_ENTRIES(f2)",
+ null,
+ DataTypes.MAP(DataTypes.INT(),
DataTypes.STRING()))
+ // nested value type
+ .testTableApiResult(
+ $("f3").mapFromEntries(),
+ CollectionUtil.map(
+ entry("one", new Integer[] {1, 2}),
+ entry("two", new Integer[] {3, 4})),
+ DataTypes.MAP(DataTypes.STRING(),
DataTypes.ARRAY(DataTypes.INT())))
+ // duplicate NULL keys are deduplicated as well
+ .testResult(
+ $("f4").mapFromEntries(),
+ "MAP_FROM_ENTRIES(f4)",
+ Collections.singletonMap(null, "b"),
+ DataTypes.MAP(DataTypes.INT(),
DataTypes.STRING()))
+ // duplicate STRING keys use the generated equality; a
NULL value is kept
+ .testResult(
+ array(
+ row("k", "a"),
+ row("k", "b"),
+ row("x",
nullOf(DataTypes.STRING())))
+ .mapFromEntries(),
+ "MAP_FROM_ENTRIES(ARRAY[ROW('k', 'a'),
ROW('k', 'b'), ROW('x', CAST(NULL AS STRING))])",
+ CollectionUtil.map(entry("k", "b"), entry("x",
null)),
+ DataTypes.MAP(DataTypes.CHAR(1).notNull(),
DataTypes.STRING())
+ .notNull())
+ // duplicate ROW keys use the generated equality; a
NULL value is kept
+ .testResult(
+ array(
+ row(row(1,
"a").cast(rowKeyType), "x"),
+ row(row(1,
"a").cast(rowKeyType), "y"),
+ row(
+ row(2,
"b").cast(rowKeyType),
+
nullOf(DataTypes.STRING())))
+ .mapFromEntries(),
+ "MAP_FROM_ENTRIES(ARRAY["
+ + "ROW(CAST(ROW(1, 'a') AS ROW(a INT,
b STRING)), 'x'), "
+ + "ROW(CAST(ROW(1, 'a') AS ROW(a INT,
b STRING)), 'y'), "
+ + "ROW(CAST(ROW(2, 'b') AS ROW(a INT,
b STRING)), CAST(NULL AS STRING))])",
+ CollectionUtil.map(
+ entry(Row.of(1, "a"), "y"),
entry(Row.of(2, "b"), null)),
+ DataTypes.MAP(rowKeyType.notNull(),
DataTypes.STRING()).notNull())
+ // duplicate ARRAY keys collapse through the generated
equality and the
+ // last entry wins, keeping its NULL value; asserted
via CARDINALITY and
+ // MAP_VALUES because Map#equals cannot compare
array-keyed maps
+ .testResult(
+ array(
+ row(array(1, 2), "x"),
+ row(array(1, 2),
nullOf(DataTypes.STRING())))
+ .mapFromEntries()
+ .cardinality(),
+ "CARDINALITY(MAP_FROM_ENTRIES("
+ + "ARRAY[ROW(ARRAY[1, 2], 'x'),
ROW(ARRAY[1, 2], CAST(NULL AS STRING))]))",
+ 1,
+ DataTypes.INT().notNull())
+ .testResult(
+ array(
+ row(array(1, 2), "x"),
+ row(array(1, 2),
nullOf(DataTypes.STRING())))
+ .mapFromEntries()
+ .mapValues(),
+ "MAP_VALUES(MAP_FROM_ENTRIES("
+ + "ARRAY[ROW(ARRAY[1, 2], 'x'),
ROW(ARRAY[1, 2], CAST(NULL AS STRING))]))",
+ new String[] {null},
+ DataTypes.ARRAY(DataTypes.STRING()).notNull())
+ // round trip with the inverse function
+ .testResult(
+ $("f0").mapFromEntries().mapEntries(),
+ "MAP_ENTRIES(MAP_FROM_ENTRIES(f0))",
+ new Row[] {Row.of(1, "one"), Row.of(2, "two")},
+ DataTypes.ARRAY(entryType)),
+ TestSetSpec.forFunction(
+ BuiltInFunctionDefinitions.MAP_FROM_ENTRIES,
"Documented examples")
+ .onFieldsWithData(1)
+ .andDataTypes(DataTypes.INT().notNull())
+ .testResult(
+ array(row(1, "one"), row(2,
"two")).mapFromEntries(),
+ "MAP_FROM_ENTRIES(ARRAY[ROW(1, 'one'), ROW(2,
'two')])",
+ CollectionUtil.map(entry(1, "one"), entry(2,
"two")),
+ DataTypes.MAP(
+ DataTypes.INT().notNull(),
+ DataTypes.CHAR(3).notNull())
+ .notNull())
+ .testResult(
+ array(row(1, "one"), row(2, "two"), row(1,
"uno")).mapFromEntries(),
+ "MAP_FROM_ENTRIES(ARRAY[ROW(1, 'one'), ROW(2,
'two'), ROW(1, 'uno')])",
+ CollectionUtil.map(entry(1, "uno"), entry(2,
"two")),
+ DataTypes.MAP(
+ DataTypes.INT().notNull(),
+ DataTypes.CHAR(3).notNull())
+ .notNull())
+ .testResult(
+ array(
+ row(nullOf(DataTypes.INT()),
"a"),
+ row(nullOf(DataTypes.INT()),
"b"))
+ .mapFromEntries(),
+ "MAP_FROM_ENTRIES(ARRAY[ROW(CAST(NULL AS INT),
'a'), ROW(CAST(NULL AS INT), 'b')])",
+ Collections.singletonMap(null, "b"),
+ DataTypes.MAP(DataTypes.INT(),
DataTypes.CHAR(1).notNull())
+ .notNull())
+ .testResult(
+ array(
+ row(1, "one"),
+ nullOf(
+ DataTypes.ROW(
+
DataTypes.FIELD(
+ "k",
DataTypes.INT()),
+
DataTypes.FIELD(
+ "v",
DataTypes.STRING()))))
+ .mapFromEntries(),
+ "MAP_FROM_ENTRIES(ARRAY[ROW(1, 'one'),
CAST(NULL AS ROW(k INT, v STRING))])",
+ null,
+ DataTypes.MAP(DataTypes.INT(),
DataTypes.STRING()))
+ // a NOT NULL array of NOT NULL entries yields a NOT
NULL map
+ .testResult(
+ array(row($("f0"), "a")).mapFromEntries(),
+ "MAP_FROM_ENTRIES(ARRAY[ROW(f0, 'a')])",
+ Collections.singletonMap(1, "a"),
+ DataTypes.MAP(
+ DataTypes.INT().notNull(),
+ DataTypes.CHAR(1).notNull())
+ .notNull()));
+ }
+
private Stream<TestSetSpec> mapUnionTestCases() {
return Stream.of(
TestSetSpec.forFunction(BuiltInFunctionDefinitions.MAP_UNION)
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapFromArraysFunction.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapFromArraysFunction.java
index 2fa789154f7..167d5e55a3e 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapFromArraysFunction.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapFromArraysFunction.java
@@ -23,6 +23,7 @@ import org.apache.flink.table.data.ArrayData;
import org.apache.flink.table.data.MapData;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.runtime.util.MapDataContainer;
import org.apache.flink.util.FlinkRuntimeException;
import javax.annotation.Nullable;
@@ -47,31 +48,6 @@ public class MapFromArraysFunction extends
BuiltInScalarFunction {
+ " is not equal to the length of the values array
"
+ valuesArray.size());
}
- return new MapDataForMapFromArrays(keysArray, valuesArray);
- }
-
- private static class MapDataForMapFromArrays implements MapData {
- private final ArrayData keyArray;
- private final ArrayData valueArray;
-
- public MapDataForMapFromArrays(ArrayData keyArray, ArrayData
valueArray) {
- this.keyArray = keyArray;
- this.valueArray = valueArray;
- }
-
- @Override
- public int size() {
- return keyArray.size();
- }
-
- @Override
- public ArrayData keyArray() {
- return keyArray;
- }
-
- @Override
- public ArrayData valueArray() {
- return valueArray;
- }
+ return new MapDataContainer(keysArray, valuesArray);
}
}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapFromEntriesFunction.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapFromEntriesFunction.java
new file mode 100644
index 00000000000..f4f2e712a25
--- /dev/null
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapFromEntriesFunction.java
@@ -0,0 +1,123 @@
+/*
+ * 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.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.runtime.util.EqualityAndHashcodeProvider;
+import org.apache.flink.table.runtime.util.MapDataContainer;
+import org.apache.flink.table.runtime.util.ObjectContainer;
+import org.apache.flink.table.types.CollectionDataType;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.CollectionUtil;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+import java.util.Map;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_FROM_ENTRIES}. */
+@Internal
+public class MapFromEntriesFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter entryElementGetter;
+ private final RowData.FieldGetter keyFieldGetter;
+ private final RowData.FieldGetter valueFieldGetter;
+
+ private final EqualityAndHashcodeProvider keyEqualityAndHashcodeProvider;
+
+ public MapFromEntriesFunction(SpecializedFunction.SpecializedContext
context) {
+ super(BuiltInFunctionDefinitions.MAP_FROM_ENTRIES, context);
+ final DataType arrayDataType =
context.getCallContext().getArgumentDataTypes().get(0);
+ final DataType entryDataType = ((CollectionDataType)
arrayDataType).getElementDataType();
+ final List<DataType> fieldDataTypes =
DataType.getFieldDataTypes(entryDataType);
+ final DataType keyDataType = fieldDataTypes.get(0);
+ final DataType valueDataType = fieldDataTypes.get(1);
+
+ entryElementGetter =
ArrayData.createElementGetter(entryDataType.getLogicalType());
+ keyFieldGetter =
RowData.createFieldGetter(keyDataType.getLogicalType(), 0);
+ valueFieldGetter =
RowData.createFieldGetter(valueDataType.getLogicalType(), 1);
+
+ keyEqualityAndHashcodeProvider =
+ new EqualityAndHashcodeProvider(context,
keyDataType.toInternal());
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ keyEqualityAndHashcodeProvider.open(context);
+ }
+
+ public @Nullable MapData eval(@Nullable ArrayData input) {
+ if (input == null) {
+ return null;
+ }
+
+ final int size = input.size();
+ // a duplicate key keeps the position of its first occurrence and the
last value wins
+ final Map<ObjectContainer, Object> entries =
+ CollectionUtil.newLinkedHashMapWithExpectedSize(size);
+ for (int pos = 0; pos < size; pos++) {
+ final RowData entry = (RowData)
entryElementGetter.getElementOrNull(input, pos);
+ if (entry == null) {
+ return null;
+ }
+ entries.put(
+ wrapKey(keyFieldGetter.getFieldOrNull(entry)),
+ valueFieldGetter.getFieldOrNull(entry));
+ }
+ final int distinctKeyCount = entries.size();
+
+ final Object[] keys = new Object[distinctKeyCount];
+ final Object[] values = new Object[distinctKeyCount];
+ int pos = 0;
+ for (Map.Entry<ObjectContainer, Object> entry : entries.entrySet()) {
+ final ObjectContainer key = entry.getKey();
+ keys[pos] = key == null ? null : key.getObject();
+ values[pos] = entry.getValue();
+ pos++;
+ }
+ return new MapDataContainer(new GenericArrayData(keys), new
GenericArrayData(values));
+ }
+
+ /**
+ * Hashes and compares the key with SQL semantics instead of {@link
Object#equals}. A {@code
+ * null} key is returned unwrapped, so all {@code null} keys are treated
as equal and collapse
+ * into a single entry.
+ */
+ private @Nullable ObjectContainer wrapKey(@Nullable Object key) {
+ if (key == null) {
+ return null;
+ }
+ return new ObjectContainer(
+ key,
+ keyEqualityAndHashcodeProvider::equals,
+ keyEqualityAndHashcodeProvider::hashCode);
+ }
+
+ @Override
+ public void close() throws Exception {
+ keyEqualityAndHashcodeProvider.close();
+ }
+}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java
index fef1f730e0e..eda8d8e9bb9 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapUnionFunction.java
@@ -26,6 +26,7 @@ import org.apache.flink.table.data.MapData;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.table.functions.FunctionContext;
import org.apache.flink.table.functions.SpecializedFunction;
+import org.apache.flink.table.runtime.util.MapDataContainer;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.KeyValueDataType;
import org.apache.flink.util.FlinkRuntimeException;
@@ -88,7 +89,7 @@ public class MapUnionFunction extends BuiltInScalarFunction {
return null;
}
if (map.size() > 0) {
- result = new MapDataForMapUnion(result, map);
+ result = mapUnion(result, map);
}
}
return result;
@@ -97,70 +98,51 @@ public class MapUnionFunction extends BuiltInScalarFunction
{
}
}
- private class MapDataForMapUnion implements MapData {
- private final GenericArrayData keysArray;
- private final GenericArrayData valuesArray;
-
- public MapDataForMapUnion(MapData map1, MapData map2) throws Throwable
{
- List<Object> keysList = new ArrayList<>();
- List<Object> valuesList = new ArrayList<>();
- boolean isKeyNullExist = false;
- ArrayData keyArray2 = map2.keyArray();
- ArrayData valueArray2 = map2.valueArray();
- for (int i = 0; i < map2.size(); i++) {
- Object key = keyElementGetter.getElementOrNull(keyArray2, i);
- if (key == null) {
- isKeyNullExist = true;
- }
- keysList.add(key);
-
valuesList.add(valueElementGetter.getElementOrNull(valueArray2, i));
+ private MapData mapUnion(MapData map1, MapData map2) throws Throwable {
+ List<Object> keysList = new ArrayList<>();
+ List<Object> valuesList = new ArrayList<>();
+ boolean isKeyNullExist = false;
+ ArrayData keyArray2 = map2.keyArray();
+ ArrayData valueArray2 = map2.valueArray();
+ for (int i = 0; i < map2.size(); i++) {
+ Object key = keyElementGetter.getElementOrNull(keyArray2, i);
+ if (key == null) {
+ isKeyNullExist = true;
}
- ArrayData keyArray1 = map1.keyArray();
- ArrayData valueArray1 = map1.valueArray();
- for (int i = 0; i < map1.size(); i++) {
- final Object key1 =
keyElementGetter.getElementOrNull(keyArray1, i);
-
- boolean keyExists = false;
- if (key1 != null) {
- for (int j = 0; j < keysList.size(); j++) {
- final Object key2 = keysList.get(j);
- if (key2 != null && (boolean)
keyEqualityHandle.invoke(key1, key2)) {
- // If key exists in map2, skip this key-value pair
- keyExists = true;
- break;
- }
+ keysList.add(key);
+ valuesList.add(valueElementGetter.getElementOrNull(valueArray2,
i));
+ }
+ ArrayData keyArray1 = map1.keyArray();
+ ArrayData valueArray1 = map1.valueArray();
+ for (int i = 0; i < map1.size(); i++) {
+ final Object key1 = keyElementGetter.getElementOrNull(keyArray1,
i);
+
+ boolean keyExists = false;
+ if (key1 != null) {
+ for (int j = 0; j < keysList.size(); j++) {
+ final Object key2 = keysList.get(j);
+ if (key2 != null && (boolean)
keyEqualityHandle.invoke(key1, key2)) {
+ // If key exists in map2, skip this key-value pair
+ keyExists = true;
+ break;
}
}
-
- if (isKeyNullExist && key1 == null) {
- continue;
- }
-
- // If key doesn't exist in map2, add the key-value pair from
map1
- if (!keyExists) {
- final Object value1 =
valueElementGetter.getElementOrNull(valueArray1, i);
- keysList.add(key1);
- valuesList.add(value1);
- }
}
- this.keysArray = new GenericArrayData(keysList.toArray());
- this.valuesArray = new GenericArrayData(valuesList.toArray());
- }
-
- @Override
- public int size() {
- return keysArray.size();
- }
- @Override
- public ArrayData keyArray() {
- return keysArray;
- }
+ if (isKeyNullExist && key1 == null) {
+ continue;
+ }
- @Override
- public ArrayData valueArray() {
- return valuesArray;
+ // If key doesn't exist in map2, add the key-value pair from map1
+ if (!keyExists) {
+ final Object value1 =
valueElementGetter.getElementOrNull(valueArray1, i);
+ keysList.add(key1);
+ valuesList.add(value1);
+ }
}
+ return new MapDataContainer(
+ new GenericArrayData(keysList.toArray()),
+ new GenericArrayData(valuesList.toArray()));
}
@Override
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/MapDataContainer.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/MapDataContainer.java
new file mode 100644
index 00000000000..336f1c9fa8e
--- /dev/null
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/MapDataContainer.java
@@ -0,0 +1,51 @@
+/*
+ * 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.runtime.util;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.MapData;
+
+/** A {@link MapData} backed directly by a key array and a value array. */
+@Internal
+public class MapDataContainer implements MapData {
+ private final ArrayData keyArray;
+ private final ArrayData valueArray;
+
+ public MapDataContainer(ArrayData keyArray, ArrayData valueArray) {
+ this.keyArray = keyArray;
+ this.valueArray = valueArray;
+ }
+
+ @Override
+ public int size() {
+ return keyArray.size();
+ }
+
+ @Override
+ public ArrayData keyArray() {
+ return keyArray;
+ }
+
+ @Override
+ public ArrayData valueArray() {
+ return valueArray;
+ }
+}
