/**

* Copyright (c) 2021 OceanBase

* OceanBase CE is licensed under Mulan PubL v2.

* You can use this software according to the terms and conditions of the Mulan PubL v2.

* You may obtain a copy of Mulan PubL v2 at:

* http://license.coscl.org.cn/MulanPubL-2.0

* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,

* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,

* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.

* See the Mulan PubL v2 for more details.

*/

#define USING_LOG_PREFIX SQL_ENG

#include "ob_sort_op_impl.h"

#include "sql/engine/ob_operator.h"

#include "sql/engine/ob_tenant_sql_memory_manager.h"

#include "storage/blocksstable/encoding/ob_encoding_query_util.h"

#include "lib/container/ob_iarray.h"

namespace oceanbase

{

using namespace common;

namespace sql

{

/************************************* start ObSortOpImpl *********************************/

ObSortOpImpl::ObAdaptiveQS::ObAdaptiveQS(common::ObIArray<ObChunkDatumStore::StoredRow *> &sort_rows,

common::ObIAllocator &alloc)

: orig_sort_rows_(sort_rows),

alloc_(alloc)

{

}

int ObSortOpImpl::ObAdaptiveQS::init(common::ObIArray<ObChunkDatumStore::StoredRow *> &sort_rows,

common::ObIAllocator &alloc, int64_t rows_begin,

int64_t rows_end, bool &can_encode)

{

int ret = OB_SUCCESS;

can_encode = true;

sort_rows_.set_allocator(&alloc);

if (rows_end - rows_begin <= 0) {

// do nothing

} else if (rows_begin < 0 || rows_end > sort_rows.count()) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(rows_begin), K(rows_end), K(sort_rows.count()), K(ret));

} else if (OB_FAIL(sort_rows_.prepare_allocate(rows_end - rows_begin))) {

LOG_WARN("failed to init", K(ret));

} else {

for (int64_t i = 0; can_encode && i < rows_end - rows_begin; i++) {

AQSItem &item = sort_rows_[i];

ObDatum cell = sort_rows.at(i + rows_begin)->cells()[0];

if (cell.is_null()) {

can_encode=false;

break;

}

item.key_ptr_ = (unsigned char *)cell.ptr_;

item.len_ = cell.len_;

item.row_ptr_ = sort_rows.at(i + rows_begin);

if (item.len_>0) item.sub_cache_[0] = item.key_ptr_[0];

if (item.len_>1) item.sub_cache_[1] = item.key_ptr_[1];

}

}

return ret;

}

/*

* AQS sort performance as follows:

*

* step1: | datas |<---------------------------+

* quicksort | |

* +--------------+-------------+ |

* | | | |

* less than(lt) equal to(eq) great than(gt) |

* | |

*step2: do radix sort and distribute buckets |

* for each buckets redo this process |

* |bucket0| ... |bucket255| |

* | |

* +-----------------------------------------+

*/

void ObSortOpImpl::ObAdaptiveQS::aqs_cps_qs(int64_t l, int64_t r,

int64_t common_prefix,

int64_t depth_limit,

int64_t cache_offset)

{

int64_t lt = l + 1, gt = r, m = (l - r) / 2 + r;

int64_t differ_at = INT64_MAX, lt_cp = INT64_MAX, gt_cp = INT64_MAX;

if ((r - l) < 16) {

insertion_sort(l, r, common_prefix, cache_offset);

//return;

} else {

// choose best pivot

if (compare_vals(m, l, differ_at, common_prefix, cache_offset) > 0) swap(m, l);

if (compare_vals(l, r-1, differ_at, common_prefix, cache_offset) > 0) swap(l, r-1);

if (compare_vals(m, l, differ_at, common_prefix, cache_offset) > 0) swap(m, l);

for (uint64_t i = l+1; i < gt; i++) {

int compare_res = compare_vals(i, l, differ_at, common_prefix, cache_offset);

if (compare_res < 0) {

if (i+1 < gt) {

__builtin_prefetch(sort_rows_.at(i+1).key_ptr_);

}

lt_cp = min(differ_at, lt_cp);

swap(i, lt);

lt++;

} else if (compare_res == 0) {

if (i+1 < gt) {

__builtin_prefetch(sort_rows_.at(i+1).key_ptr_);

}

} else {

gt_cp = min(differ_at, gt_cp);

gt--;

swap(i, gt);

i--;

}

}

lt--;

swap(lt, l);

depth_limit--;

if (lt != l) aqs_radix(l, lt, lt_cp, cache_offset, depth_limit);

if (gt != r) aqs_radix(gt, r, gt_cp, cache_offset, depth_limit);

}

}

void ObSortOpImpl::ObAdaptiveQS::insertion_sort(int64_t l, int64_t r,

int64_t common_prefix,

int64_t cache_offset)

{

for (int i = l + 1; i < r; i++)

{

int64_t idx = i;

int64_t differ_at = 0;

while ((idx - 1) >= l && compare_vals(idx, idx - 1, differ_at, common_prefix, cache_offset) < 0)

{

swap(idx, idx - 1);

idx--;

}

}

}

