Revert "Revert "Merge branch 'v6.0' of github.com:10gen/mongo into v6.0""

This reverts commit 44385b89a690f1feda55d8e888a08f33eab0a020.

13 files changed, 611 insertions, 11 deletions

diff --git a/src/mongo/db/pipeline/SConscript b/src/mongo/db/pipeline/SConscript
index 6b14f635095..16e39fc4832 100644
--- a/src/mongo/db/pipeline/SConscript
+++ b/src/mongo/db/pipeline/SConscript
@@ -99,6 +99,7 @@ env.Library(
         'expression_context.cpp',
         'expression_function.cpp',
         'expression_js_emit.cpp',
+        'expression_parser.idl',
         'expression_test_api_version.cpp',
         'expression_trigonometric.cpp',
         'javascript_execution.cpp',
@@ -106,6 +107,7 @@ env.Library(
         'variables.cpp',
     ],
     LIBDEPS=[
+        '$BUILD_DIR/mongo/crypto/fle_crypto',
         '$BUILD_DIR/mongo/db/bson/dotted_path_support',
         '$BUILD_DIR/mongo/db/commands/test_commands_enabled',
         '$BUILD_DIR/mongo/db/exec/document_value/document_value',
@@ -128,6 +130,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/mongohasher',
         '$BUILD_DIR/mongo/db/vector_clock',
+        '$BUILD_DIR/mongo/idl/idl_parser',
     ],
 )
 
diff --git a/src/mongo/db/pipeline/abt/agg_expression_visitor.cpp b/src/mongo/db/pipeline/abt/agg_expression_visitor.cpp
index 3d0c8453a47..55b44e4d37b 100644
--- a/src/mongo/db/pipeline/abt/agg_expression_visitor.cpp
+++ b/src/mongo/db/pipeline/abt/agg_expression_visitor.cpp
@@ -302,6 +302,10 @@ public:
         unsupportedExpression(expr->getOpName());
     }
 
+    void visit(const ExpressionInternalFLEEqual* expr) override final {
+        unsupportedExpression(expr->getOpName());
+    }
+
     void visit(const ExpressionMap* expr) override final {
         unsupportedExpression("$map");
     }
