This is an automated email from the ASF dual-hosted git repository. dataroaring pushed a commit to branch branch-3.0 in repository https://gitbox.apache.org/repos/asf/doris.git
commit 544aa5ec938a113d8358617f7be54eb72a24973d Author: Pxl <[email protected]> AuthorDate: Thu Sep 5 22:21:00 2024 +0800 [Compatibility](agg-state) add be_exec_version check to avoid aggregation data format changed le… (#40193) …ad core dump ## Proposed changes add be_exec_version check to avoid aggregation data format changed lead core dump ``` mysql [test]>select percentile(k3, 0.1) from d_table group by grouping sets((k1),()) order by 1; ERROR 1105 (HY000): errCode = 2, detailMessage = (hk2.dev.selectdb-in.cc)[INTERNAL_ERROR]agg state data with percentile is not supported,current_be_exec_version=7, data_be_exec_version=4, need to rebuild the data or set the be_exec_version=4 in fe.conf ``` --- ...rsion_manager.h => be_exec_version_manager.cpp} | 63 ++++++++++--------- be/src/agent/be_exec_version_manager.h | 70 +++++----------------- be/src/olap/memtable.cpp | 7 ++- be/src/olap/rowset/segment_v2/column_reader.cpp | 31 ++++------ be/src/olap/rowset/segment_v2/column_reader.h | 7 ++- be/src/olap/rowset/segment_v2/segment.cpp | 16 ++++- be/src/olap/rowset/segment_v2/segment.h | 3 + be/src/olap/schema_change.cpp | 4 +- be/src/olap/tablet_reader.h | 8 +++ be/src/olap/tablet_schema.cpp | 14 +++-- be/src/olap/tablet_schema.h | 5 +- .../aggregate_function_distinct.cpp | 3 +- .../aggregate_function_foreach.cpp | 3 +- .../aggregate_function_simple_factory.h | 3 +- be/src/vec/data_types/data_type_agg_state.h | 15 +++-- be/src/vec/olap/block_reader.cpp | 4 +- be/src/vec/olap/vertical_block_reader.cpp | 3 +- .../vec/aggregate_functions/agg_bitmap_test.cpp | 2 +- .../vec/aggregate_functions/agg_collect_test.cpp | 2 +- .../vec/aggregate_functions/agg_histogram_test.cpp | 2 +- .../aggregate_functions/agg_min_max_by_test.cpp | 2 +- .../vec/aggregate_functions/agg_min_max_test.cpp | 6 +- .../vec/aggregate_functions/agg_replace_test.cpp | 4 +- be/test/vec/aggregate_functions/agg_test.cpp | 4 +- .../aggregate_functions/vec_count_by_enum_test.cpp | 2 +- .../vec/aggregate_functions/vec_retention_test.cpp | 2 +- .../vec_sequence_match_test.cpp | 10 ++-- .../aggregate_functions/vec_window_funnel_test.cpp | 2 +- 28 files changed, 140 insertions(+), 157 deletions(-) diff --git a/be/src/agent/be_exec_version_manager.h b/be/src/agent/be_exec_version_manager.cpp similarity index 60% copy from be/src/agent/be_exec_version_manager.h copy to be/src/agent/be_exec_version_manager.cpp index a7b4e2dee20..13edb1e5bc2 100644 --- a/be/src/agent/be_exec_version_manager.h +++ b/be/src/agent/be_exec_version_manager.cpp @@ -15,36 +15,40 @@ // specific language governing permissions and limitations // under the License. -#pragma once - -#include <fmt/format.h> -#include <glog/logging.h> - -#include "common/status.h" +#include "agent/be_exec_version_manager.h" namespace doris { -class BeExecVersionManager { -public: - BeExecVersionManager() = delete; +const std::map<int, const std::set<std::string>> AGGREGATION_CHANGE_MAP = { + {AGGREGATION_2_1_VERSION, + {"window_funnel", "stddev_samp", "variance_samp", "percentile_approx_weighted", + "percentile_approx", "covar_samp", "percentile", "percentile_array"}}}; - static Status check_be_exec_version(int be_exec_version) { - if (be_exec_version > max_be_exec_version || be_exec_version < min_be_exec_version) { - return Status::InternalError( - "Received be_exec_version is not supported, be_exec_version={}, " - "min_be_exec_version={}, max_be_exec_version={}, maybe due to FE version not " - "match with BE.", - be_exec_version, min_be_exec_version, max_be_exec_version); - } - return Status::OK(); +Status BeExecVersionManager::check_be_exec_version(int be_exec_version) { + if (be_exec_version > max_be_exec_version || be_exec_version < min_be_exec_version) { + return Status::InternalError( + "Received be_exec_version is not supported, be_exec_version={}, " + "min_be_exec_version={}, max_be_exec_version={}, maybe due to FE version not " + "match with BE.", + be_exec_version, min_be_exec_version, max_be_exec_version); } + return Status::OK(); +} - static int get_newest_version() { return max_be_exec_version; } - -private: - static const int max_be_exec_version; - static const int min_be_exec_version; -}; +void BeExecVersionManager::check_agg_state_compatibility(int current_be_exec_version, + int data_be_exec_version, + std::string function_name) { + if (current_be_exec_version > AGGREGATION_2_1_VERSION && + data_be_exec_version <= AGGREGATION_2_1_VERSION && + AGGREGATION_CHANGE_MAP.find(AGGREGATION_2_1_VERSION)->second.contains(function_name)) { + throw Exception(Status::InternalError( + "agg state data with {} is not supported, " + "current_be_exec_version={}, data_be_exec_version={}, need to rebuild the data " + "or set the be_exec_version={} in fe.conf", + function_name, current_be_exec_version, data_be_exec_version, + AGGREGATION_2_1_VERSION)); + } +} /** * When we have some breaking change for execute engine, we should update be_exec_version. @@ -82,14 +86,7 @@ private: * d. change some agg function nullable property: PR #37215 * e. change variant serde to fix PR #38413 */ -constexpr inline int BeExecVersionManager::max_be_exec_version = 7; -constexpr inline int BeExecVersionManager::min_be_exec_version = 0; - -/// functional -constexpr inline int BITMAP_SERDE = 3; -constexpr inline int USE_NEW_SERDE = 4; // release on DORIS version 2.1 -constexpr inline int OLD_WAL_SERDE = 3; // use to solve compatibility issues, see pr #32299 -constexpr inline int AGG_FUNCTION_NULLABLE = 5; // change some agg nullable property: PR #37215 -constexpr inline int VARIANT_SERDE = 6; // change variant serde to fix PR #38413 +const int BeExecVersionManager::max_be_exec_version = 7; +const int BeExecVersionManager::min_be_exec_version = 0; } // namespace doris diff --git a/be/src/agent/be_exec_version_manager.h b/be/src/agent/be_exec_version_manager.h index a7b4e2dee20..34cdbbb4b74 100644 --- a/be/src/agent/be_exec_version_manager.h +++ b/be/src/agent/be_exec_version_manager.h @@ -20,24 +20,28 @@ #include <fmt/format.h> #include <glog/logging.h> +#include "common/exception.h" #include "common/status.h" namespace doris { +/// functional +constexpr inline int BITMAP_SERDE = 3; +constexpr inline int USE_NEW_SERDE = 4; // release on DORIS version 2.1 +constexpr inline int OLD_WAL_SERDE = 3; // use to solve compatibility issues, see pr #32299 +constexpr inline int AGG_FUNCTION_NULLABLE = 5; // change some agg nullable property: PR #37215 +constexpr inline int VARIANT_SERDE = 6; // change variant serde to fix PR #38413 +constexpr inline int AGGREGATION_2_1_VERSION = + 4; // some aggregation changed the data format after this version + class BeExecVersionManager { public: BeExecVersionManager() = delete; - static Status check_be_exec_version(int be_exec_version) { - if (be_exec_version > max_be_exec_version || be_exec_version < min_be_exec_version) { - return Status::InternalError( - "Received be_exec_version is not supported, be_exec_version={}, " - "min_be_exec_version={}, max_be_exec_version={}, maybe due to FE version not " - "match with BE.", - be_exec_version, min_be_exec_version, max_be_exec_version); - } - return Status::OK(); - } + static Status check_be_exec_version(int be_exec_version); + + static void check_agg_state_compatibility(int current_be_exec_version, int data_be_exec_version, + std::string function_name); static int get_newest_version() { return max_be_exec_version; } @@ -46,50 +50,4 @@ private: static const int min_be_exec_version; }; -/** - * When we have some breaking change for execute engine, we should update be_exec_version. - * NOTICE: The change could only be dont in X.Y.0 version. and if you introduced new version number N, - * remember remove version N-1's all REUSEABLE changes in master branch only. REUSEABLE means scalar or agg functions' replacement. - * If not, the old replacement will happens in the new version which is wrong. - * - * 0: not contain be_exec_version. - * 1: start from doris 1.2.0 - * a. remove ColumnString terminating zero. - * b. runtime filter use new hash method. - * 2: start from doris 2.0.0 - * a. function month/day/hour/minute/second's return type is changed to smaller type. - * b. in order to solve agg of sum/count is not compatibility during the upgrade process - * c. change the string hash method in runtime filter - * d. elt function return type change to nullable(string) - * e. add repeat_max_num in repeat function - * 3: start from doris 2.0.0 (by some mistakes) - * a. aggregation function do not serialize bitmap to string. - * b. support window funnel mode. - * 4: start from doris 2.1.0 - * a. ignore this line, window funnel mode should be enabled from 2.0. - * b. array contains/position/countequal function return nullable in less situations. - * c. cleared old version of Version 2. - * d. unix_timestamp function support timestamp with float for datetimev2, and change nullable mode. - * e. change shuffle serialize/deserialize way - * f. shrink some function's nullable mode. - * g. do local merge of remote runtime filter - * h. "now": ALWAYS_NOT_NULLABLE -> DEPEND_ON_ARGUMENTS - * - * 5: start from doris 3.0.0 - * a. change the impl of percentile (need fix) - * b. clear old version of version 3->4 - * c. change FunctionIsIPAddressInRange from AlwaysNotNullable to DependOnArguments - * d. change some agg function nullable property: PR #37215 - * e. change variant serde to fix PR #38413 - */ -constexpr inline int BeExecVersionManager::max_be_exec_version = 7; -constexpr inline int BeExecVersionManager::min_be_exec_version = 0; - -/// functional -constexpr inline int BITMAP_SERDE = 3; -constexpr inline int USE_NEW_SERDE = 4; // release on DORIS version 2.1 -constexpr inline int OLD_WAL_SERDE = 3; // use to solve compatibility issues, see pr #32299 -constexpr inline int AGG_FUNCTION_NULLABLE = 5; // change some agg nullable property: PR #37215 -constexpr inline int VARIANT_SERDE = 6; // change variant serde to fix PR #38413 - } // namespace doris diff --git a/be/src/olap/memtable.cpp b/be/src/olap/memtable.cpp index 207778becae..a8084748e5a 100644 --- a/be/src/olap/memtable.cpp +++ b/be/src/olap/memtable.cpp @@ -108,10 +108,11 @@ void MemTable::_init_agg_functions(const vectorized::Block* block) { // the aggregate function manually. function = vectorized::AggregateFunctionSimpleFactory::instance().get( "replace_load", {block->get_data_type(cid)}, - block->get_data_type(cid)->is_nullable()); + block->get_data_type(cid)->is_nullable(), + BeExecVersionManager::get_newest_version()); } else { - function = - _tablet_schema->column(cid).get_aggregate_function(vectorized::AGG_LOAD_SUFFIX); + function = _tablet_schema->column(cid).get_aggregate_function( + vectorized::AGG_LOAD_SUFFIX, _tablet_schema->column(cid).get_be_exec_version()); if (function == nullptr) { LOG(WARNING) << "column get aggregate function failed, column=" << _tablet_schema->column(cid).name(); diff --git a/be/src/olap/rowset/segment_v2/column_reader.cpp b/be/src/olap/rowset/segment_v2/column_reader.cpp index cf156ca0496..6faefc34142 100644 --- a/be/src/olap/rowset/segment_v2/column_reader.cpp +++ b/be/src/olap/rowset/segment_v2/column_reader.cpp @@ -197,6 +197,7 @@ Status ColumnReader::create_agg_state(const ColumnReaderOptions& opts, const Col auto data_type = vectorized::DataTypeFactory::instance().create_data_type(meta); const auto* agg_state_type = assert_cast<const vectorized::DataTypeAggState*>(data_type.get()); + agg_state_type->check_agg_state_compatibility(opts.be_exec_version); auto type = agg_state_type->get_serialized_type()->get_type_as_type_descriptor().type; if (read_as_string(type)) { @@ -250,14 +251,12 @@ Status ColumnReader::create(const ColumnReaderOptions& opts, const ColumnMetaPB& } ColumnReader::ColumnReader(const ColumnReaderOptions& opts, const ColumnMetaPB& meta, - uint64_t num_rows, io::FileReaderSPtr file_reader, - vectorized::DataTypePtr agg_state_ptr) + uint64_t num_rows, io::FileReaderSPtr file_reader) : _use_index_page_cache(!config::disable_storage_page_cache), _opts(opts), _num_rows(num_rows), _file_reader(std::move(file_reader)), - _dict_encoding_type(UNKNOWN_DICT_ENCODING), - _agg_state_ptr(std::move(agg_state_ptr)) { + _dict_encoding_type(UNKNOWN_DICT_ENCODING) { _meta_length = meta.length(); _meta_type = (FieldType)meta.type(); if (_meta_type == FieldType::OLAP_FIELD_TYPE_ARRAY) { @@ -278,13 +277,18 @@ ColumnReader::~ColumnReader() { Status ColumnReader::init(const ColumnMetaPB* meta) { _type_info = get_type_info(meta); + + if (meta->has_be_exec_version()) { + _be_exec_version = meta->be_exec_version(); + } + if (_type_info == nullptr) { return Status::NotSupported("unsupported typeinfo, type={}", meta->type()); } RETURN_IF_ERROR(EncodingInfo::get(_type_info.get(), meta->encoding(), &_encoding_info)); for (int i = 0; i < meta->indexes_size(); i++) { - auto& index_meta = meta->indexes(i); + const auto& index_meta = meta->indexes(i); switch (index_meta.type()) { case ORDINAL_INDEX: _ordinal_index.reset( @@ -726,21 +730,8 @@ Status ColumnReader::new_iterator(ColumnIterator** iterator) { } Status ColumnReader::new_agg_state_iterator(ColumnIterator** iterator) { - if (!_agg_state_ptr) { // meet old version ColumnMetaPB - *iterator = new FileColumnIterator(this); - return Status::OK(); - } - - const auto* agg_state_type = - assert_cast<const vectorized::DataTypeAggState*>(_agg_state_ptr.get()); - auto type = agg_state_type->get_serialized_type()->get_type_as_type_descriptor().type; - - if (read_as_string(type)) { - *iterator = new FileColumnIterator(this); - return Status::OK(); - } - - return Status::InternalError("Not supported"); + *iterator = new FileColumnIterator(this); + return Status::OK(); } Status ColumnReader::new_array_iterator(ColumnIterator** iterator) { diff --git a/be/src/olap/rowset/segment_v2/column_reader.h b/be/src/olap/rowset/segment_v2/column_reader.h index 2cbb08ecd47..6727ea7dc81 100644 --- a/be/src/olap/rowset/segment_v2/column_reader.h +++ b/be/src/olap/rowset/segment_v2/column_reader.h @@ -85,6 +85,8 @@ struct ColumnReaderOptions { bool verify_checksum = true; // for in memory olap table, use DURABLE CachePriority in page cache bool kept_in_memory = false; + + int be_exec_version = -1; }; struct ColumnIteratorOptions { @@ -206,7 +208,7 @@ public: private: ColumnReader(const ColumnReaderOptions& opts, const ColumnMetaPB& meta, uint64_t num_rows, - io::FileReaderSPtr file_reader, vectorized::DataTypePtr agg_state_ptr = nullptr); + io::FileReaderSPtr file_reader); Status init(const ColumnMetaPB* meta); // Read column inverted indexes into memory @@ -248,6 +250,7 @@ private: FieldType _meta_children_column_type; bool _meta_is_nullable; bool _use_index_page_cache; + int _be_exec_version = -1; PagePointer _meta_dict_page; CompressionTypePB _meta_compression; @@ -276,8 +279,6 @@ private: std::vector<std::unique_ptr<ColumnReader>> _sub_readers; - vectorized::DataTypePtr _agg_state_ptr; - DorisCallOnce<Status> _set_dict_encoding_type_once; }; diff --git a/be/src/olap/rowset/segment_v2/segment.cpp b/be/src/olap/rowset/segment_v2/segment.cpp index 64f58e546c2..02538783656 100644 --- a/be/src/olap/rowset/segment_v2/segment.cpp +++ b/be/src/olap/rowset/segment_v2/segment.cpp @@ -189,6 +189,9 @@ Status Segment::_open_inverted_index() { Status Segment::new_iterator(SchemaSPtr schema, const StorageReadOptions& read_options, std::unique_ptr<RowwiseIterator>* iter) { + if (read_options.runtime_state != nullptr) { + _be_exec_version = read_options.runtime_state->be_exec_version(); + } RETURN_IF_ERROR(_create_column_readers_once()); read_options.stats->total_segment_number++; @@ -502,6 +505,7 @@ Status Segment::_create_column_readers(const SegmentFooterPB& footer) { ColumnReaderOptions opts { .kept_in_memory = _tablet_schema->is_in_memory(), + .be_exec_version = _be_exec_version, }; std::unique_ptr<ColumnReader> reader; RETURN_IF_ERROR(ColumnReader::create(opts, footer.columns(iter->second), footer.num_rows(), @@ -522,8 +526,10 @@ Status Segment::_create_column_readers(const SegmentFooterPB& footer) { continue; } const ColumnMetaPB& column_pb = footer.columns(iter->second); - ColumnReaderOptions opts; - opts.kept_in_memory = _tablet_schema->is_in_memory(); + ColumnReaderOptions opts { + .kept_in_memory = _tablet_schema->is_in_memory(), + .be_exec_version = _be_exec_version, + }; std::unique_ptr<ColumnReader> reader; RETURN_IF_ERROR( ColumnReader::create(opts, column_pb, footer.num_rows(), _file_reader, &reader)); @@ -736,6 +742,9 @@ Status Segment::new_column_iterator_with_path(const TabletColumn& tablet_column, Status Segment::new_column_iterator(const TabletColumn& tablet_column, std::unique_ptr<ColumnIterator>* iter, const StorageReadOptions* opt) { + if (opt != nullptr && opt->runtime_state != nullptr) { + _be_exec_version = opt->runtime_state->be_exec_version(); + } RETURN_IF_ERROR(_create_column_readers_once()); // init column iterator by path info @@ -809,6 +818,9 @@ Status Segment::new_inverted_index_iterator(const TabletColumn& tablet_column, const TabletIndex* index_meta, const StorageReadOptions& read_options, std::unique_ptr<InvertedIndexIterator>* iter) { + if (read_options.runtime_state != nullptr) { + _be_exec_version = read_options.runtime_state->be_exec_version(); + } RETURN_IF_ERROR(_create_column_readers_once()); ColumnReader* reader = _get_column_reader(tablet_column); if (reader != nullptr && index_meta) { diff --git a/be/src/olap/rowset/segment_v2/segment.h b/be/src/olap/rowset/segment_v2/segment.h index ec438f73151..59dbe8be5ba 100644 --- a/be/src/olap/rowset/segment_v2/segment.h +++ b/be/src/olap/rowset/segment_v2/segment.h @@ -29,6 +29,7 @@ #include <unordered_map> #include <vector> +#include "agent/be_exec_version_manager.h" #include "common/status.h" // Status #include "io/fs/file_reader_writer_fwd.h" #include "io/fs/file_system.h" @@ -285,6 +286,8 @@ private: DorisCallOnce<Status> _inverted_index_file_reader_open; InvertedIndexFileInfo _idx_file_info; + + int _be_exec_version = BeExecVersionManager::get_newest_version(); }; } // namespace segment_v2 diff --git a/be/src/olap/schema_change.cpp b/be/src/olap/schema_change.cpp index 4418d02ecdf..2d459703759 100644 --- a/be/src/olap/schema_change.cpp +++ b/be/src/olap/schema_change.cpp @@ -30,6 +30,7 @@ #include <tuple> #include <utility> +#include "agent/be_exec_version_manager.h" #include "cloud/cloud_schema_change_job.h" #include "cloud/config.h" #include "common/consts.h" @@ -133,7 +134,8 @@ public: try { vectorized::AggregateFunctionPtr function = tablet_schema->column(i).get_aggregate_function( - vectorized::AGG_LOAD_SUFFIX); + vectorized::AGG_LOAD_SUFFIX, + tablet_schema->column(i).get_be_exec_version()); agg_functions.push_back(function); // create aggregate data auto* place = new char[function->size_of_data()]; diff --git a/be/src/olap/tablet_reader.h b/be/src/olap/tablet_reader.h index 18ebb9653cc..3e467cb0f81 100644 --- a/be/src/olap/tablet_reader.h +++ b/be/src/olap/tablet_reader.h @@ -29,6 +29,7 @@ #include <utility> #include <vector> +#include "agent/be_exec_version_manager.h" #include "common/status.h" #include "exprs/function_filter.h" #include "gutil/strings/substitute.h" @@ -108,6 +109,13 @@ public: !rs_splits[1].rs_reader->rowset()->rowset_meta()->is_segments_overlapping()); } + int get_be_exec_version() const { + if (runtime_state) { + return runtime_state->be_exec_version(); + } + return BeExecVersionManager::get_newest_version(); + } + void set_read_source(ReadSource read_source) { rs_splits = std::move(read_source.rs_splits); delete_predicates = std::move(read_source.delete_predicates); diff --git a/be/src/olap/tablet_schema.cpp b/be/src/olap/tablet_schema.cpp index 77e44c5fdab..9348bbc1c4f 100644 --- a/be/src/olap/tablet_schema.cpp +++ b/be/src/olap/tablet_schema.cpp @@ -673,24 +673,28 @@ bool TabletColumn::is_row_store_column() const { } vectorized::AggregateFunctionPtr TabletColumn::get_aggregate_function_union( - vectorized::DataTypePtr type) const { + vectorized::DataTypePtr type, int current_be_exec_version) const { const auto* state_type = assert_cast<const vectorized::DataTypeAggState*>(type.get()); + BeExecVersionManager::check_agg_state_compatibility( + current_be_exec_version, _be_exec_version, + state_type->get_nested_function()->get_name()); return vectorized::AggregateStateUnion::create(state_type->get_nested_function(), {type}, type); } -vectorized::AggregateFunctionPtr TabletColumn::get_aggregate_function(std::string suffix) const { +vectorized::AggregateFunctionPtr TabletColumn::get_aggregate_function( + std::string suffix, int current_be_exec_version) const { vectorized::AggregateFunctionPtr function = nullptr; auto type = vectorized::DataTypeFactory::instance().create_data_type(*this); if (type && type->get_type_as_type_descriptor().type == PrimitiveType::TYPE_AGG_STATE) { - function = get_aggregate_function_union(type); + function = get_aggregate_function_union(type, current_be_exec_version); } else { std::string origin_name = TabletColumn::get_string_by_aggregation_type(_aggregation); std::string agg_name = origin_name + suffix; std::transform(agg_name.begin(), agg_name.end(), agg_name.begin(), [](unsigned char c) { return std::tolower(c); }); - function = vectorized::AggregateFunctionSimpleFactory::instance().get(agg_name, {type}, - type->is_nullable()); + function = vectorized::AggregateFunctionSimpleFactory::instance().get( + agg_name, {type}, type->is_nullable(), BeExecVersionManager::get_newest_version()); if (!function) { LOG(WARNING) << "get column aggregate function failed, aggregation_name=" << origin_name << ", column_type=" << type->get_name(); diff --git a/be/src/olap/tablet_schema.h b/be/src/olap/tablet_schema.h index e866726ecdc..e4f319474c8 100644 --- a/be/src/olap/tablet_schema.h +++ b/be/src/olap/tablet_schema.h @@ -122,8 +122,9 @@ public: void set_path_info(const vectorized::PathInData& path); FieldAggregationMethod aggregation() const { return _aggregation; } vectorized::AggregateFunctionPtr get_aggregate_function_union( - vectorized::DataTypePtr type) const; - vectorized::AggregateFunctionPtr get_aggregate_function(std::string suffix) const; + vectorized::DataTypePtr type, int current_be_exec_version) const; + vectorized::AggregateFunctionPtr get_aggregate_function(std::string suffix, + int current_be_exec_version) const; int precision() const { return _precision; } int frac() const { return _frac; } inline bool visible() const { return _visible; } diff --git a/be/src/vec/aggregate_functions/aggregate_function_distinct.cpp b/be/src/vec/aggregate_functions/aggregate_function_distinct.cpp index f86d44b7d68..9bb2954207b 100644 --- a/be/src/vec/aggregate_functions/aggregate_function_distinct.cpp +++ b/be/src/vec/aggregate_functions/aggregate_function_distinct.cpp @@ -91,7 +91,8 @@ void register_aggregate_function_combinator_distinct(AggregateFunctionSimpleFact auto function_combinator = std::make_shared<AggregateFunctionCombinatorDistinct>(); auto transform_arguments = function_combinator->transform_arguments(nested_types); auto nested_function_name = name.substr(DISTINCT_FUNCTION_PREFIX.size()); - auto nested_function = factory.get(nested_function_name, transform_arguments); + auto nested_function = factory.get(nested_function_name, transform_arguments, false, + BeExecVersionManager::get_newest_version()); return function_combinator->transform_aggregate_function(nested_function, types, result_is_nullable); }; diff --git a/be/src/vec/aggregate_functions/aggregate_function_foreach.cpp b/be/src/vec/aggregate_functions/aggregate_function_foreach.cpp index a0fb5c94f43..ab6d0142f6a 100644 --- a/be/src/vec/aggregate_functions/aggregate_function_foreach.cpp +++ b/be/src/vec/aggregate_functions/aggregate_function_foreach.cpp @@ -45,7 +45,8 @@ void register_aggregate_function_combinator_foreach(AggregateFunctionSimpleFacto } auto nested_function_name = name.substr(0, name.size() - suffix.size()); auto nested_function = - factory.get(nested_function_name, transform_arguments, result_is_nullable); + factory.get(nested_function_name, transform_arguments, result_is_nullable, + BeExecVersionManager::get_newest_version(), false); if (!nested_function) { throw Exception( ErrorCode::INTERNAL_ERROR, diff --git a/be/src/vec/aggregate_functions/aggregate_function_simple_factory.h b/be/src/vec/aggregate_functions/aggregate_function_simple_factory.h index ae9bd070a47..cc504b9f996 100644 --- a/be/src/vec/aggregate_functions/aggregate_function_simple_factory.h +++ b/be/src/vec/aggregate_functions/aggregate_function_simple_factory.h @@ -123,8 +123,7 @@ public: } AggregateFunctionPtr get(const std::string& name, const DataTypes& argument_types, - const bool result_is_nullable = false, - int be_version = BeExecVersionManager::get_newest_version(), + const bool result_is_nullable, int be_version, bool enable_decima256 = false) { bool nullable = false; for (const auto& type : argument_types) { diff --git a/be/src/vec/data_types/data_type_agg_state.h b/be/src/vec/data_types/data_type_agg_state.h index 96b147c8e96..d7089503b01 100644 --- a/be/src/vec/data_types/data_type_agg_state.h +++ b/be/src/vec/data_types/data_type_agg_state.h @@ -41,15 +41,13 @@ public: _sub_types(std::move(sub_types)), _function_name(std::move(function_name)), _be_exec_version(be_exec_version) { - _agg_function = AggregateFunctionSimpleFactory::instance().get(_function_name, _sub_types, - _result_is_nullable); - if (_agg_function == nullptr) { + _agg_function = AggregateFunctionSimpleFactory::instance().get( + _function_name, _sub_types, _result_is_nullable, _be_exec_version); + if (_agg_function == nullptr || + !BeExecVersionManager::check_be_exec_version(be_exec_version)) { throw Exception(ErrorCode::INVALID_ARGUMENT, "DataTypeAggState function get failed, type={}", do_get_name()); } - if (!BeExecVersionManager::check_be_exec_version(be_exec_version)) { - LOG(WARNING) << "meet old agg-state, be_exec_version=" << be_exec_version; - } _agg_function->set_version(be_exec_version); _agg_serialized_type = _agg_function->get_serialized_type(); } @@ -124,6 +122,11 @@ public: DataTypePtr get_serialized_type() const { return _agg_serialized_type; } + void check_agg_state_compatibility(int read_be_exec_version) const { + BeExecVersionManager::check_agg_state_compatibility(read_be_exec_version, _be_exec_version, + get_nested_function()->get_name()); + } + private: std::string get_types_string() const { std::string types; diff --git a/be/src/vec/olap/block_reader.cpp b/be/src/vec/olap/block_reader.cpp index a606b83345d..e2b4ba39e12 100644 --- a/be/src/vec/olap/block_reader.cpp +++ b/be/src/vec/olap/block_reader.cpp @@ -179,8 +179,8 @@ Status BlockReader::_init_agg_state(const ReaderParams& read_params) { for (auto idx : _agg_columns_idx) { auto column = tablet_schema.column( read_params.origin_return_columns->at(_return_columns_loc[idx])); - AggregateFunctionPtr function = - column.get_aggregate_function(vectorized::AGG_READER_SUFFIX); + AggregateFunctionPtr function = column.get_aggregate_function( + vectorized::AGG_READER_SUFFIX, read_params.get_be_exec_version()); // to avoid coredump when something goes wrong(i.e. column missmatch) if (!function) { diff --git a/be/src/vec/olap/vertical_block_reader.cpp b/be/src/vec/olap/vertical_block_reader.cpp index 56cb3f9c1c9..5367729f637 100644 --- a/be/src/vec/olap/vertical_block_reader.cpp +++ b/be/src/vec/olap/vertical_block_reader.cpp @@ -194,7 +194,8 @@ void VerticalBlockReader::_init_agg_state(const ReaderParams& read_params) { for (size_t idx = 0; idx < _return_columns.size(); ++idx) { AggregateFunctionPtr function = tablet_schema.column(_return_columns.at(idx)) - .get_aggregate_function(vectorized::AGG_READER_SUFFIX); + .get_aggregate_function(vectorized::AGG_READER_SUFFIX, + read_params.get_be_exec_version()); DCHECK(function != nullptr); _agg_functions.push_back(function); // create aggregate data diff --git a/be/test/vec/aggregate_functions/agg_bitmap_test.cpp b/be/test/vec/aggregate_functions/agg_bitmap_test.cpp index 6ca85efe321..429b1e0640c 100644 --- a/be/test/vec/aggregate_functions/agg_bitmap_test.cpp +++ b/be/test/vec/aggregate_functions/agg_bitmap_test.cpp @@ -62,7 +62,7 @@ TEST(AggBitmapTest, bitmap_union_test) { AggregateFunctionSimpleFactory factory; register_aggregate_function_bitmap(factory); DataTypes data_types = {data_type}; - auto agg_function = factory.get(function_name, data_types); + auto agg_function = factory.get(function_name, data_types, false, -1); agg_function->set_version(3); std::unique_ptr<char[]> memory(new char[agg_function->size_of_data()]); AggregateDataPtr place = memory.get(); diff --git a/be/test/vec/aggregate_functions/agg_collect_test.cpp b/be/test/vec/aggregate_functions/agg_collect_test.cpp index d592967cbe6..eb9cb647ed4 100644 --- a/be/test/vec/aggregate_functions/agg_collect_test.cpp +++ b/be/test/vec/aggregate_functions/agg_collect_test.cpp @@ -91,7 +91,7 @@ public: DataTypes data_types = {(DataTypePtr)std::make_shared<DataType>()}; LOG(INFO) << "test_agg_collect for " << fn_name << "(" << data_types[0]->get_name() << ")"; AggregateFunctionSimpleFactory factory = AggregateFunctionSimpleFactory::instance(); - auto agg_function = factory.get(fn_name, data_types); + auto agg_function = factory.get(fn_name, data_types, false, -1); EXPECT_NE(agg_function, nullptr); std::unique_ptr<char[]> memory(new char[agg_function->size_of_data()]); diff --git a/be/test/vec/aggregate_functions/agg_histogram_test.cpp b/be/test/vec/aggregate_functions/agg_histogram_test.cpp index 069b8762952..8c9384fcea1 100644 --- a/be/test/vec/aggregate_functions/agg_histogram_test.cpp +++ b/be/test/vec/aggregate_functions/agg_histogram_test.cpp @@ -120,7 +120,7 @@ public: << "(" << data_types[0]->get_name() << ")"; AggregateFunctionSimpleFactory factory = AggregateFunctionSimpleFactory::instance(); - auto agg_function = factory.get("histogram", data_types); + auto agg_function = factory.get("histogram", data_types, false, -1); EXPECT_NE(agg_function, nullptr); std::unique_ptr<char[]> memory(new char[agg_function->size_of_data()]); diff --git a/be/test/vec/aggregate_functions/agg_min_max_by_test.cpp b/be/test/vec/aggregate_functions/agg_min_max_by_test.cpp index e1113fddf12..137f4fc70b1 100644 --- a/be/test/vec/aggregate_functions/agg_min_max_by_test.cpp +++ b/be/test/vec/aggregate_functions/agg_min_max_by_test.cpp @@ -84,7 +84,7 @@ TEST_P(AggMinMaxByTest, min_max_by_test) { DataTypes data_types = {std::make_shared<DataTypeInt32>(), i == 0 ? (DataTypePtr)std::make_shared<DataTypeInt32>() : (DataTypePtr)std::make_shared<DataTypeString>()}; - auto agg_function = factory.get(min_max_by_type, data_types); + auto agg_function = factory.get(min_max_by_type, data_types, false, -1); std::unique_ptr<char[]> memory(new char[agg_function->size_of_data()]); AggregateDataPtr place = memory.get(); agg_function->create(place); diff --git a/be/test/vec/aggregate_functions/agg_min_max_test.cpp b/be/test/vec/aggregate_functions/agg_min_max_test.cpp index fef91af4d1f..a688703efb2 100644 --- a/be/test/vec/aggregate_functions/agg_min_max_test.cpp +++ b/be/test/vec/aggregate_functions/agg_min_max_test.cpp @@ -58,7 +58,7 @@ TEST_P(AggMinMaxTest, min_max_test) { AggregateFunctionSimpleFactory factory; register_aggregate_function_minmax(factory); DataTypes data_types = {std::make_shared<DataTypeInt32>()}; - auto agg_function = factory.get(min_max_type, data_types); + auto agg_function = factory.get(min_max_type, data_types, false, -1); std::unique_ptr<char[]> memory(new char[agg_function->size_of_data()]); AggregateDataPtr place = memory.get(); agg_function->create(place); @@ -90,7 +90,7 @@ TEST_P(AggMinMaxTest, min_max_decimal_test) { AggregateFunctionSimpleFactory factory; register_aggregate_function_minmax(factory); DataTypes data_types = {data_type}; - auto agg_function = factory.get(min_max_type, data_types); + auto agg_function = factory.get(min_max_type, data_types, false, -1); std::unique_ptr<char[]> memory(new char[agg_function->size_of_data()]); AggregateDataPtr place = memory.get(); agg_function->create(place); @@ -137,7 +137,7 @@ TEST_P(AggMinMaxTest, min_max_string_test) { AggregateFunctionSimpleFactory factory; register_aggregate_function_minmax(factory); DataTypes data_types = {std::make_shared<DataTypeString>()}; - auto agg_function = factory.get(min_max_type, data_types); + auto agg_function = factory.get(min_max_type, data_types, false, -1); std::unique_ptr<char[]> memory(new char[agg_function->size_of_data()]); AggregateDataPtr place = memory.get(); agg_function->create(place); diff --git a/be/test/vec/aggregate_functions/agg_replace_test.cpp b/be/test/vec/aggregate_functions/agg_replace_test.cpp index 092b7e4fe6a..f74d63df635 100644 --- a/be/test/vec/aggregate_functions/agg_replace_test.cpp +++ b/be/test/vec/aggregate_functions/agg_replace_test.cpp @@ -219,7 +219,7 @@ public: DataTypes data_types = {data_type}; LOG(INFO) << "test_agg_replace for " << fn_name << "(" << data_types[0]->get_name() << ")"; AggregateFunctionSimpleFactory factory = AggregateFunctionSimpleFactory::instance(); - auto agg_function = factory.get(fn_name, data_types, nullable); + auto agg_function = factory.get(fn_name, data_types, nullable, -1); EXPECT_NE(agg_function, nullptr); std::unique_ptr<char[]> memory(new char[agg_function->size_of_data()]); @@ -243,7 +243,7 @@ public: DataTypes data_types = {data_type}; LOG(INFO) << "test_agg_replace for " << fn_name << "(" << data_types[0]->get_name() << ")"; AggregateFunctionSimpleFactory factory = AggregateFunctionSimpleFactory::instance(); - auto agg_function = factory.get(fn_name, data_types, nullable); + auto agg_function = factory.get(fn_name, data_types, nullable, -1); EXPECT_NE(agg_function, nullptr); std::unique_ptr<char[]> memory(new char[agg_function->size_of_data()]); diff --git a/be/test/vec/aggregate_functions/agg_test.cpp b/be/test/vec/aggregate_functions/agg_test.cpp index 53d8cd69291..21ca1aa3bd6 100644 --- a/be/test/vec/aggregate_functions/agg_test.cpp +++ b/be/test/vec/aggregate_functions/agg_test.cpp @@ -52,7 +52,7 @@ TEST(AggTest, basic_test) { register_aggregate_function_sum(factory); DataTypePtr data_type(std::make_shared<DataTypeInt32>()); DataTypes data_types = {data_type}; - auto agg_function = factory.get("sum", data_types); + auto agg_function = factory.get("sum", data_types, false, -1); std::unique_ptr<char[]> memory(new char[agg_function->size_of_data()]); AggregateDataPtr place = memory.get(); agg_function->create(place); @@ -84,7 +84,7 @@ TEST(AggTest, topn_test) { register_aggregate_function_topn(factory); DataTypes data_types = {std::make_shared<DataTypeString>(), std::make_shared<DataTypeInt32>()}; - auto agg_function = factory.get("topn", data_types); + auto agg_function = factory.get("topn", data_types, false, -1); std::unique_ptr<char[]> memory(new char[agg_function->size_of_data()]); AggregateDataPtr place = memory.get(); agg_function->create(place); diff --git a/be/test/vec/aggregate_functions/vec_count_by_enum_test.cpp b/be/test/vec/aggregate_functions/vec_count_by_enum_test.cpp index fa953b5101c..aa9c33a3eeb 100644 --- a/be/test/vec/aggregate_functions/vec_count_by_enum_test.cpp +++ b/be/test/vec/aggregate_functions/vec_count_by_enum_test.cpp @@ -43,7 +43,7 @@ public: DataTypes data_types = { std::make_shared<DataTypeNullable>(std::make_shared<DataTypeString>()), }; - agg_function = factory.get("count_by_enum", data_types, true); + agg_function = factory.get("count_by_enum", data_types, true, -1); EXPECT_NE(agg_function, nullptr); } diff --git a/be/test/vec/aggregate_functions/vec_retention_test.cpp b/be/test/vec/aggregate_functions/vec_retention_test.cpp index e96abb84b7e..537026e3937 100644 --- a/be/test/vec/aggregate_functions/vec_retention_test.cpp +++ b/be/test/vec/aggregate_functions/vec_retention_test.cpp @@ -59,7 +59,7 @@ public: std::make_shared<DataTypeUInt8>(), std::make_shared<DataTypeUInt8>(), }; - agg_function = factory.get("retention", data_types, false); + agg_function = factory.get("retention", data_types, false, -1); EXPECT_NE(agg_function, nullptr); } diff --git a/be/test/vec/aggregate_functions/vec_sequence_match_test.cpp b/be/test/vec/aggregate_functions/vec_sequence_match_test.cpp index a034304dd6b..bee948fa774 100644 --- a/be/test/vec/aggregate_functions/vec_sequence_match_test.cpp +++ b/be/test/vec/aggregate_functions/vec_sequence_match_test.cpp @@ -55,9 +55,9 @@ public: std::make_shared<DataTypeString>(), std::make_shared<DataTypeDateTime>(), std::make_shared<DataTypeUInt8>(), std::make_shared<DataTypeUInt8>(), std::make_shared<DataTypeUInt8>()}; - agg_function_sequence_match = factory.get("sequence_match", data_types, false); + agg_function_sequence_match = factory.get("sequence_match", data_types, false, -1); EXPECT_NE(agg_function_sequence_match, nullptr); - agg_function_sequence_count = factory.get("sequence_count", data_types, false); + agg_function_sequence_count = factory.get("sequence_count", data_types, false, -1); EXPECT_NE(agg_function_sequence_count, nullptr); } @@ -191,7 +191,7 @@ TEST_F(VSequenceMatchTest, testCountSerialize) { DataTypes data_types = {std::make_shared<DataTypeString>(), std::make_shared<DataTypeDateTime>(), std::make_shared<DataTypeUInt8>(), std::make_shared<DataTypeUInt8>()}; - agg_function_sequence_count = factory.get("sequence_count", data_types, false); + agg_function_sequence_count = factory.get("sequence_count", data_types, false, -1); EXPECT_NE(agg_function_sequence_count, nullptr); const int NUM_CONDS = 4; @@ -256,7 +256,7 @@ TEST_F(VSequenceMatchTest, testMatchReverseSortedSerializeMerge) { DataTypes data_types = {std::make_shared<DataTypeString>(), std::make_shared<DataTypeDateTime>(), std::make_shared<DataTypeUInt8>(), std::make_shared<DataTypeUInt8>()}; - agg_function_sequence_match = factory.get("sequence_match", data_types, false); + agg_function_sequence_match = factory.get("sequence_match", data_types, false, -1); EXPECT_NE(agg_function_sequence_match, nullptr); const int NUM_CONDS = 2; @@ -345,7 +345,7 @@ TEST_F(VSequenceMatchTest, testCountReverseSortedSerializeMerge) { DataTypes data_types = {std::make_shared<DataTypeString>(), std::make_shared<DataTypeDateTime>(), std::make_shared<DataTypeUInt8>(), std::make_shared<DataTypeUInt8>()}; - agg_function_sequence_count = factory.get("sequence_count", data_types, false); + agg_function_sequence_count = factory.get("sequence_count", data_types, false, -1); EXPECT_NE(agg_function_sequence_count, nullptr); const int NUM_CONDS = 2; diff --git a/be/test/vec/aggregate_functions/vec_window_funnel_test.cpp b/be/test/vec/aggregate_functions/vec_window_funnel_test.cpp index 5ea6ce205ba..ed019ef699b 100644 --- a/be/test/vec/aggregate_functions/vec_window_funnel_test.cpp +++ b/be/test/vec/aggregate_functions/vec_window_funnel_test.cpp @@ -58,7 +58,7 @@ public: std::make_shared<DataTypeDateTime>(), std::make_shared<DataTypeUInt8>(), std::make_shared<DataTypeUInt8>(), std::make_shared<DataTypeUInt8>(), std::make_shared<DataTypeUInt8>()}; - agg_function = factory.get("window_funnel", data_types, false); + agg_function = factory.get("window_funnel", data_types, false, -1); EXPECT_NE(agg_function, nullptr); } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
