[#22328] CDCSDK: Delete memory context after each GetChanges call

Summary:
This diff ensures that the GetChanges RPC, upon its ScopeExit, will ensure deletion of its MemoryContext to prevent leaks caused by other RPCs setting their own MemoryContext without resetting it back to the old value.

CDC uses Memory Context to allocate objects that are decoded from binary format. These memory contexts are thread-local objects. After each decode operation, this context is reset but not deleted to reduce allocation/deallocation.

This implies that each thread processing the GetChanges RPC would have one memory context saved in its thread-local storage. This is fine as long as some other RPCs don't reset this thread-local object without either deleting it or resetting it back to the old value. It seems there are other RPCs that could get executed by the same thread and they allocate a new memory context and set it to the same thread-local, leading to this leak.

As a good practice, each RPC/task that gets executed by a thread should do the following:

```
Get the old memory context
Create and Set the new memory context
Delete its memory context
Reset the old on completion of its task
```

As a safety precaution, we should also look for other instances where we are not setting the old memory context back to the thread_local.
Jira: DB-11236

Test Plan: Manual Test

Reviewers: stiwary, amartsinchyk, sergei

Reviewed By: amartsinchyk

Subscribers: ybase, ycdcxcluster

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

src/yb/cdc/cdcsdk_producer.cc

@@ -46,6 +46,7 @@
 #include "yb/tablet/transaction_participant.h"
 
 #include "yb/util/logging.h"
+#include "yb/util/scope_exit.h"
 #include "yb/util/status.h"
 #include "yb/util/status_format.h"
 
@@ -2421,6 +2422,9 @@ Status GetChangesForCDCSDK(
     int64_t* last_readable_opid_index,
     const TableId& colocated_table_id,
     const CoarseTimePoint deadline) {
+  // Delete the memory context if it was created for decoding the QLValuePB.
+  auto scope_exit = ScopeExit([&] { docdb::DeleteMemoryContextForCDCWrapper(); });
+
   OpId op_id{from_op_id.term(), from_op_id.index()};
   VLOG(1) << "GetChanges request has from_op_id: " << from_op_id.DebugString()
           << ", safe_hybrid_time: " << safe_hybrid_time_req

src/yb/docdb/doc_pg_expr.cc

@@ -116,8 +116,9 @@ class TSCallExecutor {
 
   ~TSCallExecutor() {
     // If row_ctx_ was created it is deleted with mem_ctx_, as mem_ctx_ is the parent
-    YbgSetCurrentMemoryContext(mem_ctx_);
+    YbgMemoryContext old_ctx = YbgSetCurrentMemoryContext(mem_ctx_);
     CHECK_OK(DeleteMemoryContext());
+    YbgSetCurrentMemoryContext(old_ctx);
   }
 
   Status AddColumnRef(const PgsqlColRefPB& column_ref) {

src/yb/docdb/docdb_pgapi.cc

@@ -230,6 +230,12 @@ Status SetValueFromQLBinary(
   return Status::OK();
 }
 
+void DeleteMemoryContextIfSet() {
+  if (YBCPgGetThreadLocalCurrentMemoryContext() != nullptr) {
+    YbgDeleteMemoryContext();
+  }
+}
+
 namespace {
 
 // Given a 'ql_value', interpret the binary value in it as an array of type

src/yb/docdb/docdb_pgapi.h

@@ -113,4 +113,6 @@ Status SetValueFromQLBinaryHelper(
     const std::unordered_map<uint32_t, std::vector<master::PgAttributePB>> &composite_atts_map,
     DatumMessagePB *cdc_datum_message = NULL);
 
+void DeleteMemoryContextIfSet();
+
 } // namespace yb::docdb

src/yb/docdb/docdb_util.cc

@@ -58,6 +58,8 @@ Status SetValueFromQLBinaryWrapper(
       ql_value, pg_data_type, enum_oid_label_map, composite_atts_map, cdc_datum_message);
 }
 
+void DeleteMemoryContextForCDCWrapper() { yb::docdb::DeleteMemoryContextIfSet(); }
+
 DocDBRocksDBUtil::DocDBRocksDBUtil() {}
 
 DocDBRocksDBUtil::DocDBRocksDBUtil(InitMarkerBehavior init_marker_behavior)

src/yb/docdb/docdb_util.h

@@ -38,6 +38,8 @@ Status SetValueFromQLBinaryWrapper(
     const std::unordered_map<uint32_t, std::vector<master::PgAttributePB>>& composite_atts_map,
     DatumMessagePB* cdc_datum_message = NULL);
 
+void DeleteMemoryContextForCDCWrapper();
+
 struct ExternalIntent {
   dockv::DocPath doc_path;
   std::string value;