Merge pull request #31 from pakapor/bugfix-insertSpaceColumn

Fix Insert into column name with space

Author: Enmk

clickhouse/client.cpp

@@ -201,6 +201,25 @@ void Client::Impl::ExecuteQuery(Query query) {
     }
 }
 
+std::string NameToQueryString(const std::string &input)
+{
+    std::string output;
+    output.reserve(input.size() + 2);
+    output += '`';
+
+    for (const auto & c : input) {
+        if (c == '`') {
+            //escape ` with ``
+            output.append("``");
+        } else {
+            output.push_back(c);
+        }
+    }
+
+    output += '`';
+    return output;
+}
+
 void Client::Impl::Insert(const std::string& table_name, const Block& block) {
     if (options_.ping_before_query) {
         RetryGuard([this]() { Ping(); });
@@ -211,7 +230,7 @@ void Client::Impl::Insert(const std::string& table_name, const Block& block) {
 
     // Enumerate all fields
     for (unsigned int i = 0; i < block.GetColumnCount(); i++) {
-        fields.push_back(block.GetColumnName(i));
+        fields.push_back(NameToQueryString(block.GetColumnName(i)));
     }
 
     std::stringstream fields_section;

ut/client_ut.cpp

@@ -716,6 +716,54 @@ TEST_P(ClientCase, Decimal) {
     });
 }
 
+// Test special chars in names
+TEST_P(ClientCase, ColEscapeNameTest) {
+    client_->Execute(R"sql(DROP TABLE IF EXISTS test_clickhouse_cpp."col_escape_""name_test";)sql");
+
+    client_->Execute(R"sql(CREATE TABLE IF NOT EXISTS test_clickhouse_cpp."col_escape_""name_test" ("test space" UInt64, "test "" quote" UInt64, "test ""`'[]&_\ all" UInt64) ENGINE = Memory)sql");
+
+    auto col1 = std::make_shared<ColumnUInt64>();
+    col1->Append(1);
+    col1->Append(2);
+    auto col2 = std::make_shared<ColumnUInt64>();
+    col2->Append(4);
+    col2->Append(8);
+    auto col3 = std::make_shared<ColumnUInt64>();
+    col3->Append(16);
+    col3->Append(32);
+
+    static const std::string column_names[] = {
+        "test space",
+        R"sql(test " quote)sql",
+        R"sql(test "`'[]&_\ all)sql"
+    };
+    static const auto columns_count = sizeof(column_names)/sizeof(column_names[0]);
+
+    Block block;
+    block.AppendColumn(column_names[0], col1);
+    block.AppendColumn(column_names[1], col2);
+    block.AppendColumn(column_names[2], col3);
+
+    client_->Insert(R"sql(test_clickhouse_cpp."col_escape_""name_test")sql", block);
+    client_->Select(R"sql(SELECT * FROM test_clickhouse_cpp."col_escape_""name_test")sql", [] (const Block& sblock)
+    {
+        int row = sblock.GetRowCount();
+        if (row <= 0) {return;}
+        ASSERT_EQ(columns_count, sblock.GetColumnCount());
+        for (size_t i = 0; i < columns_count; ++i) {
+            EXPECT_EQ(column_names[i], sblock.GetColumnName(i));
+        }
+
+        EXPECT_EQ(row, 2);
+        EXPECT_EQ(sblock[0]->As<ColumnUInt64>()->At(0), 1u);
+        EXPECT_EQ(sblock[0]->As<ColumnUInt64>()->At(1), 2u);
+        EXPECT_EQ(sblock[1]->As<ColumnUInt64>()->At(0), 4u);
+        EXPECT_EQ(sblock[1]->As<ColumnUInt64>()->At(1), 8u);
+        EXPECT_EQ(sblock[2]->As<ColumnUInt64>()->At(0), 16u);
+        EXPECT_EQ(sblock[2]->As<ColumnUInt64>()->At(1), 32u);
+    });
+}
+
 // Test roundtrip of DateTime64 values
 TEST_P(ClientCase, DateTime64) {
     const auto & server_info = client_->GetServerInfo();