Compact fetch schema on read is implemented. [Api 1327] (#1082)

* Schema fetch on read is implemented.

* a_type -> sample_compact_type

---------

Co-authored-by: ihsan demir <[email protected]>

Author: OzanCansel

hazelcast/generated-sources/src/hazelcast/client/protocol/codec/codecs.cpp

@@ -5821,6 +5821,22 @@ client_sendallschemas_encode(
     return msg;
 }
 
+ClientMessage
+client_fetchschema_encode(int64_t schema_id)
+{
+    size_t initial_frame_size =
+      ClientMessage::REQUEST_HEADER_LEN + ClientMessage::INT64_SIZE;
+    ClientMessage msg(initial_frame_size, true);
+    msg.set_retryable(true);
+    msg.set_operation_name("client.fetchschema");
+
+    msg.set_message_type(static_cast<int32_t>(5120));
+    msg.set_partition_id(-1);
+
+    msg.set(schema_id);
+    return msg;
+}
+
 } // namespace codec
 } // namespace protocol
 } // namespace client

hazelcast/generated-sources/src/hazelcast/client/protocol/codec/codecs.h

@@ -3092,6 +3092,12 @@ ClientMessage HAZELCAST_API
 client_sendallschemas_encode(
   const std::vector<serialization::pimpl::schema>& schemas);
 
+/**
+ * Fetches a schema from the cluster with the given schemaId
+ */
+ClientMessage HAZELCAST_API
+client_fetchschema_encode(int64_t schema_id);
+
 } // namespace codec
 } // namespace protocol
 } // namespace client

hazelcast/include/hazelcast/client/protocol/ClientMessage.h

@@ -917,6 +917,44 @@ class HAZELCAST_API ClientMessage
         return h;
     }
 
+    template<typename T>
+    typename std::
+      enable_if<std::is_same<T, serialization::pimpl::schema>::value, T>::type
+      get()
+    {
+        using namespace serialization;
+        using namespace serialization::pimpl;
+
+        // skip begin frame
+        skip_frame();
+
+        auto type_name = get<std::string>();
+
+        std::unordered_map<std::string, field_descriptor> fields;
+        {
+            skip_frame();
+
+            while (!next_frame_is_data_structure_end_frame()) {
+                skip_frame();
+
+                // skip bytes in initial frame
+                (void)rd_ptr(SIZE_OF_FRAME_LENGTH_AND_FLAGS);
+
+                auto key = get<int>();
+                auto field_name = get<std::string>();
+
+                fast_forward_to_end_frame();
+
+                fields.insert(std::make_pair(
+                  field_name, field_descriptor{ field_kind(key) }));
+            }
+        }
+
+        fast_forward_to_end_frame();
+
+        return schema{ type_name, move(fields) };
+    }
+
     /**
      * Reads the header of the current frame.
      * The cursor must be at a frame's beginning.

hazelcast/include/hazelcast/client/serialization/field_kind.h

@@ -73,4 +73,4 @@ operator<<(std::ostream&, field_kind);
 
 } // namespace serialization
 } // namespace client
-} // namespace hazelcast
+} // namespace hazelcast

hazelcast/include/hazelcast/client/serialization/pimpl/compact/compact.h

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2008-2021, Hazelcast, Inc. All Rights Reserved.
+ * Copyright (c) 2008-2023, Hazelcast, Inc. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

hazelcast/include/hazelcast/client/serialization/pimpl/compact/compact_impl.h

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2008-2021, Hazelcast, Inc. All Rights Reserved.
+ * Copyright (c) 2008-2023, Hazelcast, Inc. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -557,30 +557,18 @@ default_compact_writer::write(const T& value)
     pimpl::serialization_util::write(object_data_output_, value);
 }
 
-template<typename T>
-struct schema_of
-{
-    static schema build_schema()
-    {
-        T t;
-        schema_writer schema_writer(hz_serializer<T>::type_name());
-        serialization::compact::compact_writer writer =
-          create_compact_writer(&schema_writer);
-        serialization::hz_serializer<T>::write(t, writer);
-        return std::move(schema_writer).build();
-    }
-
-    const static schema schema_v;
-};
-
-template<typename T>
-const schema schema_of<T>::schema_v = schema_of<T>::build_schema();
-
 template<typename T>
 class class_to_schema
 {
 public:
-    static const schema& get() { return value_; }
+    static const schema* get()
+    {
+        if (is_initialized_) {
+            return &value_;
+        } else {
+            return nullptr;
+        }
+    }
 
     static void set(const T& object)
     {
@@ -613,26 +601,37 @@ template<typename T>
 T inline compact_stream_serializer::read(object_data_input& in)
 {
     int64_t schema_id = in.read<int64_t>();
-    const auto& local_schema = schema_of<T>::schema_v;
+    const schema* local_schema = class_to_schema<T>::get();
     // optimization to avoid hitting shared map in the schema_service,
     // in the case incoming data's schema is same as the local schema
-    if (schema_id == local_schema.schema_id()) {
+    if (local_schema && schema_id == local_schema->schema_id()) {
         compact::compact_reader reader =
-          create_compact_reader(*this, in, local_schema);
+          create_compact_reader(*this, in, *local_schema);
         return hz_serializer<T>::read(reader);
     }
-    // This path will run only in schema evolution case
+
     auto schema = schema_service.get(schema_id);
-    if (schema.type_name() != hz_serializer<T>::type_name()) {
+
+    if (!schema) {
+        throw exception::hazelcast_serialization{
+            "compact_stream_serializer::read",
+            boost::str(
+              boost::format(
+                "The schema can not be found with id %1% for '%2%' type") %
+              schema_id % hz_serializer<T>::type_name())
+        };
+    }
+
+    if (schema->type_name() != hz_serializer<T>::type_name()) {
         auto exception = exception::hazelcast_serialization{
             "compact_stream_serializer",
             (boost::format("Unexpected typename. expected %1%, received %2%") %
-             hz_serializer<T>::type_name() % schema.type_name())
+             hz_serializer<T>::type_name() % schema->type_name())
               .str()
         };
         BOOST_THROW_EXCEPTION(exception);
     }
-    compact::compact_reader reader = create_compact_reader(*this, in, schema);
+    compact::compact_reader reader = create_compact_reader(*this, in, *schema);
     return hz_serializer<T>::read(reader);
 }
 
@@ -642,7 +641,7 @@ void inline compact_stream_serializer::write(const T& object,
 {
     class_to_schema<T>::set(object);
 
-    const schema& schema_v = class_to_schema<T>::get();
+    const schema& schema_v = *class_to_schema<T>::get();
 
     if (!schema_service.is_schema_replicated(schema_v)) {
         out.schemas_will_be_replicated_.push_back(schema_v);

hazelcast/include/hazelcast/client/serialization/pimpl/compact/default_schema_service.h

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2008-2021, Hazelcast, Inc. All Rights Reserved.
+ * Copyright (c) 2008-2023, Hazelcast, Inc. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -61,7 +61,7 @@ class HAZELCAST_API default_schema_service
      *     <li>searching the cluster.</li>
      * </ul>
      */
