SERVER-64732 BSONColumn decompression support of interleaved arrays

Two new control bytes were introduced. One for indicating array support
and the other for array support with an array root object.

The decompressor handles both the new and the old format.

5 files changed, 1743 insertions, 701 deletions

diff --git a/src/mongo/bson/util/bsoncolumn.cpp b/src/mongo/bson/util/bsoncolumn.cpp
index ce288893f2d..7f1187c11f0 100644
--- a/src/mongo/bson/util/bsoncolumn.cpp
+++ b/src/mongo/bson/util/bsoncolumn.cpp
@@ -82,24 +82,48 @@ constexpr std::array<uint8_t, 16> kControlToScaleIndex = {
 template <typename EnterSubObjFunc, typename ElementFunc>
 class BSONObjTraversal {
 public:
-    BSONObjTraversal(EnterSubObjFunc enterFunc, ElementFunc elemFunc)
-        : _enterFunc(std::move(enterFunc)), _elemFunc(std::move(elemFunc)) {}
+    BSONObjTraversal(bool recurseIntoArrays,
+                     BSONType rootType,
+                     EnterSubObjFunc enterFunc,
+                     ElementFunc elemFunc)
+        : _enterFunc(std::move(enterFunc)),
+          _elemFunc(std::move(elemFunc)),
+          _recurseIntoArrays(recurseIntoArrays),
+          _rootType(rootType) {}
 
     bool traverse(const BSONObj& obj) {
-        return _traverse(""_sd, std::move(obj));
+        if (_recurseIntoArrays) {
+            return _traverseIntoArrays(""_sd, obj, _rootType);
+        } else {
+            return _traverseNoArrays(""_sd, obj, _rootType);
+        }
     }
 
 private:
-    bool _traverse(StringData fieldName, const BSONObj& obj) {
-        [[maybe_unused]] auto raii = _enterFunc(fieldName, obj);
+    bool _traverseNoArrays(StringData fieldName, const BSONObj& obj, BSONType type) {
+        [[maybe_unused]] auto raii = _enterFunc(fieldName, obj, type);
+
+        return std::all_of(obj.begin(), obj.end(), [this, &fieldName](auto&& elem) {
+            return elem.type() == Object
+                ? _traverseNoArrays(elem.fieldNameStringData(), elem.Obj(), Object)
+                : _elemFunc(elem);
+        });
+    }
+
+    bool _traverseIntoArrays(StringData fieldName, const BSONObj& obj, BSONType type) {
+        [[maybe_unused]] auto raii = _enterFunc(fieldName, obj, type);
+
         return std::all_of(obj.begin(), obj.end(), [this, &fieldName](auto&& elem) {
-            return elem.type() == Object ? _traverse(elem.fieldNameStringData(), elem.Obj())
-                                         : _elemFunc(elem);
+            return elem.type() == Object || elem.type() == Array
+                ? _traverseIntoArrays(elem.fieldNameStringData(), elem.Obj(), elem.type())
+                : _elemFunc(elem);
         });
     }
 
     EnterSubObjFunc _enterFunc;
     ElementFunc _elemFunc;
+    bool _recurseIntoArrays;
+    BSONType _rootType;
 };
 
 }  // namespace
@@ -214,7 +238,10 @@ BSONColumn::ElementStorage::Element BSONColumn::ElementStorage::allocate(BSONTyp
 
 struct BSONColumn::SubObjectAllocator {
 public:
-    SubObjectAllocator(ElementStorage& allocator, StringData fieldName, const BSONObj& obj)
+    SubObjectAllocator(ElementStorage& allocator,
+                       StringData fieldName,
+                       const BSONObj& obj,
+                       BSONType type)
         : _allocator(allocator) {
         // Remember size of field name for this subobject in case it ends up being an empty
         // subobject and we need to 'deallocate' it.
@@ -225,7 +252,7 @@ public:
         // Start the subobject, allocate space for the field in the parent which is BSON type byte +
         // field name + null terminator
         char* objdata = _allocator.allocate(2 + _fieldNameSize);
-        objdata[0] = Object;
+        objdata[0] = type;
         if (_fieldNameSize > 0) {
             memcpy(objdata + 1, fieldName.rawData(), _fieldNameSize);
         }
@@ -292,9 +319,15 @@ void BSONColumn::Iterator::_initialize(size_t index) {
 }
 
 void BSONColumn::Iterator::_initializeInterleaving() {
+    _interleavedArrays = *_control == bsoncolumn::kInterleavedStartControlByte ||
+        *_control == bsoncolumn::kInterleavedStartArrayRootControlByte;
+    _interleavedRootType =
+        *_control == bsoncolumn::kInterleavedStartArrayRootControlByte ? Array : Object;
     _interleavedReferenceObj = BSONObj(_control + 1);
 
-    BSONObjTraversal t([](StringData fieldName, const BSONObj& obj) { return true; },
+    BSONObjTraversal t(_interleavedArrays,
+                       _interleavedRootType,
+                       [](StringData fieldName, const BSONObj& obj, BSONType type) { return true; },
                        [this](const BSONElement& elem) {
                            _states.emplace_back();
                            _states.back()._loadLiteral(elem);
@@ -400,10 +433,12 @@ void BSONColumn::Iterator::_incrementInterleaved() {
     auto stateEnd = _states.end();
     int processed = 0;
     BSONObjTraversal t(
-        [this](StringData fieldName, const BSONObj& obj) {
+        _interleavedArrays,
+        _interleavedRootType,
+        [this](StringData fieldName, const BSONObj& obj, BSONType type) {
             // Called every time we recurse into a subobject. It makes sure we write the size and
             // EOO bytes.
-            return SubObjectAllocator(_column->_elementStorage, fieldName, obj);
+            return SubObjectAllocator(_column->_elementStorage, fieldName, obj, type);
         },
         [this, &stateIt, &stateEnd, &processed](const BSONElement& referenceField) {
             // Called for every scalar field in the reference interleaved BSONObj. We have as many
@@ -505,6 +540,20 @@ void BSONColumn::Iterator::_handleEOO() {
     _column->_fullyDecompressed = true;
 }
 
+bool BSONColumn::Iterator::_isLiteral(char control) {
+    return (control & 0xE0) == 0;
+}
+
+bool BSONColumn::Iterator::_isInterleavedStart(char control) {
+    return control == bsoncolumn::kInterleavedStartControlByteLegacy ||
+        control == bsoncolumn::kInterleavedStartControlByte ||
+        control == bsoncolumn::kInterleavedStartArrayRootControlByte;
+}
+
+uint8_t BSONColumn::Iterator::_numSimple8bBlocks(char control) {
+    return (control & 0x0F) + 1;
+}
+
 bool BSONColumn::Iterator::operator==(const Iterator& rhs) const {
     return _index == rhs._index;
 }
diff --git a/src/mongo/bson/util/bsoncolumn.h b/src/mongo/bson/util/bsoncolumn.h
index f69b52c2f92..23d4b19c7b5 100644
--- a/src/mongo/bson/util/bsoncolumn.h
+++ b/src/mongo/bson/util/bsoncolumn.h
@@ -119,19 +119,13 @@ public:
         void _handleEOO();
 
         // Checks if control byte is literal
-        static bool _isLiteral(uint8_t control) {
-            return (control & 0xE0) == 0;
-        }
+        static bool _isLiteral(char control);
 
         // Checks if control byte is interleaved mode start
-        static bool _isInterleavedStart(uint8_t control) {
-            return control == 0xF0;
-        }
+        static bool _isInterleavedStart(char control);
 
         // Returns number of Simple-8b blocks from control byte
-        static uint8_t _numSimple8bBlocks(uint8_t control) {
-            return (control & 0x0F) + 1;
-        }
+        static uint8_t _numSimple8bBlocks(char control);
 
         // Pointer to BSONColumn this Iterator is created from, this will be stale when moving the
         // BSONColumn. All iterators are invalidated on move!
@@ -213,6 +207,12 @@ public:
         // We setup a decoding state for each scalar field in this object. The object hierarchy is
         // used to re-construct with full objects with the correct hierachy to the user.
         BSONObj _interleavedReferenceObj;
+
+        // Indicates if decoding states should be opened when encountering arrays
+        bool _interleavedArrays;
+
+        // Type for root object/reference object. May be Object or Array.
+        BSONType _interleavedRootType;
     };
 
     /**
diff --git a/src/mongo/bson/util/bsoncolumn_test.cpp b/src/mongo/bson/util/bsoncolumn_test.cpp
index 17572486ea0..7ea4c16c5eb 100644
--- a/src/mongo/bson/util/bsoncolumn_test.cpp
+++ b/src/mongo/bson/util/bsoncolumn_test.cpp
@@ -172,6 +172,10 @@ public:
         return _createElement(arr);
     }
 
+    BSONElement createElementArrayAsObject(BSONArray arr) {
+        return _createElement(*static_cast<BSONObj*>(&arr));
+    }
+
     static boost::optional<uint128_t> deltaBinData(BSONElement val, BSONElement prev) {
         if (val.binaryEqualValues(prev)) {
             return uint128_t(0);
@@ -318,11 +322,21 @@ public:
         builder.appendChar(control << 4 | count);
     }
 
-    static void appendInterleavedStart(BufBuilder& builder, BSONObj reference) {
+    static void appendInterleavedStartLegacy(BufBuilder& builder, BSONObj reference) {
         builder.appendChar((char)0xF0);
         builder.appendBuf(reference.objdata(), reference.objsize());
     }
 
+    static void appendInterleavedStart(BufBuilder& builder, BSONObj reference) {
+        builder.appendChar((char)0xF1);
+        builder.appendBuf(reference.objdata(), reference.objsize());
+    }
+
+    static void appendInterleavedStartArrayRoot(BufBuilder& builder, BSONObj reference) {
+        builder.appendChar((char)0xF2);
+        builder.appendBuf(reference.objdata(), reference.objsize());
+    }
+
     template <typename T>
     static void _appendSimple8bBlock(BufBuilder& builder, boost::optional<T> val) {
         auto prev = builder.len();
@@ -393,6 +407,15 @@ public:
         ASSERT_EQ(memcmp(columnBinary.data, buf, columnBinary.length), 0);
     }
 
+    static void verifyDecompression(const BufBuilder& columnBinary,
+                                    const std::vector<BSONElement>& expected) {
+        BSONBinData bsonBinData;
+        bsonBinData.data = columnBinary.buf();
+        bsonBinData.length = columnBinary.len();
+        bsonBinData.type = Column;
+        verifyDecompression(bsonBinData, expected);
+    }
+
     static void verifyDecompression(BSONBinData columnBinary,
                                     const std::vector<BSONElement>& expected) {
         BSONObjBuilder obj;
@@ -2455,329 +2478,504 @@ TEST_F(BSONColumnTest, ArrayEqual) {
 }
 
 TEST_F(BSONColumnTest, Interleaved) {
-    BSONColumnBuilder cb("test"_sd);
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        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" << 3)),
+                                          BSONElement(),
+                                          createElementObj(BSON("y" << 4)),
+                                          createElementObj(BSON("x" << 1 << "y" << 3))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
-    std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1 << "y" << 2)),
-                                      createElementObj(BSON("x" << 1 << "y" << 3)),
-                                      createElementObj(BSON("x" << 1 << "y" << 3)),
-                                      BSONElement(),
-                                      createElementObj(BSON("y" << 4)),
-                                      createElementObj(BSON("x" << 1 << "y" << 3))};
+        BufBuilder expected;
+        appendInterleavedStartFunc(expected, elems.front().Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaInt32(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd]),
+                                deltaInt32(elems[2].Obj()["x"_sd], elems[1].Obj()["x"_sd]),
+                                boost::none,
+                                boost::none,
+                                deltaInt32(elems[5].Obj()["x"_sd], elems[2].Obj()["x"_sd])},
+                               1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaInt32(elems[1].Obj()["y"_sd], elems[0].Obj()["y"_sd]),
+                                deltaInt32(elems[2].Obj()["y"_sd], elems[1].Obj()["y"_sd]),
+                                boost::none,
+                                deltaInt32(elems[4].Obj()["y"_sd], elems[2].Obj()["y"_sd]),
+                                deltaInt32(elems[5].Obj()["y"_sd], elems[4].Obj()["y"_sd])},
+                               1);
+        appendEOO(expected);
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
 
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
+
+TEST_F(BSONColumnTest, InterleavedArray) {
+    std::vector<BSONElement> elems = {createElementObj(BSON("x" << BSON_ARRAY(1 << 2))),
+                                      createElementObj(BSON("x" << BSON_ARRAY(2 << 3)))};
 
     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]),
-                            deltaInt32(elems[2].Obj()["x"_sd], elems[1].Obj()["x"_sd]),
-                            boost::none,
-                            boost::none,
-                            deltaInt32(elems[5].Obj()["x"_sd], elems[2].Obj()["x"_sd])},
-                           1);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues,
+         deltaInt32(elems[1].Obj()["x"_sd].Array()[0], elems[0].Obj()["x"_sd].Array()[0])},
+        1);
     appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected,
-                           {kDeltaForBinaryEqualValues,
-                            deltaInt32(elems[1].Obj()["y"_sd], elems[0].Obj()["y"_sd]),
-                            deltaInt32(elems[2].Obj()["y"_sd], elems[1].Obj()["y"_sd]),
-                            boost::none,
-                            deltaInt32(elems[4].Obj()["y"_sd], elems[2].Obj()["y"_sd]),
-                            deltaInt32(elems[5].Obj()["y"_sd], elems[4].Obj()["y"_sd])},
-                           1);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues,
+         deltaInt32(elems[1].Obj()["x"_sd].Array()[1], elems[0].Obj()["x"_sd].Array()[1])},
+        1);
     appendEOO(expected);
     appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    verifyDecompression(expected, elems);
 }
 
-TEST_F(BSONColumnTest, InterleavedAfterNonInterleaved) {
-    BSONColumnBuilder cb("test"_sd);
-
-    std::vector<BSONElement> elems = {createElementInt32(1),
-                                      createElementObj(BSON("x" << 1 << "y" << 2)),
-                                      createElementObj(BSON("x" << 1 << "y" << 3)),
-                                      createElementObj(BSON("x" << 2 << "y" << 4))};
-
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+TEST_F(BSONColumnTest, InterleavedArrayRoot) {
+    std::vector<BSONElement> elems = {createElementArray(BSON_ARRAY(1 << 2)),
+                                      createElementArray(BSON_ARRAY(2 << 3))};
 
     BufBuilder expected;
-    appendLiteral(expected, elems.front());
-    appendInterleavedStart(expected, elems[1].Obj());
+    appendInterleavedStartArrayRoot(expected, elems.front().Obj());
     appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected,
-                           {kDeltaForBinaryEqualValues,
-                            deltaInt32(elems[2].Obj()["x"_sd], elems[1].Obj()["x"_sd]),
-                            deltaInt32(elems[3].Obj()["x"_sd], elems[2].Obj()["x"_sd])},
-                           1);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Array()[0], elems[0].Array()[0])},
+        1);
     appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected,
-                           {kDeltaForBinaryEqualValues,
-                            deltaInt32(elems[2].Obj()["y"_sd], elems[1].Obj()["y"_sd]),
-                            deltaInt32(elems[3].Obj()["y"_sd], elems[2].Obj()["y"_sd])},
-                           1);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Array()[1], elems[0].Array()[1])},
+        1);
     appendEOO(expected);
     appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    verifyDecompression(expected, elems);
 }
 
-TEST_F(BSONColumnTest, InterleavedLevels) {
-    BSONColumnBuilder cb("test"_sd);
-
-    std::vector<BSONElement> elems = {createElementObj(BSON("root" << BSON("x" << 1) << "y" << 2)),
-                                      createElementObj(BSON("root" << BSON("x" << 2) << "y" << 5)),
-                                      createElementObj(BSON("root" << BSON("x" << 2) << "y" << 5))};
-
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+TEST_F(BSONColumnTest, InterleavedArrayRootTypeChange) {
+    std::vector<BSONElement> elems = {createElementArray(BSON_ARRAY(1 << 2)),
+                                      createElementArrayAsObject(BSON_ARRAY(2 << 3))};
 
     BufBuilder expected;
-    appendInterleavedStart(expected, elems.front().Obj());
+    appendInterleavedStartArrayRoot(expected, elems.front().Obj());
     appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected,
-                           {kDeltaForBinaryEqualValues,
-                            deltaInt32(elems[1].Obj()["root"_sd].Obj()["x"_sd],
-                                       elems[0].Obj()["root"_sd].Obj()["x"_sd]),
-                            deltaInt32(elems[2].Obj()["root"_sd].Obj()["x"_sd],
-                                       elems[1].Obj()["root"_sd].Obj()["x"_sd])},
-                           1);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
     appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected,
-                           {kDeltaForBinaryEqualValues,
-                            deltaInt32(elems[1].Obj()["y"_sd], elems[0].Obj()["y"_sd]),
-                            deltaInt32(elems[2].Obj()["y"_sd], elems[1].Obj()["y"_sd])},
-                           1);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+    appendEOO(expected);
+
+    appendInterleavedStart(expected, elems[1].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);
+    verifyDecompression(expected, elems);
 }
 
-TEST_F(BSONColumnTest, InterleavedDoubleDifferentScale) {
-    BSONColumnBuilder cb("test"_sd);
+TEST_F(BSONColumnTest, InterleavedAfterNonInterleaved) {
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {createElementInt32(1),
+                                          createElementObj(BSON("x" << 1 << "y" << 2)),
+                                          createElementObj(BSON("x" << 1 << "y" << 3)),
+                                          createElementObj(BSON("x" << 2 << "y" << 4))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
-    std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1.0 << "y" << 2.0)),
-                                      createElementObj(BSON("x" << 1.1 << "y" << 3.0)),
-                                      createElementObj(BSON("x" << 1.2 << "y" << 2.0)),
-                                      createElementObj(BSON("x" << 1.0)),
-                                      createElementObj(BSON("x" << 1.5 << "y" << 2.0))};
+        BufBuilder expected;
+        appendLiteral(expected, elems.front());
+        appendInterleavedStartFunc(expected, elems[1].Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaInt32(elems[2].Obj()["x"_sd], elems[1].Obj()["x"_sd]),
+                                deltaInt32(elems[3].Obj()["x"_sd], elems[2].Obj()["x"_sd])},
+                               1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaInt32(elems[2].Obj()["y"_sd], elems[1].Obj()["y"_sd]),
+                                deltaInt32(elems[3].Obj()["y"_sd], elems[2].Obj()["y"_sd])},
+                               1);
+        appendEOO(expected);
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
 
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
 
-    BufBuilder expected;
-    appendInterleavedStart(expected, elems.front().Obj());
-    appendSimple8bControl(expected, 0b1010, 0b0000);
-    appendSimple8bBlocks64(expected,
-                           {kDeltaForBinaryEqualValues,
-                            deltaDouble(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd], 10),
-                            deltaDouble(elems[2].Obj()["x"_sd], elems[1].Obj()["x"_sd], 10),
-                            deltaDouble(elems[3].Obj()["x"_sd], elems[2].Obj()["x"_sd], 10),
-                            deltaDouble(elems[4].Obj()["x"_sd], elems[3].Obj()["x"_sd], 10)},
-                           1);
-    appendSimple8bControl(expected, 0b1001, 0b0000);
-    appendSimple8bBlocks64(expected,
-                           {kDeltaForBinaryEqualValues,
-                            deltaDouble(elems[1].Obj()["y"_sd], elems[0].Obj()["y"_sd], 1),
-                            deltaDouble(elems[2].Obj()["y"_sd], elems[1].Obj()["y"_sd], 1),
-                            boost::none,
-                            deltaDouble(elems[4].Obj()["y"_sd], elems[2].Obj()["y"_sd], 1)},
-                           1);
-    appendEOO(expected);
-    appendEOO(expected);
+TEST_F(BSONColumnTest, InterleavedLevels) {
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {
+            createElementObj(BSON("root" << BSON("x" << 1) << "y" << 2)),
+            createElementObj(BSON("root" << BSON("x" << 2) << "y" << 5)),
+            createElementObj(BSON("root" << BSON("x" << 2) << "y" << 5))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+        BufBuilder expected;
+        appendInterleavedStartFunc(expected, elems.front().Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaInt32(elems[1].Obj()["root"_sd].Obj()["x"_sd],
+                                           elems[0].Obj()["root"_sd].Obj()["x"_sd]),
+                                deltaInt32(elems[2].Obj()["root"_sd].Obj()["x"_sd],
+                                           elems[1].Obj()["root"_sd].Obj()["x"_sd])},
+                               1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaInt32(elems[1].Obj()["y"_sd], elems[0].Obj()["y"_sd]),
+                                deltaInt32(elems[2].Obj()["y"_sd], elems[1].Obj()["y"_sd])},
+                               1);
+        appendEOO(expected);
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
+
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
 }
 
-TEST_F(BSONColumnTest, InterleavedDoubleIncreaseScaleFromDeltaNoRescale) {
-    BSONColumnBuilder cb("test"_sd);
+TEST_F(BSONColumnTest, InterleavedDoubleDifferentScale) {
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1.0 << "y" << 2.0)),
+                                          createElementObj(BSON("x" << 1.1 << "y" << 3.0)),
+                                          createElementObj(BSON("x" << 1.2 << "y" << 2.0)),
+                                          createElementObj(BSON("x" << 1.0)),
+                                          createElementObj(BSON("x" << 1.5 << "y" << 2.0))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
-    std::vector<BSONElement> elems;
-    elems.push_back(createElementObj(BSON("x" << 1.1)));
-    elems.push_back(createElementObj(BSON("x" << 2.1)));
-    elems.push_back(createElementObj(BSON("x" << 2.2)));
-    elems.push_back(createElementObj(BSON("x" << 2.3)));
-    elems.push_back(createElementObj(BSON("x" << 3.12345678)));
+        BufBuilder expected;
+        appendInterleavedStartFunc(expected, elems.front().Obj());
+        appendSimple8bControl(expected, 0b1010, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaDouble(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd], 10),
+                                deltaDouble(elems[2].Obj()["x"_sd], elems[1].Obj()["x"_sd], 10),
+                                deltaDouble(elems[3].Obj()["x"_sd], elems[2].Obj()["x"_sd], 10),
+                                deltaDouble(elems[4].Obj()["x"_sd], elems[3].Obj()["x"_sd], 10)},
+                               1);
+        appendSimple8bControl(expected, 0b1001, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaDouble(elems[1].Obj()["y"_sd], elems[0].Obj()["y"_sd], 1),
+                                deltaDouble(elems[2].Obj()["y"_sd], elems[1].Obj()["y"_sd], 1),
+                                boost::none,
+                                deltaDouble(elems[4].Obj()["y"_sd], elems[2].Obj()["y"_sd], 1)},
+                               1);
+        appendEOO(expected);
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
 
-    for (const auto& elem : elems) {
-        cb.append(elem);
-    }
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
 
-    BufBuilder expected;
-    appendInterleavedStart(expected, elems.front().Obj());
+TEST_F(BSONColumnTest, InterleavedDoubleIncreaseScaleFromDeltaNoRescale) {
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems;
+        elems.push_back(createElementObj(BSON("x" << 1.1)));
+        elems.push_back(createElementObj(BSON("x" << 2.1)));
+        elems.push_back(createElementObj(BSON("x" << 2.2)));
+        elems.push_back(createElementObj(BSON("x" << 2.3)));
+        elems.push_back(createElementObj(BSON("x" << 3.12345678)));
+
+        if (!arrayCompression) {
+            for (const auto& elem : elems) {
+                cb.append(elem);
+            }
+        }
 
-    appendSimple8bControl(expected, 0b1010, 0b0000);
-    appendSimple8bBlocks64(expected,
-                           {kDeltaForBinaryEqualValues,
-                            deltaDouble(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd], 10),
-                            deltaDouble(elems[2].Obj()["x"_sd], elems[1].Obj()["x"_sd], 10),
-                            deltaDouble(elems[3].Obj()["x"_sd], elems[2].Obj()["x"_sd], 10)},
-                           1);
-    appendSimple8bControl(expected, 0b1101, 0b0000);
-    appendSimple8bBlocks64(
-        expected, {deltaDouble(elems[4].Obj()["x"_sd], elems[3].Obj()["x"_sd], 100000000)}, 1);
-    appendEOO(expected);
-    appendEOO(expected);
+        BufBuilder expected;
+        appendInterleavedStartFunc(expected, elems.front().Obj());
+
+        appendSimple8bControl(expected, 0b1010, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaDouble(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd], 10),
+                                deltaDouble(elems[2].Obj()["x"_sd], elems[1].Obj()["x"_sd], 10),
+                                deltaDouble(elems[3].Obj()["x"_sd], elems[2].Obj()["x"_sd], 10)},
+                               1);
+        appendSimple8bControl(expected, 0b1101, 0b0000);
+        appendSimple8bBlocks64(
+            expected, {deltaDouble(elems[4].Obj()["x"_sd], elems[3].Obj()["x"_sd], 100000000)}, 1);
+        appendEOO(expected);
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
 }
 
 TEST_F(BSONColumnTest, InterleavedMix64And128Bit) {
-    BSONColumnBuilder cb("test"_sd);
-
-    std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1 << "y"
-                                                                << "count0")),
-                                      createElementObj(BSON("x" << 2 << "y"
-                                                                << "count1")),
-                                      createElementObj(BSON("x" << 3 << "y"
-                                                                << "count2")),
-                                      createElementObj(BSON("x" << 4)),
-                                      createElementObj(BSON("x" << 5 << "y"
-                                                                << "count3"))};
-
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1 << "y"
+                                                                    << "count0")),
+                                          createElementObj(BSON("x" << 2 << "y"
+                                                                    << "count1")),
+                                          createElementObj(BSON("x" << 3 << "y"
+                                                                    << "count2")),
+                                          createElementObj(BSON("x" << 4)),
+                                          createElementObj(BSON("x" << 5 << "y"
+                                                                    << "count3"))};
+
+        if (!arrayCompression) {
+            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]),
-                            deltaInt32(elems[2].Obj()["x"_sd], elems[1].Obj()["x"_sd]),
-                            deltaInt32(elems[3].Obj()["x"_sd], elems[2].Obj()["x"_sd]),
-                            deltaInt32(elems[4].Obj()["x"_sd], elems[3].Obj()["x"_sd])},
-                           1);
-    appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks128(expected,
-                            {kDeltaForBinaryEqualValues128,
-                             deltaString(elems[1].Obj()["y"_sd], elems[0].Obj()["y"_sd]),
-                             deltaString(elems[2].Obj()["y"_sd], elems[1].Obj()["y"_sd]),
-                             boost::none,
-                             deltaString(elems[4].Obj()["y"_sd], elems[2].Obj()["y"_sd])},
-                            1);
-    appendEOO(expected);
-    appendEOO(expected);
+        BufBuilder expected;
+        appendInterleavedStartFunc(expected, elems.front().Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaInt32(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd]),
+                                deltaInt32(elems[2].Obj()["x"_sd], elems[1].Obj()["x"_sd]),
+                                deltaInt32(elems[3].Obj()["x"_sd], elems[2].Obj()["x"_sd]),
+                                deltaInt32(elems[4].Obj()["x"_sd], elems[3].Obj()["x"_sd])},
+                               1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks128(expected,
+                                {kDeltaForBinaryEqualValues128,
+                                 deltaString(elems[1].Obj()["y"_sd], elems[0].Obj()["y"_sd]),
+                                 deltaString(elems[2].Obj()["y"_sd], elems[1].Obj()["y"_sd]),
+                                 boost::none,
+                                 deltaString(elems[4].Obj()["y"_sd], elems[2].Obj()["y"_sd])},
+                                1);
+        appendEOO(expected);
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
 }
 
 TEST_F(BSONColumnTest, InterleavedWithEmptySubObj) {
-    BSONColumnBuilder cb("test"_sd);
-
-    std::vector<BSONElement> elems = {
-        createElementObj(BSON("x" << 1 << "y" << BSONObjBuilder().obj())),
-        createElementObj(BSON("x" << 2 << "y" << BSONObjBuilder().obj())),
-        createElementObj(BSON("x" << 3 << "y" << BSONObjBuilder().obj()))};
-
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {
+            createElementObj(BSON("x" << 1 << "y" << BSONObjBuilder().obj())),
+            createElementObj(BSON("x" << 2 << "y" << BSONObjBuilder().obj())),
+            createElementObj(BSON("x" << 3 << "y" << BSONObjBuilder().obj()))};
+
+        if (!arrayCompression) {
+            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]),
-                            deltaInt32(elems[2].Obj()["x"_sd], elems[1].Obj()["x"_sd])},
-                           1);
-    appendEOO(expected);
-    appendEOO(expected);
+        BufBuilder expected;
+        appendInterleavedStartFunc(expected, elems.front().Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaInt32(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd]),
+                                deltaInt32(elems[2].Obj()["x"_sd], elems[1].Obj()["x"_sd])},
+                               1);
+        appendEOO(expected);
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
 }
 
 TEST_F(BSONColumnTest, InterleavedRemoveEmptySubObj) {
-    BSONColumnBuilder cb("test"_sd);
-
-    std::vector<BSONElement> elems = {
-        createElementObj(BSON("x" << 1 << "y" << BSONObjBuilder().obj())),
-        createElementObj(BSON("x" << 2 << "y" << BSONObjBuilder().obj())),
-        createElementObj(BSON("x" << 3))};
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {
+            createElementObj(BSON("x" << 1 << "y" << BSONObjBuilder().obj())),
+            createElementObj(BSON("x" << 2 << "y" << BSONObjBuilder().obj())),
+            createElementObj(BSON("x" << 3))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+        BufBuilder expected;
+        appendInterleavedStartFunc(expected, elems.front().Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaInt32(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd])},
+                               1);
+        appendEOO(expected);
 
-    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);
-    appendEOO(expected);
+        appendInterleavedStartFunc(expected, elems[2].Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
 
-    appendInterleavedStart(expected, elems[2].Obj());
-    appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
-    appendEOO(expected);
+        appendEOO(expected);
 
-    appendEOO(expected);
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
 }
 
 TEST_F(BSONColumnTest, InterleavedAddEmptySubObj) {
-    BSONColumnBuilder cb("test"_sd);
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {
+            createElementObj(BSON("x" << 3)),
+            createElementObj(BSON("x" << 1 << "y" << BSONObjBuilder().obj()))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
+        BufBuilder expected;
+        appendInterleavedStartFunc(expected, elems.front().Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
+
+        appendInterleavedStartFunc(expected, elems[1].Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
+
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
+
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
+
+TEST_F(BSONColumnTest, InterleavedAddEmptySubArray) {
     std::vector<BSONElement> elems = {
         createElementObj(BSON("x" << 3)),
-        createElementObj(BSON("x" << 1 << "y" << BSONObjBuilder().obj()))};
-
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+        createElementObj(BSON("x" << 1 << "y" << BSONArrayBuilder().arr()))};
 
     BufBuilder expected;
     appendInterleavedStart(expected, elems.front().Obj());
@@ -2792,65 +2990,291 @@ TEST_F(BSONColumnTest, InterleavedAddEmptySubObj) {
 
     appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    verifyDecompression(expected, elems);
 }
 
 TEST_F(BSONColumnTest, InterleavedSchemaChange) {
-    BSONColumnBuilder cb("test"_sd);
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        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" << 3.0)),
+                                          createElementObj(BSON("x" << 1 << "y" << 4.0))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
-    std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1 << "y" << 2)),
-                                      createElementObj(BSON("x" << 1 << "y" << 3)),
-                                      createElementObj(BSON("x" << 1 << "y" << 3.0)),
-                                      createElementObj(BSON("x" << 1 << "y" << 4.0))};
+        BufBuilder expected;
+        appendInterleavedStartFunc(expected, elems.front().Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaInt32(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd]),
+                                deltaInt32(elems[2].Obj()["x"_sd], elems[1].Obj()["x"_sd]),
+                                deltaInt32(elems[3].Obj()["x"_sd], elems[2].Obj()["x"_sd])},
+                               1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaInt32(elems[1].Obj()["y"_sd], elems[0].Obj()["y"_sd])},
+                               1);
+        appendLiteral(expected, elems[2].Obj()["y"_sd]);
+        appendSimple8bControl(expected, 0b1001, 0b0000);
+        appendSimple8bBlocks64(
+            expected, {deltaDouble(elems[3].Obj()["y"_sd], elems[2].Obj()["y"_sd], 1)}, 1);
+        appendEOO(expected);
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
 
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
+
+TEST_F(BSONColumnTest, InterleavedObjectSchemaChange) {
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1 << "y" << BSON("z" << 2))),
+                                          createElementObj(BSON("x" << 1 << "y" << BSON("z" << 3))),
+                                          createElementObj(BSON("x" << 1 << "y" << 3))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
+
+        BufBuilder expected;
+        appendInterleavedStartFunc(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].Obj()["z"_sd],
+                                           elems[0].Obj()["y"_sd].Obj()["z"_sd])},
+                               1);
+        appendEOO(expected);
+
+        appendInterleavedStartFunc(expected, elems[2].Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
+
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
+
+TEST_F(BSONColumnTest, InterleavedObjectNameChange) {
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {
+            createElementObj(BSON("x" << 1 << "y" << BSON("z" << 2))),
+            createElementObj(BSON("x" << 1 << "y2" << BSON("z" << 3)))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
+
+        BufBuilder expected;
+        appendInterleavedStartFunc(
+            expected, BSON("x" << 1 << "y" << BSON("z" << 2) << "y2" << BSON("z" << 3)));
+        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, boost::none}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {boost::none, kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
+
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
+
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
+
+TEST_F(BSONColumnTest, InterleavedObjectEmptyObjChange) {
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {
+            createElementObj(BSON("x" << 1 << "y" << BSON("z" << 2))),
+            createElementObj(BSON("x" << 1 << "y" << BSON("z" << 3))),
+            createElementObj(BSON("x" << 1 << "y" << BSONObjBuilder().obj()))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
+
+        BufBuilder expected;
+        appendInterleavedStartFunc(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].Obj()["z"_sd],
+                                           elems[0].Obj()["y"_sd].Obj()["z"_sd])},
+                               1);
+        appendEOO(expected);
+
+        appendInterleavedStartFunc(expected, elems[2].Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
+
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
+
+TEST_F(BSONColumnTest, InterleavedObjectEmptyArrayChange) {
+    std::vector<BSONElement> elems = {
+        createElementObj(BSON("x" << 1 << "y" << BSON("z" << 2))),
+        createElementObj(BSON("x" << 1 << "y" << BSON("z" << 3))),
+        createElementObj(BSON("x" << 1 << "y" << BSONArrayBuilder().arr()))};
 
     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]),
-                            deltaInt32(elems[2].Obj()["x"_sd], elems[1].Obj()["x"_sd]),
-                            deltaInt32(elems[3].Obj()["x"_sd], elems[2].Obj()["x"_sd])},
-                           1);
-    appendSimple8bControl(expected, 0b1000, 0b0000);
     appendSimple8bBlocks64(
         expected,
-        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["y"_sd], elems[0].Obj()["y"_sd])},
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd])},
         1);
-    appendLiteral(expected, elems[2].Obj()["y"_sd]);
-    appendSimple8bControl(expected, 0b1001, 0b0000);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
     appendSimple8bBlocks64(
-        expected, {deltaDouble(elems[3].Obj()["y"_sd], elems[2].Obj()["y"_sd], 1)}, 1);
+        expected,
+        {kDeltaForBinaryEqualValues,
+         deltaInt32(elems[1].Obj()["y"_sd].Obj()["z"_sd], elems[0].Obj()["y"_sd].Obj()["z"_sd])},
+        1);
+    appendEOO(expected);
+
+    appendInterleavedStart(expected, elems[2].Obj());
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
     appendEOO(expected);
     appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    verifyDecompression(expected, elems);
 }
 
-TEST_F(BSONColumnTest, InterleavedObjectSchemaChange) {
-    BSONColumnBuilder cb("test"_sd);
+TEST_F(BSONColumnTest, InterleavedObjectNewEmptyObjMiddle) {
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        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))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
-    std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1 << "y" << BSON("z" << 2))),
-                                      createElementObj(BSON("x" << 1 << "y" << BSON("z" << 3))),
-                                      createElementObj(BSON("x" << 1 << "y" << 3))};
+        BufBuilder expected;
+        appendInterleavedStartFunc(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);
 
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+        appendInterleavedStartFunc(expected, elems[2].Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
+
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
+
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
+
+TEST_F(BSONColumnTest, InterleavedObjectNewEmptyArrayMiddle) {
+    std::vector<BSONElement> elems = {
+        createElementObj(BSON("x" << 1 << "z" << 2)),
+        createElementObj(BSON("x" << 1 << "z" << 3)),
+        createElementObj(BSON("x" << 1 << "y" << BSONArrayBuilder().arr() << "z" << 4))};
 
     BufBuilder expected;
     appendInterleavedStart(expected, elems.front().Obj());
@@ -2862,8 +3286,7 @@ TEST_F(BSONColumnTest, InterleavedObjectSchemaChange) {
     appendSimple8bControl(expected, 0b1000, 0b0000);
     appendSimple8bBlocks64(
         expected,
-        {kDeltaForBinaryEqualValues,
-         deltaInt32(elems[1].Obj()["y"_sd].Obj()["z"_sd], elems[0].Obj()["y"_sd].Obj()["z"_sd])},
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["z"_sd], elems[0].Obj()["z"_sd])},
         1);
     appendEOO(expected);
 
@@ -2873,61 +3296,155 @@ TEST_F(BSONColumnTest, InterleavedObjectSchemaChange) {
     appendSimple8bControl(expected, 0b1000, 0b0000);
     appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
     appendEOO(expected);
+
     appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    verifyDecompression(expected, elems);
 }
 
-TEST_F(BSONColumnTest, InterleavedObjectNameChange) {
-    BSONColumnBuilder cb("test"_sd);
+TEST_F(BSONColumnTest, InterleavedObjectNewEmptyObjUnderObj) {
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        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))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
+
+        BufBuilder expected;
+        appendInterleavedStartFunc(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);
 
-    std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1 << "y" << BSON("z" << 2))),
-                                      createElementObj(BSON("x" << 1 << "y2" << BSON("z" << 3)))};
+        appendInterleavedStartFunc(expected, elems[2].Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
 
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
+
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
+
+TEST_F(BSONColumnTest, InterleavedObjectNewEmptyArrayUnderObj) {
+    std::vector<BSONElement> elems = {
+        createElementObj(BSON("x" << 1 << "z" << 2)),
+        createElementObj(BSON("x" << 1 << "z" << 3)),
+        createElementObj(
+            BSON("x" << 1 << "y" << BSON("y1" << BSONArrayBuilder().arr()) << "z" << 4))};
 
     BufBuilder expected;
-    appendInterleavedStart(expected,
-                           BSON("x" << 1 << "y" << BSON("z" << 2) << "y2" << BSON("z" << 3)));
+    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, boost::none}, 1);
+    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, {boost::none, kDeltaForBinaryEqualValues}, 1);
+    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);
+    verifyDecompression(expected, elems);
 }
 
-TEST_F(BSONColumnTest, InterleavedObjectEmptyObjChange) {
-    BSONColumnBuilder cb("test"_sd);
+TEST_F(BSONColumnTest, InterleavedObjectNewEmptyObjEnd) {
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        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()))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
-    std::vector<BSONElement> elems = {
-        createElementObj(BSON("x" << 1 << "y" << BSON("z" << 2))),
-        createElementObj(BSON("x" << 1 << "y" << BSON("z" << 3))),
-        createElementObj(BSON("x" << 1 << "y" << BSONObjBuilder().obj()))};
+        BufBuilder expected;
+        appendInterleavedStartFunc(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);
 
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+        appendInterleavedStartFunc(expected, elems[2].Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
+
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
+
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
+
+TEST_F(BSONColumnTest, InterleavedObjectNewEmptyArrayEnd) {
+    std::vector<BSONElement> elems = {
+        createElementObj(BSON("x" << 1 << "y" << 2)),
+        createElementObj(BSON("x" << 1 << "y" << 3)),
+        createElementObj(BSON("x" << 1 << "y" << 4 << "z" << BSONArrayBuilder().arr()))};
 
     BufBuilder expected;
     appendInterleavedStart(expected, elems.front().Obj());
@@ -2939,36 +3456,81 @@ TEST_F(BSONColumnTest, InterleavedObjectEmptyObjChange) {
     appendSimple8bControl(expected, 0b1000, 0b0000);
     appendSimple8bBlocks64(
         expected,
-        {kDeltaForBinaryEqualValues,
-         deltaInt32(elems[1].Obj()["y"_sd].Obj()["z"_sd], elems[0].Obj()["y"_sd].Obj()["z"_sd])},
+        {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);
+    verifyDecompression(expected, elems);
 }
 
-TEST_F(BSONColumnTest, InterleavedObjectNewEmptyObjMiddle) {
-    BSONColumnBuilder cb("test"_sd);
+TEST_F(BSONColumnTest, InterleavedObjectNewEmptyObjUnderObjEnd) {
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        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" << BSON("z1" << BSONObjBuilder().obj())))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
-    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))};
+        BufBuilder expected;
+        appendInterleavedStartFunc(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);
 
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+        appendInterleavedStartFunc(expected, elems[2].Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
+
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
+
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
+
+TEST_F(BSONColumnTest, InterleavedObjectNewEmptyArrayUnderObjEnd) {
+    std::vector<BSONElement> elems = {
+        createElementObj(BSON("x" << 1 << "y" << 2)),
+        createElementObj(BSON("x" << 1 << "y" << 3)),
+        createElementObj(
+            BSON("x" << 1 << "y" << 4 << "z" << BSON("z1" << BSONArrayBuilder().arr())))};
 
     BufBuilder expected;
     appendInterleavedStart(expected, elems.front().Obj());
@@ -2980,7 +3542,7 @@ TEST_F(BSONColumnTest, InterleavedObjectNewEmptyObjMiddle) {
     appendSimple8bControl(expected, 0b1000, 0b0000);
     appendSimple8bBlocks64(
         expected,
-        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["z"_sd], elems[0].Obj()["z"_sd])},
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["y"_sd], elems[0].Obj()["y"_sd])},
         1);
     appendEOO(expected);
 
@@ -2993,26 +3555,15 @@ TEST_F(BSONColumnTest, InterleavedObjectNewEmptyObjMiddle) {
 
     appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    verifyDecompression(expected, elems);
 }
 
-TEST_F(BSONColumnTest, InterleavedObjectNewEmptyObjUnderObj) {
-    BSONColumnBuilder cb("test"_sd);
-
+TEST_F(BSONColumnTest, InterleavedObjectNewEmptyArrayUnderArrayEnd) {
     std::vector<BSONElement> elems = {
-        createElementObj(BSON("x" << 1 << "z" << 2)),
-        createElementObj(BSON("x" << 1 << "z" << 3)),
+        createElementObj(BSON("x" << 1 << "y" << 2)),
+        createElementObj(BSON("x" << 1 << "y" << 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();
-    }
+            BSON("x" << 1 << "y" << 4 << "z" << BSON_ARRAY(BSONArrayBuilder().arr())))};
 
     BufBuilder expected;
     appendInterleavedStart(expected, elems.front().Obj());
@@ -3024,7 +3575,7 @@ TEST_F(BSONColumnTest, InterleavedObjectNewEmptyObjUnderObj) {
     appendSimple8bControl(expected, 0b1000, 0b0000);
     appendSimple8bBlocks64(
         expected,
-        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["z"_sd], elems[0].Obj()["z"_sd])},
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["y"_sd], elems[0].Obj()["y"_sd])},
         1);
     appendEOO(expected);
 
@@ -3037,25 +3588,66 @@ TEST_F(BSONColumnTest, InterleavedObjectNewEmptyObjUnderObj) {
 
     appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    verifyDecompression(expected, elems);
 }
 
-TEST_F(BSONColumnTest, InterleavedObjectNewEmptyObjEnd) {
-    BSONColumnBuilder cb("test"_sd);
+TEST_F(BSONColumnTest, InterleavedObjectMissingEmptyObjMiddle) {
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        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))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
-    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()))};
+        BufBuilder expected;
+        appendInterleavedStartFunc(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);
 
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+        appendInterleavedStartFunc(expected, elems[2].Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
+
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
+
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
+
+TEST_F(BSONColumnTest, InterleavedObjectMissingEmptyArrayMiddle) {
+    std::vector<BSONElement> elems = {
+        createElementObj(BSON("x" << 1 << "y" << BSONArrayBuilder().arr() << "z" << 2)),
+        createElementObj(BSON("x" << 1 << "y" << BSONArrayBuilder().arr() << "z" << 3)),
+        createElementObj(BSON("x" << 1 << "z" << 4))};
 
     BufBuilder expected;
     appendInterleavedStart(expected, elems.front().Obj());
@@ -3067,7 +3659,7 @@ TEST_F(BSONColumnTest, InterleavedObjectNewEmptyObjEnd) {
     appendSimple8bControl(expected, 0b1000, 0b0000);
     appendSimple8bBlocks64(
         expected,
-        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["y"_sd], elems[0].Obj()["y"_sd])},
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["z"_sd], elems[0].Obj()["z"_sd])},
         1);
     appendEOO(expected);
 
@@ -3080,42 +3672,30 @@ TEST_F(BSONColumnTest, InterleavedObjectNewEmptyObjEnd) {
 
     appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    verifyDecompression(expected, elems);
 }
 
-TEST_F(BSONColumnTest, InterleavedObjectNewEmptyObjUnderObjEnd) {
-    BSONColumnBuilder cb("test"_sd);
-
+TEST_F(BSONColumnTest, InterleavedArrayMissingEmptyObjMiddle) {
     std::vector<BSONElement> elems = {
-        createElementObj(BSON("x" << 1 << "y" << 2)),
-        createElementObj(BSON("x" << 1 << "y" << 3)),
-        createElementObj(
-            BSON("x" << 1 << "y" << 4 << "z" << BSON("z1" << BSONObjBuilder().obj())))};
-
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+        createElementArray(BSON_ARRAY(1 << BSONObjBuilder().obj() << 2)),
+        createElementArray(BSON_ARRAY(1 << BSONObjBuilder().obj() << 3)),
+        createElementArray(BSON_ARRAY(1 << 4))};
 
     BufBuilder expected;
-    appendInterleavedStart(expected, elems.front().Obj());
+    appendInterleavedStartArrayRoot(expected, elems.front().Obj());
     appendSimple8bControl(expected, 0b1000, 0b0000);
     appendSimple8bBlocks64(
         expected,
-        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd])},
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Array()[0], elems[0].Array()[0])},
         1);
     appendSimple8bControl(expected, 0b1000, 0b0000);
     appendSimple8bBlocks64(
         expected,
-        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["y"_sd], elems[0].Obj()["y"_sd])},
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Array()[2], elems[0].Array()[2])},
         1);
     appendEOO(expected);
 
-    appendInterleavedStart(expected, elems[2].Obj());
+    appendInterleavedStartArrayRoot(expected, elems[2].Obj());
     appendSimple8bControl(expected, 0b1000, 0b0000);
     appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
     appendSimple8bControl(expected, 0b1000, 0b0000);
@@ -3124,26 +3704,71 @@ TEST_F(BSONColumnTest, InterleavedObjectNewEmptyObjUnderObjEnd) {
 
     appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    verifyDecompression(expected, elems);
 }
 
-TEST_F(BSONColumnTest, InterleavedObjectMissingEmptyObjMiddle) {
-    BSONColumnBuilder cb("test"_sd);
+TEST_F(BSONColumnTest, InterleavedObjectMissingEmptyObjUnderObj) {
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        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))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
+
+        BufBuilder expected;
+        appendInterleavedStartFunc(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);
+
+        appendInterleavedStartFunc(expected, elems[2].Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
+
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
+
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
 
+TEST_F(BSONColumnTest, InterleavedObjectMissingEmptyArrayUnderObj) {
     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 << "y" << BSON("y1" << BSONArrayBuilder().arr()) << "z" << 2)),
+        createElementObj(
+            BSON("x" << 1 << "y" << BSON("y1" << BSONArrayBuilder().arr()) << "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);
@@ -3167,26 +3792,15 @@ TEST_F(BSONColumnTest, InterleavedObjectMissingEmptyObjMiddle) {
 
     appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    verifyDecompression(expected, elems);
 }
 
-TEST_F(BSONColumnTest, InterleavedObjectMissingEmptyObjUnderObj) {
-    BSONColumnBuilder cb("test"_sd);
-
+TEST_F(BSONColumnTest, InterleavedObjectMissingEmptyArrayUnderArray) {
     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 << "y" << BSON_ARRAY(BSONArrayBuilder().arr()) << "z" << 2)),
+        createElementObj(BSON("x" << 1 << "y" << BSON_ARRAY(BSONArrayBuilder().arr()) << "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);
@@ -3210,26 +3824,67 @@ TEST_F(BSONColumnTest, InterleavedObjectMissingEmptyObjUnderObj) {
 
     appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    verifyDecompression(expected, elems);
 }
 
 TEST_F(BSONColumnTest, InterleavedObjectMissingEmptyObjEnd) {
-    BSONColumnBuilder cb("test"_sd);
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        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))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
+
+        BufBuilder expected;
+        appendInterleavedStartFunc(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);
 
+        appendInterleavedStartFunc(expected, elems[2].Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
+
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
+
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
+
+TEST_F(BSONColumnTest, InterleavedObjectMissingEmptyArrayEnd) {
     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" << 2 << "z" << BSONArrayBuilder().arr())),
+        createElementObj(BSON("x" << 1 << "y" << 3 << "z" << BSONArrayBuilder().arr())),
         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);
@@ -3253,26 +3908,103 @@ TEST_F(BSONColumnTest, InterleavedObjectMissingEmptyObjEnd) {
 
     appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    verifyDecompression(expected, elems);
+}
+
+TEST_F(BSONColumnTest, InterleavedArrayMissingEmptyArrayEnd) {
+    std::vector<BSONElement> elems = {
+        createElementArray(BSON_ARRAY(1 << 2 << BSONArrayBuilder().arr())),
+        createElementArray(BSON_ARRAY(1 << 3 << BSONArrayBuilder().arr())),
+        createElementArray(BSON_ARRAY(1 << 4))};
+
+    BufBuilder expected;
+    appendInterleavedStartArrayRoot(expected, elems.front().Obj());
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Array()[0], elems[0].Array()[0])},
+        1);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Array()[1], elems[0].Array()[1])},
+        1);
+    appendEOO(expected);
+
+    appendInterleavedStartArrayRoot(expected, elems[2].Obj());
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+    appendEOO(expected);
+
+    appendEOO(expected);
+
+    verifyDecompression(expected, elems);
 }
 
 TEST_F(BSONColumnTest, InterleavedObjectMissingEmptyObjUnderObjEnd) {
-    BSONColumnBuilder cb("test"_sd);
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {
+            createElementObj(
+                BSON("x" << 1 << "y" << 2 << "z" << BSON("z1" << BSONObjBuilder().obj()))),
+            createElementObj(
+                BSON("x" << 1 << "y" << 3 << "z" << BSON("z1" << BSONObjBuilder().obj()))),
+            createElementObj(BSON("x" << 1 << "y" << 4))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
+        BufBuilder expected;
+        appendInterleavedStartFunc(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);
+
+        appendInterleavedStartFunc(expected, elems[2].Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
+
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
+
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
+
+TEST_F(BSONColumnTest, InterleavedObjectMissingEmptyArrayUnderObjEnd) {
     std::vector<BSONElement> elems = {
-        createElementObj(BSON("x" << 1 << "y" << 2 << "z" << BSON("z1" << BSONObjBuilder().obj()))),
-        createElementObj(BSON("x" << 1 << "y" << 3 << "z" << BSON("z1" << BSONObjBuilder().obj()))),
+        createElementObj(
+            BSON("x" << 1 << "y" << 2 << "z" << BSON("z1" << BSONArrayBuilder().arr()))),
+        createElementObj(
+            BSON("x" << 1 << "y" << 3 << "z" << BSON("z1" << BSONArrayBuilder().arr()))),
         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);
@@ -3296,38 +4028,122 @@ TEST_F(BSONColumnTest, InterleavedObjectMissingEmptyObjUnderObjEnd) {
 
     appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    verifyDecompression(expected, elems);
+}
+
+TEST_F(BSONColumnTest, InterleavedArrayMissingEmptyObjUnderObjEnd) {
+    std::vector<BSONElement> elems = {
+        createElementArray(BSON_ARRAY(1 << 2 << BSON("z1" << BSONObjBuilder().obj()))),
+        createElementArray(BSON_ARRAY(1 << 3 << BSON("z1" << BSONObjBuilder().obj()))),
+        createElementArray(BSON_ARRAY(1 << 4))};
+
+    BufBuilder expected;
+    appendInterleavedStartArrayRoot(expected, elems.front().Obj());
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Array()[0], elems[0].Array()[0])},
+        1);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Array()[1], elems[0].Array()[1])},
+        1);
+    appendEOO(expected);
+
+    appendInterleavedStartArrayRoot(expected, elems[2].Obj());
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+    appendEOO(expected);
+
+    appendEOO(expected);
+
+    verifyDecompression(expected, elems);
 }
 
 TEST_F(BSONColumnTest, ReenterInterleaved) {
-    BSONColumnBuilder cb("test"_sd);
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1 << "y" << 2)),
+                                          createElementObj(BSON("x" << 1 << "y" << 3)),
+                                          createElementInt32(1),
+                                          createElementObj(BSON("x" << 2 << "y" << 2 << "z" << 2)),
+                                          createElementObj(BSON("x" << 5 << "y" << 3 << "z" << 3))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
-    std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1 << "y" << 2)),
-                                      createElementObj(BSON("x" << 1 << "y" << 3)),
+        BufBuilder expected;
+        appendInterleavedStartFunc(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);
+        appendLiteral(expected, elems[2]);
+        appendInterleavedStartFunc(expected, elems[3].Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaInt32(elems[4].Obj()["x"_sd], elems[3].Obj()["x"_sd])},
+                               1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaInt32(elems[4].Obj()["y"_sd], elems[3].Obj()["y"_sd])},
+                               1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaInt32(elems[4].Obj()["z"_sd], elems[3].Obj()["z"_sd])},
+                               1);
+        appendEOO(expected);
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
+
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
+
+TEST_F(BSONColumnTest, ReenterInterleavedArrayRootToObj) {
+    std::vector<BSONElement> elems = {createElementArray(BSON_ARRAY(1 << 2)),
+                                      createElementArray(BSON_ARRAY(1 << 3)),
                                       createElementInt32(1),
                                       createElementObj(BSON("x" << 2 << "y" << 2 << "z" << 2)),
                                       createElementObj(BSON("x" << 5 << "y" << 3 << "z" << 3))};
 
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
-
     BufBuilder expected;
-    appendInterleavedStart(expected, elems.front().Obj());
+    appendInterleavedStartArrayRoot(expected, elems.front().Obj());
     appendSimple8bControl(expected, 0b1000, 0b0000);
     appendSimple8bBlocks64(
         expected,
-        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd])},
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Array()[0], elems[0].Array()[0])},
         1);
     appendSimple8bControl(expected, 0b1000, 0b0000);
     appendSimple8bBlocks64(
         expected,
-        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["y"_sd], elems[0].Obj()["y"_sd])},
+        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Array()[1], elems[0].Array()[1])},
         1);
     appendEOO(expected);
     appendLiteral(expected, elems[2]);
@@ -3350,87 +4166,142 @@ TEST_F(BSONColumnTest, ReenterInterleaved) {
     appendEOO(expected);
     appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    verifyDecompression(expected, elems);
 }
 
 TEST_F(BSONColumnTest, InterleavedAlternatingMergeRight) {
-    BSONColumnBuilder cb("test"_sd);
-
-    std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1)),
-                                      createElementObj(BSON("y" << 2)),
-                                      createElementObj(BSON("z" << 3)),
-                                      createElementObj(BSON("x" << 2)),
-                                      createElementObj(BSON("y" << 3)),
-                                      createElementObj(BSON("z" << 4))};
-
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1)),
+                                          createElementObj(BSON("y" << 2)),
+                                          createElementObj(BSON("z" << 3)),
+                                          createElementObj(BSON("x" << 2)),
+                                          createElementObj(BSON("y" << 3)),
+                                          createElementObj(BSON("z" << 4))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
-    BufBuilder expected;
-    appendInterleavedStart(expected,
-                           BSON("x" << elems[0].Obj().firstElement().Int() << "y"
-                                    << elems[1].Obj().firstElement().Int() << "z"
-                                    << elems[2].Obj().firstElement().Int()));
-    appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected,
-                           {kDeltaForBinaryEqualValues,
-                            boost::none,
-                            boost::none,
-                            deltaInt32(elems[3].Obj()["x"_sd], elems[0].Obj()["x"_sd]),
-                            boost::none,
-                            boost::none},
-                           1);
-    appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected,
-                           {boost::none,
-                            kDeltaForBinaryEqualValues,
-                            boost::none,
-                            boost::none,
-                            deltaInt32(elems[4].Obj()["y"_sd], elems[1].Obj()["y"_sd]),
-                            boost::none},
-                           1);
-    appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected,
-                           {boost::none,
-                            boost::none,
-                            kDeltaForBinaryEqualValues,
-                            boost::none,
-                            boost::none,
-                            deltaInt32(elems[5].Obj()["z"_sd], elems[2].Obj()["z"_sd])},
-                           1);
-    appendEOO(expected);
-    appendEOO(expected);
+        BufBuilder expected;
+        appendInterleavedStartFunc(expected,
+                                   BSON("x" << elems[0].Obj().firstElement().Int() << "y"
+                                            << elems[1].Obj().firstElement().Int() << "z"
+                                            << elems[2].Obj().firstElement().Int()));
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                boost::none,
+                                boost::none,
+                                deltaInt32(elems[3].Obj()["x"_sd], elems[0].Obj()["x"_sd]),
+                                boost::none,
+                                boost::none},
+                               1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {boost::none,
+                                kDeltaForBinaryEqualValues,
+                                boost::none,
+                                boost::none,
+                                deltaInt32(elems[4].Obj()["y"_sd], elems[1].Obj()["y"_sd]),
+                                boost::none},
+                               1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {boost::none,
+                                boost::none,
+                                kDeltaForBinaryEqualValues,
+                                boost::none,
+                                boost::none,
+                                deltaInt32(elems[5].Obj()["z"_sd], elems[2].Obj()["z"_sd])},
+                               1);
+        appendEOO(expected);
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
 }
 
 TEST_F(BSONColumnTest, InterleavedAlternatingMergeLeftThenRight) {
-    BSONColumnBuilder cb("test"_sd);
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {createElementObj(BSON("z" << 1)),
+                                          createElementObj(BSON("y" << 2 << "z" << 2)),
+                                          createElementObj(BSON("x" << 3 << "z" << 3))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
-    std::vector<BSONElement> elems = {createElementObj(BSON("z" << 1)),
-                                      createElementObj(BSON("y" << 2 << "z" << 2)),
-                                      createElementObj(BSON("x" << 3 << "z" << 3))};
+        BufBuilder expected;
+        appendInterleavedStartFunc(expected,
+                                   BSON("y" << elems[1].Obj().firstElement().Int() << "x"
+                                            << elems[2].Obj().firstElement().Int() << "z"
+                                            << elems[0].Obj().firstElement().Int()));
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {boost::none, kDeltaForBinaryEqualValues, boost::none}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {boost::none, boost::none, kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaInt32(elems[1].Obj()["z"_sd], elems[0].Obj()["z"_sd]),
+                                deltaInt32(elems[2].Obj()["z"_sd], elems[1].Obj()["z"_sd])},
+                               1);
+        appendEOO(expected);
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
 
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
+
+TEST_F(BSONColumnTest, InterleavedMergeWithUnrelatedArray) {
+    std::vector<BSONElement> elems = {
+        createElementObj(BSON("a" << BSON_ARRAY(1 << 2) << "z" << 1)),
+        createElementObj(BSON("a" << BSON_ARRAY(1 << 2) << "y" << 2 << "z" << 2)),
+        createElementObj(BSON("a" << BSON_ARRAY(1 << 2) << "x" << 3 << "z" << 3))};
 
     BufBuilder expected;
     appendInterleavedStart(expected,
-                           BSON("y" << elems[1].Obj().firstElement().Int() << "x"
-                                    << elems[2].Obj().firstElement().Int() << "z"
-                                    << elems[0].Obj().firstElement().Int()));
+                           BSON("a" << BSON_ARRAY(1 << 2) << "y" << elems[1].Obj()["y"].Int() << "x"
+                                    << elems[2].Obj()["x"].Int() << "z"
+                                    << elems[0].Obj()["z"].Int()));
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, kDeltaForBinaryEqualValues, kDeltaForBinaryEqualValues},
+        1);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(
+        expected,
+        {kDeltaForBinaryEqualValues, kDeltaForBinaryEqualValues, kDeltaForBinaryEqualValues},
+        1);
     appendSimple8bControl(expected, 0b1000, 0b0000);
     appendSimple8bBlocks64(expected, {boost::none, kDeltaForBinaryEqualValues, boost::none}, 1);
     appendSimple8bControl(expected, 0b1000, 0b0000);
@@ -3444,153 +4315,251 @@ TEST_F(BSONColumnTest, InterleavedAlternatingMergeLeftThenRight) {
     appendEOO(expected);
     appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    verifyDecompression(expected, elems);
 }
 
-TEST_F(BSONColumnTest, InterleavedIncompatibleMerge) {
-    BSONColumnBuilder cb("test"_sd);
-
-    std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1)),
-                                      createElementObj(BSON("x" << 2 << "y" << 2)),
-                                      createElementObj(BSON("y" << 3 << "x" << 3))};
-
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+TEST_F(BSONColumnTest, InterleavedArrayAppend) {
+    std::vector<BSONElement> elems = {
+        createElementArray(BSONArrayBuilder().arr()),
+        createElementArray(BSON_ARRAY(1)),
+        createElementArray(BSON_ARRAY(1 << 2)),
+        createElementArray(BSON_ARRAY(1 << 2 << 3)),
+        createElementArray(BSON_ARRAY(1 << 2 << 3 << 4)),
+    };
 
     BufBuilder expected;
-    appendInterleavedStart(
-        expected,
-        BSON("x" << elems[0].Obj().firstElement().Int() << "y" << elems[1].Obj()["y"_sd].Int()));
+    appendLiteral(expected, elems.front());
+    appendInterleavedStartArrayRoot(expected, elems[4].Obj());
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected,
+                           {kDeltaForBinaryEqualValues,
+                            kDeltaForBinaryEqualValues,
+                            kDeltaForBinaryEqualValues,
+                            kDeltaForBinaryEqualValues},
+                           1);
+    appendSimple8bControl(expected, 0b1000, 0b0000);
+    appendSimple8bBlocks64(expected,
+                           {boost::none,
+                            kDeltaForBinaryEqualValues,
+                            kDeltaForBinaryEqualValues,
+                            kDeltaForBinaryEqualValues},
+                           1);
     appendSimple8bControl(expected, 0b1000, 0b0000);
     appendSimple8bBlocks64(
         expected,
-        {kDeltaForBinaryEqualValues, deltaInt32(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd])},
+        {boost::none, boost::none, kDeltaForBinaryEqualValues, kDeltaForBinaryEqualValues},
         1);
     appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected, {boost::none, kDeltaForBinaryEqualValues}, 1);
-    appendEOO(expected);
-    appendInterleavedStart(expected, elems[2].Obj());
-    appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
-    appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+    appendSimple8bBlocks64(
+        expected, {boost::none, boost::none, boost::none, kDeltaForBinaryEqualValues}, 1);
     appendEOO(expected);
     appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    verifyDecompression(expected, elems);
 }
 
-TEST_F(BSONColumnTest, InterleavedIncompatibleMergeMiddle) {
-    BSONColumnBuilder cb("test"_sd);
+TEST_F(BSONColumnTest, InterleavedIncompatibleMerge) {
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1)),
+                                          createElementObj(BSON("x" << 2 << "y" << 2)),
+                                          createElementObj(BSON("y" << 3 << "x" << 3))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
-    std::vector<BSONElement> elems = {
-        createElementObj(BSON("a" << 1 << "x" << 2 << "y" << 2 << "b" << 2)),
-        createElementObj(BSON("a" << 1 << "y" << 3 << "x" << 3 << "b" << 2))};
+        BufBuilder expected;
+        appendInterleavedStartFunc(expected,
+                                   BSON("x" << elems[0].Obj().firstElement().Int() << "y"
+                                            << elems[1].Obj()["y"_sd].Int()));
+        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, {boost::none, kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
+        appendInterleavedStartFunc(expected, elems[2].Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
+        appendEOO(expected);
 
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
 
-    BufBuilder expected;
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
 
-    appendInterleavedStart(expected, elems[0].Obj());
-    appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
-    appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
-    appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
-    appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
-    appendEOO(expected);
+TEST_F(BSONColumnTest, InterleavedIncompatibleMergeMiddle) {
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {
+            createElementObj(BSON("a" << 1 << "x" << 2 << "y" << 2 << "b" << 2)),
+            createElementObj(BSON("a" << 1 << "y" << 3 << "x" << 3 << "b" << 2))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
-    appendInterleavedStart(expected, elems[1].Obj());
-    appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
-    appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
-    appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
-    appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
-    appendEOO(expected);
+        BufBuilder expected;
 
-    appendEOO(expected);
+        appendInterleavedStartFunc(expected, elems[0].Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+        appendInterleavedStartFunc(expected, elems[1].Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
+
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
+
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
 }
 
 TEST_F(BSONColumnTest, InterleavedIncompatibleAfterDeterminedReference) {
-    BSONColumnBuilder cb("test"_sd);
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1)),
+                                          createElementObj(BSON("x" << 2)),
+                                          createElementObj(BSON("x" << 3)),
+                                          createElementObj(BSON("x" << 4)),
+                                          createElementObj(BSON("x" << 5)),
+                                          createElementObj(BSON("x" << 6)),
+                                          createElementObj(BSON("x" << 0 << "y" << 0))};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
-    std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1)),
-                                      createElementObj(BSON("x" << 2)),
-                                      createElementObj(BSON("x" << 3)),
-                                      createElementObj(BSON("x" << 4)),
-                                      createElementObj(BSON("x" << 5)),
-                                      createElementObj(BSON("x" << 6)),
-                                      createElementObj(BSON("x" << 0 << "y" << 0))};
+        BufBuilder expected;
+        appendInterleavedStartFunc(expected, elems.front().Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected,
+                               {kDeltaForBinaryEqualValues,
+                                deltaInt32(elems[1].Obj()["x"_sd], elems[0].Obj()["x"_sd]),
+                                deltaInt32(elems[2].Obj()["x"_sd], elems[1].Obj()["x"_sd]),
+                                deltaInt32(elems[3].Obj()["x"_sd], elems[2].Obj()["x"_sd]),
+                                deltaInt32(elems[4].Obj()["x"_sd], elems[3].Obj()["x"_sd]),
+                                deltaInt32(elems[5].Obj()["x"_sd], elems[4].Obj()["x"_sd])},
+                               1);
+        appendEOO(expected);
+        appendInterleavedStartFunc(expected, elems[6].Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
 
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+        appendEOO(expected);
 
-    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]),
-                            deltaInt32(elems[2].Obj()["x"_sd], elems[1].Obj()["x"_sd]),
-                            deltaInt32(elems[3].Obj()["x"_sd], elems[2].Obj()["x"_sd]),
-                            deltaInt32(elems[4].Obj()["x"_sd], elems[3].Obj()["x"_sd]),
-                            deltaInt32(elems[5].Obj()["x"_sd], elems[4].Obj()["x"_sd])},
-                           1);
-    appendEOO(expected);
-    appendInterleavedStart(expected, elems[6].Obj());
-    appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
-    appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
-    appendEOO(expected);
-
-    appendEOO(expected);
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
 }
 
 TEST_F(BSONColumnTest, InterleavedSkipAfterEmptySubObj) {
-    BSONColumnBuilder cb("test"_sd);
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {
+            createElementObj(
+                BSON("x" << 1 << "y" << 2 << "z" << BSON("z1" << BSONObjBuilder().obj()))),
+            BSONElement()};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
+        BufBuilder expected;
+        appendInterleavedStartFunc(expected, elems.front().Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
+        appendEOO(expected);
+
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlock64(expected, boost::none);
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
+
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
+
+TEST_F(BSONColumnTest, InterleavedSkipAfterEmptySubArray) {
     std::vector<BSONElement> elems = {
-        createElementObj(BSON("x" << 1 << "y" << 2 << "z" << BSON("z1" << BSONObjBuilder().obj()))),
+        createElementObj(
+            BSON("x" << 1 << "y" << 2 << "z" << BSON("z1" << BSONArrayBuilder().arr()))),
         BSONElement()};
 
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
-
     BufBuilder expected;
     appendInterleavedStart(expected, elems.front().Obj());
     appendSimple8bControl(expected, 0b1000, 0b0000);
@@ -3603,9 +4572,7 @@ TEST_F(BSONColumnTest, InterleavedSkipAfterEmptySubObj) {
     appendSimple8bBlock64(expected, boost::none);
     appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    verifyDecompression(expected, elems);
 }
 
 TEST_F(BSONColumnTest, ObjectEmpty) {
@@ -3633,29 +4600,51 @@ TEST_F(BSONColumnTest, ObjectEmpty) {
 }
 
 TEST_F(BSONColumnTest, ObjectEmptyAfterNonEmpty) {
-    BSONColumnBuilder cb("test"_sd);
+    auto test = [&](bool arrayCompression, auto appendInterleavedStartFunc) {
+        BSONColumnBuilder cb("test"_sd);
+
+        std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1)),
+                                          createElementObj(BSONObjBuilder().obj())};
+
+        if (!arrayCompression) {
+            for (auto elem : elems) {
+                if (!elem.eoo())
+                    cb.append(elem);
+                else
+                    cb.skip();
+            }
+        }
 
-    std::vector<BSONElement> elems = {createElementObj(BSON("x" << 1)),
-                                      createElementObj(BSONObjBuilder().obj())};
+        BufBuilder expected;
+        appendInterleavedStartFunc(expected, elems.front().Obj());
+        appendSimple8bControl(expected, 0b1000, 0b0000);
+        appendSimple8bBlock64(expected, kDeltaForBinaryEqualValues);
+        appendEOO(expected);
+        appendLiteral(expected, elems[1]);
+        appendEOO(expected);
+
+        if (!arrayCompression) {
+            auto binData = cb.finalize();
+            verifyBinary(binData, expected);
+        }
+        verifyDecompression(expected, elems);
+    };
 
-    for (auto elem : elems) {
-        if (!elem.eoo())
-            cb.append(elem);
-        else
-            cb.skip();
-    }
+    test(false, appendInterleavedStartLegacy);
+    test(true, appendInterleavedStart);
+}
+
+TEST_F(BSONColumnTest, ObjectWithOnlyEmptyObjsDoesNotStartInterleaving) {
+    std::vector<BSONElement> elems;
+    elems.push_back(createElementObj(BSON("a" << BSONObjBuilder().obj())));
+    elems.push_back(createElementObj(BSON("b" << BSONObjBuilder().obj())));
 
     BufBuilder expected;
-    appendInterleavedStart(expected, elems.front().Obj());
-    appendSimple8bControl(expected, 0b1000, 0b0000);
-    appendSimple8bBlock64(expected, kDeltaForBinaryEqualValues);
-    appendEOO(expected);
+    appendLiteral(expected, elems[0]);
     appendLiteral(expected, elems[1]);
     appendEOO(expected);
 
-    auto binData = cb.finalize();
-    verifyBinary(binData, expected);
-    verifyDecompression(binData, elems);
+    verifyDecompression(expected, elems);
 }
 
 TEST_F(BSONColumnTest, NonZeroRLEInFirstBlockAfterSimple8bBlocks) {
@@ -3842,8 +4831,9 @@ TEST_F(BSONColumnTest, NoLiteralStart) {
 TEST_F(BSONColumnTest, InvalidInterleavedCount) {
     // This test sets up an interleaved reference object with two fields but only provides one
     // interleaved substream.
+    BSONColumnBuilder cb("test"_sd);
     BufBuilder expected;
-    appendInterleavedStart(expected, BSON("a" << 1 << "b" << 1));
+    appendInterleavedStartLegacy(expected, BSON("a" << 1 << "b" << 1));
     appendSimple8bControl(expected, 0b1000, 0b0000);
     appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
     appendEOO(expected);
@@ -3855,12 +4845,12 @@ TEST_F(BSONColumnTest, InvalidInterleavedCount) {
 
 TEST_F(BSONColumnTest, InvalidInterleavedWhenAlreadyInterleaved) {
     // This tests that we handle the interleaved start byte when already in interleaved mode.
-
+    BSONColumnBuilder cb("test"_sd);
     BufBuilder expected;
-    appendInterleavedStart(expected, BSON("a" << 1 << "b" << 1));
+    appendInterleavedStartLegacy(expected, BSON("a" << 1 << "b" << 1));
     appendSimple8bControl(expected, 0b1000, 0b0000);
     appendSimple8bBlocks64(expected, {kDeltaForBinaryEqualValues}, 1);
-    appendInterleavedStart(expected, BSON("a" << 1 << "b" << 1));
+    appendInterleavedStartLegacy(expected, BSON("a" << 1 << "b" << 1));
     appendEOO(expected);
     appendEOO(expected);
 
@@ -3986,6 +4976,7 @@ TEST_F(BSONColumnTest, FTDCRoundTrip) {
         return std::string((const char*)binData.data, binData.length);
     };
 
+    // Test that we can decompress and re-compress without any data loss.
     ASSERT_EQ(roundtrip(compressed), compressed);
 }
 #endif
diff --git a/src/mongo/bson/util/bsoncolumn_util.h b/src/mongo/bson/util/bsoncolumn_util.h
index e7fe0d6aae4..9d9eec07e8e 100644
--- a/src/mongo/bson/util/bsoncolumn_util.h
+++ b/src/mongo/bson/util/bsoncolumn_util.h
@@ -33,7 +33,9 @@
 #include "mongo/platform/int128.h"
 
 namespace mongo::bsoncolumn {
-static constexpr char kInterleavedStartControlByte = (char)0xF0;
+static constexpr char kInterleavedStartControlByteLegacy = (char)0xF0;
+static constexpr char kInterleavedStartControlByte = (char)0xF1;
+static constexpr char kInterleavedStartArrayRootControlByte = (char)0xF2;
 
 inline bool isLiteralControlByte(char control) {
     return (control & 0xE0) == 0;
diff --git a/src/mongo/bson/util/bsoncolumnbuilder.cpp b/src/mongo/bson/util/bsoncolumnbuilder.cpp
index 9156b56a5f4..e9beab9162c 100644
--- a/src/mongo/bson/util/bsoncolumnbuilder.cpp
+++ b/src/mongo/bson/util/bsoncolumnbuilder.cpp
@@ -938,7 +938,7 @@ void BSONColumnBuilder::_startDetermineSubObjReference(const BSONObj& obj) {
 
 void BSONColumnBuilder::_finishDetermineSubObjReference() {
     // Done determining reference sub-object. Write this control byte and object to stream.
-    _bufBuilder.appendChar(bsoncolumn::kInterleavedStartControlByte);
+    _bufBuilder.appendChar(bsoncolumn::kInterleavedStartControlByteLegacy);
     _bufBuilder.appendBuf(_referenceSubObj.objdata(), _referenceSubObj.objsize());
     ++_numInterleavedStartWritten;