feat: allow configuring clusterMode

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)
 

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 + `()`

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)
+			}
+		})
+	}
+}

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",