From 90fa08c329528555442e7024afa28b6781aae3d7 Mon Sep 17 00:00:00 2001
From: Attila Szakacs <attila.szakacs@axoflow.com>
Date: Sun, 13 Oct 2024 00:14:05 +0200
Subject: [PATCH] grpc/clickhouse: construct query from options

Signed-off-by: Attila Szakacs <attila.szakacs@axoflow.com>
---
 modules/grpc/clickhouse/clickhouse-dest.cpp | 27 +++++++++++++++++++++
 modules/grpc/clickhouse/clickhouse-dest.hpp |  7 ++++++
 2 files changed, 34 insertions(+)

diff --git a/modules/grpc/clickhouse/clickhouse-dest.cpp b/modules/grpc/clickhouse/clickhouse-dest.cpp
index 04ba4d560..a9f50ff48 100644
--- a/modules/grpc/clickhouse/clickhouse-dest.cpp
+++ b/modules/grpc/clickhouse/clickhouse-dest.cpp
@@ -52,6 +52,12 @@ DestDriver::init()
       return false;
     }
 
+  std::string quoted_table;
+  if (!this->quote_identifier(this->table, quoted_table))
+    return false;
+
+  this->query = "INSERT INTO " + quoted_table + " FORMAT Protobuf";
+
   if (!this->schema.init())
     return false;
 
@@ -269,6 +275,27 @@ DestDriver::map_schema_type(const std::string &type_in, google::protobuf::FieldD
   return true;
 }
 
+bool
+DestDriver::quote_identifier(const std::string &identifier, std::string &quoted_identifier)
+{
+  /* https://clickhouse.com/docs/en/sql-reference/syntax#identifiers */
+
+  bool has_backtick = identifier.find('`') != std::string::npos;
+  bool has_double_quotes = identifier.find('"') != std::string::npos;
+
+  if (has_backtick && has_double_quotes)
+    {
+      msg_error("Error quoting identifier, identifier contains both backtick and double quotes",
+                log_pipe_location_tag(&this->super->super.super.super.super),
+                evt_tag_str("identifier", identifier.c_str()));
+      return false;
+    }
+
+  char quote_char = has_backtick ? '"' : '`';
+  quoted_identifier.assign(quote_char + identifier + quote_char);
+  return true;
+}
+
 
 /* C Wrappers */
 
diff --git a/modules/grpc/clickhouse/clickhouse-dest.hpp b/modules/grpc/clickhouse/clickhouse-dest.hpp
index b787b9891..54d974105 100644
--- a/modules/grpc/clickhouse/clickhouse-dest.hpp
+++ b/modules/grpc/clickhouse/clickhouse-dest.hpp
@@ -82,6 +82,11 @@ class DestDriver final : public syslogng::grpc::DestDriver
     return this->password;
   }
 
+  const std::string &get_query()
+  {
+    return this->query;
+  }
+
   Schema *get_schema()
   {
     return &this->schema;
@@ -89,6 +94,7 @@ class DestDriver final : public syslogng::grpc::DestDriver
 
 private:
   static bool map_schema_type(const std::string &type_in, google::protobuf::FieldDescriptorProto::Type &type_out);
+  bool quote_identifier(const std::string &identifier, std::string &quoted_identifier);
 
 private:
   friend class DestWorker;
@@ -98,6 +104,7 @@ class DestDriver final : public syslogng::grpc::DestDriver
   std::string table;
   std::string user;
   std::string password;
+  std::string query;
 
   Schema schema;
 };