Add api to prefetch partitions. (#1049)

Add public api to prefetch partitions
for versioned layer client. Add integration
tests to check prefetching, handling errors,
canceling, request prioritization, request
progress. Change query to return error only if all query
partitions fail. Add test to check
if all possible partitions prefetched,
even if some queries fail, check if
error returned if all queries fails.

Relates-To: OLPEDGE-1901
Signed-off-by: Liubov Didkivska <[email protected]>

Author: Liubov Didkivska

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

@@ -31,6 +31,7 @@
 #include <olp/dataservice/read/DataRequest.h>
 #include <olp/dataservice/read/DataServiceReadApi.h>
 #include <olp/dataservice/read/PartitionsRequest.h>
+#include <olp/dataservice/read/PrefetchPartitionsRequest.h>
 #include <olp/dataservice/read/PrefetchTileResult.h>
 #include <olp/dataservice/read/PrefetchTilesRequest.h>
 #include <olp/dataservice/read/TileRequest.h>
@@ -351,6 +352,58 @@ class DATASERVICE_READ_API VersionedLayerClient final {
       PrefetchTilesRequest request,
       PrefetchStatusCallback status_callback = nullptr);
 
+  /**
+   * @brief Prefetches a set of partitions asynchronously.
+   *
+   * This method downloads all partitions listed in
+   * `PrefetchPartitionsRequest`. Only partitions that are not already present
+   * in the cache are downloaded. It helps reduce the network load.
+   *
+   * @note This method does not guarantee that all partitions are available
+   * offline as the cache might overflow, and data might be evicted at any
+   * point. Use `GetData(DataRequest)` to retrieve partitions loaded by
+   * `PrefetchPartitions`.
+   *
+   * @param request The `PrefetchPartitionsRequest` instance that contains
+   * a complete set of request parameters.
+   * @param callback The `PrefetchPartitionsResponseCallback` object that is
+   * invoked if the `PrefetchPartitionsResult` instance is available or an error
+   * occurs.
+   * @param status_callback The `PrefetchPartitionsStatusCallback` object that
+   * is invoked every time a partition is fetched.
+   *
+   * @return A token that can be used to cancel this request.
+   */
+  client::CancellationToken PrefetchPartitions(
+      PrefetchPartitionsRequest request,
+      PrefetchPartitionsResponseCallback callback,
+      PrefetchPartitionsStatusCallback status_callback = nullptr);
+
+  /**
+   * @brief Prefetches a set of partitions asynchronously.
+   *
+   * This method downloads all partitions listed in
+   * `PrefetchPartitionsRequest`. Only partitions that are not already present
+   * in the cache are downloaded. It helps reduce the network load.
+   *
+   * @note This method does not guarantee that all partitions are available
+   * offline as the cache might overflow, and data might be evicted at any
+   * point. Use `GetData(DataRequest)` to retrieve partitions loaded by
+   * `PrefetchPartitions`.
+   *
+   * @param request The `PrefetchPartitionsRequest` instance that contains
+   * a complete set of request parameters.
+   * @param status_callback The `PrefetchPartitionsStatusCallback` object that
+   * is invoked every time a partition is fetched.
+   *
+   * @return `CancellableFuture` that contains the `PrefetchPartitionsResponse`
+   * instance with data or an error. You can also use `CancellableFuture` to
+   * cancel this request.
+   */
+  client::CancellableFuture<PrefetchPartitionsResponse> PrefetchPartitions(
+      PrefetchPartitionsRequest request,
+      PrefetchPartitionsStatusCallback status_callback = nullptr);
+
   /**
    * @brief Removes the partition from the mutable disk cache.
    *

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

@@ -33,20 +33,14 @@
 #include "DownloadItemsJob.h"
 #include "ExtendedApiResponse.h"
 #include "ExtendedApiResponseHelpers.h"
-#include "QueryMetadataJob.h"
+#include "QueryPartitionsJob.h"
 #include "TaskSink.h"
 #include "repositories/PartitionsRepository.h"
 
 namespace olp {
 namespace dataservice {
 namespace read {
 
-using PartitionDataHandleResult =
-    std::vector<std::pair<std::string, std::string>>;
-using PartitionsDataHandleExtendedResponse =
-    ExtendedApiResponse<PartitionDataHandleResult, client::ApiError,
-                        client::NetworkStatistics>;
-
 class PrefetchPartitionsHelper {
  public:
   using DownloadJob = DownloadItemsJob<std::string, PrefetchPartitionsResult,
@@ -60,9 +54,7 @@ class PrefetchPartitionsHelper {
       TaskSink& task_sink, size_t query_max_size, uint32_t priority) {
     client::CancellationContext execution_context;
 
-    auto query_job = std::make_shared<QueryMetadataJob<
-        std::string, std::vector<std::string>, PrefetchPartitionsResult,
-        PartitionsDataHandleExtendedResponse, PrefetchPartitionsStatus>>(
+    auto query_job = std::make_shared<QueryPartitionsJob>(
         std::move(query), nullptr, download_job, task_sink, execution_context,
         priority);
 

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

@@ -74,7 +74,18 @@ class QueryMetadataJob {
         execution_context_(execution_context),
         priority_(priority) {}
 
-  void Initialize(size_t query_count) { query_count_ = query_count; }
+  virtual ~QueryMetadataJob() = default;
+
+  virtual bool CheckIfFail() {
+    // The old behavior: when one of the query requests fails, we fail the
+    // entire prefetch.
+    return (!query_errors_.empty());
+  }
+
+  void Initialize(size_t query_count) {
+    query_count_ = query_count;
+    query_size_ = query_count;
+  }
 
   QueryResponseType Query(QueryType root, client::CancellationContext context) {
     return query_(root, context);
@@ -94,15 +105,14 @@ class QueryMetadataJob {
       if (error.GetErrorCode() == client::ErrorCode::Cancelled) {
         canceled_ = true;
       } else {
-        // The old behavior: when one of the query requests fails, we fail the
-        // entire prefetch.
-        query_error_ = error;
+        // Collect all errors.
+        query_errors_.push_back(error);
       }
     }
 
     if (!--query_count_) {
-      if (query_error_) {
-        download_job_->OnPrefetchCompleted(query_error_.value());
+      if (CheckIfFail()) {
+        download_job_->OnPrefetchCompleted(query_errors_.front());
         return;
       }
 
@@ -167,14 +177,15 @@ class QueryMetadataJob {
     }
   }
 
- private:
+ protected:
   QueryItemsFunc<ItemType, QueryType, QueryResponseType> query_;
   FilterItemsFunc<typename QueryResponseType::ResultType> filter_;
   size_t query_count_{0};
+  size_t query_size_{0};
   bool canceled_{false};
   typename QueryResponseType::ResultType query_result_;
   client::NetworkStatistics accumulated_statistics_;
-  boost::optional<client::ApiError> query_error_;
+  std::vector<client::ApiError> query_errors_;
   std::shared_ptr<
       DownloadItemsJob<ItemType, PrefetchResult, PrefetchStatusType>>
       download_job_;

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

@@ -0,0 +1,72 @@
+/*
+ * Copyright (C) 2020 HERE Europe B.V.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * License-Filename: LICENSE
+ */
+
+#pragma once
+
+#include <iterator>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include <olp/core/client/CancellationContext.h>
+#include <olp/core/logging/Log.h>
+#include <olp/dataservice/read/Types.h>
+#include "Common.h"
+#include "ExtendedApiResponse.h"
+#include "QueryMetadataJob.h"
+#include "TaskSink.h"
+
+namespace olp {
+namespace dataservice {
+namespace read {
+
+using PartitionDataHandleResult =
+    std::vector<std::pair<std::string, std::string>>;
+using PartitionsDataHandleExtendedResponse =
+    ExtendedApiResponse<PartitionDataHandleResult, client::ApiError,
+                        client::NetworkStatistics>;
+
+class QueryPartitionsJob
+    : public QueryMetadataJob<
+          std::string, std::vector<std::string>, PrefetchPartitionsResult,
+          PartitionsDataHandleExtendedResponse, PrefetchPartitionsStatus> {
+ public:
+  QueryPartitionsJob(
+      QueryItemsFunc<std::string, std::vector<std::string>,
+                     PartitionsDataHandleExtendedResponse>
+          query,
+      FilterItemsFunc<PartitionDataHandleResult> filter,
+      std::shared_ptr<DownloadItemsJob<std::string, PrefetchPartitionsResult,
+                                       PrefetchPartitionsStatus>>
+          download_job,
+      TaskSink& task_sink, client::CancellationContext execution_context,
+      uint32_t priority)
+      : QueryMetadataJob(std::move(query), std::move(filter),
+                         std::move(download_job), task_sink, execution_context,
+                         priority) {}
+
+  virtual bool CheckIfFail() {
+    // Return error only if all fails
+    return (query_errors_.size() == query_size_);
+  }
+};
+
+}  // namespace read
+}  // namespace dataservice
+}  // namespace olp

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

@@ -80,6 +80,22 @@ VersionedLayerClient::PrefetchTiles(PrefetchTilesRequest request,
   return impl_->PrefetchTiles(std::move(request), std::move(status_callback));
 }
 
+client::CancellationToken VersionedLayerClient::PrefetchPartitions(
+    PrefetchPartitionsRequest request,
+    PrefetchPartitionsResponseCallback callback,
+    PrefetchPartitionsStatusCallback status_callback) {
+  return impl_->PrefetchPartitions(std::move(request), std::move(callback),
+                                   std::move(status_callback));
+}
+
+client::CancellableFuture<PrefetchPartitionsResponse>
+VersionedLayerClient::PrefetchPartitions(
+    PrefetchPartitionsRequest request,
+    PrefetchPartitionsStatusCallback status_callback) {
+  return impl_->PrefetchPartitions(std::move(request),
+                                   std::move(status_callback));
+}
+
 client::CancellationToken VersionedLayerClient::GetData(
     TileRequest request, DataResponseCallback callback) {
   return impl_->GetData(std::move(request), std::move(callback));

tests/integration/CMakeLists.txt

@@ -30,6 +30,7 @@ set(OLP_SDK_INTEGRATIONAL_TESTS_SOURCES
     ./olp-cpp-sdk-dataservice-read/HttpResponses.h
     ./olp-cpp-sdk-dataservice-read/StreamLayerClientTest.cpp
     ./olp-cpp-sdk-dataservice-read/VersionedLayerClientTest.cpp
+    ./olp-cpp-sdk-dataservice-read/VersionedLayerClientPrefetchPartitionsTest.cpp
     ./olp-cpp-sdk-dataservice-read/VolatileLayerClientTest.cpp
     ./olp-cpp-sdk-dataservice-write/HttpResponses.h
     ./olp-cpp-sdk-dataservice-write/IndexLayerClientTest.cpp