allocate material when more than one link scan exist

src/sql/engine/table/ob_link_scan_op.cpp

@@ -294,6 +294,8 @@ int ObLinkScanOp::inner_get_next_row()
   } else if (OB_FAIL(result_->next())) {
     if (OB_ITER_END != ret) {
       LOG_WARN("failed to get next row", K(ret));
+    } else {
+      reset_result();
     }
   } else {
     const ObIArray<ObExpr *> &output = spec_.output_;
@@ -358,6 +360,7 @@ int ObLinkScanOp::inner_get_next_batch(const int64_t max_row_cnt)
   if (iter_end_) {
     brs_.size_ = 0;
     brs_.end_ = true;
+    reset_result();
   } else {
     ObEvalCtx::BatchInfoScopeGuard batch_info_guard(eval_ctx_);
     auto loop_cnt = common::min(max_row_cnt, MY_SPEC.max_batch_size_);

src/sql/optimizer/ob_log_link_scan.cpp

@@ -22,8 +22,14 @@ ObLogLinkScan::ObLogLinkScan(ObLogPlan &plan)
 int ObLogLinkScan::allocate_expr_post(ObAllocExprContext &ctx)
 {
   int ret = OB_SUCCESS;
-  for (int64_t i = 0; OB_SUCC(ret) && i < output_exprs_.count(); ++i) {
+  const ObSelectStmt *stmt = NULL;
-    ObRawExpr *expr = output_exprs_.at(i);
+  if (OB_ISNULL(get_plan()) ||
+      OB_ISNULL(stmt = static_cast<const ObSelectStmt *>(get_plan()->get_stmt()))) {
+    ret = OB_ERR_UNEXPECTED;
+    LOG_WARN("get unexpected null", K(get_plan()), K(stmt), K(ret));
+  }
+  for (int64_t i = 0; OB_SUCC(ret) && i < stmt->get_select_item_size(); ++i) {
+    ObRawExpr *expr = stmt->get_select_item(i).expr_;
     if (OB_ISNULL(expr)) {
       ret = OB_ERR_UNEXPECTED;
       LOG_WARN("null expr", K(ret));

src/sql/optimizer/ob_optimizer.cpp

@@ -622,6 +622,7 @@ int ObOptimizer::init_env_info(ObDMLStmt &stmt)
   int64_t max_table_hint = 1;
   ObDMLStmt *target_stmt = &stmt;
   ObSQLSessionInfo *session = ctx_.get_session_info();
+  int64_t link_stmt_count = 0;
   if (OB_ISNULL(target_stmt) || OB_ISNULL(session)) {
     ret = OB_ERR_UNEXPECTED;
     LOG_WARN("get unexpected null", K(ret));
@@ -645,7 +646,10 @@ int ObOptimizer::init_env_info(ObDMLStmt &stmt)
                                                session_enable_parallel,
                                                session_force_parallel_dop))) {
     LOG_WARN("failed to get session parallel info", K(ret));
+  } else if (OB_FAIL(calc_link_stmt_count(*target_stmt, link_stmt_count))) {
+    LOG_WARN("calc link stmt count failed", K(ret));
   } else {
+    ctx_.set_has_multiple_link_stmt(link_stmt_count > 1);
     parallel = ctx_.get_global_hint().get_parallel_hint();
     if (parallel <= 0) {
       parallel = ObGlobalHint::DEFAULT_PARALLEL;
@@ -794,6 +798,45 @@ int ObOptimizer::check_whether_contain_nested_sql(const ObDMLStmt &stmt)
   return ret;
 }
 
+int ObOptimizer::calc_link_stmt_count(const ObDMLStmt &stmt, int64_t &count)
+{
+  int ret = OB_SUCCESS;
+  if (stmt.is_dblink_stmt()) {
+    count += 1;
+  } else {
+    ObSEArray<ObSelectStmt *, 4> child_stmts;
+    if (OB_FAIL(stmt.get_child_stmts(child_stmts))) {
+      LOG_WARN("failed to get child stmts", K(ret));
+    }
+    for (int64_t i = 0; OB_SUCC(ret) && i < child_stmts.count(); ++i) {
+      if (OB_ISNULL(child_stmts.at(i))) {
+        ret = OB_ERR_UNEXPECTED;
+        LOG_WARN("get unexpected null", K(ret));
+      } else if (OB_FAIL(SMART_CALL(calc_link_stmt_count(*child_stmts.at(i), count)))) {
+        LOG_WARN("failed to extract column usage info", K(ret));
+      }
+    }
+    if (OB_SUCC(ret)) {
+      const common::ObIArray<TableItem*> &table_items = stmt.get_table_items();
+      for (int64_t i = 0; i < table_items.count() && OB_SUCC(ret); i++) {
+        const TableItem *table_item = table_items.at(i);
+        if (OB_ISNULL(table_item)) {
+          ret = OB_ERR_UNEXPECTED;
+          LOG_WARN("get null ptr", K(ret));
+        } else if (table_item->is_temp_table()) {
+          if (OB_ISNULL(table_item->ref_query_)) {
+            ret = OB_ERR_UNEXPECTED;
+            LOG_WARN("get null ptr", K(ret));
+          } else if (OB_FAIL(SMART_CALL(calc_link_stmt_count(*table_item->ref_query_, count)))) {
+            LOG_WARN("failed to extract column usage info", K(ret));
+          }
+        }
+      }
+    }
+  }
+  return ret;
+}
+
 int ObOptimizer::extract_column_usage_info(const ObDMLStmt &stmt)
 {
   int ret = OB_SUCCESS;

src/sql/optimizer/ob_optimizer.h

@@ -215,6 +215,7 @@ namespace sql
                              const ObColumnRefRawExpr &column_expr,
                              int64_t flag);
     int check_whether_contain_nested_sql(const ObDMLStmt &stmt);
+    int calc_link_stmt_count(const ObDMLStmt &stmt, int64_t &count);
   private:
     ObOptimizerContext &ctx_;
     DISALLOW_COPY_AND_ASSIGN(ObOptimizer);

src/sql/optimizer/ob_optimizer_context.h

@@ -148,7 +148,8 @@ ObOptimizerContext(ObSQLSessionInfo *session_info,
     nested_sql_flags_(0),
     has_for_update_(false),
     has_var_assign_(false),
-    is_var_assign_only_in_root_stmt_(false)
+    is_var_assign_only_in_root_stmt_(false),
+    has_multiple_link_stmt_(false)
   { }
   inline common::ObOptStatManager *get_opt_stat_manager() { return opt_stat_manager_; }
   inline void set_opt_stat_manager(common::ObOptStatManager *sm) { opt_stat_manager_ = sm; }
@@ -474,6 +475,8 @@ ObOptimizerContext(ObSQLSessionInfo *session_info,
   inline void set_has_var_assign(bool v) { has_var_assign_ = v; }
   inline bool is_var_assign_only_in_root_stmt() { return is_var_assign_only_in_root_stmt_; }
   inline void set_is_var_assign_only_in_root_stmt(bool v) { is_var_assign_only_in_root_stmt_ = v; }
+  inline bool has_multiple_link_stmt() const { return has_multiple_link_stmt_; }
+  inline void set_has_multiple_link_stmt(bool v) { has_multiple_link_stmt_ = v; }
 
 private:
   ObSQLSessionInfo *session_info_;
@@ -545,6 +548,7 @@ private:
   bool has_for_update_;
   bool has_var_assign_;
   bool is_var_assign_only_in_root_stmt_;
+  bool has_multiple_link_stmt_;
 };
 }
 }

src/sql/optimizer/ob_select_log_plan.cpp

@@ -3993,16 +3993,26 @@ int ObSelectLogPlan::generate_dblink_raw_plan()
     //do nothing
   } else if (OB_FAIL(allocate_link_scan_as_top(top))) {
     LOG_WARN("failed to allocate link dml as top", K(ret));
+  } else {
+    top->set_dblink_id(dblink_id);
+    ObLogLinkScan *link_scan = static_cast<ObLogLinkScan *>(top);
+    if (OB_FAIL(link_scan->set_link_stmt(stmt))) {
+      LOG_WARN("failed to set link stmt", K(ret));
+    } else if (0 == dblink_id) {
+      link_scan->set_reverse_link(true);
+    }
+  }
+  if (OB_FAIL(ret)) {
+    // do nothing
+  } else if (optimizer_context_.has_multiple_link_stmt()
+             && OB_FAIL(allocate_material_as_top(top))) {
+    LOG_WARN("allocate material above link scan failed", K(ret));
   } else if (OB_FAIL(make_candidate_plans(top))) {
     LOG_WARN("failed to make candidate plans", K(ret));
-  } else if (OB_FAIL(static_cast<ObLogLink *>(top)->set_link_stmt(stmt))) {
-    LOG_WARN("failed to set link stmt", K(ret));
   } else {
     top->mark_is_plan_root();
     top->get_plan()->set_plan_root(top);
-    top->set_dblink_id(dblink_id);
     if (0 == dblink_id) {
-      static_cast<ObLogLinkScan *>(top)->set_reverse_link(true);
       // reset dblink info, to avoid affecting the next execution flow
       query_ctx->get_query_hint_for_update().get_global_hint().reset_dblink_info_hint();
     } else {

src/sql/resolver/dml/ob_dml_stmt.h

@@ -970,6 +970,7 @@ public:
   int add_autoinc_param(share::AutoincParam &autoinc_param) { return autoinc_params_.push_back(autoinc_param); }
   inline void set_dblink_id(int64_t id) { dblink_id_ = id; }
   inline int64_t get_dblink_id() const { return dblink_id_; }
+  inline bool is_dblink_stmt() const { return OB_INVALID_ID != dblink_id_; }
   inline void set_reverse_link() { is_reverse_link_ = true; }
   inline bool is_reverse_link() const { return is_reverse_link_; }
   int add_subquery_ref(ObQueryRefRawExpr *query_ref);