[FLINK-35049][state] Implement Map Async State API for ForStStateBackend #24812

fredia · 2024-05-20T05:41:25Z

What is the purpose of the change

This PR implements Map Async State API for ForStStateBackend.

Brief change log

Introduce ForStDBIterRequest and ForStIterateOperation for Map#iterator
Introduce ForStDBBunchPutRequest for Map#putAll and Map#clear
Utilize ForStDBPutRequest for Map#put, and ForStDBGetRequest for Map#value

Verifying this change

This change added tests and can be verified as follows:

ForStDBIterateOperationTest
Othe map related tests are located in subclasses of ForStDBOperationTestBase.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): ( no)
The serializers: (no )
The runtime per-record code paths (performance sensitive): (no )
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: ( no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2024-05-20T05:47:34Z

CI report:

6d8d917 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

Zakelly

Thanks for this PR! I'm still reviewing and will put comments incrementally.

Zakelly · 2024-05-20T09:35:39Z

...atebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateRequestClassifier.java

@@ -31,13 +32,19 @@
 */
 public class ForStStateRequestClassifier implements StateRequestContainer {

+    private final StateRequestHandler stateRequestHandler;


Seems unused?

Zakelly · 2024-05-20T10:21:36Z

...nk-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBBunchPutRequest.java

+ *
+ * @param <K> The type of key in original state access request.
+ */
+public class ForStDBBunchPutRequest<K> extends ForStDBPutRequest<ContextKey<K>, Map<?, ?>> {


Seems this extension lacks practical meaning......

I'd suggest we re-organize the interfaces. ForStWriteBatchOperation manipulating some new introduced ForStSingleWriteOperation, which is wrapped and iterative given by ForStWriteBatchOperation and ForStDBPutRequest

I moved the logic of interacting with the db to the ForStDBPutRequest and ForStDBBunchPutRequest, to avoid the classifications in ForStWriteBatchOperation.

jectpro7

Hi @fredia, thanks for the PR. I left several comments, but I am still beginner of rockdb, please correct me if I am wrong :-)

jectpro7 · 2024-05-22T14:25:30Z

...nk-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBBunchPutRequest.java

+    public ForStDBBunchPutRequest(
+            ContextKey<K> key, Map value, ForStMapState table, InternalStateFuture<Void> future) {
+        super(key, value, table, future);
+        Preconditions.checkArgument(table instanceof ForStMapState);


this check seems redundant.

jectpro7 · 2024-05-23T03:39:44Z

...nk-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBBunchPutRequest.java

+        if (!find) {
+            byte[] newBytes = new byte[len + 1];
+            System.arraycopy(bytes, 0, newBytes, 0, len);
+            newBytes[len] = 1;
+            return newBytes;
+        }


From my understanding, if the byteArr is maximum value. The next should like this:

Suggested change

if (!find) {

byte[] newBytes = new byte[len + 1];

System.arraycopy(bytes, 0, newBytes, 0, len);

newBytes[len] = 1;

return newBytes;

}

if (!find) {

byte[] newBytes = new byte[len + 1];

System.arraycopy(bytes, 0, newBytes, 0, len);

newBytes[len] = Byte.MIN_VALUE + 1;

return newBytes;

}

Thanks for the suggestion, after reading the code in rocksdb, I think newBytes[len] = Byte.MAX_VALUE is enough.

inline int Slice::compare(const Slice& b) const { assert(data_ != nullptr && b.data_ != nullptr); const size_t min_len = (size_ < b.size_) ? size_ : b.size_; int r = memcmp(data_, b.data_, min_len); if (r == 0) { if (size_ < b.size_) r = -1; else if (size_ > b.size_) r = +1; } return r; }

jectpro7 · 2024-05-23T05:18:00Z

...s/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBGetRequest.java

        this.key = key;
        this.table = table;
        this.future = future;
+        this.toBoolean = toBoolean;


toBoolean seems not very clear. I would suggest to add ForStDBExistRequest for this kind of operation.

👍，I added ForStDBMapCheckRequest for this kind of operation.

jectpro7 · 2024-05-23T05:48:12Z

...-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStWriteBatchOperation.java

+                                if (request instanceof ForStDBBunchPutRequest) {
+                                    ForStDBBunchPutRequest<?> bunchPutRequest =
+                                            (ForStDBBunchPutRequest<?>) request;
+                                    byte[] primaryKey = bunchPutRequest.buildSerializedKey(null);


Sorry, I cannot understand this line, why serialize null here? Why the Bunch remove and Single remove are using different serialization method.

For map state, there are two types of key: primary key and user key.
Logically it is organized like this:
<primary key, <user key, user value>

We concatenate primary key and user key together as the key of rocksdb, and physically organized like this:
< <primary key - user key> , user value>

For Bunch remove, we will remove all user keys under one primary key, so we use the prefix to match all keys.
For Single remove, we use the bytes of <primary key-user key> to match one specific key.

masteryhx

Thanks for the PR.
PTAL my comments.

masteryhx · 2024-05-27T12:59:37Z

...-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStWriteBatchOperation.java

+                                    ForStDBBunchPutRequest<?> bunchPutRequest =
+                                            (ForStDBBunchPutRequest<?>) request;
+                                    byte[] primaryKey = bunchPutRequest.buildSerializedKey(null);
+                                    byte[] endKey = ForStDBBunchPutRequest.nextBytes(primaryKey);


I may missed something.
What's the logic of calculating and using deleteRange ?

We concatenate primary key and user key together as the key of rocksdb, and physically organized like this:

< <primary key1 - user key1> , user value1> < <primary key1 - user key2> , user value2> < <primary key1 - user key3> , user value3> < <primary key1 - user key4> , user value4>

Here, we use primary key as a prefix to calculate an interval to leverage deleteRange.

But too much deleteRange may affect the read performance, and we will consider changing it iterator.delete later.

I found that deleteRange may be ambiguous in some cases, such as when the primary key is an array of Byte. MAX_VALUE , so I changed it to iterator and delete one by one.

masteryhx · 2024-05-27T13:14:00Z

...s/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBGetRequest.java


-    private ForStDBGetRequest(K key, ForStInnerTable<K, V> table, InternalStateFuture<V> future) {
+    private final boolean toBoolean;


Could this have a subclass to make the structure more clear ?

👍I added ForStDBMapCheckRequest for this kind of operation.

masteryhx · 2024-05-27T13:19:21Z

...ink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStIterateOperation.java

+            executor.execute(
+                    () -> {
+                        // todo: config read options
+                        try (RocksIterator iter = db.newIterator(request.getColumnFamilyHandle())) {


This logic seems a bit complex.
Could you add some descriptions in some key steps or split them into methods ?

masteryhx · 2024-05-27T13:20:49Z

...s/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBPutRequest.java

    }

    public boolean valueIsNull() {
        return value == null;
    }

+    public boolean valueIsMap() {


Same as GetRequest.
Maybe we could have a more clear structure.

I moved the logic of interacting with the db to the ForStDBPutRequest and ForStDBBunchPutRequest, valueIsMap is deleted now.

masteryhx · 2024-05-27T13:21:56Z

...ink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStIterateOperation.java

+                            request.completeStateFuture(stateIterator);
+                        } catch (Exception e) {
+                            LOG.warn("Error when process iterate operation for forStDB", e);
+                            future.completeExceptionally(e);


Remember to completeExceptionally for InternalStateFuture.

masteryhx · 2024-05-27T13:22:59Z

...atebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateRequestClassifier.java

+                    dbPutRequests.add(forStMapState.buildDBBunchPutRequest(stateRequest));
+                    return;
+                }
+            case MAP_IS_EMPTY:


nit: Same as MAP_GET ?

flinkbot added the component=Runtime/StateBackends label May 20, 2024

fredia force-pushed the forst_map branch 2 times, most recently from 480b425 to 02c4d45 Compare May 20, 2024 07:57

Zakelly requested changes May 20, 2024

View reviewed changes

jectpro7 reviewed May 23, 2024

View reviewed changes

masteryhx reviewed May 27, 2024

View reviewed changes

fredia force-pushed the forst_map branch from 02c4d45 to 74526d5 Compare June 6, 2024 09:02

[FLINK-35049][state] Implement Map Async State API for ForStStateBackend

6d8d917

fredia force-pushed the forst_map branch from 74526d5 to 6d8d917 Compare June 6, 2024 11:37

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-35049][state] Implement Map Async State API for ForStStateBackend #24812

[FLINK-35049][state] Implement Map Async State API for ForStStateBackend #24812

fredia commented May 20, 2024

flinkbot commented May 20, 2024 •

edited

Zakelly left a comment

Zakelly May 20, 2024

Zakelly May 20, 2024

fredia Jun 6, 2024

jectpro7 left a comment

jectpro7 May 22, 2024

jectpro7 May 23, 2024 •

edited

fredia Jun 6, 2024 •

edited

jectpro7 May 23, 2024

fredia Jun 6, 2024

jectpro7 May 23, 2024

fredia Jun 6, 2024

masteryhx left a comment

masteryhx May 27, 2024

fredia Jun 6, 2024

fredia Jun 6, 2024

masteryhx May 27, 2024

fredia Jun 6, 2024

masteryhx May 27, 2024

masteryhx May 27, 2024

fredia Jun 6, 2024

masteryhx May 27, 2024

masteryhx May 27, 2024


		private ForStDBGetRequest(K key, ForStInnerTable<K, V> table, InternalStateFuture<V> future) {
		private final boolean toBoolean;

[FLINK-35049][state] Implement Map Async State API for ForStStateBackend #24812

Are you sure you want to change the base?

[FLINK-35049][state] Implement Map Async State API for ForStStateBackend #24812

Conversation

fredia commented May 20, 2024

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented May 20, 2024 • edited

CI report:

Zakelly left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jectpro7 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jectpro7 May 23, 2024 • edited

Choose a reason for hiding this comment

fredia Jun 6, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

masteryhx left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

flinkbot commented May 20, 2024 •

edited

jectpro7 May 23, 2024 •

edited

fredia Jun 6, 2024 •

edited