refactor(clickhouse): inline staging upload instead of WriteRecordsToStaging

Remove the WriteRecordsToStaging wrapper from avro_writer.go and call
StagingStore.Upload directly in the ClickHouse connector. This avoids
adding a new method to avro_writer.go and keeps the upload logic
colocated with the ClickHouse-specific staging setup.

Author: whites11

flow/connectors/clickhouse/avro_sync.go

@@ -3,6 +3,7 @@ package connclickhouse
 import (
 	"context"
 	"fmt"
+	"io"
 	"log/slog"
 	"strings"
 	"sync/atomic"
@@ -378,14 +379,28 @@ func (s *ClickHouseAvroSyncMethod) writeToAvroFile(
 	}
 	s3AvroFileKey = strings.TrimLeft(s3AvroFileKey, "/")
 
-	avroFile, err := ocfWriter.WriteRecordsToStaging(
-		ctx, env, s.staging, s3AvroFileKey, typeConversions, numericTruncator,
-	)
-	if err != nil {
-		return utils.AvroFile{}, fmt.Errorf("failed to write records to S3: %w", err)
+	r, w := io.Pipe()
+	defer r.Close()
+
+	var writeOcfError error
+	var numRows int64
+	go func() {
+		defer w.Close()
+		numRows, writeOcfError = ocfWriter.WriteOCF(ctx, env, w, typeConversions, numericTruncator)
+	}()
+
+	if err := s.staging.Upload(ctx, env, s3AvroFileKey, r); err != nil {
+		return utils.AvroFile{}, fmt.Errorf("failed to upload to staging: %w", err)
+	}
+	if writeOcfError != nil {
+		return utils.AvroFile{}, writeOcfError
 	}
 
-	return avroFile, nil
+	return utils.AvroFile{
+		StorageLocation: utils.AvroS3Storage,
+		FilePath:        s3AvroFileKey,
+		NumRecords:      numRows,
+	}, nil
 }
 
 func (s *ClickHouseAvroSyncMethod) SyncQRepObjects(

flow/connectors/utils/avro_writer.go

@@ -166,51 +166,6 @@ func (p *peerDBOCFWriter) WriteRecordsToS3(
 	}, nil
 }
 
-// WriteRecordsToStaging serializes records as Avro OCF and uploads via the given StagingStore.
-func (p *peerDBOCFWriter) WriteRecordsToStaging(
-	ctx context.Context,
-	env map[string]string,
-	store StagingStore,
-	key string,
-	typeConversions map[string]types.TypeConversion,
-	numericTruncator model.SnapshotTableNumericTruncator,
-) (AvroFile, error) {
-	logger := internal.LoggerFromCtx(ctx)
-
-	r, w := io.Pipe()
-	defer r.Close()
-
-	var writeOcfError error
-	var numRows int64
-
-	go func() {
-		defer func() {
-			if r := recover(); r != nil {
-				writeOcfError = fmt.Errorf("panic occurred during WriteOCF: %v", r)
-				stack := string(debug.Stack())
-				logger.Error("panic during WriteOCF", slog.Any("error", writeOcfError), slog.String("stack", stack))
-			}
-			w.Close()
-		}()
-		numRows, writeOcfError = p.WriteOCF(ctx, env, w, typeConversions, numericTruncator)
-	}()
-
-	if err := store.Upload(ctx, env, key, r); err != nil {
-		return AvroFile{}, fmt.Errorf("failed to upload to staging: %w", err)
-	}
-
-	if writeOcfError != nil {
-		logger.Error("failed to write records to OCF", slog.Any("error", writeOcfError))
-		return AvroFile{}, writeOcfError
-	}
-
-	return AvroFile{
-		StorageLocation: AvroS3Storage,
-		FilePath:        key,
-		NumRecords:      numRows,
-	}, nil
-}
-
 func (p *peerDBOCFWriter) WriteRecordsToAvroFile(ctx context.Context, env map[string]string, filePath string) (AvroFile, error) {
 	file, err := os.Create(filePath)
 	if err != nil {