void ObSortOpImpl::ObAdaptiveQS::aqs_radix(int64_t l, int64_t r,

int64_t common_prefix,

int64_t offset,

int64_t depth_limit)

{

int more_pos = l, done_pos = l;

int cache_offset = offset;

for (int i = l; i < r; i++) {

if (sort_rows_.at(i).len_ == common_prefix) {

swap(i, more_pos);

swap(more_pos, done_pos);

more_pos++;

done_pos++;

continue;

}

/*

* Update cache policy:

* we can use following model to interpret key str:

* | common prefix | key value | remians str |

* key values means the first byte after common prefix.

*

* For the cache which size is 2, there are three scenarioes:

* 1. Values in the cache are totally ineffective

* We just upate it

* 2. only last byte of the cache is effective

* we also needs to update it, since the last byte will be used to do radix sort.

* 3. first and last byte of the cache is effective

* we do not update it, since the last byte is useful for next quick sort.

*/

// NOTES: we will use sub_cache_[0] to do radix sort

int val = common_prefix - offset;

// first and last byte of cache is effective

if (val == 0) {

// only last byte of the cache is effective

} else if (val == 1) {

sort_rows_.at(i).sub_cache_[0] = sort_rows_.at(i).sub_cache_[1];

cache_offset = common_prefix + 1;

// values in the cache are totally ineffective

} else {

unsigned char *x = ((unsigned char *)(sort_rows_.at(i).key_ptr_)) + common_prefix;

sort_rows_.at(i).sub_cache_[0] = *x;

cache_offset = common_prefix + 1;

}

if (sort_rows_.at(i).len_ == common_prefix + 1) {

swap(i, more_pos);

more_pos++;

} else {

}

}

inplace_radixsort_more_bucket(done_pos, r, 7,

common_prefix, depth_limit,

cache_offset, (common_prefix - offset) != 0);

}

// use dfs to do radix sort

// reference: https://en.wikipedia.org/wiki/Radix_sort

void ObSortOpImpl::ObAdaptiveQS::inplace_radixsort_more_bucket(int64_t l, int64_t r,

int64_t div_val,

int64_t common_prefix,

int64_t depth_limit,

int64_t cache_offset,

bool update)

{

if (l >= r || l + 1 == r) {

// do nothing

} else {

if (div_val == -1) {

int more_l = l;

for (int i = l; i < r; i++) {

if (sort_rows_.at(i).len_ == common_prefix + 1) {

swap(more_l, i);

more_l++;

}

}

// update cache

if (update) {

if(more_l < r) {

__builtin_prefetch((&sort_rows_.at(more_l).len_));

__builtin_prefetch(((unsigned char *)(sort_rows_.at(more_l).key_ptr_)) + common_prefix);

}

for (int i = more_l; i < r; i++) {

unsigned char *x = ((unsigned char *)(sort_rows_.at(i).key_ptr_)) + common_prefix;

if ( i+1 < r ) {

__builtin_prefetch((&sort_rows_.at(i+1).len_));

__builtin_prefetch(((unsigned char *)(sort_rows_.at(i+1).key_ptr_)) + common_prefix);

}

sort_rows_.at(i).sub_cache_[0] = *(x + 1);

sort_rows_.at(i).sub_cache_[1] = (common_prefix + 2 == sort_rows_.at(i).len_) ? 0x00 : *(x+2);

}

}

aqs_cps_qs(more_l, r, common_prefix + 1, depth_limit, cache_offset);

return;

}

int divide_line = l;

__builtin_prefetch((&sort_rows_.at(l).sub_cache_[0]));

for (int i = l; i < r; i++) {

// byte b = index_bytes[i];

if (i+1 < r) __builtin_prefetch((&sort_rows_.at(i+1).sub_cache_[0]));

if ((sort_rows_.at(i).sub_cache_[0] & masks[div_val]) == 0) {

swap(i, divide_line);

divide_line++;

}

}

inplace_radixsort_more_bucket(l, divide_line,

div_val - 1, common_prefix,

depth_limit, cache_offset, update);

inplace_radixsort_more_bucket(divide_line, r,

div_val - 1, common_prefix,

depth_limit, cache_offset, update);

}

}

typedef int (*CompareByteFunc)(const unsigned char *s, const unsigned char *t,

int64_t length, int64_t &differ_at, int64_t cache_ends);

extern int fast_compare_simd(const unsigned char *s, const unsigned char *t,

int64_t length, int64_t &differ_at, int64_t cache_ends);

int fast_compare_normal(const unsigned char *s, const unsigned char *t,

int64_t length, int64_t &differ_at, int64_t cache_ends)

{

int cmp_ret = 0;

for (int i = 0; (cmp_ret == 0) && i < length; i++) {

if (s[i] != t[i]) {

differ_at = i + cache_ends;

cmp_ret = s[i] - t[i];

}

}

return cmp_ret;

}

CompareByteFunc get_fast_compare_func()

{

return blocksstable::is_avx512_valid()

? fast_compare_simd

: fast_compare_normal;

}

CompareByteFunc cmp_byte_func = get_fast_compare_func();

