[FLINK-35162][state] Implement WriteBatchOperation and general MultiGetOperation for ForSt #24681

ljz2051 · 2024-04-18T14:13:58Z

What is the purpose of the change

This pull request introduces the ForStValueState and support WriteBatchOperation and general MultiGetOperation for ForSt.

Brief change log

Introduces the ForStValueState
Support WriteBatchOperation and general MultiGetOperation for ForSt

Verifying this change

This change added tests and can be verified by ForStWriteBatchOperationTest and ForStGeneralMultiGetOperationTest.

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

Dependencies (does it add or upgrade a dependency): yes
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): yes
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

ljz2051 · 2024-04-18T14:15:29Z

@Zakelly @fredia @masteryhx Could you please take a look?

flinkbot · 2024-04-18T14:17:22Z

CI report:

c52f371 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 left some comments PTAL.

flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestHandler.java

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

Zakelly · 2024-04-19T02:57:53Z

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

+     * @return Processing result.
+     * @throws IOException Thrown if ForStDB access encountered an I/O related error.
+     */
+    OUT process() throws IOException;


Is this designed to be a synchronous method? Would it be better if it is an asynchronous method returning a CompletableFuture? I think this is more flexible since we don't need a specific thread waiting the result.

Good point. I agree that the future-style interface is more flexible. I have refined it.

...tebackend-forst/src/main/java/org/apache/flink/state/forst/ForStNativeMultiGetOperation.java

Zakelly · 2024-04-19T03:28:50Z

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

