From da8ddf8fd45d3957495eeb154684d277c1ffdf6e Mon Sep 17 00:00:00 2001 From: Marek Kadek Date: Mon, 20 Apr 2026 13:04:18 +0200 Subject: [PATCH 1/2] feat: allow configuring clusterMode --- .../implementations/sql/clickhouse.go | 70 +++++++++--- .../implementations/sql/clickhouse_test.go | 108 ++++++++++++++++++ webapps/console/lib/schema/destinations.tsx | 9 ++ 3 files changed, 168 insertions(+), 19 deletions(-) create mode 100644 bulker/bulkerlib/implementations/sql/clickhouse_test.go diff --git a/bulker/bulkerlib/implementations/sql/clickhouse.go b/bulker/bulkerlib/implementations/sql/clickhouse.go index 1ff57db5d..b74c0fe3e 100644 --- a/bulker/bulkerlib/implementations/sql/clickhouse.go +++ b/bulker/bulkerlib/implementations/sql/clickhouse.go @@ -39,6 +39,7 @@ const ( chLocalPrefix = "local_" chDatabaseQuery = "SELECT name FROM system.databases where name = ?" + chDatabaseEngineQuery = "SELECT engine FROM system.databases where name = ?" chClusterQuery = "SELECT max(shard_num) FROM system.clusters where cluster = ?" chCreateDatabaseTemplate = `CREATE DATABASE IF NOT EXISTS %s %s` @@ -141,18 +142,26 @@ const ( ClickHouseProtocolHTTPS ClickHouseProtocol = "https" ) +type ClickHouseClusterMode string + +const ( + ClusterModeOnCluster ClickHouseClusterMode = "on_cluster" + ClusterModeReplicatedDB ClickHouseClusterMode = "replicated_db" +) + // ClickHouseConfig dto for deserialized clickhouse config type ClickHouseConfig struct { - Protocol ClickHouseProtocol `mapstructure:"protocol,omitempty" json:"protocol,omitempty" yaml:"protocol,omitempty"` - Hosts []string `mapstructure:"hosts,omitempty" json:"hosts,omitempty" yaml:"hosts,omitempty"` - Parameters map[string]string `mapstructure:"parameters,omitempty" json:"parameters,omitempty" yaml:"parameters,omitempty"` - Username string `mapstructure:"username,omitempty" json:"username,omitempty" yaml:"username,omitempty"` - Password string `mapstructure:"password,omitempty" json:"password,omitempty" yaml:"password,omitempty"` - Database string `mapstructure:"database,omitempty" json:"database,omitempty" yaml:"database,omitempty"` - Cluster string `mapstructure:"cluster,omitempty" json:"cluster,omitempty" yaml:"cluster,omitempty"` - TLS map[string]string `mapstructure:"tls,omitempty" json:"tls,omitempty" yaml:"tls,omitempty"` - Engine *EngineConfig `mapstructure:"engine,omitempty" json:"engine,omitempty" yaml:"engine,omitempty"` - LoadAsJSON bool `mapstructure:"loadAsJson,omitempty" json:"loadAsJson,omitempty" yaml:"loadAsJson,omitempty"` + Protocol ClickHouseProtocol `mapstructure:"protocol,omitempty" json:"protocol,omitempty" yaml:"protocol,omitempty"` + Hosts []string `mapstructure:"hosts,omitempty" json:"hosts,omitempty" yaml:"hosts,omitempty"` + Parameters map[string]string `mapstructure:"parameters,omitempty" json:"parameters,omitempty" yaml:"parameters,omitempty"` + Username string `mapstructure:"username,omitempty" json:"username,omitempty" yaml:"username,omitempty"` + Password string `mapstructure:"password,omitempty" json:"password,omitempty" yaml:"password,omitempty"` + Database string `mapstructure:"database,omitempty" json:"database,omitempty" yaml:"database,omitempty"` + Cluster string `mapstructure:"cluster,omitempty" json:"cluster,omitempty" yaml:"cluster,omitempty"` + ClusterMode ClickHouseClusterMode `mapstructure:"clusterMode,omitempty" json:"clusterMode,omitempty" yaml:"clusterMode,omitempty"` + TLS map[string]string `mapstructure:"tls,omitempty" json:"tls,omitempty" yaml:"tls,omitempty"` + Engine *EngineConfig `mapstructure:"engine,omitempty" json:"engine,omitempty" yaml:"engine,omitempty"` + LoadAsJSON bool `mapstructure:"loadAsJson,omitempty" json:"loadAsJson,omitempty" yaml:"loadAsJson,omitempty"` // S3Config S3AccessKeyID string `mapstructure:"s3AccessKeyId,omitempty" json:"s3AccessKeyId,omitempty" yaml:"s3AccessKeyId,omitempty"` @@ -421,6 +430,13 @@ func (ch *ClickHouse) InitDatabase(ctx context.Context) error { if err != nil { return err } + if ch.isReplicatedDBMode() { + var engine string + err := ch.txOrDb(ctx).QueryRowContext(ctx, chDatabaseEngineQuery, ch.NamespaceName(ch.config.Database)).Scan(&engine) + if err != nil || engine != "Replicated" { + return fmt.Errorf("clusterMode=replicated_db requires database %q to use the Replicated engine; got %q (err=%v)", ch.config.Database, engine, err) + } + } if ch.config.Cluster != "" { var shardNum int err := ch.txOrDb(ctx).QueryRowContext(ctx, chClusterQuery, ch.config.Cluster).Scan(&shardNum) @@ -484,7 +500,8 @@ func (ch *ClickHouse) CreateTable(ctx context.Context, table *Table) (*Table, er }) } - //create distributed table + //create distributed table — still required in replicated_db mode for multi-shard clusters, + //since the Replicated database engine handles DDL replication within a shard but not sharding itself. if ch.distributed.Load() { return table, ch.createDistributedTableInTransaction(ctx, table) } @@ -940,9 +957,18 @@ func (ch *ClickHouse) ReplaceTable(ctx context.Context, targetTableName string, } +// isReplicatedDBMode reports whether the destination targets a database that +// uses ClickHouse's Replicated database engine. In that mode the database +// itself manages zookeeper paths, replica macros, and DDL replication, so +// table engines must be created without explicit path/replica arguments and +// without ON CLUSTER. +func (ch *ClickHouse) isReplicatedDBMode() bool { + return ch.config.Cluster != "" && ch.config.ClusterMode == ClusterModeReplicatedDB +} + // return ON CLUSTER name clause or "" if config.cluster is empty func (ch *ClickHouse) getOnClusterClause() string { - if ch.config.Cluster == "" { + if ch.config.Cluster == "" || ch.isReplicatedDBMode() { return "" } @@ -1165,6 +1191,7 @@ func (ch *ClickHouse) IsDistributed() bool { type ClickHouseCluster interface { IsDistributed() bool Config() *ClickHouseConfig + isReplicatedDBMode() bool } // TableStatementFactory is used for creating CREATE TABLE statements depends on config @@ -1175,7 +1202,7 @@ type TableStatementFactory struct { func NewTableStatementFactory(ch ClickHouseCluster) *TableStatementFactory { var onClusterClause string - if ch.Config().Cluster != "" { + if ch.Config().Cluster != "" && !ch.isReplicatedDBMode() { onClusterClause = fmt.Sprintf(chOnClusterClauseTemplate, ch.Config().Cluster) } @@ -1219,13 +1246,18 @@ func (tsf TableStatementFactory) CreateTableStatement(namespacePrefix, quotedTab } if config.Cluster != "" { - shardsMacros := "{shard}/" - if !tsf.ch.IsDistributed() { - shardsMacros = "1/" + if tsf.ch.isReplicatedDBMode() { + // Replicated database engine manages zookeeper_path and replica_name; pass none. + engineStatement = `ENGINE = Replicated` + baseEngine + `()` + } else { + shardsMacros := "{shard}/" + if !tsf.ch.IsDistributed() { + shardsMacros = "1/" + } + //create engine statement with ReplicatedReplacingMergeTree() engine. We need to replace %s with tableName on creating statement + engineStatement = `ENGINE = Replicated` + baseEngine + `('/clickhouse/tables/` + shardsMacros + config.Database + `/%s', '{replica}')` + engineStatementFormat = true } - //create engine statement with ReplicatedReplacingMergeTree() engine. We need to replace %s with tableName on creating statement - engineStatement = `ENGINE = Replicated` + baseEngine + `('/clickhouse/tables/` + shardsMacros + config.Database + `/%s', '{replica}')` - engineStatementFormat = true } else { //create table template with ReplacingMergeTree() engine engineStatement = `ENGINE = ` + baseEngine + `()` diff --git a/bulker/bulkerlib/implementations/sql/clickhouse_test.go b/bulker/bulkerlib/implementations/sql/clickhouse_test.go new file mode 100644 index 000000000..aefea4cfc --- /dev/null +++ b/bulker/bulkerlib/implementations/sql/clickhouse_test.go @@ -0,0 +1,108 @@ +package sql + +import ( + "strings" + "testing" + + "github.com/jitsucom/bulker/jitsubase/jsonorder" +) + +type fakeChCluster struct { + config *ClickHouseConfig + distributed bool +} + +func (f *fakeChCluster) IsDistributed() bool { return f.distributed } +func (f *fakeChCluster) Config() *ClickHouseConfig { return f.config } +func (f *fakeChCluster) isReplicatedDBMode() bool { + return f.config.Cluster != "" && f.config.ClusterMode == ClusterModeReplicatedDB +} + +func TestCreateTableStatement_engineVariants(t *testing.T) { + cases := []struct { + name string + config *ClickHouseConfig + distributed bool + mustContain []string + mustNotHave []string + }{ + { + name: "no cluster -> plain MergeTree", + config: &ClickHouseConfig{Database: "db"}, + mustContain: []string{"ENGINE = MergeTree()"}, + mustNotHave: []string{"Replicated", "ON CLUSTER"}, + }, + { + name: "cluster, default mode, single shard -> path uses 1/", + config: &ClickHouseConfig{Database: "db", Cluster: "c"}, + distributed: false, + mustContain: []string{"ENGINE = ReplicatedMergeTree('/clickhouse/tables/1/db/", "'{replica}')", "ON CLUSTER `c`"}, + mustNotHave: []string{"ReplicatedMergeTree()"}, + }, + { + name: "cluster, default mode, distributed -> path uses {shard}/", + config: &ClickHouseConfig{Database: "db", Cluster: "c"}, + distributed: true, + mustContain: []string{"ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/db/", "ON CLUSTER `c`"}, + mustNotHave: []string{"ReplicatedMergeTree()"}, + }, + { + name: "cluster, replicated_db mode -> path-less, no ON CLUSTER", + config: &ClickHouseConfig{Database: "db", Cluster: "c", ClusterMode: ClusterModeReplicatedDB}, + mustContain: []string{"ENGINE = ReplicatedMergeTree()"}, + mustNotHave: []string{"/clickhouse/tables/", "{replica}", "ON CLUSTER"}, + }, + { + name: "cluster, replicated_db mode, distributed -> still path-less, no ON CLUSTER", + config: &ClickHouseConfig{Database: "db", Cluster: "c", ClusterMode: ClusterModeReplicatedDB}, + distributed: true, + mustContain: []string{"ENGINE = ReplicatedMergeTree()"}, + mustNotHave: []string{"/clickhouse/tables/", "{replica}", "ON CLUSTER"}, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + cluster := &fakeChCluster{config: tc.config, distributed: tc.distributed} + tsf := NewTableStatementFactory(cluster) + table := &Table{ + Name: "events", + Columns: NewColumns(0), + PKFields: jsonorder.NewOrderedSet[string](), + } + stmt := tsf.CreateTableStatement("`db`.", "`events`", "events", "`a` String", table) + for _, want := range tc.mustContain { + if !strings.Contains(stmt, want) { + t.Fatalf("statement missing %q\nactual: %s", want, stmt) + } + } + for _, unwanted := range tc.mustNotHave { + if strings.Contains(stmt, unwanted) { + t.Fatalf("statement unexpectedly contains %q\nactual: %s", unwanted, stmt) + } + } + }) + } +} + +func TestGetOnClusterClause_replicatedDBMode(t *testing.T) { + cases := []struct { + name string + config ClickHouseConfig + want string + }{ + {name: "no cluster", config: ClickHouseConfig{}, want: ""}, + {name: "cluster, default mode", config: ClickHouseConfig{Cluster: "c"}, want: " ON CLUSTER `c` "}, + {name: "cluster, on_cluster mode", config: ClickHouseConfig{Cluster: "c", ClusterMode: ClusterModeOnCluster}, want: " ON CLUSTER `c` "}, + {name: "cluster, replicated_db mode", config: ClickHouseConfig{Cluster: "c", ClusterMode: ClusterModeReplicatedDB}, want: ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + ch := &ClickHouse{SQLAdapterBase: &SQLAdapterBase[ClickHouseConfig]{config: &tc.config}} + got := ch.getOnClusterClause() + if got != tc.want { + t.Fatalf("getOnClusterClause = %q, want %q", got, tc.want) + } + }) + } +} diff --git a/webapps/console/lib/schema/destinations.tsx b/webapps/console/lib/schema/destinations.tsx index 92cee55e6..e7d391372 100644 --- a/webapps/console/lib/schema/destinations.tsx +++ b/webapps/console/lib/schema/destinations.tsx @@ -276,6 +276,12 @@ export const ClickhouseCredentials = z.object({ .string() .optional() .describe("Name of cluster to use.
For ClickHouse Cloud or single-node setups, leave this field empty."), + clusterMode: z + .enum(["on_cluster", "replicated_db"]) + .optional() + .describe( + "Cluster mode::How tables are created on the cluster. on_cluster (default) issues ON CLUSTER + ReplicatedMergeTree with explicit ZooKeeper paths. Choose replicated_db when the target database itself uses the Replicated database engine — Jitsu will then emit path-less ReplicatedMergeTree and skip ON CLUSTER. Ignored when cluster is empty." + ), database: z.string().default("default").describe("Name of the database to use"), parameters: z .object({}) @@ -422,6 +428,9 @@ export const coreDestinations: DestinationType[] = [ loadAsJson: { hidden: true, }, + clusterMode: { + hidden: obj => !obj.cluster, + }, password: { password: true, }, From 0170c5778adbd2ac889d28a88c763aa0ff0b0ac4 Mon Sep 17 00:00:00 2001 From: Marek Kadek Date: Thu, 23 Apr 2026 13:21:06 +0200 Subject: [PATCH 2/2] address PR review --- .../implementations/sql/bulker_test.go | 17 ++- .../implementations/sql/clickhouse.go | 76 ++++++----- .../implementations/sql/clickhouse_test.go | 24 ++-- .../implementations/sql/naming_test.go | 4 +- .../ch_replicated_db_container.go | 123 ++++++++++++++++++ .../clickhouse_repl01/config.xml | 80 ++++++++++++ .../clickhouse_repl01/users.xml | 36 +++++ .../clickhouse_repl02/config.xml | 80 ++++++++++++ .../clickhouse_repl02/users.xml | 36 +++++ .../docker-compose.yml | 66 ++++++++++ webapps/console/lib/schema/destinations.tsx | 8 +- 11 files changed, 501 insertions(+), 49 deletions(-) create mode 100644 bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/ch_replicated_db_container.go create mode 100644 bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/clickhouse_repl01/config.xml create mode 100644 bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/clickhouse_repl01/users.xml create mode 100644 bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/clickhouse_repl02/config.xml create mode 100644 bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/clickhouse_repl02/users.xml create mode 100644 bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/docker-compose.yml diff --git a/bulker/bulkerlib/implementations/sql/bulker_test.go b/bulker/bulkerlib/implementations/sql/bulker_test.go index 613ec99bd..be2596178 100644 --- a/bulker/bulkerlib/implementations/sql/bulker_test.go +++ b/bulker/bulkerlib/implementations/sql/bulker_test.go @@ -14,6 +14,7 @@ import ( testcontainers2 "github.com/jitsucom/bulker/bulkerlib/implementations/sql/testcontainers" "github.com/jitsucom/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse" "github.com/jitsucom/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_noshards" + "github.com/jitsucom/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db" types2 "github.com/jitsucom/bulker/bulkerlib/types" "github.com/jitsucom/bulker/jitsubase/jsonorder" "github.com/jitsucom/bulker/jitsubase/logging" @@ -29,7 +30,7 @@ var constantTime2 = timestamp.MustParseTime(time.RFC3339Nano, "2022-08-18T15:17: const forceLeaveResultingTables = false var allBulkerConfigs = []string{BigqueryBulkerTypeId, RedshiftBulkerTypeId, RedshiftBulkerTypeId + "_iam", RedshiftBulkerTypeId + "_serverless", SnowflakeBulkerTypeId, PostgresBulkerTypeId, - MySQLBulkerTypeId, ClickHouseBulkerTypeId, ClickHouseBulkerTypeId + "_cluster", ClickHouseBulkerTypeId + "_cluster_noshards", DuckDBBulkerTypeId} + MySQLBulkerTypeId, ClickHouseBulkerTypeId, ClickHouseBulkerTypeId + "_cluster", ClickHouseBulkerTypeId + "_cluster_noshards", ClickHouseBulkerTypeId + "_replicated_db", DuckDBBulkerTypeId} var exceptBigquery []string @@ -56,6 +57,7 @@ var mysqlContainer *testcontainers2.MySQLContainer var clickhouseContainer *testcontainers2.ClickHouseContainer var clickhouseClusterContainer *clickhouse.ClickHouseClusterContainer var clickhouseClusterContainerNoShards *clickhouse_noshards.ClickHouseClusterContainerNoShards +var clickhouseReplicatedDBContainer *clickhouse_replicated_db.ClickHouseReplicatedDBContainer func init() { // uncomment to run tests locally with just one bulker type @@ -191,6 +193,19 @@ func init() { }, }} } + if utils.ArrayContains(allBulkerConfigs, ClickHouseBulkerTypeId+"_replicated_db") { + clickhouseReplicatedDBContainer, err = clickhouse_replicated_db.NewClickhouseReplicatedDBContainer(context.Background()) + if err != nil { + panic(err) + } + configRegistry[ClickHouseBulkerTypeId+"_replicated_db"] = TestConfig{BulkerType: ClickHouseBulkerTypeId, Config: ClickHouseConfig{ + Hosts: clickhouseReplicatedDBContainer.Hosts, + Username: "default", + Database: clickhouseReplicatedDBContainer.Database, + Cluster: clickhouseReplicatedDBContainer.Cluster, + DatabaseEngine: DatabaseEngineReplicated, + }} + } exceptBigquery = utils.ArrayExcluding(allBulkerConfigs, BigqueryBulkerTypeId) diff --git a/bulker/bulkerlib/implementations/sql/clickhouse.go b/bulker/bulkerlib/implementations/sql/clickhouse.go index b74c0fe3e..cdc8bc727 100644 --- a/bulker/bulkerlib/implementations/sql/clickhouse.go +++ b/bulker/bulkerlib/implementations/sql/clickhouse.go @@ -142,26 +142,34 @@ const ( ClickHouseProtocolHTTPS ClickHouseProtocol = "https" ) -type ClickHouseClusterMode string +// ClickHouseDatabaseEngine selects how the destination database stores and replicates data. +// The value is the ClickHouse database engine that Jitsu expects the target database to use. +type ClickHouseDatabaseEngine string const ( - ClusterModeOnCluster ClickHouseClusterMode = "on_cluster" - ClusterModeReplicatedDB ClickHouseClusterMode = "replicated_db" + // DatabaseEngineDefault covers the Atomic engine (self-hosted) and the Shared engine (ClickHouse Cloud). + // Tables are created as ReplicatedMergeTree with explicit ZooKeeper paths when a cluster is set, + // mirroring Jitsu's historical behaviour. + DatabaseEngineDefault ClickHouseDatabaseEngine = "default" + // DatabaseEngineReplicated targets a database created with ENGINE = Replicated(...). ClickHouse manages + // ZooKeeper paths, replica macros, and DDL replication itself, so Jitsu emits path-less Replicated* + // table engines and skips ON CLUSTER on DDL inside the database. + DatabaseEngineReplicated ClickHouseDatabaseEngine = "replicated" ) // ClickHouseConfig dto for deserialized clickhouse config type ClickHouseConfig struct { - Protocol ClickHouseProtocol `mapstructure:"protocol,omitempty" json:"protocol,omitempty" yaml:"protocol,omitempty"` - Hosts []string `mapstructure:"hosts,omitempty" json:"hosts,omitempty" yaml:"hosts,omitempty"` - Parameters map[string]string `mapstructure:"parameters,omitempty" json:"parameters,omitempty" yaml:"parameters,omitempty"` - Username string `mapstructure:"username,omitempty" json:"username,omitempty" yaml:"username,omitempty"` - Password string `mapstructure:"password,omitempty" json:"password,omitempty" yaml:"password,omitempty"` - Database string `mapstructure:"database,omitempty" json:"database,omitempty" yaml:"database,omitempty"` - Cluster string `mapstructure:"cluster,omitempty" json:"cluster,omitempty" yaml:"cluster,omitempty"` - ClusterMode ClickHouseClusterMode `mapstructure:"clusterMode,omitempty" json:"clusterMode,omitempty" yaml:"clusterMode,omitempty"` - TLS map[string]string `mapstructure:"tls,omitempty" json:"tls,omitempty" yaml:"tls,omitempty"` - Engine *EngineConfig `mapstructure:"engine,omitempty" json:"engine,omitempty" yaml:"engine,omitempty"` - LoadAsJSON bool `mapstructure:"loadAsJson,omitempty" json:"loadAsJson,omitempty" yaml:"loadAsJson,omitempty"` + Protocol ClickHouseProtocol `mapstructure:"protocol,omitempty" json:"protocol,omitempty" yaml:"protocol,omitempty"` + Hosts []string `mapstructure:"hosts,omitempty" json:"hosts,omitempty" yaml:"hosts,omitempty"` + Parameters map[string]string `mapstructure:"parameters,omitempty" json:"parameters,omitempty" yaml:"parameters,omitempty"` + Username string `mapstructure:"username,omitempty" json:"username,omitempty" yaml:"username,omitempty"` + Password string `mapstructure:"password,omitempty" json:"password,omitempty" yaml:"password,omitempty"` + Database string `mapstructure:"database,omitempty" json:"database,omitempty" yaml:"database,omitempty"` + Cluster string `mapstructure:"cluster,omitempty" json:"cluster,omitempty" yaml:"cluster,omitempty"` + DatabaseEngine ClickHouseDatabaseEngine `mapstructure:"databaseEngine,omitempty" json:"databaseEngine,omitempty" yaml:"databaseEngine,omitempty"` + TLS map[string]string `mapstructure:"tls,omitempty" json:"tls,omitempty" yaml:"tls,omitempty"` + Engine *EngineConfig `mapstructure:"engine,omitempty" json:"engine,omitempty" yaml:"engine,omitempty"` + LoadAsJSON bool `mapstructure:"loadAsJson,omitempty" json:"loadAsJson,omitempty" yaml:"loadAsJson,omitempty"` // S3Config S3AccessKeyID string `mapstructure:"s3AccessKeyId,omitempty" json:"s3AccessKeyId,omitempty" yaml:"s3AccessKeyId,omitempty"` @@ -410,7 +418,17 @@ func (ch *ClickHouse) createDatabaseIfNotExists(ctx context.Context, db string) _ = row.Scan(&dbname) } if dbname == "" { - query := fmt.Sprintf(chCreateDatabaseTemplate, db, ch.getOnClusterClause()) + // CREATE DATABASE is not executed "inside" a Replicated database, so ON CLUSTER is valid + // and required to land the new database on every node. Build the clause unconditionally + // rather than going through getOnClusterClause, which suppresses ON CLUSTER in replicated mode. + onClusterClause := "" + if ch.config.Cluster != "" { + onClusterClause = fmt.Sprintf(chOnClusterClauseTemplate, ch.config.Cluster) + } + query := fmt.Sprintf(chCreateDatabaseTemplate, db, onClusterClause) + if ch.isReplicatedDatabase() { + query += fmt.Sprintf(" ENGINE = Replicated('/clickhouse/databases/%s', '{shard}', '{replica}')", db) + } if _, err := ch.txOrDb(ctx).ExecContext(ctx, query); err != nil { return errorj.CreateSchemaError.Wrap(err, "failed to create db schema"). @@ -420,6 +438,11 @@ func (ch *ClickHouse) createDatabaseIfNotExists(ctx context.Context, db string) Statement: query, }) } + } else if ch.isReplicatedDatabase() { + var engine string + if err := ch.txOrDb(ctx).QueryRowContext(ctx, chDatabaseEngineQuery, db).Scan(&engine); err != nil || engine != "Replicated" { + return fmt.Errorf("databaseEngine=%q requires database %q to use the Replicated engine; got %q (err=%v)", DatabaseEngineReplicated, db, engine, err) + } } return nil } @@ -430,13 +453,6 @@ func (ch *ClickHouse) InitDatabase(ctx context.Context) error { if err != nil { return err } - if ch.isReplicatedDBMode() { - var engine string - err := ch.txOrDb(ctx).QueryRowContext(ctx, chDatabaseEngineQuery, ch.NamespaceName(ch.config.Database)).Scan(&engine) - if err != nil || engine != "Replicated" { - return fmt.Errorf("clusterMode=replicated_db requires database %q to use the Replicated engine; got %q (err=%v)", ch.config.Database, engine, err) - } - } if ch.config.Cluster != "" { var shardNum int err := ch.txOrDb(ctx).QueryRowContext(ctx, chClusterQuery, ch.config.Cluster).Scan(&shardNum) @@ -957,18 +973,18 @@ func (ch *ClickHouse) ReplaceTable(ctx context.Context, targetTableName string, } -// isReplicatedDBMode reports whether the destination targets a database that +// isReplicatedDatabase reports whether the destination targets a database that // uses ClickHouse's Replicated database engine. In that mode the database // itself manages zookeeper paths, replica macros, and DDL replication, so // table engines must be created without explicit path/replica arguments and -// without ON CLUSTER. -func (ch *ClickHouse) isReplicatedDBMode() bool { - return ch.config.Cluster != "" && ch.config.ClusterMode == ClusterModeReplicatedDB +// without ON CLUSTER inside the database. +func (ch *ClickHouse) isReplicatedDatabase() bool { + return ch.config.Cluster != "" && ch.config.DatabaseEngine == DatabaseEngineReplicated } // return ON CLUSTER name clause or "" if config.cluster is empty func (ch *ClickHouse) getOnClusterClause() string { - if ch.config.Cluster == "" || ch.isReplicatedDBMode() { + if ch.config.Cluster == "" || ch.isReplicatedDatabase() { return "" } @@ -1191,7 +1207,7 @@ func (ch *ClickHouse) IsDistributed() bool { type ClickHouseCluster interface { IsDistributed() bool Config() *ClickHouseConfig - isReplicatedDBMode() bool + isReplicatedDatabase() bool } // TableStatementFactory is used for creating CREATE TABLE statements depends on config @@ -1202,7 +1218,7 @@ type TableStatementFactory struct { func NewTableStatementFactory(ch ClickHouseCluster) *TableStatementFactory { var onClusterClause string - if ch.Config().Cluster != "" && !ch.isReplicatedDBMode() { + if ch.Config().Cluster != "" && !ch.isReplicatedDatabase() { onClusterClause = fmt.Sprintf(chOnClusterClauseTemplate, ch.Config().Cluster) } @@ -1246,7 +1262,7 @@ func (tsf TableStatementFactory) CreateTableStatement(namespacePrefix, quotedTab } if config.Cluster != "" { - if tsf.ch.isReplicatedDBMode() { + if tsf.ch.isReplicatedDatabase() { // Replicated database engine manages zookeeper_path and replica_name; pass none. engineStatement = `ENGINE = Replicated` + baseEngine + `()` } else { diff --git a/bulker/bulkerlib/implementations/sql/clickhouse_test.go b/bulker/bulkerlib/implementations/sql/clickhouse_test.go index aefea4cfc..6ba18f3c1 100644 --- a/bulker/bulkerlib/implementations/sql/clickhouse_test.go +++ b/bulker/bulkerlib/implementations/sql/clickhouse_test.go @@ -14,8 +14,8 @@ type fakeChCluster struct { func (f *fakeChCluster) IsDistributed() bool { return f.distributed } func (f *fakeChCluster) Config() *ClickHouseConfig { return f.config } -func (f *fakeChCluster) isReplicatedDBMode() bool { - return f.config.Cluster != "" && f.config.ClusterMode == ClusterModeReplicatedDB +func (f *fakeChCluster) isReplicatedDatabase() bool { + return f.config.Cluster != "" && f.config.DatabaseEngine == DatabaseEngineReplicated } func TestCreateTableStatement_engineVariants(t *testing.T) { @@ -33,28 +33,28 @@ func TestCreateTableStatement_engineVariants(t *testing.T) { mustNotHave: []string{"Replicated", "ON CLUSTER"}, }, { - name: "cluster, default mode, single shard -> path uses 1/", + name: "cluster, default engine, single shard -> path uses 1/", config: &ClickHouseConfig{Database: "db", Cluster: "c"}, distributed: false, mustContain: []string{"ENGINE = ReplicatedMergeTree('/clickhouse/tables/1/db/", "'{replica}')", "ON CLUSTER `c`"}, mustNotHave: []string{"ReplicatedMergeTree()"}, }, { - name: "cluster, default mode, distributed -> path uses {shard}/", + name: "cluster, default engine, distributed -> path uses {shard}/", config: &ClickHouseConfig{Database: "db", Cluster: "c"}, distributed: true, mustContain: []string{"ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/db/", "ON CLUSTER `c`"}, mustNotHave: []string{"ReplicatedMergeTree()"}, }, { - name: "cluster, replicated_db mode -> path-less, no ON CLUSTER", - config: &ClickHouseConfig{Database: "db", Cluster: "c", ClusterMode: ClusterModeReplicatedDB}, + name: "cluster, replicated database -> path-less, no ON CLUSTER", + config: &ClickHouseConfig{Database: "db", Cluster: "c", DatabaseEngine: DatabaseEngineReplicated}, mustContain: []string{"ENGINE = ReplicatedMergeTree()"}, mustNotHave: []string{"/clickhouse/tables/", "{replica}", "ON CLUSTER"}, }, { - name: "cluster, replicated_db mode, distributed -> still path-less, no ON CLUSTER", - config: &ClickHouseConfig{Database: "db", Cluster: "c", ClusterMode: ClusterModeReplicatedDB}, + name: "cluster, replicated database, distributed -> still path-less, no ON CLUSTER", + config: &ClickHouseConfig{Database: "db", Cluster: "c", DatabaseEngine: DatabaseEngineReplicated}, distributed: true, mustContain: []string{"ENGINE = ReplicatedMergeTree()"}, mustNotHave: []string{"/clickhouse/tables/", "{replica}", "ON CLUSTER"}, @@ -85,16 +85,16 @@ func TestCreateTableStatement_engineVariants(t *testing.T) { } } -func TestGetOnClusterClause_replicatedDBMode(t *testing.T) { +func TestGetOnClusterClause_replicatedDatabase(t *testing.T) { cases := []struct { name string config ClickHouseConfig want string }{ {name: "no cluster", config: ClickHouseConfig{}, want: ""}, - {name: "cluster, default mode", config: ClickHouseConfig{Cluster: "c"}, want: " ON CLUSTER `c` "}, - {name: "cluster, on_cluster mode", config: ClickHouseConfig{Cluster: "c", ClusterMode: ClusterModeOnCluster}, want: " ON CLUSTER `c` "}, - {name: "cluster, replicated_db mode", config: ClickHouseConfig{Cluster: "c", ClusterMode: ClusterModeReplicatedDB}, want: ""}, + {name: "cluster, default engine (implicit)", config: ClickHouseConfig{Cluster: "c"}, want: " ON CLUSTER `c` "}, + {name: "cluster, default engine (explicit)", config: ClickHouseConfig{Cluster: "c", DatabaseEngine: DatabaseEngineDefault}, want: " ON CLUSTER `c` "}, + {name: "cluster, replicated engine", config: ClickHouseConfig{Cluster: "c", DatabaseEngine: DatabaseEngineReplicated}, want: ""}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { diff --git a/bulker/bulkerlib/implementations/sql/naming_test.go b/bulker/bulkerlib/implementations/sql/naming_test.go index c2e4cabe8..86e25920f 100644 --- a/bulker/bulkerlib/implementations/sql/naming_test.go +++ b/bulker/bulkerlib/implementations/sql/naming_test.go @@ -21,7 +21,7 @@ func TestNaming(t *testing.T) { expectedTable: ExpectedTable{ Columns: justColumns("id", "name", "_timestamp", "column_c16da609b86c01f16a2c609eac4ccb0c", "column_12b241e808ae6c964a5bb9f1c012e63d", "秒速_センチメートル", "Université Français", "Странное Имя", "Test Name_ DROP DATABASE public_ SELECT 1 from DUAL_", "Test Name", "1test_name", "2", "_unnamed", "lorem_ipsum_dolor_sit_amet_consectetur_adipiscing_elit_sed_do_e", "camelCase", "int", "user", "select", "__ROOT__", "hash", "default", "current_time"), }, - configIds: utils.ArrayExcluding(allBulkerConfigs, RedshiftBulkerTypeId+"_serverless", RedshiftBulkerTypeId+"_iam", RedshiftBulkerTypeId, SnowflakeBulkerTypeId, BigqueryBulkerTypeId, ClickHouseBulkerTypeId, ClickHouseBulkerTypeId+"_cluster", ClickHouseBulkerTypeId+"_cluster_noshards"), + configIds: utils.ArrayExcluding(allBulkerConfigs, RedshiftBulkerTypeId+"_serverless", RedshiftBulkerTypeId+"_iam", RedshiftBulkerTypeId, SnowflakeBulkerTypeId, BigqueryBulkerTypeId, ClickHouseBulkerTypeId, ClickHouseBulkerTypeId+"_cluster", ClickHouseBulkerTypeId+"_cluster_noshards", ClickHouseBulkerTypeId+"_replicated_db"), }, { name: "naming_test1_clickhouse", @@ -34,7 +34,7 @@ func TestNaming(t *testing.T) { expectedTable: ExpectedTable{ Columns: justColumns("id", "name", "_timestamp", "column_c16da609b86c01f16a2c609eac4ccb0c", "column_12b241e808ae6c964a5bb9f1c012e63d", "秒速_センチメートル", "Université Français", "Странное Имя", "Test Name_ DROP DATABASE public_ SELECT 1 from DUAL_", "Test Name", "1test_name", "2", "_unnamed", "lorem_ipsum_dolor_sit_amet_consectetur_adipiscing_elit_sed_do_eiusmod_tempor_incididunt_ut_labore_et_dolore_magna_aliqua_ut_eni", "camelCase", "int", "user", "select", "__ROOT__", "hash", "default", "current_time"), }, - configIds: []string{ClickHouseBulkerTypeId}, + configIds: []string{ClickHouseBulkerTypeId, ClickHouseBulkerTypeId + "_replicated_db"}, }, { name: "naming_test1_case_redshift", diff --git a/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/ch_replicated_db_container.go b/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/ch_replicated_db_container.go new file mode 100644 index 000000000..4230ff39b --- /dev/null +++ b/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/ch_replicated_db_container.go @@ -0,0 +1,123 @@ +package clickhouse_replicated_db + +import ( + "context" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" + + "github.com/jitsucom/bulker/jitsubase/logging" + "github.com/testcontainers/testcontainers-go" + tc "github.com/testcontainers/testcontainers-go/modules/compose" +) + +const ( + chDatabase = "jitsu_replicated" + chCluster = "replicated_cluster" +) + +var ( + chHostsHTTP = []string{"localhost:8223", "localhost:8224"} + chHostsNative = []string{"localhost:9100", "localhost:9101"} +) + +// ClickHouseReplicatedDBContainer is a ClickHouse cluster + ZooKeeper testcontainer intended for +// exercising Jitsu's `databaseEngine=replicated` code path. Topology: 1 shard × 2 replicas — enough +// to exercise Replicated-database DDL propagation without straining shared CI resources. +type ClickHouseReplicatedDBContainer struct { + Identifier string + Container testcontainers.Container + Compose tc.ComposeStack + Context context.Context + + Cluster string + Hosts []string + HostsHTTP []string + Database string +} + +// NewClickhouseReplicatedDBContainer brings up the cluster compose and pre-creates the destination +// database with ENGINE = Replicated so the bulker test suite can connect against it. +func NewClickhouseReplicatedDBContainer(ctx context.Context) (*ClickHouseReplicatedDBContainer, error) { + composeFilePaths := "testcontainers/clickhouse_replicated_db/docker-compose.yml" + identifier := "bulker_clickhouse_replicated_db_compose" + + compose, err := tc.NewDockerComposeWith(tc.WithStackFiles(composeFilePaths), tc.StackIdentifier(identifier)) + if err != nil { + logging.Errorf("couldnt down docker compose: %s : %v", identifier, err) + } + err = compose.Down(ctx) + if err != nil { + logging.Errorf("couldnt down docker compose: %s : %v", identifier, err) + } + + compose, err = tc.NewDockerComposeWith(tc.WithStackFiles(composeFilePaths), tc.StackIdentifier(identifier)) + if err != nil { + return nil, fmt.Errorf("could not run compose file: %v - %v", composeFilePaths, err) + } + err = compose.Up(ctx, tc.Wait(true)) + if err != nil { + return nil, fmt.Errorf("could not run compose file: %v - %v", composeFilePaths, err) + } + // Pre-create the destination Replicated database. The ClickHouse Go driver opens its connection + // against config.Database, which must exist before NewClickHouse can ping. In production users + // pre-create their Replicated DB; here we mirror that one-time setup so the rest of the bulker + // test suite (which exercises tables, namespaces, schema changes) starts from the same state. + if err := createReplicatedDatabase(chHostsHTTP[0], chDatabase, chCluster); err != nil { + _ = compose.Down(ctx) + return nil, fmt.Errorf("could not create replicated database: %v", err) + } + return &ClickHouseReplicatedDBContainer{ + Identifier: identifier, + Compose: compose, + Context: ctx, + Hosts: chHostsNative, + HostsHTTP: chHostsHTTP, + Database: chDatabase, + Cluster: chCluster, + }, nil +} + +// createReplicatedDatabase issues `CREATE DATABASE IF NOT EXISTS ON CLUSTER +// ENGINE = Replicated(...)` against the first node, retrying briefly while the cluster settles. +func createReplicatedDatabase(httpAddr, db, cluster string) error { + stmt := fmt.Sprintf( + "CREATE DATABASE IF NOT EXISTS %s ON CLUSTER %s ENGINE = Replicated('/clickhouse/databases/%s', '{shard}', '{replica}')", + db, cluster, db, + ) + endpoint := "http://" + httpAddr + "/?" + url.Values{"query": []string{stmt}}.Encode() + + var lastErr error + deadline := time.Now().Add(60 * time.Second) + for time.Now().Before(deadline) { + resp, err := http.Post(endpoint, "text/plain", strings.NewReader("")) + if err != nil { + lastErr = err + time.Sleep(2 * time.Second) + continue + } + body, _ := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if resp.StatusCode == http.StatusOK { + return nil + } + lastErr = fmt.Errorf("status %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + time.Sleep(2 * time.Second) + } + return lastErr +} + +// Close terminates the underlying compose stack. +func (ch *ClickHouseReplicatedDBContainer) Close() error { + if ch.Compose != nil { + execError := ch.Compose.Down(context.Background()) + err := execError.Error + if err != nil { + return fmt.Errorf("could down docker compose: %s", ch.Identifier) + } + } + return nil +} diff --git a/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/clickhouse_repl01/config.xml b/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/clickhouse_repl01/config.xml new file mode 100644 index 000000000..0e1fe9c1d --- /dev/null +++ b/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/clickhouse_repl01/config.xml @@ -0,0 +1,80 @@ + + + + debug + true + + + + + + system + query_log
+ + + 0.0.0.0 + 8123 + 9000 + clickhouse_repl01 + 9009 + + 4096 + 3 + 100 + 8589934592 + 5368709120 + + /var/lib/clickhouse/ + /var/lib/clickhouse/tmp/ + /var/lib/clickhouse/user_files/ + + + + /etc/clickhouse-server/users.xml + + + /var/lib/clickhouse/access/ + + + default + default + UTC + false + + + + + true + + clickhouse_repl01 + 9000 + + + clickhouse_repl02 + 9000 + + + + + + + 90000 + 60000 + + zookeeper + 2181 + + + + + replicated_cluster + 01 + clickhouse_repl01 + + + + /clickhouse/task_queue/ddl + + + /var/lib/clickhouse/format_schemas/ + diff --git a/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/clickhouse_repl01/users.xml b/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/clickhouse_repl01/users.xml new file mode 100644 index 000000000..c5b2f0356 --- /dev/null +++ b/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/clickhouse_repl01/users.xml @@ -0,0 +1,36 @@ + + + + + 10000000000 + 0 + in_order + 1 + + + + + + + default + + ::/0 + + default + 1 + + + + + + + 3600 + 0 + 0 + 0 + 0 + 0 + + + + diff --git a/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/clickhouse_repl02/config.xml b/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/clickhouse_repl02/config.xml new file mode 100644 index 000000000..49da5c076 --- /dev/null +++ b/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/clickhouse_repl02/config.xml @@ -0,0 +1,80 @@ + + + + debug + true + + + + + + system + query_log
+ + + 0.0.0.0 + 8123 + 9000 + clickhouse_repl02 + 9009 + + 4096 + 3 + 100 + 8589934592 + 5368709120 + + /var/lib/clickhouse/ + /var/lib/clickhouse/tmp/ + /var/lib/clickhouse/user_files/ + + + + /etc/clickhouse-server/users.xml + + + /var/lib/clickhouse/access/ + + + default + default + UTC + false + + + + + true + + clickhouse_repl01 + 9000 + + + clickhouse_repl02 + 9000 + + + + + + + 90000 + 60000 + + zookeeper + 2181 + + + + + replicated_cluster + 01 + clickhouse_repl02 + + + + /clickhouse/task_queue/ddl + + + /var/lib/clickhouse/format_schemas/ + diff --git a/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/clickhouse_repl02/users.xml b/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/clickhouse_repl02/users.xml new file mode 100644 index 000000000..c5b2f0356 --- /dev/null +++ b/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/clickhouse_repl02/users.xml @@ -0,0 +1,36 @@ + + + + + 10000000000 + 0 + in_order + 1 + + + + + + + default + + ::/0 + + default + 1 + + + + + + + 3600 + 0 + 0 + 0 + 0 + 0 + + + + diff --git a/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/docker-compose.yml b/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/docker-compose.yml new file mode 100644 index 000000000..27a634288 --- /dev/null +++ b/bulker/bulkerlib/implementations/sql/testcontainers/clickhouse_replicated_db/docker-compose.yml @@ -0,0 +1,66 @@ +services: + zookeeper: + image: zookeeper:3.9 + container_name: zookeeper_repl + hostname: zookeeper + healthcheck: + test: ["CMD-SHELL", "zkCli.sh -server zookeeper:2181 ls /"] + interval: 1s + timeout: 3s + retries: 30 + networks: + clickhouse-replicated-network: + ipv4_address: 172.25.0.10 + clickhouse_repl01: + image: clickhouse/clickhouse-server:25.4-alpine + container_name: clickhouse_repl01 + hostname: clickhouse_repl01 + networks: + clickhouse-replicated-network: + ipv4_address: 172.25.0.11 + ports: + - "8223:8123" + - "9100:9000" + volumes: + - ./clickhouse_repl01:/etc/clickhouse-server + healthcheck: + test: [ "CMD-SHELL", "clickhouse-client --host clickhouse_repl01 --query 'SELECT 1'" ] + interval: 1s + timeout: 3s + retries: 30 + environment: + CLICKHOUSE_USER: default + CLICKHOUSE_SKIP_USER_SETUP: 1 + depends_on: + zookeeper: + condition: service_healthy + clickhouse_repl02: + image: clickhouse/clickhouse-server:25.4-alpine + container_name: clickhouse_repl02 + hostname: clickhouse_repl02 + networks: + clickhouse-replicated-network: + ipv4_address: 172.25.0.12 + ports: + - "8224:8123" + - "9101:9000" + volumes: + - ./clickhouse_repl02:/etc/clickhouse-server + healthcheck: + test: [ "CMD-SHELL", "clickhouse-client --host clickhouse_repl02 --query 'SELECT 1'" ] + interval: 1s + timeout: 3s + retries: 30 + environment: + CLICKHOUSE_USER: default + CLICKHOUSE_SKIP_USER_SETUP: 1 + depends_on: + zookeeper: + condition: service_healthy + +networks: + clickhouse-replicated-network: + name: clickhouse-replicated-network + ipam: + config: + - subnet: 172.25.0.0/24 diff --git a/webapps/console/lib/schema/destinations.tsx b/webapps/console/lib/schema/destinations.tsx index e7d391372..1ee7cee62 100644 --- a/webapps/console/lib/schema/destinations.tsx +++ b/webapps/console/lib/schema/destinations.tsx @@ -276,11 +276,11 @@ export const ClickhouseCredentials = z.object({ .string() .optional() .describe("Name of cluster to use.
For ClickHouse Cloud or single-node setups, leave this field empty."), - clusterMode: z - .enum(["on_cluster", "replicated_db"]) + databaseEngine: z + .enum(["default", "replicated"]) .optional() .describe( - "Cluster mode::How tables are created on the cluster. on_cluster (default) issues ON CLUSTER + ReplicatedMergeTree with explicit ZooKeeper paths. Choose replicated_db when the target database itself uses the Replicated database engine — Jitsu will then emit path-less ReplicatedMergeTree and skip ON CLUSTER. Ignored when cluster is empty." + "Database engine::ClickHouse database engine used by the destination database. default covers Atomic (self-hosted) and Shared (ClickHouse Cloud); Jitsu creates tables as ReplicatedMergeTree with explicit ZooKeeper paths and ON CLUSTER. Choose replicated when the destination database itself uses the Replicated engine — Jitsu then emits path-less ReplicatedMergeTree and skips ON CLUSTER on DDL inside the database (the engine replicates DDL itself). Ignored when cluster is empty." ), database: z.string().default("default").describe("Name of the database to use"), parameters: z @@ -428,7 +428,7 @@ export const coreDestinations: DestinationType[] = [ loadAsJson: { hidden: true, }, - clusterMode: { + databaseEngine: { hidden: obj => !obj.cluster, }, password: {