fix memory leak when forall batch execute

src/sql/ob_spi.cpp

@@ -4661,7 +4661,8 @@ int ObSPIService::construct_exec_params(ObPLExecCtx *ctx,
                                         const ObSqlExpression **into_exprs,
                                         int64_t into_count,
                                         ParamStore &exec_params,
-                                        ObSPIOutParams &out_params)
+                                        ObSPIOutParams &out_params,
+                                        bool is_forall)
 {
   int ret = OB_SUCCESS;
   ObObjParam result;
@@ -4723,13 +4724,15 @@ int ObSPIService::construct_exec_params(ObPLExecCtx *ctx,
       }
       if (OB_SUCC(ret)) { //执行期参数必须从PL的内存空间拷贝到SQL自己的内存空间，以防止在SQL执行过程中参数被改掉
         ObObjParam new_param = result;
-        if (result.is_pl_extend()) {
-          if (result.get_meta().get_extend_type() != PL_REF_CURSOR_TYPE) {
-            new_param.set_int_value(0);
-            OZ (pl::ObUserDefinedType::deep_copy_obj(param_allocator, result, new_param, true));
+        if (!is_forall) { // forall场景, 不做拷贝, 上层代码transform_pl_ext_type会进行paramstore整体拷贝
+          if (result.is_pl_extend()) {
+            if (result.get_meta().get_extend_type() != PL_REF_CURSOR_TYPE) {
+              new_param.set_int_value(0);
+              OZ (pl::ObUserDefinedType::deep_copy_obj(param_allocator, result, new_param, true));
+            }
+          } else {
+            OZ (deep_copy_obj(param_allocator, result, new_param));
           }
-        } else {
-          OZ (deep_copy_obj(param_allocator, result, new_param));
         }
         OX (new_param.set_need_to_check_type(true));
         OZ (exec_params.push_back(new_param));
@@ -4804,7 +4807,7 @@ int ObSPIService::inner_open(ObPLExecCtx *ctx,
 
   if (NULL == sql) {
     OZ (construct_exec_params(ctx, param_allocator, param_exprs, param_count,
-                              into_exprs, into_count, exec_params, out_params),
+                              into_exprs, into_count, exec_params, out_params, is_forall),
       K(sql), K(id), K(type), K(param_count), K(out_params), K(exec_params));
   }
 
@@ -4824,7 +4827,7 @@ int ObSPIService::inner_open(ObPLExecCtx *ctx,
         ret = OB_ERR_UNEXPECTED;
         LOG_WARN("array_binding_count is wrong", K(array_binding_count), K(ret));
       } else if (OB_FAIL(ObSQLUtils::transform_pl_ext_type(
-          exec_params, array_binding_count, param_allocator, batch_params))) {
+          exec_params, array_binding_count, param_allocator, batch_params, true))) {
         LOG_WARN("transform failed", K(ret));
       } else if (OB_ISNULL(batch_params)) {
         ret = OB_ERR_UNEXPECTED;

src/sql/ob_spi.h

@@ -626,7 +626,8 @@ private:
                                    const ObSqlExpression **into_exprs,
                                    int64_t into_count,
                                    ParamStore &exec_params,
-                                   ObSPIOutParams &out_params);
+                                   ObSPIOutParams &out_params,
+                                   bool is_forall = false);
 
   static int inner_open(pl::ObPLExecCtx *ctx,
                         ObIAllocator &param_allocator, //用于拷贝执行期参数

src/sql/ob_sql_utils.cpp

@@ -4417,7 +4417,8 @@ int ObSQLUtils::get_one_group_params(int64_t &actual_pos, ParamStore &src, Param
   return ret;
 }
 
-int ObSQLUtils::copy_params_to_array_params(int64_t query_pos, ParamStore &src, ParamStore &dst)
+int ObSQLUtils::copy_params_to_array_params(int64_t query_pos, ParamStore &src, ParamStore &dst,
+                                            ObIAllocator &alloc, bool is_forall)
 {
   int ret = OB_SUCCESS;
   for (int64_t j = 0; OB_SUCC(ret) && j < dst.count(); j++) {
@@ -4431,7 +4432,11 @@ int ObSQLUtils::copy_params_to_array_params(int64_t query_pos, ParamStore &src,
       ret = OB_ERR_UNEXPECTED;
       LOG_WARN("get unexpected null", K(ret), KPC(array_params));
     } else {
-      array_params->data_[query_pos] = src.at(j);
+      ObObjParam new_param = src.at(j);
+      if (is_forall) {
+        OZ (deep_copy_obj(alloc, src.at(j), new_param));
+      }
+      array_params->data_[query_pos] = new_param;
     }
   }
   return ret;
@@ -4463,7 +4468,7 @@ int ObSQLUtils::init_elements_info(ParamStore &src, ParamStore &dst)
 }
 
 int ObSQLUtils::transform_pl_ext_type(
-    ParamStore &src, int64_t array_binding_size, ObIAllocator &alloc, ParamStore *&dst)
+    ParamStore &src, int64_t array_binding_size, ObIAllocator &alloc, ParamStore *&dst, bool is_forall)
 {
   int ret = OB_SUCCESS;
   ParamStore *ps_ab_params = NULL;
@@ -4489,7 +4494,7 @@ int ObSQLUtils::transform_pl_ext_type(
         LOG_WARN("fail to reverse params_store", K(ret));
       } else if (OB_FAIL(get_one_group_params(actual_pos, src, temp_obj_params))) {
         LOG_WARN("get one group params failed", K(ret), K(actual_pos));
-      } else if (OB_FAIL(copy_params_to_array_params(query_pos, temp_obj_params, *dst))) {
+      } else if (OB_FAIL(copy_params_to_array_params(query_pos, temp_obj_params, *dst, alloc, is_forall))) {
         LOG_WARN("copy params to array params failed", K(ret), K(query_pos));
       } else if (query_pos == 0) {
         if (OB_FAIL(init_elements_info(src, *dst))) {

src/sql/ob_sql_utils.h

@@ -536,9 +536,9 @@ public:
                                            int64_t query_num,
                                            int64_t param_num,
                                            ParamStore &param_store);
-  static int transform_pl_ext_type(ParamStore &src, int64_t array_binding_size, ObIAllocator &alloc, ParamStore *&dst);
+  static int transform_pl_ext_type(ParamStore &src, int64_t array_binding_size, ObIAllocator &alloc, ParamStore *&dst, bool is_forall = false);
   static int get_one_group_params(int64_t &pos, ParamStore &src, ParamStore &obj_params);
-  static int copy_params_to_array_params(int64_t query_pos, ParamStore &src, ParamStore &dst);
+  static int copy_params_to_array_params(int64_t query_pos, ParamStore &src, ParamStore &dst, ObIAllocator &alloc, bool is_forall = false);
   static int init_elements_info(ParamStore &src, ParamStore &dst);
   /*-----------------------
   *  Observer no longer depends on Linux NTP service to adjust server time since 4.0.