Move consolidation calls out of StorageManager. (#4668)

This moves consolidation out of StorageManager using static methods in
`tiledb::sm::Consolidator`. The subclasses that derive from Consolidator
will still need access to the StorageManager for opening the arrays and
groups in the `consolidate` and `vacuum` overrides.

---
TYPE: NO_HISTORY
DESC: Move consolidation calls out of StorageManager.

Author: shaunrd0

tiledb/sm/c_api/tiledb.cc

@@ -58,6 +58,7 @@
 #include "tiledb/sm/c_api/api_argument_validator.h"
 #include "tiledb/sm/config/config.h"
 #include "tiledb/sm/config/config_iter.h"
+#include "tiledb/sm/consolidator/consolidator.h"
 #include "tiledb/sm/cpp_api/core_interface.h"
 #include "tiledb/sm/enums/array_type.h"
 #include "tiledb/sm/enums/encryption_type.h"
@@ -2702,13 +2703,13 @@ int32_t tiledb_array_create_with_key(
 int32_t tiledb_array_consolidate(
     tiledb_ctx_t* ctx, const char* array_uri, tiledb_config_t* config) {
   api::ensure_config_is_valid_if_present(config);
-  throw_if_not_ok(ctx->storage_manager()->array_consolidate(
+  tiledb::sm::Consolidator::array_consolidate(
       array_uri,
       tiledb::sm::EncryptionType::NO_ENCRYPTION,
       nullptr,
       0,
-      (config == nullptr) ? ctx->storage_manager()->config() :
-                            config->config()));
+      (config == nullptr) ? ctx->storage_manager()->config() : config->config(),
+      ctx->storage_manager());
   return TILEDB_OK;
 }
 
@@ -2721,13 +2722,13 @@ int32_t tiledb_array_consolidate_with_key(
     tiledb_config_t* config) {
   // Sanity checks
 
-  throw_if_not_ok(ctx->storage_manager()->array_consolidate(
+  tiledb::sm::Consolidator::array_consolidate(
       array_uri,
       static_cast<tiledb::sm::EncryptionType>(encryption_type),
       encryption_key,
       key_length,
-      (config == nullptr) ? ctx->storage_manager()->config() :
-                            config->config()));
+      (config == nullptr) ? ctx->storage_manager()->config() : config->config(),
+      ctx->storage_manager());
 
   return TILEDB_OK;
 }
@@ -2747,24 +2748,24 @@ int32_t tiledb_array_consolidate_fragments(
     uris.emplace_back(fragment_uris[i]);
   }
 
-  throw_if_not_ok(ctx->storage_manager()->fragments_consolidate(
+  tiledb::sm::Consolidator::fragments_consolidate(
       array_uri,
       tiledb::sm::EncryptionType::NO_ENCRYPTION,
       nullptr,
       0,
       uris,
-      (config == nullptr) ? ctx->storage_manager()->config() :
-                            config->config()));
+      (config == nullptr) ? ctx->storage_manager()->config() : config->config(),
+      ctx->storage_manager());
 
   return TILEDB_OK;
 }
 
 int32_t tiledb_array_vacuum(
     tiledb_ctx_t* ctx, const char* array_uri, tiledb_config_t* config) {
-  ctx->storage_manager()->array_vacuum(
+  tiledb::sm::Consolidator::array_vacuum(
       array_uri,
-      (config == nullptr) ? ctx->storage_manager()->config() :
-                            config->config());
+      (config == nullptr) ? ctx->storage_manager()->config() : config->config(),
+      ctx->storage_manager());
 
   return TILEDB_OK;
 }

tiledb/sm/consolidator/array_meta_consolidator.h

@@ -85,14 +85,14 @@ class ArrayMetaConsolidator : public Consolidator {
       const char* array_name,
       EncryptionType encryption_type,
       const void* encryption_key,
-      uint32_t key_length);
+      uint32_t key_length) override;
 
   /**
    * Performs the vacuuming operation.
    *
    * @param array_name URI of array to consolidate.
    */
