feat(clickhouse): add unified PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME env var

Add provider-agnostic PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME that works
with PEERDB_CLICKHOUSE_STAGING_PROVIDER to configure staging storage.
GCS uses the unified var exclusively. S3 falls back to the legacy
PEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME when the unified var is not set.

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

Author: whites11

flow/connectors/clickhouse/avro_sync.go

@@ -37,12 +37,12 @@ func NewClickHouseAvroSyncMethod(
 	}
 }
 
-func (s *ClickHouseAvroSyncMethod) stagingTableFunctionBuilder(ctx context.Context, avroFilePath string) (string, error) {
+func (s *ClickHouseAvroSyncMethod) s3TableFunctionBuilder(ctx context.Context, avroFilePath string) (string, error) {
 	return s.staging.TableFunctionExpr(ctx, avroFilePath, "Avro")
 }
 
 func (s *ClickHouseAvroSyncMethod) CopyStageToDestination(ctx context.Context, avroFile utils.AvroFile) error {
-	s3TableFunction, err := s.stagingTableFunctionBuilder(ctx, avroFile.FilePath)
+	s3TableFunction, err := s.s3TableFunctionBuilder(ctx, avroFile.FilePath)
 	if err != nil {
 		s.logger.Error("failed to build S3 table function",
 			slog.String("avroFilePath", avroFile.FilePath),
@@ -276,7 +276,7 @@ func (s *ClickHouseAvroSyncMethod) pushS3DataToClickHouseForSnapshot(
 
 		for i := range numParts {
 			// Get fresh credentials for each part
-			s3TableFunction, err := s.stagingTableFunctionBuilder(ctx, avroFile.FilePath)
+			s3TableFunction, err := s.s3TableFunctionBuilder(ctx, avroFile.FilePath)
 			if err != nil {
 				s.logger.Error("failed to build S3 table function",
 					slog.String("avroFilePath", avroFile.FilePath),
@@ -365,24 +365,24 @@ func (s *ClickHouseAvroSyncMethod) writeToAvroFile(
 	ocfWriter := utils.NewPeerDBOCFWriter(stream, avroSchema, ocf.ZStandard, protos.DBType_CLICKHOUSE, sizeTracker)
 	prefix := s.staging.KeyPrefix()
 
-	uuidPrefix, err := internal.PeerDBS3UuidPrefix(ctx, s.config.Env)
+	s3UuidPrefix, err := internal.PeerDBS3UuidPrefix(ctx, s.config.Env)
 	if err != nil {
 		return utils.AvroFile{}, err
 	}
 
-	var avroFileKey string
-	if uuidPrefix {
-		avroFileKey = fmt.Sprintf("%s/%s/%s/%s.avro", prefix, uuid.NewString(), flowJobName, identifierForFile)
+	var s3AvroFileKey string
+	if s3UuidPrefix {
+		s3AvroFileKey = fmt.Sprintf("%s/%s/%s/%s.avro", prefix, uuid.NewString(), flowJobName, identifierForFile)
 	} else {
-		avroFileKey = fmt.Sprintf("%s/%s/%s.avro", prefix, flowJobName, identifierForFile)
+		s3AvroFileKey = fmt.Sprintf("%s/%s/%s.avro", prefix, flowJobName, identifierForFile)
 	}
-	avroFileKey = strings.TrimLeft(avroFileKey, "/")
+	s3AvroFileKey = strings.TrimLeft(s3AvroFileKey, "/")
 
-	avroFile, err := ocfWriter.WriteRecordsToStaging(
-		ctx, env, s.staging, avroFileKey, typeConversions, numericTruncator,
+	avroFile, err := ocfWriter.WriteRecordsToS3(
+		ctx, env, s.staging, s3AvroFileKey, typeConversions, numericTruncator,
 	)
 	if err != nil {
-		return utils.AvroFile{}, fmt.Errorf("failed to write records to staging: %w", err)
+		return utils.AvroFile{}, fmt.Errorf("failed to write records to S3: %w", err)
 	}
 
 	return avroFile, nil

flow/connectors/clickhouse/clickhouse.go

@@ -108,7 +108,7 @@ func createStagingStore(
 ) (utils.StagingStore, *utils.ClickHouseS3Credentials, error) {
 	// If user provided explicit S3 config, always use S3 staging.
 	if config.S3 != nil || config.S3Path != "" || config.AccessKeyId != "" {
-		return createS3StagingStore(ctx, env, config)
+		return createS3StagingStore(ctx, env, config, "")
 	}
 
 	// Check environment-based staging provider config.
@@ -117,19 +117,26 @@ func createStagingStore(
 		return nil, nil, fmt.Errorf("failed to get staging provider: %w", err)
 	}
 
+	// Prefer unified PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME, fall back to provider-specific env vars.
+	bucketName, err := internal.PeerDBClickHouseStagingBucketName(ctx, env)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to get staging bucket name: %w", err)
+	}
+
 	switch strings.ToLower(provider) {
 	case "gcs":
-		store, err := createGCSStagingStore(ctx, env)
+		store, err := createGCSStagingStore(ctx, bucketName)
 		return store, nil, err
 	default:
-		return createS3StagingStore(ctx, env, config)
+		return createS3StagingStore(ctx, env, config, bucketName)
 	}
 }
 
 func createS3StagingStore(
 	ctx context.Context,
 	env map[string]string,
 	config *protos.ClickhouseConfig,
+	unifiedBucketName string,
 ) (utils.StagingStore, *utils.ClickHouseS3Credentials, error) {
 	var awsConfig utils.PeerAWSCredentials
 	var awsBucketPath string
@@ -157,9 +164,13 @@ func createS3StagingStore(
 		deploymentUID := internal.PeerDBDeploymentUID()
 		flowName, _ := ctx.Value(shared.FlowNameKey).(string)
 		bucketPathSuffix := fmt.Sprintf("%s/%s", url.PathEscape(deploymentUID), url.PathEscape(flowName))
-		awsBucketName, err := internal.PeerDBClickHouseAWSS3BucketName(ctx, env)
-		if err != nil {
-			return nil, nil, fmt.Errorf("failed to get PeerDB ClickHouse Bucket Name: %w", err)
+		// Prefer unified bucket name, fall back to legacy S3-specific env var.
+		awsBucketName := unifiedBucketName
+		if awsBucketName == "" {
+			awsBucketName, err = internal.PeerDBClickHouseAWSS3BucketName(ctx, env)
+			if err != nil {
+				return nil, nil, fmt.Errorf("failed to get PeerDB ClickHouse Bucket Name: %w", err)
+			}
 		}
 		if awsBucketName == "" {
 			return nil, nil, errors.New("PeerDB ClickHouse Bucket Name not set")
@@ -182,20 +193,16 @@ func createS3StagingStore(
 
 func createGCSStagingStore(
 	ctx context.Context,
-	env map[string]string,
+	bucketName string,
 ) (utils.StagingStore, error) {
-	gcsBucketName, err := internal.PeerDBClickHouseGCSBucketName(ctx, env)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get GCS bucket name: %w", err)
-	}
-	if gcsBucketName == "" {
-		return nil, errors.New("PEERDB_CLICKHOUSE_GCS_BUCKET_NAME not set but staging provider is gcs")
+	if bucketName == "" {
+		return nil, errors.New("PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME must be set when staging provider is gcs")
 	}
 
 	deploymentUID := internal.PeerDBDeploymentUID()
 	flowName, _ := ctx.Value(shared.FlowNameKey).(string)
 	bucketPath := fmt.Sprintf("gs://%s/%s/%s",
-		gcsBucketName, url.PathEscape(deploymentUID), url.PathEscape(flowName))
+		bucketName, url.PathEscape(deploymentUID), url.PathEscape(flowName))
 
 	return utils.NewGCSStagingStore(ctx, bucketPath)
 }

flow/connectors/utils/avro_writer.go

@@ -166,9 +166,8 @@ func (p *peerDBOCFWriter) WriteRecordsToS3(
 	}, nil
 }
 
-// WriteRecordsToStaging serializes records as Avro OCF and uploads via the given StagingStore.
-// This is the cloud-agnostic equivalent of WriteRecordsToS3.
-func (p *peerDBOCFWriter) WriteRecordsToStaging(
+// WriteRecordsToS3 serializes records as Avro OCF and uploads via the given StagingStore.
+func (p *peerDBOCFWriter) WriteRecordsToS3(
 	ctx context.Context,
 	env map[string]string,
 	store StagingStore,

flow/internal/dynamicconf.go

@@ -183,8 +183,8 @@ var DynamicSettings = [...]*protos.DynamicSetting{
 		TargetForSetting: protos.DynconfTarget_CLICKHOUSE,
 	},
 	{
-		Name:             "PEERDB_CLICKHOUSE_GCS_BUCKET_NAME",
-		Description:      "GCS bucket to store Avro files for mirrors with ClickHouse target",
+		Name:             "PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME",
+		Description:      "Staging bucket name for ClickHouse mirrors (provider-agnostic, preferred over legacy env vars)",
 		DefaultValue:     "",
 		ValueType:        protos.DynconfValueType_STRING,
 		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_IMMEDIATE,
@@ -730,8 +730,8 @@ func PeerDBClickHouseStagingProvider(ctx context.Context, env map[string]string)
 	return dynLookup(ctx, env, "PEERDB_CLICKHOUSE_STAGING_PROVIDER")
 }
 
-func PeerDBClickHouseGCSBucketName(ctx context.Context, env map[string]string) (string, error) {
-	return dynLookup(ctx, env, "PEERDB_CLICKHOUSE_GCS_BUCKET_NAME")
+func PeerDBClickHouseStagingBucketName(ctx context.Context, env map[string]string) (string, error) {
+	return dynLookup(ctx, env, "PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME")
 }
 
 func PeerDBS3UuidPrefix(ctx context.Context, env map[string]string) (bool, error) {