SERVER-61330 Fix detection and merging when there are mismatches of empty subobject in BSONColumnBuilder

2 files changed, 301 insertions, 15 deletions

diff --git a/src/mongo/bson/util/bsoncolumn_test.cpp b/src/mongo/bson/util/bsoncolumn_test.cpp
index 5a6b393893b..629beca5db1 100644
--- a/src/mongo/bson/util/bsoncolumn_test.cpp
+++ b/src/mongo/bson/util/bsoncolumn_test.cpp
@@ -2759,6 +2759,265 @@ TEST_F(BSONColumnTest, InterleavedObjectEmptyObjChange) {
     verifyDecompression(binData, elems);
 }
 
+TEST_F(BSONColumnTest, InterleavedObjectNewEmptyObjMiddle) {
+    BSONColumnBuilder cb("test"_sd);
+
+    std::vector<BSONElement> elems = {
+        createElementObj(BSON("x" << 1 << "z" << 2)),
+        createElementObj(BSON("x" << 1 << "z" << 3)),
+        createElementObj(BSON("x" << 1 << "y" << BSONObjBuilder().obj() << "z" << 4))};
+
+    for (auto elem : elems) {
+        if (!elem.eoo())
+            cb.append(elem);
+        else
+            cb.skip();
+    }
+
+    BufBuilder expected;
+    appendInterleavedStart(expected, elems.front().Obj());
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd])},
+        1);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["z"_sd], elems[0].Obj()["z"_sd])},
+        1);
+    appendEOO(expected);
+
+    appendInterleavedStart(expected, elems[2].Obj());
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+    appendEOO(expected);
+
+    appendEOO(expected);
+
+    auto binData = cb.finalize();
+    verifyBinary(binData, expected);
+    verifyDecompression(binData, elems);
+}
+
+TEST_F(BSONColumnTest, InterleavedObjectNewEmptyObjUnderObj) {
+    BSONColumnBuilder cb("test"_sd);
+
+    std::vector<BSONElement> elems = {
+        createElementObj(BSON("x" << 1 << "z" << 2)),
+        createElementObj(BSON("x" << 1 << "z" << 3)),
+        createElementObj(
+            BSON("x" << 1 << "y" << BSON("y1" << BSONObjBuilder().obj()) << "z" << 4))};
+
+    for (auto elem : elems) {
+        if (!elem.eoo())
+            cb.append(elem);
+        else
+            cb.skip();
+    }
+
+    BufBuilder expected;
+    appendInterleavedStart(expected, elems.front().Obj());
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd])},
+        1);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["z"_sd], elems[0].Obj()["z"_sd])},
+        1);
+    appendEOO(expected);
+
+    appendInterleavedStart(expected, elems[2].Obj());
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+    appendEOO(expected);
+
+    appendEOO(expected);
+
+    auto binData = cb.finalize();
+    verifyBinary(binData, expected);
+    verifyDecompression(binData, elems);
+}
+
+TEST_F(BSONColumnTest, InterleavedObjectNewEmptyObjEnd) {
+    BSONColumnBuilder cb("test"_sd);
+
+    std::vector<BSONElement> elems = {
+        createElementObj(BSON("x" << 1 << "y" << 2)),
+        createElementObj(BSON("x" << 1 << "y" << 3)),
+        createElementObj(BSON("x" << 1 << "y" << 4 << "z" << BSONObjBuilder().obj()))};
+
+    for (auto elem : elems) {
+        if (!elem.eoo())
+            cb.append(elem);
+        else
+            cb.skip();
+    }
+
+    BufBuilder expected;
+    appendInterleavedStart(expected, elems.front().Obj());
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd])},
+        1);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["y"_sd], elems[0].Obj()["y"_sd])},
+        1);
+    appendEOO(expected);
+
+    appendInterleavedStart(expected, elems[2].Obj());
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+    appendEOO(expected);
+
+    appendEOO(expected);
+
+    auto binData = cb.finalize();
+    verifyBinary(binData, expected);
+    verifyDecompression(binData, elems);
+}
+
+TEST_F(BSONColumnTest, InterleavedObjectMissingEmptyObjMiddle) {
+    BSONColumnBuilder cb("test"_sd);
+
+    std::vector<BSONElement> elems = {
+        createElementObj(BSON("x" << 1 << "y" << BSONObjBuilder().obj() << "z" << 2)),
+        createElementObj(BSON("x" << 1 << "y" << BSONObjBuilder().obj() << "z" << 3)),
+        createElementObj(BSON("x" << 1 << "z" << 4))};
+
+    for (auto elem : elems) {
+        if (!elem.eoo())
+            cb.append(elem);
+        else
+            cb.skip();
+    }
+
+    BufBuilder expected;
+    appendInterleavedStart(expected, elems.front().Obj());
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd])},
+        1);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["z"_sd], elems[0].Obj()["z"_sd])},
+        1);
+    appendEOO(expected);
+
+    appendInterleavedStart(expected, elems[2].Obj());
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+    appendEOO(expected);
+
+    appendEOO(expected);
+
+    auto binData = cb.finalize();
+    verifyBinary(binData, expected);
+    verifyDecompression(binData, elems);
+}
+
+TEST_F(BSONColumnTest, InterleavedObjectMissingEmptyObjUnderObj) {
+    BSONColumnBuilder cb("test"_sd);
+
+    std::vector<BSONElement> elems = {
+        createElementObj(BSON("x" << 1 << "y" << BSON("y1" << BSONObjBuilder().obj()) << "z" << 2)),
+        createElementObj(BSON("x" << 1 << "y" << BSON("y1" << BSONObjBuilder().obj()) << "z" << 3)),
+        createElementObj(BSON("x" << 1 << "z" << 4))};
+
+    for (auto elem : elems) {
+        if (!elem.eoo())
+            cb.append(elem);
+        else
+            cb.skip();
+    }
+
+    BufBuilder expected;
+    appendInterleavedStart(expected, elems.front().Obj());
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd])},
+        1);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["z"_sd], elems[0].Obj()["z"_sd])},
+        1);
+    appendEOO(expected);
+
+    appendInterleavedStart(expected, elems[2].Obj());
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+    appendEOO(expected);
+
+    appendEOO(expected);
+
+    auto binData = cb.finalize();
+    verifyBinary(binData, expected);
+    verifyDecompression(binData, elems);
+}
+
+TEST_F(BSONColumnTest, InterleavedObjectMissingEmptyObjEnd) {
+    BSONColumnBuilder cb("test"_sd);
+
+    std::vector<BSONElement> elems = {
+        createElementObj(BSON("x" << 1 << "y" << 2 << "z" << BSONObjBuilder().obj())),
+        createElementObj(BSON("x" << 1 << "y" << 3 << "z" << BSONObjBuilder().obj())),
+        createElementObj(BSON("x" << 1 << "y" << 4))};
+
+    for (auto elem : elems) {
+        if (!elem.eoo())
+            cb.append(elem);
+        else
+            cb.skip();
+    }
+
+    BufBuilder expected;
+    appendInterleavedStart(expected, elems.front().Obj());
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd])},
+        1);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["y"_sd], elems[0].Obj()["y"_sd])},
+        1);
+    appendEOO(expected);
+
+    appendInterleavedStart(expected, elems[2].Obj());
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+    appendEOO(expected);
+
+    appendEOO(expected);
+
+    auto binData = cb.finalize();
+    verifyBinary(binData, expected);
+    verifyDecompression(binData, elems);
+}
+
 TEST_F(BSONColumnTest, ReenterInterleaved) {
     BSONColumnBuilder cb("test"_sd);
 
diff --git a/src/mongo/bson/util/bsoncolumnbuilder.cpp b/src/mongo/bson/util/bsoncolumnbuilder.cpp
index 89987836807..582b4dc8497 100644
--- a/src/mongo/bson/util/bsoncolumnbuilder.cpp
+++ b/src/mongo/bson/util/bsoncolumnbuilder.cpp
@@ -67,8 +67,7 @@ bool objectIdDeltaPossible(BSONElement elem, BSONElement prev) {
                    OID::kInstanceUniqueSize);
 }
 
