SERVER-71270 In timeseries collections prevent match pushdown before project that can affect it

5 files changed, 184 insertions, 130 deletions

diff --git a/jstests/aggregation/bugs/timeseries_should_not_push_match_before_project.js b/jstests/aggregation/bugs/timeseries_should_not_push_match_before_project.js
new file mode 100644
index 00000000000..f2465988f0e
--- /dev/null
+++ b/jstests/aggregation/bugs/timeseries_should_not_push_match_before_project.js
@@ -0,0 +1,30 @@
+// Regression test for SERVER-71270.
+(function() {
+"use strict";
+load('jstests/aggregation/extras/utils.js');  // For assertArrayEq.
+const doc = {
+    _id: 0,
+    time: new Date('2019-01-18T13:24:15.443Z'),
+    tag: {},
+};
+
+db.ts.drop();
+db.coll.drop();
+
+db.createCollection('ts', {timeseries: {timeField: 'time', metaField: 'tag'}});
+db.createCollection('coll');
+
+db.ts.insertOne(doc);
+db.coll.insertOne(doc);
+const pipeline = [
+    {$project: {'time': 0}},
+    {$match: {'time': {$lte: new Date('2019-02-13T11:36:03.481Z')}}},
+];
+
+const ts = db.ts.aggregate(pipeline).toArray();
+const vanilla = db.coll.aggregate(pipeline).toArray();
+assertArrayEq({
+    actual: ts,
+    expected: vanilla,
+});
+}());
diff --git a/src/mongo/db/exec/bucket_unpacker.cpp b/src/mongo/db/exec/bucket_unpacker.cpp
index 6a819181df9..02be8fe3d64 100644
--- a/src/mongo/db/exec/bucket_unpacker.cpp
+++ b/src/mongo/db/exec/bucket_unpacker.cpp
@@ -248,6 +248,11 @@ boost::optional<StringData> checkComparisonPredicateErrors(
         return "can't handle a computed field"_sd;
     }
 
+    // We must avoid mapping predicates on fields removed by $project.
+    if (!determineIncludeField(matchExprPath, bucketSpec.behavior(), bucketSpec.fieldSet())) {
+        return "can't handle a field removed by projection"_sd;
+    }
+
     const auto isTimeField = (matchExprPath == bucketSpec.timeField());
     if (isTimeField && matchExprData.type() != BSONType::Date) {
         // Users are not allowed to insert non-date measurements into time field. So this query
@@ -885,10 +890,8 @@ std::pair<bool, BSONObj> BucketSpec::pushdownPredicate(
                   metaField.map([](StringData s) { return s.toString(); }),
                   // Since we are operating on a collection, not a query-result,
                   // there are no inclusion/exclusion projections we need to apply
-                  // to the buckets before unpacking.
-                  {},
-                  // And there are no computed projections.
-                  {},
+                  // to the buckets before unpacking. So we can use default values for the rest of
+                  // the arguments.
               },
               maxSpanSeconds,
               collationMatchesDefault,
@@ -924,7 +927,6 @@ public:
                                           int j,
                                           const BucketSpec& spec,
                                           const std::set<std::string>& unpackFieldsToIncludeExclude,
-                                          BucketUnpacker::Behavior behavior,
                                           const BSONObj& bucket,
                                           const Value& metaValue,
                                           bool includeTimeField,
@@ -976,7 +978,6 @@ public:
                                   int j,
                                   const BucketSpec& spec,
                                   const std::set<std::string>& unpackFieldsToIncludeExclude,
-                                  BucketUnpacker::Behavior behavior,
                                   const BSONObj& bucket,
                                   const Value& metaValue,
                                   bool includeTimeField,
@@ -988,7 +989,7 @@ private:
     BSONObjIterator _timeFieldIter;
 
     // Iterators used to unpack the columns of the above bucket that are populated during the reset
-    // phase according to the provided 'Behavior' and 'BucketSpec'.
+    // phase according to the provided 'BucketSpec'.
     std::vector<std::pair<std::string, BSONObjIterator>> _fieldIters;
 };
 
