Merge pull request #22208 from azat/s3-writer

src/IO/WriteBufferFromS3.cpp

@@ -112,18 +112,6 @@ void WriteBufferFromS3::finalizeImpl()
     finalized = true;
 }
 
-WriteBufferFromS3::~WriteBufferFromS3()
-{
-    try
-    {
-        finalizeImpl();
-    }
-    catch (...)
-    {
-        tryLogCurrentException(log);
-    }
-}
-
 void WriteBufferFromS3::createMultipartUpload()
 {
     Aws::S3::Model::CreateMultipartUploadRequest req;

src/IO/WriteBufferFromS3.h

@@ -64,8 +64,6 @@ public:
     /// Receives response from the server after sending all data.
     void finalize() override;
 
-    ~WriteBufferFromS3() override;
-
 private:
     bool finalized = false;
 

src/Server/TCPHandler.cpp

@@ -570,7 +570,16 @@ void TCPHandler::processInsertQuery(const Settings & connection_settings)
     /// Send block to the client - table structure.
     sendData(state.io.out->getHeader());
 
-    readData(connection_settings);
+    try
+    {
+        readData(connection_settings);
+    }
+    catch (...)
+    {
+        /// To avoid flushing from the destructor, that may lead to uncaught exception.
+        state.io.out->writeSuffix();
+        throw;
+    }
     state.io.out->writeSuffix();
 }
 

src/Storages/StorageS3.cpp

@@ -296,9 +296,18 @@ public:
 
     void writeSuffix() override
     {
-        writer->writeSuffix();
-        writer->flush();
-        write_buf->finalize();
+        try
+        {
+            writer->writeSuffix();
+            writer->flush();
+            write_buf->finalize();
+        }
+        catch (...)
+        {
+            /// Stop ParallelFormattingOutputFormat correctly.
+            writer.reset();
+            throw;
+        }
     }
 
 private:

tests/queries/0_stateless/01832_memory_write_suffix.reference

@@ -0,0 +1 @@
+1

tests/queries/0_stateless/01832_memory_write_suffix.sql

@@ -0,0 +1,7 @@
+drop table if exists data_01832;
+
+-- Memory writes from the writeSuffix() and if it will be called twice two rows
+-- will be written (since it does not reset the block).
+create table data_01832 (key Int) Engine=Memory;
+insert into data_01832 values (1);
+select * from data_01832;