engines: ReplicatedReplacingMergeTree, ReplicatedMergeTree (#2949)

Author: serprex

flow/connectors/clickhouse/normalize.go

@@ -159,14 +159,29 @@ func generateCreateTableSQLForNormalizedTable(
 	}
 
 	var engine string
-	if tableMapping == nil {
+	tmEngine := protos.TableEngine_CH_ENGINE_REPLACING_MERGE_TREE
+	if tableMapping != nil {
+		tmEngine = tableMapping.Engine
+	}
+	switch tmEngine {
+	case protos.TableEngine_CH_ENGINE_REPLACING_MERGE_TREE:
 		engine = fmt.Sprintf("ReplacingMergeTree(%s)", peerdb_clickhouse.QuoteIdentifier(versionColName))
-	} else if tableMapping.Engine == protos.TableEngine_CH_ENGINE_MERGE_TREE {
+	case protos.TableEngine_CH_ENGINE_MERGE_TREE:
 		engine = "MergeTree()"
-	} else if tableMapping.Engine == protos.TableEngine_CH_ENGINE_NULL {
+	case protos.TableEngine_CH_ENGINE_REPLICATED_REPLACING_MERGE_TREE:
+		engine = fmt.Sprintf(
+			"ReplicatedReplacingMergeTree('/clickhouse/tables/{shard}/{database}/%s','{replica}',%s)",
+			peerdb_clickhouse.EscapeStr(tableIdentifier),
+			peerdb_clickhouse.QuoteIdentifier(versionColName),
+		)
+	case protos.TableEngine_CH_ENGINE_REPLICATED_MERGE_TREE:
+		engine = fmt.Sprintf(
+			"ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/%s','{replica}',%s)",
+			peerdb_clickhouse.EscapeStr(tableIdentifier),
+			peerdb_clickhouse.QuoteIdentifier(versionColName),
+		)
+	case protos.TableEngine_CH_ENGINE_NULL:
 		engine = "Null"
-	} else {
-		engine = fmt.Sprintf("ReplacingMergeTree(%s)", peerdb_clickhouse.QuoteIdentifier(versionColName))
 	}
 
 	// add sign and version columns
@@ -179,7 +194,7 @@ func generateCreateTableSQLForNormalizedTable(
 		orderByColumns = append([]string{sourceSchemaColName}, orderByColumns...)
 	}
 
-	if tableMapping == nil || tableMapping.Engine != protos.TableEngine_CH_ENGINE_NULL {
+	if tmEngine != protos.TableEngine_CH_ENGINE_NULL {
 		if len(orderByColumns) > 0 {
 			orderByStr := strings.Join(orderByColumns, ",")
 

flow/shared/clickhouse/validation.go

@@ -12,7 +12,7 @@ import (
 	"go.temporal.io/sdk/log"
 )
 
-var acceptableTableEngines = []string{"ReplacingMergeTree", "MergeTree", "Null"}
+var acceptableTableEngines = []string{"ReplacingMergeTree", "MergeTree", "ReplicatedReplacingMergeTree", "ReplicatedMergeTree", "Null"}
 
 func CheckIfClickHouseCloudHasSharedMergeTreeEnabled(ctx context.Context, logger log.Logger,
 	conn clickhouse.Conn,

protos/flow.proto

@@ -251,6 +251,8 @@ enum TableEngine {
   CH_ENGINE_REPLACING_MERGE_TREE = 0;
   CH_ENGINE_MERGE_TREE = 1;
   CH_ENGINE_NULL = 2;
+  CH_ENGINE_REPLICATED_REPLACING_MERGE_TREE = 3;
+  CH_ENGINE_REPLICATED_MERGE_TREE = 4;
 }
 
 // protos for qrep

ui/app/mirrors/create/cdc/schemabox.tsx

@@ -245,6 +245,11 @@ export default function SchemaBox({
   const engineOptions = [
     { value: 'CH_ENGINE_REPLACING_MERGE_TREE', label: 'ReplacingMergeTree' },
     { value: 'CH_ENGINE_MERGE_TREE', label: 'MergeTree' },
+    {
+      value: 'CH_ENGINE_REPLICATED_REPLACING_MERGE_TREE',
+      label: 'ReplicatedReplacingMergeTree',
+    },
+    { value: 'CH_ENGINE_REPLICATED_MERGE_TREE', label: 'ReplicatedMergeTree' },
     { value: 'CH_ENGINE_NULL', label: 'Null' },
   ];