You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by zh...@apache.org on 2019/09/27 17:12:27 UTC
[incubator-doris] branch master updated: Replace Arena with MemPool
first step (#1899)
This is an automated email from the ASF dual-hosted git repository.
zhaoc pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-doris.git
The following commit(s) were added to refs/heads/master by this push:
new cafb9f1 Replace Arena with MemPool first step (#1899)
cafb9f1 is described below
commit cafb9f1e62873123f6b09cbe8056f5c25e2a6a85
Author: kangkaisen <ka...@apache.org>
AuthorDate: Sat Sep 28 01:12:22 2019 +0800
Replace Arena with MemPool first step (#1899)
---
be/src/exec/olap_scanner.cpp | 10 ++--
be/src/olap/aggregate_func.h | 97 ++++++++++++++++---------------
be/src/olap/field.h | 26 ++++-----
be/src/olap/memtable.cpp | 16 ++---
be/src/olap/memtable.h | 5 +-
be/src/olap/merger.cpp | 10 +++-
be/src/olap/reader.cpp | 16 ++---
be/src/olap/reader.h | 12 ++--
be/src/olap/row.h | 16 ++---
be/src/olap/schema_change.cpp | 13 +++--
be/src/olap/skiplist.h | 18 +++---
be/src/olap/task/engine_checksum_task.cpp | 11 ++--
be/src/runtime/mem_tracker.cpp | 5 --
be/test/olap/aggregate_func_test.cpp | 75 +++++++++++++-----------
be/test/olap/row_cursor_test.cpp | 12 ++--
be/test/olap/skiplist_test.cpp | 23 +++++---
16 files changed, 195 insertions(+), 170 deletions(-)
diff --git a/be/src/exec/olap_scanner.cpp b/be/src/exec/olap_scanner.cpp
index f4afb85..a1fdd6a 100644
--- a/be/src/exec/olap_scanner.cpp
+++ b/be/src/exec/olap_scanner.cpp
@@ -259,7 +259,9 @@ Status OlapScanner::get_batch(
state->batch_size() * _tuple_desc->byte_size());
bzero(tuple_buf, state->batch_size() * _tuple_desc->byte_size());
Tuple *tuple = reinterpret_cast<Tuple*>(tuple_buf);
- std::unique_ptr<Arena> arena(new Arena());
+
+ std::unique_ptr<MemTracker> tracker(new MemTracker(state->fragment_mem_tracker()->limit()));
+ std::unique_ptr<MemPool> mem_pool(new MemPool(tracker.get()));
int64_t raw_rows_threshold = raw_rows_read() + config::doris_scanner_row_num;
{
@@ -271,7 +273,7 @@ Status OlapScanner::get_batch(
break;
}
// Read one row from reader
- auto res = _reader->next_row_with_aggregation(&_read_row_cursor, arena.get(), batch->agg_object_pool(), eof);
+ auto res = _reader->next_row_with_aggregation(&_read_row_cursor, mem_pool.get(), batch->agg_object_pool(), eof);
if (res != OLAP_SUCCESS) {
return Status::InternalError("Internal Error: read storage fail.");
}
@@ -337,9 +339,9 @@ Status OlapScanner::get_batch(
}
}
- // the memory allocate by arena has been copied,
+ // the memory allocate by mem pool has been copied,
// so we should release these memory immediately
- arena.reset(new Arena());
+ mem_pool->clear();
if (VLOG_ROW_IS_ON) {
VLOG_ROW << "OlapScanner output row: " << Tuple::to_string(tuple, *_tuple_desc);
diff --git a/be/src/olap/aggregate_func.h b/be/src/olap/aggregate_func.h
index b3155cf..360d983 100644
--- a/be/src/olap/aggregate_func.h
+++ b/be/src/olap/aggregate_func.h
@@ -24,14 +24,15 @@
#include "runtime/datetime_value.h"
#include "runtime/decimalv2_value.h"
#include "runtime/string_value.h"
-#include "util/arena.h"
+#include "runtime/mem_pool.h"
+#include "runtime/mem_tracker.h"
#include "util/bitmap.h"
namespace doris {
-using AggInitFunc = void (*)(RowCursorCell* dst, const char* src, bool src_null, Arena* arena, ObjectPool* agg_pool);
-using AggUpdateFunc = void (*)(RowCursorCell* dst, const RowCursorCell& src, Arena* arena);
-using AggFinalizeFunc = void (*)(RowCursorCell* src, Arena* arena);
+using AggInitFunc = void (*)(RowCursorCell* dst, const char* src, bool src_null, MemPool* mem_pool, ObjectPool* agg_pool);
+using AggUpdateFunc = void (*)(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool);
+using AggFinalizeFunc = void (*)(RowCursorCell* src, MemPool* mem_pool);
// This class contains information about aggregate operation.
class AggregateInfo {
@@ -44,8 +45,8 @@ public:
// Memory Note: For plain memory can be allocated from arena, whose lifetime
// will last util finalize function is called. Memory allocated from heap should
// be freed in finalize functioin to avoid memory leak.
- inline void init(RowCursorCell* dst, const char* src, bool src_null, Arena* arena, ObjectPool* agg_pool) const {
- _init_fn(dst, src, src_null, arena, agg_pool);
+ inline void init(RowCursorCell* dst, const char* src, bool src_null, MemPool* mem_pool, ObjectPool* agg_pool) const {
+ _init_fn(dst, src, src_null, mem_pool, agg_pool);
}
// Update aggregated intermediate data. Data stored in engine is aggregated.
@@ -55,8 +56,8 @@ public:
// will be added to sum.
// Memory Note: Same with init function.
- inline void update(RowCursorCell* dst, const RowCursorCell& src, Arena* arena) const {
- _update_fn(dst, src, arena);
+ inline void update(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool) const {
+ _update_fn(dst, src, mem_pool);
}
// Finalize function convert intermediate context into final format. For example:
@@ -67,16 +68,16 @@ public:
// Memory Note: All heap memory allocated in init and update function should be freed
// before this function return. Memory allocated from arena will be still available
// and will be freed by client.
- inline void finalize(RowCursorCell* src, Arena* arena) const {
- _finalize_fn(src, arena);
+ inline void finalize(RowCursorCell* src, MemPool* mem_pool) const {
+ _finalize_fn(src, mem_pool);
}
FieldAggregationMethod agg_method() const { return _agg_method; }
private:
- void (*_init_fn)(RowCursorCell* dst, const char* src, bool src_null, Arena* arena, ObjectPool* agg_pool);
- void (*_update_fn)(RowCursorCell* dst, const RowCursorCell& src, Arena* arena);
- void (*_finalize_fn)(RowCursorCell* src, Arena* arena);
+ void (*_init_fn)(RowCursorCell* dst, const char* src, bool src_null, MemPool* mem_pool, ObjectPool* agg_pool);
+ void (*_update_fn)(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool);
+ void (*_finalize_fn)(RowCursorCell* src, MemPool* mem_pool);
friend class AggregateFuncResolver;
@@ -88,22 +89,22 @@ private:
template<FieldType field_type>
struct BaseAggregateFuncs {
- static void init(RowCursorCell* dst, const char* src, bool src_null, Arena* arena, ObjectPool* agg_pool) {
+ static void init(RowCursorCell* dst, const char* src, bool src_null, MemPool* mem_pool, ObjectPool* agg_pool) {
dst->set_is_null(src_null);
if (src_null) {
return;
}
const TypeInfo* _type_info = get_type_info(field_type);
- _type_info->deep_copy_with_arena(dst->mutable_cell_ptr(), src, arena);
+ _type_info->deep_copy(dst->mutable_cell_ptr(), src, mem_pool);
}
// Default update do nothing.
- static void update(RowCursorCell* dst, const RowCursorCell& src, Arena* arena) {
+ static void update(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool) {
}
// Default finalize do nothing.
- static void finalize(RowCursorCell* src, Arena* arena) {
+ static void finalize(RowCursorCell* src, MemPool* mem_pool) {
}
};
@@ -114,7 +115,7 @@ struct AggregateFuncTraits : public BaseAggregateFuncs<field_type> {
template <>
struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_NONE, OLAP_FIELD_TYPE_DECIMAL> :
public BaseAggregateFuncs<OLAP_FIELD_TYPE_DECIMAL> {
- static void init(RowCursorCell* dst, const char* src, bool src_null, Arena* arena, ObjectPool* agg_pool) {
+ static void init(RowCursorCell* dst, const char* src, bool src_null, MemPool* mem_pool, ObjectPool* agg_pool) {
dst->set_is_null(src_null);
if (src_null) {
return;
@@ -130,7 +131,7 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_NONE, OLAP_FIELD_TYPE_DECIMAL>
template <>
struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_NONE, OLAP_FIELD_TYPE_DATETIME> :
public BaseAggregateFuncs<OLAP_FIELD_TYPE_DECIMAL> {
- static void init(RowCursorCell* dst, const char* src, bool src_null, Arena* arena, ObjectPool* agg_pool) {
+ static void init(RowCursorCell* dst, const char* src, bool src_null, MemPool* mem_pool, ObjectPool* agg_pool) {
dst->set_is_null(src_null);
if (src_null) {
return;
@@ -145,7 +146,7 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_NONE, OLAP_FIELD_TYPE_DATETIME
template <>
struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_NONE, OLAP_FIELD_TYPE_DATE> :
public BaseAggregateFuncs<OLAP_FIELD_TYPE_DECIMAL> {
- static void init(RowCursorCell* dst, const char* src, bool src_null, Arena* arena, ObjectPool* agg_pool) {
+ static void init(RowCursorCell* dst, const char* src, bool src_null, MemPool* mem_pool, ObjectPool* agg_pool) {
dst->set_is_null(src_null);
if (src_null) {
return;
@@ -162,7 +163,7 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_MIN, field_type> :
public AggregateFuncTraits<OLAP_FIELD_AGGREGATION_NONE, field_type> {
typedef typename FieldTypeTraits<field_type>::CppType CppType;
- static void update(RowCursorCell* dst, const RowCursorCell& src, Arena* arena) {
+ static void update(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool) {
bool src_null = src.is_null();
// ignore null value
if (src_null) return;
@@ -182,7 +183,7 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_MIN, OLAP_FIELD_TYPE_LARGEINT>
public BaseAggregateFuncs<OLAP_FIELD_TYPE_LARGEINT> {
typedef typename FieldTypeTraits<OLAP_FIELD_TYPE_LARGEINT>::CppType CppType;
- static void update(RowCursorCell* dst, const RowCursorCell& src, Arena* arena) {
+ static void update(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool) {
bool src_null = src.is_null();
// ignore null value
if (src_null) return;
@@ -207,7 +208,7 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_MIN, OLAP_FIELD_TYPE_LARGEINT>
template <>
struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_MIN, OLAP_FIELD_TYPE_VARCHAR> :
public AggregateFuncTraits<OLAP_FIELD_AGGREGATION_NONE, OLAP_FIELD_TYPE_VARCHAR> {
- static void update(RowCursorCell* dst, const RowCursorCell& src, Arena* arena) {
+ static void update(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool) {
bool src_null = src.is_null();
// ignore null value
if (src_null) return;
@@ -217,11 +218,11 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_MIN, OLAP_FIELD_TYPE_VARCHAR>
Slice* dst_slice = reinterpret_cast<Slice*>(dst->mutable_cell_ptr());
const Slice* src_slice = reinterpret_cast<const Slice*>(src.cell_ptr());
if (dst_null || src_slice->compare(*dst_slice) < 0) {
- if (arena == nullptr || (!dst_null && dst_slice->size >= src_slice->size)) {
+ if (mem_pool == nullptr || (!dst_null && dst_slice->size >= src_slice->size)) {
memory_copy(dst_slice->data, src_slice->data, src_slice->size);
dst_slice->size = src_slice->size;
} else {
- dst_slice->data = arena->Allocate(src_slice->size);
+ dst_slice->data = (char*)mem_pool->allocate(src_slice->size);
memory_copy(dst_slice->data, src_slice->data, src_slice->size);
dst_slice->size = src_slice->size;
}
@@ -239,7 +240,7 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_MAX, field_type> :
public AggregateFuncTraits<OLAP_FIELD_AGGREGATION_NONE, field_type> {
typedef typename FieldTypeTraits<field_type>::CppType CppType;
- static void update(RowCursorCell* dst, const RowCursorCell& src, Arena* arena) {
+ static void update(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool) {
bool src_null = src.is_null();
// ignore null value
if (src_null) return;
@@ -259,7 +260,7 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_MAX, OLAP_FIELD_TYPE_LARGEINT>
public BaseAggregateFuncs<OLAP_FIELD_TYPE_LARGEINT> {
typedef typename FieldTypeTraits<OLAP_FIELD_TYPE_LARGEINT>::CppType CppType;
- static void update(RowCursorCell* dst, const RowCursorCell& src, Arena* arena) {
+ static void update(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool) {
bool src_null = src.is_null();
// ignore null value
if (src_null) return;
@@ -278,7 +279,7 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_MAX, OLAP_FIELD_TYPE_LARGEINT>
template <>
struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_MAX, OLAP_FIELD_TYPE_VARCHAR> :
public AggregateFuncTraits<OLAP_FIELD_AGGREGATION_NONE, OLAP_FIELD_TYPE_VARCHAR> {
- static void update(RowCursorCell* dst, const RowCursorCell& src, Arena* arena) {
+ static void update(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool) {
bool src_null = src.is_null();
// ignore null value
if (src_null) return;
@@ -288,11 +289,11 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_MAX, OLAP_FIELD_TYPE_VARCHAR>
Slice* dst_slice = reinterpret_cast<Slice*>(dst->mutable_cell_ptr());
const Slice* src_slice = reinterpret_cast<const Slice*>(src.cell_ptr());
if (dst_null || src_slice->compare(*dst_slice) > 0) {
- if (arena == nullptr || (!dst_null && dst_slice->size >= src_slice->size)) {
+ if (mem_pool == nullptr || (!dst_null && dst_slice->size >= src_slice->size)) {
memory_copy(dst_slice->data, src_slice->data, src_slice->size);
dst_slice->size = src_slice->size;
} else {
- dst_slice->data = arena->Allocate(src_slice->size);
+ dst_slice->data = (char*)mem_pool->allocate(src_slice->size);
memory_copy(dst_slice->data, src_slice->data, src_slice->size);
dst_slice->size = src_slice->size;
}
@@ -310,7 +311,7 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_SUM, field_type> :
public AggregateFuncTraits<OLAP_FIELD_AGGREGATION_NONE, field_type> {
typedef typename FieldTypeTraits<field_type>::CppType CppType;
- static void update(RowCursorCell* dst, const RowCursorCell& src, Arena* arena) {
+ static void update(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool) {
bool src_null = src.is_null();
if (src_null) {
return;
@@ -332,7 +333,7 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_SUM, OLAP_FIELD_TYPE_LARGEINT>
public BaseAggregateFuncs<OLAP_FIELD_TYPE_LARGEINT> {
typedef typename FieldTypeTraits<OLAP_FIELD_TYPE_LARGEINT>::CppType CppType;
- static void update(RowCursorCell* dst, const RowCursorCell& src, Arena* arena) {
+ static void update(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool) {
bool src_null = src.is_null();
if (src_null) {
return;
@@ -357,7 +358,7 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_REPLACE, field_type> :
public AggregateFuncTraits<OLAP_FIELD_AGGREGATION_NONE, field_type> {
typedef typename FieldTypeTraits<field_type>::CppType CppType;
- static void update(RowCursorCell* dst, const RowCursorCell& src, Arena* arena) {
+ static void update(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool) {
bool src_null = src.is_null();
dst->set_is_null(src_null);
if (!src_null) {
@@ -369,7 +370,7 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_REPLACE, field_type> :
template <>
struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_REPLACE, OLAP_FIELD_TYPE_VARCHAR> :
public AggregateFuncTraits<OLAP_FIELD_AGGREGATION_NONE, OLAP_FIELD_TYPE_VARCHAR> {
- static void update(RowCursorCell* dst, const RowCursorCell& src, Arena* arena) {
+ static void update(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool) {
bool dst_null = dst->is_null();
bool src_null = src.is_null();
dst->set_is_null(src_null);
@@ -379,11 +380,11 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_REPLACE, OLAP_FIELD_TYPE_VARCH
Slice* dst_slice = reinterpret_cast<Slice*>(dst->mutable_cell_ptr());
const Slice* src_slice = reinterpret_cast<const Slice*>(src.cell_ptr());
- if (arena == nullptr || (!dst_null && dst_slice->size >= src_slice->size)) {
+ if (mem_pool == nullptr || (!dst_null && dst_slice->size >= src_slice->size)) {
memory_copy(dst_slice->data, src_slice->data, src_slice->size);
dst_slice->size = src_slice->size;
} else {
- dst_slice->data = arena->Allocate(src_slice->size);
+ dst_slice->data = (char*)mem_pool->allocate(src_slice->size);
memory_copy(dst_slice->data, src_slice->data, src_slice->size);
dst_slice->size = src_slice->size;
}
@@ -399,7 +400,7 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_REPLACE, OLAP_FIELD_TYPE_CHAR>
// so when init, update hll, the src is not null
template <>
struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_HLL_UNION, OLAP_FIELD_TYPE_HLL> {
- static void init(RowCursorCell* dst, const char* src, bool src_null, Arena* arena, ObjectPool* agg_pool) {
+ static void init(RowCursorCell* dst, const char* src, bool src_null, MemPool* mem_pool, ObjectPool* agg_pool) {
DCHECK_EQ(src_null, false);
dst->set_not_null();
@@ -411,12 +412,12 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_HLL_UNION, OLAP_FIELD_TYPE_HLL
auto* hll = new HyperLogLog((const uint8_t*) src_slice->data);
dst_slice->data = reinterpret_cast<char*>(hll);
- arena->track_memory(sizeof(HyperLogLog));
+ mem_pool->mem_tracker()->consume(sizeof(HyperLogLog));
agg_pool->add(hll);
}
- static void update(RowCursorCell* dst, const RowCursorCell& src, Arena* arena) {
+ static void update(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool) {
DCHECK_EQ(src.is_null(), false);
auto* dst_slice = reinterpret_cast<Slice*>(dst->mutable_cell_ptr());
@@ -424,7 +425,7 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_HLL_UNION, OLAP_FIELD_TYPE_HLL
auto* dst_hll = reinterpret_cast<HyperLogLog*>(dst_slice->data);
// fixme(kks): trick here, need improve
- if (arena == nullptr) { // for query
+ if (mem_pool == nullptr) { // for query
HyperLogLog src_hll((const uint8_t*)src_slice->data);
dst_hll->merge(src_hll);
} else { // for stream load
@@ -434,11 +435,11 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_HLL_UNION, OLAP_FIELD_TYPE_HLL
}
// The HLL object memory will be released by ObjectPool
- static void finalize(RowCursorCell* src, Arena* arena) {
+ static void finalize(RowCursorCell* src, MemPool* mem_pool) {
auto *slice = reinterpret_cast<Slice*>(src->mutable_cell_ptr());
auto *hll = reinterpret_cast<HyperLogLog*>(slice->data);
- slice->data = arena->Allocate(HLL_COLUMN_DEFAULT_LEN);
+ slice->data = (char*)mem_pool->allocate(HLL_COLUMN_DEFAULT_LEN);
slice->size = hll->serialize((uint8_t*)slice->data);
}
};
@@ -446,7 +447,7 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_HLL_UNION, OLAP_FIELD_TYPE_HLL
// so when init, update bitmap, the src is not null
template <>
struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_BITMAP_UNION, OLAP_FIELD_TYPE_VARCHAR> {
- static void init(RowCursorCell* dst, const char* src, bool src_null, Arena* arena, ObjectPool* agg_pool) {
+ static void init(RowCursorCell* dst, const char* src, bool src_null, MemPool* mem_pool, ObjectPool* agg_pool) {
DCHECK_EQ(src_null, false);
dst->set_not_null();
auto* src_slice = reinterpret_cast<const Slice*>(src);
@@ -457,12 +458,12 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_BITMAP_UNION, OLAP_FIELD_TYPE_
auto* bitmap = new RoaringBitmap(src_slice->data);
dst_slice->data = (char*) bitmap;
- arena->track_memory(sizeof(RoaringBitmap));
+ mem_pool->mem_tracker()->consume(sizeof(RoaringBitmap));
agg_pool->add(bitmap);
}
- static void update(RowCursorCell* dst, const RowCursorCell& src, Arena* arena) {
+ static void update(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool) {
DCHECK_EQ(src.is_null(), false);
auto* dst_slice = reinterpret_cast<Slice*>(dst->mutable_cell_ptr());
@@ -470,7 +471,7 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_BITMAP_UNION, OLAP_FIELD_TYPE_
auto* dst_bitmap = reinterpret_cast<RoaringBitmap*>(dst_slice->data);
// fixme(kks): trick here, need improve
- if (arena == nullptr) { // for query
+ if (mem_pool == nullptr) { // for query
RoaringBitmap src_bitmap = RoaringBitmap(src_slice->data);
dst_bitmap->merge(src_bitmap);
} else { // for stream load
@@ -480,12 +481,12 @@ struct AggregateFuncTraits<OLAP_FIELD_AGGREGATION_BITMAP_UNION, OLAP_FIELD_TYPE_
}
// The RoaringBitmap object memory will be released by ObjectPool
- static void finalize(RowCursorCell* src, Arena *arena) {
+ static void finalize(RowCursorCell* src, MemPool* mem_pool) {
auto *slice = reinterpret_cast<Slice*>(src->mutable_cell_ptr());
auto *bitmap = reinterpret_cast<RoaringBitmap*>(slice->data);
slice->size = bitmap->size();
- slice->data = arena->Allocate(slice->size);
+ slice->data = (char*)mem_pool->allocate(slice->size);
bitmap->serialize(slice->data);
}
};
diff --git a/be/src/olap/field.h b/be/src/olap/field.h
index caecdd9..452cc8f 100644
--- a/be/src/olap/field.h
+++ b/be/src/olap/field.h
@@ -58,16 +58,16 @@ public:
inline void set_to_max(char* buf) const { return _type_info->set_to_max(buf); }
inline void set_to_min(char* buf) const { return _type_info->set_to_min(buf); }
- inline void agg_update(RowCursorCell* dest, const RowCursorCell& src, Arena* arena = nullptr) const {
- _agg_info->update(dest, src, arena);
+ inline void agg_update(RowCursorCell* dest, const RowCursorCell& src, MemPool* mem_pool = nullptr) const {
+ _agg_info->update(dest, src, mem_pool);
}
- inline void agg_finalize(RowCursorCell* dst, Arena* arena) const {
- _agg_info->finalize(dst, arena);
+ inline void agg_finalize(RowCursorCell* dst, MemPool* mem_pool) const {
+ _agg_info->finalize(dst, mem_pool);
}
- virtual void consume(RowCursorCell* dst, const char* src, bool src_null, Arena* arena, ObjectPool* agg_pool) const {
- _agg_info->init(dst, src, src_null, arena, agg_pool);
+ virtual void consume(RowCursorCell* dst, const char* src, bool src_null, MemPool* mem_pool, ObjectPool* agg_pool) const {
+ _agg_info->init(dst, src, src_null, mem_pool, agg_pool);
}
// todo(kks): Unify AggregateInfo::init method and Field::agg_init method
@@ -76,7 +76,7 @@ public:
// This functionn differs copy functionn in that if this filed
// contain aggregate information, this functionn will initialize
// destination in aggregate format, and update with srouce content.
- virtual void agg_init(RowCursorCell* dst, const RowCursorCell& src, Arena* arena, ObjectPool* agg_pool) const {
+ virtual void agg_init(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool, ObjectPool* agg_pool) const {
direct_copy(dst, src);
}
@@ -344,7 +344,7 @@ public:
}
// the char field is especial, which need the _length info when consume raw data
- void consume(RowCursorCell* dst, const char* src, bool src_null, Arena* arena, ObjectPool* agg_pool) const override {
+ void consume(RowCursorCell* dst, const char* src, bool src_null, MemPool* mem_pool, ObjectPool* agg_pool) const override {
dst->set_is_null(src_null);
if (src_null) {
return;
@@ -353,7 +353,7 @@ public:
auto* value = reinterpret_cast<const StringValue*>(src);
auto* dest_slice = (Slice*)(dst->mutable_cell_ptr());
dest_slice->size = _length;
- dest_slice->data = arena->Allocate(dest_slice->size);
+ dest_slice->data = (char*)mem_pool->allocate(dest_slice->size);
memcpy(dest_slice->data, value->ptr, value->len);
memset(dest_slice->data + value->len, 0, dest_slice->size - value->len);
}
@@ -403,8 +403,8 @@ public:
}
// bitmap storage data always not null
- void agg_init(RowCursorCell* dst, const RowCursorCell& src, Arena* arena, ObjectPool* agg_pool) const override {
- _agg_info->init(dst, (const char*)src.cell_ptr(), false, arena, agg_pool);
+ void agg_init(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool, ObjectPool* agg_pool) const override {
+ _agg_info->init(dst, (const char*)src.cell_ptr(), false, mem_pool, agg_pool);
}
char* allocate_memory(char* cell_ptr, char* variable_ptr) const override {
@@ -424,8 +424,8 @@ public:
}
// Hll storage data always not null
- void agg_init(RowCursorCell* dst, const RowCursorCell& src, Arena* arena, ObjectPool* agg_pool) const override {
- _agg_info->init(dst, (const char*)src.cell_ptr(), false, arena, agg_pool);
+ void agg_init(RowCursorCell* dst, const RowCursorCell& src, MemPool* mem_pool, ObjectPool* agg_pool) const override {
+ _agg_info->init(dst, (const char*)src.cell_ptr(), false, mem_pool, agg_pool);
}
char* allocate_memory(char* cell_ptr, char* variable_ptr) const override {
diff --git a/be/src/olap/memtable.cpp b/be/src/olap/memtable.cpp
index ae7e82b..240f57a 100644
--- a/be/src/olap/memtable.cpp
+++ b/be/src/olap/memtable.cpp
@@ -38,8 +38,10 @@ MemTable::MemTable(int64_t tablet_id, Schema* schema, const TabletSchema* tablet
_row_comparator(_schema),
_rowset_writer(rowset_writer) {
_schema_size = _schema->schema_size();
- _tuple_buf = _arena.Allocate(_schema_size);
- _skip_list = new Table(_row_comparator, &_arena);
+ _tracker.reset(new MemTracker(config::write_buffer_size * 2));
+ _mem_pool.reset(new MemPool(_tracker.get()));
+ _tuple_buf = _mem_pool->allocate(_schema_size);
+ _skip_list = new Table(_row_comparator, _mem_pool.get());
}
MemTable::~MemTable() {
@@ -56,7 +58,7 @@ int MemTable::RowCursorComparator::operator()(const char* left, const char* righ
}
size_t MemTable::memory_usage() {
- return _arena.MemoryUsage();
+ return _mem_pool->mem_tracker()->consumption();
}
void MemTable::insert(Tuple* tuple) {
@@ -68,13 +70,13 @@ void MemTable::insert(Tuple* tuple) {
bool is_null = tuple->is_null(slot->null_indicator_offset());
void* value = tuple->get_slot(slot->tuple_offset());
- _schema->column(i)->consume(&cell, (const char *)value, is_null, _skip_list->arena(), &_agg_object_pool);
+ _schema->column(i)->consume(&cell, (const char *)value, is_null, _mem_pool.get(), &_agg_object_pool);
}
bool overwritten = false;
- _skip_list->Insert(_tuple_buf, &overwritten, _keys_type);
+ _skip_list->Insert((char*)_tuple_buf, &overwritten, _keys_type);
if (!overwritten) {
- _tuple_buf = _arena.Allocate(_schema_size);
+ _tuple_buf = _mem_pool->allocate(_schema_size);
}
}
@@ -86,7 +88,7 @@ OLAPStatus MemTable::flush() {
for (it.SeekToFirst(); it.Valid(); it.Next()) {
char* row = (char*)it.key();
ContiguousRow dst_row(_schema, row);
- agg_finalize_row(&dst_row, _skip_list->arena());
+ agg_finalize_row(&dst_row, _mem_pool.get());
RETURN_NOT_OK(_rowset_writer->add_row(dst_row));
}
RETURN_NOT_OK(_rowset_writer->flush());
diff --git a/be/src/olap/memtable.h b/be/src/olap/memtable.h
index 3856c83..7ce2374 100644
--- a/be/src/olap/memtable.h
+++ b/be/src/olap/memtable.h
@@ -57,11 +57,12 @@ private:
};
RowCursorComparator _row_comparator;
- Arena _arena;
+ std::unique_ptr<MemTracker> _tracker;
+ std::unique_ptr<MemPool> _mem_pool;
ObjectPool _agg_object_pool;
typedef SkipList<char*, RowCursorComparator> Table;
- char* _tuple_buf;
+ u_int8_t* _tuple_buf;
size_t _schema_size;
Table* _skip_list;
diff --git a/be/src/olap/merger.cpp b/be/src/olap/merger.cpp
index ae1867e..d46f4b0 100644
--- a/be/src/olap/merger.cpp
+++ b/be/src/olap/merger.cpp
@@ -44,14 +44,17 @@ OLAPStatus Merger::merge_rowsets(TabletSharedPtr tablet,
RETURN_NOT_OK_LOG(row_cursor.init(tablet->tablet_schema()),
"failed to init row cursor when merging rowsets of tablet " + tablet->full_name());
row_cursor.allocate_memory_for_string_type(tablet->tablet_schema());
+
+ std::unique_ptr<MemTracker> tracker(new MemTracker(-1));
+ std::unique_ptr<MemPool> mem_pool(new MemPool(tracker.get()));
+
// The following procedure would last for long time, half of one day, etc.
int64_t output_rows = 0;
while (true) {
- Arena arena;
ObjectPool objectPool;
bool eof = false;
// Read one row into row_cursor
- RETURN_NOT_OK_LOG(reader.next_row_with_aggregation(&row_cursor, &arena, &objectPool, &eof),
+ RETURN_NOT_OK_LOG(reader.next_row_with_aggregation(&row_cursor, mem_pool.get(), &objectPool, &eof),
"failed to read next row when merging rowsets of tablet " + tablet->full_name());
if (eof) {
break;
@@ -59,6 +62,9 @@ OLAPStatus Merger::merge_rowsets(TabletSharedPtr tablet,
RETURN_NOT_OK_LOG(dst_rowset_writer->add_row(row_cursor),
"failed to write row when merging rowsets of tablet " + tablet->full_name());
output_rows++;
+ // the memory allocate by mem pool has been copied,
+ // so we should release memory immediately
+ mem_pool->clear();
}
if (stats_output != nullptr) {
diff --git a/be/src/olap/reader.cpp b/be/src/olap/reader.cpp
index 9255faa..a668cef 100644
--- a/be/src/olap/reader.cpp
+++ b/be/src/olap/reader.cpp
@@ -345,7 +345,7 @@ OLAPStatus Reader::init(const ReaderParams& read_params) {
return OLAP_SUCCESS;
}
-OLAPStatus Reader::_dup_key_next_row(RowCursor* row_cursor, Arena* arena, ObjectPool* agg_pool, bool* eof) {
+OLAPStatus Reader::_dup_key_next_row(RowCursor* row_cursor, MemPool* mem_pool, ObjectPool* agg_pool, bool* eof) {
if (UNLIKELY(_next_key == nullptr)) {
*eof = true;
return OLAP_SUCCESS;
@@ -360,12 +360,12 @@ OLAPStatus Reader::_dup_key_next_row(RowCursor* row_cursor, Arena* arena, Object
return OLAP_SUCCESS;
}
-OLAPStatus Reader::_agg_key_next_row(RowCursor* row_cursor, Arena* arena, ObjectPool* agg_pool, bool* eof) {
+OLAPStatus Reader::_agg_key_next_row(RowCursor* row_cursor, MemPool* mem_pool, ObjectPool* agg_pool, bool* eof) {
if (UNLIKELY(_next_key == nullptr)) {
*eof = true;
return OLAP_SUCCESS;
}
- init_row_with_others(row_cursor, *_next_key, arena, agg_pool);
+ init_row_with_others(row_cursor, *_next_key, mem_pool, agg_pool);
int64_t merged_count = 0;
do {
auto res = _collect_iter->next(&_next_key, &_next_delete_flag);
@@ -391,13 +391,13 @@ OLAPStatus Reader::_agg_key_next_row(RowCursor* row_cursor, Arena* arena, Object
_merged_rows += merged_count;
// For agg query, we don't need finalize agg object and directly pass agg object to agg node
if (_need_agg_finalize) {
- agg_finalize_row(_value_cids, row_cursor, arena);
+ agg_finalize_row(_value_cids, row_cursor, mem_pool);
}
return OLAP_SUCCESS;
}
-OLAPStatus Reader::_unique_key_next_row(RowCursor* row_cursor, Arena* arena, ObjectPool* agg_pool, bool* eof) {
+OLAPStatus Reader::_unique_key_next_row(RowCursor* row_cursor, MemPool* mem_pool, ObjectPool* agg_pool, bool* eof) {
*eof = false;
bool cur_delete_flag = false;
do {
@@ -407,7 +407,7 @@ OLAPStatus Reader::_unique_key_next_row(RowCursor* row_cursor, Arena* arena, Obj
}
cur_delete_flag = _next_delete_flag;
- init_row_with_others(row_cursor, *_next_key, arena, agg_pool);
+ init_row_with_others(row_cursor, *_next_key, mem_pool, agg_pool);
int64_t merged_count = 0;
while (NULL != _next_key) {
@@ -423,12 +423,12 @@ OLAPStatus Reader::_unique_key_next_row(RowCursor* row_cursor, Arena* arena, Obj
// 1. DUP_KEYS keys type has no semantic to aggregate,
// 2. to make cost of each scan round reasonable, we will control merged_count.
if (_aggregation && merged_count > config::doris_scanner_row_num) {
- agg_finalize_row(_value_cids, row_cursor, arena);
+ agg_finalize_row(_value_cids, row_cursor, mem_pool);
break;
}
// break while can NOT doing aggregation
if (!equal_row(_key_cids, *row_cursor, *_next_key)) {
- agg_finalize_row(_value_cids, row_cursor, arena);
+ agg_finalize_row(_value_cids, row_cursor, mem_pool);
break;
}
diff --git a/be/src/olap/reader.h b/be/src/olap/reader.h
index 2183cae..d07fbce 100644
--- a/be/src/olap/reader.h
+++ b/be/src/olap/reader.h
@@ -126,8 +126,8 @@ public:
// Return OLAP_SUCCESS and set `*eof` to false when next row is read into `row_cursor`.
// Return OLAP_SUCCESS and set `*eof` to true when no more rows can be read.
// Return others when unexpected error happens.
- OLAPStatus next_row_with_aggregation(RowCursor *row_cursor, Arena* arena, ObjectPool* agg_pool, bool *eof) {
- return (this->*_next_row_func)(row_cursor, arena, agg_pool, eof);
+ OLAPStatus next_row_with_aggregation(RowCursor *row_cursor, MemPool* mem_pool, ObjectPool* agg_pool, bool *eof) {
+ return (this->*_next_row_func)(row_cursor, mem_pool, agg_pool, eof);
}
uint64_t merged_rows() const {
@@ -201,9 +201,9 @@ private:
OLAPStatus _init_load_bf_columns(const ReaderParams& read_params);
- OLAPStatus _dup_key_next_row(RowCursor* row_cursor, Arena* arena, ObjectPool* agg_pool, bool* eof);
- OLAPStatus _agg_key_next_row(RowCursor* row_cursor, Arena* arena, ObjectPool* agg_pool, bool* eof);
- OLAPStatus _unique_key_next_row(RowCursor* row_cursor, Arena* arena, ObjectPool* agg_pool, bool* eof);
+ OLAPStatus _dup_key_next_row(RowCursor* row_cursor, MemPool* mem_pool, ObjectPool* agg_pool, bool* eof);
+ OLAPStatus _agg_key_next_row(RowCursor* row_cursor, MemPool* mem_pool, ObjectPool* agg_pool, bool* eof);
+ OLAPStatus _unique_key_next_row(RowCursor* row_cursor, MemPool* mem_pool, ObjectPool* agg_pool, bool* eof);
TabletSharedPtr tablet() { return _tablet; }
@@ -234,7 +234,7 @@ private:
DeleteHandler _delete_handler;
- OLAPStatus (Reader::*_next_row_func)(RowCursor* row_cursor, Arena* arena, ObjectPool* agg_pool, bool* eof) = nullptr;
+ OLAPStatus (Reader::*_next_row_func)(RowCursor* row_cursor, MemPool* mem_pool, ObjectPool* agg_pool, bool* eof) = nullptr;
bool _aggregation;
// for agg query, we don't need to finalize when scan agg object data
diff --git a/be/src/olap/row.h b/be/src/olap/row.h
index 64ea139..b6b2926 100644
--- a/be/src/olap/row.h
+++ b/be/src/olap/row.h
@@ -107,10 +107,10 @@ int index_compare_row(const LhsRowType& lhs, const RhsRowType& rhs) {
// function will first initialize destination column and then update with source column
// value.
template<typename DstRowType, typename SrcRowType>
-void init_row_with_others(DstRowType* dst, const SrcRowType& src, Arena* arena, ObjectPool* agg_pool) {
+void init_row_with_others(DstRowType* dst, const SrcRowType& src, MemPool* mem_pool, ObjectPool* agg_pool) {
for (auto cid : dst->schema()->column_ids()) {
auto dst_cell = dst->cell(cid);
- dst->schema()->column(cid)->agg_init(&dst_cell, src.cell(cid), arena, agg_pool);
+ dst->schema()->column(cid)->agg_init(&dst_cell, src.cell(cid), mem_pool, agg_pool);
}
}
@@ -145,11 +145,11 @@ void copy_row(DstRowType* dst, const SrcRowType& src, Arena* arena) {
}
template<typename DstRowType, typename SrcRowType>
-void agg_update_row(DstRowType* dst, const SrcRowType& src, Arena* arena) {
+void agg_update_row(DstRowType* dst, const SrcRowType& src, MemPool* mem_pool) {
for (uint32_t cid = dst->schema()->num_key_columns(); cid < dst->schema()->num_columns(); ++cid) {
auto dst_cell = dst->cell(cid);
auto src_cell = src.cell(cid);
- dst->schema()->column(cid)->agg_update(&dst_cell, src_cell, arena);
+ dst->schema()->column(cid)->agg_update(&dst_cell, src_cell, mem_pool);
}
}
@@ -166,18 +166,18 @@ void agg_update_row(const std::vector<uint32_t>& cids, DstRowType* dst, const Sr
}
template<typename RowType>
-void agg_finalize_row(RowType* row, Arena* arena) {
+void agg_finalize_row(RowType* row, MemPool* mem_pool) {
for (uint32_t cid = row->schema()->num_key_columns(); cid < row->schema()->num_columns(); ++cid) {
auto cell = row->cell(cid);
- row->schema()->column(cid)->agg_finalize(&cell, arena);
+ row->schema()->column(cid)->agg_finalize(&cell, mem_pool);
}
}
template<typename RowType>
-void agg_finalize_row(const std::vector<uint32_t>& ids, RowType* row, Arena* arena) {
+void agg_finalize_row(const std::vector<uint32_t>& ids, RowType* row, MemPool* mem_pool) {
for (uint32_t id : ids) {
auto cell = row->cell(id);
- row->schema()->column(id)->agg_finalize(&cell, arena);
+ row->schema()->column(id)->agg_finalize(&cell, mem_pool);
}
}
diff --git a/be/src/olap/schema_change.cpp b/be/src/olap/schema_change.cpp
index b7eed21..43baf71 100644
--- a/be/src/olap/schema_change.cpp
+++ b/be/src/olap/schema_change.cpp
@@ -568,7 +568,8 @@ bool RowBlockMerger::merge(
uint64_t* merged_rows) {
uint64_t tmp_merged_rows = 0;
RowCursor row_cursor;
- std::unique_ptr<Arena> arena(new Arena());
+ std::unique_ptr<MemTracker> tracker(new MemTracker(-1));
+ std::unique_ptr<MemPool> mem_pool(new MemPool(tracker.get()));
std::unique_ptr<ObjectPool> agg_object_pool(new ObjectPool());
if (row_cursor.init(_tablet->tablet_schema()) != OLAP_SUCCESS) {
LOG(WARNING) << "fail to init row cursor.";
@@ -579,7 +580,7 @@ bool RowBlockMerger::merge(
row_cursor.allocate_memory_for_string_type(_tablet->tablet_schema());
while (_heap.size() > 0) {
- init_row_with_others(&row_cursor, *(_heap.top().row_cursor), arena.get(), agg_object_pool.get());
+ init_row_with_others(&row_cursor, *(_heap.top().row_cursor), mem_pool.get(), agg_object_pool.get());
if (!_pop_heap()) {
goto MERGE_ERR;
@@ -599,12 +600,12 @@ bool RowBlockMerger::merge(
goto MERGE_ERR;
}
}
- agg_finalize_row(&row_cursor, arena.get());
+ agg_finalize_row(&row_cursor, mem_pool.get());
rowset_writer->add_row(row_cursor);
- // the memory allocate by arena has been copied,
- // so we should release these memory immediately
- arena.reset(new Arena());
+ // the memory allocate by mem pool has been copied,
+ // so we should release memory immediately
+ mem_pool->clear();
agg_object_pool.reset(new ObjectPool());
}
if (rowset_writer->flush() != OLAP_SUCCESS) {
diff --git a/be/src/olap/skiplist.h b/be/src/olap/skiplist.h
index 86f74cb..fb1da64 100644
--- a/be/src/olap/skiplist.h
+++ b/be/src/olap/skiplist.h
@@ -31,14 +31,12 @@
#include "common/logging.h"
#include "gen_cpp/olap_file.pb.h"
-#include "util/arena.h"
+#include "runtime/mem_pool.h"
#include "util/random.h"
#include "olap/row.h"
namespace doris {
-class Arena;
-
template<typename Key, class Comparator>
class SkipList {
private:
@@ -48,7 +46,7 @@ public:
// Create a new SkipList object that will use "cmp" for comparing keys,
// and will allocate memory using "*arena". Objects allocated in the arena
// must remain allocated for the lifetime of the skiplist object.
- explicit SkipList(Comparator cmp, Arena* arena);
+ explicit SkipList(Comparator cmp, MemPool* mem_pool);
// Insert key into the list.
// REQUIRES: nothing that compares equal to key is currently in the list.
@@ -97,14 +95,12 @@ public:
// Intentionally copyable
};
- Arena* arena() const { return arena_; }
-
private:
enum { kMaxHeight = 12 };
// Immutable after construction
Comparator const compare_;
- Arena* const arena_; // Arena used for allocations of nodes
+ MemPool* const _mem_pool; // MemPool used for allocations of nodes
Node* const head_;
@@ -186,7 +182,7 @@ struct SkipList<Key,Comparator>::Node {
template<typename Key, class Comparator>
typename SkipList<Key,Comparator>::Node*
SkipList<Key,Comparator>::NewNode(const Key& key, int height) {
- char* mem = arena_->AllocateAligned(
+ char* mem = (char*)_mem_pool->allocate(
sizeof(Node) + sizeof(std::atomic<Node*>) * (height - 1));
return new (mem) Node(key);
}
@@ -326,9 +322,9 @@ SkipList<Key,Comparator>::FindLast() const {
}
template<typename Key, class Comparator>
-SkipList<Key,Comparator>::SkipList(Comparator cmp, Arena* arena)
+SkipList<Key,Comparator>::SkipList(Comparator cmp, MemPool* mem_pool)
: compare_(cmp),
- arena_(arena),
+ _mem_pool(mem_pool),
head_(NewNode(0 /* any key will do */, kMaxHeight)),
max_height_(1),
rnd_(0xdeadbeef) {
@@ -341,7 +337,7 @@ template<typename Key, class Comparator>
void SkipList<Key, Comparator>::Aggregate(const Key& k1, const Key& k2) {
ContiguousRow dst_row(compare_._schema, k1);
ContiguousRow src_row(compare_._schema, k2);
- agg_update_row(&dst_row, src_row, arena_);
+ agg_update_row(&dst_row, src_row, _mem_pool);
}
template<typename Key, class Comparator>
diff --git a/be/src/olap/task/engine_checksum_task.cpp b/be/src/olap/task/engine_checksum_task.cpp
index 99d2333..eb2f312 100644
--- a/be/src/olap/task/engine_checksum_task.cpp
+++ b/be/src/olap/task/engine_checksum_task.cpp
@@ -104,7 +104,8 @@ OLAPStatus EngineChecksumTask::_compute_checksum() {
}
RowCursor row;
- std::unique_ptr<Arena> arena(new Arena());
+ std::unique_ptr<MemTracker> tracker(new MemTracker(-1));
+ std::unique_ptr<MemPool> mem_pool(new MemPool(tracker.get()));
std::unique_ptr<ObjectPool> agg_object_pool(new ObjectPool());
res = row.init(tablet->tablet_schema(), reader_params.return_columns);
if (res != OLAP_SUCCESS) {
@@ -116,7 +117,7 @@ OLAPStatus EngineChecksumTask::_compute_checksum() {
bool eof = false;
uint32_t row_checksum = 0;
while (true) {
- OLAPStatus res = reader.next_row_with_aggregation(&row, arena.get(), agg_object_pool.get(), &eof);
+ OLAPStatus res = reader.next_row_with_aggregation(&row, mem_pool.get(), agg_object_pool.get(), &eof);
if (res == OLAP_SUCCESS && eof) {
VLOG(3) << "reader reads to the end.";
break;
@@ -126,9 +127,9 @@ OLAPStatus EngineChecksumTask::_compute_checksum() {
}
row_checksum = hash_row(row, row_checksum);
- // the memory allocate by arena has been copied,
- // so we should release these memory immediately
- arena.reset(new Arena());
+ // the memory allocate by mem pool has been copied,
+ // so we should release memory immediately
+ mem_pool->clear();
agg_object_pool.reset(new ObjectPool());
}
diff --git a/be/src/runtime/mem_tracker.cpp b/be/src/runtime/mem_tracker.cpp
index 6a650d3..3869c04 100644
--- a/be/src/runtime/mem_tracker.cpp
+++ b/be/src/runtime/mem_tracker.cpp
@@ -184,11 +184,6 @@ MemTracker* MemTracker::CreateQueryMemTracker(const TUniqueId& id,
}
MemTracker::~MemTracker() {
- DCHECK_EQ(_consumption->current_value(), 0) << _label << "\n"
- << get_stack_trace() << "\n"
- << LogUsage("");
- // TODO chenhao
- //DCHECK(_closed) << _label;
delete _reservation_counters.load();
}
diff --git a/be/test/olap/aggregate_func_test.cpp b/be/test/olap/aggregate_func_test.cpp
index 21f2cc6..044c077 100644
--- a/be/test/olap/aggregate_func_test.cpp
+++ b/be/test/olap/aggregate_func_test.cpp
@@ -22,6 +22,8 @@
#include "common/object_pool.h"
#include "olap/decimal12.h"
#include "olap/uint24.h"
+#include "runtime/mem_pool.h"
+#include "runtime/mem_tracker.h"
namespace doris {
@@ -37,7 +39,8 @@ void test_min() {
using CppType = typename CppTypeTraits<field_type>::CppType;
char buf[64];
- Arena arena;
+ std::unique_ptr<MemTracker> tracker(new MemTracker(-1));
+ std::unique_ptr<MemPool> mem_pool(new MemPool(tracker.get()));
ObjectPool agg_object_pool;
const AggregateInfo* agg = get_aggregate_info(OLAP_FIELD_AGGREGATION_MIN, field_type);
@@ -46,7 +49,7 @@ void test_min() {
{
char val_buf[16];
*(bool*)val_buf = true;
- agg->init(&dst, val_buf, true, &arena, &agg_object_pool);
+ agg->init(&dst, val_buf, true, mem_pool.get(), &agg_object_pool);
ASSERT_TRUE(*(bool*)(buf));
}
// 100
@@ -55,7 +58,7 @@ void test_min() {
*(bool*)val_buf = false;
CppType val = 100;
memcpy(val_buf + 1, &val, sizeof(CppType));
- agg->update(&dst, val_buf, &arena);
+ agg->update(&dst, val_buf, mem_pool.get());
ASSERT_FALSE(*(bool*)(buf));
memcpy(&val, buf + 1, sizeof(CppType));
ASSERT_EQ(100, val);
@@ -66,7 +69,7 @@ void test_min() {
*(bool*)val_buf = false;
CppType val = 200;
memcpy(val_buf + 1, &val, sizeof(CppType));
- agg->update(&dst, val_buf, &arena);
+ agg->update(&dst, val_buf, mem_pool.get());
ASSERT_FALSE(*(bool*)(buf));
memcpy(&val, buf + 1, sizeof(CppType));
ASSERT_EQ(100, val);
@@ -77,7 +80,7 @@ void test_min() {
*(bool*)val_buf = false;
CppType val = 50;
memcpy(val_buf + 1, &val, sizeof(CppType));
- agg->update(&dst, val_buf, &arena);
+ agg->update(&dst, val_buf, mem_pool.get());
ASSERT_FALSE(*(bool*)(buf));
memcpy(&val, buf + 1, sizeof(CppType));
ASSERT_EQ(50, val);
@@ -86,14 +89,14 @@ void test_min() {
{
char val_buf[16];
*(bool*)val_buf = true;
- agg->update(&dst, val_buf, &arena);
+ agg->update(&dst, val_buf, mem_pool.get());
ASSERT_FALSE(*(bool*)(buf));
CppType val;
memcpy(val_buf + 1, &val, sizeof(CppType));
memcpy(&val, buf + 1, sizeof(CppType));
ASSERT_EQ(50, val);
}
- agg->finalize(&dst, &arena);
+ agg->finalize(&dst, mem_pool.get());
ASSERT_FALSE(*(bool*)(buf));
CppType val;
memcpy(&val, buf + 1, sizeof(CppType));
@@ -111,7 +114,8 @@ void test_max() {
char buf[64];
- Arena arena;
+ std::unique_ptr<MemTracker> tracker(new MemTracker(-1));
+ std::unique_ptr<MemPool> mem_pool(new MemPool(tracker.get()));
ObjectPool agg_object_pool;
const AggregateInfo* agg = get_aggregate_info(OLAP_FIELD_AGGREGATION_MAX, field_type);
@@ -120,7 +124,7 @@ void test_max() {
{
char val_buf[16];
*(bool*)val_buf = true;
- agg->init(&dst, val_buf, true, &arena, &agg_object_pool);
+ agg->init(&dst, val_buf, true, mem_pool.get(), &agg_object_pool);
ASSERT_TRUE(*(bool*)(buf));
}
// 100
@@ -129,7 +133,7 @@ void test_max() {
*(bool*)val_buf = false;
CppType val = 100;
memcpy(val_buf + 1, &val, sizeof(CppType));
- agg->update(&dst, val_buf, &arena);
+ agg->update(&dst, val_buf, mem_pool.get());
ASSERT_FALSE(*(bool*)(buf));
memcpy(&val, buf + 1, sizeof(CppType));
ASSERT_EQ(100, val);
@@ -140,7 +144,7 @@ void test_max() {
*(bool*)val_buf = false;
CppType val = 200;
memcpy(val_buf + 1, &val, sizeof(CppType));
- agg->update(&dst, val_buf, &arena);
+ agg->update(&dst, val_buf, mem_pool.get());
ASSERT_FALSE(*(bool*)(buf));
memcpy(&val, buf + 1, sizeof(CppType));
ASSERT_EQ(200, val);
@@ -151,7 +155,7 @@ void test_max() {
*(bool*)val_buf = false;
CppType val = 50;
memcpy(val_buf + 1, &val, sizeof(CppType));
- agg->update(&dst, val_buf, &arena);
+ agg->update(&dst, val_buf, mem_pool.get());
ASSERT_FALSE(*(bool*)(buf));
memcpy(&val, buf + 1, sizeof(CppType));
ASSERT_EQ(200, val);
@@ -160,13 +164,13 @@ void test_max() {
{
char val_buf[16];
*(bool*)val_buf = true;
- agg->update(&dst, val_buf, &arena);
+ agg->update(&dst, val_buf, mem_pool.get());
ASSERT_FALSE(*(bool*)(buf));
CppType val;
memcpy(&val, buf + 1, sizeof(CppType));
ASSERT_EQ(200, val);
}
- agg->finalize(&dst, &arena);
+ agg->finalize(&dst, mem_pool.get());
ASSERT_FALSE(*(bool*)(buf));
CppType val;
memcpy(&val, buf + 1, sizeof(CppType));
@@ -185,7 +189,8 @@ void test_sum() {
char buf[64];
RowCursorCell dst(buf);
- Arena arena;
+ std::unique_ptr<MemTracker> tracker(new MemTracker(-1));
+ std::unique_ptr<MemPool> mem_pool(new MemPool(tracker.get()));
ObjectPool agg_object_pool;
const AggregateInfo* agg = get_aggregate_info(OLAP_FIELD_AGGREGATION_SUM, field_type);
@@ -193,7 +198,7 @@ void test_sum() {
{
char val_buf[16];
*(bool*)val_buf = true;
- agg->init(&dst, val_buf, true, &arena, &agg_object_pool);
+ agg->init(&dst, val_buf, true, mem_pool.get(), &agg_object_pool);
ASSERT_TRUE(*(bool*)(buf));
}
// 100
@@ -202,7 +207,7 @@ void test_sum() {
*(bool*)val_buf = false;
CppType val = 100;
memcpy(val_buf + 1, &val, sizeof(CppType));
- agg->update(&dst, val_buf, &arena);
+ agg->update(&dst, val_buf, mem_pool.get());
ASSERT_FALSE(*(bool*)(buf));
memcpy(&val, buf + 1, sizeof(CppType));
ASSERT_EQ(100, val);
@@ -213,7 +218,7 @@ void test_sum() {
*(bool*)val_buf = false;
CppType val = 200;
memcpy(val_buf + 1, &val, sizeof(CppType));
- agg->update(&dst, val_buf, &arena);
+ agg->update(&dst, val_buf, mem_pool.get());
ASSERT_FALSE(*(bool*)(buf));
memcpy(&val, buf + 1, sizeof(CppType));
ASSERT_EQ(300, val);
@@ -224,7 +229,7 @@ void test_sum() {
*(bool*)val_buf = false;
CppType val = 50;
memcpy(val_buf + 1, &val, sizeof(CppType));
- agg->update(&dst, val_buf, &arena);
+ agg->update(&dst, val_buf, mem_pool.get());
ASSERT_FALSE(*(bool*)(buf));
memcpy(&val, buf + 1, sizeof(CppType));
ASSERT_EQ(350, val);
@@ -233,13 +238,13 @@ void test_sum() {
{
char val_buf[16];
*(bool*)val_buf = true;
- agg->update(&dst, val_buf, &arena);
+ agg->update(&dst, val_buf, mem_pool.get());
ASSERT_FALSE(*(bool*)(buf));
CppType val;
memcpy(&val, buf + 1, sizeof(CppType));
ASSERT_EQ(350, val);
}
- agg->finalize(&dst, &arena);
+ agg->finalize(&dst, mem_pool.get());
ASSERT_FALSE(*(bool*)(buf));
CppType val;
memcpy(&val, buf + 1, sizeof(CppType));
@@ -258,7 +263,8 @@ void test_replace() {
char buf[64];
RowCursorCell dst(buf);
- Arena arena;
+ std::unique_ptr<MemTracker> tracker(new MemTracker(-1));
+ std::unique_ptr<MemPool> mem_pool(new MemPool(tracker.get()));
ObjectPool agg_object_pool;
const AggregateInfo* agg = get_aggregate_info(OLAP_FIELD_AGGREGATION_REPLACE, field_type);
@@ -266,7 +272,7 @@ void test_replace() {
{
char val_buf[16];
*(bool*)val_buf = true;
- agg->init(&dst, val_buf, true, &arena, &agg_object_pool);
+ agg->init(&dst, val_buf, true, mem_pool.get(), &agg_object_pool);
ASSERT_TRUE(*(bool*)(buf));
}
// 100
@@ -275,7 +281,7 @@ void test_replace() {
*(bool*)val_buf = false;
CppType val = 100;
memcpy(val_buf + 1, &val, sizeof(CppType));
- agg->update(&dst, val_buf, &arena);
+ agg->update(&dst, val_buf, mem_pool.get());
ASSERT_FALSE(*(bool*)(buf));
memcpy(&val, buf + 1, sizeof(CppType));
ASSERT_EQ(100, val);
@@ -284,7 +290,7 @@ void test_replace() {
{
char val_buf[16];
*(bool*)val_buf = true;
- agg->update(&dst, val_buf, &arena);
+ agg->update(&dst, val_buf, mem_pool.get());
ASSERT_TRUE(*(bool*)(buf));
}
// 50
@@ -293,12 +299,12 @@ void test_replace() {
*(bool*)val_buf = false;
CppType val = 50;
memcpy(val_buf + 1, &val, sizeof(CppType));
- agg->update(&dst, val_buf, &arena);
+ agg->update(&dst, val_buf, mem_pool.get());
ASSERT_FALSE(*(bool*)(buf));
memcpy(&val, buf + 1, sizeof(CppType));
ASSERT_EQ(50, val);
}
- agg->finalize(&dst, &arena);
+ agg->finalize(&dst, mem_pool.get());
ASSERT_FALSE(*(bool*)(buf));
CppType val;
memcpy(&val, buf + 1, sizeof(CppType));
@@ -316,7 +322,8 @@ void test_replace_string() {
dst_slice->data = nullptr;
dst_slice->size = 0;
- Arena arena;
+ std::unique_ptr<MemTracker> tracker(new MemTracker(-1));
+ std::unique_ptr<MemPool> mem_pool(new MemPool(tracker.get()));
ObjectPool agg_object_pool;
const AggregateInfo* agg = get_aggregate_info(OLAP_FIELD_AGGREGATION_REPLACE, field_type);
@@ -326,7 +333,7 @@ void test_replace_string() {
// null
{
src_cell.set_null();
- agg->init(&dst_cell, (const char*)src_slice, true, &arena, &agg_object_pool);
+ agg->init(&dst_cell, (const char*)src_slice, true, mem_pool.get(), &agg_object_pool);
ASSERT_TRUE(dst_cell.is_null());
}
// "12345"
@@ -334,7 +341,7 @@ void test_replace_string() {
src_cell.set_not_null();
src_slice->data = (char*)"1234567890";
src_slice->size = 10;
- agg->update(&dst_cell, src_cell, &arena);
+ agg->update(&dst_cell, src_cell, mem_pool.get());
ASSERT_FALSE(dst_cell.is_null());
ASSERT_EQ(10, dst_slice->size);
ASSERT_STREQ("1234567890", dst_slice->to_string().c_str());
@@ -344,7 +351,7 @@ void test_replace_string() {
src_cell.set_not_null();
src_slice->data = (char*)"abc";
src_slice->size = 3;
- agg->update(&dst_cell, src_cell, &arena);
+ agg->update(&dst_cell, src_cell, mem_pool.get());
ASSERT_FALSE(dst_cell.is_null());
ASSERT_EQ(3, dst_slice->size);
ASSERT_STREQ("abc", dst_slice->to_string().c_str());
@@ -352,7 +359,7 @@ void test_replace_string() {
// null
{
src_cell.set_null();
- agg->update(&dst_cell, src_cell, &arena);
+ agg->update(&dst_cell, src_cell, mem_pool.get());
ASSERT_TRUE(dst_cell.is_null());
}
// "12345"
@@ -360,13 +367,13 @@ void test_replace_string() {
src_cell.set_not_null();
src_slice->data = (char*)"12345";
src_slice->size = 5;
- agg->update(&dst_cell, src_cell, &arena);
+ agg->update(&dst_cell, src_cell, mem_pool.get());
ASSERT_FALSE(dst_cell.is_null());
ASSERT_EQ(5, dst_slice->size);
ASSERT_STREQ("12345", dst_slice->to_string().c_str());
}
- agg->finalize(&dst_cell, &arena);
+ agg->finalize(&dst_cell, mem_pool.get());
ASSERT_FALSE(dst_cell.is_null());
ASSERT_EQ(5, dst_slice->size);
ASSERT_STREQ("12345", dst_slice->to_string().c_str());
diff --git a/be/test/olap/row_cursor_test.cpp b/be/test/olap/row_cursor_test.cpp
index cd26618..2a39e6d 100644
--- a/be/test/olap/row_cursor_test.cpp
+++ b/be/test/olap/row_cursor_test.cpp
@@ -448,7 +448,7 @@ TEST_F(TestRowCursor, AggregateWithoutNull) {
set_tablet_schema_for_cmp_and_aggregate(&tablet_schema);
RowCursor row;
- std::unique_ptr<Arena> arena(new Arena());
+
OLAPStatus res = row.init(tablet_schema);
ASSERT_EQ(res, OLAP_SUCCESS);
ASSERT_EQ(row.get_fixed_len(), 78);
@@ -471,8 +471,10 @@ TEST_F(TestRowCursor, AggregateWithoutNull) {
left.set_field_content(4, reinterpret_cast<char*>(&l_decimal), _mem_pool.get());
left.set_field_content(5, reinterpret_cast<char*>(&l_varchar), _mem_pool.get());
+ std::unique_ptr<MemTracker> tracker(new MemTracker(-1));
+ std::unique_ptr<MemPool> mem_pool(new MemPool(tracker.get()));
ObjectPool agg_object_pool;
- init_row_with_others(&row, left, arena.get(), &agg_object_pool);
+ init_row_with_others(&row, left, mem_pool.get(), &agg_object_pool);
RowCursor right;
res = right.init(tablet_schema);
@@ -509,7 +511,7 @@ TEST_F(TestRowCursor, AggregateWithNull) {
set_tablet_schema_for_cmp_and_aggregate(&tablet_schema);
RowCursor row;
- std::unique_ptr<Arena> arena(new Arena());
+
OLAPStatus res = row.init(tablet_schema);
ASSERT_EQ(res, OLAP_SUCCESS);
ASSERT_EQ(row.get_fixed_len(), 78);
@@ -530,8 +532,10 @@ TEST_F(TestRowCursor, AggregateWithNull) {
left.set_null(4);
left.set_field_content(5, reinterpret_cast<char*>(&l_varchar), _mem_pool.get());
+ std::unique_ptr<MemTracker> tracker(new MemTracker(-1));
+ std::unique_ptr<MemPool> mem_pool(new MemPool(tracker.get()));
ObjectPool agg_object_pool;
- init_row_with_others(&row, left, arena.get(), &agg_object_pool);
+ init_row_with_others(&row, left, mem_pool.get(), &agg_object_pool);
RowCursor right;
res = right.init(tablet_schema);
diff --git a/be/test/olap/skiplist_test.cpp b/be/test/olap/skiplist_test.cpp
index fb82c13..5b73734 100644
--- a/be/test/olap/skiplist_test.cpp
+++ b/be/test/olap/skiplist_test.cpp
@@ -23,7 +23,8 @@
#include <gtest/gtest.h>
#include "olap/schema.h"
#include "olap/utils.h"
-#include "util/arena.h"
+#include "runtime/mem_pool.h"
+#include "runtime/mem_tracker.h"
#include "util/hash_util.hpp"
#include "util/random.h"
#include "util/thread_pool.hpp"
@@ -48,9 +49,11 @@ struct TestComparator {
class SkipTest : public testing::Test {};
TEST_F(SkipTest, Empty) {
- Arena arena;
+ std::unique_ptr<MemTracker> tracker(new MemTracker(-1));
+ std::unique_ptr<MemPool> mem_pool(new MemPool(tracker.get()));
+
TestComparator cmp;
- SkipList<Key, TestComparator> list(cmp, &arena);
+ SkipList<Key, TestComparator> list(cmp, mem_pool.get());
ASSERT_TRUE(!list.Contains(10));
SkipList<Key, TestComparator>::Iterator iter(&list);
@@ -64,13 +67,15 @@ TEST_F(SkipTest, Empty) {
}
TEST_F(SkipTest, InsertAndLookup) {
+ std::unique_ptr<MemTracker> tracker(new MemTracker(-1));
+ std::unique_ptr<MemPool> mem_pool(new MemPool(tracker.get()));
+
const int N = 2000;
const int R = 5000;
Random rnd(1000);
std::set<Key> keys;
- Arena arena;
TestComparator cmp;
- SkipList<Key, TestComparator> list(cmp, &arena);
+ SkipList<Key, TestComparator> list(cmp, mem_pool.get());
for (int i = 0; i < N; i++) {
Key key = rnd.Next() % R;
if (keys.insert(key).second) {
@@ -222,14 +227,18 @@ private:
// Current state of the test
State _current;
- Arena _arena;
+ std::unique_ptr<MemTracker> _mem_tracker;
+ std::unique_ptr<MemPool> _mem_pool;
// SkipList is not protected by _mu. We just use a single writer
// thread to modify it.
SkipList<Key, TestComparator> _list;
public:
- ConcurrentTest() : _list(TestComparator(), &_arena) {}
+ ConcurrentTest():
+ _mem_tracker(new MemTracker(-1)),
+ _mem_pool(new MemPool(_mem_tracker.get())),
+ _list(TestComparator(), _mem_pool.get()){}
// REQUIRES: External synchronization
void write_step(Random* rnd) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org