Eagerly deserialize responses consisting of list of data [API-1964] (#…

…618)

* Eagerly deserialize responses consisting of list of data

We were performing lazy deserialization on APIs that return
a list of data, such  as map#values(). This was not going to work
with Compact, as after returning the list to the user, there might
be Compact serialized data on the list which we don't have the
schema on the client yet. If it was a normal response, the client
would have fetched the schema, but it is not possible with
these lazy APIs. We perform the deserialization while getting
the list items and this is a sync API. We can't perform a blocking
call there, because there is a chance that this will be executed
in the reactor thread, which would result in a deadlock.

So, the only possible way of making these APIs work with Compact
is to convert them to eager deserialization.

This PR removes lazy deserialization in everything apart from SQL,
which will be handled in another PR.

* fix mypy error

* address review comments

docs/serialization.rst

@@ -278,33 +278,6 @@ All of these APIs will work with the Compact serialization format, once it is
 promoted to the stable status.
 
 - Reading OBJECT columns of the SQL results
-- Listening for :class:`hazelcast.proxy.reliable_topic.ReliableTopic` messages
-- :func:`hazelcast.proxy.list.List.iterator`
-- :func:`hazelcast.proxy.list.List.list_iterator`
-- :func:`hazelcast.proxy.list.List.get_all`
-- :func:`hazelcast.proxy.list.List.sub_list`
-- :func:`hazelcast.proxy.map.Map.values`
-- :func:`hazelcast.proxy.map.Map.entry_set`
-- :func:`hazelcast.proxy.map.Map.execute_on_keys`
-- :func:`hazelcast.proxy.map.Map.key_set`
-- :func:`hazelcast.proxy.map.Map.project`
-- :func:`hazelcast.proxy.map.Map.execute_on_entries`
-- :func:`hazelcast.proxy.map.Map.get_all`
-- :func:`hazelcast.proxy.multi_map.MultiMap.remove_all`
-- :func:`hazelcast.proxy.multi_map.MultiMap.key_set`
-- :func:`hazelcast.proxy.multi_map.MultiMap.values`
-- :func:`hazelcast.proxy.multi_map.MultiMap.entry_set`
-- :func:`hazelcast.proxy.multi_map.MultiMap.get`
-- :func:`hazelcast.proxy.queue.Queue.iterator`
-- :func:`hazelcast.proxy.replicated_map.ReplicatedMap.values`
-- :func:`hazelcast.proxy.replicated_map.ReplicatedMap.entry_set`
-- :func:`hazelcast.proxy.replicated_map.ReplicatedMap.key_set`
-- :func:`hazelcast.proxy.set.Set.get_all`
-- :func:`hazelcast.proxy.ringbuffer.Ringbuffer.read_many`
-- :func:`hazelcast.proxy.transactional_map.TransactionalMap.values`
-- :func:`hazelcast.proxy.transactional_map.TransactionalMap.key_set`
-- :func:`hazelcast.proxy.transactional_multi_map.TransactionalMultiMap.get`
-- :func:`hazelcast.proxy.transactional_multi_map.TransactionalMultiMap.remove_all`
 
 IdentifiedDataSerializable Serialization
 ----------------------------------------

hazelcast/proxy/list.py

@@ -29,7 +29,7 @@
 from hazelcast.proxy.base import PartitionSpecificProxy, ItemEvent, ItemEventType
 from hazelcast.types import ItemType
 from hazelcast.serialization.compact import SchemaNotReplicatedError
-from hazelcast.util import check_not_none, ImmutableLazyDataList
+from hazelcast.util import check_not_none, deserialize_list_in_place
 
 
 class List(PartitionSpecificProxy["BlockingList"], typing.Generic[ItemType]):
@@ -247,9 +247,8 @@ def get_all(self) -> Future[typing.List[ItemType]]:
         """
 
         def handler(message):
-            return ImmutableLazyDataList(
-                list_get_all_codec.decode_response(message), self._to_object
-            )
+            data_list = list_get_all_codec.decode_response(message)
+            return deserialize_list_in_place(data_list, self._to_object)
 
         request = list_get_all_codec.encode_request(self.name)
         return self._invoke(request, handler)
@@ -263,9 +262,8 @@ def iterator(self) -> Future[typing.List[ItemType]]:
         """
 
         def handler(message):
-            return ImmutableLazyDataList(
-                list_iterator_codec.decode_response(message), self._to_object
-            )
+            data_list = list_iterator_codec.decode_response(message)
+            return deserialize_list_in_place(data_list, self._to_object)
 
         request = list_iterator_codec.encode_request(self.name)
         return self._invoke(request, handler)
