summaryrefslogtreecommitdiff
path: root/src/mongo/db/commands/write_commands.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/mongo/db/commands/write_commands.cpp')
-rw-r--r--src/mongo/db/commands/write_commands.cpp436
1 files changed, 157 insertions, 279 deletions
diff --git a/src/mongo/db/commands/write_commands.cpp b/src/mongo/db/commands/write_commands.cpp
index 7196c905f1b..0254baca47d 100644
--- a/src/mongo/db/commands/write_commands.cpp
+++ b/src/mongo/db/commands/write_commands.cpp
@@ -30,13 +30,11 @@
#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kDefault
#include "mongo/base/checked_cast.h"
-#include "mongo/base/error_codes.h"
#include "mongo/bson/bsonobjbuilder.h"
#include "mongo/bson/mutable/document.h"
#include "mongo/bson/mutable/element.h"
#include "mongo/db/catalog/collection_catalog.h"
#include "mongo/db/catalog/collection_operation_source.h"
-#include "mongo/db/catalog/collection_uuid_mismatch.h"
#include "mongo/db/catalog/database_holder.h"
#include "mongo/db/catalog/document_validation.h"
#include "mongo/db/client.h"
@@ -75,7 +73,6 @@
#include "mongo/db/timeseries/bucket_catalog.h"
#include "mongo/db/timeseries/bucket_compression.h"
#include "mongo/db/timeseries/timeseries_constants.h"
-#include "mongo/db/timeseries/timeseries_extended_range.h"
#include "mongo/db/timeseries/timeseries_options.h"
#include "mongo/db/timeseries/timeseries_stats.h"
#include "mongo/db/transaction_participant.h"
@@ -139,26 +136,9 @@ bool isTimeseries(OperationContext* opCtx, const Request& request) {
// collection does not yet exist, this check may return false unnecessarily. As a result, an
// insert attempt into the time-series namespace will either succeed or fail, depending on who
// wins the race.
- // Hold reference to the catalog for collection lookup without locks to be safe.
- auto catalog = CollectionCatalog::get(opCtx);
- auto coll = catalog->lookupCollectionByNamespace(opCtx, bucketNss);
- if (!coll) {
- return false;
- }
-
- if (auto options = coll->getTimeseriesOptions()) {
- uassert(ErrorCodes::InvalidOptions,
- "Time-series buckets collection is not clustered",
- coll->isClustered());
-
- uassert(ErrorCodes::InvalidOptions,
- "Time-series buckets collection is missing bucketMaxSpanSeconds",
- options->getBucketMaxSpanSeconds());
-
- return true;
- }
-
- return false;
+ return CollectionCatalog::get(opCtx)
+ ->lookupCollectionByNamespaceForRead(opCtx, bucketNss)
+ .get();
}
NamespaceString makeTimeseriesBucketsNamespace(const NamespaceString& nss) {
@@ -344,22 +324,6 @@ boost::optional<std::pair<Status, bool>> checkFailUnorderedTimeseriesInsertFailP
return boost::none;
}
-boost::optional<write_ops::WriteError> generateErrorNoTenantMigration(OperationContext* opCtx,
- const Status& status,
- int index,
- size_t numErrors) {
- constexpr size_t kMaxErrorReasonsToReport = 1;
- constexpr size_t kMaxErrorSizeToReportAfterMaxReasonsReached = 1024 * 1024;
-
- if (numErrors > kMaxErrorReasonsToReport) {
- size_t errorSize = status.reason().size();
- if (errorSize > kMaxErrorSizeToReportAfterMaxReasonsReached)
- return write_ops::WriteError(index, status.withReason(""));
- }
-
- return write_ops::WriteError(index, status);
-}
-
boost::optional<write_ops::WriteError> generateError(OperationContext* opCtx,
const Status& status,
int index,
@@ -368,16 +332,18 @@ boost::optional<write_ops::WriteError> generateError(OperationContext* opCtx,
return boost::none;
}
+ boost::optional<Status> overwrittenStatus;
+
if (status == ErrorCodes::TenantMigrationConflict) {
hangWriteBeforeWaitingForMigrationDecision.pauseWhileSet(opCtx);
- Status overwrittenStatus =
- tenant_migration_access_blocker::handleTenantMigrationConflict(opCtx, status);
+ overwrittenStatus.emplace(
+ tenant_migration_access_blocker::handleTenantMigrationConflict(opCtx, status));
// Interruption errors encountered during batch execution fail the entire batch, so throw on
// such errors here for consistency.
- if (ErrorCodes::isInterruption(overwrittenStatus)) {
- uassertStatusOK(overwrittenStatus);
+ if (ErrorCodes::isInterruption(*overwrittenStatus)) {
+ uassertStatusOK(*overwrittenStatus);
}
// Tenant migration errors, similarly to migration errors consume too much space in the
@@ -385,13 +351,25 @@ boost::optional<write_ops::WriteError> generateError(OperationContext* opCtx,
// 'handleTenantMigrationConflict' above replaces the original status, we need to manually
// truncate the new reason if the original 'status' was also truncated.
if (status.reason().empty()) {
- overwrittenStatus = overwrittenStatus.withReason("");
+ overwrittenStatus = overwrittenStatus->withReason("");
}
+ }
- return generateErrorNoTenantMigration(opCtx, overwrittenStatus, index, numErrors);
+ constexpr size_t kMaxErrorReasonsToReport = 1;
+ constexpr size_t kMaxErrorSizeToReportAfterMaxReasonsReached = 1024 * 1024;
+
+ if (numErrors > kMaxErrorReasonsToReport) {
+ size_t errorSize =
+ overwrittenStatus ? overwrittenStatus->reason().size() : status.reason().size();
+ if (errorSize > kMaxErrorSizeToReportAfterMaxReasonsReached)
+ overwrittenStatus =
+ overwrittenStatus ? overwrittenStatus->withReason("") : status.withReason("");
}
- return generateErrorNoTenantMigration(opCtx, status, index, numErrors);
+ if (overwrittenStatus)
+ return write_ops::WriteError(index, std::move(*overwrittenStatus));
+ else
+ return write_ops::WriteError(index, status);
}
template <typename T>
@@ -435,7 +413,6 @@ void populateReply(OperationContext* opCtx,
const auto& lastResult = result.results.back();
if (lastResult == ErrorCodes::StaleDbVersion ||
- lastResult == ErrorCodes::ShardCannotRefreshDueToLocksHeld ||
ErrorCodes::isStaleShardVersionError(lastResult.getStatus()) ||
ErrorCodes::isTenantMigrationError(lastResult.getStatus())) {
// For ordered:false commands we need to duplicate these error results for all ops
@@ -550,11 +527,6 @@ public:
}
write_ops::InsertCommandReply typedRun(OperationContext* opCtx) final try {
- // On debug builds, verify that the estimated size of the insert command is at least as
- // large as the size of the actual, serialized insert command. This ensures that the
- // logic which estimates the size of insert commands is correct.
- dassert(write_ops::verifySizeEstimate(request(), &unparsedRequest()));
-
transactionChecks(opCtx, ns());
if (request().getEncryptionInformation().has_value() &&
@@ -726,16 +698,16 @@ public:
OperationSource::kTimeseriesInsert));
}
- void _performTimeseriesBucketCompression(
+ TimeseriesSingleWriteResult _performTimeseriesBucketCompression(
OperationContext* opCtx, const BucketCatalog::ClosedBucket& closedBucket) const {
if (!feature_flags::gTimeseriesBucketCompression.isEnabled(
serverGlobalParams.featureCompatibility)) {
- return;
+ return {SingleWriteResult(), true};
}
// Buckets with just a single measurement is not worth compressing.
if (closedBucket.numMeasurements <= 1) {
- return;
+ return {SingleWriteResult(), true};
}
bool validateCompression = gValidateTimeseriesCompression.load();
@@ -773,8 +745,8 @@ public:
auto compressionOp =
_makeTimeseriesCompressionOp(opCtx, closedBucket.bucketId, bucketCompressionFunc);
- auto result = _getTimeseriesSingleWriteResult(write_ops_exec::performUpdates(
- opCtx, compressionOp, OperationSource::kTimeseriesBucketCompression));
+ auto result = _getTimeseriesSingleWriteResult(
+ write_ops_exec::performUpdates(opCtx, compressionOp, OperationSource::kStandard));
// Report stats, if we fail before running the transform function then just skip
// reporting.
@@ -789,6 +761,8 @@ public:
stats.onBucketClosed(*beforeSize, compressionStats);
}
}
+
+ return result;
}
/**
@@ -802,8 +776,7 @@ public:
std::vector<write_ops::WriteError>* errors,
boost::optional<repl::OpTime>* opTime,
boost::optional<OID>* electionId,
- std::vector<size_t>* docsToRetry,
- absl::flat_hash_map<int, int>& retryAttemptsForDup) const try {
+ std::vector<size_t>* docsToRetry) const try {
auto& bucketCatalog = BucketCatalog::get(opCtx);
auto metadata = bucketCatalog.getMetadata(batch->bucket());
@@ -823,18 +796,9 @@ public:
_performTimeseriesInsert(opCtx, batch, metadata, std::move(stmtIds));
if (auto error =
generateError(opCtx, output.result, start + index, errors->size())) {
- bool canContinue = output.canContinue;
- // Automatically attempts to retry on DuplicateKey error.
- if (error->getStatus().code() == ErrorCodes::DuplicateKey &&
- retryAttemptsForDup[index]++ <
- gTimeseriesInsertMaxRetriesOnDuplicates.load()) {
- docsToRetry->push_back(index);
- canContinue = true;
- } else {
- errors->emplace_back(std::move(*error));
- }
- BucketCatalog::get(opCtx).abort(batch, output.result.getStatus());
- return canContinue;
+ errors->emplace_back(std::move(*error));
+ bucketCatalog.abort(batch, output.result.getStatus());
+ return output.canContinue;
}
invariant(output.result.getValue().getN() == 1,
@@ -864,7 +828,12 @@ public:
if (closedBucket) {
// If this write closed a bucket, compress the bucket
- _performTimeseriesBucketCompression(opCtx, *closedBucket);
+ auto output = _performTimeseriesBucketCompression(opCtx, *closedBucket);
+ if (auto error =
+ generateError(opCtx, output.result, start + index, errors->size())) {
+ errors->emplace_back(std::move(*error));
+ return output.canContinue;
+ }
}
return true;
} catch (const DBException& ex) {
@@ -872,12 +841,19 @@ public:
throw;
}
- bool _commitTimeseriesBucketsAtomically(OperationContext* opCtx,
- TimeseriesBatches* batches,
- TimeseriesStmtIds&& stmtIds,
- std::vector<write_ops::WriteError>* errors,
- boost::optional<repl::OpTime>* opTime,
- boost::optional<OID>* electionId) const {
+ enum struct TimeseriesAtomicWriteResult {
+ kSuccess,
+ kContinuableError,
+ kNonContinuableError,
+ };
+
+ TimeseriesAtomicWriteResult _commitTimeseriesBucketsAtomically(
+ OperationContext* opCtx,
+ TimeseriesBatches* batches,
+ TimeseriesStmtIds&& stmtIds,
+ std::vector<write_ops::WriteError>* errors,
+ boost::optional<repl::OpTime>* opTime,
+ boost::optional<OID>* electionId) const {
auto& bucketCatalog = BucketCatalog::get(opCtx);
std::vector<std::reference_wrapper<std::shared_ptr<BucketCatalog::WriteBatch>>>
@@ -890,7 +866,7 @@ public:
}
if (batchesToCommit.empty()) {
- return true;
+ return TimeseriesAtomicWriteResult::kSuccess;
}
// Sort by bucket so that preparing the commit for each batch cannot deadlock.
@@ -916,7 +892,7 @@ public:
auto prepareCommitStatus = bucketCatalog.prepareCommit(batch);
if (!prepareCommitStatus.isOK()) {
abortStatus = prepareCommitStatus;
- return false;
+ return TimeseriesAtomicWriteResult::kContinuableError;
}
if (batch.get()->numPreviouslyCommittedMeasurements() == 0) {
@@ -933,26 +909,33 @@ public:
auto result =
write_ops_exec::performAtomicTimeseriesWrites(opCtx, insertOps, updateOps);
if (!result.isOK()) {
- if (result.code() == ErrorCodes::DuplicateKey) {
- BucketCatalog::get(opCtx).resetBucketOIDCounter();
- }
abortStatus = result;
- return false;
+ return TimeseriesAtomicWriteResult::kContinuableError;
}
getOpTimeAndElectionId(opCtx, opTime, electionId);
+ bool compressClosedBuckets = true;
for (auto batch : batchesToCommit) {
auto closedBucket = bucketCatalog.finish(
batch, BucketCatalog::CommitInfo{*opTime, *electionId});
batch.get().reset();
- if (!closedBucket) {
+ if (!closedBucket || !compressClosedBuckets) {
continue;
}
// If this write closed a bucket, compress the bucket
- _performTimeseriesBucketCompression(opCtx, *closedBucket);
+ auto ret = _performTimeseriesBucketCompression(opCtx, *closedBucket);
+ if (!ret.result.isOK()) {
+ // Don't try to compress any other buckets if we fail. We're not allowed to
+ // do more write operations.
+ compressClosedBuckets = false;
+ }
+ if (!ret.canContinue) {
+ abortStatus = ret.result.getStatus();
+ return TimeseriesAtomicWriteResult::kNonContinuableError;
+ }
}
} catch (const DBException& ex) {
abortStatus = ex.toStatus();
@@ -960,7 +943,7 @@ public:
}
batchGuard.dismiss();
- return true;
+ return TimeseriesAtomicWriteResult::kSuccess;
}
// For sharded time-series collections, we need to use the granularity from the config
@@ -985,7 +968,10 @@ public:
}
}
- std::tuple<TimeseriesBatches, TimeseriesStmtIds, size_t /* numInserted */>
+ std::tuple<TimeseriesBatches,
+ TimeseriesStmtIds,
+ size_t /* numInserted */,
+ bool /* canContinue */>
_insertIntoBucketCatalog(OperationContext* opCtx,
size_t start,
size_t numDocs,
@@ -1025,6 +1011,7 @@ public:
TimeseriesBatches batches;
TimeseriesStmtIds stmtIds;
+ bool canContinue = true;
auto insert = [&](size_t index) {
invariant(start + index < request().getDocuments().size());
@@ -1070,63 +1057,58 @@ public:
// If this insert closed buckets, rewrite to be a compressed column. If we cannot
// perform write operations at this point the bucket will be left uncompressed.
for (const auto& closedBucket : result.getValue().closedBuckets) {
+ if (!canContinue) {
+ break;
+ }
+
// If this write closed a bucket, compress the bucket
- _performTimeseriesBucketCompression(opCtx, closedBucket);
+ auto ret = _performTimeseriesBucketCompression(opCtx, closedBucket);
+ if (auto error =
+ generateError(opCtx, ret.result, start + index, errors->size())) {
+ // Bucket compression only fail when we may not try to perform any other
+ // write operation. When handleError() inside write_ops_exec.cpp return
+ // false.
+ errors->emplace_back(std::move(*error));
+ canContinue = false;
+ return false;
+ }
+ canContinue = ret.canContinue;
}
return true;
};
- try {
- if (!indices.empty()) {
- std::for_each(indices.begin(), indices.end(), insert);
- } else {
- for (size_t i = 0; i < numDocs; i++) {
- if (!insert(i) && request().getOrdered()) {
- return {std::move(batches), std::move(stmtIds), i};
- }
+ if (!indices.empty()) {
+ std::for_each(indices.begin(), indices.end(), insert);
+ } else {
+ for (size_t i = 0; i < numDocs; i++) {
+ if (!insert(i) && request().getOrdered()) {
+ return {std::move(batches), std::move(stmtIds), i, canContinue};
}
}
- } catch (const DBException& ex) {
- // Exception insert into bucket catalog, append error and wait for all batches that
- // we've already managed to write into to commit or abort. We need to wait here as
- // pointers to memory owned by this command is stored in the WriteBatch(es). This
- // ensures that no other thread may try to access this memory after this command has
- // been torn down due to the exception.
-
- boost::optional<repl::OpTime> opTime;
- boost::optional<OID> electionId;
- std::vector<size_t> docsToRetry;
- errors->emplace_back(
- *generateErrorNoTenantMigration(opCtx, ex.toStatus(), 0, errors->size()));
-
- _getTimeseriesBatchResultsNoTenantMigration(
- opCtx, batches, 0, -1, false, errors, &opTime, &electionId, &docsToRetry);
- throw;
}
-
- return {std::move(batches), std::move(stmtIds), request().getDocuments().size()};
+ return {std::move(batches),
+ std::move(stmtIds),
+ request().getDocuments().size(),
+ canContinue};
}
- template <typename ErrorGenerator>
- void _getTimeseriesBatchResultsBase(ErrorGenerator&& errorGenerator,
- OperationContext* opCtx,
- const TimeseriesBatches& batches,
- int64_t start,
- int64_t indexOfLastProcessedBatch,
- bool canContinue,
- std::vector<write_ops::WriteError>* errors,
- boost::optional<repl::OpTime>* opTime,
- boost::optional<OID>* electionId,
- std::vector<size_t>* docsToRetry = nullptr) const {
+ void _getTimeseriesBatchResults(OperationContext* opCtx,
+ const TimeseriesBatches& batches,
+ size_t start,
+ size_t indexOfLastProcessedBatch,
+ bool canContinue,
+ std::vector<write_ops::WriteError>* errors,
+ boost::optional<repl::OpTime>* opTime,
+ boost::optional<OID>* electionId,
+ std::vector<size_t>* docsToRetry = nullptr) const {
boost::optional<write_ops::WriteError> lastError;
if (!errors->empty()) {
lastError = errors->back();
}
- invariant(indexOfLastProcessedBatch == (int64_t)batches.size() || lastError);
- for (int64_t itr = 0, size = batches.size(); itr < size; ++itr) {
+ for (size_t itr = 0; itr < batches.size(); ++itr) {
const auto& [batch, index] = batches[itr];
if (!batch) {
continue;
@@ -1142,11 +1124,11 @@ public:
auto swCommitInfo = batch->getResult();
if (swCommitInfo.getStatus() == ErrorCodes::TimeseriesBucketCleared) {
- invariant(docsToRetry, "the 'docsToRetry' cannot be null");
+ tassert(6023102, "the 'docsToRetry' cannot be null", docsToRetry);
docsToRetry->push_back(index);
continue;
}
- if (auto error = errorGenerator(
+ if (auto error = generateError(
opCtx, swCommitInfo.getStatus(), start + index, errors->size())) {
errors->emplace_back(std::move(*error));
continue;
@@ -1171,76 +1153,30 @@ public:
}
}
- void _getTimeseriesBatchResults(OperationContext* opCtx,
- const TimeseriesBatches& batches,
- int64_t start,
- int64_t indexOfLastProcessedBatch,
- bool canContinue,
- std::vector<write_ops::WriteError>* errors,
- boost::optional<repl::OpTime>* opTime,
- boost::optional<OID>* electionId,
- std::vector<size_t>* docsToRetry = nullptr) const {
- auto errorGenerator =
- [](OperationContext* opCtx, const Status& status, int index, size_t numErrors) {
- return generateError(opCtx, status, index, numErrors);
- };
- _getTimeseriesBatchResultsBase(errorGenerator,
- opCtx,
- batches,
- start,
- indexOfLastProcessedBatch,
- canContinue,
- errors,
- opTime,
- electionId,
- docsToRetry);
- }
-
- void _getTimeseriesBatchResultsNoTenantMigration(
+ TimeseriesAtomicWriteResult _performOrderedTimeseriesWritesAtomically(
OperationContext* opCtx,
- const TimeseriesBatches& batches,
- int64_t start,
- int64_t indexOfLastProcessedBatch,
- bool canContinue,
std::vector<write_ops::WriteError>* errors,
boost::optional<repl::OpTime>* opTime,
boost::optional<OID>* electionId,
- std::vector<size_t>* docsToRetry = nullptr) const {
- auto errorGenerator =
- [](OperationContext* opCtx, const Status& status, int index, size_t numErrors) {
- return generateErrorNoTenantMigration(opCtx, status, index, numErrors);
- };
- _getTimeseriesBatchResultsBase(errorGenerator,
- opCtx,
- batches,
- start,
- indexOfLastProcessedBatch,
- canContinue,
- errors,
- opTime,
- electionId,
- docsToRetry);
- }
-
- bool _performOrderedTimeseriesWritesAtomically(OperationContext* opCtx,
- std::vector<write_ops::WriteError>* errors,
- boost::optional<repl::OpTime>* opTime,
- boost::optional<OID>* electionId,
- bool* containsRetry) const {
- auto [batches, stmtIds, numInserted] = _insertIntoBucketCatalog(
+ bool* containsRetry) const {
+ auto [batches, stmtIds, numInserted, canContinue] = _insertIntoBucketCatalog(
opCtx, 0, request().getDocuments().size(), {}, errors, containsRetry);
+ if (!canContinue) {
+ return TimeseriesAtomicWriteResult::kNonContinuableError;
+ }
hangTimeseriesInsertBeforeCommit.pauseWhileSet();
- if (!_commitTimeseriesBucketsAtomically(
- opCtx, &batches, std::move(stmtIds), errors, opTime, electionId)) {
- return false;
+ auto result = _commitTimeseriesBucketsAtomically(
+ opCtx, &batches, std::move(stmtIds), errors, opTime, electionId);
+ if (result != TimeseriesAtomicWriteResult::kSuccess) {
+ return result;
}
_getTimeseriesBatchResults(
opCtx, batches, 0, batches.size(), true, errors, opTime, electionId);
- return true;
+ return TimeseriesAtomicWriteResult::kSuccess;
}
/**
@@ -1251,9 +1187,19 @@ public:
boost::optional<repl::OpTime>* opTime,
boost::optional<OID>* electionId,
bool* containsRetry) const {
- if (_performOrderedTimeseriesWritesAtomically(
- opCtx, errors, opTime, electionId, containsRetry)) {
- return request().getDocuments().size();
+ auto result = _performOrderedTimeseriesWritesAtomically(
+ opCtx, errors, opTime, electionId, containsRetry);
+ switch (result) {
+ case TimeseriesAtomicWriteResult::kSuccess:
+ return request().getDocuments().size();
+ case TimeseriesAtomicWriteResult::kNonContinuableError:
+ // If we can't continue, we know that 0 were inserted since this function should
+ // guarantee that the inserts are atomic.
+ return 0;
+ case TimeseriesAtomicWriteResult::kContinuableError:
+ break;
+ default:
+ MONGO_UNREACHABLE;
}
for (size_t i = 0; i < request().getDocuments().size(); ++i) {
@@ -1272,8 +1218,6 @@ public:
* which were attempted in an update operation, but found no bucket to update. These indices
* can be passed as the 'indices' parameter in a subsequent call to this function, in order
* to to be retried.
- * In rare cases due to collision from OID generation, we will also retry inserting those
- * bucket * documents for a limited number of times.
*/
std::vector<size_t> _performUnorderedTimeseriesWrites(
OperationContext* opCtx,
@@ -1283,73 +1227,39 @@ public:
std::vector<write_ops::WriteError>* errors,
boost::optional<repl::OpTime>* opTime,
boost::optional<OID>* electionId,
- bool* containsRetry,
- absl::flat_hash_map<int, int>& retryAttemptsForDup) const {
- auto [batches, bucketStmtIds, _] =
+ bool* containsRetry) const {
+ auto [batches, bucketStmtIds, _, canContinue] =
_insertIntoBucketCatalog(opCtx, start, numDocs, indices, errors, containsRetry);
hangTimeseriesInsertBeforeCommit.pauseWhileSet();
std::vector<size_t> docsToRetry;
- bool canContinue = true;
-
-
- stdx::unordered_set<BucketCatalog::WriteBatch*> handledHere;
- int64_t handledElsewhere = 0;
- auto guard = ScopeGuard([this, &handledElsewhere, opCtx]() {
- if (handledElsewhere > 0) {
- auto& bucketCatalog = BucketCatalog::get(opCtx);
- bucketCatalog.reportMeasurementsGroupCommitted(request().getNamespace(),
- handledElsewhere);
- }
- });
+ if (!canContinue) {
+ return docsToRetry;
+ }
size_t itr = 0;
for (; itr < batches.size(); ++itr) {
auto& [batch, index] = batches[itr];
if (batch->claimCommitRights()) {
- handledHere.insert(batch.get());
auto stmtIds = isTimeseriesWriteRetryable(opCtx)
? std::move(bucketStmtIds[batch->bucket().id])
: std::vector<StmtId>{};
- try {
- canContinue = _commitTimeseriesBucket(opCtx,
- batch,
- start,
- index,
- std::move(stmtIds),
- errors,
- opTime,
- electionId,
- &docsToRetry,
- retryAttemptsForDup);
- } catch (const DBException& ex) {
- // Exception during commit, append error and wait for all our batches to
- // commit or
- // abort. We need to wait here as pointers to memory owned by this command
- // is stored in the WriteBatch(es). This ensures that no other thread may
- // try to access this memory after this command has been torn down due to
- // the exception.
- errors->emplace_back(*generateErrorNoTenantMigration(
- opCtx, ex.toStatus(), start + index, errors->size()));
- _getTimeseriesBatchResultsNoTenantMigration(opCtx,
- batches,
- 0,
- itr,
- canContinue,
- errors,
- opTime,
- electionId,
- &docsToRetry);
- throw;
- }
+
+ canContinue = _commitTimeseriesBucket(opCtx,
+ batch,
+ start,
+ index,
+ std::move(stmtIds),
+ errors,
+ opTime,
+ electionId,
+ &docsToRetry);
batch.reset();
if (!canContinue) {
break;
}
- } else if (!handledHere.contains(batch.get())) {
- ++handledElsewhere;
}
}
@@ -1370,20 +1280,9 @@ public:
boost::optional<OID>* electionId,
bool* containsRetry) const {
std::vector<size_t> docsToRetry;
- absl::flat_hash_map<int, int> retryAttemptsForDup;
do {
- docsToRetry = _performUnorderedTimeseriesWrites(opCtx,
- start,
- numDocs,
- docsToRetry,
- errors,
- opTime,
- electionId,
- containsRetry,
- retryAttemptsForDup);
- if (!retryAttemptsForDup.empty()) {
- BucketCatalog::get(opCtx).resetBucketOIDCounter();
- }
+ docsToRetry = _performUnorderedTimeseriesWrites(
+ opCtx, start, numDocs, docsToRetry, errors, opTime, electionId, containsRetry);
} while (!docsToRetry.empty());
}
@@ -1403,11 +1302,6 @@ public:
curOp.getReadWriteType());
});
- // If an expected collection UUID is provided, always fail because the user-facing
- // time-series namespace does not have a UUID.
- checkCollectionUUIDMismatch(
- opCtx, request().getNamespace(), nullptr, request().getCollectionUUID());
-
uassert(
ErrorCodes::OperationNotSupportedInTransaction,
str::stream() << "Cannot insert into a time-series collection in a multi-document "
@@ -1547,29 +1441,19 @@ public:
invariant(!_commandObj.isEmpty());
+ if (const auto& shardVersion = _commandObj.getField("shardVersion");
+ !shardVersion.eoo()) {
+ bob->append(shardVersion);
+ }
bob->append("find", _commandObj["update"].String());
extractQueryDetails(_updateOpObj, bob);
bob->append("batchSize", 1);
bob->append("singleBatch", true);
-
- if (const auto& shardVersion = _commandObj.getField("shardVersion");
- !shardVersion.eoo()) {
- bob->append(shardVersion);
- }
- if (const auto& databaseVersion = _commandObj.getField("databaseVersion");
- !databaseVersion.eoo()) {
- bob->append(databaseVersion);
- }
}
write_ops::UpdateCommandReply typedRun(OperationContext* opCtx) final try {
- // On debug builds, verify that the estimated size of the update command is at least as
- // large as the size of the actual, serialized update command. This ensures that the
- // logic which estimates the size of update commands is correct.
- dassert(write_ops::verifySizeEstimate(request(), &unparsedRequest()));
transactionChecks(opCtx, ns());
-
write_ops::UpdateCommandReply updateReply;
OperationSource source = OperationSource::kStandard;
@@ -1676,7 +1560,6 @@ public:
updateRequest.setLegacyRuntimeConstants(request().getLegacyRuntimeConstants().value_or(
Variables::generateRuntimeConstants(opCtx)));
updateRequest.setLetParameters(request().getLet());
- updateRequest.setBypassEmptyTsReplacement(request().getBypassEmptyTsReplacement());
updateRequest.setYieldPolicy(PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
updateRequest.setExplain(verbosity);
@@ -1758,11 +1641,6 @@ public:
}
write_ops::DeleteCommandReply typedRun(OperationContext* opCtx) final try {
- // On debug builds, verify that the estimated size of the deletes are at least as large
- // as the actual, serialized size. This ensures that the logic that estimates the size
- // of deletes for batch writes is correct.
- dassert(write_ops::verifySizeEstimate(request(), &unparsedRequest()));
-
transactionChecks(opCtx, ns());
write_ops::DeleteCommandReply deleteReply;
OperationSource source = OperationSource::kStandard;
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 11:41:05 +0000