Merge pull request #36587 from azat/client-reconnect

client: add a message on reconnect (under warning log level)

src/Client/ClientBase.cpp

@@ -1234,6 +1234,7 @@ void ClientBase::sendDataFrom(ReadBuffer & buf, Block & sample, const ColumnsDes
 }
 
 void ClientBase::sendDataFromPipe(Pipe&& pipe, ASTPtr parsed_query, bool have_more_data)
+try
 {
     QueryPipeline pipeline(std::move(pipe));
     PullingAsyncPipelineExecutor executor(pipeline);
@@ -1266,6 +1267,12 @@ void ClientBase::sendDataFromPipe(Pipe&& pipe, ASTPtr parsed_query, bool have_mo
     if (!have_more_data)
         connection->sendData({}, "", false);
 }
+catch (...)
+{
+    connection->sendCancel();
+    receiveEndOfQuery();
+    throw;
+}
 
 void ClientBase::sendDataFromStdin(Block & sample, const ColumnsDescription & columns_description, ASTPtr parsed_query)
 {
@@ -1406,7 +1413,15 @@ void ClientBase::processParsedSingleQuery(const String & full_query, const Strin
             apply_query_settings(*with_output->settings_ast);
 
         if (!connection->checkConnected())
+        {
+            auto poco_logs_level = Poco::Logger::parseLevel(config().getString("send_logs_level", "none"));
+            /// Print under WARNING also because it is used by clickhouse-test.
+            if (poco_logs_level >= Poco::Message::PRIO_WARNING)
+            {
+                fmt::print(stderr, "Connection lost. Reconnecting.\n");
+            }
             connect();
+        }
 
         ASTPtr input_function;
         if (insert && insert->select)

src/Client/Connection.cpp

@@ -377,9 +377,10 @@ bool Connection::ping()
 {
     // LOG_TRACE(log_wrapper.get(), "Ping");
 
-    TimeoutSetter timeout_setter(*socket, sync_request_timeout, true);
     try
     {
+        TimeoutSetter timeout_setter(*socket, sync_request_timeout, true);
+
         UInt64 pong = 0;
         writeVarUInt(Protocol::Client::Ping, *out);
         out->next();
@@ -405,6 +406,10 @@ bool Connection::ping()
     }
     catch (const Poco::Exception & e)
     {
+        /// Explicitly disconnect since ping() can receive EndOfStream,
+        /// and in this case this ping() will return false,
+        /// while next ping() may return true.
+        disconnect();
         LOG_TRACE(log_wrapper.get(), fmt::runtime(e.displayText()));
         return false;
     }

tests/queries/0_stateless/01173_transaction_control_queries.sql

@@ -4,7 +4,7 @@ drop table if exists mt2;
 create table mt1 (n Int64) engine=MergeTree order by n;
 create table mt2 (n Int64) engine=MergeTree order by n;
 
-commit; -- { serverError INVALID_TRANSACTION }
+commit; -- { serverError INVALID_TRANSACTION } -- no transaction
 rollback; -- { serverError INVALID_TRANSACTION }
 
 begin transaction;
@@ -31,7 +31,7 @@ select 'on exception before start', arraySort(groupArray(n)) from (select n from
 -- rollback on exception before start
 select functionThatDoesNotExist(); -- { serverError 46 }
 -- cannot commit after exception
-commit; -- { serverError INVALID_TRANSACTION }
+commit; -- { serverError INVALID_TRANSACTION } -- after 46
 begin transaction; -- { serverError INVALID_TRANSACTION }
 rollback;
 
@@ -42,7 +42,7 @@ select 'on exception while processing', arraySort(groupArray(n)) from (select n
 -- rollback on exception while processing
 select throwIf(100 < number) from numbers(1000); -- { serverError 395 }
 -- cannot commit after exception
-commit; -- { serverError INVALID_TRANSACTION }
+commit; -- { serverError INVALID_TRANSACTION } -- after 395
 insert into mt1 values (5); -- { serverError INVALID_TRANSACTION }
 insert into mt2 values (50); -- { serverError INVALID_TRANSACTION }
 select 1; -- { serverError INVALID_TRANSACTION }
@@ -52,10 +52,9 @@ begin transaction;
 insert into mt1 values (6);
 insert into mt2 values (60);
 select 'on session close', arraySort(groupArray(n)) from (select n from mt1 union all select * from mt2);
--- trigger reconnection by error on client, check rollback on session close
 insert into mt1 values ([1]); -- { clientError 43 }
-commit; -- { serverError INVALID_TRANSACTION }
-rollback; -- { serverError INVALID_TRANSACTION }
+-- INSERT failures does not produce client reconnect anymore, so rollback can be done
+rollback;
 
 begin transaction;
 insert into mt1 values (7);
@@ -82,19 +81,19 @@ rollback;
 
 begin transaction;
 create table m (n int) engine=Memory; -- { serverError 48 }
-commit; -- { serverError INVALID_TRANSACTION }
+commit; -- { serverError INVALID_TRANSACTION } -- after 48
 rollback;
 
 create table m (n int) engine=Memory;
 begin transaction;
 insert into m values (1); -- { serverError 48 }
 select * from m; -- { serverError INVALID_TRANSACTION }
-commit; -- { serverError INVALID_TRANSACTION }
+commit; -- { serverError INVALID_TRANSACTION } -- after 48
 rollback;
 
 begin transaction;
 select * from m; -- { serverError 48 }
-commit; -- { serverError INVALID_TRANSACTION }
+commit; -- { serverError INVALID_TRANSACTION } -- after 48
 rollback;
 
 drop table m;