-  void vacuum(const char* array_name);
+  void vacuum(const char* array_name) override;
 
  private:
   /* ********************************* */

tiledb/sm/consolidator/commits_consolidator.cc

@@ -95,8 +95,8 @@ Status CommitsConsolidator::consolidate(
 
   // Get the file name.
   auto& to_consolidate = array_dir.commit_uris_to_consolidate();
-  storage_manager_->write_consolidated_commits_file(
-      write_version, array_dir, to_consolidate);
+  Consolidator::write_consolidated_commits_file(
+      write_version, array_dir, to_consolidate, storage_manager_);
 
   return Status::Ok();
 }

tiledb/sm/consolidator/commits_consolidator.h

@@ -83,14 +83,14 @@ class CommitsConsolidator : public Consolidator {
       const char* array_name,
       EncryptionType encryption_type,
       const void* encryption_key,
-      uint32_t key_length);
+      uint32_t key_length) override;
 
   /**
    * Performs the vacuuming operation.
    *
    * @param array_name URI of array to consolidate.
    */
-  void vacuum(const char* array_name);
+  void vacuum(const char* array_name) override;
 };
 
 }  // namespace tiledb::sm

tiledb/sm/consolidator/consolidator.cc

@@ -32,12 +32,16 @@
 
 #include "tiledb/sm/consolidator/consolidator.h"
 #include "tiledb/common/logger.h"
+#include "tiledb/common/stdx_string.h"
 #include "tiledb/sm/consolidator/array_meta_consolidator.h"
 #include "tiledb/sm/consolidator/commits_consolidator.h"
 #include "tiledb/sm/consolidator/fragment_consolidator.h"
 #include "tiledb/sm/consolidator/fragment_meta_consolidator.h"
 #include "tiledb/sm/consolidator/group_meta_consolidator.h"
+#include "tiledb/sm/enums/encryption_type.h"
+#include "tiledb/sm/rest/rest_client.h"
 #include "tiledb/sm/storage_manager/storage_manager.h"
+#include "tiledb/storage_format/uri/generate_uri.h"
 
 using namespace tiledb::common;
 