@@ -337,9 +335,8 @@ def list_iterator(self, index: int = 0) -> Future[typing.List[ItemType]]:
         """
 
         def handler(message):
-            return ImmutableLazyDataList(
-                list_list_iterator_codec.decode_response(message), self._to_object
-            )
+            data_list = list_list_iterator_codec.decode_response(message)
+            return deserialize_list_in_place(data_list, self._to_object)
 
         request = list_list_iterator_codec.encode_request(self.name, index)
         return self._invoke(request, handler)
@@ -494,7 +491,8 @@ def sub_list(self, from_index: int, to_index: int) -> Future[typing.List[ItemTyp
         """
 
         def handler(message):
-            return ImmutableLazyDataList(list_sub_codec.decode_response(message), self._to_object)
+            data_list = list_sub_codec.decode_response(message)
+            return deserialize_list_in_place(data_list, self._to_object)
 
         request = list_sub_codec.encode_request(self.name, from_index, to_index)
         return self._invoke(request, handler)

hazelcast/proxy/map.py

@@ -87,8 +87,9 @@
     check_not_none,
     thread_id,
     to_millis,
-    ImmutableLazyDataList,
     IterationType,
+    deserialize_entry_list_in_place,
+    deserialize_list_in_place,
 )
 
 
@@ -563,7 +564,8 @@ def handler(message):
                     predicate.anchor_list = response["anchor_data_list"].as_anchor_list(
                         self._to_object
                     )
-                    return ImmutableLazyDataList(response["response"], self._to_object)
+                    entry_data_list = response["response"]
+                    return deserialize_entry_list_in_place(entry_data_list, self._to_object)
 
                 request = map_entries_with_paging_predicate_codec.encode_request(self.name, holder)
             else:
@@ -573,17 +575,15 @@ def handler(message):
                     return self._send_schema_and_retry(e, self.entry_set, predicate)
 
                 def handler(message):
-                    return ImmutableLazyDataList(
-                        map_entries_with_predicate_codec.decode_response(message), self._to_object
-                    )
+                    entry_data_list = map_entries_with_predicate_codec.decode_response(message)
+                    return deserialize_entry_list_in_place(entry_data_list, self._to_object)
 
                 request = map_entries_with_predicate_codec.encode_request(self.name, predicate_data)
         else:
 
             def handler(message):
-                return ImmutableLazyDataList(
-                    map_entry_set_codec.decode_response(message), self._to_object
-                )
+                entry_data_list = map_entry_set_codec.decode_response(message)
+                return deserialize_entry_list_in_place(entry_data_list, self._to_object)
 
             request = map_entry_set_codec.encode_request(self.name)
 
@@ -648,9 +648,8 @@ def execute_on_entries(
                 )
 
             def handler(message):
-                return ImmutableLazyDataList(
-                    map_execute_with_predicate_codec.decode_response(message), self._to_object
-                )
+                entry_data_list = map_execute_with_predicate_codec.decode_response(message)
+                return deserialize_entry_list_in_place(entry_data_list, self._to_object)
 
             request = map_execute_with_predicate_codec.encode_request(
                 self.name, entry_processor_data, predicate_data
@@ -664,9 +663,8 @@ def handler(message):
                 )
 
             def handler(message):
-                return ImmutableLazyDataList(
-                    map_execute_on_all_keys_codec.decode_response(message), self._to_object
-                )
+                entry_data_list = map_execute_on_all_keys_codec.decode_response(message)
+                return deserialize_entry_list_in_place(entry_data_list, self._to_object)
 
             request = map_execute_on_all_keys_codec.encode_request(self.name, entry_processor_data)
 
