Set default cf ts sz for a reused transaction (#11685)

Summary:
Set up the default column family timestamp size for a reused write committed transaction.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/11685

Test Plan: Added unit test.

Reviewed By: ltamasi

Differential Revision: D48195129

Pulled By: jowlyzhang

fbshipit-source-id: 54faa900c123fc6daa412c01490e36c10a24a678

db/write_batch.cc

@@ -746,6 +746,11 @@ size_t WriteBatchInternal::GetFirstOffset(WriteBatch* /*b*/) {
   return WriteBatchInternal::kHeader;
 }
 
+void WriteBatchInternal::SetDefaultColumnFamilyTimestampSize(
+    WriteBatch* wb, size_t default_cf_ts_sz) {
+  wb->default_cf_ts_sz_ = default_cf_ts_sz;
+}
+
 std::tuple<Status, uint32_t, size_t>
 WriteBatchInternal::GetColumnFamilyIdAndTimestampSize(
     WriteBatch* b, ColumnFamilyHandle* column_family) {

db/write_batch_internal.h

@@ -224,6 +224,9 @@ class WriteBatchInternal {
   static void SetAsLatestPersistentState(WriteBatch* b);
   static bool IsLatestPersistentState(const WriteBatch* b);
 
+  static void SetDefaultColumnFamilyTimestampSize(WriteBatch* wb,
+                                                  size_t default_cf_ts_sz);
+
   static std::tuple<Status, uint32_t, size_t> GetColumnFamilyIdAndTimestampSize(
       WriteBatch* b, ColumnFamilyHandle* column_family);
 

utilities/transactions/transaction_base.cc

@@ -110,6 +110,8 @@ void TransactionBaseImpl::Reinitialize(DB* db,
   start_time_ = dbimpl_->GetSystemClock()->NowMicros();
   indexing_enabled_ = true;
   cmp_ = GetColumnFamilyUserComparator(db_->DefaultColumnFamily());
+  WriteBatchInternal::SetDefaultColumnFamilyTimestampSize(
+      write_batch_.GetWriteBatch(), cmp_->timestamp_size());
   WriteBatchInternal::UpdateProtectionInfo(
       write_batch_.GetWriteBatch(), write_options_.protection_bytes_per_key)
       .PermitUncheckedError();

utilities/transactions/write_committed_transaction_ts_test.cc

@@ -564,6 +564,13 @@ TEST_P(WriteCommittedTxnWithTsTest, CheckKeysForConflicts) {
   ASSERT_TRUE(txn1->GetForUpdate(ReadOptions(), "foo", &dontcare).IsBusy());
   ASSERT_TRUE(called);
 
+  Transaction* reused_txn =
+      db->BeginTransaction(WriteOptions(), TransactionOptions(), txn1.get());
+  ASSERT_EQ(reused_txn, txn1.get());
+  ASSERT_OK(reused_txn->Put("foo", "v1"));
+  ASSERT_OK(reused_txn->SetCommitTimestamp(40));
+  ASSERT_OK(reused_txn->Commit());
+
   SyncPoint::GetInstance()->DisableProcessing();
   SyncPoint::GetInstance()->ClearAllCallBacks();
 }