Skip to content

Commit

Permalink
clean code
Browse files Browse the repository at this point in the history 
Signed-off-by: guo-shaoge <[email protected]>
  • Loading branch information
@guo-shaoge
guo-shaoge committed Dec 24, 2024
1 parent 9623368 commit 7deb4b4
Show file tree
Hide file tree
Showing 3 changed files with 20 additions and 213 deletions.
52 changes: 0 additions & 52 deletions dbms/src/Common/ColumnsHashingImpl.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -128,15 +128,6 @@ class HashMethodBase
using FindResult = FindResultImpl<Mapped>;
static constexpr bool has_mapped = !std::is_same<Mapped, VoidMapped>::value;
using Cache = LastElementCache<Value, consecutive_keys_optimization>;
static constexpr size_t prefetch_step = 16;

template <typename Map>
static ALWAYS_INLINE inline void prefetch(Map & map, size_t idx, const std::vector<size_t> & hashvals)
{
const auto prefetch_idx = idx + prefetch_step;
if likely (prefetch_idx < hashvals.size())
map.prefetch(hashvals[prefetch_idx]);
}

// Emplace key without hashval, and this method doesn't support prefetch.
template <typename Data>
Expand Down Expand Up @@ -185,49 +176,6 @@ class HashMethodBase
return findKeyImpl(keyHolderGetKey(key_holder), data, hashval);
}

// Emplace key using hashval, you can enable prefetch or not.
template <bool enable_prefetch = false, typename Data>
ALWAYS_INLINE inline EmplaceResult emplaceKey(
Data & data,
size_t row,
Arena & pool,
std::vector<String> & sort_key_containers,
const std::vector<size_t> & hashvals)
{
auto key_holder = static_cast<Derived &>(*this).getKeyHolder(row, &pool, sort_key_containers);
if constexpr (enable_prefetch)
{
assert(hashvals.size() == static_cast<Derived &>(*this).total_rows);
prefetch(data, row, hashvals);
return emplaceImpl(key_holder, data, hashvals[row]);
}
else
{
return emplaceImpl(key_holder, data);
}
}

template <bool enable_prefetch = false, typename Data>
ALWAYS_INLINE inline FindResult findKey(
Data & data,
size_t row,
Arena & pool,
std::vector<String> & sort_key_containers,
const std::vector<size_t> & hashvals)
{
auto key_holder = static_cast<Derived &>(*this).getKeyHolder(row, &pool, sort_key_containers);
if constexpr (enable_prefetch)
{
assert(hashvals.size() == static_cast<Derived &>(*this).total_rows);
prefetch(data, row, hashvals);
return findKeyImpl(keyHolderGetKey(key_holder), data, hashvals[row]);
}
else
{
return findKeyImpl(keyHolderGetKey(key_holder), data);
}
}

template <typename Data>
ALWAYS_INLINE inline size_t getHash(
const Data & data,
Expand Down
172 changes: 20 additions & 152 deletions dbms/src/Interpreters/Aggregator.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -721,38 +721,6 @@ void getHashVals(
}
}

template <bool only_lookup, typename Method>
std::optional<typename Method::template EmplaceOrFindKeyResult<only_lookup>::ResultType> Aggregator::emplaceOrFindKey(
Method & method,
typename Method::State & state,
size_t index,
Arena & aggregates_pool,
std::vector<std::string> & sort_key_containers,
const std::vector<size_t> & hashvals) const
{
try
{
if constexpr (only_lookup)
return state.template findKey</*enable_prefetch*/ true>(
method.data,
index,
aggregates_pool,
sort_key_containers,
hashvals);
else
return state.template emplaceKey</*enable_prefetch*/ true>(
method.data,
index,
aggregates_pool,
sort_key_containers,
hashvals);
}
catch (ResizeException &)
{
return {};
}
}

template <bool only_lookup, typename Method>
std::optional<typename Method::template EmplaceOrFindKeyResult<only_lookup>::ResultType> Aggregator::emplaceOrFindKey(
Method & method,
Expand All @@ -765,19 +733,9 @@ std::optional<typename Method::template EmplaceOrFindKeyResult<only_lookup>::Res
try
{
if constexpr (only_lookup)
return state.template findKey(
method.data,
index,
aggregates_pool,
sort_key_containers,
hashval);
return state.template findKey(method.data, index, aggregates_pool, sort_key_containers, hashval);
else
return state.template emplaceKey(
method.data,
index,
aggregates_pool,
sort_key_containers,
hashval);
return state.template emplaceKey(method.data, index, aggregates_pool, sort_key_containers, hashval);
}
catch (ResizeException &)
{
Expand Down Expand Up @@ -972,9 +930,9 @@ ALWAYS_INLINE void Aggregator::executeImplByRow(
\
if constexpr (collect_hit_rate) \
++agg_process_info.hit_row_cnt; \
\
if constexpr (enable_prefetch) \
__builtin_prefetch(aggregate_data); \
\
if constexpr (enable_prefetch) \
__builtin_prefetch(aggregate_data); \
} \
} \
\
Expand All @@ -993,33 +951,8 @@ ALWAYS_INLINE void Aggregator::executeImplByRow(
}
}

// for (size_t i = agg_process_info.start_row; i < agg_process_info.start_row + rows; ++i)
// {
// AggregateDataPtr aggregate_data = nullptr;
// if constexpr (enable_prefetch)
// {
// auto emplace_result_holder = emplaceOrFindKey<only_lookup>(
// method,
// state,
// i,
// *aggregates_pool,
// sort_key_containers,
// agg_process_info.hashvals);

