map.remove_all implementation (#566)

map.remove_all is implemented

Author: feysahin

examples/map/map_remove_all_example.py

@@ -0,0 +1,18 @@
+import hazelcast
+
+from hazelcast.predicate import between
+
+client = hazelcast.HazelcastClient()
+
+predicate_map = client.get_map("predicate-map").blocking()
+
+for i in range(10):
+    predicate_map.put("key" + str(i), i)
+
+predicate = between("this", 3, 5)
+
+predicate_map.remove_all(predicate)
+
+print(predicate_map.values())
+
+client.shutdown()

hazelcast/protocol/codec/map_remove_all_codec.py

@@ -0,0 +1,17 @@
+from hazelcast.protocol.client_message import OutboundMessage, REQUEST_HEADER_SIZE, create_initial_buffer
+from hazelcast.protocol.builtin import StringCodec
+from hazelcast.protocol.builtin import DataCodec
+
+# hex: 0x013E00
+_REQUEST_MESSAGE_TYPE = 81408
+# hex: 0x013E01
+_RESPONSE_MESSAGE_TYPE = 81409
+
+_REQUEST_INITIAL_FRAME_SIZE = REQUEST_HEADER_SIZE
+
+
+def encode_request(name, predicate):
+    buf = create_initial_buffer(_REQUEST_INITIAL_FRAME_SIZE, _REQUEST_MESSAGE_TYPE)
+    StringCodec.encode(buf, name)
+    DataCodec.encode(buf, predicate, True)
+    return OutboundMessage(buf, False)

hazelcast/proxy/map.py

@@ -70,6 +70,7 @@
     map_put_transient_with_max_idle_codec,
     map_set_with_max_idle_codec,
     map_remove_interceptor_codec,
+    map_remove_all_codec,
 )
 from hazelcast.proxy.base import (
     Proxy,
@@ -1283,6 +1284,28 @@ def remove(self, key: KeyType) -> Future[typing.Optional[ValueType]]:
 
         return self._remove_internal(key_data)
 
+    def remove_all(self, predicate: Predicate) -> Future[None]:
+        """Removes all entries which match with the supplied predicate
+
+        Args:
+            predicate: Used to select entries to be removed from map.
+
+        Returns:
+            None
+        """
+
+        check_not_none(predicate, "predicate can't be None")
+
+        try:
+            predicate_data = self._to_data(predicate)
+
+        except SchemaNotReplicatedError as e:
+            return self._send_schema_and_retry(e, self.remove_all, predicate)
+
+        request = map_remove_all_codec.encode_request(self.name, predicate_data)
+
+        return self._invoke(request)
+
     def remove_if_same(self, key: KeyType, value: ValueType) -> Future[bool]:
         """Removes the entry for a key only if it is currently mapped to a
         given value.
@@ -2215,6 +2238,9 @@ def remove(  # type: ignore[override]
     ) -> typing.Optional[ValueType]:
         return self._wrapped.remove(key).result()
 
+    def remove_all(self, predicate: Predicate) -> None:  # type: ignore[override]
+        return self._wrapped.remove_all(predicate).result()
+
     def remove_if_same(  # type: ignore[override]
         self,
         key: KeyType,

tests/integration/backward_compatible/proxy/map_test.py

@@ -495,6 +495,15 @@ def test_remove(self):
         self.assertEqual(0, self.map.size())
         self.assertFalse(self.map.contains_key("key"))
 
+    def test_remove_all_with_none_predicate(self):
+        with self.assertRaises(AssertionError):
+            self.map.remove_all(None)
+
+    def test_remove_all(self):
+        self.fill_map()
+        self.map.remove_all(predicate=sql("__key > 'key-7'"))
+        self.assertEqual(self.map.size(), 8)
+
     def test_remove_if_same_when_same(self):
         self.map.put("key", "value")