PEERDB_CLICKHOUSE_INITIAL_LOAD_ALLOW_NON_EMPTY_TABLES (#3157)

closes #3156

Author: serprex

flow/connectors/clickhouse/validate.go

@@ -41,9 +41,15 @@ func (c *ClickHouseConnector) ValidateMirrorDestination(
 		return err
 	}
 
+	initialLoadAllowNonEmptyTables, err := internal.PeerDBClickHouseInitialLoadAllowNonEmptyTables(ctx, cfg.Env)
+	if err != nil {
+		return err
+	}
+
 	if !(cfg.Resync || sourceSchemaAsDestinationColumn) {
 		if err := chvalidate.CheckIfTablesEmptyAndEngine(ctx, c.logger, c.database,
-			dstTableNames, cfg.DoInitialSnapshot, internal.PeerDBOnlyClickHouseAllowed()); err != nil {
+			dstTableNames, cfg.DoInitialSnapshot, internal.PeerDBOnlyClickHouseAllowed(), initialLoadAllowNonEmptyTables,
+		); err != nil {
 			return err
 		}
 	}

flow/internal/dynamicconf.go

@@ -306,6 +306,14 @@ var DynamicSettings = [...]*protos.DynamicSetting{
 		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_AFTER_RESUME,
 		TargetForSetting: protos.DynconfTarget_CLICKHOUSE,
 	},
+	{
+		Name:             "PEERDB_CLICKHOUSE_INITIAL_LOAD_ALLOW_NON_EMPTY_TABLES",
+		Description:      "Disables validation raising error if destination table of initial load is not empty",
+		DefaultValue:     "false",
+		ValueType:        protos.DynconfValueType_BOOL,
+		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_NEW_MIRROR,
+		TargetForSetting: protos.DynconfTarget_CLICKHOUSE,
+	},
 	{
 		Name:             "PEERDB_SKIP_SNAPSHOT_EXPORT",
 		Description:      "This avoids initial load failing due to connectivity drops, but risks data consistency unless precautions are taken",
@@ -625,6 +633,10 @@ func PeerDBClickHouseInitialLoadPartsPerPartition(ctx context.Context, env map[s
 	return dynamicConfUnsigned[uint64](ctx, env, "PEERDB_CLICKHOUSE_INITIAL_LOAD_PARTS_PER_PARTITION")
 }
 
+func PeerDBClickHouseInitialLoadAllowNonEmptyTables(ctx context.Context, env map[string]string) (bool, error) {
+	return dynamicConfBool(ctx, env, "PEERDB_CLICKHOUSE_INITIAL_LOAD_ALLOW_NON_EMPTY_TABLES")
+}
+
 func PeerDBSkipSnapshotExport(ctx context.Context, env map[string]string) (bool, error) {
 	return dynamicConfBool(ctx, env, "PEERDB_SKIP_SNAPSHOT_EXPORT")
 }

flow/shared/clickhouse/validation.go

@@ -31,7 +31,7 @@ func CheckIfClickHouseCloudHasSharedMergeTreeEnabled(ctx context.Context, logger
 }
 
 func CheckIfTablesEmptyAndEngine(ctx context.Context, logger log.Logger, conn clickhouse.Conn,
-	tables []string, initialSnapshotEnabled bool, checkForCloudSMT bool,
+	tables []string, initialSnapshotEnabled bool, checkForCloudSMT bool, allowNonEmpty bool,
 ) error {
 	queryTables := make([]string, 0, len(tables))
 	for _, table := range tables {
@@ -51,7 +51,7 @@ func CheckIfTablesEmptyAndEngine(ctx context.Context, logger log.Logger, conn cl
 		if err := rows.Scan(&tableName, &engine, &totalRows); err != nil {
 			return fmt.Errorf("failed to scan information for tables: %w", err)
 		}
-		if totalRows != 0 && initialSnapshotEnabled {
+		if !allowNonEmpty && totalRows != 0 && initialSnapshotEnabled {
 			return fmt.Errorf("table %s exists and is not empty", tableName)
 		}
 		if !slices.Contains(acceptableTableEngines, strings.TrimPrefix(engine, "Shared")) {