SERVER-57562 Check if Collection or View exist on namespace before creating timeseries bucket collection.

2 files changed, 95 insertions, 2 deletions

diff --git a/jstests/core/timeseries/timeseries_create_collection.js b/jstests/core/timeseries/timeseries_create_collection.js
new file mode 100644
index 00000000000..48d0aee4bc7
--- /dev/null
+++ b/jstests/core/timeseries/timeseries_create_collection.js
@@ -0,0 +1,74 @@
+/**
+ * Tests basic create and drop timeseries Collection behavior. Also test that we fail with
+ * NamespaceExists when namespace is already used and that we don't leave orphaned bucket collection
+ * in that case.
+ *
+ * @tags: [
+ *     assumes_no_implicit_collection_creation_after_drop,
+ *     does_not_support_transactions,
+ *     requires_fcv_49,
+ *     requires_find_command,
+ *     requires_getmore,
+ * ]
+ */
+(function() {
+'use strict';
+
+const testDB = db.getSiblingDB(jsTestName());
+assert.commandWorked(testDB.dropDatabase());
+
+const timeFieldName = 'time';
+const coll = testDB.t;
+
+// Create a timeseries collection, listCollection should show view and bucket collection
+assert.commandWorked(
+    testDB.createCollection(coll.getName(), {timeseries: {timeField: timeFieldName}}));
+let collections = assert.commandWorked(testDB.runCommand({listCollections: 1})).cursor.firstBatch;
+jsTestLog('Checking listCollections result: ' + tojson(collections));
+assert(collections.find(entry => entry.name === 'system.buckets.' + coll.getName()));
+assert(collections.find(entry => entry.name === coll.getName()));
+
+// Drop timeseries collection, both view and bucket collection should be dropped
+coll.drop();
+collections = assert.commandWorked(testDB.runCommand({listCollections: 1})).cursor.firstBatch;
+jsTestLog('Checking listCollections result: ' + tojson(collections));
+assert.isnull(collections.find(entry => entry.name === 'system.buckets.' + coll.getName()));
+assert.isnull(collections.find(entry => entry.name === coll.getName()));
+
+// Create a regular collection on the same namespace and verify result
+assert.commandWorked(testDB.createCollection(coll.getName()));
+collections = assert.commandWorked(testDB.runCommand({listCollections: 1})).cursor.firstBatch;
+jsTestLog('Checking listCollections result: ' + tojson(collections));
+assert.isnull(collections.find(entry => entry.name === 'system.buckets.' + coll.getName()));
+assert(collections.find(entry => entry.name === coll.getName()));
+
+// Create timeseries collection when regular collection already exist on namespace. Command should
+// fail with NamespaceExists and no bucket collection should be created
+assert.commandFailedWithCode(
+    testDB.createCollection(coll.getName(), {timeseries: {timeField: timeFieldName}}),
+    ErrorCodes.NamespaceExists);
+collections = assert.commandWorked(testDB.runCommand({listCollections: 1})).cursor.firstBatch;
+jsTestLog('Checking listCollections result: ' + tojson(collections));
+assert.isnull(collections.find(entry => entry.name === 'system.buckets.' + coll.getName()));
+assert(collections.find(entry => entry.name === coll.getName()));
+coll.drop();
+
+// Create a regular view on the same namespace and verify result
+testDB.getCollection("other");
+assert.commandWorked(testDB.runCommand(
+    {create: coll.getName(), viewOn: "other", pipeline: [{$match: {field: "A"}}]}));
+collections = assert.commandWorked(testDB.runCommand({listCollections: 1})).cursor.firstBatch;
+jsTestLog('Checking listCollections result: ' + tojson(collections));
+assert.isnull(collections.find(entry => entry.name === 'system.buckets.' + coll.getName()));
+assert(collections.find(entry => entry.name === coll.getName()));
+
+// Create timeseries collection when view already exist on namespace. Command should fail with
+// NamespaceExists and no bucket collection should be created
+assert.commandFailedWithCode(
+    testDB.createCollection(coll.getName(), {timeseries: {timeField: timeFieldName}}),
+    ErrorCodes.NamespaceExists);
+collections = assert.commandWorked(testDB.runCommand({listCollections: 1})).cursor.firstBatch;
+jsTestLog('Checking listCollections result: ' + tojson(collections));
+assert.isnull(collections.find(entry => entry.name === 'system.buckets.' + coll.getName()));
+assert(collections.find(entry => entry.name === coll.getName()));
+})();
diff --git a/src/mongo/db/catalog/create_collection.cpp b/src/mongo/db/catalog/create_collection.cpp
index baef2264ecd..8ebd4f8c9ee 100644
--- a/src/mongo/db/catalog/create_collection.cpp
+++ b/src/mongo/db/catalog/create_collection.cpp
@@ -194,6 +194,27 @@ Status _createTimeseries(OperationContext* opCtx,
             AutoGetDb autoDb(opCtx, bucketsNs.db(), MODE_IX);
             Lock::CollectionLock bucketsCollLock(opCtx, bucketsNs, MODE_IX);
 
+            // Check if there already exist a Collection on the namespace we will later create a
+            // view on. We're not holding a Collection lock for this Collection so we may only check
+            // if the pointer is null or not. The answer may also change at any point after this
+            // call which is fine as we properly handle an orphaned bucket collection. This check is
+            // just here to prevent it from being created in the common case.
+            if (CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, ns)) {
+                return Status(ErrorCodes::NamespaceExists,
+                              str::stream() << "Collection already exists. NS: " << ns);
+            }
+
+            auto db = autoDb.ensureDbExists();
+            if (auto view = ViewCatalog::get(db)->lookup(opCtx, ns.ns()); view) {
+                if (view->timeseries()) {
+                    return Status(ErrorCodes::NamespaceExists,
+                                  str::stream()
+                                      << "A timeseries collection already exists. NS: " << ns);
+                }
+                return Status(ErrorCodes::NamespaceExists,
+                              str::stream() << "A view already exists. NS: " << ns);
+            }
+
             if (opCtx->writesAreReplicated() &&
                 !repl::ReplicationCoordinator::get(opCtx)->canAcceptWritesFor(opCtx, bucketsNs)) {
                 // Report the error with the user provided namespace
@@ -201,8 +222,6 @@ Status _createTimeseries(OperationContext* opCtx,
                               str::stream() << "Not primary while creating collection " << ns);
             }
 
-            auto db = autoDb.ensureDbExists();
-
             WriteUnitOfWork wuow(opCtx);
             AutoStatsTracker bucketsStatsTracker(
                 opCtx,