You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sh...@apache.org on 2025/06/10 12:25:02 UTC
(flink) branch master updated: [FLINK-37920][table] Reuse same util for AsyncCalc, AsyncLookupJoin a… (#26653)
This is an automated email from the ASF dual-hosted git repository.
shengkai 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 b1148f0e2cf [FLINK-37920][table] Reuse same util for AsyncCalc, AsyncLookupJoin a… (#26653)
b1148f0e2cf is described below
commit b1148f0e2cf3671b4e02b725603ab34dbc37a3bc
Author: Shengkai <33...@users.noreply.github.com>
AuthorDate: Tue Jun 10 20:24:55 2025 +0800
[FLINK-37920][table] Reuse same util for AsyncCalc, AsyncLookupJoin a… (#26653)
---
.../functions/inference/LookupCallContext.java | 28 +--
.../plan/nodes/exec/batch/BatchExecLookupJoin.java | 17 +-
.../nodes/exec/common/CommonExecAsyncCalc.java | 7 +-
.../nodes/exec/common/CommonExecLookupJoin.java | 34 ++--
.../plan/nodes/exec/spec/MLPredictSpec.java | 8 +-
.../nodes/exec/stream/StreamExecLookupJoin.java | 25 +--
.../stream/StreamExecMLPredictTableFunction.java | 8 +-
.../physical/stream/StreamPhysicalAsyncCalc.java | 4 +-
.../StreamPhysicalMLPredictTableFunction.java | 12 +-
.../plan/rules/logical/AsyncCalcSplitRule.java | 16 +-
.../stream/StreamPhysicalAsyncCalcRule.java | 4 +-
.../utils/{AsyncUtil.java => AsyncScalarUtil.java} | 69 ++-----
.../planner/plan/utils/FunctionCallUtils.java | 206 +++++++++++++++++++++
.../table/planner/plan/utils/KeySelectorUtil.java | 16 +-
.../table/planner/plan/utils/LookupJoinUtil.java | 204 +-------------------
.../table/planner/plan/utils/MLPredictUtils.java | 59 ++++++
.../table/planner/codegen/ExpressionReducer.scala | 4 +-
.../planner/codegen/LookupJoinCodeGenerator.scala | 14 +-
.../planner/codegen/ProjectionCodeGenerator.scala | 6 +-
.../physical/common/CommonPhysicalLookupJoin.scala | 35 ++--
.../physical/stream/StreamPhysicalCalcRule.scala | 4 +-
.../nodes/exec/serde/AsyncLookupOptionsTest.java | 15 +-
...ySerdeTest.java => FunctionParamSerdeTest.java} | 24 +--
23 files changed, 425 insertions(+), 394 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/inference/LookupCallContext.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/inference/LookupCallContext.java
index 2a56ac670c6..eccd51dfc11 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/inference/LookupCallContext.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/inference/LookupCallContext.java
@@ -21,9 +21,8 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.table.catalog.DataTypeFactory;
import org.apache.flink.table.connector.source.LookupTableSource;
import org.apache.flink.table.functions.UserDefinedFunction;
-import org.apache.flink.table.planner.plan.utils.LookupJoinUtil.ConstantLookupKey;
-import org.apache.flink.table.planner.plan.utils.LookupJoinUtil.FieldRefLookupKey;
-import org.apache.flink.table.planner.plan.utils.LookupJoinUtil.LookupKey;
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils.Constant;
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils.FunctionParam;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.CallContext;
import org.apache.flink.table.types.logical.LogicalType;
@@ -35,6 +34,7 @@ import java.util.List;
import java.util.Optional;
import static org.apache.flink.table.functions.UserDefinedFunctionHelper.generateInlineFunctionName;
+import static org.apache.flink.table.planner.plan.utils.FunctionCallUtils.FieldRef;
import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.getFieldTypes;
import static org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType;
@@ -42,7 +42,7 @@ import static org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDa
@Internal
public class LookupCallContext extends AbstractSqlCallContext {
- private final List<LookupKey> lookupKeys;
+ private final List<FunctionParam> lookupKeys;
private final List<DataType> argumentDataTypes;
@@ -52,7 +52,7 @@ public class LookupCallContext extends AbstractSqlCallContext {
DataTypeFactory dataTypeFactory,
UserDefinedFunction function,
LogicalType inputType,
- List<LookupKey> lookupKeys,
+ List<FunctionParam> lookupKeys,
LogicalType lookupType) {
super(dataTypeFactory, function, generateInlineFunctionName(function), false);
this.lookupKeys = lookupKeys;
@@ -60,12 +60,12 @@ public class LookupCallContext extends AbstractSqlCallContext {
new AbstractList<>() {
@Override
public DataType get(int index) {
- final LookupKey key = getKey(index);
+ final FunctionParam key = getKey(index);
final LogicalType keyType;
- if (key instanceof ConstantLookupKey) {
- keyType = ((ConstantLookupKey) key).sourceType;
- } else if (key instanceof FieldRefLookupKey) {
- keyType = getFieldTypes(inputType).get(((FieldRefLookupKey) key).index);
+ if (key instanceof Constant) {
+ keyType = ((Constant) key).sourceType;
+ } else if (key instanceof FieldRef) {
+ keyType = getFieldTypes(inputType).get(((FieldRef) key).index);
} else {
throw new IllegalArgumentException();
}
@@ -82,12 +82,12 @@ public class LookupCallContext extends AbstractSqlCallContext {
@Override
public boolean isArgumentLiteral(int pos) {
- return getKey(pos) instanceof ConstantLookupKey;
+ return getKey(pos) instanceof Constant;
}
@Override
public boolean isArgumentNull(int pos) {
- final ConstantLookupKey key = (ConstantLookupKey) getKey(pos);
+ final Constant key = (Constant) getKey(pos);
return key.literal.isNull();
}
@@ -97,7 +97,7 @@ public class LookupCallContext extends AbstractSqlCallContext {
return Optional.empty();
}
try {
- final ConstantLookupKey key = (ConstantLookupKey) getKey(pos);
+ final Constant key = (Constant) getKey(pos);
final RexLiteral literal = key.literal;
return Optional.ofNullable(getLiteralValueAs(literal::getValueAs, clazz));
} catch (IllegalArgumentException e) {
@@ -117,7 +117,7 @@ public class LookupCallContext extends AbstractSqlCallContext {
// --------------------------------------------------------------------------------------------
- private LookupKey getKey(int pos) {
+ private FunctionParam getKey(int pos) {
return lookupKeys.get(pos);
}
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecLookupJoin.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecLookupJoin.java
index c761233c1ac..b6b8b7f0c19 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecLookupJoin.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecLookupJoin.java
@@ -33,7 +33,7 @@ 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.common.CommonExecLookupJoin;
import org.apache.flink.table.planner.plan.nodes.exec.spec.TemporalTableSourceSpec;
-import org.apache.flink.table.planner.plan.utils.LookupJoinUtil;
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils;
import org.apache.flink.table.runtime.operators.join.FlinkJoinType;
import org.apache.flink.table.types.logical.RowType;
@@ -71,10 +71,10 @@ public class BatchExecLookupJoin extends CommonExecLookupJoin
@Nullable RexNode preFilterCondition,
@Nullable RexNode remainingJoinCondition,
TemporalTableSourceSpec temporalTableSourceSpec,
- Map<Integer, LookupJoinUtil.LookupKey> lookupKeys,
+ Map<Integer, FunctionCallUtils.FunctionParam> lookupKeys,
@Nullable List<RexNode> projectionOnTemporalTable,
@Nullable RexNode filterOnTemporalTable,
- @Nullable LookupJoinUtil.AsyncLookupOptions asyncLookupOptions,
+ @Nullable FunctionCallUtils.AsyncOptions asyncLookupOptions,
InputProperty inputProperty,
RowType outputType,
String description,
@@ -111,13 +111,14 @@ public class BatchExecLookupJoin extends CommonExecLookupJoin
RexNode remainingJoinCondition,
@JsonProperty(FIELD_NAME_TEMPORAL_TABLE)
TemporalTableSourceSpec temporalTableSourceSpec,
- @JsonProperty(FIELD_NAME_LOOKUP_KEYS) Map<Integer, LookupJoinUtil.LookupKey> lookupKeys,
+ @JsonProperty(FIELD_NAME_LOOKUP_KEYS)
+ Map<Integer, FunctionCallUtils.FunctionParam> lookupKeys,
@JsonProperty(FIELD_NAME_PROJECTION_ON_TEMPORAL_TABLE) @Nullable
List<RexNode> projectionOnTemporalTable,
@JsonProperty(FIELD_NAME_FILTER_ON_TEMPORAL_TABLE) @Nullable
RexNode filterOnTemporalTable,
@JsonProperty(FIELD_NAME_ASYNC_OPTIONS) @Nullable
- LookupJoinUtil.AsyncLookupOptions asyncLookupOptions,
+ FunctionCallUtils.AsyncOptions asyncLookupOptions,
@JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty> inputProperties,
@JsonProperty(FIELD_NAME_OUTPUT_TYPE) RowType outputType,
@JsonProperty(FIELD_NAME_DESCRIPTION) String description,
@@ -158,7 +159,7 @@ public class BatchExecLookupJoin extends CommonExecLookupJoin
RelOptTable temporalTable,
ExecNodeConfig config,
ClassLoader classLoader,
- Map<Integer, LookupJoinUtil.LookupKey> allLookupKeys,
+ Map<Integer, FunctionCallUtils.FunctionParam> allLookupKeys,
TableFunction<?> syncLookupFunction,
RelBuilder relBuilder,
RowType inputRowType,
@@ -176,14 +177,14 @@ public class BatchExecLookupJoin extends CommonExecLookupJoin
RelOptTable temporalTable,
ExecNodeConfig config,
ClassLoader classLoader,
- Map<Integer, LookupJoinUtil.LookupKey> allLookupKeys,
+ Map<Integer, FunctionCallUtils.FunctionParam> allLookupKeys,
AsyncTableFunction<Object> asyncLookupFunction,
RelBuilder relBuilder,
RowType inputRowType,
RowType tableSourceRowType,
RowType resultRowType,
boolean isLeftOuterJoin,
- LookupJoinUtil.AsyncLookupOptions asyncLookupOptions) {
+ FunctionCallUtils.AsyncOptions asyncLookupOptions) {
throw new IllegalStateException(
"Batch mode should not use key-ordered async lookup joins. This is a bug. Please file an issue.");
}
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
index 1851e2ab786..55242d96404 100644
--- 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
@@ -35,7 +35,8 @@ 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.planner.plan.utils.AsyncScalarUtil;
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils;
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;
@@ -152,12 +153,12 @@ public abstract class CommonExecAsyncCalc extends ExecNodeBase<RowData>
config,
classLoader);
AsyncFunctionRunner func = new AsyncFunctionRunner(generatedFunction);
- AsyncUtil.Options options = AsyncUtil.getAsyncOptions(config);
+ FunctionCallUtils.AsyncOptions options = AsyncScalarUtil.getAsyncOptions(config);
return new AsyncWaitOperatorFactory<>(
func,
options.asyncTimeout,
options.asyncBufferCapacity,
options.asyncOutputMode,
- options.asyncRetryStrategy);
+ AsyncScalarUtil.getResultRetryStrategy(config));
}
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java
index dd09d88c4d1..f2d5d47bfe6 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java
@@ -57,6 +57,7 @@ import org.apache.flink.table.planner.plan.nodes.exec.spec.TemporalTableSourceSp
import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil;
import org.apache.flink.table.planner.plan.schema.LegacyTableSourceTable;
import org.apache.flink.table.planner.plan.schema.TableSourceTable;
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils;
import org.apache.flink.table.planner.plan.utils.LookupJoinUtil;
import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
import org.apache.flink.table.planner.utils.ShortcutUtils;
@@ -174,7 +175,7 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> {
* or field from right table.
*/
@JsonProperty(FIELD_NAME_LOOKUP_KEYS)
- private final Map<Integer, LookupJoinUtil.LookupKey> lookupKeys;
+ private final Map<Integer, FunctionCallUtils.FunctionParam> lookupKeys;
@JsonProperty(FIELD_NAME_TEMPORAL_TABLE)
private final TemporalTableSourceSpec temporalTableSourceSpec;
@@ -199,7 +200,7 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> {
@JsonProperty(FIELD_NAME_ASYNC_OPTIONS)
@JsonInclude(JsonInclude.Include.NON_NULL)
- private final @Nullable LookupJoinUtil.AsyncLookupOptions asyncLookupOptions;
+ private final @Nullable FunctionCallUtils.AsyncOptions asyncLookupOptions;
@JsonProperty(FIELD_NAME_RETRY_OPTIONS)
@JsonInclude(JsonInclude.Include.NON_NULL)
@@ -217,10 +218,10 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> {
@Nullable RexNode remainingJoinCondition,
// TODO: refactor this into TableSourceTable, once legacy TableSource is removed
TemporalTableSourceSpec temporalTableSourceSpec,
- Map<Integer, LookupJoinUtil.LookupKey> lookupKeys,
+ Map<Integer, FunctionCallUtils.FunctionParam> lookupKeys,
@Nullable List<RexNode> projectionOnTemporalTable,
@Nullable RexNode filterOnTemporalTable,
- @Nullable LookupJoinUtil.AsyncLookupOptions asyncLookupOptions,
+ @Nullable FunctionCallUtils.AsyncOptions asyncLookupOptions,
@Nullable LookupJoinUtil.RetryLookupOptions retryOptions,
ChangelogMode inputChangelogMode,
List<InputProperty> inputProperties,
@@ -375,21 +376,21 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> {
RelOptTable temporalTable,
ExecNodeConfig config,
ClassLoader classLoader,
- Map<Integer, LookupJoinUtil.LookupKey> allLookupKeys,
+ Map<Integer, FunctionCallUtils.FunctionParam> allLookupKeys,
AsyncTableFunction<Object> asyncLookupFunction,
RelBuilder relBuilder,
RowType inputRowType,
RowType tableSourceRowType,
RowType resultRowType,
boolean isLeftOuterJoin,
- LookupJoinUtil.AsyncLookupOptions asyncLookupOptions);
+ LookupJoinUtil.AsyncOptions asyncLookupOptions);
protected abstract Transformation<RowData> createSyncLookupJoinWithState(
Transformation<RowData> inputTransformation,
RelOptTable temporalTable,
ExecNodeConfig config,
ClassLoader classLoader,
- Map<Integer, LookupJoinUtil.LookupKey> allLookupKeys,
+ Map<Integer, FunctionCallUtils.FunctionParam> allLookupKeys,
TableFunction<?> syncLookupFunction,
RelBuilder relBuilder,
RowType inputRowType,
@@ -400,17 +401,16 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> {
boolean lookupKeyContainsPrimaryKey);
protected void validateLookupKeyType(
- final Map<Integer, LookupJoinUtil.LookupKey> lookupKeys,
+ final Map<Integer, FunctionCallUtils.FunctionParam> lookupKeys,
final RowType inputRowType,
final RowType tableSourceRowType) {
final List<String> imCompatibleConditions = new LinkedList<>();
lookupKeys.entrySet().stream()
- .filter(entry -> entry.getValue() instanceof LookupJoinUtil.FieldRefLookupKey)
+ .filter(entry -> entry.getValue() instanceof FunctionCallUtils.FieldRef)
.forEach(
entry -> {
int rightKey = entry.getKey();
- int leftKey =
- ((LookupJoinUtil.FieldRefLookupKey) entry.getValue()).index;
+ int leftKey = ((FunctionCallUtils.FieldRef) entry.getValue()).index;
LogicalType leftType = inputRowType.getTypeAt(leftKey);
LogicalType rightType = tableSourceRowType.getTypeAt(rightKey);
boolean isCompatible =
@@ -438,20 +438,20 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> {
RelOptTable temporalTable,
ExecNodeConfig config,
ClassLoader classLoader,
- Map<Integer, LookupJoinUtil.LookupKey> allLookupKeys,
+ Map<Integer, FunctionCallUtils.FunctionParam> allLookupKeys,
AsyncTableFunction<Object> asyncLookupFunction,
RelBuilder relBuilder,
RowType inputRowType,
RowType tableSourceRowType,
RowType resultRowType,
boolean isLeftOuterJoin,
- LookupJoinUtil.AsyncLookupOptions asyncLookupOptions,
+ LookupJoinUtil.AsyncOptions asyncLookupOptions,
@Nullable RowDataKeySelector keySelector) {
DataTypeFactory dataTypeFactory =
ShortcutUtils.unwrapContext(relBuilder).getCatalogManager().getDataTypeFactory();
- List<LookupJoinUtil.LookupKey> convertedKeys =
+ List<FunctionCallUtils.FunctionParam> convertedKeys =
Arrays.stream(LookupJoinUtil.getOrderedLookupKeys(allLookupKeys.keySet()))
.mapToObj(allLookupKeys::get)
.collect(Collectors.toList());
@@ -540,7 +540,7 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> {
RelOptTable temporalTable,
ExecNodeConfig config,
ClassLoader classLoader,
- Map<Integer, LookupJoinUtil.LookupKey> allLookupKeys,
+ Map<Integer, FunctionCallUtils.FunctionParam> allLookupKeys,
TableFunction<?> syncLookupFunction,
RelBuilder relBuilder,
RowType inputRowType,
@@ -581,7 +581,7 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> {
RelOptTable temporalTable,
ExecNodeConfig config,
ClassLoader classLoader,
- Map<Integer, LookupJoinUtil.LookupKey> allLookupKeys,
+ Map<Integer, FunctionCallUtils.FunctionParam> allLookupKeys,
TableFunction<?> syncLookupFunction,
RelBuilder relBuilder,
RowType inputRowType,
@@ -593,7 +593,7 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> {
DataTypeFactory dataTypeFactory =
ShortcutUtils.unwrapContext(relBuilder).getCatalogManager().getDataTypeFactory();
- List<LookupJoinUtil.LookupKey> convertedKeys =
+ List<FunctionCallUtils.FunctionParam> convertedKeys =
Arrays.stream(LookupJoinUtil.getOrderedLookupKeys(allLookupKeys.keySet()))
.mapToObj(allLookupKeys::get)
.collect(Collectors.toList());
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/MLPredictSpec.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/MLPredictSpec.java
index 6c44479feea..a88345c47e2 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/MLPredictSpec.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/MLPredictSpec.java
@@ -18,7 +18,7 @@
package org.apache.flink.table.planner.plan.nodes.exec.spec;
-import org.apache.flink.table.planner.plan.utils.LookupJoinUtil;
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils.FunctionParam;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
@@ -33,20 +33,20 @@ public class MLPredictSpec {
public static final String FIELD_NAME_RUNTIME_CONFIG = "runtimeConfig";
@JsonProperty(FIELD_NAME_FEATURES)
- private final List<LookupJoinUtil.LookupKey> features;
+ private final List<FunctionParam> features;
@JsonProperty(FIELD_NAME_RUNTIME_CONFIG)
private final Map<String, String> runtimeConfig;
@JsonCreator
public MLPredictSpec(
- @JsonProperty(FIELD_NAME_FEATURES) List<LookupJoinUtil.LookupKey> features,
+ @JsonProperty(FIELD_NAME_FEATURES) List<FunctionParam> features,
@JsonProperty(FIELD_NAME_RUNTIME_CONFIG) Map<String, String> runtimeConfig) {
this.features = features;
this.runtimeConfig = runtimeConfig;
}
- public List<LookupJoinUtil.LookupKey> getFeatures() {
+ public List<FunctionParam> getFeatures() {
return features;
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecLookupJoin.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecLookupJoin.java
index b55884439ab..6c3caeb5297 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecLookupJoin.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecLookupJoin.java
@@ -42,8 +42,11 @@ import org.apache.flink.table.planner.plan.nodes.exec.StateMetadata;
import org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecLookupJoin;
import org.apache.flink.table.planner.plan.nodes.exec.spec.TemporalTableSourceSpec;
import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil;
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils;
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils.AsyncOptions;
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils.FunctionParam;
import org.apache.flink.table.planner.plan.utils.KeySelectorUtil;
-import org.apache.flink.table.planner.plan.utils.LookupJoinUtil;
+import org.apache.flink.table.planner.plan.utils.LookupJoinUtil.RetryLookupOptions;
import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
import org.apache.flink.table.runtime.operators.join.FlinkJoinType;
import org.apache.flink.table.runtime.operators.join.lookup.KeyedLookupJoinWrapper;
@@ -111,13 +114,13 @@ public class StreamExecLookupJoin extends CommonExecLookupJoin
@Nullable RexNode preFilterCondition,
@Nullable RexNode remainingJoinCondition,
TemporalTableSourceSpec temporalTableSourceSpec,
- Map<Integer, LookupJoinUtil.LookupKey> lookupKeys,
+ Map<Integer, FunctionParam> lookupKeys,
@Nullable List<RexNode> projectionOnTemporalTable,
@Nullable RexNode filterOnTemporalTable,
boolean lookupKeyContainsPrimaryKey,
boolean upsertMaterialize,
- @Nullable LookupJoinUtil.AsyncLookupOptions asyncLookupOptions,
- @Nullable LookupJoinUtil.RetryLookupOptions retryOptions,
+ @Nullable AsyncOptions asyncLookupOptions,
+ @Nullable RetryLookupOptions retryOptions,
ChangelogMode inputChangelogMode,
@Nullable int[] inputUpsertKey,
InputProperty inputProperty,
@@ -162,7 +165,7 @@ public class StreamExecLookupJoin extends CommonExecLookupJoin
RexNode remainingJoinCondition,
@JsonProperty(FIELD_NAME_TEMPORAL_TABLE)
TemporalTableSourceSpec temporalTableSourceSpec,
- @JsonProperty(FIELD_NAME_LOOKUP_KEYS) Map<Integer, LookupJoinUtil.LookupKey> lookupKeys,
+ @JsonProperty(FIELD_NAME_LOOKUP_KEYS) Map<Integer, FunctionParam> lookupKeys,
@JsonProperty(FIELD_NAME_PROJECTION_ON_TEMPORAL_TABLE) @Nullable
List<RexNode> projectionOnTemporalTable,
@JsonProperty(FIELD_NAME_FILTER_ON_TEMPORAL_TABLE) @Nullable
@@ -170,10 +173,8 @@ public class StreamExecLookupJoin extends CommonExecLookupJoin
@JsonProperty(FIELD_NAME_LOOKUP_KEY_CONTAINS_PRIMARY_KEY)
boolean lookupKeyContainsPrimaryKey,
@JsonProperty(FIELD_NAME_REQUIRE_UPSERT_MATERIALIZE) boolean upsertMaterialize,
- @JsonProperty(FIELD_NAME_ASYNC_OPTIONS) @Nullable
- LookupJoinUtil.AsyncLookupOptions asyncLookupOptions,
- @JsonProperty(FIELD_NAME_RETRY_OPTIONS) @Nullable
- LookupJoinUtil.RetryLookupOptions retryOptions,
+ @JsonProperty(FIELD_NAME_ASYNC_OPTIONS) @Nullable AsyncOptions asyncLookupOptions,
+ @JsonProperty(FIELD_NAME_RETRY_OPTIONS) @Nullable RetryLookupOptions retryOptions,
@JsonProperty(FIELD_NAME_INPUT_CHANGELOG_MODE) @Nullable
ChangelogMode inputChangelogMode,
@JsonProperty(FIELD_NAME_INPUT_UPSERT_KEY) @Nullable int[] inputUpsertKey,
@@ -219,14 +220,14 @@ public class StreamExecLookupJoin extends CommonExecLookupJoin
RelOptTable temporalTable,
ExecNodeConfig config,
ClassLoader classLoader,
- Map<Integer, LookupJoinUtil.LookupKey> allLookupKeys,
+ Map<Integer, FunctionCallUtils.FunctionParam> allLookupKeys,
AsyncTableFunction<Object> asyncLookupFunction,
RelBuilder relBuilder,
RowType inputRowType,
RowType tableSourceRowType,
RowType resultRowType,
boolean isLeftOuterJoin,
- LookupJoinUtil.AsyncLookupOptions asyncLookupOptions) {
+ FunctionCallUtils.AsyncOptions asyncLookupOptions) {
RowDataKeySelector keySelector = getKeySelector(classLoader, inputRowType);
Transformation<RowData> partitionedTransform =
@@ -269,7 +270,7 @@ public class StreamExecLookupJoin extends CommonExecLookupJoin
RelOptTable temporalTable,
ExecNodeConfig config,
ClassLoader classLoader,
- Map<Integer, LookupJoinUtil.LookupKey> allLookupKeys,
+ Map<Integer, FunctionParam> allLookupKeys,
TableFunction<?> syncLookupFunction,
RelBuilder relBuilder,
RowType inputRowType,
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecMLPredictTableFunction.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecMLPredictTableFunction.java
index 3b2398a43bf..1c27ad9d252 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecMLPredictTableFunction.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecMLPredictTableFunction.java
@@ -53,7 +53,7 @@ import org.apache.flink.table.planner.plan.nodes.exec.MultipleTransformationTran
import org.apache.flink.table.planner.plan.nodes.exec.spec.MLPredictSpec;
import org.apache.flink.table.planner.plan.nodes.exec.spec.ModelSpec;
import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil;
-import org.apache.flink.table.planner.plan.utils.LookupJoinUtil;
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils;
import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
import org.apache.flink.table.runtime.collector.ListenableCollector;
import org.apache.flink.table.runtime.collector.TableFunctionResultFuture;
@@ -99,13 +99,13 @@ public class StreamExecMLPredictTableFunction extends ExecNodeBase<RowData>
private final ModelSpec modelSpec;
@JsonProperty(FIELD_NAME_ASYNC_OPTIONS)
- private final @Nullable LookupJoinUtil.AsyncLookupOptions asyncOptions;
+ private final @Nullable FunctionCallUtils.AsyncOptions asyncOptions;
public StreamExecMLPredictTableFunction(
ReadableConfig persistedConfig,
MLPredictSpec mlPredictSpec,
ModelSpec modelSpec,
- @Nullable LookupJoinUtil.AsyncLookupOptions asyncOptions,
+ @Nullable FunctionCallUtils.AsyncOptions asyncOptions,
InputProperty inputProperty,
RowType outputType,
String description) {
@@ -129,7 +129,7 @@ public class StreamExecMLPredictTableFunction extends ExecNodeBase<RowData>
@JsonProperty(FIELD_NAME_ML_PREDICT_SPEC) MLPredictSpec mlPredictSpec,
@JsonProperty(FIELD_NAME_MODEL_SPEC) ModelSpec modelSpec,
@JsonProperty(FIELD_NAME_ASYNC_OPTIONS) @Nullable
- LookupJoinUtil.AsyncLookupOptions asyncOptions,
+ FunctionCallUtils.AsyncOptions asyncOptions,
@JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty> inputProperties,
@JsonProperty(FIELD_NAME_OUTPUT_TYPE) RowType outputType,
@JsonProperty(FIELD_NAME_DESCRIPTION) String 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
index fcb1733a016..2a13d3ddf01 100644
--- 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
@@ -22,7 +22,7 @@ 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.flink.table.planner.plan.utils.AsyncScalarUtil;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelTraitSet;
@@ -70,7 +70,7 @@ public class StreamPhysicalAsyncCalc extends StreamPhysicalCalcBase {
throw new IllegalStateException(
"The condition of StreamPhysicalAsyncCalc should be null.");
}
- if (projection.stream().filter(AsyncUtil::containsAsyncCall).count() > 1) {
+ if (projection.stream().filter(AsyncScalarUtil::containsAsyncCall).count() > 1) {
throw new IllegalStateException(
"Only a single async projection is allowed in StreamPhysicalAsyncCalc.");
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalMLPredictTableFunction.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalMLPredictTableFunction.java
index 81b7470702d..9cd246d9a67 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalMLPredictTableFunction.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalMLPredictTableFunction.java
@@ -33,7 +33,9 @@ import org.apache.flink.table.planner.plan.nodes.exec.spec.ModelSpec;
import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecMLPredictTableFunction;
import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalTableFunctionScan;
import org.apache.flink.table.planner.plan.utils.ChangelogPlanUtils;
-import org.apache.flink.table.planner.plan.utils.LookupJoinUtil;
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils;
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils.FunctionParam;
+import org.apache.flink.table.planner.plan.utils.MLPredictUtils;
import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
import org.apache.flink.table.planner.utils.ShortcutUtils;
@@ -122,7 +124,7 @@ public class StreamPhysicalMLPredictTableFunction extends SingleRel implements S
for (int i = 0; i < fieldNames.size(); i++) {
column2Index.put(fieldNames.get(i), i);
}
- List<LookupJoinUtil.LookupKey> features =
+ List<FunctionParam> features =
descriptorCall.getOperands().stream()
.map(
operand -> {
@@ -136,7 +138,7 @@ public class StreamPhysicalMLPredictTableFunction extends SingleRel implements S
"Field %s is not found in input schema: %s.",
fieldName, tableCall.getType()));
}
- return new LookupJoinUtil.FieldRefLookupKey(index);
+ return new FunctionCallUtils.FieldRef(index);
} else {
throw new TableException(
String.format(
@@ -154,10 +156,10 @@ public class StreamPhysicalMLPredictTableFunction extends SingleRel implements S
return modelSpec;
}
- private LookupJoinUtil.AsyncLookupOptions buildAsyncOptions(RexModelCall modelCall) {
+ private FunctionCallUtils.AsyncOptions buildAsyncOptions(RexModelCall modelCall) {
boolean isAsyncEnabled = isAsyncMLPredict(modelCall.getModelProvider());
if (isAsyncEnabled) {
- return LookupJoinUtil.getMergedMLPredictAsyncOptions(
+ return MLPredictUtils.getMergedMLPredictAsyncOptions(
// TODO: extract runtime config
Collections.emptyMap(),
ShortcutUtils.unwrapTableConfig(getCluster()),
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
index cc7c72a909e..f10adb2038f 100644
--- 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
@@ -19,7 +19,7 @@
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.plan.utils.AsyncScalarUtil;
import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
import org.apache.calcite.plan.RelOptRule;
@@ -73,22 +73,22 @@ public class AsyncCalcSplitRule {
public static class AsyncRemoteCalcCallFinder implements RemoteCalcCallFinder {
@Override
public boolean containsRemoteCall(RexNode node) {
- return AsyncUtil.containsAsyncCall(node);
+ return AsyncScalarUtil.containsAsyncCall(node);
}
@Override
public boolean containsNonRemoteCall(RexNode node) {
- return AsyncUtil.containsNonAsyncCall(node);
+ return AsyncScalarUtil.containsNonAsyncCall(node);
}
@Override
public boolean isRemoteCall(RexNode node) {
- return AsyncUtil.isAsyncCall(node);
+ return AsyncScalarUtil.isAsyncCall(node);
}
@Override
public boolean isNonRemoteCall(RexNode node) {
- return AsyncUtil.isNonAsyncCall(node);
+ return AsyncScalarUtil.isNonAsyncCall(node);
}
@Override
@@ -109,13 +109,13 @@ public class AsyncCalcSplitRule {
private static boolean hasNestedCalls(List<RexNode> projects) {
return projects.stream()
- .filter(AsyncUtil::containsAsyncCall)
+ .filter(AsyncScalarUtil::containsAsyncCall)
.filter(expr -> expr instanceof RexCall)
.map(expr -> (RexCall) expr)
.anyMatch(
rexCall ->
rexCall.getOperands().stream()
- .anyMatch(AsyncUtil::containsAsyncCall));
+ .anyMatch(AsyncScalarUtil::containsAsyncCall));
}
/**
@@ -204,7 +204,7 @@ public class AsyncCalcSplitRule {
@Override
public boolean needConvert(RexProgram program, RexNode node, Option<State> matchState) {
- if (AsyncUtil.containsAsyncCall(node) && !matchState.get().foundMatch) {
+ if (AsyncScalarUtil.containsAsyncCall(node) && !matchState.get().foundMatch) {
matchState.get().foundMatch = true;
return true;
}
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
index d12ace81bc4..5f6f00c3f8c 100644
--- 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
@@ -21,7 +21,7 @@ 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.flink.table.planner.plan.utils.AsyncScalarUtil;
import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptRuleCall;
@@ -49,7 +49,7 @@ public class StreamPhysicalAsyncCalcRule extends ConverterRule {
public boolean matches(RelOptRuleCall call) {
FlinkLogicalCalc calc = call.rel(0);
RexProgram program = calc.getProgram();
- return program.getExprList().stream().anyMatch(AsyncUtil::containsAsyncCall);
+ return program.getExprList().stream().anyMatch(AsyncScalarUtil::containsAsyncCall);
}
@Override
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/AsyncScalarUtil.java
similarity index 69%
rename from flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/AsyncUtil.java
rename to flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/AsyncScalarUtil.java
index 61c368ff498..74b9b72309b 100644
--- 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/AsyncScalarUtil.java
@@ -32,13 +32,12 @@ 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 {
+public class AsyncScalarUtil extends FunctionCallUtils {
/**
* Checks whether it contains the specified kind of async function call in the specified node.
@@ -86,66 +85,22 @@ public class AsyncUtil {
* @param config The config from which to fetch the options
* @return Extracted options
*/
- public static AsyncUtil.Options getAsyncOptions(ExecNodeConfig config) {
- return new AsyncUtil.Options(
+ public static AsyncOptions getAsyncOptions(ExecNodeConfig config) {
+ return new AsyncOptions(
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;
- }
+ false,
+ AsyncDataStream.OutputMode.ORDERED);
}
@SuppressWarnings("unchecked")
- private static AsyncRetryStrategy<RowData> getResultRetryStrategy(
- ExecutionConfigOptions.RetryStrategy retryStrategy,
- Duration retryDelay,
- int retryMaxAttempts) {
+ public static AsyncRetryStrategy<RowData> getResultRetryStrategy(ExecNodeConfig config) {
+ ExecutionConfigOptions.RetryStrategy retryStrategy =
+ config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_SCALAR_RETRY_STRATEGY);
+ Duration retryDelay =
+ config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_SCALAR_RETRY_DELAY);
+ int retryMaxAttempts =
+ config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_SCALAR_MAX_ATTEMPTS);
// Only fixed delay is allowed at the moment, so just ignore the config.
if (retryStrategy == ExecutionConfigOptions.RetryStrategy.FIXED_DELAY) {
return new AsyncRetryStrategies.FixedDelayRetryStrategyBuilder<RowData>(
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/FunctionCallUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/FunctionCallUtils.java
new file mode 100644
index 00000000000..e295ef9f735
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/FunctionCallUtils.java
@@ -0,0 +1,206 @@
+/*
+ * 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.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.types.RowKind;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonSubTypes;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeInfo;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
+
+import org.apache.calcite.rex.RexLiteral;
+
+import java.util.Objects;
+
+/** Common utils for function call, e.g. ML_PREDICT and Lookup Join. */
+public abstract class FunctionCallUtils {
+
+ /** A field used as an equal condition when querying content from a dimension table. */
+ @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "type")
+ @JsonSubTypes({
+ @JsonSubTypes.Type(value = Constant.class),
+ @JsonSubTypes.Type(value = FieldRef.class)
+ })
+ public static class FunctionParam {
+ private FunctionParam() {
+ // sealed class
+ }
+ }
+
+ /** A {@link FunctionParam} whose value is constant. */
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ @JsonTypeName("Constant")
+ public static class Constant extends FunctionParam {
+ public static final String FIELD_NAME_SOURCE_TYPE = "sourceType";
+ public static final String FIELD_NAME_LITERAL = "literal";
+
+ @JsonProperty(FIELD_NAME_SOURCE_TYPE)
+ public final LogicalType sourceType;
+
+ @JsonProperty(FIELD_NAME_LITERAL)
+ public final RexLiteral literal;
+
+ @JsonCreator
+ public Constant(
+ @JsonProperty(FIELD_NAME_SOURCE_TYPE) LogicalType sourceType,
+ @JsonProperty(FIELD_NAME_LITERAL) RexLiteral literal) {
+ this.sourceType = sourceType;
+ this.literal = literal;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Constant that = (Constant) o;
+ return Objects.equals(sourceType, that.sourceType)
+ && Objects.equals(literal, that.literal);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(sourceType, literal);
+ }
+ }
+
+ /** A {@link FunctionParam} whose value comes from the left table field. */
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ @JsonTypeName("FieldRef")
+ public static class FieldRef extends FunctionParam {
+ public static final String FIELD_NAME_INDEX = "index";
+
+ @JsonProperty(FIELD_NAME_INDEX)
+ public final int index;
+
+ @JsonCreator
+ public FieldRef(@JsonProperty(FIELD_NAME_INDEX) int index) {
+ this.index = index;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ FieldRef that = (FieldRef) o;
+ return index == that.index;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(index);
+ }
+ }
+
+ /** AsyncLookupOptions includes async related options. */
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ @JsonTypeName("AsyncOptions")
+ public static class AsyncOptions {
+ public static final String FIELD_NAME_CAPACITY = "capacity ";
+ public static final String FIELD_NAME_TIMEOUT = "timeout";
+ public static final String FIELD_NAME_OUTPUT_MODE = "output-mode";
+ public static final String FIELD_NAME_IS_KEY_ORDERED = "is-key-ordered";
+
+ @JsonProperty(FIELD_NAME_CAPACITY)
+ public final int asyncBufferCapacity;
+
+ @JsonProperty(FIELD_NAME_TIMEOUT)
+ public final long asyncTimeout;
+
+ @JsonInclude(JsonInclude.Include.NON_DEFAULT)
+ @JsonProperty(FIELD_NAME_IS_KEY_ORDERED)
+ public final boolean keyOrdered;
+
+ @JsonProperty(FIELD_NAME_OUTPUT_MODE)
+ public final AsyncDataStream.OutputMode asyncOutputMode;
+
+ @JsonCreator
+ public AsyncOptions(
+ @JsonProperty(FIELD_NAME_CAPACITY) int asyncBufferCapacity,
+ @JsonProperty(FIELD_NAME_TIMEOUT) long asyncTimeout,
+ @JsonProperty(FIELD_NAME_IS_KEY_ORDERED) boolean keyOrdered,
+ @JsonProperty(FIELD_NAME_OUTPUT_MODE) AsyncDataStream.OutputMode asyncOutputMode) {
+ this.asyncBufferCapacity = asyncBufferCapacity;
+ this.asyncTimeout = asyncTimeout;
+ this.keyOrdered = keyOrdered;
+ this.asyncOutputMode = asyncOutputMode;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ AsyncOptions that = (AsyncOptions) o;
+ return asyncBufferCapacity == that.asyncBufferCapacity
+ && asyncTimeout == that.asyncTimeout
+ && keyOrdered == that.keyOrdered
+ && asyncOutputMode == that.asyncOutputMode;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(asyncBufferCapacity, asyncTimeout, keyOrdered, asyncOutputMode);
+ }
+
+ @Override
+ public String toString() {
+ return asyncOutputMode
+ + ", "
+ + "KEY_ORDERED: "
+ + keyOrdered
+ + ", "
+ + asyncTimeout
+ + "ms, "
+ + asyncBufferCapacity;
+ }
+ }
+
+ protected static <T> T coalesce(T t1, T t2) {
+ return t1 != null ? t1 : t2;
+ }
+
+ protected static AsyncDataStream.OutputMode convert(
+ ChangelogMode inputChangelogMode,
+ ExecutionConfigOptions.AsyncOutputMode asyncOutputMode) {
+ if (inputChangelogMode.containsOnly(RowKind.INSERT)
+ && asyncOutputMode == ExecutionConfigOptions.AsyncOutputMode.ALLOW_UNORDERED) {
+ return AsyncDataStream.OutputMode.UNORDERED;
+ }
+ return AsyncDataStream.OutputMode.ORDERED;
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/KeySelectorUtil.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/KeySelectorUtil.java
index 04bb48c06b0..96b7b339a95 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/KeySelectorUtil.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/KeySelectorUtil.java
@@ -106,7 +106,7 @@ public class KeySelectorUtil {
*/
public static RowDataKeySelector getLookupKeysSelectorFromLeftTable(
ClassLoader classLoader,
- Map<Integer, LookupJoinUtil.LookupKey> lookupKeysOfRightTable,
+ Map<Integer, FunctionCallUtils.FunctionParam> lookupKeysOfRightTable,
InternalTypeInfo<RowData> leftTableRowType) {
LogicalType[] inputFieldTypes = leftTableRowType.toRowFieldTypes();
int[] lookupKeyIndicesInOrder =
@@ -115,18 +115,18 @@ public class KeySelectorUtil {
int[] inputMapping = new int[lookupKeysOfRightTable.size()];
Arrays.fill(inputMapping, ProjectionCodeGenerator.EMPTY_INPUT_MAPPING_VALUE());
// 2. Generate all lookup keys in order.
- LookupJoinUtil.LookupKey[] orderedLookupKeys =
- new LookupJoinUtil.LookupKey[lookupKeyIndicesInOrder.length];
+ FunctionCallUtils.FunctionParam[] orderedLookupKeys =
+ new FunctionCallUtils.FunctionParam[lookupKeyIndicesInOrder.length];
// 3. Generate the logical types of all lookup keys.
LogicalType[] orderedLookupKeyLogicalTypes = new LogicalType[lookupKeysOfRightTable.size()];
int cnt = 0;
for (int idx : lookupKeyIndicesInOrder) {
- LookupJoinUtil.LookupKey key = lookupKeysOfRightTable.get(idx);
- if (key instanceof LookupJoinUtil.ConstantLookupKey) {
- LogicalType keyType = ((LookupJoinUtil.ConstantLookupKey) key).sourceType;
+ FunctionCallUtils.FunctionParam key = lookupKeysOfRightTable.get(idx);
+ if (key instanceof FunctionCallUtils.Constant) {
+ LogicalType keyType = ((FunctionCallUtils.Constant) key).sourceType;
orderedLookupKeyLogicalTypes[cnt] = keyType;
- } else if (key instanceof LookupJoinUtil.FieldRefLookupKey) {
- int leftIdx = ((LookupJoinUtil.FieldRefLookupKey) key).index;
+ } else if (key instanceof FunctionCallUtils.FieldRef) {
+ int leftIdx = ((FunctionCallUtils.FieldRef) key).index;
inputMapping[cnt] = leftIdx;
orderedLookupKeyLogicalTypes[cnt] = inputFieldTypes[leftIdx];
} else {
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java
index 0c64d1dae5a..09ce49adbc9 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LookupJoinUtil.java
@@ -23,7 +23,6 @@ import org.apache.flink.api.common.io.InputFormat;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.legacy.table.connector.source.SourceFunctionProvider;
-import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.util.retryable.RetryPredicates;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.TableException;
@@ -67,7 +66,6 @@ import org.apache.flink.table.runtime.operators.join.lookup.RetryableLookupFunct
import org.apache.flink.table.runtime.partitioner.RowDataCustomStreamPartitioner;
import org.apache.flink.table.runtime.typeutils.InternalSerializers;
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.types.RowKind;
import org.apache.flink.util.Preconditions;
@@ -75,15 +73,11 @@ import org.apache.flink.util.Preconditions;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonSubTypes;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeInfo;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
import org.apache.calcite.plan.RelOptTable;
import org.apache.calcite.rel.hint.RelHint;
-import org.apache.calcite.rex.RexLiteral;
import javax.annotation.Nullable;
@@ -110,91 +104,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
/** Utilities for lookup joins using {@link LookupTableSource}. */
@Internal
-public final class LookupJoinUtil {
-
- /** A field used as an equal condition when querying content from a dimension table. */
- @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "type")
- @JsonSubTypes({
- @JsonSubTypes.Type(value = ConstantLookupKey.class),
- @JsonSubTypes.Type(value = FieldRefLookupKey.class)
- })
- public static class LookupKey {
- private LookupKey() {
- // sealed class
- }
- }
-
- /** A {@link LookupKey} whose value is constant. */
- @JsonIgnoreProperties(ignoreUnknown = true)
- @JsonTypeName("Constant")
- public static class ConstantLookupKey extends LookupKey {
- public static final String FIELD_NAME_SOURCE_TYPE = "sourceType";
- public static final String FIELD_NAME_LITERAL = "literal";
-
- @JsonProperty(FIELD_NAME_SOURCE_TYPE)
- public final LogicalType sourceType;
-
- @JsonProperty(FIELD_NAME_LITERAL)
- public final RexLiteral literal;
-
- @JsonCreator
- public ConstantLookupKey(
- @JsonProperty(FIELD_NAME_SOURCE_TYPE) LogicalType sourceType,
- @JsonProperty(FIELD_NAME_LITERAL) RexLiteral literal) {
- this.sourceType = sourceType;
- this.literal = literal;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- ConstantLookupKey that = (ConstantLookupKey) o;
- return Objects.equals(sourceType, that.sourceType)
- && Objects.equals(literal, that.literal);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(sourceType, literal);
- }
- }
-
- /** A {@link LookupKey} whose value comes from the left table field. */
- @JsonIgnoreProperties(ignoreUnknown = true)
- @JsonTypeName("FieldRef")
- public static class FieldRefLookupKey extends LookupKey {
- public static final String FIELD_NAME_INDEX = "index";
-
- @JsonProperty(FIELD_NAME_INDEX)
- public final int index;
-
- @JsonCreator
- public FieldRefLookupKey(@JsonProperty(FIELD_NAME_INDEX) int index) {
- this.index = index;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- FieldRefLookupKey that = (FieldRefLookupKey) o;
- return index == that.index;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(index);
- }
- }
+public final class LookupJoinUtil extends FunctionCallUtils {
/** ShuffleLookupOptions includes shuffle related options. */
@JsonIgnoreProperties(ignoreUnknown = true)
@@ -203,73 +113,6 @@ public final class LookupJoinUtil {
public static final String FIELD_NAME_SHUFFLE = "shuffle";
}
- /** AsyncLookupOptions includes async related options. */
- @JsonIgnoreProperties(ignoreUnknown = true)
- @JsonTypeName("AsyncOptions")
- public static class AsyncLookupOptions {
- public static final String FIELD_NAME_CAPACITY = "capacity ";
- public static final String FIELD_NAME_TIMEOUT = "timeout";
- public static final String FIELD_NAME_OUTPUT_MODE = "output-mode";
- public static final String FIELD_NAME_IS_KEY_ORDERED = "is-key-ordered";
-
- @JsonProperty(FIELD_NAME_CAPACITY)
- public final int asyncBufferCapacity;
-
- @JsonProperty(FIELD_NAME_TIMEOUT)
- public final long asyncTimeout;
-
- @JsonInclude(JsonInclude.Include.NON_DEFAULT)
- @JsonProperty(FIELD_NAME_IS_KEY_ORDERED)
- public final boolean keyOrdered;
-
- @JsonProperty(FIELD_NAME_OUTPUT_MODE)
- public final AsyncDataStream.OutputMode asyncOutputMode;
-
- @JsonCreator
- public AsyncLookupOptions(
- @JsonProperty(FIELD_NAME_CAPACITY) int asyncBufferCapacity,
- @JsonProperty(FIELD_NAME_TIMEOUT) long asyncTimeout,
- @JsonProperty(FIELD_NAME_IS_KEY_ORDERED) boolean keyOrdered,
- @JsonProperty(FIELD_NAME_OUTPUT_MODE) AsyncDataStream.OutputMode asyncOutputMode) {
- this.asyncBufferCapacity = asyncBufferCapacity;
- this.asyncTimeout = asyncTimeout;
- this.keyOrdered = keyOrdered;
- this.asyncOutputMode = asyncOutputMode;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- AsyncLookupOptions that = (AsyncLookupOptions) o;
- return asyncBufferCapacity == that.asyncBufferCapacity
- && asyncTimeout == that.asyncTimeout
- && keyOrdered == that.keyOrdered
- && asyncOutputMode == that.asyncOutputMode;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(asyncBufferCapacity, asyncTimeout, keyOrdered, asyncOutputMode);
- }
-
- @Override
- public String toString() {
- return asyncOutputMode
- + ", "
- + "KEY_ORDERED: "
- + keyOrdered
- + ", "
- + asyncTimeout
- + "ms, "
- + asyncBufferCapacity;
- }
- }
-
/** RetryOptions includes retry lookup related options. */
@JsonIgnoreProperties(ignoreUnknown = true)
@JsonTypeName("RetryLookupOptions")
@@ -383,7 +226,7 @@ public final class LookupJoinUtil {
return lookupKeyIndicesInOrder.stream().mapToInt(Integer::intValue).toArray();
}
- public static AsyncLookupOptions getMergedAsyncOptions(
+ public static AsyncOptions getMergedAsyncOptions(
RelHint lookupHint, TableConfig config, ChangelogMode inputChangelogMode) {
Configuration confFromHint;
if (lookupHint == null) {
@@ -402,7 +245,7 @@ public final class LookupJoinUtil {
asyncOutputMode,
config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_KEY_ORDERED));
- return new AsyncLookupOptions(
+ return new AsyncOptions(
coalesce(
confFromHint.get(ASYNC_CAPACITY),
config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_LOOKUP_BUFFER_CAPACITY)),
@@ -414,31 +257,6 @@ public final class LookupJoinUtil {
convert(inputChangelogMode, asyncOutputMode));
}
- public static AsyncLookupOptions getMergedMLPredictAsyncOptions(
- Map<String, String> runtimeConfig,
- TableConfig config,
- ChangelogMode inputChangelogMode) {
- Configuration queryConf = Configuration.fromMap(runtimeConfig);
- ExecutionConfigOptions.AsyncOutputMode asyncOutputMode =
- coalesce(
- queryConf.get(ASYNC_OUTPUT_MODE),
- config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_ML_PREDICT_OUTPUT_MODE));
-
- return new AsyncLookupOptions(
- coalesce(
- queryConf.get(ASYNC_CAPACITY),
- config.get(
- ExecutionConfigOptions
- .TABLE_EXEC_ASYNC_ML_PREDICT_BUFFER_CAPACITY)),
- coalesce(
- queryConf.get(ASYNC_TIMEOUT),
- config.get(
- ExecutionConfigOptions.TABLE_EXEC_ASYNC_ML_PREDICT_TIMEOUT))
- .toMillis(),
- false,
- convert(inputChangelogMode, asyncOutputMode));
- }
-
/**
* This method determines whether async lookup is enabled according to the given lookup keys
* with considering lookup {@link RelHint} and required upsertMaterialize. Note: it will not
@@ -561,20 +379,6 @@ public final class LookupJoinUtil {
return null == async || async;
}
- private static <T> T coalesce(T t1, T t2) {
- return t1 != null ? t1 : t2;
- }
-
- private static AsyncDataStream.OutputMode convert(
- ChangelogMode inputChangelogMode,
- ExecutionConfigOptions.AsyncOutputMode asyncOutputMode) {
- if (inputChangelogMode.containsOnly(RowKind.INSERT)
- && asyncOutputMode == ExecutionConfigOptions.AsyncOutputMode.ALLOW_UNORDERED) {
- return AsyncDataStream.OutputMode.UNORDERED;
- }
- return AsyncDataStream.OutputMode.ORDERED;
- }
-
/**
* Wraps LookupFunction into a RetryableLookupFunctionDelegator to support retry. Note: only
* LookupFunction is supported.
@@ -721,7 +525,7 @@ public final class LookupJoinUtil {
PlannerBase planner,
RelOptTable table,
RowType inputRowType,
- Map<Integer, LookupKey> allLookupKeys,
+ Map<Integer, FunctionParam> allLookupKeys,
Transformation<RowData> inputTransformation,
ChangelogMode inputChangelogMode,
TransformationMetadata metadata) {
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/MLPredictUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/MLPredictUtils.java
new file mode 100644
index 00000000000..fba0cb09814
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/MLPredictUtils.java
@@ -0,0 +1,59 @@
+/*
+ * 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.configuration.Configuration;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.table.connector.ChangelogMode;
+
+import java.util.Map;
+
+import static org.apache.flink.table.api.config.LookupJoinHintOptions.ASYNC_CAPACITY;
+import static org.apache.flink.table.api.config.LookupJoinHintOptions.ASYNC_OUTPUT_MODE;
+import static org.apache.flink.table.api.config.LookupJoinHintOptions.ASYNC_TIMEOUT;
+
+/** Utils for {@code ML_PREDICT}. */
+public class MLPredictUtils extends FunctionCallUtils {
+
+ public static AsyncOptions getMergedMLPredictAsyncOptions(
+ Map<String, String> runtimeConfig,
+ TableConfig config,
+ ChangelogMode inputChangelogMode) {
+ Configuration queryConf = Configuration.fromMap(runtimeConfig);
+ ExecutionConfigOptions.AsyncOutputMode asyncOutputMode =
+ coalesce(
+ queryConf.get(ASYNC_OUTPUT_MODE),
+ config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_ML_PREDICT_OUTPUT_MODE));
+
+ return new AsyncOptions(
+ coalesce(
+ queryConf.get(ASYNC_CAPACITY),
+ config.get(
+ ExecutionConfigOptions
+ .TABLE_EXEC_ASYNC_ML_PREDICT_BUFFER_CAPACITY)),
+ coalesce(
+ queryConf.get(ASYNC_TIMEOUT),
+ config.get(
+ ExecutionConfigOptions.TABLE_EXEC_ASYNC_ML_PREDICT_TIMEOUT))
+ .toMillis(),
+ false,
+ convert(inputChangelogMode, asyncOutputMode));
+ }
+}
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 51b26f50265..940caed315d 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
@@ -27,7 +27,7 @@ import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.codegen.FunctionCodeGenerator.generateFunction
import org.apache.flink.table.planner.codegen.JsonGenerateUtils.isJsonFunctionOperand
import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable.{JSON_ARRAY, JSON_OBJECT}
-import org.apache.flink.table.planner.plan.utils.{AsyncUtil, ConstantFoldingUtil}
+import org.apache.flink.table.planner.plan.utils.{AsyncScalarUtil, ConstantFoldingUtil}
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
@@ -265,7 +265,7 @@ class ExpressionReducer(
// 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) ||
+ if containsPythonCall(e) || AsyncScalarUtil.containsAsyncCall(e) ||
!ConstantFoldingUtil.supportsConstantFolding(e) =>
nonReducibleExprs += e
None
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/LookupJoinCodeGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/LookupJoinCodeGenerator.scala
index 59547130561..3f3ca0d7f8a 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/LookupJoinCodeGenerator.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/LookupJoinCodeGenerator.scala
@@ -34,7 +34,7 @@ import org.apache.flink.table.planner.codegen.calls.BridgingFunctionGenUtil
import org.apache.flink.table.planner.codegen.calls.BridgingFunctionGenUtil.verifyFunctionAwareImplementation
import org.apache.flink.table.planner.delegation.PlannerBase
import org.apache.flink.table.planner.functions.inference.LookupCallContext
-import org.apache.flink.table.planner.plan.utils.LookupJoinUtil.{ConstantLookupKey, FieldRefLookupKey, LookupKey}
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils.{Constant, FieldRef, FunctionParam}
import org.apache.flink.table.planner.plan.utils.RexLiteralUtil
import org.apache.flink.table.planner.utils.JavaScalaConversionUtil.toScala
import org.apache.flink.table.runtime.collector.{ListenableCollector, TableFunctionResultFuture}
@@ -71,7 +71,7 @@ object LookupJoinCodeGenerator {
inputType: LogicalType,
tableSourceType: LogicalType,
returnType: LogicalType,
- lookupKeys: util.List[LookupKey],
+ lookupKeys: util.List[FunctionParam],
syncLookupFunction: TableFunction[_],
functionName: String,
fieldCopy: Boolean): GeneratedFunction[FlatMapFunction[RowData, RowData]] = {
@@ -109,7 +109,7 @@ object LookupJoinCodeGenerator {
inputType: LogicalType,
tableSourceType: LogicalType,
returnType: LogicalType,
- lookupKeys: util.List[LookupKey],
+ lookupKeys: util.List[FunctionParam],
asyncLookupFunction: AsyncTableFunction[_],
functionName: String): GeneratedTableFunctionWithDataType[AsyncFunction[RowData, AnyRef]] = {
@@ -138,7 +138,7 @@ object LookupJoinCodeGenerator {
inputType: LogicalType,
tableSourceType: LogicalType,
returnType: LogicalType,
- lookupKeys: util.List[LookupKey],
+ lookupKeys: util.List[FunctionParam],
lookupFunctionBase: Class[_],
lookupFunction: UserDefinedFunction,
functionName: String,
@@ -195,15 +195,15 @@ object LookupJoinCodeGenerator {
private def prepareOperands(
ctx: CodeGeneratorContext,
inputType: LogicalType,
- lookupKeys: util.List[LookupKey],
+ lookupKeys: util.List[FunctionParam],
fieldCopy: Boolean): Seq[GeneratedExpression] = {
lookupKeys.asScala
.map {
- case constantKey: ConstantLookupKey =>
+ case constantKey: Constant =>
val res = RexLiteralUtil.toFlinkInternalValue(constantKey.literal)
generateLiteral(ctx, res.f0, res.f1)
- case fieldKey: FieldRefLookupKey =>
+ case fieldKey: FieldRef =>
generateInputAccess(
ctx,
inputType,
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProjectionCodeGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProjectionCodeGenerator.scala
index 2d908de9c63..5ff748854c7 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProjectionCodeGenerator.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProjectionCodeGenerator.scala
@@ -26,7 +26,7 @@ import org.apache.flink.table.planner.codegen.GenerateUtils.{generateLiteral, ge
import org.apache.flink.table.planner.codegen.calls.ScalarOperatorGens
import org.apache.flink.table.planner.functions.aggfunctions._
import org.apache.flink.table.planner.plan.utils.{AggregateInfo, RexLiteralUtil}
-import org.apache.flink.table.planner.plan.utils.LookupJoinUtil.{ConstantLookupKey, LookupKey}
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils.{Constant, FunctionParam}
import org.apache.flink.table.runtime.generated.{GeneratedProjection, Projection}
import org.apache.flink.table.types.logical.{BigIntType, LogicalType, RowType}
@@ -371,7 +371,7 @@ object ProjectionCodeGenerator {
* the GeneratedProjection
*/
def generateProjectionForLookupKeysFromLeftTable(
- orderedLookupKeys: Array[LookupKey],
+ orderedLookupKeys: Array[FunctionParam],
ctx: CodeGeneratorContext,
name: String,
inputType: RowType,
@@ -382,7 +382,7 @@ object ProjectionCodeGenerator {
new Array[GeneratedExpression](orderedLookupKeys.length)
for (i <- orderedLookupKeys.indices) {
orderedLookupKeys(i) match {
- case constantLookupKey: ConstantLookupKey =>
+ case constantLookupKey: Constant =>
val res = RexLiteralUtil.toFlinkInternalValue(constantLookupKey.literal)
constantExpressions(i) = generateLiteral(ctx, res.f0, res.f1)
case _ =>
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala
index cd5def1c00a..3e66ea33d47 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala
@@ -26,6 +26,7 @@ import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalR
import org.apache.flink.table.planner.plan.schema.{IntermediateRelTable, LegacyTableSourceTable, TableSourceTable}
import org.apache.flink.table.planner.plan.utils.{ChangelogPlanUtils, ExpressionFormat, InputRefVisitor, JoinTypeUtil, LookupJoinUtil, RelExplainUtil, TemporalJoinUtil}
import org.apache.flink.table.planner.plan.utils.ExpressionFormat.ExpressionFormat
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils.{AsyncOptions, Constant, FieldRef, FunctionParam}
import org.apache.flink.table.planner.plan.utils.LookupJoinUtil._
import org.apache.flink.table.planner.plan.utils.PythonUtil.containsPythonCall
import org.apache.flink.table.planner.plan.utils.RelExplainUtil.preferExpressionFormat
@@ -94,7 +95,7 @@ abstract class CommonPhysicalLookupJoin(
extends SingleRel(cluster, traitSet, inputRel)
with FlinkRelNode {
- val allLookupKeys: Map[Int, LookupKey] = {
+ val allLookupKeys: Map[Int, FunctionParam] = {
// join key pairs from left input field index to temporal table field index
val joinKeyPairs: Array[IntPair] =
TemporalJoinUtil.getTemporalTableJoinKeyPairs(joinInfo, calcOnTemporalTable)
@@ -130,12 +131,12 @@ abstract class CommonPhysicalLookupJoin(
lazy val tableConfig: TableConfig = unwrapTableConfig(this);
- lazy val asyncOptions: Option[AsyncLookupOptions] = if (isAsyncEnabled) {
+ lazy val asyncOptions: Option[AsyncOptions] = if (isAsyncEnabled) {
Option.apply(
LookupJoinUtil.getMergedAsyncOptions(lookupHint.orNull, tableConfig, inputChangelogMode))
} else {
// do not create asyncOptions if async is not enabled
- Option.empty[AsyncLookupOptions]
+ Option.empty[AsyncOptions]
}
override def deriveRowType(): RelDataType = {
@@ -169,9 +170,9 @@ abstract class CommonPhysicalLookupJoin(
}
val lookupKeys = allLookupKeys
.map {
- case (tableField, fieldKey: FieldRefLookupKey) =>
+ case (tableField, fieldKey: FieldRef) =>
s"${tableFieldNames.get(tableField)}=${inputFieldNames(fieldKey.index)}"
- case (tableField, constantKey: ConstantLookupKey) =>
+ case (tableField, constantKey: Constant) =>
s"${tableFieldNames.get(tableField)}=${RelExplainUtil.literalToString(constantKey.literal)}"
}
.mkString(", ")
@@ -226,13 +227,13 @@ abstract class CommonPhysicalLookupJoin(
private def splitJoinCondition(
rexBuilder: RexBuilder,
leftRelDataType: RelDataType,
- leftKeys: List[LookupKey],
+ leftKeys: List[FunctionParam],
joinInfo: JoinInfo): (Option[RexNode], Option[RexNode]) = {
// indexes of left key fields
val leftKeyIndexes =
leftKeys
- .filter(k => k.isInstanceOf[FieldRefLookupKey])
- .map(k => k.asInstanceOf[FieldRefLookupKey].index)
+ .filter(k => k.isInstanceOf[FieldRef])
+ .map(k => k.asInstanceOf[FieldRef].index)
val joinPairs = joinInfo.pairs().asScala.toArray
// right lookup key index of temporal table may be duplicated in joinPairs,
// we should filter the key-pair by checking left key index.
@@ -281,8 +282,8 @@ abstract class CommonPhysicalLookupJoin(
}
/**
- * Analyze potential lookup keys (including [[ConstantLookupKey]] and [[FieldRefLookupKey]]) of
- * the temporal table from the join condition and calc program on the temporal table.
+ * Analyze potential lookup keys (including [[Constant]] and [[FieldRef]]) of the temporal table
+ * from the join condition and calc program on the temporal table.
*
* @param rexBuilder
* the RexBuilder
@@ -296,9 +297,9 @@ abstract class CommonPhysicalLookupJoin(
private def analyzeLookupKeys(
rexBuilder: RexBuilder,
joinKeyPairs: Array[IntPair],
- calcOnTemporalTable: Option[RexProgram]): Map[Int, LookupKey] = {
+ calcOnTemporalTable: Option[RexProgram]): Map[Int, FunctionParam] = {
// field_index_in_table_source => constant_lookup_key
- val constantLookupKeys = new mutable.HashMap[Int, ConstantLookupKey]
+ val constantLookupKeys = new mutable.HashMap[Int, Constant]
// analyze constant lookup keys
if (calcOnTemporalTable.isDefined && null != calcOnTemporalTable.get.getCondition) {
val program = calcOnTemporalTable.get
@@ -307,8 +308,8 @@ abstract class CommonPhysicalLookupJoin(
// presume 'A = 1 AND A = 2' will be reduced to ALWAYS_FALSE
extractConstantFieldsFromEquiCondition(condition, constantLookupKeys)
}
- val fieldRefLookupKeys = joinKeyPairs.map(p => (p.target, new FieldRefLookupKey(p.source)))
- constantLookupKeys.toMap[Int, LookupKey] ++ fieldRefLookupKeys.toMap[Int, LookupKey]
+ val fieldRefLookupKeys = joinKeyPairs.map(p => (p.target, new FieldRef(p.source)))
+ constantLookupKeys.toMap[Int, FunctionParam] ++ fieldRefLookupKeys.toMap[Int, FunctionParam]
}
/** Check if lookup key contains primary key, include constant lookup keys. */
@@ -406,7 +407,7 @@ abstract class CommonPhysicalLookupJoin(
private def extractConstantFieldsFromEquiCondition(
condition: RexNode,
- constantFieldMap: mutable.HashMap[Int, ConstantLookupKey]): Unit = condition match {
+ constantFieldMap: mutable.HashMap[Int, Constant]): Unit = condition match {
case c: RexCall if c.getKind == SqlKind.AND =>
c.getOperands.asScala.foreach(r => extractConstantField(r, constantFieldMap))
case rex: RexNode => extractConstantField(rex, constantFieldMap)
@@ -415,7 +416,7 @@ abstract class CommonPhysicalLookupJoin(
private def extractConstantField(
pred: RexNode,
- constantFieldMap: mutable.HashMap[Int, ConstantLookupKey]): Unit = pred match {
+ constantFieldMap: mutable.HashMap[Int, Constant]): Unit = pred match {
case c: RexCall if c.getKind == SqlKind.EQUALS =>
val left = c.getOperands.get(0)
val right = c.getOperands.get(1)
@@ -425,7 +426,7 @@ abstract class CommonPhysicalLookupJoin(
case _ => return // non-constant condition
}
val dataType = FlinkTypeFactory.toLogicalType(inputRef.getType)
- constantFieldMap.put(inputRef.getIndex, new ConstantLookupKey(dataType, literal))
+ constantFieldMap.put(inputRef.getIndex, new Constant(dataType, literal))
case _ => // ignore
}
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 6b18dbd46a8..e5efb155116 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,10 +20,10 @@ 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.AsyncScalarUtil
import org.apache.flink.table.planner.plan.utils.PythonUtil.containsPythonCall
+import AsyncScalarUtil.containsAsyncCall
import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelTraitSet}
import org.apache.calcite.rel.RelNode
import org.apache.calcite.rel.convert.ConverterRule
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/AsyncLookupOptionsTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/AsyncLookupOptionsTest.java
index 9ea57e07c06..539bf30af6b 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/AsyncLookupOptionsTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/AsyncLookupOptionsTest.java
@@ -25,6 +25,7 @@ import org.apache.flink.table.api.config.LookupJoinHintOptions;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.planner.hint.JoinStrategy;
import org.apache.flink.table.planner.plan.nodes.exec.spec.LookupJoinHintTestUtil;
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils;
import org.apache.flink.table.planner.plan.utils.LookupJoinUtil;
import org.apache.calcite.rel.hint.RelHint;
@@ -43,43 +44,43 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
-/** Tests for {@link LookupJoinUtil.AsyncLookupOptions}. */
+/** Tests for {@link FunctionCallUtils.AsyncOptions}. */
class AsyncLookupOptionsTest {
@Test
void testSerdeAsyncLookupOptions() throws IOException {
- LookupJoinUtil.AsyncLookupOptions asyncLookupOptions =
+ FunctionCallUtils.AsyncOptions asyncLookupOptions =
LookupJoinUtil.getMergedAsyncOptions(
LookupJoinHintTestUtil.completeLookupHint,
TableConfig.getDefault(),
ChangelogMode.insertOnly());
- testJsonRoundTrip(asyncLookupOptions, LookupJoinUtil.AsyncLookupOptions.class);
+ testJsonRoundTrip(asyncLookupOptions, FunctionCallUtils.AsyncOptions.class);
asyncLookupOptions =
LookupJoinUtil.getMergedAsyncOptions(
LookupJoinHintTestUtil.lookupHintWithAsync,
TableConfig.getDefault(),
ChangelogMode.insertOnly());
- testJsonRoundTrip(asyncLookupOptions, LookupJoinUtil.AsyncLookupOptions.class);
+ testJsonRoundTrip(asyncLookupOptions, FunctionCallUtils.AsyncOptions.class);
asyncLookupOptions =
LookupJoinUtil.getMergedAsyncOptions(
LookupJoinHintTestUtil.lookupHintWithRetry,
TableConfig.getDefault(),
ChangelogMode.insertOnly());
- testJsonRoundTrip(asyncLookupOptions, LookupJoinUtil.AsyncLookupOptions.class);
+ testJsonRoundTrip(asyncLookupOptions, FunctionCallUtils.AsyncOptions.class);
asyncLookupOptions =
LookupJoinUtil.getMergedAsyncOptions(
LookupJoinHintTestUtil.lookupHintWithTableOnly,
TableConfig.getDefault(),
ChangelogMode.insertOnly());
- testJsonRoundTrip(asyncLookupOptions, LookupJoinUtil.AsyncLookupOptions.class);
+ testJsonRoundTrip(asyncLookupOptions, FunctionCallUtils.AsyncOptions.class);
}
@Test
void testAsyncLookupOptions() {
- LookupJoinUtil.AsyncLookupOptions asyncLookupOptions =
+ FunctionCallUtils.AsyncOptions asyncLookupOptions =
LookupJoinUtil.getMergedAsyncOptions(
LookupJoinHintTestUtil.completeLookupHint,
TableConfig.getDefault(),
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LookupKeySerdeTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/FunctionParamSerdeTest.java
similarity index 72%
rename from flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LookupKeySerdeTest.java
rename to flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/FunctionParamSerdeTest.java
index 10342c49b25..eec7b71492b 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/LookupKeySerdeTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/FunctionParamSerdeTest.java
@@ -18,7 +18,7 @@
package org.apache.flink.table.planner.plan.nodes.exec.serde;
-import org.apache.flink.table.planner.plan.utils.LookupJoinUtil;
+import org.apache.flink.table.planner.plan.utils.FunctionCallUtils;
import org.apache.flink.table.types.logical.BigIntType;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectReader;
@@ -32,27 +32,27 @@ import java.io.IOException;
import static org.assertj.core.api.Assertions.assertThat;
/** Test LookupKey json ser/de. */
-class LookupKeySerdeTest {
+class FunctionParamSerdeTest {
@Test
- void testLookupKey() throws IOException {
+ void testSerdeFunctionParams() throws IOException {
SerdeContext serdeCtx = JsonSerdeTestUtil.configuredSerdeContext();
ObjectReader objectReader = CompiledPlanSerdeUtil.createJsonObjectReader(serdeCtx);
ObjectWriter objectWriter = CompiledPlanSerdeUtil.createJsonObjectWriter(serdeCtx);
- LookupJoinUtil.LookupKey[] lookupKeys =
- new LookupJoinUtil.LookupKey[] {
- new LookupJoinUtil.ConstantLookupKey(
+ FunctionCallUtils.FunctionParam[] functionParams =
+ new FunctionCallUtils.FunctionParam[] {
+ new FunctionCallUtils.Constant(
new BigIntType(),
new RexBuilder(serdeCtx.getTypeFactory()).makeLiteral("a")),
- new LookupJoinUtil.FieldRefLookupKey(3)
+ new FunctionCallUtils.FieldRef(3)
};
- for (LookupJoinUtil.LookupKey lookupKey : lookupKeys) {
- LookupJoinUtil.LookupKey result =
+ for (FunctionCallUtils.FunctionParam param : functionParams) {
+ FunctionCallUtils.FunctionParam result =
objectReader.readValue(
- objectWriter.writeValueAsString(lookupKey),
- LookupJoinUtil.LookupKey.class);
- assertThat(result).isEqualTo(lookupKey);
+ objectWriter.writeValueAsString(param),
+ FunctionCallUtils.FunctionParam.class);
+ assertThat(result).isEqualTo(param);
}
}
}