You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by le...@apache.org on 2025/03/19 06:02:52 UTC
(flink) branch master updated: [FLINK-37463][state/forst] make serialzer in ForStMapState Threadlocal (#26286)
This is an automated email from the ASF dual-hosted git repository.
leiyanfei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 1c5a295ec5f [FLINK-37463][state/forst] make serialzer in ForStMapState Threadlocal (#26286)
1c5a295ec5f is described below
commit 1c5a295ec5fc3398a542fcccb7fb6ffc99394796
Author: mayuehappy <ma...@bytedance.com>
AuthorDate: Wed Mar 19 14:02:47 2025 +0800
[FLINK-37463][state/forst] make serialzer in ForStMapState Threadlocal (#26286)
---
.../apache/flink/state/forst/ForStDBBunchPutRequest.java | 4 ++--
.../java/org/apache/flink/state/forst/ForStMapState.java | 16 ++++++++--------
2 files changed, 10 insertions(+), 10 deletions(-)
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBBunchPutRequest.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBBunchPutRequest.java
index 4a89fde78d3..6fa2e8e7240 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBBunchPutRequest.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStDBBunchPutRequest.java
@@ -44,7 +44,7 @@ import static org.apache.flink.state.forst.ForStDBIterRequest.startWithKeyPrefix
public class ForStDBBunchPutRequest<K, N, UK, UV> extends ForStDBPutRequest<K, N, Map<UK, UV>> {
/** Serializer for the user values. */
- final TypeSerializer<UV> userValueSerializer;
+ final ThreadLocal<TypeSerializer<UV>> userValueSerializer;
/** The data outputStream used for value serializer, which should be thread-safe. */
final ThreadLocal<DataOutputSerializer> valueSerializerView;
@@ -100,7 +100,7 @@ public class ForStDBBunchPutRequest<K, N, UK, UV> extends ForStDBPutRequest<K, N
public byte[] buildSerializedValue(UV singleValue) throws IOException {
DataOutputSerializer outputView = valueSerializerView.get();
outputView.clear();
- userValueSerializer.serialize(singleValue, outputView);
+ userValueSerializer.get().serialize(singleValue, outputView);
return outputView.getCopyOfBuffer();
}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStMapState.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStMapState.java
index 59b3f6d163a..43f22dc734e 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStMapState.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStMapState.java
@@ -73,10 +73,10 @@ public class ForStMapState<K, N, UK, UV> extends AbstractMapState<K, N, UK, UV>
final ThreadLocal<DataInputDeserializer> valueDeserializerView;
/** Serializer for the user keys. */
- final TypeSerializer<UK> userKeySerializer;
+ final ThreadLocal<TypeSerializer<UK>> userKeySerializer;
/** Serializer for the user values. */
- final TypeSerializer<UV> userValueSerializer;
+ final ThreadLocal<TypeSerializer<UV>> userValueSerializer;
/** Number of bytes required to prefix the key groups. */
private final int keyGroupPrefixBytes;
@@ -100,8 +100,8 @@ public class ForStMapState<K, N, UK, UV> extends AbstractMapState<K, N, UK, UV>
this.valueSerializerView = ThreadLocal.withInitial(valueSerializerViewInitializer);
this.keyDeserializerView = ThreadLocal.withInitial(keyDeserializerViewInitializer);
this.valueDeserializerView = ThreadLocal.withInitial(valueDeserializerViewInitializer);
- this.userKeySerializer = stateDescriptor.getUserKeySerializer();
- this.userValueSerializer = stateDescriptor.getSerializer();
+ this.userKeySerializer = ThreadLocal.withInitial(stateDescriptor::getUserKeySerializer);
+ this.userValueSerializer = ThreadLocal.withInitial(stateDescriptor::getSerializer);
this.keyGroupPrefixBytes = keyGroupPrefixBytes;
}
@@ -124,7 +124,7 @@ public class ForStMapState<K, N, UK, UV> extends AbstractMapState<K, N, UK, UV>
return builder.build();
}
UK userKey = (UK) contextKey.getUserKey(); // map get
- return builder.buildCompositeKeyUserKey(userKey, userKeySerializer);
+ return builder.buildCompositeKeyUserKey(userKey, userKeySerializer.get());
}
@Override
@@ -132,7 +132,7 @@ public class ForStMapState<K, N, UK, UV> extends AbstractMapState<K, N, UK, UV>
DataOutputSerializer outputView = valueSerializerView.get();
outputView.clear();
outputView.writeBoolean(false);
- userValueSerializer.serialize(value, outputView);
+ userValueSerializer.get().serialize(value, outputView);
return outputView.getCopyOfBuffer();
}
@@ -141,13 +141,13 @@ public class ForStMapState<K, N, UK, UV> extends AbstractMapState<K, N, UK, UV>
DataInputDeserializer inputView = valueDeserializerView.get();
inputView.setBuffer(valueBytes);
boolean isNull = inputView.readBoolean();
- return isNull ? null : userValueSerializer.deserialize(inputView);
+ return isNull ? null : userValueSerializer.get().deserialize(inputView);
}
public UK deserializeUserKey(byte[] userKeyBytes, int userKeyOffset) throws IOException {
DataInputDeserializer inputView = keyDeserializerView.get();
inputView.setBuffer(userKeyBytes, userKeyOffset, userKeyBytes.length - userKeyOffset);
- return userKeySerializer.deserialize(inputView);
+ return userKeySerializer.get().deserialize(inputView);
}
@Override