Support remote arrays for fragment_list consolidation

Author: ypatia

test/src/unit-capi-consolidation.cc

@@ -7169,7 +7169,7 @@ TEST_CASE_METHOD(
 TEST_CASE_METHOD(
     ConsolidationFx,
     "C API: Test consolidation, dense split fragments",
-    "[capi][consolidation][dense][split-fragments][non-rest]") {
+    "[capi][consolidation][dense][split-fragments][rest]") {
   remove_dense_array();
   create_dense_array();
   write_dense_subarray(1, 2, 1, 2);
@@ -7253,7 +7253,7 @@ TEST_CASE_METHOD(
 TEST_CASE_METHOD(
     ConsolidationFx,
     "C API: Test consolidation, sparse split fragments",
-    "[capi][consolidation][sparse][split-fragments][non-rest]") {
+    "[capi][consolidation][sparse][split-fragments][rest]") {
   remove_sparse_array();
   create_sparse_array();
   write_sparse_row(0);

test/support/src/serialization_wrappers.cc

@@ -32,9 +32,12 @@
  */
 
 #include "test/support/src/helpers.h"
+#include "tiledb/api/c_api/buffer/buffer_api_internal.h"
+#include "tiledb/api/c_api/context/context_api_internal.h"
 #include "tiledb/sm/c_api/tiledb.h"
 #include "tiledb/sm/c_api/tiledb_serialization.h"
 #include "tiledb/sm/c_api/tiledb_struct_def.h"
+#include "tiledb/sm/serialization/consolidation.h"
 #include "tiledb/sm/serialization/query.h"
 
 #ifdef TILEDB_SERIALIZATION
@@ -218,3 +221,28 @@ void tiledb_subarray_serialize(
   *subarray = deserialized_subarray;
 #endif
 }
+
+void tiledb_array_consolidation_request_wrapper(
+    tiledb_ctx_t* ctx,
+    tiledb_serialization_type_t serialize_type,
+    const std::vector<std::string>* fragment_uris_in,
+    std::vector<std::string>* fragment_uris_out) {
+  // Serialize and Deserialize
+  auto buffer = tiledb_buffer_handle_t::make_handle();
+  serialization::array_consolidation_request_serialize(
+      ctx->config(),
+      fragment_uris_in,
+      static_cast<tiledb::sm::SerializationType>(serialize_type),
+      &(buffer->buffer()));
+
+  auto [config, fragment_uris_deser] =
+      serialization::array_consolidation_request_deserialize(
+          static_cast<tiledb::sm::SerializationType>(serialize_type),
+          buffer->buffer());
+
+  tiledb_buffer_handle_t::break_handle(buffer);
+
+  if (fragment_uris_deser.has_value()) {
+    *fragment_uris_out = fragment_uris_deser.value();
+  }
+}

test/support/src/serialization_wrappers.h

@@ -35,6 +35,7 @@
 #define TILEDB_TEST_SERIALIZATION_WRAPPERS_H
 
 #include <string>
+#include <vector>
 
 #include "tiledb/sm/c_api/tiledb.h"
 #include "tiledb/sm/c_api/tiledb_serialization.h"
@@ -141,4 +142,11 @@ int tiledb_fragment_info_serialize(
  */
 void tiledb_subarray_serialize(
     tiledb_ctx_t* ctx, tiledb_array_t* array, tiledb_subarray_t** subarray);
+
+void tiledb_array_consolidation_request_wrapper(
+    tiledb_ctx_t* ctx,
+    tiledb_serialization_type_t serialize_type,
+    const std::vector<std::string>* fragment_uris_in,
+    std::vector<std::string>* fragment_uris_out);
+
 #endif  // TILEDB_TEST_SERIALIZATION_WRAPPERS_H

tiledb/sm/c_api/tiledb.cc

@@ -2318,14 +2318,32 @@ int32_t tiledb_array_create(
 
 int32_t tiledb_array_consolidate(
     tiledb_ctx_t* ctx, const char* array_uri, tiledb_config_t* config) {
+  // Validate input arguments
+  api::ensure_context_is_valid(ctx);
   api::ensure_config_is_valid_if_present(config);
+
+  auto uri = tiledb::sm::URI(array_uri);
+  if (uri.is_invalid()) {
+    throw api::CAPIStatusException(
+        "Failed to consolidate fragments; Invalid input array uri");
+  }
+
+  auto input_config = (config == nullptr) ? ctx->config() : config->config();
+  if (uri.is_tiledb() &&
+      tiledb::sm::Consolidator::mode_from_config(input_config) ==
+          tiledb::sm::ConsolidationMode::FRAGMENT) {
+    throw api::CAPIStatusException(
+        "Please use tiledb_array_consolidate_fragments API for consolidating "
+        "fragments on remote arrays.");
+  }
+
   tiledb::sm::Consolidator::array_consolidate(
       ctx->resources(),
       array_uri,
       tiledb::sm::EncryptionType::NO_ENCRYPTION,
       nullptr,
       0,
-      (config == nullptr) ? ctx->config() : config->config(),
+      input_config,
       ctx->storage_manager());
   return TILEDB_OK;
 }
@@ -2336,7 +2354,33 @@ int32_t tiledb_array_consolidate_fragments(
     const char** fragment_uris,
     const uint64_t num_fragments,
     tiledb_config_t* config) {
-  // Sanity checks
+  // Validate input arguments
+  api::ensure_context_is_valid(ctx);
+  api::ensure_config_is_valid_if_present(config);
+
+  if (fragment_uris == nullptr) {
+    throw api::CAPIStatusException(
+        "Failed to consolidate fragments; Invalid input fragment list");
+  }
+
+  auto uri = tiledb::sm::URI(array_uri);
+  if (uri.is_invalid()) {
+    throw api::CAPIStatusException(
+        "Failed to consolidate fragments; Invalid input array uri");
+  }
+
+  if (num_fragments < 1) {
+    throw api::CAPIStatusException(
+        "Failed to consolidate fragments; Invalid input number of fragments");
+  }
+
+  for (size_t i = 0; i < num_fragments; i++) {
+    if (tiledb::sm::URI(fragment_uris[i]).is_invalid()) {
+      throw api::CAPIStatusException(
+          "Failed to consolidate fragments; Invalid uri(s) in input fragment "
+          "list");
+    }
+  }
 
   // Convert the list of fragments to a vector
   std::vector<std::string> uris;

tiledb/sm/consolidator/consolidator.cc

@@ -151,16 +151,16 @@ void Consolidator::array_consolidate(
     throw ConsolidatorException("Cannot consolidate array; Invalid URI");
   }
 
-  // Check if array exists
-  if (object_type(resources, array_uri) != ObjectType::ARRAY) {
-    throw ConsolidatorException(
-        "Cannot consolidate array; Array does not exist");
-  }
-
   if (array_uri.is_tiledb()) {
     throw_if_not_ok(
         resources.rest_client()->post_consolidation_to_rest(array_uri, config));
   } else {
+      // Check if array exists
+    if (object_type(resources, array_uri) != ObjectType::ARRAY) {
+      throw ConsolidatorException(
+          "Cannot consolidate array; Array does not exist");
+    }
+
     // Get encryption key from config
     std::string encryption_key_from_cfg;
     if (!encryption_key) {
@@ -212,46 +212,54 @@ void Consolidator::fragments_consolidate(
     throw ConsolidatorException("Cannot consolidate array; Invalid URI");
   }
 
-  // Check if array exists
-  if (object_type(resources, array_uri) != ObjectType::ARRAY) {
-    throw ConsolidatorException(
-        "Cannot consolidate array; Array does not exist");
-  }
+  if (array_uri.is_tiledb()) {
+    throw_if_not_ok(
+        storage_manager->resources().rest_client()->post_consolidation_to_rest(
+            array_uri, config, &fragment_uris));
+  } else {
+      // Check if array exists
+    if (object_type(resources, array_uri) != ObjectType::ARRAY) {
+      throw ConsolidatorException(
+          "Cannot consolidate array; Array does not exist");
+    }
 
-  // Get encryption key from config
-  std::string encryption_key_from_cfg;
-  if (!encryption_key) {
-    bool found = false;
-    encryption_key_from_cfg = config.get("sm.encryption_key", &found);
-    assert(found);
-  }
+    // Get encryption key from config
+    std::string encryption_key_from_cfg;
+    if (!encryption_key) {
+      bool found = false;
+      encryption_key_from_cfg = config.get("sm.encryption_key", &found);
+      assert(found);
+    }
 
-  if (!encryption_key_from_cfg.empty()) {
-    encryption_key = encryption_key_from_cfg.c_str();
-    key_length = static_cast<uint32_t>(encryption_key_from_cfg.size());
-    std::string encryption_type_from_cfg;
-    bool found = false;
-    encryption_type_from_cfg = config.get("sm.encryption_type", &found);
-    assert(found);
-    auto [st, et] = encryption_type_enum(encryption_type_from_cfg);
-    throw_if_not_ok(st);
-    encryption_type = et.value();
-
-    if (!EncryptionKey::is_valid_key_length(
-            encryption_type,
-            static_cast<uint32_t>(encryption_key_from_cfg.size()))) {
-      encryption_key = nullptr;
-      key_length = 0;
+    if (!encryption_key_from_cfg.empty()) {
+      encryption_key = encryption_key_from_cfg.c_str();
+      key_length = static_cast<uint32_t>(encryption_key_from_cfg.size());
+      std::string encryption_type_from_cfg;
+      bool found = false;
+      encryption_type_from_cfg = config.get("sm.encryption_type", &found);
+      assert(found);
+      auto [st, et] = encryption_type_enum(encryption_type_from_cfg);
+      throw_if_not_ok(st);
+      encryption_type = et.value();
+
+      if (!EncryptionKey::is_valid_key_length(
+              encryption_type,
+              static_cast<uint32_t>(encryption_key_from_cfg.size()))) {
+        encryption_key = nullptr;
+        key_length = 0;
+      }
     }
-  }
 
-  // Consolidate
-  auto consolidator = Consolidator::create(
-      resources, ConsolidationMode::FRAGMENT, config, storage_manager);
-  auto fragment_consolidator =
-      dynamic_cast<FragmentConsolidator*>(consolidator.get());
-  throw_if_not_ok(fragment_consolidator->consolidate_fragments(
-      array_name, encryption_type, encryption_key, key_length, fragment_uris));
+    // Consolidate
+    auto fragment_consolidator =
+        make_shared<FragmentConsolidator>(HERE(), config, storage_manager);
+    throw_if_not_ok(fragment_consolidator->consolidate_fragments(
+        array_name,
+        encryption_type,
+        encryption_key,
+        key_length,
+        fragment_uris));
+  }
 }
 
 void Consolidator::write_consolidated_commits_file(

tiledb/sm/rest/rest_client.h

@@ -463,7 +463,7 @@ class RestClient {
   }
 
   /// Operation disabled in base class.
-  inline virtual Status post_consolidation_to_rest(const URI&, const Config&) {
+  inline virtual Status post_consolidation_to_rest(const URI&, const Config&, const std::vector<std::string>*) {
     throw RestClientDisabledException();
   }
 

tiledb/sm/rest/rest_client_remote.cc

@@ -1524,11 +1524,14 @@ Status RestClientRemote::ensure_json_null_delimited_string(Buffer* buffer) {
   return Status::Ok();
 }
 
-Status RestClientRemote::post_consolidation_to_rest(
-    const URI& uri, const Config& config) {
+Status RestClient::post_consolidation_to_rest(
+    const URI& uri,
+    const Config& config,
+    const std::vector<std::string>* fragment_uris) {
   Buffer buff;
-  RETURN_NOT_OK(serialization::array_consolidation_request_serialize(
-      config, serialization_type_, &buff));
+  serialization::array_consolidation_request_serialize(
+      config, fragment_uris, serialization_type_, &buff);
+
   // Wrap in a list
   BufferList serialized;
   RETURN_NOT_OK(serialized.add_buffer(std::move(buff)));

tiledb/sm/rest/rest_client_remote.h

@@ -425,10 +425,14 @@ class RestClientRemote : public RestClient {
    *
    * @param uri Array URI
    * @param config config
+   * @param fragment_uris The uris of the fragments to be consolidated if this
+   * is a request for fragment list consolidation
    * @return
    */
   Status post_consolidation_to_rest(
-      const URI& uri, const Config& config) override;
+      const URI& uri,
+      const Config& config,
+      const std::vector<std::string>* fragment_uris = nullptr) override;
 
   /**
    * Post array vacuum request to the REST server.