This is an automated email from the ASF dual-hosted git repository.
1996fanrui pushed a commit to branch release-1.20
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.20 by this push:
new d9c00f69933 [FLINK-38558] Require copy when use heap timer
d9c00f69933 is described below
commit d9c00f69933f7e2879763ada5ebc252a5cf28a95
Author: yangjunsen <[email protected]>
AuthorDate: Wed Nov 4 03:48:59 2026 +0800
[FLINK-38558] Require copy when use heap timer
---
...logDelegateEmbeddedRocksDBStateBackendTest.java | 10 +-
.../streaming/state/RocksDBKeyedStateBackend.java | 2 +-
.../state/EmbeddedRocksDBStateBackendTest.java | 26 ++++-
...ksDBHeapTimerWindowDistinctAggregateITCase.java | 113 +++++++++++++++++++++
4 files changed, 140 insertions(+), 11 deletions(-)
diff --git
a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogDelegateEmbeddedRocksDBStateBackendTest.java
b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogDelegateEmbeddedRocksDBStateBackendTest.java
index 2b49d3df55e..0b10844aef0 100644
---
a/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogDelegateEmbeddedRocksDBStateBackendTest.java
+++
b/flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogDelegateEmbeddedRocksDBStateBackendTest.java
@@ -41,6 +41,8 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
import java.nio.file.Path;
+import static org.junit.jupiter.api.Assumptions.assumeFalse;
+
/** Tests for {@link ChangelogStateBackend} delegating {@link
EmbeddedRocksDBStateBackend}. */
public class ChangelogDelegateEmbeddedRocksDBStateBackendTest
extends EmbeddedRocksDBStateBackendTest {
@@ -62,11 +64,6 @@ public class ChangelogDelegateEmbeddedRocksDBStateBackendTest
return false;
}
- @Override
- protected boolean isSafeToReuseKVState() {
- return true;
- }
-
@TestTemplate
@Disabled("The type of handle returned from snapshot() is not incremental")
public void testSharedIncrementalStateDeRegistration() {}
@@ -118,6 +115,9 @@ public class
ChangelogDelegateEmbeddedRocksDBStateBackendTest
@TestTemplate
public void testMaterializedRestorePriorityQueue() throws Exception {
+ assumeFalse(
+ useHeapTimer,
+ "Heap priority queue does not support restore test on managed
keyed state");
CheckpointStreamFactory streamFactory = createStreamFactory();
ChangelogStateBackendTestUtils.testMaterializedRestoreForPriorityQueue(
diff --git
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
index 4fea9ec0bdc..720681533ae 100644
---
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
+++
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
@@ -1052,7 +1052,7 @@ public class RocksDBKeyedStateBackend<K> extends
AbstractKeyedStateBackend<K> {
@Override
public boolean isSafeToReuseKVState() {
- return true;
+ return !(priorityQueueFactory instanceof HeapPriorityQueueSetFactory);
}
/** Rocks DB specific information about the k/v states. */
diff --git
a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackendTest.java
b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackendTest.java
index a8277fa07fc..a298958e5e9 100644
---
a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackendTest.java
+++
b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/EmbeddedRocksDBStateBackendTest.java
@@ -99,6 +99,7 @@ import java.util.stream.Collectors;
import static
org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.INCREMENTAL_RESTORE_ASYNC_COMPACT_AFTER_RESCALE;
import static
org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.USE_DELETE_FILES_IN_RANGE_DURING_RESCALING;
import static
org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.USE_INGEST_DB_RESTORE_MODE;
+import static
org.apache.flink.contrib.streaming.state.RocksDBOptions.TIMER_SERVICE_FACTORY;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -131,13 +132,15 @@ public class EmbeddedRocksDBStateBackendTest
true,
(SupplierWithException<CheckpointStorage, IOException>)
JobManagerCheckpointStorage::new,
+ false,
false
},
{
true,
(SupplierWithException<CheckpointStorage, IOException>)
JobManagerCheckpointStorage::new,
- true
+ true,
+ false
},
{
false,
@@ -148,8 +151,16 @@ public class EmbeddedRocksDBStateBackendTest
return new FileSystemCheckpointStorage(
new Path(checkpointPath), 0, -1);
},
+ false,
false
- }
+ },
+ {
+ true,
+ (SupplierWithException<CheckpointStorage, IOException>)
+ JobManagerCheckpointStorage::new,
+ false,
+ true
+ },
});
}
@@ -162,6 +173,9 @@ public class EmbeddedRocksDBStateBackendTest
@Parameter(value = 2)
public boolean useIngestDB;
+ @Parameter(value = 3)
+ public boolean useHeapTimer;
+
// Store it because we need it for the cleanup test.
private String dbPath;
private RocksDB db = null;
@@ -203,8 +217,10 @@ public class EmbeddedRocksDBStateBackendTest
Configuration configuration = new Configuration();
configuration.set(USE_INGEST_DB_RESTORE_MODE, useIngestDB);
configuration.set(
- RocksDBOptions.TIMER_SERVICE_FACTORY,
- EmbeddedRocksDBStateBackend.PriorityQueueStateType.ROCKSDB);
+ TIMER_SERVICE_FACTORY,
+ useHeapTimer
+ ?
EmbeddedRocksDBStateBackend.PriorityQueueStateType.HEAP
+ :
EmbeddedRocksDBStateBackend.PriorityQueueStateType.ROCKSDB);
configuration.set(RocksDBManualCompactionOptions.MIN_INTERVAL,
Duration.ofMillis(1));
return configuration;
}
@@ -234,7 +250,7 @@ public class EmbeddedRocksDBStateBackendTest
@Override
protected boolean isSafeToReuseKVState() {
- return true;
+ return !useHeapTimer;
}
// small safety net for instance cleanups, so that no native objects are
left
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/RocksDBHeapTimerWindowDistinctAggregateITCase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/RocksDBHeapTimerWindowDistinctAggregateITCase.java
new file mode 100644
index 00000000000..ce9d9d0ed53
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/RocksDBHeapTimerWindowDistinctAggregateITCase.java
@@ -0,0 +1,113 @@
+/*
+ * 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.table.planner.runtime.stream.sql;
+
+import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.table.planner.runtime.utils.StreamingWithStateTestBase;
+import org.apache.flink.types.Row;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests split distinct window aggregation with RocksDB state and heap
timers. */
+class RocksDBHeapTimerWindowDistinctAggregateITCase extends
StreamingWithStateTestBase {
+
+ RocksDBHeapTimerWindowDistinctAggregateITCase() {
+ super(StreamingWithStateTestBase.ROCKSDB_BACKEND());
+ }
+
+ @BeforeEach
+ @Override
+ public void before() {
+ super.before();
+
+ final EmbeddedRocksDBStateBackend stateBackend = new
EmbeddedRocksDBStateBackend();
+ stateBackend.setPriorityQueueStateType(
+ EmbeddedRocksDBStateBackend.PriorityQueueStateType.HEAP);
+ env().setStateBackend(stateBackend);
+
+ final String dataId = TestValuesTableFactory.registerData(testData());
+ tEnv().executeSql(
+ "CREATE TABLE T1 (\n"
+ + " `id` BIGINT,\n"
+ + " `ts` STRING,\n"
+ + " `card_no_id` AS MOD(`id`, 1000),\n"
+ + " `rowtime` AS TO_TIMESTAMP(`ts`),\n"
+ + " WATERMARK FOR `rowtime` AS `rowtime` -
INTERVAL '1' SECOND\n"
+ + ") WITH (\n"
+ + " 'connector' = 'values',\n"
+ + " 'data-id' = '"
+ + dataId
+ + "'\n"
+ + ")");
+ tEnv().getConfig()
+
.set(OptimizerConfigOptions.TABLE_OPTIMIZER_DISTINCT_AGG_SPLIT_ENABLED, true);
+ }
+
+ @Test
+ void testCumulateWindowDistinctCountDoesNotLoseHeapTimerKeys() throws
Exception {
+ tEnv().executeSql(
+ "CREATE TABLE sink (\n"
+ + " window_start TIMESTAMP(3),\n"
+ + " window_end TIMESTAMP(3),\n"
+ + " distinct_cnt BIGINT\n"
+ + ") WITH ('connector' = 'values')");
+
+ tEnv().executeSql(
+ "INSERT INTO sink\n"
+ + "SELECT window_start, window_end,
COUNT(DISTINCT `card_no_id`)\n"
+ + "FROM TABLE(\n"
+ + " CUMULATE(\n"
+ + " TABLE T1,\n"
+ + " DESCRIPTOR(rowtime),\n"
+ + " INTERVAL '5' SECOND,\n"
+ + " INTERVAL '20' SECOND))\n"
+ + "GROUP BY window_start, window_end")
+ .await();
+
+ assertThat(TestValuesTableFactory.getResultsAsStrings("sink"))
+ .containsExactlyInAnyOrder(
+ "+I[2020-10-10T00:00, 2020-10-10T00:00:05, 350]",
+ "+I[2020-10-10T00:00, 2020-10-10T00:00:10, 700]",
+ "+I[2020-10-10T00:00, 2020-10-10T00:00:15, 1000]",
+ "+I[2020-10-10T00:00, 2020-10-10T00:00:20, 1000]");
+ }
+
+ private static List<Row> testData() {
+ final int recordsPerSecond = 70;
+ final int windowSizeSeconds = 20;
+ final LocalDateTime start = LocalDateTime.of(2020, 10, 10, 0, 0);
+ final DateTimeFormatter formatter =
DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+ final List<Row> rows = new ArrayList<>(recordsPerSecond *
windowSizeSeconds);
+
+ for (int id = 0; id < recordsPerSecond * windowSizeSeconds; id++) {
+ rows.add(Row.of((long) id, start.plusSeconds(id /
recordsPerSecond).format(formatter)));
+ }
+ return rows;
+ }
+}