/*

* For comparsion:

* we orgnized each entry as follows:

* | len_ | cache_ | key_ptr_ | row_ptr_ |

* | |

* key row

* we will use those entry to do comparison:

* if cache cannot distinguish those two entry, we will use key_ptr

* to index key and use key to do comparison.

*/

int ObSortOpImpl::ObAdaptiveQS::compare_cache(AQSItem &l,

AQSItem &r, int64_t &differ_at,

int64_t common_prefix,

int64_t cache_offset)

{

int64_t cache_ends = cache_offset + 2;

int64_t res = 0;

for (int64_t i = common_prefix;

res == 0 && i < cache_ends; i++) {

int64_t idx = i - cache_offset;

if (l.sub_cache_[idx] != r.sub_cache_[idx]) {

differ_at = i;

res = l.sub_cache_[idx] - r.sub_cache_[idx];

}

}

if (res != 0) {

// do nothing

} else {

unsigned char *item_b = l.key_ptr_;

unsigned char *pivot_b = r.key_ptr_;

int64_t len = min(static_cast<int64_t>(l.len_), static_cast<int64_t>(r.len_));

int64_t stride = 16;

__builtin_prefetch(pivot_b + stride);

for (int64_t j = cache_ends;

res == 0 && j < len; j += stride)

{

unsigned char *pivot_key = pivot_b + j;

unsigned char *item_key = item_b + j;

__builtin_prefetch(pivot_key + stride);

if (__builtin_expect((j + stride) > len, 0)) {

for (int i = 0; res == 0 && i < len - j; i++) {

if (pivot_key[i] != item_key[i]) {

differ_at = i + j;

res = item_key[i] - pivot_key[i];

}

}

} else {

res = cmp_byte_func(item_key, pivot_key, stride, differ_at, j);

}

}

if (res != 0) {

// do nothing

} else {

differ_at = min(static_cast<int64_t>(l.len_), static_cast<int64_t>(r.len_));

res = l.len_ - r.len_;

}

}

return res;

}

int ObSortOpImpl::ObAdaptiveQS::compare_vals(int64_t l, int64_t r,

int64_t &differ_at, int64_t common_prefix,

int64_t cache_offset) {

return compare_cache(sort_rows_.at(l), sort_rows_.at(r), differ_at, common_prefix, cache_offset);

}

ObSortOpImpl::Compare::Compare()

: ret_(OB_SUCCESS), sort_collations_(nullptr), sort_cmp_funs_(nullptr),

exec_ctx_(nullptr), cmp_count_(0), cmp_start_(0), cmp_end_(0)

{

}

int ObSortOpImpl::Compare::init(

const ObIArray<ObSortFieldCollation> *sort_collations,

const ObIArray<ObSortCmpFunc> *sort_cmp_funs,

ObExecContext *exec_ctx,

bool enable_encode_sortkey)

{

int ret = OB_SUCCESS;

if (nullptr == sort_collations || nullptr == sort_cmp_funs || nullptr == exec_ctx) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(ret), KP(sort_collations), KP(sort_cmp_funs));

} else if (sort_cmp_funs->count() != sort_cmp_funs->count()) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("column count miss match", K(ret),

K(sort_cmp_funs->count()), K(sort_cmp_funs->count()));

} else {

sort_collations_ = sort_collations;

sort_cmp_funs_ = sort_cmp_funs;

exec_ctx_ = exec_ctx;

cnt_ = sort_cmp_funs_->count();

cmp_start_ = 0;

cmp_end_ = sort_cmp_funs_->count();

enable_encode_sortkey_ = enable_encode_sortkey;

}

return ret;

}

int ObSortOpImpl::Compare::fast_check_status()

{

int ret = OB_SUCCESS;

if (OB_UNLIKELY((cmp_count_++ & 8191) == 8191)) {

ret = exec_ctx_->check_status();

}

return ret;

}

bool ObSortOpImpl::Compare::operator()(

const ObChunkDatumStore::StoredRow *l,

const ObChunkDatumStore::StoredRow *r)

{

bool less = false;

int &ret = ret_;

if (OB_UNLIKELY(OB_SUCCESS != ret)) {

// already fail

} else if (!is_inited() || OB_ISNULL(l) || OB_ISNULL(r)) {

ret = !is_inited() ? OB_NOT_INIT : OB_INVALID_ARGUMENT;

LOG_WARN("not init or invalid argument", K(ret), KP(l), KP(r));

} else if (OB_FAIL(fast_check_status())) {

LOG_WARN("fast check failed", K(ret));

} else if (enable_encode_sortkey_) {

const ObDatum l_cell = l->cells()[0];

const ObDatum r_cell = r->cells()[0];

int cmp = 0;

cmp = MEMCMP(l_cell.ptr_, r_cell.ptr_, min(l_cell.len_, r_cell.len_));

less = cmp != 0 ? (cmp < 0) : (l_cell.len_ - r_cell.len_) < 0;

} else {

const ObDatum *lcells = l->cells();

const ObDatum *rcells = r->cells();

int cmp = 0;

for (int64_t i = cmp_start_; 0 == cmp && i < cmp_end_ && OB_SUCC(ret); i++) {

const ObSortFieldCollation& sort_collation = sort_collations_->at(i);

const int64_t idx = sort_collation.field_idx_;

if (OB_FAIL(sort_cmp_funs_->at(i).cmp_func_(lcells[idx], rcells[idx], cmp))) {

LOG_WARN("failed to compare", K(ret));

} else if (cmp < 0) {

less = sort_collation.is_ascending_;

} else if (cmp > 0) {

less = !sort_collation.is_ascending_;

}

}

}

return less;

}

