[ https://issues.apache.org/jira/browse/BEAM-7739?focusedWorklogId=301357&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-301357 ]
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_r317698856 ########## 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: Actually, I wonder if InMemoryReadModifyWrite was a subclass of InMemoryValue one could get rid of some of the parallel code here (e.g. bindValue vs. bindReadModifyWrite). Haven't pursued this to its conclusion though. ---------------------------------------------------------------- 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: 301357) Time Spent: 3h 20m (was: 3h 10m) > 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 20m > 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)