github-actions[bot] commented on code in PR #16870:
URL: https://github.com/apache/doris/pull/16870#discussion_r1109476755
##########
be/src/exec/base_scanner.cpp:
##########
@@ -250,6 +265,28 @@
std::move(column_ptr), slot_desc->get_data_type_ptr(),
slot_desc->col_name()));
}
+ // handle dynamic generated columns
+ if (!_full_base_schema_view->empty()) {
+ assert(_is_dynamic_schema);
+ for (size_t x = dest_block->columns(); x < _src_block.columns(); ++x) {
+ auto& column_type_name = _src_block.get_by_position(x);
+ const TColumn& tcolumn =
+
_full_base_schema_view->column_name_to_column[column_type_name.name];
+ auto original_type =
vectorized::DataTypeFactory::instance().create_data_type(tcolumn);
+ // type conflict free path, always cast to original type
+ if (!column_type_name.type->equals(*original_type)) {
+ vectorized::ColumnPtr column_ptr;
+
RETURN_IF_ERROR(vectorized::schema_util::cast_column(column_type_name,
+
original_type, &column_ptr));
+ column_type_name.column = column_ptr;
+ column_type_name.type = original_type;
+ }
+
dest_block->insert(vectorized::ColumnWithTypeAndName(std::move(column_type_name.column),
+
std::move(column_type_name.type),
Review Comment:
warning: passing result of std::move() as a const reference argument; no
move will actually happen [performance-move-const-arg]
```suggestion
column_type_name.type,
```
##########
be/src/olap/rowset/segment_v2/inverted_index_cache.h:
##########
@@ -163,5 +164,102 @@ class InvertedIndexCacheHandle {
DISALLOW_COPY_AND_ASSIGN(InvertedIndexCacheHandle);
};
+enum class InvertedIndexQueryType;
+
+class InvertedIndexQueryCacheHandle;
+
+class InvertedIndexQueryCache {
+public:
+ // cache key
+ struct CacheKey {
+ io::Path index_path; // index file path
+ std::string column_name; // column name
+ InvertedIndexQueryType query_type; // query type
+ std::wstring value; // query value
+
+ // Encode to a flat binary which can be used as LRUCache's key
+ std::string encode() const {
+ std::string key_buf(index_path.string());
+ key_buf.append("/");
+ key_buf.append(column_name);
+ key_buf.append("/");
+ key_buf.append(1, static_cast<char>(query_type));
+ key_buf.append("/");
+ key_buf.append(lucene::util::Misc::toString(value.c_str()));
+ return key_buf;
+ }
+ };
+
+ using CacheValue = roaring::Roaring;
+
+ // Create global instance of this class
+ static void create_global_cache(size_t capacity, int32_t
index_cache_percentage,
+ uint32_t num_shards = 16) {
+ DCHECK(_s_instance == nullptr);
+ static InvertedIndexQueryCache instance(capacity,
index_cache_percentage, num_shards);
+ _s_instance = &instance;
+ }
+
+ // Return global instance.
+ // Client should call create_global_cache before.
+ static InvertedIndexQueryCache* instance() { return _s_instance; }
+
+ InvertedIndexQueryCache() = delete;
+
+ InvertedIndexQueryCache(size_t capacity, int32_t index_cache_percentage,
uint32_t num_shards) {
+ _cache = std::unique_ptr<Cache>(
+ new_lru_cache("InvertedIndexQueryCache", capacity,
LRUCacheType::SIZE, num_shards));
+ }
+
+ bool lookup(const CacheKey& key, InvertedIndexQueryCacheHandle* handle);
+
+ void insert(const CacheKey& key, roaring::Roaring* bitmap,
+ InvertedIndexQueryCacheHandle* handle);
+
+private:
+ static InvertedIndexQueryCache* _s_instance;
+ std::unique_ptr<Cache> _cache {nullptr};
+};
+
+class InvertedIndexQueryCacheHandle {
+public:
+ InvertedIndexQueryCacheHandle() {}
Review Comment:
warning: use '= default' to define a trivial default constructor
[modernize-use-equals-default]
```suggestion
InvertedIndexQueryCacheHandle() = default;
```
##########
be/src/olap/task/engine_alter_tablet_task.h:
##########
@@ -39,4 +39,18 @@
std::shared_ptr<MemTrackerLimiter> _mem_tracker;
}; // EngineTask
+class EngineAlterInvertedIndexTask : public EngineTask {
+public:
+ virtual Status execute();
+
+public:
Review Comment:
warning: redundant access specifier has the same accessibility as the
previous access specifier [readability-redundant-access-specifiers]
```suggestion
```
**be/src/olap/task/engine_alter_tablet_task.h:42:** previously declared here
```cpp
public:
^
```
##########
be/src/exec/base_scanner.cpp:
##########
@@ -250,6 +265,28 @@ Status
BaseScanner::_materialize_dest_block(vectorized::Block* dest_block) {
std::move(column_ptr), slot_desc->get_data_type_ptr(),
slot_desc->col_name()));
}
+ // handle dynamic generated columns
+ if (!_full_base_schema_view->empty()) {
+ assert(_is_dynamic_schema);
+ for (size_t x = dest_block->columns(); x < _src_block.columns(); ++x) {
+ auto& column_type_name = _src_block.get_by_position(x);
+ const TColumn& tcolumn =
+
_full_base_schema_view->column_name_to_column[column_type_name.name];
+ auto original_type =
vectorized::DataTypeFactory::instance().create_data_type(tcolumn);
+ // type conflict free path, always cast to original type
+ if (!column_type_name.type->equals(*original_type)) {
+ vectorized::ColumnPtr column_ptr;
+
RETURN_IF_ERROR(vectorized::schema_util::cast_column(column_type_name,
+
original_type, &column_ptr));
+ column_type_name.column = column_ptr;
+ column_type_name.type = original_type;
+ }
+
dest_block->insert(vectorized::ColumnWithTypeAndName(std::move(column_type_name.column),
Review Comment:
warning: passing result of std::move() as a const reference argument; no
move will actually happen [performance-move-const-arg]
```suggestion
dest_block->insert(vectorized::ColumnWithTypeAndName(column_type_name.column,
```
##########
be/src/pipeline/exec/exchange_sink_buffer.cpp:
##########
@@ -155,58 +162,92 @@ Status ExchangeSinkBuffer::add_block(TransmitInfo&&
request) {
return Status::OK();
}
+Status ExchangeSinkBuffer::add_block(BroadcastTransmitInfo&& request) {
+ if (_is_finishing) {
+ return Status::OK();
+ }
+ TUniqueId ins_id = request.channel->_fragment_instance_id;
+ bool send_now = false;
+ request.block_holder->ref();
+ {
+ std::unique_lock<std::mutex>
lock(*_instance_to_package_queue_mutex[ins_id.lo]);
+ // Do not have in process rpc, directly send
+ if (_instance_to_sending_by_pipeline[ins_id.lo]) {
+ send_now = true;
+ _instance_to_sending_by_pipeline[ins_id.lo] = false;
+ }
+
_instance_to_broadcast_package_queue[ins_id.lo].emplace(std::move(request));
Review Comment:
warning: std::move of the variable 'request' of the trivially-copyable type
'doris::pipeline::BroadcastTransmitInfo' has no effect; remove std::move()
[performance-move-const-arg]
```suggestion
_instance_to_broadcast_package_queue[ins_id.lo].emplace(request);
```
##########
be/src/olap/schema_change.cpp:
##########
@@ -576,6 +579,232 @@
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {
+ _olap_data_convertor =
std::make_unique<vectorized::OlapBlockDataConvertor>();
+}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+ _olap_data_convertor.reset();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr
rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr
base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() ||
rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ _olap_data_convertor->reserve(_alter_inverted_indexs.size());
+
+ // load segments
+ SegmentCacheHandle segment_cache_handle;
+ RETURN_NOT_OK(SegmentLoader::instance()->load_segments(
+ std::static_pointer_cast<BetaRowset>(rowset_reader->rowset()),
&segment_cache_handle,
+ false));
+
+ for (auto& seg_ptr : segment_cache_handle.get_segments()) {
+ std::string segment_filename =
+ fmt::format("{}_{}.dat", rowset_meta->rowset_id().to_string(),
seg_ptr->id());
+ std::vector<ColumnId> return_columns;
+ std::vector<std::pair<int64_t, int64_t>> inverted_index_writer_signs;
+ // create inverted index writer
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto index_id = inverted_index.index_id;
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto column_idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(column_idx);
+ _olap_data_convertor->add_column_data_convertor(column);
+
+ std::unique_ptr<Field> field(FieldFactory::create(column));
+ _index_metas.emplace_back(new TabletIndex());
+ _index_metas.back()->init_from_thrift(inverted_index,
*_tablet_schema);
+ std::unique_ptr<segment_v2::InvertedIndexColumnWriter>
inverted_index_builder;
+ try {
+ RETURN_IF_ERROR(segment_v2::InvertedIndexColumnWriter::create(
+ field.get(), &inverted_index_builder,
segment_filename, segment_dir,
+ _index_metas.back().get(), fs));
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ if (inverted_index_builder) {
+ auto writer_sign = std::make_pair(seg_ptr->id(), index_id);
+ _inverted_index_builders.insert(
+ std::make_pair(writer_sign,
std::move(inverted_index_builder)));
+ inverted_index_writer_signs.push_back(writer_sign);
+ }
+ }
+
+ // create iterator for each segment
+ StorageReadOptions read_options;
+ OlapReaderStatistics stats;
+ read_options.stats = &stats;
+ read_options.tablet_schema = _tablet_schema;
+ std::unique_ptr<Schema> schema =
+ std::make_unique<Schema>(_tablet_schema->columns(),
return_columns);
+ std::unique_ptr<RowwiseIterator> iter;
+ res = seg_ptr->new_iterator(*schema, read_options, &iter);
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to create iterator[" << seg_ptr->id()
+ << "]: " << res.to_string();
+ return Status::Error<ROWSET_READER_INIT>();
+ }
+
+ std::shared_ptr<vectorized::Block> block =
+
std::make_shared<vectorized::Block>(_tablet_schema->create_block(return_columns));
+ do {
+ res = iter->next_batch(block.get());
+ if (!res.ok()) {
+ if (res.is<END_OF_FILE>()) {
+ break;
+ }
+ RETURN_NOT_OK_LOG(
+ res, "failed to read next block when schema change for
inverted index.");
+ }
+
+ // write inverted index
+ if (_write_inverted_index(iter->data_id(), block.get()) !=
Status::OK()) {
+ res = Status::Error<SCHEMA_CHANGE_INFO_INVALID>();
+ LOG(WARNING) << "failed to write block.";
+ return res;
+ }
+ block->clear_column_data();
+ } while (true);
+
+ // finish write inverted index, flush data to compound file
+ for (auto& writer_sign : inverted_index_writer_signs) {
+ try {
+ if (_inverted_index_builders[writer_sign]) {
+ _inverted_index_builders[writer_sign]->finish();
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+ }
+ }
+
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+
+ LOG(INFO) << "all row nums. source_rows=" <<
rowset_reader->rowset()->num_rows();
+ return res;
+}
+
+Status SchemaChangeForInvertedIndex::_add_nullable(
+ const std::string& column_name, const std::pair<int64_t, int64_t>&
index_writer_sign,
+ Field* field, const uint8_t* null_map, const uint8_t** ptr, size_t
num_rows) {
+ size_t offset = 0;
+ auto next_run_step = [&]() {
+ size_t step = 1;
+ for (auto i = offset + 1; i < num_rows; ++i) {
+ if (null_map[offset] == null_map[i])
Review Comment:
warning: statement should be inside braces
[readability-braces-around-statements]
```suggestion
if (null_map[offset] == null_map[i]) {
```
be/src/olap/schema_change.cpp:720:
```diff
- else
+ } else
```
##########
be/src/util/priority_work_stealing_thread_pool.hpp:
##########
@@ -97,7 +98,7 @@ class PriorityWorkStealingThreadPool : public
PriorityThreadPool {
// Any work Offer()'ed during DrainAndshutdown may or may not be processed.
void drain_and_shutdown() override {
Review Comment:
warning: only virtual member functions can be marked 'override'
[clang-diagnostic-error]
```suggestion
void drain_and_shutdown() {
```
##########
be/src/util/once.h:
##########
@@ -54,10 +55,16 @@ class DorisCallOnce {
// lambda and stores its return value. Otherwise, returns the stored
Status.
template <typename Fn>
ReturnType call(Fn fn) {
- std::call_once(_once_flag, [this, fn] {
- _status = fn();
- _has_called.store(true, std::memory_order_release);
- });
+ if (!_has_called.load(std::memory_order_acquire)) {
+ do {
+ std::lock_guard l(_mutex);
+ if (_has_called.load(std::memory_order_acquire)) break;
Review Comment:
warning: statement should be inside braces
[readability-braces-around-statements]
```suggestion
if (_has_called.load(std::memory_order_acquire)) { break;
}
```
##########
be/src/util/priority_work_stealing_thread_pool.hpp:
##########
@@ -97,7 +98,7 @@
// Any work Offer()'ed during DrainAndshutdown may or may not be processed.
void drain_and_shutdown() override {
{
- std::unique_lock<std::mutex> l(_lock);
+ std::unique_lock l(_lock);
Review Comment:
warning: use of undeclared identifier '_lock'; did you mean 'clock'?
[clang-diagnostic-error]
```suggestion
std::unique_lock l(clock);
```
**/usr/include/time.h:71:** 'clock' declared here
```cpp
extern clock_t clock (void) __THROW;
^
```
##########
be/src/vec/data_types/get_least_supertype.cpp:
##########
@@ -50,12 +50,152 @@ String get_exception_message_prefix(const DataTypes&
types) {
}
} // namespace
-DataTypePtr get_least_supertype(const DataTypes& types) {
+Status get_numeric_type(const TypeIndexSet& types, DataTypePtr* type) {
+ bool all_numbers = true;
+
+ size_t max_bits_of_signed_integer = 0;
+ size_t max_bits_of_unsigned_integer = 0;
+ size_t max_mantissa_bits_of_floating = 0;
+
+ auto maximize = [](size_t& what, size_t value) {
+ if (value > what) what = value;
Review Comment:
warning: statement should be inside braces
[readability-braces-around-statements]
```suggestion
if (value > what) { what = value;
}
```
##########
be/src/olap/schema_change.cpp:
##########
@@ -576,6 +579,232 @@ Status
VSchemaChangeWithSorting::_external_sorting(vector<RowsetSharedPtr>& src_
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
Review Comment:
warning: initializer for base class 'doris::SchemaChange' is redundant
[readability-redundant-member-init]
```suggestion
: ,
```
##########
be/src/olap/schema_change.cpp:
##########
@@ -576,6 +579,232 @@
return Status::OK();
}
+SchemaChangeForInvertedIndex::SchemaChangeForInvertedIndex(
+ const std::vector<TOlapTableIndex>& alter_inverted_indexs,
+ const TabletSchemaSPtr& tablet_schema)
+ : SchemaChange(),
+ _alter_inverted_indexs(alter_inverted_indexs),
+ _tablet_schema(tablet_schema) {
+ _olap_data_convertor =
std::make_unique<vectorized::OlapBlockDataConvertor>();
+}
+
+SchemaChangeForInvertedIndex::~SchemaChangeForInvertedIndex() {
+ VLOG_NOTICE << "~SchemaChangeForInvertedIndex()";
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+ _olap_data_convertor.reset();
+}
+
+Status SchemaChangeForInvertedIndex::process(RowsetReaderSharedPtr
rowset_reader,
+ RowsetWriter* rowset_writer,
+ TabletSharedPtr new_tablet,
+ TabletSharedPtr base_tablet,
+ TabletSchemaSPtr
base_tablet_schema) {
+ Status res = Status::OK();
+ if (rowset_reader->rowset()->empty() ||
rowset_reader->rowset()->num_rows() == 0) {
+ return Status::OK();
+ }
+
+ // create inverted index writer
+ auto rowset_meta = rowset_reader->rowset()->rowset_meta();
+ std::string segment_dir = base_tablet->tablet_path();
+ auto fs = rowset_meta->fs();
+ _olap_data_convertor->reserve(_alter_inverted_indexs.size());
+
+ // load segments
+ SegmentCacheHandle segment_cache_handle;
+ RETURN_NOT_OK(SegmentLoader::instance()->load_segments(
+ std::static_pointer_cast<BetaRowset>(rowset_reader->rowset()),
&segment_cache_handle,
+ false));
+
+ for (auto& seg_ptr : segment_cache_handle.get_segments()) {
+ std::string segment_filename =
+ fmt::format("{}_{}.dat", rowset_meta->rowset_id().to_string(),
seg_ptr->id());
+ std::vector<ColumnId> return_columns;
+ std::vector<std::pair<int64_t, int64_t>> inverted_index_writer_signs;
+ // create inverted index writer
+ for (auto& inverted_index : _alter_inverted_indexs) {
+ DCHECK_EQ(inverted_index.columns.size(), 1);
+ auto index_id = inverted_index.index_id;
+ auto column_name = inverted_index.columns[0];
+ auto column = _tablet_schema->column(column_name);
+ auto column_idx = _tablet_schema->field_index(column_name);
+ return_columns.emplace_back(column_idx);
+ _olap_data_convertor->add_column_data_convertor(column);
+
+ std::unique_ptr<Field> field(FieldFactory::create(column));
+ _index_metas.emplace_back(new TabletIndex());
+ _index_metas.back()->init_from_thrift(inverted_index,
*_tablet_schema);
+ std::unique_ptr<segment_v2::InvertedIndexColumnWriter>
inverted_index_builder;
+ try {
+ RETURN_IF_ERROR(segment_v2::InvertedIndexColumnWriter::create(
+ field.get(), &inverted_index_builder,
segment_filename, segment_dir,
+ _index_metas.back().get(), fs));
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+
+ if (inverted_index_builder) {
+ auto writer_sign = std::make_pair(seg_ptr->id(), index_id);
+ _inverted_index_builders.insert(
+ std::make_pair(writer_sign,
std::move(inverted_index_builder)));
+ inverted_index_writer_signs.push_back(writer_sign);
+ }
+ }
+
+ // create iterator for each segment
+ StorageReadOptions read_options;
+ OlapReaderStatistics stats;
+ read_options.stats = &stats;
+ read_options.tablet_schema = _tablet_schema;
+ std::unique_ptr<Schema> schema =
+ std::make_unique<Schema>(_tablet_schema->columns(),
return_columns);
+ std::unique_ptr<RowwiseIterator> iter;
+ res = seg_ptr->new_iterator(*schema, read_options, &iter);
+ if (!res.ok()) {
+ LOG(WARNING) << "failed to create iterator[" << seg_ptr->id()
+ << "]: " << res.to_string();
+ return Status::Error<ROWSET_READER_INIT>();
+ }
+
+ std::shared_ptr<vectorized::Block> block =
+
std::make_shared<vectorized::Block>(_tablet_schema->create_block(return_columns));
+ do {
+ res = iter->next_batch(block.get());
+ if (!res.ok()) {
+ if (res.is<END_OF_FILE>()) {
+ break;
+ }
+ RETURN_NOT_OK_LOG(
+ res, "failed to read next block when schema change for
inverted index.");
+ }
+
+ // write inverted index
+ if (_write_inverted_index(iter->data_id(), block.get()) !=
Status::OK()) {
+ res = Status::Error<SCHEMA_CHANGE_INFO_INVALID>();
+ LOG(WARNING) << "failed to write block.";
+ return res;
+ }
+ block->clear_column_data();
+ } while (true);
+
+ // finish write inverted index, flush data to compound file
+ for (auto& writer_sign : inverted_index_writer_signs) {
+ try {
+ if (_inverted_index_builders[writer_sign]) {
+ _inverted_index_builders[writer_sign]->finish();
+ }
+ } catch (const std::exception& e) {
+ LOG(WARNING) << "CLuceneError occured: " << e.what();
+ return Status::Error<IO_ERROR>();
+ }
+ }
+ }
+
+ _inverted_index_builders.clear();
+ _index_metas.clear();
+
+ LOG(INFO) << "all row nums. source_rows=" <<
rowset_reader->rowset()->num_rows();
+ return res;
+}
+
+Status SchemaChangeForInvertedIndex::_add_nullable(
+ const std::string& column_name, const std::pair<int64_t, int64_t>&
index_writer_sign,
+ Field* field, const uint8_t* null_map, const uint8_t** ptr, size_t
num_rows) {
+ size_t offset = 0;
+ auto next_run_step = [&]() {
+ size_t step = 1;
+ for (auto i = offset + 1; i < num_rows; ++i) {
+ if (null_map[offset] == null_map[i])
+ step++;
+ else
+ break;
Review Comment:
warning: statement should be inside braces
[readability-braces-around-statements]
```suggestion
else {
break;
}
```
##########
be/src/olap/schema_change.h:
##########
@@ -172,11 +177,45 @@ class VSchemaChangeWithSorting : public SchemaChange {
std::unique_ptr<MemTracker> _mem_tracker;
};
+class SchemaChangeForInvertedIndex : public SchemaChange {
+public:
+ explicit SchemaChangeForInvertedIndex(const std::vector<TOlapTableIndex>&
alter_inverted_indexs,
+ const TabletSchemaSPtr&
tablet_schema);
+ virtual ~SchemaChangeForInvertedIndex();
Review Comment:
warning: prefer using 'override' or (rarely) 'final' instead of 'virtual'
[modernize-use-override]
```suggestion
~SchemaChangeForInvertedIndex() override;
```
##########
be/src/olap/schema_change.h:
##########
@@ -172,11 +177,45 @@
std::unique_ptr<MemTracker> _mem_tracker;
};
+class SchemaChangeForInvertedIndex : public SchemaChange {
+public:
+ explicit SchemaChangeForInvertedIndex(const std::vector<TOlapTableIndex>&
alter_inverted_indexs,
+ const TabletSchemaSPtr&
tablet_schema);
+ virtual ~SchemaChangeForInvertedIndex();
+
+ virtual Status process(RowsetReaderSharedPtr rowset_reader, RowsetWriter*
rowset_writer,
+ TabletSharedPtr new_tablet, TabletSharedPtr
base_tablet,
+ TabletSchemaSPtr base_tablet_schema) override;
+
+private:
+ DISALLOW_COPY_AND_ASSIGN(SchemaChangeForInvertedIndex);
+ Status _write_inverted_index(int32_t segment_idx, vectorized::Block*
block);
+ Status _add_data(const std::string& column_name,
+ const std::pair<int64_t, int64_t>& index_writer_sign,
Field* field,
+ const uint8_t** ptr, size_t num_rows);
+ Status _add_nullable(const std::string& column_name,
+ const std::pair<int64_t, int64_t>& index_writer_sign,
Field* field,
+ const uint8_t* null_map, const uint8_t** ptr, size_t
num_rows);
+
+private:
Review Comment:
warning: redundant access specifier has the same accessibility as the
previous access specifier [readability-redundant-access-specifiers]
```suggestion
```
**be/src/olap/schema_change.h:189:** previously declared here
```cpp
private:
^
```
##########
be/src/olap/types.h:
##########
@@ -321,6 +323,240 @@
TypeInfoPtr _item_type_info;
const size_t _item_size;
};
+///====================== MapType Info ==========================///
+class MapTypeInfo : public TypeInfo {
+public:
+ explicit MapTypeInfo(TypeInfoPtr key_type_info, TypeInfoPtr
value_type_info)
+ : _key_type_info(std::move(key_type_info)),
+ _value_type_info(std::move(value_type_info)) {}
+ ~MapTypeInfo() override = default;
+
+ int cmp(const void* left, const void* right) const override {
+ auto l_value = reinterpret_cast<const MapValue*>(left);
+ auto r_value = reinterpret_cast<const MapValue*>(right);
+ uint32_t l_size = l_value->size();
+ uint32_t r_size = r_value->size();
+ if (l_size < r_size) {
+ return -1;
+ } else if (l_size > r_size) {
+ return 1;
+ } else {
+ return 0;
+ }
+ }
+
+ void deep_copy(void* dest, const void* src, MemPool* mem_pool) const
override { DCHECK(false); }
+
+ void direct_copy(void* dest, const void* src) const override {
CHECK(false); }
+
+ void direct_copy(uint8_t** base, void* dest, const void* src) const {
CHECK(false); }
+
+ void direct_copy_may_cut(void* dest, const void* src) const override {
direct_copy(dest, src); }
+
+ Status from_string(void* buf, const std::string& scan_key, const int
precision = 0,
+ const int scale = 0) const override {
+ return Status::Error<ErrorCode::NOT_IMPLEMENTED_ERROR>();
+ }
+
+ std::string to_string(const void* src) const override { return "{}"; }
+
+ void set_to_max(void* buf) const override {
+ DCHECK(false) << "set_to_max of list is not implemented.";
+ }
+
+ void set_to_min(void* buf) const override {
+ DCHECK(false) << "set_to_min of list is not implemented.";
+ }
+
+ // todo . is here only to need return 16 for two ptr?
+ size_t size() const override { return sizeof(MapValue); }
+
+ FieldType type() const override { return OLAP_FIELD_TYPE_MAP; }
+
+ inline const TypeInfo* get_key_type_info() const { return
_key_type_info.get(); }
+ inline const TypeInfo* get_value_type_info() const { return
_value_type_info.get(); }
+
+private:
+ TypeInfoPtr _key_type_info;
+ TypeInfoPtr _value_type_info;
+};
+
+class StructTypeInfo : public TypeInfo {
+public:
+ explicit StructTypeInfo(std::vector<TypeInfoPtr>& type_infos) {
+ for (TypeInfoPtr& type_info : type_infos) {
+ _type_infos.push_back(std::move(type_info));
+ }
+ }
+ ~StructTypeInfo() override = default;
+
+ int cmp(const void* left, const void* right) const override {
+ auto l_value = reinterpret_cast<const StructValue*>(left);
+ auto r_value = reinterpret_cast<const StructValue*>(right);
+ uint32_t l_size = l_value->size();
+ uint32_t r_size = r_value->size();
+ size_t cur = 0;
+
+ if (!l_value->has_null() && !r_value->has_null()) {
+ while (cur < l_size && cur < r_size) {
+ int result =
+ _type_infos[cur]->cmp(l_value->child_value(cur),
r_value->child_value(cur));
+ if (result != 0) {
+ return result;
+ }
+ ++cur;
+ }
+ } else {
+ while (cur < l_size && cur < r_size) {
+ if (l_value->is_null_at(cur)) {
+ if (!r_value->is_null_at(cur)) { // left is null & right
is not null
+ return -1;
+ }
+ } else if (r_value->is_null_at(cur)) { // left is not null &
right is null
+ return 1;
+ } else { // both are not null
+ int result =
_type_infos[cur]->cmp(l_value->child_value(cur),
+
r_value->child_value(cur));
+ if (result != 0) {
+ return result;
+ }
+ }
+ ++cur;
+ }
+ }
+
+ if (l_size < r_size) {
+ return -1;
+ } else if (l_size > r_size) {
+ return 1;
+ } else {
+ return 0;
+ }
+ }
+
+ void deep_copy(void* dest, const void* src, MemPool* mem_pool) const
override {
+ auto dest_value = reinterpret_cast<StructValue*>(dest);
+ auto src_value = reinterpret_cast<const StructValue*>(src);
+
+ if (src_value->size() == 0) {
+ new (dest_value) StructValue(src_value->size());
+ return;
+ }
+
+ dest_value->set_size(src_value->size());
+ dest_value->set_has_null(src_value->has_null());
+
+ size_t allocate_size = src_value->size() *
sizeof(*src_value->values());
+ // allocate memory for children value
+ for (size_t i = 0; i < src_value->size(); ++i) {
+ if (src_value->is_null_at(i)) continue;
+ allocate_size += _type_infos[i]->size();
+ }
+
+ dest_value->set_values((void**)mem_pool->allocate(allocate_size));
+ auto ptr = reinterpret_cast<uint8_t*>(dest_value->mutable_values());
+ ptr += dest_value->size() * sizeof(*dest_value->values());
+
+ for (size_t i = 0; i < src_value->size(); ++i) {
+ dest_value->set_child_value(nullptr, i);
+ if (src_value->is_null_at(i)) continue;
Review Comment:
warning: statement should be inside braces
[readability-braces-around-statements]
```suggestion
if (src_value->is_null_at(i)) { continue;
}
```
##########
be/src/olap/schema_change.h:
##########
@@ -172,11 +177,45 @@
std::unique_ptr<MemTracker> _mem_tracker;
};
+class SchemaChangeForInvertedIndex : public SchemaChange {
+public:
+ explicit SchemaChangeForInvertedIndex(const std::vector<TOlapTableIndex>&
alter_inverted_indexs,
+ const TabletSchemaSPtr&
tablet_schema);
+ virtual ~SchemaChangeForInvertedIndex();
+
+ virtual Status process(RowsetReaderSharedPtr rowset_reader, RowsetWriter*
rowset_writer,
Review Comment:
warning: 'virtual' is redundant since the function is already declared
'override' [modernize-use-override]
```suggestion
Status process(RowsetReaderSharedPtr rowset_reader, RowsetWriter*
rowset_writer,
```
##########
be/src/olap/types.h:
##########
@@ -321,6 +323,240 @@ class ArrayTypeInfo : public TypeInfo {
TypeInfoPtr _item_type_info;
const size_t _item_size;
};
+///====================== MapType Info ==========================///
+class MapTypeInfo : public TypeInfo {
+public:
+ explicit MapTypeInfo(TypeInfoPtr key_type_info, TypeInfoPtr
value_type_info)
+ : _key_type_info(std::move(key_type_info)),
+ _value_type_info(std::move(value_type_info)) {}
+ ~MapTypeInfo() override = default;
+
+ int cmp(const void* left, const void* right) const override {
+ auto l_value = reinterpret_cast<const MapValue*>(left);
+ auto r_value = reinterpret_cast<const MapValue*>(right);
+ uint32_t l_size = l_value->size();
+ uint32_t r_size = r_value->size();
+ if (l_size < r_size) {
+ return -1;
+ } else if (l_size > r_size) {
+ return 1;
+ } else {
+ return 0;
+ }
+ }
+
+ void deep_copy(void* dest, const void* src, MemPool* mem_pool) const
override { DCHECK(false); }
+
+ void direct_copy(void* dest, const void* src) const override {
CHECK(false); }
+
+ void direct_copy(uint8_t** base, void* dest, const void* src) const {
CHECK(false); }
+
+ void direct_copy_may_cut(void* dest, const void* src) const override {
direct_copy(dest, src); }
+
+ Status from_string(void* buf, const std::string& scan_key, const int
precision = 0,
+ const int scale = 0) const override {
+ return Status::Error<ErrorCode::NOT_IMPLEMENTED_ERROR>();
+ }
+
+ std::string to_string(const void* src) const override { return "{}"; }
+
+ void set_to_max(void* buf) const override {
+ DCHECK(false) << "set_to_max of list is not implemented.";
+ }
+
+ void set_to_min(void* buf) const override {
+ DCHECK(false) << "set_to_min of list is not implemented.";
+ }
+
+ // todo . is here only to need return 16 for two ptr?
+ size_t size() const override { return sizeof(MapValue); }
+
+ FieldType type() const override { return OLAP_FIELD_TYPE_MAP; }
+
+ inline const TypeInfo* get_key_type_info() const { return
_key_type_info.get(); }
+ inline const TypeInfo* get_value_type_info() const { return
_value_type_info.get(); }
+
+private:
+ TypeInfoPtr _key_type_info;
+ TypeInfoPtr _value_type_info;
+};
+
+class StructTypeInfo : public TypeInfo {
+public:
+ explicit StructTypeInfo(std::vector<TypeInfoPtr>& type_infos) {
+ for (TypeInfoPtr& type_info : type_infos) {
+ _type_infos.push_back(std::move(type_info));
+ }
+ }
+ ~StructTypeInfo() override = default;
+
+ int cmp(const void* left, const void* right) const override {
+ auto l_value = reinterpret_cast<const StructValue*>(left);
+ auto r_value = reinterpret_cast<const StructValue*>(right);
+ uint32_t l_size = l_value->size();
+ uint32_t r_size = r_value->size();
+ size_t cur = 0;
+
+ if (!l_value->has_null() && !r_value->has_null()) {
+ while (cur < l_size && cur < r_size) {
+ int result =
+ _type_infos[cur]->cmp(l_value->child_value(cur),
r_value->child_value(cur));
+ if (result != 0) {
+ return result;
+ }
+ ++cur;
+ }
+ } else {
+ while (cur < l_size && cur < r_size) {
+ if (l_value->is_null_at(cur)) {
+ if (!r_value->is_null_at(cur)) { // left is null & right
is not null
+ return -1;
+ }
+ } else if (r_value->is_null_at(cur)) { // left is not null &
right is null
+ return 1;
+ } else { // both are not null
+ int result =
_type_infos[cur]->cmp(l_value->child_value(cur),
+
r_value->child_value(cur));
+ if (result != 0) {
+ return result;
+ }
+ }
+ ++cur;
+ }
+ }
+
+ if (l_size < r_size) {
+ return -1;
+ } else if (l_size > r_size) {
+ return 1;
+ } else {
+ return 0;
+ }
+ }
+
+ void deep_copy(void* dest, const void* src, MemPool* mem_pool) const
override {
+ auto dest_value = reinterpret_cast<StructValue*>(dest);
+ auto src_value = reinterpret_cast<const StructValue*>(src);
+
+ if (src_value->size() == 0) {
+ new (dest_value) StructValue(src_value->size());
+ return;
+ }
+
+ dest_value->set_size(src_value->size());
+ dest_value->set_has_null(src_value->has_null());
+
+ size_t allocate_size = src_value->size() *
sizeof(*src_value->values());
+ // allocate memory for children value
+ for (size_t i = 0; i < src_value->size(); ++i) {
+ if (src_value->is_null_at(i)) continue;
Review Comment:
warning: statement should be inside braces
[readability-braces-around-statements]
```suggestion
if (src_value->is_null_at(i)) { continue;
}
```
##########
be/src/olap/types.h:
##########
@@ -321,6 +323,240 @@
TypeInfoPtr _item_type_info;
const size_t _item_size;
};
+///====================== MapType Info ==========================///
+class MapTypeInfo : public TypeInfo {
+public:
+ explicit MapTypeInfo(TypeInfoPtr key_type_info, TypeInfoPtr
value_type_info)
+ : _key_type_info(std::move(key_type_info)),
+ _value_type_info(std::move(value_type_info)) {}
+ ~MapTypeInfo() override = default;
+
+ int cmp(const void* left, const void* right) const override {
+ auto l_value = reinterpret_cast<const MapValue*>(left);
+ auto r_value = reinterpret_cast<const MapValue*>(right);
+ uint32_t l_size = l_value->size();
+ uint32_t r_size = r_value->size();
+ if (l_size < r_size) {
+ return -1;
+ } else if (l_size > r_size) {
+ return 1;
+ } else {
+ return 0;
+ }
+ }
+
+ void deep_copy(void* dest, const void* src, MemPool* mem_pool) const
override { DCHECK(false); }
+
+ void direct_copy(void* dest, const void* src) const override {
CHECK(false); }
+
+ void direct_copy(uint8_t** base, void* dest, const void* src) const {
CHECK(false); }
+
+ void direct_copy_may_cut(void* dest, const void* src) const override {
direct_copy(dest, src); }
+
+ Status from_string(void* buf, const std::string& scan_key, const int
precision = 0,
+ const int scale = 0) const override {
+ return Status::Error<ErrorCode::NOT_IMPLEMENTED_ERROR>();
+ }
+
+ std::string to_string(const void* src) const override { return "{}"; }
+
+ void set_to_max(void* buf) const override {
+ DCHECK(false) << "set_to_max of list is not implemented.";
+ }
+
+ void set_to_min(void* buf) const override {
+ DCHECK(false) << "set_to_min of list is not implemented.";
+ }
+
+ // todo . is here only to need return 16 for two ptr?
+ size_t size() const override { return sizeof(MapValue); }
+
+ FieldType type() const override { return OLAP_FIELD_TYPE_MAP; }
+
+ inline const TypeInfo* get_key_type_info() const { return
_key_type_info.get(); }
+ inline const TypeInfo* get_value_type_info() const { return
_value_type_info.get(); }
+
+private:
+ TypeInfoPtr _key_type_info;
+ TypeInfoPtr _value_type_info;
+};
+
+class StructTypeInfo : public TypeInfo {
+public:
+ explicit StructTypeInfo(std::vector<TypeInfoPtr>& type_infos) {
+ for (TypeInfoPtr& type_info : type_infos) {
+ _type_infos.push_back(std::move(type_info));
+ }
+ }
+ ~StructTypeInfo() override = default;
+
+ int cmp(const void* left, const void* right) const override {
+ auto l_value = reinterpret_cast<const StructValue*>(left);
+ auto r_value = reinterpret_cast<const StructValue*>(right);
+ uint32_t l_size = l_value->size();
+ uint32_t r_size = r_value->size();
+ size_t cur = 0;
+
+ if (!l_value->has_null() && !r_value->has_null()) {
+ while (cur < l_size && cur < r_size) {
+ int result =
+ _type_infos[cur]->cmp(l_value->child_value(cur),
r_value->child_value(cur));
+ if (result != 0) {
+ return result;
+ }
+ ++cur;
+ }
+ } else {
+ while (cur < l_size && cur < r_size) {
+ if (l_value->is_null_at(cur)) {
+ if (!r_value->is_null_at(cur)) { // left is null & right
is not null
+ return -1;
+ }
+ } else if (r_value->is_null_at(cur)) { // left is not null &
right is null
+ return 1;
+ } else { // both are not null
+ int result =
_type_infos[cur]->cmp(l_value->child_value(cur),
+
r_value->child_value(cur));
+ if (result != 0) {
+ return result;
+ }
+ }
+ ++cur;
+ }
+ }
+
+ if (l_size < r_size) {
+ return -1;
+ } else if (l_size > r_size) {
+ return 1;
+ } else {
+ return 0;
+ }
+ }
+
+ void deep_copy(void* dest, const void* src, MemPool* mem_pool) const
override {
+ auto dest_value = reinterpret_cast<StructValue*>(dest);
+ auto src_value = reinterpret_cast<const StructValue*>(src);
+
+ if (src_value->size() == 0) {
+ new (dest_value) StructValue(src_value->size());
+ return;
+ }
+
+ dest_value->set_size(src_value->size());
+ dest_value->set_has_null(src_value->has_null());
+
+ size_t allocate_size = src_value->size() *
sizeof(*src_value->values());
+ // allocate memory for children value
+ for (size_t i = 0; i < src_value->size(); ++i) {
+ if (src_value->is_null_at(i)) continue;
+ allocate_size += _type_infos[i]->size();
+ }
+
+ dest_value->set_values((void**)mem_pool->allocate(allocate_size));
+ auto ptr = reinterpret_cast<uint8_t*>(dest_value->mutable_values());
+ ptr += dest_value->size() * sizeof(*dest_value->values());
+
+ for (size_t i = 0; i < src_value->size(); ++i) {
+ dest_value->set_child_value(nullptr, i);
+ if (src_value->is_null_at(i)) continue;
+ dest_value->set_child_value(ptr, i);
+ ptr += _type_infos[i]->size();
+ }
+
+ // copy children value
+ for (size_t i = 0; i < src_value->size(); ++i) {
+ if (src_value->is_null_at(i)) continue;
Review Comment:
warning: statement should be inside braces
[readability-braces-around-statements]
```suggestion
if (src_value->is_null_at(i)) { continue;
}
```
##########
be/src/olap/types.h:
##########
@@ -321,6 +323,240 @@
TypeInfoPtr _item_type_info;
const size_t _item_size;
};
+///====================== MapType Info ==========================///
+class MapTypeInfo : public TypeInfo {
+public:
+ explicit MapTypeInfo(TypeInfoPtr key_type_info, TypeInfoPtr
value_type_info)
+ : _key_type_info(std::move(key_type_info)),
+ _value_type_info(std::move(value_type_info)) {}
+ ~MapTypeInfo() override = default;
+
+ int cmp(const void* left, const void* right) const override {
+ auto l_value = reinterpret_cast<const MapValue*>(left);
+ auto r_value = reinterpret_cast<const MapValue*>(right);
+ uint32_t l_size = l_value->size();
+ uint32_t r_size = r_value->size();
+ if (l_size < r_size) {
+ return -1;
+ } else if (l_size > r_size) {
+ return 1;
+ } else {
+ return 0;
+ }
+ }
+
+ void deep_copy(void* dest, const void* src, MemPool* mem_pool) const
override { DCHECK(false); }
+
+ void direct_copy(void* dest, const void* src) const override {
CHECK(false); }
+
+ void direct_copy(uint8_t** base, void* dest, const void* src) const {
CHECK(false); }
+
+ void direct_copy_may_cut(void* dest, const void* src) const override {
direct_copy(dest, src); }
+
+ Status from_string(void* buf, const std::string& scan_key, const int
precision = 0,
+ const int scale = 0) const override {
+ return Status::Error<ErrorCode::NOT_IMPLEMENTED_ERROR>();
+ }
+
+ std::string to_string(const void* src) const override { return "{}"; }
+
+ void set_to_max(void* buf) const override {
+ DCHECK(false) << "set_to_max of list is not implemented.";
+ }
+
+ void set_to_min(void* buf) const override {
+ DCHECK(false) << "set_to_min of list is not implemented.";
+ }
+
+ // todo . is here only to need return 16 for two ptr?
+ size_t size() const override { return sizeof(MapValue); }
+
+ FieldType type() const override { return OLAP_FIELD_TYPE_MAP; }
+
+ inline const TypeInfo* get_key_type_info() const { return
_key_type_info.get(); }
+ inline const TypeInfo* get_value_type_info() const { return
_value_type_info.get(); }
+
+private:
+ TypeInfoPtr _key_type_info;
+ TypeInfoPtr _value_type_info;
+};
+
+class StructTypeInfo : public TypeInfo {
+public:
+ explicit StructTypeInfo(std::vector<TypeInfoPtr>& type_infos) {
+ for (TypeInfoPtr& type_info : type_infos) {
+ _type_infos.push_back(std::move(type_info));
+ }
+ }
+ ~StructTypeInfo() override = default;
+
+ int cmp(const void* left, const void* right) const override {
+ auto l_value = reinterpret_cast<const StructValue*>(left);
+ auto r_value = reinterpret_cast<const StructValue*>(right);
+ uint32_t l_size = l_value->size();
+ uint32_t r_size = r_value->size();
+ size_t cur = 0;
+
+ if (!l_value->has_null() && !r_value->has_null()) {
+ while (cur < l_size && cur < r_size) {
+ int result =
+ _type_infos[cur]->cmp(l_value->child_value(cur),
r_value->child_value(cur));
+ if (result != 0) {
+ return result;
+ }
+ ++cur;
+ }
+ } else {
+ while (cur < l_size && cur < r_size) {
+ if (l_value->is_null_at(cur)) {
+ if (!r_value->is_null_at(cur)) { // left is null & right
is not null
+ return -1;
+ }
+ } else if (r_value->is_null_at(cur)) { // left is not null &
right is null
+ return 1;
+ } else { // both are not null
+ int result =
_type_infos[cur]->cmp(l_value->child_value(cur),
+
r_value->child_value(cur));
+ if (result != 0) {
+ return result;
+ }
+ }
+ ++cur;
+ }
+ }
+
+ if (l_size < r_size) {
+ return -1;
+ } else if (l_size > r_size) {
+ return 1;
+ } else {
+ return 0;
+ }
+ }
+
+ void deep_copy(void* dest, const void* src, MemPool* mem_pool) const
override {
+ auto dest_value = reinterpret_cast<StructValue*>(dest);
+ auto src_value = reinterpret_cast<const StructValue*>(src);
+
+ if (src_value->size() == 0) {
+ new (dest_value) StructValue(src_value->size());
+ return;
+ }
+
+ dest_value->set_size(src_value->size());
+ dest_value->set_has_null(src_value->has_null());
+
+ size_t allocate_size = src_value->size() *
sizeof(*src_value->values());
+ // allocate memory for children value
+ for (size_t i = 0; i < src_value->size(); ++i) {
+ if (src_value->is_null_at(i)) continue;
+ allocate_size += _type_infos[i]->size();
+ }
+
+ dest_value->set_values((void**)mem_pool->allocate(allocate_size));
+ auto ptr = reinterpret_cast<uint8_t*>(dest_value->mutable_values());
+ ptr += dest_value->size() * sizeof(*dest_value->values());
+
+ for (size_t i = 0; i < src_value->size(); ++i) {
+ dest_value->set_child_value(nullptr, i);
+ if (src_value->is_null_at(i)) continue;
+ dest_value->set_child_value(ptr, i);
+ ptr += _type_infos[i]->size();
+ }
+
+ // copy children value
+ for (size_t i = 0; i < src_value->size(); ++i) {
+ if (src_value->is_null_at(i)) continue;
+ _type_infos[i]->deep_copy(dest_value->mutable_child_value(i),
src_value->child_value(i),
+ mem_pool);
+ }
+ }
+
+ void direct_copy(void* dest, const void* src) const override {
+ auto dest_value = static_cast<StructValue*>(dest);
+ auto base = reinterpret_cast<uint8_t*>(dest_value->mutable_values());
+ direct_copy(&base, dest, src);
+ }
+
+ void direct_copy(uint8_t** base, void* dest, const void* src) const {
+ auto dest_value = static_cast<StructValue*>(dest);
+ auto src_value = static_cast<const StructValue*>(src);
+
+ dest_value->set_size(src_value->size());
+ dest_value->set_has_null(src_value->has_null());
+ *base += src_value->size() * sizeof(*src_value->values());
+
+ for (size_t i = 0; i < src_value->size(); ++i) {
+ dest_value->set_child_value(nullptr, i);
+ if (src_value->is_null_at(i)) continue;
Review Comment:
warning: statement should be inside braces
[readability-braces-around-statements]
```suggestion
if (src_value->is_null_at(i)) { continue;
}
```
##########
be/src/olap/tablet_schema.cpp:
##########
@@ -496,6 +528,33 @@ void TabletIndex::init_from_thrift(const TOlapTableIndex&
index,
}
}
+void TabletIndex::init_from_thrift(const TOlapTableIndex& index,
+ const std::vector<int32_t>& column_uids) {
+ _index_id = index.index_id;
+ _index_name = index.index_name;
+ _col_unique_ids = std::move(column_uids);
Review Comment:
warning: std::move of the const variable 'column_uids' has no effect; remove
std::move() or make the variable non-const [performance-move-const-arg]
```suggestion
_col_unique_ids = column_uids;
```
##########
be/src/util/async_io.h:
##########
@@ -0,0 +1,106 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <bthread/bthread.h>
+
+#include "io/fs/file_system.h"
+#include "olap/olap_define.h"
+#include "priority_thread_pool.hpp"
+#include "runtime/threadlocal.h"
+
+namespace doris {
+
+struct AsyncIOCtx {
+ int nice;
+};
+
+/**
+ * Separate task from bthread to pthread, specific for IO task.
+ */
+class AsyncIO {
+public:
+ AsyncIO() {
+ _io_thread_pool = new
PriorityThreadPool(config::doris_scanner_thread_pool_thread_num,
+
config::doris_scanner_thread_pool_queue_size,
+ "async_io_thread_pool");
+ _remote_thread_pool = new PriorityThreadPool(
+ config::doris_remote_scanner_thread_pool_thread_num,
+ config::doris_remote_scanner_thread_pool_queue_size,
"async_remote_thread_pool");
+ }
+
+ ~AsyncIO() {
+ SAFE_DELETE(_io_thread_pool);
+ SAFE_DELETE(_remote_thread_pool);
+ }
+
+ AsyncIO& operator=(const AsyncIO&) = delete;
+ AsyncIO(const AsyncIO&) = delete;
+
+ static AsyncIO& instance() {
+ static AsyncIO instance;
+ return instance;
+ }
+
+ // This function should run on the bthread, and it will put the task into
+ // thread_pool and release the bthread_worker at cv.wait. When the task is
completed,
+ // the bthread will continue to execute.
+ static void run_task(std::function<void()> fn, io::FileSystemType
file_type) {
+ DCHECK(bthread_self() != 0);
+ doris::Mutex mutex;
+ doris::ConditionVariable cv;
+ std::unique_lock l(mutex);
+
+ AsyncIOCtx* ctx =
static_cast<AsyncIOCtx*>(bthread_getspecific(btls_io_ctx_key));
+ int nice = -1;
+ if (ctx == nullptr) {
+ nice = 18;
+ } else {
+ nice = ctx->nice;
+ }
+
+ PriorityThreadPool::Task task;
+ task.priority = nice;
+ task.work_function = [&] {
+ fn();
+ std::unique_lock l(mutex);
+ cv.notify_one();
+ };
+
+ if (file_type == io::FileSystemType::S3) {
+ AsyncIO::instance().remote_thread_pool()->offer(task);
+ } else {
+ AsyncIO::instance().io_thread_pool()->offer(task);
+ }
+ cv.wait(l);
+ }
+
+ inline static bthread_key_t btls_io_ctx_key;
+
+ static void io_ctx_key_deleter(void* d) { delete
static_cast<AsyncIOCtx*>(d); }
+
+private:
+ PriorityThreadPool* _io_thread_pool = nullptr;
+ PriorityThreadPool* _remote_thread_pool = nullptr;
+
+private:
Review Comment:
warning: redundant access specifier has the same accessibility as the
previous access specifier [readability-redundant-access-specifiers]
```suggestion
```
**be/src/util/async_io.h:96:** previously declared here
```cpp
private:
^
```
##########
be/src/olap/task/engine_alter_tablet_task.h:
##########
@@ -39,4 +39,18 @@ class EngineAlterTabletTask : public EngineTask {
std::shared_ptr<MemTrackerLimiter> _mem_tracker;
}; // EngineTask
+class EngineAlterInvertedIndexTask : public EngineTask {
+public:
+ virtual Status execute();
Review Comment:
warning: prefer using 'override' or (rarely) 'final' instead of 'virtual'
[modernize-use-override]
```suggestion
Status execute() override;
```
##########
be/src/util/priority_work_stealing_thread_pool.hpp:
##########
@@ -97,7 +98,7 @@
// Any work Offer()'ed during DrainAndshutdown may or may not be processed.
void drain_and_shutdown() override {
{
- std::unique_lock<std::mutex> l(_lock);
+ std::unique_lock l(_lock);
while (get_queue_size() != 0) {
_empty_cv.wait(l);
Review Comment:
warning: use of undeclared identifier '_empty_cv' [clang-diagnostic-error]
```cpp
_empty_cv.wait(l);
^
```
##########
be/src/vec/data_types/get_least_supertype.cpp:
##########
@@ -50,12 +50,152 @@
}
} // namespace
-DataTypePtr get_least_supertype(const DataTypes& types) {
+Status get_numeric_type(const TypeIndexSet& types, DataTypePtr* type) {
+ bool all_numbers = true;
+
+ size_t max_bits_of_signed_integer = 0;
+ size_t max_bits_of_unsigned_integer = 0;
+ size_t max_mantissa_bits_of_floating = 0;
+
+ auto maximize = [](size_t& what, size_t value) {
+ if (value > what) what = value;
+ };
+
+ for (const auto& type : types) {
+ if (type == TypeIndex::UInt8)
Review Comment:
warning: statement should be inside braces
[readability-braces-around-statements]
```suggestion
if (type == TypeIndex::UInt8) {
```
be/src/vec/data_types/get_least_supertype.cpp:66:
```diff
- else if (type == TypeIndex::UInt16)
+ } else if (type == TypeIndex::UInt16)
```
##########
be/src/vec/data_types/get_least_supertype.cpp:
##########
@@ -50,12 +50,152 @@
}
} // namespace
-DataTypePtr get_least_supertype(const DataTypes& types) {
+Status get_numeric_type(const TypeIndexSet& types, DataTypePtr* type) {
+ bool all_numbers = true;
+
+ size_t max_bits_of_signed_integer = 0;
+ size_t max_bits_of_unsigned_integer = 0;
+ size_t max_mantissa_bits_of_floating = 0;
+
+ auto maximize = [](size_t& what, size_t value) {
+ if (value > what) what = value;
+ };
+
+ for (const auto& type : types) {
+ if (type == TypeIndex::UInt8)
+ maximize(max_bits_of_unsigned_integer, 8);
+ else if (type == TypeIndex::UInt16)
Review Comment:
warning: statement should be inside braces
[readability-braces-around-statements]
```suggestion
else if (type == TypeIndex::UInt16) {
```
be/src/vec/data_types/get_least_supertype.cpp:68:
```diff
- else if (type == TypeIndex::UInt32)
+ } else if (type == TypeIndex::UInt32)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
