thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-09-20 00:30:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #68335 from ClickHouse/backport/24.3/68288

Browse Source 
Backport #68288 to 24.3: Fix postgres crash
This commit is contained in:
Kseniia Sumarokova 2024-08-14 19:05:15 +02:00 committed by GitHub
commit db97a5aca6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 15 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/Processors/Sources/PostgreSQLSource.cpp
View File

@ -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);
}
@ -196,15 +196,15 @@ PostgreSQLSource<T>::~PostgreSQLSource()
tx->conn().cancel_query();
stream->close();
}
stream.reset();
tx.reset();
}
catch (...)
{
tryLogCurrentException(__PRETTY_FUNCTION__);
}
stream.reset();
tx.reset();
if (connection_holder)
connection_holder->setBroken();
}

14
src/Processors/Sources/PostgreSQLSource.h
View File

@ -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;
};