SERVER-66688 Build aggregation pipeline to identify candidate buckets given a time-series measurement

6 files changed, 355 insertions, 47 deletions

diff --git a/src/mongo/db/timeseries/SConscript b/src/mongo/db/timeseries/SConscript
index e46bf9b8177..3d0a6797307 100644
--- a/src/mongo/db/timeseries/SConscript
+++ b/src/mongo/db/timeseries/SConscript
@@ -32,6 +32,7 @@ env.Library(
         '$BUILD_DIR/mongo/bson/util/bson_column',
         '$BUILD_DIR/mongo/db/catalog/database_holder',
         '$BUILD_DIR/mongo/db/commands/server_status',
+        '$BUILD_DIR/mongo/db/dbdirectclient',
         '$BUILD_DIR/mongo/db/namespace_string',
         '$BUILD_DIR/mongo/db/server_options_core',
         '$BUILD_DIR/mongo/db/storage/storage_options',
@@ -150,6 +151,7 @@ env.CppUnitTest(
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/catalog/catalog_test_fixture',
+        '$BUILD_DIR/mongo/db/catalog/collection_crud',
         '$BUILD_DIR/mongo/db/catalog_raii',
         'bucket_catalog',
         'bucket_compression',
diff --git a/src/mongo/db/timeseries/bucket_catalog.cpp b/src/mongo/db/timeseries/bucket_catalog.cpp
index ac09c9e917f..7a0d51ad067 100644
--- a/src/mongo/db/timeseries/bucket_catalog.cpp
+++ b/src/mongo/db/timeseries/bucket_catalog.cpp
@@ -990,18 +990,14 @@ StatusWith<std::pair<BucketCatalog::BucketKey, Date_t>> BucketCatalog::_extractB
     const StringData::ComparatorInterface* comparator,
     const TimeseriesOptions& options,
     const BSONObj& doc) const {
-    auto timeElem = doc[options.getTimeField()];
-    if (!timeElem || BSONType::Date != timeElem.type()) {
-        return {ErrorCodes::BadValue,
-                str::stream() << "'" << options.getTimeField() << "' must be present and contain a "
-                              << "valid BSON UTC datetime value"};
+    auto swDocTimeAndMeta = timeseries::extractTimeAndMeta(doc, options);
+    if (!swDocTimeAndMeta.isOK()) {
+        return swDocTimeAndMeta.getStatus();
     }
-    auto time = timeElem.Date();
-
+    auto time = swDocTimeAndMeta.getValue().first;
     BSONElement metadata;
-    auto metaFieldName = options.getMetaField();
-    if (metaFieldName) {
-        metadata = doc[*metaFieldName];
+    if (auto metadataValue = swDocTimeAndMeta.getValue().second) {
+        metadata = *metadataValue;
     }
 
     // Buckets are spread across independently-lockable stripes to improve parallelism. We map a
diff --git a/src/mongo/db/timeseries/bucket_catalog_helpers.cpp b/src/mongo/db/timeseries/bucket_catalog_helpers.cpp
index 6a5d03e77bb..4b0c5cf2209 100644
--- a/src/mongo/db/timeseries/bucket_catalog_helpers.cpp
+++ b/src/mongo/db/timeseries/bucket_catalog_helpers.cpp
@@ -28,6 +28,7 @@
  */
 
 #include "mongo/db/timeseries/bucket_catalog_helpers.h"
+#include "mongo/db/dbdirectclient.h"
 #include "mongo/db/timeseries/timeseries_constants.h"
 #include "mongo/logv2/redaction.h"
 
@@ -54,6 +55,34 @@ StatusWith<std::pair<const BSONObj, const BSONObj>> extractMinAndMax(const BSONO
     return std::make_pair(minObj, maxObj);
 }
 
+BSONObj generateFindFilters(const Date_t& time,
+                            boost::optional<BSONElement> metadata,
+                            const std::string& controlMinTimePath,
+                            int64_t bucketMaxSpanSeconds) {
+    // The bucket must be uncompressed.
+    auto versionFilter = BSON(kControlVersionPath << kTimeseriesControlDefaultVersion);
+
+    // The bucket cannot be closed (aka open for new measurements).
+    auto closedFlagFilter =
+        BSON("$or" << BSON_ARRAY(BSON(kControlClosedPath << BSON("$exists" << false))
+                                 << BSON(kControlClosedPath << false)));
+
+    // The measurement meta field must match the bucket 'meta' field. If the field is not specified
+    // we can only insert into buckets which also do not have a meta field.
+    auto metaFieldFilter = (metadata && (*metadata).ok())
+        ? (*metadata).wrap(kBucketMetaFieldName)
+        : BSON(kBucketMetaFieldName << BSON("$exists" << false));
+
+    // (minimumTs <= measurementTs) && (minimumTs + maxSpanSeconds > measurementTs)
+    auto measurementMaxDifference = time - Seconds(bucketMaxSpanSeconds);
+    auto lowerBound = BSON(controlMinTimePath << BSON("$lte" << time));
+    auto upperBound = BSON(controlMinTimePath << BSON("$gt" << measurementMaxDifference));
+    auto timeRangeFilter = BSON("$and" << BSON_ARRAY(lowerBound << upperBound));
+
+    return BSON("$and" << BSON_ARRAY(versionFilter << closedFlagFilter << timeRangeFilter
+                                                   << metaFieldFilter));
+}
+
 }  // namespace
 
 StatusWith<MinMax> generateMinMaxFromBucketDoc(const BSONObj& bucketDoc,
@@ -88,4 +117,45 @@ StatusWith<Schema> generateSchemaFromBucketDoc(const BSONObj& bucketDoc,
     }
 }
 
+StatusWith<std::pair<Date_t, boost::optional<BSONElement>>> extractTimeAndMeta(
+    const BSONObj& doc, const TimeseriesOptions& options) {
+    auto timeElem = doc[options.getTimeField()];
+    if (!timeElem || BSONType::Date != timeElem.type()) {
+        return {ErrorCodes::BadValue,
+                str::stream() << "'" << options.getTimeField() << "' must be present and contain a "
+                              << "valid BSON UTC datetime value"};
+    }
+
+    auto time = timeElem.Date();
+    auto metaFieldName = options.getMetaField();
+
+    if (metaFieldName) {
+        return std::make_pair(time, doc[*metaFieldName]);
+    }
+    return std::make_pair(time, boost::none);
+}
+
+BSONObj findSuitableBucket(OperationContext* opCtx,
+                           const NamespaceString& bucketNss,
+                           const TimeseriesOptions& options,
+                           const BSONObj& measurementDoc) {
+    uassert(ErrorCodes::InvalidOptions,
+            "Missing bucketMaxSpanSeconds option.",
+            options.getBucketMaxSpanSeconds());
+
+    auto swDocTimeAndMeta = extractTimeAndMeta(measurementDoc, options);
+    if (!swDocTimeAndMeta.isOK()) {
+        return BSONObj();
+    }
+    auto [time, metadata] = swDocTimeAndMeta.getValue();
+    auto controlMinTimePath = kControlMinFieldNamePrefix.toString() + options.getTimeField();
+
+    // Generate all the filters we need to add to our 'find' query for a suitable bucket.
+    auto fullFilterExpression =
+        generateFindFilters(time, metadata, controlMinTimePath, *options.getBucketMaxSpanSeconds());
+
+    DBDirectClient client(opCtx);
+    return client.findOne(bucketNss, fullFilterExpression);
+}
+
 }  // namespace mongo::timeseries
diff --git a/src/mongo/db/timeseries/bucket_catalog_helpers.h b/src/mongo/db/timeseries/bucket_catalog_helpers.h
index 3c84124e5b2..bf1c8b2c949 100644
--- a/src/mongo/db/timeseries/bucket_catalog_helpers.h
+++ b/src/mongo/db/timeseries/bucket_catalog_helpers.h
@@ -33,6 +33,7 @@
 #include "mongo/base/string_data_comparator_interface.h"
 #include "mongo/bson/bsonobj.h"
 #include "mongo/db/timeseries/flat_bson.h"
+#include "mongo/db/timeseries/timeseries_options.h"
 
 namespace mongo::timeseries {
 
@@ -54,4 +55,37 @@ StatusWith<MinMax> generateMinMaxFromBucketDoc(const BSONObj& bucketDoc,
 StatusWith<Schema> generateSchemaFromBucketDoc(const BSONObj& bucketDoc,
                                                const StringData::ComparatorInterface* comparator);
 
+/**
+ * Extracts the time field of a measurement document and its meta field if it is present.
+ *
+ * Returns a bad status if the document is malformed.
+ */
+StatusWith<std::pair<Date_t, boost::optional<BSONElement>>> extractTimeAndMeta(
+    const BSONObj& doc, const TimeseriesOptions& options);
+
+/**
+ * Executes a 'find' query on the timeseries bucket collection to find a bucket eligible to
+ * receive a new measurement specified by a document's metadata and timestamp (measurementTs).
+ *
+ * A bucket is deemed suitable for the new measurement iff:
+ * i.   the bucket is uncompressed and not closed
+ * ii.  the meta fields match
+ * iii. the measurementTs is within the allowed time span for the bucket
+ *
+ * {$and:
+ *       [{"control.version":1},
+ *       {$or: [{"control.closed":{$exists:false}},
+ *              {"control.closed":false}]
+ *       },
+ *       {"meta":<metaValue>},
+ *       {$and: [{"control.min.time":{$lte:<measurementTs>}},
+ *               {"control.min.time":{$gt:<measurementTs - maxSpanSeconds>}}]
+ *       }]
+ * }
+ */
+BSONObj findSuitableBucket(OperationContext* opCtx,
+                           const NamespaceString& bucketNss,
+                           const TimeseriesOptions& options,
+                           const BSONObj& measurementDoc);
+
 }  // namespace mongo::timeseries
diff --git a/src/mongo/db/timeseries/bucket_catalog_helpers_test.cpp b/src/mongo/db/timeseries/bucket_catalog_helpers_test.cpp
index 97f29ddf6e5..bf2909ba9d3 100644
--- a/src/mongo/db/timeseries/bucket_catalog_helpers_test.cpp
+++ b/src/mongo/db/timeseries/bucket_catalog_helpers_test.cpp
@@ -29,6 +29,7 @@
 
 #include "mongo/bson/json.h"
 #include "mongo/db/catalog/catalog_test_fixture.h"
+#include "mongo/db/catalog/collection_write_path.h"
 #include "mongo/db/catalog/create_collection.h"
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/timeseries/bucket_catalog_helpers.h"
@@ -40,14 +41,32 @@ namespace {
 
 const NamespaceString kNss = NamespaceString("test.ts");
 
-class BucketCatalogHelpersTest : public CatalogTestFixture {};
+class BucketCatalogHelpersTest : public CatalogTestFixture {
+protected:
+    StringData _timeField = "time";
+    StringData _metaField = "mm";
+
+    void _insertIntoBucketColl(const BSONObj& bucketDoc);
+};
+
+void BucketCatalogHelpersTest::_insertIntoBucketColl(const BSONObj& bucketDoc) {
+    AutoGetCollection autoColl(operationContext(), kNss.makeTimeseriesBucketsNamespace(), MODE_IX);
+    const CollectionPtr& coll = autoColl.getCollection();
+    OpDebug* const nullOpDebug = nullptr;
+
+    {
+        WriteUnitOfWork wuow(operationContext());
+        ASSERT_OK(collection_internal::insertDocument(
+            operationContext(), coll, InsertStatement(bucketDoc), nullOpDebug));
+        wuow.commit();
+    }
+}
 
 TEST_F(BucketCatalogHelpersTest, GenerateMinMaxBadBucketDocumentsTest) {
-    ASSERT_OK(createCollection(operationContext(),
-                               kNss.db().toString(),
-                               BSON("create" << kNss.coll() << "timeseries"
-                                             << BSON("timeField"
-                                                     << "time"))));
+    ASSERT_OK(createCollection(
+        operationContext(),
+        kNss.db().toString(),
+        BSON("create" << kNss.coll() << "timeseries" << BSON("timeField" << _timeField))));
 
     AutoGetCollection autoColl(operationContext(), kNss.makeTimeseriesBucketsNamespace(), MODE_IS);
     const CollatorInterface* collator = autoColl->getDefaultCollator();
@@ -69,11 +88,10 @@ TEST_F(BucketCatalogHelpersTest, GenerateMinMaxBadBucketDocumentsTest) {
 }
 
 TEST_F(BucketCatalogHelpersTest, GenerateMinMaxTest) {
-    ASSERT_OK(createCollection(operationContext(),
-                               kNss.db().toString(),
-                               BSON("create" << kNss.coll() << "timeseries"
-                                             << BSON("timeField"
-                                                     << "time"))));
+    ASSERT_OK(createCollection(
+        operationContext(),
+        kNss.db().toString(),
+        BSON("create" << kNss.coll() << "timeseries" << BSON("timeField" << _timeField))));
 
     AutoGetCollection autoColl(operationContext(), kNss.makeTimeseriesBucketsNamespace(), MODE_IS);
     const CollatorInterface* collator = autoColl->getDefaultCollator();
@@ -105,9 +123,7 @@ TEST_F(BucketCatalogHelpersTest, GenerateMinMaxWithLowerCaseFirstCollationTest)
     ASSERT_OK(createCollection(operationContext(),
                                kNss.db().toString(),
                                BSON("create" << kNss.coll() << "timeseries"
-                                             << BSON("timeField"
-                                                     << "time")
-                                             << "collation"
+                                             << BSON("timeField" << _timeField) << "collation"
                                              << BSON("locale"
                                                      << "en_US"
                                                      << "caseFirst"
@@ -133,9 +149,7 @@ TEST_F(BucketCatalogHelpersTest, GenerateMinMaxWithUpperCaseFirstCollationTest)
     ASSERT_OK(createCollection(operationContext(),
                                kNss.db().toString(),
                                BSON("create" << kNss.coll() << "timeseries"
-                                             << BSON("timeField"
-                                                     << "time")
-                                             << "collation"
+                                             << BSON("timeField" << _timeField) << "collation"
                                              << BSON("locale"
                                                      << "en_US"
                                                      << "caseFirst"
@@ -158,11 +172,10 @@ TEST_F(BucketCatalogHelpersTest, GenerateMinMaxWithUpperCaseFirstCollationTest)
 }
 
 TEST_F(BucketCatalogHelpersTest, GenerateMinMaxSucceedsWithMixedSchemaBucketDocumentTest) {
-    ASSERT_OK(createCollection(operationContext(),
-                               kNss.db().toString(),
-                               BSON("create" << kNss.coll() << "timeseries"
-                                             << BSON("timeField"
-                                                     << "time"))));
+    ASSERT_OK(createCollection(
+        operationContext(),
+        kNss.db().toString(),
+        BSON("create" << kNss.coll() << "timeseries" << BSON("timeField" << _timeField))));
 
     AutoGetCollection autoColl(operationContext(), kNss.makeTimeseriesBucketsNamespace(), MODE_IS);
     const CollatorInterface* collator = autoColl->getDefaultCollator();
@@ -180,11 +193,10 @@ TEST_F(BucketCatalogHelpersTest, GenerateMinMaxSucceedsWithMixedSchemaBucketDocu
 }
 
 TEST_F(BucketCatalogHelpersTest, GenerateSchemaFailsWithMixedSchemaBucketDocumentTest) {
-    ASSERT_OK(createCollection(operationContext(),
-                               kNss.db().toString(),
-                               BSON("create" << kNss.coll() << "timeseries"
-                                             << BSON("timeField"
-                                                     << "time"))));
+    ASSERT_OK(createCollection(
+        operationContext(),
+        kNss.db().toString(),
+        BSON("create" << kNss.coll() << "timeseries" << BSON("timeField" << _timeField))));
 
     AutoGetCollection autoColl(operationContext(), kNss.makeTimeseriesBucketsNamespace(), MODE_IS);
     const CollatorInterface* collator = autoColl->getDefaultCollator();
@@ -202,11 +214,10 @@ TEST_F(BucketCatalogHelpersTest, GenerateSchemaFailsWithMixedSchemaBucketDocumen
 }
 
 TEST_F(BucketCatalogHelpersTest, GenerateSchemaWithInvalidMeasurementsTest) {
-    ASSERT_OK(createCollection(operationContext(),
-                               kNss.db().toString(),
-                               BSON("create" << kNss.coll() << "timeseries"
-                                             << BSON("timeField"
-                                                     << "time"))));
+    ASSERT_OK(createCollection(
+        operationContext(),
+        kNss.db().toString(),
+        BSON("create" << kNss.coll() << "timeseries" << BSON("timeField" << _timeField))));
 
     AutoGetCollection autoColl(operationContext(), kNss.makeTimeseriesBucketsNamespace(), MODE_IS);
     const CollatorInterface* collator = autoColl->getDefaultCollator();
@@ -248,11 +259,10 @@ TEST_F(BucketCatalogHelpersTest, GenerateSchemaWithInvalidMeasurementsTest) {
 }
 
 TEST_F(BucketCatalogHelpersTest, GenerateSchemaWithValidMeasurementsTest) {
-    ASSERT_OK(createCollection(operationContext(),
-                               kNss.db().toString(),
-                               BSON("create" << kNss.coll() << "timeseries"
-                                             << BSON("timeField"
-                                                     << "time"))));
+    ASSERT_OK(createCollection(
+        operationContext(),
+        kNss.db().toString(),
+        BSON("create" << kNss.coll() << "timeseries" << BSON("timeField" << _timeField))));
 
     AutoGetCollection autoColl(operationContext(), kNss.makeTimeseriesBucketsNamespace(), MODE_IS);
     const CollatorInterface* collator = autoColl->getDefaultCollator();
@@ -283,5 +293,197 @@ TEST_F(BucketCatalogHelpersTest, GenerateSchemaWithValidMeasurementsTest) {
     }
 }
 
+TEST_F(BucketCatalogHelpersTest, FindSuitableBucketForMeasurements) {
+    ASSERT_OK(createCollection(
+        operationContext(),
+        kNss.db().toString(),
+        BSON("create" << kNss.coll() << "timeseries"
+                      << BSON("timeField" << _timeField << "metaField" << _metaField))));
+
+    AutoGetCollection autoColl(operationContext(), kNss.makeTimeseriesBucketsNamespace(), MODE_IX);
+    ASSERT(autoColl->getTimeseriesOptions() && autoColl->getTimeseriesOptions()->getMetaField());
+
+    auto tsOptions = *autoColl->getTimeseriesOptions();
+    auto metaFieldName = *tsOptions.getMetaField();
+
+    std::vector<BSONObj> bucketDocs = {mongo::fromjson(
+                                           R"({
+            "_id":{"$oid":"62e7e6ec27c28d338ab29200"},
+            "control":{"version":1,"min":{"_id":1,"time":{"$date":"2021-08-01T11:00:00Z"},"a":1},
+                                   "max":{"_id":3,"time":{"$date":"2021-08-01T12:00:00Z"},"a":3},
+                       "closed":false},
+            "meta":1,
+            "data":{"time":{"0":{"$date":"2021-08-01T11:00:00Z"},
+                            "1":{"$date":"2021-08-01T11:00:00Z"},
+                            "2":{"$date":"2021-08-01T11:00:00Z"}},
+                    "a":{"0":1,"1":2,"2":3}}})"),
+                                       mongo::fromjson(
+                                           R"(
+            {"_id":{"$oid":"62e7eee4f33f295800073138"},
+            "control":{"version":1,"min":{"_id":7,"time":{"$date":"2022-08-01T12:00:00Z"},"a":1},
+                                   "max":{"_id":10,"time":{"$date":"2022-08-01T13:00:00Z"},"a":3}},
+            "meta":2,
+            "data":{"time":{"0":{"$date":"2022-08-01T12:00:00Z"},
+                            "1":{"$date":"2022-08-01T12:00:00Z"},
+                            "2":{"$date":"2022-08-01T12:00:00Z"}},
+                    "a":{"0":1,"1":2,"2":3}}})"),
+                                       mongo::fromjson(
+                                           R"({
+            "_id":{"$oid":"629e1e680958e279dc29a517"},
+            "control":{"version":1,"min":{"_id":7,"time":{"$date":"2023-08-01T13:00:00Z"},"a":1},
+                                   "max":{"_id":10,"time":{"$date":"2023-08-01T14:00:00Z"},"a":3},
+                       "closed":false},
+            "meta":3,
+            "data":{"time":{"0":{"$date":"2023-08-01T13:00:00Z"},
+                            "1":{"$date":"2023-08-01T13:00:00Z"},
+                            "2":{"$date":"2023-08-01T13:00:00Z"}},
+                    "a":{"0":1,"1":2,"2":3}}})")};
+
+    // Insert bucket documents into the system.buckets collection.
+    for (const auto& doc : bucketDocs) {
+        _insertIntoBucketColl(doc);
+    }
+
+    auto time1 = dateFromISOString("2021-08-01T11:30:00Z");
+    auto time2 = dateFromISOString("2022-08-01T12:30:00Z");
+    auto time3 = dateFromISOString("2023-08-01T13:30:00Z");
+    std::vector<BSONObj> docsWithSuitableBuckets = {
+        BSON("_id" << 1 << _timeField << time1.getValue() << _metaField << 1),
+        BSON("_id" << 2 << _timeField << time2.getValue() << _metaField << 2),
+        BSON("_id" << 3 << _timeField << time3.getValue() << _metaField << 3)};
+
+    // Iterate through the measurement documents and verify that we can find a suitable bucket to
+    // insert into.
+    for (size_t i = 0; i < docsWithSuitableBuckets.size(); ++i) {
+        const auto& doc = docsWithSuitableBuckets[i];
+        auto result = timeseries::findSuitableBucket(
+            operationContext(), kNss.makeTimeseriesBucketsNamespace(), tsOptions, doc);
+        ASSERT_FALSE(result.isEmpty());
+        ASSERT_EQ(bucketDocs[i]["_id"].OID(), result["_id"].OID());
+    }
+
+    // Verify that documents without a meta field are only eligible to be inserted into buckets with
+    // no meta field specified.
+    {
+        std::vector<BSONObj> docsWithOutMeta;
+        for (auto doc : docsWithSuitableBuckets) {
+            docsWithOutMeta.push_back(doc.removeField(_metaField));
+        }
+
+        for (size_t i = 0; i < docsWithOutMeta.size(); ++i) {
+            const auto& doc = docsWithOutMeta[i];
+            auto result = timeseries::findSuitableBucket(
+                operationContext(), kNss.makeTimeseriesBucketsNamespace(), tsOptions, doc);
+            ASSERT(result.isEmpty());
+        }
+
+        // Verify that a document with no meta field is suitable with a valid bucket also missing
+        // the meta field.
+        auto metalessDoc = BSON("_id" << 4 << _timeField << time3.getValue());
+        auto metalessBucket = mongo::fromjson(
+            R"({
+            "_id":{"$oid":"629e1e680958e279dc29a518"},
+            "control":{"version":1,"min":{"_id":7,"time":{"$date":"2023-08-01T13:00:00Z"},"a":1},
+                                   "max":{"_id":10,"time":{"$date":"2023-08-01T14:00:00Z"},"a":3},
+                       "closed":false},
+            "data":{"time":{"0":{"$date":"2023-08-01T13:00:00Z"},
+                            "1":{"$date":"2023-08-01T13:00:00Z"},
+                            "2":{"$date":"2023-08-01T13:00:00Z"}},
+                    "a":{"0":1,"1":2,"2":3}}})");
+        _insertIntoBucketColl(metalessBucket);
+
+        auto result = timeseries::findSuitableBucket(
+            operationContext(), kNss.makeTimeseriesBucketsNamespace(), tsOptions, metalessDoc);
+        ASSERT_FALSE(result.isEmpty());
+        ASSERT_EQ(metalessBucket["_id"].OID(), result["_id"].OID());
+    }
+
+    std::vector<BSONObj> docsWithoutSuitableBuckets = {
+        // Mismatching time field with corresponding bucket meta field.
+        BSON("_id" << 1 << _timeField << Date_t::now() << _metaField << 1),
+        // Mismatching meta field with corresponding bucket time range.
+        BSON("_id" << 2 << _timeField << time2.getValue() << _metaField << 100000)};
+
+    // Without a matching meta field or a time within a buckets time range, we should not find any
+    // suitable buckets.
+    for (const auto& doc : docsWithoutSuitableBuckets) {
+        BSONObj measurementMeta =
+            (doc.hasField(metaFieldName)) ? doc.getField(metaFieldName).wrap() : BSONObj();
+        auto result = timeseries::findSuitableBucket(
+            operationContext(), kNss.makeTimeseriesBucketsNamespace(), tsOptions, doc);
+        ASSERT(result.isEmpty());
+    }
+}
+
+TEST_F(BucketCatalogHelpersTest, IncompatibleBucketsForNewMeasurements) {
+    ASSERT_OK(createCollection(
+        operationContext(),
+        kNss.db().toString(),
+        BSON("create" << kNss.coll() << "timeseries"
+                      << BSON("timeField" << _timeField << "metaField" << _metaField))));
+
+    AutoGetCollection autoColl(operationContext(), kNss.makeTimeseriesBucketsNamespace(), MODE_IX);
+    ASSERT(autoColl->getTimeseriesOptions() && autoColl->getTimeseriesOptions()->getMetaField());
+    auto tsOptions = *autoColl->getTimeseriesOptions();
+
+    std::vector<BSONObj> bucketDocs = {// control.version indicates bucket is compressed.
+                                       mongo::fromjson(
+                                           R"({
+            "_id":{"$oid":"62e7e6ec27c28d338ab29200"},
+            "control":{"version":2,"min":{"_id":1,"time":{"$date":"2021-08-01T11:00:00Z"},"a":1},
+                                   "max":{"_id":3,"time":{"$date":"2021-08-01T12:00:00Z"},"a":3}},
+            "meta":1,
+            "data":{"time":{"0":{"$date":"2021-08-01T11:00:00Z"},
+                            "1":{"$date":"2021-08-01T11:00:00Z"},
+                            "2":{"$date":"2021-08-01T11:00:00Z"}},
+                    "a":{"0":1,"1":2,"2":3}}})"),
+                                       // control.closed flag is true.
+                                       mongo::fromjson(
+                                           R"(
+            {"_id":{"$oid":"62e7eee4f33f295800073138"},
+            "control":{"version":1,"min":{"_id":7,"time":{"$date":"2022-08-01T12:00:00Z"},"a":1},
+                                   "max":{"_id":10,"time":{"$date":"2022-08-01T13:00:00Z"},"a":3},
+                       "closed":true},
+            "meta":2,
+            "data":{"time":{"0":{"$date":"2022-08-01T12:00:00Z"},
+                            "1":{"$date":"2022-08-01T12:00:00Z"},
+                            "2":{"$date":"2022-08-01T12:00:00Z"}},
+                    "a":{"0":1,"1":2,"2":3}}})"),
+                                       // Compressed bucket with closed flag set.
+                                       mongo::fromjson(
+                                           R"({
+            "_id":{"$oid":"629e1e680958e279dc29a517"},
+            "control":{"version":2,"min":{"_id":7,"time":{"$date":"2023-08-01T13:00:00Z"},"a":1},
+                                   "max":{"_id":10,"time":{"$date":"2023-08-01T14:00:00Z"},"a":3},
+                       "closed":true},
+            "meta":3,
+            "data":{"time":{"0":{"$date":"2023-08-01T13:00:00Z"},
+                            "1":{"$date":"2023-08-01T13:00:00Z"},
+                            "2":{"$date":"2023-08-01T13:00:00Z"}},
+                    "a":{"0":1,"1":2,"2":3}}})")};
+
+    // Insert bucket documents into the system.buckets collection.
+    for (const auto& doc : bucketDocs) {
+        _insertIntoBucketColl(doc);
+    }
+
+    auto time1 = dateFromISOString("2021-08-01T11:30:00Z");
+    auto time2 = dateFromISOString("2022-08-01T12:30:00Z");
+    auto time3 = dateFromISOString("2023-08-01T13:30:00Z");
+    std::vector<BSONObj> validMeasurementDocs = {
+        BSON("_id" << 1 << _timeField << time1.getValue() << _metaField << 1),
+        BSON("_id" << 2 << _timeField << time2.getValue() << _metaField << 2),
+        BSON("_id" << 3 << _timeField << time3.getValue() << _metaField << 3)};
+
+    // Verify that even with matching meta fields and buckets with acceptable time ranges, if the
+    // bucket is compressed and/or closed, we should not see it as a candid bucket for future
+    // inserts.
+    for (const auto& doc : validMeasurementDocs) {
+        auto result = timeseries::findSuitableBucket(
+            operationContext(), kNss.makeTimeseriesBucketsNamespace(), tsOptions, doc);
+        ASSERT(result.isEmpty());
+    }
+}
+
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/timeseries/timeseries_constants.h b/src/mongo/db/timeseries/timeseries_constants.h
index be7d9a368f2..959cef0ec3b 100644
--- a/src/mongo/db/timeseries/timeseries_constants.h
+++ b/src/mongo/db/timeseries/timeseries_constants.h
@@ -62,6 +62,10 @@ static constexpr StringData kPartialFilterExpressionFieldName = "partialFilterEx
 static constexpr int kTimeseriesControlDefaultVersion = 1;
 static constexpr int kTimeseriesControlCompressedVersion = 2;
 
+// These are hard-coded control object subfields.
+static constexpr StringData kControlVersionPath = "control.version"_sd;
+static constexpr StringData kControlClosedPath = "control.closed"_sd;
+
 static const StringDataSet kAllowedCollectionCreationOptions{
     CreateCommand::kStorageEngineFieldName,
     CreateCommand::kIndexOptionDefaultsFieldName,