Merge pull request #441 from ilandry/clickhouse_client_move

clickhouse client move support

Author: slabko

clickhouse/client.cpp

@@ -446,7 +446,7 @@ void Client::Impl::SendBlockData(const Block& block) {
     if (compression_ == CompressionState::Enable) {
         std::unique_ptr<OutputStream> compressed_output = std::make_unique<CompressedOutput>(output_.get(), options_.max_compression_chunk_size, options_.compression_method);
         BufferedOutput buffered(std::move(compressed_output), options_.max_compression_chunk_size);
-    
+
         WriteBlock(block, buffered);
     } else {
         WriteBlock(block, *output_);
@@ -902,7 +902,7 @@ bool Client::Impl::ReadBlock(InputStream& input, Block* block) {
         if (!WireFormat::ReadString(input, &type)) {
             return false;
         }
-    
+
         if (server_info_.revision >= DBMS_MIN_REVISION_WITH_CUSTOM_SERIALIZATION) {
             uint8_t custom_format_len;
             if (!WireFormat::ReadFixed(input, &custom_format_len)) {
@@ -911,7 +911,7 @@ bool Client::Impl::ReadBlock(InputStream& input, Block* block) {
             if (custom_format_len > 0) {
                 throw UnimplementedError(std::string("unsupported custom serialization"));
             }
-        }  
+        }
 
         if (ColumnRef col = CreateColumnByType(type, create_column_settings)) {
             if (num_rows && !col->Load(&input, num_rows)) {
@@ -1101,7 +1101,7 @@ void Client::Impl::SendQuery(const Query& query, bool finalize) {
         }
         WireFormat::WriteString(*output_, std::string()); // empty string after last param
     }
- 
+
     if (finalize) {
         FinalizeQuery();
     }
@@ -1277,21 +1277,23 @@ void Client::Impl::RetryGuard(std::function<void()> func) {
 }
 
 Client::Client(const ClientOptions& opts)
-    : options_(opts)
-    , impl_(new Impl(opts))
+    : impl_(new Impl(opts))
 {
 }
 
 Client::Client(const ClientOptions& opts,
                std::unique_ptr<SocketFactory> socket_factory)
-    : options_(opts)
-    , impl_(new Impl(opts, std::move(socket_factory)))
+    : impl_(new Impl(opts, std::move(socket_factory)))
 {
 }
 
 Client::~Client()
 { }
 
+
+Client::Client(Client&&) noexcept = default;
+Client& Client::operator=(Client&&) noexcept = default;
+
 void Client::Execute(const Query& query) {
     impl_->ExecuteQuery(query);
 }

clickhouse/client.h

@@ -246,6 +246,12 @@ class Client {
             std::unique_ptr<SocketFactory> socket_factory);
     ~Client();
 
+    // movable only
+    Client(Client&&) noexcept;
+    Client& operator=(Client&&) noexcept;
+    Client(const Client&) = delete;
+    Client& operator=(const Client&) = delete;
+
     /// Intends for execute arbitrary queries.
     void Execute(const Query& query);
 
@@ -346,8 +352,6 @@ class Client {
     static Version GetVersion();
 
 private:
-    const ClientOptions options_;
-
     class Impl;
     std::unique_ptr<Impl> impl_;
 };

ut/client_ut.cpp

@@ -1998,7 +1998,7 @@ TEST_P(ClientCase, ClientName) {
 
     FlushLogs();
 
-    std::string query_log_query 
+    std::string query_log_query
         = "SELECT CAST(client_name, 'String') FROM system.query_log WHERE query_id = '" + query_id + "'";
 
     size_t total_rows = 0;
@@ -2011,3 +2011,54 @@ TEST_P(ClientCase, ClientName) {
     });
     ASSERT_GT(total_rows, 0UL) << "Query with query_id " << query_id << " is not found";
 }
+
+TEST_P(ClientCase, ClientMoveConstructor) {
+    Client client{std::move(*client_.get())};
+
+    size_t total_rows = 0;
+    client.Select("SELECT 'foobar'", [&total_rows](const Block& block) {
+        total_rows += block.GetRowCount();
+        if (block.GetRowCount() == 0) {
+            return;
+        }
+
+        ASSERT_EQ(1U, block.GetColumnCount());
+        auto col = block[0]->As<ColumnString>();
+        ASSERT_TRUE(col);
+        ASSERT_EQ(1U, col->Size());
+        EXPECT_EQ("foobar", (*col)[0]);
+    });
+    ASSERT_EQ(total_rows, 1U);
+}
+
+TEST_P(ClientCase, ClientMoveAssign) {
+    client_->Execute("SET param_session_id = 'initial'");
+
+    ClientOptions opt = GetParam();
+    Client client{opt};
+    client.Execute("SET param_session_id = 'new'");
+
+    size_t total_rows = 0;
+    client.Select("SELECT {session_id:String}", [&total_rows](const Block& block) {
+        total_rows += block.GetRowCount();
+        for (size_t i = 0; i < block.GetRowCount(); ++i) {
+            EXPECT_EQ("new", block[0]->AsStrict<ColumnString>()->At(i));
+
+        }
+    });
+    ASSERT_EQ(total_rows, 1U);
+
+    client = std::move(*client_.get());
+
+    total_rows = 0;
+    client.Select("SELECT {session_id:String}", [&total_rows](const Block& block) {
+        total_rows += block.GetRowCount();
+        for (size_t i = 0; i < block.GetRowCount(); ++i) {
+            EXPECT_EQ("initial", block[0]->AsStrict<ColumnString>()->At(i));
+            //         ^
+            // The value has changed to the session id of the original client
+        }
+    });
+    ASSERT_EQ(total_rows, 1U);
+
+}