bool ObSortOpImpl::Compare::operator()(

const common::ObIArray<ObExpr*> *l,

const ObChunkDatumStore::StoredRow *r,

ObEvalCtx &eval_ctx)

{

bool less = false;

int &ret = ret_;

if (OB_UNLIKELY(OB_SUCCESS != ret)) {

// already fail

} else if (!is_inited() || OB_ISNULL(l) || OB_ISNULL(r)) {

ret = !is_inited() ? OB_NOT_INIT : OB_INVALID_ARGUMENT;

LOG_WARN("not init or invalid argument", K(ret), KP(l), KP(r));

} else if (OB_FAIL(fast_check_status())) {

LOG_WARN("fast check failed", K(ret));

} else {

const ObDatum *rcells = r->cells();

ObDatum *other_datum = nullptr;

int cmp = 0;

const int64_t cnt = sort_cmp_funs_->count();

for (int64_t i = 0; 0 == cmp && i < cnt && OB_SUCC(ret); i++) {

const int64_t idx = sort_collations_->at(i).field_idx_;

if (OB_FAIL(l->at(idx)->eval(eval_ctx, other_datum))) {

LOG_WARN("failed to eval expr", K(ret));

} else if (OB_FAIL(sort_cmp_funs_->at(i).cmp_func_(*other_datum, rcells[idx], cmp))) {

LOG_WARN("failed to compare", K(ret));

} else {

if (cmp < 0) {

less = sort_collations_->at(i).is_ascending_;

} else if (cmp > 0) {

less = !sort_collations_->at(i).is_ascending_;

}

}

}

}

return less;

}

int ObSortOpImpl::Compare::with_ties_cmp(const common::ObIArray<ObExpr*> *l,

const ObChunkDatumStore::StoredRow *r,

ObEvalCtx &eval_ctx)

{

int cmp = 0;

int &ret = ret_;

if (OB_UNLIKELY(OB_SUCCESS != ret)) {

// already fail

} else if (!is_inited() || OB_ISNULL(l) || OB_ISNULL(r)) {

ret = !is_inited() ? OB_NOT_INIT : OB_INVALID_ARGUMENT;

LOG_WARN("not init or invalid argument", K(ret), KP(l), KP(r));

} else {

const ObDatum *rcells = r->cells();

ObDatum *other_datum = nullptr;

const int64_t cnt = sort_cmp_funs_->count();

for (int64_t i = 0; 0 == cmp && i < cnt && OB_SUCC(ret); i++) {

const int64_t idx = sort_collations_->at(i).field_idx_;

if (OB_FAIL(l->at(idx)->eval(eval_ctx, other_datum))) {

LOG_WARN("failed to eval expr", K(ret));

} else if (OB_FAIL(sort_cmp_funs_->at(i).cmp_func_(*other_datum, rcells[idx], cmp))) {

LOG_WARN("failed to compare", K(ret));

} else {

cmp = sort_collations_->at(i).is_ascending_ ? -cmp : cmp;

}

}

}

return cmp;

}

int ObSortOpImpl::Compare::with_ties_cmp(const ObChunkDatumStore::StoredRow *l,

const ObChunkDatumStore::StoredRow *r)

{

int cmp = 0;

int &ret = ret_;

if (OB_UNLIKELY(OB_SUCCESS != ret)) {

// already fail

} else if (!is_inited() || OB_ISNULL(l) || OB_ISNULL(r)) {

ret = !is_inited() ? OB_NOT_INIT : OB_INVALID_ARGUMENT;

LOG_WARN("not init or invalid argument", K(ret), KP(l), KP(r));

} else {

const ObDatum *rcells = r->cells();

const ObDatum *lcells = l->cells();

const int64_t cnt = sort_cmp_funs_->count();

for (int64_t i = 0; 0 == cmp && i < cnt && OB_SUCC(ret); i++) {

const int64_t idx = sort_collations_->at(i).field_idx_;

if (OB_FAIL(sort_cmp_funs_->at(i).cmp_func_(lcells[idx], rcells[idx], cmp))) {

LOG_WARN("failed to compare", K(ret));

} else {

cmp = sort_collations_->at(i).is_ascending_ ? -cmp : cmp;

}

}

}

return cmp;

}

// compare function for external merge sort

bool ObSortOpImpl::Compare::operator()(const ObSortOpChunk *l, const ObSortOpChunk *r)

{

bool less = false;

int &ret = ret_;

if (OB_UNLIKELY(OB_SUCCESS != ret)) {

// already fail

} else if (OB_ISNULL(l) || OB_ISNULL(r)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(ret), KP(l), KP(r));

} else {

// Return the reverse order since the heap top is the maximum element.

// NOTE: can not return !(*this)(l->row_, r->row_)

// because we should always return false if l == r.

less = (*this)(r->row_, l->row_);

}

