SERVER-74061 Add support for $out to timeseries on secondaries and sharded clusters

3 files changed, 191 insertions, 130 deletions

diff --git a/jstests/core/timeseries/libs/timeseries_agg_helpers.js b/jstests/core/timeseries/libs/timeseries_agg_helpers.js
index becad2575ae..d55951b7ac5 100644
--- a/jstests/core/timeseries/libs/timeseries_agg_helpers.js
+++ b/jstests/core/timeseries/libs/timeseries_agg_helpers.js
@@ -20,14 +20,15 @@ var TimeseriesAggTests = class {
      * @returns An array of a time-series collection and a non time-series collection,
      *     respectively in this order.
      */
-    static prepareInputCollections(numHosts, numIterations, includeIdleMeasurements = true) {
+    static prepareInputCollections(numHosts,
+                                   numIterations,
+                                   includeIdleMeasurements = true,
+                                   testDB = TimeseriesAggTests.getTestDb()) {
         const timeseriesCollOption = {timeseries: {timeField: "time", metaField: "tags"}};
 
         Random.setRandomSeed();
         const hosts = TimeseriesTest.generateHosts(numHosts);
 
-        const testDB = TimeseriesAggTests.getTestDb();
-
         // Creates a time-series input collection.
         const inColl = testDB.getCollection("in");
         inColl.drop();
@@ -78,9 +79,7 @@ var TimeseriesAggTests = class {
     /**
      * Gets an output collection object with the name 'outCollname'.
      */
-    static getOutputCollection(outCollName, shouldDrop) {
-        const testDB = TimeseriesAggTests.getTestDb();
-
+    static getOutputCollection(outCollName, shouldDrop, testDB = TimeseriesAggTests.getTestDb()) {
         let outColl = testDB.getCollection(outCollName);
         if (shouldDrop) {
             outColl.drop();
@@ -101,10 +100,16 @@ var TimeseriesAggTests = class {
      * If 'shouldDrop' is set to false, the output collection will not be dropped before executing
      * 'pipeline'.
      *
+     * If 'testDB' is set, that database will be used in the aggregation pipeline.
+     *
      * Returns sorted data by "time" field. The sorted result data will help simplify comparison
      * logic.
      */
-    static getOutputAggregateResults(inColl, pipeline, prepareAction = null, shouldDrop = true) {
+    static getOutputAggregateResults(inColl,
+                                     pipeline,
+                                     prepareAction = null,
+                                     shouldDrop = true,
+                                     testDB = TimeseriesAggTests.getTestDb()) {
         // Figures out the output collection name from the last pipeline stage.
         var outCollName = "out";
         if (pipeline[pipeline.length - 1]["$out"] != undefined) {
@@ -120,7 +125,7 @@ var TimeseriesAggTests = class {
             outCollName = pipeline[pipeline.length - 1]["$merge"].into;
         }
 
-        let outColl = TimeseriesAggTests.getOutputCollection(outCollName, shouldDrop);
+        let outColl = TimeseriesAggTests.getOutputCollection(outCollName, shouldDrop, testDB);
         if (prepareAction != null) {
             prepareAction(outColl);
         }
@@ -142,4 +147,12 @@ var TimeseriesAggTests = class {
             assert.eq(actualResults[i], expectedResults[i], actualResults);
         }
     }
+
+    static generateOutPipeline(collName, dbName, options, aggStage = null) {
+        let outStage = {$out: {db: dbName, coll: collName, timeseries: options}};
+        if (aggStage) {
+            return [aggStage, outStage];
+        }
+        return [outStage];
+    }
 };
diff --git a/jstests/core/timeseries/timeseries_out.js b/jstests/core/timeseries/timeseries_out.js
deleted file mode 100644
index a75b9a26b35..00000000000
--- a/jstests/core/timeseries/timeseries_out.js
+++ /dev/null
@@ -1,122 +0,0 @@
-/**
- * Verifies that time-series collections work as expected with $out.
- *
- *
- * @tags: [
- *   # TimeseriesAggTests doesn't handle stepdowns.
- *   does_not_support_stepdowns,
- *   # We need a timeseries collection.
- *   requires_timeseries,
- *   # TODO SERVER-74601 remove tag after support for secondaries.
- *   assumes_read_preference_unchanged,
- *   # TODO SERVER-74601 remove tag after support for sharded clusters.
- *   assumes_against_mongod_not_mongos,
- *   requires_fcv_71
- * ]
- */
-(function() {
-"use strict";
-
-load("jstests/core/timeseries/libs/timeseries_agg_helpers.js");
-
-const testDB = TimeseriesAggTests.getTestDb();
-assert.commandWorked(testDB.dropDatabase());
-const numHosts = 10;
-const numIterations = 20;
-
-let [inColl, observerInColl] = TimeseriesAggTests.prepareInputCollections(numHosts, numIterations);
-
-function generateOutPipeline(collName, options, aggStage = null) {
-    let outStage = {$out: {db: testDB.getName(), coll: collName, timeseries: options}};
-    if (aggStage) {
-        return [aggStage, outStage];
-    }
-    return [outStage];
-}
-
-function runTest(observerPipeline, actualPipeline, shouldDrop = true, valueToCheck = null) {
-    // Gets the expected results from a non time-series observer input collection.
-    const expectedResults = TimeseriesAggTests.getOutputAggregateResults(
-        observerInColl, observerPipeline, null, shouldDrop);
-
-    // Gets the actual results from a time-series input collection.
-    const actualResults =
-        TimeseriesAggTests.getOutputAggregateResults(inColl, actualPipeline, null, shouldDrop);
-
-    // Verifies that the number of measurements is same as expected.
-    TimeseriesAggTests.verifyResults(actualResults, expectedResults);
-    if (valueToCheck) {
-        for (var i = 0; i < expectedResults.length; ++i) {
-            assert.eq(actualResults[i], {"time": valueToCheck}, actualResults);
-        }
-    }
-}
-
-//  Tests that $out works with time-series collections writing to a non-timeseries collection.
-runTest([{$out: "observer_out"}], [{$out: "out"}]);
-
-//  Tests that $out creates a time-series collection when the collection does not exist.
-let actualPipeline = generateOutPipeline("out_time", {timeField: "time", metaField: "tags"});
-runTest([{$out: "observer_out"}], actualPipeline);
-
-// Tests that $out creates a time-series collection with more time-series options.
-actualPipeline = generateOutPipeline(
-    "out_time",
-    {timeField: "time", metaField: "tags", bucketMaxSpanSeconds: 100, bucketRoundingSeconds: 100});
-runTest([{$out: "observer_out"}], actualPipeline);
-
-// Change an option in the existing time-series collections.
-assert.commandWorked(testDB.runCommand({collMod: "out_time", expireAfterSeconds: 360}));
-assert.commandWorked(testDB.runCommand({collMod: "observer_out", validationLevel: "moderate"}));
-
-// Tests that a time-series collection is replaced when a time-series collection does exist.
-let newDate = new Date('1999-09-30T03:24:00');
-let observerPipeline = [{$set: {"time": newDate}}, {$out: "observer_out"}];
-actualPipeline = generateOutPipeline("out_time", {timeField: "time"}, {$set: {"time": newDate}});
-
-// Confirms that all the documents have the 'newDate' value.
-runTest(observerPipeline, actualPipeline, false, newDate);
-
-// Confirms that the original time-series collection options were preserved by $out.
-let collections = assert.commandWorked(testDB.runCommand({listCollections: 1})).cursor.firstBatch;
-let coll = collections.find(entry => entry.name === "out_time");
-assert.eq(coll["options"]["expireAfterSeconds"], 360);
-coll = collections.find(entry => entry.name === "observer_out");
-assert.eq(coll["options"]["validationLevel"], "moderate");
-
-// Tests that an error is raised when trying to create a time-series collection from a non
-// time-series collection.
-let pipeline = generateOutPipeline("observer_out", {timeField: "time"});
-assert.throwsWithCode(() => inColl.aggregate(pipeline), 7268700);
-assert.throwsWithCode(() => observerInColl.aggregate(pipeline), 7268700);
-
-// Tests that an error is raised for invalid timeseries options.
-pipeline = generateOutPipeline("out_time", {timeField: "time", invalidField: "invalid"});
-assert.throwsWithCode(() => inColl.aggregate(pipeline), 40415);
-assert.throwsWithCode(() => observerInColl.aggregate(pipeline), 40415);
-
-// Tests that an error is raised if the time-series specification changes.
-pipeline = generateOutPipeline("out_time", {timeField: "usage_guest_nice"});
-assert.throwsWithCode(() => inColl.aggregate(pipeline), 7268701);
-assert.throwsWithCode(() => observerInColl.aggregate(pipeline), 7268701);
-
-pipeline = generateOutPipeline("out_time", {timeField: "time", metaField: "usage_guest_nice"});
-assert.throwsWithCode(() => inColl.aggregate(pipeline), 7268702);
-assert.throwsWithCode(() => observerInColl.aggregate(pipeline), 7268702);
-
-// Tests that a time-series collection can be replaced with a non-timeseries collection.
-runTest([{"$out": "observer_out_time"}], [{"$out": "out_time"}]);
-
-// Tests that an error is raised if a conflicting view exists.
-assert.commandWorked(testDB.createCollection("view_out", {viewOn: "out"}));
-
-pipeline = generateOutPipeline("view_out", {timeField: "time"});
-assert.throwsWithCode(() => inColl.aggregate(pipeline), 7268703);
-assert.throwsWithCode(() => observerInColl.aggregate(pipeline), 7268703);
-
-// Test $out for time-series works with a non-existent database.
-const destDB = testDB.getSiblingDB("outDifferentDB");
-assert.commandWorked(destDB.dropDatabase());
-inColl.aggregate({$out: {db: destDB.getName(), coll: "out_time", timeseries: {timeField: "time"}}});
-assert.eq(300, destDB["out_time"].find().itcount());
-})();
diff --git a/jstests/core/timeseries/timeseries_out_non_sharded.js b/jstests/core/timeseries/timeseries_out_non_sharded.js
new file mode 100644
index 00000000000..44b749fafdf
--- /dev/null
+++ b/jstests/core/timeseries/timeseries_out_non_sharded.js
@@ -0,0 +1,170 @@
+/**
+ * Verifies that $out writes to a time-series collection from an unsharded collection.
+ * There is a test for sharded source collections in jstests/sharding/timeseries_out_sharded.js.
+ *
+ * @tags: [
+ *   # TimeseriesAggTests doesn't handle stepdowns.
+ *   does_not_support_stepdowns,
+ *   # We need a timeseries collection.
+ *   requires_timeseries,
+ *   requires_fcv_71,
+ *  featureFlagAggOutTimeseries
+ * ]
+ */
+(function() {
+"use strict";
+
+load("jstests/core/timeseries/libs/timeseries_agg_helpers.js");
+
+const numHosts = 10;
+const numIterations = 20;
+
+const testDB = TimeseriesAggTests.getTestDb();
+const dbName = testDB.getName();
+assert.commandWorked(testDB.dropDatabase());
+const targetCollName = "out_time";
+
+let [inColl, observerInColl] =
+    TimeseriesAggTests.prepareInputCollections(numHosts, numIterations, true);
+
+function runTest({
+    observer: observerPipeline,
+    timeseries: timeseriesPipeline,
+    drop: shouldDrop = true,
+    value: valueToCheck = null
+}) {
+    let expectedTSOptions = null;
+    if (!shouldDrop) {
+        // To test if an index is preserved by $out when replacing an existing collection.
+        assert.commandWorked(testDB[targetCollName].createIndex({usage_guest: 1}));
+        // To test if $out preserves the original collection options.
+        let collections = testDB.getCollectionInfos({name: targetCollName});
+        assert.eq(collections.length, 1, collections);
+        expectedTSOptions = collections[0]["options"]["timeseries"];
+    } else {
+        expectedTSOptions = timeseriesPipeline[0]["$out"]["timeseries"];
+    }
+
+    // Gets the expected results from a non time-series observer input collection.
+    const expectedResults = TimeseriesAggTests.getOutputAggregateResults(
+        observerInColl, observerPipeline, null, shouldDrop);
+
+    // Gets the actual results from a time-series input collection.
+    const actualResults =
+        TimeseriesAggTests.getOutputAggregateResults(inColl, timeseriesPipeline, null, shouldDrop);
+
+    // Verifies that the number of measurements is same as expected.
+    TimeseriesAggTests.verifyResults(actualResults, expectedResults);
+    if (valueToCheck) {
+        for (var i = 0; i < expectedResults.length; ++i) {
+            assert.eq(actualResults[i], {"time": valueToCheck}, actualResults);
+        }
+    }
+
+    let collections = testDB.getCollectionInfos({name: targetCollName});
+    assert.eq(collections.length, 1, collections);
+
+    // Verifies a time-series collection was not made, if that is expected.
+    if (!expectedTSOptions) {
+        assert(!collections[0]["options"]["timeseries"], collections);
+        return;
+    }
+
+    // Verifies the time-series options are correct, if a time-series collection is expected.
+    let actualOptions = collections[0]["options"]["timeseries"];
+    for (let option in expectedTSOptions) {
+        // Must loop through each option, since 'actualOptions' will contain default fields and
+        // values that do not exist in 'expectedTSOptions'.
+        assert.eq(expectedTSOptions[option], actualOptions[option], actualOptions);
+    }
+
+    // Verifies the original index is maintained, if $out is replacing an existing collection.
+    if (!shouldDrop) {
+        let indexSpecs = testDB[targetCollName].getIndexes();
+        assert.eq(indexSpecs.filter(index => index.name == "usage_guest_1").length, 1);
+    }
+}
+
+// Tests that $out works with a source time-series collections writing to a non-timeseries
+// collection.
+runTest({observer: [{$out: "observer_out"}], timeseries: [{$out: targetCollName}]});
+
+// Tests that $out creates a time-series collection when the collection does not exist.
+let timeseriesPipeline = TimeseriesAggTests.generateOutPipeline(
+    targetCollName, dbName, {timeField: "time", metaField: "tags"});
+runTest({observer: [{$out: "observer_out"}], timeseries: timeseriesPipeline});
+
+// Test that $out can replace an existing time-series collection without the 'timeseries' option.
+// Change an option in the existing time-series collections.
+assert.commandWorked(testDB.runCommand({collMod: targetCollName, expireAfterSeconds: 360}));
+// Run the $out stage.
+timeseriesPipeline = [{$out: targetCollName}];
+runTest({observer: [{$out: "observer_out"}], timeseries: timeseriesPipeline, drop: false});
+
+// Test that $out can replace an existing time-series collection with the 'timeseries' option.
+let newDate = new Date('1999-09-30T03:24:00');
+let observerPipeline = [{$set: {"time": newDate}}, {$out: "observer_out"}];
+timeseriesPipeline = TimeseriesAggTests.generateOutPipeline(
+    targetCollName, dbName, {timeField: "time", metaField: "tags"}, {$set: {"time": newDate}});
+// Run the $out stage and confirm all the documents have the new value.
+runTest({observer: observerPipeline, timeseries: timeseriesPipeline, drop: false, value: newDate});
+
+// Test $out to time-series succeeds with a non-existent database.
+const destDB = testDB.getSiblingDB("outDifferentDB");
+assert.commandWorked(destDB.dropDatabase());
+timeseriesPipeline =
+    TimeseriesAggTests.generateOutPipeline(targetCollName, destDB.getName(), {timeField: "time"});
+// TODO SERVER-75856 remove this conditional.
+if (FixtureHelpers.isMongos(testDB)) {  // this is not supported in mongos.
+    assert.throwsWithCode(() => inColl.aggregate(timeseriesPipeline), ErrorCodes.NamespaceNotFound);
+} else {
+    inColl.aggregate(timeseriesPipeline);
+    assert.eq(300, destDB[targetCollName].find().itcount());
+}
+
+// Tests that an error is raised when trying to create a time-series collection from a non
+// time-series collection.
+let pipeline = TimeseriesAggTests.generateOutPipeline("observer_out", dbName, {timeField: "time"});
+assert.throwsWithCode(() => inColl.aggregate(pipeline), 7268700);
+assert.throwsWithCode(() => observerInColl.aggregate(pipeline), 7268700);
+
+// Tests that an error is raised for invalid timeseries options.
+pipeline = TimeseriesAggTests.generateOutPipeline(
+    targetCollName, dbName, {timeField: "time", invalidField: "invalid"});
+assert.throwsWithCode(() => inColl.aggregate(pipeline), 40415);
+assert.throwsWithCode(() => observerInColl.aggregate(pipeline), 40415);
+
+// Tests that an error is raised if the user changes the 'timeField'.
+pipeline =
+    TimeseriesAggTests.generateOutPipeline(targetCollName, dbName, {timeField: "usage_guest_nice"});
+assert.throwsWithCode(() => inColl.aggregate(pipeline), 7406103);
+assert.throwsWithCode(() => observerInColl.aggregate(pipeline), 7406103);
+
+// Tests that an error is raised if the user changes the 'metaField'.
+pipeline = TimeseriesAggTests.generateOutPipeline(
+    targetCollName, dbName, {timeField: "time", metaField: "usage_guest_nice"});
+assert.throwsWithCode(() => inColl.aggregate(pipeline), 7406103);
+assert.throwsWithCode(() => observerInColl.aggregate(pipeline), 7406103);
+
+// Tests that an error is raised if the user changes 'bucketManSpanSeconds'.
+pipeline = TimeseriesAggTests.generateOutPipeline(
+    targetCollName,
+    dbName,
+    {timeField: "time", bucketMaxSpanSeconds: 330, bucketRoundingSeconds: 330});
+assert.throwsWithCode(() => inColl.aggregate(pipeline), 7406103);
+assert.throwsWithCode(() => observerInColl.aggregate(pipeline), 7406103);
+
+// Tests that an error is raised if the user changes 'granularity'.
+pipeline = TimeseriesAggTests.generateOutPipeline(
+    targetCollName, dbName, {timeField: "time", granularity: "minutes"});
+assert.throwsWithCode(() => inColl.aggregate(pipeline), 7406103);
+assert.throwsWithCode(() => observerInColl.aggregate(pipeline), 7406103);
+
+// Tests that an error is raised if a conflicting view exists.
+if (!FixtureHelpers.isMongos(testDB)) {  // can not shard a view.
+    assert.commandWorked(testDB.createCollection("view_out", {viewOn: "out"}));
+    pipeline = TimeseriesAggTests.generateOutPipeline("view_out", dbName, {timeField: "time"});
+    assert.throwsWithCode(() => inColl.aggregate(pipeline), 7268703);
+    assert.throwsWithCode(() => observerInColl.aggregate(pipeline), 7268703);
+}
+})();