Merge branch 'main' into ddelnano/remove-unnecessary-changes

Signed-off-by: Dom Delnano <[email protected]>

Author: ddelnano

k8s/cloud_deps/base/opensearch/operator/opensearch_operator.yaml

@@ -178,7 +178,7 @@ std::shared_ptr<px::table_store::Table> GetTableFromCsv(const std::string& filen
 
   // Construct the table.
   px::table_store::schema::Relation rel(types, names);
-  auto table = px::table_store::Table::Create("csv_table", rel);
+  auto table = px::table_store::HotColdTable::Create("csv_table", rel);
 
   // Add rowbatches to the table.
   row_idx = 0;

src/carnot/carnot_executable.cc

@@ -423,6 +423,7 @@ class ClickHouseExportSinkOperator : public Operator {
 };
 
 class OTelExportSinkOperator : public Operator {
+
  public:
   explicit OTelExportSinkOperator(int64_t id) : Operator(id, planpb::OTEL_EXPORT_SINK_OPERATOR) {}
   ~OTelExportSinkOperator() override = default;

src/carnot/plan/operators.h

@@ -1213,6 +1213,132 @@ TEST_F(LogicalPlannerTest, ClickHouseExportWithExplicitEndpoint) {
   EXPECT_TRUE(has_clickhouse_export);
 }
 
+constexpr char kClickHouseExportQuery[] = R"pxl(
+import px
+
+# Test ClickHouse export using endpoint config
+df = px.DataFrame('http_events', start_time='-10m')
+df = df[['time_', 'req_path', 'resp_status', 'resp_latency_ns']]
+px.export(df, px.otel.ClickHouseRows(table='http_events'))
+)pxl";
+
+TEST_F(LogicalPlannerTest, ClickHouseExportWithEndpointConfig) {
+  auto planner = LogicalPlanner::Create(info_).ConsumeValueOrDie();
+
+  // Create a planner state with an OTel endpoint config containing ClickHouse DSN
+  auto state = testutils::CreateTwoPEMsOneKelvinPlannerState(testutils::kHttpEventsSchema);
+
+  // Set up the endpoint config with ClickHouse DSN in the URL field
+  auto* endpoint_config = state.mutable_otel_endpoint_config();
+  endpoint_config->set_url("clickhouse_user:[email protected]:9000/pixie_db");
+  endpoint_config->set_insecure(true);
+  endpoint_config->set_timeout(10);
+
+  auto plan_or_s = planner->Plan(MakeQueryRequest(state, kClickHouseExportQuery));
+  EXPECT_OK(plan_or_s);
+  auto plan = plan_or_s.ConsumeValueOrDie();
+  EXPECT_OK(plan->ToProto());
+
+  // Verify the plan contains ClickHouse export sink operators with correct config
+  auto plan_pb = plan->ToProto().ConsumeValueOrDie();
+  bool has_clickhouse_export = false;
+
+  for (const auto& [address, agent_plan] : plan_pb.qb_address_to_plan()) {
+    for (const auto& planFragment : agent_plan.nodes()) {
+      for (const auto& planNode : planFragment.nodes()) {
+        if (planNode.op().op_type() == planpb::OperatorType::CLICKHOUSE_EXPORT_SINK_OPERATOR) {
+          const auto& clickhouse_sink_op = planNode.op().clickhouse_sink_op();
+
+          // Verify table name
+          EXPECT_EQ(clickhouse_sink_op.table_name(), "http_events");
+
+          // Verify the DSN was parsed correctly into ClickHouseConfig
+          const auto& config = clickhouse_sink_op.clickhouse_config();
+          EXPECT_EQ(config.username(), "clickhouse_user");
+          EXPECT_EQ(config.password(), "clickhouse_pass");
+          EXPECT_EQ(config.host(), "clickhouse.example.com");
+          EXPECT_EQ(config.port(), 9000);
+          EXPECT_EQ(config.database(), "pixie_db");
+
+          // Verify column mappings were created
+          EXPECT_GT(clickhouse_sink_op.column_mappings_size(), 0);
+
+          has_clickhouse_export = true;
+          break;
+        }
+      }
+      if (has_clickhouse_export) break;
+    }
+    if (has_clickhouse_export) break;
+  }
+
+  EXPECT_TRUE(has_clickhouse_export);
+}
+
+constexpr char kClickHouseExportWithExplicitEndpointQuery[] = R"pxl(
+import px
+
+# Test ClickHouse export with explicit endpoint config
+df = px.DataFrame('http_events', start_time='-10m')
+df = df[['time_', 'req_path', 'resp_status']]
+
+endpoint = px.otel.Endpoint(
+  url="explicit_user:explicit_pass@explicit-host:9001/explicit_db",
+  insecure=False,
+  timeout=20
+)
+
+px.export(df, px.otel.ClickHouseRows(table='custom_table', endpoint=endpoint))
+)pxl";
+
+TEST_F(LogicalPlannerTest, ClickHouseExportWithExplicitEndpoint) {
+  auto planner = LogicalPlanner::Create(info_).ConsumeValueOrDie();
+
+  // Create a planner state with a default endpoint config
+  auto state = testutils::CreateTwoPEMsOneKelvinPlannerState(testutils::kHttpEventsSchema);
+
+  // Set up a default endpoint config (should be overridden by explicit endpoint)
+  auto* endpoint_config = state.mutable_otel_endpoint_config();
+  endpoint_config->set_url("default_user:default_pass@default-host:9000/default_db");
+
+  auto plan_or_s = planner->Plan(MakeQueryRequest(state, kClickHouseExportWithExplicitEndpointQuery));
+  EXPECT_OK(plan_or_s);
+  auto plan = plan_or_s.ConsumeValueOrDie();
+  EXPECT_OK(plan->ToProto());
+
+  // Verify the plan uses the explicit endpoint config, not the default
+  auto plan_pb = plan->ToProto().ConsumeValueOrDie();
+  bool has_clickhouse_export = false;
+
+  for (const auto& [address, agent_plan] : plan_pb.qb_address_to_plan()) {
+    for (const auto& planFragment : agent_plan.nodes()) {
+      for (const auto& planNode : planFragment.nodes()) {
+        if (planNode.op().op_type() == planpb::OperatorType::CLICKHOUSE_EXPORT_SINK_OPERATOR) {
+          const auto& clickhouse_sink_op = planNode.op().clickhouse_sink_op();
+
+          // Verify table name
+          EXPECT_EQ(clickhouse_sink_op.table_name(), "custom_table");
+
+          // Verify the explicit endpoint was used, not the default
+          const auto& config = clickhouse_sink_op.clickhouse_config();
+          EXPECT_EQ(config.username(), "explicit_user");
+          EXPECT_EQ(config.password(), "explicit_pass");
+          EXPECT_EQ(config.host(), "explicit-host");
+          EXPECT_EQ(config.port(), 9001);
+          EXPECT_EQ(config.database(), "explicit_db");
+
+          has_clickhouse_export = true;
+          break;
+        }
+      }
+      if (has_clickhouse_export) break;
+    }
+    if (has_clickhouse_export) break;
+  }
+
+  EXPECT_TRUE(has_clickhouse_export);
+}
+
 }  // namespace planner
 }  // namespace carnot
 }  // namespace px