return less;

}

bool ObSortOpImpl::Compare::operator()(

ObChunkDatumStore::StoredRow **l,

ObChunkDatumStore::StoredRow **r)

{

bool less = false;

int &ret = ret_;

if (OB_UNLIKELY(OB_SUCCESS != ret)) {

// already fail

} else if (OB_ISNULL(l) || OB_ISNULL(r)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(ret), KP(l), KP(r));

} else {

// Return the reverse order since the heap top is the maximum element.

// NOTE: can not return !(*this)(l->row_, r->row_)

// because we should always return false if l == r.

less = (*this)(*r, *l);

}

return less;

}

ObSortOpImpl::ObSortOpImpl(ObMonitorNode &op_monitor_info)

: inited_(false), local_merge_sort_(false), need_rewind_(false),

got_first_row_(false), sorted_(false), enable_encode_sortkey_(false), mem_context_(NULL),

mem_entify_guard_(mem_context_), tenant_id_(OB_INVALID_ID), sort_collations_(nullptr),

sort_cmp_funs_(nullptr), eval_ctx_(nullptr), datum_store_(ObModIds::OB_SQL_SORT_ROW), inmem_row_size_(0), mem_check_interval_mask_(1),

row_idx_(0), heap_iter_begin_(false), imms_heap_(NULL), ems_heap_(NULL),

next_stored_row_func_(&ObSortOpImpl::array_next_stored_row),

input_rows_(OB_INVALID_ID), input_width_(OB_INVALID_ID),

profile_(ObSqlWorkAreaType::SORT_WORK_AREA), op_monitor_info_(op_monitor_info), sql_mem_processor_(profile_, op_monitor_info_),

op_type_(PHY_INVALID), op_id_(UINT64_MAX), exec_ctx_(nullptr), stored_rows_(nullptr),

io_event_observer_(nullptr), buckets_(NULL), max_bucket_cnt_(0), part_hash_nodes_(NULL),

max_node_cnt_(0), part_cnt_(0), topn_cnt_(INT64_MAX), outputted_rows_cnt_(0),

is_fetch_with_ties_(false), topn_heap_(NULL), ties_array_pos_(0), ties_array_(),

last_ties_row_(NULL), rows_(NULL)

{

}

ObSortOpImpl::~ObSortOpImpl()

{

reset();

}

// Set the note in ObPrefixSortImpl::init(): %sort_columns may be zero, to compatible with

// the wrong generated prefix sort.

int ObSortOpImpl::init(

const uint64_t tenant_id,

const ObIArray<ObSortFieldCollation> *sort_collations,

const ObIArray<ObSortCmpFunc> *sort_cmp_funs,

ObEvalCtx *eval_ctx,

ObExecContext *exec_ctx,

const bool enable_encode_sortkey /* = false*/,

const bool in_local_order /* = false */,

const bool need_rewind /* = false */,

const int64_t part_cnt /* = 0 */,

const int64_t topn_cnt /* = INT64_MAX */,

const bool is_fetch_with_ties /* = false */,

const int64_t default_block_size /* = 64KB */)

