Document redactSafePortionDollarOps(BSONObj expr) {
    MutableDocument output;
    BSONForEach(field, expr) {
        if (field.fieldName()[0] != '$')
            continue;

        auto keyword = MatchExpressionParser::parsePathAcceptingKeyword(field);
        if (!keyword) {
            continue;
        }

        switch (*keyword) {
            // These are always ok
            case PathAcceptingKeyword::TYPE:
            case PathAcceptingKeyword::REGEX:
            case PathAcceptingKeyword::OPTIONS:
            case PathAcceptingKeyword::MOD:
            case PathAcceptingKeyword::BITS_ALL_SET:
            case PathAcceptingKeyword::BITS_ALL_CLEAR:
            case PathAcceptingKeyword::BITS_ANY_SET:
            case PathAcceptingKeyword::BITS_ANY_CLEAR:
                output[field.fieldNameStringData()] = Value(field);
                break;

            // These are ok if the type of the rhs is allowed in comparisons
            case PathAcceptingKeyword::EQUALITY:
            case PathAcceptingKeyword::LESS_THAN_OR_EQUAL:
            case PathAcceptingKeyword::GREATER_THAN_OR_EQUAL:
            case PathAcceptingKeyword::LESS_THAN:
            case PathAcceptingKeyword::GREATER_THAN:
                if (isTypeRedactSafeInComparison(field.type()))
                    output[field.fieldNameStringData()] = Value(field);
                break;

            // $in must be all-or-nothing (like $or). Can't include subset of elements.
            case PathAcceptingKeyword::IN_EXPR: {
                bool allOk = true;
                BSONForEach(elem, field.Obj()) {
                    if (!isTypeRedactSafeInComparison(elem.type())) {
                        allOk = false;
                        break;
                    }
                }
                if (allOk) {
                    output[field.fieldNameStringData()] = Value(field);
                }

                break;
            }

            case PathAcceptingKeyword::ALL: {
                // $all can include subset of elements (like $and).
                vector<Value> matches;
                BSONForEach(elem, field.Obj()) {
                    // NOTE this currently doesn't allow {$all: [{$elemMatch: {...}}]}
                    if (isTypeRedactSafeInComparison(elem.type())) {
                        matches.push_back(Value(elem));
                    }
                }
                if (!matches.empty())
                    output[field.fieldNameStringData()] = Value(std::move(matches));

                break;
            }

            case PathAcceptingKeyword::ELEM_MATCH: {
                BSONObj subIn = field.Obj();
                Document subOut;
                if (subIn.firstElementFieldName()[0] == '$') {
                    subOut = redactSafePortionDollarOps(subIn);
                } else {
                    subOut = redactSafePortionTopLevel(subIn);
                }

                if (!subOut.empty())
                    output[field.fieldNameStringData()] = Value(subOut);

                break;
            }

            // These are never allowed
            case PathAcceptingKeyword::EXISTS:
            case PathAcceptingKeyword::GEO_INTERSECTS:
            case PathAcceptingKeyword::GEO_NEAR:
            case PathAcceptingKeyword::INTERNAL_EXPR_EQ:
            case PathAcceptingKeyword::INTERNAL_EXPR_GT:
            case PathAcceptingKeyword::INTERNAL_EXPR_GTE:
            case PathAcceptingKeyword::INTERNAL_EXPR_LT:
            case PathAcceptingKeyword::INTERNAL_EXPR_LTE:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_ALL_ELEM_MATCH_FROM_INDEX:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_BIN_DATA_ENCRYPTED_TYPE:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_BIN_DATA_SUBTYPE:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_EQ:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_FMOD:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_MATCH_ARRAY_INDEX:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_MAX_ITEMS:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_MAX_LENGTH:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_MIN_ITEMS:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_MIN_LENGTH:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_OBJECT_MATCH:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_TYPE:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_UNIQUE_ITEMS:
            case PathAcceptingKeyword::NOT_EQUAL:
            case PathAcceptingKeyword::NOT_IN:
            case PathAcceptingKeyword::SIZE:
            case PathAcceptingKeyword::WITHIN:
                continue;
        }
    }
    return output.freeze();
}
Document redactSafePortionDollarOps(BSONObj expr) {
    MutableDocument output;
    BSONForEach(field, expr) {
        if (field.fieldName()[0] != '$')
            continue;

        auto keyword = MatchExpressionParser::parsePathAcceptingKeyword(field);
        if (!keyword) {
            continue;
        }

        switch (*keyword) {
             
            case PathAcceptingKeyword::TYPE:
            case PathAcceptingKeyword::REGEX:
            case PathAcceptingKeyword::OPTIONS:
            case PathAcceptingKeyword::MOD:
            case PathAcceptingKeyword::BITS_ALL_SET:
            case PathAcceptingKeyword::BITS_ALL_CLEAR:
            case PathAcceptingKeyword::BITS_ANY_SET:
            case PathAcceptingKeyword::BITS_ANY_CLEAR:
                output[field.fieldNameStringData()] = Value(field);
                break;

             
            case PathAcceptingKeyword::EQUALITY:
            case PathAcceptingKeyword::LESS_THAN_OR_EQUAL:
            case PathAcceptingKeyword::GREATER_THAN_OR_EQUAL:
            case PathAcceptingKeyword::LESS_THAN:
            case PathAcceptingKeyword::GREATER_THAN:
                if (isTypeRedactSafeInComparison(field.type()))
                    output[field.fieldNameStringData()] = Value(field);
                break;

             
            case PathAcceptingKeyword::IN_EXPR: {
                bool allOk = true;
                BSONForEach(elem, field.Obj()) {
                    if (!isTypeRedactSafeInComparison(elem.type())) {
                        allOk = false;
                        break;
                    }
                }
                if (allOk) {
                    output[field.fieldNameStringData()] = Value(field);
                }

                break;
            }

            case PathAcceptingKeyword::ALL: {
                 
                vector<Value> matches;
                BSONForEach(elem, field.Obj()) {
                     
                    if (isTypeRedactSafeInComparison(elem.type())) {
                        matches.push_back(Value(elem));
                    }
                }
                if (!matches.empty())
                    output[field.fieldNameStringData()] = Value(std::move(matches));

                break;
            }

            case PathAcceptingKeyword::ELEM_MATCH: {
                BSONObj subIn = field.Obj();
                Document subOut;
                if (subIn.firstElementFieldName()[0] == '$') {
                    subOut = redactSafePortionDollarOps(subIn);
                } else {
                    subOut = redactSafePortionTopLevel(subIn);
                }

                if (!subOut.empty())
                    output[field.fieldNameStringData()] = Value(subOut);

                break;
            }

             
            case PathAcceptingKeyword::EXISTS:
            case PathAcceptingKeyword::GEO_INTERSECTS:
            case PathAcceptingKeyword::GEO_NEAR:
            case PathAcceptingKeyword::INTERNAL_EXPR_EQ:
            case PathAcceptingKeyword::INTERNAL_EXPR_GT:
            case PathAcceptingKeyword::INTERNAL_EXPR_GTE:
            case PathAcceptingKeyword::INTERNAL_EXPR_LT:
            case PathAcceptingKeyword::INTERNAL_EXPR_LTE:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_ALL_ELEM_MATCH_FROM_INDEX:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_BIN_DATA_ENCRYPTED_TYPE:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_BIN_DATA_SUBTYPE:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_EQ:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_FMOD:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_MATCH_ARRAY_INDEX:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_MAX_ITEMS:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_MAX_LENGTH:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_MIN_ITEMS:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_MIN_LENGTH:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_OBJECT_MATCH:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_TYPE:
            case PathAcceptingKeyword::INTERNAL_SCHEMA_UNIQUE_ITEMS:
            case PathAcceptingKeyword::NOT_EQUAL:
            case PathAcceptingKeyword::NOT_IN:
            case PathAcceptingKeyword::SIZE:
            case PathAcceptingKeyword::WITHIN:
                continue;
        }
    }
    return output.freeze();
}