This is an automated email from the ASF dual-hosted git repository.
mrhhsg pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 85ef38e0097 [fix](be) Support IP keys in map_agg_v2 (#68426)
85ef38e0097 is described below
commit 85ef38e0097ae8acb85ef8c2bdf970f9ea9803bf
Author: Jerry Hu <[email protected]>
AuthorDate: Tue Oct 13 11:47:12 2026 +0800
[fix](be) Support IP keys in map_agg_v2 (#68426)
### What problem does this PR solve?
Issue Number: None
Problem Summary:
`map_agg_v2` accepts IP key types during analysis, but the backend
aggregate factory did not dispatch IPv4 or IPv6 keys. Queries therefore
failed at execution with an unimplemented aggregate function error.
Adding the missing dispatch exposed an aggregate-state framing issue:
the serializer's uncompressed-size API returns an allocation bound
including the StreamVByte decoder's readable suffix, while `serialize()`
returns the logical frame end. The legacy key state wrote the whole
allocation, whereas value states used the logical length. Trimming the
key frame without version routing would break mixed-version execution,
and an exact-sized string alone does not supply the decoder's private
suffix.
This change:
- adds IPv4 and IPv6 key dispatch and preserves master's UUID key
dispatch;
- stores incoming frames in `PaddedPODArray` so decoders have a private
readable suffix;
- writes exact logical key/value frames at BE execution version 17 while
retaining the legacy key frame through version 16;
- separates transient wire-format routing from persisted aggregate-state
compatibility, so existing version-15/16 Map states remain readable and
compactable at version 17;
- propagates the negotiated execution version into nested aggregate
functions such as `map_agg_v2_foreachv2`;
- reuses one write scratch buffer and one padded read scratch buffer
across key and value frames;
- binds restricted aggregates to their original version-8 support
boundary rather than the moving global maximum, so version-15/16
`stddev_samp` queries and stored states remain usable on upgraded BEs;
and
- enables IP keys only after all backends reach execution version 17.
Master reserved version 16 for Paimon table writes; it cannot also
identify support for the new Map frame.
### Release note
Fix `map_agg_v2` with IPv4 and IPv6 keys after all backends complete the
upgrade to execution version 17. Preserve stored Map and restricted
aggregate states across the upgrade.
### Check List (For Author)
- Test:
- Unit Test: `./run-be-ut.sh -j24 --run
--filter='AggregateFunctionMapAggV2Test.*'` (8 passed)
- Regression tests: `test_restricted_aggregate_versioned_state`
(version-15 direct query and stored state read at versions 16 and 17),
`test_map_agg_v2_versioned_state` (version-16 write, version-17 read and
full compaction), and `test_map_agg_v2_ip_key` passed in an isolated
ASAN cluster
- Build: `./build.sh --be --fe -j16` passed with ASAN after merging
master
- Style: BE clang-format, check-format, build-hygiene, and PR net `git
diff --check` passed
- Static analysis: changed production C++ files had no warnings with
matching Clang 16 plus a compile-flag override. The default Clang 20
invocation could not find `stddef.h`; the BE test translation unit still
reports pre-existing `jni-util.h` static assertions under Clang 16 and
an existing `DataTypeAggState` constructor diagnostic under Clang 20
- A two-version BE cluster was not run; legacy and exact routing were
covered by BE unit tests and versioned regression queries on an upgraded
BE
- Behavior changed: Yes (version-17 exact frames and IP-key support,
legacy wire routing and persisted-state compatibility preserved)
- Does this need documentation: No
---
be/src/agent/be_exec_version_manager.cpp | 56 ++-
be/src/agent/be_exec_version_manager.h | 24 +-
be/src/exprs/aggregate/aggregate_function.h | 2 +
.../exprs/aggregate/aggregate_function_covar.cpp | 3 +-
.../aggregate/aggregate_function_distinct.cpp | 4 +-
.../exprs/aggregate/aggregate_function_foreach.cpp | 4 +-
.../aggregate/aggregate_function_foreachv2.cpp | 4 +-
.../aggregate_function_map_combinator.cpp | 5 +-
.../exprs/aggregate/aggregate_function_map_v2.cpp | 31 +-
be/src/exprs/aggregate/aggregate_function_map_v2.h | 75 ++--
.../aggregate/aggregate_function_percentile.cpp | 12 +-
.../aggregate/aggregate_function_simple_factory.h | 91 +++--
.../exprs/aggregate/aggregate_function_stddev.cpp | 6 +-
.../aggregate/aggregate_function_window_funnel.cpp | 3 +-
.../writer/paimon/paimon_write_backend_test.cpp | 7 +-
be/test/exprs/aggregate/agg_map_agg_v2_test.cpp | 415 +++++++++++++++++++++
.../main/java/org/apache/doris/common/Config.java | 4 +-
.../map/test_map_agg_v2_versioned_state.out | 6 +
.../test_restricted_aggregate_versioned_state.out | 15 +
.../data/datatype_p0/ip/test_map_agg_v2_ip_key.out | 12 +
.../map/test_map_agg_v2_versioned_state.groovy | 91 +++++
...est_restricted_aggregate_versioned_state.groovy | 72 ++++
.../datatype_p0/ip/test_map_agg_v2_ip_key.groovy | 77 ++++
23 files changed, 904 insertions(+), 115 deletions(-)
diff --git a/be/src/agent/be_exec_version_manager.cpp
b/be/src/agent/be_exec_version_manager.cpp
index 91ccb6a6a25..1ccdf978688 100644
--- a/be/src/agent/be_exec_version_manager.cpp
+++ b/be/src/agent/be_exec_version_manager.cpp
@@ -21,6 +21,26 @@
namespace doris {
+namespace {
+
+int get_function_version(const std::map<std::string, std::set<int>>&
function_versions,
+ int be_exec_version, const std::string&
function_name) {
+ auto it = function_versions.find(function_name);
+ if (it == function_versions.end()) {
+ // 0 means no version-specific behavior is registered.
+ return 0;
+ }
+
+ auto version_it = it->second.lower_bound(be_exec_version);
+ if (version_it == it->second.end()) {
+ return 0;
+ }
+
+ return *version_it;
+}
+
+} // namespace
+
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(
@@ -34,25 +54,22 @@ Status BeExecVersionManager::check_be_exec_version(int
be_exec_version) {
int BeExecVersionManager::get_function_compatibility(int be_exec_version,
std::string
function_name) {
- if (_function_restrict_map.contains(function_name) && be_exec_version !=
get_newest_version()) {
- throw Exception(Status::InternalError(
- "function {} do not support old be exec version, maybe it's
because doris are "
- "doing a rolling upgrade. newest_version={},
input_be_exec_version={}",
- function_name, get_newest_version(), be_exec_version));
- }
+ check_function_restriction(be_exec_version, function_name);
+ return get_function_version(_function_change_map, be_exec_version,
function_name);
+}
- auto it = _function_change_map.find(function_name);
- if (it == _function_change_map.end()) {
- // 0 means no compatibility issues need to be dealt with
- return 0;
- }
+int BeExecVersionManager::get_function_alternative(int be_exec_version,
std::string function_name) {
+ return get_function_version(_function_alternative_map, be_exec_version,
function_name);
+}
- auto version_it = it->second.lower_bound(be_exec_version);
- if (version_it == it->second.end()) {
- return 0;
+void BeExecVersionManager::check_function_restriction(int be_exec_version,
+ const std::string&
function_name) {
+ auto it = _function_restrict_map.find(function_name);
+ if (it != _function_restrict_map.end() && be_exec_version < it->second) {
+ throw Exception(Status::InternalError(
+ "function {} does not support be exec versions before {},
input_be_exec_version={}",
+ function_name, it->second, be_exec_version));
}
-
- return *version_it;
}
void BeExecVersionManager::check_function_compatibility(int
current_be_exec_version,
@@ -137,9 +154,12 @@ void
BeExecVersionManager::check_function_compatibility(int current_be_exec_vers
// a. distinguish Hive OpenCSVSerde row semantics from generic CSV decoding
during upgrades.
// 16: start from master
// a. support Paimon table writes.
+// 17: start from master
+// a. use exact logical frames for map_agg_v2 aggregate state serialization.
-const int BeExecVersionManager::max_be_exec_version =
SUPPORT_PAIMON_WRITE_VERSION;
+const int BeExecVersionManager::max_be_exec_version =
SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION;
const int BeExecVersionManager::min_be_exec_version = 0;
std::map<std::string, std::set<int>>
BeExecVersionManager::_function_change_map {};
-std::set<std::string> BeExecVersionManager::_function_restrict_map;
+std::map<std::string, std::set<int>>
BeExecVersionManager::_function_alternative_map {};
+std::map<std::string, int> BeExecVersionManager::_function_restrict_map;
} // namespace doris
diff --git a/be/src/agent/be_exec_version_manager.h
b/be/src/agent/be_exec_version_manager.h
index d7ec473127f..01e5f442bad 100644
--- a/be/src/agent/be_exec_version_manager.h
+++ b/be/src/agent/be_exec_version_manager.h
@@ -29,6 +29,8 @@
namespace doris {
+// Version 8 removed the old implementations of these restricted aggregate
functions.
+constexpr inline int SUPPORT_RESTRICTED_AGGREGATES_VERSION = 8;
constexpr inline int USE_NEW_FIXED_OBJECT_SERIALIZATION_VERSION = 10;
constexpr inline int SUPPORT_ICEBERG_MERGE_CARDINALITY_VERSION = 11;
constexpr inline int SUPPORT_ICEBERG_VARIANT_VERSION = 12;
@@ -36,6 +38,7 @@ constexpr inline int SUPPORT_EXTERNAL_TABLE_SINK_HASH_VERSION
= 13;
constexpr inline int SUPPORT_TIMESTAMP_NS_VERSION = 14;
constexpr inline int SUPPORT_HIVE_OPEN_CSV_VERSION = 15;
constexpr inline int SUPPORT_PAIMON_WRITE_VERSION = 16;
+constexpr inline int SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION = 17;
class BeExecVersionManager {
public:
@@ -45,6 +48,10 @@ public:
static int get_function_compatibility(int be_exec_version, std::string
function_name);
+ static int get_function_alternative(int be_exec_version, std::string
function_name);
+
+ static void check_function_restriction(int be_exec_version, const
std::string& function_name);
+
static void check_function_compatibility(int current_be_exec_version, int
data_be_exec_version,
std::string function_name);
@@ -58,10 +65,17 @@ public:
static void registe_old_function_compatibility(int breaking_old_version,
std::string function_name) {
_function_change_map[function_name].insert(breaking_old_version);
+ register_old_function_alternative(breaking_old_version, function_name);
+ }
+
+ static void register_old_function_alternative(int breaking_old_version,
+ std::string function_name) {
+ _function_alternative_map[function_name].insert(breaking_old_version);
}
- static void registe_restrict_function_compatibility(std::string
function_name) {
- _function_restrict_map.insert(function_name);
+ static void registe_restrict_function_compatibility(std::string
function_name,
+ int
first_supported_version) {
+ _function_restrict_map.emplace(function_name, first_supported_version);
}
private:
@@ -69,8 +83,10 @@ private:
static const int min_be_exec_version;
// [function name] -> [breaking change start version]
static std::map<std::string, std::set<int>> _function_change_map;
- // those function must has input newest be exec version
- static std::set<std::string> _function_restrict_map;
+ // [function name] -> [alternative implementation start version]
+ static std::map<std::string, std::set<int>> _function_alternative_map;
+ // [function name] -> [first BE exec version with the retained
implementation]
+ static std::map<std::string, int> _function_restrict_map;
};
} // namespace doris
diff --git a/be/src/exprs/aggregate/aggregate_function.h
b/be/src/exprs/aggregate/aggregate_function.h
index a6c12dcbab5..0b80839cdeb 100644
--- a/be/src/exprs/aggregate/aggregate_function.h
+++ b/be/src/exprs/aggregate/aggregate_function.h
@@ -53,6 +53,8 @@ struct AggregateFunctionAttr {
bool enable_aggregate_function_null_v2 {false};
bool new_version_percentile {false};
std::vector<std::string> column_names;
+ // Negotiated execution version propagated to nested aggregate function
creators.
+ int be_exec_version {-1};
};
template <bool nullable, typename ColVecType>
diff --git a/be/src/exprs/aggregate/aggregate_function_covar.cpp
b/be/src/exprs/aggregate/aggregate_function_covar.cpp
index d1db8b40416..f96b57f45f8 100644
--- a/be/src/exprs/aggregate/aggregate_function_covar.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_covar.cpp
@@ -56,7 +56,8 @@ void
register_aggregate_function_covar_pop(AggregateFunctionSimpleFactory& facto
}
void
register_aggregate_function_covar_samp_old(AggregateFunctionSimpleFactory&
factory) {
-
BeExecVersionManager::registe_restrict_function_compatibility("covar_samp");
+ BeExecVersionManager::registe_restrict_function_compatibility(
+ "covar_samp", SUPPORT_RESTRICTED_AGGREGATES_VERSION);
}
void register_aggregate_function_covar_samp(AggregateFunctionSimpleFactory&
factory) {
diff --git a/be/src/exprs/aggregate/aggregate_function_distinct.cpp
b/be/src/exprs/aggregate/aggregate_function_distinct.cpp
index ec9b087cd24..b8c934c33d6 100644
--- a/be/src/exprs/aggregate/aggregate_function_distinct.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_distinct.cpp
@@ -93,8 +93,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, result_type,
- false,
BeExecVersionManager::get_newest_version(), attr);
+ auto nested_function = factory.get_nested(nested_function_name,
transform_arguments,
+ result_type, false,
attr.be_exec_version, attr);
return
function_combinator->transform_aggregate_function(nested_function, types,
result_is_nullable, attr);
};
diff --git a/be/src/exprs/aggregate/aggregate_function_foreach.cpp
b/be/src/exprs/aggregate/aggregate_function_foreach.cpp
index 16ee0ea6782..14605f23821 100644
--- a/be/src/exprs/aggregate/aggregate_function_foreach.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_foreach.cpp
@@ -47,8 +47,8 @@ void
register_aggregate_function_combinator_foreach(AggregateFunctionSimpleFacto
->get_nested_type();
auto nested_function_name = name.substr(0, name.size() -
suffix.size());
auto nested_function =
- factory.get(nested_function_name, transform_arguments,
result_item_type,
- result_is_nullable,
BeExecVersionManager::get_newest_version(), attr);
+ factory.get_nested(nested_function_name, transform_arguments,
result_item_type,
+ result_is_nullable, attr.be_exec_version,
attr);
if (!nested_function) {
throw Exception(
ErrorCode::INTERNAL_ERROR,
diff --git a/be/src/exprs/aggregate/aggregate_function_foreachv2.cpp
b/be/src/exprs/aggregate/aggregate_function_foreachv2.cpp
index 695d583b7e5..f31c4dd58c1 100644
--- a/be/src/exprs/aggregate/aggregate_function_foreachv2.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_foreachv2.cpp
@@ -94,8 +94,8 @@ void
register_aggregate_function_combinator_foreachv2(AggregateFunctionSimpleFac
->get_nested_type();
auto nested_function_name = name.substr(0, name.size() -
suffix.size());
auto nested_function =
- factory.get(nested_function_name, transform_arguments,
result_item_type, true,
- BeExecVersionManager::get_newest_version(), attr);
+ factory.get_nested(nested_function_name, transform_arguments,
result_item_type,
+ true, attr.be_exec_version, attr);
if (!nested_function) {
throw Exception(
ErrorCode::INTERNAL_ERROR,
diff --git a/be/src/exprs/aggregate/aggregate_function_map_combinator.cpp
b/be/src/exprs/aggregate/aggregate_function_map_combinator.cpp
index 2d4e40aad72..ac0e46cae62 100644
--- a/be/src/exprs/aggregate/aggregate_function_map_combinator.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_map_combinator.cpp
@@ -20,7 +20,6 @@
#include <string>
#include <string_view>
-#include "agent/be_exec_version_manager.h"
#include "core/call_on_type_index.h"
#include "core/column/column_const.h"
#include "core/column/column_decimal.h"
@@ -103,9 +102,9 @@ AggregateFunctionPtr create_nested_function(const
std::string& name, const DataT
DataTypePtr nested_result_type = remove_nullable(result_value_type);
const bool nested_result_is_nullable = result_value_type->is_nullable();
- auto nested_function = AggregateFunctionSimpleFactory::instance().get(
+ auto nested_function =
AggregateFunctionSimpleFactory::instance().get_nested(
nested_function_name(name), nested_argument_types,
nested_result_type,
- nested_result_is_nullable,
BeExecVersionManager::get_newest_version(), attr);
+ nested_result_is_nullable, attr.be_exec_version, attr);
if (nested_function == nullptr) {
throw Exception(ErrorCode::INTERNAL_ERROR,
"Can not create nested aggregate function for {}",
name);
diff --git a/be/src/exprs/aggregate/aggregate_function_map_v2.cpp
b/be/src/exprs/aggregate/aggregate_function_map_v2.cpp
index b90fe60e096..9536f405344 100644
--- a/be/src/exprs/aggregate/aggregate_function_map_v2.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_map_v2.cpp
@@ -17,18 +17,21 @@
#include "exprs/aggregate/aggregate_function_map_v2.h"
+#include "agent/be_exec_version_manager.h"
#include "exprs/aggregate/helpers.h"
namespace doris {
+template <bool use_exact_key_frame>
AggregateFunctionPtr create_agg_function_map_agg_v2(const DataTypes&
argument_types,
const bool
result_is_nullable,
const
AggregateFunctionAttr& attr) {
return creator_without_type::create_ignore_nullable<
-
AggregateFunctionMapAggV2<AggregateFunctionMapAggDataV2>>(argument_types,
-
result_is_nullable, attr);
+
AggregateFunctionMapAggV2<AggregateFunctionMapAggDataV2<use_exact_key_frame>>>(
+ argument_types, result_is_nullable, attr);
}
+template <bool use_exact_key_frame>
AggregateFunctionPtr create_aggregate_function_map_agg_v2(const std::string&
name,
const DataTypes&
argument_types,
const DataTypePtr&
result_type,
@@ -57,16 +60,30 @@ AggregateFunctionPtr
create_aggregate_function_map_agg_v2(const std::string& nam
case PrimitiveType::TYPE_TIMEV2:
case PrimitiveType::TYPE_TIMESTAMPTZ:
case PrimitiveType::TYPE_UUID:
- return create_agg_function_map_agg_v2(argument_types,
result_is_nullable, attr);
+ return
create_agg_function_map_agg_v2<use_exact_key_frame>(argument_types,
+
result_is_nullable, attr);
+ case PrimitiveType::TYPE_IPV4:
+ case PrimitiveType::TYPE_IPV6:
+ if constexpr (use_exact_key_frame) {
+ return
create_agg_function_map_agg_v2<use_exact_key_frame>(argument_types,
+
result_is_nullable, attr);
+ }
+ break;
default:
- LOG(WARNING) << fmt::format("unsupported input type {} for aggregate
function {}",
- argument_types[0]->get_name(), name);
- return nullptr;
+ break;
}
+ LOG(WARNING) << fmt::format("unsupported input type {} for aggregate
function {}",
+ argument_types[0]->get_name(), name);
+ return nullptr;
}
void register_aggregate_function_map_agg_v2(AggregateFunctionSimpleFactory&
factory) {
- factory.register_function_both("map_agg_v2",
create_aggregate_function_map_agg_v2);
+ factory.register_function_both("map_agg_v2",
create_aggregate_function_map_agg_v2<true>);
+ constexpr auto old_be_exec_version =
SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION - 1;
+ factory.register_transient_alternative_function(
+ "map_agg_v2", create_aggregate_function_map_agg_v2<false>, false,
old_be_exec_version);
+ factory.register_transient_alternative_function(
+ "map_agg_v2", create_aggregate_function_map_agg_v2<false>, true,
old_be_exec_version);
}
} // namespace doris
diff --git a/be/src/exprs/aggregate/aggregate_function_map_v2.h
b/be/src/exprs/aggregate/aggregate_function_map_v2.h
index 5e62dcdfb5c..e70c1ce68a5 100644
--- a/be/src/exprs/aggregate/aggregate_function_map_v2.h
+++ b/be/src/exprs/aggregate/aggregate_function_map_v2.h
@@ -20,17 +20,22 @@
#include <glog/logging.h>
#include <parallel_hashmap/phmap.h>
+#include <string>
+
+#include "common/check.h"
#include "core/assert_cast.h"
#include "core/column/column_decimal.h"
#include "core/column/column_map.h"
#include "core/column/column_string.h"
#include "core/data_type/data_type_map.h"
#include "core/field.h"
+#include "core/pod_array.h"
#include "exprs/aggregate/aggregate_function.h"
#include "exprs/aggregate/aggregate_function_simple_factory.h"
namespace doris {
+template <bool use_exact_key_frame>
struct AggregateFunctionMapAggDataV2 {
using Map = doris::flat_hash_map<Field, int64_t>;
@@ -119,50 +124,52 @@ struct AggregateFunctionMapAggDataV2 {
}
void write(BufferWritable& buf) const {
- auto serialized_bytes =
- _key_type->get_uncompressed_serialized_bytes(*_key_column,
_be_version);
-
std::string serialized_buffer;
- serialized_buffer.resize(serialized_bytes);
-
- auto* buf_ptr = _key_type->serialize(*_key_column,
serialized_buffer.data(), _be_version);
- int64_t written_bytes = buf_ptr - serialized_buffer.data();
- DCHECK_LE(written_bytes, serialized_bytes);
-
- serialized_buffer.resize(serialized_bytes);
- buf.write_binary(serialized_buffer);
-
- serialized_bytes =
- _value_type->get_uncompressed_serialized_bytes(*_value_column,
_be_version);
-
- serialized_buffer.resize(serialized_bytes);
-
- buf_ptr = _value_type->serialize(*_value_column,
serialized_buffer.data(), _be_version);
- written_bytes = buf_ptr - serialized_buffer.data();
- DCHECK_LE(written_bytes, serialized_bytes);
-
- serialized_buffer.resize(written_bytes);
- buf.write_binary(serialized_buffer);
+ write_column(*_key_type, *_key_column, buf, serialized_buffer,
use_exact_key_frame);
+ write_column(*_value_type, *_value_column, buf, serialized_buffer,
true);
}
void read(BufferReadable& buf) {
- std::string deserialized_buffer;
+ PaddedPODArray<UInt8> frame;
+ read_column(*_key_type, &_key_column, buf, frame, use_exact_key_frame);
+ read_column(*_value_type, &_value_column, buf, frame, true);
+ }
- buf.read_binary(deserialized_buffer);
+private:
+ void write_column(const IDataType& type, const IColumn& column,
BufferWritable& buf,
+ std::string& serialized_buffer, bool use_exact_frame)
const {
+ const auto max_serialized_bytes =
+ type.get_uncompressed_serialized_bytes(column, _be_version);
+ serialized_buffer.resize(max_serialized_bytes, '\0');
+
+ const auto* end = type.serialize(column, serialized_buffer.data(),
_be_version);
+ const auto written_bytes = end - serialized_buffer.data();
+ DORIS_CHECK_LE(written_bytes, max_serialized_bytes);
+
+ if (use_exact_frame) {
+ serialized_buffer.resize(written_bytes);
+ }
+ buf.write_binary(serialized_buffer);
+ }
- const auto* ptr =
- _key_type->deserialize(deserialized_buffer.data(),
&_key_column, _be_version);
- auto read_bytes = ptr - deserialized_buffer.data();
- DCHECK_EQ(read_bytes, deserialized_buffer.size());
+ void read_column(const IDataType& type, MutableColumnPtr* column,
BufferReadable& buf,
+ PaddedPODArray<UInt8>& frame, bool use_exact_frame) const
{
+ buf.read_binary(frame);
- buf.read_binary(deserialized_buffer);
+ const auto* begin = reinterpret_cast<const char*>(frame.data());
+ const auto* frame_end = begin + frame.size();
+ const auto* end = type.deserialize(begin, column, _be_version);
+ if (use_exact_frame) {
+ DORIS_CHECK_EQ(end, frame_end);
+ return;
+ }
- ptr = _value_type->deserialize(deserialized_buffer.data(),
&_value_column, _be_version);
- read_bytes = ptr - deserialized_buffer.data();
- DCHECK_EQ(read_bytes, deserialized_buffer.size());
+ DORIS_CHECK_LE(end, frame_end);
+ const auto expected_frame_size =
+ type.get_uncompressed_serialized_bytes(**column, _be_version);
+ DORIS_CHECK_EQ(frame.size(), expected_frame_size);
}
-private:
Map _map;
Arena _arena;
IColumn::MutablePtr _key_column;
diff --git a/be/src/exprs/aggregate/aggregate_function_percentile.cpp
b/be/src/exprs/aggregate/aggregate_function_percentile.cpp
index eb829f1e318..80d7633d4cb 100644
--- a/be/src/exprs/aggregate/aggregate_function_percentile.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_percentile.cpp
@@ -91,13 +91,17 @@ void
register_aggregate_function_percentile(AggregateFunctionSimpleFactory& fact
}
void register_percentile_approx_old_function(AggregateFunctionSimpleFactory&
factory) {
-
BeExecVersionManager::registe_restrict_function_compatibility("percentile_approx");
-
BeExecVersionManager::registe_restrict_function_compatibility("percentile_approx_weighted");
+ BeExecVersionManager::registe_restrict_function_compatibility(
+ "percentile_approx", SUPPORT_RESTRICTED_AGGREGATES_VERSION);
+ BeExecVersionManager::registe_restrict_function_compatibility(
+ "percentile_approx_weighted",
SUPPORT_RESTRICTED_AGGREGATES_VERSION);
}
void
register_aggregate_function_percentile_old(AggregateFunctionSimpleFactory&
factory) {
-
BeExecVersionManager::registe_restrict_function_compatibility("percentile");
-
BeExecVersionManager::registe_restrict_function_compatibility("percentile_array");
+ BeExecVersionManager::registe_restrict_function_compatibility(
+ "percentile", SUPPORT_RESTRICTED_AGGREGATES_VERSION);
+ BeExecVersionManager::registe_restrict_function_compatibility(
+ "percentile_array", SUPPORT_RESTRICTED_AGGREGATES_VERSION);
}
void
register_aggregate_function_percentile_approx(AggregateFunctionSimpleFactory&
factory) {
diff --git a/be/src/exprs/aggregate/aggregate_function_simple_factory.h
b/be/src/exprs/aggregate/aggregate_function_simple_factory.h
index b6d8085ae92..b93b3d1d01a 100644
--- a/be/src/exprs/aggregate/aggregate_function_simple_factory.h
+++ b/be/src/exprs/aggregate/aggregate_function_simple_factory.h
@@ -122,6 +122,61 @@ public:
AggregateFunctionPtr get(const std::string& name, const DataTypes&
argument_types,
const DataTypePtr& result_type, const bool
result_is_nullable,
int be_version, AggregateFunctionAttr attr = {}) {
+ return get_impl(name, argument_types, result_type, result_is_nullable,
be_version,
+ std::move(attr), true);
+ }
+
+ AggregateFunctionPtr get_nested(const std::string& name, const DataTypes&
argument_types,
+ const DataTypePtr& result_type, const bool
result_is_nullable,
+ int be_version, AggregateFunctionAttr attr
= {}) {
+ // The wrapper has already passed top-level compatibility checks.
Nested lookup still
+ // needs versioned alternative routing, but must not reapply
direct-function restrictions.
+ return get_impl(name, argument_types, result_type, result_is_nullable,
be_version,
+ std::move(attr), false);
+ }
+
+ void register_function(const std::string& name, const Creator& creator,
bool nullable = false) {
+ if (nullable) {
+ nullable_aggregate_functions[name] = creator;
+ } else {
+ aggregate_functions[name] = creator;
+ }
+ }
+
+ void register_function_both(const std::string& name, const Creator&
creator) {
+ register_function(name, creator, false);
+ register_function(name, creator, true);
+ }
+
+ void register_alias(const std::string& name, const std::string& alias) {
+ function_alias[alias] = name;
+ for (const auto& s : combiner_names) {
+ function_alias[alias + std::string(s)] = name + std::string(s);
+ }
+ function_alias[DISTINCT_FUNCTION_PREFIX + alias] =
DISTINCT_FUNCTION_PREFIX + name;
+ }
+
+ void register_alternative_function(const std::string& name, const Creator&
creator,
+ bool nullable, int old_be_exec_version)
{
+ auto new_name = name +
BeExecVersionManager::get_function_suffix(old_be_exec_version);
+ register_function(new_name, creator, nullable);
+
BeExecVersionManager::registe_old_function_compatibility(old_be_exec_version,
name);
+ }
+
+ void register_transient_alternative_function(const std::string& name,
const Creator& creator,
+ bool nullable, int
old_be_exec_version) {
+ auto new_name = name +
BeExecVersionManager::get_function_suffix(old_be_exec_version);
+ register_function(new_name, creator, nullable);
+
BeExecVersionManager::register_old_function_alternative(old_be_exec_version,
name);
+ }
+
+ static AggregateFunctionSimpleFactory& instance();
+
+private:
+ AggregateFunctionPtr get_impl(const std::string& name, const DataTypes&
argument_types,
+ const DataTypePtr& result_type, const bool
result_is_nullable,
+ int be_version, AggregateFunctionAttr attr,
+ bool check_restriction) {
bool nullable = false;
for (const auto& type : argument_types) {
if (type->is_nullable()) {
@@ -130,7 +185,11 @@ public:
}
std::string name_str = name;
+ if (check_restriction) {
+ BeExecVersionManager::check_function_restriction(be_version,
name_str);
+ }
temporary_function_update(be_version, name_str);
+ attr.be_exec_version = be_version;
if (function_alias.contains(name)) {
name_str = function_alias[name];
@@ -158,42 +217,12 @@ public:
}
}
- void register_function(const std::string& name, const Creator& creator,
bool nullable = false) {
- if (nullable) {
- nullable_aggregate_functions[name] = creator;
- } else {
- aggregate_functions[name] = creator;
- }
- }
-
- void register_function_both(const std::string& name, const Creator&
creator) {
- register_function(name, creator, false);
- register_function(name, creator, true);
- }
-
- void register_alias(const std::string& name, const std::string& alias) {
- function_alias[alias] = name;
- for (const auto& s : combiner_names) {
- function_alias[alias + std::string(s)] = name + std::string(s);
- }
- function_alias[DISTINCT_FUNCTION_PREFIX + alias] =
DISTINCT_FUNCTION_PREFIX + name;
- }
-
- void register_alternative_function(const std::string& name, const Creator&
creator,
- bool nullable, int old_be_exec_version)
{
- auto new_name = name +
BeExecVersionManager::get_function_suffix(old_be_exec_version);
- register_function(new_name, creator, nullable);
-
BeExecVersionManager::registe_old_function_compatibility(old_be_exec_version,
name);
- }
-
void temporary_function_update(int fe_version_now, std::string& name) {
- int old_version =
BeExecVersionManager::get_function_compatibility(fe_version_now, name);
+ int old_version =
BeExecVersionManager::get_function_alternative(fe_version_now, name);
if (!old_version) {
return;
}
name = name + BeExecVersionManager::get_function_suffix(old_version);
}
-
- static AggregateFunctionSimpleFactory& instance();
};
}; // namespace doris
diff --git a/be/src/exprs/aggregate/aggregate_function_stddev.cpp
b/be/src/exprs/aggregate/aggregate_function_stddev.cpp
index f8c910f4771..85271a0a7c1 100644
--- a/be/src/exprs/aggregate/aggregate_function_stddev.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_stddev.cpp
@@ -87,8 +87,10 @@ void
register_aggregate_function_stddev_variance_pop(AggregateFunctionSimpleFact
}
void
register_aggregate_function_stddev_variance_samp_old(AggregateFunctionSimpleFactory&
factory) {
-
BeExecVersionManager::registe_restrict_function_compatibility("variance_samp");
-
BeExecVersionManager::registe_restrict_function_compatibility("stddev_samp");
+ BeExecVersionManager::registe_restrict_function_compatibility(
+ "variance_samp", SUPPORT_RESTRICTED_AGGREGATES_VERSION);
+ BeExecVersionManager::registe_restrict_function_compatibility(
+ "stddev_samp", SUPPORT_RESTRICTED_AGGREGATES_VERSION);
}
void
register_aggregate_function_stddev_variance_samp(AggregateFunctionSimpleFactory&
factory) {
diff --git a/be/src/exprs/aggregate/aggregate_function_window_funnel.cpp
b/be/src/exprs/aggregate/aggregate_function_window_funnel.cpp
index 060afb7379c..07e35174818 100644
--- a/be/src/exprs/aggregate/aggregate_function_window_funnel.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_window_funnel.cpp
@@ -56,6 +56,7 @@ void
register_aggregate_function_window_funnel(AggregateFunctionSimpleFactory& f
factory.register_alias("window_funnel_v1", "window_funnel");
}
void
register_aggregate_function_window_funnel_old(AggregateFunctionSimpleFactory&
factory) {
-
BeExecVersionManager::registe_restrict_function_compatibility("window_funnel");
+ BeExecVersionManager::registe_restrict_function_compatibility(
+ "window_funnel", SUPPORT_RESTRICTED_AGGREGATES_VERSION);
}
} // namespace doris
diff --git a/be/test/exec/sink/writer/paimon/paimon_write_backend_test.cpp
b/be/test/exec/sink/writer/paimon/paimon_write_backend_test.cpp
index d153ec81a13..d543c18282e 100644
--- a/be/test/exec/sink/writer/paimon/paimon_write_backend_test.cpp
+++ b/be/test/exec/sink/writer/paimon/paimon_write_backend_test.cpp
@@ -63,10 +63,11 @@ TEST(PaimonWriteBackendTest, ReportsConcreteBackendType) {
}
TEST(PaimonWriteBackendTest, AdvertisesPaimonWriteExecVersion) {
- EXPECT_EQ(SUPPORT_PAIMON_WRITE_VERSION,
BeExecVersionManager::get_newest_version());
+ EXPECT_LE(SUPPORT_PAIMON_WRITE_VERSION,
BeExecVersionManager::get_newest_version());
EXPECT_TRUE(BeExecVersionManager::check_be_exec_version(SUPPORT_PAIMON_WRITE_VERSION).ok());
- EXPECT_FALSE(
-
BeExecVersionManager::check_be_exec_version(SUPPORT_PAIMON_WRITE_VERSION +
1).ok());
+ EXPECT_FALSE(BeExecVersionManager::check_be_exec_version(
+ BeExecVersionManager::get_newest_version() + 1)
+ .ok());
}
TEST(PaimonTableSinkOperatorTest, InitializesAsBlockingSink) {
diff --git a/be/test/exprs/aggregate/agg_map_agg_v2_test.cpp
b/be/test/exprs/aggregate/agg_map_agg_v2_test.cpp
new file mode 100644
index 00000000000..c572b81a328
--- /dev/null
+++ b/be/test/exprs/aggregate/agg_map_agg_v2_test.cpp
@@ -0,0 +1,415 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "agent/be_exec_version_manager.h"
+#include "common/exception.h"
+#include "core/column/column_array.h"
+#include "core/column/column_map.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_string.h"
+#include "core/data_type/data_type_agg_state.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_ipv4.h"
+#include "core/data_type/data_type_ipv6.h"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "core/field.h"
+#include "core/pod_array.h"
+#include "exprs/aggregate/aggregate_function_map_v2.h"
+#include "exprs/aggregate/aggregate_function_simple_factory.h"
+#include "exprs/aggregate/aggregate_function_state_merge.h"
+#include "exprs/aggregate/aggregate_function_state_union.h"
+
+namespace doris {
+namespace {
+
+void expect_restricted_aggregate_allowed(const char* name, int
be_exec_version) {
+
EXPECT_NO_THROW(BeExecVersionManager::check_function_restriction(be_exec_version,
name));
+}
+
+void expect_restricted_aggregate_rejected(const char* name, int
be_exec_version) {
+
EXPECT_THROW(BeExecVersionManager::check_function_restriction(be_exec_version,
name),
+ Exception);
+}
+
+void check_state_union_and_merge_round_trip(const DataTypePtr& key_type,
+ const std::vector<Field>& keys) {
+ const auto be_exec_version = BeExecVersionManager::get_newest_version();
+ const DataTypes argument_types {make_nullable(key_type),
+
make_nullable(std::make_shared<DataTypeInt32>())};
+ auto state_type = std::make_shared<DataTypeAggState>(argument_types,
false, "map_agg_v2",
+ be_exec_version);
+ auto nested_function = state_type->get_nested_function();
+ ASSERT_NE(nested_function, nullptr);
+
+ auto key_column = argument_types[0]->create_column();
+ auto value_column = argument_types[1]->create_column();
+ for (size_t i = 0; i < keys.size(); ++i) {
+ key_column->insert(keys[i]);
+
value_column->insert(Field::create_field<TYPE_INT>(static_cast<Int32>(i)));
+ }
+ const IColumn* input_columns[] = {key_column.get(), value_column.get()};
+
+ Arena arena;
+ auto state_column = state_type->create_column();
+ nested_function->streaming_agg_serialize_to_column(input_columns,
state_column, keys.size(),
+ arena);
+ ASSERT_EQ(state_column->size(), keys.size());
+
+ auto union_function =
+ AggregateStateUnion::create(nested_function, DataTypes
{state_type}, state_type);
+ union_function->set_version(be_exec_version);
+ AggregateFunctionGuard union_state(union_function.get());
+ const IColumn* state_columns[] = {state_column.get()};
+ union_function->add_batch_single_place(state_column->size(),
union_state.data(), state_columns,
+ arena);
+
+ auto serialized_union = union_function->create_serialize_column();
+ std::vector<AggregateDataPtr> union_places {union_state.data()};
+ union_function->serialize_to_column(union_places, 0, serialized_union, 1);
+ ASSERT_EQ(serialized_union->size(), 1);
+
+ AggregateFunctionGuard round_tripped_union(union_function.get());
+
union_function->deserialize_and_merge_from_column_range(round_tripped_union.data(),
+ *serialized_union,
0, 0, arena);
+ auto union_result = state_type->create_column();
+ union_function->insert_result_into(round_tripped_union.data(),
*union_result);
+ ASSERT_EQ(union_result->size(), 1);
+
+ auto merge_function = AggregateStateMerge::create(nested_function,
DataTypes {state_type},
+
nested_function->get_return_type());
+ merge_function->set_version(be_exec_version);
+ AggregateFunctionGuard merged_state(merge_function.get());
+ const IColumn* union_result_columns[] = {union_result.get()};
+ merge_function->add_batch_single_place(union_result->size(),
merged_state.data(),
+ union_result_columns, arena);
+
+ auto result = nested_function->get_return_type()->create_column();
+ merge_function->insert_result_into(merged_state.data(), *result);
+ const auto& result_map = assert_cast<const ColumnMap&>(*result);
+ ASSERT_EQ(result_map.size(), 1);
+ EXPECT_EQ(result_map.get_keys().size(), keys.size());
+}
+
+template <bool use_exact_key_frame>
+std::pair<size_t, size_t> serialize_and_deserialize_state(const DataTypePtr&
key_type,
+ const
std::vector<Field>& keys,
+ int be_exec_version)
{
+ const DataTypes argument_types {make_nullable(key_type),
+
make_nullable(std::make_shared<DataTypeInt32>())};
+ AggregateFunctionMapAggDataV2<use_exact_key_frame> state(argument_types,
be_exec_version);
+ for (size_t i = 0; i < keys.size(); ++i) {
+ state.add_single(keys[i],
Field::create_field<TYPE_INT>(static_cast<Int32>(i)));
+ }
+
+ auto serialized_state = ColumnString::create();
+ BufferWritable writer(*serialized_state);
+ state.write(writer);
+ writer.commit();
+
+ PaddedPODArray<UInt8> key_frame;
+ PaddedPODArray<UInt8> value_frame;
+ BufferReadable frame_reader(serialized_state->get_data_at(0));
+ frame_reader.read_binary(key_frame);
+ frame_reader.read_binary(value_frame);
+
+ AggregateFunctionMapAggDataV2<use_exact_key_frame>
restored_state(argument_types,
+
be_exec_version);
+ BufferReadable state_reader(serialized_state->get_data_at(0));
+ restored_state.read(state_reader);
+
+ auto result_type = std::make_shared<DataTypeMap>(argument_types[0],
argument_types[1]);
+ auto result = result_type->create_column();
+ restored_state.insert_result_into(*result);
+ const auto& result_map = assert_cast<const ColumnMap&>(*result);
+ EXPECT_EQ(result_map.get_keys().size(), keys.size());
+
+ return {key_frame.size(), value_frame.size()};
+}
+
+template <bool use_exact_key_frame>
+void check_unequal_frame_round_trip(const DataTypePtr& key_type, const Field&
key,
+ const DataTypePtr& value_type, const
Field& value,
+ int be_exec_version) {
+ const DataTypes argument_types {make_nullable(key_type),
make_nullable(value_type)};
+ AggregateFunctionMapAggDataV2<use_exact_key_frame> state(argument_types,
be_exec_version);
+ state.add_single(key, value);
+
+ auto serialized_state = ColumnString::create();
+ BufferWritable writer(*serialized_state);
+ state.write(writer);
+ writer.commit();
+
+ AggregateFunctionMapAggDataV2<use_exact_key_frame>
restored_state(argument_types,
+
be_exec_version);
+ BufferReadable reader(serialized_state->get_data_at(0));
+ restored_state.read(reader);
+ EXPECT_FALSE(reader.has_remaining());
+
+ auto result_type = std::make_shared<DataTypeMap>(argument_types[0],
argument_types[1]);
+ auto result = result_type->create_column();
+ restored_state.insert_result_into(*result);
+ const auto& result_map = assert_cast<const ColumnMap&>(*result);
+ ASSERT_EQ(result_map.get_keys().size(), 1);
+ EXPECT_EQ(result_map.get_keys()[0], key);
+ EXPECT_EQ(result_map.get_values()[0], value);
+}
+
+void check_foreach_v2_state(const AggregateFunctionPtr& function, const
DataTypePtr& result_type,
+ const ColumnString& serialized, Arena& arena,
size_t key_count) {
+ AggregateFunctionGuard received_state(function.get());
+ BufferReadable reader(serialized.get_data_at(0));
+ function->deserialize(received_state.data(), reader, arena);
+ EXPECT_FALSE(reader.has_remaining());
+
+ auto result = result_type->create_column();
+ function->insert_result_into(received_state.data(), *result);
+ const auto& result_array = assert_cast<const ColumnArray&>(*result);
+ ASSERT_EQ(result_array.size(), 1);
+ ASSERT_EQ(result_array.get_offsets()[0], 1);
+ const auto& nested_map = assert_cast<const ColumnMap&>(
+ assert_cast<const
ColumnNullable&>(result_array.get_data()).get_nested_column());
+ EXPECT_EQ(nested_map.get_keys().size(), key_count);
+}
+
+void serialize_foreach_v2_key_frame(int be_exec_version, size_t&
key_frame_size) {
+ const auto key_type = make_nullable(std::make_shared<DataTypeInt32>());
+ const auto value_type = make_nullable(std::make_shared<DataTypeInt32>());
+ const DataTypes argument_types {std::make_shared<DataTypeArray>(key_type),
+
std::make_shared<DataTypeArray>(value_type)};
+ const auto map_type = std::make_shared<DataTypeMap>(key_type, value_type);
+ const auto result_type =
std::make_shared<DataTypeArray>(make_nullable(map_type));
+ auto function = AggregateFunctionSimpleFactory::instance().get(
+ "map_agg_v2_foreachv2", argument_types, result_type, false,
be_exec_version);
+ ASSERT_NE(function, nullptr);
+ function->set_version(be_exec_version);
+
+ constexpr auto key_count = SERIALIZED_MEM_SIZE_LIMIT / sizeof(Int32) + 1;
+ auto key_column = argument_types[0]->create_column();
+ auto value_column = argument_types[1]->create_column();
+ for (size_t i = 0; i < key_count; ++i) {
+ key_column->insert(Field::create_field<TYPE_ARRAY>(
+ Array {Field::create_field<TYPE_INT>(static_cast<Int32>(i +
1))}));
+ value_column->insert(Field::create_field<TYPE_ARRAY>(
+ Array {Field::create_field<TYPE_INT>(static_cast<Int32>(i))}));
+ }
+ const IColumn* input_columns[] = {key_column.get(), value_column.get()};
+
+ Arena arena;
+ AggregateFunctionGuard state(function.get());
+ for (size_t i = 0; i < key_count; ++i) {
+ function->add(state.data(), input_columns, static_cast<ssize_t>(i),
arena);
+ }
+
+ auto serialized_state = ColumnString::create();
+ BufferWritable writer(*serialized_state);
+ function->serialize(state.data(), writer);
+ writer.commit();
+
+ BufferReadable reader(serialized_state->get_data_at(0));
+ size_t dynamic_array_size = 0;
+ reader.read_binary(dynamic_array_size);
+ EXPECT_EQ(dynamic_array_size, 1);
+ PaddedPODArray<UInt8> key_frame;
+ PaddedPODArray<UInt8> value_frame;
+ reader.read_binary(key_frame);
+ reader.read_binary(value_frame);
+ EXPECT_FALSE(reader.has_remaining());
+ key_frame_size = key_frame.size();
+
+ // A newly upgraded BE must read a partial state produced using the
negotiated
+ // legacy format, rather than constructing the nested function at version
17.
+ auto receiving_function = AggregateFunctionSimpleFactory::instance().get(
+ "map_agg_v2_foreachv2", argument_types, result_type, false,
be_exec_version);
+ ASSERT_NE(receiving_function, nullptr);
+ receiving_function->set_version(be_exec_version);
+ check_foreach_v2_state(receiving_function, result_type, *serialized_state,
arena, key_count);
+
+ if (be_exec_version == SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION - 1) {
+ // Build the outer foreach envelope around an old-format nested state
to
+ // simulate a version-16 peer sending to a newly upgraded BE.
+ AggregateFunctionMapAggDataV2<false> legacy_nested({key_type,
value_type}, be_exec_version);
+ for (size_t i = 0; i < key_count; ++i) {
+
legacy_nested.add_single(Field::create_field<TYPE_INT>(static_cast<Int32>(i +
1)),
+
Field::create_field<TYPE_INT>(static_cast<Int32>(i)));
+ }
+ auto legacy_foreach_state = ColumnString::create();
+ BufferWritable legacy_writer(*legacy_foreach_state);
+ legacy_writer.write_binary(size_t {1});
+ legacy_nested.write(legacy_writer);
+ legacy_writer.commit();
+ check_foreach_v2_state(receiving_function, result_type,
*legacy_foreach_state, arena,
+ key_count);
+ }
+}
+
+} // namespace
+
+TEST(AggregateFunctionMapAggV2Test, SupportsIpKeyTypes) {
+ const DataTypes key_types {std::make_shared<DataTypeIPv4>(),
std::make_shared<DataTypeIPv6>()};
+
+ for (const auto& key_type : key_types) {
+ const DataTypes argument_types {make_nullable(key_type),
+
make_nullable(std::make_shared<DataTypeInt32>())};
+ auto function = AggregateFunctionSimpleFactory::instance().get(
+ "map_agg_v2", argument_types, nullptr, false,
+ BeExecVersionManager::get_newest_version());
+
+ ASSERT_NE(function, nullptr) << key_type->get_name();
+
+ auto legacy_function = AggregateFunctionSimpleFactory::instance().get(
+ "map_agg_v2", argument_types, nullptr, false,
+ SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION - 1);
+ EXPECT_EQ(legacy_function, nullptr) << key_type->get_name();
+ }
+}
+
+TEST(AggregateFunctionMapAggV2Test, IpStateUnionAndMergeRoundTrip) {
+ constexpr auto ipv4_key_count = SERIALIZED_MEM_SIZE_LIMIT / sizeof(IPv4) +
1;
+ std::vector<Field> ipv4_keys;
+ ipv4_keys.reserve(ipv4_key_count);
+ for (size_t i = 0; i < ipv4_key_count; ++i) {
+ ipv4_keys.push_back(Field::create_field<TYPE_IPV4>(IPv4(i + 1)));
+ }
+ check_state_union_and_merge_round_trip(std::make_shared<DataTypeIPv4>(),
ipv4_keys);
+
+ constexpr auto ipv6_key_count = SERIALIZED_MEM_SIZE_LIMIT / sizeof(IPv6) +
1;
+ std::vector<Field> ipv6_keys;
+ ipv6_keys.reserve(ipv6_key_count);
+ for (size_t i = 0; i < ipv6_key_count; ++i) {
+ ipv6_keys.push_back(Field::create_field<TYPE_IPV6>(IPv6(i + 1)));
+ }
+ check_state_union_and_merge_round_trip(std::make_shared<DataTypeIPv6>(),
ipv6_keys);
+}
+
+TEST(AggregateFunctionMapAggV2Test, ExactAndLegacyStateFrames) {
+ constexpr auto key_count = SERIALIZED_MEM_SIZE_LIMIT / sizeof(Int32) + 1;
+ std::vector<Field> keys;
+ keys.reserve(key_count);
+ for (size_t i = 0; i < key_count; ++i) {
+ keys.push_back(Field::create_field<TYPE_INT>(static_cast<Int32>(i +
1)));
+ }
+
+ const auto exact_frame_sizes = serialize_and_deserialize_state<true>(
+ std::make_shared<DataTypeInt32>(), keys,
SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION);
+ const auto legacy_frame_sizes = serialize_and_deserialize_state<false>(
+ std::make_shared<DataTypeInt32>(), keys,
SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION - 1);
+
+ EXPECT_LT(exact_frame_sizes.first, legacy_frame_sizes.first);
+ EXPECT_EQ(exact_frame_sizes.second, legacy_frame_sizes.second);
+}
+
+TEST(AggregateFunctionMapAggV2Test, ReusedScratchHandlesUnequalFrameSizes) {
+ const auto int_type = std::make_shared<DataTypeInt32>();
+ const auto string_type = std::make_shared<DataTypeString>();
+ const auto int_field = Field::create_field<TYPE_INT>(1);
+ const auto string_field =
Field::create_field<TYPE_STRING>(std::string(1024, 'x'));
+
+ check_unequal_frame_round_trip<true>(string_type, string_field, int_type,
int_field,
+
SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION);
+ check_unequal_frame_round_trip<true>(int_type, int_field, string_type,
string_field,
+
SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION);
+ check_unequal_frame_round_trip<false>(string_type, string_field, int_type,
int_field,
+
SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION - 1);
+ check_unequal_frame_round_trip<false>(int_type, int_field, string_type,
string_field,
+
SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION - 1);
+}
+
+TEST(AggregateFunctionMapAggV2Test,
PersistedStateCompatibleAcrossFrameVersion) {
+ constexpr auto legacy_version = SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION - 1;
+ const DataTypes argument_types
{make_nullable(std::make_shared<DataTypeInt32>()),
+
make_nullable(std::make_shared<DataTypeInt32>())};
+ auto legacy_state_type =
+ std::make_shared<DataTypeAggState>(argument_types, false,
"map_agg_v2", legacy_version);
+ auto current_state_type = std::make_shared<DataTypeAggState>(
+ argument_types, false, "map_agg_v2",
SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION);
+
+ EXPECT_TRUE(legacy_state_type->get_serialized_type()->equals(
+ *current_state_type->get_serialized_type()));
+ EXPECT_NO_THROW(legacy_state_type->check_function_compatibility(
+ SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION));
+}
+
+TEST(AggregateFunctionMapAggV2Test, ForeachV2UsesNegotiatedFrameVersion) {
+ size_t exact_frame_size = 0;
+ size_t legacy_frame_size = 0;
+ serialize_foreach_v2_key_frame(SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION,
exact_frame_size);
+ serialize_foreach_v2_key_frame(SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION - 1,
legacy_frame_size);
+
+ EXPECT_LT(exact_frame_size, legacy_frame_size);
+}
+
+TEST(AggregateFunctionMapAggV2Test,
NestedFunctionsPreserveLegacyWrapperAvailability) {
+ constexpr auto legacy_version = SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION - 1;
+ const auto double_type = std::make_shared<DataTypeFloat64>();
+ const auto array_type = std::make_shared<DataTypeArray>(double_type);
+ const auto result_type =
+
std::make_shared<DataTypeArray>(make_nullable(std::make_shared<DataTypeFloat64>()));
+ const std::vector<std::pair<std::string, size_t>> test_cases {
+ {"stddev_samp_foreachv2", 1},
+ {"variance_samp_foreachv2", 1},
+ {"covar_samp_foreachv2", 2},
+ };
+
+ for (const auto& [name, argument_count] : test_cases) {
+ SCOPED_TRACE(name);
+ DataTypes argument_types(argument_count, array_type);
+ AggregateFunctionPtr function;
+ EXPECT_NO_THROW(function =
AggregateFunctionSimpleFactory::instance().get(
+ name, argument_types, result_type, false,
legacy_version));
+ EXPECT_NE(function, nullptr);
+ }
+
+ EXPECT_NE(AggregateFunctionSimpleFactory::instance().get("stddev_samp",
DataTypes {double_type},
+ double_type,
false, legacy_version),
+ nullptr);
+}
+
+TEST(AggregateFunctionMapAggV2Test,
RestrictedAggregatesUseTheirOwnCompatibilityVersion) {
+ constexpr auto legacy_version = SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION - 1;
+ const auto double_type = std::make_shared<DataTypeFloat64>();
+ auto& factory = AggregateFunctionSimpleFactory::instance();
+
+ for (const auto* name :
+ {"variance_samp", "stddev_samp", "covar_samp", "percentile_approx",
+ "percentile_approx_weighted", "percentile", "percentile_array",
"window_funnel"}) {
+ SCOPED_TRACE(name);
+ expect_restricted_aggregate_rejected(name,
SUPPORT_RESTRICTED_AGGREGATES_VERSION - 1);
+ expect_restricted_aggregate_allowed(name,
SUPPORT_RESTRICTED_AGGREGATES_VERSION);
+ expect_restricted_aggregate_allowed(name, legacy_version);
+ expect_restricted_aggregate_allowed(name,
SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION);
+ }
+
+ auto direct =
+ factory.get("stddev_samp", DataTypes {double_type}, double_type,
false, legacy_version);
+ ASSERT_NE(direct, nullptr);
+ auto stored_state = std::make_shared<DataTypeAggState>(DataTypes
{double_type}, false,
+ "stddev_samp",
legacy_version);
+ EXPECT_NO_THROW(
+
stored_state->check_function_compatibility(SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION));
+}
+
+} // namespace doris
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index cec742afde5..82064f1ab07 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -2019,9 +2019,11 @@ public class Config extends ConfigBase {
public static final int HIVE_OPEN_CSV_MIN_BE_EXEC_VERSION = 15;
// Older backends do not recognize PAIMON_TABLE_SINK and cannot execute
Paimon writes.
public static final int PAIMON_WRITE_MIN_BE_EXEC_VERSION = 16;
+ // Older backends serialize map_agg_v2 keys with the legacy padded frame
format.
+ public static final int MAP_AGG_V2_EXACT_FRAME_MIN_BE_EXEC_VERSION = 17;
@ConfField(mutable = false)
- public static int max_be_exec_version = PAIMON_WRITE_MIN_BE_EXEC_VERSION;
+ public static int max_be_exec_version =
MAP_AGG_V2_EXACT_FRAME_MIN_BE_EXEC_VERSION;
/**
* Min data version of backends serialize block.
diff --git
a/regression-test/data/datatype_p0/agg_state/map/test_map_agg_v2_versioned_state.out
b/regression-test/data/datatype_p0/agg_state/map/test_map_agg_v2_versioned_state.out
new file mode 100644
index 00000000000..4c103c707d6
--- /dev/null
+++
b/regression-test/data/datatype_p0/agg_state/map/test_map_agg_v2_versioned_state.out
@@ -0,0 +1,6 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !before_compaction --
+1 4
+
+-- !after_compaction --
+1 4
diff --git
a/regression-test/data/datatype_p0/agg_state/test_restricted_aggregate_versioned_state.out
b/regression-test/data/datatype_p0/agg_state/test_restricted_aggregate_versioned_state.out
new file mode 100644
index 00000000000..38d3036eaa9
--- /dev/null
+++
b/regression-test/data/datatype_p0/agg_state/test_restricted_aggregate_versioned_state.out
@@ -0,0 +1,15 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !direct_version_15 --
+1
+
+-- !stored_version_15_at_16 --
+1
+
+-- !direct_version_16 --
+1
+
+-- !stored_version_15_at_17 --
+1
+
+-- !direct_version_17 --
+1
diff --git a/regression-test/data/datatype_p0/ip/test_map_agg_v2_ip_key.out
b/regression-test/data/datatype_p0/ip/test_map_agg_v2_ip_key.out
new file mode 100644
index 00000000000..a917817b436
--- /dev/null
+++ b/regression-test/data/datatype_p0/ip/test_map_agg_v2_ip_key.out
@@ -0,0 +1,12 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !ipv4_key --
+["192.168.0.1", "192.168.0.2"]
+
+-- !ipv6_key --
+["2001:db8::1", "2001:db8::2"]
+
+-- !ipv4_state_merge --
+65
+
+-- !ipv6_state_merge --
+17
diff --git
a/regression-test/suites/datatype_p0/agg_state/map/test_map_agg_v2_versioned_state.groovy
b/regression-test/suites/datatype_p0/agg_state/map/test_map_agg_v2_versioned_state.groovy
new file mode 100644
index 00000000000..f6b43537eec
--- /dev/null
+++
b/regression-test/suites/datatype_p0/agg_state/map/test_map_agg_v2_versioned_state.groovy
@@ -0,0 +1,91 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_map_agg_v2_versioned_state", "nonConcurrent") {
+ if (isCloudMode()) {
+ return
+ }
+
+ def originalVersion = sql("ADMIN SHOW FRONTEND CONFIG LIKE
'be_exec_version'")[0][1]
+ try {
+ sql "ADMIN SET FRONTEND CONFIG ('be_exec_version' = '16')"
+ sql "SET enable_agg_state = true"
+ sql "DROP TABLE IF EXISTS test_map_agg_v2_versioned_state"
+ sql """
+ CREATE TABLE test_map_agg_v2_versioned_state (
+ k INT,
+ state AGG_STATE<map_agg_v2(INT, INT)> GENERIC
+ ) AGGREGATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "disable_auto_compaction" = "true"
+ )
+ """
+ for (int key = 1; key <= 4; key++) {
+ sql "INSERT INTO test_map_agg_v2_versioned_state VALUES (1,
map_agg_v2_state(${key}, ${key * 10}))"
+ }
+
+ // The column and its rowsets retain version 16 after the FE switches
to version 17.
+ sql "ADMIN SET FRONTEND CONFIG ('be_exec_version' = '17')"
+ order_qt_before_compaction """
+ SELECT k, map_size(map_agg_v2_merge(state))
+ FROM test_map_agg_v2_versioned_state GROUP BY k ORDER BY k
+ """
+
+ def tablets = sql_return_maparray "SHOW TABLETS FROM
test_map_agg_v2_versioned_state"
+ assertEquals(1, tablets.size())
+ def tabletId = tablets[0].TabletId
+ def backendId = tablets[0].BackendId
+ def backendIdToIp = [:]
+ def backendIdToHttpPort = [:]
+ getBackendIpHttpPort(backendIdToIp, backendIdToHttpPort)
+ def beHost = backendIdToIp["${backendId}"]
+ def bePort = backendIdToHttpPort["${backendId}"]
+ def showTabletCompaction = {
+ def (code, stdout, stderr) = be_show_tablet_status(beHost, bePort,
tabletId)
+ assertEquals(0, code)
+ return parseJson(stdout.trim())
+ }
+ def countDataRowsets = { status ->
+ return status.rowsets.findAll { it.contains(" DATA ") }.size()
+ }
+ def beforeCount = countDataRowsets(showTabletCompaction())
+ assertTrue(beforeCount >= 2)
+
+ sql "ADMIN COMPACT TABLE test_map_agg_v2_versioned_state PARTITION
(test_map_agg_v2_versioned_state) WHERE type = 'full'"
+ def deadline = System.currentTimeMillis() + 60000L
+ def afterCompaction
+ while (System.currentTimeMillis() < deadline) {
+ afterCompaction = showTabletCompaction()
+ if (afterCompaction["last full status"] == "[OK]" &&
+ countDataRowsets(afterCompaction) < beforeCount) {
+ break
+ }
+ sleep(500)
+ }
+ assertEquals("[OK]", afterCompaction["last full status"])
+ assertTrue(countDataRowsets(afterCompaction) < beforeCount)
+
+ order_qt_after_compaction """
+ SELECT k, map_size(map_agg_v2_merge(state))
+ FROM test_map_agg_v2_versioned_state GROUP BY k ORDER BY k
+ """
+ } finally {
+ sql "ADMIN SET FRONTEND CONFIG ('be_exec_version' =
'${originalVersion}')"
+ }
+}
diff --git
a/regression-test/suites/datatype_p0/agg_state/test_restricted_aggregate_versioned_state.groovy
b/regression-test/suites/datatype_p0/agg_state/test_restricted_aggregate_versioned_state.groovy
new file mode 100644
index 00000000000..edb9d3566e1
--- /dev/null
+++
b/regression-test/suites/datatype_p0/agg_state/test_restricted_aggregate_versioned_state.groovy
@@ -0,0 +1,72 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_restricted_aggregate_versioned_state", "nonConcurrent") {
+ if (isCloudMode()) {
+ return
+ }
+
+ def originalVersion = sql("ADMIN SHOW FRONTEND CONFIG LIKE
'be_exec_version'")[0][1]
+ try {
+ sql "ADMIN SET FRONTEND CONFIG ('be_exec_version' = '15')"
+ sql "SET enable_agg_state = true"
+ sql "DROP TABLE IF EXISTS test_restricted_aggregate_versioned_source"
+ sql """
+ CREATE TABLE test_restricted_aggregate_versioned_source (
+ id INT,
+ x DOUBLE
+ ) DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+ sql "INSERT INTO test_restricted_aggregate_versioned_source VALUES (1,
1.0), (2, 2.0), (3, 3.0)"
+ order_qt_direct_version_15 """
+ SELECT stddev_samp(x) FROM
test_restricted_aggregate_versioned_source
+ """
+
+ sql "DROP TABLE IF EXISTS test_restricted_aggregate_versioned_state"
+ sql """
+ CREATE TABLE test_restricted_aggregate_versioned_state (
+ k INT,
+ state AGG_STATE<stddev_samp(DOUBLE)> GENERIC
+ ) AGGREGATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+ sql """
+ INSERT INTO test_restricted_aggregate_versioned_state
+ SELECT 1, stddev_samp_state(x) FROM
test_restricted_aggregate_versioned_source
+ """
+
+ sql "ADMIN SET FRONTEND CONFIG ('be_exec_version' = '16')"
+ order_qt_stored_version_15_at_16 """
+ SELECT stddev_samp_merge(state) FROM
test_restricted_aggregate_versioned_state
+ """
+ order_qt_direct_version_16 """
+ SELECT stddev_samp(x) FROM
test_restricted_aggregate_versioned_source
+ """
+ sql "ADMIN SET FRONTEND CONFIG ('be_exec_version' = '17')"
+ order_qt_stored_version_15_at_17 """
+ SELECT stddev_samp_merge(state) FROM
test_restricted_aggregate_versioned_state
+ """
+ order_qt_direct_version_17 """
+ SELECT stddev_samp(x) FROM
test_restricted_aggregate_versioned_source
+ """
+ } finally {
+ sql "ADMIN SET FRONTEND CONFIG ('be_exec_version' =
'${originalVersion}')"
+ }
+}
diff --git
a/regression-test/suites/datatype_p0/ip/test_map_agg_v2_ip_key.groovy
b/regression-test/suites/datatype_p0/ip/test_map_agg_v2_ip_key.groovy
new file mode 100644
index 00000000000..3f3887b775d
--- /dev/null
+++ b/regression-test/suites/datatype_p0/ip/test_map_agg_v2_ip_key.groovy
@@ -0,0 +1,77 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_map_agg_v2_ip_key") {
+ sql "DROP TABLE IF EXISTS test_map_agg_v2_ip_key"
+ sql """
+ CREATE TABLE test_map_agg_v2_ip_key (
+ id INT,
+ ipv4_key IPV4,
+ ipv6_key IPV6,
+ value INT
+ )
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+
+ sql """
+ INSERT INTO test_map_agg_v2_ip_key VALUES
+ (1, '192.168.0.1', '2001:db8::1', 10),
+ (1, '192.168.0.2', '2001:db8::2', 20)
+ """
+
+ sql """
+ INSERT INTO test_map_agg_v2_ip_key
+ SELECT
+ 2,
+ CAST(CONCAT('10.0.0.', CAST(number + 1 AS STRING)) AS IPV4),
+ CAST(CONCAT('2001:db8::', CAST(number + 1 AS STRING)) AS IPV6),
+ CAST(number AS INT)
+ FROM numbers("number" = "65")
+ """
+
+ qt_ipv4_key """
+ SELECT array_sort(map_keys(map_agg_v2(ipv4_key, value)))
+ FROM test_map_agg_v2_ip_key
+ WHERE id = 1
+ """
+
+ qt_ipv6_key """
+ SELECT array_sort(map_keys(map_agg_v2(ipv6_key, value)))
+ FROM test_map_agg_v2_ip_key
+ WHERE id = 1
+ """
+
+ qt_ipv4_state_merge """
+ SELECT map_size(map_agg_v2_merge(state))
+ FROM (
+ SELECT map_agg_v2_state(ipv4_key, value) AS state
+ FROM test_map_agg_v2_ip_key
+ WHERE id = 2
+ ) states
+ """
+
+ qt_ipv6_state_merge """
+ SELECT map_size(map_agg_v2_merge(state))
+ FROM (
+ SELECT map_agg_v2_state(ipv6_key, value) AS state
+ FROM test_map_agg_v2_ip_key
+ WHERE id = 2 AND value < 17
+ ) states
+ """
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
