upd

Author: Mryange

be/src/exec/common/agg_context_utils.h

@@ -124,7 +124,7 @@ inline void build_serialized_output_block(Block* block, ColumnRawPtrs& key_colum
 /// Unified version used by both GroupByAggContext and UngroupByAggContext.
 inline int get_slot_column_id(const AggFnEvaluator* evaluator) {
     auto ctxs = evaluator->input_exprs_ctxs();
-    CHECK(ctxs.size() == 1 && ctxs[0]->root()->is_slot_ref())
+    DCHECK(ctxs.size() == 1 && ctxs[0]->root()->is_slot_ref())
             << "input_exprs_ctxs is invalid, input_exprs_ctx[0]="
             << ctxs[0]->root()->debug_string();
     return static_cast<VSlotRef*>(ctxs[0]->root().get())->column_id();

be/src/exec/common/groupby_agg_context.cpp

@@ -683,7 +683,7 @@ Status GroupByAggContext::get_serialized_results(RuntimeState* state, Block* blo
     MutableColumns value_columns(agg_size);
     DataTypes value_data_types(agg_size);
 
-    bool mem_reuse = block->mem_reuse();
+    bool mem_reuse = make_nullable_keys.empty() && block->mem_reuse();
 
     auto key_columns = agg_context_utils::take_or_create_columns(
             block, mem_reuse, 0, key_size,
@@ -774,7 +774,7 @@ Status GroupByAggContext::get_serialized_results(RuntimeState* state, Block* blo
 
 Status GroupByAggContext::get_finalized_results(RuntimeState* state, Block* block, bool* eos,
                                                const ColumnsWithTypeAndName& columns_with_schema) {
-    bool mem_reuse = block->mem_reuse();
+    bool mem_reuse = make_nullable_keys.empty() && block->mem_reuse();
 
     size_t key_size = _groupby_expr_ctxs.size();
 

be/src/exec/common/groupby_agg_context.h

@@ -139,6 +139,11 @@ class GroupByAggContext {
     bool should_limit_output = false;
     bool enable_spill = false;
 
+    // Key columns that need nullable wrapping in output (left/full join).
+    // When non-empty, mem_reuse must be disabled in get_*_results to avoid
+    // column type mismatch after make_nullable_output_key transforms the block.
+    std::vector<size_t> make_nullable_keys;
+
     // Memory tracking for reserve estimation
     int64_t memory_usage_last_executing = 0;
 

be/src/exec/common/inline_count_agg_context.cpp

@@ -72,6 +72,8 @@ void InlineCountAggContext::emplace_into_hash_table(AggregateDataPtr* /*places*/
 // ==================== Aggregation execution ====================
 
 Status InlineCountAggContext::execute_with_serialized_key(Block* block) {
+    memory_usage_last_executing = 0;
+    SCOPED_PEAK_MEM(&memory_usage_last_executing);
     SCOPED_TIMER(_build_timer);
     DCHECK(!block->empty());
 
@@ -164,7 +166,7 @@ Status InlineCountAggContext::get_serialized_results(RuntimeState* state, Block*
     size_t key_size = _groupby_expr_ctxs.size();
     DCHECK_EQ(_agg_evaluators.size(), 1);
 
-    bool mem_reuse = block->mem_reuse();
+    bool mem_reuse = make_nullable_keys.empty() && block->mem_reuse();
 
     auto key_columns = agg_context_utils::take_or_create_columns(
             block, mem_reuse, 0, key_size,
@@ -199,7 +201,7 @@ Status InlineCountAggContext::get_serialized_results(RuntimeState* state, Block*
                             while (it != agg_method.end && num_rows < state->batch_size()) {
                                 keys[num_rows] = it.get_first();
                                 auto inline_count =
-                                        reinterpret_cast<const UInt64&>(it.get_second());
+                                        std::bit_cast<UInt64>(it.get_second());
                                 count_col.insert_data(
                                         reinterpret_cast<const char*>(&inline_count),
                                         sizeof(UInt64));
@@ -224,7 +226,7 @@ Status InlineCountAggContext::get_serialized_results(RuntimeState* state, Block*
                                             agg_method.hash_table->template get_null_key_data<
                                                     AggregateDataPtr>();
                                     auto inline_count =
-                                            reinterpret_cast<const UInt64&>(mapped);
+                                            std::bit_cast<UInt64>(mapped);
                                     count_col.insert_data(
                                             reinterpret_cast<const char*>(&inline_count),
                                             sizeof(UInt64));
@@ -251,7 +253,7 @@ Status InlineCountAggContext::get_serialized_results(RuntimeState* state, Block*
 Status InlineCountAggContext::get_finalized_results(
         RuntimeState* state, Block* block, bool* eos,
         const ColumnsWithTypeAndName& columns_with_schema) {
-    bool mem_reuse = block->mem_reuse();
+    bool mem_reuse = make_nullable_keys.empty() && block->mem_reuse();
 
     size_t key_size = _groupby_expr_ctxs.size();
 
@@ -288,7 +290,7 @@ Status InlineCountAggContext::get_finalized_results(
                                 keys[num_rows] = it.get_first();
                                 auto& mapped = it.get_second();
                                 count_column.insert_value(static_cast<Int64>(
-                                        reinterpret_cast<const UInt64&>(mapped)));
+                                        std::bit_cast<UInt64>(mapped)));
                                 ++it;
                                 ++num_rows;
                             }