Implemented VolatileLayerClient::GetPartitions() method

Added functional tests for VolatileLayerClient::GetPartitions()
Added integration tests for VolatileLayerClient::GetPartitions()

Resolves: OLPEDGE-792

Signed-off-by: Serhii Lozynskyi <[email protected]>

Author: meravingen123

olp-cpp-sdk-dataservice-read/include/olp/dataservice/read/VolatileLayerClient.h

@@ -27,7 +27,9 @@
 #include <olp/core/client/HRN.h>
 #include <olp/core/client/OlpClientSettings.h>
 #include <olp/dataservice/read/DataRequest.h>
+#include <olp/dataservice/read/PartitionsRequest.h>
 #include <olp/dataservice/read/model/Data.h>
+#include <olp/dataservice/read/model/Partitions.h>
 
 namespace olp {
 namespace dataservice {
@@ -77,6 +79,11 @@ class DATASERVICE_READ_API VolatileLayerClient final {
   using DataResult = model::Data;
   /// CallbackResponse alias
   using DataResponse = client::ApiResponse<DataResult, client::ApiError>;
+  /// PartitionResult alias
+  using PartitionsResult = model::Partitions;
+  /// CallbackResponse
+  using PartitionsResponse =
+      client::ApiResponse<PartitionsResult, client::ApiError>;
   /// Callback alias
   template <class Response>
   using Callback = std::function<void(Response response)>;
@@ -94,6 +101,16 @@ class DATASERVICE_READ_API VolatileLayerClient final {
 
   ~VolatileLayerClient();
 
+  /**
+   * @brief Fetches a list partitions for given volatile layer asynchronously.
+   * @param request contains the complete set of request parameters.
+   * @param callback will be invoked once the list of partitions is available,
+   * or an error is encountered.
+   * @return A token that can be used to cancel this request
+   */
+  client::CancellationToken GetPartitions(
+      PartitionsRequest request, Callback<PartitionsResponse> callback);
+
   /**
    * @brief GetData fetches data for a partition or data handle asynchronously.
    * If the specified partition or data handle cannot be found in the layer, the

olp-cpp-sdk-dataservice-read/src/VolatileLayerClient.cpp

@@ -17,9 +17,9 @@
  * License-Filename: LICENSE
  */
 
+#include <olp/core/porting/make_unique.h>
 #include <olp/dataservice/read/VolatileLayerClient.h>
 
-#include <olp/core/porting/make_unique.h>
 #include "VolatileLayerClientImpl.h"
 
 namespace olp {
@@ -34,6 +34,11 @@ VolatileLayerClient::VolatileLayerClient(client::HRN catalog,
 
 VolatileLayerClient::~VolatileLayerClient() = default;
 
+client::CancellationToken VolatileLayerClient::GetPartitions(
+    PartitionsRequest request, Callback<PartitionsResponse> callback) {
+  return impl_->GetPartitions(std::move(request), std::move(callback));
+}
+
 client::CancellationToken VolatileLayerClient::GetData(
     DataRequest request, Callback<DataResponse> callback) {
   return impl_->GetData(std::move(request), std::move(callback));

olp-cpp-sdk-dataservice-read/src/VolatileLayerClientImpl.cpp

@@ -23,12 +23,14 @@
 #include <olp/core/client/CancellationContext.h>
 #include <olp/core/client/OlpClientSettingsFactory.h>
 #include <olp/dataservice/read/PartitionsRequest.h>
+
+#include "TaskContext.h"
+#include "repositories/ApiRepository.h"
+#include "repositories/CatalogRepository.h"
 #include "repositories/DataRepository.h"
 #include "repositories/ExecuteOrSchedule.inl"
 #include "repositories/PartitionsRepository.h"
 
-#include "TaskContext.h"
-
 namespace olp {
 namespace dataservice {
 namespace read {
@@ -42,26 +44,68 @@ VolatileLayerClientImpl::VolatileLayerClientImpl(
     client::OlpClientSettings settings)
     : catalog_(std::move(catalog)),
       layer_id_(std::move(layer_id)),
-      settings_(std::move(settings)),
+      settings_(
+          std::make_shared<client::OlpClientSettings>(std::move(settings))),
       pending_requests_(std::make_shared<PendingRequests>()) {
-  if (!settings_.cache) {
-    settings_.cache = client::OlpClientSettingsFactory::CreateDefaultCache({});
+  if (!settings_->cache) {
+    settings_->cache = client::OlpClientSettingsFactory::CreateDefaultCache({});
   }
   // to avoid capturing task scheduler inside a task, we need a copy of settings
   // without the scheduler
-  task_scheduler_ = std::move(settings_.task_scheduler);
+  task_scheduler_ = std::move(settings_->task_scheduler);
+
+  auto cache = settings_->cache;
+
+  auto api_repo =
+      std::make_shared<repository::ApiRepository>(catalog_, settings_, cache);
+
+  auto catalog_repo = std::make_shared<repository::CatalogRepository>(
+      catalog_, api_repo, cache);
+
+  partition_repo_ = std::make_shared<repository::PartitionsRepository>(
+      catalog_, api_repo, catalog_repo, cache);
 }
 
 VolatileLayerClientImpl::~VolatileLayerClientImpl() {
   pending_requests_->CancelPendingRequests();
 }
 
+client::CancellationToken VolatileLayerClientImpl::GetPartitions(
+    PartitionsRequest request, Callback<PartitionsResponse> callback) {
+  request.WithLayerId(layer_id_);
+  client::CancellationToken token;
+  int64_t request_key = pending_requests_->GenerateRequestPlaceholder();
+  auto pending_requests = pending_requests_;
+  auto request_callback = [pending_requests, request_key,
+                           callback](PartitionsResponse response) {
+    if (pending_requests->Remove(request_key)) {
+      callback(response);
+    }
+  };
+  if (CacheWithUpdate == request.GetFetchOption()) {
+    token = partition_repo_->GetPartitions(request.WithFetchOption(CacheOnly),
+                                           request_callback);
+    auto onlineKey = pending_requests_->GenerateRequestPlaceholder();
+    pending_requests_->Insert(
+        partition_repo_->GetPartitions(
+            request.WithFetchOption(OnlineIfNotFound),
+            [pending_requests, onlineKey](PartitionsResponse) {
+              pending_requests->Remove(onlineKey);
+            }),
+        onlineKey);
+  } else {
+    token = partition_repo_->GetPartitions(request, request_callback);
+  }
+  pending_requests_->Insert(token, request_key);
+  return token;
+}
+
 client::CancellationToken VolatileLayerClientImpl::GetData(
     DataRequest request, Callback<DataResponse> callback) {
   auto add_task = [&](DataRequest& request, Callback<DataResponse> callback) {
     auto catalog = catalog_;
     auto layer_id = layer_id_;
-    auto settings = settings_;
+    auto settings = *settings_;
     auto pending_requests = pending_requests_;
 
     auto data_task = [=](client::CancellationContext context) {

olp-cpp-sdk-dataservice-read/src/VolatileLayerClientImpl.h

@@ -19,27 +19,38 @@
 
 #pragma once
 
-#include <memory>
-
 #include <olp/core/client/ApiError.h>
 #include <olp/core/client/ApiResponse.h>
 #include <olp/core/client/CancellationToken.h>
 #include <olp/core/client/HRN.h>
 #include <olp/core/client/OlpClientSettings.h>
 #include <olp/dataservice/read/DataRequest.h>
+#include <olp/dataservice/read/PartitionsRequest.h>
 #include <olp/dataservice/read/model/Data.h>
+#include <olp/dataservice/read/model/Partitions.h>
+
 #include "PendingRequests.h"
 
 namespace olp {
 namespace dataservice {
 namespace read {
 
+namespace repository {
+class CatalogRepository;
+class PartitionsRepository;
+}  // namespace repository
+
 class VolatileLayerClientImpl {
  public:
   /// DataResult alias
   using DataResult = model::Data;
   /// CallbackResponse alias
   using DataResponse = client::ApiResponse<DataResult, client::ApiError>;
+  /// PartitionResult alias
+  using PartitionsResult = model::Partitions;
+  /// CallbackResponse
+  using PartitionsResponse =
+      client::ApiResponse<PartitionsResult, client::ApiError>;
   /// Callback alias
   template <class Response>
   using Callback = std::function<void(Response response)>;
@@ -49,15 +60,19 @@ class VolatileLayerClientImpl {
 
   ~VolatileLayerClientImpl();
 
+  client::CancellationToken GetPartitions(
+      PartitionsRequest request, Callback<PartitionsResponse> callback);
+
   client::CancellationToken GetData(DataRequest request,
                                     Callback<DataResponse> callback);
 
  private:
   client::HRN catalog_;
   std::string layer_id_;
-  client::OlpClientSettings settings_;
+  std::shared_ptr<client::OlpClientSettings> settings_;
   std::shared_ptr<thread::TaskScheduler> task_scheduler_;
   std::shared_ptr<PendingRequests> pending_requests_;
+  std::shared_ptr<repository::PartitionsRepository> partition_repo_;
 };
 
 }  // namespace read

scripts/linux/fv/olp-cpp-sdk-functional-test.variables

@@ -17,3 +17,5 @@ export dataservice_read_test_versioned_prefetch_layer="hype-test-prefetch"
 export dataservice_read_test_versioned_prefetch_tile="5904591"
 export dataservice_read_test_versioned_prefetch_subpartition1="23618365"
 export dataservice_read_test_versioned_prefetch_subpartition2="1476147"
+
+export dataservice_read_volatile_layer="testlayer"

tests/functional/CMakeLists.txt

@@ -23,6 +23,7 @@ set(OLP_SDK_FUNCTIONAL_TESTS_SOURCES
     ./olp-cpp-sdk-dataservice-read/ApiTest.cpp
     ./olp-cpp-sdk-dataservice-read/CatalogClientTest.cpp
     ./olp-cpp-sdk-dataservice-read/DataserviceReadVersionedLayerClientTest.cpp
+    ./olp-cpp-sdk-dataservice-read/DataserviceReadVolatileLayerClientTest.cpp
     ./olp-cpp-sdk-dataservice-write/DataserviceWriteIndexLayerClientTest.cpp
     ./olp-cpp-sdk-dataservice-write/DataserviceWriteStreamLayerClientCacheTest.cpp
     ./olp-cpp-sdk-dataservice-write/DataserviceWriteStreamLayerClientTest.cpp