[FLINK-37724] Adds AsyncTableFunction as a fully supported UDF type #26567
+3,285
−85
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| <table class="configuration table table-bordered"> | ||
| <thead> | ||
| <tr> | ||
| <th class="text-left" style="width: 20%">Key</th> | ||
| <th class="text-left" style="width: 15%">Default</th> | ||
| <th class="text-left" style="width: 10%">Type</th> | ||
| <th class="text-left" style="width: 55%">Description</th> | ||
| </tr> | ||
| </thead> | ||
| <tbody> | ||
| <tr> | ||
| <td><h5>sink.committer.retries</h5></td> | ||
| <td style="word-wrap: break-word;">10</td> | ||
| <td>Integer</td> | ||
| <td>The number of retries a Flink application attempts for committable operations (such as transactions) on retriable errors, as specified by the sink connector, before Flink fails and potentially restarts.</td> | ||
| </tr> | ||
| </tbody> | ||
| </table> | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -409,7 +409,7 @@ public class ExecutionConfigOptions { | |
| .intType() | ||
| .defaultValue(10) | ||
| .withDescription( | ||
| "The max number of async i/o operation that the async scalar function can trigger."); | ||
|
|
||
| @Documentation.TableOption(execMode = Documentation.ExecMode.STREAMING) | ||
| public static final ConfigOption<Duration> TABLE_EXEC_ASYNC_SCALAR_TIMEOUT = | ||
|
|
@@ -443,6 +443,49 @@ public class ExecutionConfigOptions { | |
| "The max number of async retry attempts to make before task " | ||
| + "execution is failed."); | ||
|
|
||
| // ------------------------------------------------------------------------ | ||
| // Async Table Function | ||
| // ------------------------------------------------------------------------ | ||
| @Documentation.TableOption(execMode = Documentation.ExecMode.STREAMING) | ||
| public static final ConfigOption<Integer> TABLE_EXEC_ASYNC_TABLE_BUFFER_CAPACITY = | ||
| key("table.exec.async-table.buffer-capacity") | ||
| .intType() | ||
| .defaultValue(10) | ||
| .withDescription( | ||
| "The max number of async i/o operations that the async table function can trigger."); | ||
|
|
||
| @Documentation.TableOption(execMode = Documentation.ExecMode.STREAMING) | ||
| public static final ConfigOption<Duration> TABLE_EXEC_ASYNC_TABLE_TIMEOUT = | ||
| key("table.exec.async-table.timeout") | ||
| .durationType() | ||
| .defaultValue(Duration.ofMinutes(3)) | ||
| .withDescription( | ||
| "The async timeout for the asynchronous operation to complete."); | ||
|
There was a problem hiding this comment. Is this the timeout for each retry or total timeout for all retries? I suppose it's for each retry |
||
|
|
||
| @Documentation.TableOption(execMode = Documentation.ExecMode.STREAMING) | ||
| public static final ConfigOption<RetryStrategy> TABLE_EXEC_ASYNC_TABLE_RETRY_STRATEGY = | ||
| key("table.exec.async-table.retry-strategy") | ||
| .enumType(RetryStrategy.class) | ||
| .defaultValue(RetryStrategy.FIXED_DELAY) | ||
| .withDescription( | ||
| "Restart strategy which will be used, FIXED_DELAY by default."); | ||
|
|
||
| @Documentation.TableOption(execMode = Documentation.ExecMode.STREAMING) | ||
| public static final ConfigOption<Duration> TABLE_EXEC_ASYNC_TABLE_RETRY_DELAY = | ||
| key("table.exec.async-table.retry-delay") | ||
| .durationType() | ||
| .defaultValue(Duration.ofMillis(100)) | ||
| .withDescription("The delay to wait before trying again."); | ||
|
|
||
| @Documentation.TableOption(execMode = Documentation.ExecMode.STREAMING) | ||
| public static final ConfigOption<Integer> TABLE_EXEC_ASYNC_TABLE_MAX_ATTEMPTS = | ||
| key("table.exec.async-table.max-attempts") | ||
| .intType() | ||
| .defaultValue(3) | ||
| .withDescription( | ||
| "The max number of async retry attempts to make before task " | ||
| + "execution is failed."); | ||
|
|
||
| // ------------------------------------------------------------------------ | ||
| // MiniBatch Options | ||
| // ------------------------------------------------------------------------ | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -48,13 +48,17 @@ | |
| import java.lang.reflect.ParameterizedType; | ||
| import java.lang.reflect.Type; | ||
| import java.util.Arrays; | ||
| import java.util.Collection; | ||
| import java.util.HashSet; | ||
| import java.util.List; | ||
| import java.util.Optional; | ||
| import java.util.Set; | ||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| import static org.apache.flink.api.java.typeutils.TypeExtractionUtils.getAllDeclaredMethods; | ||
| import static org.apache.flink.api.java.typeutils.TypeExtractionUtils.getParameterizedType; | ||
| import static org.apache.flink.api.java.typeutils.TypeExtractionUtils.isGenericOfClass; | ||
| import static org.apache.flink.util.Preconditions.checkState; | ||
|
|
||
| /** | ||
|
|
@@ -477,11 +481,12 @@ private static void validateImplementationMethods( | |
| validateImplementationMethod(functionClass, false, false, SCALAR_EVAL); | ||
| } else if (AsyncScalarFunction.class.isAssignableFrom(functionClass)) { | ||
| validateImplementationMethod(functionClass, false, false, ASYNC_SCALAR_EVAL); | ||
| validateAsyncImplementationMethod(functionClass, false, ASYNC_SCALAR_EVAL); | ||
| } else if (TableFunction.class.isAssignableFrom(functionClass)) { | ||
| validateImplementationMethod(functionClass, true, false, TABLE_EVAL); | ||
| } else if (AsyncTableFunction.class.isAssignableFrom(functionClass)) { | ||
| validateImplementationMethod(functionClass, true, false, ASYNC_TABLE_EVAL); | ||
| validateAsyncImplementationMethod(functionClass, true, ASYNC_TABLE_EVAL); | ||
| } else if (AggregateFunction.class.isAssignableFrom(functionClass)) { | ||
| validateImplementationMethod(functionClass, true, false, AGGREGATE_ACCUMULATE); | ||
| validateImplementationMethod(functionClass, true, true, AGGREGATE_RETRACT); | ||
|
|
@@ -541,7 +546,9 @@ private static void validateImplementationMethod( | |
| } | ||
|
|
||
| private static void validateAsyncImplementationMethod( | ||
| Class<? extends UserDefinedFunction> clazz, | ||
| boolean verifyFutureContainsCollection, | ||
| String... methodNameOptions) { | ||
| final Set<String> nameSet = new HashSet<>(Arrays.asList(methodNameOptions)); | ||
| final List<Method> methods = getAllDeclaredMethods(clazz); | ||
| for (Method method : methods) { | ||
|
|
@@ -558,18 +565,31 @@ private static void validateAsyncImplementationMethod( | |
| if (method.getParameterCount() >= 1) { | ||
| Type firstParam = method.getGenericParameterTypes()[0]; | ||
| firstParam = ExtractionUtils.resolveVariableWithClassContext(clazz, firstParam); | ||
| if (isGenericOfClass(CompletableFuture.class, firstParam)) { | ||
| Optional<ParameterizedType> parameterized = getParameterizedType(firstParam); | ||
| if (!verifyFutureContainsCollection) { | ||
| foundParam = true; | ||
| } else if (parameterized.isPresent() | ||
| && parameterized.get().getActualTypeArguments().length > 0) { | ||
| firstParam = parameterized.get().getActualTypeArguments()[0]; | ||
| if (isGenericOfClass(Collection.class, firstParam)) { | ||
| foundParam = true; | ||
| } | ||
| } | ||
| } | ||
| } | ||
| if (!foundParam) { | ||
| if (!verifyFutureContainsCollection) { | ||
| throw new ValidationException( | ||
| String.format( | ||
| "Method '%s' of function class '%s' must have a first argument of type java.util.concurrent.CompletableFuture.", | ||
| method.getName(), clazz.getName())); | ||
| } else { | ||
| throw new ValidationException( | ||
| String.format( | ||
| "Method '%s' of function class '%s' must have a first argument of type java.util.concurrent.CompletableFuture<java.util.Collection>.", | ||
| method.getName(), clazz.getName())); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -43,6 +43,7 @@ | |
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.stream.Stream; | ||
|
|
||
| import static org.apache.flink.api.java.typeutils.TypeExtractionUtils.getParameterizedType; | ||
| import static org.apache.flink.table.types.extraction.ExtractionUtils.collectAnnotationsOfClass; | ||
| import static org.apache.flink.table.types.extraction.ExtractionUtils.collectAnnotationsOfMethod; | ||
| import static org.apache.flink.table.types.extraction.ExtractionUtils.extractionError; | ||
|
|
@@ -208,7 +209,8 @@ static MethodVerification createParameterVerification(boolean requireAccumulator | |
| * Verification that checks a method by parameters (arguments only) with mandatory {@link | ||
| * CompletableFuture}. | ||
| */ | ||
| static MethodVerification createParameterAndCompletableFutureVerification( | ||
| Class<?> baseClass, @Nullable Class<?> nestedArgumentClass) { | ||
| return (method, state, arguments, result) -> { | ||
| checkNoState(state); | ||
| checkScalarArgumentsOnly(arguments); | ||
|
|
@@ -220,11 +222,24 @@ static MethodVerification createParameterAndCompletableFutureVerification(Class< | |
| final Class<?> resultClass = result.toClass(); | ||
| Type genericType = method.getGenericParameterTypes()[0]; | ||
| genericType = resolveVariableWithClassContext(baseClass, genericType); | ||
| Optional<ParameterizedType> parameterized = getParameterizedType(genericType); | ||
| if (!parameterized.isPresent()) { | ||
| throw extractionError( | ||
| "The method '%s' needs generic parameters for the CompletableFuture at position %d.", | ||
| method.getName(), 0); | ||
| } | ||
| // If nestedArgumentClass is given, it is assumed to be a generic parameters of | ||
| // argumentClass, also at the position genericPos | ||
| if (nestedArgumentClass != null) { | ||
| genericType = parameterized.get().getActualTypeArguments()[0]; | ||
| parameterized = getParameterizedType(genericType); | ||
| if (!parameterized.isPresent() | ||
| || !parameterized.get().getRawType().equals(nestedArgumentClass)) { | ||
| throw extractionError( | ||
| "The method '%s' expects nested generic type CompletableFuture<%s> for the %d arg.", | ||
| method.getName(), nestedArgumentClass.getName(), 0); | ||
| } | ||
| } | ||
| final Type returnType = ((ParameterizedType) genericType).getActualTypeArguments()[0]; | ||
| Class<?> returnTypeClass = getClassFromType(returnType); | ||
| // Parameters should be validated using strict autoboxing. | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -45,6 +45,7 @@ | |
|
|
||
| import javax.annotation.Nullable; | ||
|
|
||
| import java.util.Collection; | ||
| import java.util.LinkedHashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
|
|
@@ -106,7 +107,7 @@ public static TypeInference forAsyncScalarFunction( | |
| null, | ||
| null, | ||
| createOutputFromGenericInMethod(0, 0, true), | ||
| createParameterAndCompletableFutureVerification(function, null)); | ||
| return extractTypeInference(mappingExtractor, false); | ||
| } | ||
|
|
||
|
|
@@ -172,7 +173,8 @@ public static TypeInference forAsyncTableFunction( | |
| null, | ||
| null, | ||
| createOutputFromGenericInClass(AsyncTableFunction.class, 0, true), | ||
| createParameterAndCompletableFutureVerification( | ||
| function, Collection.class)); | ||
| return extractTypeInference(mappingExtractor, false); | ||
| } | ||
|
|
||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -198,7 +198,9 @@ private static TypeStrategy deriveSystemOutputStrategy( | |
| FunctionKind functionKind, | ||
| @Nullable List<StaticArgument> staticArgs, | ||
| TypeStrategy outputStrategy) { | ||
| if (functionKind != FunctionKind.TABLE | ||
| && functionKind != FunctionKind.PROCESS_TABLE | ||
| && functionKind != FunctionKind.ASYNC_TABLE) { | ||
| return outputStrategy; | ||
| } | ||
| return new SystemOutputStrategy(functionKind, staticArgs, outputStrategy); | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is unrelated the async table function?