[FLINK-38257][SQL] Add module dynamic function loading possibility #26933

gaborgsomogyi · 2025-08-22T12:05:44Z

What is the purpose of the change

Modules like the state processor API could add SQL functions which are coming from external connectors like Kafka, Iceberg, etc... It would be good to add a generic way to add such methods. The intended end-user interaction would look like the following:

Add Kafka connector jar to the classpath
LOAD MODULE state
SELECT * FROM get_kafka_offsets(...)

In this PR I've added purely the possibility to load SQL functions with service loader.

Brief change log

Added module dynamic function loading possibility.

Verifying this change

Added new automated test.

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: no
The runtime per-record code paths (performance sensitive): no
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? not applicable

flinkbot · 2025-08-22T12:13:06Z

CI report:

032dd0e Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

zch93

The PR cleanly introduces a minimal SPI and wires the state module to load connector-owned TableFunctions at runtime via ServiceLoader, moving the built-in savepoint_metadata into its own TF class and proving the mechanism with an example + test. It achieves the stated goal with a small, well-scoped change and keeps Flink core free of connector dependencies. I wrote only some nits and questions.

zch93 · 2025-08-22T13:42:46Z

...rocessing-api/src/main/java/org/apache/flink/state/table/SavepointMetadataTableFunction.java

@@ -41,6 +46,39 @@
                                + "operator-coordinator-state-size-in-bytes BIGINT NOT NULL, "
                                + "operator-total-size-in-bytes BIGINT NOT NULL>"))
 public class SavepointMetadataTableFunction extends TableFunction<Row> {
+
+    public static final BuiltInFunctionDefinition SAVEPOINT_METADATA =


nice to keep the definition near the code. do you think we should also mark that the function should not be pre-calculated by the optimizer? i mean to make it clear that flink must always run it at query time, not try to be clever and “pre-compute” it

This one is not yet clear🤷🏻‍♂️

maybe i misunderstood what happens here with savepoint_metadata, but flink's planner can constant-fold and cache function results at plan time if it believes a function is deterministic. that is correct for pure math/string functions, but could be problematic for something like this savepoint_metadata, which read external, changing state (fs/checkpoint files). and between planning and exec, or between two exec, the metadata could change. so i just thought whether we should tell the planner this function is non-deterministic, so it will not pre-evaluate or cache it, and will always run it at exec time. idk whether the BuiltInFunctionDefinition.newBuilder() can supports to call .deterministic(false), or use and override the UDF isDeterministic()

That's fair point. I'm not fan of fixing something which is unrelated but this is so tiny that adding notDeterministic() is fine.

Though it worth the discussion separate whether the planner caches it with input params or not. I would guess yes, otherwise it just doesn't make sense from usage point of view. If the params are stored together with the result then adding notDeterministic() doesn't hurt but also not helping.

...link-state-processing-api/src/main/java/org/apache/flink/state/table/module/StateModule.java

zch93 · 2025-08-22T13:52:03Z

...link-state-processing-api/src/main/java/org/apache/flink/state/table/module/StateModule.java

-                                                    "operator-total-size-in-bytes",
-                                                    DataTypes.BIGINT().notNull()))))
-                    .build();
-
    public static final StateModule INSTANCE = new StateModule();


nit: this is not part of this pr, but it seems like the constructor does classpath scanning, and that is fine if the module is loaded once, but worth a note: if someone repeatedly constructs modules in tests, scanning happens each time

I think testing the module as it's going to behave in PROD is something which is desired even if it requires some time.

yeah, totally fair
my (out-of-scope) nit was only about to maybe add a short javadoc on StateModule noting that the constructor performs ServiceLoader discovery, so repeated instantiation re-scans the classpath, and maybe in some tests, future authors can prefer StateModule.INSTANCE over new StateModule() to avoid unintentional repeated scans

zch93 · 2025-08-22T13:56:58Z

...s/flink-state-processing-api/src/test/java/org/apache/flink/state/table/StateModuleTest.java

+    @Test
+    public void testDynamicBuiltinFunction() throws Exception {
+        Configuration config = new Configuration();
+        config.set(RUNTIME_MODE, RuntimeExecutionMode.BATCH);


do you think we should add a test with streaming mode too and add a negative test for name collisions?

Name collision test has been added but streaming would add no value here.

zch93 · 2025-08-22T14:00:26Z

...java/org/apache/flink/state/table/module/ExampleDynamicBuiltInFunctionDefinitionFactory.java

+
+import java.util.List;
+
+public class ExampleDynamicBuiltInFunctionDefinitionFactory


api looks minimal and clear.
nit: maybe we can standardize to . and document that function names returned must be unique within the module (and what happens otherwise)
nit: maybe we can add an optional ModuleContext param (classloader, config). not needed now, but helps if a provider later needs context without breaking the spi

This API is now fully experimental and later subject to change. If the time comes and ModuleContext or something else is required then we're going to add it. Finding out later needs is just not working in general 🙂
Related the names there is already a builtin function name suggestion which already given. Of course this is not documented because it's not public API and I would stick to that since it exists from ages.

makes sense, thanks for clarifying
on the “built-in function name suggestion”: understood, it is internal and long-standing, so i am fine relying on it. if feasible, a short internal comment pointing to where that suggestion lives would help new contributors

bottom line: +1 to keep the api minimal as proposed

zch93 suggested changes Aug 22, 2025

View reviewed changes

github-actions bot added the community-reviewed PR has been reviewed by the community. label Aug 22, 2025

gaborgsomogyi force-pushed the FLINK-38257 branch from 7d6027e to 2fe5c62 Compare August 22, 2025 17:33

github-actions bot added community-reviewed PR has been reviewed by the community. and removed community-reviewed PR has been reviewed by the community. labels Aug 23, 2025

gaborgsomogyi force-pushed the FLINK-38257 branch from 2fe5c62 to 6030634 Compare August 23, 2025 07:09

github-actions bot added community-reviewed PR has been reviewed by the community. and removed community-reviewed PR has been reviewed by the community. labels Aug 23, 2025

[FLINK-38257][SQL] Add module dynamic function loading possibility

032dd0e

gaborgsomogyi force-pushed the FLINK-38257 branch from 6030634 to 032dd0e Compare August 24, 2025 14:14

github-actions bot added community-reviewed PR has been reviewed by the community. and removed community-reviewed PR has been reviewed by the community. labels Aug 25, 2025


		import java.util.List;

		public class ExampleDynamicBuiltInFunctionDefinitionFactory

[FLINK-38257][SQL] Add module dynamic function loading possibility #26933

Are you sure you want to change the base?

[FLINK-38257][SQL] Add module dynamic function loading possibility #26933

Conversation

gaborgsomogyi commented Aug 22, 2025

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

Uh oh!

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

Uh oh!

CI report:

Uh oh!

zch93 left a comment • 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!

Choose a reason for hiding this comment

Uh oh!

zch93 Aug 24, 2025 • 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!

Choose a reason for hiding this comment

Uh oh!

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!

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!

zch93 Aug 24, 2025 • 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!

Uh oh!

flinkbot commented Aug 22, 2025 •

edited

Loading

zch93 left a comment •

edited

Loading

zch93 Aug 24, 2025 •

edited

Loading

zch93 Aug 24, 2025 •

edited

Loading