// HANDLE_AGG_EMPLACE_RESULT
// }
// else
// {
// auto emplace_result_holder
// = emplaceOrFindKey<only_lookup>(method, state, i, *aggregates_pool, sort_key_containers);

// HANDLE_AGG_EMPLACE_RESULT
// }
// }

size_t i = agg_process_info.start_row;
const size_t end = agg_process_info.end_row;
// const size_t end = *processed_rows - agg_process_info.start_row + 1;
const size_t mini_batch = 256;
while (i < end)
{
Expand All @@ -1028,6 +961,7 @@ ALWAYS_INLINE void Aggregator::executeImplByRow(
batch_size = end - i;

const auto cur_batch_end = i + batch_size;
// j is the row index inside each mini batch.
for (size_t j = i; j < cur_batch_end; ++j)
{
AggregateDataPtr aggregate_data = nullptr;
Expand All @@ -1038,18 +972,14 @@ ALWAYS_INLINE void Aggregator::executeImplByRow(
const size_t prefetch_idx = j + prefetch_step;
if likely (prefetch_idx < end)
{
const auto new_hashval = state.getHash(method.data, prefetch_idx, *aggregates_pool, sort_key_containers);
const auto new_hashval
= state.getHash(method.data, prefetch_idx, *aggregates_pool, sort_key_containers);
method.data.prefetch(new_hashval);
hashvals[prefetch_hash_idx] = new_hashval;
}

auto emplace_result_holder = emplaceOrFindKey<only_lookup>(
method,
state,
j,
*aggregates_pool,
sort_key_containers,
hashval);
auto emplace_result_holder
= emplaceOrFindKey<only_lookup>(method, state, j, *aggregates_pool, sort_key_containers, hashval);

HANDLE_AGG_EMPLACE_RESULT
}
Expand All @@ -1065,90 +995,29 @@ ALWAYS_INLINE void Aggregator::executeImplByRow(
if unlikely (!processed_rows.has_value())
break;

size_t processed_size = *processed_rows - i + 1;
// bool first_inst = true;
const size_t processed_size = *processed_rows - i + 1;
for (AggregateFunctionInstruction * inst = agg_process_info.aggregate_functions_instructions.data(); inst->that;
++inst)
++inst)
{
// if (first_inst)
// inst->batch_that->addBatchWithPrefetch(
// i,
// batch_size,
// places.get() + i,
// inst->state_offset,
// inst->batch_arguments,
// aggregates_pool);
// else
inst->batch_that->addBatch(
i,
processed_size,
places.get() + i,
inst->state_offset,
inst->batch_arguments,
aggregates_pool);
// first_inst = false;
i,
processed_size,
places.get() + i,
inst->state_offset,
inst->batch_arguments,
aggregates_pool);
}

i += processed_size;

if unlikely (processed_size != batch_size)
break;

i = cur_batch_end;
}

if likely (processed_rows)
agg_process_info.start_row = *processed_rows + 1;

#undef HANDLE_AGG_EMPLACE_RESULT

// if (processed_rows)
// {
// // /// Add values to the aggregate functions.
// // for (AggregateFunctionInstruction * inst = agg_process_info.aggregate_functions_instructions.data(); inst->that;
// // ++inst)
// // {
// // inst->batch_that->addBatch(
// // agg_process_info.start_row,
// // *processed_rows - agg_process_info.start_row + 1,
// // places.get(),
// // inst->state_offset,
// // inst->batch_arguments,
// // aggregates_pool);
// // }
// size_t i = agg_process_info.start_row;
// const size_t end = *processed_rows - agg_process_info.start_row + 1;
// const size_t step = 256;
// while (i < end)
// {
// size_t batch_size = step;
// if unlikely (i + batch_size > end)
// batch_size = end - i;

// bool first_inst = true;
// for (AggregateFunctionInstruction * inst = agg_process_info.aggregate_functions_instructions.data(); inst->that;
// ++inst)
// {
// if (first_inst)
// inst->batch_that->addBatchWithPrefetch(
// i,
// batch_size,
// places.get() + i,
// inst->state_offset,
// inst->batch_arguments,
// aggregates_pool);
// else
// inst->batch_that->addBatch(
// i,
// batch_size,
// places.get() + i,
// inst->state_offset,
// inst->batch_arguments,
// aggregates_pool);
// first_inst = false;
// }
// i += batch_size;
// }
// agg_process_info.start_row = *processed_rows + 1;
// }
}

void NO_INLINE
Expand Down Expand Up @@ -1993,7 +1862,6 @@ void NO_INLINE Aggregator::convertToBlocksImplFinal(
agg_keys_helpers[key_columns_vec_index]
->insertKeyIntoColumns(key, key_columns_vec[key_columns_vec_index], key_sizes_ref, params.collators);
}
// insertAggregatesIntoColumns(mapped, final_aggregate_columns_vec[key_columns_vec_index], arena);
places[data_index] = mapped;
++data_index;
});
Expand Down
9 changes: 0 additions & 9 deletions dbms/src/Interpreters/Aggregator.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1460,15 +1460,6 @@ class Aggregator
Arena * aggregates_pool,
AggProcessInfo & agg_process_info) const;

template <bool only_lookup, typename Method>
std::optional<typename Method::template EmplaceOrFindKeyResult<only_lookup>::ResultType> emplaceOrFindKey(
Method & method,
typename Method::State & state,
size_t index,
Arena & aggregates_pool,
std::vector<std::string> & sort_key_containers,
const std::vector<size_t> & hashvals) const;

template <bool only_lookup, typename Method>
std::optional<typename Method::template EmplaceOrFindKeyResult<only_lookup>::ResultType> emplaceOrFindKey(
Method & method,
Expand Down

0 comments on commit 7deb4b4

Please sign in to comment.