+    }
+
+    @Override
+    public byte[] serializeKey(ContextKey<K> contextKey) throws IOException {


Should ContextKey be thread-safe? It seems the 'read cache or serialize' logic better be called only once for each context key.

Yes. I have marked the ContextKey with @threadsafe annotation, and refactor the 'read cache or serialize' logic. Please review the ContextKey#getOrCreateSerializedKey method.

Zakelly · 2024-04-19T03:52:03Z

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

+     */
+    private @Nullable byte[] serializedKey = null;
+
+    public ContextKey(K rawKey, int keyGroup) {


Since you have provided the of static method, the constructor could be private? Well I'd prefer not providing of if you have multiple constructors....

I have remove the "ContextKey#of" static method

Zakelly · 2024-04-19T03:53:29Z

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

+ *
+ * @param <K> The type of the raw key.
+ */
+public class ContextKey<K> {


And IIUC the ContextKey will be shared across state requests? Is that means ContextKey should be attached to the RecordContext in future?

Yes, the ContextKey can be shared across state requests, and ContextKey will be attached to the RecordContext in another PR. So I move the ContextKey class to flink-runtime module from forst-module.

Well, I didn't mean that it should move to the flink-runtime. I was thinking providing a payload in RecordContext for state backend to assign customized data defined by itself. WDYT?

And by the way, the key and keygroup should be provided in RecordContext so maybe we could save some memory overhead here.

@Zakelly As we discussed offline, I have attached the keyGroup and payload (serialized key) into RecordContext , and the ContextKey is placed in forst-module. Please help to review again.

flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestHandler.java

ljz2051 · 2024-04-19T10:25:31Z

@Zakelly Thanks for your review! I have addressed all comments your mentioned.

jectpro7 · 2024-04-21T16:58:15Z

...ebackend-forst/src/main/java/org/apache/flink/state/forst/ForStGeneralMultiGetOperation.java

+                        try {
+                            ForStInnerTable<K, V> table = request.table;
+                            byte[] key = table.serializeKey(request.key);
+                            byte[] value = db.get(table.getColumnFamilyHandle(), key);


Hi @ljz2051, it creates many rpc request here, as FLIP-426 mentioned the rpc round-trip overhead is the bottleneck. It might be better by using multiGetAsList

@jectpro7 Thanks four your advice.
The MultiGet api based on remote filesystem is not currently supported by ForStDB, and more importantly, not all file systems will support the MultiGet api, so the purpose of this PR is to introduce a general MultiGet implementation that works on all file systems.

In addition, I will introduce the ForStDB native MultiGet api in another jira (FLINK-35163) to optimize remote state access.

oh, got it, good to learn. Thanks for sharing the background.

jectpro7 · 2024-04-21T17:03:04Z

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

+            }
+            db.write(writeOptions, writeBatch);
+            result.complete(null);
+        } catch (RocksDBException e) {


is there any case that the exception is not a RocksDBException? If yes, the future might not be completed. So complete it in finally might be a better choice.

I think it would be better to catch all types of exceptions here, not just RocksDBException , and complete the result future exceptionally. WDYT?

sounds great

masteryhx

Thanks for the PR!
I just left some comments. PTAL.

masteryhx · 2024-04-22T09:10:36Z

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

+            StateRequestHandler stateRequestHandler,
+            ColumnFamilyHandle columnFamily,
+            ValueStateDescriptor<V> valueStateDescriptor,
+            ThreadLocal<SerializedCompositeKeyBuilder<K>> serializedKeyBuilder,


Threadlocal parameters are a bit confusing for callers.
And also make life cycles of these variables expose to callers.
Could we avoid it and just maintain them internally ?

I think we could just pass the Supplier to ForStValueState, and keep the ThreadLocal semantics inside ForStState.

masteryhx · 2024-04-22T09:16:24Z

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

+    }
+
+    /** The Put access request for ForStDB. */
+    static class Request<K, V> {


Could it have structural relationship with other one in ForStGeneralMultiGetOperation ?
Or Could we use WriteRequest and GetRequest to distinguish them ?

They don't have structural relationship. It's a good idea that using GetRequest and PutRequest to distinguish them.

masteryhx · 2024-04-22T09:18:38Z

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

+            this.table = table;
+        }
+
+        static <K, V> Request<K, V> of(K key, V value, ForStInnerTable<K, V> table) {


A minor suggestion:
Adding some descriptions here ? Since value is nullable and we will delete it for null value which should be explicit for callers.

I have refined it.

fredia · 2024-04-23T08:25:41Z

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

+        synchronized (recordContext) {
+            if (recordContext.getPayload() == null) {
+                byte[] serializedKey = serializeKeyFunc.apply(this);
+                recordContext.setPayload(serializedKey);


Will the serializedKey and value are both store in payload?

No, only the serialized key will be cached in payload. The value cannot be reused between different state execution stage, so there is no need to cache value.

Zakelly · 2024-04-24T10:50:37Z

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

+    }
+
+    @Override
+    public CompletableFuture<Void> process() throws IOException {


It seems this process() still act as synchronous method?

I have refined it !

Zakelly

Thanks for the update , LGTM +1

masteryhx

Thanks for the update.
LGTM.

flinkbot added the component=<none> label Apr 18, 2024

ljz2051 force-pushed the FLINK-35162 branch 2 times, most recently from 2d65519 to c529291 Compare April 19, 2024 02:26

Zakelly reviewed Apr 19, 2024

View reviewed changes

ljz2051 force-pushed the FLINK-35162 branch from c529291 to 8ba3cf5 Compare April 19, 2024 10:07

ljz2051 requested a review from Zakelly April 19, 2024 10:25

jectpro7 reviewed Apr 21, 2024

View reviewed changes

masteryhx reviewed Apr 22, 2024

View reviewed changes

ljz2051 force-pushed the FLINK-35162 branch from 8ba3cf5 to 9e0f4dc Compare April 23, 2024 02:58

ljz2051 requested a review from masteryhx April 23, 2024 03:11

fredia reviewed Apr 23, 2024

View reviewed changes

ljz2051 added 2 commits April 23, 2024 20:48

[FLINK-35125][state] Enable forstjni dependency for ForStStateBackend

bfc5f5b

[FLINK-35125][state] Implement ValueState for ForStStateBackend

6cc8112

ljz2051 force-pushed the FLINK-35162 branch 2 times, most recently from 322a2ad to 3097401 Compare April 23, 2024 13:10

Zakelly reviewed Apr 24, 2024

View reviewed changes

ljz2051 added 2 commits April 25, 2024 10:05

[FLINK-35162][state]Support grouping state get and put access

4be5d0a

[FLINK-35162][state] Attach the keyGroup and payload into RecordContext

c52f371

ljz2051 force-pushed the FLINK-35162 branch from 3097401 to c52f371 Compare April 25, 2024 02:05

ljz2051 requested a review from Zakelly April 25, 2024 02:07

Zakelly approved these changes Apr 25, 2024

View reviewed changes

masteryhx approved these changes Apr 26, 2024

View reviewed changes

masteryhx closed this Apr 26, 2024

fredia mentioned this pull request Apr 30, 2024

[FLINK-35161][state] Implement StateExecutor for ForStStateBackend #24739

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-35162][state] Implement WriteBatchOperation and general MultiGetOperation for ForSt #24681

[FLINK-35162][state] Implement WriteBatchOperation and general MultiGetOperation for ForSt #24681

ljz2051 commented Apr 18, 2024

ljz2051 commented Apr 18, 2024

flinkbot commented Apr 18, 2024 •

edited

Zakelly left a comment

Zakelly Apr 19, 2024

ljz2051 Apr 19, 2024

Zakelly Apr 19, 2024

ljz2051 Apr 19, 2024

Zakelly Apr 19, 2024

ljz2051 Apr 19, 2024

Zakelly Apr 19, 2024

ljz2051 Apr 19, 2024

Zakelly Apr 22, 2024

ljz2051 Apr 23, 2024

ljz2051 commented Apr 19, 2024

jectpro7 Apr 21, 2024

ljz2051 Apr 22, 2024

jectpro7 Apr 22, 2024

jectpro7 Apr 21, 2024

ljz2051 Apr 22, 2024

jectpro7 Apr 22, 2024

masteryhx left a comment

masteryhx Apr 22, 2024

ljz2051 Apr 23, 2024

masteryhx Apr 22, 2024

ljz2051 Apr 23, 2024

masteryhx Apr 22, 2024

ljz2051 Apr 23, 2024

fredia Apr 23, 2024

ljz2051 Apr 23, 2024

Zakelly Apr 24, 2024

ljz2051 Apr 25, 2024

Zakelly left a comment

masteryhx left a comment

[FLINK-35162][state] Implement WriteBatchOperation and general MultiGetOperation for ForSt #24681

[FLINK-35162][state] Implement WriteBatchOperation and general MultiGetOperation for ForSt #24681

Conversation

ljz2051 commented Apr 18, 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

ljz2051 commented Apr 18, 2024

flinkbot commented Apr 18, 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

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

ljz2051 commented Apr 19, 2024

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

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

Zakelly left a comment

Choose a reason for hiding this comment

masteryhx left a comment

Choose a reason for hiding this comment

flinkbot commented Apr 18, 2024 •

edited