gustavodemorais commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3895386848
##########
docs/data/sql_functions.yml:
##########
@@ -925,6 +925,22 @@ collection:
- sql: MAP_ENTRIES(map)
table: MAP.mapEntries()
description: Returns an array of all entries in the given map. No order
guaranteed.
+ - sql: MAP_CONTAINS_KEY(map, key)
+ table: MAP.mapContainsKey(key)
+ description: |
+ Returns TRUE if the given key exists in the map, FALSE otherwise.
Returns NULL if the map is
+ NULL. If the search key is NULL, the function returns TRUE when the map
contains a NULL key.
+ The given key is cast implicitly to the map's key type where Flink's
implicit casting rules
+ allow it; otherwise the call fails validation.
+ e.g.
+ -- TRUE
+ MAP_CONTAINS_KEY(MAP['a', 1, 'b', 2], 'a')
+
+ -- FALSE
+ MAP_CONTAINS_KEY(MAP['a', 1, 'b', 2], 'z')
+
+ -- TRUE
+ MAP_CONTAINS_KEY(MAP[CAST(NULL AS STRING), 1], CAST(NULL AS STRING))
Review Comment:
```suggestion
Returns TRUE if the given key exists in the map, FALSE otherwise.
Returns NULL if the map is
NULL.
If the search key is NULL, the function returns TRUE when the map
contains a NULL key.
The given key is cast implicitly to the map's key type where Flink's
implicit casting rules
allow it; otherwise the call fails validation.
Examples
-- TRUE
MAP_CONTAINS_KEY(MAP['a', 1, 'b', 2], 'a')
-- FALSE
MAP_CONTAINS_KEY(MAP['a', 1, 'b', 2], 'z')
-- TRUE
MAP_CONTAINS_KEY(MAP[CAST(NULL AS STRING), 1], CAST(NULL AS STRING))
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MapFunctionITCase.java:
##########
@@ -406,6 +409,102 @@ private Stream<TestSetSpec> mapFromArraysTestCases() {
DataTypes.STRING(),
DataTypes.ARRAY(DataTypes.INT()))));
}
+ private Stream<TestSetSpec> mapContainsKeyTestCases() {
+ return Stream.of(
+ TestSetSpec.forFunction(
+ BuiltInFunctionDefinitions.MAP_CONTAINS_KEY,
"Invalid input")
+ .onFieldsWithData(CollectionUtil.map(entry("a", 1)))
+ .andDataTypes(DataTypes.MAP(DataTypes.STRING(),
DataTypes.INT()))
+ .testTableApiValidationError(
+ $("f0").mapContainsKey(true),
+ "Invalid input arguments. Expected signatures
are:\n"
+ + "MAP_CONTAINS_KEY(map <MAP>, key
<MAP KEY>)")
Review Comment:
Can we throw a better error here?
If I understand it correctly, we're throwing because the boolean type is not
valid here?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapContainsKeyFunction.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * 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.api.DataTypes;
+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.FunctionContext;
+import
org.apache.flink.table.functions.SpecializedFunction.ExpressionEvaluator;
+import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_CONTAINS_KEY}. */
+@Internal
+public class MapContainsKeyFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ExpressionEvaluator equalityEvaluator;
+ private transient MethodHandle equalityHandle;
+
+ public MapContainsKeyFunction(SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_CONTAINS_KEY, context);
+ final DataType mapDataType =
context.getCallContext().getArgumentDataTypes().get(0);
+ final DataType keyDataType = ((KeyValueDataType)
mapDataType).getKeyDataType();
+
+ keyElementGetter =
ArrayData.createElementGetter(keyDataType.getLogicalType());
+ equalityEvaluator =
+ context.createEvaluator(
+ $("key").isEqual($("needle")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("key",
keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("needle",
keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ equalityHandle = equalityEvaluator.open(context);
+ }
+
+ public @Nullable Boolean eval(@Nullable MapData map, @Nullable Object
needle) {
+ if (map == null) {
+ return null;
+ }
+ final ArrayData keys = map.keyArray();
+ final int size = map.size();
+ for (int pos = 0; pos < size; pos++) {
+ final Object key = keyElementGetter.getElementOrNull(keys, pos);
+ // NULL matches NULL here, unlike the SQL `NULL = NULL` the
evaluator would apply
+ if (needle == null && key == null) {
+ return true;
+ }
+ if (needle != null && key != null && isEqual(key, needle)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private boolean isEqual(final Object key, final Object needle) {
+ try {
+ return (boolean) equalityHandle.invoke(key, needle);
+ } catch (Throwable t) {
+ throw new FlinkRuntimeException(t);
Review Comment:
Hey you two, that's a valid discussion.
I'm +1 for keeping the throw. What actually causes this exception - a
representation bug, a broken custom RAW comparator, pathological nesting - are
edge cases and isn't "key not present," so returning false would just produce a
wrong answer indistinguishable from a real negative. That's worse than crashing.
Also, ARRAY_CONTAINS is a test-predicate too and still throws today -
changing it only here adds a new inconsistency instead of fixing one. Even if
we wanted to change the behavior for all functions, which I don't think we
should, this would require a larger discussion. For this function I'd just do
as the other ones
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapContainsKeyFunction.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.api.DataTypes;
+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.FunctionContext;
+import
org.apache.flink.table.functions.SpecializedFunction.ExpressionEvaluator;
+import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.KeyValueDataType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.lang.invoke.MethodHandle;
+
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#MAP_CONTAINS_KEY}. */
+@Internal
+public class MapContainsKeyFunction extends BuiltInScalarFunction {
+
+ private final ArrayData.ElementGetter keyElementGetter;
+ private final ExpressionEvaluator equalityEvaluator;
+ private transient MethodHandle equalityHandle;
+
+ public MapContainsKeyFunction(SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.MAP_CONTAINS_KEY, context);
+ final DataType mapDataType =
context.getCallContext().getArgumentDataTypes().get(0);
+ final DataType keyDataType = ((KeyValueDataType)
mapDataType).getKeyDataType();
+
+ keyElementGetter =
ArrayData.createElementGetter(keyDataType.getLogicalType());
+ equalityEvaluator =
+ context.createEvaluator(
+ $("key").isEqual($("needle")),
+ DataTypes.BOOLEAN(),
+ DataTypes.FIELD("key",
keyDataType.notNull().toInternal()),
+ DataTypes.FIELD("needle",
keyDataType.notNull().toInternal()));
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ equalityHandle = equalityEvaluator.open(context);
+ }
+
+ public @Nullable Boolean eval(@Nullable MapData map, @Nullable Object
needle) {
+ if (map == null) {
+ return null;
+ }
+ final ArrayData keys = map.keyArray();
+ final int size = map.size();
+ if (needle == null) {
+ // A NULL needle matches a NULL key, unlike SQL `NULL = NULL`
which yields UNKNOWN.
+ for (int pos = 0; pos < size; pos++) {
+ if (keyElementGetter.getElementOrNull(keys, pos) == null) {
+ return true;
+ }
+ }
+ } else {
+ for (int pos = 0; pos < size; pos++) {
+ final Object key = keyElementGetter.getElementOrNull(keys,
pos);
+ if (key != null && isEqual(key, needle)) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
Review Comment:
Can we simplify this?
```suggestion
public @Nullable Boolean eval(@Nullable MapData map, @Nullable Object
needle) {
if (map == null) {
return null;
}
final ArrayData keys = map.keyArray();
final int size = map.size();
for (int pos = 0; pos < size; pos++) {
final Object elementKey = keyElementGetter.getElementOrNull(keys,
pos);
// A NULL needle matches a NULL key, unlike SQL `NULL = NULL` which
yields UNKNOWN.
if (needle == null && elementKey == null) {
return true;
} else if (needle != null && elementKey != null &&
isEqual(elementKey, needle)) {
return true;
}
}
return false;
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