-// Internal recursion function for traverseLockStep() when we just need to traverse reference
-// object.
+// Traverses object and calls 'ElementFunc' on every scalar subfield encountered.
 template <typename ElementFunc>
 void _traverse(const BSONObj& reference, const ElementFunc& elemFunc) {
     for (const auto& elem : reference) {
@@ -80,6 +79,29 @@ void _traverse(const BSONObj& reference, const ElementFunc& elemFunc) {
     }
 }
 
+// Internal recursion function for traverseLockStep() when we just need to traverse reference
+// object. Like '_traverse' above but exits when an empty sub object is encountered. Returns 'true'
+// if empty subobject found.
+template <typename ElementFunc>
+bool _traverseUntilEmptyObj(const BSONObj& obj, const ElementFunc& elemFunc) {
+    for (const auto& elem : obj) {
+        if (elem.type() == Object) {
+            if (_traverseUntilEmptyObj(elem.Obj(), elemFunc)) {
+                return true;
+            }
+        } else {
+            elemFunc(elem, BSONElement());
+        }
+    }
+
+    return obj.isEmpty();
+}
+
+// Helper function for mergeObj() to detect if Object contain subfields of empty Objects
+bool _hasEmptyObj(const BSONObj& obj) {
+    return _traverseUntilEmptyObj(obj, [](const BSONElement&, const BSONElement&) {});
+}
+
 // Internal recursion function for traverseLockStep(). See documentation for traverseLockStep.
 template <typename ElementFunc>
 std::pair<BSONObj::iterator, bool> _traverseLockStep(const BSONObj& reference,
@@ -90,14 +112,7 @@ std::pair<BSONObj::iterator, bool> _traverseLockStep(const BSONObj& reference,
     for (const auto& elem : reference) {
         if (elem.type() == Object) {
             BSONObj refObj = elem.Obj();
-            bool hasIt = it != end;
-            // If refObj is empty, there must also exist an empty object on 'it' for this to be
-            // valid. First we check that we have something on 'it'
-            if (!hasIt && refObj.isEmpty()) {
-                return {it, false};
-            }
-
-            bool elemMatch = hasIt && elem.fieldNameStringData() == it->fieldNameStringData();
+            bool elemMatch = it != end && elem.fieldNameStringData() == it->fieldNameStringData();
             if (elemMatch) {
                 // If 'reference' element is Object then 'obj' must also be Object.
                 if (it->type() != Object) {
@@ -117,9 +132,12 @@ std::pair<BSONObj::iterator, bool> _traverseLockStep(const BSONObj& reference,
             } else {
                 // Assume field name at 'it' is coming later in 'reference'. Traverse as if it is
                 // missing from 'obj'. We don't increment the iterator in this case. If it is a
-                // mismatch we will detect that at end when 'it' is not at 'end'.
-                // Nothing can fail below this so traverse without all the checks.
-                _traverse(refObj, elemFunc);
+                // mismatch we will detect that at end when 'it' is not at 'end'. Nothing can fail
+                // below this so traverse without all the checks. Any empty object detected is an
+                // error.
+                if (_traverseUntilEmptyObj(refObj, elemFunc)) {
+                    return {it, false};
+                }
             }
         } else {
             // Non-object, call provided function with the two elements
@@ -196,11 +214,20 @@ bool _mergeObj(BSONObjBuilder* builder, const BSONObj& reference, const BSONObj&
             n, end, [&name](const auto& elem) { return elem.fieldNameStringData() == name; });
         if (namePos == end) {
             // Reference element does not exist in 'obj' so add it and continue merging with just
-            // this iterator incremented.
+            // this iterator incremented. Unless it is or contains an empty object which is
+            // incompatible.
+            if (refIt->type() == Object && _hasEmptyObj(refIt->Obj())) {
+                return false;
+            }
+
             builder->append(*(refIt++));
         } else {
             // Reference element do exist later in 'obj'. Add element in 'it' if it is the first
-            // time we see it, fail otherwise (incompatible ordering).
+            // time we see it, fail otherwise (incompatible ordering). Unless 'it' is or contains an
+            // empty object which is incompatible.
+            if (it->type() == Object && _hasEmptyObj(it->Obj())) {
+                return false;
+            }
             if (builder->hasField(it->fieldNameStringData())) {
                 return false;
             }