{

int ret = OB_SUCCESS;

if (is_inited()) {

ret = OB_INIT_TWICE;

LOG_WARN("init twice", K(ret));

} else if (OB_INVALID_ID == tenant_id) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument", K(ret), K(tenant_id));

} else if (OB_ISNULL(sort_collations) || OB_ISNULL(sort_cmp_funs)

|| OB_ISNULL(eval_ctx) || OB_ISNULL(exec_ctx)) {

ret = OB_INVALID_ARGUMENT;

LOG_WARN("invalid argument: argument is null", K(ret),

K(tenant_id), K(sort_collations), K(sort_cmp_funs), K(eval_ctx));

} else if (OB_FAIL(comp_.init(sort_collations, sort_cmp_funs,

exec_ctx, enable_encode_sortkey && !(part_cnt > 0)))) {

LOG_WARN("failed to init compare functions", K(ret));

} else {

local_merge_sort_ = in_local_order;

need_rewind_ = need_rewind;

enable_encode_sortkey_ = enable_encode_sortkey;

tenant_id_ = tenant_id;

sort_collations_ = sort_collations;

sort_cmp_funs_ = sort_cmp_funs;

eval_ctx_ = eval_ctx;

exec_ctx_ = exec_ctx;

part_cnt_ = part_cnt;

topn_cnt_ = topn_cnt;

use_heap_sort_ = is_topn_sort();

is_fetch_with_ties_ = is_fetch_with_ties;

int64_t batch_size = eval_ctx_->max_batch_size_;

lib::ContextParam param;

param.set_mem_attr(tenant_id, ObModIds::OB_SQL_SORT_ROW, ObCtxIds::WORK_AREA)

.set_properties(lib::USE_TL_PAGE_OPTIONAL);

if (NULL == mem_context_ && OB_FAIL(CURRENT_CONTEXT->CREATE_CONTEXT(mem_context_, param))) {

LOG_WARN("create entity failed", K(ret));

} else if (NULL == mem_context_) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("null memory entity returned", K(ret));

} else if (OB_FAIL(datum_store_.init(

INT64_MAX /* mem limit, big enough to hold all rows in memory */,

tenant_id_, ObCtxIds::WORK_AREA, ObModIds::OB_SQL_SORT_ROW,

false /*+ disable dump */,

0, /* row_extra_size */

default_block_size))) {

LOG_WARN("init row store failed", K(ret));

} else if (is_topn_sort()

&& OB_ISNULL(topn_heap_ = OB_NEWx(TopnHeap, (&mem_context_->get_malloc_allocator()),

comp_, &mem_context_->get_malloc_allocator()))) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("allocate memory failed", K(ret));

} else if (batch_size > 0

&& OB_ISNULL(stored_rows_ = static_cast<ObChunkDatumStore::StoredRow **>(

mem_context_->get_malloc_allocator().alloc(

sizeof(*stored_rows_) * batch_size)))) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("allocate memory failed", K(ret));

} else {

quick_sort_array_.set_block_allocator(

ModulePageAllocator(mem_context_->get_malloc_allocator(), "SortOpRows"));

datum_store_.set_dir_id(sql_mem_processor_.get_dir_id());

datum_store_.set_allocator(mem_context_->get_malloc_allocator());

datum_store_.set_io_event_observer(io_event_observer_);

profile_.set_exec_ctx(exec_ctx);

op_monitor_info_.otherstat_2_id_ = ObSqlMonitorStatIds::SORT_MERGE_SORT_ROUND;

op_monitor_info_.otherstat_2_value_ = 1;

ObPhysicalPlanCtx *plan_ctx = NULL;

const ObPhysicalPlan *phy_plan = nullptr;

if (OB_ISNULL(plan_ctx = GET_PHY_PLAN_CTX(*exec_ctx))) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("deserialized exec ctx without phy plan ctx set. Unexpected", K(ret));

} else if (OB_ISNULL(phy_plan = plan_ctx->get_phy_plan())) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("error unexpected, phy plan must not be nullptr", K(ret));

} else if (phy_plan->get_ddl_task_id() > 0) {

op_monitor_info_.otherstat_5_id_ = ObSqlMonitorStatIds::DDL_TASK_ID;

op_monitor_info_.otherstat_5_value_ = phy_plan->get_ddl_task_id();

}

}

if (OB_SUCC(ret)) {

inited_ = true;

if (!is_topn_sort()) {

rows_ = &quick_sort_array_;

} else {

rows_ = &(const_cast<common::ObIArray<ObChunkDatumStore::StoredRow *> &>

(topn_heap_->get_heap_data()));

}

}

}

return ret;

}

void ObSortOpImpl::reuse()

{

sorted_ = false;

iter_.reset();

quick_sort_array_.reuse();

datum_store_.reset();

inmem_row_size_ = 0;

mem_check_interval_mask_ = 1;

row_idx_ = 0;

next_stored_row_func_ = &ObSortOpImpl::array_next_stored_row;

ties_array_pos_ = 0;

if (0 != ties_array_.count()) {

for (int64_t i = 0; i < ties_array_.count(); ++i) {

sql_mem_processor_.alloc(-1 * ties_array_[i]->get_max_size());

mem_context_->get_malloc_allocator().free(ties_array_[i]);

ties_array_[i] = NULL;

}

}

ties_array_.reset();

while (!sort_chunks_.is_empty()) {

ObSortOpChunk *chunk = sort_chunks_.remove_first();

chunk->~ObSortOpChunk();

if (NULL != mem_context_) {

mem_context_->get_malloc_allocator().free(chunk);

}

}

if (NULL != imms_heap_) {

imms_heap_->reset();

}

heap_iter_begin_ = false;

if (NULL != ems_heap_) {

ems_heap_->reset();

}

if (NULL != topn_heap_) {

for (int64_t i = 0; i < topn_heap_->count(); ++i) {

sql_mem_processor_.alloc(-1 *

static_cast<SortStoredRow *>(topn_heap_->at(i))->get_max_size());

mem_context_->get_malloc_allocator().free(static_cast<SortStoredRow *>(topn_heap_->at(i)));

topn_heap_->at(i) = NULL;

}

topn_heap_->reset();

}

}

void ObSortOpImpl::unregister_profile()

{

sql_mem_processor_.unregister_profile();

}

void ObSortOpImpl::unregister_profile_if_necessary()

{

sql_mem_processor_.unregister_profile_if_necessary();

}

void ObSortOpImpl::reset()

