Skip to content

Commit 48ea89a

Browse files
kaka11chenkoarz
authored andcommitted
[Enhancement](multi-catalog) Add PredicateFilterTime, DictFilterRewriteTime, LazyReadFilteredRows profile metrics in parquet orc profiles. (apache#51248)
[Enhancement] (multi-catalog) Add PredicateFilterTime, DictFilterRewriteTime, LazyReadFilteredRows profile metrics in parquet/orc profiles.
1 parent ee53ffe commit 48ea89a

6 files changed

Lines changed: 169 additions & 137 deletions

File tree

be/src/vec/exec/format/orc/vorc_reader.cpp

Lines changed: 70 additions & 61 deletions
Original file line numberDiff line numberDiff line change
@@ -226,7 +226,9 @@ void OrcReader::_collect_profile_before_close() {
226226
COUNTER_UPDATE(_orc_profile.set_fill_column_time, _statistics.set_fill_column_time);
227227
COUNTER_UPDATE(_orc_profile.decode_value_time, _statistics.decode_value_time);
228228
COUNTER_UPDATE(_orc_profile.decode_null_map_time, _statistics.decode_null_map_time);
229-
COUNTER_UPDATE(_orc_profile.filter_block_time, _statistics.filter_block_time);
229+
COUNTER_UPDATE(_orc_profile.predicate_filter_time, _statistics.predicate_filter_time);
230+
COUNTER_UPDATE(_orc_profile.dict_filter_rewrite_time, _statistics.dict_filter_rewrite_time);
231+
COUNTER_UPDATE(_orc_profile.lazy_read_filtered_rows, _statistics.lazy_read_filtered_rows);
230232

231233
if (_file_input_stream != nullptr) {
232234
_file_input_stream->collect_profile_before_close();
@@ -260,8 +262,12 @@ void OrcReader::_init_profile() {
260262
ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecodeValueTime", orc_profile, 1);
261263
_orc_profile.decode_null_map_time =
262264
ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DecodeNullMapTime", orc_profile, 1);
263-
_orc_profile.filter_block_time =
264-
ADD_CHILD_TIMER_WITH_LEVEL(_profile, "FilterBlockTime", orc_profile, 1);
265+
_orc_profile.predicate_filter_time =
266+
ADD_CHILD_TIMER_WITH_LEVEL(_profile, "PredicateFilterTime", orc_profile, 1);
267+
_orc_profile.dict_filter_rewrite_time =
268+
ADD_CHILD_TIMER_WITH_LEVEL(_profile, "DictFilterRewriteTime", orc_profile, 1);
269+
_orc_profile.lazy_read_filtered_rows =
270+
ADD_COUNTER_WITH_LEVEL(_profile, "FilteredRowsByLazyRead", TUnit::UNIT, 1);
265271
_orc_profile.selected_row_group_count =
266272
ADD_COUNTER_WITH_LEVEL(_profile, "SelectedRowGroupCount", TUnit::UNIT, 1);
267273
_orc_profile.evaluated_row_group_count =
@@ -2006,15 +2012,18 @@ Status OrcReader::get_next_block_impl(Block* block, size_t* read_rows, bool* eof
20062012
*read_rows = 0;
20072013
return Status::OK();
20082014
}
2009-
_execute_filter_position_delete_rowids(*_filter);
20102015
{
2011-
SCOPED_RAW_TIMER(&_statistics.decode_null_map_time);
2012-
RETURN_IF_CATCH_EXCEPTION(
2013-
Block::filter_block_internal(block, columns_to_filter, *_filter));
2016+
SCOPED_RAW_TIMER(&_statistics.predicate_filter_time);
2017+
_execute_filter_position_delete_rowids(*_filter);
2018+
{
2019+
SCOPED_RAW_TIMER(&_statistics.decode_null_map_time);
2020+
RETURN_IF_CATCH_EXCEPTION(
2021+
Block::filter_block_internal(block, columns_to_filter, *_filter));
2022+
}
2023+
Block::erase_useless_column(block, column_to_keep);
2024+
RETURN_IF_ERROR(_convert_dict_cols_to_string_cols(block, &batch_vec));
2025+
*read_rows = block->rows();
20142026
}
2015-
Block::erase_useless_column(block, column_to_keep);
2016-
RETURN_IF_ERROR(_convert_dict_cols_to_string_cols(block, &batch_vec));
2017-
*read_rows = block->rows();
20182027
} else {
20192028
uint64_t rr;
20202029
SCOPED_RAW_TIMER(&_statistics.column_read_time);
@@ -2091,63 +2100,60 @@ Status OrcReader::get_next_block_impl(Block* block, size_t* read_rows, bool* eof
20912100
return Status::OK();
20922101
}
20932102

2094-
_build_delete_row_filter(block, _batch->numElements);
2095-
2096-
std::vector<uint32_t> columns_to_filter;
2097-
int column_to_keep = block->columns();
2098-
columns_to_filter.resize(column_to_keep);
2099-
for (uint32_t i = 0; i < column_to_keep; ++i) {
2100-
columns_to_filter[i] = i;
2101-
}
2102-
if (!_lazy_read_ctx.conjuncts.empty()) {
2103-
VExprContextSPtrs filter_conjuncts;
2104-
filter_conjuncts.insert(filter_conjuncts.end(), _filter_conjuncts.begin(),
2105-
_filter_conjuncts.end());
2106-
for (auto& conjunct : _dict_filter_conjuncts) {
2107-
filter_conjuncts.emplace_back(conjunct);
2108-
}
2109-
for (auto& conjunct : _non_dict_filter_conjuncts) {
2110-
filter_conjuncts.emplace_back(conjunct);
2111-
}
2112-
std::vector<IColumn::Filter*> filters;
2113-
if (_delete_rows_filter_ptr) {
2114-
filters.push_back(_delete_rows_filter_ptr.get());
2115-
}
2116-
IColumn::Filter result_filter(block->rows(), 1);
2117-
bool can_filter_all = false;
2118-
RETURN_IF_ERROR_OR_CATCH_EXCEPTION(VExprContext::execute_conjuncts(
2119-
filter_conjuncts, &filters, block, &result_filter, &can_filter_all));
2120-
if (can_filter_all) {
2121-
for (auto& col : columns_to_filter) {
2122-
std::move(*block->get_by_position(col).column).assume_mutable()->clear();
2103+
{
2104+
SCOPED_RAW_TIMER(&_statistics.predicate_filter_time);
2105+
_build_delete_row_filter(block, _batch->numElements);
2106+
2107+
std::vector<uint32_t> columns_to_filter;
2108+
int column_to_keep = block->columns();
2109+
columns_to_filter.resize(column_to_keep);
2110+
for (uint32_t i = 0; i < column_to_keep; ++i) {
2111+
columns_to_filter[i] = i;
2112+
}
2113+
if (!_lazy_read_ctx.conjuncts.empty()) {
2114+
VExprContextSPtrs filter_conjuncts;
2115+
filter_conjuncts.insert(filter_conjuncts.end(), _filter_conjuncts.begin(),
2116+
_filter_conjuncts.end());
2117+
for (auto& conjunct : _dict_filter_conjuncts) {
2118+
filter_conjuncts.emplace_back(conjunct);
21232119
}
2124-
Block::erase_useless_column(block, column_to_keep);
2125-
return _convert_dict_cols_to_string_cols(block, &batch_vec);
2126-
}
2127-
_execute_filter_position_delete_rowids(result_filter);
2128-
{
2129-
SCOPED_RAW_TIMER(&_statistics.filter_block_time);
2120+
for (auto& conjunct : _non_dict_filter_conjuncts) {
2121+
filter_conjuncts.emplace_back(conjunct);
2122+
}
2123+
std::vector<IColumn::Filter*> filters;
2124+
if (_delete_rows_filter_ptr) {
2125+
filters.push_back(_delete_rows_filter_ptr.get());
2126+
}
2127+
IColumn::Filter result_filter(block->rows(), 1);
2128+
bool can_filter_all = false;
2129+
RETURN_IF_ERROR_OR_CATCH_EXCEPTION(VExprContext::execute_conjuncts(
2130+
filter_conjuncts, &filters, block, &result_filter, &can_filter_all));
2131+
if (can_filter_all) {
2132+
for (auto& col : columns_to_filter) {
2133+
std::move(*block->get_by_position(col).column).assume_mutable()->clear();
2134+
}
2135+
Block::erase_useless_column(block, column_to_keep);
2136+
return _convert_dict_cols_to_string_cols(block, &batch_vec);
2137+
}
2138+
_execute_filter_position_delete_rowids(result_filter);
21302139
RETURN_IF_CATCH_EXCEPTION(
21312140
Block::filter_block_internal(block, columns_to_filter, result_filter));
2132-
}
2133-
Block::erase_useless_column(block, column_to_keep);
2134-
RETURN_IF_ERROR(_convert_dict_cols_to_string_cols(block, &batch_vec));
2135-
} else {
2136-
if (_delete_rows_filter_ptr) {
2137-
_execute_filter_position_delete_rowids(*_delete_rows_filter_ptr);
2138-
SCOPED_RAW_TIMER(&_statistics.filter_block_time);
2139-
RETURN_IF_CATCH_EXCEPTION(Block::filter_block_internal(block, columns_to_filter,
2140-
(*_delete_rows_filter_ptr)));
2141+
Block::erase_useless_column(block, column_to_keep);
21412142
} else {
2142-
std::unique_ptr<IColumn::Filter> filter(new IColumn::Filter(block->rows(), 1));
2143-
_execute_filter_position_delete_rowids(*filter);
2144-
SCOPED_RAW_TIMER(&_statistics.filter_block_time);
2145-
RETURN_IF_CATCH_EXCEPTION(
2146-
Block::filter_block_internal(block, columns_to_filter, (*filter)));
2143+
if (_delete_rows_filter_ptr) {
2144+
_execute_filter_position_delete_rowids(*_delete_rows_filter_ptr);
2145+
RETURN_IF_CATCH_EXCEPTION(Block::filter_block_internal(
2146+
block, columns_to_filter, (*_delete_rows_filter_ptr)));
2147+
} else {
2148+
std::unique_ptr<IColumn::Filter> filter(new IColumn::Filter(block->rows(), 1));
2149+
_execute_filter_position_delete_rowids(*filter);
2150+
RETURN_IF_CATCH_EXCEPTION(
2151+
Block::filter_block_internal(block, columns_to_filter, (*filter)));
2152+
}
2153+
Block::erase_useless_column(block, column_to_keep);
21472154
}
2148-
Block::erase_useless_column(block, column_to_keep);
2149-
RETURN_IF_ERROR(_convert_dict_cols_to_string_cols(block, &batch_vec));
21502155
}
2156+
RETURN_IF_ERROR(_convert_dict_cols_to_string_cols(block, &batch_vec));
21512157
*read_rows = block->rows();
21522158
}
21532159
return Status::OK();
@@ -2189,6 +2195,7 @@ void OrcReader::_build_delete_row_filter(const Block* block, size_t rows) {
21892195
}
21902196

21912197
Status OrcReader::filter(orc::ColumnVectorBatch& data, uint16_t* sel, uint16_t size, void* arg) {
2198+
SCOPED_RAW_TIMER(&_statistics.predicate_filter_time);
21922199
auto* block = (Block*)arg;
21932200
size_t origin_column_num = block->columns();
21942201

@@ -2289,6 +2296,7 @@ Status OrcReader::filter(orc::ColumnVectorBatch& data, uint16_t* sel, uint16_t s
22892296
sel[new_size] = i;
22902297
new_size += result_filter_data[i] ? 1 : 0;
22912298
}
2299+
_statistics.lazy_read_filtered_rows += static_cast<int64_t>(size - new_size);
22922300
data.numElements = new_size;
22932301
return Status::OK();
22942302
}
@@ -2363,6 +2371,7 @@ bool OrcReader::_can_filter_by_dict(int slot_id) {
23632371
Status OrcReader::on_string_dicts_loaded(
23642372
std::unordered_map<std::string, orc::StringDictionary*>& file_column_name_to_dict_map,
23652373
bool* is_stripe_filtered) {
2374+
SCOPED_RAW_TIMER(&_statistics.dict_filter_rewrite_time);
23662375
*is_stripe_filtered = false;
23672376
for (auto it = _dict_filter_cols.begin(); it != _dict_filter_cols.end();) {
23682377
std::string& dict_filter_col_name = it->first;

be/src/vec/exec/format/orc/vorc_reader.h

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -123,7 +123,9 @@ class OrcReader : public GenericReader {
123123
int64_t set_fill_column_time = 0;
124124
int64_t decode_value_time = 0;
125125
int64_t decode_null_map_time = 0;
126-
int64_t filter_block_time = 0;
126+
int64_t predicate_filter_time = 0;
127+
int64_t dict_filter_rewrite_time = 0;
128+
int64_t lazy_read_filtered_rows = 0;
127129
};
128130

129131
OrcReader(RuntimeProfile* profile, RuntimeState* state, const TFileScanRangeParams& params,
@@ -222,7 +224,9 @@ class OrcReader : public GenericReader {
222224
RuntimeProfile::Counter* set_fill_column_time = nullptr;
223225
RuntimeProfile::Counter* decode_value_time = nullptr;
224226
RuntimeProfile::Counter* decode_null_map_time = nullptr;
225-
RuntimeProfile::Counter* filter_block_time = nullptr;
227+
RuntimeProfile::Counter* predicate_filter_time = nullptr;
228+
RuntimeProfile::Counter* dict_filter_rewrite_time = nullptr;
229+
RuntimeProfile::Counter* lazy_read_filtered_rows = nullptr;
226230
RuntimeProfile::Counter* selected_row_group_count = nullptr;
227231
RuntimeProfile::Counter* evaluated_row_group_count = nullptr;
228232
};

0 commit comments

Comments
 (0)