@@ -1063,7 +1064,6 @@ void BucketUnpackerV1::extractSingleMeasurement(
     int j,
     const BucketSpec& spec,
     const std::set<std::string>& unpackFieldsToIncludeExclude,
-    BucketUnpacker::Behavior behavior,
     const BSONObj& bucket,
     const Value& metaValue,
     bool includeTimeField,
@@ -1078,7 +1078,7 @@ void BucketUnpackerV1::extractSingleMeasurement(
 
     for (auto&& dataElem : dataRegion) {
         const auto& colName = dataElem.fieldNameStringData();
-        if (!determineIncludeField(colName, behavior, unpackFieldsToIncludeExclude)) {
+        if (!determineIncludeField(colName, spec.behavior(), unpackFieldsToIncludeExclude)) {
             continue;
         }
         auto value = dataElem[targetIdx];
@@ -1110,7 +1110,6 @@ public:
                                   int j,
                                   const BucketSpec& spec,
                                   const std::set<std::string>& unpackFieldsToIncludeExclude,
-                                  BucketUnpacker::Behavior behavior,
                                   const BSONObj& bucket,
                                   const Value& metaValue,
                                   bool includeTimeField,
@@ -1140,7 +1139,7 @@ private:
     ColumnStore _timeColumn;
 
     // Iterators used to unpack the columns of the above bucket that are populated during the reset
-    // phase according to the provided 'Behavior' and 'BucketSpec'.
+    // phase according to the provided 'BucketSpec'.
     std::vector<ColumnStore> _fieldColumns;
 
     // Element count
@@ -1200,7 +1199,6 @@ void BucketUnpackerV2::extractSingleMeasurement(
     int j,
     const BucketSpec& spec,
     const std::set<std::string>& unpackFieldsToIncludeExclude,
-    BucketUnpacker::Behavior behavior,
     const BSONObj& bucket,
     const Value& metaValue,
     bool includeTimeField,
@@ -1236,9 +1234,11 @@ std::size_t BucketUnpackerV2::numberOfFields() {
 BucketSpec::BucketSpec(const std::string& timeField,
                        const boost::optional<std::string>& metaField,
                        const std::set<std::string>& fields,
+                       Behavior behavior,
                        const std::set<std::string>& computedProjections,
                        bool usesExtendedRange)
     : _fieldSet(fields),
+      _behavior(behavior),
       _computedMetaProjFields(computedProjections),
       _timeField(timeField),
       _timeFieldHashed(FieldNameHasher().hashedFieldName(_timeField)),
@@ -1251,6 +1251,7 @@ BucketSpec::BucketSpec(const std::string& timeField,
 
 BucketSpec::BucketSpec(const BucketSpec& other)
     : _fieldSet(other._fieldSet),
+      _behavior(other._behavior),
       _computedMetaProjFields(other._computedMetaProjFields),
       _timeField(other._timeField),
       _timeFieldHashed(HashedFieldName{_timeField, other._timeFieldHashed->hash()}),
@@ -1263,6 +1264,7 @@ BucketSpec::BucketSpec(const BucketSpec& other)
 
 BucketSpec::BucketSpec(BucketSpec&& other)
     : _fieldSet(std::move(other._fieldSet)),
+      _behavior(other._behavior),
       _computedMetaProjFields(std::move(other._computedMetaProjFields)),
       _timeField(std::move(other._timeField)),
       _timeFieldHashed(HashedFieldName{_timeField, other._timeFieldHashed->hash()}),
@@ -1276,6 +1278,7 @@ BucketSpec::BucketSpec(BucketSpec&& other)
 BucketSpec& BucketSpec::operator=(const BucketSpec& other) {
     if (&other != this) {
         _fieldSet = other._fieldSet;
+        _behavior = other._behavior;
         _computedMetaProjFields = other._computedMetaProjFields;
         _timeField = other._timeField;
         _timeFieldHashed = HashedFieldName{_timeField, other._timeFieldHashed->hash()};
@@ -1325,8 +1328,8 @@ BucketUnpacker::BucketUnpacker(BucketUnpacker&& other) = default;
 BucketUnpacker::~BucketUnpacker() = default;
 BucketUnpacker& BucketUnpacker::operator=(BucketUnpacker&& rhs) = default;
 
-BucketUnpacker::BucketUnpacker(BucketSpec spec, Behavior unpackerBehavior) {
-    setBucketSpecAndBehavior(std::move(spec), unpackerBehavior);
+BucketUnpacker::BucketUnpacker(BucketSpec spec) {
+    setBucketSpec(std::move(spec));
 }
 
 void BucketUnpacker::addComputedMetaProjFields(const std::vector<StringData>& computedFieldNames) {
@@ -1335,7 +1338,7 @@ void BucketUnpacker::addComputedMetaProjFields(const std::vector<StringData>& co
 
         // If we're already specifically including fields, we need to add the computed fields to
         // the included field set to indicate they're in the output doc.
-        if (_unpackerBehavior == BucketUnpacker::Behavior::kInclude) {
+        if (_spec.behavior() == BucketSpec::Behavior::kInclude) {
             _spec.addIncludeExcludeField(field);
         } else {
             // Since exclude is applied after addComputedMetaProjFields, we must erase the new field
@@ -1387,7 +1390,6 @@ Document BucketUnpacker::extractSingleMeasurement(int j) {
                                              j,
                                              _spec,
                                              fieldsToIncludeExcludeDuringUnpack(),
-                                             _unpackerBehavior,
                                              _bucket,
                                              _metaValue,
                                              _includeTimeField,
@@ -1501,10 +1503,10 @@ void BucketUnpacker::reset(BSONObj&& bucket, bool bucketMatchedQuery) {
             continue;
         }
 
-        // Includes a field when '_unpackerBehavior' is 'kInclude' and it's found in 'fieldSet' or
-        // _unpackerBehavior is 'kExclude' and it's not found in 'fieldSet'.
+        // Includes a field when '_spec.behavior()' is 'kInclude' and it's found in 'fieldSet' or
+        // _spec.behavior() is 'kExclude' and it's not found in 'fieldSet'.
         if (determineIncludeField(
-                colName, _unpackerBehavior, fieldsToIncludeExcludeDuringUnpack())) {
+                colName, _spec.behavior(), fieldsToIncludeExcludeDuringUnpack())) {
             _unpackingImpl->addField(elem);
         }
     }
@@ -1555,7 +1557,7 @@ int BucketUnpacker::computeMeasurementCount(const BSONObj& bucket, StringData ti
 }
 
 void BucketUnpacker::determineIncludeTimeField() {
-    const bool isInclude = _unpackerBehavior == BucketUnpacker::Behavior::kInclude;
+    const bool isInclude = _spec.behavior() == BucketSpec::Behavior::kInclude;
     const bool fieldSetContainsTime =
         _spec.fieldSet().find(_spec.timeField()) != _spec.fieldSet().end();
 
@@ -1575,22 +1577,21 @@ void BucketUnpacker::eraseMetaFromFieldSetAndDetermineIncludeMeta() {
     } else if (auto itr = _spec.fieldSet().find(*_spec.metaField());
                itr != _spec.fieldSet().end()) {
         _spec.removeIncludeExcludeField(*_spec.metaField());
-        _includeMetaField = _unpackerBehavior == BucketUnpacker::Behavior::kInclude;
+        _includeMetaField = _spec.behavior() == BucketSpec::Behavior::kInclude;
     } else {
-        _includeMetaField = _unpackerBehavior == BucketUnpacker::Behavior::kExclude;
+        _includeMetaField = _spec.behavior() == BucketSpec::Behavior::kExclude;
     }
 }
 
 void BucketUnpacker::eraseExcludedComputedMetaProjFields() {
-    if (_unpackerBehavior == BucketUnpacker::Behavior::kExclude) {
+    if (_spec.behavior() == BucketSpec::Behavior::kExclude) {
         for (const auto& field : _spec.fieldSet()) {
             _spec.eraseFromComputedMetaProjFields(field);
         }
     }
 }
 
-void BucketUnpacker::setBucketSpecAndBehavior(BucketSpec&& bucketSpec, Behavior behavior) {
-    _unpackerBehavior = behavior;
+void BucketUnpacker::setBucketSpec(BucketSpec&& bucketSpec) {
     _spec = std::move(bucketSpec);
 
     eraseMetaFromFieldSetAndDetermineIncludeMeta();
@@ -1616,7 +1617,7 @@ const std::set<std::string>& BucketUnpacker::fieldsToIncludeExcludeDuringUnpack(
 
     _unpackFieldsToIncludeExclude = std::set<std::string>();
     const auto& metaProjFields = _spec.computedMetaProjFields();
-    if (_unpackerBehavior == BucketUnpacker::Behavior::kInclude) {
+    if (_spec.behavior() == BucketSpec::Behavior::kInclude) {
         // For include, we unpack fieldSet - metaProjFields.
         for (auto&& field : _spec.fieldSet()) {
             if (metaProjFields.find(field) == metaProjFields.cend()) {
diff --git a/src/mongo/db/exec/bucket_unpacker.h b/src/mongo/db/exec/bucket_unpacker.h
index 3a9813eb87a..cfefeb2e579 100644
--- a/src/mongo/db/exec/bucket_unpacker.h
+++ b/src/mongo/db/exec/bucket_unpacker.h
@@ -54,10 +54,16 @@ namespace mongo {
  */
 class BucketSpec {
 public:
+    // When unpackin buckets with kInclude we must produce measurements that contain the
+    // set of fields. Otherwise, if the kExclude option is used, the measurements will include the
+    // set difference between all fields in the bucket and the provided fields.
+    enum class Behavior { kInclude, kExclude };
+
     BucketSpec() = default;
     BucketSpec(const std::string& timeField,
                const boost::optional<std::string>& metaField,
                const std::set<std::string>& fields = {},
+               Behavior behavior = Behavior::kExclude,
                const std::set<std::string>& computedProjections = {},
                bool usesExtendedRange = false);
     BucketSpec(const BucketSpec&);
@@ -93,6 +99,14 @@ public:
         return _fieldSet;
     }
 
+    void setBehavior(Behavior behavior) {
+        _behavior = behavior;
+    }
+
+    Behavior behavior() const {
+        return _behavior;
+    }
+
     void addComputedMetaProjFields(const StringData& field) {
         _computedMetaProjFields.emplace(field);
     }
@@ -211,6 +225,7 @@ public:
 private:
     // The set of field names in the data region that should be included or excluded.
     std::set<std::string> _fieldSet;
+    Behavior _behavior = Behavior::kExclude;
 
     // Set of computed meta field projection names. Added at the end of materialized
     // measurements.
@@ -229,10 +244,6 @@ private:
  */
 class BucketUnpacker {
 public:
-    // When BucketUnpacker is created with kInclude it must produce measurements that contain the
-    // set of fields. Otherwise, if the kExclude option is used, the measurements will include the
-    // set difference between all fields in the bucket and the provided fields.
-    enum class Behavior { kInclude, kExclude };
     /**
      * Returns the number of measurements in the bucket in O(1) time.
      */
@@ -242,7 +253,7 @@ public:
     static const std::set<StringData> reservedBucketFieldNames;
 
     BucketUnpacker();
-    BucketUnpacker(BucketSpec spec, Behavior unpackerBehavior);
+    BucketUnpacker(BucketSpec spec);
     BucketUnpacker(const BucketUnpacker& other) = delete;
     BucketUnpacker(BucketUnpacker&& other);
     ~BucketUnpacker();
@@ -274,7 +285,6 @@ public:
      */
     BucketUnpacker copy() const {
         BucketUnpacker unpackerCopy;
-        unpackerCopy._unpackerBehavior = _unpackerBehavior;
         unpackerCopy._spec = _spec;
         unpackerCopy._includeMetaField = _includeMetaField;
         unpackerCopy._includeTimeField = _includeTimeField;
@@ -286,8 +296,8 @@ public:
      */
     void reset(BSONObj&& bucket, bool bucketMatchedQuery = false);
 
-    Behavior behavior() const {
-        return _unpackerBehavior;
+    BucketSpec::Behavior behavior() const {
+        return _spec.behavior();
     }
 
     const BucketSpec& bucketSpec() const {
@@ -338,7 +348,7 @@ public:
         return std::string{timeseries::kControlMaxFieldNamePrefix} + field;
     }
 
-    void setBucketSpecAndBehavior(BucketSpec&& bucketSpec, Behavior behavior);
+    void setBucketSpec(BucketSpec&& bucketSpec);
     void setIncludeMinTimeAsMetadata();
     void setIncludeMaxTimeAsMetadata();
 
@@ -363,7 +373,6 @@ private:
     void eraseExcludedComputedMetaProjFields();
 
     BucketSpec _spec;
-    Behavior _unpackerBehavior;
 
     std::unique_ptr<UnpackingImpl> _unpackingImpl;
 
@@ -418,9 +427,9 @@ private:
  * Determines if an arbitrary field should be included in the materialized measurements.
  */
 inline bool determineIncludeField(StringData fieldName,
-                                  BucketUnpacker::Behavior unpackerBehavior,
+                                  BucketSpec::Behavior unpackerBehavior,
                                   const std::set<std::string>& unpackFieldsToIncludeExclude) {
-    const bool isInclude = unpackerBehavior == BucketUnpacker::Behavior::kInclude;
+    const bool isInclude = unpackerBehavior == BucketSpec::Behavior::kInclude;
     const bool unpackFieldsContains = unpackFieldsToIncludeExclude.find(fieldName.toString()) !=
         unpackFieldsToIncludeExclude.cend();
     return isInclude == unpackFieldsContains;
diff --git a/src/mongo/db/exec/bucket_unpacker_test.cpp b/src/mongo/db/exec/bucket_unpacker_test.cpp
index 84fbcaf3ca0..1be1929aab2 100644
--- a/src/mongo/db/exec/bucket_unpacker_test.cpp
+++ b/src/mongo/db/exec/bucket_unpacker_test.cpp
@@ -57,12 +57,12 @@ public:
      * before actually doing any unpacking.
      */
     BucketUnpacker makeBucketUnpacker(std::set<std::string> fields,
-                                      BucketUnpacker::Behavior behavior,
+                                      BucketSpec::Behavior behavior,
                                       BSONObj bucket,
                                       boost::optional<std::string> metaFieldName = boost::none) {
-        auto spec = BucketSpec{kUserDefinedTimeName.toString(), metaFieldName, std::move(fields)};
-
-        BucketUnpacker unpacker{std::move(spec), behavior};
+        auto spec =
+            BucketSpec{kUserDefinedTimeName.toString(), metaFieldName, std::move(fields), behavior};
+        BucketUnpacker unpacker{std::move(spec)};
         unpacker.reset(std::move(bucket));
         return unpacker;
     }
@@ -72,12 +72,13 @@ public:
      * the given 'bucket'. Asserts that 'reset()' throws the given 'errorCode'.
      */
     void assertUnpackerThrowsCode(std::set<std::string> fields,
-                                  BucketUnpacker::Behavior behavior,
+                                  BucketSpec::Behavior behavior,
                                   BSONObj bucket,
                                   boost::optional<std::string> metaFieldName,
                                   int errorCode) {
-        auto spec = BucketSpec{kUserDefinedTimeName.toString(), metaFieldName, std::move(fields)};
-        BucketUnpacker unpacker{std::move(spec), behavior};
+        auto spec =
+            BucketSpec{kUserDefinedTimeName.toString(), metaFieldName, std::move(fields), behavior};
+        BucketUnpacker unpacker{std::move(spec)};
         ASSERT_THROWS_CODE(unpacker.reset(std::move(bucket)), AssertionException, errorCode);
     }
 
@@ -178,7 +179,7 @@ TEST_F(BucketUnpackerTest, UnpackBasicIncludeAllMeasurementFields) {
         "a:{'0':1, '1':2}, b:{'1':1}}}");
 
     auto unpacker = makeBucketUnpacker(std::move(fields),
-                                       BucketUnpacker::Behavior::kInclude,
+                                       BucketSpec::Behavior::kInclude,
                                        std::move(bucket),
                                        kUserDefinedMetaName.toString());
 
@@ -202,7 +203,7 @@ TEST_F(BucketUnpackerTest, ExcludeASingleField) {
 
     auto test = [&](BSONObj bucket) {
         auto unpacker = makeBucketUnpacker(fields,
-                                           BucketUnpacker::Behavior::kExclude,
+                                           BucketSpec::Behavior::kExclude,
                                            std::move(bucket),
                                            kUserDefinedMetaName.toString());
 
@@ -231,7 +232,7 @@ TEST_F(BucketUnpackerTest, EmptyIncludeGetsEmptyMeasurements) {
 
     auto test = [&](BSONObj bucket) {
         auto unpacker = makeBucketUnpacker(fields,
-                                           BucketUnpacker::Behavior::kInclude,
+                                           BucketSpec::Behavior::kInclude,
                                            std::move(bucket),
                                            kUserDefinedMetaName.toString());
 
@@ -258,7 +259,7 @@ TEST_F(BucketUnpackerTest, EmptyExcludeMaterializesAllFields) {
 
     auto test = [&](BSONObj bucket) {
         auto unpacker = makeBucketUnpacker(fields,
-                                           BucketUnpacker::Behavior::kExclude,
+                                           BucketSpec::Behavior::kExclude,
                                            std::move(bucket),
                                            kUserDefinedMetaName.toString());
         ASSERT_TRUE(unpacker.hasNext());
@@ -287,7 +288,7 @@ TEST_F(BucketUnpackerTest, SparseColumnsWhereOneColumnIsExhaustedBeforeTheOther)
 
     auto test = [&](BSONObj bucket) {
         auto unpacker = makeBucketUnpacker(fields,
-                                           BucketUnpacker::Behavior::kExclude,
+                                           BucketSpec::Behavior::kExclude,
                                            std::move(bucket),
                                            kUserDefinedMetaName.toString());
         ASSERT_TRUE(unpacker.hasNext());
@@ -315,7 +316,7 @@ TEST_F(BucketUnpackerTest, UnpackBasicIncludeWithDollarPrefix) {
 
     auto test = [&](BSONObj bucket) {
         auto unpacker = makeBucketUnpacker(fields,
-                                           BucketUnpacker::Behavior::kInclude,
+                                           BucketSpec::Behavior::kInclude,
                                            std::move(bucket),
                                            kUserDefinedMetaName.toString());
         ASSERT_TRUE(unpacker.hasNext());
@@ -343,7 +344,7 @@ TEST_F(BucketUnpackerTest, BucketsWithMetadataOnly) {
 
     auto test = [&](BSONObj bucket) {
         auto unpacker = makeBucketUnpacker(fields,
-                                           BucketUnpacker::Behavior::kExclude,
+                                           BucketSpec::Behavior::kExclude,
                                            std::move(bucket),
                                            kUserDefinedMetaName.toString());
         ASSERT_TRUE(unpacker.hasNext());
@@ -370,7 +371,7 @@ TEST_F(BucketUnpackerTest, UnorderedRowKeysDoesntAffectMaterialization) {
 
     auto test = [&](BSONObj bucket) {
         auto unpacker = makeBucketUnpacker(fields,
-                                           BucketUnpacker::Behavior::kExclude,
+                                           BucketSpec::Behavior::kExclude,
                                            std::move(bucket),
                                            kUserDefinedMetaName.toString());
         ASSERT_TRUE(unpacker.hasNext());
@@ -400,7 +401,7 @@ TEST_F(BucketUnpackerTest, MissingMetaFieldDoesntMaterializeMetadata) {
 
     auto test = [&](BSONObj bucket) {
         auto unpacker = makeBucketUnpacker(fields,
-                                           BucketUnpacker::Behavior::kExclude,
+                                           BucketSpec::Behavior::kExclude,
                                            std::move(bucket),
                                            kUserDefinedMetaName.toString());
         ASSERT_TRUE(unpacker.hasNext());
@@ -428,7 +429,7 @@ TEST_F(BucketUnpackerTest, MissingMetaFieldDoesntMaterializeMetadataUnorderedKey
 
     auto test = [&](BSONObj bucket) {
         auto unpacker = makeBucketUnpacker(fields,
-                                           BucketUnpacker::Behavior::kExclude,
+                                           BucketSpec::Behavior::kExclude,
                                            std::move(bucket),
                                            kUserDefinedMetaName.toString());
         ASSERT_TRUE(unpacker.hasNext());
@@ -456,7 +457,7 @@ TEST_F(BucketUnpackerTest, ExcludedMetaFieldDoesntMaterializeMetadataWhenBucketH
 
     auto test = [&](BSONObj bucket) {
         auto unpacker = makeBucketUnpacker(fields,
-                                           BucketUnpacker::Behavior::kExclude,
+                                           BucketSpec::Behavior::kExclude,
                                            std::move(bucket),
                                            kUserDefinedMetaName.toString());
         ASSERT_TRUE(unpacker.hasNext());
@@ -484,7 +485,7 @@ TEST_F(BucketUnpackerTest, UnpackerResetThrowsOnUndefinedMeta) {
 
     auto test = [&](BSONObj bucket) {
         assertUnpackerThrowsCode(fields,
-                                 BucketUnpacker::Behavior::kExclude,
+                                 BucketSpec::Behavior::kExclude,
                                  std::move(bucket),
                                  kUserDefinedMetaName.toString(),
                                  5369600);
@@ -505,7 +506,7 @@ TEST_F(BucketUnpackerTest, UnpackerResetThrowsOnUnexpectedMeta) {
 
     auto test = [&](BSONObj bucket) {
         assertUnpackerThrowsCode(fields,
-                                 BucketUnpacker::Behavior::kExclude,
+                                 BucketSpec::Behavior::kExclude,
                                  std::move(bucket),
                                  boost::none /* no metaField provided */,
                                  5369601);
@@ -525,7 +526,7 @@ TEST_F(BucketUnpackerTest, NullMetaInBucketMaterializesAsNull) {
 
     auto test = [&](BSONObj bucket) {
         auto unpacker = makeBucketUnpacker(fields,
-                                           BucketUnpacker::Behavior::kExclude,
+                                           BucketSpec::Behavior::kExclude,
                                            std::move(bucket),
                                            kUserDefinedMetaName.toString());
         ASSERT_TRUE(unpacker.hasNext());
@@ -558,7 +559,7 @@ TEST_F(BucketUnpackerTest, GetNextHandlesMissingMetaInBucket) {
 
     auto test = [&](BSONObj bucket) {
         auto unpacker = makeBucketUnpacker(fields,
-                                           BucketUnpacker::Behavior::kExclude,
+                                           BucketSpec::Behavior::kExclude,
                                            std::move(bucket),
                                            kUserDefinedMetaName.toString());
         ASSERT_TRUE(unpacker.hasNext());
@@ -588,7 +589,7 @@ TEST_F(BucketUnpackerTest, EmptyDataRegionInBucketIsTolerated) {
 
     auto test = [&](BSONObj bucket) {
         auto unpacker = makeBucketUnpacker(
-            fields, BucketUnpacker::Behavior::kExclude, bucket, kUserDefinedMetaName.toString());
+            fields, BucketSpec::Behavior::kExclude, bucket, kUserDefinedMetaName.toString());
         ASSERT_FALSE(unpacker.hasNext());
     };
 
@@ -600,7 +601,7 @@ TEST_F(BucketUnpackerTest, UnpackerResetThrowsOnEmptyBucket) {
 
     auto bucket = Document{};
     assertUnpackerThrowsCode(std::move(fields),
-                             BucketUnpacker::Behavior::kExclude,
+                             BucketSpec::Behavior::kExclude,
                              bucket.toBson(),
                              kUserDefinedMetaName.toString(),
                              5346510);
@@ -618,48 +619,52 @@ TEST_F(BucketUnpackerTest, EraseMetaFromFieldSetAndDetermineIncludeMeta) {
     }
 })");
     auto unpacker = makeBucketUnpacker(empFields,
-                                       BucketUnpacker::Behavior::kInclude,
+                                       BucketSpec::Behavior::kInclude,
                                        std::move(bucket),
                                        kUserDefinedMetaName.toString());
 
     // Tests a spec with 'metaField' in include list.
     std::set<std::string> fields{kUserDefinedMetaName.toString()};
 
-    auto specWithMetaInclude = BucketSpec{
-        kUserDefinedTimeName.toString(), kUserDefinedMetaName.toString(), std::move(fields)};
+    auto specWithMetaInclude = BucketSpec{kUserDefinedTimeName.toString(),
+                                          kUserDefinedMetaName.toString(),
+                                          std::move(fields),
+                                          BucketSpec::Behavior::kInclude};
     // This calls eraseMetaFromFieldSetAndDetermineIncludeMeta.
-    unpacker.setBucketSpecAndBehavior(std::move(specWithMetaInclude),
-                                      BucketUnpacker::Behavior::kInclude);
+    unpacker.setBucketSpec(std::move(specWithMetaInclude));
     ASSERT_TRUE(unpacker.includeMetaField());
     ASSERT_EQ(unpacker.bucketSpec().fieldSet().count(kUserDefinedMetaName.toString()), 0);
 
     std::set<std::string> fieldsNoMetaInclude{"foo"};
     auto specWithFooInclude = BucketSpec{kUserDefinedTimeName.toString(),
                                          kUserDefinedMetaName.toString(),
-                                         std::move(fieldsNoMetaInclude)};
+                                         std::move(fieldsNoMetaInclude),
+                                         BucketSpec::Behavior::kInclude};
 
     std::set<std::string> fieldsNoMetaExclude{"foo"};
     auto specWithFooExclude = BucketSpec{kUserDefinedTimeName.toString(),
                                          kUserDefinedMetaName.toString(),
-                                         std::move(fieldsNoMetaExclude)};
+                                         std::move(fieldsNoMetaExclude),
+                                         BucketSpec::Behavior::kExclude};
 
-    unpacker.setBucketSpecAndBehavior(std::move(specWithFooExclude),
-                                      BucketUnpacker::Behavior::kExclude);
+    unpacker.setBucketSpec(std::move(specWithFooExclude));
     ASSERT_TRUE(unpacker.includeMetaField());
-    unpacker.setBucketSpecAndBehavior(std::move(specWithFooInclude),
-                                      BucketUnpacker::Behavior::kInclude);
+    unpacker.setBucketSpec(std::move(specWithFooInclude));
     ASSERT_FALSE(unpacker.includeMetaField());
 
     // Tests a spec with 'metaField' not in exclude list.
     std::set<std::string> excludeFields{};
-    auto specMetaExclude = BucketSpec{
-        kUserDefinedTimeName.toString(), kUserDefinedMetaName.toString(), std::move(excludeFields)};
+    auto specMetaExclude = BucketSpec{kUserDefinedTimeName.toString(),
+                                      kUserDefinedMetaName.toString(),
+                                      std::move(excludeFields),
+                                      BucketSpec::Behavior::kExclude};
+
     auto specMetaInclude = specMetaExclude;
-    unpacker.setBucketSpecAndBehavior(std::move(specMetaExclude),
-                                      BucketUnpacker::Behavior::kExclude);
+    specMetaInclude.setBehavior(BucketSpec::Behavior::kInclude);
+
+    unpacker.setBucketSpec(std::move(specMetaExclude));
     ASSERT_TRUE(unpacker.includeMetaField());
-    unpacker.setBucketSpecAndBehavior(std::move(specMetaInclude),
-                                      BucketUnpacker::Behavior::kInclude);
+    unpacker.setBucketSpec(std::move(specMetaInclude));
     ASSERT_FALSE(unpacker.includeMetaField());
 }
 
@@ -674,21 +679,25 @@ TEST_F(BucketUnpackerTest, DetermineIncludeTimeField) {
 })");
     std::set<std::string> unpackerFields{kUserDefinedTimeName.toString()};
     auto unpacker = makeBucketUnpacker(unpackerFields,
-                                       BucketUnpacker::Behavior::kInclude,
+                                       BucketSpec::Behavior::kInclude,
                                        std::move(bucket),
                                        kUserDefinedMetaName.toString());
 
     std::set<std::string> includeFields{kUserDefinedTimeName.toString()};
-    auto includeSpec = BucketSpec{
-        kUserDefinedTimeName.toString(), kUserDefinedMetaName.toString(), std::move(includeFields)};
+    auto includeSpec = BucketSpec{kUserDefinedTimeName.toString(),
+                                  kUserDefinedMetaName.toString(),
+                                  std::move(includeFields),
+                                  BucketSpec::Behavior::kInclude};
     // This calls determineIncludeTimeField.
-    unpacker.setBucketSpecAndBehavior(std::move(includeSpec), BucketUnpacker::Behavior::kInclude);
+    unpacker.setBucketSpec(std::move(includeSpec));
     ASSERT_TRUE(unpacker.includeTimeField());
 
     std::set<std::string> excludeFields{kUserDefinedTimeName.toString()};
-    auto excludeSpec = BucketSpec{
-        kUserDefinedTimeName.toString(), kUserDefinedMetaName.toString(), std::move(excludeFields)};
-    unpacker.setBucketSpecAndBehavior(std::move(excludeSpec), BucketUnpacker::Behavior::kExclude);
+    auto excludeSpec = BucketSpec{kUserDefinedTimeName.toString(),
+                                  kUserDefinedMetaName.toString(),
+                                  std::move(excludeFields),
+                                  BucketSpec::Behavior::kExclude};
+    unpacker.setBucketSpec(std::move(excludeSpec));
     ASSERT_FALSE(unpacker.includeTimeField());
 }
 
@@ -703,24 +712,26 @@ TEST_F(BucketUnpackerTest, DetermineIncludeFieldIncludeMode) {
                            {"data", Document{}}}
                       .toBson();
 
-    auto spec = BucketSpec{
-        kUserDefinedTimeName.toString(), kUserDefinedMetaName.toString(), std::move(fields)};
+    auto spec = BucketSpec{kUserDefinedTimeName.toString(),
+                           kUserDefinedMetaName.toString(),
+                           std::move(fields),
+                           BucketSpec::Behavior::kInclude};
 
     BucketUnpacker includeUnpacker;
-    includeUnpacker.setBucketSpecAndBehavior(std::move(spec), BucketUnpacker::Behavior::kInclude);
+    includeUnpacker.setBucketSpec(std::move(spec));
     // Need to call reset so that the private method calculateFieldsToIncludeExcludeDuringUnpack()
     // is called, and _unpackFieldsToIncludeExclude gets filled with fields.
     includeUnpacker.reset(std::move(bucket));
     // Now the spec knows which fields to include/exclude.
 
     ASSERT_TRUE(determineIncludeField(kUserDefinedTimeName,
-                                      BucketUnpacker::Behavior::kInclude,
+                                      BucketSpec::Behavior::kInclude,
                                       includeUnpacker.fieldsToIncludeExcludeDuringUnpack()));
     ASSERT_TRUE(determineIncludeField(includedMeasurementField,
-                                      BucketUnpacker::Behavior::kInclude,
+                                      BucketSpec::Behavior::kInclude,
                                       includeUnpacker.fieldsToIncludeExcludeDuringUnpack()));
     ASSERT_FALSE(determineIncludeField(excludedMeasurementField,
-                                       BucketUnpacker::Behavior::kInclude,
+                                       BucketSpec::Behavior::kInclude,
                                        includeUnpacker.fieldsToIncludeExcludeDuringUnpack()));
 }
 
@@ -735,21 +746,23 @@ TEST_F(BucketUnpackerTest, DetermineIncludeFieldExcludeMode) {
                            {"data", Document{}}}
                       .toBson();
 
-    auto spec = BucketSpec{
-        kUserDefinedTimeName.toString(), kUserDefinedMetaName.toString(), std::move(fields)};
+    auto spec = BucketSpec{kUserDefinedTimeName.toString(),
+                           kUserDefinedMetaName.toString(),
+                           std::move(fields),
+                           BucketSpec::Behavior::kExclude};
 
     BucketUnpacker excludeUnpacker;
-    excludeUnpacker.setBucketSpecAndBehavior(std::move(spec), BucketUnpacker::Behavior::kExclude);
+    excludeUnpacker.setBucketSpec(std::move(spec));
     excludeUnpacker.reset(std::move(bucket));
 
     ASSERT_FALSE(determineIncludeField(kUserDefinedTimeName,
-                                       BucketUnpacker::Behavior::kExclude,
+                                       BucketSpec::Behavior::kExclude,
                                        excludeUnpacker.fieldsToIncludeExcludeDuringUnpack()));
     ASSERT_FALSE(determineIncludeField(includedMeasurementField,
-                                       BucketUnpacker::Behavior::kExclude,
+                                       BucketSpec::Behavior::kExclude,
                                        excludeUnpacker.fieldsToIncludeExcludeDuringUnpack()));
     ASSERT_TRUE(determineIncludeField(excludedMeasurementField,
-                                      BucketUnpacker::Behavior::kExclude,
+                                      BucketSpec::Behavior::kExclude,
                                       excludeUnpacker.fieldsToIncludeExcludeDuringUnpack()));
 }
 
@@ -767,9 +780,11 @@ auto expectedTimestampObjSize(int32_t rowKeyOffset, int32_t n) {
 TEST_F(BucketUnpackerTest, ExtractSingleMeasurement) {
     std::set<std::string> fields{
         "_id", kUserDefinedMetaName.toString(), kUserDefinedTimeName.toString(), "a", "b"};
-    auto spec = BucketSpec{
-        kUserDefinedTimeName.toString(), kUserDefinedMetaName.toString(), std::move(fields)};
-    auto unpacker = BucketUnpacker{std::move(spec), BucketUnpacker::Behavior::kInclude};
+    auto spec = BucketSpec{kUserDefinedTimeName.toString(),
+                           kUserDefinedMetaName.toString(),
+                           std::move(fields),
+                           BucketSpec::Behavior::kInclude};
+    auto unpacker = BucketUnpacker{std::move(spec)};
 
     auto d1 = dateFromISOString("2020-02-17T00:00:00.000Z").getValue();
     auto d2 = dateFromISOString("2020-02-17T01:00:00.000Z").getValue();
@@ -812,9 +827,11 @@ TEST_F(BucketUnpackerTest, ExtractSingleMeasurement) {
 TEST_F(BucketUnpackerTest, ExtractSingleMeasurementSparse) {
     std::set<std::string> fields{
         "_id", kUserDefinedMetaName.toString(), kUserDefinedTimeName.toString(), "a", "b"};
-    auto spec = BucketSpec{
-        kUserDefinedTimeName.toString(), kUserDefinedMetaName.toString(), std::move(fields)};
-    auto unpacker = BucketUnpacker{std::move(spec), BucketUnpacker::Behavior::kInclude};
+    auto spec = BucketSpec{kUserDefinedTimeName.toString(),
+                           kUserDefinedMetaName.toString(),
+                           std::move(fields),
+                           BucketSpec::Behavior::kInclude};
+    auto unpacker = BucketUnpacker{std::move(spec)};
 
     auto d1 = dateFromISOString("2020-02-17T00:00:00.000Z").getValue();
     auto d2 = dateFromISOString("2020-02-17T01:00:00.000Z").getValue();
@@ -902,7 +919,7 @@ TEST_F(BucketUnpackerTest, TamperedCompressedCountLess) {
     auto modifiedCompressedBucket = modifyCompressedBucketElementCount(*compressedBucket, -1);
 
     auto unpacker = makeBucketUnpacker(std::move(fields),
-                                       BucketUnpacker::Behavior::kInclude,
+                                       BucketSpec::Behavior::kInclude,
                                        std::move(modifiedCompressedBucket),
                                        kUserDefinedMetaName.toString());
 
@@ -938,7 +955,7 @@ TEST_F(BucketUnpackerTest, TamperedCompressedCountMore) {
     auto modifiedCompressedBucket = modifyCompressedBucketElementCount(*compressedBucket, 1);
 
     auto unpacker = makeBucketUnpacker(std::move(fields),
-                                       BucketUnpacker::Behavior::kInclude,
+                                       BucketSpec::Behavior::kInclude,
                                        std::move(modifiedCompressedBucket),
                                        kUserDefinedMetaName.toString());
 
@@ -974,7 +991,7 @@ TEST_F(BucketUnpackerTest, TamperedCompressedCountMissing) {
     auto modifiedCompressedBucket = modifyCompressedBucketElementCount(*compressedBucket, 0);
 
     auto unpacker = makeBucketUnpacker(std::move(fields),
-                                       BucketUnpacker::Behavior::kInclude,
+                                       BucketSpec::Behavior::kInclude,
                                        std::move(modifiedCompressedBucket),
                                        kUserDefinedMetaName.toString());
 
@@ -1012,7 +1029,7 @@ TEST_F(BucketUnpackerTest, TamperedCompressedElementMismatchDataField) {
         modifyCompressedBucketRemoveLastInField(*compressedBucket, "a"_sd);
 
     auto unpacker = makeBucketUnpacker(std::move(fields),
-                                       BucketUnpacker::Behavior::kInclude,
+                                       BucketSpec::Behavior::kInclude,
                                        std::move(modifiedCompressedBucket),
                                        kUserDefinedMetaName.toString());
 
@@ -1048,7 +1065,7 @@ TEST_F(BucketUnpackerTest, TamperedCompressedElementMismatchTimeField) {
         modifyCompressedBucketRemoveLastInField(*compressedBucket, "time"_sd);
 
     auto unpacker = makeBucketUnpacker(std::move(fields),
-                                       BucketUnpacker::Behavior::kInclude,
+                                       BucketSpec::Behavior::kInclude,
                                        std::move(modifiedCompressedBucket),
                                        kUserDefinedMetaName.toString());
 
diff --git a/src/mongo/db/pipeline/document_source_internal_unpack_bucket.cpp b/src/mongo/db/pipeline/document_source_internal_unpack_bucket.cpp
index 89959bb0b49..c0b12367968 100644
--- a/src/mongo/db/pipeline/document_source_internal_unpack_bucket.cpp
+++ b/src/mongo/db/pipeline/document_source_internal_unpack_bucket.cpp
@@ -289,7 +289,6 @@ boost::intrusive_ptr<DocumentSource> DocumentSourceInternalUnpackBucket::createF
 
     // If neither "include" nor "exclude" is specified, the default is "exclude": [] and
     // if that's the case, no field will be added to 'bucketSpec.fieldSet' in the for-loop below.
-    BucketUnpacker::Behavior unpackerBehavior = BucketUnpacker::Behavior::kExclude;
     BucketSpec bucketSpec;
     auto hasIncludeExclude = false;
     auto hasTimeField = false;
@@ -323,8 +322,8 @@ boost::intrusive_ptr<DocumentSource> DocumentSourceInternalUnpackBucket::createF
                         field.find('.') == std::string::npos);
                 bucketSpec.addIncludeExcludeField(field);
             }
-            unpackerBehavior = fieldName == kInclude ? BucketUnpacker::Behavior::kInclude
-                                                     : BucketUnpacker::Behavior::kExclude;
+            bucketSpec.setBehavior(fieldName == kInclude ? BucketSpec::Behavior::kInclude
+                                                         : BucketSpec::Behavior::kExclude);
             hasIncludeExclude = true;
         } else if (fieldName == kAssumeNoMixedSchemaData) {
             uassert(6067202,
@@ -418,13 +417,12 @@ boost::intrusive_ptr<DocumentSource> DocumentSourceInternalUnpackBucket::createF
             "The $_internalUnpackBucket stage requires a bucketMaxSpanSeconds parameter",
             hasBucketMaxSpanSeconds);
 
-    return make_intrusive<DocumentSourceInternalUnpackBucket>(
-        expCtx,
-        BucketUnpacker{std::move(bucketSpec), unpackerBehavior},
-        bucketMaxSpanSeconds,
-        eventFilterBson,
-        wholeBucketFilterBson,
-        assumeClean);
+    return make_intrusive<DocumentSourceInternalUnpackBucket>(expCtx,
+                                                              BucketUnpacker{std::move(bucketSpec)},
+                                                              bucketMaxSpanSeconds,
+                                                              eventFilterBson,
+                                                              wholeBucketFilterBson,
+                                                              assumeClean);
 }
 
 boost::intrusive_ptr<DocumentSource> DocumentSourceInternalUnpackBucket::createFromBsonExternal(
@@ -471,7 +469,7 @@ boost::intrusive_ptr<DocumentSource> DocumentSourceInternalUnpackBucket::createF
 
     return make_intrusive<DocumentSourceInternalUnpackBucket>(
         expCtx,
-        BucketUnpacker{std::move(bucketSpec), BucketUnpacker::Behavior::kExclude},
+        BucketUnpacker{std::move(bucketSpec)},
         // Using the bucket span associated with the default granularity seconds.
         timeseries::getMaxSpanSecondsFromGranularity(BucketGranularityEnum::Seconds),
         assumeClean);
@@ -481,16 +479,16 @@ void DocumentSourceInternalUnpackBucket::serializeToArray(
     std::vector<Value>& array, boost::optional<ExplainOptions::Verbosity> explain) const {
     MutableDocument out;
     auto behavior =
-        _bucketUnpacker.behavior() == BucketUnpacker::Behavior::kInclude ? kInclude : kExclude;
+        _bucketUnpacker.behavior() == BucketSpec::Behavior::kInclude ? kInclude : kExclude;
     const auto& spec = _bucketUnpacker.bucketSpec();
     std::vector<Value> fields;
     for (auto&& field : spec.fieldSet()) {
         fields.emplace_back(field);
     }
     if (((_bucketUnpacker.includeMetaField() &&
-          _bucketUnpacker.behavior() == BucketUnpacker::Behavior::kInclude) ||
+          _bucketUnpacker.behavior() == BucketSpec::Behavior::kInclude) ||
          (!_bucketUnpacker.includeMetaField() &&
-          _bucketUnpacker.behavior() == BucketUnpacker::Behavior::kExclude && spec.metaField())) &&
+          _bucketUnpacker.behavior() == BucketSpec::Behavior::kExclude && spec.metaField())) &&
         std::find(spec.computedMetaProjFields().cbegin(),
                   spec.computedMetaProjFields().cend(),
                   *spec.metaField()) == spec.computedMetaProjFields().cend())
@@ -650,9 +648,8 @@ void DocumentSourceInternalUnpackBucket::internalizeProject(const BSONObj& proje
     // Update '_bucketUnpacker' state with the new fields and behavior.
     auto spec = _bucketUnpacker.bucketSpec();
     spec.setFieldSet(fields);
-    _bucketUnpacker.setBucketSpecAndBehavior(std::move(spec),
-                                             isInclusion ? BucketUnpacker::Behavior::kInclude
-                                                         : BucketUnpacker::Behavior::kExclude);
+    spec.setBehavior(isInclusion ? BucketSpec::Behavior::kInclude : BucketSpec::Behavior::kExclude);
+    _bucketUnpacker.setBucketSpec(std::move(spec));
 }
 
 std::pair<BSONObj, bool> DocumentSourceInternalUnpackBucket::extractOrBuildProjectToInternalize(
@@ -1220,10 +1217,10 @@ Pipeline::SourceContainer::iterator DocumentSourceInternalUnpackBucket::doOptimi
         // interested in $count.
         auto deps = getRestPipelineDependencies(itr, container);
         if (deps.hasNoRequirements()) {
-            _bucketUnpacker.setBucketSpecAndBehavior({_bucketUnpacker.bucketSpec().timeField(),
-                                                      _bucketUnpacker.bucketSpec().metaField(),
-                                                      {}},
-                                                     BucketUnpacker::Behavior::kInclude);
+            _bucketUnpacker.setBucketSpec({_bucketUnpacker.bucketSpec().timeField(),
+                                           _bucketUnpacker.bucketSpec().metaField(),
+                                           {},
+                                           BucketSpec::Behavior::kInclude});
 
             // Keep going for next optimization.
         }