Merge pull request #15984 from ClickHouse/fix_insert_mv_dangling_ref

Fix possibly dangling reference to target table of MV

src/Interpreters/InterpreterInsertQuery.cpp

@@ -413,6 +413,11 @@ BlockIO InterpreterInsertQuery::execute()
         res.out = std::move(out_streams.at(0));
 
     res.pipeline.addStorageHolder(table);
+    if (const auto * mv = dynamic_cast<const StorageMaterializedView *>(table.get()))
+    {
+        if (auto inner_table = mv->tryGetTargetTable())
+            res.pipeline.addStorageHolder(inner_table);
+    }
 
     return res;
 }

src/Storages/StorageBuffer.cpp

@@ -239,6 +239,7 @@ Pipe StorageBuffer::read(
         }
 
         pipe_from_dst.addTableLock(destination_lock);
+        pipe_from_dst.addStorageHolder(destination);
     }
 
     Pipe pipe_from_buffers;

src/Storages/StorageMaterializeMySQL.cpp

@@ -82,6 +82,7 @@ Pipe StorageMaterializeMySQL::read(
     }
 
     Pipe pipe = nested_storage->read(require_columns_name, nested_metadata, query_info, context, processed_stage, max_block_size, num_streams);
+    pipe.addTableLock(lock);
 
     if (!expressions->children.empty() && !pipe.empty())
     {

src/Storages/StorageMaterializedView.cpp

@@ -124,6 +124,7 @@ Pipe StorageMaterializedView::read(
 
     Pipe pipe = storage->read(column_names, metadata_snapshot, query_info, context, processed_stage, max_block_size, num_streams);
     pipe.addTableLock(lock);
+    pipe.addStorageHolder(storage);
 
     return pipe;
 }

src/Storages/StorageMerge.cpp

@@ -344,6 +344,7 @@ Pipe StorageMerge::createSources(
         convertingSourceStream(header, metadata_snapshot, *modified_context, modified_query_info.query, pipe, processed_stage);
 
         pipe.addTableLock(struct_lock);
+        pipe.addStorageHolder(storage);
         pipe.addInterpreterContext(modified_context);
     }