Merge pull request #222 from den818/SERVER_LOGS

Receiving  server logs

Author: Enmk

clickhouse/client.cpp

@@ -443,6 +443,24 @@ bool Client::Impl::ReceivePacket(uint64_t* server_packet) {
         return false;
     }
 
+    case ServerCodes::Log: {
+        // log tag
+        if (!WireFormat::SkipString(*input_)) {
+            return false;
+        }
+        Block block;
+
+        // Use uncompressed stream since log blocks usually contain only one row
+        if (!ReadBlock(*input_, &block)) {
+            return false;
+        }
+
+        if (events_) {
+            events_->OnServerLog(block);
+        }
+        return true;
+    }
+
     case ServerCodes::TableColumns: {
         // external table name
         if (!WireFormat::SkipString(*input_)) {

clickhouse/query.h

@@ -57,6 +57,12 @@ class QueryEvents {
 
     virtual void OnProgress(const Progress& progress) = 0;
 
+    /** Handle query execution logs provided by server.
+     *  Amount of logs regulated by `send_logs_level` setting.
+     *  By-default only `fatal` log events are sent to the client side.
+     */
+    virtual void OnServerLog(const Block& block) = 0;
+
     virtual void OnFinish() = 0;
 };
 
@@ -65,6 +71,7 @@ using ExceptionCallback        = std::function<void(const Exception& e)>;
 using ProgressCallback         = std::function<void(const Progress& progress)>;
 using SelectCallback           = std::function<void(const Block& block)>;
 using SelectCancelableCallback = std::function<bool(const Block& block)>;
+using SelectServerLogCallback  = std::function<bool(const Block& block)>;
 
 
 class Query : public QueryEvents {
@@ -122,6 +129,12 @@ class Query : public QueryEvents {
         return *this;
     }
 
+    /// Set handler for receiving a server log of query exceution.
+    inline Query& OnServerLog(SelectServerLogCallback cb) {
+        select_server_log_cb_ = std::move(cb);
+        return *this;
+    }
+
     static const std::string default_query_id;
 
 private:
@@ -155,6 +168,12 @@ class Query : public QueryEvents {
         }
     }
 
+    void OnServerLog(const Block& block) override {
+        if (select_server_log_cb_) {
+            select_server_log_cb_(block);
+        }
+    }
+
     void OnFinish() override {
     }
 
@@ -166,6 +185,7 @@ class Query : public QueryEvents {
     ProgressCallback progress_cb_;
     SelectCallback select_cb_;
     SelectCancelableCallback select_cancelable_cb_;
+    SelectServerLogCallback select_server_log_cb_;
 };
 
 }

ut/client_ut.cpp

@@ -1097,6 +1097,24 @@ TEST_P(ClientCase, QuerySettings) {
     EXPECT_THROW(client_->Execute(query), ServerException);
 }
 
+TEST_P(ClientCase, ServerLogs) {
+
+    Block block;
+    createTableWithOneColumn<ColumnString>(block);
+
+    size_t received_row_count = 0;
+    Query query("INSERT INTO " + table_name + " (*) VALUES (\'Foo\'), (\'Bar\')" );
+    query.SetSetting("send_logs_level", {"trace"});
+    query.OnServerLog([&](const Block& block) {
+        received_row_count += block.GetRowCount();
+        return true;
+    });
+    client_->Execute(query);
+
+    EXPECT_GT(received_row_count, 0U);
+}
+
+
 const auto LocalHostEndpoint = ClientOptions()
         .SetHost(           getEnvOrDefault("CLICKHOUSE_HOST",     "localhost"))
         .SetPort(   getEnvOrDefault<size_t>("CLICKHOUSE_PORT",     "9000"))