From 5a6090ad05117c76a4b37071a6362f30f395b235 Mon Sep 17 00:00:00 2001
From: kssenii <sumarokovakseniia@mail.ru>
Date: Tue, 13 Aug 2024 16:25:07 +0200
Subject: [PATCH] Fix

---
 src/Processors/Sources/PostgreSQLSource.cpp | 12 ++++++------
 src/Processors/Sources/PostgreSQLSource.h   | 14 +++++++++-----
 2 files changed, 15 insertions(+), 11 deletions(-)

diff --git a/src/Processors/Sources/PostgreSQLSource.cpp b/src/Processors/Sources/PostgreSQLSource.cpp
index a3d6fd691d8..b9bda46bd10 100644
--- a/src/Processors/Sources/PostgreSQLSource.cpp
+++ b/src/Processors/Sources/PostgreSQLSource.cpp
@@ -35,9 +35,9 @@ PostgreSQLSource<T>::PostgreSQLSource(
     const Block & sample_block,
     UInt64 max_block_size_)
     : ISource(sample_block.cloneEmpty())
-    , query_str(query_str_)
     , max_block_size(max_block_size_)
     , connection_holder(std::move(connection_holder_))
+    , query_str(query_str_)
 {
     init(sample_block);
 }
@@ -51,10 +51,10 @@ PostgreSQLSource<T>::PostgreSQLSource(
     UInt64 max_block_size_,
     bool auto_commit_)
     : ISource(sample_block.cloneEmpty())
-    , query_str(query_str_)
-    , tx(std::move(tx_))
     , max_block_size(max_block_size_)
     , auto_commit(auto_commit_)
+    , query_str(query_str_)
+    , tx(std::move(tx_))
 {
     init(sample_block);
 }
@@ -204,15 +204,15 @@ PostgreSQLSource<T>::~PostgreSQLSource()
                   */
                 stream->close();
             }
-
-            stream.reset();
-            tx.reset();
         }
         catch (...)
         {
             tryLogCurrentException(__PRETTY_FUNCTION__);
         }
 
+        stream.reset();
+        tx.reset();
+
         if (connection_holder)
             connection_holder->setBroken();
     }
diff --git a/src/Processors/Sources/PostgreSQLSource.h b/src/Processors/Sources/PostgreSQLSource.h
index 8a648ae8bb5..319c5d8d7c2 100644
--- a/src/Processors/Sources/PostgreSQLSource.h
+++ b/src/Processors/Sources/PostgreSQLSource.h
@@ -38,14 +38,12 @@ protected:
         UInt64 max_block_size_,
         bool auto_commit_);
 
-    String query_str;
-    std::shared_ptr<T> tx;
-    std::unique_ptr<pqxx::stream_from> stream;
-
     Status prepare() override;
 
-    void onStart();
     Chunk generate() override;
+
+    void onStart();
+
     void onFinish();
 
 private:
@@ -61,6 +59,12 @@ private:
     postgres::ConnectionHolderPtr connection_holder;
 
     std::unordered_map<size_t, PostgreSQLArrayInfo> array_info;
+
+protected:
+    String query_str;
+    /// tx and stream must be destroyed before connection_holder.
+    std::shared_ptr<T> tx;
+    std::unique_ptr<pqxx::stream_from> stream;
 };