grpc/clickhouse: construct query from options

Signed-off-by: Attila Szakacs <[email protected]>

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 */
 

modules/grpc/clickhouse/clickhouse-dest.hpp

@@ -82,13 +82,19 @@ class DestDriver final : public syslogng::grpc::DestDriver
     return this->password;
   }
 
+  const std::string &get_query()
+  {
+    return this->query;
+  }
+
   Schema *get_schema()
   {
     return &this->schema;
   }
 
 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;
 };