You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by za...@apache.org on 2024/12/10 03:28:48 UTC
(flink) branch master updated (b32181bb95d -> e016051e2c1)
This is an automated email from the ASF dual-hosted git repository.
zakelly pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from b32181bb95d [FLINK-36451][runtime] Makes hasLeadership and confirmLeadership work asynchronously
new d2826649370 [hotfix][state] Make state iterator adaptor tolerate null iterable input
new e016051e2c1 [FLINK-36860][State/v2] Make internal list state inherit the internal merging state
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../asyncprocessing/AsyncExecutionController.java | 20 ++--
.../runtime/asyncprocessing/StateRequestType.java | 5 +-
.../flink/runtime/state/v2/AbstractListState.java | 13 +++
.../state/v2/adaptor/CompleteStateIterator.java | 13 ++-
.../runtime/state/v2/adaptor/ListStateAdaptor.java | 20 ++++
.../v2/internal/InternalAggregatingState.java | 2 +-
.../state/v2/internal/InternalAppendingState.java | 31 +-----
.../state/v2/internal/InternalListState.java | 4 +-
.../state/v2/internal/InternalReducingState.java | 4 +-
.../state/v2/internal/InternalStateAccessible.java | 33 +++---
.../flink/runtime/state/v2/ttl/TtlListState.java | 10 ++
.../forst/ForStDBMultiRawMergePutRequest.java | 57 ++++++++++
...leGetRequest.java => ForStDBRawGetRequest.java} | 17 ++-
.../apache/flink/state/forst/ForStListState.java | 115 ++++++++++++++++++++-
.../state/forst/ForStStateRequestClassifier.java | 32 ++++++
.../flink/state/forst/ForStStateRequestType.java | 15 +--
.../flink/state/forst/ForStListStateTest.java | 101 ++++++++++++++++++
.../flink/state/forst/ForStStateTestBase.java | 110 ++++++++++++++++++++
18 files changed, 521 insertions(+), 81 deletions(-)
copy flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/CacheLimitPolicy.java => flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalStateAccessible.java (57%)
create mode 100644 flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBMultiRawMergePutRequest.java
copy flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/{ForStDBSingleGetRequest.java => ForStDBRawGetRequest.java} (75%)
copy flink-formats/flink-compress/src/main/java/org/apache/flink/formats/compress/extractor/Extractor.java => flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateRequestType.java (71%)
create mode 100644 flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStListStateTest.java
create mode 100644 flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateTestBase.java
(flink) 01/02: [hotfix][state] Make state iterator adaptor tolerate null iterable input
Posted by za...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zakelly pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d28266493709750af49d011bf4cef7c34c86b41d
Author: Zakelly <za...@gmail.com>
AuthorDate: Sat Dec 7 12:22:03 2024 +0800
[hotfix][state] Make state iterator adaptor tolerate null iterable input
---
.../runtime/state/v2/adaptor/CompleteStateIterator.java | 13 ++++++++++---
1 file changed, 10 insertions(+), 3 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/adaptor/CompleteStateIterator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/adaptor/CompleteStateIterator.java
index bf2f3f23a09..11bd86b6937 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/adaptor/CompleteStateIterator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/adaptor/CompleteStateIterator.java
@@ -22,6 +22,8 @@ import org.apache.flink.api.common.state.v2.StateFuture;
import org.apache.flink.api.common.state.v2.StateIterator;
import org.apache.flink.core.state.StateFutureUtils;
+import javax.annotation.Nullable;
+
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -35,9 +37,14 @@ public class CompleteStateIterator<T> implements StateIterator<T> {
final Iterator<T> iterator;
final boolean empty;
- public CompleteStateIterator(Iterable<T> iterable) {
- this.iterator = iterable.iterator();
- this.empty = !iterator.hasNext();
+ public CompleteStateIterator(@Nullable Iterable<T> iterable) {
+ if (iterable == null) {
+ this.iterator = Collections.emptyIterator();
+ this.empty = true;
+ } else {
+ this.iterator = iterable.iterator();
+ this.empty = !iterator.hasNext();
+ }
}
@Override
(flink) 02/02: [FLINK-36860][State/v2] Make internal list state inherit the internal merging state
Posted by za...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zakelly pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e016051e2c18227bb700e292c10c3a5fec25a588
Author: Zakelly <za...@gmail.com>
AuthorDate: Sat Dec 7 12:25:12 2024 +0800
[FLINK-36860][State/v2] Make internal list state inherit the internal merging state
---
.../asyncprocessing/AsyncExecutionController.java | 20 ++--
.../runtime/asyncprocessing/StateRequestType.java | 5 +-
.../flink/runtime/state/v2/AbstractListState.java | 13 +++
.../runtime/state/v2/adaptor/ListStateAdaptor.java | 20 ++++
.../v2/internal/InternalAggregatingState.java | 2 +-
.../state/v2/internal/InternalAppendingState.java | 31 +-----
.../state/v2/internal/InternalListState.java | 4 +-
.../state/v2/internal/InternalReducingState.java | 4 +-
...ListState.java => InternalStateAccessible.java} | 41 ++++++--
.../flink/runtime/state/v2/ttl/TtlListState.java | 10 ++
.../forst/ForStDBMultiRawMergePutRequest.java | 57 ++++++++++
.../flink/state/forst/ForStDBRawGetRequest.java | 32 ++++--
.../apache/flink/state/forst/ForStListState.java | 115 ++++++++++++++++++++-
.../state/forst/ForStStateRequestClassifier.java | 32 ++++++
.../flink/state/forst/ForStStateRequestType.java | 21 ++--
.../flink/state/forst/ForStListStateTest.java | 101 ++++++++++++++++++
.../flink/state/forst/ForStStateTestBase.java | 110 ++++++++++++++++++++
17 files changed, 542 insertions(+), 76 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java
index 3af58543dcf..19aee130b51 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java
@@ -291,13 +291,21 @@ public class AsyncExecutionController<K> implements StateRequestHandler, Closeab
public <IN, OUT> OUT handleRequestSync(
State state, StateRequestType type, @Nullable IN payload) {
InternalStateFuture<OUT> stateFuture = handleRequest(state, type, payload);
- while (!stateFuture.isDone()) {
- try {
- mailboxExecutor.yield();
- } catch (InterruptedException e) {
- LOG.warn("Error while waiting for state future to complete.", e);
- throw new RuntimeException("Error while waiting for state future to complete.", e);
+ // Trigger since we are waiting the result.
+ triggerIfNeeded(true);
+ try {
+ while (!stateFuture.isDone()) {
+ if (!mailboxExecutor.tryYield()) {
+ // We force trigger the buffer if the executor is not fully loaded.
+ if (!stateExecutor.fullyLoaded()) {
+ triggerIfNeeded(true);
+ }
+ waitForNewMails();
+ }
}
+ } catch (InterruptedException ignored) {
+ // ignore the interrupted exception to avoid throwing fatal error when the task cancel
+ // or exit.
}
return stateFuture.get();
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestType.java b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestType.java
index 504115a48fa..9ef2d167c3f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestType.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestType.java
@@ -112,5 +112,8 @@ public enum StateRequestType {
AGGREGATING_GET,
/** Add element to aggregating state by {@link AggregatingState#asyncAdd(Object)}. */
- AGGREGATING_ADD
+ AGGREGATING_ADD,
+
+ /** Defined by different state backends. */
+ CUSTOMIZED
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractListState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractListState.java
index 8631f954237..5c7a6ba6bf3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractListState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/AbstractListState.java
@@ -24,6 +24,7 @@ import org.apache.flink.runtime.asyncprocessing.StateRequestHandler;
import org.apache.flink.runtime.asyncprocessing.StateRequestType;
import org.apache.flink.runtime.state.v2.internal.InternalListState;
+import java.util.Collection;
import java.util.List;
/**
@@ -83,4 +84,16 @@ public class AbstractListState<K, N, V> extends AbstractKeyedState<K, N, V>
public void addAll(List<V> values) {
handleRequestSync(StateRequestType.LIST_ADD_ALL, values);
}
+
+ @Override
+ public StateFuture<Void> asyncMergeNamespaces(N target, Collection<N> sources) {
+ throw new UnsupportedOperationException(
+ getClass() + " has not implement the asyncMergeNamespaces().");
+ }
+
+ @Override
+ public void mergeNamespaces(N target, Collection<N> sources) {
+ throw new UnsupportedOperationException(
+ getClass() + " has not implement the mergeNamespaces().");
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/adaptor/ListStateAdaptor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/adaptor/ListStateAdaptor.java
index 7b1b494eefd..df76566ce51 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/adaptor/ListStateAdaptor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/adaptor/ListStateAdaptor.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.state.v2.StateIterator;
import org.apache.flink.core.state.StateFutureUtils;
import org.apache.flink.runtime.state.v2.internal.InternalListState;
+import java.util.Collection;
import java.util.List;
/**
@@ -114,4 +115,23 @@ public class ListStateAdaptor<K, N, V>
throw new RuntimeException("Error while adding value to raw ListState", e);
}
}
+
+ @Override
+ public StateFuture<Void> asyncMergeNamespaces(N target, Collection<N> sources) {
+ try {
+ delegatedState.mergeNamespaces(target, sources);
+ } catch (Exception e) {
+ throw new RuntimeException("Error while merging namespaces in raw ListState", e);
+ }
+ return StateFutureUtils.completedVoidFuture();
+ }
+
+ @Override
+ public void mergeNamespaces(N target, Collection<N> sources) {
+ try {
+ delegatedState.mergeNamespaces(target, sources);
+ } catch (Exception e) {
+ throw new RuntimeException("Error while merging namespaces in raw ListState", e);
+ }
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalAggregatingState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalAggregatingState.java
index bc469825d85..39e4a4bffc6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalAggregatingState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalAggregatingState.java
@@ -31,4 +31,4 @@ import org.apache.flink.api.common.state.v2.AggregatingState;
public interface InternalAggregatingState<K, N, IN, ACC, OUT>
extends InternalMergingState<K, N, IN, ACC, OUT, OUT>,
AggregatingState<IN, OUT>,
- InternalKeyedState<K, N, ACC> {}
+ InternalStateAccessible<ACC> {}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalAppendingState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalAppendingState.java
index 7e938c6fbe9..6b9b323d8fe 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalAppendingState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalAppendingState.java
@@ -18,7 +18,6 @@
package org.apache.flink.runtime.state.v2.internal;
import org.apache.flink.api.common.state.v2.AppendingState;
-import org.apache.flink.api.common.state.v2.StateFuture;
/**
* This class defines the internal interface for appending state.
@@ -31,32 +30,4 @@ import org.apache.flink.api.common.state.v2.StateFuture;
* @param <SYNCOUT> Type of the value that can be retrieved from the state by synchronous interface.
*/
public interface InternalAppendingState<K, N, IN, SV, OUT, SYNCOUT>
- extends InternalKeyedState<K, N, SV>, AppendingState<IN, OUT, SYNCOUT> {
- /**
- * Get internally stored value.
- *
- * @return internally stored value.
- */
- StateFuture<SV> asyncGetInternal();
-
- /**
- * Update internally stored value.
- *
- * @param valueToStore new value to store.
- */
- StateFuture<Void> asyncUpdateInternal(SV valueToStore);
-
- /**
- * Get internally stored value.
- *
- * @return internally stored value.
- */
- SV getInternal();
-
- /**
- * Update internally stored value.
- *
- * @param valueToStore new value to store.
- */
- void updateInternal(SV valueToStore);
-}
+ extends InternalKeyedState<K, N, SV>, AppendingState<IN, OUT, SYNCOUT> {}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalListState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalListState.java
index efeccc7539a..7c0e0272392 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalListState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalListState.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.state.v2.internal;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.state.v2.ListState;
+import org.apache.flink.api.common.state.v2.StateIterator;
/**
* This class defines the internal interface for list state.
@@ -29,4 +30,5 @@ import org.apache.flink.api.common.state.v2.ListState;
* @param <V> The type of the intermediate state.
*/
@Internal
-public interface InternalListState<K, N, V> extends InternalKeyedState<K, N, V>, ListState<V> {}
+public interface InternalListState<K, N, V>
+ extends InternalMergingState<K, N, V, V, StateIterator<V>, Iterable<V>>, ListState<V> {}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalReducingState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalReducingState.java
index bdebbfc2b2d..43e2eea794f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalReducingState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalReducingState.java
@@ -27,4 +27,6 @@ import org.apache.flink.api.common.state.v2.ReducingState;
* @param <T> Type of the value in the operator state.
*/
public interface InternalReducingState<K, N, T>
- extends InternalAggregatingState<K, N, T, T, T>, ReducingState<T> {}
+ extends InternalAggregatingState<K, N, T, T, T>,
+ ReducingState<T>,
+ InternalStateAccessible<T> {}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalListState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalStateAccessible.java
similarity index 55%
copy from flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalListState.java
copy to flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalStateAccessible.java
index efeccc7539a..376227436e9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalListState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalStateAccessible.java
@@ -18,15 +18,34 @@
package org.apache.flink.runtime.state.v2.internal;
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.state.v2.ListState;
+import org.apache.flink.api.common.state.v2.StateFuture;
-/**
- * This class defines the internal interface for list state.
- *
- * @param <K> The type of key the state is associated to.
- * @param <N> The namespace type.
- * @param <V> The type of the intermediate state.
- */
-@Internal
-public interface InternalListState<K, N, V> extends InternalKeyedState<K, N, V>, ListState<V> {}
+public interface InternalStateAccessible<SV> {
+ /**
+ * Get internally stored value.
+ *
+ * @return internally stored value.
+ */
+ StateFuture<SV> asyncGetInternal();
+
+ /**
+ * Update internally stored value.
+ *
+ * @param valueToStore new value to store.
+ */
+ StateFuture<Void> asyncUpdateInternal(SV valueToStore);
+
+ /**
+ * Get internally stored value.
+ *
+ * @return internally stored value.
+ */
+ SV getInternal();
+
+ /**
+ * Update internally stored value.
+ *
+ * @param valueToStore new value to store.
+ */
+ void updateInternal(SV valueToStore);
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/ttl/TtlListState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/ttl/TtlListState.java
index 7cd52632866..c41b1d76b5b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/ttl/TtlListState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/ttl/TtlListState.java
@@ -123,6 +123,16 @@ class TtlListState<K, N, T>
return withTs;
}
+ @Override
+ public StateFuture<Void> asyncMergeNamespaces(N target, Collection<N> sources) {
+ return original.asyncMergeNamespaces(target, sources);
+ }
+
+ @Override
+ public void mergeNamespaces(N target, Collection<N> sources) {
+ original.mergeNamespaces(target, sources);
+ }
+
private class IteratorWithCleanup implements Iterator<T> {
private final Iterator<TtlValue<T>> originalIterator;
private boolean anyUnexpired = false;
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBMultiRawMergePutRequest.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBMultiRawMergePutRequest.java
new file mode 100644
index 00000000000..1f4aa8482f7
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBMultiRawMergePutRequest.java
@@ -0,0 +1,57 @@
+/*
+ * 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.state.forst;
+
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.forstdb.RocksDB;
+import org.forstdb.RocksDBException;
+
+import java.io.IOException;
+import java.util.Collection;
+
+/**
+ * The Put access request for ForStDB.
+ *
+ * @param <K> The type of key in put access request.
+ * @param <N> The type of namespace in put access request.
+ * @param <V> The type of value in put access request.
+ */
+public class ForStDBMultiRawMergePutRequest<K, N, V> extends ForStDBPutRequest<K, N, V> {
+
+ final Collection<byte[]> rawValue;
+
+ ForStDBMultiRawMergePutRequest(
+ ContextKey<K, N> key,
+ Collection<byte[]> value,
+ ForStInnerTable<K, N, V> table,
+ InternalStateFuture<Void> future) {
+ super(key, null, true, table, future);
+ this.rawValue = value;
+ }
+
+ @Override
+ public void process(ForStDBWriteBatchWrapper writeBatchWrapper, RocksDB db)
+ throws IOException, RocksDBException {
+ byte[] key = buildSerializedKey();
+ for (byte[] value : rawValue) {
+ writeBatchWrapper.merge(table.getColumnFamilyHandle(), key, value);
+ }
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalListState.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBRawGetRequest.java
similarity index 51%
copy from flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalListState.java
copy to flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBRawGetRequest.java
index efeccc7539a..4aa1a6ddc4a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalListState.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBRawGetRequest.java
@@ -16,17 +16,31 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.state.v2.internal;
+package org.apache.flink.state.forst;
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.state.v2.ListState;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import java.io.IOException;
+import java.util.List;
/**
- * This class defines the internal interface for list state.
+ * The Get access request for ForStDB.
*
- * @param <K> The type of key the state is associated to.
- * @param <N> The namespace type.
- * @param <V> The type of the intermediate state.
+ * @param <K> The type of key in get access request.
+ * @param <N> The type of namespace in get access request.
+ * @param <V> The type of value returned by get request.
*/
-@Internal
-public interface InternalListState<K, N, V> extends InternalKeyedState<K, N, V>, ListState<V> {}
+public class ForStDBRawGetRequest<K, N, V> extends ForStDBGetRequest<K, N, List<V>, byte[]> {
+
+ ForStDBRawGetRequest(
+ ContextKey<K, N> key,
+ ForStInnerTable<K, N, List<V>> table,
+ InternalStateFuture<byte[]> future) {
+ super(key, table, future);
+ }
+
+ @Override
+ public void completeStateFuture(byte[] bytesValue) throws IOException {
+ future.complete(bytesValue);
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStListState.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStListState.java
index 8bca44b4a67..f3bfc29cb17 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStListState.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStListState.java
@@ -19,11 +19,14 @@
package org.apache.flink.state.forst;
import org.apache.flink.api.common.state.v2.ListState;
+import org.apache.flink.api.common.state.v2.StateFuture;
import org.apache.flink.api.common.state.v2.StateIterator;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.core.memory.DataOutputSerializer;
import org.apache.flink.core.state.InternalStateFuture;
+import org.apache.flink.core.state.StateFutureUtils;
import org.apache.flink.runtime.asyncprocessing.RecordContext;
import org.apache.flink.runtime.asyncprocessing.StateRequest;
import org.apache.flink.runtime.asyncprocessing.StateRequestHandler;
@@ -33,12 +36,14 @@ import org.apache.flink.runtime.state.VoidNamespace;
import org.apache.flink.runtime.state.VoidNamespaceSerializer;
import org.apache.flink.runtime.state.v2.AbstractListState;
import org.apache.flink.runtime.state.v2.ListStateDescriptor;
-import org.apache.flink.util.Preconditions;
import org.forstdb.ColumnFamilyHandle;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
+import java.util.Iterator;
import java.util.List;
import java.util.function.Supplier;
@@ -129,15 +134,25 @@ public class ForStListState<K, N, V> extends AbstractListState<K, N, V>
@SuppressWarnings("unchecked")
@Override
- public ForStDBGetRequest<K, N, List<V>, StateIterator<V>> buildDBGetRequest(
+ public ForStDBGetRequest<K, N, List<V>, ?> buildDBGetRequest(
StateRequest<?, ?, ?, ?> stateRequest) {
- Preconditions.checkArgument(stateRequest.getRequestType() == StateRequestType.LIST_GET);
ContextKey<K, N> contextKey =
new ContextKey<>(
(RecordContext<K>) stateRequest.getRecordContext(),
(N) stateRequest.getNamespace());
- return new ForStDBListGetRequest<>(
- contextKey, this, (InternalStateFuture<StateIterator<V>>) stateRequest.getFuture());
+ switch (stateRequest.getRequestType()) {
+ case LIST_GET:
+ return new ForStDBListGetRequest<>(
+ contextKey,
+ this,
+ (InternalStateFuture<StateIterator<V>>) stateRequest.getFuture());
+ case CUSTOMIZED:
+ // must be LIST_GET_RAW
+ return new ForStDBRawGetRequest<>(
+ contextKey, this, (InternalStateFuture<byte[]>) stateRequest.getFuture());
+ default:
+ throw new UnsupportedOperationException();
+ }
}
@SuppressWarnings("unchecked")
@@ -166,6 +181,14 @@ public class ForStListState<K, N, V> extends AbstractListState<K, N, V>
value = (List<V>) stateRequest.getPayload();
merge = true;
break;
+ case CUSTOMIZED:
+ // must be LIST_ADD_ALL_RAW
+ return new ForStDBMultiRawMergePutRequest<>(
+ contextKey,
+ ((Tuple2<ForStStateRequestType, List<byte[]>>) stateRequest.getPayload())
+ .f1,
+ this,
+ (InternalStateFuture<Void>) stateRequest.getFuture());
default:
throw new IllegalArgumentException();
}
@@ -177,4 +200,86 @@ public class ForStListState<K, N, V> extends AbstractListState<K, N, V>
contextKey, value, this, (InternalStateFuture<Void>) stateRequest.getFuture());
}
}
+
+ @Override
+ public StateFuture<Void> asyncMergeNamespaces(N target, Collection<N> sources) {
+ if (sources == null || sources.isEmpty()) {
+ return StateFutureUtils.completedVoidFuture();
+ }
+ // phase 1: read from the sources and target
+ List<StateFuture<byte[]>> futures = new ArrayList<>(sources.size());
+ for (N source : sources) {
+ if (source != null) {
+ setCurrentNamespace(source);
+ futures.add(
+ handleRequest(
+ StateRequestType.CUSTOMIZED,
+ Tuple2.of(ForStStateRequestType.LIST_GET_RAW, null)));
+ }
+ }
+ // phase 2: merge the sources to the target
+ return StateFutureUtils.combineAll(futures)
+ .thenCompose(
+ values -> {
+ List<StateFuture<Void>> updateFutures =
+ new ArrayList<>(sources.size() + 1);
+ List<byte[]> validValues = new ArrayList<>(sources.size());
+ Iterator<byte[]> valueIterator = values.iterator();
+ for (N source : sources) {
+ byte[] value = valueIterator.next();
+ if (value != null) {
+ validValues.add(value);
+ setCurrentNamespace(source);
+ updateFutures.add(asyncClear());
+ }
+ }
+ if (!validValues.isEmpty()) {
+ setCurrentNamespace(target);
+ updateFutures.add(
+ handleRequest(
+ StateRequestType.CUSTOMIZED,
+ Tuple2.of(
+ ForStStateRequestType.MERGE_ALL_RAW,
+ validValues)));
+ }
+ return StateFutureUtils.combineAll(updateFutures);
+ })
+ .thenAccept(ignores -> {});
+ }
+
+ @Override
+ public void mergeNamespaces(N target, Collection<N> sources) {
+ if (sources == null || sources.isEmpty()) {
+ return;
+ }
+ try {
+ // merge the sources to the target
+ List<byte[]> validValues = new ArrayList<>(sources.size());
+ for (N source : sources) {
+ if (source != null) {
+ setCurrentNamespace(source);
+ byte[] oldValue =
+ handleRequestSync(
+ StateRequestType.CUSTOMIZED,
+ Tuple2.of(ForStStateRequestType.LIST_GET_RAW, null));
+
+ if (oldValue != null) {
+ setCurrentNamespace(source);
+ clear();
+ validValues.add(oldValue);
+ }
+ }
+ }
+
+ // if something came out of merging the sources, merge it or write it to the target
+ if (!validValues.isEmpty()) {
+ setCurrentNamespace(target);
+ handleRequestSync(
+ StateRequestType.CUSTOMIZED,
+ Tuple2.of(ForStStateRequestType.MERGE_ALL_RAW, validValues));
+ }
+ } catch (Exception e) {
+ throw new RuntimeException("merge namespace fail.", e);
+ }
+ }
}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateRequestClassifier.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateRequestClassifier.java
index 3cbe3dd005f..98b3a6438e3 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateRequestClassifier.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateRequestClassifier.java
@@ -18,6 +18,7 @@
package org.apache.flink.state.forst;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.asyncprocessing.StateRequest;
import org.apache.flink.runtime.asyncprocessing.StateRequestContainer;
import org.apache.flink.runtime.asyncprocessing.StateRequestType;
@@ -120,12 +121,43 @@ public class ForStStateRequestClassifier implements StateRequestContainer {
+ " doesn't yet support the clear method.");
}
}
+ case CUSTOMIZED:
+ {
+ handleCustomizedStateRequests(stateRequest);
+ return;
+ }
default:
throw new UnsupportedOperationException(
"Unsupported state request type:" + stateRequestType);
}
}
+ @SuppressWarnings("unchecked")
+ private void handleCustomizedStateRequests(StateRequest<?, ?, ?, ?> stateRequest) {
+ Tuple2<ForStStateRequestType, ?> payload =
+ (Tuple2<ForStStateRequestType, ?>) stateRequest.getPayload();
+ ForStStateRequestType requestType = payload.f0;
+ switch (requestType) {
+ case LIST_GET_RAW:
+ {
+ ForStListState<?, ?, ?> forStListState =
+ (ForStListState<?, ?, ?>) stateRequest.getState();
+ dbGetRequests.add(forStListState.buildDBGetRequest(stateRequest));
+ return;
+ }
+ case MERGE_ALL_RAW:
+ {
+ ForStListState<?, ?, ?> forStListState =
+ (ForStListState<?, ?, ?>) stateRequest.getState();
+ dbPutRequests.add(forStListState.buildDBPutRequest(stateRequest));
+ return;
+ }
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported customized state request type:" + requestType);
+ }
+ }
+
public List<ForStDBGetRequest<?, ?, ?, ?>> pollDbGetRequests() {
return dbGetRequests;
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalListState.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateRequestType.java
similarity index 63%
copy from flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalListState.java
copy to flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateRequestType.java
index efeccc7539a..33c9568957c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/internal/InternalListState.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateRequestType.java
@@ -16,17 +16,16 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.state.v2.internal;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.state.v2.ListState;
+package org.apache.flink.state.forst;
/**
- * This class defines the internal interface for list state.
- *
- * @param <K> The type of key the state is associated to.
- * @param <N> The namespace type.
- * @param <V> The type of the intermediate state.
+ * ForSt customized state request other than {@link
+ * org.apache.flink.runtime.asyncprocessing.StateRequestType}.
*/
-@Internal
-public interface InternalListState<K, N, V> extends InternalKeyedState<K, N, V>, ListState<V> {}
+public enum ForStStateRequestType {
+ /** Get the list in raw bytes without deserialization. */
+ LIST_GET_RAW,
+
+ /** Merge a list of raw bytes. */
+ MERGE_ALL_RAW
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStListStateTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStListStateTest.java
new file mode 100644
index 00000000000..68552b89c66
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStListStateTest.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.state.forst;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.runtime.state.v2.ListStateDescriptor;
+import org.apache.flink.runtime.state.v2.internal.InternalListState;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link ForStListState}. */
+public class ForStListStateTest extends ForStStateTestBase {
+
+ @Test
+ public void testMergeNamespace() throws Exception {
+ ListStateDescriptor<Integer> descriptor =
+ new ListStateDescriptor<>("testState", BasicTypeInfo.INT_TYPE_INFO);
+ InternalListState<String, Integer, Integer> listState =
+ keyedBackend.createState(1, IntSerializer.INSTANCE, descriptor);
+
+ setCurrentContext("test", "test");
+ for (int i = 0; i < 10; i++) {
+ listState.setCurrentNamespace(i);
+ listState.asyncAdd(i);
+ }
+ drain();
+
+ setCurrentContext("test", "test");
+ for (int i = 0; i < 10; i++) {
+ listState.setCurrentNamespace(i);
+ Iterable<Integer> list = listState.get();
+ assertThat(list).containsExactly(i);
+ }
+ drain();
+
+ // 1~20
+ ArrayList<Integer> namespaces = new ArrayList<>();
+ for (int i = 1; i < 20; i++) {
+ namespaces.add(i);
+ }
+
+ setCurrentContext("test", "test");
+ listState
+ .asyncMergeNamespaces(0, namespaces)
+ .thenAccept(
+ (e) -> {
+ listState.setCurrentNamespace(0);
+ Iterable<Integer> list = listState.get();
+ assertThat(list).containsExactly(0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
+
+ for (int i = 1; i < 10; i++) {
+ listState.setCurrentNamespace(i);
+ assertThat(listState.get()).isNullOrEmpty();
+ }
+ });
+ drain();
+
+ // test sync method
+ setCurrentContext("test", "test");
+ for (int i = 10; i < 20; i++) {
+ listState.setCurrentNamespace(i);
+ listState.add(i);
+ }
+ drain();
+
+ setCurrentContext("test", "test");
+ listState.mergeNamespaces(0, namespaces);
+ listState.setCurrentNamespace(0);
+ Iterable<Integer> list = listState.get();
+ assertThat(list)
+ .containsExactly(
+ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19);
+
+ for (int i = 1; i < 20; i++) {
+ listState.setCurrentNamespace(i);
+ assertThat(listState.get()).isNullOrEmpty();
+ }
+ drain();
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateTestBase.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateTestBase.java
new file mode 100644
index 00000000000..06b9a601045
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateTestBase.java
@@ -0,0 +1,110 @@
+/*
+ * 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.state.forst;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.runtime.asyncprocessing.AsyncExecutionController;
+import org.apache.flink.runtime.asyncprocessing.RecordContext;
+import org.apache.flink.runtime.operators.testutils.MockEnvironment;
+import org.apache.flink.runtime.util.TestingTaskManagerRuntimeInfo;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor;
+import org.apache.flink.streaming.runtime.tasks.mailbox.MailboxExecutorImpl;
+import org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailboxImpl;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+
+import static org.apache.flink.state.forst.ForStTestUtils.createKeyedStateBackend;
+
+/**
+ * A base class for all forst state tests, providing building the AEC and StateBackend logic, and
+ * some tool methods.
+ */
+public class ForStStateTestBase {
+
+ protected ForStKeyedStateBackend<String> keyedBackend;
+
+ protected AsyncExecutionController<String> aec;
+
+ protected MailboxExecutor mailboxExecutor;
+
+ protected RecordContext<String> context;
+
+ @BeforeEach
+ public void setup(@TempDir File temporaryFolder) throws IOException {
+ FileSystem.initialize(new Configuration(), null);
+ Configuration configuration = new Configuration();
+ configuration.set(ForStOptions.REMOTE_DIRECTORY, temporaryFolder.toURI().toString());
+ ForStStateBackend forStStateBackend =
+ new ForStStateBackend().configure(configuration, null);
+
+ keyedBackend =
+ createKeyedStateBackend(
+ forStStateBackend,
+ getMockEnvironment(temporaryFolder),
+ StringSerializer.INSTANCE);
+
+ mailboxExecutor =
+ new MailboxExecutorImpl(
+ new TaskMailboxImpl(), 0, StreamTaskActionExecutor.IMMEDIATE);
+
+ aec =
+ new AsyncExecutionController<>(
+ mailboxExecutor,
+ (a, b) -> {},
+ keyedBackend.createStateExecutor(),
+ 1,
+ 100,
+ 0,
+ 1,
+ null);
+ keyedBackend.setup(aec);
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ keyedBackend.close();
+ }
+
+ protected void setCurrentContext(Object record, String key) {
+ context = aec.buildContext(record, key);
+ context.retain();
+ aec.setCurrentContext(context);
+ }
+
+ protected void drain() {
+ context.release();
+ aec.drainInflightRecords(0);
+ }
+
+ private static MockEnvironment getMockEnvironment(File tempDir) {
+ return MockEnvironment.builder()
+ .setUserCodeClassLoader(ForStStateBackendConfigTest.class.getClassLoader())
+ .setTaskManagerRuntimeInfo(
+ new TestingTaskManagerRuntimeInfo(new Configuration(), tempDir))
+ .build();
+ }
+}