diff --git a/src/mongo/db/pipeline/dependencies.cpp b/src/mongo/db/pipeline/dependencies.cpp
index 8b60a31637c..d2a5563c7c7 100644
--- a/src/mongo/db/pipeline/dependencies.cpp
+++ b/src/mongo/db/pipeline/dependencies.cpp
@@ -37,6 +37,13 @@
 
 namespace mongo {
 
+std::list<std::string> DepsTracker::sortedFields() const {
+    // Use a special comparator to put parent fieldpaths before their children.
+    std::list<std::string> sortedFields(fields.begin(), fields.end());
+    sortedFields.sort(PathPrefixComparator());
+    return sortedFields;
+}
+
 BSONObj DepsTracker::toProjectionWithoutMetadata(
     TruncateToRootLevel truncationBehavior /*= TruncateToRootLevel::no*/) const {
     BSONObjBuilder bb;
@@ -52,17 +59,21 @@ BSONObj DepsTracker::toProjectionWithoutMetadata(
         return bb.obj();
     }
 
+    // Go through dependency fieldpaths to find the minimal set of projections that cover the
+    // dependencies. For example, the dependencies ["a.b", "a.b.c.g", "c", "c.d", "f"] would be
+    // minimally covered by the projection {"a.b": 1, "c": 1, "f": 1}. The key operation here is
+    // folding dependencies into ancestor dependencies, wherever possible. This is assisted by a
+    // special sort in DepsTracker::sortedFields that treats '.' as the first char and thus places
+    // parent paths directly before their children.
     bool idSpecified = false;
     std::string last;
-    for (const auto& field : fields) {
+    for (const auto& field : sortedFields()) {
         if (str::startsWith(field, "_id") && (field.size() == 3 || field[3] == '.')) {
             idSpecified = true;
         }
 
         if (!last.empty() && str::startsWith(field, last)) {
-            // we are including a parent of *it so we don't need to include this field
-            // explicitly. This logic relies on on set iterators going in lexicographic order so
-            // that a string is always directly before of all fields it prefixes.
+            // We are including a parent of this field, so we can skip this field.
             continue;
         }
 
@@ -96,4 +107,36 @@ void DepsTracker::setNeedsMetadata(DocumentMetadataFields::MetaType type, bool r
     invariant(required || !_metadataDeps[type]);
     _metadataDeps[type] = required;
 }
+
+// Returns true if the lhs value should sort before the rhs, false otherwise.
+bool PathPrefixComparator::operator()(const std::string& lhs, const std::string& rhs) const {
+    constexpr char dot = '.';
+
+    for (size_t pos = 0, len = std::min(lhs.size(), rhs.size()); pos < len; ++pos) {
+        // Below, we explicitly choose unsigned char because the usual const char& returned by
+        // operator[] is actually signed on x86 and will incorrectly order unicode characters.
+        unsigned char lchar = lhs[pos], rchar = rhs[pos];
+        if (lchar == rchar) {
+            continue;
+        }
+
+        // Consider the path delimiter '.' as being less than all other characters, so that
+        // paths sort directly before any paths they prefix and directly after any paths
+        // which prefix them.
+        if (lchar == dot) {
+            return true;
+        } else if (rchar == dot) {
+            return false;
+        }
+
+        // Otherwise, default to normal character comparison.
+        return lchar < rchar;
+    }
+
+    // If we get here, then we have reached the end of lhs and/or rhs and all of their path
+    // segments up to this point match. If lhs is shorter than rhs, then lhs prefixes rhs
+    // and should sort before it.
+    return lhs.size() < rhs.size();
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/dependencies.h b/src/mongo/db/pipeline/dependencies.h
index bda3bf9b243..3c892de8181 100644
--- a/src/mongo/db/pipeline/dependencies.h
+++ b/src/mongo/db/pipeline/dependencies.h
@@ -184,6 +184,11 @@ struct DepsTracker {
         }
     }
 
+    /**
+     * Return fieldpaths ordered such that a parent is immediately before its children.
+     */
+    std::list<std::string> sortedFields() const;
+
     std::set<std::string> fields;    // Names of needed fields in dotted notation.
     std::set<Variables::Id> vars;    // IDs of referenced variables.
     bool needWholeDocument = false;  // If true, ignore 'fields'; the whole document is needed.
@@ -201,4 +206,13 @@ private:
     // dependency analysis.
     QueryMetadataBitSet _metadataDeps;
 };
+
+
+/** Custom comparator that orders fieldpath strings by path prefix first, then by field.
+ * This ensures that a parent field is ordered directly before its children.
+ */
+struct PathPrefixComparator {
+    /* Returns true if the lhs value should sort before the rhs, false otherwise. */
+    bool operator()(const std::string& lhs, const std::string& rhs) const;
+};
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/dependencies_test.cpp b/src/mongo/db/pipeline/dependencies_test.cpp
index f366ad3ce1d..938130b91bd 100644
--- a/src/mongo/db/pipeline/dependencies_test.cpp
+++ b/src/mongo/db/pipeline/dependencies_test.cpp
@@ -162,6 +162,13 @@ TEST(DependenciesToProjectionTest, ShouldIncludeFieldEvenIfSuffixOfAnotherFieldW
                       BSON("a" << 1 << "ab" << 1 << "_id" << 0));
 }
 
+TEST(DependenciesToProjectionTest, ExcludeIndirectDescendants) {
+    const char* array[] = {"a.b", "_id", "a.b.c.d.e"};
+    DepsTracker deps;
+    deps.fields = arrayToSet(array);
+    ASSERT_BSONOBJ_EQ(deps.toProjectionWithoutMetadata(), BSON("_id" << 1 << "a.b" << 1));
+}
+
 TEST(DependenciesToProjectionTest, ShouldIncludeIdIfNeeded) {
     const char* array[] = {"a", "_id"};
     DepsTracker deps;
@@ -199,6 +206,27 @@ TEST(DependenciesToProjectionTest, ShouldIncludeFieldPrefixedByIdWhenIdSubfieldI
                       BSON("_id.a" << 1 << "_id_a" << 1 << "a" << 1));
 }
 
+// SERVER-66418
+TEST(DependenciesToProjectionTest, ChildCoveredByParentWithSpecialChars) {
+    // without "_id"
+    {
+        // This is an important test case because '-' is one of the few chars before '.' in utf-8.
+        const char* array[] = {"a", "a-b", "a.b"};
+        DepsTracker deps;
+        deps.fields = arrayToSet(array);
+        ASSERT_BSONOBJ_EQ(deps.toProjectionWithoutMetadata(),
+                          BSON("a" << 1 << "a-b" << 1 << "_id" << 0));
+    }
+    // with "_id"
+    {
+        const char* array[] = {"_id", "a", "a-b", "a.b"};
+        DepsTracker deps;
+        deps.fields = arrayToSet(array);
+        ASSERT_BSONOBJ_EQ(deps.toProjectionWithoutMetadata(),
+                          BSON("_id" << 1 << "a" << 1 << "a-b" << 1));
+    }
+}
+
 TEST(DependenciesToProjectionTest, ShouldOutputEmptyObjectIfEntireDocumentNeeded) {
     const char* array[] = {"a"};  // fields ignored with needWholeDocument
     DepsTracker deps;
@@ -259,5 +287,56 @@ TEST(DependenciesToProjectionTest,
     ASSERT_TRUE(deps.metadataDeps()[DocumentMetadataFields::kTextScore]);
 }
 
+TEST(DependenciesToProjectionTest, SortFieldPaths) {
+    const char* array[] = {"",
+                           "A",
+                           "_id",
+                           "a",
+                           "a.b",
+                           "a.b.c",
+                           "a.c",
+                           // '-' char in utf-8 comes before '.' but our special fieldpath sort
+                           // puts '.' first so that children directly follow their parents.
+                           "a-b",
+                           "a-b.ear",
+                           "a-bear",
+                           "a-bear.",
+                           "a🌲",
+                           "b",
+                           "b.a"
+                           "b.aa"
+                           "b.🌲d"};
+    DepsTracker deps;
+    deps.fields = arrayToSet(array);
+    // our custom sort will restore the ordering above
+    std::list<std::string> fieldPathSorted = deps.sortedFields();
+    auto itr = fieldPathSorted.begin();
+    for (unsigned long i = 0; i < fieldPathSorted.size(); i++) {
+        ASSERT_EQ(*itr, array[i]);
+        ++itr;
+    }
+}
+
+TEST(DependenciesToProjectionTest, PathLessThan) {
+    auto lessThan = PathPrefixComparator();
+    ASSERT_FALSE(lessThan("a", "a"));
+    ASSERT_TRUE(lessThan("a", "aa"));
+    ASSERT_TRUE(lessThan("a", "b"));
+    ASSERT_TRUE(lessThan("", "a"));
+    ASSERT_TRUE(lessThan("Aa", "aa"));
+    ASSERT_TRUE(lessThan("a.b", "ab"));
+    ASSERT_TRUE(lessThan("a.b", "a-b"));  // SERVER-66418
+    ASSERT_TRUE(lessThan("a.b", "a b"));  // SERVER-66418
+    // verify the difference from the standard sort
+    ASSERT_TRUE(std::string("a.b") > std::string("a-b"));
+    ASSERT_TRUE(std::string("a.b") > std::string("a b"));
+    // test unicode behavior
+    ASSERT_TRUE(lessThan("a.b", "a🌲"));
+    ASSERT_TRUE(lessThan("a.b", "a🌲b"));
+    ASSERT_TRUE(lessThan("🌲", "🌳"));  // U+1F332 < U+1F333
+    ASSERT_TRUE(lessThan("🌲", "🌲.b"));
+    ASSERT_FALSE(lessThan("🌲.b", "🌲"));
+}
+
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/document_source_densify.cpp b/src/mongo/db/pipeline/document_source_densify.cpp
index 0325dbaa775..740ef193431 100644
--- a/src/mongo/db/pipeline/document_source_densify.cpp
+++ b/src/mongo/db/pipeline/document_source_densify.cpp
@@ -56,6 +56,10 @@ RangeStatement RangeStatement::parse(RangeSpec spec) {
 
     optional<TimeUnit> unit = [&]() {
         if (auto unit = spec.getUnit()) {
+            uassert(6586400,
+                    "The step parameter in a range statement must be a whole number when "
+                    "densifying a date range",
+                    step.integral64Bit());
             return optional<TimeUnit>(parseTimeUnit(unit.get()));
         } else {
             return optional<TimeUnit>(boost::none);
@@ -275,8 +279,8 @@ DocumentSourceInternalDensify::DocGenerator::DocGenerator(DensifyValue min,
         // Extra checks for date step + unit.
         tassert(5733501, "Unit must be specified with a date step", _range.getUnit());
         tassert(5733505,
-                "Step must be representable as an integer for date densification",
-                _range.getStep().integral());
+                "Step must be a whole number for date densification",
+                _range.getStep().integral64Bit());
     } else {
         tassert(5733506, "Unit must not be specified with non-date values", !_range.getUnit());
     }
@@ -877,7 +881,7 @@ DensifyValue DensifyValue::increment(const RangeStatement& range) const {
             },
             [&](Date_t date) {
                 return DensifyValue(dateAdd(
-                    date, range.getUnit().value(), range.getStep().getDouble(), timezone()));
+                    date, range.getUnit().value(), range.getStep().coerceToLong(), timezone()));
             }},
         _value);
 }