@@ -121,12 +125,207 @@ Status Consolidator::consolidate(
 }
 
 void Consolidator::vacuum([[maybe_unused]] const char* array_name) {
-  throw Status_ConsolidatorError("Cannot vacuum; Invalid object");
+  throw ConsolidatorException("Cannot vacuum; Invalid object");
+}
+
+void Consolidator::array_consolidate(
+    const char* array_name,
+    EncryptionType encryption_type,
+    const void* encryption_key,
+    uint32_t key_length,
+    const Config& config,
+    StorageManager* storage_manager) {
+  // Check array URI
+  URI array_uri(array_name);
+  if (array_uri.is_invalid()) {
+    throw ConsolidatorException("Cannot consolidate array; Invalid URI");
+  }
+
+  // Check if array exists
+  ObjectType obj_type;
+  throw_if_not_ok(storage_manager->object_type(array_uri, &obj_type));
+
+  if (obj_type != ObjectType::ARRAY) {
+    throw ConsolidatorException(
+        "Cannot consolidate array; Array does not exist");
+  }
+
+  if (array_uri.is_tiledb()) {
+    throw_if_not_ok(storage_manager->rest_client()->post_consolidation_to_rest(
+        array_uri, config));
+  } else {
+    // 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;
+      }
+    }
+
+    // Consolidate
+    auto mode = Consolidator::mode_from_config(config);
+    auto consolidator = Consolidator::create(mode, config, storage_manager);
+    throw_if_not_ok(consolidator->consolidate(
+        array_name, encryption_type, encryption_key, key_length));
+  }
+}
+
+void Consolidator::fragments_consolidate(
+    const char* array_name,
+    EncryptionType encryption_type,
+    const void* encryption_key,
+    uint32_t key_length,
+    const std::vector<std::string> fragment_uris,
+    const Config& config,
+    StorageManager* storage_manager) {
+  // Check array URI
+  URI array_uri(array_name);
+  if (array_uri.is_invalid()) {
+    throw ConsolidatorException("Cannot consolidate array; Invalid URI");
+  }
+
+  // Check if array exists
+  ObjectType obj_type;
+  throw_if_not_ok(storage_manager->object_type(array_uri, &obj_type));
+
+  if (obj_type != 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);
+  }
+
+  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(
+      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));
+}
+
+void Consolidator::write_consolidated_commits_file(
+    format_version_t write_version,
+    ArrayDirectory array_dir,
+    const std::vector<URI>& commit_uris,
+    StorageManager* storage_manager) {
+  // Compute the file name.
+  auto name = storage_format::generate_consolidated_fragment_name(
+      commit_uris.front(), commit_uris.back(), write_version);
+
+  // Compute size of consolidated file. Save the sizes of the files to re-use
+  // below.
+  storage_size_t total_size = 0;
+  const auto base_uri_size = array_dir.uri().to_string().size();
+  std::vector<storage_size_t> file_sizes(commit_uris.size());
+  for (uint64_t i = 0; i < commit_uris.size(); i++) {
+    const auto& uri = commit_uris[i];
+    total_size += uri.to_string().size() - base_uri_size + 1;
+
+    // If the file is a delete, add the file size to the count and the size of
+    // the size variable.
+    if (stdx::string::ends_with(
+            uri.to_string(), constants::delete_file_suffix)) {
+      throw_if_not_ok(storage_manager->vfs()->file_size(uri, &file_sizes[i]));
+      total_size += file_sizes[i];
+      total_size += sizeof(storage_size_t);
+    }
+  }
+
+  // Write consolidated file, URIs are relative to the array URI.
+  std::vector<uint8_t> data(total_size);
+  storage_size_t file_index = 0;
+  for (uint64_t i = 0; i < commit_uris.size(); i++) {
+    // Add the uri.
+    const auto& uri = commit_uris[i];
+    std::string relative_uri = uri.to_string().substr(base_uri_size) + "\n";
+    memcpy(&data[file_index], relative_uri.data(), relative_uri.size());
+    file_index += relative_uri.size();
+
+    // For deletes, read the delete condition to the output file.
+    if (stdx::string::ends_with(
+            uri.to_string(), constants::delete_file_suffix)) {
+      memcpy(&data[file_index], &file_sizes[i], sizeof(storage_size_t));
+      file_index += sizeof(storage_size_t);
+      throw_if_not_ok(storage_manager->vfs()->read(
+          uri, 0, &data[file_index], file_sizes[i]));
+      file_index += file_sizes[i];
+    }
+  }
+
+  // Write the file to storage.
+  URI consolidated_commits_uri =
+      array_dir.get_commits_dir(write_version)
+          .join_path(name + constants::con_commits_file_suffix);
+  throw_if_not_ok(storage_manager->vfs()->write(
+      consolidated_commits_uri, data.data(), data.size()));
+  throw_if_not_ok(storage_manager->vfs()->close_file(consolidated_commits_uri));
+}
+
+void Consolidator::array_vacuum(
+    const char* array_name,
+    const Config& config,
+    StorageManager* storage_manager) {
+  URI array_uri(array_name);
+  if (array_uri.is_tiledb()) {
+    throw_if_not_ok(
+        storage_manager->rest_client()->post_vacuum_to_rest(array_uri, config));
+    return;
+  }
+
+  auto mode = Consolidator::mode_from_config(config, true);
+  auto consolidator = Consolidator::create(mode, config, storage_manager);
+  consolidator->vacuum(array_name);
 }
 
 void Consolidator::check_array_uri(const char* array_name) {
   if (URI(array_name).is_tiledb()) {
-    throw std::logic_error("Consolidation is not supported for remote arrays.");
+    throw ConsolidatorException(
+        "Consolidation is not supported for remote arrays.");
   }
 }