Merge pull request #15645 from ClickHouse/read-from-storage-step-field-order

Ensure resource destruction order in ReadFromStorageStep.

src/Processors/Pipe.h

@@ -101,7 +101,7 @@ public:
     void setQuota(const std::shared_ptr<const EnabledQuota> & quota);
 
     /// Do not allow to change the table while the processors of pipe are alive.
-    void addTableLock(const TableLockHolder & lock) { holder.table_locks.push_back(lock); }
+    void addTableLock(TableLockHolder lock) { holder.table_locks.emplace_back(std::move(lock)); }
     /// This methods are from QueryPipeline. Needed to make conversion from pipeline to pipe possible.
     void addInterpreterContext(std::shared_ptr<Context> context) { holder.interpreter_context.emplace_back(std::move(context)); }
     void addStorageHolder(StoragePtr storage) { holder.storage_holders.emplace_back(std::move(storage)); }

src/Processors/QueryPipeline.h

@@ -101,7 +101,7 @@ public:
 
     const Block & getHeader() const { return pipe.getHeader(); }
 
-    void addTableLock(const TableLockHolder & lock) { pipe.addTableLock(lock); }
+    void addTableLock(TableLockHolder lock) { pipe.addTableLock(std::move(lock)); }
     void addInterpreterContext(std::shared_ptr<Context> context) { pipe.addInterpreterContext(std::move(context)); }
     void addStorageHolder(StoragePtr storage) { pipe.addStorageHolder(std::move(storage)); }
     void addQueryPlan(std::unique_ptr<QueryPlan> plan) { pipe.addQueryPlan(std::move(plan)); }

src/Processors/QueryPlan/ReadFromStorageStep.cpp

@@ -12,30 +12,18 @@ namespace DB
 {
 
 ReadFromStorageStep::ReadFromStorageStep(
-    TableLockHolder table_lock_,
-    StorageMetadataPtr metadata_snapshot_,
-    StreamLocalLimits & limits_,
-    SizeLimits & leaf_limits_,
-    std::shared_ptr<const EnabledQuota> quota_,
-    StoragePtr storage_,
-    const Names & required_columns_,
-    const SelectQueryInfo & query_info_,
-    std::shared_ptr<Context> context_,
-    QueryProcessingStage::Enum processing_stage_,
-    size_t max_block_size_,
-    size_t max_streams_)
-    : table_lock(std::move(table_lock_))
-    , metadata_snapshot(std::move(metadata_snapshot_))
-    , limits(limits_)
-    , leaf_limits(leaf_limits_)
-    , quota(std::move(quota_))
-    , storage(std::move(storage_))
-    , required_columns(required_columns_)
-    , query_info(query_info_)
-    , context(std::move(context_))
-    , processing_stage(processing_stage_)
-    , max_block_size(max_block_size_)
-    , max_streams(max_streams_)
+    TableLockHolder table_lock,
+    StorageMetadataPtr metadata_snapshot,
+    StreamLocalLimits & limits,
+    SizeLimits & leaf_limits,
+    std::shared_ptr<const EnabledQuota> quota,
+    StoragePtr storage,
+    const Names & required_columns,
+    const SelectQueryInfo & query_info,
+    std::shared_ptr<Context> context,
+    QueryProcessingStage::Enum processing_stage,
+    size_t max_block_size,
+    size_t max_streams)
 {
     /// Note: we read from storage in constructor of step because we don't know real header before reading.
     /// It will be fixed when storage return QueryPlanStep itself.
@@ -83,9 +71,6 @@ ReadFromStorageStep::ReadFromStorageStep(
     pipeline = std::make_unique<QueryPipeline>();
     QueryPipelineProcessorsCollector collector(*pipeline, this);
 
-    /// Table lock is stored inside pipeline here.
-    pipeline->addTableLock(table_lock);
-
     pipe.setLimits(limits);
 
     /**
@@ -103,8 +88,11 @@ ReadFromStorageStep::ReadFromStorageStep(
 
     pipeline->init(std::move(pipe));
 
+    /// Add resources to pipeline. The order is important.
+    /// Add in reverse order of destruction. Pipeline will be destroyed at the end in case of exception.
     pipeline->addInterpreterContext(std::move(context));
     pipeline->addStorageHolder(std::move(storage));
+    pipeline->addTableLock(std::move(table_lock));
 
     processors = collector.detachProcessors();
 

src/Processors/QueryPlan/ReadFromStorageStep.h

@@ -45,20 +45,6 @@ public:
     void describePipeline(FormatSettings & settings) const override;
 
 private:
-    TableLockHolder table_lock;
-    StorageMetadataPtr metadata_snapshot;
-    StreamLocalLimits limits;
-    SizeLimits leaf_limits;
-    std::shared_ptr<const EnabledQuota> quota;
-
-    StoragePtr storage;
-    const Names & required_columns;
-    const SelectQueryInfo & query_info;
-    std::shared_ptr<Context> context;
-    QueryProcessingStage::Enum processing_stage;
-    size_t max_block_size;
-    size_t max_streams;
-
     QueryPipelinePtr pipeline;
     Processors processors;
 };