[FLINK-37724] Adds AsyncTableFunction as a fully supported UDF type #26567

AlanConfluent · 2025-05-15T21:52:15Z

What is the purpose of the change

This aims to implement FLIP-498 https://cwiki.apache.org/confluence/display/FLINK/FLIP-498%3A+AsyncTableFunction+for+async+table+function+support.

In particular AsyncTableFunctions can now be defined as:

public class TestTableFunction extends AsyncTableFunction<String> {
    public void eval(CompletableFuture<Collection<String>> result, Integer i) {
        result.complete(Arrays.asList("Row1 " + i, "Row2 " + i));
    }
}

They can be registered in the catalog as with other UDFs, such as with:

 tEnv.createTemporarySystemFunction("func", new TestTableFunction());

There are a few new configs:

Name (Prefix table.exec.async-table)	Meaning
buffer-capacity	The number of outstanding requests the operator allows at once
timeout	The total time which can pass before the invocation (including retries) is considered timed out and task execution is failed
retry-strategy	FIXED_DELAY is for a retry after a fixed amount of time
retry-delay	The time to wait between retries for the FIXED_DELAY strategy. Could be the base delay time for a (not yet proposed) exponential backoff.
max-attempts	The maximum number of attempts while retrying.

Brief change log

Adds AsyncTableFunction support to type system, codegen, and runtime

Verifying this change

This change added tests and can be verified as follows:

Tested Udf validation (UserDefinedFunctionHelperTest)
Tested type inference (TypeInferenceExtractorTest.java)
Tested code generation for AsyncTableFunction
Tested new split rules for new async correlates
ITCases new async correlate use

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

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
- It should only affect the new code and shouldn't have detrimental effects to existing AsyncTableFunction use with lookup joins
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
- More documentation will follow as well beyond the configs

flinkbot · 2025-05-15T21:54:58Z

CI report:

e1b0ed6 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

lihaosky

Thanks Alan for the contribution! I made a brief pass and left some initial comments

lihaosky · 2025-05-22T20:13:43Z

docs/layouts/shortcodes/generated/sink_configuration.html

@@ -0,0 +1,18 @@
+<table class="configuration table table-bordered">


This is unrelated the async table function?

lihaosky · 2025-05-22T20:18:53Z

flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractionUtils.java

+    public static boolean isGenericOfClass(Class<?> clazz, Type type) {
+        Optional<ParameterizedType> parameterized = getParameterizedType(type);
+        return clazz.equals(type)
+                || parameterized.isPresent() && clazz.equals(parameterized.get().getRawType());
+    }
+
+    /**
+     * Returns an optional of a ParameterizedType, if that's what the type is.
+     *
+     * @param type The type to check
+     * @return optional which is present if the type is a ParameterizedType
+     */
+    public static Optional<ParameterizedType> getParameterizedType(Type type) {


Add unit test for these two function?

lihaosky · 2025-05-22T20:31:49Z

flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractionUtils.java

@@ -373,4 +374,28 @@ public static void validateLambdaType(Class<?> baseClass, Type t) {
                            + "Otherwise the type has to be specified explicitly using type information.");
        }
    }
+
+    /**
+     * Will return true if the type of the given generic class type.


Suggested change

* Will return true if the type of the given generic class type.

* Will return true if the type of the given generic class type matches clazz.

lihaosky · 2025-05-22T20:35:15Z

...k-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java

+                    .durationType()
+                    .defaultValue(Duration.ofMinutes(3))
+                    .withDescription(
+                            "The async timeout for the asynchronous operation to complete.");


Is this the timeout for each retry or total timeout for all retries? I suppose it's for each retry

lihaosky · 2025-05-22T20:51:48Z

...ble-common/src/test/java/org/apache/flink/table/functions/UserDefinedFunctionHelperTest.java

    /** Implementation method isn't void. */
    public static class NoFutureAsyncScalarFunction extends AsyncScalarFunction {
        public void eval(int i) {}
    }

