[FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to clarify the concepts #2768

shixiaogang · 2016-11-08T02:58:26Z

Changes in the definition of State and StateDescriptor:

Add get() in the State interface.
Remove type serializers of state values from StateDescriptors.
Add SimpleStateDescriptor to simplify the construction of ValueStateDescriptor, ReducingStateDescriptor and FoldingStateDescriptor.
Changes the definition of KeyedStateBackend and AbstractKeyedStateBackend accordingly.
Modify the implementation of ListStateDescriptor accordingly.

Changes in HeapStateBackend:

Let AbstractHeapState not implement the State interface. The clear() method now is removed from AbstractHeapState.
Add HeapSimpleState to simplify the implementation of HeapValueState, HeapReducingState and HeapFoldingState.
Change the implementation of HeapValueState, HeapReducingState and HeapFoldingState accordingly.

Changes in RocksDBStateBackend:

Let AbstractRocksDBState not implement the State interface, removing the clear() method. Now, AbstractRocksDBState does not depend on the types of State and StateDescriptor any more.
Add RocksDBSimpleState to simplify the implementation of RocksDBValueState, RocksDBReducingState and RocksDBFoldingState.
Change the implementation of RocksDBValueState, RocksDBReducingState and RocksDBFoldingState accordingly.

Others:

Update the usage of States in the implementation of window operators.
Update the usage of States in unit tests.

=== UPDATE ===

Remove clear() method from State
Add a new interface called UpdatableState
Reformat the code

StephanEwen · 2016-11-08T13:28:17Z

...end-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/AbstractRocksDBState.java

-			RocksDBKeyedStateBackend<K> backend) {
-
+			RocksDBKeyedStateBackend<K> backend
+	) {


Please stick with the original code style here

StephanEwen · 2016-11-08T13:31:24Z

...ckend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBSimpleState.java

+			if (valueBytes == null) {
+				return stateDesc.getDefaultValue();
+			}
+			return valueSerializer.deserialize(new DataInputViewStreamWrapper(new ByteArrayInputStream(valueBytes)));


Given that this is used only in single threaded settings, we should try to reuse the DataInputViewStreamWrapper and ByteArrayInputStream.

It may help reduce the number of object allocation, but i think the brought benefits are fewer that brought by the re-usage of ByteArrayOutputStream. Because the byte array inside ByteArrayInputStream is provided by other objects.

What do you think?

StephanEwen · 2016-11-08T13:33:12Z

flink-core/src/main/java/org/apache/flink/api/common/state/ListStateDescriptor.java

+		out.defaultWriteObject();
+	}
+
+	private void readObject(final ObjectInputStream in) throws IOException, ClassNotFoundException {


I think this method is not necessary, because it only triggers default serialization anyways.

aljoscha

Overall, I like these changes a lot! Very good work.

I had some inline comments, mostly about missing Javadoc. One thing I'd like to see changed is the additional V generic parameter in a lot of the state accessor methods, I don't think we need that one and I would like it if we don't add additional generic parameters if we don't have to.

aljoscha · 2016-11-22T14:43:19Z

flink-core/src/main/java/org/apache/flink/api/common/state/SimpleStateDescriptor.java

+	private transient TypeInformation<T> typeInfo;
+
+	/** The default value returned by the state when no other value is bound to a key */
+	protected transient T defaultValue;


We might think about renaming this because for FoldingState it's not a default value but the initial accumulation value. (Just a suggestion, not strictly necessary)

aljoscha · 2016-11-22T14:45:19Z

flink-core/src/main/java/org/apache/flink/api/common/state/State.java

+	 * partitioned the returned value is the same for all inputs in a given
+	 * operator instance. If state partitioning is applied, the value returned
+	 * depends on the current operator input, as the operator maintains an
+	 * independent state for each partition.


This is incorrect, we don't maintain state by partition but by key. (This is not your fault, it was always like this on ValueState and you copied it from there, I just discovered it now when reading your code.)

aljoscha · 2016-11-22T14:46:47Z

flink-core/src/main/java/org/apache/flink/api/common/state/ValueState.java

+	 *
 	 * @throws IOException Thrown if the system cannot access the state.
 	 */
+	@Deprecated


Please also add a @deprecated description in the Javadoc, pointing to use get().

aljoscha · 2016-11-22T14:50:54Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapSimpleState.java

+import java.util.Map;
+
+/**
+ * Heap-backed partitioned states whose values are not composited and is snapshotted into files.


the is snapshotted into files portion is not always true, and also not necessary here.

aljoscha · 2016-11-22T16:52:19Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedStateBackend.java

 	 */
 	@SuppressWarnings({"rawtypes", "unchecked"})
-	<N, S extends State> S getPartitionedState(
+	<N, V, S extends State<V>> S getPartitionedState(


I think we don't need the additional V parameter. It's never used and State<?> works just as while while State<V> just pretends to add more type safety. Because both ways work I would prefer to go with the solution that doesn't add additional generic parameters to methods.

Same holds for mergePartitionedStates().

aljoscha · 2016-11-22T16:58:12Z

...main/java/org/apache/flink/streaming/api/functions/query/AbstractQueryableStateOperator.java

 */
 @Internal
-abstract class AbstractQueryableStateOperator<S extends State, IN>
+abstract class AbstractQueryableStateOperator<V, S extends State<V>, IN>


I think here the same thing I mentioned earlier holds, we don't really need the additional V parameter.

aljoscha · 2016-11-22T16:59:23Z

...ming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java

 	 * @throws Exception Thrown, if the state backend cannot create the key/value state.
 	 */
-	protected <S extends State> S getPartitionedState(StateDescriptor<S, ?> stateDescriptor) throws Exception {
+	protected <V, S extends State<V>> S getPartitionedState(StateDescriptor<S> stateDescriptor) throws Exception {


Here, the same thing I mentioned on KeyedStateBackend holds, we don't need the additional V parameter.

aljoscha · 2016-11-22T17:00:19Z

...-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/Trigger.java

 		 *                                       function (function is not part os a KeyedStream).
 		 */
-		<S extends State> S getPartitionedState(StateDescriptor<S, ?> stateDescriptor);
+		<V, S extends State<V>> S getPartitionedState(StateDescriptor<S> stateDescriptor);


Same thing as elsewhere: we don't need V.

shixiaogang · 2016-11-23T03:49:47Z

@aljoscha Thanks for your review. I have updated the PR according to your suggestion.

aljoscha · 2016-11-23T11:15:05Z

@shixiaogang Thanks for the swift update! The changes look very good now though I found one last thing that could be problematic. This is changing the behaviour of FoldingState when it comes to the default value. I'll quickly open a PR that verifies the current behaviour.

shixiaogang · 2016-11-24T08:31:38Z

Oh... I added another field to make the code more clear, but I did not notice the serialization problem. Thanks very much for your reminder.

Your solution does work though the concept of "defaultValue" in folding states is a little confusing. Another solution to let FoldingStateDescriptor implement its own serialization method. And I prefer this one.

What do you think? @aljoscha

aljoscha · 2016-11-24T10:06:18Z

@shixiaogang you can also do that, but then the machinery in SimpleStateDescriptor is only used by ValueStateDescriptor because this is really the only descriptor that has a default value. (In my original implementation it was like this and only ValueStateDescriptor had the serialisation logic and a default value.)

shixiaogang · 2016-11-28T10:07:08Z

I moved default value from SimpleStateDescriptor to ValueStateDescriptor. Now only ValueStateDescriptors have default values. The serialization methods may contain some duplicated code, but i think it's acceptable.

I also modify the implementation of HeapReducingStates. The first value will be copied before being put into the heap.

@aljoscha What do you think of these changes?

aljoscha · 2016-11-28T10:21:58Z

Thanks for updating, @shixiaogang. I'll look at this today!

For the change to reducing state it would be better to have this as a separate issue/commit/PR. @fhueske you already opened an issue for this, right? Could you please point us to that?

shixiaogang · 2016-12-01T08:16:27Z

I rebased the branch to resolve the conflicts with the master branch.

…lt value

shixiaogang · 2017-01-18T01:50:34Z

Despite the changes in the state descriptors, the Flink jobs can restore from old versioned snapshots now.

StephanEwen · 2017-01-25T15:13:08Z

Hi @shixiaogang !

I went through this pull request and below are a few thoughts. The pull request changes many things together. Some can work, and for others I would suggest to do it differently.
Let me know what you think about this:

Changing the `StateDescriptor`

This is probably a good cleanup. It does currently break the backwards compatibility, because Flink 1.1 wrote the state descriptors into the checkpoints. Flink 1.2 and 1.3 do not do that any more, but currently rely on the unchanged StateDescriptor classes for resuming Flink-1.1-savepoints.

We added a way to define "migration" classes, meaning we can store the old StateDescriptor classes in a migration package where they are dynamically loaded only as proxies when resuming a Flink-1.1 savepoint. That way we can change the classes and maintain backwards compatibility.

Removing `clear()` from`State`

I think it would be nice to keep clear() on the base State interface. Can you explain why you want to remove it? In my opinion, every state needs to be able to clear, so it makes sense to have this on the case interface.

If this is in preparation for the MapState, then the MapState can simply override the clear() method with different logic.

I think that the MapState needs to support clear as well, where it deletes the sub-map for the current key.
On the HeapStateBackend, this is quite easy, when we assume that each (key/namespace) has a map as the value (and the complete map can be dropped)
For the RocksDBStateBackend, it is a bit more expensive, and would correspond to a range-iteration-and-deletes.

If an application decides that the MapState clear() is to expensive, it can decide to not call it. But we should still support it for cases where it is necessary.

Changing the `State` interface

I would like to not change State to State<T> in the Flink master now, because it cases warnings in all parts of the code (that suddenly use raw types) and for some user programs.

While this would be done for Flink 2.0, it would make merging simpler if we don't change it now.

StephanEwen · 2017-01-26T16:38:49Z

Update: Just saw that you already did the migration support.

Will merge the StateDescriptor refactoring. Let's discuss the other two types of changes separately.

StephanEwen · 2017-01-31T15:46:57Z

I posted a rebased version of the state descriptor refactoring here: #3243

shixiaogang · 2017-02-04T02:00:30Z

@StephanEwen Thanks a lot for your comments.

Removing clear() from State

This change is suggested by @aljoscha who wants to let broadcast states share the same interface (see the discussion in FLINK-5023) . As mentioned, the broadcast states are read-only in some cases. Hence it's suggested not to provide the clear() method in the base State interface.

Changing the State interface

The State interface is typed because I want to provide the get() method for all states so that we can retrieve the data in the state (under the current key for keyed states). The functionality is already provided by all states except ValueState who provides the same functionality with the value() method. Providing the method for all states can help reduce some duplicated code in the implementation. It also makes sense for read-only states mentioned above.

aljoscha · 2017-02-04T16:02:00Z

In fact, my original suggestion was to add a new interface ReadableState:

interface ReadableState<T> {
  T get();
}

and leave all the existing interfaces as they are. That way we would have the least amount of changes in existing code and still have a common interface for state that can be read. (Note that ReadableState does not extends State on purpose, because of State.clear().)

shixiaogang · 2017-02-06T09:19:19Z

@aljoscha That way, it's very confusing that a ReadableState is not a State. Hence I made State read-only and introduced the UpdatableState interface who extends State with the method clear().

These changes (mainly the introduction of the get() method) are intended to remove the duplicated code. As they have little relationship with the implementation of map states. I think it's okay not to change these interfaces now.

But I prefer to rethink the state hierarchy in the near future because there exists too much duplicated code now.

shixiaogang · 2017-02-14T06:29:05Z

Close the pull request because the state descriptor now is refactored with the introduction of composited serializers (See FLINK-5790).

shixiaogang changed the title ~~[FLINK-5023 & FLINK-5024] Add SimpleStateDescriptor to clarify the concepts~~ [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to clarify the concepts Nov 8, 2016

StephanEwen reviewed Nov 8, 2016

View reviewed changes

aljoscha requested changes Nov 22, 2016

View reviewed changes

shixiaogang force-pushed the flink-5023 branch from d006d31 to 7062b58 Compare November 23, 2016 03:44

aljoscha mentioned this pull request Nov 23, 2016

Add Tests in StateBackendTestBase that verify Default-Value Behaviour #2855

Closed

shixiaogang force-pushed the flink-5023 branch from e78264b to e885c7d Compare December 1, 2016 08:06

xiaogang.sxg added 8 commits January 16, 2017 18:12

Refactor the interface of State and StateDescriptor

941159e

Add updatable state

71ef451

Remove unused type parameters

2709b34

Make folding states return null when the state is empty

b8f0794

Revert the unit test to reflect the behavior of folding states' defau…

156ba95

…lt value

Use default value as the initial accumulation value in folding states

c5971f8

Remove default value from SimpleStateDescriptor

b4966ed

Make new descriptors compatible with old savepoints

4666207

shixiaogang force-pushed the flink-5023 branch from e885c7d to 4666207 Compare January 17, 2017 11:16

StephanEwen mentioned this pull request Jan 31, 2017

[FLINK-5024] [core] Refactor the interface of State and StateDescriptor #3243

Closed

shixiaogang closed this Feb 14, 2017

shixiaogang deleted the flink-5023 branch February 15, 2017 03:40

rmetzger added the component=Runtime/StateBackends label Mar 14, 2019

[FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to clarify the concepts #2768

[FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to clarify the concepts #2768

Uh oh!

Conversation

shixiaogang commented Nov 8, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aljoscha left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

shixiaogang commented Nov 23, 2016

Uh oh!

aljoscha commented Nov 23, 2016

Uh oh!

shixiaogang commented Nov 24, 2016

Uh oh!

aljoscha commented Nov 24, 2016

Uh oh!

shixiaogang commented Nov 28, 2016

Uh oh!

aljoscha commented Nov 28, 2016

Uh oh!

shixiaogang commented Dec 1, 2016

Uh oh!

shixiaogang commented Jan 18, 2017

Uh oh!

StephanEwen commented Jan 25, 2017

Changing the StateDescriptor

Removing clear() fromState

Changing the State interface

Uh oh!

StephanEwen commented Jan 26, 2017

Uh oh!

StephanEwen commented Jan 31, 2017

Uh oh!

shixiaogang commented Feb 4, 2017

Uh oh!

aljoscha commented Feb 4, 2017

Uh oh!

shixiaogang commented Feb 6, 2017

Uh oh!

shixiaogang commented Feb 14, 2017

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

shixiaogang commented Nov 8, 2016 •

edited

Loading

Changing the `StateDescriptor`

Removing `clear()` from`State`

Changing the `State` interface