This is an automated email from the ASF dual-hosted git repository. hangxiang pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2c5078bc9051a9ffadcc404b1829aa46b8f82776 Author: Hangxiang Yu <[email protected]> AuthorDate: Wed Apr 17 10:51:31 2024 +0800 [FLINK-34987][state] Implement Async Value State (#24651) --- .../flink/runtime/state/v2/InternalValueState.java | 49 ++++++++++++++++++ .../runtime/state/v2/ValueStateDescriptor.java | 60 ++++++++++++++++++++++ .../AsyncExecutionControllerTest.java | 29 +++-------- 3 files changed, 116 insertions(+), 22 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalValueState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalValueState.java new file mode 100644 index 00000000000..72122bf4cf2 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/InternalValueState.java @@ -0,0 +1,49 @@ +/* + * 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.v2; + +import org.apache.flink.api.common.state.v2.StateFuture; +import org.apache.flink.api.common.state.v2.ValueState; +import org.apache.flink.runtime.asyncprocessing.AsyncExecutionController; +import org.apache.flink.runtime.asyncprocessing.StateRequestType; + +/** + * A default implementation of {@link ValueState} which delegates all async requests to {@link + * AsyncExecutionController}. + * + * @param <K> The type of key the state is associated to. + * @param <V> The type of values kept internally in state. + */ +public class InternalValueState<K, V> extends InternalKeyedState<K, V> implements ValueState<V> { + + public InternalValueState( + AsyncExecutionController<K> asyncExecutionController, + ValueStateDescriptor<V> valueStateDescriptor) { + super(asyncExecutionController, valueStateDescriptor); + } + + @Override + public final StateFuture<V> asyncValue() { + return handleRequest(StateRequestType.VALUE_GET, null); + } + + @Override + public final StateFuture<Void> asyncUpdate(V value) { + return handleRequest(StateRequestType.VALUE_UPDATE, value); + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/ValueStateDescriptor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/ValueStateDescriptor.java new file mode 100644 index 00000000000..aa63e76ab5b --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/ValueStateDescriptor.java @@ -0,0 +1,60 @@ +/* + * 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.v2; + +import org.apache.flink.api.common.serialization.SerializerConfig; +import org.apache.flink.api.common.state.v2.ValueState; +import org.apache.flink.api.common.typeinfo.TypeInformation; + +/** + * {@link StateDescriptor} for {@link ValueState}. This can be used to create partitioned value + * state internally. + * + * @param <T> The type of the values that the value state can hold. + */ +public class ValueStateDescriptor<T> extends StateDescriptor<T> { + + /** + * Creates a new {@code ValueStateDescriptor} with the given stateId and type. + * + * @param stateId The (unique) stateId for the state. + * @param typeInfo The type of the values in the state. + */ + public ValueStateDescriptor(String stateId, TypeInformation<T> typeInfo) { + super(stateId, typeInfo); + } + + /** + * Creates a new {@code ValueStateDescriptor} with the given stateId and type. + * + * @param stateId The (unique) stateId for the state. + * @param typeInfo The type of the values in the state. + * @param serializerConfig The serializer related config used to generate {@code + * TypeSerializer}. + */ + public ValueStateDescriptor( + String stateId, TypeInformation<T> typeInfo, SerializerConfig serializerConfig) { + super(stateId, typeInfo, serializerConfig); + } + + @Override + public Type getType() { + return Type.VALUE; + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java index d36a5e53ae6..9f1e1a203a5 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java @@ -18,14 +18,16 @@ package org.apache.flink.runtime.asyncprocessing; -import org.apache.flink.api.common.state.v2.StateFuture; -import org.apache.flink.api.common.state.v2.ValueState; +import org.apache.flink.api.common.typeinfo.BasicTypeInfo; +import org.apache.flink.api.common.typeutils.base.IntSerializer; import org.apache.flink.core.state.StateFutureUtils; import org.apache.flink.runtime.mailbox.SyncMailboxExecutor; import org.apache.flink.runtime.state.AsyncKeyedStateBackend; import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend; import org.apache.flink.runtime.state.OperatorStateBackend; import org.apache.flink.runtime.state.StateBackend; +import org.apache.flink.runtime.state.v2.InternalValueState; +import org.apache.flink.runtime.state.v2.ValueStateDescriptor; import org.apache.flink.util.Preconditions; import org.junit.jupiter.api.BeforeEach; @@ -368,32 +370,15 @@ class AsyncExecutionControllerTest { } } - static class TestValueState implements ValueState<Integer> { - - private final AsyncExecutionController<String> asyncExecutionController; + static class TestValueState extends InternalValueState<String, Integer> { private final TestUnderlyingState underlyingState; public TestValueState( AsyncExecutionController<String> aec, TestUnderlyingState underlyingState) { - this.asyncExecutionController = aec; + super(aec, new ValueStateDescriptor<>("test-value-state", BasicTypeInfo.INT_TYPE_INFO)); this.underlyingState = underlyingState; - } - - @Override - public StateFuture<Void> asyncClear() { - return asyncExecutionController.handleRequest(this, StateRequestType.CLEAR, null); - } - - @Override - public StateFuture<Integer> asyncValue() { - return asyncExecutionController.handleRequest(this, StateRequestType.VALUE_GET, null); - } - - @Override - public StateFuture<Void> asyncUpdate(Integer value) { - return asyncExecutionController.handleRequest( - this, StateRequestType.VALUE_UPDATE, value); + assertThat(this.getValueSerializer()).isEqualTo(IntSerializer.INSTANCE); } }
