Avoid using kInAtomicGroup tag for single-cf op (#4981)

Summary:
if an operation just involves a single column family, then we do
not have to set the kInAtomicGroup tag when writing to MANIFEST. This change
can fix a compatibility test failure, i.e. 5.15 and earlier cannot recognize
kInAtomicGroup tag.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/4981

Differential Revision: D14072687

Pulled By: riversand963

fbshipit-source-id: 46b0c61e399f16c6b7169de0b33430d0ed90d6d4

db/db_impl.cc

@@ -3337,12 +3337,15 @@ Status DBImpl::IngestExternalFiles(
         edit_lists.push_back(edit_list);
         ++num_entries;
       }
-      // Mark the version edits as an atomic group
-      for (auto& edits : edit_lists) {
-        assert(edits.size() == 1);
-        edits[0]->MarkAtomicGroup(--num_entries);
+      // Mark the version edits as an atomic group if the number of version
+      // edits exceeds 1.
+      if (cfds_to_commit.size() > 1) {
+        for (auto& edits : edit_lists) {
+          assert(edits.size() == 1);
+          edits[0]->MarkAtomicGroup(--num_entries);
+        }
+        assert(0 == num_entries);
       }
-      assert(0 == num_entries);
       status =
           versions_->LogAndApply(cfds_to_commit, mutable_cf_options_list,
                                  edit_lists, &mutex_, directories_.GetDbDir());

db/memtable_list.cc

@@ -573,12 +573,15 @@ Status InstallMemtableAtomicFlushResults(
     ++num_entries;
     edit_lists.emplace_back(edits);
   }
-  // Mark the version edits as an atomic group
-  for (auto& edits : edit_lists) {
-    assert(edits.size() == 1);
-    edits[0]->MarkAtomicGroup(--num_entries);
+  // Mark the version edits as an atomic group if the number of version edits
+  // exceeds 1.
+  if (cfds.size() > 1) {
+    for (auto& edits : edit_lists) {
+      assert(edits.size() == 1);
+      edits[0]->MarkAtomicGroup(--num_entries);
+    }
+    assert(0 == num_entries);
   }
-  assert(0 == num_entries);
 
   // this can release and reacquire the mutex.
   s = vset->LogAndApply(cfds, mutable_cf_options_list, edit_lists, mu,

include/rocksdb/db.h

@@ -926,6 +926,8 @@ class DB {
   // If atomic flush is enabled, Flush(options, column_families) will flush all
   // column families specified in 'column_families' up to the latest sequence
   // number at the time when flush is requested.
+  // Note that RocksDB 5.15 and earlier may not be able to open later versions
+  // with atomic flush enabled.
   virtual Status Flush(
       const FlushOptions& options,
       const std::vector<ColumnFamilyHandle*>& column_families) = 0;
@@ -1067,6 +1069,8 @@ class DB {
   // column families with iterators, iterator on one column family may return
   // ingested data, while iterator on other column family returns old data.
   // Users can use snapshot for a consistent view of data.
+  // If your db ingests multiple SST files using this API, i.e. args.size()
+  // > 1, then RocksDB 5.15 and earlier will not be able to open it.
   //
   // REQUIRES: each arg corresponds to a different column family: namely, for
   // 0 <= i < j < len(args), args[i].column_family != args[j].column_family.

tools/check_format_compatible.sh

@@ -56,7 +56,7 @@ declare -a backward_compatible_checkout_objs=("2.2.fb.branch" "2.3.fb.branch" "2
 declare -a forward_compatible_checkout_objs=("3.10.fb" "3.11.fb" "3.12.fb" "3.13.fb" "4.0.fb" "4.1.fb" "4.2.fb" "4.3.fb" "4.4.fb" "4.5.fb" "4.6.fb" "4.7.fb" "4.8.fb" "4.9.fb" "4.10.fb" "4.11.fb" "4.12.fb" "4.13.fb" "5.0.fb" "5.1.fb" "5.2.fb" "5.3.fb" "5.4.fb" "5.5.fb" "5.6.fb" "5.7.fb" "5.8.fb" "5.9.fb" "5.10.fb")
 declare -a forward_compatible_with_options_checkout_objs=("5.11.fb" "5.12.fb" "5.13.fb" "5.14.fb")
 declare -a checkout_objs=(${backward_compatible_checkout_objs[@]} ${forward_compatible_checkout_objs[@]} ${forward_compatible_with_options_checkout_objs[@]})
-declare -a extern_sst_ingestion_compatible_checkout_objs=("5.14.fb" "5.15.fb" "5.16.fb" "5.17.fb")
+declare -a extern_sst_ingestion_compatible_checkout_objs=("5.14.fb" "5.15.fb" "5.16.fb" "5.17.fb" "5.18.fb")
 
 generate_db()
 {