Cosmos DB聚合管道$match报错：多阶段$match不支持问题解决

问题背景 #

你这段代码在本地运行正常，但部署到Cosmos DB后执行var totalRecords = query.Count()时遇到了明确的错误：

{ "_t": "OKMongoResponse", "ok": 0, "code": 118, "errmsg": "$match is currently only supported when it is the first and only stage of the aggregation pipeline. Please restructure your query to combine multiple $match stages into a single $match stage.", "$err": "$match is currently only supported when it is the first and only stage of the aggregation pipeline. Please restructure your query to combine multiple $match stages into a single $match stage." }

问题根源在ApplyFilter方法里：你循环调用Where添加过滤条件，每个Where都会被Cosmos DB转化为独立的$match聚合阶段，而Cosmos DB不允许管道中存在多个分散的$match阶段，必须合并为一个。

解决方案思路 #

我们需要把多个过滤条件合并成单一的Where子句，让Cosmos DB只生成一个$match阶段，完全符合它的聚合规则。

重构后的代码 #

首先修改ApplyFilter方法，通过表达式合并来构建统一的过滤条件：

private IQueryable<ConnectionGetDto> ApplyFilter(IQueryable<ConnectionGetDto> query, IEnumerable<PermittedCommunicationType> permittedCommunication)
{
    if (permittedCommunication == null || !permittedCommunication.Any())
        return query;

    // 初始化一个恒真表达式，后续逐个AND追加条件
    Expression<Func<ConnectionGetDto, bool>> combinedFilter = c => true;

    foreach (var p in permittedCommunication)
    {
        switch (p)
        {
            case PermittedCommunicationType.Call:
                combinedFilter = combinedFilter.And(c => c.PermittedCommunication.Call);
                break;
            case PermittedCommunicationType.Email:
                combinedFilter = combinedFilter.And(c => c.PermittedCommunication.Email);
                break;
            case PermittedCommunicationType.Im:
                combinedFilter = combinedFilter.And(c => c.PermittedCommunication.Im);
                break;
            case PermittedCommunicationType.Letter:
                combinedFilter = combinedFilter.And(c => c.PermittedCommunication.Letter);
                break;
            case PermittedCommunicationType.Sms:
                combinedFilter = combinedFilter.And(c => c.PermittedCommunication.Sms);
                break;
        }
    }

    return query.Where(combinedFilter);
}

还需要添加一个表达式合并的扩展方法，用来处理LINQ表达式的参数替换与逻辑合并：

public static class ExpressionExtensions
{
    public static Expression<Func<T, bool>> And<T>(this Expression<Func<T, bool>> left, Expression<Func<T, bool>> right)
    {
        var parameter = Expression.Parameter(typeof(T));
        var visitor = new ReplaceParameterVisitor(left.Parameters[0], parameter);

        var leftBody = visitor.Visit(left.Body);
        var rightBody = visitor.Visit(right.Body);

        return Expression.Lambda<Func<T, bool>>(Expression.AndAlso(leftBody, rightBody), parameter);
    }

    private class ReplaceParameterVisitor : ExpressionVisitor
    {
        private readonly ParameterExpression _oldParam;
        private readonly ParameterExpression _newParam;

        public ReplaceParameterVisitor(ParameterExpression oldParam, ParameterExpression newParam)
        {
            _oldParam = oldParam;
            _newParam = newParam;
        }

        protected override Expression VisitParameter(ParameterExpression node)
        {
            return node == _oldParam ? _newParam : base.VisitParameter(node);
        }
    }
}

另外，优化GetConnections方法的异步逻辑——原来的Task.Run包裹LINQ查询是多余的，Cosmos DB的查询是延迟执行的，应该在最终执行计数时用异步方法：

public async Task GetConnections(string enterpriseId, IEnumerable<PermittedCommunicationType> permittedCommunication) 
{ 
    IQueryable<ConnectionGetDto> query = from conn in _connectionCollection.AsQueryable() 
                                         where conn.EnterpriseId == enterpriseId 
                                         join consumer in _consumerCollection.AsQueryable() on conn.ConsumerId equals consumer.Id into joined 
                                         select new ConnectionGetDto 
                                         { 
                                             Id = conn.Id, 
                                             AdditionalData = conn.AdditionalData, 
                                             FirstName = joined.First().Profile.FirstName, 
                                         }; 

    query = ApplyFilter(query, permittedCommunication); 

    // 使用异步计数方法，符合异步编程规范
    var totalRecords = await query.CountAsync(); 
}

为什么这样能解决问题 #

原来的循环添加Where会生成多个独立的$match阶段，而合并成单一Where后，所有过滤条件会被组合成一个完整的$match阶段，完美适配Cosmos DB的聚合管道要求，自然就不会再触发118错误了。

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