From 121dd427e8c94e5a2177d9dce7139b823d13e975 Mon Sep 17 00:00:00 2001
From: LiuYoung00 <liuyanglo_ol@163.com>
Date: Wed, 22 Mar 2023 04:42:13 +0000
Subject: [PATCH] [to #47632376]add pipe row memory limit

---
 src/pl/ob_pl.cpp                                         | 7 +++++++
 src/pl/ob_pl_allocator.h                                 | 9 ++++++++-
 src/share/parameter/ob_parameter_seed.ipp                | 3 +++
 .../r/mysql/all_virtual_sys_parameter_stat.result        | 1 +
 4 files changed, 19 insertions(+), 1 deletion(-)

diff --git a/src/pl/ob_pl.cpp b/src/pl/ob_pl.cpp
index 9ea0fe4e9..726940d6c 100644
--- a/src/pl/ob_pl.cpp
+++ b/src/pl/ob_pl.cpp
@@ -2400,6 +2400,13 @@ int ObPLExecState::final(int ret)
     }
   }
 
+  if (OB_FAIL(ret) && func_.get_ret_type().is_composite_type() && result_.is_ext() && func_.is_pipelined()) {
+    tmp_ret = ObUserDefinedType::destruct_obj(result_, ctx_.exec_ctx_->get_my_session());
+    if (OB_SUCCESS != tmp_ret) {
+      LOG_WARN("failed to destruct pl object", K(tmp_ret));
+    }
+  }
+
   if (OB_NOT_NULL(top_context_)
       && top_context_->get_exec_stack().count() > 0
       && top_context_->get_exec_stack().at(
diff --git a/src/pl/ob_pl_allocator.h b/src/pl/ob_pl_allocator.h
index 9e4277e3c..6b06ab5e0 100644
--- a/src/pl/ob_pl_allocator.h
+++ b/src/pl/ob_pl_allocator.h
@@ -22,11 +22,16 @@ namespace oceanbase
 {
 namespace pl
 {
+/* ObPLAllocator can be dynamically scaled， to reduce memory consumption introduced by deep copy
+ *  when curr_.used > next_threshold_ or (src copy size) * 2
+ *  we alloc a new mem and reset old
+ *
+ */
 class ObPLAllocator : public common::ObIAllocator
 {
 public:
   ObPLAllocator()
-    : allocator1_(ObModIds::OB_PL_TEMP, OB_MALLOC_NORMAL_BLOCK_SIZE, MTL_ID()),
+    : allocator1_(ObModIds::OB_PL_TEMP, OB_MALLOC_NORMAL_BLOCK_SIZE, MTL_ID()), // PLAllocator use current tenant memory set by MTL_ID()
       allocator2_(ObModIds::OB_PL_TEMP, OB_MALLOC_NORMAL_BLOCK_SIZE, MTL_ID()),
       curr_(&allocator1_),
       backup_(&allocator2_),
@@ -55,6 +60,8 @@ public:
 
   ObIAllocator* get_allocator() { return curr_; }
 
+  int64_t get_used() { return allocator1_.used() + allocator2_.used(); }
+
 private:
   const static int PL_ALLOC_THRESHOLD = 1024 * 1024; // 1M
 
diff --git a/src/share/parameter/ob_parameter_seed.ipp b/src/share/parameter/ob_parameter_seed.ipp
index 4f5a72196..b29665bdf 100644
--- a/src/share/parameter/ob_parameter_seed.ipp
+++ b/src/share/parameter/ob_parameter_seed.ipp
@@ -1397,3 +1397,6 @@ DEF_BOOL(_enable_transaction_internal_routing, OB_TENANT_PARAMETER, "True",
 DEF_STR(_load_tde_encrypt_engine, OB_CLUSTER_PARAMETER, "NONE",
         "load the engine that meet the security classification requirement to encrypt data.  default NONE",
         ObParameterAttr(Section::OBSERVER, Source::DEFAULT, EditLevel::DYNAMIC_EFFECTIVE));
+DEF_INT(_pipelined_table_function_memory_limit, OB_TENANT_PARAMETER, "524288000", "[1024,18446744073709551615]",
+        "pipeline table function result set memory size limit. default 524288000 (500M), Range: [1024,18446744073709551615]",
+        ObParameterAttr(Section::OBSERVER, Source::DEFAULT, EditLevel::DYNAMIC_EFFECTIVE));
diff --git a/tools/deploy/mysql_test/test_suite/inner_table/r/mysql/all_virtual_sys_parameter_stat.result b/tools/deploy/mysql_test/test_suite/inner_table/r/mysql/all_virtual_sys_parameter_stat.result
index fdb93d5ad..767d3e6cc 100644
--- a/tools/deploy/mysql_test/test_suite/inner_table/r/mysql/all_virtual_sys_parameter_stat.result
+++ b/tools/deploy/mysql_test/test_suite/inner_table/r/mysql/all_virtual_sys_parameter_stat.result
@@ -303,6 +303,7 @@ _ob_trans_rpc_timeout
 _parallel_max_active_sessions
 _parallel_min_message_pool
 _parallel_server_sleep_time
+_pipelined_table_function_memory_limit
 _print_sample_ppm
 _private_buffer_size
 _pushdown_storage_level