src/carnot/planner/logical_planner_test.cc

@@ -313,6 +313,30 @@ class ClickHouseRows : public QLObject {
   std::string table_name_;
 };
 
+class ClickHouseRows : public QLObject {
+ public:
+  static constexpr TypeDescriptor ClickHouseRowsType = {
+      /* name */ "ClickHouseRows",
+      /* type */ QLObjectType::kClickHouseRows,
+  };
+
+  static StatusOr<std::shared_ptr<ClickHouseRows>> Create(
+      ASTVisitor* ast_visitor, const std::string& table_name);
+
+  static bool IsClickHouseRows(const QLObjectPtr& obj) {
+    return obj->type() == ClickHouseRowsType.type();
+  }
+
+  const std::string& table_name() const { return table_name_; }
+
+ protected:
+  ClickHouseRows(ASTVisitor* ast_visitor, std::string table_name)
+      : QLObject(ClickHouseRowsType, ast_visitor), table_name_(std::move(table_name)) {}
+
+ private:
+  std::string table_name_;
+};
+
 }  // namespace compiler
 }  // namespace planner
 }  // namespace carnot

src/carnot/planner/objects/otel.h

@@ -50,6 +50,7 @@ pl_cc_library(
         "//src/vizier/funcs:cc_library",
         "//src/vizier/funcs/context:cc_library",
         "//src/vizier/services/agent/shared/base:cc_library",
+        "//src/vizier/services/metadata/local:cc_library",
         "@com_github_grpc_grpc//:grpc++",
     ],
 )

src/carnot/planpb/plan.pb.go

@@ -27,6 +27,7 @@
 #include "src/shared/schema/utils.h"
 #include "src/table_store/table_store.h"
 #include "src/vizier/funcs/funcs.h"
+#include "src/vizier/services/metadata/local/local_metadata_service.h"
 
 DEFINE_int32(
     table_store_data_limit, gflags::Int32FromEnv("PL_TABLE_STORE_DATA_LIMIT_MB", 1024 + 256),