[#26666] docdb: Skip tracking readonly transactions in recently_applied_ map

Summary:
We currently track all recently applied transactions in the recently_applied_ map in
transaction participant in order to calculate the min_replay_txn_start_ht used to filter
transactions during transaction loading in tablet bootstrap.

It is posible to have a table where the only transactions performed on it are readonly, i.e.
no WRITE_OPs with write pairs. In this case, since we do not replicate any write pairs, the
retryable requests max_replicated_op_id is never updated. The op id we use to filter the
recently_applied_ map with to clean it up is at most max_replicated_op_id, so we end up never
removing entries from the map, and the size of this map grows indefinitely.

This diff changes the behavior to never add transactions that have no batches with any write pairs
to recently_applied_, to avoid this issue.

Memory consumed by the recently_applied_ map was also added to mem tracker.
Jira: DB-16044

Test Plan:
Added tests:
`./yb_build.sh --cxx-test pgwrapper_pg_mini-test --gtest_filter PgMiniTestSingleNode.TestAppliedTransactionsStateReadOnly`
`./yb_build.sh --cxx-test pgwrapper_pg_mini-test --gtest_filter PgMiniTest.TestAppliedTransactionsStateInFlight`

Reviewers: sergei, bkolagani

Reviewed By: sergei, bkolagani

Subscribers: bkolagani, rthallam, ybase, yql

Differential Revision: https://phorge.dev.yugabyte.com/D43077

Author: es1024

src/yb/tablet/running_transaction.cc

@@ -691,5 +691,13 @@ bool RunningTransaction::IsTxnLoadedWithCDC() const {
   return is_txn_loaded_with_cdc_;
 }
 
+void RunningTransaction::MarkHasRetryableRequestsReplicated() {
+  has_retryable_requests_replicated_ = true;
+}
+
+bool RunningTransaction::HasRetryableRequestsReplicated() const {
+  return has_retryable_requests_replicated_;
+}
+
 } // namespace tablet
 } // namespace yb

src/yb/tablet/running_transaction.h

@@ -154,6 +154,10 @@ class RunningTransaction : public std::enable_shared_from_this<RunningTransactio
 
   bool IsTxnLoadedWithCDC() const;
 
+  void MarkHasRetryableRequestsReplicated();
+
+  bool HasRetryableRequestsReplicated() const;
+
  private:
   static boost::optional<TransactionStatus> GetStatusAt(
       HybridTime time,
@@ -237,6 +241,9 @@ class RunningTransaction : public std::enable_shared_from_this<RunningTransactio
   // Identification marker for transactions that are loaded on tablet bootstrap with CDC
   // enbled.
   bool is_txn_loaded_with_cdc_ = false;
+
+  // Whether or not transaction has any batches replicated by retryable requests.
+  bool has_retryable_requests_replicated_ = false;
 };
 
 Status MakeAbortedStatus(const TransactionId& id);

src/yb/tablet/tablet.cc

