You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ab...@apache.org on 2026/09/15 10:12:17 UTC
(fluss) branch main updated: [server] remove multi-key partitions from cache correctly (#3882)
This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 7a1214710 [server] remove multi-key partitions from cache correctly (#3882)
7a1214710 is described below
commit 7a12147109ccfa2f4a347640662b5dc1498bfd79
Author: Junfan Zhang <zu...@apache.org>
AuthorDate: Tue Sep 15 18:12:13 2026 +0800
[server] remove multi-key partitions from cache correctly (#3882)
* [server] remove multi-key partitions from cache correctly
* addressed comments by anton
* rollback -1 case
---------
Co-authored-by: zhangjunfan <zh...@qiyi.com>
---
.../server/coordinator/AutoPartitionManager.java | 47 +++++++++++++++++++++-
.../coordinator/AutoPartitionManagerTest.java | 35 ++++++++++++++++
2 files changed, 81 insertions(+), 1 deletion(-)
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java
index 277b72bf3..39a36f772 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java
@@ -354,7 +354,10 @@ public class AutoPartitionManager implements AutoCloseable {
lock,
() -> {
if (autoPartitionTables.containsKey(tableId)) {
- partitionsByTable.get(tableId).remove(partitionName);
+ removePartitionFromPartitionsByTable(
+ autoPartitionTables.get(tableId),
+ partitionsByTable.get(tableId),
+ partitionName);
}
});
}
@@ -399,6 +402,24 @@ public class AutoPartitionManager implements AutoCloseable {
}
}
+ private void removePartitionFromPartitionsByTable(
+ TableInfo tableInfo,
+ NavigableMap<String, Set<String>> partitionMap,
+ String partitionName) {
+ if (tableInfo.getPartitionKeys().size() > 1) {
+ String autoPartitionValue = extractAutoPartitionValue(tableInfo, partitionName);
+ Set<String> partitionSet = partitionMap.get(autoPartitionValue);
+ if (partitionSet != null) {
+ partitionSet.remove(partitionName);
+ if (partitionSet.isEmpty()) {
+ partitionMap.remove(autoPartitionValue);
+ }
+ }
+ } else {
+ partitionMap.remove(partitionName);
+ }
+ }
+
private void doAutoPartition() {
Instant now = clock.instant();
inLock(lock, () -> doAutoPartition(now, autoPartitionTables.keySet(), false));
@@ -679,6 +700,30 @@ public class AutoPartitionManager implements AutoCloseable {
return autoCreateDayPartitionDelayMinutes.get(tableId);
}
+ /** Returns a snapshot of the cached partitions for the specified table. */
+ @VisibleForTesting
+ @Nullable
+ protected Map<String, Set<String>> getPartitionsByTable(long tableId) {
+ return inLock(
+ lock,
+ () -> {
+ TreeMap<String, Set<String>> partitionMap = partitionsByTable.get(tableId);
+ if (partitionMap == null) {
+ return null;
+ }
+
+ Map<String, Set<String>> snapshot = new TreeMap<>();
+ partitionMap.forEach(
+ (partitionTime, partitionSet) ->
+ snapshot.put(
+ partitionTime,
+ partitionSet == null
+ ? null
+ : new HashSet<>(partitionSet)));
+ return snapshot;
+ });
+ }
+
@Override
public void close() throws Exception {
if (isClosed.compareAndSet(false, true)) {
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java
index 6f470255a..7b75e4605 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java
@@ -680,6 +680,41 @@ class AutoPartitionManagerTest {
assertThat(partitions.keySet()).contains("20250420", "20250421", "20250422");
}
+ @Test
+ void testRemovePartitionFromMultiplePartitionKeysTable() throws Exception {
+ ManuallyTriggeredScheduledExecutorService periodicExecutor =
+ new ManuallyTriggeredScheduledExecutorService();
+ AutoPartitionManager autoPartitionManager =
+ new AutoPartitionManager(
+ new TestingServerMetadataCache(3),
+ metadataManager,
+ remoteDirDynamicLoader,
+ new Configuration(),
+ disabledCapacityController(),
+ new ManualClock(0L),
+ periodicExecutor);
+
+ TableInfo table = createPartitionedTable(2, 0, AutoPartitionTimeUnit.DAY, true);
+ long tableId = table.getTableId();
+ autoPartitionManager.addAutoPartitionTable(table, false);
+ autoPartitionManager.addPartition(tableId, "20250419$A");
+ autoPartitionManager.addPartition(tableId, "20250419$B");
+
+ assertThat(autoPartitionManager.getPartitionsByTable(tableId))
+ .containsEntry(
+ "20250419", new HashSet<>(Arrays.asList("20250419$A", "20250419$B")));
+
+ autoPartitionManager.removePartition(tableId, "20250419$A");
+
+ assertThat(autoPartitionManager.getPartitionsByTable(tableId))
+ .containsEntry("20250419", new HashSet<>(Arrays.asList("20250419$B")));
+
+ autoPartitionManager.removePartition(tableId, "20250419$B");
+
+ assertThat(autoPartitionManager.getPartitionsByTable(tableId))
+ .doesNotContainKey("20250419");
+ }
+
/**
* Test if AutoPartitionManager.createPartition applies maxBucketLimit per partition while
* adding new partition automatically.