[FLINK-40355][table] Add new MAP_CONTAINS_KEY function by VasShabu · Pull Request #28970 · apache/flink

VasShabu · 2026-08-13T10:19:44Z

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.

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?

Yes (please specify the tool below)

Generated-by: Claude Opus 5

flinkbot · 2026-08-13T10:24:20Z

CI report:

119ef1b Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

davidradl · 2026-08-14T13:57:45Z

+        try {
+            return (boolean) equalityHandle.invoke(key, needle);
+        } catch (Throwable t) {
+            throw new FlinkRuntimeException(t);


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.

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.

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.

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

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.

davidradl · 2026-08-14T14:01:00Z

+    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.


when we say fails validation - I assume we should return false as per my other 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

gustavodemorais

Thanks for working on this, @VasShabu. I've added some comments

gustavodemorais · 2026-08-31T14:38:36Z

+                                $("f0").mapContainsKey(true),
+                                "Invalid input arguments. Expected signatures are:\n"
+                                        + "MAP_CONTAINS_KEY(map <MAP>, key <MAP KEY>)")


Can we throw a better error here?

If I understand it correctly, we're throwing because the boolean type is not valid here?

Not a blocker but can't we tell the type and be more specific as to why it failed

gustavodemorais · 2026-08-31T16:14:17Z

+        try {
+            return (boolean) equalityHandle.invoke(key, needle);
+        } catch (Throwable t) {
+            throw new FlinkRuntimeException(t);


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

gustavodemorais · 2026-08-31T16:14:43Z

+    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;
+    }


Can we simplify this? Something like

Suggested change

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;

}

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;

}

}

@gustavodemorais it was before, however it is less performant
since it is runtime I suggested to change it
#28970 (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

raminqaf

@VasShabu Thanks for the contribution. Left some comments!

raminqaf · 2026-09-03T14:41:35Z

+/**
+ * Specific {@link ArgumentTypeStrategy} for {@link BuiltInFunctionDefinitions#MAP_CONTAINS_KEY}.
+ */


Maybe mention for which specific argument you are inferring

raminqaf · 2026-09-03T14:51:06Z

+        try {
+            return (boolean) equalityHandle.invoke(key, needle);
+        } catch (Throwable t) {
+            throw new FlinkRuntimeException(t);


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.

# Conflicts: # flink-python/pyflink/table/expression.py # flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java # flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MapFunctionITCase.java

…small issues

…changes + simplify hotpath

gustavodemorais

Hey @VasShabu, thanks for the update! I think we're almost there. Added two more suggestions

gustavodemorais · 2026-09-04T10:00:58Z

+                                DataTypes.MAP(DataTypes.INT(), DataTypes.STRING()),
+                                DataTypes.BOOLEAN())
+                        .expectErrorMessage(
+                                "Invalid input arguments. Expected signatures are:\n"


Test should be failing since you updated the msg to "Unsupported argument type.."

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.

gustavodemorais · 2026-09-04T10:05:42Z

+    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;
+    }


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

…nputtyeStrat fix

snuyanzin · 2026-09-04T13:08:43Z

    }

-    /**
+    /*    *


why do we need this change?

VasShabu · 2026-09-08T14:03:36Z

@flinkbot run azure

VasShabu force-pushed the mapContainsKeyImplementation branch from b7b48ba to fb849a3 Compare August 14, 2026 10:59

davidradl reviewed Aug 14, 2026

View reviewed changes

snuyanzin reviewed Aug 22, 2026

View reviewed changes

Comment thread ...-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java Outdated

snuyanzin reviewed Aug 22, 2026

View reviewed changes

Comment thread .../flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java Outdated

snuyanzin reviewed Aug 22, 2026

View reviewed changes

Comment thread .../main/java/org/apache/flink/table/types/inference/strategies/MapKeyArgumentTypeStrategy.java

snuyanzin reviewed Aug 22, 2026

View reviewed changes

Comment thread ...me/src/main/java/org/apache/flink/table/runtime/functions/scalar/MapContainsKeyFunction.java Outdated

snuyanzin reviewed Aug 25, 2026

View reviewed changes

Comment thread docs/data/sql_functions.yml Outdated

snuyanzin reviewed Aug 25, 2026

View reviewed changes

Comment thread docs/data/sql_functions_zh.yml Outdated

gustavodemorais reviewed Aug 31, 2026

View reviewed changes

raminqaf reviewed Sep 3, 2026

View reviewed changes

VasShabu added 5 commits September 3, 2026 17:04

[FLINK-40355][table] altered Docs, made it more detailed

3d2b5fd

[FLINK-40355][table] improved docs clarity, improved hotpath + other …

77209df

…small issues

[FLINK-40355][table] Solved small nit e.g.

bf00b15

[FLINK-40355][table] made invalid type error more descriptive + docs …

f1dab7e

…changes + simplify hotpath

github-actions Bot added the community-reviewed PR has been reviewed by the community. label Sep 3, 2026

VasShabu force-pushed the mapContainsKeyImplementation branch from 8912224 to f1dab7e Compare September 3, 2026 16:22

gustavodemorais reviewed Sep 4, 2026

View reviewed changes

[FLINK-40355][table] added example type casting, optimised hotpath, i…

fc8f2d2

…nputtyeStrat fix

snuyanzin reviewed Sep 4, 2026

View reviewed changes

[FLINK-40355][table] fixed small nit with comments.

119ef1b

VasShabu force-pushed the mapContainsKeyImplementation branch from 64edf98 to 119ef1b Compare September 8, 2026 14:35

                   }
-                  /**
+                  /*    *

Uh oh!

Conversation

VasShabu commented Aug 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Was generative AI tooling used to co-author this PR?

Uh oh!

flinkbot commented Aug 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

gustavodemorais left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gustavodemorais Aug 31, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

snuyanzin Sep 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

raminqaf left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

gustavodemorais left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

VasShabu commented Sep 8, 2026

Uh oh!

Reviewers

Assignees

Labels

VasShabu commented Aug 13, 2026 •

edited

Loading

flinkbot commented Aug 13, 2026 •

edited

Loading

gustavodemorais Aug 31, 2026 •

edited

Loading

snuyanzin Sep 3, 2026 •

edited

Loading