Merge pull request #231 from den818/SETTINGS

Implementation per query settings

Author: Enmk

clickhouse/client.cpp

@@ -36,8 +36,9 @@
 #define DBMS_MIN_REVISION_WITH_LOW_CARDINALITY_TYPE     54405
 #define DBMS_MIN_REVISION_WITH_COLUMN_DEFAULTS_METADATA 54410
 #define DBMS_MIN_REVISION_WITH_CLIENT_WRITE_INFO        54420
+#define DBMS_MIN_REVISION_WITH_SETTINGS_SERIALIZED_AS_STRINGS 54429
 
-#define REVISION DBMS_MIN_REVISION_WITH_CLIENT_WRITE_INFO
+#define REVISION  DBMS_MIN_REVISION_WITH_SETTINGS_SERIALIZED_AS_STRINGS
 
 namespace clickhouse {
 
@@ -131,7 +132,7 @@ class Client::Impl {
 
     bool ReceivePacket(uint64_t* server_packet = nullptr);
 
-    void SendQuery(const std::string& query, const std::string& query_id);
+    void SendQuery(const Query& query);
 
     void SendData(const Block& block);
 
@@ -230,7 +231,7 @@ void Client::Impl::ExecuteQuery(Query query) {
         RetryGuard([this]() { Ping(); });
     }
 
-    SendQuery(query.GetText(), query.GetQueryID());
+    SendQuery(query);
 
     while (ReceivePacket()) {
         ;
@@ -272,7 +273,8 @@ void Client::Impl::Insert(const std::string& table_name, const std::string& quer
         }
     }
 
-    SendQuery("INSERT INTO " + table_name + " ( " + fields_section.str() + " ) VALUES", query_id);
+    Query query("INSERT INTO " + table_name + " ( " + fields_section.str() + " ) VALUES", query_id);
+    SendQuery(query);
 
     uint64_t server_packet;
     // Receive data packet.
@@ -608,9 +610,9 @@ void Client::Impl::SendCancel() {
     output_->Flush();
 }
 
-void Client::Impl::SendQuery(const std::string& query, const std::string& query_id) {
+void Client::Impl::SendQuery(const Query& query) {
     WireFormat::WriteUInt64(*output_, ClientCodes::Query);
-    WireFormat::WriteString(*output_, query_id);
+    WireFormat::WriteString(*output_, query.GetQueryID());
 
     /// Client info.
     if (server_info_.revision >= DBMS_MIN_REVISION_WITH_CLIENT_INFO) {
@@ -643,15 +645,24 @@ void Client::Impl::SendQuery(const std::string& query, const std::string& query_
         }
     }
 
-    /// Per query settings.
-    //if (settings)
-    //    settings->serialize(*out);
-    //else
+    /// Per query settings
+    if (server_info_.revision >= DBMS_MIN_REVISION_WITH_SETTINGS_SERIALIZED_AS_STRINGS) {
+        for(const auto& [name, field] : query.GetQuerySettings()) {
+            WireFormat::WriteString(*output_, name);
+            WireFormat::WriteVarint64(*output_, field.flags);
+            WireFormat::WriteString(*output_, field.value);
+        }
+    }
+    else if (query.GetQuerySettings().size() > 0) {
+        // Current implementation works only for server version >= v20.1.2.4-stable, since we do not implement binary settings serialization.
+        throw UnimplementedError(std::string("Can't send query settings to a server, server version is too old"));
+    }
+    // Empty string signals end of serialized settings
     WireFormat::WriteString(*output_, std::string());
 
     WireFormat::WriteUInt64(*output_, Stages::Complete);
     WireFormat::WriteUInt64(*output_, compression_);
-    WireFormat::WriteString(*output_, query);
+    WireFormat::WriteString(*output_, query.GetText());
     // Send empty block as marker of
     // end of data
     SendData(Block());

clickhouse/query.h

@@ -7,32 +7,21 @@
 #include <functional>
 #include <memory>
 #include <string>
+#include <unordered_map>
 
 namespace clickhouse {
 
-/**
- * Settings of individual query.
- */
-struct QuerySettings {
-    /// Maximum thread to use on the server-side to process a query. Default - let the server choose.
-    int max_threads = 0;
-    /// Compute min and max values of the result.
-    bool extremes = false;
-    /// Silently skip unavailable shards.
-    bool skip_unavailable_shards = false;
-    /// Write statistics about read rows, bytes, time elapsed, etc.
-    bool output_format_write_statistics = true;
-    /// Use client timezone for interpreting DateTime string values, instead of adopting server timezone.
-    bool use_client_time_zone = false;
-
-    // connect_timeout
-    // max_block_size
-    // distributed_group_by_no_merge = false
-    // strict_insert_defaults = 0
-    // network_compression_method = LZ4
-    // priority = 0
+struct QuerySettingsField {
+    enum Flags : uint64_t
+    {
+        IMPORTANT = 0x01,
+        CUSTOM = 0x02,
+    };
+    std::string value;
+    uint64_t flags{0};
 };
 
+using QuerySettings = std::unordered_map<std::string, QuerySettingsField>;
 
 struct Profile {
     uint64_t rows = 0;
@@ -94,6 +83,22 @@ class Query : public QueryEvents {
         return query_id_;
     }
 
+    inline const QuerySettings& GetQuerySettings() const {
+        return query_settings_;
+    }
+
+    /// Set per query settings
+    inline Query& SetQuerySettings(QuerySettings query_settings) {
+        query_settings_ = std::move(query_settings);
+        return *this;
+    }
+
+    /// Set per query setting
+    inline Query& SetSetting(const std::string& key, const QuerySettingsField& value) {
+        query_settings_[key] = value;
+        return *this;
+    }
+
     /// Set handler for receiving result data.
     inline Query& OnData(SelectCallback cb) {
         select_cb_ = std::move(cb);
@@ -111,7 +116,6 @@ class Query : public QueryEvents {
         return *this;
     }
 
-
     /// Set handler for receiving a progress of query exceution.
     inline Query& OnProgress(ProgressCallback cb) {
         progress_cb_ = std::move(cb);
@@ -157,6 +161,7 @@ class Query : public QueryEvents {
 private:
     const std::string query_;
     const std::string query_id_;
+    QuerySettings query_settings_;
     ExceptionCallback exception_cb_;
     ProgressCallback progress_cb_;
     SelectCallback select_cb_;

ut/client_ut.cpp

@@ -1039,6 +1039,64 @@ TEST_P(ClientCase, OnProgress) {
     EXPECT_LE(received_progress->written_bytes, 10000u);
 }
 
+TEST_P(ClientCase, QuerySettings) {
+    client_->Execute("DROP TEMPORARY TABLE IF EXISTS test_clickhouse_query_settings_table_1;");
+    client_->Execute("CREATE TEMPORARY TABLE IF NOT EXISTS test_clickhouse_query_settings_table_1 ( id  Int64 )");
+
+    client_->Execute("DROP TEMPORARY TABLE IF EXISTS test_clickhouse_query_settings_table_2;");
+    client_->Execute("CREATE TEMPORARY TABLE IF NOT EXISTS test_clickhouse_query_settings_table_2 ( id  Int64, value Int64 )");
+
+    client_->Execute("INSERT INTO test_clickhouse_query_settings_table_1 (*) VALUES (1)");
+
+    Query query("SELECT value "
+                "FROM test_clickhouse_query_settings_table_1 "
+                "LEFT OUTER JOIN test_clickhouse_query_settings_table_2 "
+                "ON test_clickhouse_query_settings_table_1.id = test_clickhouse_query_settings_table_2.id");
+
+
+    bool checked = false;
+
+    query.SetSetting("join_use_nulls", {"1"});
+
+    query.OnData(
+        [&](const Block& block) {
+            if (block.GetRowCount() == 0)
+                return;
+            ASSERT_EQ(1U, block.GetColumnCount());
+            ASSERT_EQ(1U, block.GetRowCount());
+            ASSERT_TRUE(block[0]->GetType().IsEqual(Type::CreateNullable(Type::CreateSimple<int64_t>())));
+            auto cl = block[0]->As<ColumnNullable>();
+            EXPECT_TRUE(cl->IsNull(0));
+            checked = true;
+        });
+    client_->Execute(query);
+
+    EXPECT_TRUE(checked);
+
+    query.SetSetting("join_use_nulls", {"0"});
+
+    query.OnData(
+        [&](const Block& block) {
+            if (block.GetRowCount() == 0)
+                return;
+            ASSERT_EQ(1U, block.GetColumnCount());
+            ASSERT_EQ(1U, block.GetRowCount());
+            ASSERT_TRUE(block[0]->GetType().IsEqual(Type::CreateSimple<int64_t>()));
+            auto cl = block[0]->As<ColumnInt64>();
+            EXPECT_EQ(cl->At(0), 0);
+            checked = true;
+        }
+    );
+    checked = false;
+    client_->Execute(query);
+
+    EXPECT_TRUE(checked);
+
+    query.SetSetting("wrong_setting_name", {"0", QuerySettingsField::IMPORTANT});
+
+    EXPECT_THROW(client_->Execute(query), ServerException);
+}
+
 const auto LocalHostEndpoint = ClientOptions()
         .SetHost(           getEnvOrDefault("CLICKHOUSE_HOST",     "localhost"))
         .SetPort(   getEnvOrDefault<size_t>("CLICKHOUSE_PORT",     "9000"))