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 ce009886976 [FLINK-40708][table] Support ValueView in PTFs
ce009886976 is described below
commit ce00988697662748e75ce83f6e6382f465c33e00
Author: Timo Walther <[email protected]>
AuthorDate: Wed Sep 30 16:55:09 2026 +0200
[FLINK-40708][table] Support ValueView in PTFs
This closes #29269.
---
docs/content.zh/docs/dev/table/functions/ptfs.md | 126 ++++++++++++++---
docs/content/docs/dev/table/functions/ptfs.md | 126 ++++++++++++++---
.../apache/flink/table/api/dataview/DataView.java | 1 +
.../apache/flink/table/api/dataview/ValueView.java | 155 ++++++++++++++++++++
.../table/functions/ProcessTableFunction.java | 124 ++++++++++------
.../table/types/extraction/DataTypeExtractor.java | 26 ++++
.../types/extraction/DataTypeExtractorTest.java | 50 +++++++
.../codegen/ProcessTableRunnerGenerator.scala | 42 +++---
.../stream/ProcessTableFunctionSemanticTests.java | 2 +
.../stream/ProcessTableFunctionTestPrograms.java | 37 +++++
.../exec/stream/ProcessTableFunctionTestUtils.java | 47 ++++++
.../table/runtime/dataview/DataViewUtils.java | 8 ++
.../table/runtime/dataview/StateValueView.java | 86 +++++++++++
.../runtime/generated/ProcessTableRunner.java | 157 ++++++++++++++-------
.../process/AbstractProcessTableOperator.java | 34 ++++-
.../table/runtime/dataview/DataViewUtilsTest.java | 70 +++++++++
.../functions/ProcessTableFunctionTestHarness.java | 77 +++++-----
.../runtime/functions/ValueViewStateConverter.java | 61 ++++++++
.../ProcessTableFunctionTestHarnessTest.java | 155 ++++++++++++++++++++
tools/maven/suppressions.xml | 1 +
20 files changed, 1193 insertions(+), 192 deletions(-)
diff --git a/docs/content.zh/docs/dev/table/functions/ptfs.md
b/docs/content.zh/docs/dev/table/functions/ptfs.md
index 5e7289e2ee7..ce4cd36c6d8 100644
--- a/docs/content.zh/docs/dev/table/functions/ptfs.md
+++ b/docs/content.zh/docs/dev/table/functions/ptfs.md
@@ -712,31 +712,103 @@ class CountingFunction extends
ProcessTableFunction<String> {
Flink's state backends provide different types of state to efficiently handle
large state.
-Currently, PTFs support three types of state:
-
-- **Value state**: Represents a single value.
-- **List state**: Represents a list of values, supporting operations like
appending, removing, and iterating.
-- **Map state**: Represents a map (key-value pair) for efficient lookups,
modifications, and removal of individual entries.
-
-By default, state entries in a PTF are represented as value state. This means
that every state entry is fully read from
-the state backend when the evaluation method is called, and the value is
written back to the state backend once the
-evaluation method finishes.
+Every state entry can store data ranging from a single scalar, a composite
type, or even a whole list or map. What
+differs is *how* the data is accessed:
+
+- **Eager value state**: The value follows a Read-Modify-Write cycle. On
access the complete value is deserialized, and
+ on update it is serialized again - regardless of how much of it is actually
read or changed.
+- **Value view** (`org.apache.flink.table.api.dataview.ValueView`): A value
with lazy access. The complete value
+ is only deserialized on read and serialized on update.
+- **List view** (`org.apache.flink.table.api.dataview.ListView`): A list of
values with lazy access for each element,
+ supporting operations like appending, removing, and iterating.
+- **Map view** (`org.apache.flink.table.api.dataview.MapView`): A map
(key-value pair) with lazy access for each pair,
+ supporting efficient lookups, modifications, and removal of individual
entries.
+
+By default, a state entry in a PTF is represented as eager value state. Value
state is not limited to a single scalar. It can also
+hold a whole list or map (for example a `Row` or POJO with an `ARRAY` or `MAP`
field). In that case the *complete* collection is
+deserialized when `eval()` is called and serialized again when it returns, no
matter how many elements are actually accessed.
+
+{{< tabs "2739eeed-3d13-455c-8e2f-5e164da9f844" >}}
+{{< tab "Java" >}}
+```java
+// Function that stores a whole list as eager value state.
+// The entire list is deserialized on entry and serialized on exit of every
call.
+class HistoryFunction extends ProcessTableFunction<String> {
+ public static class MyState {
+ public List<String> events = new ArrayList<>();
+ }
-To optimize state access and avoid unnecessary (de)serialization, state
entries can be declared as:
-- `org.apache.flink.table.api.dataview.ListView` (for list state)
-- `org.apache.flink.table.api.dataview.MapView` (for map state)
+ public void eval(
+ @StateHint MyState memory,
+ @ArgumentHint(SET_SEMANTIC_TABLE) Row input
+ ) {
+ memory.events.add(input.getFieldAs("eventId"));
+ collect("Seen " + memory.events.size() + " events");
+ }
+}
+```
+{{< /tab >}}
+{{< /tabs >}}
-These provide direct views to the underlying Flink state backend.
+`ValueView`, `ListView`, and `MapView` are the lazy counterparts. They provide
direct views to the underlying Flink
+state backend and are preferred over eager value state, especially when state
is accessed conditionally or is too large
+to fit into memory.
-For example, when using a `MapView`, accessing a value via `MapView#get` will
only deserialize the value associated with
-the specified key. This allows for efficient access to individual entries
without needing to load the entire map. This
-approach is particularly useful when the map does not fit entirely into memory.
+For example, a `ValueView` only deserializes its value when
`ValueView#getValue` is called and only serializes it when
+`ValueView#setValue` or `ValueView#clear` is called. A `ListView` appends or
iterates elements without materializing the
+full list, and a `MapView` accesses a value via `MapView#get` by only
deserializing the value associated with the
+specified key. This allows for efficient access to individual entries without
needing to load the entire collection.
{{< hint info >}}
State TTL is applied individually to each entry in a list or map, allowing for
fine-grained expiration control over state
elements.
{{< /hint >}}
+The following example demonstrates how to declare and use a `ValueView` for
counting events per user.
+
+{{< tabs "2837eeed-3d13-455c-8e2f-5e164da9f844" >}}
+{{< tab "Java" >}}
+```java
+// Function that uses a value view for counting events per user with lazy
state access
+class CountingFunction extends ProcessTableFunction<String> {
+ public void eval(
+ @StateHint ValueView<Integer> count,
+ @ArgumentHint(SET_SEMANTIC_TABLE) Row input
+ ) {
+ Integer currentCount = count.getValue();
+ if (currentCount == null) {
+ currentCount = 0;
+ }
+ count.setValue(currentCount + 1);
+ collect("Count for user: " + (currentCount + 1));
+ }
+}
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+The `ValueView` value type is reflectively extracted. If reflection is not
feasible - such as when a `Row` object is
+involved - a type hint can be provided. In contrast to list and map views, the
hint defines the value type directly.
+
+{{< tabs "2937eeed-3d13-455c-8e2f-5e164da9f844" >}}
+{{< tab "Java" >}}
+```java
+// Function that uses a value view of a row
+class CountingFunction extends ProcessTableFunction<String> {
+ public void eval(
+ @StateHint(type = @DataTypeHint("ROW<count INT>")) ValueView<Row> count,
+ @ArgumentHint(SET_SEMANTIC_TABLE) Row input
+ ) {
+ Row v = count.getValue();
+ Integer c = (v == null) ? 0 : v.getFieldAs("count");
+ count.setValue(Row.of(c + 1));
+ collect("Count for user: " + (c + 1));
+ }
+}
+```
+{{< /tab >}}
+{{< /tabs >}}
+
The following example demonstrates how to declare and use a `MapView`. It
assumes the PTF processes a table with the
schema `(userId, eventId, ...)`, partitioned by `userId`, with a high
cardinality of distinct `eventId` values. For this
use case, it is generally recommended to partition the table by both `userId`
and `eventId`. For example purposes, the
@@ -820,6 +892,28 @@ class CountingFunction extends
ProcessTableFunction<String> {
{{< /tab >}}
{{< /tabs >}}
+We recommend the use of the [`VARIANT`]({{< ref
"docs/sql/reference/data-types" >}}#variant) data type for future schema
+evolution. For example, a `ValueView<Variant>` can lazily store
semi-structured data whose schema may
+change over time:
+
+{{< tabs "3037eeed-3d13-455c-8e2f-5e164da9f844" >}}
+{{< tab "Java" >}}
+```java
+// Function that lazily stores the latest semi-structured payload per key
+class VariantFunction extends ProcessTableFunction<String> {
+ public void eval(
+ @StateHint ValueView<Variant> last,
+ @ArgumentHint(SET_SEMANTIC_TABLE) Row input
+ ) {
+ Variant previous = last.getValue();
+ last.setValue(input.getFieldAs("payload"));
+ collect(previous == null ? "First payload" : "Updated payload");
+ }
+}
+```
+{{< /tab >}}
+{{< /tabs >}}
+
{{< top >}}
Time and Timers
diff --git a/docs/content/docs/dev/table/functions/ptfs.md
b/docs/content/docs/dev/table/functions/ptfs.md
index 0939d1fdfb8..03878239a69 100644
--- a/docs/content/docs/dev/table/functions/ptfs.md
+++ b/docs/content/docs/dev/table/functions/ptfs.md
@@ -712,31 +712,103 @@ class CountingFunction extends
ProcessTableFunction<String> {
Flink's state backends provide different types of state to efficiently handle
large state.
-Currently, PTFs support three types of state:
-
-- **Value state**: Represents a single value.
-- **List state**: Represents a list of values, supporting operations like
appending, removing, and iterating.
-- **Map state**: Represents a map (key-value pair) for efficient lookups,
modifications, and removal of individual entries.
-
-By default, state entries in a PTF are represented as value state. This means
that every state entry is fully read from
-the state backend when the evaluation method is called, and the value is
written back to the state backend once the
-evaluation method finishes.
+Every state entry can store data ranging from a single scalar, a composite
type, or even a whole list or map. What
+differs is *how* the data is accessed:
+
+- **Eager value state**: The value follows a Read-Modify-Write cycle. On
access the complete value is deserialized, and
+ on update it is serialized again - regardless of how much of it is actually
read or changed.
+- **Value view** (`org.apache.flink.table.api.dataview.ValueView`): A value
with lazy access. The complete value
+ is only deserialized on read and serialized on update.
+- **List view** (`org.apache.flink.table.api.dataview.ListView`): A list of
values with lazy access for each element,
+ supporting operations like appending, removing, and iterating.
+- **Map view** (`org.apache.flink.table.api.dataview.MapView`): A map
(key-value pair) with lazy access for each pair,
+ supporting efficient lookups, modifications, and removal of individual
entries.
+
+By default, a state entry in a PTF is represented as eager value state. Value
state is not limited to a single scalar. It can also
+hold a whole list or map (for example a `Row` or POJO with an `ARRAY` or `MAP`
field). In that case the *complete* collection is
+deserialized when `eval()` is called and serialized again when it returns, no
matter how many elements are actually accessed.
+
+{{< tabs "2739eeed-3d13-455c-8e2f-5e164da9f844" >}}
+{{< tab "Java" >}}
+```java
+// Function that stores a whole list as eager value state.
+// The entire list is deserialized on entry and serialized on exit of every
call.
+class HistoryFunction extends ProcessTableFunction<String> {
+ public static class MyState {
+ public List<String> events = new ArrayList<>();
+ }
-To optimize state access and avoid unnecessary (de)serialization, state
entries can be declared as:
-- `org.apache.flink.table.api.dataview.ListView` (for list state)
-- `org.apache.flink.table.api.dataview.MapView` (for map state)
+ public void eval(
+ @StateHint MyState memory,
+ @ArgumentHint(SET_SEMANTIC_TABLE) Row input
+ ) {
+ memory.events.add(input.getFieldAs("eventId"));
+ collect("Seen " + memory.events.size() + " events");
+ }
+}
+```
+{{< /tab >}}
+{{< /tabs >}}
-These provide direct views to the underlying Flink state backend.
+`ValueView`, `ListView`, and `MapView` are the lazy counterparts. They provide
direct views to the underlying Flink
+state backend and are preferred over eager value state, especially when state
is accessed conditionally or is too large
+to fit into memory.
-For example, when using a `MapView`, accessing a value via `MapView#get` will
only deserialize the value associated with
-the specified key. This allows for efficient access to individual entries
without needing to load the entire map. This
-approach is particularly useful when the map does not fit entirely into memory.
+For example, a `ValueView` only deserializes its value when
`ValueView#getValue` is called and only serializes it when
+`ValueView#setValue` or `ValueView#clear` is called. A `ListView` appends or
iterates elements without materializing the
+full list, and a `MapView` accesses a value via `MapView#get` by only
deserializing the value associated with the
+specified key. This allows for efficient access to individual entries without
needing to load the entire collection.
{{< hint info >}}
State TTL is applied individually to each entry in a list or map, allowing for
fine-grained expiration control over state
elements.
{{< /hint >}}
+The following example demonstrates how to declare and use a `ValueView` for
counting events per user.
+
+{{< tabs "2837eeed-3d13-455c-8e2f-5e164da9f844" >}}
+{{< tab "Java" >}}
+```java
+// Function that uses a value view for counting events per user with lazy
state access
+class CountingFunction extends ProcessTableFunction<String> {
+ public void eval(
+ @StateHint ValueView<Integer> count,
+ @ArgumentHint(SET_SEMANTIC_TABLE) Row input
+ ) {
+ Integer currentCount = count.getValue();
+ if (currentCount == null) {
+ currentCount = 0;
+ }
+ count.setValue(currentCount + 1);
+ collect("Count for user: " + (currentCount + 1));
+ }
+}
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+The `ValueView` value type is reflectively extracted. If reflection is not
feasible - such as when a `Row` object is
+involved - a type hint can be provided. In contrast to list and map views, the
hint defines the value type directly.
+
+{{< tabs "2937eeed-3d13-455c-8e2f-5e164da9f844" >}}
+{{< tab "Java" >}}
+```java
+// Function that uses a value view of a row
+class CountingFunction extends ProcessTableFunction<String> {
+ public void eval(
+ @StateHint(type = @DataTypeHint("ROW<count INT>")) ValueView<Row> count,
+ @ArgumentHint(SET_SEMANTIC_TABLE) Row input
+ ) {
+ Row v = count.getValue();
+ Integer c = (v == null) ? 0 : v.getFieldAs("count");
+ count.setValue(Row.of(c + 1));
+ collect("Count for user: " + (c + 1));
+ }
+}
+```
+{{< /tab >}}
+{{< /tabs >}}
+
The following example demonstrates how to declare and use a `MapView`. It
assumes the PTF processes a table with the
schema `(userId, eventId, ...)`, partitioned by `userId`, with a high
cardinality of distinct `eventId` values. For this
use case, it is generally recommended to partition the table by both `userId`
and `eventId`. For example purposes, the
@@ -820,6 +892,28 @@ class CountingFunction extends
ProcessTableFunction<String> {
{{< /tab >}}
{{< /tabs >}}
+We recommend the use of the [`VARIANT`]({{< ref
"docs/sql/reference/data-types" >}}#variant) data type for future schema
+evolution. For example, a `ValueView<Variant>` can lazily store
semi-structured data whose schema may
+change over time:
+
+{{< tabs "3037eeed-3d13-455c-8e2f-5e164da9f844" >}}
+{{< tab "Java" >}}
+```java
+// Function that lazily stores the latest semi-structured payload per key
+class VariantFunction extends ProcessTableFunction<String> {
+ public void eval(
+ @StateHint ValueView<Variant> last,
+ @ArgumentHint(SET_SEMANTIC_TABLE) Row input
+ ) {
+ Variant previous = last.getValue();
+ last.setValue(input.getFieldAs("payload"));
+ collect(previous == null ? "First payload" : "Updated payload");
+ }
+}
+```
+{{< /tab >}}
+{{< /tabs >}}
+
{{< top >}}
Time and Timers
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/dataview/DataView.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/dataview/DataView.java
index a3873dcfbd9..878bedb8071 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/dataview/DataView.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/dataview/DataView.java
@@ -30,6 +30,7 @@ import org.apache.flink.table.functions.ProcessTableFunction;
*
* @see ListView
* @see MapView
+ * @see ValueView
*/
@PublicEvolving
public interface DataView {
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/dataview/ValueView.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/dataview/ValueView.java
new file mode 100644
index 00000000000..f272500dd6f
--- /dev/null
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/dataview/ValueView.java
@@ -0,0 +1,155 @@
+/*
+ * 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.api.dataview;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.functions.ProcessTableFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.types.Row;
+
+import java.util.Objects;
+
+/**
+ * A {@link DataView} that provides lazy access to a value in state entries.
+ *
+ * <p>A {@link ValueView} can be backed by a Java heap object or can leverage
Flink's state backends
+ * depending on the context. In many unbounded data scenarios, the {@link
ValueView} delegates all
+ * calls to a {@link ValueState} instead of the heap object.
+ *
+ * <p>For process table functions, the view can be used as a top-level state
entry. Data views in
+ * PTFs are always backed by state.
+ *
+ * <p>In contrast to eager value state (i.e. a top-level {@code Row} or POJO
state entry which
+ * follows a Read-Modify-Write cycle on every call), a {@link ValueView} only
deserializes the value
+ * when {@link #getValue()} is accessed and only serializes it when {@link
#setValue(Object)} or
+ * {@link #clear()} is called. Thus, it should be preferred for state that is
accessed
+ * conditionally.
+ *
+ * <p>Note: {@link ValueView} is not supported in accumulators of aggregating
functions.
+ *
+ * <p>Note: A {@link ValueView} does not support null values. Storing a null
value via {@link
+ * #setValue(Object)} is equal to calling {@link #clear()}.
+ *
+ * <p>The {@link DataType} of the view's value is reflectively extracted from
the state definition.
+ * This includes the generic argument {@code T} of this class. If reflective
extraction is not
+ * successful, it is possible to use a {@link DataTypeHint} on top of the
state entry. In contrast
+ * to {@link ListView} and {@link MapView}, the hint defines the value type
{@code T} directly.
+ *
+ * <p>The following example shows how to specify a {@link
ProcessTableFunction} with a {@link
+ * ValueView}:
+ *
+ * <pre>{@code
+ * public class CountingFunction extends ProcessTableFunction<String> {
+ * public void eval(@StateHint ValueView<Integer> count,
@ArgumentHint(SET_SEMANTIC_TABLE) Row input) {
+ * Integer c = count.getValue();
+ * if (c == null) {
+ * c = 0;
+ * }
+ * count.setValue(c + 1);
+ * collect("Count: " + (c + 1));
+ * }
+ * }
+ * }</pre>
+ *
+ * <p>If reflective extraction is not possible - for example when a {@link
Row} is used - a {@link
+ * DataTypeHint} can define the value type directly:
+ *
+ * <pre>{@code
+ * public class CountingFunction extends ProcessTableFunction<String> {
+ * public void eval(
+ * @StateHint(type = @DataTypeHint("ROW<count INT>")) ValueView<Row>
count,
+ * @ArgumentHint(SET_SEMANTIC_TABLE) Row input) {
+ * Row v = count.getValue();
+ * Integer c = (v == null) ? 0 : v.getFieldAs("count");
+ * count.setValue(Row.of(c + 1));
+ * collect("Count: " + (c + 1));
+ * }
+ * }
+ * }</pre>
+ *
+ * @param <T> value type
+ */
+@PublicEvolving
+public class ValueView<T> implements DataView {
+
+ private T value;
+
+ /**
+ * Creates a value view.
+ *
+ * <p>The {@link DataType} of the contained value is reflectively
extracted.
+ */
+ public ValueView() {
+ // default constructor
+ }
+
+ /** Returns the current value of this view or {@code null} if no value has
been set. */
+ public T getValue() {
+ return value;
+ }
+
+ /**
+ * Updates the view to the given value.
+ *
+ * <p>Setting a null value is equal to calling {@link #clear()}.
+ */
+ public void setValue(T value) {
+ this.value = value;
+ }
+
+ /** Returns {@code true} if this view holds no value (i.e. {@link
#getValue()} is null). */
+ public boolean isEmpty() {
+ return getValue() == null;
+ }
+
+ /** Removes the value from this view. */
+ @Override
+ public void clear() {
+ value = null;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof ValueView)) {
+ return false;
+ }
+ final ValueView<?> valueView = (ValueView<?>) o;
+ return Objects.equals(getValue(), valueView.getValue());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(getValue());
+ }
+
+ //
--------------------------------------------------------------------------------------------
+ // Utilities
+ //
--------------------------------------------------------------------------------------------
+
+ /** Utility method for creating a {@link DataType} of {@link ValueView}
explicitly. */
+ public static DataType newValueViewDataType(DataType valueDataType) {
+ return DataTypes.STRUCTURED(ValueView.class, DataTypes.FIELD("value",
valueDataType));
+ }
+}
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java
index ed9d243d269..a1a8e39d886 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java
@@ -24,12 +24,15 @@ import org.apache.flink.table.annotation.ArgumentTrait;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.annotation.StateHint;
+import org.apache.flink.table.api.dataview.DataView;
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.types.extraction.TypeInferenceExtractor;
import org.apache.flink.table.types.inference.TypeInference;
+import org.apache.flink.types.variant.Variant;
import org.apache.flink.util.Collector;
import java.time.Instant;
@@ -268,62 +271,56 @@ import java.time.LocalDateTime;
* }
* }</pre>
*
- * <h2>Efficiency and Design Principles</h2>
- *
- * <p>A stateful function also means that data layout and data retention
should be well thought
- * through. An ever-growing state can happen by an unlimited number of
partitions (i.e. an open
- * keyspace) or even within a partition. Consider setting a {@link
StateHint#ttl()} or call {@link
- * Context#clearAllState()} eventually:
- *
- * <pre>{@code
- * // Function that waits for a second event coming in BUT with better state
efficiency
- * class CountingFunction extends ProcessTableFunction<String> {
- * public static class SeenState {
- * public String first;
- * }
- *
- * public void eval(Context ctx, @StateHint(ttl = "1 day") SeenState memory,
@ArgumentHint(SET_SEMANTIC_TABLE) Row input) {
- * if (memory.first == null) {
- * memory.first = input.toString();
- * } else {
- * collect("Event 1: " + memory.first + " and Event 2: " +
input.toString());
- * ctx.clearAllState();
- * }
- * }
- * }
- * }</pre>
- *
* <h2>Large State</h2>
*
* <p>Flink's state backends provide different types of state to efficiently
handle large state.
*
- * <p>Currently, PTFs support three types of state:
+ * <p>Every state entry can store data ranging from a single scalar, a
composite type, or even a
+ * whole list or map. What differs is <i>how</i> the data is accessed:
*
* <ul>
- * <li><b>Value state</b>: Represents a single value.
- * <li><b>List state</b>: Represents a list of values, supporting operations
like appending,
- * removing, and iterating.
- * <li><b>Map state</b>: Represents a map (key-value pair) for efficient
lookups, modifications,
- * and removal of individual entries.
+ * <li><b>Eager value state</b>: The value follows a Read-Modify-Write
cycle. On access the
+ * complete value is deserialized, and on update it is serialized again
- regardless of how
+ * much of it is actually read or changed.
+ * <li><b>Value view</b>: A value with lazy access via {@link ValueView}.
The complete value is
+ * only deserialized on read and serialized on update.
+ * <li><b>List view</b>: A list of values with lazy access for each element
via {@link ListView},
+ * supporting operations like appending, removing, and iterating.
+ * <li><b>Map view</b>: A map (key-value pair) with lazy access for each
pair via {@link MapView},
+ * supporting efficient lookups, modifications, and removal of
individual entries.
* </ul>
*
- * <p>By default, state entries in a PTF are represented as value state. This
means that every state
- * entry is fully read from the state backend when the evaluation method is
called, and the value is
- * written back to the state backend once the evaluation method finishes.
+ * <p>By default, a state entry in a PTF is represented as eager value state.
Value state is not
+ * limited to a single scalar. It can also hold a whole list or map (for
example a {@code Row} or
+ * POJO with an {@code ARRAY} or {@code MAP} field). In that case the
<i>complete</i> collection is
+ * deserialized when {@code eval()} is called and serialized again when it
returns, no matter how
+ * many elements are actually accessed.
*
- * <p>To optimize state access and avoid unnecessary (de)serialization, state
entries can be
- * declared as {@link ListView} or {@link MapView}. These provide direct views
to the underlying
- * Flink state backend.
+ * <p>{@link ValueView}, {@link ListView}, and {@link MapView} are the lazy
counterparts. They
+ * provide direct views to the underlying Flink state backend and are
preferred over eager value
+ * state, especially when state is accessed conditionally or is too large to
fit into memory.
*
- * <p>For example, when using a {@link MapView}, accessing a value via {@link
MapView#get(Object)}
- * will only deserialize the value associated with the specified key. This
allows for efficient
- * access to individual entries without needing to load the entire map. This
approach is
- * particularly useful when the map does not fit entirely into memory.
+ * <p>For example, a {@link ListView} appends or iterates elements without
materializing the full
+ * list, and a {@link MapView} accesses a value via {@link
MapView#get(Object)} by only
+ * deserializing the value associated with the specified key. This allows for
efficient access to
+ * individual entries without needing to load the entire collection.
*
* <p>State TTL is applied individually to each entry in a list or map,
allowing for fine-grained
* expiration control over state elements.
*
* <pre>{@code
+ * // Function that uses a value view for counting events per user with lazy
state access
+ * class CountingFunction extends ProcessTableFunction<String> {
+ * public void eval(@StateHint ValueView<Integer> count,
@ArgumentHint(SET_SEMANTIC_TABLE) Row input) {
+ * Integer currentCount = count.getValue();
+ * if (currentCount == null) {
+ * currentCount = 0;
+ * }
+ * count.setValue(currentCount + 1);
+ * collect("Count for user: " + (currentCount + 1));
+ * }
+ * }
+ *
* // Function that uses a map view for storing a large map for an event
history per user
* class HistoryFunction extends ProcessTableFunction<String> {
* public void eval(@StateHint MapView<String, Integer> largeMemory,
@ArgumentHint(SET_SEMANTIC_TABLE) Row input) {
@@ -341,6 +338,46 @@ import java.time.LocalDateTime;
* }
* }</pre>
*
+ * <h2>Efficiency and Design Principles</h2>
+ *
+ * <p>A stateful function also means that data layout and data retention
should be well thought
+ * through. An ever-growing state can happen by an unlimited number of
partitions (i.e. an open
+ * keyspace) or even within a partition. Consider setting a {@link
StateHint#ttl()} or call {@link
+ * Context#clearAllState()} eventually:
+ *
+ * <pre>{@code
+ * // Function that waits for a second event coming in BUT with better state
efficiency
+ * class CountingFunction extends ProcessTableFunction<String> {
+ * public static class SeenState {
+ * public String first;
+ * }
+ *
+ * public void eval(Context ctx, @StateHint(ttl = "1 day") SeenState memory,
@ArgumentHint(SET_SEMANTIC_TABLE) Row input) {
+ * if (memory.first == null) {
+ * memory.first = input.toString();
+ * } else {
+ * collect("Event 1: " + memory.first + " and Event 2: " +
input.toString());
+ * ctx.clearAllState();
+ * }
+ * }
+ * }
+ * }</pre>
+ *
+ * <p>We recommend the use of the {@code VARIANT} data type for future schema
evolution. For
+ * example, a {@link ValueView} of {@link Variant} can lazily store
semi-structured data whose
+ * schema may change over time:
+ *
+ * <pre>{@code
+ * // Function that lazily stores the latest semi-structured payload per key
+ * class VariantFunction extends ProcessTableFunction<String> {
+ * public void eval(@StateHint ValueView<Variant> last,
@ArgumentHint(SET_SEMANTIC_TABLE) Row input) {
+ * Variant previous = last.getValue();
+ * last.setValue(input.getFieldAs("payload"));
+ * collect(previous == null ? "First payload" : "Updated payload");
+ * }
+ * }
+ * }</pre>
+ *
* <h1>Time and Timers</h1>
*
* <p>A PTF supports event time natively. Time-based services are available
via {@link
@@ -543,8 +580,9 @@ public abstract class ProcessTableFunction<T> extends
UserDefinedFunction {
/**
* Clears the given state entry within the virtual partition once the
eval() method returns.
*
- * <p>Semantically this is equal to setting all fields of the state
entry to null shortly
- * before the eval() method returns.
+ * <p>Semantically, this is equal to calling {@link DataView#clear()}
on the state entry if
+ * backed by a data view. For eager value state, semantically this is
equal to setting all
+ * fields of the state entry to null shortly before the eval() method
returns.
*
* @param stateName name of the state entry; either reflectively
extracted or manually
* defined via {@link StateHint#name()}.
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/DataTypeExtractor.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/DataTypeExtractor.java
index a3cabcc660b..4066b6acee4 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/DataTypeExtractor.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/DataTypeExtractor.java
@@ -27,6 +27,7 @@ import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.dataview.DataView;
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.data.ArrayData;
import org.apache.flink.table.data.DecimalData;
@@ -40,6 +41,7 @@ import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.KeyValueDataType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.StructuredType;
import org.apache.flink.table.types.utils.ClassDataTypeConverter;
import org.apache.flink.table.types.utils.TypeInfoDataTypeConverter;
import org.apache.flink.types.Row;
@@ -710,6 +712,17 @@ public final class DataTypeExtractor {
return dataType;
}
+ // A value view accepts any value type T (including composite ones),
so it is handled
+ // before the composite check below. If the type went through regular
extraction logic it
+ // is already the structured value view type, otherwise the annotated
type directly defines
+ // the value type T and is mapped to the underlying value.
+ if (ValueView.class.isAssignableFrom(clazz)) {
+ if (isValueViewStructuredType(dataType.getLogicalType())) {
+ return dataType;
+ }
+ return ValueView.newValueViewDataType(dataType);
+ }
+
// data type went through regular extraction logic
if (isCompositeType(dataType.getLogicalType())) {
return dataType;
@@ -733,4 +746,17 @@ public final class DataTypeExtractor {
throw extractionError("Invalid data view: %s", clazz.getName());
}
}
+
+ /**
+ * Checks whether the given type is already the structured type of a
{@link ValueView} (i.e. it
+ * went through regular reflective extraction) as opposed to an annotated
value type that still
+ * needs to be wrapped.
+ */
+ private static boolean isValueViewStructuredType(LogicalType type) {
+ return type instanceof StructuredType
+ && ((StructuredType) type)
+ .getImplementationClass()
+ .map(ValueView.class::isAssignableFrom)
+ .orElse(false);
+ }
}
diff --git
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/DataTypeExtractorTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/DataTypeExtractorTest.java
index f5e4423b9de..b07cd279c0d 100644
---
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/DataTypeExtractorTest.java
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/DataTypeExtractorTest.java
@@ -28,6 +28,7 @@ import org.apache.flink.table.api.DataTypes;
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.data.RowData;
import org.apache.flink.table.functions.TableFunction;
@@ -451,6 +452,38 @@ class DataTypeExtractorTest {
TestSpec.forType("Invalid data view",
AccumulatorWithInvalidView.class)
.expectErrorMessage(
"Annotated list views should have a logical
type of ARRAY."),
+ TestSpec.forType(
+ "Value view with default extraction",
+ AccumulatorWithDefaultValueView.class)
+ .expectDataType(
+ DataTypes.STRUCTURED(
+ AccumulatorWithDefaultValueView.class,
+ DataTypes.FIELD(
+ "valueView",
+
ValueView.newValueViewDataType(DataTypes.INT())))),
+ TestSpec.forType(
+ "Value view with custom extraction for atomic
type",
+ AccumulatorWithCustomValueView.class)
+ .expectDataType(
+ DataTypes.STRUCTURED(
+ AccumulatorWithCustomValueView.class,
+ DataTypes.FIELD(
+ "valueView",
+
ValueView.newValueViewDataType(DataTypes.INT())))),
+ TestSpec.forType(
+ "Value view with custom extraction for
composite type",
+ AccumulatorWithRowValueView.class)
+ .expectDataType(
+ DataTypes.STRUCTURED(
+ AccumulatorWithRowValueView.class,
+ DataTypes.FIELD(
+ "valueView",
+ ValueView.newValueViewDataType(
+ DataTypes.ROW(
+
DataTypes.FIELD(
+ "s",
DataTypes.STRING()),
+
DataTypes.FIELD(
+ "i",
DataTypes.INT())))))),
TestSpec.forGeneric(
"Assigning constructor for tuples",
TableFunction.class,
@@ -1076,6 +1109,23 @@ class DataTypeExtractorTest {
public ListView<?> listView;
}
+ /** Accumulator with default extraction for value view. */
+ public static class AccumulatorWithDefaultValueView {
+ public ValueView<Integer> valueView;
+ }
+
+ /** Accumulator with custom extraction for value view of an atomic type. */
+ public static class AccumulatorWithCustomValueView {
+ @DataTypeHint("INT")
+ public ValueView<?> valueView;
+ }
+
+ /** Accumulator with custom extraction for value view of a composite type.
*/
+ public static class AccumulatorWithRowValueView {
+ @DataTypeHint("ROW<s STRING, i INT>")
+ public ValueView<Row> valueView;
+ }
+
//
--------------------------------------------------------------------------------------------
/** Table function that uses a big tuple with constructor defined field
order. */
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProcessTableRunnerGenerator.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProcessTableRunnerGenerator.scala
index f2a93723348..1f41f2a6fa6 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProcessTableRunnerGenerator.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProcessTableRunnerGenerator.scala
@@ -18,9 +18,9 @@
package org.apache.flink.table.planner.codegen
import org.apache.flink.api.common.functions.OpenContext
-import org.apache.flink.api.common.state.{ListState, MapState}
+import org.apache.flink.api.common.state.{ListState, MapState, ValueState}
import org.apache.flink.table.api.ValidationException
-import org.apache.flink.table.api.dataview.{DataView, ListView, MapView}
+import org.apache.flink.table.api.dataview.{DataView, ListView, MapView,
ValueView}
import org.apache.flink.table.connector.ChangelogMode
import org.apache.flink.table.data.RowData
import org.apache.flink.table.data.conversion.RowRowConverter
@@ -38,6 +38,7 @@ import
org.apache.flink.table.planner.utils.JavaScalaConversionUtil.toScala
import org.apache.flink.table.runtime.dataview.DataViewUtils
import org.apache.flink.table.runtime.dataview.StateListView.KeyedStateListView
import
org.apache.flink.table.runtime.dataview.StateMapView.KeyedStateMapViewWithKeysNotNull
+import
org.apache.flink.table.runtime.dataview.StateValueView.KeyedStateValueView
import org.apache.flink.table.runtime.generated.{GeneratedProcessTableRunner,
ProcessTableRunner}
import org.apache.flink.table.types.DataType
import org.apache.flink.table.types.extraction.ExtractionUtils
@@ -115,18 +116,11 @@ object ProcessTableRunnerGenerator {
stateDataTypes.foreach(ExtractionUtils.checkStateDataType)
val stateHandlesTerm = "stateHandles"
- val valueStateToFunctionTerm = "valueStateToFunction"
- val stateClearedTerm = "stateCleared"
- val valueStateFromFunctionTerm = "valueStateFromFunction"
val stateEntries = stateInfos.asScala.values.zipWithIndex.toSeq
val externalStateOperands =
- generateStateToFunction(ctx, stateEntries, stateHandlesTerm,
valueStateToFunctionTerm)
- val stateFromFunctionCode = generateStateFromFunction(
- ctx,
- stateEntries,
- externalStateOperands,
- stateClearedTerm,
- valueStateFromFunctionTerm)
+ generateStateToFunction(ctx, stateEntries, stateHandlesTerm)
+ val stateFromFunctionCode =
+ generateStateFromFunction(ctx, stateEntries, externalStateOperands,
stateHandlesTerm)
// Generate result collector
val resultCollectorTerm =
@@ -206,8 +200,7 @@ object ProcessTableRunnerGenerator {
private def generateStateToFunction(
ctx: CodeGeneratorContext,
stateEntries: Seq[(StateInfo, Int)],
- stateHandlesTerm: String,
- valueStateToFunctionTerm: String): Seq[GeneratedExpression] = {
+ stateHandlesTerm: String): Seq[GeneratedExpression] = {
stateEntries.map {
case (stateInfo, pos) =>
val stateDataType = stateInfo.getDataType
@@ -228,7 +221,7 @@ object ProcessTableRunnerGenerator {
DataViewUtils.checkForInvalidDataViews(stateType)
generateValueStateToFunction(
ctx,
- valueStateToFunctionTerm,
+ stateHandlesTerm,
pos,
externalStateTypeTerm,
externalStateTerm,
@@ -253,23 +246,27 @@ object ProcessTableRunnerGenerator {
(className[KeyedStateListView[_, _]], className[ListState[_]])
} else if (DataViewUtils.isDataView(stateType, classOf[MapView[_, _]])) {
(className[KeyedStateMapViewWithKeysNotNull[_, _, _]],
className[MapState[_, _]])
+ } else if (DataViewUtils.isDataView(stateType, classOf[ValueView[_]])) {
+ (className[KeyedStateValueView[_, _]], className[ValueState[_]])
+ } else {
+ throw new IllegalStateException(s"Unsupported data view state type:
$stateType")
}
val openCode =
s"""
- |$externalStateTerm = new $constructor(($stateHandleTypeTerm)
$stateHandlesTerm[$pos]);
+ |$externalStateTerm = new $constructor(($stateHandleTypeTerm)
$stateHandlesTerm[$pos].getState());
""".stripMargin
ctx.addReusableOpenStatement(openCode)
}
private def generateValueStateToFunction(
ctx: CodeGeneratorContext,
- valueStateToFunctionTerm: String,
+ stateHandlesTerm: String,
pos: Int,
externalStateTypeTerm: String,
externalStateTerm: String,
stateDataType: DataType): String = {
- val stateEntryTerm = s"$valueStateToFunctionTerm[$pos]"
+ val stateEntryTerm = s"$stateHandlesTerm[$pos].getToFunction()"
val converterCode = genToExternalConverter(ctx, stateDataType,
stateEntryTerm)
val constructorCode = stateDataType.getConversionClass match {
@@ -299,8 +296,7 @@ object ProcessTableRunnerGenerator {
ctx: CodeGeneratorContext,
stateEntries: Seq[(StateInfo, Int)],
externalStateOperands: Seq[GeneratedExpression],
- stateClearedTerm: String,
- stateFromFunctionTerm: String): String = {
+ stateHandlesTerm: String): String = {
stateEntries
.map {
case (stateInfo, pos) =>
@@ -310,10 +306,10 @@ object ProcessTableRunnerGenerator {
if (DataViewUtils.isDataView(stateType, classOf[DataView])) {
NO_CODE
} else {
- val stateEntryTerm = s"$stateFromFunctionTerm[$pos]"
+ val stateHandleTerm = s"$stateHandlesTerm[$pos]"
val externalStateOperandTerm =
externalStateOperands(pos).resultTerm
- s"$stateEntryTerm = $stateClearedTerm[$pos] ? null : " +
- s"${genToInternalConverter(ctx,
stateDataType)(externalStateOperandTerm)};"
+ s"$stateHandleTerm.setFromFunction($stateHandleTerm.isCleared() ?
null : " +
+ s"${genToInternalConverter(ctx,
stateDataType)(externalStateOperandTerm)});"
}
}
.filter(c => c != NO_CODE)
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 45020941d47..bb893881a74 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
@@ -88,6 +88,8 @@ public class ProcessTableFunctionSemanticTests extends
SemanticTestBase {
ProcessTableFunctionTestPrograms.PROCESS_CHAINED_TIME_TABLE_API,
ProcessTableFunctionTestPrograms.PROCESS_INVALID_ROW_SEMANTIC_TABLE_TIMERS,
ProcessTableFunctionTestPrograms.PROCESS_INVALID_PASS_THROUGH_TIMERS,
+ ProcessTableFunctionTestPrograms.PROCESS_VALUE_STATE,
+ ProcessTableFunctionTestPrograms.PROCESS_COMPLEX_VALUE_STATE,
ProcessTableFunctionTestPrograms.PROCESS_LIST_STATE,
ProcessTableFunctionTestPrograms.PROCESS_MAP_STATE,
ProcessTableFunctionTestPrograms.PROCESS_MULTI_INPUT,
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 d9cc5e3a357..72a051edab3 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
@@ -25,6 +25,7 @@ import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctio
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ChainedReceivingFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ChainedSendingFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ClearStateFunction;
+import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ComplexValueViewFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ContextFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.DescriptorFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.EmptyArgFunction;
@@ -73,6 +74,7 @@ import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctio
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.UpdatingJoinFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.UpdatingRetractFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.UpdatingUpsertFunction;
+import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ValueViewFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.VariantFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.VariantStateFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.VariantTableArgFunction;
@@ -1503,6 +1505,41 @@ public class ProcessTableFunctionTestPrograms {
"Timers are not supported in the current PTF
declaration.")
.build();
+ public static final TableTestProgram PROCESS_VALUE_STATE =
+ TableTestProgram.of("process-value-state", "value view state
entry")
+ .setupTemporarySystemFunction("f", ValueViewFunction.class)
+ .setupSql(MULTI_VALUES)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(KEYED_BASE_SINK_SCHEMA)
+ .consumedValues(
+ "+I[Bob, {null,
KeyedStateValueView, +I[Bob, 12]}]",
+ "+I[Alice, {null,
KeyedStateValueView, +I[Alice, 42]}]",
+ "+I[Bob, {1, KeyedStateValueView,
+I[Bob, 99]}]",
+ "+I[Bob, {2, KeyedStateValueView,
+I[Bob, 100]}]",
+ "+I[Alice, {1,
KeyedStateValueView, +I[Alice, 400]}]")
+ .build())
+ .runSql("INSERT INTO sink SELECT * FROM f(r => TABLE t
PARTITION BY name)")
+ .build();
+
+ public static final TableTestProgram PROCESS_COMPLEX_VALUE_STATE =
+ TableTestProgram.of(
+ "process-complex-value-state", "value view with
composite value type")
+ .setupTemporarySystemFunction("f",
ComplexValueViewFunction.class)
+ .setupSql(MULTI_VALUES)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(KEYED_BASE_SINK_SCHEMA)
+ .consumedValues(
+ "+I[Bob, {null,
KeyedStateValueView, +I[Bob, 12]}]",
+ "+I[Alice, {null,
KeyedStateValueView, +I[Alice, 42]}]",
+ "+I[Bob,
{(1970-01-01T00:00:00.001Z,[12]), KeyedStateValueView, +I[Bob, 99]}]",
+ "+I[Bob,
{(1970-01-01T00:00:00.002Z,[12, 99]), KeyedStateValueView, +I[Bob, 100]}]",
+ "+I[Alice,
{(1970-01-01T00:00:00.001Z,[42]), KeyedStateValueView, +I[Alice, 400]}]")
+ .build())
+ .runSql("INSERT INTO sink SELECT * FROM f(r => TABLE t
PARTITION BY name)")
+ .build();
+
public static final TableTestProgram PROCESS_LIST_STATE =
TableTestProgram.of("process-list-state", "list view state entry")
.setupTemporarySystemFunction("f", ListStateFunction.class)
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 6bbcf9a926d..330cce0346d 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
@@ -19,6 +19,7 @@
package org.apache.flink.table.planner.plan.nodes.exec.stream;
import org.apache.flink.api.java.tuple.Tuple1;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.annotation.ArgumentHint;
import org.apache.flink.table.annotation.ArgumentTrait;
import org.apache.flink.table.annotation.DataTypeHint;
@@ -27,6 +28,7 @@ import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableRuntimeException;
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.data.RowData;
@@ -862,6 +864,51 @@ public class ProcessTableFunctionTestUtils {
}
}
+ /** Testing function. */
+ public static class ValueViewFunction extends
AppendProcessTableFunctionBase {
+ public void eval(
+ Context ctx,
+ @StateHint ValueView<Integer> s,
+ @ArgumentHint(SET_SEMANTIC_TABLE) Row r)
+ throws Exception {
+ collectObjects(s.getValue(), s.getClass().getSimpleName(), r);
+
+ // get
+ Integer count = s.getValue();
+ if (count == null) {
+ count = 0;
+ }
+
+ // update
+ s.setValue(count + 1);
+
+ // clear
+ if (count == 2) {
+ ctx.clearState("s");
+ }
+ }
+ }
+
+ /** Testing function with a value view of a nested composite type. */
+ public static class ComplexValueViewFunction extends
AppendProcessTableFunctionBase {
+ public void eval(
+ Context ctx,
+ @StateHint ValueView<Tuple2<Instant, List<Integer>>> s,
+ @ArgumentHint(SET_SEMANTIC_TABLE) Row r)
+ throws Exception {
+ collectObjects(s.getValue(), s.getClass().getSimpleName(), r);
+
+ // get
+ final Tuple2<Instant, List<Integer>> current = s.getValue();
+ final List<Integer> scores =
+ current == null ? new ArrayList<>() : new
ArrayList<>(current.f1);
+
+ // update
+ scores.add(r.getFieldAs("score"));
+ s.setValue(Tuple2.of(Instant.ofEpochMilli(scores.size()), scores));
+ }
+ }
+
/** Testing function. */
public static class MapStateFunction extends
AppendProcessTableFunctionBase {
public void eval(
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/dataview/DataViewUtils.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/dataview/DataViewUtils.java
index 94fa2a1c2b5..3a87559d53f 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/dataview/DataViewUtils.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/dataview/DataViewUtils.java
@@ -25,6 +25,7 @@ import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.api.dataview.DataView;
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.data.binary.LazyBinaryFormat;
import org.apache.flink.table.dataview.NullSerializer;
import org.apache.flink.table.runtime.typeutils.ExternalSerializer;
@@ -123,6 +124,13 @@ public final class DataViewUtils {
*/
public static DataType adjustDataViews(
DataType accumulatorDataType, boolean hasStateBackedDataViews) {
+ // ValueView is only supported as a top-level state entry of a
ProcessTableFunction.
+ // This check catches it at any position (whole accumulator, field, or
nested).
+ if (hasNested(accumulatorDataType.getLogicalType(), t -> isDataView(t,
ValueView.class))) {
+ throw new ValidationException(
+ "ValueView is not supported in accumulators of aggregating
functions. "
+ + "Use it as a top-level state entry of a
ProcessTableFunction instead.");
+ }
final Function<DataType, TypeSerializer<?>> serializer;
if (hasStateBackedDataViews) {
serializer = dataType -> NullSerializer.INSTANCE;
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/dataview/StateValueView.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/dataview/StateValueView.java
new file mode 100644
index 00000000000..a10f27632a7
--- /dev/null
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/dataview/StateValueView.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.dataview;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.table.api.dataview.ValueView;
+
+import java.io.IOException;
+
+/**
+ * {@link StateValueView} is a {@link ValueView} which is implemented using
state backends.
+ *
+ * @param <EV> the external type of the value in the {@link ValueView}
+ */
+@Internal
+public abstract class StateValueView<N, EV> extends ValueView<EV> implements
StateDataView<N> {
+
+ @Override
+ public EV getValue() {
+ try {
+ return getValueState().value();
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to get value.", e);
+ }
+ }
+
+ @Override
+ public void setValue(EV value) {
+ try {
+ if (value == null) {
+ getValueState().clear();
+ } else {
+ getValueState().update(value);
+ }
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to set value.", e);
+ }
+ }
+
+ @Override
+ public void clear() {
+ getValueState().clear();
+ }
+
+ protected abstract ValueState<EV> getValueState();
+
+ /**
+ * {@link KeyedStateValueView} is a default implementation of {@link
StateValueView} whose
+ * underlying representation is a keyed state.
+ */
+ public static final class KeyedStateValueView<N, T> extends
StateValueView<N, T> {
+
+ private final ValueState<T> valueState;
+
+ public KeyedStateValueView(ValueState<T> valueState) {
+ this.valueState = valueState;
+ }
+
+ @Override
+ public void setCurrentNamespace(N namespace) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ protected ValueState<T> getValueState() {
+ return valueState;
+ }
+ }
+}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/ProcessTableRunner.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/ProcessTableRunner.java
index 5477fc4523f..e2768af12cc 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/ProcessTableRunner.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/ProcessTableRunner.java
@@ -25,6 +25,7 @@ import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.functions.ProcessTableFunction;
+import
org.apache.flink.table.runtime.generated.ProcessTableRunner.StateHandle.Kind;
import
org.apache.flink.table.runtime.operators.process.AbstractProcessTableOperator;
import
org.apache.flink.table.runtime.operators.process.AbstractProcessTableOperator.RunnerContext;
import
org.apache.flink.table.runtime.operators.process.AbstractProcessTableOperator.RunnerOnTimerContext;
@@ -36,7 +37,6 @@ import org.apache.flink.util.function.RunnableWithException;
import javax.annotation.Nullable;
import java.io.IOException;
-import java.util.Arrays;
/**
* Abstraction of code-generated calls to {@link ProcessTableFunction} to be
used within {@link
@@ -45,10 +45,9 @@ import java.util.Arrays;
@Internal
public abstract class ProcessTableRunner extends AbstractRichFunction {
- // Constant references after initialization
- protected State[] stateHandles;
- private HashFunction[] stateHashCode;
- private RecordEqualiser[] stateEquals;
+ // Constant references after initialization.
+ // Accessed by generated code.
+ protected StateHandle[] stateHandles;
private boolean emitRowtime;
// Contexts
@@ -68,30 +67,14 @@ public abstract class ProcessTableRunner extends
AbstractRichFunction {
private @Nullable Long rowtime;
private @Nullable StringData timerName;
- /** State entries to be converted into external data structure; null if
state is empty. */
- protected RowData[] valueStateToFunction;
-
- /**
- * Reference to whether the state has been cleared within the function; if
yes, a conversion
- * from external to internal data structure is not necessary anymore.
- */
- protected boolean[] stateCleared;
-
- /** State ready for persistence; null if {@link #stateCleared} was true
during conversion. */
- protected RowData[] valueStateFromFunction;
-
public void initialize(
- State[] stateHandles,
- HashFunction[] stateHashCode,
- RecordEqualiser[] stateEquals,
+ StateHandle[] stateHandles,
boolean emitRowtime,
RunnerContext runnerContext,
RunnerOnTimerContext runnerOnTimerContext,
PassThroughCollectorBase evalCollector,
PassAllCollector onTimerCollector) {
this.stateHandles = stateHandles;
- this.stateHashCode = stateHashCode;
- this.stateEquals = stateEquals;
this.emitRowtime = emitRowtime;
// Accessed by generated code
@@ -99,9 +82,6 @@ public abstract class ProcessTableRunner extends
AbstractRichFunction {
this.runnerOnTimerContext = runnerOnTimerContext;
this.evalCollector = evalCollector;
this.onTimerCollector = onTimerCollector;
- this.valueStateToFunction = new RowData[stateHandles.length];
- this.stateCleared = new boolean[stateHandles.length];
- this.valueStateFromFunction = new RowData[stateHandles.length];
}
public void ingestTableEvent(int pos, RowData row, int timeColumn, long
watermark) {
@@ -131,11 +111,13 @@ public abstract class ProcessTableRunner extends
AbstractRichFunction {
}
public void clearAllState() {
- Arrays.fill(stateCleared, true);
+ for (StateHandle stateHandle : stateHandles) {
+ stateHandle.cleared = true;
+ }
}
public void clearState(int statePos) {
- stateCleared[statePos] = true;
+ stateHandles[statePos].cleared = true;
}
public long getTableWatermark() {
@@ -165,11 +147,12 @@ public abstract class ProcessTableRunner extends
AbstractRichFunction {
private void processMethod(RunnableWithException method) throws Exception {
if (stateHandles.length > 0) {
// For each function call:
- // - the state is read from Flink
- // - converted into external data structure
+ // - eager value state is read from Flink and converted into
external data structure
+ // - views are cheap to set up (lazy access to Flink state)
// - evaluated
- // - converted into internal data structure (if not cleared)
- // - the state is written into Flink
+ // - eager value state is converted into internal data structure
(if not cleared)
+ // and written into Flink
+ // - views may only need to be cleared if requested
moveStateToFunction();
method.run();
moveStateFromFunction();
@@ -180,41 +163,38 @@ public abstract class ProcessTableRunner extends
AbstractRichFunction {
@SuppressWarnings("unchecked")
private void moveStateToFunction() throws IOException {
- Arrays.fill(stateCleared, false);
- for (int i = 0; i < stateHandles.length; i++) {
- final State stateHandle = stateHandles[i];
- if (!(stateHandle instanceof ValueState)) {
+ for (StateHandle stateHandle : stateHandles) {
+ stateHandle.cleared = false;
+ if (stateHandle.kind != Kind.EAGER_VALUE) {
+ // Views access Flink state lazily; nothing to move eagerly.
continue;
}
- final ValueState<RowData> valueState = (ValueState<RowData>)
stateHandle;
- final RowData value = valueState.value();
- valueStateToFunction[i] = value;
+ final ValueState<RowData> valueState = (ValueState<RowData>)
stateHandle.state;
+ stateHandle.toFunction = valueState.value();
}
}
@SuppressWarnings("unchecked")
private void moveStateFromFunction() throws IOException {
- for (int i = 0; i < stateHandles.length; i++) {
- final State stateHandle = stateHandles[i];
- if (stateHandle instanceof ValueState) {
- moveValueStateFromFunction((ValueState<RowData>) stateHandle,
i);
- } else {
- if (stateCleared[i]) {
- stateHandle.clear();
- }
+ for (StateHandle stateHandle : stateHandles) {
+ if (stateHandle.kind == Kind.EAGER_VALUE) {
+ moveValueStateFromFunction(stateHandle);
+ } else if (stateHandle.cleared) {
+ stateHandle.state.clear();
}
}
}
- private void moveValueStateFromFunction(ValueState<RowData> valueState,
int pos)
- throws IOException {
- final RowData fromFunction = valueStateFromFunction[pos];
+ private void moveValueStateFromFunction(StateHandle stateHandle) throws
IOException {
+ @SuppressWarnings("unchecked")
+ final ValueState<RowData> valueState = (ValueState<RowData>)
stateHandle.state;
+ final RowData fromFunction = stateHandle.fromFunction;
if (fromFunction == null || isEmpty(fromFunction)) {
valueState.clear();
} else {
- final HashFunction hashCode = stateHashCode[pos];
- final RecordEqualiser equals = stateEquals[pos];
- final RowData toFunction = valueStateToFunction[pos];
+ final HashFunction hashCode = stateHandle.hashFunction;
+ final RecordEqualiser equals = stateHandle.equaliser;
+ final RowData toFunction = stateHandle.toFunction;
// Reduce state updates by checking if something has changed
if (toFunction == null
|| hashCode.hashCode(toFunction) !=
hashCode.hashCode(fromFunction)
@@ -232,4 +212,77 @@ public abstract class ProcessTableRunner extends
AbstractRichFunction {
}
return row.getRowKind() == RowKind.INSERT;
}
+
+ //
--------------------------------------------------------------------------------------------
+
+ /**
+ * A bundle of everything the runner needs to handle a single state entry.
+ *
+ * <p>The {@link Kind} determines how the entry is moved to/from the
function. For all view
+ * kinds ({@link Kind#MAP_VIEW}, {@link Kind#LIST_VIEW}, {@link
Kind#VALUE_VIEW}) the view
+ * accesses Flink state lazily, so {@link #hashFunction} and {@link
#equaliser} are {@code null}
+ * and the scratch fields are unused. Only {@link Kind#EAGER_VALUE}
follows a Read-Modify-Write
+ * cycle and uses the hash function, equaliser, and per-invocation scratch
fields.
+ */
+ @Internal
+ public static final class StateHandle {
+
+ /** Kind of a state entry. */
+ public enum Kind {
+ MAP_VIEW,
+ LIST_VIEW,
+ VALUE_VIEW,
+ EAGER_VALUE
+ }
+
+ private final Kind kind;
+ private final State state;
+ private final @Nullable HashFunction hashFunction;
+ private final @Nullable RecordEqualiser equaliser;
+
+ // Per-invocation scratch, only used for EAGER_VALUE.
+
+ /** State entry to be converted into external data structure; null if
state is empty. */
+ private @Nullable RowData toFunction;
+
+ /** State ready for persistence; null if {@link #cleared} was true
during conversion. */
+ private @Nullable RowData fromFunction;
+
+ /**
+ * Whether the state has been cleared within the function; if yes, a
conversion from
+ * external to internal data structure is not necessary anymore.
+ */
+ private boolean cleared;
+
+ public StateHandle(
+ Kind kind,
+ State state,
+ @Nullable HashFunction hashFunction,
+ @Nullable RecordEqualiser equaliser) {
+ this.kind = kind;
+ this.state = state;
+ this.hashFunction = hashFunction;
+ this.equaliser = equaliser;
+ }
+
+ public Kind getKind() {
+ return kind;
+ }
+
+ public State getState() {
+ return state;
+ }
+
+ public @Nullable RowData getToFunction() {
+ return toFunction;
+ }
+
+ public void setFromFunction(@Nullable RowData fromFunction) {
+ this.fromFunction = fromFunction;
+ }
+
+ public boolean isCleared() {
+ return cleared;
+ }
+ }
}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java
index d3b58b5c886..b850bbd884c 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java
@@ -43,6 +43,7 @@ import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.table.api.TableRuntimeException;
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.connector.ChangelogMode;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
@@ -52,6 +53,7 @@ import org.apache.flink.table.functions.TableSemantics;
import org.apache.flink.table.runtime.dataview.DataViewUtils;
import org.apache.flink.table.runtime.generated.HashFunction;
import org.apache.flink.table.runtime.generated.ProcessTableRunner;
+import org.apache.flink.table.runtime.generated.ProcessTableRunner.StateHandle;
import org.apache.flink.table.runtime.generated.RecordEqualiser;
import
org.apache.flink.table.runtime.operators.process.TimeConverter.InstantTimeConverter;
import
org.apache.flink.table.runtime.operators.process.TimeConverter.LocalDateTimeConverter;
@@ -92,7 +94,7 @@ public abstract class AbstractProcessTableOperator extends
AbstractStreamOperato
private transient PassThroughCollectorBase evalCollector;
private transient PassAllCollector onTimerCollector;
private transient StateDescriptor<?, ?>[] stateDescriptors;
- private transient State[] stateHandles;
+ private transient StateHandle[] stateHandles;
private transient @Nullable MapState<StringData, Long> namedTimersMapState;
private transient @Nullable InternalTimerService<StringData>
namedTimerService;
@@ -133,8 +135,6 @@ public abstract class AbstractProcessTableOperator extends
AbstractStreamOperato
processTableRunner.initialize(
stateHandles,
- stateHashCode,
- stateEquals,
shouldEmitRowtime(),
runnerContext,
runnerOnTimerContext,
@@ -375,6 +375,10 @@ public abstract class AbstractProcessTableOperator extends
AbstractStreamOperato
stateName,
ExternalSerializer.of(keyDataType),
ExternalSerializer.of(valueDataType));
+ } else if (DataViewUtils.isDataView(type, ValueView.class)) {
+ final DataType valueDataType = dataType.getChildren().get(0);
+ stateDescriptor =
+ new ValueStateDescriptor<>(stateName,
ExternalSerializer.of(valueDataType));
} else {
stateDescriptor =
new ValueStateDescriptor<>(stateName,
InternalSerializers.create(type));
@@ -391,7 +395,7 @@ public abstract class AbstractProcessTableOperator extends
AbstractStreamOperato
}
private void setStateHandles() {
- final State[] stateHandles = new State[stateDescriptors.length];
+ final StateHandle[] stateHandles = new
StateHandle[stateDescriptors.length];
for (int i = 0; i < stateDescriptors.length; i++) {
final KeyedStateStore keyedStateStore =
getKeyedStateStore().orElseThrow(IllegalStateException::new);
@@ -408,11 +412,31 @@ public abstract class AbstractProcessTableOperator
extends AbstractStreamOperato
} else {
throw new IllegalStateException("Unknown state descriptor:" +
stateDescriptor);
}
- stateHandles[i] = stateHandle;
+ // The kind must be derived from the declared type, not the
descriptor, because both
+ // eager value state and value views use a ValueStateDescriptor.
+ final LogicalType type =
stateInfos.get(i).getDataType().getLogicalType();
+ final StateHandle.Kind kind = deriveStateKind(type);
+ // Hash function and equaliser are only used by eager value state
to reduce state
+ // updates; views access state lazily and don't need them.
+ final boolean isEagerValue = kind == StateHandle.Kind.EAGER_VALUE;
+ final HashFunction hashFunction = isEagerValue ? stateHashCode[i]
: null;
+ final RecordEqualiser equaliser = isEagerValue ? stateEquals[i] :
null;
+ stateHandles[i] = new StateHandle(kind, stateHandle, hashFunction,
equaliser);
}
this.stateHandles = stateHandles;
}
+ private static StateHandle.Kind deriveStateKind(LogicalType type) {
+ if (DataViewUtils.isDataView(type, ListView.class)) {
+ return StateHandle.Kind.LIST_VIEW;
+ } else if (DataViewUtils.isDataView(type, MapView.class)) {
+ return StateHandle.Kind.MAP_VIEW;
+ } else if (DataViewUtils.isDataView(type, ValueView.class)) {
+ return StateHandle.Kind.VALUE_VIEW;
+ }
+ return StateHandle.Kind.EAGER_VALUE;
+ }
+
private boolean shouldEmitRowtime() {
return !tableSemantics.isEmpty()
&& tableSemantics.stream().allMatch(input ->
input.timeColumn() != -1);
diff --git
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/dataview/DataViewUtilsTest.java
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/dataview/DataViewUtilsTest.java
new file mode 100644
index 00000000000..1d8cf9e4689
--- /dev/null
+++
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/dataview/DataViewUtilsTest.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.dataview;
+
+import org.apache.flink.table.api.DataTypes;
+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.types.DataType;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link DataViewUtils}. */
+class DataViewUtilsTest {
+
+ @Test
+ void testValueViewInAccumulatorFieldIsRejected() {
+ final DataType accumulatorDataType =
+ DataTypes.ROW(
+ DataTypes.FIELD("v",
ValueView.newValueViewDataType(DataTypes.INT())));
+
+ assertThatThrownBy(() ->
DataViewUtils.adjustDataViews(accumulatorDataType, true))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining(
+ "ValueView is not supported in accumulators of
aggregating functions");
+ }
+
+ @Test
+ void testValueViewAsWholeAccumulatorIsRejected() {
+ final DataType accumulatorDataType =
ValueView.newValueViewDataType(DataTypes.INT());
+
+ assertThatThrownBy(() ->
DataViewUtils.adjustDataViews(accumulatorDataType, false))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining(
+ "ValueView is not supported in accumulators of
aggregating functions");
+ }
+
+ @Test
+ void testListAndMapViewsInAccumulatorAreAllowed() {
+ final DataType accumulatorDataType =
+ DataTypes.ROW(
+ DataTypes.FIELD("l",
ListView.newListViewDataType(DataTypes.INT())),
+ DataTypes.FIELD(
+ "m",
+ MapView.newMapViewDataType(DataTypes.STRING(),
DataTypes.INT())));
+
+ assertThatCode(() ->
DataViewUtils.adjustDataViews(accumulatorDataType, true))
+ .doesNotThrowAnyException();
+ }
+}
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 42034b08b69..5204ad57868 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
@@ -24,6 +24,7 @@ import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableRuntimeException;
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.data.RowData;
@@ -1774,47 +1775,49 @@ public class ProcessTableFunctionTestHarness<OUT>
implements AutoCloseable {
return allArguments;
}
- /** Creates appropriate StateConverter for the given state data type.
*/
+ /** Creates the appropriate {@link StateConverter} for the given state
data type. */
private StateConverter createStateConverter(
DataType stateDataType, ClassLoader classLoader) {
- DataType resolvedType =
-
ListView.class.isAssignableFrom(stateDataType.getConversionClass())
- || MapView.class.isAssignableFrom(
- stateDataType.getConversionClass())
- ? stateDataType.getChildren().get(0)
- : stateDataType;
-
- LogicalType logicalType = resolvedType.getLogicalType();
-
- if (logicalType instanceof ArrayType) {
- DataType elementType = resolvedType.getChildren().get(0);
- DataStructureConverter<Object, Object> elementConverter =
- DataStructureConverters.getConverter(elementType);
- elementConverter.open(classLoader);
- return new ListViewStateConverter((ArrayType) logicalType,
elementConverter);
- } else if (logicalType instanceof MapType) {
- DataType keyType = resolvedType.getChildren().get(0);
- DataType valueType = resolvedType.getChildren().get(1);
- DataStructureConverter<Object, Object> keyConverter =
- DataStructureConverters.getConverter(keyType);
- DataStructureConverter<Object, Object> valueConverter =
- DataStructureConverters.getConverter(valueType);
- keyConverter.open(classLoader);
- valueConverter.open(classLoader);
+ final Class<?> conversionClass =
stateDataType.getConversionClass();
+
+ // Data views expose their element / key-value / value type as the
single child of the
+ // structured view type. Dispatch on the view class explicitly so
that a composite value
+ // type (e.g. a ValueView<Row>) is not misrouted to a plain value
state converter.
+ if (ListView.class.isAssignableFrom(conversionClass)) {
+ final DataType arrayType = stateDataType.getChildren().get(0);
+ final DataType elementType = arrayType.getChildren().get(0);
+ return new ListViewStateConverter(
+ (ArrayType) arrayType.getLogicalType(),
+ openConverter(elementType, classLoader));
+ } else if (MapView.class.isAssignableFrom(conversionClass)) {
+ final DataType mapType = stateDataType.getChildren().get(0);
+ final DataType keyType = mapType.getChildren().get(0);
+ final DataType valueType = mapType.getChildren().get(1);
return new MapViewStateConverter(
- (MapType) logicalType, keyConverter, valueConverter);
- } else if (logicalType instanceof RowType) {
- DataStructureConverter<Object, Object> converter =
- DataStructureConverters.getConverter(resolvedType);
- converter.open(classLoader);
- return new RowStateConverter((RowType) logicalType, converter);
- } else {
- DataStructureConverter<Object, Object> converter =
- DataStructureConverters.getConverter(resolvedType);
- converter.open(classLoader);
- return new StructuredTypeStateConverter(
- resolvedType.getConversionClass(), converter);
+ (MapType) mapType.getLogicalType(),
+ openConverter(keyType, classLoader),
+ openConverter(valueType, classLoader));
+ } else if (ValueView.class.isAssignableFrom(conversionClass)) {
+ final DataType valueType = stateDataType.getChildren().get(0);
+ return new ValueViewStateConverter(openConverter(valueType,
classLoader));
+ }
+
+ // Eager value state is represented as a Row or POJO directly.
+ final LogicalType logicalType = stateDataType.getLogicalType();
+ if (logicalType instanceof RowType) {
+ return new RowStateConverter(
+ (RowType) logicalType, openConverter(stateDataType,
classLoader));
}
+ return new StructuredTypeStateConverter(
+ conversionClass, openConverter(stateDataType,
classLoader));
+ }
+
+ private static DataStructureConverter<Object, Object> openConverter(
+ DataType dataType, ClassLoader classLoader) {
+ final DataStructureConverter<Object, Object> converter =
+ DataStructureConverters.getConverter(dataType);
+ converter.open(classLoader);
+ return converter;
}
/** Checks if an argument name is a system-reserved argument. */
diff --git
a/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ValueViewStateConverter.java
b/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ValueViewStateConverter.java
new file mode 100644
index 00000000000..f30a30ff353
--- /dev/null
+++
b/flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ValueViewStateConverter.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.dataview.ValueView;
+import org.apache.flink.table.data.conversion.DataStructureConverter;
+
+/**
+ * Converter for ValueView state.
+ *
+ * <p>Converts between external ValueView objects and the internal
representation of the contained
+ * value. An empty view is represented by a {@code null} internal value.
+ */
+@Internal
+class ValueViewStateConverter implements StateConverter {
+
+ private final DataStructureConverter<Object, Object> valueConverter;
+
+ ValueViewStateConverter(DataStructureConverter<Object, Object>
valueConverter) {
+ this.valueConverter = valueConverter;
+ }
+
+ @Override
+ public Object toInternal(Object external) {
+ ValueView<?> valueView = (ValueView<?>) external;
+ Object value = valueView.getValue();
+ return value == null ? null : valueConverter.toInternal(value);
+ }
+
+ @Override
+ public Object toExternal(Object internal) {
+ ValueView<Object> valueView = new ValueView<>();
+ if (internal != null) {
+ valueView.setValue(valueConverter.toExternal(internal));
+ }
+ return valueView;
+ }
+
+ @Override
+ public Object createNewInternalState() {
+ // An empty value view is represented by a null internal value.
+ return null;
+ }
+}
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 ef124d2efca..6e079414bee 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
@@ -27,6 +27,7 @@ import org.apache.flink.table.api.TableRuntimeException;
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.connector.ChangelogMode;
import org.apache.flink.table.functions.ProcessTableFunction;
import org.apache.flink.table.functions.TableSemantics;
@@ -404,6 +405,57 @@ class ProcessTableFunctionTestHarnessTest {
}
}
+ /** PTF with ValueView state - counts rows per partition using lazy value
state. */
+ @DataTypeHint("ROW<count INT>")
+ public static class PTFWithValueViewState extends
ProcessTableFunction<Row> {
+ public void eval(
+ @StateHint ValueView<Integer> count,
+ @ArgumentHint(ArgumentTrait.SET_SEMANTIC_TABLE) Row input) {
+ Integer current = count.getValue();
+ if (current == null) {
+ current = 0;
+ }
+ current += 1;
+ count.setValue(current);
+ collect(Row.of(current));
+ }
+ }
+
+ /**
+ * PTF with ValueView state that accesses state conditionally to
demonstrate lazy access:
+ * positive values are stored, a zero clears the state via {@code
setValue(null)}, and negative
+ * values don't touch the state at all.
+ */
+ @DataTypeHint("ROW<value INT>")
+ public static class PTFWithConditionalValueViewState extends
ProcessTableFunction<Row> {
+ public void eval(
+ @StateHint ValueView<Integer> memory,
+ @ArgumentHint(ArgumentTrait.SET_SEMANTIC_TABLE) Row input) {
+ Integer value = input.getFieldAs("value");
+ if (value == 0) {
+ memory.setValue(null);
+ } else if (value > 0) {
+ memory.setValue(value);
+ }
+ // negative values: no state access at all
+ collect(Row.of(value));
+ }
+ }
+
+ /** PTF with a ValueView of a row-typed value declared via a data type
hint. */
+ @DataTypeHint("ROW<count INT>")
+ public static class PTFWithValueViewRowState extends
ProcessTableFunction<Row> {
+ public void eval(
+ @StateHint(type = @DataTypeHint("ROW<count INT>"))
ValueView<Row> count,
+ @ArgumentHint(ArgumentTrait.SET_SEMANTIC_TABLE) Row input) {
+ Row v = count.getValue();
+ Integer c = (v == null) ? 0 : v.getFieldAs("count");
+ c += 1;
+ count.setValue(Row.of(c));
+ collect(Row.of(c));
+ }
+ }
+
/** PTF with Row state - mirrors the doc example using Row as state type.
*/
@DataTypeHint("ROW<count BIGINT>")
public static class PTFWithRowState extends ProcessTableFunction<Row> {
@@ -1819,6 +1871,109 @@ class ProcessTableFunctionTestHarnessTest {
harness.close();
}
+ @Test
+ void testValueViewState() throws Exception {
+ ProcessTableFunctionTestHarness<Row> harness =
+
ProcessTableFunctionTestHarness.ofClass(PTFWithValueViewState.class)
+ .withTableArgument(
+ TableArgument.forName("input")
+ .type(DataTypes.of("ROW<key STRING>"))
+ .partitionBy("key")
+ .build())
+ .build();
+
+ harness.processElementForTable("input", Row.of("A"));
+ assertThat(harness.getOutput()).containsExactly(Row.of("A", 1));
+
+ harness.processElementForTable("input", Row.of("A"));
+ assertThat(harness.getOutput().get(1)).isEqualTo(Row.of("A", 2));
+
+ harness.processElementForTable("input", Row.of("B"));
+ assertThat(harness.getOutput().get(2)).isEqualTo(Row.of("B", 1));
+
+ ValueView<Integer> stateA = harness.getStateForKey("count",
Row.of("A"));
+ assertThat(stateA.getValue()).isEqualTo(2);
+ ValueView<Integer> stateB = harness.getStateForKey("count",
Row.of("B"));
+ assertThat(stateB.getValue()).isEqualTo(1);
+
+ harness.close();
+ }
+
+ @Test
+ void testValueViewWithRowValue() throws Exception {
+ ProcessTableFunctionTestHarness<Row> harness =
+
ProcessTableFunctionTestHarness.ofClass(PTFWithValueViewRowState.class)
+ .withTableArgument(
+ TableArgument.forName("input")
+ .type(DataTypes.of("ROW<key STRING>"))
+ .partitionBy("key")
+ .build())
+ .build();
+
+ harness.processElementForTable("input", Row.of("A"));
+ assertThat(harness.getOutput()).containsExactly(Row.of("A", 1));
+
+ harness.processElementForTable("input", Row.of("A"));
+ assertThat(harness.getOutput().get(1)).isEqualTo(Row.of("A", 2));
+
+ // The state itself holds a row value.
+ ValueView<Row> state = harness.getStateForKey("count", Row.of("A"));
+ assertThat(state.getValue()).isEqualTo(Row.of(2));
+
+ harness.close();
+ }
+
+ @Test
+ void testInitialStateWithValueView() throws Exception {
+ ValueView<Integer> initialValue = new ValueView<>();
+ initialValue.setValue(100);
+
+ ProcessTableFunctionTestHarness<Row> harness =
+
ProcessTableFunctionTestHarness.ofClass(PTFWithValueViewState.class)
+ .withTableArgument(
+ TableArgument.forName("input")
+ .type(DataTypes.of("ROW<key STRING>"))
+ .partitionBy("key")
+ .build())
+ .withInitialStateForKey("count", Row.of("A"),
initialValue)
+ .build();
+
+ ValueView<Integer> state = harness.getStateForKey("count",
Row.of("A"));
+ assertThat(state.getValue()).isEqualTo(100);
+
+ harness.processElementForTable("input", Row.of("A"));
+ assertThat(harness.getOutput()).containsExactly(Row.of("A", 101));
+
+ harness.close();
+ }
+
+ @Test
+ void testValueViewLazyAccessAndClear() throws Exception {
+ ProcessTableFunctionTestHarness<Row> harness =
+
ProcessTableFunctionTestHarness.ofClass(PTFWithConditionalValueViewState.class)
+ .withTableArgument(
+ TableArgument.forName("input")
+ .type(DataTypes.of("ROW<key STRING,
value INT>"))
+ .partitionBy("key")
+ .build())
+ .build();
+
+ // Negative value: state is never accessed, so it stays unset (empty).
+ harness.processElementForTable("input", Row.of("A", -1));
+ assertThat((Object) harness.getStateForKey("memory",
Row.of("A"))).isNull();
+
+ // Positive value: state is written.
+ harness.processElementForTable("input", Row.of("A", 5));
+ ValueView<Integer> state = harness.getStateForKey("memory",
Row.of("A"));
+ assertThat(state.getValue()).isEqualTo(5);
+
+ // Zero: state is cleared via setValue(null).
+ harness.processElementForTable("input", Row.of("A", 0));
+ assertThat((Object) harness.getStateForKey("memory",
Row.of("A"))).isNull();
+
+ harness.close();
+ }
+
@Test
void testInitialStateKeyArityMismatch() {
Exception exception =
diff --git a/tools/maven/suppressions.xml b/tools/maven/suppressions.xml
index 1e4287b1e72..957d80bf068 100644
--- a/tools/maven/suppressions.xml
+++ b/tools/maven/suppressions.xml
@@ -41,6 +41,7 @@ under the License.
<suppress files="NFAITCase.java" checks="FileLength"/>
<suppress files="HyperLogLogPlusPlus.java" checks="FileLength"/>
<suppress files="BuiltInFunctionDefinitions.java"
checks="FileLength"/>
+ <suppress files="ProcessTableFunctionTestHarnessTest.java"
checks="FileLength"/>
<!-- Legacy mockito usages -->
<!-- Updated this suppress files section due to FLINK-36063,
which migrated many classes from flink-streaming-java to flink-runtime. This
allows runtime access to stream graph. -->
