diff --git a/bulker/bulkerlib/implementations/sql/bulker_test.go b/bulker/bulkerlib/implementations/sql/bulker_test.go index d50f5a42a..9c879079f 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 1ff57db5d..cdc8bc727 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,34 @@ const ( ClickHouseProtocolHTTPS ClickHouseProtocol = "https" ) +// 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 ( + // 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"` - 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"` @@ -401,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"). @@ -411,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 } @@ -484,7 +516,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 +973,18 @@ func (ch *ClickHouse) ReplaceTable(ctx context.Context, targetTableName string, } +// 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 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 == "" { + if ch.config.Cluster == "" || ch.isReplicatedDatabase() { return "" } @@ -1165,6 +1207,7 @@ func (ch *ClickHouse) IsDistributed() bool { type ClickHouseCluster interface { IsDistributed() bool Config() *ClickHouseConfig + isReplicatedDatabase() bool } // TableStatementFactory is used for creating CREATE TABLE statements depends on config @@ -1175,7 +1218,7 @@ type TableStatementFactory struct { func NewTableStatementFactory(ch ClickHouseCluster) *TableStatementFactory { var onClusterClause string - if ch.Config().Cluster != "" { + if ch.Config().Cluster != "" && !ch.isReplicatedDatabase() { onClusterClause = fmt.Sprintf(chOnClusterClauseTemplate, ch.Config().Cluster) } @@ -1219,13 +1262,18 @@ func (tsf TableStatementFactory) CreateTableStatement(namespacePrefix, quotedTab } if config.Cluster != "" { - shardsMacros := "{shard}/" - if !tsf.ch.IsDistributed() { - shardsMacros = "1/" + if tsf.ch.isReplicatedDatabase() { + // 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..6ba18f3c1 --- /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) isReplicatedDatabase() bool { + return f.config.Cluster != "" && f.config.DatabaseEngine == DatabaseEngineReplicated +} + +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 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 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 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 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"}, + }, + } + + 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_replicatedDatabase(t *testing.T) { + cases := []struct { + name string + config ClickHouseConfig + want string + }{ + {name: "no cluster", config: ClickHouseConfig{}, 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) { + 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/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 bf56dab71..9d3513d5d 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."), + databaseEngine: z + .enum(["default", "replicated"]) + .optional() + .describe( + "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 .object({}) @@ -422,6 +428,9 @@ export const coreDestinations: DestinationType[] = [ loadAsJson: { hidden: true, }, + databaseEngine: { + hidden: obj => !obj.cluster, + }, password: { password: true, },