[carnot] Add ClickHouse source and export-sink plan operators

Author: ddelnano

bazel/container_images.bzl

@@ -14,7 +14,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-load("@io_bazel_rules_docker//container:container.bzl", "container_pull")
+load("@io_bazel_rules_docker//container:container.bzl", "container_pull", "container_image", "container_layer")
 
 # When adding an image here, first add it to scripts/regclient/regbot_deps.yaml
 # Once that is in, trigger the github workflow that mirrors the required image
@@ -367,3 +367,12 @@ def stirling_test_images():
         repository = "golang_1_22_grpc_server_with_buildinfo",
         digest = "sha256:67adba5e8513670fa37bd042862e7844f26239e8d2997ed8c3b0aa527bc04cc3",
     )
+
+    # ClickHouse server image for testing.
+    # clickhouse/clickhouse-server:25.7-alpine
+    container_pull(
+        name = "clickhouse_server_base_image",
+        registry = "docker.io",
+        repository = "clickhouse/clickhouse-server",
+        digest = "sha256:60c53a520a1caad6555eb6772a8a9c91bb09774c1c7ec87e3371ea3da254eeab",
+    )

src/carnot/carnot.cc

@@ -181,6 +181,8 @@ Status CarnotImpl::RegisterUDFsInPlanFragment(exec::ExecState* exec_state, plan:
       .OnUDTFSource(no_op)
       .OnEmptySource(no_op)
       .OnOTelSink(no_op)
+      .OnClickHouseSource(no_op)
+      .OnClickHouseExportSink(no_op)
       .Walk(pf);
 }
 

src/carnot/exec/BUILD.bazel

@@ -46,6 +46,7 @@ pl_cc_library(
         "//src/shared/types:cc_library",
         "//src/table_store/table:cc_library",
         "@com_github_apache_arrow//:arrow",
+        "@com_github_clickhouse_clickhouse_cpp//:clickhouse_cpp",
         "@com_github_grpc_grpc//:grpc++",
         "@com_github_opentelemetry_proto//:logs_service_grpc_cc",
         "@com_github_opentelemetry_proto//:metrics_service_grpc_cc",
@@ -300,3 +301,46 @@ pl_cc_test(
         "@com_github_grpc_grpc//:grpc++_test",
     ],
 )
+
+pl_cc_test(
+    name = "clickhouse_source_node_test",
+    timeout = "long",
+    srcs = ["clickhouse_source_node_test.cc"],
+    data = [
+        "//src/stirling/source_connectors/socket_tracer/testing/container_images/clickhouse",
+    ],
+    tags = [
+        "exclusive",
+        "requires_bpf",
+    ],
+    deps = [
+        ":cc_library",
+        ":exec_node_test_helpers",
+        ":test_utils",
+        "//src/carnot/planpb:plan_testutils",
+        "//src/common/testing/test_utils:cc_library",
+        "@com_github_clickhouse_clickhouse_cpp//:clickhouse_cpp",
+    ],
+)
+
+pl_cc_test(
+    name = "clickhouse_export_sink_node_test",
+    timeout = "long",
+    srcs = ["clickhouse_export_sink_node_test.cc"],
+    data = [
+        "//src/stirling/source_connectors/socket_tracer/testing/container_images/clickhouse",
+    ],
+    tags = [
+        "exclusive",
+        "requires_bpf",
+    ],
+    deps = [
+        ":cc_library",
+        ":exec_node_test_helpers",
+        ":test_utils",
+        "//src/carnot/plan:cc_library",
+        "//src/carnot/planpb:plan_pl_cc_proto",
+        "//src/common/testing/test_utils:cc_library",
+        "@com_github_clickhouse_clickhouse_cpp//:clickhouse_cpp",
+    ],
+)

src/carnot/exec/exec_graph.cc

@@ -24,6 +24,8 @@
 #include <unordered_map>
 
 #include "src/carnot/exec/agg_node.h"
