Always chunk Avro on uncompressed bytes (#4165)

It's been running for some time without any issues

Closes DBI-581

Author: ilidemi

flow/connectors/clickhouse/avro_sync.go

@@ -193,14 +193,6 @@ func (s *ClickHouseAvroSyncMethod) pushDataToS3ForSnapshot(
 		return nil, 0, err
 	}
 
-	trackUncompressed, err := internal.PeerDBS3ChunkOnUncompressed(ctx, config.Env)
-	if err != nil {
-		return nil, 0, err
-	}
-	if config.SourceType == protos.DBType_MONGO {
-		trackUncompressed = true
-	}
-
 	s.logger.Info("writing avro chunks to S3 start",
 		slog.String("partitionId", partition.PartitionId),
 		slog.Int64("bytesPerAvroFile", bytesPerAvroFile))
@@ -210,7 +202,7 @@ func (s *ClickHouseAvroSyncMethod) pushDataToS3ForSnapshot(
 		substream := model.NewQRecordStream(0)
 		substream.SetSchema(schema)
 		substream.SetSchemaDebug(stream.SchemaDebug())
-		sizeTracker := model.QRecordAvroChunkSizeTracker{TrackUncompressed: trackUncompressed}
+		sizeTracker := model.QRecordAvroChunkSizeTracker{}
 		go func() {
 			recordsDone := true
 			for record := range stream.Records {

flow/connectors/utils/avro_writer.go

@@ -19,7 +19,6 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/internal"
 	"github.com/PeerDB-io/peerdb/flow/model"
 	"github.com/PeerDB-io/peerdb/flow/pkg/common"
-	"github.com/PeerDB-io/peerdb/flow/shared"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
@@ -125,13 +124,7 @@ func (p *peerDBOCFWriter) WriteRecordsToS3(
 			}
 			w.Close()
 		}()
-		var writer io.Writer
-		if p.sizeTracker == nil || p.sizeTracker.TrackUncompressed {
-			writer = w
-		} else {
-			writer = shared.NewWatchWriter(w, &p.sizeTracker.Bytes)
-		}
-		numRows, writeOcfError = p.WriteOCF(ctx, env, writer, typeConversions, numericTruncator)
+		numRows, writeOcfError = p.WriteOCF(ctx, env, w, typeConversions, numericTruncator)
 	}()
 
 	partSize, err := internal.PeerDBS3PartSize(ctx, env)
@@ -255,7 +248,7 @@ func (p *peerDBOCFWriter) writeRecordsToOCFWriter(
 		return 0, err
 	}
 
-	calcSize := p.sizeTracker != nil && p.sizeTracker.TrackUncompressed
+	calcSize := p.sizeTracker != nil
 	for qrecord := range p.stream.Records {
 		if err := ctx.Err(); err != nil {
 			return numRows.Load(), err

flow/internal/dynamicconf.go

@@ -193,20 +193,12 @@ var DynamicSettings = [...]*protos.DynamicSetting{
 	},
 	{
 		Name:             "PEERDB_S3_BYTES_PER_AVRO_FILE",
-		Description:      "S3 upload chunk size in bytes, needed for large unpartitioned initial loads.",
+		Description:      "S3 upload chunk size in bytes before compression, needed for large unpartitioned initial loads.",
 		DefaultValue:     "1000000000",
 		ValueType:        protos.DynconfValueType_INT,
 		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_IMMEDIATE,
 		TargetForSetting: protos.DynconfTarget_CLICKHOUSE,
 	},
-	{
-		Name:             "PEERDB_S3_CHUNK_ON_UNCOMPRESSED",
-		Description:      "Track uncompressed Avro size during initial load, following the in-memory size on the destination.",
-		DefaultValue:     "true",
-		ValueType:        protos.DynconfValueType_BOOL,
-		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_IMMEDIATE,
-		TargetForSetting: protos.DynconfTarget_CLICKHOUSE,
-	},
 	{
 		Name:             "PEERDB_QUEUE_FORCE_TOPIC_CREATION",
 		Description:      "Force auto topic creation in mirrors, applies to Kafka and PubSub mirrors",
@@ -789,10 +781,6 @@ func PeerDBMetricsRecordAggregatesEnabled(ctx context.Context, env map[string]st
 	return dynamicConfBool(ctx, env, "PEERDB_METRICS_RECORD_AGGREGATES_ENABLED")
 }
 
-func PeerDBS3ChunkOnUncompressed(ctx context.Context, env map[string]string) (bool, error) {
-	return dynamicConfBool(ctx, env, "PEERDB_S3_CHUNK_ON_UNCOMPRESSED")
-}
-
 func PeerDBPostgresApplyCtidBlockPartitioning(ctx context.Context, env map[string]string) (bool, error) {
 	return dynamicConfBool(ctx, env, "PEERDB_POSTGRES_APPLY_CTID_BLOCK_PARTITIONING_OVERRIDE")
 }

flow/model/conversion_avro.go

@@ -110,8 +110,7 @@ type QRecordAvroSchemaDefinition struct {
 }
 
 type QRecordAvroChunkSizeTracker struct {
-	TrackUncompressed bool
-	Bytes             atomic.Int64
+	Bytes atomic.Int64
 }
 
 func GetAvroSchemaDefinition(