MongoDB C# Driver如何基于FullDocument过滤WatchAsync变更流？

问题背景 #

原有基于Oplog的过滤逻辑可正常筛选符合条件的新增、更新、删除文档，但切换到collection.WatchAsync实现时遇到两处障碍：

1. 无法直接将FilterDefinition<BsonDocument>类型的文档过滤器应用到ChangeStreamDocument<BsonDocument>对象
2. 尝试渲染过滤器时抛出System.MissingMethodException（提示ChangeStreamDocumentSerializer无参构造函数不存在）

原有Oplog过滤代码：

// filter generated by settings
FilterDefinition<BsonDocument> filter = GetFilterFromOverriddenMethod();
// interested only in u, d and filtered i documents
filter = Builders<BsonDocument>.Filter.Or(
  Builders<BsonDocument>.Filter.In("op", new[] { "u", "d" }),
  Builders<BsonDocument>.Filter.And(
    Builders<BsonDocument>.Filter.Eq("op", "i"),
    filter
  )
);
// filter applied on oplog collection
filter = Builders<BsonDocument>.Filter.And(
  Builders<BsonDocument>.Filter.Eq("ns", "dbName.collectionName"),
  Builders<BsonDocument>.Filter.Gt("ts", lastId),
  filter
);

可行解决方案 #

方案1：手动映射过滤器到变更流文档结构 #

变更流的ChangeStreamDocument结构与Oplog文档差异较大，需将原有针对业务文档的过滤器，映射到变更流的对应字段（新增/更新用FullDocument，删除用DocumentKey），同时匹配变更类型：

// 获取原有业务文档过滤器
var docFilter = GetFilterFromOverriddenMethod();

// 构建变更流专属过滤条件
var changeStreamFilter = Builders<ChangeStreamDocument<BsonDocument>>.Filter.Or(
    // 匹配更新操作，校验更新后的完整文档
    Builders<ChangeStreamDocument<BsonDocument>>.Filter.And(
        Builders<ChangeStreamDocument<BsonDocument>>.Filter.Eq(c => c.OperationType, ChangeStreamOperationType.Update),
        Builders<ChangeStreamDocument<BsonDocument>>.Filter.ElemMatch(c => c.FullDocument, docFilter)
    ),
    // 匹配删除操作（仅能通过DocumentKey校验，若原有过滤器含非主键条件需额外处理）
    Builders<ChangeStreamDocument<BsonDocument>>.Filter.Eq(c => c.OperationType, ChangeStreamOperationType.Delete),
    // 匹配插入操作，校验插入的完整文档
    Builders<ChangeStreamDocument<BsonDocument>>.Filter.And(
        Builders<ChangeStreamDocument<BsonDocument>>.Filter.Eq(c => c.OperationType, ChangeStreamOperationType.Insert),
        Builders<ChangeStreamDocument<BsonDocument>>.Filter.ElemMatch(c => c.FullDocument, docFilter)
    )
);

// 启动变更流，开启UpdateLookup以获取更新后的完整文档
var cursor = await collection.WatchAsync(
    new ChangeStreamOptions { FullDocument = ChangeStreamFullDocumentOption.UpdateLookup }, 
    changeStreamFilter
);

方案2：渲染原有过滤器为BsonDocument后适配变更流路径 #

若不想手动重构过滤器逻辑，可先将原有FilterDefinition<BsonDocument>渲染为BsonDocument，再调整字段路径指向变更流的FullDocument：

var docFilter = GetFilterFromOverriddenMethod();
// 使用BsonDocument序列化器渲染原有过滤器
var bsonSerializer = BsonSerializer.SerializerRegistry.GetSerializer<BsonDocument>();
var renderedDocFilter = docFilter.Render(bsonSerializer, BsonSerializer.SerializerRegistry);

// 调整过滤器字段路径，指向变更流的fullDocument字段
var adjustedInsertUpdateFilter = new BsonDocument("$expr", new BsonDocument("$and", new BsonArray
{
    new BsonDocument("$in", new BsonArray { "$operationType", new BsonArray { "insert", "update" } }),
    renderedDocFilter.ReplaceKeyNames(k => $"fullDocument.{k}")
}));

// 单独处理删除操作
var deleteFilter = new BsonDocument("operationType", "delete");

// 合并所有过滤条件
var changeStreamFilter = Builders<BsonDocument>.Filter.Or(adjustedInsertUpdateFilter, deleteFilter);

// 构建变更流管道并启动
var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<BsonDocument>>()
    .Match(changeStreamFilter);

var cursor = await collection.WatchAsync(
    pipeline, 
    new ChangeStreamOptions { FullDocument = ChangeStreamFullDocumentOption.UpdateLookup }
);

其中ReplaceKeyNames为自定义工具方法，用于批量修改过滤器的字段前缀：

public static BsonDocument ReplaceKeyNames(this BsonDocument doc, Func<string, string> keyMapper)
{
    var newDoc = new BsonDocument();
    foreach (var element in doc.Elements)
    {
        if (element.Value is BsonDocument subDoc)
        {
            newDoc.Add(keyMapper(element.Name), subDoc.ReplaceKeyNames(keyMapper));
        }
        else if (element.Value is BsonArray array)
        {
            var newArray = new BsonArray();
            foreach (var item in array)
            {
                newArray.Add(item is BsonDocument arrayDoc ? arrayDoc.ReplaceKeyNames(keyMapper) : item);
            }
            newDoc.Add(keyMapper(element.Name), newArray);
        }
        else
        {
            newDoc.Add(keyMapper(element.Name), element.Value);
        }
    }
    return newDoc;
}

关键注意事项 #

• 变更流OperationType与Oplogop字段的对应关系：insert→i，update→u，delete→d
• 必须设置ChangeStreamFullDocumentOption.UpdateLookup，才能在更新操作中获取完整的文档内容
• 删除操作无法获取完整业务文档，仅能通过DocumentKey匹配，若原有过滤器包含非主键条件，需提前在业务逻辑中做额外处理

内容的提问来源于stack exchange，提问作者Criton