@@ -891,7 +895,7 @@ DensifyValue DensifyValue::decrement(const RangeStatement& range) const {
             },
             [&](Date_t date) {
                 return DensifyValue(dateAdd(
-                    date, range.getUnit().value(), -range.getStep().getDouble(), timezone()));
+                    date, range.getUnit().value(), -range.getStep().coerceToLong(), timezone()));
             }},
         _value);
 }
@@ -906,7 +910,7 @@ bool DensifyValue::isOnStepRelativeTo(DensifyValue base, RangeStatement range) c
             },
             [&](Date_t date) {
                 auto unit = range.getUnit().value();
-                double step = range.getStep().getDouble();
+                long long step = range.getStep().coerceToLong();
                 auto baseDate = base.getDate();
 
                 // Months, quarters and years have variable lengths depending on leap days
diff --git a/src/mongo/db/pipeline/document_source_densify_test.cpp b/src/mongo/db/pipeline/document_source_densify_test.cpp
index 79b16303eb9..c1a0218e9eb 100644
--- a/src/mongo/db/pipeline/document_source_densify_test.cpp
+++ b/src/mongo/db/pipeline/document_source_densify_test.cpp
@@ -374,7 +374,7 @@ DEATH_TEST(DensifyGeneratorTest, DateMinMustBeLessThanMax, "lower or equal to")
                        5733502);
 }
 
