support filtering out deletion operation types for mongo

flow/connectors/mongo/cdc.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"slices"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -17,6 +18,7 @@ import (
 	"go.opentelemetry.io/otel/trace"
 
 	"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"
@@ -25,6 +27,24 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
+type operationType string
+
+const (
+	operationTypeInsert  operationType = "insert"
+	operationTypeUpdate  operationType = "update"
+	operationTypeReplace operationType = "replace"
+	operationTypeDelete  operationType = "delete"
+)
+
+func parseOperationType(s string) (operationType, bool) {
+	switch op := operationType(s); op {
+	case operationTypeInsert, operationTypeUpdate, operationTypeReplace, operationTypeDelete:
+		return op, true
+	default:
+		return "", false
+	}
+}
+
 type Namespace struct {
 	Db   string `bson:"db"`
 	Coll string `bson:"coll"`
@@ -117,7 +137,7 @@ func (c *MongoConnector) SetupReplication(
 		SetComment("PeerDB changeStream").
 		SetFullDocument(options.UpdateLookup)
 
-	pipeline, err := createPipeline(nil)
+	pipeline, err := createPipeline(nil, nil)
 	if err != nil {
 		return model.SetupReplicationResult{}, fmt.Errorf("failed to create changestream pipeline: %w", err)
 	}
@@ -201,7 +221,7 @@ func (c *MongoConnector) PullRecords(
 		changeStreamOpts.SetResumeAfter(resumeToken)
 	}
 
-	pipeline, err := createPipeline(req.TableNameMapping)
+	pipeline, err := createPipeline(req.TableNameMapping, c.excludedOps)
 	if err != nil {
 		return err
 	}
@@ -444,8 +464,8 @@ func (c *MongoConnector) PullRecords(
 		}
 
 		items := model.NewMongoRecordItems(2)
-		switch changeEvent.OperationType {
-		case "insert":
+		switch operationType(changeEvent.OperationType) {
+		case operationTypeInsert:
 			if err := addRecordItems(changeEvent.DocumentKey, changeEvent.FullDocument, &items, sourceTableName); err != nil {
 				return fmt.Errorf("failed to process document: %w", err)
 			}
@@ -458,7 +478,7 @@ func (c *MongoConnector) PullRecords(
 			}); err != nil {
 				return fmt.Errorf("failed to add insert record: %w", err)
 			}
-		case "update", "replace":
+		case operationTypeUpdate, operationTypeReplace:
 			if err := addRecordItems(changeEvent.DocumentKey, changeEvent.FullDocument, &items, sourceTableName); err != nil {
 				return fmt.Errorf("failed to process document: %w", err)
 			}
@@ -471,7 +491,7 @@ func (c *MongoConnector) PullRecords(
 			}); err != nil {
 				return fmt.Errorf("failed to add update record: %w", err)
 			}
-		case "delete":
+		case operationTypeDelete:
 			if err := addRecordItems(changeEvent.DocumentKey, changeEvent.FullDocument, &items, sourceTableName); err != nil {
 				return fmt.Errorf("failed to process document: %w", err)
 			}
@@ -496,7 +516,7 @@ func (c *MongoConnector) PullRecords(
 	return nil
 }
 
-func createPipeline(tableNameMapping map[string]model.NameAndExclude) (mongo.Pipeline, error) {
+func createPipeline(tableNameMapping map[string]model.NameAndExclude, excludedOps []operationType) (mongo.Pipeline, error) {
 	pipeline := mongo.Pipeline{}
 
 	// filter out events from tables that are not in the mapping
@@ -528,6 +548,13 @@ func createPipeline(tableNameMapping map[string]model.NameAndExclude) (mongo.Pip
 		}}})
 	}
 
+	// filter out excluded operation types
+	if len(excludedOps) > 0 {
+		pipeline = append(pipeline, bson.D{{Key: "$match", Value: bson.D{
+			{Key: "operationType", Value: bson.D{{Key: "$nin", Value: excludedOps}}},
+		}}})
+	}
+
 	// Mongo recommends using '$project' first to reduce change event size, and only use
 	// '$changeStreamSplitLargeEvent' in the pipeline if still necessary. Given the document
 	// themselves have a 16MB limit, project required fields for now for code simplicity.
@@ -569,7 +596,28 @@ func (c *MongoConnector) FinishExport(any) error {
 	return nil
 }
 
-func (c *MongoConnector) SetupReplConn(context.Context, map[string]string) error {
+func (c *MongoConnector) SetupReplConn(ctx context.Context, env map[string]string) error {
+	// Unlike Postgres, MongoDB doesn't need a dedicated replication connection:
+	// change streams are cursors served by the connector's pooled client.
+	// Since SetupReplConn is called once per SyncFlow activity, resolving
+	// dynamic config here avoids per-batch catalog reads.
+	excludedOps, err := internal.PeerDBMongoDBExcludedOperationTypes(ctx, env)
+	if err != nil {
+		return fmt.Errorf("failed to get excluded operation types: %w", err)
+	}
+	c.excludedOps = make([]operationType, 0, len(excludedOps))
+	for _, op := range excludedOps {
+		if parsed, ok := parseOperationType(op); ok {
+			if !slices.Contains(c.excludedOps, parsed) {
+				c.excludedOps = append(c.excludedOps, parsed)
+			}
+		} else {
+			c.logger.Warn("ignoring invalid operation type in exclusion list", slog.String("operationType", op))
+		}
+	}
+	if len(c.excludedOps) > 0 {
+		c.logger.Info("excluding operation types from replication", slog.Any("operationTypes", c.excludedOps))
+	}
 	return nil
 }
 

flow/connectors/mongo/cdc_test.go

@@ -177,6 +177,109 @@ func TestResumeTokenHelpersRoundTrip(t *testing.T) {
 	require.Equal(t, toBsonTs(ts), bsonTs)
 }
 
+func TestCreatePipeline(t *testing.T) {
+	tableNameMapping := map[string]model.NameAndExclude{"db.coll": {Name: "db_coll"}}
+
+	// lookupInPipeline returns the first value found at the given path in any pipeline stage.
+	lookupInPipeline := func(t *testing.T, pipeline mongo.Pipeline, path ...string) bson.RawValue {
+		t.Helper()
+		for _, stage := range pipeline {
+			raw, err := bson.Marshal(stage)
+			require.NoError(t, err)
+			if value, err := bson.Raw(raw).LookupErr(path...); err == nil {
+				return value
+			}
+		}
+		return bson.RawValue{}
+	}
+
+	excludedOpsInPipeline := func(t *testing.T, pipeline mongo.Pipeline) []string {
+		t.Helper()
+		ninValue := lookupInPipeline(t, pipeline, "$match", "operationType", "$nin")
+		if ninValue.IsZero() {
+			return nil
+		}
+		values, err := ninValue.Array().Values()
+		require.NoError(t, err)
+		ops := make([]string, 0, len(values))
+		for _, value := range values {
+			ops = append(ops, value.StringValue())
+		}
+		return ops
+	}
+
+	requireProjectFields := func(t *testing.T, pipeline mongo.Pipeline) {
+		t.Helper()
+		for _, field := range []string{"operationType", "clusterTime", "documentKey", "fullDocument", "ns"} {
+			require.False(t, lookupInPipeline(t, pipeline, "$project", field).IsZero())
+		}
+	}
+
+	t.Run("pipeline without filters", func(t *testing.T) {
+		pipeline, err := createPipeline(nil, nil)
+		require.NoError(t, err)
+
+		requireProjectFields(t, pipeline)
+		require.True(t, lookupInPipeline(t, pipeline, "$match").IsZero())
+	})
+
+	t.Run("pipeline with table mapping", func(t *testing.T) {
+		pipeline, err := createPipeline(tableNameMapping, nil)
+		require.NoError(t, err)
+
+		requireProjectFields(t, pipeline)
+		require.Equal(t, "db", lookupInPipeline(t, pipeline, "$match", "$or", "0", "$and", "0", "ns.db").StringValue())
+		require.Equal(t, "coll", lookupInPipeline(t, pipeline, "$match", "$or", "0", "$and", "1", "ns.coll", "$in", "0").StringValue())
+		require.Empty(t, excludedOpsInPipeline(t, pipeline))
+	})
+
+	t.Run("pipeline with excluded operation types", func(t *testing.T) {
+		pipeline, err := createPipeline(nil, []operationType{operationTypeDelete})
+		require.NoError(t, err)
+
+		requireProjectFields(t, pipeline)
+		require.Equal(t, []string{"delete"}, excludedOpsInPipeline(t, pipeline))
+		require.Empty(t, lookupInPipeline(t, pipeline, "$match", "$or", "0", "$and", "1", "ns.coll", "$in", "0"))
+		require.Empty(t, lookupInPipeline(t, pipeline, "$match", "$or", "0", "$and", "0", "ns.db"))
+	})
+}
+
+func TestExcludedOperationTypes(t *testing.T) {
+	envKey := "PEERDB_MONGODB_EXCLUDED_OPERATION_TYPES"
+	resolve := func(t *testing.T, env map[string]string) ([]operationType, error) {
+		t.Helper()
+		connector := &MongoConnector{logger: internal.LoggerFromCtx(t.Context())}
+		if err := connector.SetupReplConn(t.Context(), env); err != nil {
+			return nil, err
+		}
+		return connector.excludedOps, nil
+	}
+
+	t.Run("valid operation types", func(t *testing.T) {
+		ops, err := resolve(t, map[string]string{envKey: "Delete, UPDATE"})
+		require.NoError(t, err)
+		require.Equal(t, []operationType{operationTypeDelete, operationTypeUpdate}, ops)
+	})
+
+	t.Run("empty", func(t *testing.T) {
+		ops, err := resolve(t, map[string]string{envKey: ""})
+		require.NoError(t, err)
+		require.Empty(t, ops)
+	})
+
+	t.Run("invalid operation type is ignored", func(t *testing.T) {
+		ops, err := resolve(t, map[string]string{envKey: "delete,drop"})
+		require.NoError(t, err)
+		require.Equal(t, []operationType{operationTypeDelete}, ops)
+	})
+
+	t.Run("bad input is ignored", func(t *testing.T) {
+		ops, err := resolve(t, map[string]string{envKey: "123, bad input"})
+		require.NoError(t, err)
+		require.Empty(t, ops)
+	})
+}
+
 func TestDecodeEvent(t *testing.T) {
 	id := bson.NewObjectID()
 	insertTs := time.Now().UTC()

flow/connectors/mongo/mongo.go

@@ -66,6 +66,7 @@ type MongoConnector struct {
 	client             *mongo.Client
 	ssh                *utils.SSHTunnel
 	createChangeStream createChangeStreamFunc
+	excludedOps        []operationType
 	totalBytesRead     atomic.Int64
 	deltaBytesRead     atomic.Int64
 }

flow/connectors/mysql/cdc.go

@@ -1253,7 +1253,8 @@ func parseIncidentEvent(data []byte) (uint16, string) {
 }
 
 func (c *MySqlConnector) recordColumnTypeChange(ctx context.Context, otelManager *otel_metrics.OtelManager,
-	table, column string, from, to types.QValueKind, eventType string) {
+	table, column string, from, to types.QValueKind, eventType string,
+) {
 	key := fmt.Sprintf("%s.%s.%s.%s", table, column, from, to)
 	if _, ok := c.warnedTypeChanges.LoadOrStore(key, struct{}{}); !ok {
 		c.logger.Warn("column type change detected, not propagating",

flow/e2e/mongo_test.go

@@ -575,6 +575,53 @@ func (s MongoClickhouseSuite) Test_CDC() {
 	RequireEnvCanceled(t, env)
 }
 
+func (s MongoClickhouseSuite) Test_CDC_Excluded_Operation_Types() {
+	t := s.T()
+
+	srcDatabase := GetTestDatabase(s.Suffix())
+	srcTable := "test_excluded_op_types"
+	dstTable := "test_excluded_op_types_dst"
+
+	connectionGen := FlowConnectionGenerationConfig{
+		FlowJobName:   AddSuffix(s, srcTable),
+		TableMappings: TableMappings(s, srcTable, dstTable),
+		Destination:   s.Peer().Name,
+	}
+	flowConnConfig := s.generateFlowConnectionConfigsDefaultEnv(connectionGen)
+	flowConnConfig.Env["PEERDB_MONGODB_EXCLUDED_OPERATION_TYPES"] = "delete"
+
+	adminClient := s.Source().(*MongoSource).AdminClient()
+	err := adminClient.Database(srcDatabase).CreateCollection(t.Context(), srcTable)
+	require.NoError(t, err)
+
+	collection := adminClient.Database(srcDatabase).Collection(srcTable)
+
+	tc := NewTemporalClient(t)
+	env := ExecutePeerflow(t, tc, flowConnConfig)
+	SetupCDCFlowStatusQuery(t, env, flowConnConfig)
+
+	for i := range 2 {
+		insertRes, err := collection.InsertOne(t.Context(), bson.D{bson.E{Key: "key", Value: i}}, options.InsertOne())
+		require.NoError(t, err)
+		require.True(t, insertRes.Acknowledged)
+	}
+	EnvWaitForEqualTablesWithNames(env, s, "insert events", srcTable, dstTable, "_id,doc")
+
+	deleteRes, err := collection.DeleteOne(t.Context(), bson.D{bson.E{Key: "key", Value: 0}}, options.DeleteOne())
+	require.NoError(t, err)
+	require.Equal(t, int64(1), deleteRes.DeletedCount)
+
+	insertRes, err := collection.InsertOne(t.Context(), bson.D{bson.E{Key: "key", Value: 2}}, options.InsertOne())
+	require.NoError(t, err)
+	require.True(t, insertRes.Acknowledged)
+
+	// destination keeps all 3 rows (GetRows filters on FINAL and _peerdb_is_deleted = 0)
+	EnvWaitForCount(env, s, "insert after delete", dstTable, "_id,doc", 3)
+
+	env.Cancel(t.Context())
+	RequireEnvCanceled(t, env)
+}
+
 func (s MongoClickhouseSuite) Test_Document_With_Dots_In_Keys() {
 	t := s.T()
 

flow/internal/dynamicconf.go

@@ -509,6 +509,15 @@ var DynamicSettings = [...]*protos.DynamicSetting{
 		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_AFTER_RESUME,
 		TargetForSetting: protos.DynconfTarget_ALL,
 	},
+	{
+		Name: "PEERDB_MONGODB_EXCLUDED_OPERATION_TYPES",
+		Description: "Comma-separated list of MongoDB change stream operation types to exclude from CDC " +
+			"(allowed values: insert, update, replace, delete)",
+		DefaultValue:     "",
+		ValueType:        protos.DynconfValueType_STRING,
+		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_AFTER_RESUME,
+		TargetForSetting: protos.DynconfTarget_ALL,
+	},
 }
 
 var DynamicIndex = func() map[string]int {
@@ -901,3 +910,17 @@ func PeerDBOffloadPartitionRanges(ctx context.Context, env map[string]string) (b
 func PeerDBPGAutomatedSchemaDump(ctx context.Context, env map[string]string) (bool, error) {
 	return dynamicConfBool(ctx, env, "PEERDB_PG_AUTOMATED_SCHEMA_DUMP")
 }
+
+func PeerDBMongoDBExcludedOperationTypes(ctx context.Context, env map[string]string) ([]string, error) {
+	value, err := dynLookup(ctx, env, "PEERDB_MONGODB_EXCLUDED_OPERATION_TYPES")
+	if err != nil {
+		return nil, err
+	}
+	var ops []string
+	for op := range strings.SplitSeq(value, ",") {
+		if op := strings.ToLower(strings.TrimSpace(op)); op != "" {
+			ops = append(ops, op)
+		}
+	}
+	return ops, nil
+}