Fix out of order consolidation. (#4597)

The following issue TileDB-Inc/TileDB-Py#1885
reported that running consolidation in this order invalidates an array:
- consolidate fragments
- consolidate commits
- vacuum fragments
- vacuum commits

This fixes two issues:
- We sometimes would not delete the commit files when vacuuming
fragments.
- We would vacuum an ignore file when a consolidated commits file still
needed it.

---
TYPE: BUG
DESC: Fix out of order consolidation.

(cherry picked from commit 65d1ce4)

test/src/unit-capi-consolidation.cc

@@ -7217,4 +7217,30 @@ TEST_CASE_METHOD(
                   "progress can be made") == msg);
 
   remove_sparse_string_array();
+}
+
+TEST_CASE_METHOD(
+    ConsolidationFx,
+    "C API: Test consolidation, fragments/commits out of order",
+    "[capi][consolidation][fragments-commits][out-of-order]") {
+#ifdef TILEDB_SERIALIZATION
+  serialize_ = GENERATE(true, false);
+  if (serialize_) {
+    refactored_query_v2_ = GENERATE(true, false);
+  }
+#endif
+
+  remove_sparse_array();
+  create_sparse_array();
+
+  write_sparse_full();
+  write_sparse_unordered();
+  consolidate_sparse("fragments");
+  consolidate_sparse("commits");
+  vacuum_sparse("fragments");
+  vacuum_sparse("commits");
+  read_sparse_full_unordered();
+  check_commits_dir_sparse(1, 0, 1);
+
+  remove_sparse_array();
 }

tiledb/sm/array/array_directory.cc

@@ -829,11 +829,17 @@ ArrayDirectory::load_consolidated_commit_uris(
     for (auto& meta_file : meta_files) {
       std::stringstream ss(meta_file.second);
       uint64_t count = 0;
+      bool all_in_set = true;
       for (std::string uri_str; std::getline(ss, uri_str);) {
-        count += uris_set.count(uri_.to_string() + uri_str);
+        if (uris_set.count(uri_.to_string() + uri_str) > 0) {
+          count++;
+        } else {
+          all_in_set = false;
+          break;
+        }
       }
 
-      if (count == uris_set.size()) {
+      if (all_in_set && count == uris_set.size()) {
         for (auto& uri : commits_dir_uris) {
           if (stdx::string::ends_with(
                   uri.to_string(), constants::con_commits_file_suffix)) {

tiledb/sm/consolidator/fragment_consolidator.cc

@@ -314,10 +314,9 @@ void FragmentConsolidator::vacuum(const char* array_name) {
     array_dir.write_commit_ignore_file(commit_uris_to_ignore);
   }
 
-  // Delete the commit and vacuum files
+  // Delete the vacuum files.
   auto vfs = storage_manager_->vfs();
   auto compute_tp = storage_manager_->compute_tp();
-  vfs->remove_files(compute_tp, filtered_fragment_uris.commit_uris_to_vacuum());
   vfs->remove_files(
       compute_tp, filtered_fragment_uris.fragment_vac_uris_to_vacuum());
 
@@ -326,6 +325,14 @@ void FragmentConsolidator::vacuum(const char* array_name) {
       compute_tp, 0, fragment_uris_to_vacuum.size(), [&](size_t i) {
         RETURN_NOT_OK(vfs->remove_dir(fragment_uris_to_vacuum[i]));
 
+        // Remove the commit file, if present.
+        auto commit_uri = array_dir.get_commit_uri(fragment_uris_to_vacuum[i]);
+        bool is_file = false;
+        RETURN_NOT_OK(vfs->is_file(commit_uri, &is_file));
+        if (is_file) {
+          RETURN_NOT_OK(vfs->remove_file(commit_uri));
+        }
+
         return Status::Ok();
       }));
 }