chore(clickhouse): use replicated merge tree (#293)

* chore: migrate to replicated merge tree * chore: templatize replicated merge tree prefix --------- Signed-off-by: Prashant Shahi <[email protected]>
SigNoz · Apr 4, 2024 · 2916d1c · 2916d1c
1 parent e738876
commit 2916d1c
Show file tree

Hide file tree

Showing 28 changed files with 65 additions and 37 deletions.
diff --git a/cmd/signozschemamigrator/migrate.go b/cmd/signozschemamigrator/migrate.go
@@ -41,6 +41,7 @@ func main() {
 	f.String("cluster-name", "cluster", "Cluster name to use while running migrations")
 	f.Bool("disable-duration-sort-feature", false, "Flag to disable the duration sort feature. Defaults to false.")
 	f.Bool("disable-timestamp-sort-feature", false, "Flag to disable the timestamp sort feature. Defaults to false.")
+	f.Bool("replication", false, "Flag to enable replication. Defaults to false.")
 	f.Bool("verbose", false, "Flag to enable verbose logging. Defaults to false.")
 
 	err := f.Parse(os.Args[1:])
@@ -73,6 +74,11 @@ func main() {
 		logger.Fatal("Failed to get verbose flag from args", zap.Error(err))
 	}
 
+	replicationEnabled, err := f.GetBool("replication")
+	if err != nil {
+		logger.Fatal("Failed to get replication flag from args", zap.Error(err))
+	}
+
 	if dsn == "" {
 		logger.Fatal("dsn is a required field")
 	}
@@ -99,7 +105,17 @@ func main() {
 	}
 	logger.Info("Successfully set env var SIGNOZ_CLUSTER ", zap.String("cluster-name", clusterNameFromEnv))
 
-	manager, err := migrationmanager.New(dsn, clusterName, disableDurationSortFeature, disableTimestampSortFeature, verboseLoggingEnabled)
+	// set SIGNOZ_REPLICATED env var so that golang-migrate can use it
+	// the value of this env would replace all occurences of {{.SIGNOZ_REPLICATED}} in the migration files
+	if replicationEnabled {
+		logger.Info("Setting env var SIGNOZ_REPLICATED", zap.Bool("replication", replicationEnabled))
+		err = os.Setenv("SIGNOZ_REPLICATED", "Replicated")
+		if err != nil {
+			logger.Fatal("Failed to set env var SIGNOZ_REPLICATED", zap.Error(err))
+		}
+	}
+
+	manager, err := migrationmanager.New(dsn, clusterName, disableDurationSortFeature, disableTimestampSortFeature, verboseLoggingEnabled, replicationEnabled)
 	if err != nil {
 		logger.Fatal("Failed to create migration manager", zap.Error(err))
 	}

diff --git a/migrationmanager/manager.go b/migrationmanager/manager.go
@@ -17,14 +17,15 @@ type MigrationManager struct {
 	logger    *zap.Logger
 }
 
-func New(dsn string, clusterName string, isDurationSortFeatureDisabled, isTimestampSortFeatureDisabled, verboseLoggingEnabled bool) (*MigrationManager, error) {
+func New(dsn string, clusterName string, isDurationSortFeatureDisabled, isTimestampSortFeatureDisabled, verboseLoggingEnabled, replicationEnabled bool) (*MigrationManager, error) {
 	logger := zap.L().With(zap.String("component", "migrationmanager"))
 	cfg := migrators.MigratorConfig{
 		DSN:                            dsn,
 		ClusterName:                    clusterName,
 		IsDurationSortFeatureDisabled:  isDurationSortFeatureDisabled,
 		IsTimestampSortFeatureDisabled: isTimestampSortFeatureDisabled,
 		VerboseLoggingEnabled:          verboseLoggingEnabled,
+		ReplicationEnabled:             replicationEnabled,
 	}
 
 	logsMigrator, err := createNewMigrator("logs", cfg)

diff --git a/migrationmanager/migrators/basemigrator/migrator.go b/migrationmanager/migrators/basemigrator/migrator.go
@@ -85,7 +85,7 @@ func (m *BaseMigrator) runSqlMigrations(ctx context.Context, migrationFolder, da
 }
 
 func (m *BaseMigrator) buildClickhouseMigrateURL(database string) (string, error) {
-	var clickhouseUrl string
+	var clickhouseUrl, migrationsTableEngine string
 	parsedURL, err := url.Parse(m.Cfg.DSN)
 	if err != nil {
 		return "", err
@@ -102,10 +102,16 @@ func (m *BaseMigrator) buildClickhouseMigrateURL(database string) (string, error
 	username := paramMap["username"]
 	password := paramMap["password"]
 
+	if m.Cfg.ReplicationEnabled {
+		migrationsTableEngine = "ReplicatedMergeTree"
+	} else {
+		migrationsTableEngine = "MergeTree"
+	}
+
 	if len(username) > 0 && len(password) > 0 {
-		clickhouseUrl = fmt.Sprintf("clickhouse://%s:%s@%s/%s?x-multi-statement=true&x-cluster-name=%s&x-migrations-table=schema_migrations&x-migrations-table-engine=MergeTree", username[0], password[0], host, database, m.Cfg.ClusterName)
+		clickhouseUrl = fmt.Sprintf("clickhouse://%s:%s@%s/%s?x-multi-statement=true&x-cluster-name=%s&x-migrations-table=schema_migrations&x-migrations-table-engine=%s", username[0], password[0], host, database, m.Cfg.ClusterName, migrationsTableEngine)
 	} else {
-		clickhouseUrl = fmt.Sprintf("clickhouse://%s/%s?x-multi-statement=true&x-cluster-name=%s&x-migrations-table=schema_migrations&x-migrations-table-engine=MergeTree", host, database, m.Cfg.ClusterName)
+		clickhouseUrl = fmt.Sprintf("clickhouse://%s/%s?x-multi-statement=true&x-cluster-name=%s&x-migrations-table=schema_migrations&x-migrations-table-engine=%s", host, database, m.Cfg.ClusterName, migrationsTableEngine)
 	}
 	return clickhouseUrl, nil
 }

diff --git a/migrationmanager/migrators/logs/migrations/000001_init_db.up.sql b/migrationmanager/migrators/logs/migrations/000001_init_db.up.sql
@@ -17,7 +17,7 @@ CREATE TABLE IF NOT EXISTS signoz_logs.logs ON CLUSTER {{.SIGNOZ_CLUSTER}} (
 	attributes_float64_key Array(String) CODEC(ZSTD(1)),
 	attributes_float64_value Array(Float64) CODEC(ZSTD(1)),
 	INDEX body_idx body TYPE tokenbf_v1(10240, 3, 0) GRANULARITY 4
-) ENGINE MergeTree
+) ENGINE {{.SIGNOZ_REPLICATED}}MergeTree
 PARTITION BY toDate(timestamp / 1000000000)
 ORDER BY (timestamp, id)
 TTL toDateTime(timestamp / 1000000000) + INTERVAL 1296000 SECOND DELETE;
@@ -27,15 +27,15 @@ TTL toDateTime(timestamp / 1000000000) + INTERVAL 1296000 SECOND DELETE;
 CREATE TABLE IF NOT EXISTS signoz_logs.logs_atrribute_keys ON CLUSTER {{.SIGNOZ_CLUSTER}} (
 name String,
 datatype String
-)ENGINE = ReplacingMergeTree
+)ENGINE = {{.SIGNOZ_REPLICATED}}ReplacingMergeTree
 ORDER BY (name, datatype);
 
 
 
 CREATE TABLE IF NOT EXISTS signoz_logs.logs_resource_keys ON CLUSTER {{.SIGNOZ_CLUSTER}} (
 name String,
 datatype String
-)ENGINE = ReplacingMergeTree
+)ENGINE = {{.SIGNOZ_REPLICATED}}ReplacingMergeTree
 ORDER BY (name, datatype);
 
 

diff --git a/migrationmanager/migrators/logs/migrations/000006_tag_attributes.up.sql b/migrationmanager/migrators/logs/migrations/000006_tag_attributes.up.sql
@@ -6,7 +6,7 @@ CREATE TABLE IF NOT EXISTS signoz_logs.tag_attributes ON CLUSTER {{.SIGNOZ_CLUST
     stringTagValue String CODEC(ZSTD(1)),
     int64TagValue Nullable(Int64) CODEC(ZSTD(1)),
     float64TagValue Nullable(Float64) CODEC(ZSTD(1))
-) ENGINE ReplacingMergeTree
+) ENGINE {{.SIGNOZ_REPLICATED}}ReplacingMergeTree
 ORDER BY (tagKey, tagType, tagDataType, stringTagValue, int64TagValue, float64TagValue)
 TTL toDateTime(timestamp) + INTERVAL 172800 SECOND DELETE
 SETTINGS ttl_only_drop_parts = 1, allow_nullable_key = 1;

diff --git a/migrationmanager/migrators/logs/migrations/000009_add_usage.up.sql b/migrationmanager/migrators/logs/migrations/000009_add_usage.up.sql
@@ -4,7 +4,7 @@ CREATE TABLE IF NOT EXISTS signoz_logs.usage ON CLUSTER {{ .SIGNOZ_CLUSTER }} (
     exporter_id String,
     timestamp DateTime,
     data String
-) ENGINE MergeTree()
+) ENGINE {{.SIGNOZ_REPLICATED}}MergeTree()
 ORDER BY (tenant, collector_id, exporter_id, timestamp)
 TTL timestamp + INTERVAL 3 DAY;
 

diff --git a/migrationmanager/migrators/metrics/migrations/000001_init_db.up.sql b/migrationmanager/migrators/metrics/migrations/000001_init_db.up.sql
@@ -4,7 +4,7 @@ CREATE TABLE IF NOT EXISTS signoz_metrics.samples_v2 ON CLUSTER {{.SIGNOZ_CLUSTE
                         timestamp_ms Int64 Codec(DoubleDelta, LZ4),
                         value Float64 Codec(Gorilla, LZ4)
                 )
-                ENGINE = MergeTree
+                ENGINE = {{.SIGNOZ_REPLICATED}}MergeTree
                         PARTITION BY toDate(timestamp_ms / 1000)
                         ORDER BY (metric_name, fingerprint, timestamp_ms)
                          TTL toDateTime(timestamp_ms/1000) + INTERVAL 2592000 SECOND DELETE;
@@ -21,7 +21,7 @@ CREATE TABLE IF NOT EXISTS signoz_metrics.time_series_v2 ON CLUSTER {{.SIGNOZ_CL
                         timestamp_ms Int64 Codec(DoubleDelta, LZ4),
                         labels String Codec(ZSTD(5))
                 )
-                ENGINE = ReplacingMergeTree
+                ENGINE = {{.SIGNOZ_REPLICATED}}ReplacingMergeTree
                         PARTITION BY toDate(timestamp_ms / 1000)
                         ORDER BY (metric_name, fingerprint)
                         TTL toDateTime(timestamp_ms/1000) + INTERVAL 2592000 SECOND DELETE;
@@ -48,7 +48,7 @@ CREATE TABLE IF NOT EXISTS signoz_metrics.time_series_v3 ON CLUSTER {{.SIGNOZ_CL
                         timestamp_ms Int64 CODEC(Delta, ZSTD),
                         labels String CODEC(ZSTD(5))
                 )
-                ENGINE = ReplacingMergeTree
+                ENGINE = {{.SIGNOZ_REPLICATED}}ReplacingMergeTree
                         PARTITION BY toDate(timestamp_ms / 1000)
                         ORDER BY (env, temporality, metric_name, fingerprint);
 

diff --git a/migrationmanager/migrators/metrics/migrations/000003_ts_v4.up.sql b/migrationmanager/migrators/metrics/migrations/000003_ts_v4.up.sql
@@ -11,7 +11,7 @@ CREATE TABLE IF NOT EXISTS signoz_metrics.time_series_v4 ON CLUSTER {{.SIGNOZ_CL
     labels String CODEC(ZSTD(5)),
     INDEX idx_labels labels TYPE ngrambf_v1(4, 1024, 3, 0) GRANULARITY 1
 )
-ENGINE = ReplacingMergeTree
+ENGINE = {{.SIGNOZ_REPLICATED}}ReplacingMergeTree
         PARTITION BY toDate(unix_milli / 1000)
         ORDER BY (env, temporality, metric_name, fingerprint, unix_milli)
         TTL toDateTime(unix_milli/1000) + INTERVAL 2592000 SECOND DELETE

diff --git a/migrationmanager/migrators/metrics/migrations/000004_ts_v4_high_precision.up.sql b/migrationmanager/migrators/metrics/migrations/000004_ts_v4_high_precision.up.sql
@@ -11,7 +11,7 @@ CREATE TABLE IF NOT EXISTS signoz_metrics.time_series_v4_6hrs ON CLUSTER {{.SIGN
     labels String CODEC(ZSTD(5)),
     INDEX idx_labels labels TYPE ngrambf_v1(4, 1024, 3, 0) GRANULARITY 1
 )
-ENGINE = ReplacingMergeTree
+ENGINE = {{.SIGNOZ_REPLICATED}}ReplacingMergeTree
         PARTITION BY toDate(unix_milli / 1000)
         ORDER BY (env, temporality, metric_name, fingerprint, unix_milli)
         TTL toDateTime(unix_milli/1000) + INTERVAL 2592000 SECOND DELETE
@@ -32,7 +32,7 @@ CREATE TABLE IF NOT EXISTS signoz_metrics.time_series_v4_1day ON CLUSTER {{.SIGN
     labels String CODEC(ZSTD(5)),
     INDEX idx_labels labels TYPE ngrambf_v1(4, 1024, 3, 0) GRANULARITY 1
 )
-ENGINE = ReplacingMergeTree
+ENGINE = {{.SIGNOZ_REPLICATED}}ReplacingMergeTree
         PARTITION BY toDate(unix_milli / 1000)
         ORDER BY (env, temporality, metric_name, fingerprint, unix_milli)
         TTL toDateTime(unix_milli/1000) + INTERVAL 2592000 SECOND DELETE

diff --git a/migrationmanager/migrators/metrics/migrations/000005_samples_v4.up.sql b/migrationmanager/migrators/metrics/migrations/000005_samples_v4.up.sql
@@ -6,7 +6,7 @@ CREATE TABLE IF NOT EXISTS signoz_metrics.samples_v4 ON CLUSTER {{.SIGNOZ_CLUSTE
     unix_milli Int64 CODEC(DoubleDelta, ZSTD),
     value Float64 Codec(Gorilla, ZSTD)
 )
-ENGINE = MergeTree
+ENGINE = {{.SIGNOZ_REPLICATED}}MergeTree
         PARTITION BY toDate(unix_milli / 1000)
         ORDER BY (env, temporality, metric_name, fingerprint, unix_milli)
         TTL toDateTime(unix_milli/1000) + INTERVAL 2592000 SECOND DELETE

diff --git a/migrationmanager/migrators/metrics/migrations/000006_add_usage.up.sql b/migrationmanager/migrators/metrics/migrations/000006_add_usage.up.sql
@@ -4,7 +4,7 @@ CREATE TABLE IF NOT EXISTS signoz_metrics.usage ON CLUSTER {{ .SIGNOZ_CLUSTER }}
     exporter_id String,
     timestamp DateTime,
     data String
-) ENGINE MergeTree()
+) ENGINE {{.SIGNOZ_REPLICATED}}MergeTree()
 ORDER BY (tenant, collector_id, exporter_id, timestamp)
 TTL timestamp + INTERVAL 3 DAY;
 

diff --git a/migrationmanager/migrators/metrics/migrations/000007_exp_hist.up.sql b/migrationmanager/migrators/metrics/migrations/000007_exp_hist.up.sql
@@ -10,7 +10,7 @@ CREATE TABLE IF NOT EXISTS signoz_metrics.exp_hist ON CLUSTER {{.SIGNOZ_CLUSTER}
     max Float64 Codec(Gorilla, ZSTD),
     sketch AggregateFunction(quantilesDD(0.01, 0.5, 0.75, 0.9, 0.95, 0.99), UInt64) CODEC(ZSTD(1))
 )
-ENGINE = MergeTree
+ENGINE = {{.SIGNOZ_REPLICATED}}MergeTree
         PARTITION BY toDate(unix_milli / 1000)
         ORDER BY (env, temporality, metric_name, fingerprint, unix_milli)
         TTL toDateTime(unix_milli/1000) + INTERVAL 2592000 SECOND DELETE

diff --git a/migrationmanager/migrators/migrators.go b/migrationmanager/migrators/migrators.go
@@ -12,6 +12,7 @@ type MigratorConfig struct {
 	IsDurationSortFeatureDisabled  bool
 	IsTimestampSortFeatureDisabled bool
 	VerboseLoggingEnabled          bool
+	ReplicationEnabled             bool
 }
 
 type Migrator interface {

diff --git a/migrationmanager/migrators/traces/migrations/000001_init_db.up.sql b/migrationmanager/migrators/traces/migrations/000001_init_db.up.sql
@@ -26,6 +26,6 @@ CREATE TABLE IF NOT EXISTS signoz_traces.signoz_index ON CLUSTER {{.SIGNOZ_CLUST
   INDEX idx_tagsKeys tagsKeys TYPE bloom_filter(0.01) GRANULARITY 64,
   INDEX idx_tagsValues tagsValues TYPE bloom_filter(0.01) GRANULARITY 64,
   INDEX idx_duration durationNano TYPE minmax GRANULARITY 1
-) ENGINE MergeTree
+) ENGINE {{.SIGNOZ_REPLICATED}}MergeTree
 PARTITION BY toDate(timestamp)
 ORDER BY (serviceName, -toUnixTimestamp(timestamp));
diff --git a/migrationmanager/migrators/traces/migrations/000003_init_schema.up.sql b/migrationmanager/migrators/traces/migrations/000003_init_schema.up.sql
@@ -13,7 +13,7 @@ CREATE TABLE IF NOT EXISTS signoz_traces.signoz_error_index ON CLUSTER {{.SIGNOZ
   INDEX idx_service serviceName TYPE bloom_filter GRANULARITY 4,
   INDEX idx_message exceptionMessage TYPE bloom_filter GRANULARITY 4,
   INDEX idx_type exceptionType TYPE bloom_filter GRANULARITY 4
-) ENGINE MergeTree
+) ENGINE {{.SIGNOZ_REPLICATED}}MergeTree
 PARTITION BY toDate(timestamp)
 ORDER BY (exceptionType, exceptionMessage, serviceName, -toUnixTimestamp(timestamp));
 
diff --git a/migrationmanager/migrators/traces/migrations/000007_init_signoz_index_v2.up.sql b/migrationmanager/migrators/traces/migrations/000007_init_signoz_index_v2.up.sql
@@ -39,7 +39,7 @@ CREATE TABLE IF NOT EXISTS signoz_traces.signoz_index_v2 ON CLUSTER {{.SIGNOZ_CL
   INDEX idx_httpHost httpHost TYPE bloom_filter GRANULARITY 4,
   INDEX idx_httpMethod httpMethod TYPE bloom_filter GRANULARITY 4,
   INDEX idx_timestamp timestamp TYPE minmax GRANULARITY 1
-) ENGINE MergeTree
+) ENGINE {{.SIGNOZ_REPLICATED}}MergeTree
 PARTITION BY toDate(timestamp)
 PRIMARY KEY (serviceName, hasError, toStartOfHour(timestamp), name)
 ORDER BY (serviceName, hasError, toStartOfHour(timestamp), name, timestamp)

diff --git a/migrationmanager/migrators/traces/migrations/000008_init_signoz_spans.up.sql b/migrationmanager/migrators/traces/migrations/000008_init_signoz_spans.up.sql
@@ -2,7 +2,7 @@ CREATE TABLE IF NOT EXISTS signoz_traces.signoz_spans ON CLUSTER {{.SIGNOZ_CLUST
     timestamp DateTime64(9) CODEC(DoubleDelta, LZ4),
     traceID FixedString(32) CODEC(ZSTD(1)),
     model String CODEC(ZSTD(9))
-) ENGINE MergeTree
+) ENGINE {{.SIGNOZ_REPLICATED}}MergeTree
 PARTITION BY toDate(timestamp)
 ORDER BY traceID
 TTL toDateTime(timestamp) + INTERVAL 1296000 SECOND DELETE

diff --git a/migrationmanager/migrators/traces/migrations/000009_init_signoz_error_index.up.sql b/migrationmanager/migrators/traces/migrations/000009_init_signoz_error_index.up.sql
@@ -13,7 +13,7 @@ CREATE TABLE IF NOT EXISTS signoz_traces.signoz_error_index ON CLUSTER {{.SIGNOZ
   INDEX idx_service serviceName TYPE bloom_filter GRANULARITY 4,
   INDEX idx_message exceptionMessage TYPE bloom_filter GRANULARITY 4,
   INDEX idx_type exceptionType TYPE bloom_filter GRANULARITY 4
-) ENGINE MergeTree
+) ENGINE {{.SIGNOZ_REPLICATED}}MergeTree
 PARTITION BY toDate(timestamp)
 ORDER BY (serviceName, exceptionType, exceptionMessage, timestamp);
 
diff --git a/migrationmanager/migrators/traces/migrations/000011_init_durationMV.up.sql b/migrationmanager/migrators/traces/migrations/000011_init_durationMV.up.sql
@@ -31,7 +31,7 @@ CREATE TABLE IF NOT EXISTS signoz_traces.durationSort ON CLUSTER {{.SIGNOZ_CLUST
   INDEX idx_httpHost httpHost TYPE bloom_filter GRANULARITY 4,
   INDEX idx_httpMethod httpMethod TYPE bloom_filter GRANULARITY 4,
   INDEX idx_timestamp timestamp TYPE minmax GRANULARITY 1
-) ENGINE MergeTree
+) ENGINE {{.SIGNOZ_REPLICATED}}MergeTree
 PARTITION BY toDate(timestamp)
 ORDER BY (durationNano, timestamp)
 TTL toDateTime(timestamp) + INTERVAL 1296000 SECOND DELETE

diff --git a/migrationmanager/migrators/traces/migrations/000012_init_signoz_error_index_v2.up.sql b/migrationmanager/migrators/traces/migrations/000012_init_signoz_error_index_v2.up.sql
@@ -10,7 +10,7 @@ CREATE TABLE IF NOT EXISTS signoz_traces.signoz_error_index_v2 ON CLUSTER {{.SIG
   exceptionStacktrace String CODEC(ZSTD(1)),
   exceptionEscaped bool CODEC(T64, ZSTD(1)),
   INDEX idx_error_id errorID TYPE bloom_filter GRANULARITY 4
-) ENGINE MergeTree
+) ENGINE {{.SIGNOZ_REPLICATED}}MergeTree
 PARTITION BY toDate(timestamp)
 ORDER BY (timestamp, groupID)
 TTL toDateTime(timestamp) + INTERVAL 1296000 SECOND DELETE;
diff --git a/migrationmanager/migrators/traces/migrations/000014_add_usage_explorer_mv.up.sql b/migrationmanager/migrators/traces/migrations/000014_add_usage_explorer_mv.up.sql
@@ -2,7 +2,7 @@ CREATE TABLE IF NOT EXISTS signoz_traces.usage_explorer ON CLUSTER {{.SIGNOZ_CLU
   timestamp DateTime64(9) CODEC(DoubleDelta, LZ4),
   service_name LowCardinality(String) CODEC(ZSTD(1)),
   count UInt64 CODEC(T64, ZSTD(1))
-) ENGINE SummingMergeTree
+) ENGINE {{.SIGNOZ_REPLICATED}}SummingMergeTree
 PARTITION BY toDate(timestamp)
 ORDER BY (timestamp, service_name)
 TTL toDateTime(timestamp) + INTERVAL 1296000 SECOND DELETE;

diff --git a/migrationmanager/migrators/traces/migrations/000015_top_level_operations_mv.up.sql b/migrationmanager/migrators/traces/migrations/000015_top_level_operations_mv.up.sql
@@ -1,7 +1,7 @@
 CREATE TABLE IF NOT EXISTS signoz_traces.top_level_operations ON CLUSTER {{.SIGNOZ_CLUSTER}} (
     name LowCardinality(String) CODEC(ZSTD(1)),
     serviceName LowCardinality(String) CODEC(ZSTD(1))
-) ENGINE ReplacingMergeTree
+) ENGINE {{.SIGNOZ_REPLICATED}}ReplacingMergeTree
 ORDER BY (serviceName, name);
 
 

diff --git a/migrationmanager/migrators/traces/migrations/000016_dependency_graph_minutes.up.sql b/migrationmanager/migrators/traces/migrations/000016_dependency_graph_minutes.up.sql
@@ -5,7 +5,7 @@ CREATE TABLE IF NOT EXISTS signoz_traces.dependency_graph_minutes ON CLUSTER {{.
     error_count SimpleAggregateFunction(sum, UInt64) CODEC(T64, ZSTD(1)),
     total_count SimpleAggregateFunction(sum, UInt64) CODEC(T64, ZSTD(1)),
     timestamp DateTime CODEC(DoubleDelta, LZ4)
-) ENGINE AggregatingMergeTree
+) ENGINE {{.SIGNOZ_REPLICATED}}AggregatingMergeTree
 PARTITION BY toDate(timestamp)
 ORDER BY (timestamp, src, dest)
 TTL toDateTime(timestamp) + INTERVAL 1296000 SECOND DELETE;

diff --git a/migrationmanager/migrators/traces/migrations/000022_dependency_graph_minutes_v2.up.sql b/migrationmanager/migrators/traces/migrations/000022_dependency_graph_minutes_v2.up.sql
@@ -8,7 +8,7 @@ CREATE TABLE IF NOT EXISTS signoz_traces.dependency_graph_minutes_v2 ON CLUSTER
     deployment_environment LowCardinality(String) CODEC(ZSTD(1)),
     k8s_cluster_name LowCardinality(String) CODEC(ZSTD(1)),
     k8s_namespace_name LowCardinality(String) CODEC(ZSTD(1))
-) ENGINE AggregatingMergeTree
+) ENGINE {{.SIGNOZ_REPLICATED}}AggregatingMergeTree
 PARTITION BY toDate(timestamp)
 ORDER BY (timestamp, src, dest, deployment_environment, k8s_cluster_name, k8s_namespace_name)
 TTL toDateTime(timestamp) + INTERVAL 1296000 SECOND DELETE;

diff --git a/migrationmanager/migrators/traces/migrations/000023_span_attributes_mv.up.sql b/migrationmanager/migrators/traces/migrations/000023_span_attributes_mv.up.sql
@@ -6,7 +6,7 @@ CREATE TABLE IF NOT EXISTS signoz_traces.span_attributes ON CLUSTER {{.SIGNOZ_CL
     stringTagValue String CODEC(ZSTD(1)),
     float64TagValue Nullable(Float64) CODEC(ZSTD(1)),
     isColumn bool CODEC(ZSTD(1)),
-) ENGINE ReplacingMergeTree
+) ENGINE {{.SIGNOZ_REPLICATED}}ReplacingMergeTree
 ORDER BY (tagKey, tagType, dataType, stringTagValue, float64TagValue, isColumn)
 TTL toDateTime(timestamp) + INTERVAL 172800 SECOND DELETE
 SETTINGS ttl_only_drop_parts = 1, allow_nullable_key = 1;

diff --git a/migrationmanager/migrators/traces/migrations/000024_span_attributes_keys.up.sql b/migrationmanager/migrators/traces/migrations/000024_span_attributes_keys.up.sql
@@ -3,7 +3,7 @@ CREATE TABLE IF NOT EXISTS signoz_traces.span_attributes_keys ON CLUSTER {{.SIGN
     tagType Enum('tag', 'resource') CODEC(ZSTD(1)),
     dataType Enum('string', 'bool', 'float64') CODEC(ZSTD(1)),
     isColumn bool CODEC(ZSTD(1)),
-) ENGINE ReplacingMergeTree
+) ENGINE {{.SIGNOZ_REPLICATED}}ReplacingMergeTree
 ORDER BY (tagKey, tagType, dataType, isColumn);
 
 CREATE TABLE IF NOT EXISTS signoz_traces.distributed_span_attributes_keys ON CLUSTER {{.SIGNOZ_CLUSTER}} AS signoz_traces.span_attributes_keys

diff --git a/migrationmanager/migrators/traces/migrations/000026_add_usage.up.sql b/migrationmanager/migrators/traces/migrations/000026_add_usage.up.sql
@@ -4,7 +4,7 @@ CREATE TABLE IF NOT EXISTS signoz_traces.usage ON CLUSTER {{ .SIGNOZ_CLUSTER }}
     exporter_id String,
     timestamp DateTime,
     data String
-) ENGINE MergeTree()
+) ENGINE {{.SIGNOZ_REPLICATED}}MergeTree()
 ORDER BY (tenant, collector_id, exporter_id, timestamp)
 TTL timestamp + INTERVAL 3 DAY;
 

diff --git a/migrationmanager/migrators/traces/migrator.go b/migrationmanager/migrators/traces/migrator.go
@@ -39,7 +39,7 @@ func (m *TracesMigrator) initFeatures() error {
 			return err
 		}
 	} else {
-		err := enableDurationSortFeature(m.DB, m.Cfg.ClusterName)
+		err := enableDurationSortFeature(m.DB, m.Cfg.ClusterName, m.Cfg.ReplicationEnabled)
 		if err != nil {
 			return err
 		}
@@ -58,7 +58,11 @@ func (m *TracesMigrator) initFeatures() error {
 	return nil
 }
 
-func enableDurationSortFeature(db clickhouse.Conn, cluster string) error {
+func enableDurationSortFeature(db clickhouse.Conn, cluster string, replicationEnabled bool) error {
+	signozReplicated := ""
+	if replicationEnabled {
+		signozReplicated = "Replicated"
+	}
 	err := db.Exec(context.Background(), fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s.%s ON CLUSTER %s( 
 		timestamp DateTime64(9) CODEC(DoubleDelta, LZ4),
 		traceID FixedString(32) CODEC(ZSTD(1)),
@@ -101,11 +105,11 @@ func enableDurationSortFeature(db clickhouse.Conn, cluster string) error {
 		INDEX idx_timestamp timestamp TYPE minmax GRANULARITY 1,
 		INDEX idx_rpcMethod rpcMethod TYPE bloom_filter GRANULARITY 4,
 		INDEX idx_responseStatusCode responseStatusCode TYPE set(0) GRANULARITY 1,
-		) ENGINE MergeTree()
+		) ENGINE %sMergeTree()
 		PARTITION BY toDate(timestamp)
 		ORDER BY (durationNano, timestamp)
 		TTL toDateTime(timestamp) + INTERVAL 1296000 SECOND DELETE
-		SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1`, clickhousetracesexporter.DefaultTraceDatabase, clickhousetracesexporter.DefaultDurationSortTable, cluster))
+		SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1`, clickhousetracesexporter.DefaultTraceDatabase, clickhousetracesexporter.DefaultDurationSortTable, cluster, signozReplicated))
 	if err != nil {
 		return err
 	}