This is an automated email from the ASF dual-hosted git repository.
zclllyybb 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 8be719da4fa [fix](be) Publish nullable hash state after creation
(#66102)
8be719da4fa is described below
commit 8be719da4fa8eaf17a40a059fdcfa984bc152abd
Author: HappenLee <[email protected]>
AuthorDate: Thu Jul 30 12:30:58 2026 +0800
[fix](be) Publish nullable hash state after creation (#66102)
Nullable aggregation hash tables marked the null-key entry as present
before its aggregate-state creator completed. If allocation or
aggregate-state construction threw, cleanup treated uninitialized
null-key storage as a valid state and could crash while destroying it.
Publishing the flag only after the creator returns fixes that case, but
nullable LIMIT creators perform additional potentially throwing work
after aggregate-state construction. If top-N column insertion or heap
maintenance throws, the flag remains unpublished while a live
resource-owning aggregate state would otherwise be left behind.
This change preserves delayed flag publication and gives the LIMIT
null-key creators a strong exception guarantee: construct into a local
pointer, perform post-construction LIMIT maintenance, destroy the
completed state if that work throws, and publish the mapped pointer only
on success. Null-key storage is value-initialized, and the same
publication ordering is applied to StringHashMap batch paths. Unit tests
cover immediate creator failure, construction followed by post-create
failure and rollback, and successful retry for ordinary nullable and
StringHashMap paths.
---
be/src/exec/common/agg_utils.h | 13 ++
be/src/exec/common/columns_hashing.h | 9 +-
be/src/exec/common/hash_table/hash_map_context.h | 12 +-
be/src/exec/operator/aggregation_sink_operator.cpp | 20 ++-
.../operator/streaming_aggregation_operator.cpp | 9 +-
be/test/exec/hash_map/hash_table_method_test.cpp | 141 +++++++++++++++++++++
6 files changed, 182 insertions(+), 22 deletions(-)
diff --git a/be/src/exec/common/agg_utils.h b/be/src/exec/common/agg_utils.h
index b613a68a5ab..6c89684d436 100644
--- a/be/src/exec/common/agg_utils.h
+++ b/be/src/exec/common/agg_utils.h
@@ -17,6 +17,7 @@
#pragma once
+#include <utility>
#include <variant>
#include <vector>
@@ -28,6 +29,18 @@
namespace doris {
+template <typename PostCreate, typename Rollback>
+void commit_aggregate_state(AggregateDataPtr& mapped, AggregateDataPtr
new_state,
+ PostCreate&& post_create, Rollback&& rollback) {
+ try {
+ std::forward<PostCreate>(post_create)();
+ } catch (...) {
+ std::forward<Rollback>(rollback)(new_state);
+ throw;
+ }
+ mapped = new_state;
+}
+
template <typename T>
using AggData = PHHashMap<T, AggregateDataPtr, HashCRC32<T>>;
template <typename T>
diff --git a/be/src/exec/common/columns_hashing.h
b/be/src/exec/common/columns_hashing.h
index ee6a76445df..dc346097dc3 100644
--- a/be/src/exec/common/columns_hashing.h
+++ b/be/src/exec/common/columns_hashing.h
@@ -128,18 +128,17 @@ struct HashMethodSingleLowNullableColumn : public
SingleColumnMethod {
size_t hash_value, Func&& f,
CreatorForNull&& null_creator) {
if (key_column->is_null_at(row)) {
- bool has_null_key = data.has_null_key_data();
- data.has_null_key_data() = true;
-
if constexpr (std::is_same_v<Mapped, void>) {
- if (!has_null_key) {
+ if (!data.has_null_key_data()) {
std::forward<CreatorForNull>(null_creator)();
+ data.has_null_key_data() = true;
}
return nullptr;
} else {
- if (!has_null_key) {
+ if (!data.has_null_key_data()) {
std::forward<CreatorForNull>(null_creator)(
data.template get_null_key_data<Mapped>());
+ data.has_null_key_data() = true;
}
return &data.template get_null_key_data<Mapped>();
}
diff --git a/be/src/exec/common/hash_table/hash_map_context.h
b/be/src/exec/common/hash_table/hash_map_context.h
index 804d2e48020..fb8184f9fea 100644
--- a/be/src/exec/common/hash_table/hash_map_context.h
+++ b/be/src/exec/common/hash_table/hash_map_context.h
@@ -506,11 +506,10 @@ void process_submap_emplace(Submap& submap, const
uint32_t* indices, size_t coun
pre_handler(row);
if constexpr (is_nullable) {
if (state.key_column->is_null_at(row)) {
- bool has_null_key = hash_table.has_null_key_data();
- hash_table.has_null_key_data() = true;
- if (!has_null_key) {
+ if (!hash_table.has_null_key_data()) {
std::forward<FF>(creator_for_null_key)(
hash_table.template get_null_key_data<Mapped>());
+ hash_table.has_null_key_data() = true;
}
result_handler(row, hash_table.template
get_null_key_data<Mapped>());
continue;
@@ -552,10 +551,9 @@ void process_submap_emplace_void(Submap& submap, const
uint32_t* indices, size_t
pre_handler(row);
if constexpr (is_nullable) {
if (state.key_column->is_null_at(row)) {
- bool has_null_key = hash_table.has_null_key_data();
- hash_table.has_null_key_data() = true;
- if (!has_null_key) {
+ if (!hash_table.has_null_key_data()) {
std::forward<FF>(creator_for_null_key)();
+ hash_table.has_null_key_data() = true;
}
continue;
}
@@ -1164,7 +1162,7 @@ struct DataWithNullKey : public Base {
protected:
bool has_null_key = false;
- Base::Value null_key_data;
+ Base::Value null_key_data {};
};
/// Single low cardinality column.
diff --git a/be/src/exec/operator/aggregation_sink_operator.cpp
b/be/src/exec/operator/aggregation_sink_operator.cpp
index c77563e9751..f4e74be9995 100644
--- a/be/src/exec/operator/aggregation_sink_operator.cpp
+++ b/be/src/exec/operator/aggregation_sink_operator.cpp
@@ -749,16 +749,22 @@ bool
AggSinkLocalState::_emplace_into_hash_table_limit(AggregateDataPtr* places,
};
auto creator_for_null_key = [&](auto& mapped) {
- mapped =
Base::_shared_state->agg_arena_pool.aligned_alloc(
- Base::_parent->template
cast<AggSinkOperatorX>()
-
._total_size_of_aggregate_states,
- Base::_parent->template
cast<AggSinkOperatorX>()
- ._align_aggregate_states);
- auto st = _create_agg_status(mapped);
+ auto* new_state =
+
Base::_shared_state->agg_arena_pool.aligned_alloc(
+ Base::_parent->template
cast<AggSinkOperatorX>()
+
._total_size_of_aggregate_states,
+ Base::_parent->template
cast<AggSinkOperatorX>()
+
._align_aggregate_states);
+ auto st = _create_agg_status(new_state);
if (!st) {
throw Exception(st.code(),
st.to_string());
}
- _shared_state->refresh_top_limit(i,
key_columns);
+ commit_aggregate_state(
+ mapped, new_state,
+ [&] {
_shared_state->refresh_top_limit(i, key_columns); },
+ [&](auto* state) {
+
static_cast<void>(_destroy_agg_status(state));
+ });
};
SCOPED_TIMER(_hash_table_emplace_timer);
diff --git a/be/src/exec/operator/streaming_aggregation_operator.cpp
b/be/src/exec/operator/streaming_aggregation_operator.cpp
index 28fc383bace..31ae0d843e4 100644
--- a/be/src/exec/operator/streaming_aggregation_operator.cpp
+++ b/be/src/exec/operator/streaming_aggregation_operator.cpp
@@ -763,16 +763,19 @@ bool
StreamingAggLocalState::_emplace_into_hash_table_limit(AggregateDataPtr* pl
};
auto creator_for_null_key = [&](auto& mapped) {
- mapped = _agg_arena_pool.aligned_alloc(
+ auto* new_state =
_agg_arena_pool.aligned_alloc(
Base::_parent->template
cast<StreamingAggOperatorX>()
._total_size_of_aggregate_states,
Base::_parent->template
cast<StreamingAggOperatorX>()
._align_aggregate_states);
- auto st = _create_agg_status(mapped);
+ auto st = _create_agg_status(new_state);
if (!st) {
throw Exception(st.code(),
st.to_string());
}
- _refresh_limit_heap(i, key_columns);
+ commit_aggregate_state(
+ mapped, new_state,
+ [&] { _refresh_limit_heap(i,
key_columns); },
+ [&](auto* state) {
_destroy_agg_status(state); });
};
SCOPED_TIMER(_hash_table_emplace_timer);
diff --git a/be/test/exec/hash_map/hash_table_method_test.cpp
b/be/test/exec/hash_map/hash_table_method_test.cpp
index de2f0a7ac63..0fcab89040d 100644
--- a/be/test/exec/hash_map/hash_table_method_test.cpp
+++ b/be/test/exec/hash_map/hash_table_method_test.cpp
@@ -17,6 +17,7 @@
#include <gtest/gtest.h>
+#include <new>
#include <set>
#include "core/data_type/data_type_number.h"
@@ -25,6 +26,7 @@
#include "exec/common/hash_table/hash.h"
#include "exec/common/hash_table/hash_map_context.h"
#include "exec/common/hash_table/ph_hash_map.h"
+#include "exec/common/hash_table/ph_hash_set.h"
#include "testutil/column_helper.h"
namespace doris {
@@ -254,6 +256,145 @@ static AggregateDataPtr make_mapped(size_t val) {
return reinterpret_cast<AggregateDataPtr>(val);
}
+struct TrackedNullAggregateState {
+ explicit TrackedNullAggregateState(size_t& destroy_count_) :
destroy_count(destroy_count_) {}
+ ~TrackedNullAggregateState() { ++destroy_count; }
+
+ size_t& destroy_count;
+};
+
+static void create_null_state_then_fail(AggregateDataPtr& mapped, void*
storage,
+ size_t& destroy_count) {
+ auto* new_state = new (storage) TrackedNullAggregateState(destroy_count);
+ commit_aggregate_state(
+ mapped, reinterpret_cast<AggregateDataPtr>(new_state),
+ [] {
+ throw Exception(ErrorCode::INTERNAL_ERROR,
+ "post-construction null key creation failed");
+ },
+ [](AggregateDataPtr state) {
+
reinterpret_cast<TrackedNullAggregateState*>(state)->~TrackedNullAggregateState();
+ });
+}
+
+TEST(HashTableMethodTest, testNullableNullKeyCreationExceptionSafety) {
+ using NullableMethod =
+ MethodSingleNullableColumn<MethodOneNumber<UInt32,
AggDataNullable<UInt32>>>;
+ NullableMethod method;
+ using State = NullableMethod::State;
+
+ auto col = ColumnHelper::create_nullable_column<DataTypeInt32>({0}, {1});
+ ColumnRawPtrs key_columns = {col.get()};
+ State state(key_columns);
+ method.init_serialized_keys(key_columns, 1);
+
+ EXPECT_THROW(method.lazy_emplace(
+ state, 0, [](const auto&, auto&, auto&) {},
+ [](auto&) {
+ throw Exception(ErrorCode::INTERNAL_ERROR, "null
key creation failed");
+ }),
+ Exception);
+ EXPECT_FALSE(method.hash_table->has_null_key_data());
+ EXPECT_TRUE(method.hash_table->empty());
+ EXPECT_FALSE(method.find(state, 0).is_found());
+
+ alignas(TrackedNullAggregateState) char
state_storage[sizeof(TrackedNullAggregateState)];
+ size_t destroy_count = 0;
+ EXPECT_THROW(method.lazy_emplace(
+ state, 0, [](const auto&, auto&, auto&) {},
+ [&](auto& null_mapped) {
+ create_null_state_then_fail(null_mapped,
state_storage, destroy_count);
+ }),
+ Exception);
+ EXPECT_EQ(destroy_count, 1);
+ EXPECT_FALSE(method.hash_table->has_null_key_data());
+ EXPECT_EQ(method.hash_table->get_null_key_data<AggregateDataPtr>(),
nullptr);
+ EXPECT_TRUE(method.hash_table->empty());
+ EXPECT_FALSE(method.find(state, 0).is_found());
+
+ auto* mapped = method.lazy_emplace(
+ state, 0, [](const auto&, auto&, auto&) {},
+ [](auto& null_mapped) { null_mapped = make_mapped(123); });
+ ASSERT_NE(mapped, nullptr);
+ EXPECT_EQ(*mapped, make_mapped(123));
+ EXPECT_TRUE(method.hash_table->has_null_key_data());
+ EXPECT_EQ(method.hash_table->size(), 1);
+}
+
+TEST(HashTableMethodTest, testNullableVoidNullKeyCreationExceptionSafety) {
+ using NullableMethod = MethodSingleNullableColumn<
+ MethodOneNumber<UInt32, DataWithNullKey<PHHashSet<UInt32,
HashCRC32<UInt32>>>>>;
+ NullableMethod method;
+ using State = NullableMethod::State;
+
+ auto col = ColumnHelper::create_nullable_column<DataTypeInt32>({0}, {1});
+ ColumnRawPtrs key_columns = {col.get()};
+ State state(key_columns);
+ method.init_serialized_keys(key_columns, 1);
+
+ EXPECT_THROW(
+ method.lazy_emplace(
+ state, 0, [](const auto&, auto&, auto&) {},
+ [] { throw Exception(ErrorCode::INTERNAL_ERROR, "null key
creation failed"); }),
+ Exception);
+ EXPECT_FALSE(method.hash_table->has_null_key_data());
+ EXPECT_TRUE(method.hash_table->empty());
+
+ method.lazy_emplace(
+ state, 0, [](const auto&, auto&, auto&) {}, [] {});
+ EXPECT_TRUE(method.hash_table->has_null_key_data());
+ EXPECT_EQ(method.hash_table->size(), 1);
+}
+
+TEST(HashTableMethodTest,
testNullableStringBatchNullKeyCreationExceptionSafety) {
+ using NullableMethod = MethodSingleNullableColumn<
+ MethodStringNoCache<AggregatedDataWithNullableShortStringKey>>;
+ NullableMethod method;
+ using State = NullableMethod::State;
+
+ auto col = ColumnHelper::create_nullable_column<DataTypeString>({""}, {1});
+ ColumnRawPtrs key_columns = {col.get()};
+ State state(key_columns);
+ method.init_serialized_keys(key_columns, 1);
+
+ EXPECT_THROW(lazy_emplace_batch(
+ method, state, 1, [](const auto&, auto&, auto&) {},
+ [](auto&) {
+ throw Exception(ErrorCode::INTERNAL_ERROR, "null
key creation failed");
+ },
+ [](uint32_t, auto&) {}),
+ Exception);
+ EXPECT_FALSE(method.hash_table->has_null_key_data());
+ EXPECT_TRUE(method.hash_table->empty());
+
+ alignas(TrackedNullAggregateState) char
state_storage[sizeof(TrackedNullAggregateState)];
+ size_t destroy_count = 0;
+ EXPECT_THROW(lazy_emplace_batch(
+ method, state, 1, [](const auto&, auto&, auto&) {},
+ [&](auto& null_mapped) {
+ create_null_state_then_fail(null_mapped,
state_storage, destroy_count);
+ },
+ [](uint32_t, auto&) {}),
+ Exception);
+ EXPECT_EQ(destroy_count, 1);
+ EXPECT_FALSE(method.hash_table->has_null_key_data());
+ EXPECT_EQ(method.hash_table->get_null_key_data<AggregateDataPtr>(),
nullptr);
+ EXPECT_TRUE(method.hash_table->empty());
+
+ bool result_handled = false;
+ lazy_emplace_batch(
+ method, state, 1, [](const auto&, auto&, auto&) {},
+ [](auto& null_mapped) { null_mapped = make_mapped(456); },
+ [&](uint32_t row, auto& mapped) {
+ EXPECT_EQ(row, 0);
+ EXPECT_EQ(mapped, make_mapped(456));
+ result_handled = true;
+ });
+ EXPECT_TRUE(result_handled);
+ EXPECT_TRUE(method.hash_table->has_null_key_data());
+ EXPECT_EQ(method.hash_table->size(), 1);
+}
+
// ========== MethodOneNumber<UInt32, AggData<UInt32>> ==========
// AggData<UInt32> = PHHashMap<UInt32, AggregateDataPtr, HashCRC32<UInt32>>
TEST(HashTableMethodTest, testMethodOneNumberAggInsertFindForEach) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
