[
https://issues.apache.org/jira/browse/BEAM-7739?focusedWorklogId=301355&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-301355
]
ASF GitHub Bot logged work on BEAM-7739:
----------------------------------------
Author: ASF GitHub Bot
Created on: 26/Aug/19 17:08
Start Date: 26/Aug/19 17:08
Worklog Time Spent: 10m
Work Description: robertwb commented on pull request #9067: [BEAM-7739]
Implement ReadModifyWriteState in Python SDK
URL: https://github.com/apache/beam/pull/9067#discussion_r317697628
##########
File path:
runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryStateInternals.java
##########
@@ -166,9 +172,62 @@ public WatermarkHoldState bindWatermark(
}
}
- /** An {@link InMemoryState} implementation of {@link ValueState}. */
+
+ /** An {@link InMemoryState} implementation of {@link ReadModifyWriteState}.
*/
+ public static final class InMemoryReadModifyWrite<T>
+ implements ReadModifyWriteState<T>,
InMemoryState<InMemoryReadModifyWrite<T>> {
+ private final Coder<T> coder;
+
+ private boolean isCleared = true;
+ private @Nullable T value = null;
+
+ public InMemoryReadModifyWrite(Coder<T> coder) {
+ this.coder = coder;
+ }
+
+ @Override
+ public void clear() {
+ // Even though we're clearing we can't remove this from the in-memory
state map, since
+ // other users may already have a handle on this Value.
+ value = null;
+ isCleared = true;
+ }
+
+ @Override
+ public InMemoryReadModifyWrite<T> readLater() {
+ return this;
+ }
+
+ @Override
+ public T read() {
+ return value;
+ }
+
+ @Override
+ public void write(T input) {
+ isCleared = false;
+ this.value = input;
+ }
+
+ @Override
+ public InMemoryReadModifyWrite<T> copy() {
+ InMemoryReadModifyWrite<T> that = new InMemoryReadModifyWrite<>(coder);
+ if (!this.isCleared) {
+ that.isCleared = this.isCleared;
+ that.value = uncheckedClone(coder, this.value);
+ }
+ return that;
+ }
+
+ @Override
+ public boolean isCleared() {
+ return isCleared;
+ }
+ }
+
+ /** An {@link InMemoryState} implementation of {@link ReadModifyWriteState}.
*/
public static final class InMemoryValue<T>
- implements ValueState<T>, InMemoryState<InMemoryValue<T>> {
+ implements ValueState<T>, InMemoryState<InMemoryValue<T>> {
Review comment:
Seems like this could be mostly merged with InMemoryReadModifyWrite
(possibly requiring a common baseclass).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
-------------------
Worklog Id: (was: 301355)
Time Spent: 3h 10m (was: 3h)
> Add ValueState in Python sdk
> ----------------------------
>
> Key: BEAM-7739
> URL: https://issues.apache.org/jira/browse/BEAM-7739
> Project: Beam
> Issue Type: New Feature
> Components: sdk-py-core
> Reporter: Rakesh Kumar
> Assignee: Rakesh Kumar
> Priority: Major
> Time Spent: 3h 10m
> Remaining Estimate: 0h
>
> Currently ValueState is missing from Python Sdks but it is existing in Java
> sdks.
--
This message was sent by Atlassian Jira
(v8.3.2#803003)
