#pragma once #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include namespace DB { namespace ErrorCodes { extern const int UNKNOWN_AGGREGATED_DATA_VARIANT; extern const int NOT_ENOUGH_SPACE; } class IBlockOutputStream; /** Different data structures that can be used for aggregation * For efficiency, the aggregation data itself is put into the pool. * Data and pool ownership (states of aggregate functions) * is acquired later - in `convertToBlocks` function, by the ColumnAggregateFunction object. * * Most data structures exist in two versions: normal and two-level (TwoLevel). * A two-level hash table works a little slower with a small number of different keys, * but with a large number of different keys scales better, because it allows * parallelize some operations (merging, post-processing) in a natural way. * * To ensure efficient work over a wide range of conditions, * first single-level hash tables are used, * and when the number of different keys is large enough, * they are converted to two-level ones. * * PS. There are many different approaches to the effective implementation of parallel and distributed aggregation, * best suited for different cases, and this approach is just one of them, chosen for a combination of reasons. */ using AggregatedDataWithoutKey = AggregateDataPtr; using AggregatedDataWithUInt8Key = FixedHashMap; using AggregatedDataWithUInt16Key = FixedHashMap; using AggregatedDataWithUInt64Key = HashMap>; using AggregatedDataWithShortStringKey = StringHashMap; using AggregatedDataWithStringKey = HashMapWithSavedHash; using AggregatedDataWithKeys128 = HashMap; using AggregatedDataWithKeys256 = HashMap; using AggregatedDataWithUInt64KeyTwoLevel = TwoLevelHashMap>; using AggregatedDataWithShortStringKeyTwoLevel = TwoLevelStringHashMap; using AggregatedDataWithStringKeyTwoLevel = TwoLevelHashMapWithSavedHash; using AggregatedDataWithKeys128TwoLevel = TwoLevelHashMap; using AggregatedDataWithKeys256TwoLevel = TwoLevelHashMap; /** Variants with better hash function, using more than 32 bits for hash. * Using for merging phase of external aggregation, where number of keys may be far greater than 4 billion, * but we keep in memory and merge only sub-partition of them simultaneously. * TODO We need to switch for better hash function not only for external aggregation, * but also for huge aggregation results on machines with terabytes of RAM. */ using AggregatedDataWithUInt64KeyHash64 = HashMap>; using AggregatedDataWithStringKeyHash64 = HashMapWithSavedHash; using AggregatedDataWithKeys128Hash64 = HashMap; using AggregatedDataWithKeys256Hash64 = HashMap; template struct AggregationDataWithNullKey : public Base { using Base::Base; bool & hasNullKeyData() { return has_null_key_data; } AggregateDataPtr & getNullKeyData() { return null_key_data; } bool hasNullKeyData() const { return has_null_key_data; } const AggregateDataPtr & getNullKeyData() const { return null_key_data; } size_t size() const { return Base::size() + (has_null_key_data ? 1 : 0); } bool empty() const { return Base::empty() && !has_null_key_data; } void clear() { Base::clear(); has_null_key_data = false; } void clearAndShrink() { Base::clearAndShrink(); has_null_key_data = false; } private: bool has_null_key_data = false; AggregateDataPtr null_key_data = nullptr; }; template struct AggregationDataWithNullKeyTwoLevel : public Base { using Base::impls; AggregationDataWithNullKeyTwoLevel() {} template explicit AggregationDataWithNullKeyTwoLevel(const Other & other) : Base(other) { impls[0].hasNullKeyData() = other.hasNullKeyData(); impls[0].getNullKeyData() = other.getNullKeyData(); } bool & hasNullKeyData() { return impls[0].hasNullKeyData(); } AggregateDataPtr & getNullKeyData() { return impls[0].getNullKeyData(); } bool hasNullKeyData() const { return impls[0].hasNullKeyData(); } const AggregateDataPtr & getNullKeyData() const { return impls[0].getNullKeyData(); } }; template using HashTableWithNullKey = AggregationDataWithNullKey>; template using StringHashTableWithNullKey = AggregationDataWithNullKey>; using AggregatedDataWithNullableUInt8Key = AggregationDataWithNullKey; using AggregatedDataWithNullableUInt16Key = AggregationDataWithNullKey; using AggregatedDataWithNullableUInt64Key = AggregationDataWithNullKey; using AggregatedDataWithNullableStringKey = AggregationDataWithNullKey; using AggregatedDataWithNullableUInt64KeyTwoLevel = AggregationDataWithNullKeyTwoLevel< TwoLevelHashMap, TwoLevelHashTableGrower<>, HashTableAllocator, HashTableWithNullKey>>; using AggregatedDataWithNullableShortStringKeyTwoLevel = AggregationDataWithNullKeyTwoLevel< TwoLevelStringHashMap>; using AggregatedDataWithNullableStringKeyTwoLevel = AggregationDataWithNullKeyTwoLevel< TwoLevelHashMapWithSavedHash, TwoLevelHashTableGrower<>, HashTableAllocator, HashTableWithNullKey>>; /// For the case where there is one numeric key. /// FieldType is UInt8/16/32/64 for any type with corresponding bit width. template struct AggregationMethodOneNumber { using Data = TData; using Key = typename Data::key_type; using Mapped = typename Data::mapped_type; Data data; AggregationMethodOneNumber() {} template AggregationMethodOneNumber(const Other & other) : data(other.data) {} /// To use one `Method` in different threads, use different `State`. using State = ColumnsHashing::HashMethodOneNumber; /// Use optimization for low cardinality. static const bool low_cardinality_optimization = false; // Insert the key from the hash table into columns. static void insertKeyIntoColumns(const Key & key, MutableColumns & key_columns, const Sizes & /*key_sizes*/) { auto key_holder = reinterpret_cast(&key); auto column = static_cast(key_columns[0].get()); column->insertRawData(key_holder); } }; /// For the case where there is one string key. template struct AggregationMethodString { using Data = TData; using Key = typename Data::key_type; using Mapped = typename Data::mapped_type; Data data; AggregationMethodString() {} template AggregationMethodString(const Other & other) : data(other.data) {} using State = ColumnsHashing::HashMethodString; static const bool low_cardinality_optimization = false; static void insertKeyIntoColumns(const StringRef & key, MutableColumns & key_columns, const Sizes &) { key_columns[0]->insertData(key.data, key.size); } }; /// Same as above but without cache template struct AggregationMethodStringNoCache { using Data = TData; using Key = typename Data::key_type; using Mapped = typename Data::mapped_type; Data data; AggregationMethodStringNoCache() {} template AggregationMethodStringNoCache(const Other & other) : data(other.data) {} using State = ColumnsHashing::HashMethodString; static const bool low_cardinality_optimization = false; static void insertKeyIntoColumns(const StringRef & key, MutableColumns & key_columns, const Sizes &) { key_columns[0]->insertData(key.data, key.size); } }; /// For the case where there is one fixed-length string key. template struct AggregationMethodFixedString { using Data = TData; using Key = typename Data::key_type; using Mapped = typename Data::mapped_type; Data data; AggregationMethodFixedString() {} template AggregationMethodFixedString(const Other & other) : data(other.data) {} using State = ColumnsHashing::HashMethodFixedString; static const bool low_cardinality_optimization = false; static void insertKeyIntoColumns(const StringRef & key, MutableColumns & key_columns, const Sizes &) { key_columns[0]->insertData(key.data, key.size); } }; /// Same as above but without cache template struct AggregationMethodFixedStringNoCache { using Data = TData; using Key = typename Data::key_type; using Mapped = typename Data::mapped_type; Data data; AggregationMethodFixedStringNoCache() {} template AggregationMethodFixedStringNoCache(const Other & other) : data(other.data) {} using State = ColumnsHashing::HashMethodFixedString; static const bool low_cardinality_optimization = false; static void insertKeyIntoColumns(const StringRef & key, MutableColumns & key_columns, const Sizes &) { key_columns[0]->insertData(key.data, key.size); } }; /// Single low cardinality column. template struct AggregationMethodSingleLowCardinalityColumn : public SingleColumnMethod { using Base = SingleColumnMethod; using BaseState = typename Base::State; using Data = typename Base::Data; using Key = typename Base::Key; using Mapped = typename Base::Mapped; using Base::data; AggregationMethodSingleLowCardinalityColumn() = default; template explicit AggregationMethodSingleLowCardinalityColumn(const Other & other) : Base(other) {} using State = ColumnsHashing::HashMethodSingleLowCardinalityColumn; static const bool low_cardinality_optimization = true; static void insertKeyIntoColumns(const Key & key, MutableColumns & key_columns_low_cardinality, const Sizes & /*key_sizes*/) { auto col = assert_cast(key_columns_low_cardinality[0].get()); if constexpr (std::is_same_v) { col->insertData(key.data, key.size); } else { col->insertData(reinterpret_cast(&key), sizeof(key)); } } }; /// For the case where all keys are of fixed length, and they fit in N (for example, 128) bits. template struct AggregationMethodKeysFixed { using Data = TData; using Key = typename Data::key_type; using Mapped = typename Data::mapped_type; static constexpr bool has_nullable_keys = has_nullable_keys_; static constexpr bool has_low_cardinality = has_low_cardinality_; Data data; AggregationMethodKeysFixed() {} template AggregationMethodKeysFixed(const Other & other) : data(other.data) {} using State = ColumnsHashing::HashMethodKeysFixed; static const bool low_cardinality_optimization = false; static void insertKeyIntoColumns(const Key & key, MutableColumns & key_columns, const Sizes & key_sizes) { size_t keys_size = key_columns.size(); static constexpr auto bitmap_size = has_nullable_keys ? std::tuple_size>::value : 0; /// In any hash key value, column values to be read start just after the bitmap, if it exists. size_t pos = bitmap_size; for (size_t i = 0; i < keys_size; ++i) { IColumn * observed_column; ColumnUInt8 * null_map; bool column_nullable = false; if constexpr (has_nullable_keys) column_nullable = isColumnNullable(*key_columns[i]); /// If we have a nullable column, get its nested column and its null map. if (column_nullable) { ColumnNullable & nullable_col = assert_cast(*key_columns[i]); observed_column = &nullable_col.getNestedColumn(); null_map = assert_cast(&nullable_col.getNullMapColumn()); } else { observed_column = key_columns[i].get(); null_map = nullptr; } bool is_null = false; if (column_nullable) { /// The current column is nullable. Check if the value of the /// corresponding key is nullable. Update the null map accordingly. size_t bucket = i / 8; size_t offset = i % 8; UInt8 val = (reinterpret_cast(&key)[bucket] >> offset) & 1; null_map->insertValue(val); is_null = val == 1; } if (has_nullable_keys && is_null) observed_column->insertDefault(); else { size_t size = key_sizes[i]; observed_column->insertData(reinterpret_cast(&key) + pos, size); pos += size; } } } }; /** Aggregates by concatenating serialized key values. * The serialized value differs in that it uniquely allows to deserialize it, having only the position with which it starts. * That is, for example, for strings, it contains first the serialized length of the string, and then the bytes. * Therefore, when aggregating by several strings, there is no ambiguity. */ template struct AggregationMethodSerialized { using Data = TData; using Key = typename Data::key_type; using Mapped = typename Data::mapped_type; Data data; AggregationMethodSerialized() {} template AggregationMethodSerialized(const Other & other) : data(other.data) {} using State = ColumnsHashing::HashMethodSerialized; static const bool low_cardinality_optimization = false; static void insertKeyIntoColumns(const StringRef & key, MutableColumns & key_columns, const Sizes &) { auto pos = key.data; for (auto & column : key_columns) pos = column->deserializeAndInsertFromArena(pos); } }; class Aggregator; using ColumnsHashing::HashMethodContext; using ColumnsHashing::HashMethodContextPtr; struct AggregatedDataVariants : private boost::noncopyable { /** Working with states of aggregate functions in the pool is arranged in the following (inconvenient) way: * - when aggregating, states are created in the pool using IAggregateFunction::create (inside - `placement new` of arbitrary structure); * - they must then be destroyed using IAggregateFunction::destroy (inside - calling the destructor of arbitrary structure); * - if aggregation is complete, then, in the Aggregator::convertToBlocks function, pointers to the states of aggregate functions * are written to ColumnAggregateFunction; ColumnAggregateFunction "acquires ownership" of them, that is - calls `destroy` in its destructor. * - if during the aggregation, before call to Aggregator::convertToBlocks, an exception was thrown, * then the states of aggregate functions must still be destroyed, * otherwise, for complex states (eg, AggregateFunctionUniq), there will be memory leaks; * - in this case, to destroy states, the destructor calls Aggregator::destroyAggregateStates method, * but only if the variable aggregator (see below) is not nullptr; * - that is, until you transfer ownership of the aggregate function states in the ColumnAggregateFunction, set the variable `aggregator`, * so that when an exception occurs, the states are correctly destroyed. * * PS. This can be corrected by making a pool that knows about which states of aggregate functions and in which order are put in it, and knows how to destroy them. * But this can hardly be done simply because it is planned to put variable-length strings into the same pool. * In this case, the pool will not be able to know with what offsets objects are stored. */ Aggregator * aggregator = nullptr; size_t keys_size{}; /// Number of keys. NOTE do we need this field? Sizes key_sizes; /// Dimensions of keys, if keys of fixed length /// Pools for states of aggregate functions. Ownership will be later transferred to ColumnAggregateFunction. Arenas aggregates_pools; Arena * aggregates_pool{}; /// The pool that is currently used for allocation. /** Specialization for the case when there are no keys, and for keys not fitted into max_rows_to_group_by. */ AggregatedDataWithoutKey without_key = nullptr; // Disable consecutive key optimization for Uint8/16, because they use a FixedHashMap // and the lookup there is almost free, so we don't need to cache the last lookup result std::unique_ptr> key8; std::unique_ptr> key16; std::unique_ptr> key32; std::unique_ptr> key64; std::unique_ptr> key_string; std::unique_ptr> key_fixed_string; std::unique_ptr> keys128; std::unique_ptr> keys256; std::unique_ptr> serialized; std::unique_ptr> key32_two_level; std::unique_ptr> key64_two_level; std::unique_ptr> key_string_two_level; std::unique_ptr> key_fixed_string_two_level; std::unique_ptr> keys128_two_level; std::unique_ptr> keys256_two_level; std::unique_ptr> serialized_two_level; std::unique_ptr> key64_hash64; std::unique_ptr> key_string_hash64; std::unique_ptr> key_fixed_string_hash64; std::unique_ptr> keys128_hash64; std::unique_ptr> keys256_hash64; std::unique_ptr> serialized_hash64; /// Support for nullable keys. std::unique_ptr> nullable_keys128; std::unique_ptr> nullable_keys256; std::unique_ptr> nullable_keys128_two_level; std::unique_ptr> nullable_keys256_two_level; /// Support for low cardinality. std::unique_ptr>> low_cardinality_key8; std::unique_ptr>> low_cardinality_key16; std::unique_ptr>> low_cardinality_key32; std::unique_ptr>> low_cardinality_key64; std::unique_ptr>> low_cardinality_key_string; std::unique_ptr>> low_cardinality_key_fixed_string; std::unique_ptr>> low_cardinality_key32_two_level; std::unique_ptr>> low_cardinality_key64_two_level; std::unique_ptr>> low_cardinality_key_string_two_level; std::unique_ptr>> low_cardinality_key_fixed_string_two_level; std::unique_ptr> low_cardinality_keys128; std::unique_ptr> low_cardinality_keys256; std::unique_ptr> low_cardinality_keys128_two_level; std::unique_ptr> low_cardinality_keys256_two_level; /// In this and similar macros, the option without_key is not considered. #define APPLY_FOR_AGGREGATED_VARIANTS(M) \ M(key8, false) \ M(key16, false) \ M(key32, false) \ M(key64, false) \ M(key_string, false) \ M(key_fixed_string, false) \ M(keys128, false) \ M(keys256, false) \ M(serialized, false) \ M(key32_two_level, true) \ M(key64_two_level, true) \ M(key_string_two_level, true) \ M(key_fixed_string_two_level, true) \ M(keys128_two_level, true) \ M(keys256_two_level, true) \ M(serialized_two_level, true) \ M(key64_hash64, false) \ M(key_string_hash64, false) \ M(key_fixed_string_hash64, false) \ M(keys128_hash64, false) \ M(keys256_hash64, false) \ M(serialized_hash64, false) \ M(nullable_keys128, false) \ M(nullable_keys256, false) \ M(nullable_keys128_two_level, true) \ M(nullable_keys256_two_level, true) \ M(low_cardinality_key8, false) \ M(low_cardinality_key16, false) \ M(low_cardinality_key32, false) \ M(low_cardinality_key64, false) \ M(low_cardinality_keys128, false) \ M(low_cardinality_keys256, false) \ M(low_cardinality_key_string, false) \ M(low_cardinality_key_fixed_string, false) \ M(low_cardinality_key32_two_level, true) \ M(low_cardinality_key64_two_level, true) \ M(low_cardinality_keys128_two_level, true) \ M(low_cardinality_keys256_two_level, true) \ M(low_cardinality_key_string_two_level, true) \ M(low_cardinality_key_fixed_string_two_level, true) \ enum class Type { EMPTY = 0, without_key, #define M(NAME, IS_TWO_LEVEL) NAME, APPLY_FOR_AGGREGATED_VARIANTS(M) #undef M }; Type type = Type::EMPTY; AggregatedDataVariants() : aggregates_pools(1, std::make_shared()), aggregates_pool(aggregates_pools.back().get()) {} bool empty() const { return type == Type::EMPTY; } void invalidate() { type = Type::EMPTY; } ~AggregatedDataVariants(); void init(Type type_) { switch (type_) { case Type::EMPTY: break; case Type::without_key: break; #define M(NAME, IS_TWO_LEVEL) \ case Type::NAME: NAME = std::make_unique(); break; APPLY_FOR_AGGREGATED_VARIANTS(M) #undef M } type = type_; } /// Number of rows (different keys). size_t size() const { switch (type) { case Type::EMPTY: return 0; case Type::without_key: return 1; #define M(NAME, IS_TWO_LEVEL) \ case Type::NAME: return NAME->data.size() + (without_key != nullptr); APPLY_FOR_AGGREGATED_VARIANTS(M) #undef M } __builtin_unreachable(); } /// The size without taking into account the row in which data is written for the calculation of TOTALS. size_t sizeWithoutOverflowRow() const { switch (type) { case Type::EMPTY: return 0; case Type::without_key: return 1; #define M(NAME, IS_TWO_LEVEL) \ case Type::NAME: return NAME->data.size(); APPLY_FOR_AGGREGATED_VARIANTS(M) #undef M } __builtin_unreachable(); } const char * getMethodName() const { switch (type) { case Type::EMPTY: return "EMPTY"; case Type::without_key: return "without_key"; #define M(NAME, IS_TWO_LEVEL) \ case Type::NAME: return #NAME; APPLY_FOR_AGGREGATED_VARIANTS(M) #undef M } __builtin_unreachable(); } bool isTwoLevel() const { switch (type) { case Type::EMPTY: return false; case Type::without_key: return false; #define M(NAME, IS_TWO_LEVEL) \ case Type::NAME: return IS_TWO_LEVEL; APPLY_FOR_AGGREGATED_VARIANTS(M) #undef M } __builtin_unreachable(); } #define APPLY_FOR_VARIANTS_CONVERTIBLE_TO_TWO_LEVEL(M) \ M(key32) \ M(key64) \ M(key_string) \ M(key_fixed_string) \ M(keys128) \ M(keys256) \ M(serialized) \ M(nullable_keys128) \ M(nullable_keys256) \ M(low_cardinality_key32) \ M(low_cardinality_key64) \ M(low_cardinality_keys128) \ M(low_cardinality_keys256) \ M(low_cardinality_key_string) \ M(low_cardinality_key_fixed_string) \ #define APPLY_FOR_VARIANTS_NOT_CONVERTIBLE_TO_TWO_LEVEL(M) \ M(key8) \ M(key16) \ M(key64_hash64) \ M(key_string_hash64)\ M(key_fixed_string_hash64) \ M(keys128_hash64) \ M(keys256_hash64) \ M(serialized_hash64) \ M(low_cardinality_key8) \ M(low_cardinality_key16) \ #define APPLY_FOR_VARIANTS_SINGLE_LEVEL(M) \ APPLY_FOR_VARIANTS_NOT_CONVERTIBLE_TO_TWO_LEVEL(M) \ APPLY_FOR_VARIANTS_CONVERTIBLE_TO_TWO_LEVEL(M) \ bool isConvertibleToTwoLevel() const { switch (type) { #define M(NAME) \ case Type::NAME: return true; APPLY_FOR_VARIANTS_CONVERTIBLE_TO_TWO_LEVEL(M) #undef M default: return false; } } void convertToTwoLevel(); #define APPLY_FOR_VARIANTS_TWO_LEVEL(M) \ M(key32_two_level) \ M(key64_two_level) \ M(key_string_two_level) \ M(key_fixed_string_two_level) \ M(keys128_two_level) \ M(keys256_two_level) \ M(serialized_two_level) \ M(nullable_keys128_two_level) \ M(nullable_keys256_two_level) \ M(low_cardinality_key32_two_level) \ M(low_cardinality_key64_two_level) \ M(low_cardinality_keys128_two_level) \ M(low_cardinality_keys256_two_level) \ M(low_cardinality_key_string_two_level) \ M(low_cardinality_key_fixed_string_two_level) \ #define APPLY_FOR_LOW_CARDINALITY_VARIANTS(M) \ M(low_cardinality_key8) \ M(low_cardinality_key16) \ M(low_cardinality_key32) \ M(low_cardinality_key64) \ M(low_cardinality_keys128) \ M(low_cardinality_keys256) \ M(low_cardinality_key_string) \ M(low_cardinality_key_fixed_string) \ M(low_cardinality_key32_two_level) \ M(low_cardinality_key64_two_level) \ M(low_cardinality_keys128_two_level) \ M(low_cardinality_keys256_two_level) \ M(low_cardinality_key_string_two_level) \ M(low_cardinality_key_fixed_string_two_level) \ bool isLowCardinality() { switch (type) { #define M(NAME) \ case Type::NAME: return true; APPLY_FOR_LOW_CARDINALITY_VARIANTS(M) #undef M default: return false; } } static HashMethodContextPtr createCache(Type type, const HashMethodContext::Settings & settings) { switch (type) { case Type::without_key: return nullptr; #define M(NAME, IS_TWO_LEVEL) \ case Type::NAME: \ { \ using TPtr ## NAME = decltype(AggregatedDataVariants::NAME); \ using T ## NAME = typename TPtr ## NAME ::element_type; \ return T ## NAME ::State::createContext(settings); \ } APPLY_FOR_AGGREGATED_VARIANTS(M) #undef M default: throw Exception("Unknown aggregated data variant.", ErrorCodes::UNKNOWN_AGGREGATED_DATA_VARIANT); } } }; using AggregatedDataVariantsPtr = std::shared_ptr; using ManyAggregatedDataVariants = std::vector; using ManyAggregatedDataVariantsPtr = std::shared_ptr; /** How are "total" values calculated with WITH TOTALS? * (For more details, see TotalsHavingBlockInputStream.) * * In the absence of group_by_overflow_mode = 'any', the data is aggregated as usual, but the states of the aggregate functions are not finalized. * Later, the aggregate function states for all rows (passed through HAVING) are merged into one - this will be TOTALS. * * If there is group_by_overflow_mode = 'any', the data is aggregated as usual, except for the keys that did not fit in max_rows_to_group_by. * For these keys, the data is aggregated into one additional row - see below under the names `overflow_row`, `overflows`... * Later, the aggregate function states for all rows (passed through HAVING) are merged into one, * also overflow_row is added or not added (depending on the totals_mode setting) also - this will be TOTALS. */ /** Aggregates the source of the blocks. */ class Aggregator { public: struct Params { /// Data structure of source blocks. Block src_header; /// Data structure of intermediate blocks before merge. Block intermediate_header; /// What to count. const ColumnNumbers keys; const AggregateDescriptions aggregates; const size_t keys_size; const size_t aggregates_size; /// The settings of approximate calculation of GROUP BY. const bool overflow_row; /// Do we need to put into AggregatedDataVariants::without_key aggregates for keys that are not in max_rows_to_group_by. const size_t max_rows_to_group_by; const OverflowMode group_by_overflow_mode; /// Two-level aggregation settings (used for a large number of keys). /** With how many keys or the size of the aggregation state in bytes, * two-level aggregation begins to be used. Enough to reach of at least one of the thresholds. * 0 - the corresponding threshold is not specified. */ const size_t group_by_two_level_threshold; const size_t group_by_two_level_threshold_bytes; /// Settings to flush temporary data to the filesystem (external aggregation). const size_t max_bytes_before_external_group_by; /// 0 - do not use external aggregation. /// Return empty result when aggregating without keys on empty set. bool empty_result_for_aggregation_by_empty_set; const std::string tmp_path; /// Settings is used to determine cache size. No threads are created. size_t max_threads; const size_t min_free_disk_space; Params( const Block & src_header_, const ColumnNumbers & keys_, const AggregateDescriptions & aggregates_, bool overflow_row_, size_t max_rows_to_group_by_, OverflowMode group_by_overflow_mode_, size_t group_by_two_level_threshold_, size_t group_by_two_level_threshold_bytes_, size_t max_bytes_before_external_group_by_, bool empty_result_for_aggregation_by_empty_set_, const std::string & tmp_path_, size_t max_threads_, size_t min_free_disk_space_) : src_header(src_header_), keys(keys_), aggregates(aggregates_), keys_size(keys.size()), aggregates_size(aggregates.size()), overflow_row(overflow_row_), max_rows_to_group_by(max_rows_to_group_by_), group_by_overflow_mode(group_by_overflow_mode_), group_by_two_level_threshold(group_by_two_level_threshold_), group_by_two_level_threshold_bytes(group_by_two_level_threshold_bytes_), max_bytes_before_external_group_by(max_bytes_before_external_group_by_), empty_result_for_aggregation_by_empty_set(empty_result_for_aggregation_by_empty_set_), tmp_path(tmp_path_), max_threads(max_threads_), min_free_disk_space(min_free_disk_space_) { } /// Only parameters that matter during merge. Params(const Block & intermediate_header_, const ColumnNumbers & keys_, const AggregateDescriptions & aggregates_, bool overflow_row_, size_t max_threads_) : Params(Block(), keys_, aggregates_, overflow_row_, 0, OverflowMode::THROW, 0, 0, 0, false, "", max_threads_, 0) { intermediate_header = intermediate_header_; } }; Aggregator(const Params & params_); /// Aggregate the source. Get the result in the form of one of the data structures. void execute(const BlockInputStreamPtr & stream, AggregatedDataVariants & result); using AggregateColumns = std::vector; using AggregateColumnsData = std::vector; using AggregateColumnsConstData = std::vector; using AggregateFunctionsPlainPtrs = std::vector; /// Process one block. Return false if the processing should be aborted (with group_by_overflow_mode = 'break'). bool executeOnBlock(const Block & block, AggregatedDataVariants & result, ColumnRawPtrs & key_columns, AggregateColumns & aggregate_columns, /// Passed to not create them anew for each block bool & no_more_keys); bool executeOnBlock(Columns columns, UInt64 num_rows, AggregatedDataVariants & result, ColumnRawPtrs & key_columns, AggregateColumns & aggregate_columns, /// Passed to not create them anew for each block bool & no_more_keys); /** Convert the aggregation data structure into a block. * If overflow_row = true, then aggregates for rows that are not included in max_rows_to_group_by are put in the first block. * * If final = false, then ColumnAggregateFunction is created as the aggregation columns with the state of the calculations, * which can then be combined with other states (for distributed query processing). * If final = true, then columns with ready values are created as aggregate columns. */ BlocksList convertToBlocks(AggregatedDataVariants & data_variants, bool final, size_t max_threads) const; /** Merge several aggregation data structures and output the result as a block stream. */ std::unique_ptr mergeAndConvertToBlocks(ManyAggregatedDataVariants & data_variants, bool final, size_t max_threads) const; ManyAggregatedDataVariants prepareVariantsToMerge(ManyAggregatedDataVariants & data_variants) const; /** Merge the stream of partially aggregated blocks into one data structure. * (Pre-aggregate several blocks that represent the result of independent aggregations from remote servers.) */ void mergeStream(const BlockInputStreamPtr & stream, AggregatedDataVariants & result, size_t max_threads); using BucketToBlocks = std::map; /// Merge partially aggregated blocks separated to buckets into one data structure. void mergeBlocks(BucketToBlocks bucket_to_blocks, AggregatedDataVariants & result, size_t max_threads); /// Merge several partially aggregated blocks into one. /// Precondition: for all blocks block.info.is_overflows flag must be the same. /// (either all blocks are from overflow data or none blocks are). /// The resulting block has the same value of is_overflows flag. Block mergeBlocks(BlocksList & blocks, bool final); /** Split block with partially-aggregated data to many blocks, as if two-level method of aggregation was used. * This is needed to simplify merging of that data with other results, that are already two-level. */ std::vector convertBlockToTwoLevel(const Block & block); using CancellationHook = std::function; /** Set a function that checks whether the current task can be aborted. */ void setCancellationHook(const CancellationHook cancellation_hook); /// For external aggregation. void writeToTemporaryFile(AggregatedDataVariants & data_variants); bool hasTemporaryFiles() const { return !temporary_files.empty(); } struct TemporaryFiles { std::vector> files; size_t sum_size_uncompressed = 0; size_t sum_size_compressed = 0; mutable std::mutex mutex; bool empty() const { std::lock_guard lock(mutex); return files.empty(); } }; const TemporaryFiles & getTemporaryFiles() const { return temporary_files; } /// Get data structure of the result. Block getHeader(bool final) const; protected: friend struct AggregatedDataVariants; friend class MergingAndConvertingBlockInputStream; friend class ConvertingAggregatedToChunksTransform; friend class ConvertingAggregatedToChunksSource; Params params; AggregatedDataVariants::Type method_chosen; Sizes key_sizes; HashMethodContextPtr aggregation_state_cache; AggregateFunctionsPlainPtrs aggregate_functions; /** This array serves two purposes. * * 1. Function arguments are collected side by side, and they do not need to be collected from different places. Also the array is made zero-terminated. * The inner loop (for the case without_key) is almost twice as compact; performance gain of about 30%. * * 2. Calling a function by pointer is better than a virtual call, because in the case of a virtual call, * GCC 5.1.2 generates code that, at each iteration of the loop, reloads the function address from memory into the register * (the offset value in the virtual function table). */ struct AggregateFunctionInstruction { const IAggregateFunction * that; IAggregateFunction::AddFunc func; size_t state_offset; const IColumn ** arguments; const IAggregateFunction * batch_that; const IColumn ** batch_arguments; const UInt64 * offsets = nullptr; }; using AggregateFunctionInstructions = std::vector; Sizes offsets_of_aggregate_states; /// The offset to the n-th aggregate function in a row of aggregate functions. size_t total_size_of_aggregate_states = 0; /// The total size of the row from the aggregate functions. // add info to track alignment requirement // If there are states whose alignmentment are v1, ..vn, align_aggregate_states will be max(v1, ... vn) size_t align_aggregate_states = 1; bool all_aggregates_has_trivial_destructor = false; /// How many RAM were used to process the query before processing the first block. Int64 memory_usage_before_aggregation = 0; std::mutex mutex; Logger * log = &Logger::get("Aggregator"); /// Returns true if you can abort the current task. CancellationHook isCancelled; /// For external aggregation. TemporaryFiles temporary_files; /** Select the aggregation method based on the number and types of keys. */ AggregatedDataVariants::Type chooseAggregationMethod(); /** Create states of aggregate functions for one key. */ void createAggregateStates(AggregateDataPtr & aggregate_data) const; /** Call `destroy` methods for states of aggregate functions. * Used in the exception handler for aggregation, since RAII in this case is not applicable. */ void destroyAllAggregateStates(AggregatedDataVariants & result); /// Process one data block, aggregate the data into a hash table. template void executeImpl( Method & method, Arena * aggregates_pool, size_t rows, ColumnRawPtrs & key_columns, AggregateFunctionInstruction * aggregate_instructions, bool no_more_keys, AggregateDataPtr overflow_row) const; /// Specialization for a particular value no_more_keys. template void executeImplCase( Method & method, typename Method::State & state, Arena * aggregates_pool, size_t rows, AggregateFunctionInstruction * aggregate_instructions, AggregateDataPtr overflow_row) const; template void executeImplBatch( Method & method, typename Method::State & state, Arena * aggregates_pool, size_t rows, AggregateFunctionInstruction * aggregate_instructions) const; /// For case when there are no keys (all aggregate into one row). void executeWithoutKeyImpl( AggregatedDataWithoutKey & res, size_t rows, AggregateFunctionInstruction * aggregate_instructions, Arena * arena) const; template void writeToTemporaryFileImpl( AggregatedDataVariants & data_variants, Method & method, IBlockOutputStream & out); protected: /// Merge NULL key data from hash table `src` into `dst`. template void mergeDataNullKey( Table & table_dst, Table & table_src, Arena * arena) const; /// Merge data from hash table `src` into `dst`. template void mergeDataImpl( Table & table_dst, Table & table_src, Arena * arena) const; /// Merge data from hash table `src` into `dst`, but only for keys that already exist in dst. In other cases, merge the data into `overflows`. template void mergeDataNoMoreKeysImpl( Table & table_dst, AggregatedDataWithoutKey & overflows, Table & table_src, Arena * arena) const; /// Same, but ignores the rest of the keys. template void mergeDataOnlyExistingKeysImpl( Table & table_dst, Table & table_src, Arena * arena) const; void mergeWithoutKeyDataImpl( ManyAggregatedDataVariants & non_empty_data) const; template void mergeSingleLevelDataImpl( ManyAggregatedDataVariants & non_empty_data) const; template void convertToBlockImpl( Method & method, Table & data, MutableColumns & key_columns, AggregateColumnsData & aggregate_columns, MutableColumns & final_aggregate_columns, bool final) const; template void convertToBlockImplFinal( Method & method, Table & data, MutableColumns & key_columns, MutableColumns & final_aggregate_columns) const; template void convertToBlockImplNotFinal( Method & method, Table & data, MutableColumns & key_columns, AggregateColumnsData & aggregate_columns) const; template Block prepareBlockAndFill( AggregatedDataVariants & data_variants, bool final, size_t rows, Filler && filler) const; template Block convertOneBucketToBlock( AggregatedDataVariants & data_variants, Method & method, bool final, size_t bucket) const; Block mergeAndConvertOneBucketToBlock( ManyAggregatedDataVariants & variants, Arena * arena, bool final, size_t bucket, std::atomic * is_cancelled = nullptr) const; Block prepareBlockAndFillWithoutKey(AggregatedDataVariants & data_variants, bool final, bool is_overflows) const; Block prepareBlockAndFillSingleLevel(AggregatedDataVariants & data_variants, bool final) const; BlocksList prepareBlocksAndFillTwoLevel(AggregatedDataVariants & data_variants, bool final, ThreadPool * thread_pool) const; template BlocksList prepareBlocksAndFillTwoLevelImpl( AggregatedDataVariants & data_variants, Method & method, bool final, ThreadPool * thread_pool) const; template void mergeStreamsImplCase( Block & block, Arena * aggregates_pool, Method & method, Table & data, AggregateDataPtr overflow_row) const; template void mergeStreamsImpl( Block & block, Arena * aggregates_pool, Method & method, Table & data, AggregateDataPtr overflow_row, bool no_more_keys) const; void mergeWithoutKeyStreamsImpl( Block & block, AggregatedDataVariants & result) const; template void mergeBucketImpl( ManyAggregatedDataVariants & data, Int32 bucket, Arena * arena, std::atomic * is_cancelled = nullptr) const; template void convertBlockToTwoLevelImpl( Method & method, Arena * pool, ColumnRawPtrs & key_columns, const Block & source, std::vector & destinations) const; template void destroyImpl(Table & table) const; void destroyWithoutKey( AggregatedDataVariants & result) const; /** Checks constraints on the maximum number of keys for aggregation. * If it is exceeded, then, depending on the group_by_overflow_mode, either * - throws an exception; * - returns false, which means that execution must be aborted; * - sets the variable no_more_keys to true. */ bool checkLimits(size_t result_size, bool & no_more_keys) const; }; /** Get the aggregation variant by its type. */ template Method & getDataVariant(AggregatedDataVariants & variants); #define M(NAME, IS_TWO_LEVEL) \ template <> inline decltype(AggregatedDataVariants::NAME)::element_type & getDataVariant(AggregatedDataVariants & variants) { return *variants.NAME; } APPLY_FOR_AGGREGATED_VARIANTS(M) #undef M }