You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2024/01/15 14:41:48 UTC
(flink) branch master updated: [FLINK-34027][table] Introduce AsyncScalarFunction as a new UDF type (#23975)
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new b666de974b5 [FLINK-34027][table] Introduce AsyncScalarFunction as a new UDF type (#23975)
b666de974b5 is described below
commit b666de974b5dbc6e0d6710baececb1d6ede9b769
Author: Alan Sheinberg <57...@users.noreply.github.com>
AuthorDate: Mon Jan 15 06:41:41 2024 -0800
[FLINK-34027][table] Introduce AsyncScalarFunction as a new UDF type (#23975)
---
.../generated/execution_config_configuration.html | 30 ++
.../table/api/config/ExecutionConfigOptions.java | 53 ++
.../flink/table/functions/AsyncScalarFunction.java | 103 ++++
.../apache/flink/table/functions/FunctionKind.java | 2 +
.../flink/table/functions/UserDefinedFunction.java | 1 +
.../table/functions/UserDefinedFunctionHelper.java | 42 ++
.../table/types/extraction/DataTypeExtractor.java | 41 ++
.../table/types/extraction/ExtractionUtils.java | 128 +++++
.../types/extraction/FunctionMappingExtractor.java | 93 +++-
.../types/extraction/TypeInferenceExtractor.java | 19 +
.../functions/UserDefinedFunctionHelperTest.java | 59 +++
.../types/extraction/DataTypeExtractorTest.java | 31 ++
.../types/extraction/ExtractionUtilsTest.java | 117 +++++
.../extraction/TypeInferenceExtractorTest.java | 149 ++++++
.../catalog/FunctionCatalogOperatorTable.java | 1 +
.../table/planner/codegen/AsyncCodeGenerator.java | 225 +++++++++
.../functions/bridging/BridgingSqlFunction.java | 4 +-
.../nodes/exec/common/CommonExecAsyncCalc.java | 163 ++++++
.../nodes/exec/stream/StreamExecAsyncCalc.java | 82 +++
.../physical/stream/StreamPhysicalAsyncCalc.java | 85 ++++
.../plan/rules/logical/AsyncCalcSplitRule.java | 203 ++++++++
.../rules/logical/PythonCorrelateSplitRule.java | 3 +-
.../stream/StreamPhysicalAsyncCalcRule.java | 64 +++
.../flink/table/planner/plan/utils/AsyncUtil.java | 188 +++++++
.../planner/plan/utils/ExecNodeMetadataUtil.java | 2 +
.../table/planner/codegen/ExpressionReducer.scala | 19 +-
.../codegen/calls/BridgingFunctionGenUtil.scala | 40 +-
.../planner/plan/rules/FlinkStreamRuleSets.scala | 36 +-
.../plan/rules/logical/PythonCalcSplitRule.scala | 463 +----------------
.../plan/rules/logical/RemoteCalcCallFinder.java} | 28 +-
...lcSplitRule.scala => RemoteCalcSplitRule.scala} | 293 +++++------
.../physical/stream/StreamPhysicalCalcRule.scala | 5 +-
.../planner/codegen/AsyncCodeGeneratorTest.java | 202 ++++++++
.../plan/rules/logical/AsyncCalcSplitRuleTest.java | 276 ++++++++++
.../runtime/stream/table/AsyncCalcITCase.java | 361 +++++++++++++
.../plan/rules/logical/AsyncCalcSplitRuleTest.xml | 556 +++++++++++++++++++++
.../operators/calc/async/AsyncFunctionRunner.java | 68 +++
.../calc/async/DelegatingAsyncResultFuture.java | 84 ++++
.../operators/calc/async/RetryPredicates.java | 55 ++
39 files changed, 3752 insertions(+), 622 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/execution_config_configuration.html b/docs/layouts/shortcodes/generated/execution_config_configuration.html
index c29ad615227..1ce188e0729 100644
--- a/docs/layouts/shortcodes/generated/execution_config_configuration.html
+++ b/docs/layouts/shortcodes/generated/execution_config_configuration.html
@@ -26,6 +26,36 @@
<td>Duration</td>
<td>The async timeout for the asynchronous operation to complete.</td>
</tr>
+ <tr>
+ <td><h5>table.exec.async-scalar.buffer-capacity</h5><br> <span class="label label-primary">Streaming</span></td>
+ <td style="word-wrap: break-word;">10</td>
+ <td>Integer</td>
+ <td>The max number of async i/o operation that the async lookup join can trigger.</td>
+ </tr>
+ <tr>
+ <td><h5>table.exec.async-scalar.max-attempts</h5><br> <span class="label label-primary">Streaming</span></td>
+ <td style="word-wrap: break-word;">3</td>
+ <td>Integer</td>
+ <td>The max number of async retry attempts to make before task execution is failed.</td>
+ </tr>
+ <tr>
+ <td><h5>table.exec.async-scalar.retry-delay</h5><br> <span class="label label-primary">Streaming</span></td>
+ <td style="word-wrap: break-word;">100 ms</td>
+ <td>Duration</td>
+ <td>The delay to wait before trying again.</td>
+ </tr>
+ <tr>
+ <td><h5>table.exec.async-scalar.retry-strategy</h5><br> <span class="label label-primary">Streaming</span></td>
+ <td style="word-wrap: break-word;">FIXED_DELAY</td>
+ <td><p>Enum</p></td>
+ <td>Restart strategy which will be used, FIXED_DELAY by default.<br /><br />Possible values:<ul><li>"NO_RETRY"</li><li>"FIXED_DELAY"</li></ul></td>
+ </tr>
+ <tr>
+ <td><h5>table.exec.async-scalar.timeout</h5><br> <span class="label label-primary">Streaming</span></td>
+ <td style="word-wrap: break-word;">3 min</td>
+ <td>Duration</td>
+ <td>The async timeout for the asynchronous operation to complete.</td>
+ </tr>
<tr>
<td><h5>table.exec.deduplicate.insert-update-after-sensitive-enabled</h5><br> <span class="label label-primary">Streaming</span></td>
<td style="word-wrap: break-word;">true</td>
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java
index 1d0dd1fc557..e211f9c4607 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java
@@ -344,6 +344,49 @@ public class ExecutionConfigOptions {
+ "If set to ALLOW_UNORDERED, will attempt to use {@see AsyncDataStream.OutputMode.UNORDERED} when it does not "
+ "affect the correctness of the result, otherwise ORDERED will be still used.");
+ // ------------------------------------------------------------------------
+ // Async Scalar Function
+ // ------------------------------------------------------------------------
+ @Documentation.TableOption(execMode = Documentation.ExecMode.STREAMING)
+ public static final ConfigOption<Integer> TABLE_EXEC_ASYNC_SCALAR_BUFFER_CAPACITY =
+ key("table.exec.async-scalar.buffer-capacity")
+ .intType()
+ .defaultValue(10)
+ .withDescription(
+ "The max number of async i/o operation that the async lookup join can trigger.");
+
+ @Documentation.TableOption(execMode = Documentation.ExecMode.STREAMING)
+ public static final ConfigOption<Duration> TABLE_EXEC_ASYNC_SCALAR_TIMEOUT =
+ key("table.exec.async-scalar.timeout")
+ .durationType()
+ .defaultValue(Duration.ofMinutes(3))
+ .withDescription(
+ "The async timeout for the asynchronous operation to complete.");
+
+ @Documentation.TableOption(execMode = Documentation.ExecMode.STREAMING)
+ public static final ConfigOption<RetryStrategy> TABLE_EXEC_ASYNC_SCALAR_RETRY_STRATEGY =
+ key("table.exec.async-scalar.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_SCALAR_RETRY_DELAY =
+ key("table.exec.async-scalar.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_SCALAR_MAX_ATTEMPTS =
+ key("table.exec.async-scalar.max-attempts")
+ .intType()
+ .defaultValue(3)
+ .withDescription(
+ "The max number of async retry attempts to make before task "
+ + "execution is failed.");
+
// ------------------------------------------------------------------------
// MiniBatch Options
// ------------------------------------------------------------------------
@@ -692,6 +735,16 @@ public class ExecutionConfigOptions {
ALLOW_UNORDERED
}
+ /** Retry strategy in the case of failure. */
+ @PublicEvolving
+ public enum RetryStrategy {
+ /** When a failure occurs, don't retry. */
+ NO_RETRY,
+
+ /** A fixed delay before retrying again. */
+ FIXED_DELAY
+ }
+
/** Determine if CAST operates using the legacy behaviour or the new one. */
@Deprecated
public enum LegacyCastBehaviour implements DescribedEnum {
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/AsyncScalarFunction.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/AsyncScalarFunction.java
new file mode 100644
index 00000000000..0ebfeed5557
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/AsyncScalarFunction.java
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.functions;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.catalog.DataTypeFactory;
+import org.apache.flink.table.types.extraction.TypeInferenceExtractor;
+import org.apache.flink.table.types.inference.TypeInference;
+
+/**
+ * Base class for a user-defined scalar function which returns results asynchronously. A
+ * user-defined scalar function maps zero, one, or multiple scalar values to a new scalar value.
+ *
+ * <p>The behavior of a {@link AsyncScalarFunction} can be defined by implementing a custom
+ * evaluation method. An evaluation method must be declared publicly and named <code>eval</code>.
+ * Evaluation methods can also be overloaded by implementing multiple methods named <code>eval
+ * </code>. The first argument of the method must be a {@link
+ * java.util.concurrent.CompletableFuture} of the return type. The method body must complete the
+ * future either if there is a result or if it completes exceptionally.
+ *
+ * <p>By default, input and output data types are automatically extracted using reflection. If the
+ * reflective information is not sufficient, it can be supported and enriched with {@link
+ * org.apache.flink.table.annotation.DataTypeHint} and {@link
+ * org.apache.flink.table.annotation.FunctionHint} annotations.
+ *
+ * <p>The following examples show how to specify an async scalar function:
+ *
+ * <pre>{@code
+ * // a function that accepts two INT arguments and computes a sum
+ * class SumFunction extends AsyncScalarFunction {
+ * public void eval(CompletableFuture<Integer> future, Integer a, Integer b) {
+ * return future.complete(a + b);
+ * }
+ * }
+ *
+ * // a function that accepts either INT NOT NULL or BOOLEAN NOT NULL and computes a STRING
+ * class StringifyFunction extends AsyncScalarFunction {
+ * public void eval(CompletableFuture<String> future, int i) {
+ * return future.complete(String.valueOf(i));
+ * }
+ * public void eval(CompletableFuture<String> future, boolean b) {
+ * return future.complete(String.valueOf(b));
+ * }
+ * }
+ *
+ * // a function that accepts either INT or BOOLEAN and computes a STRING using function hints
+ * @FunctionHint(input = [@DataTypeHint("INT")])
+ * @FunctionHint(input = [@DataTypeHint("BOOLEAN")])
+ * class StringifyFunction extends AsyncScalarFunction {
+ * public void eval(CompletableFuture<String> future, Object o) {
+ * return future.complete(o.toString());
+ * }
+ * }
+ *
+ * // a function that accepts any data type as argument and computes a STRING
+ * class StringifyFunction extends AsyncScalarFunction {
+ * public void eval(CompletableFuture<String> future,
+ * @DataTypeHint(inputGroup = InputGroup.ANY) Object o) {
+ * return future.complete(o.toString());
+ * }
+ * }
+ *
+ * // a function that accepts an arbitrary number of BIGINT values and computes a DECIMAL(10, 4)
+ * class SumFunction extends AsyncScalarFunction {
+ * public void eval(@DataTypeHint("DECIMAL(10, 4)") CompletableFuture<BigDecimal> future,
+ * Long... values) {
+ * // ...
+ * }
+ * }
+ * }</pre>
+ *
+ * <p>For storing a user-defined function in a catalog, the class must have a default constructor
+ * and must be instantiable during runtime. Anonymous functions in Table API can only be persisted
+ * if the function is not stateful (i.e. containing only transient and static fields).
+ */
+@PublicEvolving
+public class AsyncScalarFunction extends UserDefinedFunction {
+ @Override
+ public final FunctionKind getKind() {
+ return FunctionKind.ASYNC_SCALAR;
+ }
+
+ @Override
+ public TypeInference getTypeInference(DataTypeFactory typeFactory) {
+ return TypeInferenceExtractor.forAsyncScalarFunction(typeFactory, getClass());
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/FunctionKind.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/FunctionKind.java
index 32fc9fbcf23..025a4fee026 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/FunctionKind.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/FunctionKind.java
@@ -25,6 +25,8 @@ import org.apache.flink.annotation.PublicEvolving;
public enum FunctionKind {
SCALAR,
+ ASYNC_SCALAR,
+
TABLE,
ASYNC_TABLE,
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunction.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunction.java
index 0231835079a..5dd8ca08460 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunction.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunction.java
@@ -50,6 +50,7 @@ import static org.apache.flink.table.functions.UserDefinedFunctionHelper.isClass
* </ul>
*
* @see ScalarFunction
+ * @see AsyncScalarFunction
* @see TableFunction
* @see AsyncTableFunction
* @see AggregateFunction
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java
index 279d9baeb7b..354146d70c3 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java
@@ -44,10 +44,13 @@ import javax.annotation.Nullable;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import static org.apache.flink.api.java.typeutils.TypeExtractionUtils.getAllDeclaredMethods;
@@ -67,6 +70,8 @@ public final class UserDefinedFunctionHelper {
public static final String SCALAR_EVAL = "eval";
+ public static final String ASYNC_SCALAR_EVAL = "eval";
+
public static final String TABLE_EVAL = "eval";
public static final String AGGREGATE_ACCUMULATE = "accumulate";
@@ -459,6 +464,9 @@ public final class UserDefinedFunctionHelper {
Class<? extends UserDefinedFunction> functionClass) {
if (ScalarFunction.class.isAssignableFrom(functionClass)) {
validateImplementationMethod(functionClass, false, false, SCALAR_EVAL);
+ } else if (AsyncScalarFunction.class.isAssignableFrom(functionClass)) {
+ validateImplementationMethod(functionClass, false, false, ASYNC_SCALAR_EVAL);
+ validateAsyncImplementationMethod(functionClass, ASYNC_SCALAR_EVAL);
} else if (TableFunction.class.isAssignableFrom(functionClass)) {
validateImplementationMethod(functionClass, true, false, TABLE_EVAL);
} else if (AsyncTableFunction.class.isAssignableFrom(functionClass)) {
@@ -521,6 +529,40 @@ public final class UserDefinedFunctionHelper {
}
}
+ private static void validateAsyncImplementationMethod(
+ Class<? extends UserDefinedFunction> clazz, String... methodNameOptions) {
+ final Set<String> nameSet = new HashSet<>(Arrays.asList(methodNameOptions));
+ final List<Method> methods = getAllDeclaredMethods(clazz);
+ for (Method method : methods) {
+ if (!nameSet.contains(method.getName())) {
+ continue;
+ }
+ if (!method.getReturnType().equals(Void.TYPE)) {
+ throw new ValidationException(
+ String.format(
+ "Method '%s' of function class '%s' must be void.",
+ method.getName(), clazz.getName()));
+ }
+ boolean foundParam = false;
+ if (method.getParameterCount() >= 1) {
+ Type firstParam = method.getGenericParameterTypes()[0];
+ firstParam = ExtractionUtils.resolveVariableWithClassContext(clazz, firstParam);
+ if (CompletableFuture.class.equals(firstParam)
+ || firstParam instanceof ParameterizedType
+ && CompletableFuture.class.equals(
+ ((ParameterizedType) firstParam).getRawType())) {
+ foundParam = true;
+ }
+ }
+ if (!foundParam) {
+ 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()));
+ }
+ }
+ }
+
/** Checks if a user-defined function can be easily instantiated. */
private static void validateInstantiation(Class<?> clazz, boolean requiresDefaultConstructor) {
if (!InstantiationUtil.isPublic(clazz)) {
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/DataTypeExtractor.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/DataTypeExtractor.java
index afa8a8670df..e1f0122d10e 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/DataTypeExtractor.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/DataTypeExtractor.java
@@ -68,6 +68,7 @@ import static org.apache.flink.table.types.extraction.ExtractionUtils.extraction
import static org.apache.flink.table.types.extraction.ExtractionUtils.hasInvokableConstructor;
import static org.apache.flink.table.types.extraction.ExtractionUtils.isStructuredFieldMutable;
import static org.apache.flink.table.types.extraction.ExtractionUtils.resolveVariable;
+import static org.apache.flink.table.types.extraction.ExtractionUtils.resolveVariableWithClassContext;
import static org.apache.flink.table.types.extraction.ExtractionUtils.toClass;
import static org.apache.flink.table.types.extraction.ExtractionUtils.validateStructuredClass;
import static org.apache.flink.table.types.extraction.ExtractionUtils.validateStructuredFieldReadability;
@@ -155,6 +156,46 @@ public final class DataTypeExtractor {
paramPos, method.getName(), baseClass.getName()));
}
+ /**
+ * Extracts a data type from a method parameter by considering surrounding classes and parameter
+ * annotation. This version assumes that the parameter is a generic type, and uses the generic
+ * position type as the extracted data type. For example, if the parameter is a
+ * CompletableFuture<Long> and genericPos is 0, it will extract Long.
+ */
+ public static DataType extractFromGenericMethodParameter(
+ DataTypeFactory typeFactory,
+ Class<?> baseClass,
+ Method method,
+ int paramPos,
+ int genericPos) {
+
+ Type parameterType = method.getGenericParameterTypes()[paramPos];
+ parameterType = resolveVariableWithClassContext(baseClass, parameterType);
+ if (!(parameterType instanceof ParameterizedType)) {
+ throw extractionError(
+ "The method '%s' needs generic parameters for the %d arg.",
+ method.getName(), paramPos);
+ }
+ final Type genericParameterType =
+ ((ParameterizedType) parameterType).getActualTypeArguments()[genericPos];
+ final Parameter parameter = method.getParameters()[paramPos];
+ final DataTypeHint hint = parameter.getAnnotation(DataTypeHint.class);
+ final DataTypeTemplate template;
+ if (hint != null) {
+ template = DataTypeTemplate.fromAnnotation(typeFactory, hint);
+ } else {
+ template = DataTypeTemplate.fromDefaults();
+ }
+ return extractDataTypeWithClassContext(
+ typeFactory,
+ template,
+ baseClass,
+ genericParameterType,
+ String.format(
+ " in generic parameter %d of method '%s' in class '%s'",
+ paramPos, method.getName(), baseClass.getName()));
+ }
+
/**
* Extracts a data type from a method return type by considering surrounding classes and method
* annotation.
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/ExtractionUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/ExtractionUtils.java
index 1b27c469667..b0b9a74f636 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/ExtractionUtils.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/ExtractionUtils.java
@@ -37,9 +37,11 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.io.InputStream;
import java.lang.annotation.Annotation;
+import java.lang.reflect.Array;
import java.lang.reflect.Constructor;
import java.lang.reflect.Executable;
import java.lang.reflect.Field;
+import java.lang.reflect.GenericArrayType;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.lang.reflect.Parameter;
@@ -348,6 +350,53 @@ public final class ExtractionUtils {
}
}
+ /** Resolves a variable type while accepting a context for resolution. */
+ public static Type resolveVariableWithClassContext(@Nullable Type contextType, Type type) {
+ final List<Type> typeHierarchy;
+ if (contextType != null) {
+ typeHierarchy = collectTypeHierarchy(contextType);
+ } else {
+ typeHierarchy = Collections.emptyList();
+ }
+ if (!containsTypeVariable(type)) {
+ return type;
+ }
+ if (type instanceof TypeVariable) {
+ return resolveVariable(typeHierarchy, (TypeVariable<?>) type);
+ } else if (type instanceof ParameterizedType) {
+ ParameterizedType parameterizedType = (ParameterizedType) type;
+ List<Type> paramTypes = new ArrayList<>();
+ for (Type paramType : parameterizedType.getActualTypeArguments()) {
+ paramType = resolveVariableWithClassContext(contextType, paramType);
+ paramTypes.add(paramType);
+ }
+ return new ParameterizedTypeImpl(
+ paramTypes.toArray(paramTypes.toArray(new Type[0])),
+ parameterizedType.getRawType(),
+ parameterizedType.getOwnerType());
+ } else if (type instanceof GenericArrayType) {
+ Type componentType =
+ resolveVariableWithClassContext(
+ contextType, ((GenericArrayType) type).getGenericComponentType());
+ return new GenericArrayTypeImpl(componentType);
+ } else {
+ return type;
+ }
+ }
+
+ /** Gets the associated class type from a Type parameter. */
+ public static Class<?> getClassFromType(Type type) {
+ if (type instanceof ParameterizedType) {
+ return (Class<?>) ((ParameterizedType) type).getRawType();
+ } else if (type instanceof GenericArrayType) {
+ return Array.newInstance(
+ (Class<?>) ((GenericArrayType) type).getGenericComponentType(), 0)
+ .getClass();
+ }
+ // Otherwise assume it's a basic type that can be cast.
+ return (Class<?>) type;
+ }
+
// --------------------------------------------------------------------------------------------
// Methods intended for this package
// --------------------------------------------------------------------------------------------
@@ -1002,6 +1051,85 @@ public final class ExtractionUtils {
return Class.forName(name, initialize, classLoader);
}
+ /** Utility to know if the type contains a type variable that needs to be resolved. */
+ private static boolean containsTypeVariable(Type type) {
+ if (type instanceof TypeVariable) {
+ return true;
+ } else if (type instanceof ParameterizedType) {
+ return Arrays.stream(((ParameterizedType) type).getActualTypeArguments())
+ .anyMatch(ExtractionUtils::containsTypeVariable);
+ } else if (type instanceof GenericArrayType) {
+ return containsTypeVariable(((GenericArrayType) type).getGenericComponentType());
+ }
+ // WildcardType does not contain a type variable, and we don't consider it resolvable.
+ return false;
+ }
+
+ /**
+ * {@link ParameterizedType} we use for resolving types, so that if you resolve the type
+ * CompletableFuture<T>, we can create resolve the parameter and return
+ * CompletableFuture<Long>.
+ */
+ private static class ParameterizedTypeImpl implements ParameterizedType {
+
+ private final Type[] actualTypeArguments;
+ private final Type rawType;
+ private final Type ownerType;
+
+ public ParameterizedTypeImpl(Type[] actualTypeArguments, Type rawType, Type ownerType) {
+ this.actualTypeArguments = actualTypeArguments;
+ this.rawType = rawType;
+ this.ownerType = ownerType;
+ }
+
+ @Override
+ public Type[] getActualTypeArguments() {
+ return actualTypeArguments;
+ }
+
+ @Override
+ public Type getRawType() {
+ return rawType;
+ }
+
+ @Override
+ public Type getOwnerType() {
+ return ownerType;
+ }
+
+ @Override
+ public String toString() {
+ List<String> actualTypes =
+ Arrays.stream(getActualTypeArguments())
+ .map(Type::getTypeName)
+ .collect(Collectors.toList());
+ return getRawType().getTypeName() + "<" + String.join(", ", actualTypes) + ">";
+ }
+ }
+
+ /**
+ * {@link GenericArrayType} we use for resolving types, so that if you resolve the type
+ * List<T[]>, we can create resolve the parameter and return List<Long[]>.
+ */
+ private static class GenericArrayTypeImpl implements GenericArrayType {
+
+ private final Type genericComponentType;
+
+ public GenericArrayTypeImpl(Type genericComponentType) {
+ this.genericComponentType = genericComponentType;
+ }
+
+ @Override
+ public Type getGenericComponentType() {
+ return genericComponentType;
+ }
+
+ @Override
+ public String getTypeName() {
+ return getGenericComponentType().getTypeName() + "[]";
+ }
+ }
+
// --------------------------------------------------------------------------------------------
private ExtractionUtils() {
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/FunctionMappingExtractor.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/FunctionMappingExtractor.java
index f172069f7ad..26909257645 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/FunctionMappingExtractor.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/FunctionMappingExtractor.java
@@ -29,9 +29,12 @@ import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable;
import java.lang.reflect.Method;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -39,8 +42,10 @@ import java.util.stream.Stream;
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;
+import static org.apache.flink.table.types.extraction.ExtractionUtils.getClassFromType;
import static org.apache.flink.table.types.extraction.ExtractionUtils.isAssignable;
import static org.apache.flink.table.types.extraction.ExtractionUtils.isInvokable;
+import static org.apache.flink.table.types.extraction.ExtractionUtils.resolveVariableWithClassContext;
/**
* Utility for extracting function mappings from signature to result, e.g. from (INT, STRING) to
@@ -143,21 +148,10 @@ final class FunctionMappingExtractor extends BaseMappingExtractor {
boolean allowDataTypeHint) {
return (extractor, method) -> {
if (allowDataTypeHint) {
- final Set<DataTypeHint> dataTypeHints = new HashSet<>();
- dataTypeHints.addAll(collectAnnotationsOfMethod(DataTypeHint.class, method));
- dataTypeHints.addAll(
- collectAnnotationsOfClass(
- DataTypeHint.class, extractor.getFunctionClass()));
- if (dataTypeHints.size() > 1) {
- throw extractionError(
- "More than one data type hint found for output of function. "
- + "Please use a function hint instead.");
+ Optional<FunctionResultTemplate> hints = extractHints(extractor, method);
+ if (hints.isPresent()) {
+ return hints.get();
}
- if (dataTypeHints.size() == 1) {
- return FunctionTemplate.createResultTemplate(
- extractor.typeFactory, dataTypeHints.iterator().next());
- }
- // otherwise continue with regular extraction
}
final DataType dataType =
DataTypeExtractor.extractFromGeneric(
@@ -169,6 +163,53 @@ final class FunctionMappingExtractor extends BaseMappingExtractor {
};
}
+ /**
+ * Extraction that uses a generic type variable of a method parameter for producing a {@link
+ * FunctionResultTemplate}.
+ *
+ * <p>If enabled, a {@link DataTypeHint} from method or class has higher priority.
+ */
+ static ResultExtraction createGenericResultExtractionFromMethod(
+ int paramPos, int genericPos, boolean allowDataTypeHint) {
+ return (extractor, method) -> {
+ if (allowDataTypeHint) {
+ Optional<FunctionResultTemplate> hints = extractHints(extractor, method);
+ if (hints.isPresent()) {
+ return hints.get();
+ }
+ }
+ final DataType dataType =
+ DataTypeExtractor.extractFromGenericMethodParameter(
+ extractor.typeFactory,
+ extractor.getFunctionClass(),
+ method,
+ paramPos,
+ genericPos);
+ return FunctionResultTemplate.of(dataType);
+ };
+ }
+
+ /** Uses hints to extract functional template. */
+ private static Optional<FunctionResultTemplate> extractHints(
+ BaseMappingExtractor extractor, Method method) {
+ final Set<DataTypeHint> dataTypeHints = new HashSet<>();
+ dataTypeHints.addAll(collectAnnotationsOfMethod(DataTypeHint.class, method));
+ dataTypeHints.addAll(
+ collectAnnotationsOfClass(DataTypeHint.class, extractor.getFunctionClass()));
+ if (dataTypeHints.size() > 1) {
+ throw extractionError(
+ "More than one data type hint found for output of function. "
+ + "Please use a function hint instead.");
+ }
+ if (dataTypeHints.size() == 1) {
+ return Optional.ofNullable(
+ FunctionTemplate.createResultTemplate(
+ extractor.typeFactory, dataTypeHints.iterator().next()));
+ }
+ // otherwise continue with regular extraction
+ return Optional.empty();
+ }
+
/** Verification that checks a method by parameters and return type. */
static MethodVerification createParameterAndReturnTypeVerification() {
return (method, signature, result) -> {
@@ -208,6 +249,30 @@ final class FunctionMappingExtractor extends BaseMappingExtractor {
};
}
+ /** Verification that checks a method by parameters including an additional first parameter. */
+ static MethodVerification createGenericParameterWithArgumentAndReturnTypeVerification(
+ Class<?> baseClass, Class<?> argumentClass, int paramPos, int genericPos) {
+ return (method, signature, result) -> {
+ final Class<?>[] parameters =
+ Stream.concat(Stream.of(argumentClass), signature.stream())
+ .toArray(Class<?>[]::new);
+
+ Type genericType = method.getGenericParameterTypes()[paramPos];
+ genericType = resolveVariableWithClassContext(baseClass, genericType);
+ if (!(genericType instanceof ParameterizedType)) {
+ throw extractionError(
+ "The method '%s' needs generic parameters for the %d arg.",
+ method.getName(), paramPos);
+ }
+ Type returnType =
+ ((ParameterizedType) genericType).getActualTypeArguments()[genericPos];
+ Class<?> returnClazz = getClassFromType(returnType);
+ if (!(isInvokable(method, parameters) && isAssignable(result, returnClazz, true))) {
+ throw createMethodNotFoundError(method.getName(), parameters, null);
+ }
+ };
+ }
+
/** Verification that checks a method by parameters. */
static MethodVerification createParameterVerification() {
return (method, signature, result) -> {
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/TypeInferenceExtractor.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/TypeInferenceExtractor.java
index 445e152607d..2ba0cc25b5b 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/TypeInferenceExtractor.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/TypeInferenceExtractor.java
@@ -23,6 +23,7 @@ import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.catalog.DataTypeFactory;
import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.AsyncScalarFunction;
import org.apache.flink.table.functions.AsyncTableFunction;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.TableAggregateFunction;
@@ -49,7 +50,9 @@ import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import static org.apache.flink.table.types.extraction.ExtractionUtils.extractionError;
+import static org.apache.flink.table.types.extraction.FunctionMappingExtractor.createGenericParameterWithArgumentAndReturnTypeVerification;
import static org.apache.flink.table.types.extraction.FunctionMappingExtractor.createGenericResultExtraction;
+import static org.apache.flink.table.types.extraction.FunctionMappingExtractor.createGenericResultExtractionFromMethod;
import static org.apache.flink.table.types.extraction.FunctionMappingExtractor.createParameterAndReturnTypeVerification;
import static org.apache.flink.table.types.extraction.FunctionMappingExtractor.createParameterSignatureExtraction;
import static org.apache.flink.table.types.extraction.FunctionMappingExtractor.createParameterVerification;
@@ -85,6 +88,22 @@ public final class TypeInferenceExtractor {
return extractTypeInference(mappingExtractor);
}
+ /** Extracts a type inference from a {@link AsyncScalarFunction}. */
+ public static TypeInference forAsyncScalarFunction(
+ DataTypeFactory typeFactory, Class<? extends AsyncScalarFunction> function) {
+ final FunctionMappingExtractor mappingExtractor =
+ new FunctionMappingExtractor(
+ typeFactory,
+ function,
+ UserDefinedFunctionHelper.ASYNC_SCALAR_EVAL,
+ createParameterSignatureExtraction(1),
+ null,
+ createGenericResultExtractionFromMethod(0, 0, true),
+ createGenericParameterWithArgumentAndReturnTypeVerification(
+ function, CompletableFuture.class, 0, 0));
+ return extractTypeInference(mappingExtractor);
+ }
+
/** Extracts a type inference from a {@link AggregateFunction}. */
public static TypeInference forAggregateFunction(
DataTypeFactory typeFactory, Class<? extends AggregateFunction<?, ?>> function) {
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/functions/UserDefinedFunctionHelperTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/functions/UserDefinedFunctionHelperTest.java
index 80fb05809a1..50a9c01b574 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/functions/UserDefinedFunctionHelperTest.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/functions/UserDefinedFunctionHelperTest.java
@@ -35,6 +35,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
@@ -133,6 +134,33 @@ class UserDefinedFunctionHelperTest {
"Method 'eval' of function class '"
+ PrivateMethodScalarFunction.class.getName()
+ "' is not public."),
+ TestSpec.forClass(ValidAsyncScalarFunction.class).expectSuccess(),
+ TestSpec.forInstance(new ValidAsyncScalarFunction()).expectSuccess(),
+ TestSpec.forClass(PrivateAsyncScalarFunction.class)
+ .expectErrorMessage(
+ "Function class '"
+ + PrivateAsyncScalarFunction.class.getName()
+ + "' is not public."),
+ TestSpec.forClass(MissingImplementationAsyncScalarFunction.class)
+ .expectErrorMessage(
+ "Function class '"
+ + MissingImplementationAsyncScalarFunction.class.getName()
+ + "' does not implement a method named 'eval'."),
+ TestSpec.forClass(PrivateMethodAsyncScalarFunction.class)
+ .expectErrorMessage(
+ "Method 'eval' of function class '"
+ + PrivateMethodAsyncScalarFunction.class.getName()
+ + "' is not public."),
+ TestSpec.forClass(NonVoidAsyncScalarFunction.class)
+ .expectErrorMessage(
+ "Method 'eval' of function class '"
+ + NonVoidAsyncScalarFunction.class.getName()
+ + "' must be void."),
+ TestSpec.forClass(NoFutureAsyncScalarFunction.class)
+ .expectErrorMessage(
+ "Method 'eval' of function class '"
+ + NoFutureAsyncScalarFunction.class.getName()
+ + "' must have a first argument of type java.util.concurrent.CompletableFuture."),
TestSpec.forInstance(new ValidTableAggregateFunction()).expectSuccess(),
TestSpec.forInstance(new MissingEmitTableAggregateFunction())
.expectErrorMessage(
@@ -267,6 +295,37 @@ class UserDefinedFunctionHelperTest {
}
}
+ /** Valid scalar function. */
+ public static class ValidAsyncScalarFunction extends AsyncScalarFunction {
+ public void eval(CompletableFuture<Integer> future, int i) {}
+ }
+
+ private static class PrivateAsyncScalarFunction extends AsyncScalarFunction {
+ public void eval(CompletableFuture<Integer> future, int i) {}
+ }
+
+ /** No implementation method. */
+ public static class MissingImplementationAsyncScalarFunction extends AsyncScalarFunction {
+ // nothing to do
+ }
+
+ /** Implementation method is private. */
+ public static class PrivateMethodAsyncScalarFunction extends AsyncScalarFunction {
+ private void eval(CompletableFuture<Integer> future, int i) {}
+ }
+
+ /** Implementation method isn't void. */
+ public static class NonVoidAsyncScalarFunction extends AsyncScalarFunction {
+ public String eval(CompletableFuture<Integer> future, int i) {
+ return "";
+ }
+ }
+
+ /** Implementation method isn't void. */
+ public static class NoFutureAsyncScalarFunction extends AsyncScalarFunction {
+ public void eval(int i) {}
+ }
+
/** Valid table aggregate function. */
public static class ValidTableAggregateFunction extends TableAggregateFunction<String, String> {
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/DataTypeExtractorTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/DataTypeExtractorTest.java
index 24d21fd767e..eb3888ea1a3 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/DataTypeExtractorTest.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/DataTypeExtractorTest.java
@@ -59,6 +59,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
import java.util.stream.Stream;
@@ -378,6 +379,12 @@ class DataTypeExtractorTest {
// method with generic return type
TestSpec.forMethodOutput(IntegerVarArg.class).expectDataType(DataTypes.INT()),
+ // method with parameter with a generic itself
+ TestSpec.forGenericMethodParameter(CompletableFutureVarArg.class, 0, 0)
+ .expectDataType(DataTypes.BIGINT()),
+ // method with the future return type as a generic
+ TestSpec.forGenericMethodParameter(CompletableFutureGeneric.class, 0, 0)
+ .expectDataType(DataTypes.BIGINT()),
TestSpec.forType(
"Structured type with invalid constructor",
SimplePojoWithInvalidConstructor.class)
@@ -554,6 +561,20 @@ class DataTypeExtractorTest {
return forMethodParameter(null, clazz, paramPos);
}
+ static TestSpec forGenericMethodParameter(
+ String description, Class<?> clazz, int paramPos, int genericPos) {
+ final Method method = clazz.getMethods()[0];
+ return new TestSpec(
+ description,
+ (lookup) ->
+ DataTypeExtractor.extractFromGenericMethodParameter(
+ lookup, clazz, method, paramPos, genericPos));
+ }
+
+ static TestSpec forGenericMethodParameter(Class<?> clazz, int paramPos, int genericPos) {
+ return forGenericMethodParameter(null, clazz, paramPos, genericPos);
+ }
+
static TestSpec forMethodOutput(String description, Class<?> clazz) {
final Method method = clazz.getMethods()[0];
return new TestSpec(
@@ -967,6 +988,16 @@ class DataTypeExtractorTest {
// nothing to do
}
+ public static class CompletableFutureVarArg extends VarArgMethod<CompletableFuture<Long>> {
+ // nothing to do
+ }
+
+ public static class CompletableFutureBase<T> {
+ public void eval(CompletableFuture<T> i, int arg) {}
+ }
+
+ public static class CompletableFutureGeneric extends CompletableFutureBase<Long> {}
+
// --------------------------------------------------------------------------------------------
private static class RawTypeGeneric {
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/ExtractionUtilsTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/ExtractionUtilsTest.java
new file mode 100644
index 00000000000..80b12fb9a95
--- /dev/null
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/ExtractionUtilsTest.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.types.extraction;
+
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.GenericArrayType;
+import java.lang.reflect.Method;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+
+/** Tests {@link ExtractionUtils}. */
+public class ExtractionUtilsTest {
+
+ @Test
+ void testResolveParameters() {
+ List<Method> methods = ExtractionUtils.collectMethods(LongClass.class, "method");
+ Method method = methods.get(0);
+ Type longType =
+ ExtractionUtils.resolveVariableWithClassContext(
+ LongClass.class, method.getGenericParameterTypes()[0]);
+ Type futureType =
+ ExtractionUtils.resolveVariableWithClassContext(
+ LongClass.class, method.getGenericParameterTypes()[1]);
+ Type listOfFutures =
+ ExtractionUtils.resolveVariableWithClassContext(
+ LongClass.class, method.getGenericParameterTypes()[2]);
+ Type arrayType =
+ ExtractionUtils.resolveVariableWithClassContext(
+ LongClass.class, method.getGenericParameterTypes()[3]);
+ assertThat(longType).isEqualTo(Long.class);
+ assertThat(futureType).isInstanceOf(ParameterizedType.class);
+ assertThat(((ParameterizedType) futureType).getRawType())
+ .isEqualTo(CompletableFuture.class);
+ assertThat(((ParameterizedType) futureType).getActualTypeArguments()[0])
+ .isEqualTo(Long.class);
+ assertThat(listOfFutures).isInstanceOf(ParameterizedType.class);
+ assertThat(((ParameterizedType) listOfFutures).getRawType()).isEqualTo(List.class);
+ assertThat(((ParameterizedType) listOfFutures).getActualTypeArguments()[0])
+ .isInstanceOf(ParameterizedType.class);
+ ParameterizedType innerFuture =
+ ((ParameterizedType)
+ ((ParameterizedType) listOfFutures).getActualTypeArguments()[0]);
+ assertThat(innerFuture.getRawType()).isEqualTo(CompletableFuture.class);
+ assertThat(innerFuture.getActualTypeArguments()[0]).isEqualTo(Long.class);
+ assertThat(arrayType).isInstanceOf(GenericArrayType.class);
+ assertThat(((GenericArrayType) arrayType).getGenericComponentType()).isEqualTo(Long.class);
+ }
+
+ @Test
+ void testResolveParametersDeeper() {
+ List<Method> methods = ExtractionUtils.collectMethods(FutureClass.class, "method");
+ Method method = methods.get(0);
+ Type futureType =
+ ExtractionUtils.resolveVariableWithClassContext(
+ FutureClass.class, method.getGenericParameterTypes()[0]);
+ Type listOfFutures =
+ ExtractionUtils.resolveVariableWithClassContext(
+ FutureClass.class, method.getGenericParameterTypes()[1]);
+ assertThat(futureType).isInstanceOf(ParameterizedType.class);
+ assertThat(((ParameterizedType) futureType).getRawType())
+ .isEqualTo(CompletableFuture.class);
+ assertThat(((ParameterizedType) futureType).getActualTypeArguments()[0])
+ .isEqualTo(Long.class);
+ assertThat(listOfFutures).isInstanceOf(ParameterizedType.class);
+ assertThat(((ParameterizedType) listOfFutures).getRawType()).isEqualTo(List.class);
+ assertThat(((ParameterizedType) listOfFutures).getActualTypeArguments()[0])
+ .isInstanceOf(ParameterizedType.class);
+ ParameterizedType innerFuture =
+ ((ParameterizedType)
+ ((ParameterizedType) listOfFutures).getActualTypeArguments()[0]);
+ assertThat(innerFuture.getRawType()).isEqualTo(CompletableFuture.class);
+ assertThat(innerFuture.getActualTypeArguments()[0]).isEqualTo(Long.class);
+ }
+
+ /** Test function. */
+ public static class ClassBase<T> {
+
+ public void method(
+ T generic,
+ CompletableFuture<T> genericFuture,
+ List<CompletableFuture<T>> listOfGenericFuture,
+ T[] array) {}
+ }
+
+ /** Test function. */
+ public static class LongClass extends ClassBase<Long> {}
+
+ /** Test function. */
+ public static class ClassBase2<T> {
+
+ public void method(T generic, List<T> list) {}
+ }
+
+ /** Test function. */
+ public static class FutureClass extends ClassBase2<CompletableFuture<Long>> {}
+}
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java
index 13fb65f0b2e..97e08fc2a2a 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java
@@ -27,6 +27,7 @@ import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.AsyncScalarFunction;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.table.functions.TableFunction;
@@ -51,6 +52,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
import java.util.stream.Stream;
@@ -165,6 +167,15 @@ class TypeInferenceExtractorTest {
new String[0], new ArgumentTypeStrategy[0]),
TypeStrategies.explicit(DataTypes.INT())),
+ // no arguments async
+ TestSpec.forAsyncScalarFunction(ZeroArgFunctionAsync.class)
+ .expectNamedArguments()
+ .expectTypedArguments()
+ .expectOutputMapping(
+ InputTypeStrategies.sequence(
+ new String[0], new ArgumentTypeStrategy[0]),
+ TypeStrategies.explicit(DataTypes.INT())),
+
// test primitive arguments extraction
TestSpec.forScalarFunction(MixedArgFunction.class)
.expectNamedArguments("i", "d")
@@ -180,6 +191,21 @@ class TypeInferenceExtractorTest {
}),
TypeStrategies.explicit(DataTypes.INT())),
+ // test primitive arguments extraction async
+ TestSpec.forAsyncScalarFunction(MixedArgFunctionAsync.class)
+ .expectNamedArguments("i", "d")
+ .expectTypedArguments(
+ DataTypes.INT().notNull().bridgedTo(int.class), DataTypes.DOUBLE())
+ .expectOutputMapping(
+ InputTypeStrategies.sequence(
+ new String[] {"i", "d"},
+ new ArgumentTypeStrategy[] {
+ InputTypeStrategies.explicit(
+ DataTypes.INT().notNull().bridgedTo(int.class)),
+ InputTypeStrategies.explicit(DataTypes.DOUBLE())
+ }),
+ TypeStrategies.explicit(DataTypes.INT())),
+
// test overloaded arguments extraction
TestSpec.forScalarFunction(OverloadedFunction.class)
.expectOutputMapping(
@@ -200,6 +226,25 @@ class TypeInferenceExtractorTest {
TypeStrategies.explicit(
DataTypes.BIGINT().notNull().bridgedTo(long.class))),
+ // test overloaded arguments extraction async
+ TestSpec.forAsyncScalarFunction(OverloadedFunctionAsync.class)
+ .expectOutputMapping(
+ InputTypeStrategies.sequence(
+ new String[] {"i", "d"},
+ new ArgumentTypeStrategy[] {
+ InputTypeStrategies.explicit(
+ DataTypes.INT().notNull().bridgedTo(int.class)),
+ InputTypeStrategies.explicit(DataTypes.DOUBLE())
+ }),
+ TypeStrategies.explicit(DataTypes.INT()))
+ .expectOutputMapping(
+ InputTypeStrategies.sequence(
+ new String[] {"s"},
+ new ArgumentTypeStrategy[] {
+ InputTypeStrategies.explicit(DataTypes.STRING())
+ }),
+ TypeStrategies.explicit(DataTypes.BIGINT())),
+
// test varying arguments extraction
TestSpec.forScalarFunction(VarArgFunction.class)
.expectOutputMapping(
@@ -213,6 +258,19 @@ class TypeInferenceExtractorTest {
}),
TypeStrategies.explicit(DataTypes.STRING())),
+ // test varying arguments extraction async
+ TestSpec.forAsyncScalarFunction(VarArgFunctionAsync.class)
+ .expectOutputMapping(
+ InputTypeStrategies.varyingSequence(
+ new String[] {"i", "more"},
+ new ArgumentTypeStrategy[] {
+ InputTypeStrategies.explicit(
+ DataTypes.INT().notNull().bridgedTo(int.class)),
+ InputTypeStrategies.explicit(
+ DataTypes.INT().notNull().bridgedTo(int.class))
+ }),
+ TypeStrategies.explicit(DataTypes.STRING())),
+
// test varying arguments extraction with byte
TestSpec.forScalarFunction(VarArgWithByteFunction.class)
.expectOutputMapping(
@@ -226,6 +284,19 @@ class TypeInferenceExtractorTest {
}),
TypeStrategies.explicit(DataTypes.STRING())),
+ // test varying arguments extraction with byte async
+ TestSpec.forAsyncScalarFunction(VarArgWithByteFunctionAsync.class)
+ .expectOutputMapping(
+ InputTypeStrategies.varyingSequence(
+ new String[] {"bytes"},
+ new ArgumentTypeStrategy[] {
+ InputTypeStrategies.explicit(
+ DataTypes.TINYINT()
+ .notNull()
+ .bridgedTo(byte.class))
+ }),
+ TypeStrategies.explicit(DataTypes.STRING())),
+
// output hint with input extraction
TestSpec.forScalarFunction(ExtractWithOutputHintFunction.class)
.expectNamedArguments("i")
@@ -238,6 +309,18 @@ class TypeInferenceExtractorTest {
}),
TypeStrategies.explicit(DataTypes.INT())),
+ // output hint with input extraction
+ TestSpec.forAsyncScalarFunction(ExtractWithOutputHintFunctionAsync.class)
+ .expectNamedArguments("i")
+ .expectTypedArguments(DataTypes.INT())
+ .expectOutputMapping(
+ InputTypeStrategies.sequence(
+ new String[] {"i"},
+ new ArgumentTypeStrategy[] {
+ InputTypeStrategies.explicit(DataTypes.INT())
+ }),
+ TypeStrategies.explicit(DataTypes.INT())),
+
// output extraction with input hints
TestSpec.forScalarFunction(ExtractWithInputHintFunction.class)
.expectNamedArguments("i", "b")
@@ -326,6 +409,12 @@ class TypeInferenceExtractorTest {
"Considering all hints, the method should comply with the signature:\n"
+ "java.lang.String eval(int[])"),
+ // mismatch between hints and implementation regarding return type
+ TestSpec.forAsyncScalarFunction(InvalidMethodScalarFunctionAsync.class)
+ .expectErrorMessage(
+ "Considering all hints, the method should comply with the signature:\n"
+ + "eval(java.util.concurrent.CompletableFuture, int[])"),
+
// mismatch between hints and implementation regarding accumulator
TestSpec.forAggregateFunction(InvalidMethodAggregateFunction.class)
.expectErrorMessage(
@@ -430,6 +519,17 @@ class TypeInferenceExtractorTest {
DataTypeHintOnScalarFunction.class)
.expectNamedArguments()
.expectTypedArguments()
+ .expectOutputMapping(
+ InputTypeStrategies.sequence(
+ new String[] {}, new ArgumentTypeStrategy[] {}),
+ TypeStrategies.explicit(
+ DataTypes.ROW(DataTypes.FIELD("i", DataTypes.INT()))
+ .bridgedTo(RowData.class))),
+ TestSpec.forAsyncScalarFunction(
+ "A data type hint on the method is used for enriching (not a function output hint)",
+ DataTypeHintOnScalarFunctionAsync.class)
+ .expectNamedArguments()
+ .expectTypedArguments()
.expectOutputMapping(
InputTypeStrategies.sequence(
new String[] {}, new ArgumentTypeStrategy[] {}),
@@ -809,6 +909,19 @@ class TypeInferenceExtractorTest {
new DataTypeFactoryMock(), function));
}
+ static TestSpec forAsyncScalarFunction(Class<? extends AsyncScalarFunction> function) {
+ return forAsyncScalarFunction(null, function);
+ }
+
+ static TestSpec forAsyncScalarFunction(
+ String description, Class<? extends AsyncScalarFunction> function) {
+ return new TestSpec(
+ description == null ? function.getSimpleName() : description,
+ () ->
+ TypeInferenceExtractor.forAsyncScalarFunction(
+ new DataTypeFactoryMock(), function));
+ }
+
static TestSpec forAggregateFunction(Class<? extends AggregateFunction<?, ?>> function) {
return new TestSpec(
function.getSimpleName(),
@@ -1436,4 +1549,40 @@ class TypeInferenceExtractorTest {
return null;
}
}
+
+ private static class ZeroArgFunctionAsync extends AsyncScalarFunction {
+ public void eval(CompletableFuture<Integer> f) {}
+ }
+
+ private static class MixedArgFunctionAsync extends AsyncScalarFunction {
+ public void eval(CompletableFuture<Integer> f, int i, Double d) {}
+ }
+
+ private static class OverloadedFunctionAsync extends AsyncScalarFunction {
+ public void eval(CompletableFuture<Integer> f, int i, Double d) {}
+
+ public void eval(CompletableFuture<Long> f, String s) {}
+ }
+
+ private static class VarArgFunctionAsync extends AsyncScalarFunction {
+ public void eval(CompletableFuture<String> f, int i, int... more) {}
+ }
+
+ private static class VarArgWithByteFunctionAsync extends AsyncScalarFunction {
+ public void eval(CompletableFuture<String> f, byte... bytes) {}
+ }
+
+ @FunctionHint(output = @DataTypeHint("INT"))
+ private static class ExtractWithOutputHintFunctionAsync extends AsyncScalarFunction {
+ public void eval(CompletableFuture<Object> f, Integer i) {}
+ }
+
+ @FunctionHint(output = @DataTypeHint("STRING"))
+ private static class InvalidMethodScalarFunctionAsync extends AsyncScalarFunction {
+ public void eval(CompletableFuture<Long> f, int[] i) {}
+ }
+
+ private static class DataTypeHintOnScalarFunctionAsync extends AsyncScalarFunction {
+ public void eval(@DataTypeHint("ROW<i INT>") CompletableFuture<RowData> f) {}
+ }
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/catalog/FunctionCatalogOperatorTable.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/catalog/FunctionCatalogOperatorTable.java
index 4e5bb766cbb..7b148fdde38 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/catalog/FunctionCatalogOperatorTable.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/catalog/FunctionCatalogOperatorTable.java
@@ -193,6 +193,7 @@ public class FunctionCatalogOperatorTable implements SqlOperatorTable {
if (kind == FunctionKind.TABLE) {
return true;
} else if (kind == FunctionKind.SCALAR
+ || kind == FunctionKind.ASYNC_SCALAR
|| kind == FunctionKind.AGGREGATE
|| kind == FunctionKind.TABLE_AGGREGATE) {
if (category != null && category.isTableFunction()) {
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/AsyncCodeGenerator.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/AsyncCodeGenerator.java
new file mode 100644
index 00000000000..fad669b7e55
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/AsyncCodeGenerator.java
@@ -0,0 +1,225 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.codegen;
+
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.functions.async.AsyncFunction;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.functions.FunctionKind;
+import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction;
+import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
+import org.apache.flink.table.runtime.generated.GeneratedFunction;
+import org.apache.flink.table.runtime.operators.calc.async.DelegatingAsyncResultFuture;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexVisitorImpl;
+import org.apache.commons.text.StringSubstitutor;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * Generates an {@link AsyncFunction} which can be used to evaluate calc projections from an async
+ * scalar function.
+ */
+public class AsyncCodeGenerator {
+
+ public static final String DEFAULT_EXCEPTION_TERM = "e";
+ public static final String DEFAULT_DELEGATING_FUTURE_TERM = "f";
+
+ /**
+ * Creates a generated function which produces an {@link AsyncFunction} which executes the calc
+ * projections.
+ *
+ * @param name The name used to generate the underlying function name
+ * @param inputType The RowType of the input RowData
+ * @param returnType The RowType of the resulting RowData
+ * @param retainHeader If the header of the row should be retained
+ * @param calcProjection The list of projections to be executed by this function
+ * @param tableConfig The table configuration
+ * @param classLoader The classloader to use while resolving classes
+ * @return A {@link GeneratedFunction} returning an {@link AsyncFunction} executing the given
+ * list of projections
+ */
+ public static GeneratedFunction<AsyncFunction<RowData, RowData>> generateFunction(
+ String name,
+ RowType inputType,
+ RowType returnType,
+ List<RexNode> calcProjection,
+ boolean retainHeader,
+ ReadableConfig tableConfig,
+ ClassLoader classLoader) {
+ CodeGeneratorContext ctx = new CodeGeneratorContext(tableConfig, classLoader);
+ String processCode =
+ generateProcessCode(
+ ctx,
+ inputType,
+ returnType,
+ calcProjection,
+ retainHeader,
+ CodeGenUtils.DEFAULT_INPUT1_TERM(),
+ CodeGenUtils.DEFAULT_COLLECTOR_TERM(),
+ DEFAULT_EXCEPTION_TERM,
+ CodeGenUtils.DEFAULT_OUT_RECORD_TERM(),
+ DEFAULT_DELEGATING_FUTURE_TERM);
+ return FunctionCodeGenerator.generateFunction(
+ ctx,
+ name,
+ getFunctionClass(),
+ processCode,
+ returnType,
+ inputType,
+ CodeGenUtils.DEFAULT_INPUT1_TERM(),
+ JavaScalaConversionUtil.toScala(Optional.empty()),
+ JavaScalaConversionUtil.toScala(Optional.empty()),
+ CodeGenUtils.DEFAULT_COLLECTOR_TERM(),
+ CodeGenUtils.DEFAULT_CONTEXT_TERM());
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Class<AsyncFunction<RowData, RowData>> getFunctionClass() {
+ return (Class<AsyncFunction<RowData, RowData>>) (Object) AsyncFunction.class;
+ }
+
+ private static String generateProcessCode(
+ CodeGeneratorContext ctx,
+ RowType inputType,
+ RowType outRowType,
+ List<RexNode> projection,
+ boolean retainHeader,
+ String inputTerm,
+ String collectorTerm,
+ String errorTerm,
+ String recordTerm,
+ String delegatingFutureTerm) {
+
+ projection.forEach(n -> n.accept(new AsyncScalarFunctionsValidator()));
+
+ ExprCodeGenerator exprGenerator =
+ new ExprCodeGenerator(ctx, false)
+ .bindInput(
+ inputType,
+ inputTerm,
+ JavaScalaConversionUtil.toScala(Optional.empty()));
+
+ List<GeneratedExpression> projectionExprs =
+ projection.stream()
+ .map(exprGenerator::generateExpression)
+ .collect(Collectors.toList());
+ int syncIndex = 0;
+ int index = 0;
+ StringBuilder outputs = new StringBuilder();
+ StringBuilder syncInvocations = new StringBuilder();
+ StringBuilder asyncInvocation = new StringBuilder();
+ if (retainHeader) {
+ outputs.append(String.format("%s.setRowKind(rowKind);\n", recordTerm, inputTerm));
+ }
+ for (GeneratedExpression fieldExpr : projectionExprs) {
+ if (fieldExpr.resultTerm().isEmpty()) {
+ outputs.append(
+ String.format("%s.setField(%d, resultObject);\n", recordTerm, index));
+ asyncInvocation.append(fieldExpr.code());
+ } else {
+ outputs.append(
+ String.format(
+ "%s.setField(%d, %s.getSynchronousResult(%d));\n",
+ recordTerm, index, delegatingFutureTerm, syncIndex));
+ syncInvocations.append(fieldExpr.code());
+ syncInvocations.append(
+ String.format(
+ "%s.addSynchronousResult(%s);\n",
+ delegatingFutureTerm, fieldExpr.resultTerm()));
+ syncIndex++;
+ }
+ index++;
+ }
+
+ Map<String, String> values = new HashMap<>();
+ values.put("delegatingFutureTerm", delegatingFutureTerm);
+ values.put("delegatingFutureType", DelegatingAsyncResultFuture.class.getCanonicalName());
+ values.put("collectorTerm", collectorTerm);
+ values.put("typeTerm", GenericRowData.class.getCanonicalName());
+ values.put("recordTerm", recordTerm);
+ values.put("inputTerm", inputTerm);
+ values.put("fieldCount", Integer.toString(LogicalTypeChecks.getFieldCount(outRowType)));
+ values.put("outputs", outputs.toString());
+ values.put("syncInvocations", syncInvocations.toString());
+ values.put("asyncInvocation", asyncInvocation.toString());
+ values.put("errorTerm", errorTerm);
+
+ return StringSubstitutor.replace(
+ String.join(
+ "\n",
+ new String[] {
+ "final ${delegatingFutureType} ${delegatingFutureTerm} ",
+ " = new ${delegatingFutureType}(${collectorTerm});",
+ "final org.apache.flink.types.RowKind rowKind = ${inputTerm}.getRowKind();\n",
+ "try {",
+ " java.util.function.Function<Object, ${typeTerm}> outputFactory = ",
+ " new java.util.function.Function<Object, ${typeTerm}>() {",
+ " @Override",
+ " public ${typeTerm} apply(Object resultObject) {",
+ " final ${typeTerm} ${recordTerm} = new ${typeTerm}(${fieldCount});",
+ " ${outputs}",
+ " return ${recordTerm};",
+ " }",
+ " };",
+ "",
+ " ${delegatingFutureTerm}.setOutputFactory(outputFactory);",
+ // Ensure that sync invocations come first so that we know that they're
+ // available when the async callback occurs.
+ " ${syncInvocations}",
+ " ${asyncInvocation}",
+ "",
+ "} catch (Throwable ${errorTerm}) {",
+ " ${collectorTerm}.completeExceptionally(${errorTerm});",
+ "}"
+ }),
+ values);
+ }
+
+ private static class AsyncScalarFunctionsValidator extends RexVisitorImpl<Void> {
+ public AsyncScalarFunctionsValidator() {
+ super(true);
+ }
+
+ @Override
+ public Void visitCall(RexCall call) {
+ super.visitCall(call);
+
+ if (call.getOperator() instanceof BridgingSqlFunction
+ && ((BridgingSqlFunction) call.getOperator()).getDefinition().getKind()
+ != FunctionKind.ASYNC_SCALAR) {
+ throw new CodeGenException(
+ "Invalid use of function "
+ + call.getOperator()
+ + "."
+ + "Code generation should only be done with async calls");
+ }
+ return null;
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlFunction.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlFunction.java
index 5521917f13b..7c12066b711 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlFunction.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlFunction.java
@@ -116,7 +116,9 @@ public class BridgingSqlFunction extends SqlFunction {
TypeInference typeInference) {
final FunctionKind functionKind = resolvedFunction.getDefinition().getKind();
checkState(
- functionKind == FunctionKind.SCALAR || functionKind == FunctionKind.TABLE,
+ functionKind == FunctionKind.SCALAR
+ || functionKind == FunctionKind.ASYNC_SCALAR
+ || functionKind == FunctionKind.TABLE,
"Scalar or table function kind expected.");
if (functionKind == FunctionKind.TABLE) {
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecAsyncCalc.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecAsyncCalc.java
new file mode 100644
index 00000000000..1851e2ab786
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecAsyncCalc.java
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.nodes.exec.common;
+
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.functions.async.AsyncFunction;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.async.AsyncWaitOperatorFactory;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.codegen.AsyncCodeGenerator;
+import org.apache.flink.table.planner.delegation.PlannerBase;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeConfig;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeContext;
+import org.apache.flink.table.planner.plan.nodes.exec.InputProperty;
+import org.apache.flink.table.planner.plan.nodes.exec.SingleTransformationTranslator;
+import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil;
+import org.apache.flink.table.planner.plan.utils.AsyncUtil;
+import org.apache.flink.table.runtime.generated.GeneratedFunction;
+import org.apache.flink.table.runtime.operators.calc.async.AsyncFunctionRunner;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/** Base class for exec Async Calc. */
+public abstract class CommonExecAsyncCalc extends ExecNodeBase<RowData>
+ implements SingleTransformationTranslator<RowData> {
+
+ public static final String ASYNC_CALC_TRANSFORMATION = "async-calc";
+
+ public static final String FIELD_NAME_PROJECTION = "projection";
+
+ @JsonProperty(FIELD_NAME_PROJECTION)
+ private final List<RexNode> projection;
+
+ public CommonExecAsyncCalc(
+ int id,
+ ExecNodeContext context,
+ ReadableConfig persistedConfig,
+ List<RexNode> projection,
+ List<InputProperty> inputProperties,
+ RowType outputType,
+ String description) {
+ super(id, context, persistedConfig, inputProperties, outputType, description);
+ checkArgument(inputProperties.size() == 1);
+ this.projection = checkNotNull(projection);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ protected Transformation<RowData> translateToPlanInternal(
+ PlannerBase planner, ExecNodeConfig config) {
+ final ExecEdge inputEdge = getInputEdges().get(0);
+ final Transformation<RowData> inputTransform =
+ (Transformation<RowData>) inputEdge.translateToPlan(planner);
+ return createAsyncOneInputTransformation(
+ inputTransform, config, planner.getFlinkContext().getClassLoader());
+ }
+
+ private OneInputTransformation<RowData, RowData> createAsyncOneInputTransformation(
+ Transformation<RowData> inputTransform,
+ ExecNodeConfig config,
+ ClassLoader classLoader) {
+ List<RexCall> asyncRexCalls =
+ projection.stream()
+ .filter(x -> x instanceof RexCall)
+ .map(x -> (RexCall) x)
+ .collect(Collectors.toList());
+
+ List<Integer> forwardedFields =
+ projection.stream()
+ .filter(x -> x instanceof RexInputRef)
+ .map(x -> ((RexInputRef) x).getIndex())
+ .collect(Collectors.toList());
+ final ExecEdge inputEdge = getInputEdges().get(0);
+ RowType inputRowType =
+ RowType.of(inputEdge.getOutputType().getChildren().toArray(new LogicalType[0]));
+
+ List<LogicalType> forwardedFieldsLogicalTypes =
+ forwardedFields.stream()
+ .map(inputRowType.getChildren()::get)
+ .collect(Collectors.toList());
+ List<LogicalType> asyncCallLogicalTypes =
+ asyncRexCalls.stream()
+ .map(node -> FlinkTypeFactory.toLogicalType(node.getType()))
+ .collect(Collectors.toList());
+ List<LogicalType> fieldsLogicalTypes = new ArrayList<>();
+ fieldsLogicalTypes.addAll(forwardedFieldsLogicalTypes);
+ fieldsLogicalTypes.addAll(asyncCallLogicalTypes);
+ InternalTypeInfo<RowData> asyncOperatorResultTypeInfo =
+ InternalTypeInfo.ofFields(fieldsLogicalTypes.toArray(new LogicalType[0]));
+ OneInputStreamOperatorFactory<RowData, RowData> factory =
+ getAsyncFunctionOperator(config, classLoader, inputRowType);
+ return ExecNodeUtil.createOneInputTransformation(
+ inputTransform,
+ createTransformationMeta(ASYNC_CALC_TRANSFORMATION, config),
+ factory,
+ asyncOperatorResultTypeInfo,
+ inputTransform.getParallelism(),
+ false);
+ }
+
+ private OneInputStreamOperatorFactory<RowData, RowData> getAsyncFunctionOperator(
+ ExecNodeConfig config, ClassLoader classLoader, RowType inputRowType) {
+
+ RowType resultTypeInfo =
+ RowType.of(
+ projection.stream()
+ .map(node -> FlinkTypeFactory.toLogicalType(node.getType()))
+ .toArray(LogicalType[]::new));
+
+ GeneratedFunction<AsyncFunction<RowData, RowData>> generatedFunction =
+ AsyncCodeGenerator.generateFunction(
+ "AsyncScalarFunction",
+ inputRowType,
+ resultTypeInfo,
+ projection,
+ true,
+ config,
+ classLoader);
+ AsyncFunctionRunner func = new AsyncFunctionRunner(generatedFunction);
+ AsyncUtil.Options options = AsyncUtil.getAsyncOptions(config);
+ return new AsyncWaitOperatorFactory<>(
+ func,
+ options.asyncTimeout,
+ options.asyncBufferCapacity,
+ options.asyncOutputMode,
+ options.asyncRetryStrategy);
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecAsyncCalc.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecAsyncCalc.java
new file mode 100644
index 00000000000..997d7cdb492
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecAsyncCalc.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.nodes.exec.stream;
+
+import org.apache.flink.FlinkVersion;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeContext;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeMetadata;
+import org.apache.flink.table.planner.plan.nodes.exec.InputProperty;
+import org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecAsyncCalc;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import org.apache.calcite.rex.RexNode;
+
+import java.util.Collections;
+import java.util.List;
+
+/** Stream {@link ExecNode} for {@link org.apache.flink.table.functions.AsyncScalarFunction}. */
+@ExecNodeMetadata(
+ name = "stream-exec-async-calc",
+ version = 1,
+ producedTransformations = CommonExecAsyncCalc.ASYNC_CALC_TRANSFORMATION,
+ minPlanVersion = FlinkVersion.v1_19,
+ minStateVersion = FlinkVersion.v1_19,
+ consumedOptions = {
+ "table.exec.async-scalar.buffer-capacity",
+ "table.exec.async-scalar.timeout",
+ "table.exec.async-scalar.retry-strategy",
+ "table.exec.async-scalar.retry-delay",
+ "table.exec.async-scalar.max-attempts",
+ })
+public class StreamExecAsyncCalc extends CommonExecAsyncCalc implements StreamExecNode<RowData> {
+
+ public StreamExecAsyncCalc(
+ ReadableConfig tableConfig,
+ List<RexNode> projection,
+ InputProperty inputProperty,
+ RowType outputType,
+ String description) {
+ this(
+ ExecNodeContext.newNodeId(),
+ ExecNodeContext.newContext(StreamExecAsyncCalc.class),
+ ExecNodeContext.newPersistedConfig(StreamExecAsyncCalc.class, tableConfig),
+ projection,
+ Collections.singletonList(inputProperty),
+ outputType,
+ description);
+ }
+
+ @JsonCreator
+ public StreamExecAsyncCalc(
+ @JsonProperty(FIELD_NAME_ID) int id,
+ @JsonProperty(FIELD_NAME_TYPE) ExecNodeContext context,
+ @JsonProperty(FIELD_NAME_CONFIGURATION) ReadableConfig persistedConfig,
+ @JsonProperty(FIELD_NAME_PROJECTION) List<RexNode> projection,
+ @JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty> inputProperties,
+ @JsonProperty(FIELD_NAME_OUTPUT_TYPE) RowType outputType,
+ @JsonProperty(FIELD_NAME_DESCRIPTION) String description) {
+ super(id, context, persistedConfig, projection, inputProperties, outputType, description);
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalAsyncCalc.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalAsyncCalc.java
new file mode 100644
index 00000000000..fcb1733a016
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalAsyncCalc.java
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.nodes.physical.stream;
+
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import org.apache.flink.table.planner.plan.nodes.exec.InputProperty;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecAsyncCalc;
+import org.apache.flink.table.planner.plan.utils.AsyncUtil;
+
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig;
+
+/** Stream physical RelNode for {@link org.apache.flink.table.functions.AsyncScalarFunction}. */
+public class StreamPhysicalAsyncCalc extends StreamPhysicalCalcBase {
+ private final RelOptCluster cluster;
+ private final RexProgram calcProgram;
+ private final RelDataType outputRowType;
+
+ public StreamPhysicalAsyncCalc(
+ RelOptCluster cluster,
+ RelTraitSet traitSet,
+ RelNode inputRel,
+ RexProgram calcProgram,
+ RelDataType outputRowType) {
+ super(cluster, traitSet, inputRel, calcProgram, outputRowType);
+ this.cluster = cluster;
+ this.calcProgram = calcProgram;
+ this.outputRowType = outputRowType;
+ }
+
+ @Override
+ public Calc copy(RelTraitSet traitSet, RelNode child, RexProgram program) {
+ return new StreamPhysicalAsyncCalc(cluster, traitSet, child, program, outputRowType);
+ }
+
+ @Override
+ public ExecNode<?> translateToExecNode() {
+ List<RexNode> projection =
+ calcProgram.getProjectList().stream()
+ .map(calcProgram::expandLocalRef)
+ .collect(Collectors.toList());
+ if (calcProgram.getCondition() != null) {
+ throw new IllegalStateException(
+ "The condition of StreamPhysicalAsyncCalc should be null.");
+ }
+ if (projection.stream().filter(AsyncUtil::containsAsyncCall).count() > 1) {
+ throw new IllegalStateException(
+ "Only a single async projection is allowed in StreamPhysicalAsyncCalc.");
+ }
+
+ return new StreamExecAsyncCalc(
+ unwrapTableConfig(this),
+ projection,
+ InputProperty.DEFAULT,
+ FlinkTypeFactory.toLogicalRowType(getRowType()),
+ getRelDetailedDescription());
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/AsyncCalcSplitRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/AsyncCalcSplitRule.java
new file mode 100644
index 00000000000..308447c610c
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/AsyncCalcSplitRule.java
@@ -0,0 +1,203 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.logical;
+
+import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalCalc;
+import org.apache.flink.table.planner.plan.utils.AsyncUtil;
+import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexProgram;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import scala.Option;
+
+/**
+ * Defines split rules for async calc nodes. These largely exist to isolate and simplify the calls
+ * to the async function from other calc operations, so that the operator can handle just that
+ * functionality.
+ */
+public class AsyncCalcSplitRule {
+
+ private static final RemoteCalcCallFinder ASYNC_CALL_FINDER = new AsyncRemoteCalcCallFinder();
+ public static final RelOptRule SPLIT_CONDITION =
+ new RemoteCalcSplitConditionRule(ASYNC_CALL_FINDER);
+ public static final RelOptRule SPLIT_PROJECT =
+ new RemoteCalcSplitProjectionRule(ASYNC_CALL_FINDER);
+ public static final RelOptRule SPLIT_PROJECTION_REX_FIELD =
+ new RemoteCalcSplitProjectionRexFieldRule(ASYNC_CALL_FINDER);
+ public static final RelOptRule SPLIT_CONDITION_REX_FIELD =
+ new RemoteCalcSplitConditionRexFieldRule(ASYNC_CALL_FINDER);
+ public static final RelOptRule EXPAND_PROJECT =
+ new RemoteCalcExpandProjectRule(ASYNC_CALL_FINDER);
+ public static final RelOptRule PUSH_CONDITION =
+ new RemoteCalcPushConditionRule(ASYNC_CALL_FINDER);
+ public static final RelOptRule REWRITE_PROJECT =
+ new RemoteCalcRewriteProjectionRule(ASYNC_CALL_FINDER);
+ public static final RelOptRule NESTED_SPLIT = new AsyncCalcSplitNestedRule(ASYNC_CALL_FINDER);
+ public static final RelOptRule ONE_PER_CALC_SPLIT =
+ new AsyncCalcSplitOnePerCalcRule(ASYNC_CALL_FINDER);
+
+ /**
+ * An Async implementation of {@link RemoteCalcCallFinder} which finds uses of {@link
+ * org.apache.flink.table.functions.AsyncScalarFunction}.
+ */
+ public static class AsyncRemoteCalcCallFinder implements RemoteCalcCallFinder {
+ @Override
+ public boolean containsRemoteCall(RexNode node) {
+ return AsyncUtil.containsAsyncCall(node);
+ }
+
+ @Override
+ public boolean containsNonRemoteCall(RexNode node) {
+ return AsyncUtil.containsNonAsyncCall(node);
+ }
+
+ @Override
+ public boolean isRemoteCall(RexNode node) {
+ return AsyncUtil.isAsyncCall(node);
+ }
+
+ @Override
+ public boolean isNonRemoteCall(RexNode node) {
+ return AsyncUtil.isNonAsyncCall(node);
+ }
+ }
+
+ private static boolean hasNestedCalls(List<RexNode> projects) {
+ return projects.stream()
+ .filter(AsyncUtil::containsAsyncCall)
+ .filter(expr -> expr instanceof RexCall)
+ .map(expr -> (RexCall) expr)
+ .anyMatch(
+ rexCall ->
+ rexCall.getOperands().stream()
+ .anyMatch(AsyncUtil::containsAsyncCall));
+ }
+
+ /**
+ * Splits nested call <- asyncCall chains so that nothing is immediately waiting on an async
+ * call in a single calc.
+ *
+ * <p>For Example: Calc(select=[syncCall(asyncCall()]) -> Source
+ *
+ * <p>becomes
+ *
+ * <p>Calc(select=[syncCall(f0)]) -> AsyncCalc(select=[asyncCall() as f0]) -> Source
+ */
+ public static class AsyncCalcSplitNestedRule extends RemoteCalcSplitRuleBase<Void> {
+
+ public AsyncCalcSplitNestedRule(RemoteCalcCallFinder callFinder) {
+ super("AsyncCalcSplitNestedRule", callFinder);
+ }
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ FlinkLogicalCalc calc = call.rel(0);
+
+ // Matches if we have nested remote calls
+ List<RexNode> projects =
+ calc.getProgram().getProjectList().stream()
+ .map(calc.getProgram()::expandLocalRef)
+ .collect(Collectors.toList());
+ return hasNestedCalls(projects);
+ }
+
+ // We convert not on the outermost call, but anything within it
+ @Override
+ public boolean needConvert(RexProgram program, RexNode node, Option<Void> matchState) {
+ return node instanceof RexCall
+ && !((RexCall) node)
+ .getOperands().stream().anyMatch(callFinder()::containsRemoteCall);
+ }
+
+ @Override
+ public SplitComponents split(RexProgram program, ScalarFunctionSplitter splitter) {
+ return new SplitComponents(
+ JavaScalaConversionUtil.toScala(Optional.<RexNode>empty()),
+ JavaScalaConversionUtil.toScala(
+ Optional.ofNullable(program.getCondition())
+ .map(program::expandLocalRef)),
+ JavaScalaConversionUtil.toScala(
+ program.getProjectList().stream()
+ .map(program::expandLocalRef)
+ .map(n -> n.accept(splitter))
+ .collect(Collectors.toList())));
+ }
+ }
+
+ /**
+ * Splits async calls if there are multiple across projections, so that there's one per calc.
+ * This assumes that the nested rule has been run first, so there is just one per projection.
+ *
+ * <p>For Example: Calc(select=[asyncCall(), asyncCall()]) -> Source
+ *
+ * <p>becomes
+ *
+ * <p>AsyncCalc(select=[asyncCall(), f0]) -> AsyncCalc(select=[asyncCall() as f0]) -> Source
+ */
+ public static class AsyncCalcSplitOnePerCalcRule
+ extends RemoteCalcSplitProjectionRuleBase<AsyncCalcSplitOnePerCalcRule.State> {
+
+ public AsyncCalcSplitOnePerCalcRule(RemoteCalcCallFinder callFinder) {
+ super("AsyncCalcSplitOnePerCalcRule", callFinder);
+ }
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ FlinkLogicalCalc calc = call.rel(0);
+ List<RexNode> projects =
+ calc.getProgram().getProjectList().stream()
+ .map(calc.getProgram()::expandLocalRef)
+ .collect(Collectors.toList());
+
+ // If this has no nested calls, then this can be called to split up separate projections
+ // into two different calcs. We don't want the splitter to be called to with nested
+ // calls since it won't behave correctly, so this must be used in conjunction with the
+ // nested rule.
+ return !hasNestedCalls(projects)
+ && projects.stream().filter(callFinder()::containsRemoteCall).count() >= 2;
+ }
+
+ @Override
+ public boolean needConvert(RexProgram program, RexNode node, Option<State> matchState) {
+ if (AsyncUtil.containsAsyncCall(node) && !matchState.get().foundMatch) {
+ matchState.get().foundMatch = true;
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public Option<State> getMatchState() {
+ return Option.apply(new State());
+ }
+
+ /** State object used to keep track of whether a match has been found yet. */
+ public static class State {
+ boolean foundMatch = false;
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PythonCorrelateSplitRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PythonCorrelateSplitRule.java
index 1eef306c648..986f0fc538c 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PythonCorrelateSplitRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PythonCorrelateSplitRule.java
@@ -245,7 +245,8 @@ public class PythonCorrelateSplitRule extends RelOptRule {
// separate node
return node instanceof RexFieldAccess;
}
- });
+ },
+ new PythonRemoteCalcCallFinder());
}
@Override
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalAsyncCalcRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalAsyncCalcRule.java
new file mode 100644
index 00000000000..d12ace81bc4
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalAsyncCalcRule.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.physical.stream;
+
+import org.apache.flink.table.planner.plan.nodes.FlinkConventions;
+import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalAsyncCalc;
+import org.apache.flink.table.planner.plan.utils.AsyncUtil;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.convert.ConverterRule;
+import org.apache.calcite.rex.RexProgram;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/** A physical rule for identifying logical async Calcs and converting it to a physical RelNode. */
+public class StreamPhysicalAsyncCalcRule extends ConverterRule {
+ public static final RelOptRule INSTANCE =
+ new StreamPhysicalAsyncCalcRule(
+ Config.INSTANCE.withConversion(
+ FlinkLogicalCalc.class,
+ FlinkConventions.LOGICAL(),
+ FlinkConventions.STREAM_PHYSICAL(),
+ "StreamPhysicalAsyncCalcRule"));
+
+ protected StreamPhysicalAsyncCalcRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ FlinkLogicalCalc calc = call.rel(0);
+ RexProgram program = calc.getProgram();
+ return program.getExprList().stream().anyMatch(AsyncUtil::containsAsyncCall);
+ }
+
+ @Override
+ public @Nullable RelNode convert(RelNode rel) {
+ FlinkLogicalCalc calc = (FlinkLogicalCalc) rel;
+ RelTraitSet traitSet = rel.getTraitSet().replace(FlinkConventions.STREAM_PHYSICAL());
+ RelNode newInput = RelOptRule.convert(calc.getInput(), FlinkConventions.STREAM_PHYSICAL());
+
+ return new StreamPhysicalAsyncCalc(
+ rel.getCluster(), traitSet, newInput, calc.getProgram(), rel.getRowType());
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/AsyncUtil.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/AsyncUtil.java
new file mode 100644
index 00000000000..61c368ff498
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/AsyncUtil.java
@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.utils;
+
+import org.apache.flink.streaming.api.datastream.AsyncDataStream;
+import org.apache.flink.streaming.api.functions.async.AsyncRetryStrategy;
+import org.apache.flink.streaming.util.retryable.AsyncRetryStrategies;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.functions.FunctionKind;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeConfig;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
+
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexNode;
+
+import java.time.Duration;
+import java.util.Objects;
+
+import static org.apache.flink.table.runtime.operators.calc.async.RetryPredicates.ANY_EXCEPTION;
+import static org.apache.flink.table.runtime.operators.calc.async.RetryPredicates.EMPTY_RESPONSE;
+
+/** Contains utilities for {@link org.apache.flink.table.functions.AsyncScalarFunction}. */
+public class AsyncUtil {
+
+ /**
+ * Checks whether it contains the specified kind of async function call in the specified node.
+ *
+ * @param node the RexNode to check
+ * @return true if it contains an async function call in the specified node.
+ */
+ public static boolean containsAsyncCall(RexNode node) {
+ return node.accept(new FunctionFinder(true, true));
+ }
+
+ /**
+ * Checks whether it contains non-async function call in the specified node.
+ *
+ * @param node the RexNode to check
+ * @return true if it contains a non-async function call in the specified node.
+ */
+ public static boolean containsNonAsyncCall(RexNode node) {
+ return node.accept(new FunctionFinder(false, true));
+ }
+
+ /**
+ * Checks whether the specified node is the specified kind of async function call.
+ *
+ * @param node the RexNode to check
+ * @return true if the specified node is an async function call.
+ */
+ public static boolean isAsyncCall(RexNode node) {
+ return node.accept(new FunctionFinder(true, false));
+ }
+
+ /**
+ * Checks whether the specified node is a non-async function call.
+ *
+ * @param node the RexNode to check
+ * @return true if the specified node is a non-async function call.
+ */
+ public static boolean isNonAsyncCall(RexNode node) {
+ return node.accept(new FunctionFinder(false, false));
+ }
+
+ /**
+ * Gets the options required to run the operator.
+ *
+ * @param config The config from which to fetch the options
+ * @return Extracted options
+ */
+ public static AsyncUtil.Options getAsyncOptions(ExecNodeConfig config) {
+ return new AsyncUtil.Options(
+ config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_SCALAR_BUFFER_CAPACITY),
+ config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_SCALAR_TIMEOUT).toMillis(),
+ AsyncDataStream.OutputMode.ORDERED,
+ getResultRetryStrategy(
+ config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_SCALAR_RETRY_STRATEGY),
+ config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_SCALAR_RETRY_DELAY),
+ config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_SCALAR_MAX_ATTEMPTS)));
+ }
+
+ /** Options for configuring async behavior. */
+ public static class Options {
+
+ public final int asyncBufferCapacity;
+ public final long asyncTimeout;
+ public final AsyncDataStream.OutputMode asyncOutputMode;
+ public final AsyncRetryStrategy<RowData> asyncRetryStrategy;
+
+ public Options(
+ int asyncBufferCapacity,
+ long asyncTimeout,
+ AsyncDataStream.OutputMode asyncOutputMode,
+ AsyncRetryStrategy<RowData> asyncRetryStrategy) {
+ this.asyncBufferCapacity = asyncBufferCapacity;
+ this.asyncTimeout = asyncTimeout;
+ this.asyncOutputMode = asyncOutputMode;
+ this.asyncRetryStrategy = asyncRetryStrategy;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Options that = (Options) o;
+ return asyncBufferCapacity == that.asyncBufferCapacity
+ && asyncTimeout == that.asyncTimeout
+ && asyncOutputMode == that.asyncOutputMode;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(asyncBufferCapacity, asyncTimeout, asyncOutputMode);
+ }
+
+ @Override
+ public String toString() {
+ return asyncOutputMode + ", " + asyncTimeout + "ms, " + asyncBufferCapacity;
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static AsyncRetryStrategy<RowData> getResultRetryStrategy(
+ ExecutionConfigOptions.RetryStrategy retryStrategy,
+ Duration retryDelay,
+ int retryMaxAttempts) {
+ // Only fixed delay is allowed at the moment, so just ignore the config.
+ if (retryStrategy == ExecutionConfigOptions.RetryStrategy.FIXED_DELAY) {
+ return new AsyncRetryStrategies.FixedDelayRetryStrategyBuilder<RowData>(
+ retryMaxAttempts, retryDelay.toMillis())
+ .ifResult(EMPTY_RESPONSE)
+ .ifException(ANY_EXCEPTION)
+ .build();
+ }
+ return AsyncRetryStrategies.NO_RETRY_STRATEGY;
+ }
+
+ private static class FunctionFinder extends RexDefaultVisitor<Boolean> {
+
+ private final boolean findAsyncCall;
+ private final boolean recursive;
+
+ public FunctionFinder(boolean findAsyncCall, boolean recursive) {
+ this.findAsyncCall = findAsyncCall;
+ this.recursive = recursive;
+ }
+
+ @Override
+ public Boolean visitNode(RexNode rexNode) {
+ return false;
+ }
+
+ private boolean isImmediateAsyncCall(RexCall call) {
+ FunctionDefinition definition = ShortcutUtils.unwrapFunctionDefinition(call);
+ return definition != null && definition.getKind() == FunctionKind.ASYNC_SCALAR;
+ }
+
+ @Override
+ public Boolean visitCall(RexCall call) {
+ boolean isImmediateAsyncCall = isImmediateAsyncCall(call);
+ return findAsyncCall == isImmediateAsyncCall
+ || (recursive
+ && call.getOperands().stream().anyMatch(node -> node.accept(this)));
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/ExecNodeMetadataUtil.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/ExecNodeMetadataUtil.java
index b478f637fc9..efded08a82e 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/ExecNodeMetadataUtil.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/ExecNodeMetadataUtil.java
@@ -29,6 +29,7 @@ import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeMetadata;
import org.apache.flink.table.planner.plan.nodes.exec.MultipleExecNodeMetadata;
import org.apache.flink.table.planner.plan.nodes.exec.serde.JsonSerdeUtil;
+import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecAsyncCalc;
import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc;
import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecChangelogNormalize;
import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCorrelate;
@@ -139,6 +140,7 @@ public final class ExecNodeMetadataUtil {
add(StreamExecWindowRank.class);
add(StreamExecWindowTableFunction.class);
add(StreamExecPythonCalc.class);
+ add(StreamExecAsyncCalc.class);
add(StreamExecPythonCorrelate.class);
add(StreamExecPythonGroupAggregate.class);
add(StreamExecPythonGroupWindowAggregate.class);
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala
index 40a3df64057..e8da1a85579 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala
@@ -26,6 +26,7 @@ import org.apache.flink.table.functions.{FunctionContext, UserDefinedFunction}
import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.codegen.FunctionCodeGenerator.generateFunction
import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable.{JSON_ARRAY, JSON_OBJECT}
+import org.apache.flink.table.planner.plan.utils.AsyncUtil
import org.apache.flink.table.planner.plan.utils.PythonUtil.containsPythonCall
import org.apache.flink.table.planner.utils.JavaScalaConversionUtil.toScala
import org.apache.flink.table.planner.utils.Logging
@@ -65,9 +66,9 @@ class ExpressionReducer(
constExprs: java.util.List[RexNode],
reducedValues: java.util.List[RexNode]): Unit = {
- val pythonUDFExprs = new ListBuffer[RexNode]()
+ val remoteUDFExprs = new ListBuffer[RexNode]()
- val literals = skipAndValidateExprs(rexBuilder, constExprs, pythonUDFExprs)
+ val literals = skipAndValidateExprs(rexBuilder, constExprs, remoteUDFExprs)
val literalTypes = literals.map(e => FlinkTypeFactory.toLogicalType(e.getType))
val resultType = RowType.of(literalTypes: _*)
@@ -131,8 +132,8 @@ class ExpressionReducer(
while (i < constExprs.size()) {
val unreduced = constExprs.get(i)
// use eq to compare reference
- if (pythonUDFExprs.exists(_ eq unreduced)) {
- // if contains python function then just insert the original expression.
+ if (remoteUDFExprs.exists(_ eq unreduced)) {
+ // if contains async function then just insert the original expression.
reducedValues.add(unreduced)
} else
unreduced match {
@@ -253,15 +254,15 @@ class ExpressionReducer(
private def skipAndValidateExprs(
rexBuilder: RexBuilder,
constExprs: java.util.List[RexNode],
- pythonUDFExprs: ListBuffer[RexNode]): List[RexNode] = {
+ remoteUDFExprs: ListBuffer[RexNode]): List[RexNode] = {
constExprs.asScala
.map(e => (e.getType.getSqlTypeName, e))
.flatMap {
- // Skip expressions that contain python functions because it's quite expensive to
- // call Python UDFs during optimization phase. They will be optimized during the runtime.
- case (_, e) if containsPythonCall(e) =>
- pythonUDFExprs += e
+ // Skip expressions that contain python or async functions because it's quite expensive to
+ // call async UDFs during optimization phase. They will be optimized during the runtime.
+ case (_, e) if containsPythonCall(e) || AsyncUtil.containsAsyncCall(e) =>
+ remoteUDFExprs += e
None
// we don't support object literals yet, we skip those constant expressions
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingFunctionGenUtil.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingFunctionGenUtil.scala
index 87c666ca5dc..5ad355e19ee 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingFunctionGenUtil.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingFunctionGenUtil.scala
@@ -27,7 +27,7 @@ import org.apache.flink.table.expressions.ApiExpressionUtils.{typeLiteral, unres
import org.apache.flink.table.expressions.Expression
import org.apache.flink.table.functions._
import org.apache.flink.table.functions.SpecializedFunction.{ExpressionEvaluator, ExpressionEvaluatorFactory}
-import org.apache.flink.table.functions.UserDefinedFunctionHelper.{validateClassForRuntime, ASYNC_TABLE_EVAL, SCALAR_EVAL, TABLE_EVAL}
+import org.apache.flink.table.functions.UserDefinedFunctionHelper.{validateClassForRuntime, ASYNC_SCALAR_EVAL, ASYNC_TABLE_EVAL, SCALAR_EVAL, TABLE_EVAL}
import org.apache.flink.table.planner.calcite.{FlinkTypeFactory, RexFactory}
import org.apache.flink.table.planner.codegen._
import org.apache.flink.table.planner.codegen.CodeGenUtils._
@@ -47,6 +47,8 @@ import org.apache.flink.table.types.logical.utils.LogicalTypeChecks.isCompositeT
import org.apache.flink.table.types.utils.DataTypeUtils.{isInternal, validateInputDataType, validateOutputDataType}
import org.apache.flink.util.Preconditions
+import AsyncCodeGenerator.DEFAULT_DELEGATING_FUTURE_TERM
+
import java.util.concurrent.CompletableFuture
import scala.collection.JavaConverters._
@@ -147,6 +149,13 @@ object BridgingFunctionGenUtil {
skipIfArgsNull)
} else if (udf.getKind == FunctionKind.ASYNC_TABLE) {
generateAsyncTableFunctionCall(functionTerm, externalOperands, returnType)
+ } else if (udf.getKind == FunctionKind.ASYNC_SCALAR) {
+ generateAsyncScalarFunctionCall(
+ ctx,
+ functionTerm,
+ externalOperands,
+ returnType,
+ outputDataType)
} else {
generateScalarFunctionCall(ctx, functionTerm, externalOperands, outputDataType)
}
@@ -210,6 +219,28 @@ object BridgingFunctionGenUtil {
GeneratedExpression(NO_CODE, NEVER_NULL, functionCallCode, outputType)
}
+ private def generateAsyncScalarFunctionCall(
+ ctx: CodeGeneratorContext,
+ functionTerm: String,
+ externalOperands: Seq[GeneratedExpression],
+ outputType: LogicalType,
+ outputDataType: DataType): GeneratedExpression = {
+ val converterTerm = ctx.addReusableConverter(outputDataType)
+ val functionCallCode =
+ s"""
+ |${externalOperands.map(_.code).mkString("\n")}
+ |if (${externalOperands.map(_.nullTerm).mkString(" || ")}) {
+ | $DEFAULT_DELEGATING_FUTURE_TERM.createAsyncFuture($converterTerm).complete(null);
+ |} else {
+ | $functionTerm.eval(
+ | $DEFAULT_DELEGATING_FUTURE_TERM.createAsyncFuture($converterTerm),
+ | ${externalOperands.map(_.resultTerm).mkString(", ")});
+ |}
+ |""".stripMargin
+
+ GeneratedExpression(NO_CODE, NEVER_NULL, functionCallCode, outputType)
+ }
+
/**
* Generates a collector that converts the output of a table function (possibly as an atomic type)
* into an internal row type. Returns a collector term for referencing the collector.
@@ -389,6 +420,13 @@ object BridgingFunctionGenUtil {
functionName)
} else if (udf.getKind == FunctionKind.SCALAR) {
verifyImplementation(SCALAR_EVAL, argumentDataTypes, Some(outputDataType), udf, functionName)
+ } else if (udf.getKind == FunctionKind.ASYNC_SCALAR) {
+ verifyImplementation(
+ ASYNC_SCALAR_EVAL,
+ DataTypes.NULL.bridgedTo(classOf[CompletableFuture[_]]) +: argumentDataTypes,
+ None,
+ udf,
+ functionName)
} else {
throw new CodeGenException(
s"Unsupported function kind '${udf.getKind}' for function '$functionName'.")
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
index 7e337799ccd..f079a8aca28 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
@@ -372,17 +372,48 @@ object FlinkStreamRuleSets {
PushFilterInCalcIntoTableSourceScanRule.INSTANCE,
// Rule that rewrites temporal join with extracted primary key
TemporalJoinRewriteWithUniqueKeyRule.INSTANCE,
- // Rule that splits python ScalarFunctions from java/scala ScalarFunctions.
+ // Avoids accessing a field from the result (condition).
PythonCalcSplitRule.SPLIT_CONDITION_REX_FIELD,
+ // Avoids accessing a field from the result (projection).
PythonCalcSplitRule.SPLIT_PROJECTION_REX_FIELD,
+ // Avoids dealing with a python call in the condition.
PythonCalcSplitRule.SPLIT_CONDITION,
+ // Avoids dealing with Java calls in the same Calc as python calls.
PythonCalcSplitRule.SPLIT_PROJECT,
+ // Splits calcs which contain both general Python functions and pandas Python functions
PythonCalcSplitRule.SPLIT_PANDAS_IN_PROJECT,
+ // Avoid accessing a field as input to an async call with a single calc.
PythonCalcSplitRule.EXPAND_PROJECT,
+ // Avoid having any condition in a python calc by pushing it first.
PythonCalcSplitRule.PUSH_CONDITION,
+ // Orders the projections so that input references are first, followed by python calls.
PythonCalcSplitRule.REWRITE_PROJECT,
+ // Renames the field names of the Flatten calc which is right after a calc representing a
+ // Python Map operation to the output names of the map function
PythonMapRenameRule.INSTANCE,
- PythonMapMergeRule.INSTANCE
+ // Merges Python calc used in Map operation, Flatten calcs and Python calcs used in Map
+ // xoperation together
+ PythonMapMergeRule.INSTANCE,
+ // Similar to the python rules above, the goal is to limit complexity of calcs which
+ // have async calls so that the implementation can be simplified to handle a single async call.
+ // Avoids accessing a field from an asynchronous result (condition).
+ AsyncCalcSplitRule.SPLIT_CONDITION_REX_FIELD,
+ // Avoids accessing a field from an asynchronous result (projection).
+ AsyncCalcSplitRule.SPLIT_PROJECTION_REX_FIELD,
+ // Avoids dealing with an async call in the condition.
+ AsyncCalcSplitRule.SPLIT_CONDITION,
+ // Avoids dealing with Java calls in the same Calc as Async calls.
+ AsyncCalcSplitRule.SPLIT_PROJECT,
+ // Avoid accessing a field as input to an async call with a single calc.
+ AsyncCalcSplitRule.EXPAND_PROJECT,
+ // Avoid having any condition in an async calc by pushing it first.
+ AsyncCalcSplitRule.PUSH_CONDITION,
+ // Orders the projections so that input references are first, followed by async calls.
+ AsyncCalcSplitRule.REWRITE_PROJECT,
+ // Avoid async calls which call async calls.
+ AsyncCalcSplitRule.NESTED_SPLIT,
+ // Avoid having async calls in multiple projections in a single calc.
+ AsyncCalcSplitRule.ONE_PER_CALC_SPLIT
)
/** RuleSet to do physical optimize for stream */
@@ -400,6 +431,7 @@ object FlinkStreamRuleSets {
// calc
StreamPhysicalCalcRule.INSTANCE,
StreamPhysicalPythonCalcRule.INSTANCE,
+ StreamPhysicalAsyncCalcRule.INSTANCE,
// union
StreamPhysicalUnionRule.INSTANCE,
// sort
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRule.scala
index 725966ac510..1756b9eb5e8 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRule.scala
@@ -20,7 +20,7 @@ package org.apache.flink.table.planner.plan.rules.logical
import org.apache.flink.table.functions.ScalarFunction
import org.apache.flink.table.functions.python.PythonFunctionKind
import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalCalc
-import org.apache.flink.table.planner.plan.utils.{InputRefVisitor, RexDefaultVisitor}
+import org.apache.flink.table.planner.plan.utils.{InputRefVisitor, PythonUtil, RexDefaultVisitor}
import org.apache.flink.table.planner.plan.utils.PythonUtil.{containsNonPythonCall, containsPythonCall, isNonPythonCall, isPythonCall}
import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
@@ -34,245 +34,14 @@ import scala.collection.JavaConversions._
import scala.collection.JavaConverters._
import scala.collection.mutable
-/**
- * Base rule that splits [[FlinkLogicalCalc]] into multiple [[FlinkLogicalCalc]]s. It is mainly to
- * ensure that each [[FlinkLogicalCalc]] only contains Java/Scala [[ScalarFunction]]s or Python
- * [[ScalarFunction]]s.
- */
-abstract class PythonCalcSplitRuleBase(description: String)
- extends RelOptRule(operand(classOf[FlinkLogicalCalc], any), description) {
-
- override def onMatch(call: RelOptRuleCall): Unit = {
- val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
- val input = calc.getInput
- val rexBuilder = call.builder().getRexBuilder
- val program = calc.getProgram
- val extractedRexNodes = new mutable.ArrayBuffer[RexNode]()
-
- val extractedFunctionOffset = input.getRowType.getFieldCount
- val splitter = new ScalarFunctionSplitter(
- program,
- rexBuilder,
- extractedFunctionOffset,
- extractedRexNodes,
- new Function[RexNode, Boolean] {
- override def apply(node: RexNode): Boolean = needConvert(program, node)
- })
-
- val (bottomCalcCondition, topCalcCondition, topCalcProjects) = split(program, splitter)
- val accessedFields =
- extractRefInputFields(topCalcProjects, topCalcCondition, extractedFunctionOffset)
-
- val bottomCalcProjects =
- accessedFields.map(RexInputRef.of(_, input.getRowType)) ++ extractedRexNodes
- val bottomCalcFieldNames = SqlValidatorUtil.uniquify(
- accessedFields.map(i => input.getRowType.getFieldNames.get(i)).toSeq ++
- extractedRexNodes.indices.map("f" + _),
- rexBuilder.getTypeFactory.getTypeSystem.isSchemaCaseSensitive
- )
-
- val bottomCalc = new FlinkLogicalCalc(
- calc.getCluster,
- calc.getTraitSet,
- input,
- RexProgram.create(
- input.getRowType,
- bottomCalcProjects.toList,
- bottomCalcCondition.orNull,
- bottomCalcFieldNames,
- rexBuilder))
-
- val inputRewriter = new ExtractedFunctionInputRewriter(
- calc.getCluster.getRexBuilder,
- extractedFunctionOffset,
- accessedFields)
- val topCalc = new FlinkLogicalCalc(
- calc.getCluster,
- calc.getTraitSet,
- bottomCalc,
- RexProgram.create(
- bottomCalc.getRowType,
- topCalcProjects.map(_.accept(inputRewriter)),
- topCalcCondition.map(_.accept(inputRewriter)).orNull,
- calc.getRowType,
- rexBuilder))
-
- call.transformTo(topCalc)
- }
-
- /** Extracts the indices of the input fields referred by the specified projects and condition. */
- private def extractRefInputFields(
- projects: Seq[RexNode],
- condition: Option[RexNode],
- inputFieldsCount: Int): Array[Int] = {
- val visitor = new InputRefVisitor
-
- // extract referenced input fields from projections
- projects.foreach(exp => exp.accept(visitor))
-
- // extract referenced input fields from condition
- condition.foreach(_.accept(visitor))
-
- // fields of indexes greater than inputFieldsCount is the extracted functions and
- // should be filtered as they are not from the original input
- visitor.getFields.filter(_ < inputFieldsCount)
- }
-
- /** Returns true if need to convert the specified node. */
- def needConvert(program: RexProgram, node: RexNode): Boolean
-
- /**
- * Splits the specified [[RexProgram]] using the specified [[ScalarFunctionSplitter]]. It returns
- * a triple of (bottom calc condition, top calc condition, top calc projects) as the split result.
- */
- def split(
- program: RexProgram,
- splitter: ScalarFunctionSplitter): (Option[RexNode], Option[RexNode], Seq[RexNode])
-}
-
-/**
- * Rule that splits [[FlinkLogicalCalc]]s which contain Python functions in the condition into
- * multiple [[FlinkLogicalCalc]]s. After this rule is applied, there will be no Python functions in
- * the condition of the [[FlinkLogicalCalc]]s.
- */
-object PythonCalcSplitConditionRule
- extends PythonCalcSplitRuleBase("PythonCalcSplitConditionRule") {
-
- override def matches(call: RelOptRuleCall): Boolean = {
- val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
-
- // matches if it contains Python functions in condition
- Option(calc.getProgram.getCondition)
- .map(calc.getProgram.expandLocalRef)
- .exists(containsPythonCall(_))
- }
-
- override def needConvert(program: RexProgram, node: RexNode): Boolean = isPythonCall(node)
-
- override def split(
- program: RexProgram,
- splitter: ScalarFunctionSplitter): (Option[RexNode], Option[RexNode], Seq[RexNode]) = {
- (
- None,
- Option(program.getCondition).map(program.expandLocalRef(_).accept(splitter)),
- program.getProjectList.map(program.expandLocalRef))
- }
-}
-
-abstract class PythonCalcSplitProjectionRuleBase(description: String)
- extends PythonCalcSplitRuleBase(description) {
-
- override def split(
- program: RexProgram,
- splitter: ScalarFunctionSplitter): (Option[RexNode], Option[RexNode], Seq[RexNode]) = {
- (
- Option(program.getCondition).map(program.expandLocalRef),
- None,
- program.getProjectList.map(program.expandLocalRef(_).accept(splitter)))
- }
-}
-
-abstract class PythonCalcSplitRexFieldRuleBase(description: String)
- extends PythonCalcSplitRuleBase(description) {
-
- override def needConvert(program: RexProgram, node: RexNode): Boolean = {
- node match {
- case x: RexFieldAccess =>
- x.getReferenceExpr match {
- case y: RexLocalRef if containsPythonCall(program.expandLocalRef(y)) => true
- case _ => false
- }
- case _ => false
- }
- }
-
- protected def containsFieldAccessAfterPythonCall(node: RexNode): Boolean = {
- node match {
- case call: RexCall => call.getOperands.exists(containsFieldAccessAfterPythonCall)
- case x: RexFieldAccess => containsPythonCall(x.getReferenceExpr)
- case _ => false
- }
- }
-}
-
-/**
- * Rule that splits the RexField with the input of Python function contained in the projection of
- * [[FlinkLogicalCalc]]s.
- */
-object PythonCalcSplitProjectionRexFieldRule
- extends PythonCalcSplitRexFieldRuleBase("PythonCalcSplitProjectionRexFieldRule") {
-
- override def matches(call: RelOptRuleCall): Boolean = {
- val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
-
- val projects = calc.getProgram.getProjectList.map(calc.getProgram.expandLocalRef)
- projects.exists(containsFieldAccessAfterPythonCall)
- }
-
- override def split(
- program: RexProgram,
- splitter: ScalarFunctionSplitter): (Option[RexNode], Option[RexNode], Seq[RexNode]) = {
- (
- Option(program.getCondition).map(program.expandLocalRef),
- None,
- program.getProjectList.map(_.accept(splitter)))
- }
-}
-
-/**
- * Rule that splits the RexField with the input of Python function contained in the condition of
- * [[FlinkLogicalCalc]]s.
- */
-object PythonCalcSplitConditionRexFieldRule
- extends PythonCalcSplitRexFieldRuleBase("PythonCalcSplitConditionRexFieldRule") {
-
- override def matches(call: RelOptRuleCall): Boolean = {
- val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
-
- Option(calc.getProgram.getCondition)
- .map(calc.getProgram.expandLocalRef)
- .exists(containsFieldAccessAfterPythonCall)
- }
-
- override def split(
- program: RexProgram,
- splitter: ScalarFunctionSplitter): (Option[RexNode], Option[RexNode], Seq[RexNode]) = {
- (
- None,
- Option(program.getCondition).map(_.accept(splitter)),
- program.getProjectList.map(_.accept(splitter)))
- }
-}
-
-/**
- * Rule that splits [[FlinkLogicalCalc]]s which contain both Java functions and Python functions in
- * the projection into multiple [[FlinkLogicalCalc]]s. After this rule is applied, it will only
- * contain Python functions or Java functions in the projection of each [[FlinkLogicalCalc]].
- */
-object PythonCalcSplitProjectionRule
- extends PythonCalcSplitProjectionRuleBase("PythonCalcSplitProjectionRule") {
-
- override def matches(call: RelOptRuleCall): Boolean = {
- val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
- val projects = calc.getProgram.getProjectList.map(calc.getProgram.expandLocalRef)
-
- // matches if it contains both Python functions and Java functions in the projection
- projects.exists(containsPythonCall(_)) && projects.exists(containsNonPythonCall)
- }
-
- override def needConvert(program: RexProgram, node: RexNode): Boolean = {
- program.getProjectList.map(program.expandLocalRef).exists(isNonPythonCall) == isPythonCall(node)
- }
-}
-
/**
* Rule that splits [[FlinkLogicalCalc]]s which contain both general Python functions and pandas
* Python functions in the projection into multiple [[FlinkLogicalCalc]]s. After this rule is
* applied, it will only contain general Python functions or pandas Python functions in the
* projection of each [[FlinkLogicalCalc]].
*/
-object PythonCalcSplitPandasInProjectionRule
- extends PythonCalcSplitProjectionRuleBase("PythonCalcSplitPandasInProjectionRule") {
+class PythonCalcSplitPandasInProjectionRule(callFinder: RemoteCalcCallFinder)
+ extends RemoteCalcSplitProjectionRuleBase("PythonCalcSplitPandasInProjectionRule", callFinder) {
override def matches(call: RelOptRuleCall): Boolean = {
val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
@@ -284,7 +53,10 @@ object PythonCalcSplitPandasInProjectionRule
projects.exists(containsPythonCall(_, PythonFunctionKind.PANDAS))
}
- override def needConvert(program: RexProgram, node: RexNode): Boolean = {
+ override def needConvert(
+ program: RexProgram,
+ node: RexNode,
+ matchState: Option[Nothing]): Boolean = {
program.getProjectList
.map(program.expandLocalRef)
.exists(isPythonCall(_, PythonFunctionKind.GENERAL)) == isPythonCall(
@@ -293,222 +65,37 @@ object PythonCalcSplitPandasInProjectionRule
}
}
-/**
- * Rule that expands the RexFieldAccess inputs of Python functions contained in the projection of
- * [[FlinkLogicalCalc]]s.
- */
-object PythonCalcExpandProjectRule extends PythonCalcSplitRuleBase("PythonCalcExpandProjectRule") {
-
- override def matches(call: RelOptRuleCall): Boolean = {
- val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
- val projects = calc.getProgram.getProjectList.map(calc.getProgram.expandLocalRef)
-
- projects.exists(containsPythonCall(_)) && projects.exists(containsFieldAccessInputs)
+class PythonRemoteCalcCallFinder extends RemoteCalcCallFinder {
+ override def containsRemoteCall(node: RexNode): Boolean = {
+ PythonUtil.containsPythonCall(node)
}
- override def needConvert(program: RexProgram, node: RexNode): Boolean =
- node.isInstanceOf[RexFieldAccess]
-
- override def split(
- program: RexProgram,
- splitter: ScalarFunctionSplitter): (Option[RexNode], Option[RexNode], Seq[RexNode]) = {
- (
- Option(program.getCondition).map(program.expandLocalRef),
- None,
- program.getProjectList.map(program.expandLocalRef(_).accept(splitter)))
- }
-
- private def containsFieldAccessInputs(node: RexNode): Boolean = {
- node match {
- case call: RexCall => call.getOperands.exists(containsFieldAccessInputs)
- case _: RexFieldAccess => true
- case _ => false
- }
+ override def containsNonRemoteCall(node: RexNode): Boolean = {
+ PythonUtil.containsNonPythonCall(node)
}
-}
-
-/**
- * Rule that pushes the condition of [[FlinkLogicalCalc]]s before it for the [[FlinkLogicalCalc]]s
- * which contain Python functions in the projection.
- */
-object PythonCalcPushConditionRule extends PythonCalcSplitRuleBase("PythonCalcPushConditionRule") {
-
- override def matches(call: RelOptRuleCall): Boolean = {
- val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
- val projects = calc.getProgram.getProjectList.map(calc.getProgram.expandLocalRef)
- // matches if all the following conditions hold true:
- // 1) the condition is not null
- // 2) it contains Python functions in the projection
- calc.getProgram.getCondition != null && projects.exists(containsPythonCall(_))
+ override def isRemoteCall(node: RexNode): Boolean = {
+ PythonUtil.isPythonCall(node)
}
- override def needConvert(program: RexProgram, node: RexNode): Boolean = isNonPythonCall(node)
-
- override def split(
- program: RexProgram,
- splitter: ScalarFunctionSplitter): (Option[RexNode], Option[RexNode], Seq[RexNode]) = {
- (
- Option(program.getCondition).map(program.expandLocalRef),
- None,
- program.getProjectList.map(program.expandLocalRef))
+ override def isNonRemoteCall(node: RexNode): Boolean = {
+ PythonUtil.isNonPythonCall(node)
}
}
-/**
- * Rule that ensures that it only contains [[RexInputRef]]s at the beginning of the project list and
- * [[RexCall]]s at the end of the project list for [[FlinkLogicalCalc]]s which contain Python
- * functions in the projection. This rule exists to keep DataStreamPythonCalc as simple as possible
- * and ensures that it only needs to handle the Python function execution.
- */
-object PythonCalcRewriteProjectionRule
- extends PythonCalcSplitRuleBase("PythonCalcRewriteProjectionRule") {
-
- override def matches(call: RelOptRuleCall): Boolean = {
- val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
- val projects = calc.getProgram.getProjectList.map(calc.getProgram.expandLocalRef)
-
- // matches if all the following conditions hold true:
- // 1) it contains Python functions in the projection
- // 2) it contains RexNodes besides RexInputRef and RexCall or
- // not all the RexCalls lying at the end of the project list
- projects.exists(containsPythonCall(_)) &&
- (projects.exists(expr => !expr.isInstanceOf[RexCall] && !expr.isInstanceOf[RexInputRef]) ||
- projects.indexWhere(_.isInstanceOf[RexCall]) <
- projects.lastIndexWhere(_.isInstanceOf[RexInputRef]))
- }
-
- override def needConvert(program: RexProgram, node: RexNode): Boolean = isPythonCall(node)
-
- override def split(
- program: RexProgram,
- splitter: ScalarFunctionSplitter): (Option[RexNode], Option[RexNode], Seq[RexNode]) = {
- (None, None, program.getProjectList.map(program.expandLocalRef(_).accept(splitter)))
- }
-}
-
-private class ScalarFunctionSplitter(
- program: RexProgram,
- rexBuilder: RexBuilder,
- extractedFunctionOffset: Int,
- extractedRexNodes: mutable.ArrayBuffer[RexNode],
- needConvert: Function[RexNode, Boolean])
- extends RexDefaultVisitor[RexNode] {
-
- private var fieldsRexCall: Map[Int, Int] = Map[Int, Int]()
-
- override def visitCall(call: RexCall): RexNode = {
- if (needConvert(call)) {
- getExtractedRexNode(call)
- } else {
- call.clone(call.getType, call.getOperands.asScala.map(_.accept(this)))
- }
- }
-
- override def visitFieldAccess(fieldAccess: RexFieldAccess): RexNode = {
- if (needConvert(fieldAccess)) {
- val expr = fieldAccess.getReferenceExpr
- expr match {
- case localRef: RexLocalRef if containsPythonCall(program.expandLocalRef(localRef)) =>
- getExtractedRexFieldAccess(fieldAccess, localRef.getIndex)
- case _: RexCorrelVariable =>
- val field = fieldAccess.getField
- new RexInputRef(field.getIndex, field.getType)
- case _ =>
- val newFieldAccess =
- rexBuilder.makeFieldAccess(expr.accept(this), fieldAccess.getField.getIndex)
- getExtractedRexNode(newFieldAccess)
- }
- } else {
- fieldAccess
- }
- }
-
- override def visitLocalRef(localRef: RexLocalRef): RexNode = {
- program.getExprList.get(localRef.getIndex).accept(this)
- }
-
- override def visitNode(rexNode: RexNode): RexNode = rexNode
-
- private def getExtractedRexNode(node: RexNode): RexNode = {
- val newNode = new RexInputRef(extractedFunctionOffset + extractedRexNodes.length, node.getType)
- extractedRexNodes.append(node)
- newNode
- }
-
- private def getExtractedRexFieldAccess(node: RexFieldAccess, rexCallIndex: Int): RexNode = {
- val pythonCall: RexCall =
- program.expandLocalRef(node.getReferenceExpr.asInstanceOf[RexLocalRef]).asInstanceOf[RexCall]
- if (!fieldsRexCall.contains(rexCallIndex)) {
- extractedRexNodes.append(pythonCall)
- fieldsRexCall += rexCallIndex -> (extractedFunctionOffset + extractedRexNodes.length - 1)
- }
- rexBuilder.makeFieldAccess(
- new RexInputRef(fieldsRexCall(rexCallIndex), pythonCall.getType),
- node.getField.getIndex)
- }
-}
-
-/**
- * Rewrite field accesses of a RexNode as not all the fields from the original input are forwarded:
- * 1) Fields of index greater than or equal to extractedFunctionOffset refer to the extracted
- * function. 2) Fields of index less than extractedFunctionOffset refer to the original input field.
- *
- * @param rexBuilder
- * the RexBuilder
- * @param extractedFunctionOffset
- * the original start offset of the extracted functions
- * @param accessedFields
- * the accessed fields which will be forwarded
- */
-private class ExtractedFunctionInputRewriter(
- rexBuilder: RexBuilder,
- extractedFunctionOffset: Int,
- accessedFields: Array[Int])
- extends RexDefaultVisitor[RexNode] {
-
- /** old input fields ref index -> new input fields ref index mappings */
- private val fieldMap: Map[Int, Int] = accessedFields.zipWithIndex.toMap
-
- override def visitInputRef(inputRef: RexInputRef): RexNode = {
- if (inputRef.getIndex >= extractedFunctionOffset) {
- new RexInputRef(
- inputRef.getIndex - extractedFunctionOffset + accessedFields.length,
- inputRef.getType)
- } else {
- new RexInputRef(
- fieldMap.getOrElse(
- inputRef.getIndex,
- throw new IllegalArgumentException("input field contains invalid index")),
- inputRef.getType)
- }
- }
-
- override def visitCall(call: RexCall): RexNode = {
- call.clone(call.getType, call.getOperands.asScala.map(_.accept(this)))
- }
-
- override def visitFieldAccess(fieldAccess: RexFieldAccess): RexNode = {
- rexBuilder.makeFieldAccess(
- fieldAccess.getReferenceExpr.accept(this),
- fieldAccess.getField.getIndex)
- }
-
- override def visitNode(rexNode: RexNode): RexNode = rexNode
-}
-
object PythonCalcSplitRule {
/**
* These rules should be applied sequentially in the order of SPLIT_CONDITION, SPLIT_PROJECT,
* SPLIT_PANDAS_IN_PROJECT, EXPAND_PROJECT, PUSH_CONDITION and REWRITE_PROJECT.
*/
- val SPLIT_CONDITION: RelOptRule = PythonCalcSplitConditionRule
- val SPLIT_PROJECT: RelOptRule = PythonCalcSplitProjectionRule
- val SPLIT_PANDAS_IN_PROJECT: RelOptRule = PythonCalcSplitPandasInProjectionRule
- val SPLIT_PROJECTION_REX_FIELD: RelOptRule = PythonCalcSplitProjectionRexFieldRule
- val SPLIT_CONDITION_REX_FIELD: RelOptRule = PythonCalcSplitConditionRexFieldRule
- val EXPAND_PROJECT: RelOptRule = PythonCalcExpandProjectRule
- val PUSH_CONDITION: RelOptRule = PythonCalcPushConditionRule
- val REWRITE_PROJECT: RelOptRule = PythonCalcRewriteProjectionRule
+ private val callFinder = new PythonRemoteCalcCallFinder()
+ val SPLIT_CONDITION: RelOptRule = new RemoteCalcSplitConditionRule(callFinder)
+ val SPLIT_PROJECT: RelOptRule = new RemoteCalcSplitProjectionRule(callFinder)
+ val SPLIT_PANDAS_IN_PROJECT: RelOptRule = new PythonCalcSplitPandasInProjectionRule(callFinder)
+ val SPLIT_PROJECTION_REX_FIELD: RelOptRule = new RemoteCalcSplitProjectionRexFieldRule(callFinder)
+ val SPLIT_CONDITION_REX_FIELD: RelOptRule = new RemoteCalcSplitConditionRexFieldRule(callFinder)
+ val EXPAND_PROJECT: RelOptRule = new RemoteCalcExpandProjectRule(callFinder)
+ val PUSH_CONDITION: RelOptRule = new RemoteCalcPushConditionRule(callFinder)
+ val REWRITE_PROJECT: RelOptRule = new RemoteCalcRewriteProjectionRule(callFinder)
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/FunctionKind.java b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcCallFinder.java
similarity index 56%
copy from flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/FunctionKind.java
copy to flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcCallFinder.java
index 32fc9fbcf23..5209ab1b14b 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/FunctionKind.java
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcCallFinder.java
@@ -16,22 +16,22 @@
* limitations under the License.
*/
-package org.apache.flink.table.functions;
+package org.apache.flink.table.planner.plan.rules.logical;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.calcite.rex.RexNode;
-/** Categorizes the semantics of a {@link FunctionDefinition}. */
-@PublicEvolving
-public enum FunctionKind {
- SCALAR,
-
- TABLE,
-
- ASYNC_TABLE,
-
- AGGREGATE,
+/**
+ * Finds whether a {@link RexNode} contains remote calls or not.
+ */
+public interface RemoteCalcCallFinder {
- TABLE_AGGREGATE,
+ // If the node contains directly or indirectly a remote call.
+ boolean containsRemoteCall(RexNode node);
+ // If the node contains directly or indirectly a non-remote call.
+ boolean containsNonRemoteCall(RexNode node);
+ // If the node contains directly a remote call.
+ boolean isRemoteCall(RexNode node);
- OTHER
+ // If the node contains directly a non-remote call.
+ boolean isNonRemoteCall(RexNode node);
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcSplitRule.scala
similarity index 63%
copy from flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRule.scala
copy to flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcSplitRule.scala
index 725966ac510..62a8a7ea369 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCalcSplitRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/RemoteCalcSplitRule.scala
@@ -18,10 +18,8 @@
package org.apache.flink.table.planner.plan.rules.logical
import org.apache.flink.table.functions.ScalarFunction
-import org.apache.flink.table.functions.python.PythonFunctionKind
import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalCalc
import org.apache.flink.table.planner.plan.utils.{InputRefVisitor, RexDefaultVisitor}
-import org.apache.flink.table.planner.plan.utils.PythonUtil.{containsNonPythonCall, containsPythonCall, isNonPythonCall, isPythonCall}
import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
import org.apache.calcite.plan.RelOptRule.{any, operand}
@@ -36,12 +34,24 @@ import scala.collection.mutable
/**
* Base rule that splits [[FlinkLogicalCalc]] into multiple [[FlinkLogicalCalc]]s. It is mainly to
- * ensure that each [[FlinkLogicalCalc]] only contains Java/Scala [[ScalarFunction]]s or Python
+ * ensure that each [[FlinkLogicalCalc]] only contains Java/Scala [[ScalarFunction]]s or Remote
* [[ScalarFunction]]s.
*/
-abstract class PythonCalcSplitRuleBase(description: String)
+abstract class RemoteCalcSplitRuleBase[T](
+ description: String,
+ protected val callFinder: RemoteCalcCallFinder)
extends RelOptRule(operand(classOf[FlinkLogicalCalc], any), description) {
+ // Consider the rules to be equal if they are the same class and their call finders are the same
+ // class.
+ override def equals(obj: Any): Boolean = {
+ obj match {
+ case base: RemoteCalcSplitRuleBase[_] =>
+ super.equals(base) && callFinder.getClass.equals(base.callFinder.getClass)
+ case _ => false
+ }
+ }
+
override def onMatch(call: RelOptRuleCall): Unit = {
val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
val input = calc.getInput
@@ -50,18 +60,23 @@ abstract class PythonCalcSplitRuleBase(description: String)
val extractedRexNodes = new mutable.ArrayBuffer[RexNode]()
val extractedFunctionOffset = input.getRowType.getFieldCount
+ val matchState = getMatchState;
val splitter = new ScalarFunctionSplitter(
program,
rexBuilder,
extractedFunctionOffset,
extractedRexNodes,
new Function[RexNode, Boolean] {
- override def apply(node: RexNode): Boolean = needConvert(program, node)
- })
+ override def apply(node: RexNode): Boolean = needConvert(program, node, matchState)
+ },
+ callFinder)
- val (bottomCalcCondition, topCalcCondition, topCalcProjects) = split(program, splitter)
+ val splitComponents = split(program, splitter)
val accessedFields =
- extractRefInputFields(topCalcProjects, topCalcCondition, extractedFunctionOffset)
+ extractRefInputFields(
+ splitComponents.topCalcProjects,
+ splitComponents.topCalcCondition,
+ extractedFunctionOffset)
val bottomCalcProjects =
accessedFields.map(RexInputRef.of(_, input.getRowType)) ++ extractedRexNodes
@@ -78,7 +93,7 @@ abstract class PythonCalcSplitRuleBase(description: String)
RexProgram.create(
input.getRowType,
bottomCalcProjects.toList,
- bottomCalcCondition.orNull,
+ splitComponents.bottomCalcCondition.orNull,
bottomCalcFieldNames,
rexBuilder))
@@ -92,10 +107,11 @@ abstract class PythonCalcSplitRuleBase(description: String)
bottomCalc,
RexProgram.create(
bottomCalc.getRowType,
- topCalcProjects.map(_.accept(inputRewriter)),
- topCalcCondition.map(_.accept(inputRewriter)).orNull,
+ splitComponents.topCalcProjects.map(_.accept(inputRewriter)),
+ splitComponents.topCalcCondition.map(_.accept(inputRewriter)).orNull,
calc.getRowType,
- rexBuilder))
+ rexBuilder
+ ))
call.transformTo(topCalc)
}
@@ -119,100 +135,123 @@ abstract class PythonCalcSplitRuleBase(description: String)
}
/** Returns true if need to convert the specified node. */
- def needConvert(program: RexProgram, node: RexNode): Boolean
+ def getMatchState: Option[T] = {
+ Option.empty
+ }
+
+ /** Returns true if need to convert the specified node. */
+ def needConvert(program: RexProgram, node: RexNode, matchState: Option[T]): Boolean
/**
* Splits the specified [[RexProgram]] using the specified [[ScalarFunctionSplitter]]. It returns
* a triple of (bottom calc condition, top calc condition, top calc projects) as the split result.
*/
- def split(
- program: RexProgram,
- splitter: ScalarFunctionSplitter): (Option[RexNode], Option[RexNode], Seq[RexNode])
+ def split(program: RexProgram, splitter: ScalarFunctionSplitter): SplitComponents
}
/**
- * Rule that splits [[FlinkLogicalCalc]]s which contain Python functions in the condition into
- * multiple [[FlinkLogicalCalc]]s. After this rule is applied, there will be no Python functions in
+ * The components for splitting a single calc into two calcs.
+ *
+ * @param bottomCalcCondition
+ * Which RexNode should be a condition in the bottom calc
+ * @param topCalcCondition
+ * Which RexNode should be a condition in the top calc
+ * @param topCalcProjects
+ * Which RexNodes should be projections in the top calc
+ */
+class SplitComponents(
+ var bottomCalcCondition: Option[RexNode],
+ var topCalcCondition: Option[RexNode],
+ var topCalcProjects: Seq[RexNode]) {}
+
+/**
+ * Rule that splits [[FlinkLogicalCalc]]s which contain Remote functions in the condition into
+ * multiple [[FlinkLogicalCalc]]s. After this rule is applied, there will be no Remote functions in
* the condition of the [[FlinkLogicalCalc]]s.
*/
-object PythonCalcSplitConditionRule
- extends PythonCalcSplitRuleBase("PythonCalcSplitConditionRule") {
+class RemoteCalcSplitConditionRule(callFinder: RemoteCalcCallFinder)
+ extends RemoteCalcSplitRuleBase("RemoteCalcSplitConditionRule", callFinder) {
override def matches(call: RelOptRuleCall): Boolean = {
val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
- // matches if it contains Python functions in condition
+ // matches if it contains Remote functions in condition
Option(calc.getProgram.getCondition)
.map(calc.getProgram.expandLocalRef)
- .exists(containsPythonCall(_))
+ .exists(callFinder.containsRemoteCall)
}
- override def needConvert(program: RexProgram, node: RexNode): Boolean = isPythonCall(node)
-
- override def split(
+ override def needConvert(
program: RexProgram,
- splitter: ScalarFunctionSplitter): (Option[RexNode], Option[RexNode], Seq[RexNode]) = {
- (
+ node: RexNode,
+ matchState: Option[Nothing]): Boolean =
+ callFinder.isRemoteCall(node)
+
+ override def split(program: RexProgram, splitter: ScalarFunctionSplitter): SplitComponents = {
+ new SplitComponents(
None,
Option(program.getCondition).map(program.expandLocalRef(_).accept(splitter)),
program.getProjectList.map(program.expandLocalRef))
}
}
-abstract class PythonCalcSplitProjectionRuleBase(description: String)
- extends PythonCalcSplitRuleBase(description) {
+abstract class RemoteCalcSplitProjectionRuleBase[T](
+ description: String,
+ callFinder: RemoteCalcCallFinder)
+ extends RemoteCalcSplitRuleBase[T](description, callFinder) {
- override def split(
- program: RexProgram,
- splitter: ScalarFunctionSplitter): (Option[RexNode], Option[RexNode], Seq[RexNode]) = {
- (
+ override def split(program: RexProgram, splitter: ScalarFunctionSplitter): SplitComponents = {
+ new SplitComponents(
Option(program.getCondition).map(program.expandLocalRef),
None,
program.getProjectList.map(program.expandLocalRef(_).accept(splitter)))
}
}
-abstract class PythonCalcSplitRexFieldRuleBase(description: String)
- extends PythonCalcSplitRuleBase(description) {
+abstract class RemoteCalcSplitRexFieldRuleBase(
+ description: String,
+ callFinder: RemoteCalcCallFinder)
+ extends RemoteCalcSplitRuleBase(description, callFinder) {
- override def needConvert(program: RexProgram, node: RexNode): Boolean = {
+ override def needConvert(
+ program: RexProgram,
+ node: RexNode,
+ matchState: Option[Nothing]): Boolean = {
node match {
case x: RexFieldAccess =>
x.getReferenceExpr match {
- case y: RexLocalRef if containsPythonCall(program.expandLocalRef(y)) => true
+ case y: RexLocalRef if callFinder.containsRemoteCall(program.expandLocalRef(y)) => true
case _ => false
}
case _ => false
}
}
- protected def containsFieldAccessAfterPythonCall(node: RexNode): Boolean = {
+ protected def containsFieldAccessAfterRemoteCall(node: RexNode): Boolean = {
node match {
- case call: RexCall => call.getOperands.exists(containsFieldAccessAfterPythonCall)
- case x: RexFieldAccess => containsPythonCall(x.getReferenceExpr)
+ case call: RexCall => call.getOperands.exists(containsFieldAccessAfterRemoteCall)
+ case x: RexFieldAccess => callFinder.containsRemoteCall(x.getReferenceExpr)
case _ => false
}
}
}
/**
- * Rule that splits the RexField with the input of Python function contained in the projection of
+ * Rule that splits the RexField with the input of Remote function contained in the projection of
* [[FlinkLogicalCalc]]s.
*/
-object PythonCalcSplitProjectionRexFieldRule
- extends PythonCalcSplitRexFieldRuleBase("PythonCalcSplitProjectionRexFieldRule") {
+class RemoteCalcSplitProjectionRexFieldRule(callFinder: RemoteCalcCallFinder)
+ extends RemoteCalcSplitRexFieldRuleBase("RemoteCalcSplitProjectionRexFieldRule", callFinder) {
override def matches(call: RelOptRuleCall): Boolean = {
val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
val projects = calc.getProgram.getProjectList.map(calc.getProgram.expandLocalRef)
- projects.exists(containsFieldAccessAfterPythonCall)
+ projects.exists(containsFieldAccessAfterRemoteCall)
}
- override def split(
- program: RexProgram,
- splitter: ScalarFunctionSplitter): (Option[RexNode], Option[RexNode], Seq[RexNode]) = {
- (
+ override def split(program: RexProgram, splitter: ScalarFunctionSplitter): SplitComponents = {
+ new SplitComponents(
Option(program.getCondition).map(program.expandLocalRef),
None,
program.getProjectList.map(_.accept(splitter)))
@@ -220,24 +259,22 @@ object PythonCalcSplitProjectionRexFieldRule
}
/**
- * Rule that splits the RexField with the input of Python function contained in the condition of
+ * Rule that splits the RexField with the input of Remote function contained in the condition of
* [[FlinkLogicalCalc]]s.
*/
-object PythonCalcSplitConditionRexFieldRule
- extends PythonCalcSplitRexFieldRuleBase("PythonCalcSplitConditionRexFieldRule") {
+class RemoteCalcSplitConditionRexFieldRule(callFinder: RemoteCalcCallFinder)
+ extends RemoteCalcSplitRexFieldRuleBase("RemoteCalcSplitConditionRexFieldRule", callFinder) {
override def matches(call: RelOptRuleCall): Boolean = {
val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
Option(calc.getProgram.getCondition)
.map(calc.getProgram.expandLocalRef)
- .exists(containsFieldAccessAfterPythonCall)
+ .exists(containsFieldAccessAfterRemoteCall)
}
- override def split(
- program: RexProgram,
- splitter: ScalarFunctionSplitter): (Option[RexNode], Option[RexNode], Seq[RexNode]) = {
- (
+ override def split(program: RexProgram, splitter: ScalarFunctionSplitter): SplitComponents = {
+ new SplitComponents(
None,
Option(program.getCondition).map(_.accept(splitter)),
program.getProjectList.map(_.accept(splitter)))
@@ -245,74 +282,54 @@ object PythonCalcSplitConditionRexFieldRule
}
/**
- * Rule that splits [[FlinkLogicalCalc]]s which contain both Java functions and Python functions in
+ * Rule that splits [[FlinkLogicalCalc]]s which contain both Java functions and Remote functions in
* the projection into multiple [[FlinkLogicalCalc]]s. After this rule is applied, it will only
- * contain Python functions or Java functions in the projection of each [[FlinkLogicalCalc]].
- */
-object PythonCalcSplitProjectionRule
- extends PythonCalcSplitProjectionRuleBase("PythonCalcSplitProjectionRule") {
-
- override def matches(call: RelOptRuleCall): Boolean = {
- val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
- val projects = calc.getProgram.getProjectList.map(calc.getProgram.expandLocalRef)
-
- // matches if it contains both Python functions and Java functions in the projection
- projects.exists(containsPythonCall(_)) && projects.exists(containsNonPythonCall)
- }
-
- override def needConvert(program: RexProgram, node: RexNode): Boolean = {
- program.getProjectList.map(program.expandLocalRef).exists(isNonPythonCall) == isPythonCall(node)
- }
-}
-
-/**
- * Rule that splits [[FlinkLogicalCalc]]s which contain both general Python functions and pandas
- * Python functions in the projection into multiple [[FlinkLogicalCalc]]s. After this rule is
- * applied, it will only contain general Python functions or pandas Python functions in the
- * projection of each [[FlinkLogicalCalc]].
+ * contain Remote functions or Java functions in the projection of each [[FlinkLogicalCalc]].
*/
-object PythonCalcSplitPandasInProjectionRule
- extends PythonCalcSplitProjectionRuleBase("PythonCalcSplitPandasInProjectionRule") {
+class RemoteCalcSplitProjectionRule(callFinder: RemoteCalcCallFinder)
+ extends RemoteCalcSplitProjectionRuleBase("RemoteCalcSplitProjectionRule", callFinder) {
override def matches(call: RelOptRuleCall): Boolean = {
val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
val projects = calc.getProgram.getProjectList.map(calc.getProgram.expandLocalRef)
- // matches if it contains both general Python functions and
- // pandas Python functions in the projection
- projects.exists(containsPythonCall(_, PythonFunctionKind.GENERAL)) &&
- projects.exists(containsPythonCall(_, PythonFunctionKind.PANDAS))
+ // matches if it contains both Remote functions and Java functions in the projection
+ projects.exists(callFinder.containsRemoteCall) && projects.exists(
+ callFinder.containsNonRemoteCall)
}
- override def needConvert(program: RexProgram, node: RexNode): Boolean = {
+ override def needConvert(
+ program: RexProgram,
+ node: RexNode,
+ matchState: Option[Nothing]): Boolean = {
program.getProjectList
.map(program.expandLocalRef)
- .exists(isPythonCall(_, PythonFunctionKind.GENERAL)) == isPythonCall(
- node,
- PythonFunctionKind.PANDAS)
+ .exists(callFinder.isNonRemoteCall) == callFinder.isRemoteCall(node)
}
}
/**
- * Rule that expands the RexFieldAccess inputs of Python functions contained in the projection of
+ * Rule that expands the RexFieldAccess inputs of Remote functions contained in the projection of
* [[FlinkLogicalCalc]]s.
*/
-object PythonCalcExpandProjectRule extends PythonCalcSplitRuleBase("PythonCalcExpandProjectRule") {
+class RemoteCalcExpandProjectRule(callFinder: RemoteCalcCallFinder)
+ extends RemoteCalcSplitRuleBase("RemoteCalcExpandProjectRule", callFinder) {
override def matches(call: RelOptRuleCall): Boolean = {
val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
val projects = calc.getProgram.getProjectList.map(calc.getProgram.expandLocalRef)
- projects.exists(containsPythonCall(_)) && projects.exists(containsFieldAccessInputs)
+ projects.exists(callFinder.containsRemoteCall) && projects.exists(containsFieldAccessInputs)
}
- override def needConvert(program: RexProgram, node: RexNode): Boolean =
+ override def needConvert(
+ program: RexProgram,
+ node: RexNode,
+ matchState: Option[Nothing]): Boolean =
node.isInstanceOf[RexFieldAccess]
- override def split(
- program: RexProgram,
- splitter: ScalarFunctionSplitter): (Option[RexNode], Option[RexNode], Seq[RexNode]) = {
- (
+ override def split(program: RexProgram, splitter: ScalarFunctionSplitter): SplitComponents = {
+ new SplitComponents(
Option(program.getCondition).map(program.expandLocalRef),
None,
program.getProjectList.map(program.expandLocalRef(_).accept(splitter)))
@@ -329,9 +346,10 @@ object PythonCalcExpandProjectRule extends PythonCalcSplitRuleBase("PythonCalcEx
/**
* Rule that pushes the condition of [[FlinkLogicalCalc]]s before it for the [[FlinkLogicalCalc]]s
- * which contain Python functions in the projection.
+ * which contain Remote functions in the projection.
*/
-object PythonCalcPushConditionRule extends PythonCalcSplitRuleBase("PythonCalcPushConditionRule") {
+class RemoteCalcPushConditionRule(callFinder: RemoteCalcCallFinder)
+ extends RemoteCalcSplitRuleBase("RemoteCalcPushConditionRule", callFinder) {
override def matches(call: RelOptRuleCall): Boolean = {
val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
@@ -339,16 +357,18 @@ object PythonCalcPushConditionRule extends PythonCalcSplitRuleBase("PythonCalcPu
// matches if all the following conditions hold true:
// 1) the condition is not null
- // 2) it contains Python functions in the projection
- calc.getProgram.getCondition != null && projects.exists(containsPythonCall(_))
+ // 2) it contains Remote functions in the projection
+ calc.getProgram.getCondition != null && projects.exists(callFinder.containsRemoteCall)
}
- override def needConvert(program: RexProgram, node: RexNode): Boolean = isNonPythonCall(node)
-
- override def split(
+ override def needConvert(
program: RexProgram,
- splitter: ScalarFunctionSplitter): (Option[RexNode], Option[RexNode], Seq[RexNode]) = {
- (
+ node: RexNode,
+ matchState: Option[Nothing]): Boolean =
+ callFinder.isNonRemoteCall(node)
+
+ override def split(program: RexProgram, splitter: ScalarFunctionSplitter): SplitComponents = {
+ new SplitComponents(
Option(program.getCondition).map(program.expandLocalRef),
None,
program.getProjectList.map(program.expandLocalRef))
@@ -357,42 +377,48 @@ object PythonCalcPushConditionRule extends PythonCalcSplitRuleBase("PythonCalcPu
/**
* Rule that ensures that it only contains [[RexInputRef]]s at the beginning of the project list and
- * [[RexCall]]s at the end of the project list for [[FlinkLogicalCalc]]s which contain Python
- * functions in the projection. This rule exists to keep DataStreamPythonCalc as simple as possible
- * and ensures that it only needs to handle the Python function execution.
+ * [[RexCall]]s at the end of the project list for [[FlinkLogicalCalc]]s which contain Remote
+ * functions in the projection. This rule exists to keep implementations as simple as possible and
+ * ensures that it only needs to handle the Remote function execution.
*/
-object PythonCalcRewriteProjectionRule
- extends PythonCalcSplitRuleBase("PythonCalcRewriteProjectionRule") {
+class RemoteCalcRewriteProjectionRule(callFinder: RemoteCalcCallFinder)
+ extends RemoteCalcSplitRuleBase("RemoteCalcRewriteProjectionRule", callFinder) {
override def matches(call: RelOptRuleCall): Boolean = {
val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
val projects = calc.getProgram.getProjectList.map(calc.getProgram.expandLocalRef)
// matches if all the following conditions hold true:
- // 1) it contains Python functions in the projection
+ // 1) it contains Remote functions in the projection
// 2) it contains RexNodes besides RexInputRef and RexCall or
// not all the RexCalls lying at the end of the project list
- projects.exists(containsPythonCall(_)) &&
+ projects.exists(callFinder.containsRemoteCall) &&
(projects.exists(expr => !expr.isInstanceOf[RexCall] && !expr.isInstanceOf[RexInputRef]) ||
projects.indexWhere(_.isInstanceOf[RexCall]) <
projects.lastIndexWhere(_.isInstanceOf[RexInputRef]))
}
- override def needConvert(program: RexProgram, node: RexNode): Boolean = isPythonCall(node)
-
- override def split(
+ override def needConvert(
program: RexProgram,
- splitter: ScalarFunctionSplitter): (Option[RexNode], Option[RexNode], Seq[RexNode]) = {
- (None, None, program.getProjectList.map(program.expandLocalRef(_).accept(splitter)))
+ node: RexNode,
+ matchState: Option[Nothing]): Boolean =
+ callFinder.isRemoteCall(node)
+
+ override def split(program: RexProgram, splitter: ScalarFunctionSplitter): SplitComponents = {
+ new SplitComponents(
+ None,
+ None,
+ program.getProjectList.map(program.expandLocalRef(_).accept(splitter)))
}
}
-private class ScalarFunctionSplitter(
+class ScalarFunctionSplitter(
program: RexProgram,
rexBuilder: RexBuilder,
extractedFunctionOffset: Int,
extractedRexNodes: mutable.ArrayBuffer[RexNode],
- needConvert: Function[RexNode, Boolean])
+ needConvert: Function[RexNode, Boolean],
+ callFinder: RemoteCalcCallFinder)
extends RexDefaultVisitor[RexNode] {
private var fieldsRexCall: Map[Int, Int] = Map[Int, Int]()
@@ -409,7 +435,8 @@ private class ScalarFunctionSplitter(
if (needConvert(fieldAccess)) {
val expr = fieldAccess.getReferenceExpr
expr match {
- case localRef: RexLocalRef if containsPythonCall(program.expandLocalRef(localRef)) =>
+ case localRef: RexLocalRef
+ if callFinder.containsRemoteCall(program.expandLocalRef(localRef)) =>
getExtractedRexFieldAccess(fieldAccess, localRef.getIndex)
case _: RexCorrelVariable =>
val field = fieldAccess.getField
@@ -437,14 +464,14 @@ private class ScalarFunctionSplitter(
}
private def getExtractedRexFieldAccess(node: RexFieldAccess, rexCallIndex: Int): RexNode = {
- val pythonCall: RexCall =
+ val remoteCall: RexCall =
program.expandLocalRef(node.getReferenceExpr.asInstanceOf[RexLocalRef]).asInstanceOf[RexCall]
if (!fieldsRexCall.contains(rexCallIndex)) {
- extractedRexNodes.append(pythonCall)
+ extractedRexNodes.append(remoteCall)
fieldsRexCall += rexCallIndex -> (extractedFunctionOffset + extractedRexNodes.length - 1)
}
rexBuilder.makeFieldAccess(
- new RexInputRef(fieldsRexCall(rexCallIndex), pythonCall.getType),
+ new RexInputRef(fieldsRexCall(rexCallIndex), remoteCall.getType),
node.getField.getIndex)
}
}
@@ -496,19 +523,3 @@ private class ExtractedFunctionInputRewriter(
override def visitNode(rexNode: RexNode): RexNode = rexNode
}
-
-object PythonCalcSplitRule {
-
- /**
- * These rules should be applied sequentially in the order of SPLIT_CONDITION, SPLIT_PROJECT,
- * SPLIT_PANDAS_IN_PROJECT, EXPAND_PROJECT, PUSH_CONDITION and REWRITE_PROJECT.
- */
- val SPLIT_CONDITION: RelOptRule = PythonCalcSplitConditionRule
- val SPLIT_PROJECT: RelOptRule = PythonCalcSplitProjectionRule
- val SPLIT_PANDAS_IN_PROJECT: RelOptRule = PythonCalcSplitPandasInProjectionRule
- val SPLIT_PROJECTION_REX_FIELD: RelOptRule = PythonCalcSplitProjectionRexFieldRule
- val SPLIT_CONDITION_REX_FIELD: RelOptRule = PythonCalcSplitConditionRexFieldRule
- val EXPAND_PROJECT: RelOptRule = PythonCalcExpandProjectRule
- val PUSH_CONDITION: RelOptRule = PythonCalcPushConditionRule
- val REWRITE_PROJECT: RelOptRule = PythonCalcRewriteProjectionRule
-}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalCalcRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalCalcRule.scala
index 1571b23d24b..6b18dbd46a8 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalCalcRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalCalcRule.scala
@@ -20,6 +20,8 @@ package org.apache.flink.table.planner.plan.rules.physical.stream
import org.apache.flink.table.planner.plan.nodes.FlinkConventions
import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalCalc
import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalCalc
+import org.apache.flink.table.planner.plan.utils.AsyncUtil
+import org.apache.flink.table.planner.plan.utils.AsyncUtil.containsAsyncCall
import org.apache.flink.table.planner.plan.utils.PythonUtil.containsPythonCall
import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelTraitSet}
@@ -35,7 +37,8 @@ class StreamPhysicalCalcRule(config: Config) extends ConverterRule(config) {
override def matches(call: RelOptRuleCall): Boolean = {
val calc: FlinkLogicalCalc = call.rel(0)
val program = calc.getProgram
- !program.getExprList.asScala.exists(containsPythonCall(_))
+ !program.getExprList.asScala.exists(containsPythonCall(_)) &&
+ !program.getExprList.asScala.exists(containsAsyncCall)
}
def convert(rel: RelNode): RelNode = {
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/AsyncCodeGeneratorTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/AsyncCodeGeneratorTest.java
new file mode 100644
index 00000000000..79f867c2843
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/AsyncCodeGeneratorTest.java
@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.codegen;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.async.AsyncFunction;
+import org.apache.flink.streaming.api.functions.async.ResultFuture;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.functions.AsyncScalarFunction;
+import org.apache.flink.table.planner.calcite.SqlToRexConverter;
+import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
+import org.apache.flink.table.planner.utils.PlannerMocks;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
+import org.apache.flink.table.runtime.generated.GeneratedFunction;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexNode;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link AsyncCodeGenerator}. */
+public class AsyncCodeGeneratorTest {
+
+ private static final RowType INPUT_TYPE =
+ RowType.of(new IntType(), new BigIntType(), new VarCharType());
+
+ private PlannerMocks plannerMocks;
+ private SqlToRexConverter converter;
+
+ private RelDataType tableRowType;
+
+ @BeforeEach
+ public void before() {
+ plannerMocks = PlannerMocks.create();
+ tableRowType =
+ plannerMocks
+ .getPlannerContext()
+ .getTypeFactory()
+ .buildRelNodeRowType(
+ JavaScalaConversionUtil.toScala(Arrays.asList("f1", "f2", "f3")),
+ JavaScalaConversionUtil.toScala(
+ Arrays.asList(
+ new IntType(),
+ new BigIntType(),
+ new VarCharType())));
+ ShortcutUtils.unwrapContext(plannerMocks.getPlanner().createToRelContext().getCluster());
+ converter =
+ ShortcutUtils.unwrapContext(
+ plannerMocks.getPlanner().createToRelContext().getCluster())
+ .getRexFactory()
+ .createSqlToRexConverter(tableRowType, null);
+
+ plannerMocks
+ .getFunctionCatalog()
+ .registerTemporarySystemFunction("myfunc", new AsyncFunc(), false);
+ plannerMocks
+ .getFunctionCatalog()
+ .registerTemporarySystemFunction("myfunc_error", new AsyncFuncError(), false);
+ }
+
+ @Test
+ public void testStringReturnType() throws Exception {
+ RowData rowData =
+ execute(
+ "myFunc(f1, f2, f3)",
+ RowType.of(new VarCharType()),
+ GenericRowData.of(2, 3L, StringData.fromString("foo")));
+ assertThat(rowData).isEqualTo(GenericRowData.of(StringData.fromString("complete foo 4 6")));
+ }
+
+ @Test
+ public void testTwoReturnTypes_passThroughFirst() throws Exception {
+ RowData rowData =
+ execute(
+ Arrays.asList("f2", "myFunc(f1, f2, f3)"),
+ RowType.of(new VarCharType(), new BigIntType()),
+ GenericRowData.of(2, 3L, StringData.fromString("foo")));
+ assertThat(rowData)
+ .isEqualTo(GenericRowData.of(3L, StringData.fromString("complete foo 4 6")));
+ }
+
+ @Test
+ public void testTwoReturnTypes_passThroughSecond() throws Exception {
+ RowData rowData =
+ execute(
+ Arrays.asList("myFunc(f1, f2, f3)", "f2"),
+ RowType.of(new VarCharType(), new BigIntType()),
+ GenericRowData.of(2, 3L, StringData.fromString("foo")));
+ assertThat(rowData)
+ .isEqualTo(GenericRowData.of(StringData.fromString("complete foo 4 6"), 3L));
+ }
+
+ @Test
+ public void testError() throws Exception {
+ CompletableFuture<Collection<RowData>> future =
+ executeFuture(
+ Arrays.asList("myFunc_error(f1, f2, f3)"),
+ RowType.of(new VarCharType(), new BigIntType()),
+ GenericRowData.of(2, 3L, StringData.fromString("foo")));
+ assertThat(future).isCompletedExceptionally();
+ assertThatThrownBy(future::get).cause().hasMessage("Error!");
+ }
+
+ private RowData execute(String sqlExpression, RowType resultType, RowData input)
+ throws Exception {
+ return execute(Arrays.asList(sqlExpression), resultType, input);
+ }
+
+ private RowData execute(List<String> sqlExpressions, RowType resultType, RowData input)
+ throws Exception {
+ Collection<RowData> result = executeFuture(sqlExpressions, resultType, input).get();
+ assertThat(result).hasSize(1);
+ return result.iterator().next();
+ }
+
+ private CompletableFuture<Collection<RowData>> executeFuture(
+ List<String> sqlExpressions, RowType resultType, RowData input) throws Exception {
+ List<RexNode> nodes =
+ sqlExpressions.stream()
+ .map(sql -> converter.convertToRexNode(sql))
+ .collect(Collectors.toList());
+ GeneratedFunction<AsyncFunction<RowData, RowData>> function =
+ AsyncCodeGenerator.generateFunction(
+ "name",
+ INPUT_TYPE,
+ resultType,
+ nodes,
+ true,
+ new Configuration(),
+ Thread.currentThread().getContextClassLoader());
+ AsyncFunction<RowData, RowData> asyncFunction =
+ function.newInstance(Thread.currentThread().getContextClassLoader());
+ TestResultFuture resultFuture = new TestResultFuture();
+ asyncFunction.asyncInvoke(input, resultFuture);
+ return resultFuture.getResult();
+ }
+
+ /** Test function. */
+ public static final class AsyncFunc extends AsyncScalarFunction {
+ public void eval(CompletableFuture<String> f, Integer i, Long l, String s) {
+ f.complete("complete " + s + " " + (i * i) + " " + (2 * l));
+ }
+ }
+
+ /** Test function. */
+ public static final class AsyncFuncError extends AsyncScalarFunction {
+ public void eval(CompletableFuture<String> f, Integer i, Long l, String s) {
+ f.completeExceptionally(new RuntimeException("Error!"));
+ }
+ }
+
+ /** Test result future. */
+ public static final class TestResultFuture implements ResultFuture<RowData> {
+
+ CompletableFuture<Collection<RowData>> data = new CompletableFuture<>();
+
+ @Override
+ public void complete(Collection<RowData> result) {
+ data.complete(result);
+ }
+
+ @Override
+ public void completeExceptionally(Throwable error) {
+ data.completeExceptionally(error);
+ }
+
+ public CompletableFuture<Collection<RowData>> getResult() {
+ return data;
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/AsyncCalcSplitRuleTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/AsyncCalcSplitRuleTest.java
new file mode 100644
index 00000000000..41fef430ed8
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/AsyncCalcSplitRuleTest.java
@@ -0,0 +1,276 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.logical;
+
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.functions.AsyncScalarFunction;
+import org.apache.flink.table.planner.plan.optimize.program.BatchOptimizeContext;
+import org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram;
+import org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgramBuilder;
+import org.apache.flink.table.planner.plan.optimize.program.HEP_RULES_EXECUTION_TYPE;
+import org.apache.flink.table.planner.plan.rules.FlinkStreamRuleSets;
+import org.apache.flink.table.planner.utils.TableTestBase;
+import org.apache.flink.table.planner.utils.TableTestUtil;
+import org.apache.flink.types.Row;
+
+import org.apache.calcite.plan.hep.HepMatchOrder;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CompletableFuture;
+
+/** Test for {@link AsyncCalcSplitRule}. */
+public class AsyncCalcSplitRuleTest extends TableTestBase {
+
+ private TableTestUtil util = streamTestUtil(TableConfig.getDefault());
+
+ @BeforeEach
+ public void setup() {
+ FlinkChainedProgram programs = new FlinkChainedProgram<BatchOptimizeContext>();
+ programs.addLast(
+ "logical_rewrite",
+ FlinkHepRuleSetProgramBuilder.newBuilder()
+ .setHepRulesExecutionType(HEP_RULES_EXECUTION_TYPE.RULE_SEQUENCE())
+ .setHepMatchOrder(HepMatchOrder.BOTTOM_UP)
+ .add(FlinkStreamRuleSets.LOGICAL_REWRITE())
+ .build());
+
+ TableEnvironment tEnv = util.getTableEnv();
+ tEnv.executeSql(
+ "CREATE TABLE MyTable (\n"
+ + " a int,\n"
+ + " b bigint,\n"
+ + " c string,\n"
+ + " d ARRAY<INT NOT NULL>\n"
+ + ") ;");
+
+ tEnv.executeSql(
+ "CREATE TABLE MyTable2 (\n"
+ + " a2 int,\n"
+ + " b2 bigint,\n"
+ + " c2 string,\n"
+ + " d2 ARRAY<INT NOT NULL>\n"
+ + ") ;");
+
+ util.addTemporarySystemFunction("func1", new Func1());
+ util.addTemporarySystemFunction("func2", new Func2());
+ util.addTemporarySystemFunction("func3", new Func3());
+ util.addTemporarySystemFunction("func4", new Func4());
+ util.addTemporarySystemFunction("func5", new Func5());
+ }
+
+ @Test
+ public void testSingleCall() {
+ String sqlQuery = "SELECT func1(a) FROM MyTable";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testLiteralPlusTableSelect() {
+ String sqlQuery = "SELECT 'foo', func1(a) FROM MyTable";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testFieldPlusTableSelect() {
+ String sqlQuery = "SELECT a, func1(a) from MyTable";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testTwoCalls() {
+ String sqlQuery = "SELECT func1(a), func1(a) from MyTable";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testFourCalls() {
+ String sqlQuery = "SELECT func1(a), func2(a), func1(a), func2(a) from MyTable";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testNestedCalls() {
+ String sqlQuery = "SELECT func1(func1(func1(a))) from MyTable";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testThreeNestedCalls() {
+ String sqlQuery = "SELECT func1(func1(a)), func1(func1(func1(a))), func1(a) from MyTable";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testPassedToOtherUDF() {
+ String sqlQuery = "SELECT Concat(func2(a), 'foo') from MyTable";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testJustCall() {
+ String sqlQuery = "SELECT func1(1)";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testWhereCondition() {
+ String sqlQuery = "SELECT a from MyTable where REGEXP(func2(a), 'string (2|3)')";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testWhereConditionAndProjection() {
+ String sqlQuery = "SELECT func2(a) from MyTable where REGEXP(func2(a), 'val (2|3)')";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testWhereConditionWithInts() {
+ String sqlQuery = "SELECT a from MyTable where func1(a) >= 12";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testAggregate() {
+ String sqlQuery = "SELECT a, func3(count(*)) from MyTable group by a";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testSelectCallWithIntArray() {
+ String sqlQuery = "SELECT func4(d) from MyTable";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testFieldAccessAfter() {
+ String sqlQuery = "SELECT func5(a).f0 from MyTable";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testFieldOperand() {
+ String sqlQuery = "SELECT func1(func5(a).f0) from MyTable";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testInnerJoinWithFuncInOn() {
+ String sqlQuery =
+ "SELECT a from MyTable INNER JOIN MyTable2 ON func2(a) = func2(a2) AND "
+ + "REGEXP(func2(a), 'string (2|4)')";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testInnerJoinWithFuncProjection() {
+ String sqlQuery = "SELECT func1(a) from MyTable INNER JOIN MyTable2 ON a = a2";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testInnerJoinWithFuncInWhere() {
+ String sqlQuery =
+ "SELECT a from MyTable INNER JOIN MyTable2 ON a = a2 "
+ + "WHERE REGEXP(func2(a), 'val (2|3)')";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testLeftJoinWithFuncInOn() {
+ String sqlQuery = "SELECT a, a2 from MyTable LEFT JOIN MyTable2 ON func1(a) = func1(a2)";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testLeftJoinWithFuncInWhere() {
+ String sqlQuery =
+ "SELECT a, a2 from MyTable LEFT JOIN MyTable2 ON a = a2 "
+ + "WHERE REGEXP(func2(a), 'string (2|3)')";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testRightJoinWithFuncInOn() {
+ String sqlQuery = "SELECT a, a2 from MyTable RIGHT JOIN MyTable2 ON func1(a) = func1(a2)";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testProjectCallInSubquery() {
+ String sqlQuery =
+ "SELECT blah FROM (SELECT func2(a) as blah from MyTable) "
+ + "WHERE REGEXP(blah, 'string (2|3)')";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testWhereConditionCallInSubquery() {
+ String sqlQuery =
+ "SELECT blah FROM (select a as blah from MyTable "
+ + "WHERE REGEXP(func2(a), 'string (2|3)'))";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ @Test
+ public void testWhereNotInSubquery() {
+ String sqlQuery = "SELECT func1(a) FROM MyTable where a not in (select a2 from MyTable2)";
+ util.verifyRelPlan(sqlQuery);
+ }
+
+ /** Test function. */
+ public static class Func1 extends AsyncScalarFunction {
+ public void eval(CompletableFuture<Integer> future, Integer param) {
+ future.complete(param + 10);
+ }
+ }
+
+ /** Test function. */
+ public static class Func2 extends AsyncScalarFunction {
+ public void eval(CompletableFuture<String> future, Integer param) {
+ future.complete("string " + param + 10);
+ }
+ }
+
+ /** Test function. */
+ public static class Func3 extends AsyncScalarFunction {
+ public void eval(CompletableFuture<Long> future, Long param) {
+ future.complete(param + 10);
+ }
+ }
+
+ /** Test function. */
+ public static class Func4 extends AsyncScalarFunction {
+ public void eval(CompletableFuture<int[]> future, int[] param) {
+ future.complete(param);
+ }
+ }
+
+ /** Test function. */
+ public static class Func5 extends AsyncScalarFunction {
+
+ @DataTypeHint("ROW<f0 INT, f1 String>")
+ public void eval(CompletableFuture<Row> future, Integer a) {
+ future.complete(Row.of(a + 1, "" + (a * a)));
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/table/AsyncCalcITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/table/AsyncCalcITCase.java
new file mode 100644
index 00000000000..418d686e2b2
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/table/AsyncCalcITCase.java
@@ -0,0 +1,361 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.runtime.stream.table;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.table.functions.AsyncScalarFunction;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.planner.runtime.utils.StreamingTestBase;
+import org.apache.flink.types.Row;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT Case tests for {@link AsyncScalarFunction}. */
+public class AsyncCalcITCase extends StreamingTestBase {
+
+ private TableEnvironment tEnv;
+
+ @BeforeEach
+ public void before() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(2);
+ tEnv = StreamTableEnvironment.create(env, EnvironmentSettings.inStreamingMode());
+ tEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_ASYNC_SCALAR_BUFFER_CAPACITY, 2);
+ tEnv.getConfig()
+ .set(ExecutionConfigOptions.TABLE_EXEC_ASYNC_SCALAR_TIMEOUT, Duration.ofMinutes(1));
+ }
+
+ @Test
+ public void testSimpleTableSelect() {
+ Table t1 = tEnv.fromValues(1, 2, 3).as("f1");
+ tEnv.createTemporaryView("t1", t1);
+ tEnv.createTemporarySystemFunction("func", new AsyncFunc());
+ final List<Row> results = executeSql("select func(f1) from t1");
+ final List<Row> expectedRows =
+ Arrays.asList(Row.of("val 1"), Row.of("val 2"), Row.of("val 3"));
+ assertThat(results).containsSequence(expectedRows);
+ }
+
+ @Test
+ public void testLiteralPlusTableSelect() {
+ Table t1 = tEnv.fromValues(1, 2, 3).as("f1");
+ tEnv.createTemporaryView("t1", t1);
+ tEnv.createTemporarySystemFunction("func", new AsyncFunc());
+ final List<Row> results = executeSql("select 'foo', func(f1) from t1");
+ final List<Row> expectedRows =
+ Arrays.asList(
+ Row.of("foo", "val 1"), Row.of("foo", "val 2"), Row.of("foo", "val 3"));
+ assertThat(results).containsSequence(expectedRows);
+ }
+
+ @Test
+ public void testFieldPlusTableSelect() {
+ Table t1 = tEnv.fromValues(1, 2, 3).as("f1");
+ tEnv.createTemporaryView("t1", t1);
+ tEnv.createTemporarySystemFunction("func", new AsyncFunc());
+ final List<Row> results = executeSql("select f1, func(f1) from t1");
+ final List<Row> expectedRows =
+ Arrays.asList(Row.of(1, "val 1"), Row.of(2, "val 2"), Row.of(3, "val 3"));
+ assertThat(results).containsSequence(expectedRows);
+ }
+
+ @Test
+ public void testTwoCalls() {
+ Table t1 = tEnv.fromValues(1, 2, 3).as("f1");
+ tEnv.createTemporaryView("t1", t1);
+ tEnv.createTemporarySystemFunction("func", new AsyncFunc());
+ final List<Row> results = executeSql("select func(f1), func(f1) from t1");
+ final List<Row> expectedRows =
+ Arrays.asList(
+ Row.of("val 1", "val 1"),
+ Row.of("val 2", "val 2"),
+ Row.of("val 3", "val 3"));
+ assertThat(results).containsSequence(expectedRows);
+ }
+
+ @Test
+ public void testThreeNestedCalls() {
+ Table t1 = tEnv.fromValues(1, 2, 3).as("f1");
+ tEnv.createTemporaryView("t1", t1);
+ tEnv.createTemporarySystemFunction("func", new AsyncFuncAdd10());
+ final List<Row> results =
+ executeSql("select func(func(f1)), func(func(func(f1))), func(f1) from t1");
+ final List<Row> expectedRows =
+ Arrays.asList(Row.of(21, 31, 11), Row.of(22, 32, 12), Row.of(23, 33, 13));
+ assertThat(results).containsSequence(expectedRows);
+ }
+
+ @Test
+ public void testPassedToOtherUDF() {
+ Table t1 = tEnv.fromValues(1, 2, 3).as("f1");
+ tEnv.createTemporaryView("t1", t1);
+ tEnv.createTemporarySystemFunction("func", new AsyncFunc());
+ final List<Row> results = executeSql("select Concat(func(f1), 'foo') from t1");
+ final List<Row> expectedRows =
+ Arrays.asList(Row.of("val 1foo"), Row.of("val 2foo"), Row.of("val 3foo"));
+ assertThat(results).containsSequence(expectedRows);
+ }
+
+ @Test
+ public void testJustCall() {
+ Table t1 = tEnv.fromValues(1, 2, 3).as("f1");
+ tEnv.createTemporaryView("t1", t1);
+ tEnv.createTemporarySystemFunction("func", new AsyncFunc());
+ final List<Row> results = executeSql("select func(1)");
+ final List<Row> expectedRows = Collections.singletonList(Row.of("val 1"));
+ assertThat(results).containsSequence(expectedRows);
+ }
+
+ @Test
+ public void testWhereConditionAndProjection() {
+ Table t1 = tEnv.fromValues(1, 2, 3).as("f1");
+ tEnv.createTemporaryView("t1", t1);
+ tEnv.createTemporarySystemFunction("func", new AsyncFunc());
+ final List<Row> results =
+ executeSql("select func(f1) from t1 where REGEXP(func(f1), 'val (2|3)')");
+ final List<Row> expectedRows = Arrays.asList(Row.of("val 2"), Row.of("val 3"));
+ assertThat(results).containsSequence(expectedRows);
+ }
+
+ @Test
+ public void testFieldAccessAfter() {
+ Table t1 = tEnv.fromValues(2).as("f1");
+ tEnv.createTemporaryView("t1", t1);
+ tEnv.createTemporarySystemFunction("func", new AsyncFuncRow());
+ final List<Row> results = executeSql("select func(f1).f0 from t1");
+ final List<Row> expectedRows = Collections.singletonList(Row.of(3));
+ assertThat(results).containsSequence(expectedRows);
+ }
+
+ @Test
+ public void testFieldOperand() {
+ Table t1 = tEnv.fromValues(2).as("f1");
+ tEnv.createTemporaryView("t1", t1);
+ tEnv.createTemporarySystemFunction("func", new AsyncFuncRow());
+ tEnv.createTemporarySystemFunction("func2", new AsyncFuncAdd10());
+ Table structs = tEnv.sqlQuery("select func(f1) from t1");
+ tEnv.createTemporaryView("t2", structs);
+ final List<Row> results = executeSql("select func2(t2.f0) from t2");
+ final List<Row> expectedRows = Collections.singletonList(Row.of(13));
+ assertThat(results).containsSequence(expectedRows);
+ }
+
+ @Test
+ public void testOverload() {
+ Table t1 = tEnv.fromValues(1).as("f1");
+ tEnv.createTemporaryView("t1", t1);
+ tEnv.createTemporarySystemFunction("func", new AsyncFuncOverload());
+ final List<Row> results = executeSql("select func(f1), func(cast(f1 as String)) from t1");
+ final List<Row> expectedRows =
+ Collections.singletonList(Row.of("int version 1", "string version 1"));
+ assertThat(results).containsSequence(expectedRows);
+ }
+
+ @Test
+ public void testMultiLayerGeneric() {
+ Table t1 = tEnv.fromValues(1).as("f1");
+ tEnv.createTemporaryView("t1", t1);
+ tEnv.createTemporarySystemFunction("func", new LongAsyncFuncGeneric());
+ final List<Row> results = executeSql("select func(f1) from t1");
+ final List<Row> expectedRows = Collections.singletonList(Row.of((Object) new Long[] {11L}));
+ assertThat(results).containsSequence(expectedRows);
+ }
+
+ @Test
+ public void testMultiLayerMoreGeneric() {
+ Table t1 = tEnv.fromValues(1).as("f1");
+ tEnv.createTemporaryView("t1", t1);
+ tEnv.createTemporarySystemFunction("func", new LongAsyncFuncMoreGeneric());
+ final List<Row> results = executeSql("select func(f1) from t1");
+ final List<Row> expectedRows = Collections.singletonList(Row.of((Object) new Long[] {11L}));
+ assertThat(results).containsSequence(expectedRows);
+ }
+
+ @Test
+ public void testFailures() {
+ // If there is a failure after hitting the end of the input, then it doesn't retry. Having
+ // the buffer = 1 triggers the end input only after completion.
+ tEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_ASYNC_SCALAR_BUFFER_CAPACITY, 1);
+ Table t1 = tEnv.fromValues(1).as("f1");
+ tEnv.createTemporaryView("t1", t1);
+ AsyncFuncFail func = new AsyncFuncFail(2);
+ tEnv.createTemporarySystemFunction("func", func);
+ final List<Row> results = executeSql("select func(f1) from t1");
+ final List<Row> expectedRows = Collections.singletonList(Row.of(3));
+ assertThat(results).containsSequence(expectedRows);
+ }
+
+ private List<Row> executeSql(String sql) {
+ TableResult result = tEnv.executeSql(sql);
+ final List<Row> rows = new ArrayList<>();
+ result.collect().forEachRemaining(rows::add);
+ return rows;
+ }
+
+ /** Test function. */
+ public static class AsyncFunc extends AsyncFuncBase {
+
+ private static final long serialVersionUID = 1L;
+
+ public void eval(CompletableFuture<String> future, Integer param) {
+ executor.schedule(() -> future.complete("val " + param), 10, TimeUnit.MILLISECONDS);
+ }
+ }
+
+ /** Test function. */
+ public static class AsyncFuncAdd10 extends AsyncFuncBase {
+
+ private static final long serialVersionUID = 2L;
+
+ public void eval(CompletableFuture<Integer> future, Integer param) {
+ executor.schedule(() -> future.complete(param + 10), 10, TimeUnit.MILLISECONDS);
+ }
+ }
+
+ /** Test function. */
+ public static class AsyncFuncOverload extends AsyncFuncBase {
+
+ private static final long serialVersionUID = 3L;
+
+ public void eval(CompletableFuture<String> future, Integer param) {
+ executor.schedule(
+ () -> future.complete("int version " + param), 10, TimeUnit.MILLISECONDS);
+ }
+
+ public void eval(CompletableFuture<String> future, String param) {
+ executor.schedule(
+ () -> future.complete("string version " + param), 10, TimeUnit.MILLISECONDS);
+ }
+ }
+
+ /** Test function. */
+ public static class AsyncFuncRow extends AsyncScalarFunction {
+
+ @DataTypeHint("ROW<f0 INT, f1 String>")
+ public void eval(CompletableFuture<Row> future, int a) {
+ future.complete(Row.of(a + 1, "" + (a * a)));
+ }
+ }
+
+ /** Test function. */
+ public static class AsyncFuncFail extends AsyncFuncBase implements Serializable {
+
+ private static final long serialVersionUID = 8996145425452974113L;
+
+ private final int numFailures;
+ private final AtomicInteger failures = new AtomicInteger(0);
+
+ public AsyncFuncFail(int numFailures) {
+ this.numFailures = numFailures;
+ }
+
+ public void eval(CompletableFuture<Integer> future, int ignoredA) {
+ if (failures.getAndIncrement() < numFailures) {
+ future.completeExceptionally(new RuntimeException("Error " + failures.get()));
+ return;
+ }
+ future.complete(failures.get());
+ }
+ }
+
+ /** Test function. */
+ public abstract static class AsyncFuncGeneric<T> extends AsyncFuncBase {
+
+ private static final long serialVersionUID = 3L;
+
+ abstract T[] newT(int param);
+
+ public void eval(CompletableFuture<T[]> future, Integer param) {
+ executor.schedule(() -> future.complete(newT(param)), 10, TimeUnit.MILLISECONDS);
+ }
+ }
+
+ /** Test function. */
+ public static class LongAsyncFuncGeneric extends AsyncFuncGeneric<Long> {
+ @Override
+ Long[] newT(int param) {
+ Long[] result = new Long[1];
+ result[0] = 10L + param;
+ return result;
+ }
+ }
+
+ /** Test function. */
+ public abstract static class AsyncFuncMoreGeneric<T> extends AsyncFuncBase {
+
+ private static final long serialVersionUID = 3L;
+
+ abstract void finish(T future, int param);
+
+ public void eval(T future, Integer param) {
+ executor.schedule(() -> finish(future, param), 10, TimeUnit.MILLISECONDS);
+ }
+ }
+
+ /** Test function. */
+ public static class LongAsyncFuncMoreGeneric
+ extends AsyncFuncMoreGeneric<CompletableFuture<Long[]>> {
+ @Override
+ void finish(CompletableFuture<Long[]> future, int param) {
+ Long[] result = new Long[1];
+ result[0] = 10L + param;
+ future.complete(result);
+ }
+ }
+
+ /** Test function. */
+ public static class AsyncFuncBase extends AsyncScalarFunction {
+
+ protected ScheduledExecutorService executor;
+
+ @Override
+ public void open(FunctionContext context) {
+ executor = Executors.newSingleThreadScheduledExecutor();
+ }
+
+ @Override
+ public void close() {
+ if (null != executor && !executor.isShutdown()) {
+ executor.shutdownNow();
+ }
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/AsyncCalcSplitRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/AsyncCalcSplitRuleTest.xml
new file mode 100644
index 00000000000..649b53d3705
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/AsyncCalcSplitRuleTest.xml
@@ -0,0 +1,556 @@
+<?xml version="1.0" ?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to you under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+<Root>
+ <TestCase name="testSingleCall">
+ <Resource name="sql">
+ <![CDATA[SELECT func1(a) FROM MyTable]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[func1($0)])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+AsyncCalc(select=[func1(a) AS EXPR$0])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testLiteralPlusTableSelect">
+ <Resource name="sql">
+ <![CDATA[SELECT 'foo', func1(a) FROM MyTable]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[_UTF-16LE'foo'], EXPR$1=[func1($0)])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=['foo' AS EXPR$0, f0 AS EXPR$1])
++- AsyncCalc(select=[func1(a) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testFieldPlusTableSelect">
+ <Resource name="sql">
+ <![CDATA[SELECT a, func1(a) from MyTable]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], EXPR$1=[func1($0)])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+AsyncCalc(select=[a, func1(a) AS EXPR$1])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testTwoCalls">
+ <Resource name="sql">
+ <![CDATA[SELECT func1(a), func1(a) from MyTable]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[func1($0)], EXPR$1=[func1($0)])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+AsyncCalc(select=[f0 AS EXPR$0, func1(a) AS EXPR$1])
++- AsyncCalc(select=[a, func1(a) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testFourCalls">
+ <Resource name="sql">
+ <![CDATA[SELECT func1(a), func2(a), func1(a), func2(a) from MyTable]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[func1($0)], EXPR$1=[func2($0)], EXPR$2=[func1($0)], EXPR$3=[func2($0)])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+AsyncCalc(select=[f0 AS EXPR$0, f00 AS EXPR$1, f01 AS EXPR$2, func2(a) AS EXPR$3])
++- AsyncCalc(select=[f0, f00, a, func1(a) AS f01])
+ +- AsyncCalc(select=[f0, a, func2(a) AS f00])
+ +- AsyncCalc(select=[a, func1(a) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testNestedCalls">
+ <Resource name="sql">
+ <![CDATA[SELECT func1(func1(func1(a))) from MyTable]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[func1(func1(func1($0)))])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+AsyncCalc(select=[func1(f0) AS EXPR$0])
++- AsyncCalc(select=[func1(f0) AS f0])
+ +- AsyncCalc(select=[func1(a) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testThreeNestedCalls">
+ <Resource name="sql">
+ <![CDATA[SELECT func1(func1(a)), func1(func1(func1(a))), func1(a) from MyTable]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[func1(func1($0))], EXPR$1=[func1(func1(func1($0)))], EXPR$2=[func1($0)])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+AsyncCalc(select=[f0 AS EXPR$0, func1(f1) AS EXPR$1, f2 AS EXPR$2])
++- AsyncCalc(select=[f2, f0, func1(f1) AS f1])
+ +- AsyncCalc(select=[f2, f1, func1(f0) AS f0])
+ +- AsyncCalc(select=[f0, f00 AS f1, func1(a) AS f2])
+ +- AsyncCalc(select=[f0, a, func1(a) AS f00])
+ +- AsyncCalc(select=[a, func1(a) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testPassedToOtherUDF">
+ <Resource name="sql">
+ <![CDATA[SELECT Concat(func2(a), 'foo') from MyTable]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[CONCAT(func2($0), _UTF-16LE'foo')])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[CONCAT(f0, 'foo') AS EXPR$0])
++- AsyncCalc(select=[func2(a) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJustCall">
+ <Resource name="sql">
+ <![CDATA[SELECT func1(1)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[func1(1)])
++- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+AsyncCalc(select=[func1(1) AS EXPR$0])
++- Values(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testWhereCondition">
+ <Resource name="sql">
+ <![CDATA[SELECT a from MyTable where REGEXP(func2(a), 'string (2|3)')]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0])
++- LogicalFilter(condition=[REGEXP(func2($0), _UTF-16LE'string (2|3)')])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[a], where=[REGEXP(f0, 'string (2|3)')])
++- AsyncCalc(select=[a, func2(a) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testWhereConditionAndProjection">
+ <Resource name="sql">
+ <![CDATA[SELECT func2(a) from MyTable where REGEXP(func2(a), 'val (2|3)')]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[func2($0)])
++- LogicalFilter(condition=[REGEXP(func2($0), _UTF-16LE'val (2|3)')])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+AsyncCalc(select=[func2(a) AS EXPR$0])
++- Calc(select=[a], where=[REGEXP(f0, 'val (2|3)')])
+ +- AsyncCalc(select=[a, func2(a) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testWhereConditionWithInts">
+ <Resource name="sql">
+ <![CDATA[SELECT a from MyTable where func1(a) >= 12]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0])
++- LogicalFilter(condition=[>=(func1($0), 12)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[a], where=[>=(f0, 12)])
++- AsyncCalc(select=[a, func1(a) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testAggregate">
+ <Resource name="sql">
+ <![CDATA[SELECT a, func3(count(*)) from MyTable group by a]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], EXPR$1=[func3($1)])
++- LogicalAggregate(group=[{0}], agg#0=[COUNT()])
+ +- LogicalProject(a=[$0])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+AsyncCalc(select=[a, func3($f1) AS EXPR$1])
++- GroupAggregate(groupBy=[a], select=[a, COUNT(*) AS $f1])
+ +- Exchange(distribution=[hash[a]])
+ +- Calc(select=[a])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testSelectCallWithIntArray">
+ <Resource name="sql">
+ <![CDATA[SELECT func4(d) from MyTable]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[func4($3)])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+AsyncCalc(select=[func4(d) AS EXPR$0])
++- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+]]>
+ </Resource>
+ </TestCase>
+
+ <TestCase name="testFieldAccessAfter">
+ <Resource name="sql">
+ <![CDATA[SELECT func5(a).f0 from MyTable]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[func5($0).f0])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[f0.f0 AS EXPR$0])
++- AsyncCalc(select=[func5(a) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+]]>
+ </Resource>
+ </TestCase>
+
+ <TestCase name="testFieldOperand">
+ <Resource name="sql">
+ <![CDATA[SELECT func1(func5(a).f0) from MyTable]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[func1(func5($0).f0)])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+AsyncCalc(select=[func1(f0) AS EXPR$0])
++- Calc(select=[f0.f0 AS f0])
+ +- AsyncCalc(select=[func5(a) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+]]>
+ </Resource>
+ </TestCase>
+
+ <TestCase name="testInnerJoinWithFuncInOn">
+ <Resource name="sql">
+ <![CDATA[SELECT a from MyTable INNER JOIN MyTable2 ON func2(a) = func2(a2) AND REGEXP(func2(a), 'string (2|4)')]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0])
++- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], a2=[$5], b2=[$6], c2=[$7], d2=[$8])
+ +- LogicalJoin(condition=[AND(=($4, $9), REGEXP(func2($0), _UTF-16LE'string (2|4)'))], joinType=[inner])
+ :- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], $f4=[func2($0)])
+ : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+ +- LogicalProject(a2=[$0], b2=[$1], c2=[$2], d2=[$3], $f4=[func2($0)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[a])
++- Join(joinType=[InnerJoin], where=[=($f4, $f40)], select=[a, $f4, $f40], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
+ :- Exchange(distribution=[hash[$f4]])
+ : +- AsyncCalc(select=[a, func2(a) AS $f4])
+ : +- Calc(select=[a], where=[REGEXP(f0, 'string (2|4)')])
+ : +- AsyncCalc(select=[a, func2(a) AS f0])
+ : +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+ +- Exchange(distribution=[hash[$f4]])
+ +- AsyncCalc(select=[func2(a2) AS $f4])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a2, b2, c2, d2])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testInnerJoinWithFuncProjection">
+ <Resource name="sql">
+ <![CDATA[SELECT func1(a) from MyTable INNER JOIN MyTable2 ON a = a2]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[func1($0)])
++- LogicalJoin(condition=[=($0, $4)], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+AsyncCalc(select=[func1(a) AS EXPR$0])
++- Join(joinType=[InnerJoin], where=[=(a, a2)], select=[a, a2], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
+ :- Exchange(distribution=[hash[a]])
+ : +- Calc(select=[a])
+ : +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+ +- Exchange(distribution=[hash[a2]])
+ +- Calc(select=[a2])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a2, b2, c2, d2])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testInnerJoinWithFuncInWhere">
+ <Resource name="sql">
+ <![CDATA[SELECT a from MyTable INNER JOIN MyTable2 ON a = a2 WHERE REGEXP(func2(a), 'val (2|3)')]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0])
++- LogicalFilter(condition=[REGEXP(func2($0), _UTF-16LE'val (2|3)')])
+ +- LogicalJoin(condition=[=($0, $4)], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[a])
++- Join(joinType=[InnerJoin], where=[=(a, a2)], select=[a, a2], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
+ :- Exchange(distribution=[hash[a]])
+ : +- Calc(select=[a], where=[REGEXP(f0, 'val (2|3)')])
+ : +- AsyncCalc(select=[a, func2(a) AS f0])
+ : +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+ +- Exchange(distribution=[hash[a2]])
+ +- Calc(select=[a2], where=[REGEXP(f0, 'val (2|3)')])
+ +- AsyncCalc(select=[a2, func2(a2) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a2, b2, c2, d2])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testLeftJoinWithFuncInOn">
+ <Resource name="sql">
+ <![CDATA[SELECT a, a2 from MyTable LEFT JOIN MyTable2 ON func1(a) = func1(a2)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], a2=[$4])
++- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], a2=[$5], b2=[$6], c2=[$7], d2=[$8])
+ +- LogicalJoin(condition=[=($4, $9)], joinType=[left])
+ :- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], $f4=[func1($0)])
+ : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+ +- LogicalProject(a2=[$0], b2=[$1], c2=[$2], d2=[$3], $f4=[func1($0)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[a, a2])
++- Join(joinType=[LeftOuterJoin], where=[=($f4, $f40)], select=[a, $f4, a2, $f40], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
+ :- Exchange(distribution=[hash[$f4]])
+ : +- AsyncCalc(select=[a, func1(a) AS $f4])
+ : +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+ +- Exchange(distribution=[hash[$f4]])
+ +- AsyncCalc(select=[a2, func1(a2) AS $f4])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a2, b2, c2, d2])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testLeftJoinWithFuncInWhere">
+ <Resource name="sql">
+ <![CDATA[SELECT a, a2 from MyTable LEFT JOIN MyTable2 ON a = a2 WHERE REGEXP(func2(a), 'string (2|3)')]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], a2=[$4])
++- LogicalFilter(condition=[REGEXP(func2($0), _UTF-16LE'string (2|3)')])
+ +- LogicalJoin(condition=[=($0, $4)], joinType=[left])
+ :- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Join(joinType=[LeftOuterJoin], where=[=(a, a2)], select=[a, a2], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
+:- Exchange(distribution=[hash[a]])
+: +- Calc(select=[a], where=[REGEXP(f0, 'string (2|3)')])
+: +- AsyncCalc(select=[a, func2(a) AS f0])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
++- Exchange(distribution=[hash[a2]])
+ +- Calc(select=[a2])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a2, b2, c2, d2])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testRightJoinWithFuncInOn">
+ <Resource name="sql">
+ <![CDATA[SELECT a, a2 from MyTable RIGHT JOIN MyTable2 ON func1(a) = func1(a2)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], a2=[$4])
++- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], a2=[$5], b2=[$6], c2=[$7], d2=[$8])
+ +- LogicalJoin(condition=[=($4, $9)], joinType=[right])
+ :- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], $f4=[func1($0)])
+ : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+ +- LogicalProject(a2=[$0], b2=[$1], c2=[$2], d2=[$3], $f4=[func1($0)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[a, a2])
++- Join(joinType=[RightOuterJoin], where=[=($f4, $f40)], select=[a, $f4, a2, $f40], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
+ :- Exchange(distribution=[hash[$f4]])
+ : +- AsyncCalc(select=[a, func1(a) AS $f4])
+ : +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+ +- Exchange(distribution=[hash[$f4]])
+ +- AsyncCalc(select=[a2, func1(a2) AS $f4])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a2, b2, c2, d2])
+]]>
+ </Resource>
+ </TestCase>
+
+ <TestCase name="testProjectCallInSubquery">
+ <Resource name="sql">
+ <![CDATA[SELECT blah FROM (SELECT func2(a) as blah from MyTable) WHERE REGEXP(blah, 'string (2|3)')]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(blah=[$0])
++- LogicalFilter(condition=[REGEXP($0, _UTF-16LE'string (2|3)')])
+ +- LogicalProject(blah=[func2($0)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+AsyncCalc(select=[func2(a) AS blah])
++- Calc(select=[a], where=[REGEXP(f0, 'string (2|3)')])
+ +- AsyncCalc(select=[a, func2(a) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testWhereConditionCallInSubquery">
+ <Resource name="sql">
+ <![CDATA[SELECT blah FROM (select a as blah from MyTable WHERE REGEXP(func2(a), 'string (2|3)'))]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(blah=[$0])
++- LogicalProject(blah=[$0])
+ +- LogicalFilter(condition=[REGEXP(func2($0), _UTF-16LE'string (2|3)')])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[a], where=[REGEXP(f0, 'string (2|3)')])
++- AsyncCalc(select=[a, func2(a) AS f0])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testWhereNotInSubquery">
+ <Resource name="sql">
+ <![CDATA[SELECT func1(a) FROM MyTable where a not in (select a2 from MyTable2)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[func1($0)])
++- LogicalFilter(condition=[NOT(IN($0, {
+LogicalProject(a2=[$0])
+ LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
+}))])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+AsyncCalc(select=[func1(a) AS EXPR$0])
++- Join(joinType=[LeftAntiJoin], where=[OR(IS NULL(a), IS NULL(a2), =(a, a2))], select=[a], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
+ :- Exchange(distribution=[single])
+ : +- Calc(select=[a])
+ : +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
+ +- Exchange(distribution=[single])
+ +- Calc(select=[a2])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a2, b2, c2, d2])
+]]>
+ </Resource>
+ </TestCase>
+</Root>
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/calc/async/AsyncFunctionRunner.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/calc/async/AsyncFunctionRunner.java
new file mode 100644
index 00000000000..5687572fd9b
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/calc/async/AsyncFunctionRunner.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.calc.async;
+
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.functions.util.FunctionUtils;
+import org.apache.flink.streaming.api.functions.async.AsyncFunction;
+import org.apache.flink.streaming.api.functions.async.ResultFuture;
+import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.GeneratedFunction;
+
+/**
+ * Async function runner for {@link org.apache.flink.table.functions.AsyncScalarFunction}, which
+ * takes the generated function, instantiates it, and then calls its lifecycle methods.
+ */
+public class AsyncFunctionRunner extends RichAsyncFunction<RowData, RowData> {
+
+ private static final long serialVersionUID = -6664660022391632123L;
+
+ private final GeneratedFunction<AsyncFunction<RowData, RowData>> generatedFetcher;
+
+ private transient AsyncFunction<RowData, RowData> fetcher;
+
+ public AsyncFunctionRunner(
+ GeneratedFunction<AsyncFunction<RowData, RowData>> generatedFetcher) {
+ this.generatedFetcher = generatedFetcher;
+ }
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {
+ super.open(openContext);
+ fetcher = generatedFetcher.newInstance(getRuntimeContext().getUserCodeClassLoader());
+ FunctionUtils.setFunctionRuntimeContext(fetcher, getRuntimeContext());
+ FunctionUtils.openFunction(fetcher, openContext);
+ }
+
+ @Override
+ public void asyncInvoke(RowData input, ResultFuture<RowData> resultFuture) {
+ try {
+ fetcher.asyncInvoke(input, resultFuture);
+ } catch (Throwable t) {
+ resultFuture.completeExceptionally(t);
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ FunctionUtils.closeFunction(fetcher);
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/calc/async/DelegatingAsyncResultFuture.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/calc/async/DelegatingAsyncResultFuture.java
new file mode 100644
index 00000000000..2a42a3414bd
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/calc/async/DelegatingAsyncResultFuture.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.calc.async;
+
+import org.apache.flink.streaming.api.functions.async.ResultFuture;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.conversion.DataStructureConverter;
+import org.apache.flink.util.Preconditions;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+/**
+ * Inspired by {@link org.apache.flink.table.runtime.operators.join.lookup.DelegatingResultFuture}
+ * for {@link org.apache.flink.table.functions.AsyncScalarFunction}.
+ */
+public class DelegatingAsyncResultFuture implements BiConsumer<Object, Throwable> {
+
+ private final ResultFuture<Object> delegatedResultFuture;
+ private final List<Object> synchronousResults = new ArrayList<>();
+ private Function<Object, RowData> outputFactory;
+ private CompletableFuture<Object> future;
+ private CompletableFuture<Object> convertedFuture;
+
+ public DelegatingAsyncResultFuture(ResultFuture<Object> delegatedResultFuture) {
+ this.delegatedResultFuture = delegatedResultFuture;
+ }
+
+ public synchronized void addSynchronousResult(Object object) {
+ synchronousResults.add(object);
+ }
+
+ public synchronized Object getSynchronousResult(int index) {
+ return synchronousResults.get(index);
+ }
+
+ public void setOutputFactory(Function<Object, RowData> outputFactory) {
+ this.outputFactory = outputFactory;
+ }
+
+ public CompletableFuture<?> createAsyncFuture(
+ DataStructureConverter<Object, Object> converter) {
+ Preconditions.checkState(future == null);
+ Preconditions.checkState(convertedFuture == null);
+ Preconditions.checkNotNull(outputFactory);
+ future = new CompletableFuture<>();
+ convertedFuture = future.thenApply(converter::toInternal);
+ this.convertedFuture.whenComplete(this);
+ return future;
+ }
+
+ @Override
+ public void accept(Object o, Throwable throwable) {
+ if (throwable != null) {
+ delegatedResultFuture.completeExceptionally(throwable);
+ } else {
+ try {
+ RowData rowData = outputFactory.apply(o);
+ delegatedResultFuture.complete(java.util.Collections.singleton(rowData));
+ } catch (Throwable t) {
+ delegatedResultFuture.completeExceptionally(t);
+ }
+ }
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/calc/async/RetryPredicates.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/calc/async/RetryPredicates.java
new file mode 100644
index 00000000000..1f781e51a9f
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/calc/async/RetryPredicates.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.calc.async;
+
+import org.apache.flink.table.data.RowData;
+
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.function.Predicate;
+
+/** Contains retry predicates used to determine if a result or error should result in a retry. */
+public class RetryPredicates {
+
+ public static final EmptyResponseResultStrategy EMPTY_RESPONSE =
+ new EmptyResponseResultStrategy();
+ public static final Predicate<Throwable> ANY_EXCEPTION = new AnyExceptionStrategy();
+
+ /** Returns true if the response is null or empty. */
+ public static class EmptyResponseResultStrategy
+ implements Predicate<Collection<RowData>>, Serializable {
+ private static final long serialVersionUID = 5065281655787318565L;
+
+ @Override
+ public boolean test(Collection<RowData> c) {
+ return null == c || c.isEmpty();
+ }
+ }
+
+ /** Returns true for any exception. */
+ public static class AnyExceptionStrategy implements Predicate<Throwable>, Serializable {
+
+ private static final long serialVersionUID = -46468466280154192L;
+
+ @Override
+ public boolean test(Throwable throwable) {
+ return true;
+ }
+ }
+}