-    schema get(int64_t schemaId);
+    std::shared_ptr<schema> get(int64_t schemaId);
 
     /**
      * Replicates schema on the cluster

hazelcast/include/hazelcast/client/serialization/pimpl/compact/schema.h

@@ -62,4 +62,4 @@ operator<<(std::ostream& os, const schema& schema);
 } // namespace pimpl
 } // namespace serialization
 } // namespace client
-} // namespace hazelcast
+} // namespace hazelcast

hazelcast/src/hazelcast/client/compact.cpp

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2008-2021, Hazelcast, Inc. All Rights Reserved.
+ * Copyright (c) 2008-2023, Hazelcast, Inc. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -99,7 +99,7 @@ operator<<(std::ostream& os, field_kind kind)
             os << "INT64";
             break;
         case field_kind::ARRAY_OF_INT64:
-            os << "ARRAY_OF_INT16";
+            os << "ARRAY_OF_INT64";
             break;
         case field_kind::FLOAT32:
             os << "FLOAT32";
@@ -1882,17 +1882,43 @@ default_schema_service::default_schema_service(spi::ClientContext& context)
 {
 }
 
-schema
+std::shared_ptr<schema>
 default_schema_service::get(int64_t schemaId)
 {
     auto ptr = replicateds_.get(schemaId);
 
-    if (!ptr) {
-        throw exception::illegal_state{ "default_schema_service::get",
-                                        "Schema doesn't exist for this type" };
+    if (ptr) {
+        return ptr;
+    }
+
+    auto logger = context_.get_logger();
+    if (logger.enabled(logger::level::finest)) {
+        logger.log(
+          logger::level::finest,
+          boost::str(boost::format("Could not find schema id %1% locally, will "
+                                   "search on the cluster %1%") %
+                     schemaId));
+    }
+
+    using namespace protocol::codec;
+
+    auto request_message = client_fetchschema_encode(schemaId);
+
+    auto invocation = spi::impl::ClientInvocation::create(
+      context_, request_message, SERVICE_NAME);
+    auto message = invocation->invoke().get();
+
+    message.skip_frame();
+    auto sch = message.get_nullable<schema>();
+
+    std::shared_ptr<schema> schema_ptr;
+
+    if (sch) {
+        schema_ptr = std::make_shared<schema>(std::move(*sch));
+        replicateds_.put_if_absent(schemaId, schema_ptr);
     }
 
-    return *ptr;
+    return schema_ptr;
 }
 
 void

hazelcast/src/hazelcast/client/network.cpp

@@ -714,13 +714,11 @@ ClientConnectionManagerImpl::initialize_client_on_cluster(
             std::lock_guard<std::recursive_mutex> guard(client_state_mutex_);
 
             if (target_cluster_id == cluster_id_) {
-                if (logger_.enabled(hazelcast::logger::level::fine)) {
-                    logger_.log(
-                      hazelcast::logger::level::fine,
-                      (boost::format("Client state is sent to cluster: %1%") %
-                       target_cluster_id)
-                        .str());
-                }
+                HZ_LOG(logger_,
+                       fine,
+                       (boost::format("Client state is sent to cluster: %1%") %
+                        target_cluster_id)
+                         .str());
 
                 client_state_ = client_state::INITIALIZED_ON_CLUSTER;
                 fire_life_cycle_event(lifecycle_event::CLIENT_CONNECTED);