[ntuple] add support for large objects in streamer field

Author: jblomer

tree/ntuple/inc/ROOT/RField.hxx

@@ -287,7 +287,7 @@ public:
    size_t GetValueSize() const final;
    size_t GetAlignment() const final;
    // As of field version 1, the byte stream contains the byte count stack for large objects (see binary specs)
-   std::uint32_t GetFieldVersion() const final { return 0; }
+   std::uint32_t GetFieldVersion() const final { return 1; }
    std::uint32_t GetTypeVersion() const final;
    std::uint32_t GetTypeChecksum() const final;
    TClass *GetClass() const { return fClass; }

tree/ntuple/src/RFieldMeta.cxx

@@ -19,6 +19,7 @@
 #include <ROOT/RFieldBase.hxx>
 #include <ROOT/RFieldUtils.hxx>
 #include <ROOT/RFieldVisitor.hxx>
+#include <ROOT/RNTupleSerialize.hxx>
 #include <ROOT/RNTupleUtils.hxx>
 #include <ROOT/RSpan.hxx>
 
@@ -1316,18 +1317,29 @@ std::size_t ROOT::RStreamerField::AppendImpl(const void *from)
                              [this](TVirtualStreamerInfo *info) { fStreamerInfos[info->GetNumber()] = info; });
    fClass->Streamer(const_cast<void *>(from), buffer);
 
-   auto nbytes = buffer.Length();
+   const auto nbytes = buffer.Length();
+   std::size_t szBufCounts = 0;
    R__ASSERT(nbytes >= 0);
    if (static_cast<std::size_t>(nbytes) > kMaxSmallBuffer) {
-      throw RException(R__FAIL("large objects (>1GiB) not supported by the version 0 streamer field"));
+      const std::uint64_t nCounts = buffer.GetByteCounts().size();
+      szBufCounts = sizeof(std::uint64_t) * (2 * nCounts + 1);
+      auto bufCounts = Internal::MakeUninitArray<unsigned char>(szBufCounts);
+      std::size_t pos = Internal::RNTupleSerializer::SerializeUInt64(nCounts, bufCounts.get());
+      for (const auto &[bcountLoc, bcountVal] : buffer.GetByteCounts()) {
+         pos += Internal::RNTupleSerializer::SerializeUInt64(bcountLoc, bufCounts.get() + pos);
+         pos += Internal::RNTupleSerializer::SerializeUInt64(bcountVal, bufCounts.get() + pos);
+      }
+      assert(pos == szBufCounts);
+      fAuxiliaryColumn->AppendV(bufCounts.get(), szBufCounts);
+      fIndex += szBufCounts;
    } else {
       assert(buffer.GetByteCounts().empty());
    }
 
    fAuxiliaryColumn->AppendV(buffer.Buffer(), buffer.Length());
    fIndex += nbytes;
    fPrincipalColumn->Append(&fIndex);
-   return nbytes + fPrincipalColumn->GetElement()->GetPackedSize();
+   return szBufCounts + nbytes + fPrincipalColumn->GetElement()->GetPackedSize();
 }
 
 void ROOT::RStreamerField::ReadGlobalImpl(ROOT::NTupleSize_t globalIndex, void *to)
@@ -1336,10 +1348,47 @@ void ROOT::RStreamerField::ReadGlobalImpl(ROOT::NTupleSize_t globalIndex, void *
    ROOT::NTupleSize_t nbytes;
    fPrincipalColumn->GetCollectionInfo(globalIndex, &collectionStart, &nbytes);
 
-   if (nbytes > kMaxSmallBuffer)
-      throw RException(R__FAIL("large objects (>1GiB) not supported by the version 0 streamer field"));
+   TBufferFile::ByteCountFinder_t byteCounts;
+   if (nbytes > kMaxSmallBuffer) {
+      unsigned char bufNCounts[sizeof(std::uint64_t)];
+      fAuxiliaryColumn->ReadV(collectionStart, sizeof(std::uint64_t), bufNCounts);
+      nbytes -= sizeof(std::uint64_t);
+      collectionStart = collectionStart + sizeof(std::uint64_t);
+
+      std::uint64_t nCounts;
+      Internal::RNTupleSerializer::DeserializeUInt64(bufNCounts, nCounts);
+      if (nCounts > (std::numeric_limits<std::size_t>::max() / sizeof(std::uint64_t)) / 2 - 1)
+         throw RException(R__FAIL("invalid byte count size in streamer field: " + std::to_string(nCounts)));
+      const std::size_t szBufCounts = sizeof(std::uint64_t) * (2 * nCounts);
+      if (szBufCounts > nbytes) {
+         throw RException(R__FAIL("invalid byte count size in streamer field: nCounts=" + std::to_string(nCounts) +
+                                  " szBufCounts=" + std::to_string(szBufCounts) + " nbytes=" + std::to_string(nbytes)));
+      }
+
+      auto bufCounts = Internal::MakeUninitArray<unsigned char>(szBufCounts);
+      fAuxiliaryColumn->ReadV(collectionStart, szBufCounts, bufCounts.get());
+      nbytes -= szBufCounts;
+      collectionStart = collectionStart + szBufCounts;
+
+      byteCounts.reserve(nCounts);
+      std::size_t pos = 0;
+      for (std::uint64_t i = 0; i < nCounts; ++i) {
+         std::uint64_t bcountLoc, bcountVal;
+         pos += Internal::RNTupleSerializer::DeserializeUInt64(bufCounts.get() + pos, bcountLoc);
+         pos += Internal::RNTupleSerializer::DeserializeUInt64(bufCounts.get() + pos, bcountVal);
+         if ((bcountLoc > nbytes) || (bcountVal > nbytes) || (nbytes - bcountVal < bcountLoc)) {
+            throw RException(
+               R__FAIL("invalid byte count record: " + std::to_string(bcountLoc) + ", " + std::to_string(bcountVal)));
+         }
+         byteCounts.emplace(bcountLoc, bcountVal);
+      }
+      assert(pos == szBufCounts);
+      if (byteCounts.size() != nCounts)
+         throw RException(R__FAIL("duplicate byte counts"));
+   }
 
    TBufferFile buffer(TBuffer::kRead, nbytes);
+   buffer.SetByteCounts(std::move(byteCounts));
    fAuxiliaryColumn->ReadV(collectionStart, nbytes, buffer.Buffer());
    fClass->Streamer(to, buffer);
 }

tree/ntuple/test/rfield_streamer.cxx

@@ -462,7 +462,7 @@ TEST(RField, StreamerFieldVersion)
       if (version < 2) {
          const auto &f = reader->GetModel().GetConstField("f");
          EXPECT_TRUE(dynamic_cast<const ROOT::RStreamerField *>(&f));
-         EXPECT_EQ(0u, f.GetFieldVersion());
+         EXPECT_EQ(1u, f.GetFieldVersion());
          EXPECT_EQ(version, f.GetOnDiskFieldVersion());
          EXPECT_EQ(2u, f.GetTypeVersion());
          EXPECT_EQ(137u, f.GetOnDiskTypeVersion());