You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "VasShabu (via GitHub)" <gi...@apache.org> on 2026/08/13 10:19:46 UTC
[PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
VasShabu opened a new pull request, #28970:
URL: https://github.com/apache/flink/pull/28970
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## What is the purpose of the change
This pull request add a new function maoContainsKey function, which which given a key which check if it is present within a map.
## Brief change log
- added new function to builtin function definitions
- created a new inputType strategy
- created reference in baseExpressions
- created references in expressio.py
- added function to expressions.rst
- created new scalar function, mapContainsKey function.
- created test cases
-
## Verifying this change
Please make sure both new and modified tests in this PR follow [the conventions for tests defined in our code quality guide](https://flink.apache.org/how-to-contribute/code-style-and-quality-common/#7-testing).
This change added tests and can be verified as follows:
/mvnw -o -pl flink-table/flink-table-planner -Dtest='JsonFunctionsITCase' -Dsurefire.failIfNoSpecifiedTests=false -Dcheckstyle.skip=true -Dspotless.check.skip=true -Drat.skip=true -e -Denforcer.skip=true test
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (100MB)*
- *Extended integration test for recovery after master (JobManager) failure*
- *Added test that validates that TaskInfo is transferred only once across recoveries*
- *Manually verified the change by running a 4 node cluster with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (don't know)
- The runtime per-record code paths (performance sensitive): (yes)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (java docs, sqlfunction yml file, expression.py)
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change the checkbox below to `[X]` and replace the placeholder in the "Generated-by"
line with the tool name and version. Otherwise remove the "Generated-by" line.
See the ASF Generative Tooling Guidance for details:
https://www.apache.org/legal/generative-tooling.html
You are responsible for the quality and correctness of every change in this PR
regardless of the tooling used. Low-effort AI-generated PRs will be closed. See
AGENTS.md for the full guidance.
-->
- [X] Yes (please specify the tool below)
Generated-by: Claude Opus 5
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3784303841
##########
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:
In the past we have not wanted to issue errors in SQL as it would end the job. I suggest we return false and log in this error case. In the error case they are not equal.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3836607509
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java:
##########
@@ -209,6 +212,21 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL)
"org.apache.flink.table.runtime.functions.scalar.MapEntriesFunction")
.build();
+ public static final BuiltInFunctionDefinition MAP_CONTAINS_KEY =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("MAP_CONTAINS_KEY")
+ .kind(SCALAR)
+ .inputTypeStrategy(
+ sequence(
+ Arrays.asList("map", "key"),
+ Arrays.asList(logical(LogicalTypeRoot.MAP), MAP_KEY_ARG)))
Review Comment:
```suggestion
List.of("map", "key"),
List.of(logical(LogicalTypeRoot.MAP), MAP_KEY_ARG)))
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3933092637
##########
docs/data/sql_functions.yml:
##########
@@ -925,6 +925,23 @@ 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.
Review Comment:
added example to all docs.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3836645671
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java:
##########
@@ -1967,6 +1968,26 @@ public OutType mapEntries() {
return toApiSpecificExpression(unresolvedCall(MAP_ENTRIES, toExpr()));
}
+ /**
+ * Returns {@code TRUE} if the given key exists in the map, {@code FALSE} otherwise. Returns
+ * {@code NULL} if the map is {@code NULL}.
+ *
+ * <p>A {@code NULL} key matches a {@code NULL} key in the map. The given key is cast implicitly
Review Comment:
>If the search key is NULL, the function returns TRUE when the map contains a NULL key
this probably sounds a little bit better
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3934274728
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java:
##########
@@ -2575,7 +2597,7 @@ public OutType jsonType(String path) {
return toApiSpecificExpression(unresolvedCall(JSON_TYPE, toExpr(), valueLiteral(path)));
}
- /**
+ /* *
Review Comment:
why do we need this change?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3795647977
##########
docs/data/sql_functions.yml:
##########
@@ -925,6 +925,21 @@ 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. A NULL key matches a NULL key in the map. 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.
Review Comment:
This validation error is called during the planning stage i think when a map and a key is given with different types on the keys
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3925618928
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapContainsKeyFunction.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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 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;
+ }
Review Comment:
How about?
```suggestion
// A NULL needle matches a NULL key, unlike SQL `NULL = NULL` which yields UNKNOWN.
if (keysEqual(elementKey, needle)) {
return true;
}
```
```java
private boolean keysEqual(@Nullable Object elementKey, @Nullable Object needle) {
// A NULL needle matches a NULL key, unlike SQL `NULL = NULL` which yields UNKNOWN.
return needle == null ? elementKey == null : elementKey != null && isEqual(elementKey, needle);
}
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3925689670
##########
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:
@gustavodemorais it was before, however it is less performant
since it is runtime I suggested to change it
https://github.com/apache/flink/pull/28970#discussion_r3836727901
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on PR #28970:
URL: https://github.com/apache/flink/pull/28970#issuecomment-5582759900
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin merged PR #28970:
URL: https://github.com/apache/flink/pull/28970
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3987846796
##########
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:
Do we need the `else`?
```suggestion
return false;
}
for (int pos = 0; pos < size; pos++) {
final Object key = keyElementGetter.getElementOrNull(keys, pos);
if (key != null && isEqual(key, 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3925781009
##########
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;
+ }
+ }
Review Comment:
Agree. We know before hand if `needle` is null or not and we can decide based on that.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3880054722
##########
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:
Hi there,
As this is a function is a test to say does the map contain a key function - if it fails I assume that it doesn't contain the key function, so false would seem reasonable. What sort of errors can we get here, that we would reasonably want to surface?
At the same time I am keen we should follow the convention of the other functions. Though we could argue ARRAY_CONTAINS is a test also, so should not error, but ARRAY_REMOVE is a mutation , which if it fails should reasonably be an error.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3987846796
##########
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:
Do we need the `else`?
```suggestion
}
for (int pos = 0; pos < size; pos++) {
final Object key = keyElementGetter.getElementOrNull(keys, pos);
if (key != null && isEqual(key, 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3896287782
##########
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:
Not a blocker but can't we tell the type and be more specific as to why it failed
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on PR #28970:
URL: https://github.com/apache/flink/pull/28970#issuecomment-5585489121
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3925689670
##########
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:
@gustavodemorais it was before, however it is less performat
since it is runtime I suggested to change it
https://github.com/apache/flink/pull/28970#discussion_r3836727901
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3795591077
##########
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:
The catch Throwable here mirrors other builtin function's implementation, functions like :
- ARRAY_CONTAINS
- ARRAY_POSITION
- ARRAY_REMOVE
I was also talking to my mentor about logging and how it is not great in streaming as we are processing millions of records and logs can get full really quickly and how you dont know easily what error causes a specific entry in a log.
Please let me know your thoughts, intrested to hear more, I see that there are two options, either keep this implementation or alter other functions aswell to ensure that we are returning False as you say and logging. However if we decide to do this to the naked eye some invalid input that would cause an error would be hidden by a silent failure only to be discovered in a log.
Would love to hear you thoughts regarding this.
Thanks
Vas.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #28970:
URL: https://github.com/apache/flink/pull/28970#issuecomment-5279083181
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b7b48badc7acdf85d0b2314c86d9d603b087688d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "b7b48badc7acdf85d0b2314c86d9d603b087688d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b7b48badc7acdf85d0b2314c86d9d603b087688d UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3853895482
##########
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.
+ eg.
Review Comment:
```suggestion
e.g.
```
nit
##########
docs/data/sql_functions_zh.yml:
##########
@@ -1052,6 +1052,22 @@ collection:
- sql: MAP_ENTRIES(map)
table: MAP.mapEntries()
description: 以数组形式返回 map 中的所有 entry，不保证顺序。
+ - 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.
+ eg.
Review Comment:
```suggestion
e.g.
```
nit
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3784333121
##########
docs/data/sql_functions.yml:
##########
@@ -925,6 +925,21 @@ 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. A NULL key matches a NULL key in the map. 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.
Review Comment:
when we say fails validation - I assume we should return false as per my other comment
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on PR #28970:
URL: https://github.com/apache/flink/pull/28970#issuecomment-5586385774
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3784326359
##########
docs/data/sql_functions.yml:
##########
@@ -925,6 +925,21 @@ 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)
Review Comment:
I am curious if the parameter key is a castable to the key in the map, does this work? Can we add tests and document what we find around this edge case?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on PR #28970:
URL: https://github.com/apache/flink/pull/28970#issuecomment-5682105326
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3836680831
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/MapKeyArgumentTypeStrategy.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.api.DataTypes;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionDefinition;
+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.MapType;
+
+import java.util.Optional;
+
+import static org.apache.flink.table.types.logical.utils.LogicalTypeCasts.supportsImplicitCast;
+
+/**
+ * Specific {@link ArgumentTypeStrategy} for {@link BuiltInFunctionDefinitions#MAP_CONTAINS_KEY}.
+ */
+@Internal
+class MapKeyArgumentTypeStrategy implements ArgumentTypeStrategy {
+
+ @Override
+ public Optional<DataType> inferArgumentType(
+ CallContext callContext, int argumentPos, boolean throwOnFailure) {
+ final MapType mapType =
+ (MapType) callContext.getArgumentDataTypes().get(0).getLogicalType();
+ final LogicalType actualKeyType =
+ callContext.getArgumentDataTypes().get(argumentPos).getLogicalType();
+ LogicalType expectedKeyType = mapType.getKeyType();
Review Comment:
extract
```java
callContext.getArgumentDataTypes()
```
into a separate variable
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3836727901
##########
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;
+ }
+ }
Review Comment:
```suggestion
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;
}
}
}
```
isn't this more performant, given the fact that this is the code executing during runtime
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3925618928
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapContainsKeyFunction.java:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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 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;
+ }
Review Comment:
How about?
```suggestion
// A NULL needle matches a NULL key, unlike SQL `NULL = NULL` which yields UNKNOWN.
if (keysEqual(elementKey, needle)) {
return true;
}
```
```java
private boolean keysEqual(@Nullable Object elementKey, @Nullable Object needle) {
// A NULL needle matches a NULL key, unlike SQL `NULL = NULL` which yields UNKNOWN.
return needle == null ? elementKey == null : elementKey != null && isEqual(elementKey, needle);
}
```
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/MapKeyArgumentTypeStrategy.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.api.DataTypes;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionDefinition;
+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.MapType;
+
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.flink.table.types.logical.utils.LogicalTypeCasts.supportsImplicitCast;
+
+/**
+ * Specific {@link ArgumentTypeStrategy} for {@link BuiltInFunctionDefinitions#MAP_CONTAINS_KEY}.
+ */
Review Comment:
Maybe mention for which specific argument you are inferring
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/MapKeyArgumentTypeStrategy.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.api.DataTypes;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionDefinition;
+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.MapType;
+
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.flink.table.types.logical.utils.LogicalTypeCasts.supportsImplicitCast;
+
+/**
+ * Specific {@link ArgumentTypeStrategy} for {@link BuiltInFunctionDefinitions#MAP_CONTAINS_KEY}.
+ */
+@Internal
+class MapKeyArgumentTypeStrategy implements ArgumentTypeStrategy {
+
+ @Override
+ public Optional<DataType> inferArgumentType(
+ CallContext callContext, int argumentPos, boolean throwOnFailure) {
+ List<DataType> argumentTypes = callContext.getArgumentDataTypes();
+ final MapType mapType = (MapType) argumentTypes.get(0).getLogicalType();
Review Comment:
Extract `0` to a const and call it something like `MAP_ARGUMENT_POS`
##########
docs/data/sql_functions.yml:
##########
@@ -925,6 +925,23 @@ 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.
Review Comment:
Can you please provide an example for this too?
> 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.
##########
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:
In practice it almost never fires. The caller only calls `isEqual` when both key and needle are non-null (the `needle != null && elementKey != null` guard), and generated equality on two internal values of the same type is exception-free. So it's mostly defensive. It would realistically only trigger on an internal codegen or runtime bug while comparing two keys, or a JVM Error bubbling up through invoke.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "VasShabu (via GitHub)" <gi...@apache.org>.
VasShabu commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3933194138
##########
flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/InputTypeStrategiesTest.java:
##########
@@ -644,6 +644,43 @@ ANY, explicit(DataTypes.INT())
.expectArgumentTypes(
DataTypes.ARRAY(DataTypes.INT().notNull()).notNull(),
DataTypes.INT()),
+ TestSpec.forStrategy(
+ "MapKey argument type strategy implicitly casts the key",
+ sequence(
+ logical(LogicalTypeRoot.MAP),
+ SpecificInputTypeStrategies.MAP_KEY_ARG))
+ .calledWithArgumentTypes(
+ DataTypes.MAP(DataTypes.BIGINT().notNull(), DataTypes.STRING()),
+ DataTypes.INT().notNull())
+ .expectSignature("f(<MAP>, <MAP KEY>)")
+ .expectArgumentTypes(
+ DataTypes.MAP(DataTypes.BIGINT().notNull(), DataTypes.STRING()),
+ DataTypes.BIGINT().notNull()),
+ TestSpec.forStrategy(
+ "MapKey argument type strategy widens a NOT NULL key type "
+ + "for a nullable argument",
+ sequence(
+ logical(LogicalTypeRoot.MAP),
+ SpecificInputTypeStrategies.MAP_KEY_ARG))
+ .calledWithArgumentTypes(
+ DataTypes.MAP(DataTypes.BIGINT().notNull(), DataTypes.STRING())
+ .notNull(),
+ DataTypes.BIGINT())
+ .expectArgumentTypes(
+ DataTypes.MAP(DataTypes.BIGINT().notNull(), DataTypes.STRING())
+ .notNull(),
+ DataTypes.BIGINT()),
+ TestSpec.forStrategy(
+ "MapKey argument type strategy rejects a key that cannot be cast",
+ sequence(
+ logical(LogicalTypeRoot.MAP),
+ SpecificInputTypeStrategies.MAP_KEY_ARG))
+ .calledWithArgumentTypes(
+ DataTypes.MAP(DataTypes.INT(), DataTypes.STRING()),
+ DataTypes.BOOLEAN())
+ .expectErrorMessage(
+ "Invalid input arguments. Expected signatures are:\n"
Review Comment:
ah yes, I forgot to run the inputTypeStrategiesTest, before commiting I ran mapFunctionITCases, will make sure I run all tests relating to a feature prior to commit.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3932919191
##########
flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/InputTypeStrategiesTest.java:
##########
@@ -644,6 +644,43 @@ ANY, explicit(DataTypes.INT())
.expectArgumentTypes(
DataTypes.ARRAY(DataTypes.INT().notNull()).notNull(),
DataTypes.INT()),
+ TestSpec.forStrategy(
+ "MapKey argument type strategy implicitly casts the key",
+ sequence(
+ logical(LogicalTypeRoot.MAP),
+ SpecificInputTypeStrategies.MAP_KEY_ARG))
+ .calledWithArgumentTypes(
+ DataTypes.MAP(DataTypes.BIGINT().notNull(), DataTypes.STRING()),
+ DataTypes.INT().notNull())
+ .expectSignature("f(<MAP>, <MAP KEY>)")
+ .expectArgumentTypes(
+ DataTypes.MAP(DataTypes.BIGINT().notNull(), DataTypes.STRING()),
+ DataTypes.BIGINT().notNull()),
+ TestSpec.forStrategy(
+ "MapKey argument type strategy widens a NOT NULL key type "
+ + "for a nullable argument",
+ sequence(
+ logical(LogicalTypeRoot.MAP),
+ SpecificInputTypeStrategies.MAP_KEY_ARG))
+ .calledWithArgumentTypes(
+ DataTypes.MAP(DataTypes.BIGINT().notNull(), DataTypes.STRING())
+ .notNull(),
+ DataTypes.BIGINT())
+ .expectArgumentTypes(
+ DataTypes.MAP(DataTypes.BIGINT().notNull(), DataTypes.STRING())
+ .notNull(),
+ DataTypes.BIGINT()),
+ TestSpec.forStrategy(
+ "MapKey argument type strategy rejects a key that cannot be cast",
+ sequence(
+ logical(LogicalTypeRoot.MAP),
+ SpecificInputTypeStrategies.MAP_KEY_ARG))
+ .calledWithArgumentTypes(
+ DataTypes.MAP(DataTypes.INT(), DataTypes.STRING()),
+ DataTypes.BOOLEAN())
+ .expectErrorMessage(
+ "Invalid input arguments. Expected signatures are:\n"
Review Comment:
Test should be failing since you updated the msg to "Unsupported argument type.."
##########
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:
We have an additional if condition for each element in the smaller version. I think this is small performance diff since we also do things like "isEqual(elementKey, needle)" here. However, I agree it's a performance diff and Ramin seems to also be +1 to the previous version. I'd say @VasShabu can rollback to your suggestion and maybe only add a very short comment to explain why. Useful since there are other functions like ArrayContainsFunction that use the shorter version and one might wonder why we went with the lengthy version
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3896259792
##########
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? Something like
```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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40355][table] Add new MAP_CONTAINS_KEY function [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #28970:
URL: https://github.com/apache/flink/pull/28970#discussion_r3784326359
##########
docs/data/sql_functions.yml:
##########
@@ -925,6 +925,21 @@ 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)
Review Comment:
I am curious if the parameter key is a castable to the key in the map, does this work? Can we add tests and document what we find around this edge case?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org