+#include "src/carnot/exec/clickhouse_export_sink_node.h"
+#include "src/carnot/exec/clickhouse_source_node.h"
 #include "src/carnot/exec/empty_source_node.h"
 #include "src/carnot/exec/equijoin_node.h"
 #include "src/carnot/exec/exec_node.h"
@@ -108,6 +110,14 @@ Status ExecutionGraph::Init(table_store::schema::Schema* schema, plan::PlanState
       .OnOTelSink([&](auto& node) {
         return OnOperatorImpl<plan::OTelExportSinkOperator, OTelExportSinkNode>(node, &descriptors);
       })
+      .OnClickHouseSource([&](auto& node) {
+        return OnOperatorImpl<plan::ClickHouseSourceOperator, ClickHouseSourceNode>(node,
+                                                                                    &descriptors);
+      })
+      .OnClickHouseExportSink([&](auto& node) {
+        return OnOperatorImpl<plan::ClickHouseExportSinkOperator, ClickHouseExportSinkNode>(node,
+                                                                                             &descriptors);
+      })
       .Walk(pf_);
 }
 

src/carnot/planner/compiler_state/compiler_state.h

@@ -119,7 +119,8 @@ class CompilerState : public NotCopyable {
                 int64_t max_output_rows_per_table, std::string_view result_address,
                 std::string_view result_ssl_targetname, const RedactionOptions& redaction_options,
                 std::unique_ptr<planpb::OTelEndpointConfig> endpoint_config,
-                std::unique_ptr<PluginConfig> plugin_config, DebugInfo debug_info)
+                std::unique_ptr<PluginConfig> plugin_config, DebugInfo debug_info,
+                std::unique_ptr<planpb::ClickHouseConfig> clickhouse_config = nullptr)
       : relation_map_(std::move(relation_map)),
         table_names_to_sensitive_columns_(table_names_to_sensitive_columns),
         registry_info_(registry_info),
@@ -130,7 +131,8 @@ class CompilerState : public NotCopyable {
         redaction_options_(redaction_options),
         endpoint_config_(std::move(endpoint_config)),
         plugin_config_(std::move(plugin_config)),
-        debug_info_(std::move(debug_info)) {}
+        debug_info_(std::move(debug_info)),
+        clickhouse_config_(std::move(clickhouse_config)) {}
 
   CompilerState() = delete;
 
@@ -175,6 +177,7 @@ class CompilerState : public NotCopyable {
   planpb::OTelEndpointConfig* endpoint_config() { return endpoint_config_.get(); }
   PluginConfig* plugin_config() { return plugin_config_.get(); }
   const DebugInfo& debug_info() { return debug_info_; }
+  planpb::ClickHouseConfig* clickhouse_config() { return clickhouse_config_.get(); }
 
  private:
   std::unique_ptr<RelationMap> relation_map_;
@@ -191,6 +194,7 @@ class CompilerState : public NotCopyable {
   std::unique_ptr<planpb::OTelEndpointConfig> endpoint_config_ = nullptr;
   std::unique_ptr<PluginConfig> plugin_config_ = nullptr;
   DebugInfo debug_info_;
+  std::unique_ptr<planpb::ClickHouseConfig> clickhouse_config_ = nullptr;
 };
 
 }  // namespace planner

src/carnot/planner/ir/BUILD.bazel

@@ -47,6 +47,7 @@ pl_cc_library(
         "//src/carnot/planpb:plan_pl_cc_proto",
         "//src/shared/metadata:cc_library",
         "//src/shared/metadatapb:metadata_pl_cc_proto",
+        "@com_github_clickhouse_clickhouse_cpp//:clickhouse_cpp",
         "@com_github_vinzenz_libpypa//:libpypa",
     ],
 )
@@ -67,6 +68,14 @@ pl_cc_test(
     ],
 )
 
