[#21832] docdb: Batch metrics updates for YCQL reads

Summary:
In 52a7dca, metric updates were batched for YSQL
reads to avoid performance degradation from contention updating rocksdb metrics.

This diff applies the same batching to YCQL reads.
Jira: DB-10734

Test Plan:
Jenkins.

Also confirmed that the following code path is no longer triggered by temporarily adding stack trace
tracking in yb::AtomicGauge<>::IncrementBy:

```
    @          0x2a253d9  yb::AtomicGauge<>::IncrementBy()
    @          0x33a882b  rocksdb::StatisticsMetricImpl::recordTick()
    @          0x32c73a5  rocksdb::DBIter::Next()
    ....
    @          0x34f41b6  yb::tablet::AbstractTablet::HandleQLReadRequest()
```

Reviewers: arybochkin

Reviewed By: arybochkin

Subscribers: rthallam, hsunder, ybase, yql, bogdan

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

src/yb/docdb/cql_operation.cc

@@ -1699,7 +1699,8 @@ Status QLReadOperation::Execute(const YQLStorageIf& ql_storage,
                                 const DocReadContext& doc_read_context,
                                 std::reference_wrapper<const ScopedRWOperation> pending_op,
                                 QLResultSet* resultset,
-                                HybridTime* restart_read_ht) {
+                                HybridTime* restart_read_ht,
+                                const docdb::DocDBStatistics* statistics) {
   auto se = ScopeExit([resultset] {
     resultset->Complete();
   });
@@ -1744,7 +1745,7 @@ Status QLReadOperation::Execute(const YQLStorageIf& ql_storage,
       &static_row_spec));
   RETURN_NOT_OK(ql_storage.GetIterator(
       request_, full_projection, doc_read_context, txn_op_context_, read_operation_data,
-      *spec, pending_op, &iter));
+      *spec, pending_op, &iter, statistics));
   VTRACE(1, "Initialized iterator");
 
   QLTableRow static_row;
@@ -1759,7 +1760,7 @@ Status QLReadOperation::Execute(const YQLStorageIf& ql_storage,
     std::unique_ptr<YQLRowwiseIteratorIf> static_row_iter;
     RETURN_NOT_OK(ql_storage.GetIterator(
         request_, static_projection, doc_read_context, txn_op_context_, read_operation_data,
-        *static_row_spec, pending_op, &static_row_iter));
+        *static_row_spec, pending_op, &static_row_iter, statistics));
     RETURN_NOT_OK(static_row_iter->FetchNext(&static_row));
   }
 

src/yb/docdb/cql_operation.h

@@ -19,6 +19,7 @@
 #include "yb/docdb/doc_expr.h"
 #include "yb/dockv/doc_key.h"
 #include "yb/docdb/doc_operation.h"
+#include "yb/docdb/docdb_statistics.h"
 #include "yb/docdb/intent_aware_iterator.h"
 
 #include "yb/util/operation_counter.h"
@@ -226,7 +227,8 @@ class QLReadOperation : public DocExprExecutor {
                  const DocReadContext& doc_read_context,
                  std::reference_wrapper<const ScopedRWOperation> pending_op,
                  qlexpr::QLResultSet* result_set,
-                 HybridTime* restart_read_ht);
+                 HybridTime* restart_read_ht,
+                 const docdb::DocDBStatistics* statistics);
 
   Status PopulateResultSet(const std::unique_ptr<qlexpr::QLScanSpec>& spec,
                            const qlexpr::QLTableRow& table_row,

src/yb/docdb/doc_operation-test.cc

@@ -356,7 +356,7 @@ SubDocKey(DocKey(0x0000, [1], []), [ColumnId(3); HT{ <max> w: 2 }]) -> 4
     auto pending_op = ScopedRWOperation::TEST_Create();
     EXPECT_OK(read_op.Execute(
         ql_storage, ReadOperationData::FromSingleReadTime(read_time), doc_read_context, pending_op,
-        &resultset, &read_restart_ht));
+        &resultset, &read_restart_ht, nullptr /* statistics */));
     EXPECT_FALSE(read_restart_ht.is_valid());
 
     // Transfer the column values from result set to rowblock.

src/yb/docdb/ql_rocksdb_storage.cc

@@ -48,9 +48,11 @@ Status QLRocksDBStorage::GetIterator(
     const ReadOperationData& read_operation_data,
     const qlexpr::QLScanSpec& spec,
     std::reference_wrapper<const ScopedRWOperation> pending_op,
-    std::unique_ptr<YQLRowwiseIteratorIf> *iter) const {
+    std::unique_ptr<YQLRowwiseIteratorIf> *iter,
+    const docdb::DocDBStatistics* statistics) const {
   auto doc_iter = std::make_unique<DocRowwiseIterator>(
-      projection, doc_read_context, txn_op_context, doc_db_, read_operation_data, pending_op);
+      projection, doc_read_context, txn_op_context, doc_db_, read_operation_data, pending_op,
+      statistics);
   RETURN_NOT_OK(doc_iter->Init(spec));
   *iter = std::move(doc_iter);
   return Status::OK();

