You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ro...@apache.org on 2025/10/15 06:38:09 UTC
(flink) 02/02: [FLINK-38464] Introduce OrderedMultiSetState
This is an automated email from the ASF dual-hosted git repository.
roman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 83db8f73f79cb2787da05bc329bec276a4e52d08
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Tue Sep 30 16:56:34 2025 +0200
[FLINK-38464] Introduce OrderedMultiSetState
---
.../AdaptiveSequencedMultiSetState.java | 201 +++++++
.../SequencedMultiSetState.java | 227 ++++++++
.../SequencedMultiSetStateConfig.java | 118 ++++
.../SequencedMultiSetStateContext.java | 52 ++
.../sequencedmultisetstate/TimeSelector.java | 41 ++
.../ValueStateMultiSetState.java | 207 +++++++
.../linked/LinkedMultiSetState.java | 323 ++++++++++
.../sequencedmultisetstate/linked/MetaSqnInfo.java | 58 ++
.../linked/MetaSqnInfoSerializer.java | 101 ++++
.../sequencedmultisetstate/linked/Node.java | 124 ++++
.../linked/NodeSerializer.java | 210 +++++++
.../sequencedmultisetstate/linked/RowDataKey.java | 81 +++
.../linked/RowDataKeySerializer.java | 136 +++++
.../linked/RowDataKeySerializerSnapshot.java | 153 +++++
.../sequencedmultisetstate/linked/RowSqnInfo.java | 66 +++
.../linked/RowSqnInfoSerializer.java | 101 ++++
.../SequencedMultiSetStateTest.java | 647 +++++++++++++++++++++
.../linked/MetaSqnInfoSerializerTest.java | 46 ++
.../linked/NodeSerializerTest.java | 53 ++
.../linked/RowDataKeySerializerTest.java | 104 ++++
.../linked/RowSqnInfoSerializerTest.java | 46 ++
flink-tests/pom.xml | 8 +
.../TypeSerializerTestCoverageTest.java | 8 +
23 files changed, 3111 insertions(+)
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/AdaptiveSequencedMultiSetState.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/AdaptiveSequencedMultiSetState.java
new file mode 100644
index 00000000000..8abfa76a8dc
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/AdaptiveSequencedMultiSetState.java
@@ -0,0 +1,201 @@
+/*
+ * 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.sequencedmultisetstate;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.sequencedmultisetstate.linked.LinkedMultiSetState;
+import org.apache.flink.util.function.FunctionWithException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.function.Function;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * An {@link SequencedMultiSetState} that switches dynamically between {@link
+ * ValueStateMultiSetState} and {@link LinkedMultiSetState} based on the number of elements.
+ */
+class AdaptiveSequencedMultiSetState implements SequencedMultiSetState<RowData> {
+ private static final Logger LOG = LoggerFactory.getLogger(AdaptiveSequencedMultiSetState.class);
+
+ private final ValueStateMultiSetState smallState;
+ private final LinkedMultiSetState largeState;
+ private final long switchToLargeThreshold;
+ private final long switchToSmallThreshold;
+
+ AdaptiveSequencedMultiSetState(
+ ValueStateMultiSetState smallState,
+ LinkedMultiSetState largeState,
+ long switchToLargeThreshold,
+ long switchToSmallThreshold) {
+ checkArgument(switchToLargeThreshold > switchToSmallThreshold);
+ this.smallState = smallState;
+ this.largeState = largeState;
+ this.switchToLargeThreshold = switchToLargeThreshold;
+ this.switchToSmallThreshold = switchToSmallThreshold;
+ LOG.info(
+ "Created {} with thresholds: {}=>large, {}=>small",
+ this.getClass().getSimpleName(),
+ switchToLargeThreshold,
+ switchToSmallThreshold);
+ }
+
+ @Override
+ public StateChangeInfo<RowData> add(RowData element, long timestamp) throws Exception {
+ return execute(
+ state -> state.add(element, timestamp), StateChangeInfo::getSizeAfter, "add");
+ }
+
+ @Override
+ public StateChangeInfo<RowData> append(RowData element, long timestamp) throws Exception {
+ return execute(
+ state -> state.append(element, timestamp), StateChangeInfo::getSizeAfter, "append");
+ }
+
+ @Override
+ public Iterator<Tuple2<RowData, Long>> iterator() throws Exception {
+ if (smallState.isEmpty()) {
+ return largeState.iterator();
+ } else {
+ return smallState.iterator();
+ }
+ }
+
+ @Override
+ public boolean isEmpty() throws IOException {
+ // large state check is faster
+ return largeState.isEmpty() && smallState.isEmpty();
+ }
+
+ public StateChangeInfo<RowData> remove(RowData element) throws Exception {
+ return execute(state -> state.remove(element), StateChangeInfo::getSizeAfter, "remove");
+ }
+
+ @Override
+ public void clear() {
+ clearCache();
+ smallState.clear();
+ largeState.clear();
+ }
+
+ @Override
+ public void loadCache() throws IOException {
+ smallState.loadCache();
+ largeState.loadCache();
+ }
+
+ @Override
+ public void clearCache() {
+ smallState.clearCache();
+ largeState.clearCache();
+ }
+
+ private <T> T execute(
+ FunctionWithException<SequencedMultiSetState<RowData>, T, Exception> stateOp,
+ Function<T, Long> getNewSize,
+ String action)
+ throws Exception {
+
+ final boolean isUsingLarge = isIsUsingLargeState();
+
+ // start with small state, i.e. choose smallState when both are empty
+ SequencedMultiSetState<RowData> currentState = isUsingLarge ? largeState : smallState;
+ SequencedMultiSetState<RowData> otherState = isUsingLarge ? smallState : largeState;
+
+ T result = stateOp.apply(currentState);
+ final long sizeAfter = getNewSize.apply(result);
+
+ final boolean thresholdReached =
+ isUsingLarge
+ ? sizeAfter <= switchToSmallThreshold
+ : sizeAfter >= switchToLargeThreshold;
+
+ if (thresholdReached) {
+ LOG.debug(
+ "Switch {} -> {} because '{}' resulted in state size reaching {} elements",
+ currentState.getClass().getSimpleName(),
+ otherState.getClass().getSimpleName(),
+ action,
+ sizeAfter);
+ switchState(currentState, otherState);
+ }
+
+ clearCache();
+ return result;
+ }
+
+ @VisibleForTesting
+ boolean isIsUsingLargeState() throws IOException {
+ smallState.loadCache();
+ if (!smallState.isEmpty()) {
+ return false;
+ }
+ largeState.loadCache();
+ return !largeState.isEmpty();
+ }
+
+ private void switchState(
+ SequencedMultiSetState<RowData> src, SequencedMultiSetState<RowData> dst)
+ throws Exception {
+ Iterator<Tuple2<RowData, Long>> it = src.iterator();
+ while (it.hasNext()) {
+ Tuple2<RowData, Long> next = it.next();
+ dst.append(next.f0, next.f1);
+ }
+ src.clear();
+ }
+
+ public static AdaptiveSequencedMultiSetState create(
+ SequencedMultiSetStateConfig sequencedMultiSetStateConfig,
+ String backendTypeIdentifier,
+ ValueStateMultiSetState smallState,
+ LinkedMultiSetState largeState) {
+ return new AdaptiveSequencedMultiSetState(
+ smallState,
+ largeState,
+ sequencedMultiSetStateConfig
+ .getAdaptiveHighThresholdOverride()
+ .orElse(
+ isHeap(backendTypeIdentifier)
+ ? ADAPTIVE_HEAP_HIGH_THRESHOLD
+ : ADAPTIVE_ROCKSDB_HIGH_THRESHOLD),
+ sequencedMultiSetStateConfig
+ .getAdaptiveLowThresholdOverride()
+ .orElse(
+ isHeap(backendTypeIdentifier)
+ ? ADAPTIVE_HEAP_LOW_THRESHOLD
+ : ADAPTIVE_ROCKSDB_LOW_THRESHOLD));
+ }
+
+ private static final long ADAPTIVE_HEAP_HIGH_THRESHOLD = 400;
+ private static final long ADAPTIVE_HEAP_LOW_THRESHOLD = 300;
+ private static final long ADAPTIVE_ROCKSDB_HIGH_THRESHOLD = 50;
+ private static final long ADAPTIVE_ROCKSDB_LOW_THRESHOLD = 40;
+
+ private static boolean isHeap(String stateBackend) {
+ String trim = stateBackend.trim();
+ return trim.equalsIgnoreCase("hashmap") || trim.equalsIgnoreCase("heap");
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/SequencedMultiSetState.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/SequencedMultiSetState.java
new file mode 100644
index 00000000000..5f3140b6d24
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/SequencedMultiSetState.java
@@ -0,0 +1,227 @@
+/*
+ * 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.sequencedmultisetstate;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.sequencedmultisetstate.linked.LinkedMultiSetState;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * This class represents an interface for managing an ordered multi-set state in Apache Flink. It
+ * provides methods to add, append, and remove elements while maintaining insertion order.
+ *
+ * <p>The state supports two types of semantics for adding elements:
+ *
+ * <ul>
+ * <li><b>Normal Set Semantics:</b> Replaces an existing matching element with the new one.
+ * <li><b>Multi-Set Semantics:</b> Appends the new element, allowing duplicates.
+ * </ul>
+ *
+ * <p>Removal operations are supported with different result types, indicating the outcome of the
+ * removal process, such as whether all elements were removed, the last added element was removed,
+ * or no elements were removed.
+ *
+ * @param <T> The type of elements stored in the state.
+ */
+@Internal
+public interface SequencedMultiSetState<T> {
+
+ /**
+ * Add the given element using the normal (non-multi) set semantics: if a matching element
+ * exists already, replace it (the timestamp is updated).
+ */
+ StateChangeInfo<T> add(T element, long timestamp) throws Exception;
+
+ /** Add the given element using the multi-set semantics, i.e. append. */
+ StateChangeInfo<T> append(T element, long timestamp) throws Exception;
+
+ /**
+ * Remove the given element. If there are multiple instances of the same element, remove the
+ * first one in insertion order.
+ */
+ StateChangeInfo<T> remove(T element) throws Exception;
+
+ /** Represents a result of a state changing operation. */
+ class StateChangeInfo<T> {
+ private final StateChangeType changeType;
+ private final long sizeBefore;
+ private final long sizeAfter;
+ @Nullable private final T payload; // depends on the change type
+
+ public static <T> StateChangeInfo<T> forAddition(long sizeBefore, long sizeAfter) {
+ return new StateChangeInfo<>(sizeBefore, sizeAfter, StateChangeType.ADDITION, null);
+ }
+
+ public static <T> StateChangeInfo<T> forRemovedLastAdded(
+ long sizeBefore, long sizeAfter, T payload) {
+ return new StateChangeInfo<>(
+ sizeBefore, sizeAfter, StateChangeType.REMOVAL_LAST_ADDED, payload);
+ }
+
+ public static <T> StateChangeInfo<T> forRemovedOther(long sizeBefore, long sizeAfter) {
+ return new StateChangeInfo<>(
+ sizeBefore, sizeAfter, StateChangeType.REMOVAL_OTHER, null);
+ }
+
+ public static <T> StateChangeInfo<T> forAllRemoved(
+ long sizeBefore, long sizeAfter, T payload) {
+ return new StateChangeInfo<>(
+ sizeBefore, sizeAfter, StateChangeType.REMOVAL_ALL, payload);
+ }
+
+ public static <T> StateChangeInfo<T> forRemovalNotFound(long size) {
+ return new StateChangeInfo<>(size, size, StateChangeType.REMOVAL_NOT_FOUND, null);
+ }
+
+ private StateChangeInfo(
+ long sizeBefore, long sizeAfter, StateChangeType changeType, @Nullable T payload) {
+ changeType.validate(sizeBefore, sizeAfter, payload);
+ this.sizeBefore = sizeBefore;
+ this.sizeAfter = sizeAfter;
+ this.changeType = changeType;
+ this.payload = payload;
+ }
+
+ public long getSizeAfter() {
+ return sizeAfter;
+ }
+
+ public boolean wasEmpty() {
+ return sizeBefore == 0;
+ }
+
+ public StateChangeType getChangeType() {
+ return changeType;
+ }
+
+ /** The payload depends on the {@link StateChangeType}. */
+ public Optional<T> getPayload() {
+ return Optional.ofNullable(payload);
+ }
+ }
+
+ /** Get iterator over all remaining elements and their timestamps, in order of insertion. */
+ Iterator<Tuple2<T, Long>> iterator() throws Exception;
+
+ /** Tells whether any state exists (in the given key context). */
+ boolean isEmpty() throws IOException;
+
+ /** Clear the state (in the current key context). */
+ void clear();
+
+ /** Load cache. */
+ void loadCache() throws IOException;
+
+ /** Clear caches. */
+ void clearCache();
+
+ /** Removal Result Type. */
+ enum StateChangeType {
+ /**
+ * An element was added or appended to the state. The result will not contain any elements.
+ */
+ ADDITION {
+ @Override
+ public <T> void validate(long sizeBefore, long sizeAfter, T payload) {
+ checkArgument(sizeAfter == sizeBefore + 1 || sizeAfter == sizeBefore);
+ }
+ },
+ /**
+ * Nothing was removed (e.g. as a result of TTL or not matching key), the result will not
+ * contain any elements.
+ */
+ REMOVAL_NOT_FOUND {
+ @Override
+ public <T> void validate(long sizeBefore, long sizeAfter, T payload) {
+ checkArgument(sizeAfter == sizeBefore);
+ checkArgument(payload == null);
+ }
+ },
+ /** All elements were removed. The result will contain the last removed element. */
+ REMOVAL_ALL {
+ @Override
+ public <T> void validate(long sizeBefore, long sizeAfter, T payload) {
+ checkArgument(sizeBefore > 0);
+ checkArgument(sizeAfter == 0);
+ checkNotNull(payload);
+ }
+ },
+ /**
+ * The most recently added element was removed. The result will contain the element added
+ * before it.
+ */
+ REMOVAL_LAST_ADDED {
+ @Override
+ public <T> void validate(long sizeBefore, long sizeAfter, T payload) {
+ checkArgument(sizeAfter == sizeBefore - 1);
+ checkNotNull(payload);
+ }
+ },
+ /**
+ * An element was removed, it was not the most recently added, there are more elements. The
+ * result will not contain any elements
+ */
+ REMOVAL_OTHER {
+ @Override
+ public <T> void validate(long sizeBefore, long sizeAfter, T payload) {
+ checkArgument(sizeAfter == sizeBefore - 1);
+ checkArgument(payload == null);
+ }
+ };
+
+ public abstract <T> void validate(long sizeBefore, long sizeAfter, T payload);
+ }
+
+ enum Strategy {
+ VALUE_STATE,
+ MAP_STATE,
+ ADAPTIVE
+ }
+
+ static SequencedMultiSetState<RowData> create(
+ SequencedMultiSetStateContext parameters,
+ RuntimeContext ctx,
+ String backendTypeIdentifier) {
+ switch (parameters.config.getStrategy()) {
+ case MAP_STATE:
+ return LinkedMultiSetState.create(parameters, ctx);
+ case VALUE_STATE:
+ return ValueStateMultiSetState.create(parameters, ctx);
+ case ADAPTIVE:
+ return AdaptiveSequencedMultiSetState.create(
+ parameters.config,
+ backendTypeIdentifier,
+ ValueStateMultiSetState.create(parameters, ctx),
+ LinkedMultiSetState.create(parameters, ctx));
+ default:
+ throw new UnsupportedOperationException(parameters.config.getStrategy().name());
+ }
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/SequencedMultiSetStateConfig.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/SequencedMultiSetStateConfig.java
new file mode 100644
index 00000000000..a6b16c343d8
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/SequencedMultiSetStateConfig.java
@@ -0,0 +1,118 @@
+/*
+ * 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.sequencedmultisetstate;
+
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.streaming.api.TimeDomain;
+
+import javax.annotation.Nullable;
+
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/** Configuration for {@link SequencedMultiSetState}. */
+public class SequencedMultiSetStateConfig {
+
+ private final SequencedMultiSetState.Strategy strategy;
+ private final @Nullable Long adaptiveHighThresholdOverride;
+ private final @Nullable Long adaptiveLowThresholdOverride;
+ private final StateTtlConfig ttlConfig;
+ private final TimeSelector ttlTimeSelector;
+
+ public SequencedMultiSetStateConfig(
+ SequencedMultiSetState.Strategy strategy,
+ @Nullable Long adaptiveHighThresholdOverride,
+ @Nullable Long adaptiveLowThresholdOverride,
+ StateTtlConfig ttlConfig,
+ TimeDomain ttlTimeDomain) {
+ this(
+ strategy,
+ adaptiveHighThresholdOverride,
+ adaptiveLowThresholdOverride,
+ ttlConfig,
+ TimeSelector.getTimeDomain(ttlTimeDomain));
+ }
+
+ public SequencedMultiSetStateConfig(
+ SequencedMultiSetState.Strategy strategy,
+ @Nullable Long adaptiveHighThresholdOverride,
+ @Nullable Long adaptiveLowThresholdOverride,
+ StateTtlConfig ttlConfig,
+ TimeSelector ttlTimeSelector) {
+ checkArgument(
+ !ttlConfig.isEnabled(),
+ "TTL is not supported"); // https://issues.apache.org/jira/browse/FLINK-38463
+ this.strategy = strategy;
+ this.adaptiveHighThresholdOverride = adaptiveHighThresholdOverride;
+ this.adaptiveLowThresholdOverride = adaptiveLowThresholdOverride;
+ this.ttlConfig = ttlConfig;
+ this.ttlTimeSelector = ttlTimeSelector;
+ }
+
+ public static SequencedMultiSetStateConfig defaults(
+ TimeDomain ttlTimeDomain, StateTtlConfig ttlConfig) {
+ return forValue(ttlTimeDomain, ttlConfig);
+ }
+
+ public static SequencedMultiSetStateConfig forMap(
+ TimeDomain ttlTimeDomain, StateTtlConfig ttlConfig) {
+ return new SequencedMultiSetStateConfig(
+ SequencedMultiSetState.Strategy.MAP_STATE, null, null, ttlConfig, ttlTimeDomain);
+ }
+
+ public static SequencedMultiSetStateConfig forValue(
+ TimeDomain ttlTimeDomain, StateTtlConfig ttl) {
+ return new SequencedMultiSetStateConfig(
+ SequencedMultiSetState.Strategy.VALUE_STATE, null, null, ttl, ttlTimeDomain);
+ }
+
+ public static SequencedMultiSetStateConfig adaptive(
+ TimeDomain ttlTimeDomain,
+ long adaptiveHighThresholdOverride,
+ long adaptiveLowThresholdOverride,
+ StateTtlConfig ttl) {
+ return new SequencedMultiSetStateConfig(
+ SequencedMultiSetState.Strategy.ADAPTIVE,
+ adaptiveHighThresholdOverride,
+ adaptiveLowThresholdOverride,
+ ttl,
+ ttlTimeDomain);
+ }
+
+ public TimeSelector getTimeSelector() {
+ return ttlTimeSelector;
+ }
+
+ public SequencedMultiSetState.Strategy getStrategy() {
+ return strategy;
+ }
+
+ public Optional<Long> getAdaptiveHighThresholdOverride() {
+ return Optional.ofNullable(adaptiveHighThresholdOverride);
+ }
+
+ public Optional<Long> getAdaptiveLowThresholdOverride() {
+ return Optional.ofNullable(adaptiveLowThresholdOverride);
+ }
+
+ public StateTtlConfig getTtlConfig() {
+ return ttlConfig;
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/SequencedMultiSetStateContext.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/SequencedMultiSetStateContext.java
new file mode 100644
index 00000000000..b7412d2abcd
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/SequencedMultiSetStateContext.java
@@ -0,0 +1,52 @@
+/*
+ * 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.sequencedmultisetstate;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.GeneratedHashFunction;
+import org.apache.flink.table.runtime.generated.GeneratedRecordEqualiser;
+
+import java.util.function.Function;
+
+/** {@link SequencedMultiSetState} (creation) context. */
+public class SequencedMultiSetStateContext {
+
+ public final SequencedMultiSetStateConfig config;
+ public final TypeSerializer<RowData> keySerializer;
+ public final GeneratedRecordEqualiser generatedKeyEqualiser;
+ public final GeneratedHashFunction generatedKeyHashFunction;
+ public final TypeSerializer<RowData> recordSerializer;
+ public final Function<RowData, RowData> keyExtractor;
+
+ public SequencedMultiSetStateContext(
+ TypeSerializer<RowData> keySerializer,
+ GeneratedRecordEqualiser generatedKeyEqualiser,
+ GeneratedHashFunction generatedKeyHashFunction,
+ TypeSerializer<RowData> recordSerializer,
+ Function<RowData, RowData> keyExtractor,
+ SequencedMultiSetStateConfig config) {
+ this.keySerializer = keySerializer;
+ this.generatedKeyEqualiser = generatedKeyEqualiser;
+ this.generatedKeyHashFunction = generatedKeyHashFunction;
+ this.recordSerializer = recordSerializer;
+ this.keyExtractor = keyExtractor;
+ this.config = config;
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/TimeSelector.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/TimeSelector.java
new file mode 100644
index 00000000000..40fe6f456e6
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/TimeSelector.java
@@ -0,0 +1,41 @@
+/*
+ * 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.sequencedmultisetstate;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.util.clock.SystemClock;
+
+@Internal
+@FunctionalInterface
+public interface TimeSelector {
+
+ long getTimestamp(long elementTimestamp);
+
+ static TimeSelector getTimeDomain(TimeDomain timeDomain) {
+ switch (timeDomain) {
+ case EVENT_TIME:
+ return elementTimestamp -> elementTimestamp;
+ case PROCESSING_TIME:
+ return elementTimestamp -> SystemClock.getInstance().absoluteTimeMillis();
+ default:
+ throw new IllegalStateException("unknown time domain: " + timeDomain);
+ }
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/ValueStateMultiSetState.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/ValueStateMultiSetState.java
new file mode 100644
index 00000000000..3bc88dc67e3
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/ValueStateMultiSetState.java
@@ -0,0 +1,207 @@
+/*
+ * 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.sequencedmultisetstate;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.RecordEqualiser;
+import org.apache.flink.types.RowKind;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.function.Function;
+
+/**
+ * Simple implementation of {@link SequencedMultiSetState} based on plain {@code ValueState<List>}.
+ */
+class ValueStateMultiSetState implements SequencedMultiSetState<RowData> {
+
+ private final ValueState<List<Tuple2<RowData, Long>>> valuesState;
+ private final RecordEqualiser keyEqualiser;
+ private final Function<RowData, RowData> keyExtractor;
+ private final TimeSelector timeSelector;
+ private List<Tuple2<RowData, Long>> cache;
+
+ ValueStateMultiSetState(
+ ValueState<List<Tuple2<RowData, Long>>> valuesState,
+ RecordEqualiser keyEqualiser,
+ Function<RowData, RowData> keyExtractor,
+ TimeSelector timeSelector) {
+ this.valuesState = valuesState;
+ this.keyEqualiser = keyEqualiser;
+ this.keyExtractor = keyExtractor;
+ this.timeSelector = timeSelector;
+ }
+
+ public static ValueStateMultiSetState create(
+ SequencedMultiSetStateContext p, RuntimeContext ctx) {
+ //noinspection rawtypes,unchecked
+ return new ValueStateMultiSetState(
+ ctx.getState(
+ new ValueStateDescriptor<>(
+ "list",
+ new ListSerializer<>(
+ new TupleSerializer(
+ Tuple2.class,
+ new TypeSerializer[] {
+ p.recordSerializer, LongSerializer.INSTANCE
+ })))),
+ p.generatedKeyEqualiser.newInstance(ctx.getUserCodeClassLoader()),
+ p.keyExtractor,
+ p.config.getTimeSelector());
+ }
+
+ @Override
+ public StateChangeInfo<RowData> add(RowData row, long ts) throws Exception {
+ normalizeRowKind(row);
+ final Tuple2<RowData, Long> toAdd = Tuple2.of(row, timeSelector.getTimestamp(ts));
+ final RowData key = asKey(row);
+ final List<Tuple2<RowData, Long>> list = maybeReadState();
+ final long oldSize = list.size();
+
+ int idx = Integer.MIN_VALUE;
+ int i = 0;
+ for (Tuple2<RowData, Long> t : list) {
+ if (keyEqualiser.equals(asKey(t.f0), key)) {
+ idx = i;
+ break;
+ }
+ i++;
+ }
+ if (idx < 0) {
+ list.add(toAdd);
+ } else {
+ list.set(idx, toAdd);
+ }
+ valuesState.update(list);
+ return StateChangeInfo.forAddition(oldSize, list.size());
+ }
+
+ @Override
+ public StateChangeInfo<RowData> append(RowData row, long timestamp) throws Exception {
+ normalizeRowKind(row);
+ List<Tuple2<RowData, Long>> values = maybeReadState();
+ final long oldSize = values.size();
+ values.add(Tuple2.of(row, timeSelector.getTimestamp(timestamp)));
+ valuesState.update(values);
+ return StateChangeInfo.forAddition(oldSize, values.size());
+ }
+
+ @Override
+ public Iterator<Tuple2<RowData, Long>> iterator() throws Exception {
+ return maybeReadState().iterator();
+ }
+
+ @Override
+ public StateChangeInfo<RowData> remove(RowData row) throws Exception {
+ normalizeRowKind(row);
+ final RowData key = asKey(row);
+ final List<Tuple2<RowData, Long>> list = maybeReadState();
+ final int oldSize = list.size();
+
+ int dropIdx = Integer.MIN_VALUE;
+ RowData last = null;
+ int i = 0;
+ for (Tuple2<RowData, Long> t : list) {
+ if (keyEqualiser.equals(key, asKey(t.f0))) {
+ dropIdx = i;
+ break;
+ } else {
+ last = t.f0;
+ }
+ i++;
+ }
+ final RowData removed;
+ if (dropIdx >= 0) {
+ list.remove(dropIdx);
+ removed = row;
+ valuesState.update(list);
+ } else {
+ removed = null;
+ }
+ return toRemovalResult(oldSize, list.size(), dropIdx, removed, last);
+ }
+
+ @Override
+ public void loadCache() throws IOException {
+ cache = readState();
+ }
+
+ @Override
+ public void clearCache() {
+ cache = null;
+ }
+
+ private List<Tuple2<RowData, Long>> maybeReadState() throws IOException {
+ if (cache != null) {
+ return cache;
+ }
+ return readState();
+ }
+
+ private List<Tuple2<RowData, Long>> readState() throws IOException {
+ List<Tuple2<RowData, Long>> value = valuesState.value();
+ if (value == null) {
+ value = new ArrayList<>();
+ }
+ return value;
+ }
+
+ @Override
+ public void clear() {
+ clearCache();
+ valuesState.clear();
+ }
+
+ @Override
+ public boolean isEmpty() throws IOException {
+ List<Tuple2<RowData, Long>> list = cache == null ? valuesState.value() : cache;
+ return list == null || list.isEmpty();
+ }
+
+ private RowData asKey(RowData row) {
+ return keyExtractor.apply(row);
+ }
+
+ private static void normalizeRowKind(RowData row) {
+ row.setRowKind(RowKind.INSERT);
+ }
+
+ private static StateChangeInfo<RowData> toRemovalResult(
+ long oldSize, long newSize, int dropIdx, RowData removed, RowData last) {
+ if (dropIdx < 0) {
+ return StateChangeInfo.forRemovalNotFound(oldSize);
+ } else if (newSize == 0) {
+ return StateChangeInfo.forAllRemoved(oldSize, newSize, removed);
+ } else if (dropIdx == oldSize - 1) {
+ return StateChangeInfo.forRemovedLastAdded(oldSize, newSize, last);
+ } else {
+ return StateChangeInfo.forRemovedOther(oldSize, newSize);
+ }
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/LinkedMultiSetState.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/LinkedMultiSetState.java
new file mode 100644
index 00000000000..840fe007ebf
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/LinkedMultiSetState.java
@@ -0,0 +1,323 @@
+/*
+ * 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.sequencedmultisetstate.linked;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.state.MapState;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.HashFunction;
+import org.apache.flink.table.runtime.generated.RecordEqualiser;
+import org.apache.flink.table.runtime.sequencedmultisetstate.SequencedMultiSetState;
+import org.apache.flink.table.runtime.sequencedmultisetstate.SequencedMultiSetStateContext;
+import org.apache.flink.table.runtime.sequencedmultisetstate.TimeSelector;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.function.Function;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * This class implements an ordered multi-set state backend using Flink's state primitives. It
+ * maintains the insertion order of elements and supports operations such as adding, appending, and
+ * removing elements. The state is backed by Flink's `MapState` and `ValueState` to store and manage
+ * the relationships between rows and sequence numbers (SQNs).
+ *
+ * <p>Key features of this state implementation:
+ *
+ * <ul>
+ * <li>Maintains insertion order of elements using a doubly-linked list structure.
+ * <li>Supports both normal set semantics (replacing existing elements) and multi-set semantics
+ * (allowing duplicates).
+ * <li>Efficiently tracks the highest sequence number and links between elements for fast
+ * traversal and updates.
+ * <li>Provides methods to add, append, and remove elements with appropriate handling of state
+ * transitions.
+ * </ul>
+ *
+ * <p>Note: This implementation is marked as {@code @Internal} and is intended for internal use
+ * within Flink. It may be subject to changes in future versions.
+ *
+ * <p>Usage:
+ *
+ * <ul>
+ * <li>Use the {@link #add(RowData, long)} method to add an element, replacing any existing
+ * matching element.
+ * <li>Use the {@link #append(RowData, long)} method to append an element, allowing duplicates.
+ * <li>Use the {@link #remove(RowData)} method to remove an element, with detailed removal result
+ * types.
+ * </ul>
+ *
+ * @see SequencedMultiSetState
+ * @see org.apache.flink.api.common.state.MapState
+ * @see org.apache.flink.api.common.state.ValueState
+ */
+@Internal
+public class LinkedMultiSetState implements SequencedMultiSetState<RowData> {
+
+ // maps rows to SQNs (first and last SQN for a row (same in case of upsert key))
+ private final MapState<RowDataKey, RowSqnInfo> rowToSqnState;
+ // maps SQNs to Nodes, which comprise a doubly-linked list
+ private final MapState<Long, Node> sqnToNodeState;
+ // highest sequence number; also latest emitted downstream
+ private final ValueState<MetaSqnInfo> highestSqnAndSizeState;
+
+ private final RecordEqualiser keyEqualiser;
+ private final HashFunction keyHashFunction;
+ private final Function<RowData, RowData> keyExtractor;
+ private final TimeSelector timeSelector;
+
+ private LinkedMultiSetState(
+ MapState<RowDataKey, RowSqnInfo> rowToSqnState,
+ MapState<Long, Node> sqnToNodeState,
+ ValueState<MetaSqnInfo> highestSqnAndSizeState,
+ RecordEqualiser keyEqualiser,
+ HashFunction keyHashFunction,
+ Function<RowData, RowData> keyExtractor,
+ TimeSelector timeSelector) {
+ this.rowToSqnState = checkNotNull(rowToSqnState);
+ this.sqnToNodeState = checkNotNull(sqnToNodeState);
+ this.highestSqnAndSizeState = checkNotNull(highestSqnAndSizeState);
+ this.keyEqualiser = checkNotNull(keyEqualiser);
+ this.keyHashFunction = checkNotNull(keyHashFunction);
+ this.keyExtractor = keyExtractor;
+ this.timeSelector = timeSelector;
+ }
+
+ public static LinkedMultiSetState create(SequencedMultiSetStateContext p, RuntimeContext ctx) {
+
+ RecordEqualiser keyEqualiser =
+ p.generatedKeyEqualiser.newInstance(ctx.getUserCodeClassLoader());
+ HashFunction keyHashFunction =
+ p.generatedKeyHashFunction.newInstance(ctx.getUserCodeClassLoader());
+
+ MapState<RowDataKey, RowSqnInfo> rowToSqnState =
+ ctx.getMapState(
+ new MapStateDescriptor<>(
+ "rowToSqnState",
+ new RowDataKeySerializer(
+ p.keySerializer,
+ keyEqualiser,
+ keyHashFunction,
+ p.generatedKeyEqualiser,
+ p.generatedKeyHashFunction),
+ new RowSqnInfoSerializer()));
+ MapState<Long, Node> sqnToNodeState =
+ ctx.getMapState(
+ new MapStateDescriptor<>(
+ "sqnToNodeState",
+ LongSerializer.INSTANCE,
+ new NodeSerializer(p.recordSerializer)));
+
+ ValueState<MetaSqnInfo> highestSqnState =
+ ctx.getState(
+ new ValueStateDescriptor<>("highestSqnState", new MetaSqnInfoSerializer()));
+ return new LinkedMultiSetState(
+ rowToSqnState,
+ sqnToNodeState,
+ highestSqnState,
+ keyEqualiser,
+ keyHashFunction,
+ p.keyExtractor,
+ p.config.getTimeSelector());
+ }
+
+ @Override
+ public StateChangeInfo<RowData> add(RowData row, long timestamp) throws Exception {
+ final RowDataKey key = toKey(row);
+ final MetaSqnInfo highSqnAndSize = highestSqnAndSizeState.value();
+ final Long highSqn = highSqnAndSize == null ? null : highSqnAndSize.highSqn;
+ final long oldSize = highSqnAndSize == null ? 0 : highSqnAndSize.size;
+ final RowSqnInfo rowSqnInfo = rowToSqnState.get(key);
+ final Long rowSqn = rowSqnInfo == null ? null : rowToSqnState.get(key).firstSqn;
+ final boolean isNewRowKey = rowSqn == null; // it's a 1st such record 'row'
+ final boolean isNewContextKey = highSqn == null; // 1st a record for current context key
+
+ final Long oldSqn;
+ final long newSqn;
+ final long newSize;
+
+ if (isNewContextKey && isNewRowKey) {
+ // no state at all for this context key
+ oldSqn = null;
+ newSqn = 0;
+ newSize = 1;
+ } else if (isNewRowKey) {
+ // add new rowKey "to the end"
+ oldSqn = null;
+ newSqn = highSqn + 1;
+ newSize = oldSize + 1;
+ } else {
+ // replace an existing row by rowKey
+ oldSqn = newSqn = rowSqn;
+ newSize = oldSize;
+ }
+
+ timestamp = timeSelector.getTimestamp(timestamp);
+
+ sqnToNodeState.put(
+ newSqn,
+ isNewRowKey
+ ? new Node(row, newSqn, highSqn, null, null, timestamp)
+ : sqnToNodeState.get(oldSqn).withRow(row, timestamp));
+ highestSqnAndSizeState.update(MetaSqnInfo.of(newSqn, newSize));
+ if (isNewRowKey) {
+ rowToSqnState.put(key, RowSqnInfo.ofSingle(newSqn));
+ if (!isNewContextKey) {
+ sqnToNodeState.put(highSqn, sqnToNodeState.get(highSqn).withNext(newSqn));
+ }
+ }
+ return StateChangeInfo.forAddition(oldSize, newSize);
+ }
+
+ @Override
+ public StateChangeInfo<RowData> append(RowData row, long timestamp) throws Exception {
+ final RowDataKey key = toKey(row);
+ final MetaSqnInfo highSqnAndSize = highestSqnAndSizeState.value();
+ final Long highSqn = highSqnAndSize == null ? null : highSqnAndSize.highSqn;
+ final long oldSize = highSqnAndSize == null ? 0 : highSqnAndSize.size;
+ final boolean existed = highSqn != null;
+ final long newSqn = (existed ? highSqn + 1 : 0);
+ final Node newNode =
+ new Node(
+ row,
+ newSqn,
+ highSqn, /*next*/
+ null, /*nextForRecord*/
+ null,
+ timeSelector.getTimestamp(timestamp));
+ final long newSize = oldSize + 1;
+
+ final RowSqnInfo sqnInfo = existed ? rowToSqnState.get(key) : null;
+ final Long rowSqn = sqnInfo == null ? null : sqnInfo.firstSqn;
+ if (rowSqn == null) {
+ rowToSqnState.put(key, RowSqnInfo.ofSingle(newSqn));
+ } else {
+ rowToSqnState.put(key, RowSqnInfo.of(rowSqn, newSqn));
+ sqnToNodeState.put(
+ sqnInfo.lastSqn, sqnToNodeState.get(sqnInfo.lastSqn).withNextForRecord(newSqn));
+ }
+ highestSqnAndSizeState.update(MetaSqnInfo.of(newSqn, newSize));
+ sqnToNodeState.put(newSqn, newNode);
+ if (existed) {
+ sqnToNodeState.put(highSqn, sqnToNodeState.get(highSqn).withNext(newSqn));
+ }
+ return StateChangeInfo.forAddition(oldSize, newSize);
+ }
+
+ public StateChangeInfo<RowData> remove(RowData row) throws Exception {
+ final RowDataKey key = toKey(row);
+ final RowSqnInfo sqnInfo = rowToSqnState.get(key);
+ final Long rowSqn = sqnInfo == null ? null : sqnInfo.firstSqn;
+ final MetaSqnInfo highSqnStateAndSize = highestSqnAndSizeState.value();
+ final long oldSize = highSqnStateAndSize == null ? 0L : highSqnStateAndSize.size;
+ if (rowSqn == null) {
+ return StateChangeInfo.forRemovalNotFound(oldSize);
+ }
+ final Node node = sqnToNodeState.get(rowSqn);
+
+ final Node prev = removeNode(node, key, highSqnStateAndSize);
+
+ if (!node.isHighestSqn()) {
+ return StateChangeInfo.forRemovedOther(oldSize, oldSize - 1);
+ } else if (prev == null) {
+ return StateChangeInfo.forAllRemoved(oldSize, oldSize - 1, row);
+ } else {
+ return StateChangeInfo.forRemovedLastAdded(oldSize, oldSize - 1, prev.row);
+ }
+ }
+
+ @Override
+ public void clear() {
+ clearCache();
+ sqnToNodeState.clear();
+ highestSqnAndSizeState.clear();
+ rowToSqnState.clear();
+ }
+
+ @Override
+ public void loadCache() {}
+
+ @Override
+ public void clearCache() {}
+
+ private Node removeNode(Node node, RowDataKey key, MetaSqnInfo highSqnStateAndSize)
+ throws Exception {
+
+ if (node.isLowestSqn() && node.isHighestSqn()) {
+ // fast track: if last record for PK then cleanup everything and return
+ clear();
+ return null;
+ }
+
+ sqnToNodeState.remove(node.getSqn());
+ highestSqnAndSizeState.update(
+ MetaSqnInfo.of(
+ node.isHighestSqn() ? node.prevSqn : highSqnStateAndSize.highSqn,
+ highSqnStateAndSize.size - 1));
+ if (node.isLastForRecord()) {
+ rowToSqnState.remove(key);
+ } else {
+ rowToSqnState.put(key, rowToSqnState.get(key).withFirstSqn(node.nextSqnForRecord));
+ }
+ // link prev node to next
+ Node prev = null;
+ if (node.hasPrev()) {
+ prev = sqnToNodeState.get(node.prevSqn).withNext(node.nextSqn);
+ sqnToNodeState.put(node.prevSqn, prev);
+ }
+ // link next node to prev
+ if (node.hasNext()) {
+ sqnToNodeState.put(
+ node.nextSqn, sqnToNodeState.get(node.nextSqn).withPrev(node.prevSqn));
+ }
+ return prev;
+ }
+
+ @Override
+ public Iterator<Tuple2<RowData, Long>> iterator() throws Exception {
+ // this can be implemented more efficiently
+ // however, the expected use case is to migrate all the values either to or from the memory
+ // state backend, so loading all into memory seems fine
+ List<Node> list = new ArrayList<>();
+ for (Node node : sqnToNodeState.values()) {
+ list.add(node);
+ }
+ list.sort(Comparator.comparingLong(Node::getSqn));
+ return list.stream().map(node -> Tuple2.of(node.row, node.timestamp)).iterator();
+ }
+
+ @Override
+ public boolean isEmpty() throws IOException {
+ return highestSqnAndSizeState.value() == null;
+ }
+
+ private RowDataKey toKey(RowData row0) {
+ return RowDataKey.toKey(keyExtractor.apply(row0), keyEqualiser, keyHashFunction);
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/MetaSqnInfo.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/MetaSqnInfo.java
new file mode 100644
index 00000000000..a7354872ece
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/MetaSqnInfo.java
@@ -0,0 +1,58 @@
+/*
+ * 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.sequencedmultisetstate.linked;
+
+import org.apache.flink.util.Preconditions;
+
+import java.util.Objects;
+
+/** Stores first and last SQN for a record. */
+class MetaSqnInfo {
+ public final long highSqn;
+ public final long size;
+
+ public MetaSqnInfo(long highSqn, long size) {
+ Preconditions.checkArgument(size >= 0);
+ this.highSqn = highSqn;
+ this.size = size;
+ }
+
+ public static MetaSqnInfo of(long first, long last) {
+ return new MetaSqnInfo(first, last);
+ }
+
+ @Override
+ public String toString() {
+ return "MetaSqnInfo{" + "firstSqn=" + highSqn + ", lastSqn=" + size + '}';
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof MetaSqnInfo)) {
+ return false;
+ }
+ MetaSqnInfo that = (MetaSqnInfo) o;
+ return highSqn == that.highSqn && size == that.size;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(highSqn, size);
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/MetaSqnInfoSerializer.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/MetaSqnInfoSerializer.java
new file mode 100644
index 00000000000..4819d6d4dfd
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/MetaSqnInfoSerializer.java
@@ -0,0 +1,101 @@
+/*
+ * 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.sequencedmultisetstate.linked;
+
+import org.apache.flink.api.common.typeutils.CompositeSerializer;
+import org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+
+@SuppressWarnings("ClassEscapesDefinedScope")
+public class MetaSqnInfoSerializer extends CompositeSerializer<MetaSqnInfo> {
+
+ public MetaSqnInfoSerializer() {
+ this(null, LongSerializer.INSTANCE, LongSerializer.INSTANCE);
+ }
+
+ protected MetaSqnInfoSerializer(
+ PrecomputedParameters precomputed, TypeSerializer<?>... fieldSerializers) {
+ super(
+ PrecomputedParameters.precompute(
+ true, true, (TypeSerializer<Object>[]) fieldSerializers),
+ fieldSerializers);
+ }
+
+ @Override
+ public MetaSqnInfo createInstance(Object... values) {
+ return new MetaSqnInfo((Long) values[0], (Long) values[1]);
+ }
+
+ @Override
+ protected void setField(MetaSqnInfo sqnInfo, int index, Object fieldValue) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ protected Object getField(MetaSqnInfo value, int index) {
+ switch (index) {
+ case 0:
+ return value.highSqn;
+ case 1:
+ return value.size;
+ default:
+ throw new IllegalArgumentException("invalid index: " + index);
+ }
+ }
+
+ @Override
+ protected CompositeSerializer<MetaSqnInfo> createSerializerInstance(
+ PrecomputedParameters precomputed, TypeSerializer<?>... originalSerializers) {
+ return new MetaSqnInfoSerializer(precomputed, originalSerializers);
+ }
+
+ @Override
+ public TypeSerializerSnapshot<MetaSqnInfo> snapshotConfiguration() {
+ return new MetaSqnInfoSerializerSnapshot(this);
+ }
+
+ public static class MetaSqnInfoSerializerSnapshot
+ extends CompositeTypeSerializerSnapshot<MetaSqnInfo, MetaSqnInfoSerializer> {
+
+ @SuppressWarnings("unused")
+ public MetaSqnInfoSerializerSnapshot() {}
+
+ MetaSqnInfoSerializerSnapshot(MetaSqnInfoSerializer serializer) {
+ super(serializer);
+ }
+
+ @Override
+ protected int getCurrentOuterSnapshotVersion() {
+ return 0;
+ }
+
+ @Override
+ protected TypeSerializer<?>[] getNestedSerializers(MetaSqnInfoSerializer outerSerializer) {
+ return new TypeSerializer[] {LongSerializer.INSTANCE, LongSerializer.INSTANCE};
+ }
+
+ @Override
+ protected MetaSqnInfoSerializer createOuterSerializerWithNestedSerializers(
+ TypeSerializer<?>[] nestedSerializers) {
+ return new MetaSqnInfoSerializer(null, nestedSerializers);
+ }
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/Node.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/Node.java
new file mode 100644
index 00000000000..232437ab310
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/Node.java
@@ -0,0 +1,124 @@
+/*
+ * 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.sequencedmultisetstate.linked;
+
+import org.apache.flink.table.data.RowData;
+
+import java.util.Objects;
+
+class Node {
+ final RowData row;
+ private final long sqn;
+ final Long prevSqn;
+ final Long nextSqn;
+ final Long nextSqnForRecord;
+ final Long timestamp; // for future TTL support
+
+ Node(RowData row, long sqn, Long prevSqn, Long nextSqn, Long nextSqnForRecord, Long timestamp) {
+ this.row = row;
+ this.sqn = sqn;
+ this.prevSqn = prevSqn;
+ this.nextSqn = nextSqn;
+ this.nextSqnForRecord = nextSqnForRecord;
+ this.timestamp = timestamp;
+ }
+
+ public boolean isLastForRecord() {
+ return nextSqnForRecord == null;
+ }
+
+ public boolean isLowestSqn() {
+ return !hasPrev();
+ }
+
+ public boolean isHighestSqn() {
+ return !hasNext();
+ }
+
+ public boolean hasPrev() {
+ return prevSqn != null;
+ }
+
+ public boolean hasNext() {
+ return nextSqn != null;
+ }
+
+ public Node withNextForRecord(Long nextSeqNoForRecord) {
+ return new Node(row, sqn, prevSqn, nextSqn, nextSeqNoForRecord, timestamp);
+ }
+
+ public Node withNext(Long nextSeqNo) {
+ return new Node(row, sqn, prevSqn, nextSeqNo, nextSqnForRecord, timestamp);
+ }
+
+ public Node withPrev(Long prevSeqNo) {
+ return new Node(row, sqn, prevSeqNo, nextSqn, nextSqnForRecord, timestamp);
+ }
+
+ public Node withRow(RowData row, long timestamp) {
+ return new Node(row, sqn, prevSqn, nextSqn, nextSqnForRecord, timestamp);
+ }
+
+ public RowData getRow() {
+ return row;
+ }
+
+ public long getSqn() {
+ return sqn;
+ }
+
+ public Long getPrevSqn() {
+ return prevSqn;
+ }
+
+ public Long getNextSqn() {
+ return nextSqn;
+ }
+
+ public Long getNextSqnForRecord() {
+ return nextSqnForRecord;
+ }
+
+ public Long getTimestamp() {
+ return timestamp;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof Node)) {
+ return false;
+ }
+ Node node = (Node) o;
+ // do not compare row data since:
+ // 1. the type might be different after deserialization, e.g. GenericRowData vs
+ // BinaryRowData
+ // 2. proper comparison requires (generated) equalizer
+ // 3. equals is only used in tests (as opposed to RowDataKey)
+ return sqn == node.sqn
+ && Objects.equals(prevSqn, node.prevSqn)
+ && Objects.equals(nextSqn, node.nextSqn)
+ && Objects.equals(nextSqnForRecord, node.nextSqnForRecord);
+ }
+
+ @Override
+ public int hashCode() {
+ // rowData is ignored - see equals
+ return Objects.hash(sqn, prevSqn, nextSqn, nextSqnForRecord);
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/NodeSerializer.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/NodeSerializer.java
new file mode 100644
index 00000000000..aa6577e4fed
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/NodeSerializer.java
@@ -0,0 +1,210 @@
+/*
+ * 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.sequencedmultisetstate.linked;
+
+import org.apache.flink.api.common.typeutils.CompositeSerializer;
+import org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.java.typeutils.runtime.NullableSerializer;
+import org.apache.flink.table.data.RowData;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** {@link TypeSerializer} for {@link Node}. */
+@SuppressWarnings("NullableProblems")
+public class NodeSerializer extends CompositeSerializer<Node> {
+
+ private static final LongSerializer LONG_SERIALIZER = LongSerializer.INSTANCE;
+ private static final TypeSerializer<?> NULLABLE_LONG_SERIALIZER =
+ NullableSerializer.wrap(LONG_SERIALIZER, true);
+
+ public NodeSerializer(TypeSerializer<RowData> serializer) {
+ this(null, NodeField.getFieldSerializers(serializer));
+ }
+
+ protected NodeSerializer(
+ PrecomputedParameters precomputed, TypeSerializer<?>[] originalSerializers) {
+ //noinspection unchecked
+ super(
+ PrecomputedParameters.precompute(
+ true, true, (TypeSerializer<Object>[]) originalSerializers),
+ originalSerializers);
+ }
+
+ private NodeSerializer(TypeSerializer<?>[] nestedSerializers) {
+ this(null, nestedSerializers);
+ }
+
+ @Override
+ public Node createInstance(Object... values) {
+ return new Node(
+ NodeField.ROW.get(values),
+ NodeField.SEQ_NO.get(values),
+ NodeField.PREV_SEQ_NO.get(values),
+ NodeField.NEXT_SEQ_NO.get(values),
+ NodeField.NEXT_SEQ_NO_FOR_RECORD.get(values),
+ NodeField.TIMESTAMP.get(values));
+ }
+
+ @Override
+ public boolean isImmutableType() {
+ return true;
+ }
+
+ @Override
+ protected Object getField(Node node, int index) {
+ return NodeField.get(node, index);
+ }
+
+ @Override
+ protected CompositeSerializer<Node> createSerializerInstance(
+ PrecomputedParameters precomputed, TypeSerializer<?>... originalSerializers) {
+ return new NodeSerializer(precomputed, originalSerializers);
+ }
+
+ @Override
+ public TypeSerializerSnapshot<Node> snapshotConfiguration() {
+ return new NodeSerializerSnapshot(this);
+ }
+
+ @Override
+ protected void setField(Node value, int index, Object fieldValue) {
+ throw new UnsupportedOperationException();
+ }
+
+ private enum NodeField {
+ ROW {
+ @Override
+ Object get(Node node) {
+ return node.getRow();
+ }
+
+ @Override
+ public TypeSerializer<?> getSerializer(TypeSerializer<RowData> serializer) {
+ return serializer;
+ }
+ },
+ SEQ_NO {
+ @Override
+ Object get(Node node) {
+ return node.getSqn();
+ }
+
+ @Override
+ public TypeSerializer<?> getSerializer(TypeSerializer<RowData> serializer) {
+ return LONG_SERIALIZER;
+ }
+ },
+ PREV_SEQ_NO {
+ @Override
+ Object get(Node node) {
+ return node.getPrevSqn();
+ }
+
+ @Override
+ public TypeSerializer<?> getSerializer(TypeSerializer<RowData> serializer) {
+ return NULLABLE_LONG_SERIALIZER;
+ }
+ },
+ NEXT_SEQ_NO {
+ @Override
+ Object get(Node node) {
+ return node.getNextSqn();
+ }
+
+ @Override
+ public TypeSerializer<?> getSerializer(TypeSerializer<RowData> serializer) {
+ return NULLABLE_LONG_SERIALIZER;
+ }
+ },
+ NEXT_SEQ_NO_FOR_RECORD {
+ @Override
+ Object get(Node node) {
+ return node.getNextSqnForRecord();
+ }
+
+ @Override
+ public TypeSerializer<?> getSerializer(TypeSerializer<RowData> serializer) {
+ return NULLABLE_LONG_SERIALIZER;
+ }
+ },
+ TIMESTAMP {
+ @Override
+ Object get(Node node) {
+ return node.getTimestamp();
+ }
+
+ @Override
+ public TypeSerializer<?> getSerializer(TypeSerializer<RowData> serializer) {
+ return LONG_SERIALIZER;
+ }
+ };
+
+ private static TypeSerializer<?>[] getFieldSerializers(TypeSerializer<RowData> serializer) {
+ List<TypeSerializer<?>> result = new ArrayList<>();
+ for (NodeField field : values()) {
+ result.add(field.getSerializer(serializer));
+ }
+ return result.toArray(new TypeSerializer[0]);
+ }
+
+ public abstract TypeSerializer<?> getSerializer(TypeSerializer<RowData> serializer);
+
+ abstract Object get(Node node);
+
+ <T> T get(Object... values) {
+ //noinspection unchecked
+ return (T) values[ordinal()];
+ }
+
+ public static Object get(Node node, int field) {
+ return values()[field].get(node);
+ }
+ }
+
+ /** {@link TypeSerializerSnapshot} of {@link NodeSerializerSnapshot}. */
+ public static class NodeSerializerSnapshot
+ extends CompositeTypeSerializerSnapshot<Node, NodeSerializer> {
+ @SuppressWarnings("unused")
+ public NodeSerializerSnapshot() {}
+
+ NodeSerializerSnapshot(NodeSerializer nodeSerializer) {
+ super(nodeSerializer);
+ }
+
+ @Override
+ protected int getCurrentOuterSnapshotVersion() {
+ return 0;
+ }
+
+ @Override
+ protected TypeSerializer<?>[] getNestedSerializers(NodeSerializer outerSerializer) {
+ return outerSerializer.fieldSerializers;
+ }
+
+ @Override
+ protected NodeSerializer createOuterSerializerWithNestedSerializers(
+ TypeSerializer<?>[] nestedSerializers) {
+ return new NodeSerializer(nestedSerializers);
+ }
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowDataKey.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowDataKey.java
new file mode 100644
index 00000000000..9e449a5a42c
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowDataKey.java
@@ -0,0 +1,81 @@
+/*
+ * 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.sequencedmultisetstate.linked;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.HashFunction;
+import org.apache.flink.table.runtime.generated.RecordEqualiser;
+import org.apache.flink.types.RowKind;
+
+import static org.apache.flink.types.RowKind.INSERT;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * This class wraps keys of type {@link RowData} for the following purposes:
+ *
+ * <ol>
+ * <li>Fix the {@link RowKind} to be the same in all keys.
+ * <li>Project the fields in case of upsert key.
+ * <li>Fix {@link Object#equals(Object)} and hashCode for heap state backend.
+ * <li>Potentially fix mutability for heap state backend (by copying using serializer)
+ * </ol>
+ */
+@Internal
+class RowDataKey {
+ private final RecordEqualiser equaliser;
+ private final HashFunction hashFunction;
+ final RowData rowData;
+
+ RowDataKey(RecordEqualiser equaliser, HashFunction hashFunction) {
+ this.equaliser = checkNotNull(equaliser);
+ this.hashFunction = checkNotNull(hashFunction);
+ this.rowData = null;
+ }
+
+ public RowDataKey(RowData rowData, RecordEqualiser equaliser, HashFunction hashFunction) {
+ this.equaliser = checkNotNull(equaliser);
+ this.hashFunction = checkNotNull(hashFunction);
+ this.rowData = checkNotNull(rowData);
+ }
+
+ public static RowDataKey toKeyNotProjected(
+ RowData row, RecordEqualiser equaliser, HashFunction hasher) {
+ return toKey(row, equaliser, hasher);
+ }
+
+ public static RowDataKey toKey(RowData row, RecordEqualiser equaliser, HashFunction hasher) {
+ row.setRowKind(INSERT);
+ return new RowDataKey(row, equaliser, hasher);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof RowDataKey)) {
+ return false;
+ }
+ RowDataKey other = (RowDataKey) o;
+ return equaliser.equals(rowData, other.rowData);
+ }
+
+ @Override
+ public int hashCode() {
+ return hashFunction.hashCode(rowData);
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowDataKeySerializer.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowDataKeySerializer.java
new file mode 100644
index 00000000000..9db229160cc
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowDataKeySerializer.java
@@ -0,0 +1,136 @@
+/*
+ * 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.sequencedmultisetstate.linked;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.GeneratedHashFunction;
+import org.apache.flink.table.runtime.generated.GeneratedRecordEqualiser;
+import org.apache.flink.table.runtime.generated.HashFunction;
+import org.apache.flink.table.runtime.generated.RecordEqualiser;
+
+import java.io.IOException;
+import java.util.Objects;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/** {@link TypeSerializer} for {@link RowDataKey}. */
+@Internal
+public class RowDataKeySerializer extends TypeSerializer<RowDataKey> {
+ final TypeSerializer<RowData> serializer;
+ final GeneratedRecordEqualiser equaliser; // used to snapshot
+ final GeneratedHashFunction hashFunction; // used to snapshot
+ final RecordEqualiser equalizerInstance; // passed to restored keys
+ final HashFunction hashFunctionInstance; // passed to restored keys
+
+ public RowDataKeySerializer(
+ TypeSerializer<RowData> serializer,
+ RecordEqualiser equalizerInstance,
+ HashFunction hashFunctionInstance,
+ GeneratedRecordEqualiser equaliser,
+ GeneratedHashFunction hashFunction) {
+ this.serializer = checkNotNull(serializer);
+ this.equalizerInstance = checkNotNull(equalizerInstance);
+ this.hashFunctionInstance = checkNotNull(hashFunctionInstance);
+ this.equaliser = checkNotNull(equaliser);
+ this.hashFunction = checkNotNull(hashFunction);
+ }
+
+ @Override
+ public boolean isImmutableType() {
+ return false;
+ }
+
+ @Override
+ public TypeSerializer<RowDataKey> duplicate() {
+ return new RowDataKeySerializer(
+ serializer.duplicate(),
+ equalizerInstance,
+ hashFunctionInstance,
+ equaliser,
+ hashFunction);
+ }
+
+ @Override
+ public RowDataKey createInstance() {
+ return new RowDataKey(equalizerInstance, hashFunctionInstance);
+ }
+
+ @Override
+ public RowDataKey copy(RowDataKey from) {
+ return RowDataKey.toKeyNotProjected(
+ serializer.copy(from.rowData), equalizerInstance, hashFunctionInstance);
+ }
+
+ @Override
+ public RowDataKey copy(RowDataKey from, RowDataKey reuse) {
+ return copy(from);
+ }
+
+ @Override
+ public int getLength() {
+ return serializer.getLength();
+ }
+
+ @Override
+ public void serialize(RowDataKey record, DataOutputView target) throws IOException {
+ serializer.serialize(record.rowData, target);
+ }
+
+ @Override
+ public RowDataKey deserialize(DataInputView source) throws IOException {
+ return RowDataKey.toKeyNotProjected(
+ serializer.deserialize(source), equalizerInstance, hashFunctionInstance);
+ }
+
+ @Override
+ public RowDataKey deserialize(RowDataKey reuse, DataInputView source) throws IOException {
+ return deserialize(source);
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws IOException {
+ serializer.copy(source, target);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof RowDataKeySerializer) {
+ RowDataKeySerializer other = (RowDataKeySerializer) obj;
+ return serializer.equals(other.serializer)
+ && equalizerInstance.equals(other.equalizerInstance)
+ && hashFunctionInstance.equals(other.hashFunctionInstance);
+ }
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(serializer, equalizerInstance, hashFunctionInstance);
+ }
+
+ @Override
+ public TypeSerializerSnapshot<RowDataKey> snapshotConfiguration() {
+ return new RowDataKeySerializerSnapshot(this);
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowDataKeySerializerSnapshot.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowDataKeySerializerSnapshot.java
new file mode 100644
index 00000000000..08d9888e18b
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowDataKeySerializerSnapshot.java
@@ -0,0 +1,153 @@
+/*
+ * 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.sequencedmultisetstate.linked;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.GeneratedHashFunction;
+import org.apache.flink.table.runtime.generated.GeneratedRecordEqualiser;
+import org.apache.flink.table.runtime.generated.HashFunction;
+import org.apache.flink.table.runtime.generated.RecordEqualiser;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+
+import static org.apache.flink.api.common.typeutils.TypeSerializerSnapshot.writeVersionedSnapshot;
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/** {@link TypeSerializerSnapshot} of {@link RowDataKeySerializer}. */
+public class RowDataKeySerializerSnapshot implements TypeSerializerSnapshot<RowDataKey> {
+
+ private RowDataKeySerializer serializer;
+ private TypeSerializerSnapshot<RowData> restoredRowDataSerializerSnapshot;
+
+ @SuppressWarnings("unused")
+ public RowDataKeySerializerSnapshot() {
+ // this constructor is used when restoring from a checkpoint/savepoint.
+ }
+
+ public RowDataKeySerializerSnapshot(RowDataKeySerializer serializer) {
+ this.serializer = checkNotNull(serializer);
+ }
+
+ @Override
+ public int getCurrentVersion() {
+ return 0;
+ }
+
+ @Override
+ public void writeSnapshot(DataOutputView out) throws IOException {
+ store(serializer.equaliser, out);
+ store(serializer.hashFunction, out);
+ writeVersionedSnapshot(out, serializer.serializer.snapshotConfiguration());
+ }
+
+ @Override
+ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader)
+ throws IOException {
+ checkArgument(readVersion == 0, "Unexpected version: " + readVersion);
+
+ GeneratedRecordEqualiser equaliser = restore(in, userCodeClassLoader);
+ GeneratedHashFunction hashFunction = restore(in, userCodeClassLoader);
+
+ restoredRowDataSerializerSnapshot =
+ TypeSerializerSnapshot.readVersionedSnapshot(in, userCodeClassLoader);
+
+ serializer =
+ new RowDataKeySerializer(
+ restoredRowDataSerializerSnapshot.restoreSerializer(),
+ equaliser.newInstance(userCodeClassLoader),
+ hashFunction.newInstance(userCodeClassLoader),
+ equaliser,
+ hashFunction);
+ }
+
+ private static void store(Object object, DataOutputView out) throws IOException {
+ byte[] bytes = InstantiationUtil.serializeObject(object);
+ out.writeInt(bytes.length);
+ out.write(bytes);
+ }
+
+ private <T> T restore(DataInputView in, ClassLoader classLoader) throws IOException {
+ int len = in.readInt();
+ byte[] bytes = new byte[len];
+ in.read(bytes);
+ try {
+ return InstantiationUtil.deserializeObject(bytes, classLoader); // here
+ } catch (ClassNotFoundException e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public TypeSerializer<RowDataKey> restoreSerializer() {
+ return serializer;
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<RowDataKey> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<RowDataKey> oldSerializerSnapshot) {
+ if (!(oldSerializerSnapshot instanceof RowDataKeySerializerSnapshot)) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+
+ RowDataKeySerializerSnapshot old = (RowDataKeySerializerSnapshot) oldSerializerSnapshot;
+
+ TypeSerializerSchemaCompatibility<RowData> compatibility =
+ old.restoredRowDataSerializerSnapshot.resolveSchemaCompatibility(
+ old.serializer.serializer.snapshotConfiguration());
+
+ return mapToOuterCompatibility(
+ compatibility,
+ serializer.equalizerInstance,
+ serializer.hashFunctionInstance,
+ serializer.equaliser,
+ serializer.hashFunction);
+ }
+
+ private static TypeSerializerSchemaCompatibility<RowDataKey> mapToOuterCompatibility(
+ TypeSerializerSchemaCompatibility<RowData> rowDataCmp,
+ RecordEqualiser equaliserInstance,
+ HashFunction hashFunctionInstance,
+ GeneratedRecordEqualiser equaliser,
+ GeneratedHashFunction hashFunction) {
+ if (rowDataCmp.isCompatibleAsIs()) {
+ return TypeSerializerSchemaCompatibility.compatibleAsIs();
+ } else if (rowDataCmp.isCompatibleAfterMigration()) {
+ return TypeSerializerSchemaCompatibility.compatibleAfterMigration();
+ } else if (rowDataCmp.isCompatibleWithReconfiguredSerializer()) {
+ return TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(
+ new RowDataKeySerializer(
+ rowDataCmp.getReconfiguredSerializer(),
+ equaliserInstance,
+ hashFunctionInstance,
+ equaliser,
+ hashFunction));
+ } else if (rowDataCmp.isIncompatible()) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ } else {
+ throw new UnsupportedOperationException("Unknown compatibility mode: " + rowDataCmp);
+ }
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowSqnInfo.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowSqnInfo.java
new file mode 100644
index 00000000000..b8bcc438b68
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowSqnInfo.java
@@ -0,0 +1,66 @@
+/*
+ * 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.sequencedmultisetstate.linked;
+
+import org.apache.flink.util.Preconditions;
+
+import java.util.Objects;
+
+/** Stores first and las SQN for a record. */
+class RowSqnInfo {
+ public final long firstSqn;
+ public final long lastSqn;
+
+ public RowSqnInfo(long firstSqn, long lastSqn) {
+ Preconditions.checkArgument(firstSqn <= lastSqn);
+ this.firstSqn = firstSqn;
+ this.lastSqn = lastSqn;
+ }
+
+ public static RowSqnInfo ofSingle(long sqn) {
+ return of(sqn, sqn);
+ }
+
+ public static RowSqnInfo of(long first, long last) {
+ return new RowSqnInfo(first, last);
+ }
+
+ @Override
+ public String toString() {
+ return "RowSqnInfo{" + "firstSqn=" + firstSqn + ", lastSqn=" + lastSqn + '}';
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof RowSqnInfo)) {
+ return false;
+ }
+ RowSqnInfo that = (RowSqnInfo) o;
+ return firstSqn == that.firstSqn && lastSqn == that.lastSqn;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(firstSqn, lastSqn);
+ }
+
+ public RowSqnInfo withFirstSqn(long firstSqn) {
+ return RowSqnInfo.of(firstSqn, lastSqn);
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowSqnInfoSerializer.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowSqnInfoSerializer.java
new file mode 100644
index 00000000000..348071a22cd
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowSqnInfoSerializer.java
@@ -0,0 +1,101 @@
+/*
+ * 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.sequencedmultisetstate.linked;
+
+import org.apache.flink.api.common.typeutils.CompositeSerializer;
+import org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+
+@SuppressWarnings("ClassEscapesDefinedScope")
+public class RowSqnInfoSerializer extends CompositeSerializer<RowSqnInfo> {
+
+ public RowSqnInfoSerializer() {
+ this(null, LongSerializer.INSTANCE, LongSerializer.INSTANCE);
+ }
+
+ protected RowSqnInfoSerializer(
+ PrecomputedParameters precomputed, TypeSerializer<?>... fieldSerializers) {
+ super(
+ PrecomputedParameters.precompute(
+ true, true, (TypeSerializer<Object>[]) fieldSerializers),
+ fieldSerializers);
+ }
+
+ @Override
+ public RowSqnInfo createInstance(Object... values) {
+ return new RowSqnInfo((Long) values[0], (Long) values[1]);
+ }
+
+ @Override
+ protected void setField(RowSqnInfo sqnInfo, int index, Object fieldValue) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ protected Object getField(RowSqnInfo value, int index) {
+ switch (index) {
+ case 0:
+ return value.firstSqn;
+ case 1:
+ return value.lastSqn;
+ default:
+ throw new IllegalArgumentException("invalid index: " + index);
+ }
+ }
+
+ @Override
+ protected CompositeSerializer<RowSqnInfo> createSerializerInstance(
+ PrecomputedParameters precomputed, TypeSerializer<?>... originalSerializers) {
+ return new RowSqnInfoSerializer(precomputed, originalSerializers);
+ }
+
+ @Override
+ public TypeSerializerSnapshot<RowSqnInfo> snapshotConfiguration() {
+ return new RowSqnInfoSerializerSnapshot(this);
+ }
+
+ public static class RowSqnInfoSerializerSnapshot
+ extends CompositeTypeSerializerSnapshot<RowSqnInfo, RowSqnInfoSerializer> {
+
+ @SuppressWarnings("unused")
+ public RowSqnInfoSerializerSnapshot() {}
+
+ RowSqnInfoSerializerSnapshot(RowSqnInfoSerializer serializer) {
+ super(serializer);
+ }
+
+ @Override
+ protected int getCurrentOuterSnapshotVersion() {
+ return 0;
+ }
+
+ @Override
+ protected TypeSerializer<?>[] getNestedSerializers(RowSqnInfoSerializer outerSerializer) {
+ return new TypeSerializer[] {LongSerializer.INSTANCE, LongSerializer.INSTANCE};
+ }
+
+ @Override
+ protected RowSqnInfoSerializer createOuterSerializerWithNestedSerializers(
+ TypeSerializer<?>[] nestedSerializers) {
+ return new RowSqnInfoSerializer(null, nestedSerializers);
+ }
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/sequencedmultisetstate/SequencedMultiSetStateTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/sequencedmultisetstate/SequencedMultiSetStateTest.java
new file mode 100644
index 00000000000..f31db7c80b1
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/sequencedmultisetstate/SequencedMultiSetStateTest.java
@@ -0,0 +1,647 @@
+/*
+ * 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.sequencedmultisetstate;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.functions.SerializerFactory;
+import org.apache.flink.api.common.serialization.SerializerConfigImpl;
+import org.apache.flink.api.common.state.KeyedStateStore;
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.runtime.externalresource.ExternalResourceInfoProvider;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.operators.testutils.MockEnvironment;
+import org.apache.flink.runtime.operators.testutils.MockEnvironmentBuilder;
+import org.apache.flink.runtime.query.KvStateRegistry;
+import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
+import org.apache.flink.runtime.state.DefaultKeyedStateStore;
+import org.apache.flink.runtime.state.InternalKeyContext;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.KeyedStateBackendParametersImpl;
+import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
+import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
+import org.apache.flink.streaming.api.TimeDomain;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.utils.ProjectedRowData;
+import org.apache.flink.table.runtime.generated.GeneratedHashFunction;
+import org.apache.flink.table.runtime.generated.GeneratedRecordEqualiser;
+import org.apache.flink.table.runtime.generated.HashFunction;
+import org.apache.flink.table.runtime.generated.RecordEqualiser;
+import org.apache.flink.table.runtime.sequencedmultisetstate.SequencedMultiSetState.StateChangeInfo;
+import org.apache.flink.table.runtime.sequencedmultisetstate.SequencedMultiSetState.StateChangeType;
+import org.apache.flink.table.runtime.sequencedmultisetstate.SequencedMultiSetState.Strategy;
+import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.function.BiConsumerWithException;
+import org.apache.flink.util.function.ThrowingConsumer;
+import org.apache.flink.util.function.TriFunctionWithException;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.LongStream;
+
+import static org.apache.flink.table.runtime.sequencedmultisetstate.SequencedMultiSetState.StateChangeType.REMOVAL_ALL;
+import static org.apache.flink.table.runtime.sequencedmultisetstate.SequencedMultiSetState.StateChangeType.REMOVAL_LAST_ADDED;
+import static org.apache.flink.table.runtime.sequencedmultisetstate.SequencedMultiSetState.StateChangeType.REMOVAL_NOT_FOUND;
+import static org.apache.flink.table.runtime.sequencedmultisetstate.SequencedMultiSetState.StateChangeType.REMOVAL_OTHER;
+import static org.apache.flink.table.runtime.util.StreamRecordUtils.row;
+import static org.apache.flink.table.runtime.util.StreamRecordUtils.rowOfKind;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/** Test for various implementations of {@link SequencedMultiSetState}. */
+@SuppressWarnings({"SameParameterValue", "unused"})
+@ExtendWith(ParameterizedTestExtension.class)
+public class SequencedMultiSetStateTest {
+
+ @Parameter(0)
+ private Strategy strategy;
+
+ @Parameter(1)
+ private long adaptiveLowThresholdOverride;
+
+ @Parameter(2)
+ private long adaptiveHighThresholdOverride;
+
+ @Parameters(name = "strategy={0}, lowThreshold={1}, highThreshold={2}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ new Object[] {Strategy.VALUE_STATE, -1, -1},
+ new Object[] {Strategy.MAP_STATE, -1, -1},
+ new Object[] {Strategy.ADAPTIVE, 0, 1},
+ new Object[] {Strategy.ADAPTIVE, 1, 2},
+ new Object[] {Strategy.ADAPTIVE, 0, 10},
+ new Object[] {Strategy.ADAPTIVE, 9, 10},
+ };
+ }
+
+ // for simplicity, all tests use string type only, with row key being the 1st column
+ private static final LogicalType VARCHAR = DataTypes.VARCHAR(50).getLogicalType();
+ public static final int KEY_POS = 0;
+
+ @TestTemplate
+ public void testBasicFlow() throws Exception {
+ runTest(
+ (state, keyContext) -> {
+ keyContext.setCurrentKey("sk1");
+ assertTrue(state.isEmpty());
+
+ state.add(row("key", "value"), 1L);
+ assertFalse(state.isEmpty());
+
+ keyContext.setCurrentKey("sk2");
+ assertTrue(state.isEmpty());
+
+ keyContext.setCurrentKey("sk1");
+ state.clear();
+ assertStateContents(state);
+ });
+ }
+
+ @TestTemplate
+ public void testAppend() throws Exception {
+ runTest(
+ state -> {
+ // should always keep appending
+ state.append(row("k1", "x"), 777L);
+ assertStateContents(state, Tuple2.of(row("k1", "x"), 777L));
+
+ state.append(row("k1", "x"), 778L);
+ assertStateContents(
+ state,
+ Tuple2.of(row("k1", "x"), 777L),
+ Tuple2.of(row("k1", "x"), 778L));
+
+ state.append(row("k2", "y"), 779L);
+ assertStateContents(
+ state,
+ Tuple2.of(row("k1", "x"), 777L),
+ Tuple2.of(row("k1", "x"), 778L),
+ Tuple2.of(row("k2", "y"), 779L));
+
+ state.append(row("k1", "x"), 777L);
+ assertStateContents(
+ state,
+ Tuple2.of(row("k1", "x"), 777L),
+ Tuple2.of(row("k1", "x"), 778L),
+ Tuple2.of(row("k2", "y"), 779L),
+ Tuple2.of(row("k1", "x"), 777L));
+ });
+ }
+
+ @TestTemplate
+ public void testAdd() throws Exception {
+ runTest(
+ state -> {
+ state.add(row("k1", "x"), 777L);
+ assertStateContents(state, row("k1", "x"), 777L);
+
+ state.add(row("k2", "x"), 777L);
+ assertStateContents(
+ state,
+ Tuple2.of(row("k1", "x"), 777L),
+ Tuple2.of(row("k2", "x"), 777L));
+
+ state.add(row("k2", "y"), 778L);
+ assertStateContents(
+ state,
+ Tuple2.of(row("k1", "x"), 777L),
+ Tuple2.of(row("k2", "y"), 778L));
+
+ state.add(row("k1", "y"), 778L);
+ assertStateContents(
+ state,
+ Tuple2.of(row("k1", "y"), 778L),
+ Tuple2.of(row("k2", "y"), 778L));
+ });
+ }
+
+ @TestTemplate
+ public void testRemove() throws Exception {
+ runTest(
+ state -> {
+ removeAndAssert(state, row("key1"), REMOVAL_NOT_FOUND);
+
+ state.add(row("key1", "value"), 777L);
+ state.add(row("key2", "value"), 777L);
+ state.add(row("key3", "value"), 777L);
+ state.add(row("key4", "value"), 777L);
+
+ removeAndAssert(state, row("key999"), REMOVAL_NOT_FOUND);
+ removeAndAssert(state, row("key4"), REMOVAL_LAST_ADDED, row("key3", "value"));
+ removeAndAssert(state, row("key3"), REMOVAL_LAST_ADDED, row("key2", "value"));
+ removeAndAssert(state, row("key1"), REMOVAL_OTHER);
+ removeAndAssert(
+ state,
+ row("key2", "value-to-return"),
+ REMOVAL_ALL,
+ // value-to-return should be returned, not the original value
+ // according to the current logic of Flink operators
+ row("key2", "value-to-return"));
+
+ // shouldn't fail e.g. due to bad pointers
+ removeAndAssert(state, row("key1"), REMOVAL_NOT_FOUND);
+ removeAndAssert(state, row("key2"), REMOVAL_NOT_FOUND);
+ });
+ }
+
+ @TestTemplate
+ public void testAddAfterRemovingTail() throws Exception {
+ runTest(
+ state -> {
+ state.add(row("key1", "value-1"), 777L);
+ state.add(row("key2", "value-2"), 777L);
+
+ // key1 is the tail now - remove it and then add
+ removeAndAssert(state, row("key1"), REMOVAL_OTHER, row("key1", "value-1"));
+ state.add(row("key1", "value-1"), 777L);
+
+ // key2 is the tail now - remove it and then add
+ removeAndAssert(state, row("key2"), REMOVAL_OTHER, row("key2", "value-2"));
+ state.add(row("key2", "value-2"), 777L);
+ });
+ }
+
+ @TestTemplate
+ public void testRemoveFirstAppended() throws Exception {
+ runTest(
+ state -> {
+ state.append(row("key", "value-1"), 777L);
+ state.append(row("key", "value-2"), 778L);
+ state.append(row("key", "value-3"), 779L);
+
+ removeAndAssert(state, row("key"), REMOVAL_OTHER);
+ assertStateContents(
+ state,
+ Tuple2.of(row("key", "value-2"), 778L),
+ Tuple2.of(row("key", "value-3"), 779L));
+
+ removeAndAssert(state, row("key"), REMOVAL_OTHER);
+ assertStateContents(state, Tuple2.of(row("key", "value-3"), 779L));
+
+ removeAndAssert(state, row("key"), REMOVAL_ALL, row("key"));
+ assertTrue(state.isEmpty());
+ });
+ }
+
+ @TestTemplate
+ public void testRemoveWithInterleavingRowAppended() throws Exception {
+ runTest(
+ state -> {
+ state.append(row("key1", "value"), 777L); // sqn = 1
+ state.append(row("key2", "value"), 777L); // sqn = 2
+ state.append(row("key2", "value"), 778L); // sqn = 3
+ removeAndAssert(state, row("key2"), REMOVAL_OTHER, row("key2", "value"));
+ removeAndAssert(state, row("key2"), REMOVAL_LAST_ADDED, row("key1", "value"));
+ removeAndAssert(
+ state,
+ row("key1", "value-to-return"),
+ REMOVAL_ALL,
+ row("key1", "value-to-return"));
+ });
+ }
+
+ /** Test that loading and clearing the cache doesn't impact correctness. */
+ @TestTemplate
+ public void testCaching() throws Exception {
+ runTest(
+ (state, ctx) -> {
+ ctx.setCurrentKey("sk1");
+ state.add(row("key", "value-1"), 777L);
+ state.clearCache();
+ assertFalse(state.isEmpty());
+
+ ctx.setCurrentKey("sk2");
+ state.loadCache();
+ assertTrue(state.isEmpty());
+ });
+ }
+
+ /** Test that loading and clearing the cache doesn't impact correctness. */
+ @TestTemplate
+ public void testKeyExtraction() throws Exception {
+ final Function<RowData, RowData> keyExtractor =
+ row -> ProjectedRowData.from(new int[] {1}).replaceRow(row);
+
+ runTest(
+ (state, ctx) -> {
+ ctx.setCurrentKey("sk1");
+ state.add(row("value-123", "key"), 777L);
+ assertFalse(state.isEmpty());
+ StateChangeInfo<RowData> ret = state.remove(row("value-456", "key"));
+ Tuple3.of(ret.getSizeAfter(), ret.getChangeType(), ret.getPayload());
+ assertTrue(state.isEmpty());
+ },
+ keyExtractor,
+ 0);
+ }
+
+ /** Test that row kind is not taken into account when matching the rows. */
+ @TestTemplate
+ public void testRowKindNormalization() throws Exception {
+ runTest(
+ state -> {
+ for (RowKind firstKind : RowKind.values()) {
+ for (RowKind secondKind : RowKind.values()) {
+
+ state.append(rowOfKind(firstKind, "key", "value"), 778L);
+ state.remove(rowOfKind(secondKind, "key", "value"));
+ assertTrue(state.isEmpty());
+
+ state.add(rowOfKind(firstKind, "key", "value"), 777L);
+ state.remove(rowOfKind(secondKind, "key", "value"));
+ assertTrue(state.isEmpty());
+
+ state.add(rowOfKind(firstKind, "key", "value"), 777L);
+ state.add(rowOfKind(secondKind, "key", "value"), 778L);
+ assertStateContents(state, Tuple2.of(row("key", "value"), 778L));
+ state.clear();
+ }
+ }
+ });
+ }
+
+ @TestTemplate
+ public void testAdaptivity() throws Exception {
+ assumeTrue(strategy == Strategy.ADAPTIVE);
+ final long totalSize = adaptiveHighThresholdOverride * 2;
+ runTest(
+ (state, ctx) -> {
+ AdaptiveSequencedMultiSetState ad = (AdaptiveSequencedMultiSetState) state;
+ int runningSize = 0;
+
+ ctx.setCurrentKey("k1");
+ assertFalse(ad.isIsUsingLargeState(), "should start with value state");
+ for (; runningSize < totalSize; runningSize++) {
+ assertEquals(
+ runningSize >= adaptiveHighThresholdOverride,
+ ad.isIsUsingLargeState(),
+ "should switch after reaching high threshold");
+ ad.append(row("key", "value"), runningSize /* timestamp */);
+ }
+
+ ctx.setCurrentKey("k2");
+ assertFalse(ad.isIsUsingLargeState(), "should not mix different context keys");
+
+ ctx.setCurrentKey("k1");
+ assertTrue(ad.isIsUsingLargeState(), "should not mix different context keys");
+
+ // remove until hitting the threshold - shouldn't trigger switch
+ for (; runningSize > adaptiveLowThresholdOverride + 1; runningSize--) {
+ ad.remove(row("key"));
+ assertTrue(
+ ad.isIsUsingLargeState(),
+ "should switch back after reaching low threshold");
+ }
+ // trigger switch
+ ad.remove(row("key"));
+ runningSize--;
+ assertFalse(
+ ad.isIsUsingLargeState(),
+ "should switch back after reaching low threshold");
+ // verify the order of the migrated elements by looking at their timestamps
+ //noinspection unchecked
+ assertStateContents(
+ state,
+ LongStream.range(totalSize - runningSize, totalSize)
+ .mapToObj(ts -> Tuple2.of(row("key", "value"), ts))
+ .toArray(Tuple2[]::new));
+ for (; runningSize > 0; runningSize--) {
+ assertFalse(
+ ad.isIsUsingLargeState(),
+ "should switch back after reaching low threshold");
+ ad.remove(row("key"));
+ }
+ assertTrue(ad.isEmpty());
+ assertEquals(0, runningSize);
+
+ for (; runningSize < totalSize; runningSize++) {
+ assertEquals(
+ runningSize >= adaptiveHighThresholdOverride,
+ ad.isIsUsingLargeState(),
+ "should switch after reaching high threshold");
+ ad.add(row(Integer.toString(runningSize), "value"), 777L);
+ }
+ assertTrue(ad.isIsUsingLargeState());
+
+ state.clear();
+ assertFalse(
+ ad.isIsUsingLargeState(), "should switch to value state after clear");
+ });
+ }
+
+ @TestTemplate
+ public void testAddReturnValues() throws Exception {
+ testReturnValues(SequencedMultiSetState::add);
+ }
+
+ @TestTemplate
+ public void testAppendReturnValues() throws Exception {
+ testReturnValues(SequencedMultiSetState::append);
+ }
+
+ private void testReturnValues(
+ TriFunctionWithException<
+ SequencedMultiSetState<RowData>,
+ RowData,
+ Long,
+ StateChangeInfo<RowData>,
+ Exception>
+ updateFn)
+ throws Exception {
+ runTest(
+ state -> {
+ StateChangeInfo<RowData> ret;
+
+ ret = updateFn.apply(state, row("key-1", "value"), 777L);
+ assertEquals(StateChangeType.ADDITION, ret.getChangeType());
+ assertEquals(1, ret.getSizeAfter());
+ assertTrue(ret.wasEmpty());
+
+ ret = updateFn.apply(state, row("key-2", "value"), 777L);
+ assertEquals(StateChangeType.ADDITION, ret.getChangeType());
+ assertEquals(2, ret.getSizeAfter());
+ assertFalse(ret.wasEmpty());
+
+ removeAndAssert(state, row("key-1"), REMOVAL_OTHER);
+ removeAndAssert(state, row("key-2"), REMOVAL_ALL, row("key-2"));
+
+ ret = updateFn.apply(state, row("key-3", "value"), 777L);
+ assertEquals(StateChangeType.ADDITION, ret.getChangeType());
+ assertEquals(1, ret.getSizeAfter());
+ assertTrue(ret.wasEmpty());
+ });
+ }
+
+ private void runTest(ThrowingConsumer<SequencedMultiSetState<RowData>, Exception> test)
+ throws Exception {
+ runTest(
+ (state, keyContext) -> {
+ keyContext.setCurrentKey("key1");
+ test.accept(state);
+ });
+ }
+
+ private void runTest(
+ BiConsumerWithException<
+ SequencedMultiSetState<RowData>, InternalKeyContext<String>, Exception>
+ test)
+ throws Exception {
+ runTest(test, Function.identity(), KEY_POS);
+ }
+
+ private void runTest(
+ BiConsumerWithException<
+ SequencedMultiSetState<RowData>, InternalKeyContext<String>, Exception>
+ test,
+ Function<RowData, RowData> keyExtractor,
+ int keyPos)
+ throws Exception {
+ SequencedMultiSetStateContext p =
+ new SequencedMultiSetStateContext(
+ new RowDataSerializer(VARCHAR),
+ new MyGeneratedEqualiser(keyPos),
+ new MyGeneratedHashFunction(keyPos),
+ new RowDataSerializer(VARCHAR, VARCHAR),
+ keyExtractor,
+ new SequencedMultiSetStateConfig(
+ strategy,
+ adaptiveHighThresholdOverride,
+ adaptiveLowThresholdOverride,
+ StateTtlConfig.DISABLED,
+ TimeDomain.EVENT_TIME));
+
+ MockEnvironment env = new MockEnvironmentBuilder().build();
+
+ AbstractKeyedStateBackend<String> stateBackend =
+ getKeyedStateBackend(env, StringSerializer.INSTANCE);
+
+ RuntimeContext ctx =
+ new StreamingRuntimeContext(
+ env,
+ Collections.emptyMap(),
+ UnregisteredMetricsGroup.createOperatorMetricGroup(),
+ new OperatorID(),
+ new TestProcessingTimeService(),
+ getKeyedStateStore(stateBackend),
+ ExternalResourceInfoProvider.NO_EXTERNAL_RESOURCES);
+
+ test.accept(SequencedMultiSetState.create(p, ctx, "hashmap"), stateBackend);
+ }
+
+ private static KeyedStateStore getKeyedStateStore(KeyedStateBackend<String> stateBackend) {
+ return new DefaultKeyedStateStore(
+ stateBackend,
+ new SerializerFactory() {
+ @Override
+ public <T> TypeSerializer<T> createSerializer(TypeInformation<T> ti) {
+ return ti.createSerializer(new SerializerConfigImpl());
+ }
+ });
+ }
+
+ private static <T> AbstractKeyedStateBackend<T> getKeyedStateBackend(
+ MockEnvironment env, TypeSerializer<T> keySerializer) throws IOException {
+ String op = "test-operator";
+ JobID jobId = new JobID();
+ JobVertexID jobVertexId = new JobVertexID();
+ KeyGroupRange emptyKeyGroupRange = KeyGroupRange.of(0, 10);
+ int numberOfKeyGroups = emptyKeyGroupRange.getNumberOfKeyGroups();
+
+ return new HashMapStateBackend()
+ .createKeyedStateBackend(
+ new KeyedStateBackendParametersImpl<>(
+ env,
+ jobId,
+ op,
+ keySerializer,
+ numberOfKeyGroups,
+ emptyKeyGroupRange,
+ new KvStateRegistry().createTaskRegistry(jobId, jobVertexId),
+ TtlTimeProvider.DEFAULT,
+ new UnregisteredMetricsGroup(),
+ Collections.emptyList(),
+ new CloseableRegistry()));
+ }
+
+ private static class MyGeneratedEqualiser extends GeneratedRecordEqualiser {
+
+ private final int keyPos;
+
+ public MyGeneratedEqualiser(int keyPos) {
+ super("", "", new Object[0]);
+ this.keyPos = keyPos;
+ }
+
+ @Override
+ public RecordEqualiser newInstance(ClassLoader classLoader) {
+ return new TestRecordEqualiser(keyPos);
+ }
+ }
+
+ private static class MyGeneratedHashFunction extends GeneratedHashFunction {
+
+ private final int keyPos;
+
+ public MyGeneratedHashFunction(int keyPos) {
+ super("", "", new Object[0], new Configuration());
+ this.keyPos = keyPos;
+ }
+
+ @Override
+ public HashFunction newInstance(ClassLoader classLoader) {
+ return new TestRecordEqualiser(keyPos);
+ }
+ }
+
+ private static class TestRecordEqualiser implements RecordEqualiser, HashFunction {
+
+ private final int keyPos;
+
+ private TestRecordEqualiser(int keyPos) {
+ this.keyPos = keyPos;
+ }
+
+ @Override
+ public boolean equals(RowData row1, RowData row2) {
+ return row1.getRowKind() == row2.getRowKind()
+ && row1.getString(keyPos).equals(row2.getString(keyPos));
+ }
+
+ @Override
+ public int hashCode(Object data) {
+ RowData rd = (RowData) data;
+ return Objects.hash(rd.getRowKind(), rd.getString(keyPos));
+ }
+ }
+
+ private static void assertStateContents(
+ SequencedMultiSetState<RowData> state, RowData rowData, Long timestamp)
+ throws Exception {
+ assertStateContents(state, Tuple2.of(rowData, timestamp));
+ }
+
+ @SafeVarargs
+ private static void assertStateContents(
+ SequencedMultiSetState<RowData> state, Tuple2<RowData, Long>... expectedArr)
+ throws Exception {
+ List<Tuple2<RowData, Long>> actual = new ArrayList<>();
+ state.iterator().forEachRemaining(actual::add);
+ assertEquals(expectedArr.length == 0, state.isEmpty());
+ assertEquals(expectedArr.length, actual.size());
+ Assertions.assertArrayEquals(expectedArr, actual.toArray());
+ }
+
+ private static void removeAndAssert(
+ SequencedMultiSetState<RowData> state,
+ RowData key,
+ StateChangeType expectedResultType,
+ RowData... expectedReturnedRow)
+ throws Exception {
+ StateChangeInfo<RowData> ret = state.remove(key);
+
+ assertEquals(expectedResultType, ret.getChangeType());
+ switch (ret.getChangeType()) {
+ case REMOVAL_NOT_FOUND:
+ assertEquals(Optional.empty(), ret.getPayload());
+ break;
+ case REMOVAL_ALL:
+ assertEquals(0, ret.getSizeAfter());
+ assertTrue(state.isEmpty(), "state is expected to be empty");
+ assertEquals(Optional.of(expectedReturnedRow[0]), ret.getPayload());
+ break;
+ case REMOVAL_OTHER:
+ assertFalse(state.isEmpty(), "state is expected to be non-empty");
+ assertEquals(Optional.empty(), ret.getPayload());
+ break;
+ case REMOVAL_LAST_ADDED:
+ assertFalse(state.isEmpty(), "state is expected to be non-empty");
+ assertEquals(Optional.of(expectedReturnedRow[0]), ret.getPayload());
+ break;
+ }
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/MetaSqnInfoSerializerTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/MetaSqnInfoSerializerTest.java
new file mode 100644
index 00000000000..92bf9d29aaf
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/MetaSqnInfoSerializerTest.java
@@ -0,0 +1,46 @@
+/*
+ * 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.sequencedmultisetstate.linked;
+
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+
+/** Test for {@link MetaSqnInfoSerializer}. */
+class MetaSqnInfoSerializerTest extends SerializerTestBase<MetaSqnInfo> {
+
+ @Override
+ protected TypeSerializer<MetaSqnInfo> createSerializer() {
+ return new MetaSqnInfoSerializer();
+ }
+
+ @Override
+ protected int getLength() {
+ return 2 * Long.SIZE / 8;
+ }
+
+ @Override
+ protected Class<MetaSqnInfo> getTypeClass() {
+ return MetaSqnInfo.class;
+ }
+
+ @Override
+ protected MetaSqnInfo[] getTestData() {
+ return new MetaSqnInfo[] {MetaSqnInfo.of(1L, 2L), MetaSqnInfo.of(1L, 1L)};
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/NodeSerializerTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/NodeSerializerTest.java
new file mode 100644
index 00000000000..1dbdaf3b355
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/NodeSerializerTest.java
@@ -0,0 +1,53 @@
+/*
+ * 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.sequencedmultisetstate.linked;
+
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
+import org.apache.flink.table.runtime.util.StreamRecordUtils;
+import org.apache.flink.table.types.logical.IntType;
+
+/** Test for {@link RowDataKeySerializer}. */
+public class NodeSerializerTest extends SerializerTestBase<Node> {
+
+ @Override
+ protected TypeSerializer<Node> createSerializer() {
+ return new NodeSerializer(new RowDataSerializer(new IntType()));
+ }
+
+ @Override
+ protected int getLength() {
+ return -1;
+ }
+
+ @Override
+ protected Class<Node> getTypeClass() {
+ return Node.class;
+ }
+
+ @Override
+ protected Node[] getTestData() {
+ return new Node[] {
+ new Node(StreamRecordUtils.row(1), 1L, null, 2L, 2L, 1L),
+ new Node(StreamRecordUtils.row(2), 2L, 1L, 3L, 3L, 2L),
+ new Node(StreamRecordUtils.row(3), 3L, 2L, null, null, 3L),
+ };
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowDataKeySerializerTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowDataKeySerializerTest.java
new file mode 100644
index 00000000000..f78faf9f343
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowDataKeySerializerTest.java
@@ -0,0 +1,104 @@
+/*
+ * 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.sequencedmultisetstate.linked;
+
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.GeneratedHashFunction;
+import org.apache.flink.table.runtime.generated.GeneratedRecordEqualiser;
+import org.apache.flink.table.runtime.generated.HashFunction;
+import org.apache.flink.table.runtime.generated.RecordEqualiser;
+import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
+import org.apache.flink.table.runtime.util.StreamRecordUtils;
+import org.apache.flink.table.types.logical.IntType;
+
+import java.util.Objects;
+
+/** Test for {@link RowDataKeySerializer}. */
+public class RowDataKeySerializerTest extends SerializerTestBase<RowDataKey> {
+
+ private final TestRecordEqualiser equaliser = new TestRecordEqualiser();
+
+ @Override
+ protected TypeSerializer<RowDataKey> createSerializer() {
+ return new RowDataKeySerializer(
+ new RowDataSerializer(new IntType()),
+ equaliser,
+ equaliser,
+ EQUALISER,
+ HASH_FUNCTION);
+ }
+
+ @Override
+ protected int getLength() {
+ return -1;
+ }
+
+ @Override
+ protected Class<RowDataKey> getTypeClass() {
+ return RowDataKey.class;
+ }
+
+ @Override
+ protected RowDataKey[] getTestData() {
+ return new RowDataKey[] {new RowDataKey(StreamRecordUtils.row(123), equaliser, equaliser)};
+ }
+
+ static final GeneratedRecordEqualiser EQUALISER =
+ new GeneratedRecordEqualiser("", "", new Object[0]) {
+
+ @Override
+ public RecordEqualiser newInstance(ClassLoader classLoader) {
+ return new TestRecordEqualiser();
+ }
+ };
+
+ static final GeneratedHashFunction HASH_FUNCTION =
+ new GeneratedHashFunction("", "", new Object[0], new Configuration()) {
+ @Override
+ public HashFunction newInstance(ClassLoader classLoader) {
+ return new TestRecordEqualiser();
+ }
+ };
+
+ private static class TestRecordEqualiser implements RecordEqualiser, HashFunction {
+ @Override
+ public boolean equals(RowData row1, RowData row2) {
+ return row1.getRowKind() == row2.getRowKind() && row1.getInt(0) == row2.getInt(0);
+ }
+
+ @Override
+ public int hashCode(Object data) {
+ RowData rd = (RowData) data;
+ return Objects.hash(rd.getRowKind(), rd.getInt(0));
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj instanceof TestRecordEqualiser;
+ }
+
+ @Override
+ public int hashCode() {
+ return 0;
+ }
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowSqnInfoSerializerTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowSqnInfoSerializerTest.java
new file mode 100644
index 00000000000..fd584c3862f
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/sequencedmultisetstate/linked/RowSqnInfoSerializerTest.java
@@ -0,0 +1,46 @@
+/*
+ * 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.sequencedmultisetstate.linked;
+
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+
+/** Test for {@link RowSqnInfoSerializer}. */
+class RowSqnInfoSerializerTest extends SerializerTestBase<RowSqnInfo> {
+
+ @Override
+ protected TypeSerializer<RowSqnInfo> createSerializer() {
+ return new RowSqnInfoSerializer();
+ }
+
+ @Override
+ protected int getLength() {
+ return 2 * Long.SIZE / 8;
+ }
+
+ @Override
+ protected Class<RowSqnInfo> getTypeClass() {
+ return RowSqnInfo.class;
+ }
+
+ @Override
+ protected RowSqnInfo[] getTestData() {
+ return new RowSqnInfo[] {RowSqnInfo.of(1L, 2L), RowSqnInfo.of(1L, 1L)};
+ }
+}
diff --git a/flink-tests/pom.xml b/flink-tests/pom.xml
index 1988f8e63b5..e4e9de6e3dd 100644
--- a/flink-tests/pom.xml
+++ b/flink-tests/pom.xml
@@ -193,6 +193,14 @@ under the License.
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-runtime</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-runtime</artifactId>
diff --git a/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java b/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java
index 691460a4012..976cef6349b 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java
@@ -74,6 +74,10 @@ import org.apache.flink.table.dataview.MapViewSerializer;
import org.apache.flink.table.dataview.NullAwareMapSerializer;
import org.apache.flink.table.dataview.NullSerializer;
import org.apache.flink.table.runtime.operators.window.CountWindow;
+import org.apache.flink.table.runtime.sequencedmultisetstate.linked.MetaSqnInfoSerializer;
+import org.apache.flink.table.runtime.sequencedmultisetstate.linked.NodeSerializer;
+import org.apache.flink.table.runtime.sequencedmultisetstate.linked.RowDataKeySerializer;
+import org.apache.flink.table.runtime.sequencedmultisetstate.linked.RowSqnInfoSerializer;
import org.apache.flink.table.runtime.typeutils.ArrayDataSerializer;
import org.apache.flink.table.runtime.typeutils.BinaryRowDataSerializer;
import org.apache.flink.table.runtime.typeutils.DecimalDataSerializer;
@@ -255,6 +259,10 @@ public class TypeSerializerTestCoverageTest extends TestLogger {
// KeyAndValueSerializer shouldn't be used to serialize data to state and
// doesn't need to ensure upgrade compatibility.
"org.apache.flink.streaming.api.operators.sortpartition.KeyAndValueSerializer",
+ RowDataKeySerializer.class.getName(),
+ NodeSerializer.class.getName(),
+ RowSqnInfoSerializer.class.getName(),
+ MetaSqnInfoSerializer.class.getName(),
SetSerializer.class.getName());
// check if a test exists for each type serializer