You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "fredia (via GitHub)" <gi...@apache.org> on 2024/09/05 07:03:50 UTC
Re: [PR] [FLINK-34975] Implement Aggregating Async State API for ForStStateBackend [flink]
fredia commented on code in PR #25125:
URL: https://github.com/apache/flink/pull/25125#discussion_r1744908188
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestType.java:
##########
@@ -111,6 +111,6 @@ public enum StateRequestType {
/** Get value from aggregating state by {@link AggregatingState#asyncGet()}. */
AGGREGATING_GET,
- /** Add element to aggregating state by {@link AggregatingState#asyncAdd(Object)}. */
- AGGREGATING_ADD
+ /** Put element to aggregating state by {@link AggregatingState#asyncAdd(Object)}. */
+ AGGREGATING_PUT
Review Comment:
Why was `AGGREGATING_ADD` renamed here?
##########
flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateExecutorTest.java:
##########
@@ -120,6 +120,113 @@ public void testExecuteValueStateRequest() throws Exception {
forStStateExecutor.shutdown();
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testExecuteAggregatingStateRequest() throws Exception {
+ ForStStateExecutor forStStateExecutor = new ForStStateExecutor(4, db, new WriteOptions());
+ ForStAggregatingState<String, ?, Integer, Integer, Integer> state =
+ buildForStSumAggregateState("agg-state-1");
+
+ StateRequestContainer stateRequestContainer =
+ forStStateExecutor.createStateRequestContainer();
+ assertTrue(stateRequestContainer.isEmpty());
+
+ // 1. init aggregateValue for every 1000 key
+ int keyNum = 1000;
+ for (int i = 0; i < keyNum; i++) {
+ stateRequestContainer.offer(
+ buildStateRequest(
+ state, StateRequestType.AGGREGATING_PUT, "" + i, i, "record" + i));
+ }
+
+ forStStateExecutor.executeBatchRequests(stateRequestContainer).get();
+
+ // 2. get all value and verify
+ List<StateRequest<String, Integer, Integer>> requests = new ArrayList<>();
+ stateRequestContainer = forStStateExecutor.createStateRequestContainer();
+ for (int i = 0; i < keyNum; i++) {
+ StateRequest<String, Integer, Integer> r =
+ (StateRequest<String, Integer, Integer>)
+ buildStateRequest(
+ state,
+ StateRequestType.AGGREGATING_GET,
+ "" + i,
+ null,
+ "record" + i);
+ requests.add(r);
+ stateRequestContainer.offer(r);
+ }
+
+ forStStateExecutor.executeBatchRequests(stateRequestContainer).get();
+
+ for (StateRequest<String, Integer, Integer> request : requests) {
+ assertThat(request.getRequestType()).isEqualTo(StateRequestType.AGGREGATING_GET);
+ String key = request.getRecordContext().getKey();
+ assertThat(request.getRecordContext().getRecord()).isEqualTo("record" + key);
+ assertThat(((TestStateFuture<Integer>) request.getFuture()).getCompletedResult())
+ .isEqualTo(Integer.parseInt(key));
+ }
+
+ // 3. add more value for the aggregate state
+ stateRequestContainer = forStStateExecutor.createStateRequestContainer();
+ int addCnt = 10;
+ for (int i = 0; i < keyNum; i++) {
+ for (int j = 0; j < addCnt; j++) {
+ StateRequest<String, Integer, Integer> r =
+ (StateRequest<String, Integer, Integer>)
+ buildStateRequest(
+ state, StateRequestType.AGGREGATING_PUT, "" + i, 1, i * 2);
+ requests.add(r);
+ stateRequestContainer.offer(r);
+ }
+ }
+
+ // clear first 100 state
+ for (int i = 0; i < 100; i++) {
+ StateRequest<String, Integer, Integer> r =
+ (StateRequest<String, Integer, Integer>)
+ buildStateRequest(
+ state, StateRequestType.CLEAR, "" + i, null, "record" + i);
+ requests.add(r);
+ stateRequestContainer.offer(r);
+ }
+
+ forStStateExecutor.executeBatchRequests(stateRequestContainer).get();
+
+ // 4. read and verify the updated aggregate state
+ requests = new ArrayList<>();
+ stateRequestContainer = forStStateExecutor.createStateRequestContainer();
+ for (int i = 0; i < keyNum; i++) {
+ StateRequest<String, Integer, Integer> r =
+ (StateRequest<String, Integer, Integer>)
+ buildStateRequest(
+ state,
+ StateRequestType.AGGREGATING_GET,
+ "" + i,
+ null,
+ "record" + i);
+ requests.add(r);
+ stateRequestContainer.offer(r);
+ }
+
+ forStStateExecutor.executeBatchRequests(stateRequestContainer).get();
+
+ for (int i = 0; i < 100; i++) {
+ StateRequest<String, Integer, Integer> request = requests.get(i);
+ assertThat(request.getRequestType()).isEqualTo(StateRequestType.AGGREGATING_GET);
+ assertThat(((TestStateFuture<Integer>) request.getFuture()).getCompletedResult())
+ .isEqualTo(null);
+ }
+ for (int i = 100; i < keyNum; i++) {
+ StateRequest<String, Integer, Integer> request = requests.get(i);
+ assertThat(request.getRequestType()).isEqualTo(StateRequestType.AGGREGATING_GET);
+ String key = request.getRecordContext().getKey();
+ assertThat(request.getRecordContext().getRecord()).isEqualTo("record" + key);
+ assertThat(((TestStateFuture<Integer>) request.getFuture()).getCompletedResult())
+ .isEqualTo(1);
Review Comment:
IIUC, the result should be equal to `key+1`, could you please illustrate why this is equal to `1`?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org