PEERDB_SNOWFLAKE_AUTO_COMPRESS (#3137)

This allows us to skip the auto-compression enabled by default for
Snowflake
Snowflake seems to struggle with certain files that get auto-compressed
and then fail to decompress:
```
Invalid data encountered during decompression for file: 'not_file',
compression type used: 'ZSTD', cause: 'Data corruption detected'
```

closes #3123

---------

Co-authored-by: Stoica Alexandru <[email protected]>

Author: serprex

flow/connectors/snowflake/qrep_avro_sync.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/PeerDB-io/peerdb/flow/connectors/utils"
 	"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"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
@@ -199,12 +200,18 @@ func (s *SnowflakeAvroSyncHandler) putFileToStage(ctx context.Context, avroFile
 		return nil
 	}
 
-	putCmd := fmt.Sprintf("PUT file://%s @%s", avroFile.FilePath, stage)
+	autoCompressionStr := ""
+	autoCompression, err := internal.PeerDBSnowflakeAutoCompress(ctx, s.config.Env)
+	if err != nil {
+		s.logger.Warn("Failed to load PEERDB_SNOWFLAKE_AUTO_COMPRESS, proceeding without", slog.Any("error", err))
+	} else if !autoCompression {
+		autoCompressionStr = " AUTO_COMPRESS=FALSE"
+	}
 
-	if _, err := s.ExecContext(ctx, putCmd); err != nil {
+	if _, err := s.ExecContext(ctx, fmt.Sprintf("PUT file://%s @%s%s", avroFile.FilePath, stage, autoCompressionStr)); err != nil {
 		return fmt.Errorf("failed to put file to stage: %w", err)
 	}
 
-	s.logger.Info(fmt.Sprintf("put file %s to stage %s", avroFile.FilePath, stage))
+	s.logger.Info("put file to stage", slog.String("file", avroFile.FilePath), slog.String("stage", stage))
 	return nil
 }

flow/internal/dynamicconf.go

@@ -34,8 +34,8 @@ var DynamicSettings = [...]*protos.DynamicSetting{
 	},
 	{
 		Name: "PEERDB_NORMALIZE_CHANNEL_BUFFER_SIZE",
-		Description: "Advanced setting: changes buffer size of channel PeerDB uses for queueing normalizing, " +
-			"use with PEERDB_PARALLEL_SYNC_NORMALIZE",
+		Description: "Advanced setting: changes buffer size of channel PeerDB uses for queueing normalization, " +
+			"use with PEERDB_ENABLE_PARALLEL_SYNC_NORMALIZE",
 		DefaultValue:     "128",
 		ValueType:        protos.DynconfValueType_INT,
 		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_AFTER_RESUME,
@@ -120,14 +120,6 @@ var DynamicSettings = [...]*protos.DynamicSetting{
 		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_NEW_MIRROR,
 		TargetForSetting: protos.DynconfTarget_ALL,
 	},
-	{
-		Name:             "PEERDB_CLICKHOUSE_BINARY_FORMAT",
-		Description:      "Binary field encoding on clickhouse destination; either raw, hex, or base64",
-		DefaultValue:     "raw",
-		ValueType:        protos.DynconfValueType_STRING,
-		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_AFTER_RESUME,
-		TargetForSetting: protos.DynconfTarget_CLICKHOUSE,
-	},
 	{
 		Name:             "PEERDB_SNOWFLAKE_MERGE_PARALLELISM",
 		Description:      "Parallel MERGE statements to run for CDC mirrors with Snowflake targets. -1 for no limit",
@@ -136,6 +128,22 @@ var DynamicSettings = [...]*protos.DynamicSetting{
 		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_IMMEDIATE,
 		TargetForSetting: protos.DynconfTarget_SNOWFLAKE,
 	},
+	{
+		Name:             "PEERDB_SNOWFLAKE_AUTO_COMPRESS",
+		Description:      "AUTO_COMPRESS option when uploading to Snowflake",
+		DefaultValue:     "true",
+		ValueType:        protos.DynconfValueType_BOOL,
+		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_AFTER_RESUME,
+		TargetForSetting: protos.DynconfTarget_SNOWFLAKE,
+	},
+	{
+		Name:             "PEERDB_CLICKHOUSE_BINARY_FORMAT",
+		Description:      "Binary field encoding on clickhouse destination; either raw, hex, or base64",
+		DefaultValue:     "raw",
+		ValueType:        protos.DynconfValueType_STRING,
+		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_AFTER_RESUME,
+		TargetForSetting: protos.DynconfTarget_CLICKHOUSE,
+	},
 	{
 		Name:             "PEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME",
 		Description:      "S3 buckets to store Avro files for mirrors with ClickHouse target",
@@ -554,6 +562,10 @@ func PeerDBSnowflakeMergeParallelism(ctx context.Context, env map[string]string)
 	return dynamicConfSigned[int64](ctx, env, "PEERDB_SNOWFLAKE_MERGE_PARALLELISM")
 }
 
+func PeerDBSnowflakeAutoCompress(ctx context.Context, env map[string]string) (bool, error) {
+	return dynamicConfBool(ctx, env, "PEERDB_SNOWFLAKE_AUTO_COMPRESS")
+}
+
 func PeerDBClickHouseAWSS3BucketName(ctx context.Context, env map[string]string) (string, error) {
 	return dynLookup(ctx, env, "PEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME")
 }