SERVER-60676 Improve robustness when parsing invalid BSONColumn binary or compressed v2 buckets

* Changed BSONColumn::operator[] interface so we can differentiate skipped element from out of bounds
* Replaced BadValue error codes in BSONColumn for unique assertion ids
* Throw exception when interleaved mode exits unexpectedly when it shouldn't
* Test and handle buckets where the compressed columns contain different amount of fields

1 files changed, 10 insertions, 4 deletions

diff --git a/src/mongo/db/exec/bucket_unpacker.cpp b/src/mongo/db/exec/bucket_unpacker.cpp
index ba7d5554630..87aa9f18c20 100644
--- a/src/mongo/db/exec/bucket_unpacker.cpp
+++ b/src/mongo/db/exec/bucket_unpacker.cpp
@@ -277,6 +277,9 @@ bool BucketUnpackerV2::getNext(MutableDocument& measurement,
     }
 
     for (auto& fieldColumn : _fieldColumns) {
+        uassert(6067601,
+                "Bucket unexpectedly contained fewer values than count",
+                fieldColumn.it != fieldColumn.column.end());
         const BSONElement& elem = *(fieldColumn.it++);
         // EOO represents missing field
         if (!elem.eoo()) {
@@ -296,9 +299,10 @@ void BucketUnpackerV2::extractSingleMeasurement(MutableDocument& measurement,
                                                 bool includeTimeField,
                                                 bool includeMetaField) {
     if (includeTimeField) {
-        BSONElement val = _timeColumn.column[j];
-        uassert(6067500, "Bucket unexpectedly contained fewer values than count", !val.eoo());
-        measurement.addField(_timeColumn.column.name(), Value{val});
+        auto val = _timeColumn.column[j];
+        uassert(
+            6067500, "Bucket unexpectedly contained fewer values than count", val && !val->eoo());
+        measurement.addField(_timeColumn.column.name(), Value{*val});
     }
 
     if (includeMetaField && !metaValue.isNull()) {
@@ -307,7 +311,9 @@ void BucketUnpackerV2::extractSingleMeasurement(MutableDocument& measurement,
 
     if (includeTimeField) {
         for (auto& fieldColumn : _fieldColumns) {
-            measurement.addField(fieldColumn.column.name(), Value{fieldColumn.column[j]});
+            auto val = fieldColumn.column[j];
+            uassert(6067600, "Bucket unexpectedly contained fewer values than count", val);
+            measurement.addField(fieldColumn.column.name(), Value{*val});
         }
     }
 }