feat: introduce random sleep before clickhouse loads

warehouse/integrations/clickhouse/clickhouse.go

@@ -11,6 +11,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math/rand"
 	"net/url"
 	"os"
 	"path"
@@ -160,6 +161,7 @@ type Clickhouse struct {
 		numWorkersDownloadLoadFiles int
 		s3EngineEnabledWorkspaceIDs []string
 		slowQueryThreshold          time.Duration
+		randomLoadDelay             func(string) time.Duration
 	}
 }
 
@@ -234,6 +236,15 @@ func New(conf *config.Config, log logger.Logger, stat stats.Stats) *Clickhouse {
 	ch.config.numWorkersDownloadLoadFiles = conf.GetInt("Warehouse.clickhouse.numWorkersDownloadLoadFiles", 8)
 	ch.config.s3EngineEnabledWorkspaceIDs = conf.GetStringSlice("Warehouse.clickhouse.s3EngineEnabledWorkspaceIDs", nil)
 	ch.config.slowQueryThreshold = conf.GetDuration("Warehouse.clickhouse.slowQueryThreshold", 5, time.Minute)
+	ch.config.randomLoadDelay = func(destinationID string) time.Duration {
+		maxDelay := conf.GetDurationVar(
+			5,
+			time.Second,
+			fmt.Sprintf("Warehouse.clickhouse.%s.maxLoadDelay", destinationID),
+			"Warehouse.clickhouse.maxLoadDelay",
+		)
+		return time.Duration(float64(maxDelay) * (1 - rand.Float64()))
+	}
 
 	return ch
 }
@@ -491,6 +502,9 @@ func (ch *Clickhouse) typecastDataFromType(data, dataType string) interface{} {
 
 // loadTable loads table to clickhouse from the load files
 func (ch *Clickhouse) loadTable(ctx context.Context, tableName string, tableSchemaInUpload model.TableSchema) (err error) {
+	if misc.SleepCtx(ctx, ch.config.randomLoadDelay(ch.Warehouse.Destination.ID)) != nil {
+		return nil
+	}
 	if ch.UseS3CopyEngineForLoading() {
 		return ch.loadByCopyCommand(ctx, tableName, tableSchemaInUpload)
 	}