@@ -730,9 +728,8 @@ def execute_on_keys(
             return self._send_schema_and_retry(e, self.execute_on_keys, keys, entry_processor)
 
         def handler(message):
-            return ImmutableLazyDataList(
-                map_execute_on_keys_codec.decode_response(message), self._to_object
-            )
+            entry_data_list = map_execute_on_keys_codec.decode_response(message)
+            return deserialize_entry_list_in_place(entry_data_list, self._to_object)
 
         request = map_execute_on_keys_codec.encode_request(
             self.name, entry_processor_data, key_list
@@ -943,7 +940,8 @@ def handler(message):
                     predicate.anchor_list = response["anchor_data_list"].as_anchor_list(
                         self._to_object
                     )
-                    return ImmutableLazyDataList(response["response"], self._to_object)
+                    data_list = response["response"]
+                    return deserialize_list_in_place(data_list, self._to_object)
 
                 request = map_key_set_with_paging_predicate_codec.encode_request(self.name, holder)
             else:
@@ -953,17 +951,15 @@ def handler(message):
                     return self._send_schema_and_retry(e, self.key_set, predicate)
 
                 def handler(message):
-                    return ImmutableLazyDataList(
-                        map_key_set_with_predicate_codec.decode_response(message), self._to_object
-                    )
+                    data_list = map_key_set_with_predicate_codec.decode_response(message)
+                    return deserialize_list_in_place(data_list, self._to_object)
 
                 request = map_key_set_with_predicate_codec.encode_request(self.name, predicate_data)
         else:
 
             def handler(message):
-                return ImmutableLazyDataList(
-                    map_key_set_codec.decode_response(message), self._to_object
-                )
+                data_list = map_key_set_codec.decode_response(message)
+                return deserialize_list_in_place(data_list, self._to_object)
 
             request = map_key_set_codec.encode_request(self.name)
 
@@ -1067,9 +1063,8 @@ def project(
                 return self._send_schema_and_retry(e, self.project, projection, predicate)
 
             def handler(message):
-                return ImmutableLazyDataList(
-                    map_project_with_predicate_codec.decode_response(message), self._to_object
-                )
+                data_list = map_project_with_predicate_codec.decode_response(message)
+                return deserialize_list_in_place(data_list, self._to_object)
 
             request = map_project_with_predicate_codec.encode_request(
                 self.name, projection_data, predicate_data
@@ -1081,9 +1076,8 @@ def handler(message):
                 return self._send_schema_and_retry(e, self.project, projection, predicate)
 
             def handler(message):
-                return ImmutableLazyDataList(
-                    map_project_codec.decode_response(message), self._to_object
-                )
+                data_list = map_project_codec.decode_response(message)
+                return deserialize_list_in_place(data_list, self._to_object)
 
             request = map_project_codec.encode_request(self.name, projection_data)
 
@@ -1652,7 +1646,8 @@ def handler(message):
                     predicate.anchor_list = response["anchor_data_list"].as_anchor_list(
                         self._to_object
                     )
-                    return ImmutableLazyDataList(response["response"], self._to_object)
+                    data_list = response["response"]
+                    return deserialize_list_in_place(data_list, self._to_object)
 
                 request = map_values_with_paging_predicate_codec.encode_request(self.name, holder)
             else:
@@ -1662,17 +1657,15 @@ def handler(message):
                     return self._send_schema_and_retry(e, self.values, predicate)
 
                 def handler(message):
-                    return ImmutableLazyDataList(
-                        map_values_with_predicate_codec.decode_response(message), self._to_object
-                    )
+                    data_list = map_values_with_predicate_codec.decode_response(message)
+                    return deserialize_list_in_place(data_list, self._to_object)
 
                 request = map_values_with_predicate_codec.encode_request(self.name, predicate_data)
         else:
 
             def handler(message):
-                return ImmutableLazyDataList(
-                    map_values_codec.decode_response(message), self._to_object
-                )
+                data_list = map_values_codec.decode_response(message)
+                return deserialize_list_in_place(data_list, self._to_object)
 
             request = map_values_codec.encode_request(self.name)
 
@@ -1698,9 +1691,8 @@ def _get_all_internal(self, partition_to_keys, futures=None):
             futures = []
 
         def handler(message):
-            return ImmutableLazyDataList(
-                map_get_all_codec.decode_response(message), self._to_object
-            )
+            entry_data_list = map_get_all_codec.decode_response(message)
+            return deserialize_entry_list_in_place(entry_data_list, self._to_object)
 
         for partition_id, key_dict in partition_to_keys.items():
             request = map_get_all_codec.encode_request(self.name, key_dict.values())

hazelcast/proxy/multi_map.py

@@ -31,8 +31,13 @@
 from hazelcast.types import ValueType, KeyType
 from hazelcast.serialization.data import Data
 from hazelcast.serialization.compact import SchemaNotReplicatedError
-from hazelcast.util import check_not_none, thread_id, to_millis, ImmutableLazyDataList
-
+from hazelcast.util import (
+    check_not_none,
+    thread_id,
+    to_millis,
+    deserialize_list_in_place,
+    deserialize_entry_list_in_place,
+)
 
 EntryEventCallable = typing.Callable[[EntryEvent[KeyType, ValueType]], None]
 
@@ -213,9 +218,8 @@ def entry_set(self) -> Future[typing.List[typing.Tuple[KeyType, ValueType]]]:
         """
 
         def handler(message):
-            return ImmutableLazyDataList(
-                multi_map_entry_set_codec.decode_response(message), self._to_object
-            )
+            entry_data_list = multi_map_entry_set_codec.decode_response(message)
+            return deserialize_entry_list_in_place(entry_data_list, self._to_object)
 
         request = multi_map_entry_set_codec.encode_request(self.name)
         return self._invoke(request, handler)
@@ -246,9 +250,8 @@ def get(self, key: KeyType) -> Future[typing.Optional[typing.List[ValueType]]]:
             return self._send_schema_and_retry(e, self.get, key)
 
         def handler(message):
-            return ImmutableLazyDataList(
-                multi_map_get_codec.decode_response(message), self._to_object
-            )
+            data_list = multi_map_get_codec.decode_response(message)
+            return deserialize_list_in_place(data_list, self._to_object)
 
         request = multi_map_get_codec.encode_request(self.name, key_data, thread_id())
         return self._invoke_on_key(request, key_data, handler)
@@ -314,9 +317,8 @@ def key_set(self) -> Future[typing.List[KeyType]]:
         """
 
         def handler(message):
-            return ImmutableLazyDataList(
-                multi_map_key_set_codec.decode_response(message), self._to_object
-            )
+            data_list = multi_map_key_set_codec.decode_response(message)
+            return deserialize_list_in_place(data_list, self._to_object)
 
         request = multi_map_key_set_codec.encode_request(self.name)
         return self._invoke(request, handler)
@@ -410,9 +412,8 @@ def remove_all(self, key: KeyType) -> Future[typing.List[ValueType]]:
         check_not_none(key, "key can't be None")
 
         def handler(message):
-            return ImmutableLazyDataList(
-                multi_map_remove_codec.decode_response(message), self._to_object
-            )
+            data_list = multi_map_remove_codec.decode_response(message)
+            return deserialize_list_in_place(data_list, self._to_object)
 
         try:
             key_data = self._to_data(key)
@@ -551,9 +552,8 @@ def values(self) -> Future[typing.List[ValueType]]:
         """
 
         def handler(message):
-            return ImmutableLazyDataList(
-                multi_map_values_codec.decode_response(message), self._to_object
-            )
+            data_list = multi_map_values_codec.decode_response(message)
+            return deserialize_list_in_place(data_list, self._to_object)
 
         request = multi_map_values_codec.encode_request(self.name)
         return self._invoke(request, handler)

hazelcast/proxy/queue.py

@@ -26,7 +26,7 @@
 from hazelcast.proxy.base import PartitionSpecificProxy, ItemEvent, ItemEventType
 from hazelcast.types import ItemType
 from hazelcast.serialization.compact import SchemaNotReplicatedError
-from hazelcast.util import check_not_none, to_millis, ImmutableLazyDataList
+from hazelcast.util import check_not_none, to_millis, deserialize_list_in_place
 
 
 class Queue(PartitionSpecificProxy["BlockingQueue"], typing.Generic[ItemType]):
@@ -202,9 +202,8 @@ def iterator(self) -> Future[typing.List[ItemType]]:
         """
 
         def handler(message):
-            return ImmutableLazyDataList(
-                queue_iterator_codec.decode_response(message), self._to_object
-            )
+            data_list = queue_iterator_codec.decode_response(message)
+            return deserialize_list_in_place(data_list, self._to_object)
 
         request = queue_iterator_codec.encode_request(self.name)
         return self._invoke(request, handler)

hazelcast/proxy/reliable_topic.py

@@ -226,7 +226,7 @@ def _handle_next_batch(self, future):
             result = future.result()
 
             # Check if there are any messages lost since the last read
-            # and whether or not the listener can tolerate that.
+            # and whether the listener can tolerate that.
             lost_count = (result.next_sequence_to_read_from - result.read_count) - self._sequence
             if lost_count != 0 and not self._is_loss_tolerable(lost_count):
                 self.cancel()
@@ -254,7 +254,7 @@ def _handle_next_batch(self, future):
 
                     topic_message = TopicMessage(
                         self._topic_name,
-                        self._to_object(message.payload),
+                        message.payload,
                         message.publish_time,
                         member,
                     )