hcq/oceanbase
hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

remove ObOperator::get_next_row() in ObSelectIntoOp::inner_open()

Browse Source
This commit is contained in:
obdev 2022-12-23 07:37:51 +00:00 committed by ob-robot
parent 190932ebb3
commit 35eec7741d
2 changed files with 82 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/sql/code_generator/ob_static_engine_cg.cpp
View File

@ -5752,6 +5752,7 @@ int ObStaticEngineCG::generate_spec(ObLogSelectInto &op, ObSelectIntoSpec &spec,
spec.into_type_ = op.get_into_type(); spec.into_type_ = op.get_into_type();
spec.closed_cht_ = op.get_closed_cht(); spec.closed_cht_ = op.get_closed_cht();
spec.is_optional_ = op.get_is_optional(); spec.is_optional_ = op.get_is_optional();
spec.plan_->need_drive_dml_query_ = true;
} }
} }
return ret; return ret;

127
src/sql/engine/basic/ob_select_into_op.cpp
View File

@ -137,44 +137,22 @@ int ObSelectIntoOp::inner_open()
} }
} }
} }
return ret;
}
int ObSelectIntoOp::inner_get_next_row()
{
int ret = OB_SUCCESS;
int64_t row_count = 0;
const ObItemType into_type = MY_SPEC.into_type_;
ObPhysicalPlanCtx *phy_plan_ctx = NULL;
if (OB_ISNULL(phy_plan_ctx = ctx_.get_physical_plan_ctx())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get phy_plan_ctx failed", K(ret));
}
while (OB_SUCC(ret) && row_count < top_limit_cnt_) { while (OB_SUCC(ret) && row_count < top_limit_cnt_) {
if (is_vectorized()) { clear_evaluated_flag();
int64_t batch_size = MIN(spec_.max_batch_size_, top_limit_cnt_ - row_count); if (OB_FAIL(child_->get_next_row())) {
const ObBatchRows *brs = NULL;
if (OB_FAIL(ObOperator::get_next_batch(batch_size, brs))) {
LOG_WARN("get next batch failed", K(ret));
} else if (brs->size_ > 0) {
row_count += brs->size_ - brs->skip_->accumulate_bit_cnt(brs->size_);
if (T_INTO_OUTFILE == into_type) {
if (OB_FAIL(into_outfile_batch(*brs))) {
LOG_WARN("into outfile batch failed", K(ret));
}
} else {
ObEvalCtx::BatchInfoScopeGuard guard(eval_ctx_);
guard.set_batch_size(brs->size_);
for (int64_t i = 0; OB_SUCC(ret) && i < brs->size_; i++) {
if (brs->skip_->contain(i)) {
continue;
}
guard.set_batch_idx(i);
if (T_INTO_VARIABLES == into_type) {
if (OB_FAIL(into_varlist())) {
LOG_WARN("into varlist failed", K(ret));
}
} else {
if (OB_FAIL(into_dumpfile())) {
LOG_WARN("into dumpfile failed", K(ret));
}
}
}
}
}
if (OB_SUCC(ret) && brs->end_) {
ret = OB_ITER_END;
}
} else {
if (OB_FAIL(ObOperator::get_next_row())) {
if (OB_LIKELY(OB_ITER_END == ret)) { if (OB_LIKELY(OB_ITER_END == ret)) {
} else { } else {
LOG_WARN("get next row failed", K(ret)); LOG_WARN("get next row failed", K(ret));
@ -195,7 +173,6 @@ int ObSelectIntoOp::inner_open()
} }
} }
} }
}
if (OB_SUCC(ret) || OB_ITER_END == ret) { // if into user variables or into dumpfile, must be one row if (OB_SUCC(ret) || OB_ITER_END == ret) { // if into user variables or into dumpfile, must be one row
if ((T_INTO_VARIABLES == into_type || T_INTO_DUMPFILE == into_type) && row_count > 1) { if ((T_INTO_VARIABLES == into_type || T_INTO_DUMPFILE == into_type) && row_count > 1) {
ret = OB_ERR_TOO_MANY_ROWS; ret = OB_ERR_TOO_MANY_ROWS;
@ -203,50 +180,76 @@ int ObSelectIntoOp::inner_open()
} }
} }
} //end while } //end while
if (OB_ITER_END == ret || OB_SUCC(ret)) { // set affected rows
if (OB_ITER_END == ret) {
ret = OB_SUCCESS;
}
if (OB_SUCC(ret)) { // set affected rows
phy_plan_ctx->set_affected_rows(row_count); phy_plan_ctx->set_affected_rows(row_count);
} }
return ret; return ret;
} }
int ObSelectIntoOp::inner_get_next_row()
{
int ret = OB_SUCCESS;
clear_evaluated_flag();
if (OB_FAIL(child_->get_next_row())) {
if (OB_ITER_END != ret) {
LOG_WARN("get next row failed", K(ret));
} else {
// OB_ITER_END
}
}
return ret;
}
int ObSelectIntoOp::inner_get_next_batch(const int64_t max_row_cnt) int ObSelectIntoOp::inner_get_next_batch(const int64_t max_row_cnt)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
clear_evaluated_flag();
const ObBatchRows *child_brs = NULL; const ObBatchRows *child_brs = NULL;
if (OB_FAIL(child_->get_next_batch(max_row_cnt, child_brs))) { int64_t batch_size = min(max_row_cnt, MY_SPEC.max_batch_size_);
int64_t row_count = 0;
const ObItemType into_type = MY_SPEC.into_type_;
ObPhysicalPlanCtx *phy_plan_ctx = NULL;
if (OB_ISNULL(phy_plan_ctx = ctx_.get_physical_plan_ctx())) {
ret = OB_ERR_UNEXPECTED;
LOG_WARN("get phy_plan_ctx failed", K(ret));
}
bool stop_loop = false;
bool is_iter_end = false;
while (OB_SUCC(ret) && !stop_loop) {
clear_evaluated_flag();
int64_t rowkey_batch_size = min(batch_size, top_limit_cnt_ - row_count);
if (OB_FAIL(child_->get_next_batch(rowkey_batch_size, child_brs))) {
LOG_WARN("get next batch failed", K(ret)); LOG_WARN("get next batch failed", K(ret));
} else { } else {
brs_.size_ = child_brs->size_; brs_.size_ = child_brs->size_;
brs_.end_ = child_brs->end_; brs_.end_ = child_brs->end_;
is_iter_end = brs_.end_ && 0 == brs_.size_;
if (brs_.size_ > 0) { if (brs_.size_ > 0) {
brs_.skip_->deep_copy(*(child_brs->skip_), brs_.size_); brs_.skip_->deep_copy(*(child_brs->skip_), brs_.size_);
const ObIArray<ObExpr*> &select_exprs = MY_SPEC.output_; row_count += brs_.size_ - brs_.skip_->accumulate_bit_cnt(brs_.size_);
for (int64_t i = 0; i < select_exprs.count() && OB_SUCC(ret); i++) { if (T_INTO_OUTFILE == into_type) {
if (OB_FAIL(select_exprs.at(i)->eval_batch(eval_ctx_, *brs_.skip_, brs_.size_))) { if (OB_FAIL(into_outfile_batch(brs_))) {
LOG_WARN("eval expr failed", K(ret)); LOG_WARN("into outfile batch failed", K(ret));
}
} else {
ObEvalCtx::BatchInfoScopeGuard guard(eval_ctx_);
guard.set_batch_size(brs_.size_);
for (int64_t i = 0; OB_SUCC(ret) && i < brs_.size_; i++) {
if (brs_.skip_->contain(i)) {
continue;
}
guard.set_batch_idx(i);
if (T_INTO_VARIABLES == into_type) {
if (OB_FAIL(into_varlist())) {
LOG_WARN("into varlist failed", K(ret));
}
} else {
if (OB_FAIL(into_dumpfile())) {
LOG_WARN("into dumpfile failed", K(ret));
} }
} }
} }
} }
}
}
if (is_iter_end || row_count >= top_limit_cnt_) {
stop_loop = true;
}
if (OB_SUCC(ret) || is_iter_end) { // if into user variables or into dumpfile, must be one row
if ((T_INTO_VARIABLES == into_type || T_INTO_DUMPFILE == into_type) && row_count > 1) {
ret = OB_ERR_TOO_MANY_ROWS;
LOG_WARN("more than one row for into variables or into dumpfile", K(ret), K(row_count));
}
}
} //end while
if (OB_SUCC(ret)) { // set affected rows
phy_plan_ctx->set_affected_rows(row_count);
}
return ret; return ret;
} }