You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by shixiaogang <gi...@git.apache.org> on 2016/11/08 02:58:31 UTC
[GitHub] flink pull request #2768: [FLINK-5023 & FLINK-5024] Add SimpleStateDescripto...
GitHub user shixiaogang opened a pull request:
https://github.com/apache/flink/pull/2768
[FLINK-5023 & FLINK-5024] Add SimpleStateDescriptor to clarify the concepts
Changes in the definition of `State` and `StateDescriptor`:
- Add `get()` in the `State` interface.
- Remove type serializers of state values from `StateDescriptor`s.
- 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 `State`s in the implementation of window operators.
- Update the usage of `State`s in unit tests.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/alibaba/flink flink-5023
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/2768.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2768
----
commit 007a93b454e693bc3662f540ac5f33e899ce9058
Author: xiaogang.sxg <xi...@alibaba-inc.com>
Date: 2016-11-08T02:38:22Z
Refactor the interface of State and StateDescriptor
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to cla...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/2768
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.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to cla...
Posted by shixiaogang <gi...@git.apache.org>.
Github user shixiaogang commented on the issue:
https://github.com/apache/flink/pull/2768
I rebased the branch to resolve the conflicts with the master branch.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/2768#discussion_r89162070
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/Trigger.java ---
@@ -179,12 +179,13 @@ public TriggerResult onMerge(W window, OnMergeContext ctx) throws Exception {
*
* @param stateDescriptor The StateDescriptor that contains the name and type of the
* state that is being accessed.
+ * @param <V> The type of the values in the state.
* @param <S> The type of the state.
* @return The partitioned state object.
* @throws UnsupportedOperationException Thrown, if no partitioned state is available for the
* 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);
--- End diff --
Same thing as elsewhere: we don't need `V`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/2768#discussion_r89161821
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java ---
@@ -487,7 +487,7 @@ protected ProcessingTimeService getProcessingTimeService() {
* @throws IllegalStateException Thrown, if the key/value state was already initialized.
* @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 {
--- End diff --
Here, the same thing I mentioned on `KeyedStateBackend` holds, we don't need the additional `V` parameter.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to cla...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/2768
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()`.)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/2768#discussion_r86987978
--- Diff: flink-core/src/main/java/org/apache/flink/api/common/state/ListStateDescriptor.java ---
@@ -84,21 +138,37 @@ public boolean equals(Object o) {
ListStateDescriptor<?> that = (ListStateDescriptor<?>) o;
- return serializer.equals(that.serializer) && name.equals(that.name);
-
+ return elemTypeSerializer.equals(that.elemTypeSerializer) && name.equals(that.name);
}
@Override
public int hashCode() {
- int result = serializer.hashCode();
+ int result = elemTypeSerializer.hashCode();
result = 31 * result + name.hashCode();
return result;
}
@Override
public String toString() {
return "ListStateDescriptor{" +
- "serializer=" + serializer +
+ "elem serializer=" + elemTypeSerializer +
'}';
}
+
+ // ------------------------------------------------------------------------
+ // Serialization
+ // ------------------------------------------------------------------------
+
+ private void writeObject(final ObjectOutputStream out) throws IOException {
+ // make sure we have a serializer before the type information gets lost
+ initializeSerializerUnlessSet(new ExecutionConfig());
+
+ // write all the non-transient fields
+ out.defaultWriteObject();
+ }
+
+ private void readObject(final ObjectInputStream in) throws IOException, ClassNotFoundException {
--- End diff --
I think this method is not necessary, because it only triggers default serialization anyways.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to cla...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/2768
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?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to cla...
Posted by shixiaogang <gi...@git.apache.org>.
Github user shixiaogang commented on the issue:
https://github.com/apache/flink/pull/2768
Close the pull request because the state descriptor now is refactored with the introduction of composited serializers (See [FLINK-5790](https://issues.apache.org/jira/browse/FLINK-5790)).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to cla...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/2768
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.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/2768#discussion_r86987689
--- Diff: flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBSimpleState.java ---
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.contrib.streaming.state;
+
+import org.apache.flink.api.common.state.SimpleStateDescriptor;
+import org.apache.flink.api.common.state.State;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.util.Preconditions;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.WriteOptions;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+
+/**
+ * The implementation that stores simple states in RocksDB.
+ *
+ * @param <K> The type of the key.
+ * @param <N> The type of the namespace.
+ * @param <V> The type of value that the state state stores.
+ */
+public class RocksDBSimpleState<K, N, V> extends AbstractRocksDBState<K, N> implements State<V> {
+
+ /** State descriptor from which to create this state instance */
+ protected final SimpleStateDescriptor<V, ? extends State<V>> stateDesc;
+
+ /** Serializer for the values */
+ protected final TypeSerializer<V> valueSerializer;
+
+ /**
+ * We disable writes to the write-ahead-log here. We can't have these in the base class
+ * because JNI segfaults for some reason if they are.
+ */
+ protected final WriteOptions writeOptions;
+
+ /**
+ * Creates a new {@code RocksDBSimpleState}.
+ *
+ * @param namespaceSerializer The serializer for the namespace.
+ * @param stateDesc The state identifier for the state. This contains name and can create a default state value.
+ */
+ public RocksDBSimpleState(ColumnFamilyHandle columnFamily,
+ TypeSerializer<N> namespaceSerializer,
+ SimpleStateDescriptor<V, ? extends State<V>> stateDesc,
+ RocksDBKeyedStateBackend<K> backend
+ ) {
+ super(columnFamily, namespaceSerializer, backend);
+
+ this.stateDesc = Preconditions.checkNotNull(stateDesc, "State Descriptor");
+ this.valueSerializer = stateDesc.getSerializer();
+
+ writeOptions = new WriteOptions();
+ writeOptions.setDisableWAL(true);
+ }
+
+ @Override
+ public V get() {
+ try {
+ writeCurrentKeyWithGroupAndNamespace();
+ byte[] key = keySerializationStream.toByteArray();
+ byte[] valueBytes = backend.db.get(columnFamily, key);
+ if (valueBytes == null) {
+ return stateDesc.getDefaultValue();
+ }
+ return valueSerializer.deserialize(new DataInputViewStreamWrapper(new ByteArrayInputStream(valueBytes)));
--- End diff --
Given that this is used only in single threaded settings, we should try to reuse the `DataInputViewStreamWrapper` and `ByteArrayInputStream`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to cla...
Posted by shixiaogang <gi...@git.apache.org>.
Github user shixiaogang commented on the issue:
https://github.com/apache/flink/pull/2768
Despite the changes in the state descriptors, the Flink jobs can restore from old versioned snapshots now.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to cla...
Posted by shixiaogang <gi...@git.apache.org>.
Github user shixiaogang commented on the issue:
https://github.com/apache/flink/pull/2768
@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](https://issues.apache.org/jira/browse/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.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to cla...
Posted by shixiaogang <gi...@git.apache.org>.
Github user shixiaogang commented on the issue:
https://github.com/apache/flink/pull/2768
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
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/2768#discussion_r89129638
--- Diff: flink-core/src/main/java/org/apache/flink/api/common/state/ValueState.java ---
@@ -37,30 +37,24 @@
* @param <T> Type of the value in the state.
*/
@PublicEvolving
-public interface ValueState<T> extends State {
-
+public interface ValueState<T> extends UpdatableState<T> {
/**
- * Returns the current value for the state. When the state is not
- * 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.
- *
- * @return The operator state value corresponding to the current input.
- *
+ * Returns the current value for the state. The method performs the same
+ * functionality as {@link State#get()}.
+ *
* @throws IOException Thrown if the system cannot access the state.
*/
+ @Deprecated
--- End diff --
Please also add a `@deprecated` description in the Javadoc, pointing to use `get()`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/2768#discussion_r89160249
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedStateBackend.java ---
@@ -69,25 +69,26 @@
* @param stateDescriptor The identifier for the state. This contains name and can create a default state value.
* @param <N> The type of the namespace.
+ * @param <V> The type of the values in the state.
* @param <S> The type of the state.
*
* @return A new key/value state backed by this backend.
*
* @throws Exception Exceptions may occur during initialization of the state and should be forwarded.
*/
@SuppressWarnings({"rawtypes", "unchecked"})
- <N, S extends State> S getPartitionedState(
+ <N, V, S extends State<V>> S getPartitionedState(
--- End diff --
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()`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to cla...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/2768
@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.)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to cla...
Posted by shixiaogang <gi...@git.apache.org>.
Github user shixiaogang commented on the issue:
https://github.com/apache/flink/pull/2768
I moved default value from `SimpleStateDescriptor` to `ValueStateDescriptor`. Now only `ValueStateDescriptor`s have default values. The serialization methods may contain some duplicated code, but i think it's acceptable.
I also modify the implementation of `HeapReducingState`s. The first value will be copied before being put into the heap.
@aljoscha What do you think of these changes?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor...
Posted by shixiaogang <gi...@git.apache.org>.
Github user shixiaogang closed the pull request at:
https://github.com/apache/flink/pull/2768
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to cla...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/2768
I posted a rebased version of the state descriptor refactoring here: #3243
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to cla...
Posted by shixiaogang <gi...@git.apache.org>.
Github user shixiaogang commented on the issue:
https://github.com/apache/flink/pull/2768
@aljoscha Thanks for your review. I have updated the PR according to your suggestion.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/2768#discussion_r89128914
--- Diff: flink-core/src/main/java/org/apache/flink/api/common/state/SimpleStateDescriptor.java ---
@@ -0,0 +1,226 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.common.state;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Base class for the descriptors of simple states. A {@code SimpleStateDescriptor} is used
+ * for creating partitioned simple states whose values are not composited.
+ *
+ * <p>Subclasses must correctly implement {@link #equals(Object)} and {@link #hashCode()}.
+ *
+ * @param <T> The type of the value in the state.
+ * @param <S> The type of the created state.
+ */
+@PublicEvolving
+public abstract class SimpleStateDescriptor<T, S extends State<T>> extends StateDescriptor<S> {
+ private static final long serialVersionUID = 1L;
+
+ /** The serializer for the type. May be eagerly initialized in the constructor,
+ * or lazily once the type is serialized or an ExecutionConfig is provided. */
+ protected TypeSerializer<T> typeSerializer;
+
+ /** The type information describing the value type. Only used to lazily create the serializer
+ * and dropped during serialization */
+ 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;
--- End diff --
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)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to cla...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/2768
@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.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor to cla...
Posted by shixiaogang <gi...@git.apache.org>.
Github user shixiaogang commented on the issue:
https://github.com/apache/flink/pull/2768
@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.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/2768#discussion_r89129366
--- Diff: flink-core/src/main/java/org/apache/flink/api/common/state/State.java ---
@@ -20,19 +20,30 @@
import org.apache.flink.annotation.PublicEvolving;
+import java.io.IOException;
+
/**
* Interface that different types of partitioned state must implement.
*
* <p>The state is only accessible by functions applied on a KeyedDataStream. The key is
* automatically supplied by the system, so the function always sees the value mapped to the
* key of the current element. That way, the system can handle stream and state partitioning
* consistently together.
+ *
+ * @param <T> The type of the values in the state.
*/
@PublicEvolving
-public interface State {
-
+public interface State<T> {
/**
- * Removes the value mapped under the current key.
+ * Returns the current value for the state. When the state is not
+ * 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.
--- End diff --
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.)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/2768#discussion_r89161552
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/query/AbstractQueryableStateOperator.java ---
@@ -28,16 +28,17 @@
/**
* Internal operator handling queryable state instances (setup and update).
*
+ * @param <V> Value type
* @param <S> State type
* @param <IN> Input type
*/
@Internal
-abstract class AbstractQueryableStateOperator<S extends State, IN>
+abstract class AbstractQueryableStateOperator<V, S extends State<V>, IN>
--- End diff --
I think here the same thing I mentioned earlier holds, we don't really need the additional `V` parameter.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2768: [FLINK-5023][FLINK-5024] Add SimpleStateDescriptor...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/2768#discussion_r89130528
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapSimpleState.java ---
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.state.heap;
+
+import org.apache.flink.api.common.state.SimpleStateDescriptor;
+import org.apache.flink.api.common.state.State;
+import org.apache.flink.api.common.state.UpdatableState;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.query.netty.message.KvStateRequestSerializer;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.util.Preconditions;
+
+import java.util.Map;
+
+/**
+ * Heap-backed partitioned states whose values are not composited and is snapshotted into files.
--- End diff --
the `is snapshotted into files` portion is not always true, and also not necessary here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---