From 25e75b0fe8e42666b3b2e302d1747b8e49a8b70a Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 25 Aug 2026 03:17:17 +0000 Subject: [PATCH 1/4] Initial plan From 345f58c20714f7e95fa97b27b7500e497e8ba14c Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 25 Aug 2026 03:19:23 +0000 Subject: [PATCH 2/4] fix: key schema DDL guard per database.table instead of process-global Replace process-global std::once_flag with a std::mutex + std::set keyed by "database.table". DDL now runs exactly once per unique table across all DMXHostEngine instances in the same process, fixing the bug where only the first engine's table was created. Co-authored-by: zaoxing <2923149+zaoxing@users.noreply.github.com> --- native/csrc/clickhouse_client.cpp | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/native/csrc/clickhouse_client.cpp b/native/csrc/clickhouse_client.cpp index 454aa2901..8819ef998 100644 --- a/native/csrc/clickhouse_client.cpp +++ b/native/csrc/clickhouse_client.cpp @@ -8,6 +8,7 @@ #include #include #include +#include #include #include #include @@ -27,7 +28,8 @@ thread_local bool tl_cleaned = false; thread_local std::string tl_db; thread_local std::string tl_table; -std::once_flag g_schema_once; +std::mutex g_schema_mutex; +std::set g_schema_inited_tables; // --------------------- SQL helpers --------------------- @@ -416,8 +418,15 @@ void ClickHouseInsertStage::ThreadInit(int /*thread_idx*/, const ClickHouseClien } try { - // DDL init once globally (copy cfg/opts into the call_once closure) - std::call_once(g_schema_once, [cfg, opts]() { RunSchemaInitOnce(cfg, opts); }); + // DDL init once per database.table (not process-global) + { + const std::string table_key = cfg.database + "." + cfg.table; + std::lock_guard lock(g_schema_mutex); + if (g_schema_inited_tables.find(table_key) == g_schema_inited_tables.end()) { + RunSchemaInitOnce(cfg, opts); + g_schema_inited_tables.insert(table_key); + } + } // Per-thread client tl_client = std::make_unique(opts); From 948eb502374fa5b9b004a821ca7f3c09853a657b Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 25 Aug 2026 04:27:27 +0000 Subject: [PATCH 3/4] fix(clickhouse): split DB/table schema init, use null-byte separator for table key Co-authored-by: zaoxing <2923149+zaoxing@users.noreply.github.com> --- native/csrc/clickhouse_client.cpp | 25 ++++++++++++++++++------- 1 file changed, 18 insertions(+), 7 deletions(-) diff --git a/native/csrc/clickhouse_client.cpp b/native/csrc/clickhouse_client.cpp index 8819ef998..bdbf7d98d 100644 --- a/native/csrc/clickhouse_client.cpp +++ b/native/csrc/clickhouse_client.cpp @@ -29,6 +29,7 @@ thread_local std::string tl_db; thread_local std::string tl_table; std::mutex g_schema_mutex; +std::set g_schema_inited_dbs; std::set g_schema_inited_tables; // --------------------- SQL helpers --------------------- @@ -273,12 +274,9 @@ void ApplySessionSettings(clickhouse::Client& client, const ClickHouseClientConf } } -void RunSchemaInitOnce(const ClickHouseClientConfig& cfg, const clickhouse::ClientOptions& opts) { +void RunDbInitOnce(const ClickHouseClientConfig& cfg, const clickhouse::ClientOptions& opts) { auto client = std::make_unique(opts); - const std::string db_q = QuoteIdent(cfg.database); - const std::string table_q = QuoteIdent(cfg.table); - const std::string fq_table_q = db_q + "." + table_q; if (cfg.drop_existing_database) { client->Execute("DROP DATABASE IF EXISTS " + db_q); @@ -287,6 +285,14 @@ void RunSchemaInitOnce(const ClickHouseClientConfig& cfg, const clickhouse::Clie if (cfg.create_database_if_missing || cfg.drop_existing_database) { client->Execute("CREATE DATABASE IF NOT EXISTS " + db_q); } +} + +void RunTableInitOnce(const ClickHouseClientConfig& cfg, const clickhouse::ClientOptions& opts) { + auto client = std::make_unique(opts); + + const std::string db_q = QuoteIdent(cfg.database); + const std::string table_q = QuoteIdent(cfg.table); + const std::string fq_table_q = db_q + "." + table_q; // Schema: // dtype: String @@ -418,12 +424,17 @@ void ClickHouseInsertStage::ThreadInit(int /*thread_idx*/, const ClickHouseClien } try { - // DDL init once per database.table (not process-global) + // DDL init: DB-level ops guarded once per database, + // table-level ops guarded once per database\0table (null separator avoids dot collisions) { - const std::string table_key = cfg.database + "." + cfg.table; std::lock_guard lock(g_schema_mutex); + if (g_schema_inited_dbs.find(cfg.database) == g_schema_inited_dbs.end()) { + RunDbInitOnce(cfg, opts); + g_schema_inited_dbs.insert(cfg.database); + } + const std::string table_key = cfg.database + '\0' + cfg.table; if (g_schema_inited_tables.find(table_key) == g_schema_inited_tables.end()) { - RunSchemaInitOnce(cfg, opts); + RunTableInitOnce(cfg, opts); g_schema_inited_tables.insert(table_key); } } From a533fa1f45dcfc4fe03db57e3fabeb8103ec0192 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 25 Aug 2026 04:44:46 +0000 Subject: [PATCH 4/4] fix(clickhouse): include endpoint in schema-init keys, fix DB init guard, remove dead schema_once Co-authored-by: zaoxing <2923149+zaoxing@users.noreply.github.com> --- native/csrc/clickhouse_client.cpp | 23 ++++++++++++++++++----- native/csrc/clickhouse_client.h | 2 -- native/csrc/dmx_host_engine.h | 1 - 3 files changed, 18 insertions(+), 8 deletions(-) diff --git a/native/csrc/clickhouse_client.cpp b/native/csrc/clickhouse_client.cpp index 698372357..4b3fd34ed 100644 --- a/native/csrc/clickhouse_client.cpp +++ b/native/csrc/clickhouse_client.cpp @@ -503,15 +503,28 @@ void ClickHouseInsertStage::ThreadInit(int thread_idx, const ClickHouseClientCon } try { - // DDL init: DB-level ops guarded once per database, - // table-level ops guarded once per database\0table (null separator avoids dot collisions) + // DDL init: DB-level ops guarded once per (host, port, secure, database), + // table-level ops guarded once per (host, port, secure, database, table). + // Null-byte separators avoid collisions with identifiers that contain dots. + // The endpoint prefix ensures engines targeting different servers are not + // incorrectly treated as already initialised. { + const std::string endpoint_prefix = + cfg.host + '\0' + std::to_string(cfg.port) + '\0' + + (cfg.secure ? "1" : "0") + '\0'; + const std::string db_key = endpoint_prefix + cfg.database; + const std::string table_key = db_key + '\0' + cfg.table; + std::lock_guard lock(g_schema_mutex); - if (g_schema_inited_dbs.find(cfg.database) == g_schema_inited_dbs.end()) { + if (g_schema_inited_dbs.find(db_key) == g_schema_inited_dbs.end()) { RunDbInitOnce(cfg, opts); - g_schema_inited_dbs.insert(cfg.database); + // Only mark as initialised when DB-level DDL was (or would be) executed. + // If neither flag is set, no DDL runs; skip the mark so a later engine + // that does set one of these flags is not incorrectly suppressed. + if (cfg.create_database_if_missing || cfg.drop_existing_database) { + g_schema_inited_dbs.insert(db_key); + } } - const std::string table_key = cfg.database + '\0' + cfg.table; if (g_schema_inited_tables.find(table_key) == g_schema_inited_tables.end()) { RunTableInitOnce(cfg, opts); g_schema_inited_tables.insert(table_key); diff --git a/native/csrc/clickhouse_client.h b/native/csrc/clickhouse_client.h index 6093d4ef7..527670809 100644 --- a/native/csrc/clickhouse_client.h +++ b/native/csrc/clickhouse_client.h @@ -97,8 +97,6 @@ struct ClickHouseClientConfig { int receive_timeout_ms = 0; int send_timeout_ms = 0; - std::shared_ptr schema_once = - std::make_shared(); std::shared_ptr runtime_metrics = std::make_shared(); }; diff --git a/native/csrc/dmx_host_engine.h b/native/csrc/dmx_host_engine.h index 11f63da44..2e08c3053 100644 --- a/native/csrc/dmx_host_engine.h +++ b/native/csrc/dmx_host_engine.h @@ -73,7 +73,6 @@ class DMXHostEngine : public DMXHostEngineBase { } auto config = std::any_cast(stage.thread_init_config); - config.schema_once = std::make_shared(); auto metrics = std::make_shared(stage.parallelism); config.runtime_metrics = metrics;