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 2b71c2f8965 [FLINK-41065][table] Allow process table functions to
exclude state entries per call
2b71c2f8965 is described below
commit 2b71c2f89651763059a082a1df2c740b66e3e421
Author: Timo Walther <[email protected]>
AuthorDate: Mon Oct 26 09:42:13 2026 +0100
[FLINK-41065][table] Allow process table functions to exclude state entries
per call
This closes #29566.
---
.../table/types/inference/StateTypeStrategy.java | 27 ++++-
.../table/types/inference/TypeInferenceUtil.java | 51 +++++----
.../stream/ProcessTableFunctionSemanticTests.java | 1 +
.../stream/ProcessTableFunctionTestPrograms.java | 23 ++++
.../exec/stream/ProcessTableFunctionTestUtils.java | 76 +++++++++++++
.../functions/ptf/BuiltInProcessTableFunction.java | 119 ++++++++++++++++++---
.../functions/ProcessTableFunctionTestHarness.java | 6 +-
.../ProcessTableFunctionTestHarnessTest.java | 69 ++++++++++++
8 files changed, 331 insertions(+), 41 deletions(-)
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/StateTypeStrategy.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/StateTypeStrategy.java
index 9936c6d4d56..88108906466 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/StateTypeStrategy.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/StateTypeStrategy.java
@@ -19,13 +19,25 @@
package org.apache.flink.table.types.inference;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.functions.ProcessTableFunction;
+import org.apache.flink.table.types.DataType;
import javax.annotation.Nullable;
import java.time.Duration;
import java.util.Optional;
-/** Strategy for inferring a function call's intermediate result data type
(i.e. state entry). */
+/**
+ * Strategy for inferring a function call's intermediate result data type
(i.e. state entry).
+ *
+ * <p>For {@link ProcessTableFunction}, returning {@link Optional#empty()}
from {@link
+ * #inferType(CallContext)} excludes the state entry from a specific call. A
function's {@code
+ * eval()} and {@code onTimer()} methods must not declare a parameter for it.
This allows declaring
+ * a superset of state entries while the strategy decides what a call needs.
+ *
+ * <p>For aggregating functions, returning {@link Optional#empty()} from {@link
+ * #inferType(CallContext)} is not supported as the accumulator must always be
inferable.
+ */
@PublicEvolving
public interface StateTypeStrategy extends TypeStrategy {
@@ -42,6 +54,19 @@ public interface StateTypeStrategy extends TypeStrategy {
return new DefaultStateTypeStrategy(typeStrategy, timeToLive,
broadcast);
}
+ /**
+ * Infers the data type of the state entry from the given function call.
+ *
+ * <p>For {@link ProcessTableFunction}, returning {@link Optional#empty()}
means that the state
+ * entry is not used by this call. It is then excluded from the call's
state, and the function's
+ * {@code eval()} and {@code onTimer()} methods must not declare a
parameter for it.
+ *
+ * <p>For aggregating functions, the accumulator must always be inferable.
Returning {@link
+ * Optional#empty()} fails with a validation exception.
+ */
+ @Override
+ Optional<DataType> inferType(CallContext callContext);
+
/**
* The time-to-live (TTL) duration that automatically cleans up the state
entry.
*
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TypeInferenceUtil.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TypeInferenceUtil.java
index fa3598c8273..a2360dedcfd 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TypeInferenceUtil.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TypeInferenceUtil.java
@@ -23,6 +23,7 @@ import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.DataTypeFactory;
import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.functions.FunctionKind;
import org.apache.flink.table.functions.ModelSemantics;
import org.apache.flink.table.functions.TableSemantics;
import org.apache.flink.table.functions.UserDefinedFunctionHelper;
@@ -36,7 +37,6 @@ import javax.annotation.Nullable;
import java.time.Duration;
import java.util.LinkedHashMap;
import java.util.List;
-import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;
@@ -183,21 +183,20 @@ public final class TypeInferenceUtil {
/**
* Infers {@link StateInfo}s using the given {@link StateTypeStrategy}s.
It assumes that input
* arguments have been adapted before if necessary.
+ *
+ * <p>State entries whose strategy returns {@link Optional#empty()} are
excluded from the
+ * result. This allows a function to declare more state entries than a
specific call needs. Only
+ * process table functions can exclude state entries; for all other
functions every state entry,
+ * such as the accumulator of an aggregating function, must be inferred.
*/
public static LinkedHashMap<String, StateInfo> inferStateInfos(
CallContext callContext, LinkedHashMap<String, StateTypeStrategy>
stateTypeStrategies) {
- return stateTypeStrategies.entrySet().stream()
- .map(
- e ->
- Map.entry(
- e.getKey(),
- inferStateInfo(callContext,
e.getKey(), e.getValue())))
- .collect(
- Collectors.toMap(
- Map.Entry::getKey,
- Map.Entry::getValue,
- (x, y) -> y,
- LinkedHashMap::new));
+ final LinkedHashMap<String, StateInfo> stateInfos = new
LinkedHashMap<>();
+ stateTypeStrategies.forEach(
+ (name, strategy) ->
+ inferStateInfo(callContext, name, strategy)
+ .ifPresent(stateInfo -> stateInfos.put(name,
stateInfo)));
+ return stateInfos;
}
/** Generates a signature of the given {@link FunctionDefinition}. */
@@ -583,24 +582,32 @@ public final class TypeInferenceUtil {
return castCallContext;
}
- private static StateInfo inferStateInfo(
+ private static Optional<StateInfo> inferStateInfo(
CallContext callContext, String name, StateTypeStrategy
stateTypeStrategy) {
final DataType stateType =
stateTypeStrategy.inferType(callContext).orElse(null);
- if (stateType == null || isUnknown(stateType)) {
- final String errorMessage;
+ if (stateType == null) {
+ if (callContext.getFunctionDefinition().getKind() ==
FunctionKind.PROCESS_TABLE) {
+ return Optional.empty();
+ }
if
(name.equals(UserDefinedFunctionHelper.DEFAULT_ACCUMULATOR_NAME)) {
- errorMessage = "Could not infer an accumulator type for the
given arguments.";
- } else {
- errorMessage =
- String.format("Could not infer a data type for state
entry '%s'.", name);
+ throw new ValidationException(
+ "Could not infer an accumulator type for the given
arguments.");
}
- throw new ValidationException(errorMessage);
+ throw new ValidationException(
+ String.format("Could not infer a data type for state entry
'%s'.", name));
+ }
+ if (isUnknown(stateType)) {
+ throw new ValidationException(
+ String.format(
+ "Could not infer a data type for state entry '%s'.
"
+ + "Untyped NULL received.",
+ name));
}
final Duration ttl =
stateTypeStrategy.getTimeToLive(callContext).orElse(null);
final boolean broadcast = stateTypeStrategy.isBroadcast();
- return new StateInfo(stateType, ttl, broadcast);
+ return Optional.of(new StateInfo(stateType, ttl, broadcast));
}
private static boolean isUnknown(DataType dataType) {
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
index 1faf06beb60..44713492267 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
@@ -86,6 +86,7 @@ public class ProcessTableFunctionSemanticTests extends
SemanticTestBase {
ProcessTableFunctionTestPrograms.PROCESS_OPTIONAL_PARTITION_BY_TIME,
ProcessTableFunctionTestPrograms.PROCESS_OPTIONAL_ON_TIME,
ProcessTableFunctionTestPrograms.PROCESS_POJO_STATE_TIME,
+ ProcessTableFunctionTestPrograms.PROCESS_OPTIONAL_STATE,
ProcessTableFunctionTestPrograms.PROCESS_EAGER_AND_VALUE_VIEW_STATE_TIME,
ProcessTableFunctionTestPrograms.PROCESS_CHAINED_TIME,
ProcessTableFunctionTestPrograms.PROCESS_CHAINED_TIME_TABLE_API,
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
index d2512eb0e54..d4ce7d8ab2b 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
@@ -61,6 +61,7 @@ import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctio
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.NotifyStatefulSetsRulesFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.OptionalOnTimeFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.OptionalPartitionOnTimeFunction;
+import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.OptionalStateFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.OrderByFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.PojoArgsFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.PojoCreatingFunction;
@@ -1462,6 +1463,28 @@ public class ProcessTableFunctionTestPrograms {
"INSERT INTO sink SELECT * FROM f(r => TABLE t
PARTITION BY name, on_time => DESCRIPTOR(ts))")
.build();
+ public static final TableTestProgram PROCESS_OPTIONAL_STATE =
+ TableTestProgram.of(
+ "process-optional-state",
+ "state entries excluded by the type inference
depending on the call")
+ .setupTemporarySystemFunction("f",
OptionalStateFunction.class)
+ .setupTableSource(TIMED_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(KEYED_BASE_SINK_SCHEMA)
+ .consumedValues(
+ "+I[Bob, {+I[6]}]",
+ "+I[Alice, {+I[1]}]",
+ "+I[Bob, {+I[6], +I[6]}]",
+ "+I[Alice, {+I[1], +I[1]}]")
+ .build())
+ .runSql(
+ "INSERT INTO sink "
+ + "SELECT name, `out` FROM f(r => TABLE t
PARTITION BY name, mode => 1, uid => 'one', on_time => DESCRIPTOR(ts)) "
+ + "UNION ALL "
+ + "SELECT name, `out` FROM f(r => TABLE t
PARTITION BY name, mode => 2, uid => 'two', on_time => DESCRIPTOR(ts))")
+ .build();
+
public static final TableTestProgram
PROCESS_EAGER_AND_VALUE_VIEW_STATE_TIME =
TableTestProgram.of(
"process-eager-and-value-view-state-time",
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
index 1122c322a0e..eaf3a27f661 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
@@ -39,6 +39,7 @@ import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.TableSemantics;
import
org.apache.flink.table.runtime.operators.process.AbstractProcessTableOperator.RunnerContext;
import org.apache.flink.table.test.program.SourceTestStep;
+import org.apache.flink.table.types.inference.StateTypeStrategy;
import org.apache.flink.table.types.inference.StaticArgument;
import org.apache.flink.table.types.inference.StaticArgumentTrait;
import org.apache.flink.table.types.inference.TypeInference;
@@ -55,6 +56,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.EnumSet;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -843,6 +845,80 @@ public class ProcessTableFunctionTestUtils {
}
}
+ /**
+ * Testing function that declares two state entries but only uses the
second one if the {@code
+ * mode} argument is 2. The state type strategy of the unused entry
returns {@link
+ * Optional#empty()}, such that {@code eval()} and {@code onTimer()} are
resolved to the
+ * overload without a parameter for it. Both overloads count the processed
rows and emit the
+ * state once at the end of the input.
+ */
+ public static class OptionalStateFunction extends
AppendProcessTableFunctionBase {
+ @Override
+ public TypeInference getTypeInference(DataTypeFactory typeFactory) {
+ final LinkedHashMap<String, StateTypeStrategy> stateStrategies =
new LinkedHashMap<>();
+ stateStrategies.put(
+ "s1",
+ StateTypeStrategy.of(
+ callContext ->
+ Optional.of(
+ DataTypes.ROW(DataTypes.FIELD("i",
DataTypes.INT())))));
+ stateStrategies.put(
+ "s2",
+ StateTypeStrategy.of(
+ callContext ->
+ callContext.getArgumentValue(1,
Integer.class).orElse(1) == 2
+ ? Optional.of(
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ "s",
DataTypes.STRING())))
+ : Optional.empty()));
+ return TypeInference.newBuilder()
+ .staticArguments(
+ StaticArgument.table(
+ "r",
+ Row.class,
+ false,
+ EnumSet.of(
+
StaticArgumentTrait.SET_SEMANTIC_TABLE,
+
StaticArgumentTrait.REQUIRE_ON_TIME)),
+ StaticArgument.scalar("mode", DataTypes.INT(),
false))
+ .stateTypeStrategies(stateStrategies)
+ .outputTypeStrategy(
+ callContext ->
+ Optional.of(
+ DataTypes.ROW(
+ DataTypes.FIELD("out",
DataTypes.STRING()))))
+ .build();
+ }
+
+ public void eval(Context ctx, Row s1, Row r, Integer mode) {
+ mutateFirstState(ctx, s1);
+ }
+
+ public void eval(Context ctx, Row s1, Row s2, Row r, Integer mode) {
+ mutateFirstState(ctx, s1);
+ s2.setField("s", String.valueOf(s1.<Integer>getFieldAs("i")));
+ }
+
+ public void onTimer(OnTimerContext ctx, Row s1) {
+ collectObjects(s1);
+ }
+
+ public void onTimer(OnTimerContext ctx, Row s1, Row s2) {
+ collectObjects(s1, s2);
+ }
+
+ private void mutateFirstState(Context ctx, Row s1) {
+ final Integer i = s1.getFieldAs("i");
+ if (i == null) {
+ ctx.timeContext(Long.class).registerOnTime("end",
Long.MAX_VALUE);
+ s1.setField("i", 1);
+ } else {
+ s1.setField("i", i + 1);
+ }
+ }
+ }
+
/**
* Testing function that combines eager value state (a POJO) and a {@link
ValueView} and
* accesses both from eval() and onTimer().
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/BuiltInProcessTableFunction.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/BuiltInProcessTableFunction.java
index 1ac2e2e03e7..5cecc79866f 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/BuiltInProcessTableFunction.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/BuiltInProcessTableFunction.java
@@ -19,21 +19,32 @@
package org.apache.flink.table.runtime.functions.ptf;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.dataview.ListView;
+import org.apache.flink.table.api.dataview.MapView;
+import org.apache.flink.table.api.dataview.ValueView;
import org.apache.flink.table.catalog.DataTypeFactory;
import org.apache.flink.table.functions.BuiltInFunctionDefinition;
import org.apache.flink.table.functions.FunctionRequirement;
import org.apache.flink.table.functions.ProcessTableFunction;
import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.runtime.dataview.DataViewUtils;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.CallContext;
import org.apache.flink.table.types.inference.StateTypeStrategy;
import org.apache.flink.table.types.inference.TypeInference;
import org.apache.flink.table.types.inference.TypeStrategies;
+import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.utils.DataTypeUtils;
+import javax.annotation.Nullable;
+
+import java.time.Duration;
import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Optional;
import java.util.Set;
+import java.util.function.UnaryOperator;
import java.util.stream.Collectors;
/**
@@ -59,6 +70,18 @@ public abstract class BuiltInProcessTableFunction<T> extends
ProcessTableFunctio
protected BuiltInProcessTableFunction(
final BuiltInFunctionDefinition definition, final
SpecializedContext context) {
+ this(definition, context, null);
+ }
+
+ /**
+ * Creates a function whose state entries are declared by the specialized
function itself
+ * instead of the definition. Use {@link #internalState(Map.Entry[])} or
{@link
+ * #customState(Map.Entry[])} to create the strategies.
+ */
+ protected BuiltInProcessTableFunction(
+ final BuiltInFunctionDefinition definition,
+ final SpecializedContext context,
+ @Nullable final LinkedHashMap<String, StateTypeStrategy>
internalStateStrategies) {
this.definition = definition;
final CallContext callContext = context.getCallContext();
final TypeInference definitionTypeInference =
@@ -76,7 +99,85 @@ public abstract class BuiltInProcessTableFunction<T> extends
ProcessTableFunctio
.orElseThrow(IllegalStateException::new);
this.internalStateStrategies =
- toInternalStateStrategies(definitionTypeInference,
callContext);
+ internalStateStrategies != null
+ ? internalStateStrategies
+ : toInternalStateStrategies(definitionTypeInference);
+ }
+
+ /**
+ * Declares the state entries of a specialized function in order. The
strategies may infer
+ * external data types, they are converted to internal ones.
+ */
+ @SafeVarargs
+ protected static LinkedHashMap<String, StateTypeStrategy> internalState(
+ final Map.Entry<String, StateTypeStrategy>... entries) {
+ return state(BuiltInProcessTableFunction::toInternalStateStrategy,
entries);
+ }
+
+ /**
+ * Declares the state entries of a specialized function in order. The
strategies must already
+ * infer the data types that the function works on, no internal conversion
is applied.
+ */
+ @SafeVarargs
+ protected static LinkedHashMap<String, StateTypeStrategy> customState(
+ final Map.Entry<String, StateTypeStrategy>... entries) {
+ return state(UnaryOperator.identity(), entries);
+ }
+
+ private static LinkedHashMap<String, StateTypeStrategy> state(
+ final UnaryOperator<StateTypeStrategy> mapper,
+ final Map.Entry<String, StateTypeStrategy>[] entries) {
+ final LinkedHashMap<String, StateTypeStrategy> result = new
LinkedHashMap<>();
+ for (final Map.Entry<String, StateTypeStrategy> entry : entries) {
+ result.put(entry.getKey(), mapper.apply(entry.getValue()));
+ }
+ return result;
+ }
+
+ private static StateTypeStrategy toInternalStateStrategy(final
StateTypeStrategy strategy) {
+ return new StateTypeStrategy() {
+ @Override
+ public Optional<DataType> inferType(final CallContext callContext)
{
+ return strategy.inferType(callContext)
+ .map(BuiltInProcessTableFunction::toInternalStateType);
+ }
+
+ @Override
+ public Optional<Duration> getTimeToLive(final CallContext
callContext) {
+ return strategy.getTimeToLive(callContext);
+ }
+
+ @Override
+ public boolean isBroadcast() {
+ return strategy.isBroadcast();
+ }
+ };
+ }
+
+ /**
+ * Converts a state type to its internal representation. For {@link
ValueView}/{@link
+ * MapView}/{@link ListView} only the value/key/element types are
internalized; the view wrapper
+ * is preserved, unlike {@link DataTypeUtils#toInternalDataType} which
would rewrite the whole
+ * view to {@code RowData}.
+ */
+ private static DataType toInternalStateType(final DataType stateType) {
+ final LogicalType logicalType = stateType.getLogicalType();
+ if (DataViewUtils.isDataView(logicalType, ValueView.class)) {
+ final DataType valueType = stateType.getChildren().get(0);
+ return
ValueView.newValueViewDataType(DataTypeUtils.toInternalDataType(valueType));
+ }
+ if (DataViewUtils.isDataView(logicalType, MapView.class)) {
+ final DataType mapType = stateType.getChildren().get(0);
+ return MapView.newMapViewDataType(
+
DataTypeUtils.toInternalDataType(mapType.getChildren().get(0)),
+
DataTypeUtils.toInternalDataType(mapType.getChildren().get(1)));
+ }
+ if (DataViewUtils.isDataView(logicalType, ListView.class)) {
+ final DataType arrayType = stateType.getChildren().get(0);
+ return ListView.newListViewDataType(
+
DataTypeUtils.toInternalDataType(arrayType.getChildren().get(0)));
+ }
+ return DataTypeUtils.toInternalDataType(stateType);
}
// Uses deprecated typedArguments() because staticArguments() does not
apply the conversion
@@ -95,23 +196,13 @@ public abstract class BuiltInProcessTableFunction<T>
extends ProcessTableFunctio
.build();
}
- /** Wraps each state type strategy to produce internal data types,
preserving TTL. */
+ /** Wraps each state type strategy to produce internal data types. */
private static LinkedHashMap<String, StateTypeStrategy>
toInternalStateStrategies(
- final TypeInference definitionTypeInference, final CallContext
callContext) {
+ final TypeInference definitionTypeInference) {
final LinkedHashMap<String, StateTypeStrategy> result = new
LinkedHashMap<>();
definitionTypeInference
.getStateTypeStrategies()
- .forEach(
- (name, strategy) ->
- result.put(
- name,
- StateTypeStrategy.of(
- ctx ->
- strategy.inferType(ctx)
- .map(
-
DataTypeUtils
-
::toInternalDataType),
-
strategy.getTimeToLive(callContext).orElse(null))));
+ .forEach((name, strategy) -> result.put(name,
toInternalStateStrategy(strategy)));
return result;
}
diff --git
a/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java
b/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java
index 5204ad57868..b3efb9ef31e 100644
---
a/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java
+++
b/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java
@@ -1756,10 +1756,8 @@ public class ProcessTableFunctionTestHarness<OUT>
implements AutoCloseable {
Optional<DataType> dataTypeOpt =
strategy.inferType(callContext);
if (dataTypeOpt.isEmpty()) {
- throw new IllegalStateException(
- String.format(
- "Could not infer data type for state
parameter '%s'",
- stateName));
+ // state entry is not used by this call
+ continue;
}
DataType stateDataType = dataTypeOpt.get();
diff --git
a/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarnessTest.java
b/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarnessTest.java
index 6e079414bee..ddd279a58cd 100644
---
a/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarnessTest.java
+++
b/flink-table/flink-table-test-utils/src/test/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarnessTest.java
@@ -28,10 +28,15 @@ import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.api.dataview.ListView;
import org.apache.flink.table.api.dataview.MapView;
import org.apache.flink.table.api.dataview.ValueView;
+import org.apache.flink.table.catalog.DataTypeFactory;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.functions.ProcessTableFunction;
import org.apache.flink.table.functions.TableSemantics;
import
org.apache.flink.table.runtime.functions.ProcessTableFunctionTestHarness.TableArgument;
+import org.apache.flink.table.types.inference.StateTypeStrategy;
+import org.apache.flink.table.types.inference.StaticArgument;
+import org.apache.flink.table.types.inference.StaticArgumentTrait;
+import org.apache.flink.table.types.inference.TypeInference;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
@@ -42,9 +47,12 @@ import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.util.ArrayList;
+import java.util.EnumSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
import static org.assertj.core.api.Assertions.assertThat;
@@ -495,6 +503,43 @@ class ProcessTableFunctionTestHarnessTest {
}
}
+ /** PTF that declares two state entries of which the type inference
excludes the second one. */
+ public static class PTFWithExcludedState extends ProcessTableFunction<Row>
{
+ @Override
+ public TypeInference getTypeInference(DataTypeFactory typeFactory) {
+ final LinkedHashMap<String, StateTypeStrategy> stateStrategies =
new LinkedHashMap<>();
+ stateStrategies.put(
+ "counter",
+ StateTypeStrategy.of(
+ callContext ->
+ Optional.of(
+ DataTypes.ROW(
+ DataTypes.FIELD("count",
DataTypes.INT())))));
+ stateStrategies.put("unused", StateTypeStrategy.of(callContext ->
Optional.empty()));
+ return TypeInference.newBuilder()
+ .staticArguments(
+ StaticArgument.table(
+ "input",
+ Row.class,
+ false,
+
EnumSet.of(StaticArgumentTrait.SET_SEMANTIC_TABLE)))
+ .stateTypeStrategies(stateStrategies)
+ .outputTypeStrategy(
+ callContext ->
+ Optional.of(
+ DataTypes.ROW(
+ DataTypes.FIELD("count",
DataTypes.INT()))))
+ .build();
+ }
+
+ public void eval(Row counter, Row input) {
+ final Integer count = counter.getFieldAs("count");
+ final int newCount = count == null ? 1 : count + 1;
+ counter.setField("count", newCount);
+ collect(Row.of(newCount));
+ }
+ }
+
//
-------------------------------------------------------------------------
// Builder Configuration Tests
//
-------------------------------------------------------------------------
@@ -1822,6 +1867,30 @@ class ProcessTableFunctionTestHarnessTest {
harness.close();
}
+ @Test
+ void testStateEntryExcludedByTypeInference() throws Exception {
+ ProcessTableFunctionTestHarness<Row> harness =
+
ProcessTableFunctionTestHarness.ofClass(PTFWithExcludedState.class)
+ .withTableArgument(
+ TableArgument.forName("input")
+ .type(DataTypes.of("ROW<key STRING,
value INT>"))
+ .partitionBy("key")
+ .build())
+ .build();
+
+ harness.processElementForTable("input", Row.of("A", 10));
+ harness.processElementForTable("input", Row.of("A", 20));
+ harness.processElementForTable("input", Row.of("B", 5));
+
+ assertThat(harness.getOutput())
+ .containsExactly(Row.of("A", 1), Row.of("A", 2), Row.of("B",
1));
+
+ final Row counterA = harness.getStateForKey("counter", Row.of("A"));
+ assertThat(counterA.<Integer>getFieldAs("count")).isEqualTo(2);
+
+ harness.close();
+ }
+
@Test
void testInitialStateWithListView() throws Exception {
ListView<Integer> initialList = new ListView<>();
