Skip to content

[FLINK-40355][table] Add new MAP_CONTAINS_KEY function - #28970

Open
VasShabu wants to merge 2 commits into
apache:masterfrom
VasShabu:mapContainsKeyImplementation
Open

[FLINK-40355][table] Add new MAP_CONTAINS_KEY function#28970
VasShabu wants to merge 2 commits into
apache:masterfrom
VasShabu:mapContainsKeyImplementation

Conversation

@VasShabu

@VasShabu VasShabu commented Aug 13, 2026

Copy link
Copy Markdown
Contributor

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

flinkbot commented Aug 13, 2026

Copy link
Copy Markdown
Collaborator

CI report:

Bot commands The @flinkbot bot supports the following commands:
  • @flinkbot run azure re-run the last Azure build

@VasShabu
VasShabu force-pushed the mapContainsKeyImplementation branch from b7b48ba to fb849a3 Compare August 14, 2026 10:59
try {
return (boolean) equalityHandle.invoke(key, needle);
} catch (Throwable t) {
throw new FlinkRuntimeException(t);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

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.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

when we say fails validation - I assume we should return false as per my other comment

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants