Add a new transaction option for large transaction optimization (facebook#13582)

cbi42 · facebook-github-bot · commit e3b7dd7b5642 · 2025-05-02T12:16:02.000-07:00
Summary: I added `TransactionDBOptions::txn_commit_bypass_memtable_threshold` previously but per DB option is not dynamically changeable. Adding it as a per transaction option to make it easier to use. The option naming is updated to make it easier for customer to understand `large_txn_commit_optimize_threshold`. The transaction DB option `TransactionDBOptions::txn_commit_bypass_memtable_threshold` is marked as deprecated. Pull Request resolved: facebook#13582 Test Plan: - new unit test - updated stress test to use this new transaction option Reviewed By: jowlyzhang Differential Revision: D73960981 Pulled By: cbi42 fbshipit-source-id: 406f6e0f5f4eb6b336976f9a93b0bc08e61a9662
diff --git a/db_stress_tool/db_stress_test_base.cc b/db_stress_tool/db_stress_test_base.cc
@@ -838,11 +838,15 @@ Status StressTest::NewTxn(WriteOptions& write_opts, ThreadState* thread,
         FLAGS_use_only_the_last_commit_time_batch_for_recovery;
     txn_options.lock_timeout = 600000;  // 10 min
     txn_options.deadlock_detect = true;
-    if (FLAGS_commit_bypass_memtable_one_in > 0) {
+    if (FLAGS_commit_bypass_memtable_one_in > 0 &&
+        thread->rand.OneIn(FLAGS_commit_bypass_memtable_one_in)) {
       assert(FLAGS_txn_write_policy == 0);
       assert(FLAGS_user_timestamp_size == 0);
-      txn_options.commit_bypass_memtable =
-          thread->rand.OneIn(FLAGS_commit_bypass_memtable_one_in);
+      if (thread->rand.OneIn(2)) {
+        txn_options.commit_bypass_memtable = true;
+      } else {
+        txn_options.large_txn_commit_optimize_threshold = 1;
+      }
       if (commit_bypass_memtable) {
         *commit_bypass_memtable = txn_options.commit_bypass_memtable;
       }
diff --git a/include/rocksdb/utilities/transaction_db.h b/include/rocksdb/utilities/transaction_db.h
@@ -254,10 +254,12 @@ struct TransactionDBOptions {
   // for more details.
   std::vector<std::shared_ptr<SecondaryIndex>> secondary_indices;
 
-  // EXPERIMENTAL, SUBJECT TO CHANGE
+  // Deprecated, this option may be removed in the future.
+  // Use TransactionOptions::large_txn_commit_optimize_threshold instead.
+  //
   // This option is only valid for write committed. If the number of updates in
-  // a transaction exceeds this threshold, then the transaction commit will skip
-  // insertions into memtable as an optimization to reduce commit latency.
+  // a transaction is at least this threshold, then the transaction commit will
+  // skip insertions into memtable as an optimization to reduce commit latency.
   // See comment for TransactionOptions::commit_bypass_memtable for more detail.
   // Setting TransactionOptions::commit_bypass_memtable to true takes precedence
   // over this option.
@@ -387,6 +389,13 @@ struct TransactionOptions {
   // can cause flush/compaction to report `num_single_del_mismatch` due to
   // consecutive SingleDeletes.
   bool commit_bypass_memtable = false;
+
+  // EXPERIMENTAL, SUBJECT TO CHANGE
+  // When the number of updates in a transaction is at least this threshold,
+  // we will enable optimizations for commiting a large transaction. See
+  // comment for `commit_bypass_memtable` for more optimization detail.
+  uint32_t large_txn_commit_optimize_threshold =
+      std::numeric_limits<uint32_t>::max();
 };
 
 // The per-write optimizations that do not involve transactions. TransactionDB
diff --git a/unreleased_history/new_features/per-txn-threshold.md b/unreleased_history/new_features/per-txn-threshold.md
@@ -0,0 +1 @@
+* Add new experimental `TransactionOptions::large_txn_commit_optimize_threshold` to enable optimizations for large transaction commit with per transaction threshold. `TransactionDBOptions::txn_commit_bypass_memtable_threshold` is deprecated in favor of this transaction option.
diff --git a/utilities/transactions/pessimistic_transaction.cc b/utilities/transactions/pessimistic_transaction.cc
@@ -105,7 +105,12 @@ void PessimisticTransaction::Initialize(const TransactionOptions& txn_options) {
   commit_timestamp_ = kMaxTxnTimestamp;
 
   if (txn_options.commit_bypass_memtable) {
-    commit_bypass_memtable_threshold_ = 0;
+    // No need to optimize for empty transction
+    commit_bypass_memtable_threshold_ = 1;
+  } else if (txn_options.large_txn_commit_optimize_threshold !=
+             std::numeric_limits<uint32_t>::max()) {
+    commit_bypass_memtable_threshold_ =
+        txn_options.large_txn_commit_optimize_threshold;
   } else {
     commit_bypass_memtable_threshold_ =
         db_options.txn_commit_bypass_memtable_threshold;
@@ -887,7 +892,7 @@ Status WriteCommittedTxn::CommitInternal() {
   // any operations appended to this working_batch will be ignored from WAL
   working_batch->MarkWalTerminationPoint();
 
-  bool bypass_memtable = wb->Count() > commit_bypass_memtable_threshold_;
+  bool bypass_memtable = wb->Count() >= commit_bypass_memtable_threshold_;
   if (!bypass_memtable) {
     // insert prepared batch into Memtable only skipping WAL.
     // Memtable will ignore BeginPrepare/EndPrepare markers
diff --git a/utilities/transactions/transaction_test.cc b/utilities/transactions/transaction_test.cc
@@ -9414,8 +9414,8 @@ TEST_P(CommitBypassMemtableTest, ThresholdTxnDBOption) {
   ASSERT_OK(txn1->Commit());
   ASSERT_TRUE(commit_bypass_memtable);
 
-  // Below threshold
-  for (auto num_ops : {threshold, threshold + 1}) {
+  // Test threshold behavior
+  for (auto num_ops : {threshold - 1, threshold}) {
     commit_bypass_memtable = false;
     txn_opts.commit_bypass_memtable = false;
     auto txn = txn_db->BeginTransaction(wopts, txn_opts, txn1);
@@ -9427,16 +9427,16 @@ TEST_P(CommitBypassMemtableTest, ThresholdTxnDBOption) {
     }
     ASSERT_OK(txn->Prepare());
     ASSERT_OK(txn->Commit());
-    ASSERT_EQ(commit_bypass_memtable, num_ops > threshold);
+    ASSERT_EQ(commit_bypass_memtable, num_ops >= threshold);
     delete txn;
   }
 
   // Repeat the same test with updates to two CFs
   std::vector<std::string> cfs = {"pk", "sk"};
   CreateColumnFamilies(cfs, options);
 
-  // Below threshold
-  for (auto num_ops : {threshold, threshold + 1}) {
+  // Test threshold behavior with CFs
+  for (auto num_ops : {threshold - 1, threshold}) {
     commit_bypass_memtable = false;
     txn_opts.commit_bypass_memtable = false;
     auto txn_cf = txn_db->BeginTransaction(wopts, txn_opts, nullptr);
@@ -9447,11 +9447,129 @@ TEST_P(CommitBypassMemtableTest, ThresholdTxnDBOption) {
     }
     ASSERT_OK(txn_cf->Prepare());
     ASSERT_OK(txn_cf->Commit());
-    ASSERT_EQ(commit_bypass_memtable, num_ops > threshold);
+    ASSERT_EQ(commit_bypass_memtable, num_ops >= threshold);
     delete txn_cf;
   }
 }
 
+TEST_P(CommitBypassMemtableTest, OptimizeLargeTxnCommitThreshold) {
+  // Tests TransactionOptions::large_txn_commit_optimize_threshold
+  const uint32_t threshold = 10;
+  SetUpTransactionDB();
+  bool commit_bypass_memtable = false;
+  SyncPoint::GetInstance()->SetCallBack(
+      "WriteCommittedTxn::CommitInternal:bypass_memtable",
+      [&](void* arg) { commit_bypass_memtable = *(static_cast<bool*>(arg)); });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  // Test with transaction option only
+  WriteOptions wopts;
+  TransactionOptions txn_opts;
+  txn_opts.large_txn_commit_optimize_threshold = threshold;
+
+  // Test with transaction below threshold
+  auto txn1 = txn_db->BeginTransaction(wopts, txn_opts, nullptr);
+  ASSERT_OK(txn1->SetName("xid1"));
+  ASSERT_OK(txn1->Put("k1", "v1"));
+  ASSERT_OK(txn1->Prepare());
+  ASSERT_OK(txn1->Commit());
+  ASSERT_FALSE(commit_bypass_memtable);
+  delete txn1;
+
+  // Test with transaction at threshold
+  txn1 = txn_db->BeginTransaction(wopts, txn_opts, nullptr);
+  ASSERT_OK(txn1->SetName("xid2"));
+  for (uint32_t i = 0; i < threshold; ++i) {
+    ASSERT_OK(
+        txn1->Put("key" + std::to_string(i), "value" + std::to_string(i)));
+  }
+  ASSERT_OK(txn1->Prepare());
+  ASSERT_OK(txn1->Commit());
+  ASSERT_TRUE(commit_bypass_memtable);
+  delete txn1;
+
+  // Test with both DB option and transaction option - transaction option should
+  // take precedence
+  SetUpTransactionDB(/*threshold=*/threshold * 2);
+
+  // Transaction option is lower than DB option, should use transaction option
+  txn_opts.large_txn_commit_optimize_threshold = threshold;
+  txn1 = txn_db->BeginTransaction(wopts, txn_opts, nullptr);
+  ASSERT_OK(txn1->SetName("xid3"));
+  for (uint32_t i = 0; i < threshold; ++i) {
+    ASSERT_OK(
+        txn1->Put("key" + std::to_string(i), "value" + std::to_string(i)));
+  }
+  ASSERT_OK(txn1->Prepare());
+  commit_bypass_memtable = false;
+  ASSERT_OK(txn1->Commit());
+  ASSERT_TRUE(commit_bypass_memtable);
+  delete txn1;
+
+  // Transaction option is higher than DB option, should use transaction option
+  txn_opts.large_txn_commit_optimize_threshold = threshold * 3;
+  txn1 = txn_db->BeginTransaction(wopts, txn_opts, nullptr);
+  ASSERT_OK(txn1->SetName("xid4"));
+  for (uint32_t i = 0; i < threshold * 3 - 1; ++i) {
+    ASSERT_OK(
+        txn1->Put("key" + std::to_string(i), "value" + std::to_string(i)));
+  }
+  ASSERT_OK(txn1->Prepare());
+  commit_bypass_memtable = false;
+  ASSERT_OK(txn1->Commit());
+  ASSERT_FALSE(commit_bypass_memtable);
+  delete txn1;
+
+  SetUpTransactionDB();
+  // Test with multiple column families
+  std::vector<std::string> cfs = {"pk", "sk"};
+  CreateColumnFamilies(cfs, options);
+
+  txn_opts.large_txn_commit_optimize_threshold = threshold;
+
+  // Below threshold
+  auto txn_cf = txn_db->BeginTransaction(wopts, txn_opts, nullptr);
+  ASSERT_OK(txn_cf->SetName("xid_cf_below"));
+  for (uint32_t i = 0; i < threshold - 1; ++i) {
+    ASSERT_OK(txn_cf->Put(handles_[i % 2], "key" + std::to_string(i),
+                          "value" + std::to_string(i)));
+  }
+  ASSERT_OK(txn_cf->Prepare());
+  commit_bypass_memtable = false;
+  ASSERT_OK(txn_cf->Commit());
+  ASSERT_FALSE(commit_bypass_memtable);
+  delete txn_cf;
+
+  // At threshold
+  txn_cf = txn_db->BeginTransaction(wopts, txn_opts, nullptr);
+  ASSERT_OK(txn_cf->SetName("xid_cf_at_threshold"));
+  for (uint32_t i = 0; i < threshold; ++i) {
+    ASSERT_OK(txn_cf->Put(handles_[i % 2], "key" + std::to_string(i),
+                          "value" + std::to_string(i)));
+  }
+  ASSERT_OK(txn_cf->Prepare());
+  commit_bypass_memtable = false;
+  ASSERT_OK(txn_cf->Commit());
+  ASSERT_TRUE(commit_bypass_memtable);
+  delete txn_cf;
+
+  // Test that commit_bypass_memtable takes precedence over
+  // large_txn_commit_optimize_threshold
+  txn_opts.large_txn_commit_optimize_threshold =
+      threshold * 10;                      // High threshold
+  txn_opts.commit_bypass_memtable = true;  // Should override threshold
+
+  txn_cf = txn_db->BeginTransaction(wopts, txn_opts, nullptr);
+  ASSERT_OK(txn_cf->SetName("xid_cf_precedence"));
+  ASSERT_OK(txn_cf->Put(handles_[0], "key1", "value1"));  // Just one operation
+  ASSERT_OK(txn_cf->Prepare());
+  commit_bypass_memtable = false;
+  ASSERT_OK(txn_cf->Commit());
+  ASSERT_TRUE(commit_bypass_memtable);  // Should be true because of
+                                        // commit_bypass_memtable
+  delete txn_cf;
+}
+
 TEST_P(CommitBypassMemtableTest, AtomicFlushTest) {
   const uint32_t threshold = 10;
   SetUpTransactionDB(/*threshold=*/threshold, /*atomic_flush=*/true);

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	+* Add new experimental `TransactionOptions::large_txn_commit_optimize_threshold` to enable optimizations for large transaction commit with per transaction threshold. `TransactionDBOptions::txn_commit_bypass_memtable_threshold` is deprecated in favor of this transaction option.