-DEATH_TEST(DensifyGeneratorTest, DateStepMustBeInt, "integer") {
+DEATH_TEST(DensifyGeneratorTest, DateStepMustBeInt, "whole number") {
     size_t counter = 0;
     ASSERT_THROWS_CODE(GenClass(makeDate("2021-01-01T00:00:00.000Z"),
                                 RangeStatement(Value(1.5),
diff --git a/src/mongo/db/pipeline/expression.cpp b/src/mongo/db/pipeline/expression.cpp
index cc3728c024f..bf0915deddd 100644
--- a/src/mongo/db/pipeline/expression.cpp
+++ b/src/mongo/db/pipeline/expression.cpp
@@ -39,6 +39,9 @@
 #include <utility>
 #include <vector>
 
+#include "mongo/bson/bsonmisc.h"
+#include "mongo/bson/bsontypes.h"
+#include "mongo/crypto/fle_crypto.h"
 #include "mongo/db/bson/dotted_path_support.h"
 #include "mongo/db/commands/feature_compatibility_version_documentation.h"
 #include "mongo/db/exec/document_value/document.h"
@@ -46,6 +49,7 @@
 #include "mongo/db/hasher.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/pipeline/expression_context.h"
+#include "mongo/db/pipeline/expression_parser_gen.h"
 #include "mongo/db/pipeline/variable_validation.h"
 #include "mongo/db/query/datetime/date_time_support.h"
 #include "mongo/db/query/sort_pattern.h"
@@ -3742,6 +3746,123 @@ const char* ExpressionLog10::getOpName() const {
     return "$log10";
 }
 
+/* ----------------------- ExpressionInternalFLEEqual ---------------------------- */
+constexpr auto kInternalFleEq = "$_internalFleEq"_sd;
+
+ExpressionInternalFLEEqual::ExpressionInternalFLEEqual(ExpressionContext* const expCtx,
+                                                       boost::intrusive_ptr<Expression> field,
+                                                       ConstDataRange serverToken,
+                                                       int64_t contentionFactor,
+                                                       ConstDataRange edcToken)
+    : Expression(expCtx, {std::move(field)}),
+      _serverToken(PrfBlockfromCDR(serverToken)),
+      _edcToken(PrfBlockfromCDR(edcToken)),
+      _contentionFactor(contentionFactor) {
+    expCtx->sbeCompatible = false;
+
+    auto tokens =
+        EDCServerCollection::generateEDCTokens(ConstDataRange(_edcToken), _contentionFactor);
+
+    for (auto& token : tokens) {
+        _cachedEDCTokens.insert(std::move(token.data));
+    }
+}
+
+void ExpressionInternalFLEEqual::_doAddDependencies(DepsTracker* deps) const {
+    for (auto&& operand : _children) {
+        operand->addDependencies(deps);
+    }
+}
+
+REGISTER_EXPRESSION_WITH_MIN_VERSION(_internalFleEq,
+                                     ExpressionInternalFLEEqual::parse,
+                                     AllowedWithApiStrict::kAlways,
+                                     AllowedWithClientType::kAny,
+                                     multiversion::FeatureCompatibilityVersion::kVersion_6_0);
+
+intrusive_ptr<Expression> ExpressionInternalFLEEqual::parse(ExpressionContext* const expCtx,
+                                                            BSONElement expr,
+                                                            const VariablesParseState& vps) {
+
+    IDLParserErrorContext ctx(kInternalFleEq);
+    auto fleEq = InternalFleEqStruct::parse(ctx, expr.Obj());
+
+    auto fieldExpr = Expression::parseOperand(expCtx, fleEq.getField().getElement(), vps);
+
+    auto serverTokenPair = fromEncryptedConstDataRange(fleEq.getServerEncryptionToken());
+
+    uassert(6672405,
+            "Invalid server token",
+            serverTokenPair.first == EncryptedBinDataType::kFLE2TransientRaw &&
+                serverTokenPair.second.length() == sizeof(PrfBlock));
+
+    auto edcTokenPair = fromEncryptedConstDataRange(fleEq.getEdcDerivedToken());
+
+    uassert(6672406,
+            "Invalid edc token",
+            edcTokenPair.first == EncryptedBinDataType::kFLE2TransientRaw &&
+                edcTokenPair.second.length() == sizeof(PrfBlock));
+
+
+    auto cf = fleEq.getMaxCounter();
+    uassert(6672408, "Contention factor must be between 0 and 10000", cf >= 0 && cf < 10000);
+
+    return new ExpressionInternalFLEEqual(expCtx,
+                                          std::move(fieldExpr),
+                                          serverTokenPair.second,
+                                          fleEq.getMaxCounter(),
+                                          edcTokenPair.second);
+}
+
+Value toValue(const std::array<std::uint8_t, 32>& buf) {
+    auto vec = toEncryptedVector(EncryptedBinDataType::kFLE2TransientRaw, buf);
+    return Value(BSONBinData(vec.data(), vec.size(), BinDataType::Encrypt));
+}
+
+Value ExpressionInternalFLEEqual::serialize(bool explain) const {
+    return Value(Document{{kInternalFleEq,
+                           Document{{"field", _children[0]->serialize(explain)},
+                                    {"edc", toValue(_edcToken)},
+                                    {"counter", Value(static_cast<long long>(_contentionFactor))},
+                                    {"server", toValue(_serverToken)}}}});
+}
+
+Value ExpressionInternalFLEEqual::evaluate(const Document& root, Variables* variables) const {
+    // Inputs
+    // 1. Value for FLE2IndexedEqualityEncryptedValue field
+
+    Value fieldValue = _children[0]->evaluate(root, variables);
+
+    if (fieldValue.nullish()) {
+        return Value(BSONNULL);
+    }
+
+    if (fieldValue.getType() != BinData) {
+        return Value(false);
+    }
+
+    auto fieldValuePair = fromEncryptedBinData(fieldValue);
+
+    uassert(6672407,
+            "Invalid encrypted indexed field",
+            fieldValuePair.first == EncryptedBinDataType::kFLE2EqualityIndexedValue);
+
+    // Value matches if
+    // 1. Decrypt field is successful
+    // 2. EDC_u Token is in GenTokens(EDC Token, ContentionFactor)
+    //
+    auto swIndexed =
+        EDCServerCollection::decryptAndParse(ConstDataRange(_serverToken), fieldValuePair.second);
+    uassertStatusOK(swIndexed);
+    auto indexed = swIndexed.getValue();
+
+    return Value(_cachedEDCTokens.count(indexed.edc.data) == 1);
+}
+
+const char* ExpressionInternalFLEEqual::getOpName() const {
+    return kInternalFleEq.rawData();
+}
+
 /* ------------------------ ExpressionNary ----------------------------- */
 
 /**
diff --git a/src/mongo/db/pipeline/expression.h b/src/mongo/db/pipeline/expression.h
index fa59e89ecf2..9cce6d0b1e2 100644
--- a/src/mongo/db/pipeline/expression.h
+++ b/src/mongo/db/pipeline/expression.h
@@ -29,6 +29,7 @@
 
 #pragma once
 
+#include "mongo/base/data_range.h"
 #include "mongo/platform/basic.h"
 
 #include <algorithm>
@@ -2197,6 +2198,38 @@ public:
     }
 };
 
+class ExpressionInternalFLEEqual final : public Expression {
+public:
+    ExpressionInternalFLEEqual(ExpressionContext* expCtx,
+                               boost::intrusive_ptr<Expression> field,
+                               ConstDataRange serverToken,
+                               int64_t contentionFactor,
+                               ConstDataRange edcToken);
+    Value serialize(bool explain) const final;
+
+    Value evaluate(const Document& root, Variables* variables) const final;
+    const char* getOpName() const;
+
+    static boost::intrusive_ptr<Expression> parse(ExpressionContext* expCtx,
+                                                  BSONElement expr,
+                                                  const VariablesParseState& vps);
+    void _doAddDependencies(DepsTracker* deps) const final;
+
+    void acceptVisitor(ExpressionMutableVisitor* visitor) final {
+        return visitor->visit(this);
+    }
+
+    void acceptVisitor(ExpressionConstVisitor* visitor) const final {
+        return visitor->visit(this);
+    }
+
+private:
+    std::array<std::uint8_t, 32> _serverToken;
+    std::array<std::uint8_t, 32> _edcToken;
+    int64_t _contentionFactor;
+    stdx::unordered_set<std::array<std::uint8_t, 32>> _cachedEDCTokens;
+};
+
 class ExpressionMap final : public Expression {
 public:
     ExpressionMap(
diff --git a/src/mongo/db/pipeline/expression_parser.idl b/src/mongo/db/pipeline/expression_parser.idl
new file mode 100644
index 00000000000..9f1cde70856
--- /dev/null
+++ b/src/mongo/db/pipeline/expression_parser.idl
@@ -0,0 +1,57 @@
+# Copyright (C) 2022-present MongoDB, Inc.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the Server Side Public License, version 1,
+# as published by MongoDB, Inc.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# Server Side Public License for more details.
+#
+# You should have received a copy of the Server Side Public License
+# along with this program. If not, see
+# <http://www.mongodb.com/licensing/server-side-public-license>.
+#
+# As a special exception, the copyright holders give permission to link the
+# code of portions of this program with the OpenSSL library under certain
+# conditions as described in each individual source file and distribute
+# linked combinations including the program with the OpenSSL library. You
+# must comply with the Server Side Public License in all respects for
+# all of the code used other than as permitted herein. If you modify file(s)
+# with this exception, you may extend this exception to your version of the
+# file(s), but you are not obligated to do so. If you do not wish to do so,
+# delete this exception statement from your version. If you delete this
+# exception statement from all source files in the program, then also delete
+# it in the license file.
+
+global:
+    cpp_namespace: "mongo"
+
+imports:
+    - "mongo/idl/basic_types.idl"
+
+structs:
+
+    InternalFleEqStruct:
+        description: "Struct for $_internalFleEq"
+        strict: true
+        fields:
+          field:
+            description: "Expression"
+            type: IDLAnyType
+            cpp_name: field
+          edc:
+            description: "EDCDerivedFromDataToken"
+            type: bindata_encrypt
+            cpp_name: edcDerivedToken
+          server:
+            description: "ServerDataEncryptionLevel1Token"
+            type: bindata_encrypt
+            cpp_name: serverEncryptionToken
+          counter:
+            description: "Queryable Encryption max counter"
+            type: long
+            cpp_name: maxCounter
+
+
diff --git a/src/mongo/db/pipeline/expression_test.cpp b/src/mongo/db/pipeline/expression_test.cpp
index 63eb5cc884c..fd6f1c3490e 100644
--- a/src/mongo/db/pipeline/expression_test.cpp
+++ b/src/mongo/db/pipeline/expression_test.cpp
@@ -169,6 +169,7 @@ void parseAndVerifyResults(
     ASSERT_VALUE_EQ(expr->evaluate({}, &expCtx.variables), expected);
 }
 
+
 /* ------------------------- ExpressionArrayToObject -------------------------- */
 
 TEST(ExpressionArrayToObjectTest, KVFormatSimple) {
@@ -3715,4 +3716,240 @@ TEST(ExpressionCondTest, ConstantCondShouldOptimizeWithNonConstantBranches) {
     ASSERT_BSONOBJ_BINARY_EQ(expectedResult, expressionToBson(optimizedExprCond));
 }
 
+TEST(ExpressionFLETest, BadInputs) {
+
+    auto expCtx = ExpressionContextForTest();
+    auto vps = expCtx.variablesParseState;
+    {
+        auto expr = fromjson("{$_internalFleEq: 12}");
+        ASSERT_THROWS_CODE(ExpressionInternalFLEEqual::parse(&expCtx, expr.firstElement(), vps),
+                           DBException,
+                           10065);
+    }
+}
+
+// Test we return true if it matches
+TEST(ExpressionFLETest, TestBinData) {
+    auto expCtx = ExpressionContextForTest();
+    auto vps = expCtx.variablesParseState;
+
+    {
+        auto expr = fromjson(R"({$_internalFleEq: {
+        field: {
+            "$binary": {
+                "base64":
+                "BxI0VngSNJh2EjQSNFZ4kBIQ0JE8aMUFkPk5sSTVqfdNNfjqUfQQ1Uoj0BBcthrWoe9wyU3cN6zmWaQBPJ97t0ZPbecnMsU736yXre6cBO4Zdt/wThtY+v5+7vFgNnWpgRP0e+vam6QPmLvbBrO0LdsvAPTGW4yqwnzCIXCoEg7QPGfbfAXKPDTNenBfRlawiblmTOhO/6ljKotWsMp22q/rpHrn9IEIeJmecwuuPIJ7EA+XYQ3hOKVccYf2ogoK73+8xD/Vul83Qvr84Q8afc4QUMVs8A==",
+                    "subType": "6"
+            }
+        },
+        server: {
+            "$binary": {
+                "base64": "COuac/eRLYakKX6B0vZ1r3QodOQFfjqJD+xlGiPu4/Ps",
+                "subType": "6"
+            }
+        },
+        counter: {
+            "$numberLong": "3"
+        },
+        edc: {
+            "$binary": {
+                "base64": "CEWSmQID7SfwyAUI3ZkSFkATKryDQfnxXEOGad5d4Rsg",
+                "subType": "6"
+            }
+        }    } })");
+        auto exprFle = ExpressionInternalFLEEqual::parse(&expCtx, expr.firstElement(), vps);
+
+        ASSERT_VALUE_EQ(exprFle->evaluate({}, &expCtx.variables), Value(true));
+    }
+
+    // Negative: Use wrong server token
+    {
+        auto expr = fromjson(R"({$_internalFleEq: {
+        field: {
+            "$binary": {
+                "base64":
+                "BxI0VngSNJh2EjQSNFZ4kBIQ0JE8aMUFkPk5sSTVqfdNNfjqUfQQ1Uoj0BBcthrWoe9wyU3cN6zmWaQBPJ97t0ZPbecnMsU736yXre6cBO4Zdt/wThtY+v5+7vFgNnWpgRP0e+vam6QPmLvbBrO0LdsvAPTGW4yqwnzCIXCoEg7QPGfbfAXKPDTNenBfRlawiblmTOhO/6ljKotWsMp22q/rpHrn9IEIeJmecwuuPIJ7EA+XYQ3hOKVccYf2ogoK73+8xD/Vul83Qvr84Q8afc4QUMVs8A==",
+                    "subType": "6"
+            }
+        },
+        server: {
+            "$binary": {
+                "base64": "COuac/eRLYakKX6B0vZ1r3QodOQFfjqJD+xlGiPu4/Ps",
+                "subType": "6"
+            }
+        },
+        counter: {
+            "$numberLong": "3"
+        },
+        edc: {
+            "$binary": {
+                "base64": "CEWSMQID7SFWYAUI3ZKSFKATKRYDQFNXXEOGAD5D4RSG",
+                "subType": "6"
+            }
+        }    } })");
+        auto exprFle = ExpressionInternalFLEEqual::parse(&expCtx, expr.firstElement(), vps);
+
+        ASSERT_VALUE_EQ(exprFle->evaluate({}, &expCtx.variables), Value(false));
+    }
+
+    // Negative: Use wrong edc token
+    {
+        auto expr = fromjson(R"({$_internalFleEq: {
+        field: {
+            "$binary": {
+                "base64":
+                "BxI0VngSNJh2EjQSNFZ4kBIQ0JE8aMUFkPk5sSTVqfdNNfjqUfQQ1Uoj0BBcthrWoe9wyU3cN6zmWaQBPJ97t0ZPbecnMsU736yXre6cBO4Zdt/wThtY+v5+7vFgNnWpgRP0e+vam6QPmLvbBrO0LdsvAPTGW4yqwnzCIXCoEg7QPGfbfAXKPDTNenBfRlawiblmTOhO/6ljKotWsMp22q/rpHrn9IEIeJmecwuuPIJ7EA+XYQ3hOKVccYf2ogoK73+8xD/Vul83Qvr84Q8afc4QUMVs8A==",
+                    "subType": "6"
+            }
+        },
+        server: {
+            "$binary": {
+                "base64": "COUAC/ERLYAKKX6B0VZ1R3QODOQFFJQJD+XLGIPU4/PS",
+                "subType": "6"
+            }
+        },
+        counter: {
+            "$numberLong": "3"
+        },
+        edc: {
+            "$binary": {
+                "base64": "CEWSmQID7SfwyAUI3ZkSFkATKryDQfnxXEOGad5d4Rsg",
+                "subType": "6"
+            }
+        }    } })");
+        auto exprFle = ExpressionInternalFLEEqual::parse(&expCtx, expr.firstElement(), vps);
+
+        ASSERT_THROWS_CODE(
+            exprFle->evaluate({}, &expCtx.variables), DBException, ErrorCodes::Overflow);
+    }
+}
+
+TEST(ExpressionFLETest, TestBinData_ContentionFactor) {
+    auto expCtx = ExpressionContextForTest();
+    auto vps = expCtx.variablesParseState;
+
+    // Use the wrong contention factor - 0
+    {
+        auto expr = fromjson(R"({$_internalFleEq: {
+        field: {
+            "$binary": {
+                "base64":
+                "BxI0VngSNJh2EjQSNFZ4kBIQ5+Wa5+SZafJeRUDGdLNx+i2ADDkyV2qA90Xcve7FqltoDm1PllSSgUS4fYtw3XDjzoNZrFFg8LfG2wH0HYbLMswv681KJpmEw7+RXy4CcPVFgoRFt24N13p7jT+pqu2oQAHAoxYTy/TsiAyY4RnAMiXYGg3hWz4AO/WxHNSyq6B6kX5d7x/hrXvppsZDc2Pmhd+c5xmovlv5RPj7wnNld13kYcMluztjNswiCH05hM/kp2/P7kw30iVnbz0SZxn1FjjCug==",
+                    "subType": "6"
+            }
+        },
+        server: {
+            "$binary": {
+                "base64": "COuac/eRLYakKX6B0vZ1r3QodOQFfjqJD+xlGiPu4/Ps",
+                "subType": "6"
+            }
+        },
+        counter: {
+            "$numberLong": "0"
+        },
+        edc: {
+            "$binary": {
+                "base64": "CEWSmQID7SfwyAUI3ZkSFkATKryDQfnxXEOGad5d4Rsg",
+                "subType": "6"
+            }
+        }    } })");
+        auto exprFle = ExpressionInternalFLEEqual::parse(&expCtx, expr.firstElement(), vps);
+
+        ASSERT_VALUE_EQ(exprFle->evaluate({}, &expCtx.variables), Value(false));
+    }
+
+    // Use the right contention factor - 50
+    {
+        auto expr = fromjson(R"({$_internalFleEq: {
+        field: {
+            "$binary": {
+                "base64":
+"BxI0VngSNJh2EjQSNFZ4kBIQ5+Wa5+SZafJeRUDGdLNx+i2ADDkyV2qA90Xcve7FqltoDm1PllSSgUS4fYtw3XDjzoNZrFFg8LfG2wH0HYbLMswv681KJpmEw7+RXy4CcPVFgoRFt24N13p7jT+pqu2oQAHAoxYTy/TsiAyY4RnAMiXYGg3hWz4AO/WxHNSyq6B6kX5d7x/hrXvppsZDc2Pmhd+c5xmovlv5RPj7wnNld13kYcMluztjNswiCH05hM/kp2/P7kw30iVnbz0SZxn1FjjCug==",
+                    "subType": "6"
+            }
+        },
+        server: {
+            "$binary": {
+                "base64": "COuac/eRLYakKX6B0vZ1r3QodOQFfjqJD+xlGiPu4/Ps",
+                "subType": "6"
+            }
+        },
+        counter: {
+            "$numberLong": "50"
+        },
+        edc: {
+            "$binary": {
+                "base64": "CEWSmQID7SfwyAUI3ZkSFkATKryDQfnxXEOGad5d4Rsg",
+                "subType": "6"
+            }
+        }    } })");
+        auto exprFle = ExpressionInternalFLEEqual::parse(&expCtx, expr.firstElement(), vps);
+
+        ASSERT_VALUE_EQ(exprFle->evaluate({}, &expCtx.variables), Value(true));
+    }
+}
+
+TEST(ExpressionFLETest, TestBinData_RoundTrip) {
+    auto expCtx = ExpressionContextForTest();
+    auto vps = expCtx.variablesParseState;
+
+    auto expr = fromjson(R"({$_internalFleEq: {
+    field: {
+        "$binary": {
+            "base64":
+            "BxI0VngSNJh2EjQSNFZ4kBIQ0JE8aMUFkPk5sSTVqfdNNfjqUfQQ1Uoj0BBcthrWoe9wyU3cN6zmWaQBPJ97t0ZPbecnMsU736yXre6cBO4Zdt/wThtY+v5+7vFgNnWpgRP0e+vam6QPmLvbBrO0LdsvAPTGW4yqwnzCIXCoEg7QPGfbfAXKPDTNenBfRlawiblmTOhO/6ljKotWsMp22q/rpHrn9IEIeJmecwuuPIJ7EA+XYQ3hOKVccYf2ogoK73+8xD/Vul83Qvr84Q8afc4QUMVs8A==",
+                "subType": "6"
+        }
+    },
+    server: {
+        "$binary": {
+            "base64": "COuac/eRLYakKX6B0vZ1r3QodOQFfjqJD+xlGiPu4/Ps",
+            "subType": "6"
+        }
+    },
+    counter: {
+        "$numberLong": "3"
+    },
+    edc: {
+        "$binary": {
+            "base64": "CEWSmQID7SfwyAUI3ZkSFkATKryDQfnxXEOGad5d4Rsg",
+            "subType": "6"
+        }
+    }    } })");
+    auto exprFle = ExpressionInternalFLEEqual::parse(&expCtx, expr.firstElement(), vps);
+
+    ASSERT_VALUE_EQ(exprFle->evaluate({}, &expCtx.variables), Value(true));
+
+    // Verify it round trips
+    auto value = exprFle->serialize(false);
+
+    auto roundTripExpr = fromjson(R"({$_internalFleEq: {
+    field: {
+        "$const" : { "$binary": {
+            "base64":
+            "BxI0VngSNJh2EjQSNFZ4kBIQ0JE8aMUFkPk5sSTVqfdNNfjqUfQQ1Uoj0BBcthrWoe9wyU3cN6zmWaQBPJ97t0ZPbecnMsU736yXre6cBO4Zdt/wThtY+v5+7vFgNnWpgRP0e+vam6QPmLvbBrO0LdsvAPTGW4yqwnzCIXCoEg7QPGfbfAXKPDTNenBfRlawiblmTOhO/6ljKotWsMp22q/rpHrn9IEIeJmecwuuPIJ7EA+XYQ3hOKVccYf2ogoK73+8xD/Vul83Qvr84Q8afc4QUMVs8A==",
+                "subType": "6"
+        }}
+    },
+    edc: {
+        "$binary": {
+            "base64": "CEWSmQID7SfwyAUI3ZkSFkATKryDQfnxXEOGad5d4Rsg",
+            "subType": "6"
+        }
+    },
+    counter: {
+        "$numberLong": "3"
+    },
+    server: {
+        "$binary": {
+            "base64": "COuac/eRLYakKX6B0vZ1r3QodOQFfjqJD+xlGiPu4/Ps",
+            "subType": "6"
+        }
+    }
+        } })");
+
+
+    ASSERT_BSONOBJ_EQ(value.getDocument().toBson(), roundTripExpr);
+}
+
 }  // namespace ExpressionTests
