[Bugfix] obkv bugfix

src/observer/table/ob_table_context.cpp

@@ -345,6 +345,23 @@ int ObTableCtx::check_entity()
   return ret;
 }
 
+bool ObTableCtx::has_exist_in_columns(const ObIArray<ObString> &columns,
+                                      const ObString &name,
+                                      int64_t *idx /* =nullptr */) const
+{
+  bool exist = false;
+  int64_t num = columns.count();
+  for (int64_t i = 0; i < num && !exist; i++) {
+    if (0 == name.case_compare(columns.at(i))) {
+      exist = true;
+      if (idx != NULL) {
+        *idx = i;
+      }
+    }
+  }
+  return exist;
+}
+
 int ObTableCtx::generate_key_range(const ObIArray<ObNewRange> &scan_ranges)
 {
   int ret = OB_SUCCESS;
@@ -495,7 +512,7 @@ int ObTableCtx::init_scan(const ObTableQuery &query,
         if (OB_ISNULL(column_schema)) {
           ret = OB_ERR_UNEXPECTED;
           LOG_WARN("column schema is NULL", K(ret));
-        } else if (has_exist_in_array(select_columns, column_schema->get_column_name_str())) {
+        } else if (has_exist_in_columns(select_columns, column_schema->get_column_name_str())) {
           if (OB_FAIL(select_col_ids_.push_back(column_schema->get_column_id()))) {
             LOG_WARN("fail to add column id", K(ret));
           } else if (OB_FAIL(select_metas_.push_back(column_schema->get_meta_type()))) {

src/observer/table/ob_table_context.h

@@ -284,6 +284,9 @@ private:
   int check_rowkey(ObRowkey &rowkey);
   int check_properties(ObIArray<std::pair<ObString, ObObj>> &properties);
   int check_entity();
+  bool has_exist_in_columns(const common::ObIArray<common::ObString>& columns,
+                            const common::ObString &name,
+                            int64_t *idx = nullptr) const;
 private:
   bool is_init_;
   common::ObIAllocator &allocator_;

src/observer/table/ob_table_executor.h

@@ -31,11 +31,18 @@ public:
         is_opened_(false)
   {
   }
-  virtual ~ObTableApiExecutor() {}
+  virtual ~ObTableApiExecutor()
+  {
+    if (OB_NOT_NULL(child_)) {
+      child_->~ObTableApiExecutor();
+      child_ = nullptr;
+    }
+  }
 public:
   virtual int open() = 0;
   virtual int get_next_row() = 0;
   virtual int close() = 0;
+  virtual void destroy() = 0;
 public:
   void set_parent(ObTableApiExecutor *parent);
   void set_child(ObTableApiExecutor *child);

src/observer/table/ob_table_insert_up_executor.h

@@ -61,10 +61,21 @@ public:
         cur_idx_(0)
   {
   }
+  virtual ~ObTableApiInsertUpExecutor()
+  {
+    destroy();
+  }
 public:
   virtual int open();
   virtual int get_next_row();
   virtual int close();
+  virtual void destroy()
+  {
+    // destroy
+    conflict_checker_.destroy();
+    upd_rtctx_.cleanup();
+    ObTableApiModifyExecutor::destroy();
+  }
 public:
   OB_INLINE bool is_insert_duplicated()
   {

src/observer/table/ob_table_modify_executor.h

@@ -64,10 +64,16 @@ public:
   }
   virtual ~ObTableApiModifyExecutor()
   {
+    destroy();
   }
 public:
   virtual int open() override;
   virtual int close() override;
+  virtual void destroy() override
+  {
+    dml_rtctx_.cleanup();
+  }
+
   int init_das_ref();
   int submit_all_dml_task();
   int init_das_dml_rtdef(const sql::ObDASDMLBaseCtDef &das_ctdef,

src/observer/table/ob_table_op_wrapper.cpp

@@ -92,12 +92,14 @@ int ObTableOpWrapper::process_op_with_spec(ObTableCtx &tb_ctx,
         && OB_FAIL(process_affected_entity(tb_ctx, *spec, *executor, op_result))) {
       LOG_WARN("fail to process affected entity", K(ret), K(tb_ctx));
     }
+  }
+
   int tmp_ret = OB_SUCCESS;
   if (OB_SUCCESS != (tmp_ret = executor->close())) {
     LOG_WARN("fail to close executor", K(tmp_ret));
     ret = COVER_SUCC(tmp_ret);
   }
-  }
+
   op_result.set_errno(ret);
   op_result.set_type(tb_ctx.get_opertion_type());
   spec->destroy_executor(executor);

src/observer/table/ob_table_replace_executor.h

@@ -55,10 +55,20 @@ public:
         cur_idx_(0)
   {
   }
+  virtual ~ObTableApiReplaceExecutor()
+  {
+    destroy();
+  }
 public:
   virtual int open();
   virtual int get_next_row();
   virtual int close();
+  virtual void destroy()
+  {
+    // destroy
+    conflict_checker_.destroy();
+    ObTableApiModifyExecutor::destroy();
+  }
 private:
   const static int64_t DEFAULT_REPLACE_BATCH_ROW_COUNT = 1000L;
   OB_INLINE const common::ObIArray<sql::ObExpr *>& get_primary_table_new_row()

src/observer/table/ob_table_scan_executor.h

@@ -51,6 +51,7 @@ public:
   int open() override;
   int get_next_row() override;
   int close() override;
+  void destroy() override {}
 public:
   OB_INLINE const ObTableApiScanSpec& get_spec() const { return scan_spec_; }
   OB_INLINE void reset()

src/share/table/ob_table.cpp

@@ -307,6 +307,21 @@ int64_t ObTableEntity::hash_rowkey() const
   return hash_value;
 }
 
+bool ObTableEntity::has_exist_in_properties(const ObString &name, int64_t *idx /* =nullptr */) const
+{
+  bool exist = false;
+  int64_t num = properties_names_.count();
+  for (int64_t i = 0; i < num && !exist; i++) {
+    if (0 == name.case_compare(properties_names_.at(i))) {
+      exist = true;
+      if (idx != NULL) {
+        *idx = i;
+      }
+    }
+  }
+  return exist;
+}
+
 int ObTableEntity::get_property(const ObString &prop_name, ObObj &prop_value) const
 {
   int ret = OB_SUCCESS;
@@ -315,7 +330,7 @@ int ObTableEntity::get_property(const ObString &prop_name, ObObj &prop_value) co
     LOG_WARN("property name should not be empty string", K(ret), K(prop_name));
   } else {
     int64_t idx = -1;
-    if (has_exist_in_array(properties_names_, prop_name, &idx)) {
+    if (has_exist_in_properties(prop_name, &idx)) {
       prop_value = properties_values_.at(idx);
     } else {
       ret = OB_SEARCH_NOT_FOUND;
@@ -332,7 +347,7 @@ int ObTableEntity::set_property(const ObString &prop_name, const ObObj &prop_val
     LOG_WARN("property name should not be empty string", K(ret), K(prop_name));
   } else {
     int64_t idx = -1;
-    if (has_exist_in_array(properties_names_, prop_name, &idx)) {
+    if (has_exist_in_properties(prop_name, &idx)) {
       properties_values_.at(idx) = prop_value;
     } else {
       if (OB_FAIL(properties_names_.push_back(prop_name))) {

src/share/table/ob_table.h

@@ -120,6 +120,8 @@ public:
   const ObIArray<ObObj> &get_properties_values() const { return properties_values_; }
   const ObSEArray<ObObj, 8> &get_rowkey_objs() const { return rowkey_; };
   DECLARE_TO_STRING;
+private:
+  bool has_exist_in_properties(const ObString &name, int64_t *idx = nullptr) const;
 private:
   ObSEArray<ObObj, 8> rowkey_;
   ObSEArray<ObString, 8> properties_names_;