Bugfix/issue 3662 fix hash map key mapped memory issue (#3667) (#628)

Author: yl-lisen

src/Interpreters/Streaming/Aggregator.cpp

@@ -1821,6 +1821,9 @@ void NO_INLINE Aggregator::mergeDataImpl(
     {
         if (inserted)
         {
+            /// exception-safety - if you can not allocate memory or create states, then destructors will not be called.
+            dst = nullptr;
+
             /// If there are multiple sources, there are more than one AggregatedDataVariant. Aggregator always creates a new AggregatedDataVariant and merge all other
             /// AggregatedDataVariants to the new created one. After finalize(), it does not clean up aggregate state except the new create AggregatedDataVariant.
             /// If it does not alloc new memory for the 'dst' (i.e. aggregate state of the new AggregatedDataVariant which get destroyed after finalize()) but reuse
@@ -3064,17 +3067,17 @@ void Aggregator::serializeAggregateStates(const AggregateDataPtr & place, WriteB
 
 void Aggregator::deserializeAggregateStates(AggregateDataPtr & place, ReadBuffer & rb, Arena * arena) const
 {
+    /// exception-safety - if you can not allocate memory or create states, then destructors will not be called.
+    place = nullptr;
+
     UInt8 has_states;
     readIntBinary(has_states, rb);
     if (has_states)
     {
-        if (!place)
-        {
-            /// Allocate states for all aggregate functions
-            AggregateDataPtr aggregate_data = arena->alignedAlloc(total_size_of_aggregate_states, align_aggregate_states);
-            createAggregateStates(aggregate_data);
-            place = aggregate_data;
-        }
+        /// Allocate states for all aggregate functions
+        AggregateDataPtr aggregate_data = arena->alignedAlloc(total_size_of_aggregate_states, align_aggregate_states);
+        createAggregateStates(aggregate_data);
+        place = aggregate_data;
 
         for (size_t i = 0; i < params.aggregates_size; ++i)
             aggregate_functions[i]->deserialize(place + offsets_of_aggregate_states[i], rb, std::nullopt, arena);

src/Processors/Transforms/Streaming/ChangelogConvertTransform.cpp

@@ -163,7 +163,8 @@ void ChangelogConvertTransform::work()
         const auto & columns = chunk.getColumns();
         for (auto key_col_pos : key_column_positions)
         {
-            materialized_columns.push_back(columns[key_col_pos]->convertToFullColumnIfConst());
+            /// Matierlize Sparse/Const/LowCardinality columns
+            materialized_columns.push_back(columns[key_col_pos]->convertToFullIfNeeded());
             key_columns.push_back(materialized_columns.back().get());
         }
 
@@ -433,8 +434,9 @@ void ChangelogConvertTransform::recover(CheckpointContextPtr ckpt_ctx)
         index.deserialize(
             /*MappedDeserializer*/
             [&](std::unique_ptr<RowRefWithRefCount<LightChunk>> & mapped_, Arena &, ReadBuffer & rb_) {
-                mapped_ = std::make_unique<RowRefWithRefCount<LightChunk>>();
-                mapped_->deserialize(&source_chunks, deserialized_indices_to_blocks, rb_);
+                auto new_mapped = std::make_unique<RowRefWithRefCount<LightChunk>>();
+                new_mapped->deserialize(&source_chunks, deserialized_indices_to_blocks, rb_);
+                mapped_ = std::move(new_mapped);
             },
             pool,
             rb);