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

[CP] [CP] [CP] Fix rollup SIGSEGV when building stored row

Browse Source
This commit is contained in:
stdliu
2022-06-06 14:20:53 +08:00
committed by wangzelin.wzl
parent bc26b2f163
commit d1636b9051
24 changed files with 184 additions and 324 deletions
Download Patch File Download Diff File Expand all files Collapse all files

202
src/sql/engine/basic/ob_chunk_datum_store.cpp
View File

@ -57,113 +57,6 @@ void point2pointer(T*& dst_pointer, B* dst_base, T* src_pointer, const B* src_ba
dst_pointer = reinterpret_cast<T*>(reinterpret_cast<char*>(dst_base) + reinterpret_cast<intptr_t>(src_pointer) -
reinterpret_cast<const char*>(src_base));
}
} // namespace chunk_datum_store
int ObChunkDatumStore::StoredRow::copy_datums(const common::ObIArray<ObExpr*>& exprs, ObEvalCtx& ctx, char* buf,
const int64_t size, const int64_t row_size, const uint32_t row_extend_size)
{
int ret = OB_SUCCESS;
if (payload_ != buf || size < 0) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), KP(payload_), KP(buf), K(size));
} else {
cnt_ = static_cast<uint32_t>(exprs.count());
int64_t pos = sizeof(ObDatum) * cnt_ + row_extend_size;
row_size_ = static_cast<int32_t>(row_size);
for (int64_t i = 0; OB_SUCC(ret) && i < cnt_; ++i) {
ObDatum& in_datum = static_cast<ObDatum&>(exprs.at(i)->locate_expr_datum(ctx));
ObDatum* datum = new (&cells()[i]) ObDatum();
if (OB_FAIL(datum->deep_copy(in_datum, buf, size, pos))) {
LOG_WARN("failed to copy datum", K(ret), K(i), K(pos), K(size), K(row_size), K(in_datum), K(*datum));
} else {
LOG_DEBUG("succ to copy_datums", K(cnt_), K(i), K(size), K(row_size), K(in_datum), K(*datum));
}
}
}
return ret;
}
int ObChunkDatumStore::StoredRow::copy_datums(const common::ObIArray<ObExpr*>& exprs, ObEvalCtx& ctx, int64_t& row_size,
char* buf, const int64_t max_buf_size, const uint32_t row_extend_size)
{
int ret = OB_SUCCESS;
if (max_buf_size < 0) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), KP(payload_), KP(buf), K(max_buf_size));
} else {
cnt_ = static_cast<uint32_t>(exprs.count());
int64_t pos = sizeof(ObDatum) * cnt_ + row_extend_size + sizeof(StoredRow);
ObDatum* datums = cells();
if (pos > max_buf_size) {
ret = OB_BUF_NOT_ENOUGH;
} else {
for (int64_t i = 0; OB_SUCC(ret) && i < cnt_; ++i) {
ObDatum* in_datum = nullptr;
ObDatum* datum = (ObDatum*)datums;
if (OB_FAIL(exprs.at(i)->eval(ctx, in_datum))) {
LOG_WARN("failed to eval datum", K(ret));
} else if (OB_FAIL(datum->deep_copy(*in_datum, buf, max_buf_size, pos))) {
if (OB_BUF_NOT_ENOUGH != ret) {
LOG_WARN("failed to copy datum", K(ret), K(i), K(pos), K(max_buf_size), KP(in_datum), K(*datum));
}
} else {
LOG_DEBUG("succ to copy_datums", K(cnt_), K(i), K(max_buf_size), KP(in_datum), K(*datum));
}
++datums;
}
row_size_ = static_cast<int32_t>(pos);
row_size = row_size_;
}
}
return ret;
}
int ObChunkDatumStore::StoredRow::copy_datums(common::ObDatum** datums, const int64_t cnt, char* buf,
const int64_t size, const int64_t row_size, const uint32_t row_extend_size)
{
int ret = OB_SUCCESS;
if (payload_ != buf || size < 0 || nullptr == datums) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), KP(payload_), KP(buf), K(size), K(datums));
} else {
cnt_ = static_cast<uint32_t>(cnt);
int64_t pos = sizeof(ObDatum) * cnt_ + row_extend_size;
row_size_ = static_cast<int32_t>(row_size);
for (int64_t i = 0; OB_SUCC(ret) && i < cnt_; ++i) {
if (nullptr == datums[i]) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), KP(payload_), KP(buf), K(size), K(datums));
} else {
ObDatum* datum = new (&cells()[i]) ObDatum();
if (OB_FAIL(datum->deep_copy(*datums[i], buf, size, pos))) {
LOG_WARN(
"failed to copy datum", K(ret), K(i), K(pos), K(size), K(row_size), K(*datums[i]), K(datums[i]->len_));
}
}
}
}
return ret;
}
int ObChunkDatumStore::StoredRow::copy_datums(common::ObDatum* datums, const int64_t cnt, char* buf, const int64_t size,
const int64_t row_size, const uint32_t row_extend_size)
{
int ret = OB_SUCCESS;
if (payload_ != buf || size < 0 || nullptr == datums) {
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), KP(payload_), KP(buf), K(size), K(datums));
} else {
cnt_ = static_cast<uint32_t>(cnt);
row_size_ = static_cast<int32_t>(row_size);
MEMCPY(buf, static_cast<const void*>(datums), size);
int64_t pos = sizeof(ObDatum) * cnt_ + row_extend_size;
for (int64_t i = 0; i < cnt_; ++i) {
cells()[i].ptr_ = buf + pos;
pos += cells()[i].len_;
}
}
return ret;
}
int ObChunkDatumStore::StoredRow::to_expr(const common::ObIArray<ObExpr*>& exprs, ObEvalCtx& ctx) const
@ -230,6 +123,54 @@ void ObChunkDatumStore::StoredRow::swizzling(char* base /*= NULL*/)
}
}
int ObChunkDatumStore::StoredRow::build(StoredRow *&sr, const ObExprPtrIArray &exprs, ObEvalCtx &ctx, char *buf,
const int64_t buf_len, const uint32_t extra_size /* = 0 */)
{
int ret = OB_SUCCESS;
sr = reinterpret_cast<StoredRow *>(buf);
int64_t pos = sizeof(*sr) + sizeof(ObDatum) * exprs.count() + extra_size;
if (pos > buf_len) {
ret = OB_BUF_NOT_ENOUGH;
} else {
sr->cnt_ = exprs.count();
ObDatum *datums = sr->cells();
for (int64_t i = 0; i < exprs.count() && OB_SUCC(ret); i++) {
ObExpr *expr = exprs.at(i);
ObDatum *in_datum = NULL;
if (OB_UNLIKELY(NULL == expr)) {
// Set datum to NULL for NULL expr
datums[i].set_null();
} else if (OB_FAIL(expr->eval(ctx, in_datum))) {
LOG_WARN("expression evaluate failed", K(ret));
} else {
ret = datums[i].deep_copy(*in_datum, buf, buf_len, pos);
}
}
if (OB_SUCC(ret)) {
sr->row_size_ = static_cast<int32_t>(pos);
}
}
return ret;
}
int ObChunkDatumStore::StoredRow::build(StoredRow *&sr, const ObExprPtrIArray &exprs, ObEvalCtx &ctx,
common::ObIAllocator &alloc, const uint32_t extra_size /* = 0 */)
{
int ret = OB_SUCCESS;
int64_t size = 0;
char *buf = NULL;
if (OB_FAIL(Block::row_store_size(exprs, ctx, size, extra_size))) {
LOG_WARN("get row store size failed", K(ret));
} else if (NULL == (buf = static_cast<char *>(alloc.alloc(size)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
LOG_WARN("allocate memory failed", K(ret), K(size));
} else if (OB_FAIL(build(sr, exprs, ctx, buf, size, extra_size))) {
LOG_WARN("build stored row failed", K(ret));
}
return ret;
}
int ObChunkDatumStore::Block::add_row(const common::ObIArray<ObExpr*>& exprs, ObEvalCtx& ctx, const int64_t row_size,
uint32_t row_extend_size, StoredRow** stored_row)
{
@ -242,18 +183,15 @@ int ObChunkDatumStore::Block::add_row(const common::ObIArray<ObExpr*>& exprs, Ob
ret = OB_BUF_NOT_ENOUGH;
LOG_WARN("buffer not enough", K(row_size), "remain", buf->remain());
} else {
StoredRow* sr = (StoredRow*)buf->head();
if (OB_FAIL(sr->copy_datums(
exprs, ctx, buf->head() + ROW_HEAD_SIZE, row_size - ROW_HEAD_SIZE, row_size, row_extend_size))) {
LOG_WARN("copy row failed", K(ret), K(row_size));
StoredRow *sr = NULL;
if (OB_FAIL(StoredRow::build(sr, exprs, ctx, buf->head(), row_size, row_extend_size))) {
LOG_WARN("build stored row failed", K(ret));
} else if (OB_FAIL(buf->advance(row_size))) {
LOG_WARN("fill buffer head failed", K(ret), K(buf), K(row_size));
} else {
if (OB_FAIL(buf->advance(row_size))) {
LOG_WARN("fill buffer head failed", K(ret), K(buf), K(row_size));
} else {
rows_++;
if (NULL != stored_row) {
*stored_row = sr;
}
rows_++;
if (NULL != stored_row) {
*stored_row = sr;
}
}
}
@ -268,25 +206,17 @@ int ObChunkDatumStore::Block::append_row(const common::ObIArray<ObExpr*>& exprs,
ret = OB_INVALID_ARGUMENT;
LOG_WARN("invalid argument", K(ret), K(buf));
} else {
int64_t max_size = buf->remain();
if (ROW_HEAD_SIZE > max_size) {
ret = OB_BUF_NOT_ENOUGH;
StoredRow *sr = NULL;
if (OB_FAIL(StoredRow::build(sr, exprs, *ctx, buf->head(), buf->remain(), row_extend_size))) {
if (OB_BUF_NOT_ENOUGH != ret) {
LOG_WARN("build stored row failed", K(ret));
}
} else if (OB_FAIL(buf->advance(sr->row_size_))) {
LOG_WARN("buffer advance failed", K(ret));
} else {
StoredRow* sr = (StoredRow*)buf->head();
int64_t row_size = 0;
if (OB_FAIL(sr->copy_datums(exprs, *ctx, row_size, buf->head(), max_size, row_extend_size))) {
if (OB_BUF_NOT_ENOUGH != ret) {
LOG_WARN("copy row failed", K(ret), K(max_size));
}
} else {
if (OB_FAIL(buf->advance(row_size))) {
LOG_WARN("fill buffer head failed", K(ret), K(buf));
} else {
++rows_;
if (NULL != stored_row) {
*stored_row = sr;
}
}
++rows_;
if (NULL != stored_row) {
*stored_row = sr;
}
}
}

137
src/sql/engine/basic/ob_chunk_datum_store.h
View File

@ -34,36 +34,6 @@ class ObChunkDatumStore {
OB_UNIS_VERSION_V(1);
public:
static inline int row_copy_size(const common::ObIArray<ObExpr*>& exprs, ObEvalCtx& ctx, int64_t& size)
{
int ret = OB_SUCCESS;
common::ObDatum* datum = nullptr;
size = DATUM_SIZE * exprs.count();
for (int64_t i = 0; i < exprs.count() && OB_SUCC(ret); ++i) {
if (OB_FAIL(exprs.at(i)->eval(ctx, datum))) {
SQL_ENG_LOG(WARN, "failed to eval expr datum", KPC(exprs.at(i)), K(ret));
} else {
size += datum->len_;
}
}
return ret;
}
static inline int64_t row_copy_size(common::ObDatum** datums, const int64_t cnt)
{
int64_t size = DATUM_SIZE * cnt;
for (int64_t i = 0; i < cnt; ++i) {
size += datums[i]->len_;
}
return size;
}
static inline int64_t row_copy_size(common::ObDatum* datums, const int64_t cnt)
{
int64_t size = DATUM_SIZE * cnt;
for (int64_t i = 0; i < cnt; ++i) {
size += datums[i].len_;
}
return size;
}
/*
* StoredRow memory layout
* N Datum + extend_size(can be 0) + real data
@ -75,16 +45,22 @@ public:
struct StoredRow {
StoredRow() : cnt_(0), row_size_(0)
{}
int copy_datums(const common::ObIArray<ObExpr*>& exprs, ObEvalCtx& ctx, char* buf, const int64_t size,
const int64_t row_size, const uint32_t row_extend_size);
int copy_datums(const common::ObIArray<ObExpr*>& exprs, ObEvalCtx& ctx, int64_t& row_size, char* buf,
const int64_t max_buf_size, const uint32_t row_extend_size);
int copy_datums(common::ObDatum** datums, const int64_t cnt, char* buf, const int64_t size, const int64_t row_size,
const uint32_t row_extend_size);
int copy_datums(common::ObDatum* datums, const int64_t cnt, char* buf, const int64_t size, const int64_t row_size,
const uint32_t row_extend_size);
int to_expr(const common::ObIArray<ObExpr*>& exprs, ObEvalCtx& ctx) const;
int to_expr(const common::ObIArray<ObExpr*>& exprs, ObEvalCtx& ctx, int64_t count) const;
// Build a stored row by exprs.
// @param [out] sr, result stored row
// @param epxrs,
// @param ctx
// @param buf
// @param buf_len, use Block::row_store_size() to detect the needed buffer size.
// @param extra_size, extra store size
// @param unswizzling
// @return OB_SUCCESS or OB_BUF_NOT_ENOUGH if buf not enough
static int build(StoredRow *&sr, const ObExprPtrIArray &exprs, ObEvalCtx &ctx, char *buf, const int64_t buf_len,
const uint32_t extra_size = 0);
static int build(StoredRow *&sr, const ObExprPtrIArray &exprs, ObEvalCtx &ctx, common::ObIAllocator &alloc,
const uint32_t extra_size = 0);
inline common::ObDatum* cells()
{
@ -117,7 +93,6 @@ public:
* 2) Provide reuse mode, memory can be reused
* 3) Provide conversion from StoredRow to ObIArray<ObExpr*>
*/
template <typename T = ObChunkDatumStore::StoredRow>
class LastStoredRow {
public:
LastStoredRow(ObIAllocator& alloc)
@ -138,13 +113,13 @@ public:
char* buf = NULL;
int64_t row_size = 0;
int64_t buffer_len = 0;
T* new_row = NULL;
StoredRow* new_row = NULL;
if (0 == exprs.count()) {
// no column. scenario like distinct 1
} else if (OB_FAIL(ObChunkDatumStore::row_copy_size(exprs, ctx, row_size))) {
SQL_ENG_LOG(WARN, "failed to calc copy size", K(ret));
} else {
int64_t head_size = sizeof(T);
int64_t head_size = sizeof(StoredRow);
reuse = OB_ISNULL(store_row_) ? false : reuse && (max_size_ >= row_size + head_size + extra_size);
if (reuse && OB_NOT_NULL(store_row_)) {
// switch buffer for write
@ -163,24 +138,21 @@ public:
} else if (OB_ISNULL(buf2 = reinterpret_cast<char*>(alloc_.alloc(buffer_len)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
SQL_ENG_LOG(ERROR, "alloc buf failed", K(ret));
} else if (OB_ISNULL(pre_alloc_row1_ = new (buf1) T())) {
} else if (OB_ISNULL(pre_alloc_row1_ = new (buf1) StoredRow())) {
ret = OB_ALLOCATE_MEMORY_FAILED;
SQL_ENG_LOG(ERROR, "failed to new row", K(ret));
} else if (OB_ISNULL(pre_alloc_row2_ = new (buf2) T())) {
} else if (OB_ISNULL(pre_alloc_row2_ = new (buf2) StoredRow())) {
ret = OB_ALLOCATE_MEMORY_FAILED;
SQL_ENG_LOG(ERROR, "failed to new row", K(ret));
} else {
buf = buf1;
new_row = pre_alloc_row1_;
}
}
if (OB_SUCC(ret)) {
int64_t pos = head_size;
if (OB_FAIL(new_row->copy_datums(exprs, ctx, buf + pos, buffer_len - head_size, row_size, extra_size))) {
SQL_ENG_LOG(WARN, "failed to deep copy row", K(ret), K(buffer_len), K(row_size));
if (OB_FAIL(StoredRow::build(store_row_, exprs, ctx, buf, buffer_len, extra_size))) {
SQL_ENG_LOG(WARN, "failed to build stored row", K(ret), K(buffer_len), K(row_size));
} else {
max_size_ = buffer_len;
store_row_ = new_row;
}
}
}
@ -192,9 +164,9 @@ public:
bool reuse = reuse_;
char* buf = NULL;
int64_t buffer_len = 0;
T* new_row = NULL;
StoredRow* new_row = NULL;
int64_t row_size = row.row_size_;
int64_t head_size = sizeof(T);
int64_t head_size = sizeof(StoredRow);
reuse = OB_ISNULL(store_row_) ? false : reuse && (max_size_ >= row_size + head_size + extra_size);
if (reuse && OB_NOT_NULL(store_row_)) {
buf = reinterpret_cast<char*>(store_row_);
@ -205,20 +177,15 @@ public:
if (OB_ISNULL(buf = reinterpret_cast<char*>(alloc_.alloc(buffer_len)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
SQL_ENG_LOG(ERROR, "alloc buf failed", K(ret));
} else if (OB_ISNULL(new_row = new (buf) T())) {
} else if (OB_ISNULL(new_row = new (buf) StoredRow())) {
ret = OB_ALLOCATE_MEMORY_FAILED;
SQL_ENG_LOG(ERROR, "failed to new row", K(ret));
}
}
if (OB_SUCC(ret)) {
int64_t pos = head_size;
if (OB_FAIL(new_row->copy_datums(const_cast<common::ObDatum*>(row.cells()),
row.cnt_,
buf + pos,
buffer_len - head_size,
row_size,
extra_size))) {
SQL_ENG_LOG(WARN, "failed to deep copy row", K(ret), K(buffer_len), K(row_size));
if (OB_FAIL(new_row->assign(&row))) {
SQL_ENG_LOG(WARN, "stored row assign failed", K(ret));
} else {
max_size_ = buffer_len;
store_row_ = new_row;
@ -226,7 +193,7 @@ public:
}
return ret;
}
void set_store_row(T* in_store_row)
void set_store_row(StoredRow* in_store_row)
{
store_row_ = in_store_row;
}
@ -237,18 +204,17 @@ public:
}
TO_STRING_KV(K_(max_size), K_(reuse), KPC_(store_row));
T* store_row_;
StoredRow* store_row_;
ObIAllocator& alloc_;
int64_t max_size_;
bool reuse_;
private:
// To avoid writing memory overwrite, alloc 2 row for alternate writing
T* pre_alloc_row1_;
T* pre_alloc_row2_;
StoredRow* pre_alloc_row1_;
StoredRow* pre_alloc_row2_;
};
template <typename T = ObChunkDatumStore::StoredRow>
class ShadowStoredRow {
public:
ShadowStoredRow() : alloc_(nullptr), store_row_(nullptr), saved_(false)
@ -261,7 +227,7 @@ public:
int init(common::ObIAllocator& allocator, int64_t datum_cnt)
{
int ret = OB_SUCCESS;
int64_t buffer_len = datum_cnt * sizeof(ObDatum) + sizeof(T);
int64_t buffer_len = datum_cnt * sizeof(ObDatum) + sizeof(StoredRow);
char* buf = nullptr;
if (NULL != alloc_) {
ret = common::OB_INIT_TWICE;
@ -271,9 +237,9 @@ public:
SQL_ENG_LOG(ERROR, "alloc buf failed", K(ret));
} else {
alloc_ = &allocator;
store_row_ = new (buf) T();
store_row_ = new (buf) StoredRow();
store_row_->cnt_ = datum_cnt;
store_row_->row_size_ = datum_cnt * sizeof(ObDatum);
store_row_->row_size_ = datum_cnt * sizeof(ObDatum) + sizeof(StoredRow);
saved_ = false;
}
return ret;
@ -322,7 +288,7 @@ public:
saved_ = false;
}
T* get_store_row() const
StoredRow* get_store_row() const
{
return store_row_;
}
@ -330,7 +296,7 @@ public:
private:
common::ObIAllocator* alloc_;
T* store_row_;
StoredRow* store_row_;
bool saved_;
};
@ -370,18 +336,10 @@ public:
}
// following interface for ObDatum only,unused for now
static int64_t inline min_buf_size(common::ObDatum** datums, const int64_t cnt)
{
return BlockBuffer::HEAD_SIZE + sizeof(BlockBuffer) + row_store_size(datums, cnt);
}
static int64_t inline min_buf_size(common::ObDatum* datums, const int64_t cnt)
{
return BlockBuffer::HEAD_SIZE + sizeof(BlockBuffer) + row_store_size(datums, cnt);
}
static int64_t inline row_store_size(common::ObDatum** datums, const int64_t cnt, uint32_t row_extend_size = 0)
{
return ROW_HEAD_SIZE + row_extend_size + ObChunkDatumStore::row_copy_size(datums, cnt);
}
static int64_t inline row_store_size(common::ObDatum* datums, const int64_t cnt, uint32_t row_extend_size = 0)
{
return ROW_HEAD_SIZE + row_extend_size + ObChunkDatumStore::row_copy_size(datums, cnt);
@ -970,6 +928,30 @@ private:
callback_->free(size);
}
static inline int row_copy_size(const common::ObIArray<ObExpr *> &exprs, ObEvalCtx &ctx, int64_t &size)
{
int ret = OB_SUCCESS;
common::ObDatum *datum = nullptr;
size = DATUM_SIZE * exprs.count();
for (int64_t i = 0; i < exprs.count() && OB_SUCC(ret); ++i) {
if (OB_FAIL(exprs.at(i)->eval(ctx, datum))) {
SQL_ENG_LOG(WARN, "failed to eval expr datum", KPC(exprs.at(i)), K(ret));
} else {
size += datum->len_;
}
}
return ret;
}
static inline int64_t row_copy_size(common::ObDatum *datums, const int64_t cnt)
{
int64_t size = DATUM_SIZE * cnt;
for (int64_t i = 0; i < cnt; ++i) {
size += datums[i].len_;
}
return size;
}
private:
bool inited_;
uint64_t tenant_id_;
@ -1011,6 +993,8 @@ private:
DISALLOW_COPY_AND_ASSIGN(ObChunkDatumStore);
};
typedef ObChunkDatumStore::StoredRow ObStoredDatumRow;
inline int ObChunkDatumStore::BlockBuffer::advance(int64_t size)
{
int ret = common::OB_SUCCESS;
@ -1026,7 +1010,6 @@ inline int ObChunkDatumStore::BlockBuffer::advance(int64_t size)
}
return ret;
}
} // end namespace sql
} // end namespace oceanbase

2
src/sql/engine/basic/ob_limit_op.h
View File

@ -68,7 +68,7 @@ private:
int64_t total_cnt_;
bool is_percent_first_;
ObChunkDatumStore::LastStoredRow<> pre_sort_columns_;
ObChunkDatumStore::LastStoredRow pre_sort_columns_;
};
} // end namespace sql