diff --git a/src/mongo/db/pipeline/expression_visitor.h b/src/mongo/db/pipeline/expression_visitor.h
index 46ad3ee6295..6b7c4fc4cdd 100644
--- a/src/mongo/db/pipeline/expression_visitor.h
+++ b/src/mongo/db/pipeline/expression_visitor.h
@@ -153,6 +153,7 @@ class ExpressionHyperbolicSine;
 class ExpressionInternalFindSlice;
 class ExpressionInternalFindPositional;
 class ExpressionInternalFindElemMatch;
+class ExpressionInternalFLEEqual;
 class ExpressionInternalJsEmit;
 class ExpressionFunction;
 class ExpressionDegreesToRadians;
@@ -245,6 +246,7 @@ public:
     virtual void visit(expression_walker::MaybeConstPtr<IsConst, ExpressionLn>) = 0;
     virtual void visit(expression_walker::MaybeConstPtr<IsConst, ExpressionLog>) = 0;
     virtual void visit(expression_walker::MaybeConstPtr<IsConst, ExpressionLog10>) = 0;
+    virtual void visit(expression_walker::MaybeConstPtr<IsConst, ExpressionInternalFLEEqual>) = 0;
     virtual void visit(expression_walker::MaybeConstPtr<IsConst, ExpressionMap>) = 0;
     virtual void visit(expression_walker::MaybeConstPtr<IsConst, ExpressionMeta>) = 0;
     virtual void visit(expression_walker::MaybeConstPtr<IsConst, ExpressionMod>) = 0;
