summaryrefslogtreecommitdiff
path: root/src/mongo/db/timeseries/bucket_catalog.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/mongo/db/timeseries/bucket_catalog.cpp')
-rw-r--r--src/mongo/db/timeseries/bucket_catalog.cpp204
1 files changed, 113 insertions, 91 deletions
diff --git a/src/mongo/db/timeseries/bucket_catalog.cpp b/src/mongo/db/timeseries/bucket_catalog.cpp
index c9d3bc2f76a..10e0e2df903 100644
--- a/src/mongo/db/timeseries/bucket_catalog.cpp
+++ b/src/mongo/db/timeseries/bucket_catalog.cpp
@@ -38,7 +38,6 @@
#include "mongo/db/commands/server_status.h"
#include "mongo/db/concurrency/write_conflict_exception.h"
#include "mongo/db/operation_context.h"
-#include "mongo/db/timeseries/metadata.h"
#include "mongo/db/timeseries/timeseries_options.h"
#include "mongo/platform/compiler.h"
#include "mongo/stdx/thread.h"
@@ -47,14 +46,11 @@
namespace mongo {
namespace {
+void normalizeArray(BSONArrayBuilder* builder, const BSONObj& obj);
+void normalizeObject(BSONObjBuilder* builder, const BSONObj& obj);
+
const auto getBucketCatalog = ServiceContext::declareDecoration<BucketCatalog>();
MONGO_FAIL_POINT_DEFINE(hangTimeseriesDirectModificationBeforeWriteConflict);
-MONGO_FAIL_POINT_DEFINE(hangWaitingForConflictingPreparedBatch);
-
-Mutex _bucketIdGenLock =
- MONGO_MAKE_LATCH(HierarchicalAcquisitionLevel(0), "bucket_catalog_internal::_bucketIdGenLock");
-PseudoRandom _bucketIdGenPRNG(SecureRandom().nextInt64());
-AtomicWord<uint64_t> _bucketIdGenCounter{static_cast<uint64_t>(_bucketIdGenPRNG.nextInt64())};
uint8_t numDigits(uint32_t num) {
uint8_t numDigits = 0;
@@ -65,6 +61,81 @@ uint8_t numDigits(uint32_t num) {
return numDigits;
}
+void normalizeArray(BSONArrayBuilder* builder, const BSONObj& obj) {
+ for (auto& arrayElem : obj) {
+ if (arrayElem.type() == BSONType::Array) {
+ BSONArrayBuilder subArray = builder->subarrayStart();
+ normalizeArray(&subArray, arrayElem.Obj());
+ } else if (arrayElem.type() == BSONType::Object) {
+ BSONObjBuilder subObject = builder->subobjStart();
+ normalizeObject(&subObject, arrayElem.Obj());
+ } else {
+ builder->append(arrayElem);
+ }
+ }
+}
+
+void normalizeObject(BSONObjBuilder* builder, const BSONObj& obj) {
+ // BSONObjIteratorSorted provides an abstraction similar to what this function does. However it
+ // is using a lexical comparison that is slower than just doing a binary comparison of the field
+ // names. That is all we need here as we are looking to create something that is binary
+ // comparable no matter of field order provided by the user.
+
+ // Helper that extracts the necessary data from a BSONElement that we can sort and re-construct
+ // the same BSONElement from.
+ struct Field {
+ BSONElement element() const {
+ return BSONElement(fieldName.rawData() - 1, // Include type byte before field name
+ fieldName.size() + 1, // Include null terminator after field name
+ totalSize);
+ }
+ bool operator<(const Field& rhs) const {
+ return fieldName < rhs.fieldName;
+ }
+ StringData fieldName;
+ int totalSize;
+ };
+
+ // Put all elements in a buffer, sort it and then continue normalize in sorted order
+ auto num = obj.nFields();
+ static constexpr std::size_t kNumStaticFields = 16;
+ boost::container::small_vector<Field, kNumStaticFields> fields;
+ fields.resize(num);
+ BSONObjIterator bsonIt(obj);
+ int i = 0;
+ while (bsonIt.more()) {
+ auto elem = bsonIt.next();
+ fields[i++] = {elem.fieldNameStringData(), elem.size()};
+ }
+ auto it = fields.begin();
+ auto end = fields.end();
+ std::sort(it, end);
+ for (; it != end; ++it) {
+ auto elem = it->element();
+ if (elem.type() == BSONType::Array) {
+ BSONArrayBuilder subArray(builder->subarrayStart(elem.fieldNameStringData()));
+ normalizeArray(&subArray, elem.Obj());
+ } else if (elem.type() == BSONType::Object) {
+ BSONObjBuilder subObject(builder->subobjStart(elem.fieldNameStringData()));
+ normalizeObject(&subObject, elem.Obj());
+ } else {
+ builder->append(elem);
+ }
+ }
+}
+
+void normalizeTopLevel(BSONObjBuilder* builder, const BSONElement& elem) {
+ if (elem.type() == BSONType::Array) {
+ BSONArrayBuilder subArray(builder->subarrayStart(elem.fieldNameStringData()));
+ normalizeArray(&subArray, elem.Obj());
+ } else if (elem.type() == BSONType::Object) {
+ BSONObjBuilder subObject(builder->subobjStart(elem.fieldNameStringData()));
+ normalizeObject(&subObject, elem.Obj());
+ } else {
+ builder->append(elem);
+ }
+}
+
OperationId getOpId(OperationContext* opCtx,
BucketCatalog::CombineWithInsertsFromOtherClients combine) {
switch (combine) {
@@ -84,7 +155,7 @@ BSONObj buildControlMinTimestampDoc(StringData timeField, Date_t roundedTime) {
}
std::pair<OID, Date_t> generateBucketId(const Date_t& time, const TimeseriesOptions& options) {
- OID oid;
+ OID bucketId = OID::gen();
// We round the measurement timestamp down to the nearest minute, hour, or day depending on the
// granularity. We do this for two reasons. The first is so that if measurements come in
@@ -94,35 +165,32 @@ std::pair<OID, Date_t> generateBucketId(const Date_t& time, const TimeseriesOpti
// what the bucket timestamp will be, so we can route measurements to the right shard chunk.
auto roundedTime = timeseries::roundTimestampToGranularity(time, options.getGranularity());
int64_t const roundedSeconds = durationCount<Seconds>(roundedTime.toDurationSinceEpoch());
- oid.setTimestamp(roundedSeconds);
-
- // Now, if we used the standard OID generation method for the remaining bytes we could end up
- // with lots of bucket OID collisions. Consider the case where we have the granularity set to
- // 'Hours'. This means we will round down to the nearest day, so any bucket generated on the
- // same machine on the same day will have the same timestamp portion and unique instance portion
- // of the OID. Only the increment would differ. Since we only use 3 bytes for the increment
- // portion, we run a serious risk of overflow if we are generating lots of buckets.
+ bucketId.setTimestamp(roundedSeconds);
+
+ // Now, if we stopped here we could end up with bucket OID collisions. Consider the case where
+ // we have the granularity set to 'Hours'. This means we will round down to the nearest day, so
+ // any bucket generated on the same machine on the same day will have the same timestamp portion
+ // and unique instance portion of the OID. Only the increment will differ. Since we only use 3
+ // bytes for the increment portion, we run a serious risk of overflow if we are generating lots
+ // of buckets.
//
- // To address this, we'll instead use a PRNG to generate the rest of the bytes. With 8 bytes of
- // randomness, we should have a pretty low chance of collisions. The limit of the birthday
- // paradox converges to roughly the square root of the size of the space, so we would need a few
- // billion buckets with the same timestamp to expect collisions. In the rare case that we do get
- // a collision, we can (and do) simply regenerate the bucket _id at a higher level.
- uint64_t bits = BigEndian<uint64_t>::store(_bucketIdGenCounter.addAndFetch(1));
-
- OID::InstanceUnique instance;
- const auto instanceBuf = static_cast<uint8_t*>(instance.bytes);
- std::memcpy(instanceBuf, &bits, OID::kInstanceUniqueSize);
-
- OID::Increment increment;
- const auto incrementBuf = static_cast<uint8_t*>(increment.bytes);
- uint8_t* bitsBuf = (uint8_t*)&bits;
- std::memcpy(incrementBuf, &(bitsBuf)[OID::kInstanceUniqueSize], OID::kIncrementSize);
-
- oid.setInstanceUnique(instance);
- oid.setIncrement(increment);
+ // To address this, we'll take the difference between the actual timestamp and the rounded
+ // timestamp and add it to the instance portion of the OID to ensure we can't have a collision.
+ // for timestamps generated on the same machine.
+ //
+ // This leaves open the possibility that in the case of step-down/step-up, we could get a
+ // collision if the old primary and the new primary have unique instance bits that differ by
+ // less than the maximum rounding difference. This is quite unlikely though, and can be resolved
+ // by restarting the new primary. It remains an open question whether we can fix this in a
+ // better way.
+ // TODO (SERVER-61412): Avoid time-series bucket OID collisions after election
+ auto instance = bucketId.getInstanceUnique();
+ uint32_t sum = DataView(reinterpret_cast<char*>(instance.bytes)).read<uint32_t>(1) +
+ (durationCount<Seconds>(time.toDurationSinceEpoch()) - roundedSeconds);
+ DataView(reinterpret_cast<char*>(instance.bytes)).write<uint32_t>(sum, 1);
+ bucketId.setInstanceUnique(instance);
- return {oid, roundedTime};
+ return {bucketId, roundedTime};
}
Status getTimeseriesBucketClearedError(const OID& bucketId,
@@ -202,12 +270,6 @@ void BucketCatalog::ExecutionStatsController::incNumMeasurementsCommitted(long l
_globalStats->numMeasurementsCommitted.fetchAndAddRelaxed(increment);
}
-void BucketCatalog::ExecutionStatsController::incNumMeasurementsGroupCommitted(
- long long increment) {
- _collectionStats->numMeasurementsGroupCommitted.fetchAndAddRelaxed(increment);
- _globalStats->numMeasurementsGroupCommitted.fetchAndAddRelaxed(increment);
-}
-
class BucketCatalog::Bucket {
public:
friend class BucketCatalog;
@@ -684,11 +746,7 @@ Status BucketCatalog::prepareCommit(std::shared_ptr<WriteBatch> batch) {
_useBucketInState(&stripe, stripeLock, batch->bucket().id, BucketState::kPrepared);
if (batch->finished()) {
- // Someone may have aborted it while we were waiting. Since we have the prepared batch, we
- // should now be able to fully abort the bucket.
- if (bucket) {
- _abort(&stripe, stripeLock, batch, getBatchStatus());
- }
+ // Someone may have aborted it while we were waiting.
return getBatchStatus();
} else if (!bucket) {
_abort(&stripe, stripeLock, batch, getTimeseriesBucketClearedError(batch->bucket().id));
@@ -853,8 +911,6 @@ void BucketCatalog::_appendExecutionStatsToBuilder(const ExecutionStats* stats,
if (commits) {
builder->appendNumber("avgNumMeasurementsPerCommit", measurementsCommitted / commits);
}
- builder->appendNumber("numMeasurementsGroupCommitted",
- stats->numMeasurementsGroupCommitted.load());
}
@@ -867,16 +923,6 @@ void BucketCatalog::appendGlobalExecutionStats(BSONObjBuilder* builder) const {
_appendExecutionStatsToBuilder(&_globalExecutionStats, builder);
}
-void BucketCatalog::resetBucketOIDCounter() {
- stdx::lock_guard lk{_bucketIdGenLock};
- _bucketIdGenCounter.store(static_cast<uint64_t>(_bucketIdGenPRNG.nextInt64()));
-}
-
-void BucketCatalog::reportMeasurementsGroupCommitted(const NamespaceString& ns, int64_t count) {
- auto stats = _getExecutionStats(ns);
- stats.incNumMeasurementsGroupCommitted(count);
-}
-
BucketCatalog::BucketMetadata::BucketMetadata(BSONElement elem,
const StringData::ComparatorInterface* comparator)
: _metadataElement(elem), _comparator(comparator) {
@@ -884,7 +930,7 @@ BucketCatalog::BucketMetadata::BucketMetadata(BSONElement elem,
BSONObjBuilder objBuilder;
// We will get an object of equal size, just with reordered fields.
objBuilder.bb().reserveBytes(_metadataElement.size());
- timeseries::metadata::normalize(_metadataElement, objBuilder);
+ normalizeTopLevel(&objBuilder, _metadataElement);
_metadata = objBuilder.obj();
}
// Updates the BSONElement to refer to the copied BSONObj.
@@ -899,9 +945,8 @@ const BSONObj& BucketCatalog::BucketMetadata::toBSON() const {
return _metadata;
}
-boost::optional<StringData> BucketCatalog::BucketMetadata::getMetaField() const {
- return _metadataElement ? boost::make_optional(_metadataElement.fieldNameStringData())
- : boost::none;
+StringData BucketCatalog::BucketMetadata::getMetaField() const {
+ return StringData(_metadataElement.fieldName());
}
const StringData::ComparatorInterface* BucketCatalog::BucketMetadata::getComparator() const {
@@ -1006,10 +1051,6 @@ void BucketCatalog::_waitToCommitBatch(Stripe* stripe, const std::shared_ptr<Wri
}
}
- // We only hit this failpoint when there are conflicting prepared batches on the same
- // bucket.
- hangWaitingForConflictingPreparedBatch.pauseWhileSet();
-
// We have to wait for someone else to finish.
current->getResult().getStatus().ignore(); // We don't care about the result.
}
@@ -1067,7 +1108,7 @@ void BucketCatalog::_abort(Stripe* stripe,
// user is doing with it, but we need to keep the bucket around until
// that batch is finished.
if (auto& prepared = bucket->_preparedBatch) {
- if (batch && prepared == batch) {
+ if (prepared == batch) {
// We own the prepared batch, so we can go ahead and abort it and remove the bucket.
prepared->_abort(status);
prepared.reset();
@@ -1078,8 +1119,6 @@ void BucketCatalog::_abort(Stripe* stripe,
if (doRemove) {
[[maybe_unused]] bool removed = _removeBucket(stripe, stripeLock, bucket);
- } else {
- _setBucketState(bucket->id(), BucketState::kCleared);
}
}
@@ -1123,31 +1162,14 @@ BucketCatalog::Bucket* BucketCatalog::_allocateBucket(Stripe* stripe,
const CreationInfo& info) {
_expireIdleBuckets(stripe, stripeLock, info.stats, info.closedBuckets);
+ auto [bucketId, roundedTime] = generateBucketId(info.time, info.options);
- // In rare cases duplicate bucket _id fields can be generated in the same stripe and fail to be
- // inserted. We will perform a limited number of retries to minimize the probability of
- // collision.
- auto maxRetries = gTimeseriesInsertMaxRetriesOnDuplicates.load();
- OID bucketId;
- Date_t roundedTime;
- stdx::unordered_map<OID, std::unique_ptr<Bucket>, OID::Hasher>::iterator it;
- bool inserted = false;
- for (int retryAttempts = 0; !inserted && retryAttempts < maxRetries; ++retryAttempts) {
- std::tie(bucketId, roundedTime) = generateBucketId(info.time, info.options);
- std::tie(it, inserted) = stripe->allBuckets.try_emplace(
- bucketId, std::make_unique<Bucket>(bucketId, info.stripe));
- if (!inserted) {
- resetBucketOIDCounter();
- }
- }
- uassert(6130900,
- "Unable to insert documents due to internal OID generation collision. Increase the "
- "value of server parameter 'timeseriesInsertMaxRetriesOnDuplicates' and try again",
- inserted);
-
+ auto [it, inserted] =
+ stripe->allBuckets.try_emplace(bucketId, std::make_unique<Bucket>(bucketId, info.stripe));
+ tassert(6130900, "Expected bucket to be inserted", inserted);
Bucket* bucket = it->second.get();
stripe->openBuckets[info.key] = bucket;
- _initializeBucketState(it->first);
+ _initializeBucketState(bucketId);
if (info.openedDuetoMetadata) {
info.stats.incNumBucketsOpenedDueToMetadata();
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 11:03:30 +0000