update

BiteTheDDDDt · BiteTheDDDDt · commit 1e4bddae73d8 · 2026-03-02T11:59:16.000+08:00
diff --git a/be/src/pipeline/exec/aggregation_sink_operator.cpp b/be/src/pipeline/exec/aggregation_sink_operator.cpp
@@ -649,12 +649,10 @@ bool AggSinkLocalState::_emplace_into_hash_table_limit(vectorized::AggregateData
                             };
 
                             SCOPED_TIMER(_hash_table_emplace_timer);
-                            vectorized::lazy_emplace_batch(agg_method, state, num_rows, creator,
-                                                           creator_for_null_key,
-                                                           [&](uint32_t row, auto& mapped) {
-                                                               i = row;
-                                                               places[row] = mapped;
-                                                           });
+                            vectorized::lazy_emplace_batch(
+                                    agg_method, state, num_rows, creator, creator_for_null_key,
+                                    [&](uint32_t row) { i = row; },
+                                    [&](uint32_t row, auto& mapped) { places[row] = mapped; });
                             COUNTER_UPDATE(_hash_table_input_counter, num_rows);
                             return true;
                         }
diff --git a/be/src/pipeline/exec/streaming_aggregation_operator.cpp b/be/src/pipeline/exec/streaming_aggregation_operator.cpp
@@ -728,12 +728,10 @@ bool StreamingAggLocalState::_emplace_into_hash_table_limit(vectorized::Aggregat
                             };
 
                             SCOPED_TIMER(_hash_table_emplace_timer);
-                            vectorized::lazy_emplace_batch(agg_method, state, num_rows, creator,
-                                                           creator_for_null_key,
-                                                           [&](uint32_t row, auto& mapped) {
-                                                               i = row;
-                                                               places[row] = mapped;
-                                                           });
+                            vectorized::lazy_emplace_batch(
+                                    agg_method, state, num_rows, creator, creator_for_null_key,
+                                    [&](uint32_t row) { i = row; },
+                                    [&](uint32_t row, auto& mapped) { places[row] = mapped; });
                             COUNTER_UPDATE(_hash_table_input_counter, num_rows);
                             return true;
                         }
diff --git a/be/src/vec/common/hash_table/hash_map_context.h b/be/src/vec/common/hash_table/hash_map_context.h
@@ -112,22 +112,21 @@ struct MethodBaseInner {
     }
 
     template <bool read>
