SERVER-57334 Create desugaring namespace for DocumentSourceDensify

12 files changed, 1370 insertions, 386 deletions

diff --git a/jstests/aggregation/sources/densify/desugar.js b/jstests/aggregation/sources/densify/desugar.js
new file mode 100644
index 00000000000..b1f7c11154d
--- /dev/null
+++ b/jstests/aggregation/sources/densify/desugar.js
@@ -0,0 +1,140 @@
+/**
+ * Test how $densify desugars.
+ *
+ * @tags: [
+ *   # Needed as $densify is a 51 feature.
+ *   requires_fcv_51,
+ *   # We're testing the explain plan, not the query results, so the facet passthrough would fail.
+ *   do_not_wrap_aggregations_in_facets,
+ * ]
+ */
+(function() {
+"use strict";
+
+const featureEnabled =
+    assert.commandWorked(db.adminCommand({getParameter: 1, featureFlagDensify: 1}))
+        .featureFlagDensify.value;
+if (!featureEnabled) {
+    jsTestLog("Skipping test because the densify feature flag is disabled");
+    return;
+}
+
+load("jstests/libs/fixture_helpers.js");
+
+const coll = db[jsTestName()];
+coll.insert({});
+
+// Use .explain() to see what the stage desugars to.
+// The result is formatted as explain-output, which differs from MQL syntax in some cases:
+// for example {$sort: {a: 1}} explains as {$sort: {sortKey: {a: 1}}}.
+function desugar(stage) {
+    const result = coll.explain().aggregate([
+        // prevent stages from being absorbed into the .find() layer
+        {$_internalInhibitOptimization: {}},
+        stage,
+    ]);
+
+    assert.commandWorked(result);
+    // We proceed by cases based on topology.
+    if (!FixtureHelpers.isMongos(db)) {
+        assert(Array.isArray(result.stages), result);
+        // The first two stages should be the .find() cursor and the inhibit-optimization stage;
+        // the rest of the stages are what the user's 'stage' expanded to.
+        assert(result.stages[0].$cursor, result);
+        assert(result.stages[1].$_internalInhibitOptimization, result);
+        return result.stages.slice(2);
+    } else {
+        if (result.splitPipeline) {
+            assert(result.splitPipeline.shardsPart[0].$_internalInhibitOptimization, result);
+            assert.eq(result.splitPipeline.shardsPart.length, 2);
+            assert.eq(result.splitPipeline.mergerPart.length, 1);
+            return [result.splitPipeline.shardsPart[1], result.splitPipeline.mergerPart[0]];
+        } else if (result.stages) {
+            // Required for aggregation_mongos_passthrough.
+            assert(Array.isArray(result.stages), result);
+            // The first two stages should be the .find() cursor and the inhibit-optimization stage;
+            // the rest of the stages are what the user's 'stage' expanded to.
+            assert(result.stages[0].$cursor, result);
+            assert(result.stages[1].$_internalInhibitOptimization, result);
+            return result.stages.slice(2);
+        } else {
+            // Required for aggregation_one_shard_sharded_collections.
+            assert(Array.isArray(result.shards["shard-rs0"].stages), result);
+            assert(result.shards["shard-rs0"].stages[0].$cursor, result);
+            assert(result.shards["shard-rs0"].stages[1].$_internalInhibitOptimization, result);
+            return result.shards["shard-rs0"].stages.slice(2);
+        }
+    }
+}
+
+// Implicit partition fields and sort are generated.
+assert.eq(desugar({$densify: {field: "a", range: {step: 1.0, bounds: "full"}}}), [
+    {$sort: {sortKey: {a: 1}}},
+    {$_internalDensify: {field: "a", partitionByFields: [], range: {step: 1.0, bounds: "full"}}},
+]);
+
+// PartitionByFields are prepended to the sortKey if "partition" is specified.
+assert.eq(
+    desugar({
+        $densify:
+            {field: "a", partitionByFields: ["b", "c"], range: {step: 1.0, bounds: "partition"}}
+    }),
+    [
+        {$sort: {sortKey: {b: 1, c: 1, a: 1}}},
+        {
+            $_internalDensify:
+                {field: "a", partitionByFields: ["b", "c"], range: {step: 1.0, bounds: "partition"}}
+        },
+    ]);
+
+// PartitionByFields are not prepended to the sortKey if "full" is specified.
+assert.eq(
+    desugar({
+        $densify: {field: "a", partitionByFields: ["b", "c"], range: {step: 1.0, bounds: "full"}}
+    }),
+    [
+        {$sort: {sortKey: {a: 1}}},
+        {
+            $_internalDensify:
+                {field: "a", partitionByFields: ["b", "c"], range: {step: 1.0, bounds: "full"}}
+        },
+    ]);
+
+// PartitionByFields are prepended to the sortKey if numeric bounds are specified.
+assert.eq(
+    desugar({
+        $densify: {field: "a", partitionByFields: ["b", "c"], range: {step: 1.0, bounds: [-10, 0]}}
+    }),
+    [
+        {$sort: {sortKey: {b: 1, c: 1, a: 1}}},
+        {
+            $_internalDensify:
+                {field: "a", partitionByFields: ["b", "c"], range: {step: 1.0, bounds: [-10, 0]}}
+        },
+    ]);
+
+// PartitionByFields are prepended to the sortKey if date bounds are specified.
+assert.eq(
+    desugar({
+        $densify: {
+            field: "a",
+            partitionByFields: ["b", "c"],
+            range:
+                {step: 1.0, bounds: [new Date("2020-01-03"), new Date("2020-01-04")], unit: "day"}
+        }
+    }),
+    [
+        {$sort: {sortKey: {b: 1, c: 1, a: 1}}},
+        {
+            $_internalDensify: {
+                field: "a",
+                partitionByFields: ["b", "c"],
+                range: {
+                    step: 1.0,
+                    bounds: [new Date("2020-01-03"), new Date("2020-01-04")],
+                    unit: "day"
+                }
+            }
+        },
+    ]);
+})();
diff --git a/jstests/aggregation/sources/densify/internal_parse.js b/jstests/aggregation/sources/densify/internal_parse.js
new file mode 100644
index 00000000000..d837c765af4
--- /dev/null
+++ b/jstests/aggregation/sources/densify/internal_parse.js
@@ -0,0 +1,39 @@
+/**
+ * Test the syntax of $_internalDensify.
+ * @tags: [
+ *     # Needed as $densify is a 51 feature.
+ *     requires_fcv_51,
+ *     # Our method of connecting via an internal client requries an unsharded topology.
+ *     assumes_unsharded_collection,
+ *     assumes_against_mongod_not_mongos,
+ *     assumes_read_preference_unchanged,
+ * ]
+ */
+
+(function() {
+"use strict";
+
+load("jstests/aggregation/sources/densify/libs/parse_util.js");
+
+const dbName = jsTestName();
+const testDB = db.getSiblingDB(dbName);
+testDB.dropDatabase();
+const collName = jsTestName();
+
+const testInternalClient = (function createInternalClient() {
+    const connInternal = new Mongo(testDB.getMongo().host);
+    const curDB = connInternal.getDB(dbName);
+    assert.commandWorked(curDB.runCommand({
+        ["hello"]: 1,
+        internalClient: {minWireVersion: NumberInt(0), maxWireVersion: NumberInt(7)}
+    }));
+    return connInternal;
+})();
+
+const internalDB = testInternalClient.getDB(dbName);
+const internalColl = internalDB[collName];
+
+parseUtil(internalDB, internalColl, "$_internalDensify", {
+    writeConcern: {w: "majority"},
+});
+})();
diff --git a/jstests/aggregation/sources/densify/libs/parse_util.js b/jstests/aggregation/sources/densify/libs/parse_util.js
new file mode 100644
index 00000000000..9e51df53445
--- /dev/null
+++ b/jstests/aggregation/sources/densify/libs/parse_util.js
@@ -0,0 +1,210 @@
+/**
+ * Utility for testing the parsing of densify-like commands. I.e. $_internalDensify and $densify.
+ */
+
+let parseUtil = (function(db, coll, stageName, options = {}) {
+    function run(stage, extraCommandArgs = options) {
+        return coll.runCommand(Object.merge(
+            {aggregate: coll.getName(), pipeline: [stage], cursor: {}}, extraCommandArgs));
+    }
+
+    function runTest(stageName) {
+        const featureEnabled =
+            assert.commandWorked(db.adminCommand({getParameter: 1, featureFlagDensify: 1}))
+                .featureFlagDensify.value;
+        if (!featureEnabled) {
+            jsTestLog("Skipping test because the densify feature flag is disabled");
+            return;
+        }
+
+        // Required fields.
+        const kIDLRequiredFieldErrorCode = 40414;
+        assert.commandFailedWithCode(
+            run({[stageName]: {field: "a", range: {step: 1.0}}}),
+            kIDLRequiredFieldErrorCode,
+            "BSON field '$densify.range.bounds' is missing but a required field");
+        assert.commandFailedWithCode(run({[stageName]: {range: {step: 1.0, bounds: "full"}}}),
+                                     kIDLRequiredFieldErrorCode,
+                                     "BSON field '$densify.field' is missing but a required field");
+        assert.commandFailedWithCode(
+            run({[stageName]: {field: "a", range: {bounds: "full"}}}),
+            kIDLRequiredFieldErrorCode,
+            "BSON field '$densify.range.step' is missing but a required field");
+
+        // Wrong types
+        assert.commandFailedWithCode(
+            run({[stageName]: {field: 1.0, range: {step: 1.0, bounds: "full"}}}),
+            ErrorCodes.TypeMismatch,
+            "BSON field '$densify.field' is the wrong type 'double', expected type 'string'");
+        assert.commandFailedWithCode(
+            run({[stageName]: {field: "a", range: {step: "invalid", bounds: "full"}}}),
+            ErrorCodes.TypeMismatch,
+            "BSON field '$densify.range.step' is the wrong type 'string', expected types '[int, decimal, double, long']");
+        assert.commandFailedWithCode(
+            run({
+                [stageName]:
+                    {field: "a", partitionByFields: "incorrect", range: {step: 1.0, bounds: "full"}}
+            }),
+            ErrorCodes.TypeMismatch,
+            "BSON field '$densify.partitionByFields' is the wrong type 'string', expected type 'array'");
+        assert.commandFailedWithCode(
+            run({
+                [stageName]: {
+                    field: "a",
+                    range: {
+                        step: 1.0,
+                        bounds: [new Date("2020-01-01"), new Date("2020-01-02")],
+                        unit: 1000
+                    }
+                }
+            }),
+            ErrorCodes.TypeMismatch,
+            "BSON field '$densify.range.unit' is the wrong type 'double', expected type 'string'");
+
+        // Logical errors
+        // Too few bounds
+        assert.commandFailedWithCode(
+            run({
+                [stageName]: {
+                    field: "a",
+                    range: {step: 1.0, bounds: [new Date("2020-01-01")], unit: "second"}
+                }
+            }),
+            5733403,
+            "a bounding array in a range statement must have exactly two elements");
+        // Too many elements
+        assert.commandFailedWithCode(
+            run({[stageName]: {field: "a", range: {step: 1.0, bounds: [0, 1, 2]}}}),
+            5733403,
+            "a bounding array in a range statement must have exactly two elements");
+        // Negative step
+        assert.commandFailedWithCode(
+            run({[stageName]: {field: "a", range: {step: -1.0, bounds: [0, 1]}}}),
+            5733401,
+            "the step parameter in a range statement must be a strictly positive numeric value");
+        // Field path expression instead of field path
+        assert.commandFailedWithCode(
+            run({[stageName]: {field: "$a", range: {step: 1.0, bounds: [0, 1]}}}),
+            16410,
+            "FieldPath field names may not start with '$'. Consider using $getField or $setField.");
+        // Field path expression instead of field path
+        assert.commandFailedWithCode(
+            run({
+                [stageName]:
+                    {field: "a", partitionByFields: ["$b"], range: {step: 1.0, bounds: [0, 1]}}
+            }),
+            16410,
+            "FieldPath field names may not start with '$'. Consider using $getField or $setField.");
+        // Partition bounds but not partitionByFields
+        assert.commandFailedWithCode(
+            run({
+                [stageName]:
+                    {field: "a", partitionByFields: [], range: {step: 1.0, bounds: "partition"}}
+            }),
+            5733408,
+            "one may not specify the bounds as 'partition' without specifying a non-empty array of partitionByFields. You may have meant to specify 'full' bounds.");
+        assert.commandFailedWithCode(
+            run({[stageName]: {field: "a", range: {step: 1.0, bounds: "partition"}}}),
+            5733408,
+            "one may not specify the bounds as 'partition' without specifying a non-empty array of partitionByFields. You may have meant to specify 'full' bounds.");
+        // Left bound greater than right bound
+        assert.commandFailedWithCode(
+            run({[stageName]: {field: "a", range: {step: 1.0, bounds: [2, 1]}}}),
+            5733402,
+            "the bounds in a range statement must be the string 'full', 'partition', or an ascending array of two numbers or two dates");
+        assert.commandFailedWithCode(
+            run({
+                [stageName]: {
+                    field: "a",
+                    range: {
+                        step: 1.0,
+                        bounds: [new Date("2020-01-01"), new Date("2019-01-01")],
+                        unit: "second"
+                    }
+                }
+            }),
+            5733402,
+            "the bounds in a range statement must be the string 'full', 'partition', or an ascending array of two numbers or two dates");
+        // Unit with numeric bounds
+        assert.commandFailedWithCode(
+            run({[stageName]: {field: "a", range: {step: 1.0, bounds: [1, 2], unit: "second"}}}),
+            5733409,
+            "numeric bounds may not have unit parameter");
+        // Mixed numeric and date bounds
+        assert.commandFailedWithCode(
+            run({
+                [stageName]: {field: "a", range: {step: 1.0, bounds: [1, new Date("2020-01-01")]}}
+            }),
+            5733406,
+            "a bounding array must contain either both dates or both numeric types");
+        assert.commandFailedWithCode(
+            run({
+                [stageName]: {
+                    field: "a",
+                    range: {step: 1.0, bounds: [new Date("2020-01-01"), 1], unit: "second"}
+                }
+            }),
+            5733402,
+            "a bounding array must be an ascending array of either two dates or two numbers");
+
+        // Positive test cases
+        assert.commandWorked(run({[stageName]: {field: "a", range: {step: 1.0, bounds: [1, 2]}}}));
+        // TODO SERVER-57337: Enable this parsing test.
+        /* assert.commandWorked(run({
+         *     [stageName]: {
+         *         field: "a",
+         *         range: {
+         *             step: 1.0,
+         *             bounds: [new Date("2020-01-01"), new Date("2021-01-01")],
+         *             unit: "second"
+         *         }
+         *     }
+         * })); */
+        // TODO SERVER-57337 SERVER-57344: Enable this parsing test.
+        /* assert.commandWorked(run({
+         *     [stageName]: {
+         *         field: "a",
+         *         partitionByFields: ["b", "c"],
+         *         range: {
+         *             step: 1.0,
+         *             bounds: [new Date("2020-01-01"), new Date("2021-01-01")],
+         *             unit: "second"
+         *         }
+         *     }
+         * })); */
+        // TODO SERVER-57337 SERVER-57344: Enable this parsing test.
+        /* assert.commandWorked(run({
+         *     [stageName]: {
+         *         field: "a",
+         *         partitionByFields: ["b", "c"],
+         *         range: {step: 1.0, bounds: "partition", unit: "second"}
+         *     }
+         * })); */
+        // TODO SERVER-57337 SERVER-57344: Enable this parsing test.
+        /* assert.commandWorked(run({
+         *     [stageName]: {
+         *         field: "a",
+         *         partitionByFields: ["b", "c"],
+         *         range: {step: 1.0, bounds: "full", unit: "second"}
+         *     }
+         * })); */
+        // TODO SERVER-57344: Enable this parsing test.
+        /* assert.commandWorked(run({
+         *     [stageName]:
+         *         {field: "a", partitionByFields: ["b", "c"], range: {step: 1.0, bounds: "full"}}
+         * })); */
+        // TODO SERVER-57344: Enable this parsing test.
+        /* assert.commandWorked(run({
+            [stageName]: {
+                field: "a",
+                partitionByFields: [
+                    "b",
+                ],
+                range: { step: 1.0, bounds: "partition" }
+            }
+           })); */
+        assert.commandWorked(run({[stageName]: {field: "a", range: {step: 1.0, bounds: "full"}}}));
+    }
+
+    runTest(stageName);
+});
diff --git a/jstests/aggregation/sources/densify/parse.js b/jstests/aggregation/sources/densify/parse.js
new file mode 100644
index 00000000000..58154ed47c7
--- /dev/null
+++ b/jstests/aggregation/sources/densify/parse.js
@@ -0,0 +1,18 @@
+/**
+ * Test the syntax of $densify.
+ * @tags: [
+ *   # Needed as $densify is a 51 feature.
+ *   requires_fcv_51,
+ * ]
+ */
+
+(function() {
+"use strict";
+
+load("jstests/aggregation/sources/densify/libs/parse_util.js");
+
+const coll = db.densify_parse;
+coll.drop();
+
+parseUtil(db, coll, "$densify");
+})();
diff --git a/src/mongo/db/pipeline/SConscript b/src/mongo/db/pipeline/SConscript
index 351a062dca1..463b05d0f35 100644
--- a/src/mongo/db/pipeline/SConscript
+++ b/src/mongo/db/pipeline/SConscript
@@ -391,6 +391,7 @@ env.Library(
     source=[
         'document_source_change_stream.idl',
         'document_source_coll_stats.idl',
+        'document_source_densify.idl',
         'document_source_list_sessions.idl',
         'document_source_merge.idl',
         'document_source_merge_modes.idl',
diff --git a/src/mongo/db/pipeline/document_source_densify.cpp b/src/mongo/db/pipeline/document_source_densify.cpp
index 003b5718d1b..d48c3bbfbf5 100644
--- a/src/mongo/db/pipeline/document_source_densify.cpp
+++ b/src/mongo/db/pipeline/document_source_densify.cpp
@@ -28,36 +28,212 @@
  */
 
 #include "mongo/db/pipeline/document_source_densify.h"
+#include "mongo/db/pipeline/document_source_sort.h"
 #include "mongo/db/pipeline/field_path.h"
 #include "mongo/stdx/variant.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/visit_helper.h"
 
+using boost::intrusive_ptr;
+using boost::optional;
+using std::list;
+using SortPatternPart = mongo::SortPattern::SortPatternPart;
+using NumericBounds = mongo::RangeStatement::NumericBounds;
+using DateBounds = mongo::RangeStatement::DateBounds;
+using Full = mongo::RangeStatement::Full;
+using Partition = mongo::RangeStatement::Partition;
+
 namespace mongo {
+
+RangeStatement RangeStatement::parse(RangeSpec spec) {
+    Value step = spec.getStep();
+    ValueComparator comp = ValueComparator();
+    uassert(5733401,
+            "The step parameter in a range statement must be a strictly positive numeric value",
+            step.numeric() && comp.evaluate(step > Value(0)));
+
+    optional<TimeUnit> unit = [&]() {
+        if (auto unit = spec.getUnit()) {
+            return optional<TimeUnit>(parseTimeUnit(unit.get()));
+        } else {
+            return optional<TimeUnit>(boost::none);
+        }
+    }();
+
+    Bounds bounds = [&]() {
+        BSONElement bounds = spec.getBounds().getElement();
+        switch (bounds.type()) {
+            case mongo::Array: {
+                std::vector<BSONElement> array = bounds.Array();
+
+                uassert(5733403,
+                        "A bounding array in a range statement must have exactly two elements",
+                        array.size() == 2);
+                uassert(5733402,
+                        "A bounding array must be an ascending array of either two dates or two "
+                        "numbers",
+                        comp.evaluate(Value(array[0]) <= Value(array[1])));
+                if (array[0].isNumber()) {
+                    uassert(5733409, "Numeric bounds may not have unit parameter", !unit);
+                    uassert(5733406,
+                            "A bounding array must contain either both dates or both numeric types",
+                            array[1].isNumber());
+                    return Bounds(std::pair<Value, Value>(Value(array[0]), Value(array[1])));
+                } else if (array[0].type() == mongo::Date) {
+                    uassert(5733405,
+                            "A bounding array must contain either both dates or both numeric types",
+                            array[1].type() == mongo::Date);
+                    uassert(5733410, "A bounding array of dates must specify a unit", unit);
+                    return Bounds(std::pair<Date_t, Date_t>(array[0].date(), array[1].date()));
+                }
+            }
+            case mongo::String: {
+                if (bounds.str() == kValFull)
+                    return Bounds(Full());
+                else if (bounds.str() == kValPartition)
+                    return Bounds(Partition());
+            }
+            default:
+                uasserted(5733404,
+                          "The bounds in a range statement must be the string \'full\', "
+                          "\'partition\', or an ascending array of two numbers or two dates");
+        }
+    }();
+
+    RangeStatement range = RangeStatement(step, bounds, unit);
+    return range;
+}
+
+REGISTER_DOCUMENT_SOURCE_CONDITIONALLY(
+    densify,
+    LiteParsedDocumentSourceDefault::parse,
+    document_source_densify::createFromBson,
+    AllowedWithApiStrict::kNeverInVersion1,
+    AllowedWithClientType::kAny,
+    ServerGlobalParams::FeatureCompatibility::Version::kVersion51,
+    ::mongo::feature_flags::gFeatureFlagDensify.isEnabledAndIgnoreFCV());
+
+REGISTER_DOCUMENT_SOURCE_CONDITIONALLY(
+    _internalDensify,
+    LiteParsedDocumentSourceDefault::parse,
+    DocumentSourceInternalDensify::createFromBson,
+    AllowedWithApiStrict::kInternal,
+    AllowedWithClientType::kInternal,
+    ServerGlobalParams::FeatureCompatibility::Version::kVersion51,
+    ::mongo::feature_flags::gFeatureFlagDensify.isEnabledAndIgnoreFCV());
+
 namespace document_source_densify {
-// TODO SERVER-57334 Translation logic goes here.
+
+list<intrusive_ptr<DocumentSource>> createFromBsonInternal(
+    BSONElement elem,
+    const intrusive_ptr<ExpressionContext>& expCtx,
+    StringData stageName,
+    bool isInternal) {
+    uassert(ErrorCodes::FailedToParse,
+            str::stream() << "The " << stageName << " stage specification must be an object, found "
+                          << typeName(elem.type()),
+            elem.type() == BSONType::Object);
+
+    auto spec = DensifySpec::parse(IDLParserErrorContext(stageName), elem.embeddedObject());
+    auto rangeStatement = RangeStatement::parse(spec.getRange());
+
+    list<FieldPath> partitions;
+    if (spec.getPartitionByFields()) {
+        auto partitionFields = (*spec.getPartitionByFields());
+        for (auto partitionField : partitionFields)
+            partitions.push_back(FieldPath(partitionField));
+    }
+
+    FieldPath field = FieldPath(spec.getField());
+
+    if (stdx::holds_alternative<RangeStatement::Partition>(rangeStatement.getBounds()) &&
+        partitions.empty())
+        uasserted(5733408,
+                  "One cannot specify the bounds as 'partition' without specifying a non-empty "
+                  "array of partitionByFields. You may have meant to specify 'full' bounds.");
+
+    return create(std::move(expCtx),
+                  std::move(partitions),
+                  std::move(field),
+                  std::move(rangeStatement),
+                  isInternal);
+}
+
+list<intrusive_ptr<DocumentSource>> createFromBson(BSONElement elem,
+                                                   const intrusive_ptr<ExpressionContext>& expCtx) {
+    return createFromBsonInternal(elem, expCtx, kStageName, false);
+}
+
+list<intrusive_ptr<DocumentSource>> create(const intrusive_ptr<ExpressionContext>& expCtx,
+                                           list<FieldPath> partitions,
+                                           FieldPath field,
+                                           RangeStatement rangeStatement,
+                                           bool isInternal) {
+    list<intrusive_ptr<DocumentSource>> results;
+
+    // If we're creating an internal stage then we must not desugar and produce a sort stage in
+    // addition.
+    if (!isInternal) {
+        // Add partition fields to sort spec.
+        std::vector<SortPatternPart> sortParts;
+        // We do not add partitions to the sort spec if the range is "full".
+        if (!stdx::holds_alternative<Full>(rangeStatement.getBounds())) {
+            for (auto partition : partitions) {
+                SortPatternPart part;
+                part.fieldPath = partition.fullPath();
+                sortParts.push_back(std::move(part));
+            }
+        }
+
+        // Add field path to sort spec.
+        SortPatternPart part;
+        part.fieldPath = field.fullPath();
+        sortParts.push_back(std::move(part));
+
+        // Constructing resulting stages.
+        results.push_back(DocumentSourceSort::create(expCtx, SortPattern{sortParts}));
+    }
+
+    // Constructing resulting stages.
+    results.push_back(
+        make_intrusive<DocumentSourceInternalDensify>(expCtx, field, partitions, rangeStatement));
+
+    return results;
+}
 }  // namespace document_source_densify
 
 namespace {
-double getDensifyDouble(const Document& doc, const FieldPath& path) {
+Value getDensifyValue(const Document& doc, const FieldPath& path) {
     Value val = doc.getNestedField(path);
-    uassert(5733201, "Densify field type must be double", val.numeric());
-    return val.getDouble();
+    uassert(5733201, "Densify field type must be numeric", val.numeric());
+    return val;
+}
+
+Value addValues(Value lhs, Value rhs) {
+    return uassertStatusOK(ExpressionAdd::apply(lhs, rhs));
+}
+
+Value subtractValues(Value lhs, Value rhs) {
+    return uassertStatusOK(ExpressionSubtract::apply(lhs, rhs));
+}
+
+Value floorValue(Value operand) {
+    return uassertStatusOK(ExpressionFloor::apply(operand));
 }
 }  // namespace
 
 DocumentSourceInternalDensify::DocGenerator::DocGenerator(
     DocumentSourceInternalDensify::DensifyValueType min,
-    DocumentSourceInternalDensify::DensifyValueType max,
-    StepSpec step,
+    RangeStatement range,
     FieldPath fieldName,
     boost::optional<Document> includeFields,
-    boost::optional<Document> finalDoc)
-    : _step(std::move(step)),
+    boost::optional<Document> finalDoc,
+    ValueComparator comp)
+    : _comp(std::move(comp)),
+      _range(std::move(range)),
       _path(std::move(fieldName.fullPath())),
       _finalDoc(std::move(finalDoc)),
-      _min(std::move(min)),
-      _max(std::move(max)) {
+      _min(std::move(min)) {
 
     if (includeFields) {
         _includeFields = *includeFields;
@@ -85,32 +261,34 @@ DocumentSourceInternalDensify::DocGenerator::DocGenerator(
         }
     }
 
-    tassert(5733305, "DocGenerator step must be positive", _step.step > 0);
+    tassert(
+        5733305, "DocGenerator step must be positive", _comp.evaluate(_range.getStep() > Value(0)));
     stdx::visit(
         visit_helper::Overloaded{
-            [&](const double doubleMin) {
+            [&](const Value val) {
                 tassert(5733304,
                         "DocGenerator all values must be same type",
-                        stdx::holds_alternative<double>(_max));
+                        stdx::holds_alternative<NumericBounds>(_range.getBounds()));
+                NumericBounds bounds = stdx::get<NumericBounds>(_range.getBounds());
                 tassert(5733303,
                         "DocGenerator min must be lower or equal to max",
-                        stdx::get<double>(_max) >= doubleMin);
-                tassert(5733506,
-                        "Unit and tz must not be specified with non-date values",
-                        !_step.unit && !_step.tz);
+                        _comp.evaluate(bounds.second >= val));
+                tassert(
+                    5733506, "Unit must not be specified with non-date values", !_range.getUnit());
             },
             [&](const Date_t dateMin) {
                 tassert(5733500,
                         "DocGenerator all values must be same type",
-                        stdx::holds_alternative<Date_t>(_max));
-                tassert(5733501, "Unit must be specified with a date step", _step.unit);
+                        stdx::holds_alternative<DateBounds>(_range.getBounds()));
+                DateBounds bounds = stdx::get<DateBounds>(_range.getBounds());
+                tassert(5733501, "Unit must be specified with a date step", _range.getUnit());
+                Value floorStep = floorValue(_range.getStep());
                 tassert(5733505,
                         "Step must be an integer for date densification",
-                        floor(_step.step) == _step.step);
+                        _comp.evaluate(floorStep == _range.getStep()));
                 tassert(5733502,
                         "DocGenerator min must be lower or equal to max",
-                        stdx::get<Date_t>(_max) >= dateMin);
-                tassert(5733504, "DocGenerator with dates requires a time zone", _step.tz);
+                        bounds.second >= dateMin);
             },
         },
         _min);
@@ -130,19 +308,24 @@ Document DocumentSourceInternalDensify::DocGenerator::getNextDocument() {
     Value valueToAdd;
     stdx::visit(
         visit_helper::Overloaded{
-            [&](double doubleVal) {
-                valueToAdd = Value(doubleVal);
-                doubleVal += _step.step;
-                if (doubleVal > stdx::get<double>(_max)) {
+            [&](Value val) {
+                valueToAdd = val;
+                Value nextValue = addValues(val, _range.getStep());
+                NumericBounds bounds = stdx::get<NumericBounds>(_range.getBounds());
+                if (_comp.evaluate(nextValue > bounds.second)) {
                     _state =
                         _finalDoc ? GeneratorState::kReturningFinalDocument : GeneratorState::kDone;
                 }
-                _min = doubleVal;
+                _min = nextValue;
             },
             [&](Date_t dateVal) {
                 valueToAdd = Value(dateVal);
-                dateVal = dateAdd(dateVal, _step.unit.get(), _step.step, _step.tz.get());
-                if (dateVal > stdx::get<Date_t>(_max)) {
+                dateVal = dateAdd(dateVal,
+                                  _range.getUnit().get(),
+                                  _range.getStep().getDouble(),
+                                  TimeZoneDatabase::utcZone());
+                DateBounds bounds = stdx::get<DateBounds>(_range.getBounds());
+                if (dateVal > bounds.second) {
                     _state =
                         _finalDoc ? GeneratorState::kReturningFinalDocument : GeneratorState::kDone;
                 }
@@ -160,71 +343,64 @@ bool DocumentSourceInternalDensify::DocGenerator::done() const {
     return _state == GeneratorState::kDone;
 }
 
-DocumentSourceInternalDensify::DocumentSourceInternalDensify(
-    const boost::intrusive_ptr<ExpressionContext>& pExpCtx,
-    double step,
-    FieldPath path,
-    boost::optional<std::pair<DensifyValueType, DensifyValueType>> range)
-    : DocumentSource(kStageName, pExpCtx), _step{step}, _path(std::move(path)) {
-    if (range) {
-        _rangeMin = range->first;
-        _rangeMax = range->second;
-        _densifyType = TypeOfDensify::kExplicitRange;
-    } else {
-        _densifyType = TypeOfDensify::kFull;
-    }
-}
-
-DocumentSource::GetNextResult DocumentSourceInternalDensify::densifyAfterEOF() {
-    // Once we have hit an EOF, if the last seen value (_rangeMin) plus the step is greater
+// TODO SERVER-57344: Execution flow should be refactored such that std::visits are done in these
+// functions instead of the doGetNext(). This is to avoid the need to pass NumericBounds to avoid
+// duplicate std::visits in functions like this.
+DocumentSource::GetNextResult DocumentSourceInternalDensify::densifyAfterEOF(NumericBounds bounds) {
+    // Once we have hit an EOF, if the last seen value (_current) plus the step is greater
     // than or equal to the rangeMax, that means we have finished densifying
     // over the explicit range so we just return an EOF. Otherwise, we finish
     // densifying over the rest of the range.
-    if (stdx::get<double>(*_rangeMin) + _step.step >= stdx::get<double>(*_rangeMax)) {
+    if (compareValues(addValues(stdx::get<Value>(*_current), _range.getStep()) >= bounds.second)) {
         _densifyState = DensifyState::kDensifyDone;
         return DocumentSource::GetNextResult::makeEOF();
     } else {
-        _docGenerator = DocGenerator(stdx::get<double>(*_rangeMin) + _step.step,
-                                     *_rangeMax,
-                                     _step,
-                                     _path,
+        _docGenerator = DocGenerator(addValues(stdx::get<Value>(*_current), _range.getStep()),
+                                     _range,
+                                     _field,
+                                     boost::none,
                                      boost::none,
-                                     boost::none);
+                                     pExpCtx->getValueComparator());
         _densifyState = DensifyState::kHaveGenerator;
         return _docGenerator->getNextDocument();
     }
 }
 
-DocumentSource::GetNextResult DocumentSourceInternalDensify::processDocAboveMinBound(double val,
-                                                                                     Document doc) {
+DocumentSource::GetNextResult DocumentSourceInternalDensify::processDocAboveMinBound(
+    Value val, NumericBounds bounds, Document doc) {
     // If we are above the range, there must be more left to densify.
     // Otherwise the state would be kDoneDensify and this function would not be reached.
     tassert(8423306,
-            "Cannot be in this state if _rangeMin is greater than _rangeMax.",
-            *_rangeMin <= *_rangeMax);
-    auto rem = valOffsetFromStep(val, stdx::get<double>(*_rangeMin), _step.step);
+            "Cannot be in this state if _current is greater than the upper bound.",
+            compareValues(stdx::get<Value>(*_current) <= bounds.second));
+    auto rem = valOffsetFromStep(val, stdx::get<Value>(*_current), _range.getStep());
     // If val is on the step we need to subtract the step to avoid returning the doc twice.
-    if (rem == 0) {
-        val = val - _step.step;
+    if (compareValues(rem == Value(0))) {
+        val = subtractValues(val, _range.getStep());
     }
-    _docGenerator = DocGenerator(*_rangeMin,
-                                 std::min(val, stdx::get<double>(*_rangeMax)),
-                                 _step,
-                                 _path,
-                                 boost::none,
-                                 std::move(doc));
+    Value upperBound = (compareValues(val <= bounds.second)) ? val : bounds.second;
+
+    _docGenerator =
+        DocGenerator(*_current,
+                     RangeStatement(_range.getStep(),
+                                    NumericBounds(stdx::get<Value>(*_current), upperBound),
+                                    _range.getUnit()),
+                     _field,
+                     boost::none,
+                     std::move(doc),
+                     pExpCtx->getValueComparator());
     Document nextFromGen = _docGenerator->getNextDocument();
-    _rangeMin = getDensifyDouble(nextFromGen, _path);
+    _current = getDensifyValue(nextFromGen, _field);
     _densifyState = DensifyState::kHaveGenerator;
     // If the doc generator is done it will be deleted and the state will be kNeedGen.
-    resetDocGen();
+    resetDocGen(bounds);
     return nextFromGen;
 }
 
 /** Checks if the generator is done, changes states accordingly. */
-void DocumentSourceInternalDensify::resetDocGen() {
+void DocumentSourceInternalDensify::resetDocGen(NumericBounds bounds) {
     if (_docGenerator->done()) {
-        if (_rangeMin >= _rangeMax) {
+        if (compareValues(stdx::get<Value>(*_current) >= bounds.second)) {
             _densifyState = DensifyState::kDensifyDone;
         } else {
             _densifyState = DensifyState::kNeedGen;
@@ -234,13 +410,14 @@ void DocumentSourceInternalDensify::resetDocGen() {
 }
 
 
-DocumentSourceInternalDensify::ValComparedToRange DocumentSourceInternalDensify::processRangeNum(
-    double val, double rangeMin, double rangeMax) {
-    if (val < rangeMin) {
+DocumentSourceInternalDensify::ValComparedToRange DocumentSourceInternalDensify::processRange(
+    Value val, Value current, NumericBounds bounds) {
+    int comparison = compareValues(val, current);
+    if (comparison < 0) {
         return DocumentSourceInternalDensify::ValComparedToRange::kBelow;
-    } else if (val == rangeMin) {
+    } else if (comparison == 0) {
         return DocumentSourceInternalDensify::ValComparedToRange::kRangeMin;
-    } else if (val > rangeMin && val <= rangeMax) {
+    } else if (compareValues(val <= bounds.second)) {
         return DocumentSourceInternalDensify::ValComparedToRange::kInside;
     } else {
         return DocumentSourceInternalDensify::ValComparedToRange::kAbove;
@@ -249,35 +426,42 @@ DocumentSourceInternalDensify::ValComparedToRange DocumentSourceInternalDensify:
 
 DocumentSource::GetNextResult DocumentSourceInternalDensify::handleSourceExhausted() {
     _eof = true;
-    switch (_densifyType) {
-        case TypeOfDensify::kFull: {
-            _densifyState = DensifyState::kDensifyDone;
-            return DocumentSource::GetNextResult::makeEOF();
-        }
-        case TypeOfDensify::kExplicitRange: {
-            // The _rangeMin is treated as the last seen value. Therefore, when creating document
-            // generators we pass in the _rangeMin + the step in order to avoid returning the same
-            // document twice. However, if we have yet to densify, we do not want to skip the
-            // current value of _rangeMin, so the step is decremented here to avoid that.
-            if (_densifyState == DensifyState::kUninitializedOrBelowRange) {
-                _rangeMin = stdx::get<double>(*_rangeMin) - _step.step;
-            }
-            return densifyAfterEOF();
-        }
-        case TypeOfDensify::kPartition: {
-            // TODO SERVER-57340 and SERVER-57342
-            tasserted(5734000, "TypefDensify should not be kPartition");
-            break;
-        }
-        default: { MONGO_UNREACHABLE; }
-    }
+    return stdx::visit(
+        visit_helper::Overloaded{
+            [&](RangeStatement::Full full) {
+                _densifyState = DensifyState::kDensifyDone;
+                return DocumentSource::GetNextResult::makeEOF();
+            },
+            [&](RangeStatement::Partition part) {
+                MONGO_UNREACHABLE;
+                return DocumentSource::GetNextResult::makeEOF();
+            },
+            [&](RangeStatement::DateBounds bounds) {
+                // TODO SERVER-57340 and SERVER-57342
+                tasserted(5734000, "Type of densify should not be kPartition");
+                return DocumentSource::GetNextResult::makeEOF();
+            },
+            [&](RangeStatement::NumericBounds bounds) {
+                // The _current is treated as the last seen value. Therefore, when creating document
+                // generators we pass in the _current + the step in order to avoid returning the
+                // same document twice. However, if we have yet to densify, we do not want to skip
+                // the current value of _current, so the step is decremented here to avoid that.
+                if (_densifyState == DensifyState::kUninitializedOrBelowRange) {
+                    _current = subtractValues(stdx::get<Value>(*_current), _range.getStep());
+                }
+                return densifyAfterEOF(bounds);
+            },
+        },
+        _range.getBounds());
 }
 
-DocumentSource::GetNextResult DocumentSourceInternalDensify::handleNeedGenFull(
-    Document currentDoc) {
-    // Note that _rangeMax is not the global max, its only the max up to the current document.
-    if (stdx::get<double>(*_rangeMin) + _step.step >= stdx::get<double>(*_rangeMax)) {
-        _rangeMin = stdx::get<double>(*_rangeMax);
+DocumentSource::GetNextResult DocumentSourceInternalDensify::handleNeedGenFull(Document currentDoc,
+                                                                               Value max) {
+    // Note that max is not the global max, its only the max up to the current document.
+    Value currentPlusStep = addValues(stdx::get<Value>(*_current), _range.getStep());
+
+    if (compareValues(currentPlusStep >= max)) {
+        _current = max;
         return currentDoc;
     }
 
@@ -285,19 +469,18 @@ DocumentSource::GetNextResult DocumentSourceInternalDensify::handleNeedGenFull(
     // value in the document minus the min is divisible by the step. If it is we
     // subtract step from max. This is neccessary so we don't generate the final
     // document twice.
-    auto maxAdjusted = valOffsetFromStep(stdx::get<double>(*_rangeMax),
-                                         stdx::get<double>(*_rangeMin),
-                                         _step.step) == 0
-        ? stdx::get<double>(*_rangeMax) - _step.step
-        : stdx::get<double>(*_rangeMax);
-
-    _docGenerator =
-        DocumentSourceInternalDensify::DocGenerator(stdx::get<double>(*_rangeMin) + _step.step,
-                                                    maxAdjusted,
-                                                    _step,
-                                                    _path,
-                                                    boost::none,
-                                                    std::move(currentDoc));
+    auto offsetFromStep = valOffsetFromStep(max, stdx::get<Value>(*_current), _range.getStep());
+    auto maxAdjusted =
+        compareValues(offsetFromStep == Value(0)) ? subtractValues(max, _range.getStep()) : max;
+
+    Value newCurrent = addValues(stdx::get<Value>(*_current), _range.getStep());
+    _docGenerator = DocumentSourceInternalDensify::DocGenerator(
+        DensifyValueType(newCurrent),
+        RangeStatement(_range.getStep(), NumericBounds(newCurrent, maxAdjusted), _range.getUnit()),
+        _field,
+        boost::none,
+        std::move(currentDoc),
+        pExpCtx->getValueComparator());
 
     _densifyState = DensifyState::kHaveGenerator;
     auto nextDoc = _docGenerator->getNextDocument();
@@ -305,32 +488,33 @@ DocumentSource::GetNextResult DocumentSourceInternalDensify::handleNeedGenFull(
         _docGenerator = boost::none;
         _densifyState = DensifyState::kNeedGen;
     }
-    _rangeMin = getDensifyDouble(nextDoc, _path);
+    _current = getDensifyValue(nextDoc, _field);
 
     return nextDoc;
 }
 
 DocumentSource::GetNextResult DocumentSourceInternalDensify::handleNeedGenExplicit(
-    Document currentDoc, double val) {
-    auto where = processRangeNum(val, stdx::get<double>(*_rangeMin), stdx::get<double>(*_rangeMax));
+    Document currentDoc, Value val, NumericBounds bounds) {
+    auto where = processRange(val, stdx::get<Value>(*_current), bounds);
     switch (where) {
         case ValComparedToRange::kInside: {
-            _rangeMin = stdx::get<double>(*_rangeMin) + _step.step;
-            if (stdx::get<double>(*_rangeMin) == val) {
+            _current = addValues(stdx::get<Value>(*_current), _range.getStep());
+            if (compareValues(stdx::get<Value>(*_current) == val)) {
                 return currentDoc;
             }
-            return processDocAboveMinBound(val, currentDoc);
+            return processDocAboveMinBound(val, bounds, currentDoc);
         }
         case ValComparedToRange::kAbove: {
-            _rangeMin = stdx::get<double>(*_rangeMin) + _step.step;
-            if (stdx::get<double>(*_rangeMin) > stdx::get<double>(*_rangeMax)) {
+            _current = addValues(stdx::get<Value>(*_current), _range.getStep());
+            _current = stdx::get<Value>(*_current), _range.getStep();
+            if (compareValues(stdx::get<Value>(*_current) > bounds.second)) {
                 _densifyState = DensifyState::kDensifyDone;
                 return currentDoc;
             }
-            return processDocAboveMinBound(val, currentDoc);
+            return processDocAboveMinBound(val, bounds, currentDoc);
         }
         case ValComparedToRange::kRangeMin: {
-            _rangeMin = stdx::get<double>(*_rangeMin) + _step.step;
+            _current = addValues(stdx::get<Value>(*_current), _range.getStep());
             _densifyState = DensifyState::kUninitializedOrBelowRange;
             return currentDoc;
         }
@@ -341,6 +525,31 @@ DocumentSource::GetNextResult DocumentSourceInternalDensify::handleNeedGenExplic
         default: { MONGO_UNREACHABLE; }
     }
 }
+boost::intrusive_ptr<DocumentSource> DocumentSourceInternalDensify::createFromBson(
+    BSONElement elem, const boost::intrusive_ptr<ExpressionContext>& expCtx) {
+    auto results = document_source_densify::createFromBsonInternal(elem, expCtx, kStageName, true);
+    tassert(5733413,
+            "When creating an $_internalDensify stage, only one stage should be returned",
+            results.size() == 1);
+    return results.front();
+}
+
+Value DocumentSourceInternalDensify::serialize(
+    boost::optional<ExplainOptions::Verbosity> explain) const {
+    MutableDocument spec;
+    spec[kFieldFieldName] = Value(_field.fullPath());
+    std::vector<Value> serializedPartitionByFields(_partitions.size());
+    std::transform(_partitions.begin(),
+                   _partitions.end(),
+                   serializedPartitionByFields.begin(),
+                   [&](FieldPath field) -> Value { return Value(field.fullPath()); });
+    spec[kPartitionByFieldsFieldName] = Value(serializedPartitionByFields);
+    spec[kRangeFieldName] = _range.serialize();
+    MutableDocument out;
+    out[getSourceName()] = Value(spec.freeze());
+
+    return Value(out.freezeToValue());
+}
 
 DocumentSource::GetNextResult DocumentSourceInternalDensify::doGetNext() {
     switch (_densifyState) {
@@ -357,42 +566,49 @@ DocumentSource::GetNextResult DocumentSourceInternalDensify::doGetNext() {
             }
 
             auto doc = nextDoc.getDocument();
-            if (doc.getNestedField(_path).missing()) {
+            if (doc.getNestedField(_field).missing()) {
                 // The densify field is not present, let document pass unmodified.
                 return nextDoc;
             }
-            double val = getDensifyDouble(doc, _path);
 
-            switch (_densifyType) {
-                case TypeOfDensify::kFull: {
-                    _rangeMin = val;
-                    _densifyState = DensifyState::kNeedGen;
-                    return nextDoc;
-                }
-                case TypeOfDensify::kExplicitRange: {
-                    auto where = processRangeNum(
-                        val, stdx::get<double>(*_rangeMin), stdx::get<double>(*_rangeMax));
-                    switch (where) {
-                        case ValComparedToRange::kInside: {
-                            return processDocAboveMinBound(val, nextDoc.getDocument());
-                        }
-                        case ValComparedToRange::kAbove: {
-                            return processDocAboveMinBound(val, nextDoc.getDocument());
-                        }
-                        case ValComparedToRange::kRangeMin: {
-                            _rangeMin = stdx::get<double>(*_rangeMin) + _step.step;
-                            return nextDoc;
-                        }
-                        case ValComparedToRange::kBelow: {
-                            return nextDoc;
+            Value val = getDensifyValue(doc, _field);
+
+            return stdx::visit(
+                visit_helper::Overloaded{
+                    [&](Full full) {
+                        _current = val;
+                        _densifyState = DensifyState::kNeedGen;
+                        return nextDoc;
+                    },
+                    [&](Partition partition) {
+                        tasserted(5734001, "Type of densify should not be 'partition'");
+                        return DocumentSource::GetNextResult::makeEOF();
+                    },
+                    [&](DateBounds bounds) {
+                        tasserted(5733412, "Type of densify should not be date bounds");
+                        return DocumentSource::GetNextResult::makeEOF();
+                    },
+                    [&](NumericBounds bounds) {
+                        auto where = processRange(val, stdx::get<Value>(*_current), bounds);
+                        switch (where) {
+                            case ValComparedToRange::kInside: {
+                                return processDocAboveMinBound(val, bounds, nextDoc.getDocument());
+                            }
+                            case ValComparedToRange::kAbove: {
+                                return processDocAboveMinBound(val, bounds, nextDoc.getDocument());
+                            }
+                            case ValComparedToRange::kRangeMin: {
+                                _current = addValues(stdx::get<Value>(*_current), _range.getStep());
+                                return nextDoc;
+                            }
+                            case ValComparedToRange::kBelow: {
+                                return nextDoc;
+                            }
                         }
-                    }
-                }
-                case TypeOfDensify::kPartition: {
-                    // TODO SERVER-57340 and SERVER-57342
-                    tasserted(5734001, "TypefDensify should not be kPartition");
-                }
-            }
+                        tasserted(5733414, "One of the switch statements should have been hit.");
+                        return DocumentSource::GetNextResult::makeEOF();
+                    }},
+                _range.getBounds());
         }
         case DensifyState::kNeedGen: {
             tassert(8423305, "Document generator must not exist in this state.", !_docGenerator);
@@ -406,26 +622,31 @@ DocumentSource::GetNextResult DocumentSourceInternalDensify::doGetNext() {
             }
 
             auto currentDoc = nextDoc.getDocument();
-            if (currentDoc.getNestedField(_path).missing()) {
+            if (currentDoc.getNestedField(_field).missing()) {
                 // The densify field is not present, let document pass unmodified.
                 return nextDoc;
             }
-            double val = getDensifyDouble(currentDoc, _path);
-
-            switch (_densifyType) {
-                case TypeOfDensify::kFull: {
-                    _rangeMin = *_rangeMin;
-                    _rangeMax = val;
-                    return handleNeedGenFull(currentDoc);
-                }
-                case TypeOfDensify::kExplicitRange: {
-                    return handleNeedGenExplicit(nextDoc.getDocument(), val);
-                }
-                case TypeOfDensify::kPartition: {
-                    // TODO SERVER-57340 and SERVER-57342
-                    tasserted(5734003, "TypefDensify should not be kPartition");
-                }
-            }
+            Value val = getDensifyValue(currentDoc, _field);
+
+            return stdx::visit(
+                visit_helper::Overloaded{
+                    [&](Full full) {
+                        _current = *_current;
+                        return handleNeedGenFull(currentDoc, val);
+                    },
+                    [&](Partition partition) {
+                        // TODO SERVER-57340 and SERVER-57342
+                        tasserted(5734003, "Type of densify should not be kPartition");
+                        return DocumentSource::GetNextResult::makeEOF();
+                    },
+                    [&](DateBounds bounds) {
+                        MONGO_UNREACHABLE;
+                        return DocumentSource::GetNextResult::makeEOF();
+                    },
+                    [&](NumericBounds bounds) {
+                        return handleNeedGenExplicit(nextDoc.getDocument(), val, bounds);
+                    }},
+                _range.getBounds());
         }
         case DensifyState::kHaveGenerator: {
             tassert(5733203,
@@ -434,38 +655,45 @@ DocumentSource::GetNextResult DocumentSourceInternalDensify::doGetNext() {
 
             auto generatedDoc = _docGenerator->getNextDocument();
 
-            switch (_densifyType) {
-                case TypeOfDensify::kFull: {
-                    if (_docGenerator->done()) {
-                        _docGenerator = boost::none;
-                        _densifyState = DensifyState::kNeedGen;
-                    }
-                    _rangeMin = getDensifyDouble(generatedDoc, _path);
-                    return generatedDoc;
-                }
-                case TypeOfDensify::kExplicitRange: {
-                    auto val = getDensifyDouble(generatedDoc, _path);
-                    // Only want to update the rangeMin if the value - current is divisible by the
-                    // step.
-                    auto rem = valOffsetFromStep(val, stdx::get<double>(*_rangeMin), _step.step);
-                    if (rem == 0) {
-                        _rangeMin = val;
-                    }
-                    resetDocGen();
-                    return generatedDoc;
-                }
-                case TypeOfDensify::kPartition: {
-                    // TODO SERVER-57340 and SERVER-57342
-                    tasserted(5734004, "TypefDensify should not be kPartition");
-                }
-            }
+            return stdx::visit(
+                visit_helper::Overloaded{
+                    [&](Full full) {
+                        if (_docGenerator->done()) {
+                            _docGenerator = boost::none;
+                            _densifyState = DensifyState::kNeedGen;
+                        }
+                        _current = getDensifyValue(generatedDoc, _field);
+                        return GetNextResult(std::move(generatedDoc));
+                    },
+                    [&](Partition partition) {
+                        // TODO SERVER-57340 and SERVER-57342
+                        tasserted(5734004, "Type of densify should not be kPartition");
+                        return DocumentSource::GetNextResult::makeEOF();
+                    },
+                    [&](DateBounds bounds) {
+                        MONGO_UNREACHABLE;
+                        return DocumentSource::GetNextResult::makeEOF();
+                    },
+                    [&](NumericBounds bounds) {
+                        auto val = getDensifyValue(generatedDoc, _field);
+                        // Only want to update the rangeMin if the value - current is divisible by
+                        // the step.
+                        auto rem =
+                            valOffsetFromStep(val, stdx::get<Value>(*_current), _range.getStep());
+                        if (compareValues(rem == Value(0))) {
+                            _current = val;
+                        }
+                        resetDocGen(bounds);
+                        return GetNextResult(std::move(generatedDoc));
+                    }},
+                _range.getBounds());
         }
         case DensifyState::kDensifyDone: {
             // In the full range, this should only return EOF.
-            /** In the explicit range we finish densifying over the range
-             and any remaining documents is passed to the next stage */
+            // In the explicit range we finish densifying over the range and any remaining documents
+            // is passed to the next stage.
             auto doc = pSource->getNext();
-            if (_densifyType == TypeOfDensify::kFull) {
+            if (stdx::holds_alternative<Full>(_range.getBounds())) {
                 tassert(5734005,
                         "GetNextResult must be EOF in kDensifyDone and kFull state",
                         !doc.isAdvanced());
diff --git a/src/mongo/db/pipeline/document_source_densify.h b/src/mongo/db/pipeline/document_source_densify.h
index f29e111c4da..e10a91ca71c 100644
--- a/src/mongo/db/pipeline/document_source_densify.h
+++ b/src/mongo/db/pipeline/document_source_densify.h
@@ -31,54 +31,159 @@
 
 #include "mongo/db/exec/document_value/value.h"
 #include "mongo/db/pipeline/document_source.h"
+#include "mongo/db/pipeline/document_source_densify_gen.h"
+#include "mongo/db/pipeline/expression.h"
 #include "mongo/db/pipeline/field_path.h"
 #include "mongo/db/query/datetime/date_time_support.h"
 #include "mongo/util/time_support.h"
+#include "mongo/util/visit_helper.h"
 
 namespace mongo {
 
+class RangeStatement {
+public:
+    static constexpr StringData kArgUnit = "unit"_sd;
+    static constexpr StringData kArgBounds = "bounds"_sd;
+    static constexpr StringData kArgStep = "step"_sd;
+
+    static constexpr StringData kValFull = "full"_sd;
+    static constexpr StringData kValPartition = "partition"_sd;
+
+    struct Full {};
+    struct Partition {};
+    typedef std::pair<Date_t, Date_t> DateBounds;
+    typedef std::pair<Value, Value> NumericBounds;
+    using Bounds = stdx::variant<Full, Partition, DateBounds, NumericBounds>;
+
+    Bounds getBounds() {
+        return _bounds;
+    }
+
+    boost::optional<TimeUnit> getUnit() {
+        return _unit;
+    }
+
+    Value getStep() {
+        return _step;
+    }
+
+    RangeStatement(Value step, Bounds bounds, boost::optional<TimeUnit> unit)
+        : _step(step), _bounds(bounds), _unit(unit) {}
+
+    static RangeStatement parse(RangeSpec spec);
+
+    Value serialize() const {
+        MutableDocument spec;
+        spec[kArgStep] = _step;
+        spec[kArgBounds] = stdx::visit(
+            visit_helper::Overloaded{[&](Full full) { return Value(kValFull); },
+                                     [&](Partition partition) { return Value(kValPartition); },
+                                     [&](std::pair<Date_t, Date_t> dates) {
+                                         return Value({Value(dates.first), Value(dates.second)});
+                                     },
+                                     [&](std::pair<Value, Value> vals) {
+                                         return Value({vals.first, vals.second});
+                                     }},
+            _bounds);
+        if (_unit)
+            spec[kArgUnit] = Value(serializeTimeUnit(*_unit));
+        return spec.freezeToValue();
+    }
+
+private:
+    Value _step;
+    Bounds _bounds;
+    boost::optional<TimeUnit> _unit = boost::none;
+};
+
 namespace document_source_densify {
-// TODO SERVER-57334 Translation logic goes here.
-}
+constexpr StringData kStageName = "$densify"_sd;
+
+/**
+ * The 'internal' parameter specifies whether or not we create a sort stage that is required for
+ * correct execution of an _internalDensify stage.
+ */
+std::list<boost::intrusive_ptr<DocumentSource>> createFromBsonInternal(
+    BSONElement elem,
+    const boost::intrusive_ptr<ExpressionContext>& pExpCtx,
+    StringData stageName,
+    bool isInternal);
+std::list<boost::intrusive_ptr<DocumentSource>> createFromBson(
+    BSONElement elem, const boost::intrusive_ptr<ExpressionContext>& pExpCtx);
+
+/**
+ * The 'internal' parameter specifies whether or not we create a sort stage that is required for
+ * correct execution of an _internalDensify stage.
+ */
+std::list<boost::intrusive_ptr<DocumentSource>> create(
+    const boost::intrusive_ptr<ExpressionContext>& expCtx,
+    std::list<FieldPath> partitions,
+    FieldPath field,
+    RangeStatement rangeStatement,
+    bool isInternal);
+}  // namespace document_source_densify
 
 class DocumentSourceInternalDensify final : public DocumentSource {
 public:
     static constexpr StringData kStageName = "$_internalDensify"_sd;
-
-    using DensifyValueType = stdx::variant<double, Date_t>;
-    struct StepSpec {
-        double step;
-        boost::optional<TimeUnit> unit;
-        boost::optional<TimeZone> tz;
+    static constexpr StringData kPartitionByFieldsFieldName = "partitionByFields"_sd;
+    static constexpr StringData kFieldFieldName = "field"_sd;
+    static constexpr StringData kRangeFieldName = "range"_sd;
+
+    using DensifyValueType = stdx::variant<Value, Date_t>;
+
+    DocumentSourceInternalDensify(const boost::intrusive_ptr<ExpressionContext>& pExpCtx,
+                                  const FieldPath& field,
+                                  const std::list<FieldPath>& partitions,
+                                  const RangeStatement& range)
+        : DocumentSource(kStageName, pExpCtx),
+          _field(std::move(field)),
+          _partitions(std::move(partitions)),
+          _range(std::move(range)) {
+        _current = stdx::visit(
+            visit_helper::Overloaded{
+                [&](RangeStatement::Full full) -> boost::optional<DensifyValueType> {
+                    return boost::none;
+                },
+                [&](RangeStatement::Partition partition) -> boost::optional<DensifyValueType> {
+                    return boost::none;
+                },
+                [&](RangeStatement::DateBounds bounds) -> boost::optional<DensifyValueType> {
+                    return DensifyValueType(bounds.first);
+                },
+                [&](RangeStatement::NumericBounds bounds) -> boost::optional<DensifyValueType> {
+                    return DensifyValueType(bounds.first);
+                }},
+            _range.getBounds());
     };
+
     class DocGenerator {
     public:
-        DocGenerator(DensifyValueType min,
-                     DensifyValueType max,
-                     StepSpec step,
+        DocGenerator(DensifyValueType current,
+                     RangeStatement range,
                      FieldPath fieldName,
                      boost::optional<Document> includeFields,
-                     boost::optional<Document> finalDoc);
+                     boost::optional<Document> finalDoc,
+                     ValueComparator comp);
         Document getNextDocument();
         bool done() const;
 
     private:
-        StepSpec _step;
+        ValueComparator _comp;
+        RangeStatement _range;
         // The field to add to 'includeFields' to generate a document.
         FieldPath _path;
         Document _includeFields;
-        // The document that is equal to or larger than '_max' that prompted the creation of this
-        // generator. Will be returned after the final generated document. Can be boost::none if we
-        // are generating the values at the end of the range.
+        // The document that is equal to or larger than the upper bound that prompted the creation
+        // of this generator. Will be returned after the final generated document. Can be
+        // boost::none if we are generating the values at the end of the range.
         boost::optional<Document> _finalDoc;
         // The minimum value that this generator will create, therefore the next generated document
         // will have this value.
         DensifyValueType _min;
-        // The maximum value that will be generated. This value is inclusive.
-        DensifyValueType _max;
 
         enum class GeneratorState {
-            // Generating documents between '_min' and '_max'.
+            // Generating documents between '_min' and the upper bound.
             kGeneratingDocuments,
             // Generated all necessary documents, waiting for a final 'getNextDocument()' call.
             kReturningFinalDocument,
@@ -114,10 +219,8 @@ public:
     const char* getSourceName() const final {
         return kStageName.rawData();
     }
-    Value serialize(boost::optional<ExplainOptions::Verbosity> explain = boost::none) const final {
-        // TODO SERVER-57334
-        return Value(DOC("$densify" << Document()));
-    }
+
+    Value serialize(boost::optional<ExplainOptions::Verbosity> explain = boost::none) const final;
 
     DepsTracker::State getDependencies(DepsTracker* deps) const final {
         return DepsTracker::State::SEE_NEXT;
@@ -137,54 +240,77 @@ private:
         kAbove,
     };
 
+    bool compareValues(Value::DeferredComparison deferredComparison) {
+        return pExpCtx->getValueComparator().evaluate(deferredComparison);
+    }
+
+    int compareValues(const Value& lhs, const Value& rhs) {
+        return pExpCtx->getValueComparator().compare(lhs, rhs);
+    }
+
     /**
-       Decides whether or not to build a DocGen and return the first document generated
-       or return the current doc if the rangeMin + step is greater than rangeMax.
-    */
-    DocumentSource::GetNextResult handleNeedGenFull(Document currentDoc);
+     * Decides whether or not to build a DocGen and return the first document generated or return
+     * the current doc if the rangeMin + step is greater than rangeMax.
+     */
+    DocumentSource::GetNextResult handleNeedGenFull(Document currentDoc, Value max);
 
     /**
-        Checks where the current doc's value lies compared to the range and
-        creates the correct DocGen if needed and returns the next doc.
-    */
-    DocumentSource::GetNextResult handleNeedGenExplicit(Document currentDoc, double val);
-
-    /** Takes care of when an EOF has been hit for the explicit case.
-    It checks if we have finished densifying over the range, and if so changes the
-    state to be kDensify done. Otherwise it builds a new generator that will finish
-    densifying over the range and changes the state to kHaveGen. */
-    DocumentSource::GetNextResult densifyAfterEOF();
-
-    /** Creates a document generator based on the value passed in, the current
-    _rangeMin, and the _rangeMax. Once created, the state changes to kHaveGenerator and the first
-    document from the generator is returned. */
-    DocumentSource::GetNextResult processDocAboveMinBound(double val, Document doc);
-
-    /** Takes in a value and checks if the value is below, on the bottom, inside, or above the
-    range, and returns the equivelant state from ValComparedToRange. */
-    ValComparedToRange processRangeNum(double val, double rangeMin, double rangeMax);
-
-    /** Handles when the pSource has been exhausted. In the full
-    case we are done with the densification process and the state becomes kDensifyDone, however in
-    the explicit case we may still need to densify over the remainder of the range, so the
-    densifyAfterEOF() function is called. */
-    DocumentSource::GetNextResult handleSourceExhausted();
+     * Checks where the current doc's value lies compared to the range and creates the correct
+     * DocGen if needed and returns the next doc.
+     */
+    DocumentSource::GetNextResult handleNeedGenExplicit(Document currentDoc,
+                                                        Value val,
+                                                        RangeStatement::NumericBounds bounds);
+
+    /**
+     * Takes care of when an EOF has been hit for the explicit case. It checks if we have finished
+     * densifying over the range, and if so changes the state to be kDensify done. Otherwise it
+     * builds a new generator that will finish densifying over the range and changes the state to
+     * kHaveGen.
+     */
+    DocumentSource::GetNextResult densifyAfterEOF(RangeStatement::NumericBounds);
+
+    /**
+     * Creates a document generator based on the value passed in, the current _current, and the
+     * NumericBounds. Once created, the state changes to kHaveGenerator and the first document from
+     * the generator is returned.
+     */
+    DocumentSource::GetNextResult processDocAboveMinBound(Value val,
+                                                          RangeStatement::NumericBounds bounds,
+                                                          Document doc);
+
+    /**
+     * Takes in a value and checks if the value is below, on the bottom, inside, or above the
+     * range, and returns the equivelant state from ValComparedToRange.
+     */
+    ValComparedToRange processRange(Value val, Value current, RangeStatement::NumericBounds bounds);
 
-    /** Checks if the current document generator is done. If it is and we have finished densifying,
-    it changes the state to be kDensifyDone. If there is more to densify, the state becomes
-    kNeedGen. The generator is also deleted. */
-    void resetDocGen();
+    /**
+     * Handles when the pSource has been exhausted. In the full case we are done with the
+     * densification process and the state becomes kDensifyDone, however in the explicit case we may
+     * still need to densify over the remainder of the range, so the densifyAfterEOF() function is
+     * called.
+     */
+    DocumentSource::GetNextResult handleSourceExhausted();
 
-    auto valOffsetFromStep(double val, double sub, double step) {
-        return fmod((val - sub), step);
+    /**
+     * Checks if the current document generator is done. If it is and we have finished densifying,
+     * it changes the state to be kDensifyDone. If there is more to densify, the state becomes
+     * kNeedGen. The generator is also deleted.
+     */
+    void resetDocGen(RangeStatement::NumericBounds bounds);
+
+    auto valOffsetFromStep(Value val, Value sub, Value step) {
+        Value diff = uassertStatusOK(ExpressionSubtract::apply(val, sub));
+        return uassertStatusOK(ExpressionMod::apply(diff, step));
     }
     boost::optional<DocGenerator> _docGenerator = boost::none;
 
-    // The minimum value that the document generator will create, therefore the next generated
-    // document will have this value.
-    // This is also used as last seen value by the explicit case.
-    boost::optional<DensifyValueType> _rangeMin = boost::none;
-    boost::optional<DensifyValueType> _rangeMax = boost::none;
+    /**
+     * The minimum value that the document generator will create, therefore the next generated
+     * document will have this value. This is also used as last seen value by the explicit case.
+     */
+    boost::optional<DensifyValueType> _current = boost::none;
 
     bool _eof = false;
 
@@ -196,10 +322,10 @@ private:
         kPartition,
     };
 
-    StepSpec _step;
-    FieldPath _path;
-
     DensifyState _densifyState = DensifyState::kUninitializedOrBelowRange;
     TypeOfDensify _densifyType;
+    FieldPath _field;
+    std::list<FieldPath> _partitions;
+    RangeStatement _range;
 };
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/document_source_densify.idl b/src/mongo/db/pipeline/document_source_densify.idl
new file mode 100644
index 00000000000..a531d261d1a
--- /dev/null
+++ b/src/mongo/db/pipeline/document_source_densify.idl
@@ -0,0 +1,65 @@
+# Copyright (C) 2021-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"
+  - "mongo/db/pipeline/value.idl"
+structs:
+  RangeSpec:
+    description: Specification for a range based densification.
+    strict: true
+    fields:
+      step:
+        description: "Step value. Assumed to be a number unless 'unit' is specified."
+        type: numeric
+        optional: false
+      unit:
+        description: "The time unit to be used for 'step' if densifying over dates."
+        optional: true
+        type: string
+      bounds:
+        description: "The bounds over which to densify. Defaults to 'full'."
+        optional: false
+        type: IDLAnyType
+  DensifySpec:
+    description: Specification for a $densify stage.
+    strict: true
+    fields:
+      field:
+        description: The field on which we are densifying.
+        type: string
+        optional: false
+      partitionByFields:
+        description: The field(s) that will be used as the partition keys.
+        type: array<string>
+        optional: true
+      range:
+        description: Specification for range based densification.
+        type: RangeSpec
+        optional: false
diff --git a/src/mongo/db/pipeline/document_source_densify_test.cpp b/src/mongo/db/pipeline/document_source_densify_test.cpp
index 8d1454eb37a..4d256e19797 100644
--- a/src/mongo/db/pipeline/document_source_densify_test.cpp
+++ b/src/mongo/db/pipeline/document_source_densify_test.cpp
@@ -33,6 +33,7 @@
 #include "mongo/db/pipeline/aggregation_context_fixture.h"
 #include "mongo/db/pipeline/document_source_densify.h"
 #include "mongo/db/pipeline/document_source_mock.h"
+#include "mongo/db/pipeline/pipeline.h"
 #include "mongo/unittest/death_test.h"
 #include "mongo/unittest/unittest.h"
 #include <utility>
@@ -40,9 +41,30 @@
 namespace mongo {
 namespace {
 
+using DateBounds = RangeStatement::DateBounds;
+using NumericBounds = RangeStatement::NumericBounds;
+using Full = RangeStatement::Full;
 using GenClass = DocumentSourceInternalDensify::DocGenerator;
 using DensifyFullNumericTest = AggregationContextFixture;
 using DensifyExplicitNumericTest = AggregationContextFixture;
+using DensifyCloneTest = AggregationContextFixture;
+
+MONGO_INITIALIZER_GENERAL(turnOnDensifyFlag,
+                          ("AllFailPointsRegistered"),
+                          ("BeginDocumentSourceRegistration",
+                           "addToDocSourceParserMap__internalDensify"))
+(InitializerContext*) {
+    const auto& spMap = ServerParameterSet::getGlobal()->getMap();
+    const auto& spIt = spMap.find("featureFlagDensify");
+    invariant(spIt != spMap.end());
+
+    auto* sp = spIt->second;
+    invariant(sp);
+    BSONObjBuilder bob;
+    sp->appendSupportingRoundtrip(nullptr, bob, "featureFlagDensify");
+    // Set to the new value
+    uassertStatusOK(sp->set(BSON("featureFlagDensify" << true).firstElement()));
+}
 
 Date_t makeDate(std::string dateStr) {
     auto statusDate = dateFromISOString(dateStr);
@@ -53,28 +75,53 @@ Date_t makeDate(std::string dateStr) {
 DEATH_TEST(DensifyGeneratorTest, ErrorsIfMinOverMax, "lower or equal to max") {
     Document doc{{"a", 1}};
     ASSERT_THROWS_CODE(
-        GenClass(1, 0, {1, boost::none}, "path", doc, doc), AssertionException, 5733303);
+        GenClass(Value(1),
+                 RangeStatement(Value(1), NumericBounds(Value(1), Value(0)), boost::none),
+                 "path",
+                 doc,
+                 doc,
+                 ValueComparator()),
+        AssertionException,
+        5733303);
 }
 
 DEATH_TEST(DensifyGeneratorTest, ErrorsIfStepIsZero, "be positive") {
     Document doc{{"a", 1}};
-    ASSERT_THROWS_CODE(GenClass(0, 1, {0, boost::none, boost::none}, "path", doc, doc),
-                       AssertionException,
-                       5733305);
+    ASSERT_THROWS_CODE(
+        GenClass(Value(0),
+                 RangeStatement(Value(0), NumericBounds(Value(0), Value(1)), boost::none),
+                 "path",
+                 doc,
+                 doc,
+                 ValueComparator()),
+        AssertionException,
+        5733305);
 }
 
 DEATH_TEST(DensifyGeneratorTest, ErrorsOnMixedValues, "same type") {
     Document doc{{"a", 1}};
-    ASSERT_THROWS_CODE(GenClass(0, Date_t::max(), {1, boost::none, boost::none}, "path", doc, doc),
-                       AssertionException,
-                       5733300);
+    ASSERT_THROWS_CODE(
+        GenClass(Date_t::max(),
+                 RangeStatement(Value(1), NumericBounds(Value(0), Value(1)), boost::none),
+                 "path",
+                 doc,
+                 doc,
+                 ValueComparator()),
+        AssertionException,
+        5733300);
 }
 
 DEATH_TEST(DensifyGeneratorTest, ErrorsIfFieldExistsInDocument, "cannot include field") {
     Document doc{{"path", 1}};
-    ASSERT_THROWS_CODE(GenClass(0, 1, {1, boost::none, boost::none}, "path", doc, doc),
-                       AssertionException,
-                       5733306);
+    ASSERT_THROWS_CODE(
+        GenClass(Value(0),
+                 RangeStatement(Value(1), NumericBounds(Value(0), Value(1)), boost::none),
+                 "path",
+                 doc,
+                 doc,
+                 ValueComparator()),
+        AssertionException,
+        5733306);
 }
 
 DEATH_TEST(DensifyGeneratorTest, ErrorsIfFieldExistsButIsArray, "cannot include field") {
@@ -83,9 +130,15 @@ DEATH_TEST(DensifyGeneratorTest, ErrorsIfFieldExistsButIsArray, "cannot include
     docArray.push_back(doc);
     docArray.push_back(doc);
     Document preservedFields{{"arr", Value(docArray)}};
-    ASSERT_THROWS_CODE(GenClass(0, 1, {1, boost::none, boost::none}, "arr", preservedFields, doc),
-                       AssertionException,
-                       5733306);
+    ASSERT_THROWS_CODE(
+        GenClass(Value(0),
+                 RangeStatement(Value(1), NumericBounds(Value(0), Value(1)), boost::none),
+                 "arr",
+                 preservedFields,
+                 doc,
+                 ValueComparator()),
+        AssertionException,
+        5733306);
 }
 
 TEST(DensifyGeneratorTest, ErrorsIfFieldIsInArray) {
@@ -95,21 +148,38 @@ TEST(DensifyGeneratorTest, ErrorsIfFieldIsInArray) {
     docArray.push_back(doc);
     Document preservedFields{{"arr", Value(docArray)}};
     ASSERT_THROWS_CODE(
-        GenClass(0, 1, {1, boost::none, boost::none}, "arr.path", preservedFields, doc),
+        GenClass(Value(0),
+                 RangeStatement(Value(1), NumericBounds(Value(0), Value(1)), boost::none),
+                 "arr.path",
+                 preservedFields,
+                 doc,
+                 ValueComparator()),
         AssertionException,
         5733307);
 }
 
 TEST(DensifyGeneratorTest, ErrorsIfPrefixOfFieldExists) {
     Document doc{{"a", 2}};
-    ASSERT_THROWS_CODE(GenClass(1, 1, {1, boost::none, boost::none}, "a.b", doc, doc),
-                       AssertionException,
-                       5733308);
+    ASSERT_THROWS_CODE(
+        GenClass(Value(1),
+                 RangeStatement(Value(1), NumericBounds(Value(1), Value(1)), boost::none),
+                 "a.b",
+                 doc,
+                 doc,
+                 ValueComparator()),
+        AssertionException,
+        5733308);
 }
 
 TEST(DensifyGeneratorTest, GeneratesNumericDocumentCorrectly) {
     Document doc{{"a", 2}};
-    auto generator = GenClass(1, 1, {1, boost::none, boost::none}, "a", Document(), doc);
+    auto generator =
+        GenClass(Value(1),
+                 RangeStatement(Value(1), NumericBounds(Value(1), Value(1)), boost::none),
+                 "a",
+                 Document(),
+                 doc,
+                 ValueComparator());
     ASSERT_FALSE(generator.done());
     Document docOne{{"a", 1}};
     ASSERT_DOCUMENT_EQ(docOne, generator.getNextDocument());
@@ -119,7 +189,13 @@ TEST(DensifyGeneratorTest, GeneratesNumericDocumentCorrectly) {
 }
 
 TEST(DensifyGeneratorTest, GeneratesNumericDocumentCorrectlyWithoutFinalDoc) {
-    auto generator = GenClass(1, 1, {1, boost::none, boost::none}, "a", Document(), boost::none);
+    auto generator =
+        GenClass(Value(1),
+                 RangeStatement(Value(1), NumericBounds(Value(1), Value(1)), boost::none),
+                 "a",
+                 Document(),
+                 boost::none,
+                 ValueComparator());
     ASSERT_FALSE(generator.done());
     Document docOne{{"a", 1}};
     ASSERT_DOCUMENT_EQ(docOne, generator.getNextDocument());
@@ -129,7 +205,13 @@ TEST(DensifyGeneratorTest, GeneratesNumericDocumentCorrectlyWithoutFinalDoc) {
 TEST(DensifyGeneratorTest, PreservesIncludeFields) {
     Document doc{{"a", 2}, {"b", 2}, {"c", 2}};
     Document preserveFields{{"b", 1}, {"c", 1}};
-    auto generator = GenClass(1, 1, {1, boost::none, boost::none}, "a", preserveFields, doc);
+    auto generator =
+        GenClass(Value(1),
+                 RangeStatement(Value(1), NumericBounds(Value(1), Value(1)), boost::none),
+                 "a",
+                 preserveFields,
+                 doc,
+                 ValueComparator());
     ASSERT_FALSE(generator.done());
     Document docOne{{"b", 1}, {"c", 1}, {"a", 1}};
     ASSERT_DOCUMENT_EQ(docOne, generator.getNextDocument());
@@ -140,7 +222,13 @@ TEST(DensifyGeneratorTest, PreservesIncludeFields) {
 
 TEST(DensifyGeneratorTest, GeneratesNumberOfNumericDocumentsCorrectly) {
     Document doc{{"a", 83}};
-    auto generator = GenClass(0, 10, {2, boost::none, boost::none}, "a", Document(), doc);
+    auto generator =
+        GenClass(Value(0),
+                 RangeStatement(Value(2), NumericBounds(Value(0), Value(10)), boost::none),
+                 "a",
+                 Document(),
+                 doc,
+                 ValueComparator());
     for (int curVal = 0; curVal <= 10; curVal += 2) {
         ASSERT_FALSE(generator.done());
         Document nextDoc{{"a", curVal}};
@@ -154,7 +242,13 @@ TEST(DensifyGeneratorTest, GeneratesNumberOfNumericDocumentsCorrectly) {
 TEST(DensifyGeneratorTest, WorksWithNonIntegerStepAndPreserveFields) {
     Document doc{{"a", 2}, {"b", 2}, {"c", 2}};
     Document preserveFields{{"b", 1}, {"c", 1}};
-    auto generator = GenClass(0, 10, {1.3, boost::none, boost::none}, "a", preserveFields, doc);
+    auto generator =
+        GenClass(Value(0),
+                 RangeStatement(Value(1.3), NumericBounds(Value(0), Value(10)), boost::none),
+                 "a",
+                 preserveFields,
+                 doc,
+                 ValueComparator());
     for (double curVal = 0; curVal <= 10; curVal += 1.3) {
         ASSERT_FALSE(generator.done());
         Document nextDoc{{"b", 1}, {"c", 1}, {"a", curVal}};
@@ -167,7 +261,13 @@ TEST(DensifyGeneratorTest, WorksWithNonIntegerStepAndPreserveFields) {
 
 TEST(DensifyGeneratorTest, GeneratesOffsetFromMaxDocsCorrectly) {
     Document doc{{"a", 83}};
-    auto generator = GenClass(1, 11, {2, boost::none, boost::none}, "a", Document(), doc);
+    auto generator =
+        GenClass(Value(1),
+                 RangeStatement(Value(2), NumericBounds(Value(1), Value(11)), boost::none),
+                 "a",
+                 Document(),
+                 doc,
+                 ValueComparator());
     for (int curVal = 1; curVal <= 11; curVal += 2) {
         ASSERT_FALSE(generator.done());
         Document nextDoc{{"a", curVal}};
@@ -181,7 +281,13 @@ TEST(DensifyGeneratorTest, GeneratesOffsetFromMaxDocsCorrectly) {
 TEST(DensifyGeneratorTest, GeneratesAtDottedPathCorrectly) {
     Document doc{{"a", 83}};
     Document preservedFields{{"a", Document{{"b", 1}}}};
-    auto generator = GenClass(1, 11, {2, boost::none, boost::none}, "a.c", preservedFields, doc);
+    auto generator =
+        GenClass(Value(1),
+                 RangeStatement(Value(2), NumericBounds(Value(1), Value(11)), boost::none),
+                 "a.c",
+                 preservedFields,
+                 doc,
+                 ValueComparator());
     for (int curVal = 1; curVal <= 11; curVal += 2) {
         ASSERT_FALSE(generator.done());
         Document nextDoc{{"a", Document{{"b", 1}, {"c", curVal}}}};
@@ -192,7 +298,12 @@ TEST(DensifyGeneratorTest, GeneratesAtDottedPathCorrectly) {
     ASSERT_TRUE(generator.done());
     // Test deeply nested fields.
     Document secondPreservedFields{{"a", Document{{"b", 1}}}};
-    generator = GenClass(1, 11, {2, boost::none, boost::none}, "a.c.d", secondPreservedFields, doc);
+    generator = GenClass(Value(1),
+                         RangeStatement(Value(2), NumericBounds(Value(1), Value(11)), boost::none),
+                         "a.c.d",
+                         secondPreservedFields,
+                         doc,
+                         ValueComparator());
     for (int curVal = 1; curVal <= 11; curVal += 2) {
         ASSERT_FALSE(generator.done());
         Document nextDoc{{"a", Document{{"b", 1}, {"c", Document{{"d", curVal}}}}}};
@@ -203,60 +314,56 @@ TEST(DensifyGeneratorTest, GeneratesAtDottedPathCorrectly) {
     ASSERT_TRUE(generator.done());
 }
 
-DEATH_TEST(DensifyGeneratorTest, FailsIfDatesAndTZNotProvided, "time zone") {
-    ASSERT_THROWS_CODE(GenClass(makeDate("2021-01-01T00:00:00.000Z"),
-                                makeDate("2021-01-01T00:00:02.000Z"),
-                                {1, TimeUnit::second, boost::none},
-                                "a",
-                                Document(),
-                                Document()),
-                       AssertionException,
-                       5733504);
-}
-
 DEATH_TEST(DensifyGeneratorTest, FailsIfDatesAndUnitNotProvided, "date step") {
     ASSERT_THROWS_CODE(GenClass(makeDate("2021-01-01T00:00:00.000Z"),
-                                makeDate("2021-01-01T00:00:02.000Z"),
-                                {1, boost::none, TimeZoneDatabase::utcZone()},
+                                RangeStatement(Value(1),
+                                               DateBounds(makeDate("2021-01-01T00:00:00.000Z"),
+                                                          makeDate("2021-01-01T00:00:02.000Z")),
+                                               boost::none),
                                 "a",
                                 Document(),
-                                Document()),
+                                Document(),
+                                ValueComparator()),
                        AssertionException,
                        5733501);
 }
 
 DEATH_TEST(DensifyGeneratorTest, FailsIfNumberAndUnitProvided, "non-date") {
     ASSERT_THROWS_CODE(
-        GenClass(1, 10, {1, TimeUnit::second, boost::none}, "a", Document(), Document()),
-        AssertionException,
-        5733506);
-}
-
-DEATH_TEST(DensifyGeneratorTest, FailsIfNumberAndTZProvided, "non-date") {
-    ASSERT_THROWS_CODE(
-        GenClass(1, 10, {1, boost::none, TimeZoneDatabase::utcZone()}, "a", Document(), Document()),
+        GenClass(Value(1),
+                 RangeStatement(Value(1), NumericBounds(Value(1), Value(10)), TimeUnit::second),
+                 "a",
+                 Document(),
+                 Document(),
+                 ValueComparator()),
         AssertionException,
         5733506);
 }
 
 DEATH_TEST(DensifyGeneratorTest, DateMinMustBeLessThanMax, "lower or equal to") {
     ASSERT_THROWS_CODE(GenClass(makeDate("2021-01-01T00:00:02.000Z"),
-                                makeDate("2021-01-01T00:00:01.000Z"),
-                                {1, TimeUnit::second, TimeZoneDatabase::utcZone()},
+                                RangeStatement(Value(1),
+                                               DateBounds(makeDate("2021-01-01T00:00:02.000Z"),
+                                                          makeDate("2021-01-01T00:00:01.000Z")),
+                                               TimeUnit::second),
                                 "a",
                                 Document(),
-                                Document()),
+                                Document(),
+                                ValueComparator()),
                        AssertionException,
                        5733502);
 }
 
 DEATH_TEST(DensifyGeneratorTest, DateStepMustBeInt, "integer") {
     ASSERT_THROWS_CODE(GenClass(makeDate("2021-01-01T00:00:00.000Z"),
-                                makeDate("2021-01-01T00:00:01.000Z"),
-                                {1.5, TimeUnit::second, TimeZoneDatabase::utcZone()},
+                                RangeStatement(Value(1.5),
+                                               DateBounds(makeDate("2021-01-01T00:00:00.000Z"),
+                                                          makeDate("2021-01-01T00:00:01.000Z")),
+                                               TimeUnit::second),
                                 "a",
                                 Document(),
-                                Document()),
+                                Document(),
+                                ValueComparator()),
                        AssertionException,
                        5733505);
 }
@@ -265,11 +372,14 @@ TEST(DensifyGeneratorTest, GeneratesDatesBySecondCorrectly) {
     Document doc{{"a", 83}};
     std::string dateBase = "2021-01-01T00:00:";
     auto generator = GenClass(makeDate("2021-01-01T00:00:01.000Z"),
-                              makeDate("2021-01-01T00:00:11.00Z"),
-                              {2, TimeUnit::second, TimeZoneDatabase::utcZone()},
+                              RangeStatement(Value(2),
+                                             DateBounds(makeDate("2021-01-01T00:00:01.000Z"),
+                                                        makeDate("2021-01-01T00:00:11.000Z")),
+                                             TimeUnit::second),
                               "a",
                               Document(),
-                              doc);
+                              doc,
+                              ValueComparator());
     for (int curVal = 1; curVal <= 11; curVal += 2) {
         auto appendStr = std::to_string(curVal);
         appendStr.insert(appendStr.begin(), 2 - appendStr.length(), '0');
@@ -286,11 +396,14 @@ TEST(DensifyGeneratorTest, GeneratesDatesByHourCorrectly) {
     Document doc{{"a", 83}};
     std::string dateBase = "2021-01-01T";
     auto generator = GenClass(makeDate("2021-01-01T01:00:00.000Z"),
-                              makeDate("2021-01-01T15:00:00.00Z"),
-                              {2, TimeUnit::hour, TimeZoneDatabase::utcZone()},
+                              RangeStatement(Value(2),
+                                             DateBounds(makeDate("2021-01-01T01:00:00.000Z"),
+                                                        makeDate("2021-01-01T15:00:00.000Z")),
+                                             TimeUnit::hour),
                               "a",
                               Document(),
-                              doc);
+                              doc,
+                              ValueComparator());
     for (int curVal = 1; curVal <= 15; curVal += 2) {
         auto appendStr = std::to_string(curVal);
         appendStr.insert(appendStr.begin(), 2 - appendStr.length(), '0');
@@ -307,11 +420,14 @@ TEST(DensifyGeneratorTest, GeneratesDatesByMonthCorrectly) {
     Document doc{{"a", 83}};
     std::string dateBase = "2021-";
     auto generator = GenClass(makeDate("2021-01-01T01:00:00.000Z"),
-                              makeDate("2021-10-01T00:00:00.00Z"),
-                              {2, TimeUnit::month, TimeZoneDatabase::utcZone()},
+                              RangeStatement(Value(2),
+                                             DateBounds(makeDate("2021-01-01T00:00:00.000Z"),
+                                                        makeDate("2021-10-01T00:00:00.000Z")),
+                                             TimeUnit::month),
                               "a",
                               Document(),
-                              doc);
+                              doc,
+                              ValueComparator());
     for (int curVal = 1; curVal <= 10; curVal += 2) {
         auto appendStr = std::to_string(curVal);
         appendStr.insert(appendStr.begin(), 2 - appendStr.length(), '0');
@@ -324,7 +440,8 @@ TEST(DensifyGeneratorTest, GeneratesDatesByMonthCorrectly) {
     ASSERT_TRUE(generator.done());
 }
 TEST_F(DensifyFullNumericTest, DensifySingleValue) {
-    auto densify = DocumentSourceInternalDensify(getExpCtx(), 2, "a");
+    auto densify = DocumentSourceInternalDensify(
+        getExpCtx(), "a", std::list<FieldPath>(), RangeStatement(Value(2), Full(), boost::none));
     auto source = DocumentSourceMock::createForTest({"{a: 1}"}, getExpCtx());
     densify.setSource(source.get());
 
@@ -336,7 +453,8 @@ TEST_F(DensifyFullNumericTest, DensifySingleValue) {
 }
 
 TEST_F(DensifyFullNumericTest, DensifyValuesCorrectlyWithDuplicates) {
-    auto densify = DocumentSourceInternalDensify(getExpCtx(), 2, "a");
+    auto densify = DocumentSourceInternalDensify(
+        getExpCtx(), "a", std::list<FieldPath>(), RangeStatement(Value(2), Full(), boost::none));
     auto source = DocumentSourceMock::createForTest(
         {"{a: 1}", "{a: 1}", "{a: 1}", "{a: 3}", "{a: 7}", "{a: 7}", "{a: 7}"}, getExpCtx());
     densify.setSource(source.get());
@@ -376,7 +494,8 @@ TEST_F(DensifyFullNumericTest, DensifyValuesCorrectlyWithDuplicates) {
 }
 
 TEST_F(DensifyFullNumericTest, DensifyValuesCorrectlyOffStep) {
-    auto densify = DocumentSourceInternalDensify(getExpCtx(), 3, "a");
+    auto densify = DocumentSourceInternalDensify(
+        getExpCtx(), "a", std::list<FieldPath>(), RangeStatement(Value(3), Full(), boost::none));
     auto source = DocumentSourceMock::createForTest({"{a: 1}", "{a: 9}"}, getExpCtx());
     densify.setSource(source.get());
 
@@ -403,7 +522,8 @@ TEST_F(DensifyFullNumericTest, DensifyValuesCorrectlyOffStep) {
 }
 
 TEST_F(DensifyFullNumericTest, DensifyValuesCorrectlyOnStep) {
-    auto densify = DocumentSourceInternalDensify(getExpCtx(), 2, "a");
+    auto densify = DocumentSourceInternalDensify(
+        getExpCtx(), "a", std::list<FieldPath>(), RangeStatement(Value(2), Full(), boost::none));
     auto source = DocumentSourceMock::createForTest({"{a: 1}", "{a: 9}"}, getExpCtx());
     densify.setSource(source.get());
 
@@ -434,10 +554,10 @@ TEST_F(DensifyFullNumericTest, DensifyValuesCorrectlyOnStep) {
     ASSERT(densify.getNext().isEOF());
 }
 
-
 TEST_F(DensifyFullNumericTest,
        NoDensificationIfStepIsGreaterThanDocumentDifferenceMultipleDocuments) {
-    auto densify = DocumentSourceInternalDensify(getExpCtx(), 2, "a");
+    auto densify = DocumentSourceInternalDensify(
+        getExpCtx(), "a", std::list<FieldPath>(), RangeStatement(Value(2), Full(), boost::none));
     auto source =
         DocumentSourceMock::createForTest({"{a: 1}", "{a : 2}", "{a: 3}", "{a: 4}"}, getExpCtx());
     densify.setSource(source.get());
@@ -462,7 +582,8 @@ TEST_F(DensifyFullNumericTest,
 }
 
 TEST_F(DensifyFullNumericTest, DensificationFieldMissing) {
-    auto densify = DocumentSourceInternalDensify(getExpCtx(), 10, "a");
+    auto densify = DocumentSourceInternalDensify(
+        getExpCtx(), "a", std::list<FieldPath>(), RangeStatement(Value(10), Full(), boost::none));
     auto source = DocumentSourceMock::createForTest(
         {"{b: 1}", "{a: 1}", "{a: 20}", "{b: 2}", "{b: 3}"}, getExpCtx());
     densify.setSource(source.get());
@@ -494,7 +615,8 @@ TEST_F(DensifyFullNumericTest, DensificationFieldMissing) {
 }
 
 TEST_F(DensifyFullNumericTest, NoDensificationIfStepGreaterThanDocumentDifference) {
-    auto densify = DocumentSourceInternalDensify(getExpCtx(), 10, "a");
+    auto densify = DocumentSourceInternalDensify(
+        getExpCtx(), "a", std::list<FieldPath>(), RangeStatement(Value(10), Full(), boost::none));
     auto source = DocumentSourceMock::createForTest({"{a: 1}", "{a: 9}"}, getExpCtx());
     densify.setSource(source.get());
 
@@ -509,7 +631,8 @@ TEST_F(DensifyFullNumericTest, NoDensificationIfStepGreaterThanDocumentDifferenc
 }
 
 TEST_F(DensifyFullNumericTest, DensifyOverDocumentsWithGaps) {
-    auto densify = DocumentSourceInternalDensify(getExpCtx(), 3, "a");
+    auto densify = DocumentSourceInternalDensify(
+        getExpCtx(), "a", std::list<FieldPath>(), RangeStatement(Value(3), Full(), boost::none));
     auto source = DocumentSourceMock::createForTest(
         {"{a: 1}", "{a: 2}", "{a : 3}", "{a : 4}", "{a : 9}", "{a : 10}", "{a : 15}"}, getExpCtx());
     densify.setSource(source.get());
@@ -555,10 +678,9 @@ TEST_F(DensifyFullNumericTest, DensifyOverDocumentsWithGaps) {
 TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeStartingBelowRange) {
     auto densify = DocumentSourceInternalDensify(
         getExpCtx(),
-        2,
         "a",
-        std::make_pair(DocumentSourceInternalDensify::DensifyValueType(5),
-                       DocumentSourceInternalDensify::DensifyValueType(15)));
+        std::list<FieldPath>(),
+        RangeStatement(Value(2), NumericBounds(Value(5), Value(15)), boost::none));
     auto source = DocumentSourceMock::createForTest(
         {"{a: 0}", "{a: 1}", "{a: 8}", "{a: 13}", "{a: 19}"}, getExpCtx());
     densify.setSource(source.get());
@@ -614,10 +736,9 @@ TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeStar
 TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeStepStartinOnMinRange) {
     auto densify = DocumentSourceInternalDensify(
         getExpCtx(),
-        1,
         "a",
-        std::make_pair(DocumentSourceInternalDensify::DensifyValueType(0),
-                       DocumentSourceInternalDensify::DensifyValueType(4)));
+        std::list<FieldPath>(),
+        RangeStatement(Value(1), NumericBounds(Value(0), Value(4)), boost::none));
     auto source = DocumentSourceMock::createForTest(
         {"{a: 0}", "{a: 1}", "{a: 8}", "{a: 13}", "{a: 19}"}, getExpCtx());
     densify.setSource(source.get());
@@ -664,10 +785,9 @@ TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeStep
 TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeStartingInsideRange) {
     auto densify = DocumentSourceInternalDensify(
         getExpCtx(),
-        1,
         "a",
-        std::make_pair(DocumentSourceInternalDensify::DensifyValueType(0),
-                       DocumentSourceInternalDensify::DensifyValueType(4)));
+        std::list<FieldPath>(),
+        RangeStatement(Value(1), NumericBounds(Value(0), Value(4)), boost::none));
     auto source =
         DocumentSourceMock::createForTest({"{a: 1}", "{a: 2}", "{a: 4}", "{a: 6}"}, getExpCtx());
     densify.setSource(source.get());
@@ -706,10 +826,9 @@ TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeStar
 TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeOnlyInsideRange) {
     auto densify = DocumentSourceInternalDensify(
         getExpCtx(),
-        1,
         "a",
-        std::make_pair(DocumentSourceInternalDensify::DensifyValueType(0),
-                       DocumentSourceInternalDensify::DensifyValueType(4)));
+        std::list<FieldPath>(),
+        RangeStatement(Value(1), NumericBounds(Value(0), Value(4)), boost::none));
     auto source = DocumentSourceMock::createForTest({"{a: 1}"}, getExpCtx());
     densify.setSource(source.get());
 
@@ -743,10 +862,9 @@ TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeOnly
 TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeStartingAboveRange) {
     auto densify = DocumentSourceInternalDensify(
         getExpCtx(),
-        1,
         "a",
-        std::make_pair(DocumentSourceInternalDensify::DensifyValueType(0),
-                       DocumentSourceInternalDensify::DensifyValueType(4)));
+        std::list<FieldPath>(),
+        RangeStatement(Value(1), NumericBounds(Value(0), Value(4)), boost::none));
     auto source = DocumentSourceMock::createForTest({"{a: 5}", "{a: 6}"}, getExpCtx());
     densify.setSource(source.get());
 
@@ -788,10 +906,9 @@ TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeStar
 TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeStartingInsideOffStep) {
     auto densify = DocumentSourceInternalDensify(
         getExpCtx(),
-        2,
         "a",
-        std::make_pair(DocumentSourceInternalDensify::DensifyValueType(0),
-                       DocumentSourceInternalDensify::DensifyValueType(4)));
+        std::list<FieldPath>(),
+        RangeStatement(Value(2), NumericBounds(Value(0), Value(4)), boost::none));
     auto source = DocumentSourceMock::createForTest({"{a: 1}", "{a: 6}"}, getExpCtx());
     densify.setSource(source.get());
 
@@ -826,10 +943,9 @@ TEST_F(DensifyExplicitNumericTest,
        CorrectlyDensifiesForNumericExplicitRangeStartingInsideWithDupes) {
     auto densify = DocumentSourceInternalDensify(
         getExpCtx(),
-        1,
         "a",
-        std::make_pair(DocumentSourceInternalDensify::DensifyValueType(0),
-                       DocumentSourceInternalDensify::DensifyValueType(4)));
+        std::list<FieldPath>(),
+        RangeStatement(Value(1), NumericBounds(Value(0), Value(4)), boost::none));
     auto source = DocumentSourceMock::createForTest({"{a: 1}", "{a: 1}", "{a: 6}"}, getExpCtx());
     densify.setSource(source.get());
 
@@ -871,10 +987,9 @@ TEST_F(DensifyExplicitNumericTest,
 TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeWithDupesWithinSource) {
     auto densify = DocumentSourceInternalDensify(
         getExpCtx(),
-        1,
         "a",
-        std::make_pair(DocumentSourceInternalDensify::DensifyValueType(0),
-                       DocumentSourceInternalDensify::DensifyValueType(20)));
+        std::list<FieldPath>(),
+        RangeStatement(Value(1), NumericBounds(Value(0), Value(20)), boost::none));
     auto source = DocumentSourceMock::createForTest(
         {"{a: 1}", "{a: 7}", "{a: 7}", "{a: 7}", "{a: 15}"}, getExpCtx());
     densify.setSource(source.get());
@@ -917,10 +1032,9 @@ TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeWith
 TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeAfterHitsEOF) {
     auto densify = DocumentSourceInternalDensify(
         getExpCtx(),
-        1,
         "a",
-        std::make_pair(DocumentSourceInternalDensify::DensifyValueType(2),
-                       DocumentSourceInternalDensify::DensifyValueType(5)));
+        std::list<FieldPath>(),
+        RangeStatement(Value(1), NumericBounds(Value(2), Value(5)), boost::none));
     auto source = DocumentSourceMock::createForTest({"{a: 0}"}, getExpCtx());
     densify.setSource(source.get());
 
@@ -954,10 +1068,9 @@ TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeAfte
 TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeWhenFieldIsMissing) {
     auto densify = DocumentSourceInternalDensify(
         getExpCtx(),
-        1,
         "a",
-        std::make_pair(DocumentSourceInternalDensify::DensifyValueType(0),
-                       DocumentSourceInternalDensify::DensifyValueType(4)));
+        std::list<FieldPath>(),
+        RangeStatement(Value(1), NumericBounds(Value(0), Value(4)), boost::none));
     auto source = DocumentSourceMock::createForTest({"{b: 2}", "{b: 6}", "{a: 1}"}, getExpCtx());
     densify.setSource(source.get());
 
@@ -999,10 +1112,9 @@ TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeWhen
 TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeStepLargerThanRange) {
     auto densify = DocumentSourceInternalDensify(
         getExpCtx(),
-        6,
         "a",
-        std::make_pair(DocumentSourceInternalDensify::DensifyValueType(0),
-                       DocumentSourceInternalDensify::DensifyValueType(4)));
+        std::list<FieldPath>(),
+        RangeStatement(Value(6), NumericBounds(Value(0), Value(4)), boost::none));
     auto source = DocumentSourceMock::createForTest({"{a: 1}", "{a: 6}"}, getExpCtx());
     densify.setSource(source.get());
 
@@ -1028,10 +1140,9 @@ TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeStep
 TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeHitEOFNearMax) {
     auto densify = DocumentSourceInternalDensify(
         getExpCtx(),
-        2,
         "a",
-        std::make_pair(DocumentSourceInternalDensify::DensifyValueType(0),
-                       DocumentSourceInternalDensify::DensifyValueType(3)));
+        std::list<FieldPath>(),
+        RangeStatement(Value(2), NumericBounds(Value(0), Value(3)), boost::none));
     auto source = DocumentSourceMock::createForTest({"{a: 0}", "{a: 2}"}, getExpCtx());
     densify.setSource(source.get());
 
@@ -1053,15 +1164,25 @@ TEST_F(DensifyExplicitNumericTest, CorrectlyDensifiesForNumericExplicitRangeHitE
 TEST_F(DensifyExplicitNumericTest, DensificationForNumericValuesErrorsIfFieldIsNotNumeric) {
     auto densify = DocumentSourceInternalDensify(
         getExpCtx(),
-        6,
         "a",
-        std::make_pair(DocumentSourceInternalDensify::DensifyValueType(0),
-                       DocumentSourceInternalDensify::DensifyValueType(4)));
+        std::list<FieldPath>(),
+        RangeStatement(Value(6), NumericBounds(Value(0), Value(4)), boost::none));
     auto source =
         DocumentSourceMock::createForTest({"{a: \"should be numeric\"}", "{a: 6}"}, getExpCtx());
     densify.setSource(source.get());
     ASSERT_THROWS_CODE(densify.getNext(), AssertionException, 5733201);
 }
 
+TEST_F(DensifyCloneTest, InternalDesnifyCanBeCloned) {
+
+    std::list<boost::intrusive_ptr<DocumentSource>> sources;
+    sources.push_back(make_intrusive<DocumentSourceInternalDensify>(
+        getExpCtx(),
+        "a",
+        std::list<FieldPath>(),
+        RangeStatement(Value(2), NumericBounds(Value(0), Value(3)), boost::none)));
+    auto pipe = Pipeline::create(sources, getExpCtx());
+    auto clonedPipe = pipe->clone();
+}
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/expression.cpp b/src/mongo/db/pipeline/expression.cpp
index 60b87e9de07..ab21accfcd0 100644
--- a/src/mongo/db/pipeline/expression.cpp
+++ b/src/mongo/db/pipeline/expression.cpp
@@ -2610,6 +2610,22 @@ void ExpressionFilter::_doAddDependencies(DepsTracker* deps) const {
 
 /* ------------------------- ExpressionFloor -------------------------- */
 
+StatusWith<Value> ExpressionFloor::apply(Value arg) {
+    if (!arg.numeric()) {
+        return Status{ErrorCodes::Error(5733411), "Floor must take a numeric argument"};
+    }
+    switch (arg.getType()) {
+        case NumberDouble:
+            return Value(std::floor(arg.getDouble()));
+        case NumberDecimal:
+            // Round toward the nearest decimal with a zero exponent in the negative direction.
+            return Value(arg.getDecimal().quantize(Decimal128::kNormalizedZero,
+                                                   Decimal128::kRoundTowardNegative));
+        default:
+            return arg;
+    }
+}
+
 Value ExpressionFloor::evaluateNumericArg(const Value& numericArg) const {
     // There's no point in taking the floor of integers or longs, it will have no effect.
     switch (numericArg.getType()) {
@@ -3008,28 +3024,28 @@ void ExpressionMeta::_doAddDependencies(DepsTracker* deps) const {
 
 /* ----------------------- ExpressionMod ---------------------------- */
 
-Value ExpressionMod::evaluate(const Document& root, Variables* variables) const {
-    Value lhs = _children[0]->evaluate(root, variables);
-    Value rhs = _children[1]->evaluate(root, variables);
-
+StatusWith<Value> ExpressionMod::apply(Value lhs, Value rhs) {
     BSONType leftType = lhs.getType();
     BSONType rightType = rhs.getType();
 
     if (lhs.numeric() && rhs.numeric()) {
-        auto assertNonZero = [](bool isZero) { uassert(16610, "can't $mod by zero", !isZero); };
 
         // If either side is decimal, perform the operation in decimal.
         if (leftType == NumberDecimal || rightType == NumberDecimal) {
             Decimal128 left = lhs.coerceToDecimal();
             Decimal128 right = rhs.coerceToDecimal();
-            assertNonZero(right.isZero());
+            if (right.isZero()) {
+                return Status(ErrorCodes::Error(5733415), str::stream() << "can't $mod by zero");
+            }
 
             return Value(left.modulo(right));
         }
 
         // ensure we aren't modding by 0
         double right = rhs.coerceToDouble();
-        assertNonZero(right == 0);
+        if (right == 0) {
+            return Status(ErrorCodes::Error(16610), str::stream() << "can't $mod by zero");
+        };
 
         if (leftType == NumberDouble || (rightType == NumberDouble && !rhs.integral())) {
             // Need to do fmod. Integer-valued double case is handled below.
@@ -3052,11 +3068,17 @@ Value ExpressionMod::evaluate(const Document& root, Variables* variables) const
     } else if (lhs.nullish() || rhs.nullish()) {
         return Value(BSONNULL);
     } else {
-        uasserted(16611,
-                  str::stream() << "$mod only supports numeric types, not "
-                                << typeName(lhs.getType()) << " and " << typeName(rhs.getType()));
+        return Status(ErrorCodes::Error(16611),
+                      str::stream()
+                          << "$mod only supports numeric types, not " << typeName(lhs.getType())
+                          << " and " << typeName(rhs.getType()));
     }
 }
+Value ExpressionMod::evaluate(const Document& root, Variables* variables) const {
+    Value lhs = _children[0]->evaluate(root, variables);
+    Value rhs = _children[1]->evaluate(root, variables);
+    return uassertStatusOK(apply(lhs, rhs));
+}
 
 REGISTER_STABLE_EXPRESSION(mod, ExpressionMod::parse);
 const char* ExpressionMod::getOpName() const {
diff --git a/src/mongo/db/pipeline/expression.h b/src/mongo/db/pipeline/expression.h
index 87abdbbda7d..041c3cbe794 100644
--- a/src/mongo/db/pipeline/expression.h
+++ b/src/mongo/db/pipeline/expression.h
@@ -1891,6 +1891,8 @@ public:
     explicit ExpressionFloor(ExpressionContext* const expCtx, ExpressionVector&& children)
         : ExpressionSingleNumericArg<ExpressionFloor>(expCtx, std::move(children)) {}
 
+    static StatusWith<Value> apply(Value lhs);
+
     Value evaluateNumericArg(const Value& numericArg) const final;
     const char* getOpName() const final;
 
@@ -2299,6 +2301,8 @@ public:
     ExpressionMod(ExpressionContext* const expCtx, ExpressionVector&& children)
         : ExpressionFixedArity<ExpressionMod, 2>(expCtx, std::move(children)) {}
 
+    static StatusWith<Value> apply(Value lhs, Value rhs);
+
     Value evaluate(const Document& root, Variables* variables) const final;
     const char* getOpName() const final;
 
diff --git a/src/mongo/db/pipeline/value.idl b/src/mongo/db/pipeline/value.idl
index 7c92868e9d0..806001964d5 100644
--- a/src/mongo/db/pipeline/value.idl
+++ b/src/mongo/db/pipeline/value.idl
@@ -38,3 +38,13 @@ types:
     cpp_type: Value
     serializer: Value::serializeForIDL
     deserializer: Value::deserializeForIDL
+  numeric:
+    bson_serialization_type:
+    - int
+    - decimal
+    - double
+    - long
+    description: A value with any numeric BSON type.
+    cpp_type: Value
+    serializer: Value::serializeForIDL
+    deserializer: Value::deserializeForIDL