You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ha...@apache.org on 2023/08/31 03:22:51 UTC
[flink] branch master updated: [FLINK-32975][state] Enhance equals() for all MapState's iterator (#23305)
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
The following commit(s) were added to refs/heads/master by this push:
new aa8d93ea239 [FLINK-32975][state] Enhance equals() for all MapState's iterator (#23305)
aa8d93ea239 is described below
commit aa8d93ea239f5be79066b7e5caad08d966c86ab2
Author: xiarui <xi...@gmail.com>
AuthorDate: Mon Aug 28 18:00:09 2023 +0800
[FLINK-32975][state] Enhance equals() for all MapState's iterator (#23305)
---
.../flink/runtime/state/StateBackendTestBase.java | 35 ++++++++++++++++++++++
.../flink/state/changelog/ChangelogMapState.java | 11 +++++++
.../contrib/streaming/state/RocksDBMapState.java | 10 +++++++
3 files changed, 56 insertions(+)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
index 27eadd320ef..8f2495646f1 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
@@ -97,6 +97,7 @@ import org.junit.rules.ExpectedException;
import java.io.IOException;
import java.io.Serializable;
+import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -3591,6 +3592,40 @@ public abstract class StateBackendTestBase<B extends AbstractStateBackend> exten
}
}
+ /** Verify that iterator of {@link MapState} compares on the content. */
+ @Test
+ public void testMapStateEntryCompare() throws Exception {
+ MapStateDescriptor<Integer, Long> stateDesc1 =
+ new MapStateDescriptor<>("map-state-1", Integer.class, Long.class);
+ MapStateDescriptor<Integer, Long> stateDesc2 =
+ new MapStateDescriptor<>("map-state-2", Integer.class, Long.class);
+
+ CheckpointableKeyedStateBackend<Integer> backend =
+ createKeyedBackend(IntSerializer.INSTANCE);
+
+ try {
+ MapState<Integer, Long> state1 =
+ backend.getPartitionedState(
+ VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, stateDesc1);
+ MapState<Integer, Long> state2 =
+ backend.getPartitionedState(
+ VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, stateDesc2);
+
+ Map.Entry<Integer, Long> expectedEntry = new AbstractMap.SimpleEntry<>(0, 10L);
+ backend.setCurrentKey(1);
+ state1.put(expectedEntry.getKey(), expectedEntry.getValue());
+ state2.put(expectedEntry.getKey(), expectedEntry.getValue());
+
+ assertEquals(state1.entries().iterator().next(), expectedEntry);
+ assertEquals(state2.entries().iterator().next(), expectedEntry);
+
+ assertEquals(state1.entries().iterator().next(), state2.entries().iterator().next());
+ } finally {
+ IOUtils.closeQuietly(backend);
+ backend.dispose();
+ }
+ }
+
/** Verify that {@link ValueStateDescriptor} allows {@code null} as default. */
@Test
public void testValueStateNullAsDefaultValue() throws Exception {
diff --git a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogMapState.java b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogMapState.java
index 02ce3b73bf2..f2829655203 100644
--- a/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogMapState.java
+++ b/flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogMapState.java
@@ -35,6 +35,7 @@ import org.apache.flink.util.function.ThrowingConsumer;
import java.io.IOException;
import java.util.Iterator;
import java.util.Map;
+import java.util.Objects;
/**
* Delegated partitioned {@link MapState} that forwards changes to {@link StateChange} upon {@link
@@ -83,6 +84,16 @@ class ChangelogMapState<K, N, UK, UV>
}
return oldValue;
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof Map.Entry)) {
+ return false;
+ }
+ Map.Entry<?, ?> e = (Map.Entry<?, ?>) o;
+ return Objects.equals(entry.getKey(), e.getKey())
+ && Objects.equals(entry.getValue(), e.getValue());
+ }
};
}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBMapState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBMapState.java
index e7a3a653632..58a58d8f157 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBMapState.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBMapState.java
@@ -51,6 +51,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.Map;
+import java.util.Objects;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -537,6 +538,15 @@ class RocksDBMapState<K, N, UK, UV> extends AbstractRocksDBState<K, N, Map<UK, U
return oldValue;
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof Map.Entry)) {
+ return false;
+ }
+ Map.Entry<?, ?> e = (Map.Entry<?, ?>) o;
+ return Objects.equals(getKey(), e.getKey()) && Objects.equals(getValue(), e.getValue());
+ }
}
/** An auxiliary utility to scan all entries under the given key. */