remove deprecated mongodb dynconf settings (#4297)

Remove the following env and their corresponding logic:
- PEERDB_MONGODB_DIRECT_BSON_CONVERTER
- PEERDB_MONGODB_PARALLEL_SNAPSHOTTING

Both are performance optimizations and have since stabilized; so we can
remove the legacy code.

Author: jgao54

flow/connectors/mongo/cdc.go

@@ -248,10 +248,7 @@ func (c *MongoConnector) PullRecords(
 		}
 	}
 
-	converter, err := NewBsonConverter(ctx, req.Env)
-	if err != nil {
-		return fmt.Errorf("failed to create converter: %w", err)
-	}
+	converter := NewDirectBsonConverter()
 	addRecordItems := func(documentKey bson.Raw, fullDocument bson.Raw, items *model.RecordItems, tableName string) error {
 		if len(documentKey) > 0 {
 			rv := documentKey.Lookup(DefaultDocumentKeyColumnName)

flow/connectors/mongo/cdc_test.go

@@ -143,7 +143,6 @@ func TestChangeStreamIdleConnectionAdvancesOffset(t *testing.T) {
 		TableNameSchemaMapping: map[string]*protos.TableSchema{},
 		MaxBatchSize:           10000,
 		IdleTimeout:            time.Minute,
-		Env:                    map[string]string{"PEERDB_MONGODB_DIRECT_BSON_CONVERTER": "true"},
 	}
 	drainMongoCDCRecordsAsync(t, req.RecordStream)
 

flow/connectors/mongo/codec.go

@@ -13,7 +13,6 @@ 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/otel_metrics"
 	"github.com/PeerDB-io/peerdb/flow/pkg/common"
@@ -40,15 +39,6 @@ func (c *MongoConnector) GetQRepPartitions(
 		return fullTablePartition, nil
 	}
 
-	parallelSnapshotting, err := internal.PeerDBMongoDBParallelSnapshotting(ctx, config.Env)
-	if err != nil {
-		c.logger.Warn("failed to get parallel snapshotting config", slog.Any("error", err))
-	}
-	if !parallelSnapshotting {
-		c.logger.Info("parallel snapshotting disabled, falling back to full table partition")
-		return fullTablePartition, nil
-	}
-
 	if config.NumRowsPerPartition <= 0 {
 		return nil, fmt.Errorf("num rows per partition must be greater than 0")
 	} else if last != nil && last.Range != nil {
@@ -163,10 +153,7 @@ func (c *MongoConnector) PullQRepRecords(
 	}
 	defer cursor.Close(ctx)
 
-	converter, err := NewBsonConverter(ctx, config.Env)
-	if err != nil {
-		return 0, 0, fmt.Errorf("failed to create bson converter: %w", err)
-	}
+	converter := NewDirectBsonConverter()
 	for cursor.Next(ctx) {
 		record, err := QValuesFromBsonRaw(cursor.Current, config.Version, converter, config.WatermarkTable)
 		if err != nil {

flow/connectors/mongo/qrep.go

@@ -1,7 +1,6 @@
 package connmongo
 
 import (
-	"context"
 	"encoding/base64"
 	"encoding/hex"
 	"fmt"
@@ -13,80 +12,30 @@ import (
 	"go.mongodb.org/mongo-driver/v2/bson"
 	"go.mongodb.org/mongo-driver/v2/x/bsonx/bsoncore"
 
-	"github.com/PeerDB-io/peerdb/flow/internal"
 	"github.com/PeerDB-io/peerdb/flow/shared"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
 type BsonToQValueConverter interface {
-	// QValueStringFromId QValueStringFromId converts a raw _id value to a QValueString.
+	// QValueStringFromId converts a raw _id value to a QValueString.
 	QValueStringFromId(id bson.RawValue, version uint32) (types.QValueString, error)
 	// QValueJSONFromDocument converts a raw BSON document to a QValueJSON.
 	QValueJSONFromDocument(raw bson.Raw) (types.QValueJSON, error)
 }
 
-func NewBsonConverter(ctx context.Context, env map[string]string) (BsonToQValueConverter, error) {
-	direct, err := internal.PeerDBMongoDBDirectBsonConverter(ctx, env)
-	if err != nil {
-		return nil, err
-	}
-
-	if !direct {
-		return &LegacyBsonConverter{
-			api: CreateExtendedJSONMarshaler(),
-		}, nil
-	}
+// DirectBsonConverter converts BSON directly to JSON string without intermediate deserialization,
+// it uses jsoniter.Stream to build JSON output incrementally into a reusable buffer (to avoid allocation)
+type DirectBsonConverter struct {
+	stream *jsoniter.Stream
+}
 
+func NewDirectBsonConverter() *DirectBsonConverter {
 	return &DirectBsonConverter{
 		// technically we write JSON directly via raw stream methods and do not use jsoniter's
 		// config-driven serialization specified here, but this config is applied consistent with our
-		// custom serialization (and used by LegacyBsonConverter) so specifying it here for consistency
+		// custom serialization so specifying it here for consistency
 		stream: jsoniter.NewStream(jsoniter.ConfigCompatibleWithStandardLibrary, nil, 512),
-	}, nil
-}
-
-// LegacyBsonConverter converts BSON to JSON by first deserializing BSON to bson.D,
-// and then serialize to JSON string via json-iterator.
-type LegacyBsonConverter struct {
-	api jsoniter.API
-}
-
-func (c *LegacyBsonConverter) QValueJSONFromDocument(raw bson.Raw) (types.QValueJSON, error) {
-	var d bson.D
-	if err := bson.Unmarshal(raw, &d); err != nil {
-		return types.QValueJSON{}, fmt.Errorf("failed to unmarshal document: %w", err)
-	}
-	jsonb, err := c.api.Marshal(d)
-	if err != nil {
-		return types.QValueJSON{}, fmt.Errorf("failed to marshal document: %w", err)
 	}
-	return types.QValueJSON{Val: string(jsonb)}, nil
-}
-
-func (c *LegacyBsonConverter) QValueStringFromId(id bson.RawValue, version uint32) (types.QValueString, error) {
-	if version >= shared.InternalVersion_MongoDBIdWithoutRedundantQuotes {
-		switch id.Type {
-		case bson.TypeObjectID:
-			return types.QValueString{Val: id.ObjectID().Hex()}, nil
-		case bson.TypeString:
-			return types.QValueString{Val: id.StringValue()}, nil
-		}
-	}
-	var val any
-	if err := id.Unmarshal(&val); err != nil {
-		return types.QValueString{}, fmt.Errorf("failed to unmarshal %s: %w", DefaultDocumentKeyColumnName, err)
-	}
-	jsonb, err := c.api.Marshal(val)
-	if err != nil {
-		return types.QValueString{}, fmt.Errorf("failed to marshal %s: %w", DefaultDocumentKeyColumnName, err)
-	}
-	return types.QValueString{Val: string(jsonb)}, nil
-}
-
-// DirectBsonConverter converts BSON directly to JSON string without intermediate deserialization,
-// it uses jsoniter.Stream to build JSON output incrementally into a reusable buffer (to avoid allocation)
-type DirectBsonConverter struct {
-	stream *jsoniter.Stream
 }
 
 func (c *DirectBsonConverter) QValueJSONFromDocument(raw bson.Raw) (types.QValueJSON, error) {
@@ -276,8 +225,15 @@ func rawValueToJSON(v bsoncore.Value, stream *jsoniter.Stream) error {
 	return nil
 }
 
-// writeFloat64JSON matches the encoding behavior of encodeCustom + writeFloat64WithExplicitDecimal:
-// NaN/Inf → quoted strings, integer-valued floats → explicit ".0", others → standard notation
+// Assume (and test) that values outside of these limits will come out in scientific notation
+// and will be parsed as floats either way
+var (
+	floatLimit    = math.Pow10(21)
+	floatNegLimit = -floatLimit
+)
+
+// writeFloat64JSON encodes NaN/Inf as quoted strings, integer-valued floats with an explicit
+// ".0" suffix (to hint ClickHouse to parse as float), and other values in standard notation.
 func writeFloat64JSON(stream *jsoniter.Stream, v float64) {
 	if math.IsNaN(v) {
 		stream.WriteRaw(`"NaN"`)

flow/connectors/mongo/qvalue_convert.go

@@ -1,7 +1,6 @@
 package connmongo
 
 import (
-	"context"
 	"encoding/base64"
 	"fmt"
 	"math"
@@ -631,22 +630,15 @@ func TestMarshalDocument(t *testing.T) {
 		},
 	}
 
-	directConverter, err := NewBsonConverter(context.Background(), map[string]string{"PEERDB_MONGODB_DIRECT_BSON_CONVERTER": "true"})
-	require.NoError(t, err)
-	legacyConverter, err := NewBsonConverter(context.Background(), map[string]string{"PEERDB_MONGODB_DIRECT_BSON_CONVERTER": "false"})
-	require.NoError(t, err)
+	converter := NewDirectBsonConverter()
 	for _, test := range tests {
 		t.Run(test.desc, func(t *testing.T) {
 			inputRaw, err := bson.Marshal(test.input)
 			require.NoError(t, err)
 
-			directResult, err := directConverter.QValueJSONFromDocument(inputRaw)
-			require.NoError(t, err)
-			require.Equal(t, test.expected, directResult.Val)
-
-			legacyResult, err := legacyConverter.QValueJSONFromDocument(inputRaw)
+			result, err := converter.QValueJSONFromDocument(inputRaw)
 			require.NoError(t, err)
-			require.Equal(t, directResult, legacyResult)
+			require.Equal(t, test.expected, result.Val)
 		})
 	}
 }
@@ -664,8 +656,7 @@ func TestMarshalId(t *testing.T) {
 		require.NoError(t, err)
 		return bson.Raw(raw).Lookup("_id")
 	}
-	converter, err := NewBsonConverter(context.Background(), nil)
-	require.NoError(t, err)
+	converter := NewDirectBsonConverter()
 
 	objectId, err := bson.ObjectIDFromHex("6893edbecb1f9508891bbb84")
 	require.NoError(t, err)
@@ -704,8 +695,7 @@ func TestMarshalId(t *testing.T) {
 // Tests that floats of all magnitudes are marshaled into a reasonable length and have a signifier
 // that they're not integers
 func TestMarshalFloatLengths(t *testing.T) {
-	converter, err := NewBsonConverter(context.Background(), nil)
-	require.NoError(t, err)
+	converter := NewDirectBsonConverter()
 	maxExponent := 309
 	require.Equal(t, math.Inf(1), math.Pow10(maxExponent), "exponent range should cover +Inf")   //nolint:testifylint
 	require.Equal(t, math.Inf(-1), -math.Pow10(maxExponent), "exponent range should cover -Inf") //nolint:testifylint
@@ -754,8 +744,7 @@ func TestMarshalFloatLengths(t *testing.T) {
 }
 
 func TestQValuesFromBsonRawInvalidIds(t *testing.T) {
-	converter, err := NewBsonConverter(t.Context(), map[string]string{"PEERDB_MONGODB_DIRECT_BSON_CONVERTER": "true"})
-	require.NoError(t, err)
+	converter := NewDirectBsonConverter()
 
 	t.Run("null _id is rejected", func(t *testing.T) {
 		raw, err := bson.Marshal(bson.D{{Key: "_id", Value: nil}})

flow/connectors/mongo/codec_test.go …/connectors/mongo/qvalue_convert_test.goflow/connectors/mongo/codec_test.go renamed to flow/connectors/mongo/qvalue_convert_test.go flow/connectors/mongo/codec_test.go renamed to flow/connectors/mongo/qvalue_convert_test.go

@@ -71,10 +71,7 @@ func (s *MongoSource) GetRows(ctx context.Context, suffix, table, cols string) (
 		Records: nil,
 	}
 
-	converter, err := connmongo.NewBsonConverter(ctx, nil)
-	if err != nil {
-		return nil, err
-	}
+	converter := connmongo.NewDirectBsonConverter()
 
 	for cursor.Next(ctx) {
 		record, err := connmongo.QValuesFromBsonRaw(cursor.Current, shared.InternalVersion_Latest, converter, table)