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

Fix error result of window function vec2.0

Browse Source
This commit is contained in:
Zach41
2024-06-28 08:29:00 +00:00
committed by ob-robot
parent a27e86cfaa
commit a873193ea0
1 changed files with 26 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
src/sql/engine/window_function/ob_window_function_vec_op.cpp
View File

@ -3070,10 +3070,10 @@ int ObWindowFunctionVecOp::rd_merge_result(PartialMerge &part_res, WinFuncInfo &
ResFmt *res_data = static_cast<ResFmt *>(wf_expr->get_vector(eval_ctx_));
ObEvalCtx::BatchInfoScopeGuard guard(eval_ctx_);
if (patch_first_) {
for (int i = 0; OB_SUCC(ret) && i < batch_size; i++) {
null_payload = rd_patch_->first_row_->is_null(rd_col_id);
rd_patch_->first_row_->get_cell_payload(*rd_coord_row_meta_, rd_col_id, payload, len);
for (int i = 0; !null_payload && OB_SUCC(ret) && i < batch_size; i++) {
guard.set_batch_idx(i);
null_payload = rd_patch_->first_row_->is_null(rd_col_id);
rd_patch_->first_row_->get_cell_payload(*rd_coord_row_meta_, rd_col_id, payload, len);
if (is_rank && i >= first_row_same_order_upper_bound) {
int64_t rank_patch = rd_patch_->first_row_frame_offset();
if (OB_FAIL(part_res.template add_rank<ResFmt>(info, true, nullptr, 0, rank_patch))) {
@ -3085,11 +3085,11 @@ int ObWindowFunctionVecOp::rd_merge_result(PartialMerge &part_res, WinFuncInfo &
}
}
if (OB_SUCC(ret) && patch_last_) {
for (int i = 0; OB_SUCC(ret) && i < batch_size; i++) {
null_payload = rd_patch_->last_row_->is_null(rd_col_id);
rd_patch_->last_row_->get_cell_payload(*rd_coord_row_meta_, rd_col_id, payload, len);
for (int i = 0; !null_payload && OB_SUCC(ret) && i < batch_size; i++) {
guard.set_batch_idx(i);
if (i >= last_row_same_order_lower_bound) {
null_payload = rd_patch_->last_row_->is_null(rd_col_id);
rd_patch_->last_row_->get_cell_payload(*rd_coord_row_meta_, rd_col_id, payload, len);
if (OB_FAIL(part_res.template merge<ResFmt>(info, null_payload, payload, len))) {
LOG_WARN("merge last row's patch failed", K(ret));
}
@ -3135,6 +3135,8 @@ int ObWindowFunctionVecOp::rd_apply_patches(const int64_t max_row_cnt)
LOG_WARN("find first row same order lower bound idx failed", K(ret));
} else if (patch_last_ && OB_FAIL(rd_find_last_row_lower_bound(max_row_cnt, last_row_same_order_lower))) {
LOG_WARN("find last row same order upper bound idx failed", K(ret));
} else {
LOG_TRACE("rd apply patch", K(first_row_same_order_upper), K(last_row_same_order_lower));
}
for (int i = 0; OB_SUCC(ret) && i < MY_SPEC.rd_wfs_.count(); i++) {
WinFuncInfo &info = const_cast<WinFuncInfo &>(MY_SPEC.wf_infos_.at(MY_SPEC.rd_wfs_.at(i)));
@ -3244,7 +3246,9 @@ int ObWindowFunctionVecOp::rd_find_first_row_upper_bound(int64_t batch_size, int
for (int i = 0; OB_SUCC(ret) && i < MY_SPEC.rd_sort_collations_.count(); i++) {
int64_t tmp_bound = -1;
ObExpr *sort_expr = MY_SPEC.rd_coord_exprs_.at(i);
NullSafeRowCmpFunc cmp_fn = sort_expr->basic_funcs_->row_null_first_cmp_;
NullSafeRowCmpFunc cmp_fn = (MY_SPEC.rd_sort_collations_.at(i).null_pos_ == NULL_FIRST ?
sort_expr->basic_funcs_->row_null_first_cmp_ :
sort_expr->basic_funcs_->row_null_last_cmp_);
rd_patch_->first_row_->get_cell_payload(rd_patch_->row_meta_, i, val, val_len);
val_isnull = rd_patch_->first_row_->is_null(i);
VectorRangeUtil::NullSafeCmp cmp_op(sort_expr->obj_meta_, cmp_fn, val, val_len, val_isnull,
@ -3281,7 +3285,9 @@ int ObWindowFunctionVecOp::rd_find_last_row_lower_bound(int64_t batch_size, int6
int64_t tmp_bound = -1;
int64_t field_idx = MY_SPEC.rd_sort_collations_.at(i).field_idx_;
ObExpr *sort_expr = MY_SPEC.all_expr_.at(field_idx);
NullSafeRowCmpFunc cmp_fn = sort_expr->basic_funcs_->row_null_first_cmp_;
NullSafeRowCmpFunc cmp_fn = (MY_SPEC.rd_sort_collations_.at(i).null_pos_ == NULL_FIRST ?
sort_expr->basic_funcs_->row_null_first_cmp_ :
sort_expr->basic_funcs_->row_null_last_cmp_);
rd_patch_->last_row_->get_cell_payload(rd_patch_->row_meta_, i, val, val_len);
val_isnull = rd_patch_->last_row_->is_null(i);
VectorRangeUtil::NullSafeCmp cmp_op(sort_expr->obj_meta_, cmp_fn, val, val_len, val_isnull,
@ -3473,7 +3479,7 @@ int32_t WinFuncColExpr::non_aggr_reserved_row_size() const
if (!wf_expr_->is_aggregate_expr()) {
if (is_fixed_length_vec(vec_tc) || vec_tc == VEC_TC_NUMBER) {
ret_size = ObDatum::get_reserved_size(
ObDatum::get_obj_datum_map_type(wf_info_.expr_->datum_meta_.type_));
ObDatum::get_obj_datum_map_type(wf_info_.expr_->datum_meta_.type_), wf_info_.expr_->datum_meta_.precision_);
} else {
ret_size = sizeof(char *) + sizeof(uint32_t); // <char *, len>
}
@ -3952,6 +3958,17 @@ int ObWindowFunctionVecSpec::rd_generate_patch(RDWinFuncPXPieceMsgCtx &msg_ctx,
msg_ctx.infos_.at(i)->last_row_ = patch_pairs.at(i).second;
}
}
#ifndef NDEBUG
for (int i = 0; i < msg_ctx.infos_.count(); i++) {
RDWinFuncPXPartialInfo *info = msg_ctx.infos_.at(i);
if (info->first_row_ == nullptr) { break; }
CompactRow2STR first_row(info->row_meta_, *info->first_row_, &rd_coord_exprs_);
CompactRow2STR last_row(info->row_meta_, *info->last_row_, &rd_coord_exprs_);
int64_t first_row_extra = *reinterpret_cast<int64_t *>(info->first_row_->get_extra_payload(info->row_meta_));
int64_t last_row_extra = *reinterpret_cast<int64_t *>(info->last_row_->get_extra_payload(info->row_meta_));
LOG_INFO("after generating patch", K(i), K(first_row), K(last_row), K(first_row_extra), K(last_row_extra));
}
#endif
return ret;
}