{

sql_mem_processor_.unregister_profile();

iter_.reset();

reuse();

quick_sort_array_.reset();

datum_store_.reset();

inmem_row_size_ = 0;

local_merge_sort_ = false;

need_rewind_ = false;

sorted_ = false;

got_first_row_ = false;

comp_.reset();

max_bucket_cnt_ = 0;

max_node_cnt_ = 0;

part_cnt_ = 0;

topn_cnt_ = INT64_MAX;

outputted_rows_cnt_ = 0;

is_fetch_with_ties_ = false;

rows_ = NULL;

ties_array_pos_ = 0;

if (0 != ties_array_.count()) {

for (int64_t i = 0; i < ties_array_.count(); ++i) {

mem_context_->get_malloc_allocator().free(ties_array_[i]);

ties_array_[i] = NULL;

}

}

ties_array_.reset();

if (NULL != mem_context_) {

if (NULL != imms_heap_) {

imms_heap_->~IMMSHeap();

mem_context_->get_malloc_allocator().free(imms_heap_);

imms_heap_ = NULL;

}

if (NULL != ems_heap_) {

ems_heap_->~EMSHeap();

mem_context_->get_malloc_allocator().free(ems_heap_);

ems_heap_ = NULL;

}

if (NULL != stored_rows_) {

mem_context_->get_malloc_allocator().free(stored_rows_);

stored_rows_ = NULL;

}

if (NULL != buckets_) {

mem_context_->get_malloc_allocator().free(buckets_);

buckets_ = NULL;

}

if (NULL != part_hash_nodes_) {

mem_context_->get_malloc_allocator().free(part_hash_nodes_);

part_hash_nodes_ = NULL;

}

if (NULL != topn_heap_) {

for (int64_t i = 0; i < topn_heap_->count(); ++i) {

mem_context_->get_malloc_allocator().free(static_cast<SortStoredRow *>(topn_heap_->at(i)));

topn_heap_->at(i) = NULL;

}

topn_heap_->~TopnHeap();

mem_context_->get_malloc_allocator().free(topn_heap_);

topn_heap_ = NULL;

}

if (NULL != last_ties_row_) {

mem_context_->get_malloc_allocator().free(last_ties_row_);

last_ties_row_ = NULL;

}

// can not destroy mem_entify here, the memory may hold by %iter_ or %datum_store_

}

inited_ = false;

io_event_observer_ = nullptr;

}

template <typename Input>

int ObSortOpImpl::build_chunk(const int64_t level, Input &input, int64_t extra_size)

{

int ret = OB_SUCCESS;

const int64_t curr_time = ObTimeUtility::fast_current_time();

int64_t stored_row_cnt = 0;

ObChunkDatumStore *datum_store = NULL;

const ObChunkDatumStore::StoredRow *src_store_row = NULL;

ObChunkDatumStore::StoredRow *dst_store_row = NULL;

ObSortOpChunk *chunk = NULL;

if (!is_inited()) {

ret = OB_NOT_INIT;

LOG_WARN("not init", K(ret));

} else if (OB_ISNULL(chunk = OB_NEWx(ObSortOpChunk,

(&mem_context_->get_malloc_allocator()), level))) {

ret = OB_ALLOCATE_MEMORY_FAILED;

LOG_WARN("allocate memory failed", K(ret));

} else if (OB_FAIL(chunk->datum_store_.init(1/*+ mem limit, small limit for dump immediately */,

tenant_id_, ObCtxIds::WORK_AREA, ObModIds::OB_SQL_SORT_ROW,

true/*+ enable dump */, extra_size/* for InMemoryTopnSort */))) {

LOG_WARN("init row store failed", K(ret));

} else {

chunk->datum_store_.set_dir_id(sql_mem_processor_.get_dir_id());

chunk->datum_store_.set_allocator(mem_context_->get_malloc_allocator());

chunk->datum_store_.set_callback(&sql_mem_processor_);

chunk->datum_store_.set_io_event_observer(io_event_observer_);

while (OB_SUCC(ret)) {

if (!is_fetch_with_ties_ && stored_row_cnt >= topn_cnt_) {

break;

} else if (OB_FAIL(input(datum_store, src_store_row))) {

if (OB_ITER_END != ret) {

LOG_WARN("get input row failed", K(ret));

} else {

ret = OB_SUCCESS;

}

break;

} else if (OB_FAIL(chunk->datum_store_.add_row(*src_store_row, &dst_store_row))) {

LOG_WARN("copy row to row store failed");

} else {

stored_row_cnt++;

op_monitor_info_.otherstat_1_id_ = ObSqlMonitorStatIds::SORT_SORTED_ROW_COUNT;

op_monitor_info_.otherstat_1_value_ += 1;

}

}

// 必须强制先dump，然后finish dump才有效

if (OB_FAIL(ret)) {

} else if (OB_FAIL(chunk->datum_store_.dump(false, true))) {

LOG_WARN("failed to dump row store", K(ret));

} else if (OB_FAIL(chunk->datum_store_.finish_add_row(true/*+ need dump */))) {

LOG_WARN("finish add row failed", K(ret));

} else {

const int64_t sort_io_time = ObTimeUtility::fast_current_time() - curr_time;

op_monitor_info_.otherstat_4_id_ = ObSqlMonitorStatIds::SORT_DUMP_DATA_TIME;

op_monitor_info_.otherstat_4_value_ += sort_io_time;

LOG_TRACE("dump sort file",

"level", level,

"rows", chunk->datum_store_.get_row_cnt(),

"file_size", chunk->datum_store_.get_file_size(),

"memory_hold", chunk->datum_store_.get_mem_hold(),

"mem_used", mem_context_->used());

}

}

if (OB_SUCC(ret)) {

// In increase sort, chunk->level_ may less than the last of sort chunks.

// insert the chunk to the upper bound the level.

ObSortOpChunk *pos = sort_chunks_.get_last();

for ( ; pos != sort_chunks_.get_header() && pos->level_ > level; pos = pos->get_prev()) {

}

pos = pos->get_next();

if (!sort_chunks_.add_before(pos, chunk)) {

ret = OB_ERR_UNEXPECTED;

LOG_WARN("add link node to list failed", K(ret));

}

}

if (OB_SUCCESS != ret && NULL != chunk) {

chunk->~ObSortOpChunk();

mem_context_->get_malloc_allocator().free(chunk);

chunk = NULL;

}

return ret;

}

