feat(kv-ir): Integrate QueryHandler into the IR stream Deserializer.

components/core/CMakeLists.txt

@@ -461,6 +461,8 @@ set(SOURCE_FILES_unitTest
         src/clp/ffi/ir_stream/search/QueryHandler.hpp
         src/clp/ffi/ir_stream/search/QueryHandlerImpl.cpp
         src/clp/ffi/ir_stream/search/QueryHandlerImpl.hpp
+        src/clp/ffi/ir_stream/search/QueryHandlerReq.hpp
+        src/clp/ffi/ir_stream/search/test/test_deserializer_integration.cpp
         src/clp/ffi/ir_stream/search/test/test_QueryHandlerImpl.cpp
         src/clp/ffi/ir_stream/search/test/test_utils.cpp
         src/clp/ffi/ir_stream/search/test/utils.cpp

components/core/src/clp/ffi/ir_stream/Deserializer.hpp

@@ -6,7 +6,7 @@
 #include <memory>
 #include <string>
 #include <system_error>
-#include <tuple>
+#include <utility>
 #include <vector>
 
 #include <nlohmann/json.hpp>
@@ -20,37 +20,61 @@
 #include "IrUnitHandlerInterface.hpp"
 #include "IrUnitType.hpp"
 #include "protocol_constants.hpp"
+#include "search/QueryHandlerReq.hpp"
 #include "utils.hpp"
 
 namespace clp::ffi::ir_stream {
 /**
  * A deserializer for reading IR units from a CLP kv-pair IR stream. An IR unit handler should be
- * provided to perform user-defined operations on each deserialized IR unit.
+ * provided to perform user-defined operations on each deserialized IR unit. Additionally, a query
+ * handler can be provided to handle queries and column projections.
  *
  * NOTE: This class is designed only to provide deserialization functionalities. Callers are
  * responsible for maintaining a `ReaderInterface` to input IR bytes from an I/O stream.
  *
  * @tparam IrUnitHandler
+ * @tparam QueryHandlerType
  */
-template <IrUnitHandlerInterface IrUnitHandler>
+template <
+        IrUnitHandlerInterface IrUnitHandler,
+        search::QueryHandlerReq QueryHandlerType = search::EmptyQueryHandler>
 requires(std::move_constructible<IrUnitHandler>)
 class Deserializer {
 public:
     // Factory function
     /**
-     * Creates a deserializer by reading the stream's preamble from the given reader.
+     * Creates a deserializer with an empty query handler (for use when the deserializer won't be
+     * used to perform queries or column projections).
      * @param reader
      * @param ir_unit_handler
-     * @return A result containing the deserializer or an error code indicating the failure:
-     * - std::errc::result_out_of_range if the IR stream is truncated
-     * - std::errc::protocol_error if the IR stream is corrupted
-     * - std::errc::protocol_not_supported if either:
-     *   - the IR stream contains an unsupported metadata format;
-     *   - the IR stream's version is unsupported;
-     *   - or the IR stream's user-defined metadata is not a JSON object.
+     * @return A result containing the deserializer on success, or an error code indicating the
+     * failure:
+     * - Forwards `create_generic`'s return values.
      */
     [[nodiscard]] static auto create(ReaderInterface& reader, IrUnitHandler ir_unit_handler)
-            -> OUTCOME_V2_NAMESPACE::std_result<Deserializer>;
+            -> OUTCOME_V2_NAMESPACE::std_result<Deserializer>
+    requires std::is_same_v<QueryHandlerType, search::EmptyQueryHandler>
+    {
+        return create_generic(reader, std::move(ir_unit_handler), {});
+    }
+
+    /**
+     * Creates a deserializer with a query handler (for use when the deserializer will be used to
+     * perform queries or column projections).
+     * @param reader
+     * @param ir_unit_handler
+     * @param query_handler
+     * @return A result containing the deserializer on success, or an error code indicating the
+     * failure:
+     * - Forwards `create_generic`'s return values.
+     */
+    [[nodiscard]] static auto
+    create(ReaderInterface& reader, IrUnitHandler ir_unit_handler, QueryHandlerType query_handler)
+            -> OUTCOME_V2_NAMESPACE::std_result<Deserializer>
+    requires search::IsNonEmptyQueryHandler<QueryHandlerType>::value
+    {
+        return create_generic(reader, std::move(ir_unit_handler), std::move(query_handler));
+    }
 
     // Delete copy constructor and assignment
     Deserializer(Deserializer const&) = delete;
@@ -65,34 +89,45 @@ class Deserializer {
 
     // Methods
     /**
-     * Deserializes the stream from the given reader up to and including the next log event IR unit.
+     * Deserializes the stream from the given reader up to and including the next log event IR unit,
+     * and invokes the user-defined IR unit handler according to the deserialized IR unit type.
+     *
+     * NOTE: If the deserialized IR unit is `IrUnitType::LogEvent` and the query handler is not
+     * `search::EmptyQueryHandler`, `handle_log_event` will only be invoked if the query handler
+
+     * returns `search::AstEvaluationResult::True`.
+     *
      * @param reader
      * @return Forwards `deserialize_tag`s return values if no tag bytes can be read to determine
      * the next IR unit type.
      * @return std::errc::protocol_not_supported if the IR unit type is not supported.
      * @return std::errc::operation_not_permitted if the deserializer already reached the end of
      * stream by deserializing an end-of-stream IR unit in the previous calls.
-     * @return IRUnitType::LogEvent if a log event IR unit is deserialized, or an error code
+     * @return IrUnitType::LogEvent if a log event IR unit is deserialized, or an error code
      * indicating the failure:
      * - Forwards `deserialize_ir_unit_kv_pair_log_event`'s return values if it failed to
      *   deserialize and construct the log event.
      * - Forwards `handle_log_event`'s return values from the user-defined IR unit handler on
      *   unit handling failure.
-     * @return IRUnitType::SchemaTreeNodeInsertion if a schema tree node insertion IR unit is
+     * - Forwards `search::QueryHandler::evaluate_kv_pair_log_event`'s return values on failure, if
+     *   `QueryHandlerType` is not `search::EmptyQueryHandler`.
+     * @return IrUnitType::SchemaTreeNodeInsertion if a schema tree node insertion IR unit is
      * deserialized, or an error code indicating the failure:
      * - Forwards `deserialize_ir_unit_schema_tree_node_insertion`'s return values if it failed to
      *   deserialize and construct the schema tree node locator.
      * - Forwards `handle_schema_tree_node_insertion`'s return values from the user-defined IR unit
      *   handler on unit handling failure.
+     * - Forwards `search::QueryHandler::update_partially_resolved_columns`'s return values on
+     *   failure, if `QueryHandlerType` is not `search::EmptyQueryHandler`.
      * - std::errc::protocol_error if the deserialized schema tree node already exists in the schema
      *   tree.
-     * @return IRUnitType::UtcOffsetChange if a UTC offset change IR unit is deserialized, or an
+     * @return IrUnitType::UtcOffsetChange if a UTC offset change IR unit is deserialized, or an
      * error code indicating the failure:
      * - Forwards `deserialize_ir_unit_utc_offset_change`'s return values if it failed to
      *   deserialize the UTC offset.
      * - Forwards `handle_utc_offset_change`'s return values from the user-defined IR unit handler
      *   on unit handling failure.
-     * @return IRUnitType::EndOfStream if an end-of-stream IR unit is deserialized, or an error code
+     * @return IrUnitType::EndOfStream if an end-of-stream IR unit is deserialized, or an error code
      * indicating the failure:
      * - Forwards `handle_end_of_stream`'s return values from the user-defined IR unit handler on
      *   unit handling failure.
@@ -118,10 +153,35 @@ class Deserializer {
     [[nodiscard]] auto get_metadata() const -> nlohmann::json const& { return m_metadata; }
 
 private:
+    // Factory function
+    /**
+     * Creates a deserializer by reading the stream's preamble from the given reader.
+     * @param reader
+     * @param ir_unit_handler
+     * @param query_handler
+     * @return A result containing the deserializer or an error code indicating the failure:
+     * - std::errc::result_out_of_range if the IR stream is truncated
+     * - std::errc::protocol_error if the IR stream is corrupted
+     * - std::errc::protocol_not_supported if either:
+     *   - the IR stream contains an unsupported metadata format;
+     *   - the IR stream's version is unsupported;
+     *   - or the IR stream's user-defined metadata is not a JSON object.
+     */
+    [[nodiscard]] static auto create_generic(
+            ReaderInterface& reader,
+            IrUnitHandler ir_unit_handler,
+            QueryHandlerType query_handler
+    ) -> OUTCOME_V2_NAMESPACE::std_result<Deserializer>;
+
     // Constructor
-    Deserializer(IrUnitHandler ir_unit_handler, nlohmann::json metadata)
+    Deserializer(
+            IrUnitHandler ir_unit_handler,
+            nlohmann::json metadata,
+            QueryHandlerType query_handler
+    )
             : m_ir_unit_handler{std::move(ir_unit_handler)},
-              m_metadata(std::move(metadata)) {}
+              m_metadata(std::move(metadata)),
+              m_query_handler{std::move(query_handler)} {}
 
     // Variables
     std::shared_ptr<SchemaTree> m_auto_gen_keys_schema_tree{std::make_shared<SchemaTree>()};
@@ -130,12 +190,16 @@ class Deserializer {
     UtcOffset m_utc_offset{0};
     IrUnitHandler m_ir_unit_handler;
     bool m_is_complete{false};
+    [[no_unique_address]] QueryHandlerType m_query_handler;
 };
 
-template <IrUnitHandlerInterface IrUnitHandler>
+template <IrUnitHandlerInterface IrUnitHandler, search::QueryHandlerReq QueryHandlerType>
 requires(std::move_constructible<IrUnitHandler>)
-auto Deserializer<IrUnitHandler>::create(ReaderInterface& reader, IrUnitHandler ir_unit_handler)
-        -> OUTCOME_V2_NAMESPACE::std_result<Deserializer> {
+auto Deserializer<IrUnitHandler, QueryHandlerType>::create_generic(
+        ReaderInterface& reader,
+        IrUnitHandler ir_unit_handler,
+        QueryHandlerType query_handler
+) -> OUTCOME_V2_NAMESPACE::std_result<Deserializer> {
     bool is_four_byte_encoded{};
     if (auto const err{get_encoding_type(reader, is_four_byte_encoded)};
         IRErrorCode::IRErrorCode_Success != err)
@@ -176,13 +240,44 @@ auto Deserializer<IrUnitHandler>::create(ReaderInterface& reader, IrUnitHandler
         return std::errc::protocol_not_supported;
     }
 
-    return Deserializer{std::move(ir_unit_handler), std::move(metadata_json)};
+    return Deserializer{
+            std::move(ir_unit_handler),
+            std::move(metadata_json),
+            std::move(query_handler)
+    };
 }
 
+/**
+ * Wrapper for `Deserializer`'s factory function to enable automatic type deduction.
+ * @param reader
+ * @param ir_unit_handler
+ * @return Forwards `Deserializer::create`'s return values.
+ */
 template <IrUnitHandlerInterface IrUnitHandler>
+requires std::is_move_constructible_v<IrUnitHandler>
+[[nodiscard]] auto make_deserializer(ReaderInterface& reader, IrUnitHandler ir_unit_handler)
+        -> OUTCOME_V2_NAMESPACE::std_result<Deserializer<IrUnitHandler>>;
+
+/**
+ * Wrapper for `Deserializer`'s factory function to enable automatic type deduction.
+ * @param reader
+ * @param ir_unit_handler
+ * @param query_handler
+ * @return Forwards `Deserializer::create`'s return values.
+ */
+template <IrUnitHandlerInterface IrUnitHandler, search::QueryHandlerReq QueryHandlerType>
+requires std::move_constructible<IrUnitHandler>
+[[nodiscard]] auto make_deserializer(
+        ReaderInterface& reader,
+        IrUnitHandler ir_unit_handler,
+        QueryHandlerType query_handler
+) -> OUTCOME_V2_NAMESPACE::std_result<Deserializer<IrUnitHandler, QueryHandlerType>>;
+
+template <IrUnitHandlerInterface IrUnitHandler, search::QueryHandlerReq QueryHandlerType>
 requires(std::move_constructible<IrUnitHandler>)
-auto Deserializer<IrUnitHandler>::deserialize_next_ir_unit(ReaderInterface& reader)
-        -> OUTCOME_V2_NAMESPACE::std_result<IrUnitType> {
+auto Deserializer<IrUnitHandler, QueryHandlerType>::deserialize_next_ir_unit(
+        ReaderInterface& reader
+) -> OUTCOME_V2_NAMESPACE::std_result<IrUnitType> {
     if (is_stream_completed()) {
         return std::errc::operation_not_permitted;
     }
@@ -211,6 +306,14 @@ auto Deserializer<IrUnitHandler>::deserialize_next_ir_unit(ReaderInterface& read
                 return result.error();
             }
 
+            if constexpr (search::IsNonEmptyQueryHandler<QueryHandlerType>::value) {
+                if (search::AstEvaluationResult::True
+                    != OUTCOME_TRYX(m_query_handler.evaluate_kv_pair_log_event(result.value())))
+                {
+                    break;
+                }
+            }
+
             if (auto const err{m_ir_unit_handler.handle_log_event(std::move(result.value()))};
                 IRErrorCode::IRErrorCode_Success != err)
             {
@@ -237,7 +340,15 @@ auto Deserializer<IrUnitHandler>::deserialize_next_ir_unit(ReaderInterface& read
                 return std::errc::protocol_error;
             }
 
-            std::ignore = schema_tree_to_insert->insert_node(node_locator);
+            auto const node_id{schema_tree_to_insert->insert_node(node_locator)};
+
+            if constexpr (search::IsNonEmptyQueryHandler<QueryHandlerType>::value) {
+                OUTCOME_TRYV(m_query_handler.update_partially_resolved_columns(
+                        is_auto_generated,
+                        node_locator,
+                        node_id
+                ));
+            }
 
             if (auto const err{m_ir_unit_handler.handle_schema_tree_node_insertion(
                         is_auto_generated,
@@ -286,6 +397,27 @@ auto Deserializer<IrUnitHandler>::deserialize_next_ir_unit(ReaderInterface& read
 
     return ir_unit_type;
 }
+
+template <IrUnitHandlerInterface IrUnitHandler>
+requires std::is_move_constructible_v<IrUnitHandler>
+[[nodiscard]] auto make_deserializer(ReaderInterface& reader, IrUnitHandler ir_unit_handler)
+        -> OUTCOME_V2_NAMESPACE::std_result<Deserializer<IrUnitHandler>> {
+    return Deserializer<IrUnitHandler>::create(reader, std::move(ir_unit_handler));
+}
+
+template <IrUnitHandlerInterface IrUnitHandler, search::QueryHandlerReq QueryHandlerType>
+requires std::move_constructible<IrUnitHandler>
+[[nodiscard]] auto make_deserializer(
+        ReaderInterface& reader,
+        IrUnitHandler ir_unit_handler,
+        QueryHandlerType query_handler
+) -> OUTCOME_V2_NAMESPACE::std_result<Deserializer<IrUnitHandler, QueryHandlerType>> {
+    return Deserializer<IrUnitHandler, QueryHandlerType>::create(
+            reader,
+            std::move(ir_unit_handler),
+            std::move(query_handler)
+    );
+}
 }  // namespace clp::ffi::ir_stream
 
 #endif  // CLP_FFI_IR_STREAM_DESERIALIZER_HPP

components/core/src/clp/ffi/ir_stream/search/QueryHandlerReq.hpp

@@ -0,0 +1,47 @@
+#ifndef CLP_FFI_IR_STREAM_SEARCH_QUERYHANDLERREQ_HPP
+#define CLP_FFI_IR_STREAM_SEARCH_QUERYHANDLERREQ_HPP
+
+#include <type_traits>
+
+#include "NewProjectedSchemaTreeNodeCallbackReq.hpp"
+#include "QueryHandler.hpp"
+
+namespace clp::ffi::ir_stream::search {
+/**
+ * Defines an empty query handler that can be used with `clp::ffi::ir_stream::Deserializer` to
+ * deserialize an IR stream without performing any query evaluation.
+ */
+struct EmptyQueryHandler {};
+
+/**
+ * A type trait to determine if a given type is an instantiation of
+ * `clp::ffi::ir_stream::search::QueryHandler`.
+ * @tparam T The type to check.
+ */
+template <typename T>
+struct IsNonEmptyQueryHandler : std::false_type {};
+
+/**
+ * Specialization of `IsNonEmptyQueryHandler` for `clp::ffi::ir_stream::search::QueryHandler`.
+ * @tparam NewProjectedSchemaTreeNodeCallbackType
+ */
+template <NewProjectedSchemaTreeNodeCallbackReq NewProjectedSchemaTreeNodeCallbackType>
+struct IsNonEmptyQueryHandler<QueryHandler<NewProjectedSchemaTreeNodeCallbackType>>
+        : std::true_type {};
+
+/**
+ * Requirements for a query handler that can be used with `clp::ffi::ir_stream::Deserializer`. A
+ * valid query handler must be:
+ *
+ * - an `EmptyQueryHandler` or satisfy the `IsNonEmptyQueryHandler` trait.
+ * - move constructible.
+ *
+ * @tparam QueryHandlerType The type to check.
+ */
+template <typename QueryHandlerType>
+concept QueryHandlerReq = (std::is_same_v<QueryHandlerType, EmptyQueryHandler>
+                           || IsNonEmptyQueryHandler<QueryHandlerType>::value)
+                          && std::is_move_constructible_v<QueryHandlerType>;
+}  // namespace clp::ffi::ir_stream::search
+
+#endif  // CLP_FFI_IR_STREAM_SEARCH_QUERYHANDLERREQ_HPP