+pl_cc_test(
+    name = "clickhouse_export_sink_ir_test",
+    srcs = ["clickhouse_export_sink_ir_test.cc"],
+    deps = [
+        "//src/carnot/planner/compiler:test_utils",
+    ],
+)
+
 pl_cc_test(
     name = "pattern_match_test",
     srcs = ["pattern_match_test.cc"],

src/carnot/planner/ir/all_ir_nodes.h

@@ -20,6 +20,8 @@
 
 #include "src/carnot/planner/ir/blocking_agg_ir.h"
 #include "src/carnot/planner/ir/bool_ir.h"
+#include "src/carnot/planner/ir/clickhouse_source_ir.h"
+#include "src/carnot/planner/ir/clickhouse_export_sink_ir.h"
 #include "src/carnot/planner/ir/column_ir.h"
 #include "src/carnot/planner/ir/data_ir.h"
 #include "src/carnot/planner/ir/drop_ir.h"

src/carnot/planner/ir/operators.inl

@@ -37,5 +37,7 @@ PX_CARNOT_IR_NODE(Rolling)
 PX_CARNOT_IR_NODE(Stream)
 PX_CARNOT_IR_NODE(EmptySource)
 PX_CARNOT_IR_NODE(OTelExportSink)
+PX_CARNOT_IR_NODE(ClickHouseSource)
+PX_CARNOT_IR_NODE(ClickHouseExportSink)
 
 #endif

src/carnot/planner/ir/pattern_match.h

@@ -160,6 +160,10 @@ inline ClassMatch<IRNodeType::kOTelExportSink> OTelExportSink() {
   return ClassMatch<IRNodeType::kOTelExportSink>();
 }
 
+inline ClassMatch<IRNodeType::kClickHouseExportSink> ClickHouseExportSink() {
+  return ClassMatch<IRNodeType::kClickHouseExportSink>();
+}
+
 inline ClassMatch<IRNodeType::kEmptySource> EmptySource() {
   return ClassMatch<IRNodeType::kEmptySource>();
 }
@@ -266,7 +270,7 @@ struct ResultSink : public ParentMatch {
 
   bool Match(const IRNode* node) const override {
     return ExternalGRPCSink().Match(node) || MemorySink().Match(node) ||
-           OTelExportSink().Match(node);
+           OTelExportSink().Match(node) || ClickHouseExportSink().Match(node);
   }
 };
 

src/carnot/planner/logical_planner.cc

@@ -97,6 +97,18 @@ StatusOr<std::unique_ptr<CompilerState>> CreateCompilerState(
   for (const auto& debug_info_pb : logical_state.debug_info().otel_debug_attributes()) {
     debug_info.otel_debug_attrs.push_back({debug_info_pb.name(), debug_info_pb.value()});
   }
+
+  std::unique_ptr<planpb::ClickHouseConfig> clickhouse_config = nullptr;
+  if (logical_state.has_clickhouse_config()) {
+    clickhouse_config = std::make_unique<planpb::ClickHouseConfig>();
+    clickhouse_config->set_hostname(logical_state.clickhouse_config().hostname());
+    clickhouse_config->set_host(logical_state.clickhouse_config().host());
+    clickhouse_config->set_port(logical_state.clickhouse_config().port());
+    clickhouse_config->set_username(logical_state.clickhouse_config().username());
+    clickhouse_config->set_password(logical_state.clickhouse_config().password());
+    clickhouse_config->set_database(logical_state.clickhouse_config().database());
+  }
+
   // Create a CompilerState obj using the relation map and grabbing the current time.
   return std::make_unique<planner::CompilerState>(
       std::move(rel_map), sensitive_columns, registry_info, px::CurrentTimeNS(),
@@ -105,7 +117,8 @@ StatusOr<std::unique_ptr<CompilerState>> CreateCompilerState(
       // TODO(philkuz) add an endpoint config to logical_state and pass that in here.
       RedactionOptionsFromPb(logical_state.redaction_options()), std::move(otel_endpoint_config),
       // TODO(philkuz) propagate the otel debug attributes here.
-      std::move(plugin_config), debug_info);
+      std::move(plugin_config), debug_info,
+      std::move(clickhouse_config));
 }
 
 StatusOr<std::unique_ptr<LogicalPlanner>> LogicalPlanner::Create(const udfspb::UDFInfo& udf_info) {