-    ALWAYS_INLINE void prefetch(size_t i) {
+    void prefetch(size_t i) {
         if (LIKELY(i + HASH_MAP_PREFETCH_DIST < hash_values.size())) {
             hash_table->template prefetch<read>(keys[i + HASH_MAP_PREFETCH_DIST],
                                                 hash_values[i + HASH_MAP_PREFETCH_DIST]);
         }
     }
 
     template <typename State>
-    ALWAYS_INLINE auto find(State& state, size_t i) {
+    auto find(State& state, size_t i) {
         prefetch<true>(i);
         return state.find_key_with_hash(*hash_table, i, keys[i], hash_values[i]);
     }
 
     template <typename State, typename F, typename FF>
-    ALWAYS_INLINE auto lazy_emplace(State& state, size_t i, F&& creator,
-                                    FF&& creator_for_null_key) {
+    auto lazy_emplace(State& state, size_t i, F&& creator, FF&& creator_for_null_key) {
         prefetch<false>(i);
         return state.lazy_emplace_key(*hash_table, i, keys[i], hash_values[i], creator,
                                       creator_for_null_key);
@@ -432,7 +431,7 @@ struct IsNullableStringHashMap<DataWithNullKey<StringHashMap<Mapped, Allocator>>
 
 /// Helper: get the underlying StringHashTable from a hash table (handles DataWithNullKey wrapper).
 template <typename HashMap>
-ALWAYS_INLINE auto& get_string_hash_table(HashMap& data) {
+auto& get_string_hash_table(HashMap& data) {
     return data;
 }
 
@@ -441,7 +440,7 @@ ALWAYS_INLINE auto& get_string_hash_table(HashMap& data) {
 /// Returns a pair: (converted_key, needs_copy).
 /// For groups 0 and 5, the key must be a non-const copy (lazy_emplace_if_zero takes Key&).
 template <int GroupIdx>
-ALWAYS_INLINE auto convert_key_for_submap(const StringRef& origin) {
+auto convert_key_for_submap(const StringRef& origin) {
     if constexpr (GroupIdx == 0) {
         return StringRef(origin); // copy — m0 needs non-const Key&
     } else if constexpr (GroupIdx == 1) {
@@ -459,7 +458,7 @@ ALWAYS_INLINE auto convert_key_for_submap(const StringRef& origin) {
 
 /// Hash value to use for a given group. Group 0 (empty string) always uses hash=0.
 template <int GroupIdx, typename HashValues>
-ALWAYS_INLINE size_t hash_for_group(const HashValues& hash_values, uint32_t row) {
+size_t hash_for_group(const HashValues& hash_values, uint32_t row) {
     if constexpr (GroupIdx == 0) {
         return 0;
     } else {
@@ -474,13 +473,14 @@ static constexpr bool group_needs_prefetch = (GroupIdx != 0);
 
 /// Process one sub-table group for emplace with result_handler.
 /// Handles nullable null-key check, prefetch, key conversion, and emplace.
+/// pre_handler(row) is called before each emplace, allowing callers to set per-row state
+/// (e.g., current row index used inside creator lambdas).
 template <int GroupIdx, bool is_nullable, typename Submap, typename HashMethodType, typename State,
-          typename HashMap, typename F, typename FF, typename ResultHandler>
-ALWAYS_INLINE void process_submap_emplace(Submap& submap, const uint32_t* indices, size_t count,
-                                          HashMethodType& agg_method, State& state,
-                                          HashMap& hash_table, F&& creator,
-                                          FF&& creator_for_null_key,
-                                          ResultHandler&& result_handler) {
+          typename HashMap, typename F, typename FF, typename PreHandler, typename ResultHandler>
+void process_submap_emplace(Submap& submap, const uint32_t* indices, size_t count,
+                            HashMethodType& agg_method, State& state, HashMap& hash_table,
+                            F&& creator, FF&& creator_for_null_key, PreHandler&& pre_handler,
+                            ResultHandler&& result_handler) {
     using Mapped = typename HashMethodType::Mapped;
     for (size_t j = 0; j < count; j++) {
         if constexpr (group_needs_prefetch<GroupIdx>) {
@@ -490,6 +490,7 @@ ALWAYS_INLINE void process_submap_emplace(Submap& submap, const uint32_t* indice
             }
         }
         uint32_t row = indices[j];
+        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();
@@ -524,11 +525,9 @@ ALWAYS_INLINE void process_submap_emplace(Submap& submap, const uint32_t* indice
 /// pre_handler(row) is called before each emplace.
 template <int GroupIdx, bool is_nullable, typename Submap, typename HashMethodType, typename State,
           typename HashMap, typename F, typename FF, typename PreHandler>
-ALWAYS_INLINE void process_submap_emplace_void(Submap& submap, const uint32_t* indices,
-                                               size_t count, HashMethodType& agg_method,
-                                               State& state, HashMap& hash_table, F&& creator,
-                                               FF&& creator_for_null_key,
-                                               PreHandler&& pre_handler) {
+void process_submap_emplace_void(Submap& submap, const uint32_t* indices, size_t count,
+                                 HashMethodType& agg_method, State& state, HashMap& hash_table,
+                                 F&& creator, FF&& creator_for_null_key, PreHandler&& pre_handler) {
     for (size_t j = 0; j < count; j++) {
         if constexpr (group_needs_prefetch<GroupIdx>) {
             if (j + HASH_MAP_PREFETCH_DIST < count) {
@@ -566,9 +565,9 @@ ALWAYS_INLINE void process_submap_emplace_void(Submap& submap, const uint32_t* i
 /// Process one sub-table group for find with result_handler.
 template <int GroupIdx, bool is_nullable, typename Submap, typename HashMethodType, typename State,
           typename HashMap, typename ResultHandler>
-ALWAYS_INLINE void process_submap_find(Submap& submap, const uint32_t* indices, size_t count,
-                                       HashMethodType& agg_method, State& state,
-                                       HashMap& hash_table, ResultHandler&& result_handler) {
+void process_submap_find(Submap& submap, const uint32_t* indices, size_t count,
+                         HashMethodType& agg_method, State& state, HashMap& hash_table,
+                         ResultHandler&& result_handler) {
     using Mapped = typename HashMethodType::Mapped;
     using FindResult = typename ColumnsHashing::columns_hashing_impl::FindResultImpl<Mapped>;
     for (size_t j = 0; j < count; j++) {
@@ -606,10 +605,14 @@ ALWAYS_INLINE void process_submap_find(Submap& submap, const uint32_t* indices,
 
 /// Batch emplace helper: for StringHashMap, directly accesses sub-tables bypassing dispatch();
 /// for other hash maps, does per-row loop with standard prefetch.
-/// The result_handler is called for each row: result_handler(row_index, emplace_result)
-template <typename HashMethodType, typename State, typename F, typename FF, typename ResultHandler>
+/// pre_handler(row) is called before each emplace, allowing callers to set per-row state
+/// (e.g., current row index used inside creator lambdas).
+/// result_handler(row_index, mapped) is called after each emplace.
+template <typename HashMethodType, typename State, typename F, typename FF, typename PreHandler,
+          typename ResultHandler>
 void lazy_emplace_batch(HashMethodType& agg_method, State& state, uint32_t num_rows, F&& creator,
-                        FF&& creator_for_null_key, ResultHandler&& result_handler) {
+                        FF&& creator_for_null_key, PreHandler&& pre_handler,
+                        ResultHandler&& result_handler) {
     if constexpr (HashMethodType::is_string_hash_map()) {
         using HashMap = typename HashMethodType::HashMapType;
         constexpr bool is_nullable = IsNullableStringHashMap<HashMap>::value;
@@ -622,22 +625,33 @@ void lazy_emplace_batch(HashMethodType& agg_method, State& state, uint32_t num_r
             constexpr int G = decltype(group_idx)::value;
             const auto& indices = groups.group_row_indices[G];
             if (!indices.empty()) {
-                process_submap_emplace<G, is_nullable>(submap, indices.data(), indices.size(),
-                                                       agg_method, state, hash_table, creator,
-                                                       creator_for_null_key, result_handler);
+                process_submap_emplace<G, is_nullable>(
+                        submap, indices.data(), indices.size(), agg_method, state, hash_table,
+                        creator, creator_for_null_key, pre_handler, result_handler);
             }
         });
     } else {
         // Standard per-row loop with ahead prefetch
         for (uint32_t i = 0; i < num_rows; ++i) {
             agg_method.template prefetch<false>(i);
+            pre_handler(i);
             result_handler(i, *state.lazy_emplace_key(*agg_method.hash_table, i, agg_method.keys[i],
                                                       agg_method.hash_values[i], creator,
                                                       creator_for_null_key));
         }
     }
 }
 
+/// Convenience overload without pre_handler (uses no-op).
+template <typename HashMethodType, typename State, typename F, typename FF, typename ResultHandler>
+void lazy_emplace_batch(HashMethodType& agg_method, State& state, uint32_t num_rows, F&& creator,
+                        FF&& creator_for_null_key, ResultHandler&& result_handler) {
+    lazy_emplace_batch(
+            agg_method, state, num_rows, std::forward<F>(creator),
+            std::forward<FF>(creator_for_null_key), [](uint32_t) {},
+            std::forward<ResultHandler>(result_handler));
+}
+
 /// Batch emplace helper (void version): like lazy_emplace_batch but ignores the return value.
 /// pre_handler(row) is called before each emplace, allowing callers to update captured state
 /// (e.g., the current row index used inside creator lambdas).