@@ -1707,7 +1707,8 @@ Status Tablet::WriteTransactionalBatch(
   }
   boost::container::small_vector<uint8_t, 16> encoded_replicated_batch_idx_set;
   auto prepare_batch_data = VERIFY_RESULT(transaction_participant()->PrepareBatchData(
-      transaction_id, batch_idx, &encoded_replicated_batch_idx_set));
+      transaction_id, batch_idx, &encoded_replicated_batch_idx_set,
+      !put_batch.write_pairs().empty()));
   if (!prepare_batch_data) {
     // If metadata is missing it could be caused by aborted and removed transaction.
     // In this case we should not add new intents for it.

src/yb/tablet/transaction_participant.cc

@@ -462,7 +462,8 @@ class TransactionParticipant::Impl
 
   Result<boost::optional<std::pair<IsolationLevel, TransactionalBatchData>>> PrepareBatchData(
       const TransactionId& id, size_t batch_idx,
-      boost::container::small_vector_base<uint8_t>* encoded_replicated_batches) {
+      boost::container::small_vector_base<uint8_t>* encoded_replicated_batches,
+      bool has_write_pairs) {
     // We are not trying to cleanup intents here because we don't know whether this transaction
     // has intents of not.
     auto lock_and_iterator = VERIFY_RESULT(LockAndFind(
@@ -475,6 +476,9 @@ class TransactionParticipant::Impl
     }
     auto& transaction = lock_and_iterator.transaction();
     transaction.AddReplicatedBatch(batch_idx, encoded_replicated_batches);
+    if (has_write_pairs) {
+      transaction.MarkHasRetryableRequestsReplicated();
+    }
     return std::make_pair(transaction.metadata().isolation, transaction.last_batch_data());
   }
 
@@ -1957,6 +1961,9 @@ class TransactionParticipant::Impl
     if (GetLatestCheckPointUnlocked() != OpId::Max()) {
       txn->SetTxnLoadedWithCDC();
     }
+    if (last_batch_data.hybrid_time) {
+      txn->MarkHasRetryableRequestsReplicated();
+    }
     transactions_.insert(txn);
     mem_tracker_->Consume(kRunningTransactionSize);
     TransactionsModifiedUnlocked(&min_running_notifier);
@@ -1991,7 +1998,13 @@ class TransactionParticipant::Impl
     recently_removed_transactions_cleanup_queue_.push_back({transaction.id(), now + 15s});
     LOG_IF_WITH_PREFIX(DFATAL, !recently_removed_transactions_.insert(transaction.id()).second)
         << "Transaction removed twice: " << transaction.id();
-    AddRecentlyAppliedTransaction(transaction.start_ht(), transaction.GetApplyOpId());
+    if (transaction.HasRetryableRequestsReplicated()) {
+      AddRecentlyAppliedTransaction(transaction.start_ht(), transaction.GetApplyOpId());
+    } else {
+      VLOG_WITH_PREFIX(2)
+          << "Transaction " << transaction.id() << " has no write pairs, not adding to recently "
+          << "applied transactions map";
+    }
     transactions_.erase(it);
     mem_tracker_->Release(kRunningTransactionSize);
     TransactionsModifiedUnlocked(min_running_notifier);
@@ -2581,8 +2594,9 @@ Result<TransactionMetadata> TransactionParticipant::PrepareMetadata(
 Result<boost::optional<std::pair<IsolationLevel, TransactionalBatchData>>>
     TransactionParticipant::PrepareBatchData(
     const TransactionId& id, size_t batch_idx,
-    boost::container::small_vector_base<uint8_t>* encoded_replicated_batches) {
-  return impl_->PrepareBatchData(id, batch_idx, encoded_replicated_batches);
+    boost::container::small_vector_base<uint8_t>* encoded_replicated_batches,
+    bool has_write_pairs) {
+  return impl_->PrepareBatchData(id, batch_idx, encoded_replicated_batches, has_write_pairs);
 }
 
 void TransactionParticipant::BatchReplicated(

src/yb/tablet/transaction_participant.h

@@ -139,7 +139,8 @@ class TransactionParticipant : public TransactionStatusManager {
   // Returns boost::none when transaction is unknown.
   Result<boost::optional<std::pair<IsolationLevel, TransactionalBatchData>>> PrepareBatchData(
       const TransactionId& id, size_t batch_idx,
-      boost::container::small_vector_base<uint8_t>* encoded_replicated_batches);
+      boost::container::small_vector_base<uint8_t>* encoded_replicated_batches,
+      bool has_write_pairs);
 
   void BatchReplicated(const TransactionId& id, const TransactionalBatchData& data);
 

src/yb/yql/pgwrapper/pg_mini-test.cc

@@ -134,6 +134,7 @@ METRIC_DECLARE_entity(tablet);
 METRIC_DECLARE_gauge_uint64(aborted_transactions_pending_cleanup);
 METRIC_DECLARE_histogram(handler_latency_outbound_transfer);
 METRIC_DECLARE_gauge_int64(rpc_busy_reactors);
+METRIC_DECLARE_gauge_uint64(wal_replayable_applied_transactions);
 
 namespace yb::pgwrapper {
 namespace {
@@ -2552,6 +2553,151 @@ TEST_F(PgMiniTestSingleNode, TestBootstrapOnAppliedTransactionWithIntents) {
   ASSERT_EQ(res, 1);
 }
 
+TEST_F(PgMiniTestSingleNode, TestAppliedTransactionsStateReadOnly) {
+  constexpr size_t kIters = 100;
+
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_delete_intents_sst_files) = false;
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_use_bootstrap_intent_ht_filter) = true;
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_TEST_no_schedule_remove_intents) = true;
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_TEST_disable_flush_on_shutdown) = true;
+
+  auto conn = ASSERT_RESULT(Connect());
+
+  LOG(INFO) << "Creating tables";
+  ASSERT_OK(conn.Execute("CREATE TABLE test1(a int primary key) SPLIT INTO 1 TABLETS"));
+
+  tablet::TabletPeerPtr test1_peer = nullptr;
+  {
+    const auto& peers = ListTabletPeers(cluster_.get(), ListPeersFilter::kLeaders);
+    for (auto peer : peers) {
+      if (peer->shared_tablet()->regular_db()) {
+        test1_peer = peer;
+        break;
+      }
+    }
+    ASSERT_NE(test1_peer, nullptr);
+  }
+  std::string test1_tablet_id = test1_peer->shared_tablet()->tablet_id();
+
+  ASSERT_OK(conn.StartTransaction(IsolationLevel::SNAPSHOT_ISOLATION));
+  ASSERT_OK(conn.Execute("INSERT INTO test1(a) VALUES (0)"));
+  ASSERT_OK(conn.CommitTransaction());
+
+  ASSERT_OK(test1_peer->shared_tablet()->Flush(tablet::FlushMode::kSync));
+  ASSERT_OK(test1_peer->FlushBootstrapState());
+
+  ASSERT_OK(conn.Execute("CREATE TABLE test2(a int references test1(a)) SPLIT INTO 1 TABLETS"));
+  for (size_t i = 0; i < kIters; ++i) {
+    ASSERT_OK(conn.StartTransaction(IsolationLevel::SNAPSHOT_ISOLATION));
+    ASSERT_OK(conn.Execute("INSERT INTO test2(a) VALUES (0)"));
+    ASSERT_OK(conn.CommitTransaction());
+  }
+
+  ASSERT_EQ(
+      GetMetricOpt<AtomicGauge<uint64_t>>(
+          *test1_peer->shared_tablet(), METRIC_wal_replayable_applied_transactions)->value(),
+      1);
+
+  LOG(INFO) << "Restarting cluster";
+  ASSERT_OK(RestartCluster());
+
+  for (auto peer : ListTabletPeers(cluster_.get(), ListPeersFilter::kAll)) {
+    if (!peer->shared_tablet()->regular_db()) {
+      continue;
+    }
+    auto metric_value = GetMetricOpt<AtomicGauge<uint64_t>>(
+        *peer->shared_tablet(), METRIC_wal_replayable_applied_transactions)->value();
+    if (peer->shared_tablet()->tablet_id() == test1_tablet_id) {
+      ASSERT_EQ(metric_value, 1);
+    } else {
+      ASSERT_EQ(metric_value, kIters);
+    }
+  }
+
+  conn = ASSERT_RESULT(Connect());
+  auto res = ASSERT_RESULT(conn.FetchRow<PGUint64>("SELECT COUNT(*) FROM test1"));
+  ASSERT_EQ(res, 1);
+  res = ASSERT_RESULT(conn.FetchRow<PGUint64>("SELECT COUNT(*) FROM test2"));
+  ASSERT_EQ(res, kIters);
+}
+
+TEST_F(PgMiniTest, TestAppliedTransactionsStateInFlight) {
+  const auto kApplyWait = 5s * kTimeMultiplier;
+
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_delete_intents_sst_files) = false;
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_use_bootstrap_intent_ht_filter) = true;
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_TEST_no_schedule_remove_intents) = true;
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_TEST_disable_flush_on_shutdown) = true;
+
+  auto conn1 = ASSERT_RESULT(Connect());
+  auto conn2 = ASSERT_RESULT(Connect());
+  auto conn3 = ASSERT_RESULT(Connect());
+
+  LOG(INFO) << "Creating table";
+  ASSERT_OK(conn1.Execute("CREATE TABLE test(a int) SPLIT INTO 1 TABLETS"));
+
+  const auto& pg_ts_uuid = cluster_->mini_tablet_server(kPgTsIndex)->server()->permanent_uuid();
+  tablet::TabletPeerPtr tablet_peer = nullptr;
+  for (auto peer : ListTabletPeers(cluster_.get(), ListPeersFilter::kNonLeaders)) {
+    if (peer->shared_tablet()->regular_db() && peer->permanent_uuid() != pg_ts_uuid) {
+      tablet_peer = peer;
+      break;
+    }
+  }
+  ASSERT_NE(tablet_peer, nullptr);
+
+  tserver::MiniTabletServer* tablet_server = nullptr;
+  for (size_t i = 0; i < NumTabletServers(); ++i) {
+    auto* ts = cluster_->mini_tablet_server(i);
+    if (ts->server()->permanent_uuid() == tablet_peer->permanent_uuid()) {
+      tablet_server = ts;
+      break;
+    }
+  }
+  ASSERT_NE(tablet_server, nullptr);
+
+  ASSERT_OK(conn1.StartTransaction(IsolationLevel::SNAPSHOT_ISOLATION));
+  ASSERT_OK(conn1.Execute("INSERT INTO test(a) VALUES (0)"));
+  ASSERT_OK(conn1.CommitTransaction());
+
+  ASSERT_OK(conn1.StartTransaction(IsolationLevel::SNAPSHOT_ISOLATION));
+  ASSERT_OK(conn1.Execute("INSERT INTO test(a) VALUES (1)"));
+
+  ASSERT_OK(conn2.StartTransaction(IsolationLevel::SNAPSHOT_ISOLATION));
+  ASSERT_OK(conn2.Execute("INSERT INTO test(a) VALUES (2)"));
+  ASSERT_OK(conn2.FetchRow<PGUint32>("SELECT a FROM test WHERE a = 0 FOR KEY SHARE"));
+
+  ASSERT_OK(conn3.StartTransaction(IsolationLevel::SNAPSHOT_ISOLATION));
+  ASSERT_OK(conn3.FetchRow<PGUint32>("SELECT a FROM test WHERE a = 0 FOR KEY SHARE"));
+
+  ASSERT_OK(tablet_peer->shared_tablet()->Flush(tablet::FlushMode::kSync));
+
+  ASSERT_OK(tablet_server->Restart());
+  ASSERT_OK(tablet_server->WaitStarted());
+
+  ASSERT_OK(conn1.CommitTransaction());
+  ASSERT_OK(conn2.CommitTransaction());
+  ASSERT_OK(conn3.CommitTransaction());
+
+  // Wait for apply.
+  SleepFor(kApplyWait);
+
+  std::unordered_map<std::string, uint64_t> metric_values;
+  for (auto peer : ListTabletPeers(cluster_.get(), ListPeersFilter::kAll)) {
+    if (peer->shared_tablet()->regular_db()) {
+      metric_values[peer->permanent_uuid()] = GetMetricOpt<AtomicGauge<uint64_t>>(
+          *peer->shared_tablet(), METRIC_wal_replayable_applied_transactions)->value();
+    }
+  }
+
+  // Expecting metric value to be 3 on all peers: the intial insert + conn1/conn2 transactions.
+  // conn3 transaction is readonly and not added to map.
+  LOG(INFO) << "Metric values: " << CollectionToString(metric_values);
+  for (const auto& [_, value] : metric_values) {
+    ASSERT_EQ(value, 3);
+  }
+}
+
 Status MockAbortFailure(
     const yb::tserver::PgFinishTransactionRequestPB* req,
     yb::tserver::PgFinishTransactionResponsePB* resp, yb::rpc::RpcContext* context) {