// 如果发现需要dump，则

// 1 重新获取可用内存大小

// 2 检查是否还需要dump

// 3 如果需要dump，分三种情况

// 3.0 cache_size <= mem_bound 全内存(这里表示之前预估不准确，同时有足够内存可用)

// 申请是否有更多内存可用，决定是否需要dump

// 3.0.1 申请内存大于等于cache size，则不dump

// 3.0.2 申请内存小于cache size，则dump，返回的是算one-pass size

// 3.1 未超过cache size，则直接dump

// 3.2 超过了cache size，则采用2*size方式申请内存，one-pass内存

// 然后继续，和之前逻辑一样

// 所以这里会导致最开始dump的partition one-pass内存较少，后面倍数cache size关系的one-pass更大

int ObSortOpImpl::preprocess_dump(bool &dumped)

{

int ret = OB_SUCCESS;

dumped = false;

if (OB_FAIL(sql_mem_processor_.get_max_available_mem_size(

&mem_context_->get_malloc_allocator()))) {

LOG_WARN("failed to get max available memory size", K(ret));

} else if (OB_FAIL(sql_mem_processor_.update_used_mem_size(mem_context_->used()))) {

LOG_WARN("failed to update used memory size", K(ret));

} else {

dumped = need_dump();

if (dumped) {

if (!sql_mem_processor_.is_auto_mgr()) {

// 如果dump在非auto管理模式也需要注册到workarea

if (OB_FAIL(sql_mem_processor_.extend_max_memory_size(

&mem_context_->get_malloc_allocator(),

[&](int64_t max_memory_size) {

UNUSED(max_memory_size);

return need_dump();

},

dumped, mem_context_->used()))) {

LOG_WARN("failed to extend memory size", K(ret));

}

} else if (profile_.get_cache_size() < profile_.get_global_bound_size()) {

// in-memory：所有数据都可以缓存，即global bound size比较大，则继续看是否有更多内存可用

if (OB_FAIL(sql_mem_processor_.extend_max_memory_size(

&mem_context_->get_malloc_allocator(),

[&](int64_t max_memory_size) {

UNUSED(max_memory_size);

return need_dump();

},

dumped, mem_context_->used()))) {

LOG_WARN("failed to extend memory size", K(ret));

}

LOG_TRACE("trace sort need dump", K(dumped), K(mem_context_->used()),

K(get_memory_limit()), K(profile_.get_cache_size()), K(profile_.get_expect_size()));

} else {

// one-pass

if (profile_.get_cache_size() <=

datum_store_.get_mem_hold() + datum_store_.get_file_size()) {

// 总体数据量超过cache size，说明估算的cache不准确，需要重新估算one-pass size，按照2*cache_size处理

if (OB_FAIL(sql_mem_processor_.update_cache_size(&mem_context_->get_malloc_allocator(),

profile_.get_cache_size() * EXTEND_MULTIPLE))) {

LOG_WARN("failed to update cache size", K(ret), K(profile_.get_cache_size()));

} else {

dumped = need_dump();

}

} else { }

}

LOG_INFO("trace sort need dump", K(dumped), K(mem_context_->used()), K(get_memory_limit()),

K(profile_.get_cache_size()), K(profile_.get_expect_size()),

K(sql_mem_processor_.get_data_size()));

}

}

return ret;

}

int ObSortOpImpl::before_add_row()

{

int ret = OB_SUCCESS;

int64_t sort_force_dump_rows = - EVENT_CALL(EventTable::EN_SORT_IMPL_FORCE_DO_DUMP);

if (!is_inited()) {

ret = OB_NOT_INIT;

LOG_WARN("not init", K(ret));

} else if (OB_UNLIKELY(!got_first_row_)) {

if (!comp_.is_inited() && OB_FAIL(comp_.init(sort_collations_, sort_cmp_funs_,

exec_ctx_, enable_encode_sortkey_ && !(part_cnt_ > 0)))) {

LOG_WARN("init compare failed", K(ret));

} else {

got_first_row_ = true;

int64_t size = OB_INVALID_ID == input_rows_ ? 0 : input_rows_ * input_width_;

if (OB_FAIL(sql_mem_processor_.init(

&mem_context_->get_malloc_allocator(),

tenant_id_,

size, op_monitor_info_.op_type_, op_monitor_info_.op_id_, exec_ctx_))) {

LOG_WARN("failed to init sql mem processor", K(ret));

} else {

datum_store_.set_dir_id(sql_mem_processor_.get_dir_id());

datum_store_.set_callback(&sql_mem_processor_);

datum_store_.set_io_event_observer(io_event_observer_);

}

}

}