+    /** Implementation method isn't void. */


first param isn't CompletableFuture<Collection>?

lihaosky · 2025-05-22T20:51:54Z

...ble-common/src/test/java/org/apache/flink/table/functions/UserDefinedFunctionHelperTest.java

+        public void eval(int i) {}
+    }
+
+    /** Implementation method isn't void. */


lihaosky · 2025-05-22T21:01:37Z

...ain/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecAsyncCorrelate.java

+        minPlanVersion = FlinkVersion.v1_19,
+        minStateVersion = FlinkVersion.v1_19)


Not exist yet in 1.19?

lihaosky · 2025-05-22T21:08:52Z

...nner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/AsyncCalcSplitRule.java

@@ -71,24 +72,35 @@ public class AsyncCalcSplitRule {
     * org.apache.flink.table.functions.AsyncScalarFunction}.


as well as AsyncTableFunction?

lihaosky · 2025-05-22T21:09:46Z

...nner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/AsyncCalcSplitRule.java

+        public AsyncRemoteCalcCallFinder() {
+            this(FunctionKind.ASYNC_SCALAR);
+        }


nit: Not sure if we want to default to scalar function given both are supported now. Requiring functionKind is probably fine

lihaosky · 2025-05-22T21:30:29Z

...test/java/org/apache/flink/table/planner/plan/rules/logical/AsyncCorrelateSplitRuleTest.java

+        String sqlQuery = "select * FROM MyTable, LATERAL TABLE(asyncTableFunc(scalar(a)))";
+        util.verifyRelPlan(sqlQuery);
+    }
+


Can you add

@Test public void testCorrelateWithCast() { String sqlQuery = "select * FROM MyTable, LATERAL TABLE(asyncTableFunc(cast(cast(a as int) as int)))"; util.verifyRelPlan(sqlQuery); }

There was a fix I made internally around cast

flinkbot added the component=<none> label May 15, 2025

[FLINK-37724] Adds AsyncTableFunction as a fully supported UDF type

e1b0ed6

AlanConfluent force-pushed the FLINK-37724 branch from 41f1eb3 to e1b0ed6 Compare May 16, 2025 19:41

lihaosky reviewed May 22, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[FLINK-37724] Adds AsyncTableFunction as a fully supported UDF type #26567

[FLINK-37724] Adds AsyncTableFunction as a fully supported UDF type #26567

AlanConfluent commented May 15, 2025

Uh oh!

flinkbot commented May 15, 2025 •

edited

Loading

Uh oh!

lihaosky left a comment

Uh oh!

lihaosky May 22, 2025

Uh oh!

lihaosky May 22, 2025

Uh oh!

lihaosky May 22, 2025

Uh oh!

lihaosky May 22, 2025

Uh oh!

lihaosky May 22, 2025

Uh oh!

lihaosky May 22, 2025

Uh oh!

lihaosky May 22, 2025

Uh oh!

lihaosky May 22, 2025

Uh oh!

lihaosky May 22, 2025

Uh oh!

lihaosky May 22, 2025

Uh oh!

Uh oh!

		@@ -0,0 +1,18 @@
		<table class="configuration table table-bordered">

	* Will return true if the type of the given generic class type.
	* Will return true if the type of the given generic class type matches clazz.

		minPlanVersion = FlinkVersion.v1_19,
		minStateVersion = FlinkVersion.v1_19)

		@@ -71,24 +72,35 @@ public class AsyncCalcSplitRule {
		* org.apache.flink.table.functions.AsyncScalarFunction}.

[FLINK-37724] Adds AsyncTableFunction as a fully supported UDF type #26567

Are you sure you want to change the base?

[FLINK-37724] Adds AsyncTableFunction as a fully supported UDF type #26567

Conversation

AlanConfluent commented May 15, 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 May 15, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

lihaosky 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!

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!

flinkbot commented May 15, 2025 •

edited

Loading