src/yb/docdb/ql_rocksdb_storage.h

@@ -40,7 +40,8 @@ class QLRocksDBStorage : public YQLStorageIf {
       const ReadOperationData& read_operation_data,
       const qlexpr::QLScanSpec& spec,
       std::reference_wrapper<const ScopedRWOperation> pending_op,
-      std::unique_ptr<YQLRowwiseIteratorIf> *iter) const override;
+      std::unique_ptr<YQLRowwiseIteratorIf> *iter,
+      const docdb::DocDBStatistics* statistics = nullptr) const override;
 
   Status BuildYQLScanSpec(
       const QLReadRequestPB& request,

src/yb/docdb/ql_storage_interface.h

@@ -47,7 +47,8 @@ class YQLStorageIf {
       const ReadOperationData& read_operation_data,
       const qlexpr::QLScanSpec& spec,
       std::reference_wrapper<const ScopedRWOperation> pending_op,
-      std::unique_ptr<YQLRowwiseIteratorIf>* iter) const = 0;
+      std::unique_ptr<YQLRowwiseIteratorIf>* iter,
+      const DocDBStatistics* statistics = nullptr) const = 0;
 
   virtual Status BuildYQLScanSpec(
       const QLReadRequestPB& request,

src/yb/master/system_tablet.cc

@@ -73,7 +73,7 @@ Status SystemTablet::HandleQLReadRequest(const docdb::ReadOperationData& read_op
   auto pending_op = ScopedRWOperation();
   return AbstractTablet::HandleQLReadRequest(
       read_operation_data, ql_read_request, TransactionOperationContext(), YQLTable(), pending_op,
-      result, rows_data);
+      result, rows_data, nullptr /* statistics */);
 }
 
 Status SystemTablet::CreatePagingStateForRead(const QLReadRequestPB& ql_read_request,

src/yb/master/yql_virtual_table.cc

@@ -63,7 +63,8 @@ Status YQLVirtualTable::GetIterator(
     const docdb::ReadOperationData& read_operation_data,
     const qlexpr::QLScanSpec& spec,
     std::reference_wrapper<const ScopedRWOperation> pending_op,
-    std::unique_ptr<docdb::YQLRowwiseIteratorIf>* iter) const {
+    std::unique_ptr<docdb::YQLRowwiseIteratorIf>* iter,
+    const docdb::DocDBStatistics* statistics) const {
   // Acquire shared lock on catalog manager to verify it is still the leader and metadata will
   // not change.
   SCOPED_LEADER_SHARED_LOCK(l, master_->catalog_manager_impl());

src/yb/master/yql_virtual_table.h

@@ -57,7 +57,8 @@ class YQLVirtualTable : public docdb::YQLStorageIf {
       const docdb::ReadOperationData& read_operation_data,
       const qlexpr::QLScanSpec& spec,
       std::reference_wrapper<const ScopedRWOperation> pending_op,
-      std::unique_ptr<docdb::YQLRowwiseIteratorIf>* iter) const override;
+      std::unique_ptr<docdb::YQLRowwiseIteratorIf>* iter,
+      const docdb::DocDBStatistics* statistics = nullptr) const override;
 
   Status BuildYQLScanSpec(
       const QLReadRequestPB& request,

src/yb/tablet/abstract_tablet.cc

@@ -42,7 +42,8 @@ Status AbstractTablet::HandleQLReadRequest(
     const docdb::YQLStorageIf& ql_storage,
     std::reference_wrapper<const ScopedRWOperation> pending_op,
     QLReadRequestResult* result,
-    WriteBuffer* rows_data) {
+    WriteBuffer* rows_data,
+    const docdb::DocDBStatistics* statistics) {
   // TODO(Robert): verify that all key column values are provided
   docdb::QLReadOperation doc_op(ql_read_request, txn_op_context);
 
@@ -55,7 +56,7 @@ Status AbstractTablet::HandleQLReadRequest(
   TRACE("Start Execute");
   const Status s = doc_op.Execute(
       ql_storage, read_operation_data, *doc_read_context, pending_op, &resultset,
-      &result->restart_read_ht);
+      &result->restart_read_ht, statistics);
   TRACE("Done Execute");
   if (!s.ok()) {
     if (s.IsQLError()) {

src/yb/tablet/abstract_tablet.h

@@ -70,7 +70,8 @@ class AbstractTablet {
       const docdb::YQLStorageIf& ql_storage,
       std::reference_wrapper<const ScopedRWOperation> pending_op,
       QLReadRequestResult* result,
-      WriteBuffer* rows_data);
+      WriteBuffer* rows_data,
+      const docdb::DocDBStatistics* statistics);
 
   virtual Status CreatePagingStateForRead(const QLReadRequestPB& ql_read_request,
                                                   const size_t row_count,

src/yb/tablet/tablet.cc

@@ -336,18 +336,96 @@ bool IsSortedAscendingEncoded(
 
 namespace {
 
-thread_local docdb::DocDBStatistics scoped_docdb_statistics;
-thread_local ScopedTabletMetrics scoped_tablet_metrics;
+class MetricsScope {
+ public:
+  MetricsScope(TabletMetrics& global_metrics, rocksdb::Statistics& regulardb_statistics,
+               rocksdb::Statistics& intentsdb_statistics) :
+      MetricsScope(
+          global_metrics, regulardb_statistics, intentsdb_statistics,
+          GetAtomicFlag(&FLAGS_batch_tablet_metrics_update)) { }
+
+  ~MetricsScope() {
+    if (!IsBatchedMetricsUpdate()) {
+      return;
+    }
+
+    if (GetAtomicFlag(&FLAGS_dump_metrics_to_trace)) {
+      TraceScopedMetrics();
+    }
 
-void TraceScopedMetrics() {
-  std::stringstream ss;
-  ss << "Metric changes:\n";
-  size_t changes = scoped_docdb_statistics.Dump(&ss);
-  changes += scoped_tablet_metrics.Dump(&ss);
-  if (changes > 0) {
-    TRACE(ss.str());
+    scoped_tablet_metrics_.MergeAndClear(&global_metrics_);
+    scoped_docdb_statistics_.MergeAndClear(&regulardb_statistics_, &intentsdb_statistics_);
   }
-}
+
+  docdb::DocDBStatistics* statistics() {
+    return statistics_;
+  }
+
+  TabletMetrics* metrics() {
+    return &metrics_;
+  }
+
+ protected:
+  static thread_local docdb::DocDBStatistics scoped_docdb_statistics_;
+  static thread_local ScopedTabletMetrics scoped_tablet_metrics_;
+
+  bool IsBatchedMetricsUpdate() {
+    return statistics_ != nullptr;
+  }
+
+ private:
+  MetricsScope(TabletMetrics& global_metrics, rocksdb::Statistics& regulardb_statistics,
+               rocksdb::Statistics& intentsdb_statistics, bool batch_metrics_update) :
+      global_metrics_(global_metrics), regulardb_statistics_(regulardb_statistics),
+      intentsdb_statistics_(intentsdb_statistics),
+      statistics_(batch_metrics_update ? &scoped_docdb_statistics_ : nullptr),
+      metrics_(batch_metrics_update ? scoped_tablet_metrics_ : global_metrics_) { }
+
+  void TraceScopedMetrics() {
+    std::stringstream ss;
+    ss << "Metric changes:\n";
+    size_t changes = scoped_docdb_statistics_.Dump(&ss);
+    changes += scoped_tablet_metrics_.Dump(&ss);
+    if (changes > 0) {
+      TRACE(ss.str());
+    }
+  }
+
+  TabletMetrics& global_metrics_;
+  rocksdb::Statistics& regulardb_statistics_;
+  rocksdb::Statistics& intentsdb_statistics_;
+
+  docdb::DocDBStatistics* statistics_;
+  TabletMetrics& metrics_;
+};
+
+class YSQLMetricsScope : public MetricsScope {
+ public:
+  YSQLMetricsScope(TabletMetrics& global_metrics, rocksdb::Statistics& regulardb_statistics,
+                   rocksdb::Statistics& intentsdb_statistics,
+                   PgsqlMetricsCaptureType metrics_capture, PgsqlResponsePB& pgsql_response):
+      MetricsScope(global_metrics, regulardb_statistics, intentsdb_statistics),
+      metrics_capture_(metrics_capture), pgsql_response_(pgsql_response) { }
+
+  ~YSQLMetricsScope() {
+    if (!IsBatchedMetricsUpdate()) {
+      return;
+    }
+
+    if (GetAtomicFlag(&FLAGS_ysql_analyze_dump_metrics) &&
+        metrics_capture_ == PgsqlMetricsCaptureType::PGSQL_METRICS_CAPTURE_ALL) {
+      scoped_tablet_metrics_.CopyToPgsqlResponse(&pgsql_response_);
+      scoped_docdb_statistics_.CopyToPgsqlResponse(&pgsql_response_);
+    }
+  }
+
+ private:
+  PgsqlMetricsCaptureType metrics_capture_;
+  PgsqlResponsePB& pgsql_response_;
+};
+
+thread_local docdb::DocDBStatistics MetricsScope::scoped_docdb_statistics_;
+thread_local ScopedTabletMetrics MetricsScope::scoped_tablet_metrics_;
 
 std::string MakeTabletLogPrefix(
     const TabletId& tablet_id, const std::string& log_prefix_suffix) {
@@ -915,7 +993,6 @@ Status Tablet::OpenKeyValueTablet() {
     intents_db_->ListenFilesChanged(std::bind(&Tablet::CleanupIntentFiles, this));
   }
 
-  ql_storage_.reset(new docdb::QLRocksDBStorage(doc_db()));
   if (transaction_participant_) {
     // We need to set the "cdc_sdk_min_checkpoint_op_id" so that intents don't get
     // garbage collected after transactions are loaded.
@@ -1590,8 +1667,13 @@ Status Tablet::HandleQLReadRequest(
   auto scoped_read_operation = CreateScopedRWOperationNotBlockingRocksDbShutdownStart(
       read_operation_data.deadline);
   RETURN_NOT_OK(scoped_read_operation);
+
+  MetricsScope metrics_scope(*metrics_, *regulardb_statistics_, *intentsdb_statistics_);
+
   ScopedTabletMetricsLatencyTracker metrics_tracker(
-      metrics_.get(), TabletEventStats::kQlReadLatency);
+      metrics_scope.metrics(), TabletEventStats::kQlReadLatency);
+
+  docdb::QLRocksDBStorage storage{doc_db(metrics_scope.metrics())};
 
   bool schema_version_compatible = IsSchemaVersionCompatible(
       metadata()->schema_version(), ql_read_request.schema_version(),
@@ -1603,8 +1685,8 @@ Status Tablet::HandleQLReadRequest(
         CreateTransactionOperationContext(transaction_metadata, /* is_ysql_catalog_table */ false);
     RETURN_NOT_OK(txn_op_ctx);
     status = AbstractTablet::HandleQLReadRequest(
-        read_operation_data, ql_read_request, *txn_op_ctx, *ql_storage_, scoped_read_operation,
-        result, rows_data);
+        read_operation_data, ql_read_request, *txn_op_ctx, storage, scoped_read_operation,
+        result, rows_data, metrics_scope.statistics());
 
     schema_version_compatible = IsSchemaVersionCompatible(
         metadata()->schema_version(), ql_read_request.schema_version(),
@@ -1687,33 +1769,15 @@ Status Tablet::HandlePgsqlReadRequest(
       read_operation_data.deadline);
   RETURN_NOT_OK(scoped_read_operation);
 
-  docdb::DocDBStatistics* statistics = nullptr;
-  TabletMetrics* metrics = metrics_.get();
-
-  if (GetAtomicFlag(&FLAGS_batch_tablet_metrics_update)) {
-    statistics = &scoped_docdb_statistics;
-    metrics = &scoped_tablet_metrics;
-  }
+  YSQLMetricsScope metrics_scope(
+      *metrics_, *regulardb_statistics_, *intentsdb_statistics_,
+      pgsql_read_request.metrics_capture(), result->response);
 
   auto status = DoHandlePgsqlReadRequest(
-      &scoped_read_operation, statistics, metrics, read_operation_data,
-      is_explicit_request_read_time, pgsql_read_request, transaction_metadata,
+      &scoped_read_operation, metrics_scope.statistics(), metrics_scope.metrics(),
+      read_operation_data, is_explicit_request_read_time, pgsql_read_request, transaction_metadata,
       subtransaction_metadata, result);
 
-  if (statistics) {
-    if (GetAtomicFlag(&FLAGS_dump_metrics_to_trace)) {
-      TraceScopedMetrics();
-    }
-    auto metrics_capture = pgsql_read_request.metrics_capture();
-    if (GetAtomicFlag(&FLAGS_ysql_analyze_dump_metrics) &&
-        metrics_capture == PgsqlMetricsCaptureType::PGSQL_METRICS_CAPTURE_ALL) {
-      statistics->CopyToPgsqlResponse(&result->response);
-      scoped_tablet_metrics.CopyToPgsqlResponse(&result->response);
-    }
-    scoped_tablet_metrics.MergeAndClear(metrics_.get());
-    statistics->MergeAndClear(regulardb_statistics_.get(), intentsdb_statistics_.get());
-  }
-
   return status;
 }
 

src/yb/tablet/tablet.h

@@ -1126,8 +1126,6 @@ class Tablet : public AbstractTablet,
   // Optional key bounds (see docdb::KeyBounds) served by this tablet.
   docdb::KeyBounds key_bounds_;
 
-  std::unique_ptr<docdb::YQLStorageIf> ql_storage_;
-
   // This is for docdb fine-grained locking.
   docdb::SharedLockManager shared_lock_manager_;