fix(clickhouse): add opt-out for staging cleanup after QRep flow

PR #4200 (StagingStore abstraction) unified the avro upload and staging
cleanup paths onto the staging store's real bucket prefix. As a side
effect, CleanupQRepFlow now always deletes staging avro files once a
ClickHouse snapshot/QRep flow completes.

Previously cleanup keyed off config.StagingPath, which is empty for
ClickHouse mirrors that don't set SnapshotStagingPath (they get their
bucket from PEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME), so the prefix guard
`strings.HasPrefix(stagingPath, "s3://")` was false and cleanup was a
no-op. Staging avro files were effectively retained. Some deployments
have downstream pipelines that consume those files, and the upgrade
silently broke them by deleting the files as soon as the mirror
completed.

Keep cleanup as the default (it's the correct behavior and avoids
unbounded staging growth), but add PEERDB_CLICKHOUSE_SKIP_STAGING_CLEANUP
to opt out and retain staging artifacts.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: dtunikov

flow/connectors/clickhouse/qrep.go

@@ -6,6 +6,7 @@ import (
 	"log/slog"
 
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
+	"github.com/PeerDB-io/peerdb/flow/internal"
 	"github.com/PeerDB-io/peerdb/flow/model"
 	"github.com/PeerDB-io/peerdb/flow/shared"
 )
@@ -44,6 +45,17 @@ func (c *ClickHouseConnector) ConsolidateQRepPartitions(_ context.Context, confi
 // CleanupQRepFlow function for clickhouse connector
 func (c *ClickHouseConnector) CleanupQRepFlow(ctx context.Context, config *protos.QRepConfig) error {
 	flowName := config.FlowJobName
+
+	skipCleanup, err := internal.PeerDBClickHouseSkipStagingCleanup(ctx, config.Env)
+	if err != nil {
+		return fmt.Errorf("failed to read staging cleanup config: %w", err)
+	}
+	if skipCleanup {
+		c.logger.Info("Skipping staging cleanup after QRepFlow (PEERDB_CLICKHOUSE_SKIP_STAGING_CLEANUP enabled)",
+			slog.String("stagingPath", c.staging.BucketPath()), slog.String("flowName", flowName))
+		return nil
+	}
+
 	c.logger.Info("Cleaning up stage after QRepFlow",
 		slog.String("stagingPath", c.staging.BucketPath()), slog.String("flowName", flowName))
 

flow/internal/dynamicconf.go

@@ -198,6 +198,15 @@ var DynamicSettings = [...]*protos.DynamicSetting{
 		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_IMMEDIATE,
 		TargetForSetting: protos.DynconfTarget_CLICKHOUSE,
 	},
+	{
+		Name: "PEERDB_CLICKHOUSE_SKIP_STAGING_CLEANUP",
+		Description: "Skip deleting ClickHouse staging avro files after a QRep/snapshot flow completes. " +
+			"Enable to retain staging artifacts for downstream pipelines that consume them",
+		DefaultValue:     "false",
+		ValueType:        protos.DynconfValueType_BOOL,
+		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_IMMEDIATE,
+		TargetForSetting: protos.DynconfTarget_CLICKHOUSE,
+	},
 	{
 		Name:             "PEERDB_S3_UUID_PREFIX",
 		Description:      "Use random UUID as prefix instead of flow name, can help partitioning on non-AWS based s3 providers",
@@ -758,6 +767,10 @@ func PeerDBClickHouseStagingBucketName(ctx context.Context, env map[string]strin
 	return dynLookup(ctx, env, "PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME")
 }
 
+func PeerDBClickHouseSkipStagingCleanup(ctx context.Context, env map[string]string) (bool, error) {
+	return dynamicConfBool(ctx, env, "PEERDB_CLICKHOUSE_SKIP_STAGING_CLEANUP")
+}
+
 func PeerDBClickHouseAWSS3BucketName(ctx context.Context, env map[string]string) (string, error) {
 	return dynLookup(ctx, env, "PEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME")
 }