@@ -424,6 +426,7 @@ struct SelectiveConstExpressionVisitorBase : public ExpressionConstVisitor {
     void visit(const ExpressionLn*) override {}
     void visit(const ExpressionLog*) override {}
     void visit(const ExpressionLog10*) override {}
+    void visit(const ExpressionInternalFLEEqual*) override {}
     void visit(const ExpressionMap*) override {}
     void visit(const ExpressionMeta*) override {}
     void visit(const ExpressionMod*) override {}
diff --git a/src/mongo/db/pipeline/pipeline_d.cpp b/src/mongo/db/pipeline/pipeline_d.cpp
index 85c3e82839a..b21e8635c51 100644
--- a/src/mongo/db/pipeline/pipeline_d.cpp
+++ b/src/mongo/db/pipeline/pipeline_d.cpp
@@ -1451,7 +1451,9 @@ PipelineD::buildInnerQueryExecutorGeneric(const MultipleCollectionAccessor& coll
                                     // This produces {$const: maxBucketSpanSeconds}
                                     make_intrusive<ExpressionConstant>(
                                         expCtx.get(),
-                                        Value{unpack->getBucketMaxSpanSeconds() * 1000}))),
+                                        Value{static_cast<long long>(
+                                                  unpack->getBucketMaxSpanSeconds()) *
+                                              1000}))),
                             expCtx);
                         pipeline->_sources.insert(
                             unpackIter,