diff --git a/flow/connectors/mongo/cdc.go b/flow/connectors/mongo/cdc.go index 6afe3aa83..57eecf297 100644 --- a/flow/connectors/mongo/cdc.go +++ b/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 } diff --git a/flow/connectors/mongo/cdc_test.go b/flow/connectors/mongo/cdc_test.go index 4ef20da3c..a76ad61ab 100644 --- a/flow/connectors/mongo/cdc_test.go +++ b/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() diff --git a/flow/connectors/mongo/mongo.go b/flow/connectors/mongo/mongo.go index 212fd6808..668b3e8cd 100644 --- a/flow/connectors/mongo/mongo.go +++ b/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 } diff --git a/flow/connectors/mysql/cdc.go b/flow/connectors/mysql/cdc.go index 7bc3c44de..7e80c92ca 100644 --- a/flow/connectors/mysql/cdc.go +++ b/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", diff --git a/flow/e2e/mongo_test.go b/flow/e2e/mongo_test.go index e8c22a9ee..3dcaa3270 100644 --- a/flow/e2e/mongo_test.go +++ b/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() diff --git a/flow/internal/dynamicconf.go b/flow/internal/dynamicconf.go index bd96c5077..f2aed28f0 100644 --- a/flow/internal/dynamicconf.go +++ b/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 +}