Throw exception if server doesn't commited insert

Author: alesapin

clickhouse/client.cpp

@@ -248,9 +248,16 @@ void Client::Impl::Insert(const std::string& table_name, const Block& block) {
     SendData(Block());
 
     // Wait for EOS.
-    while (ReceivePacket()) {
+    uint64_t eos_packet{0};
+    while (ReceivePacket(&eos_packet)) {
         ;
     }
+
+    if (eos_packet != ServerCodes::EndOfStream && eos_packet != ServerCodes::Exception
+        && eos_packet != ServerCodes::Log && options_.rethrow_exceptions) {
+        throw std::runtime_error(std::string{"unexpected packet from server while receiving end of query, expected (expected Exception, EndOfStream or Log, got: "}
+                            + (eos_packet ? std::to_string(eos_packet) : "nothing") + ")");
+    }
 }
 
 void Client::Impl::Ping() {
@@ -495,23 +502,29 @@ bool Client::Impl::ReceiveException(bool rethrow) {
     std::unique_ptr<Exception> e(new Exception);
     Exception* current = e.get();
 
+    bool exception_received = true;
     do {
         bool has_nested = false;
 
         if (!WireFormat::ReadFixed(&input_, &current->code)) {
-            return false;
+           exception_received = false;
+           break;
         }
         if (!WireFormat::ReadString(&input_, &current->name)) {
-            return false;
+            exception_received = false;
+            break;
         }
         if (!WireFormat::ReadString(&input_, &current->display_text)) {
-            return false;
+            exception_received = false;
+            break;
         }
         if (!WireFormat::ReadString(&input_, &current->stack_trace)) {
-            return false;
+            exception_received = false;
+            break;
         }
         if (!WireFormat::ReadFixed(&input_, &has_nested)) {
-            return false;
+            exception_received = false;
+            break;
         }
 
         if (has_nested) {
@@ -530,7 +543,7 @@ bool Client::Impl::ReceiveException(bool rethrow) {
         throw ServerException(std::move(e));
     }
 
-    return true;
+    return exception_received;
 }
 
 void Client::Impl::SendCancel() {

clickhouse/protocol.h

@@ -5,15 +5,17 @@ namespace clickhouse {
     /// То, что передаёт сервер.
     namespace ServerCodes {
         enum {
-            Hello       = 0,    /// Имя, версия, ревизия.
-            Data        = 1,    /// Блок данных со сжатием или без.
-            Exception   = 2,    /// Исключение во время обработки запроса.
-            Progress    = 3,    /// Прогресс выполнения запроса: строк считано, байт считано.
-            Pong        = 4,    /// Ответ на Ping.
-            EndOfStream = 5,    /// Все пакеты были переданы.
-            ProfileInfo = 6,    /// Пакет с профайлинговой информацией.
-            Totals      = 7,    /// Блок данных с тотальными значениями, со сжатием или без.
-            Extremes    = 8,    /// Блок данных с минимумами и максимумами, аналогично.
+            Hello                = 0,    /// Имя, версия, ревизия.
+            Data                 = 1,    /// Блок данных со сжатием или без.
+            Exception            = 2,    /// Исключение во время обработки запроса.
+            Progress             = 3,    /// Прогресс выполнения запроса: строк считано, байт считано.
+            Pong                 = 4,    /// Ответ на Ping.
+            EndOfStream          = 5,    /// Все пакеты были переданы.
+            ProfileInfo          = 6,    /// Пакет с профайлинговой информацией.
+            Totals               = 7,    /// Блок данных с тотальными значениями, со сжатием или без.
+            Extremes             = 8,    /// Блок данных с минимумами и максимумами, аналогично.
+            TablesStatusResponse = 9,    /// Ответ на запрос TableStatus.
+            Log                  = 10,   /// Системный лог исполнения запроса.
         };
     }