From 7486b7a7e665127f0edb0199ff82f6207adb5e65 Mon Sep 17 00:00:00 2001 From: JasMehta08 Date: Sun, 2 Aug 2026 01:45:53 +0530 Subject: [PATCH] [ntuple] Add RPageSourceS3 for reading ntuples from S3 --- tree/ntuple/inc/ROOT/RPageStorageS3.hxx | 73 +++ tree/ntuple/src/RPageStorage.cxx | 10 +- tree/ntuple/src/RPageStorageS3.cxx | 271 +++++++++ tree/ntuple/test/ntuple_storage_s3.cxx | 731 +++++++++++++++++++++++- 4 files changed, 1082 insertions(+), 3 deletions(-) diff --git a/tree/ntuple/inc/ROOT/RPageStorageS3.hxx b/tree/ntuple/inc/ROOT/RPageStorageS3.hxx index d8370ab713c67..80dea8912996d 100644 --- a/tree/ntuple/inc/ROOT/RPageStorageS3.hxx +++ b/tree/ntuple/inc/ROOT/RPageStorageS3.hxx @@ -17,11 +17,14 @@ #include #include #include +#include +#include #include #include #include #include +#include namespace ROOT { namespace Experimental { @@ -114,6 +117,10 @@ RResult ParseS3Url(std::string_view uri); Currently implements Mode B (one sealed page per S3 object, kTypeObject64 locators). Mode A (multiple packed pages per object, kTypeMulti locators) will be added separately. +Prefer calling RNTupleWriter::CommitDataset() explicitly to letting the writer's destructor do it: a +destructor cannot propagate an exception, so a failed footer or anchor upload is only logged and leaves +the ntuple without an anchor, i.e. unreadable. + \warning The S3 backend is experimental and under active development. */ // clang-format on @@ -161,6 +168,72 @@ public: CloneAsHidden(std::string_view name, const ROOT::RNTupleWriteOptions &opts) const final; }; // class RPageSinkS3 +// clang-format off +/** +\class ROOT::Experimental::Internal::RPageSourceS3 +\ingroup NTuple +\brief Storage provider that reads ntuple pages from S3-compatible object storage. + +Counterpart of RPageSinkS3: implements Mode B reads (one sealed page per S3 object, kTypeObject64 +locators). Pages are fetched one object at a time; batching them into concurrent GETs is left to a +follow-up. + +\warning The S3 backend is experimental and under active development. +*/ +// clang-format on +class RPageSourceS3 : public ROOT::Internal::RPageSource { +private: + /// HTTP base URL for this ntuple (derived from the s3 scheme URI); never has a trailing slash + std::string fBaseUrl; + /// Connection used by everything that runs on the calling thread: the anchor, header and footer in + /// LoadStructureImpl, the page lists in LoadPageListImpl and single pages in LoadSealedPageImpl. + /// Reused across objects so curl keeps it alive instead of re-handshaking per object. + ROOT::Internal::RCurlConnection fConnection; + /// Connection used exclusively by LoadClusters(), which the cluster pool calls on its own I/O thread. + /// A libcurl easy handle carries per-request state and must not be driven by two threads at once, so + /// the prefetch path needs a handle of its own rather than sharing fConnection. + ROOT::Internal::RCurlConnection fClusterConnection; + /// Anchor metadata, fetched and parsed in LoadStructureImpl + RNTupleAnchorS3 fAnchor; + /// Populated by LoadStructureImpl and AttachImpl, moved out at the end of AttachImpl + ROOT::Internal::RNTupleDescriptorBuilder fDescriptorBuilder; + + /// Resolve a numeric object ID to its full HTTP URL through the anchor's URL template + std::string MakeObjectUrl(std::uint64_t objId) const; + /// Download `size` bytes from `url` into the caller-provided `buffer` via an HTTP GET request. + /// The connection is explicit because the caller's thread determines which one may be used. + void GetObject(ROOT::Internal::RCurlConnection &connection, const std::string &url, unsigned char *buffer, + std::size_t size); + + /// Tag to select the internal constructor that takes an already-resolved base URL. + struct RFromBaseUrl {}; + /// Internal constructor used by CloneImpl: the public constructor derives the base URL by parsing an + /// s3 scheme URI, whereas a clone of an open source already has one and must not re-parse it. + RPageSourceS3(std::string_view ntupleName, std::string_view baseUrl, const ROOT::RNTupleReadOptions &options, + RFromBaseUrl); + + void LoadPageListImpl(const RNTupleLocator &locator, unsigned char *buffer) final; + void LoadSealedPageImpl(const RNTupleLocator &locator, RSealedPage &sealedPage) final; + +protected: + void LoadStructureImpl() final; + ROOT::RNTupleDescriptor AttachImpl() final; + /// The cloned page source opens its own pair of HTTP connections to the same base URL. + std::unique_ptr CloneImpl() const final; + +public: + RPageSourceS3(std::string_view ntupleName, std::string_view uri, const ROOT::RNTupleReadOptions &options); + ~RPageSourceS3() override; + + std::vector> + LoadClusters(std::span clusterKeys) final; + + void LoadStreamerInfo() final; + + std::unique_ptr OpenWithDifferentAnchor(const ROOT::Internal::RNTupleLink &anchorLink, + const ROOT::RNTupleReadOptions &options = {}) final; +}; // class RPageSourceS3 + } // namespace Internal } // namespace Experimental } // namespace ROOT diff --git a/tree/ntuple/src/RPageStorage.cxx b/tree/ntuple/src/RPageStorage.cxx index 0d7be23860605..225ad05de6e6c 100644 --- a/tree/ntuple/src/RPageStorage.cxx +++ b/tree/ntuple/src/RPageStorage.cxx @@ -192,8 +192,14 @@ ROOT::Internal::RPageSource::Create(std::string_view ntupleName, std::string_vie throw RException(R__FAIL("This RNTuple build does not support DAOS.")); #endif - if (ROOT::StartsWith(location, "ntpl+s3+http://") || ROOT::StartsWith(location, "ntpl+s3+https://")) - throw RException(R__FAIL("S3 read support is not yet implemented.")); + if (ROOT::StartsWith(location, "ntpl+s3+http://") || ROOT::StartsWith(location, "ntpl+s3+https://")) { +#ifdef R__ENABLE_S3 + return std::make_unique(ntupleName, location, options); +#else + throw RException(R__FAIL("This RNTuple build does not support S3. Rebuild ROOT with the 'curl' " + "cmake option enabled (-Dcurl=ON) to enable the S3 backend.")); +#endif + } return std::make_unique(ntupleName, location, options); } diff --git a/tree/ntuple/src/RPageStorageS3.cxx b/tree/ntuple/src/RPageStorageS3.cxx index da4afb9546b27..d347abef7bca0 100644 --- a/tree/ntuple/src/RPageStorageS3.cxx +++ b/tree/ntuple/src/RPageStorageS3.cxx @@ -12,8 +12,11 @@ #include +#include #include #include +#include +#include #include #include #include @@ -23,14 +26,18 @@ #include #include +#include #include #include #include #include #include +#include using ROOT::Internal::MakeUninitArray; using ROOT::Internal::RNTupleCompressor; +using ROOT::Internal::RNTupleDecompressor; +using ROOT::Internal::RNTupleSerializer; /// Field-by-field equality check across all data members. bool ROOT::Experimental::Internal::RNTupleAnchorS3::operator==(const RNTupleAnchorS3 &other) const @@ -337,3 +344,267 @@ ROOT::Experimental::Internal::RPageSinkS3::CloneAsHidden(std::string_view name, return std::unique_ptr(new RPageSinkS3(name, cloneBaseUrl, opts, RFromBaseUrl{})); } + +// RPageSourceS3 + +namespace { +/// The anchor is a small JSON document, but its size comes from a server-supplied content-length, +/// so bound it before using it as an allocation size. +constexpr std::uint64_t kMaxAnchorSize = 1024 * 1024; + +/// Bounds the header and footer sizes taken from the anchor, so that summing them cannot overflow. +/// The checksum proves they arrived intact, not that they are sensible. +constexpr std::uint64_t kMaxEnvelopeSize = 1024 * 1024 * 1024; +} // anonymous namespace + +ROOT::Experimental::Internal::RPageSourceS3::RPageSourceS3(std::string_view ntupleName, std::string_view uri, + const ROOT::RNTupleReadOptions &options) + : RPageSourceS3(ntupleName, ParseS3Url(uri).Unwrap(), options, RFromBaseUrl{}) +{ +} + +ROOT::Experimental::Internal::RPageSourceS3::RPageSourceS3(std::string_view ntupleName, std::string_view baseUrl, + const ROOT::RNTupleReadOptions &options, RFromBaseUrl) + : RPageSource(ntupleName, options), fBaseUrl(baseUrl), fConnection(fBaseUrl), fClusterConnection(fBaseUrl) +{ + fConnection.SetCredentialsFromEnvironment(); + fClusterConnection.SetCredentialsFromEnvironment(); + // Enable the counters before any I/O happens, so that LoadStructureImpl() can already use them. + EnableDefaultMetrics("RPageSourceS3"); +} + +ROOT::Experimental::Internal::RPageSourceS3::~RPageSourceS3() +{ + // The cluster pool's I/O thread calls back into this source, so it has to be joined before any of + // the members below (in particular fConnection) are destroyed. + StopClusterPoolBackgroundThread(); +} + +std::string ROOT::Experimental::Internal::RPageSourceS3::MakeObjectUrl(std::uint64_t objId) const +{ + // Resolve the template the writer recorded rather than assuming the default one. Same substitution + // as RPageSinkS3::CloneAsHidden(). + std::string url = fAnchor.GetUrlTemplate(); + + auto pos = url.find("${baseurl}"); + if (pos != std::string::npos) + url.replace(pos, std::strlen("${baseurl}"), fBaseUrl); + + pos = url.find("${objid}"); + if (pos != std::string::npos) + url.replace(pos, std::strlen("${objid}"), std::to_string(objId)); + + return url; +} + +void ROOT::Experimental::Internal::RPageSourceS3::GetObject(ROOT::Internal::RCurlConnection &connection, + const std::string &url, unsigned char *buffer, + std::size_t size) +{ + // Retarget the connection rather than opening a new one, so curl keeps it alive across objects. The + // caller picks which one: a curl easy handle serves one thread at a time. + connection.SetUrl(url).ThrowOnError(); + + // One range from offset 0 covers the whole object. RCurlConnection also copes with a server that + // ignores the range and replies 200 with the full body. + ROOT::Internal::RCurlConnection::RUserRange range; + range.fDestination = buffer; + range.fOffset = 0; + range.fLength = size; + + auto status = connection.SendRangesReq(1, &range); + if (!status) + throw ROOT::RException(R__FAIL("S3 GET failed for " + url + ": " + status.fStatusMsg)); + if (range.fNBytesRecv != size) { + throw ROOT::RException(R__FAIL("S3 GET for " + url + " returned " + std::to_string(range.fNBytesRecv) + + " bytes, expected " + std::to_string(size))); + } +} + +void ROOT::Experimental::Internal::RPageSourceS3::LoadStructureImpl() +{ + // The anchor lives at the base URL. Its size is not known upfront, so ask for it with a HEAD request + // before downloading it. + std::uint64_t anchorSize = 0; + { + fConnection.SetUrl(fBaseUrl).ThrowOnError(); + auto headStatus = fConnection.SendHeadReq(anchorSize); + if (!headStatus) + throw ROOT::RException(R__FAIL("S3 HEAD failed for anchor at " + fBaseUrl + ": " + headStatus.fStatusMsg)); + if (anchorSize == ROOT::Internal::RCurlConnection::kUnknownSize) + throw ROOT::RException(R__FAIL("S3 HEAD returned no content-length for anchor at " + fBaseUrl)); + // anchorSize comes from the server, so bound it before allocating. + if (anchorSize > kMaxAnchorSize) { + throw ROOT::RException(R__FAIL("S3 anchor at " + fBaseUrl + " is implausibly large (" + + std::to_string(anchorSize) + " bytes, limit is " + + std::to_string(kMaxAnchorSize) + "); refusing to read it")); + } + } + + auto anchorBuffer = MakeUninitArray(anchorSize); + GetObject(fConnection, fBaseUrl, anchorBuffer.get(), static_cast(anchorSize)); + + const std::string anchorJson(reinterpret_cast(anchorBuffer.get()), + static_cast(anchorSize)); + auto anchorResult = RNTupleAnchorS3::CreateFromJSON(anchorJson); + if (!anchorResult) + throw ROOT::RException(R__FORWARD_ERROR(anchorResult)); + fAnchor = anchorResult.Inspect(); + + // The envelope sizes are used to size a buffer and to offset into it; bound them so that summing them + // below cannot overflow. + if (fAnchor.GetNBytesHeader() > kMaxEnvelopeSize || fAnchor.GetNBytesFooter() > kMaxEnvelopeSize || + fAnchor.GetLenHeader() > kMaxEnvelopeSize || fAnchor.GetLenFooter() > kMaxEnvelopeSize) { + throw ROOT::RException(R__FAIL("S3 anchor at " + fBaseUrl + + " declares an implausible header or footer size " + "(limit is " + + std::to_string(kMaxEnvelopeSize) + " bytes); refusing to read it")); + } + + fDescriptorBuilder.SetVersion(fAnchor.GetVersionEpoch(), fAnchor.GetVersionMajor(), fAnchor.GetVersionMinor(), + fAnchor.GetVersionPatch()); + fDescriptorBuilder.SetOnDiskHeaderSize(fAnchor.GetNBytesHeader()); + fDescriptorBuilder.AddToOnDiskFooterSize(fAnchor.GetNBytesFooter()); + + // Reserve enough space for the compressed and the uncompressed header/footer (see AttachImpl) + const auto bufSize = + fAnchor.GetNBytesHeader() + fAnchor.GetNBytesFooter() + std::max(fAnchor.GetLenHeader(), fAnchor.GetLenFooter()); + fStructureBuffer.fBuffer = MakeUninitArray(bufSize); + fStructureBuffer.fPtrHeader = fStructureBuffer.fBuffer.get(); + fStructureBuffer.fPtrFooter = fStructureBuffer.fBuffer.get() + fAnchor.GetNBytesHeader(); + + { + Detail::RNTupleAtomicTimer timer(fCounters->fTimeWallRead, fCounters->fTimeCpuRead); + GetObject(fConnection, MakeObjectUrl(fAnchor.GetHeaderObjId()), + reinterpret_cast(fStructureBuffer.fPtrHeader), fAnchor.GetNBytesHeader()); + GetObject(fConnection, MakeObjectUrl(fAnchor.GetFooterObjId()), + reinterpret_cast(fStructureBuffer.fPtrFooter), fAnchor.GetNBytesFooter()); + // fNRead counts byte ranges read, so the three GETs count and the preceding HEAD does not. + fCounters->fNRead.Add(3); + } +} + +ROOT::RNTupleDescriptor ROOT::Experimental::Internal::RPageSourceS3::AttachImpl() +{ + auto unzipBuf = reinterpret_cast(fStructureBuffer.fPtrFooter) + fAnchor.GetNBytesFooter(); + + RNTupleDecompressor::Unzip(fStructureBuffer.fPtrHeader, fAnchor.GetNBytesHeader(), fAnchor.GetLenHeader(), unzipBuf); + RNTupleSerializer::DeserializeHeader(unzipBuf, fAnchor.GetLenHeader(), fDescriptorBuilder); + + // The name locates nothing here, but comparing it asserts that the URL points at the data set the + // caller meant. An empty name opts out. + const auto &storedName = fDescriptorBuilder.GetDescriptor().GetName(); + if (!fNTupleName.empty() && storedName != fNTupleName) { + throw ROOT::RException( + R__FAIL("the S3 ntuple at " + fBaseUrl + " is named '" + storedName + "', not '" + fNTupleName + "'")); + } + + RNTupleDecompressor::Unzip(fStructureBuffer.fPtrFooter, fAnchor.GetNBytesFooter(), fAnchor.GetLenFooter(), unzipBuf); + RNTupleSerializer::DeserializeFooter(unzipBuf, fAnchor.GetLenFooter(), fDescriptorBuilder); + + return fDescriptorBuilder.MoveDescriptor(); +} + +void ROOT::Experimental::Internal::RPageSourceS3::LoadPageListImpl(const RNTupleLocator &locator, unsigned char *buffer) +{ + const auto objId = locator.GetPosition().GetLocation(); + GetObject(fConnection, MakeObjectUrl(objId), buffer, locator.GetNBytesOnStorage()); +} + +void ROOT::Experimental::Internal::RPageSourceS3::LoadSealedPageImpl(const RNTupleLocator &locator, + RSealedPage &sealedPage) +{ + Detail::RNTupleAtomicTimer timer(fCounters->fTimeWallRead, fCounters->fTimeCpuRead); + const auto objId = locator.GetPosition().GetLocation(); + // The const_cast is safe: the buffer belongs to the caller, who provided it for us to fill. + GetObject(fConnection, MakeObjectUrl(objId), + const_cast(reinterpret_cast(sealedPage.GetBuffer())), + sealedPage.GetBufferSize()); +} + +std::vector> +ROOT::Experimental::Internal::RPageSourceS3::LoadClusters(std::span clusterKeys) +{ + /// Where one page of a cluster lives in S3 and how much space it takes in the cluster buffer. + struct RS3SealedPageLocator { + ROOT::DescriptorId_t fColumnId = 0; + ROOT::NTupleSize_t fPageNo = 0; + std::uint64_t fObjId = 0; + std::uint64_t fBufferSize = 0; ///< page payload + checksum (if available) + }; + + fCounters->fNClusterLoaded.Add(clusterKeys.size()); + + std::vector> clusters; + + for (const auto &clusterKey : clusterKeys) { + std::vector onDiskPages; + std::uint64_t clusterBufSize = 0; + + auto pageZeroMap = std::make_unique(); + PrepareLoadCluster( + clusterKey, *pageZeroMap, + [&](ROOT::DescriptorId_t physicalColumnId, ROOT::NTupleSize_t pageNo, + const ROOT::RClusterDescriptor::RPageInfo &pageInfo) { + const auto &pageLocator = pageInfo.GetLocator(); + const auto objId = pageLocator.GetPosition().GetLocation(); + const auto pageBufferSize = pageLocator.GetNBytesOnStorage() + pageInfo.HasChecksum() * kNBytesPageChecksum; + onDiskPages.emplace_back(RS3SealedPageLocator{physicalColumnId, pageNo, objId, pageBufferSize}); + clusterBufSize += pageBufferSize; + }); + + auto clusterBuffer = new unsigned char[clusterBufSize]; + auto pageMap = + std::make_unique(std::unique_ptr(clusterBuffer)); + + { + // One GET per page in Mode B; batching them concurrently is a follow-up. This runs on the + // cluster pool's I/O thread, hence fClusterConnection: the caller may be using fConnection. + Detail::RNTupleAtomicTimer timer(fCounters->fTimeWallRead, fCounters->fTimeCpuRead); + auto pageBuffer = clusterBuffer; + for (const auto &sealedLoc : onDiskPages) { + ROOT::Internal::ROnDiskPage::Key key(sealedLoc.fColumnId, sealedLoc.fPageNo); + pageMap->Register(key, ROOT::Internal::ROnDiskPage(pageBuffer, sealedLoc.fBufferSize)); + GetObject(fClusterConnection, MakeObjectUrl(sealedLoc.fObjId), pageBuffer, sealedLoc.fBufferSize); + pageBuffer += sealedLoc.fBufferSize; + } + } + + fCounters->fNPageRead.Add(onDiskPages.size()); + fCounters->fSzReadPayload.Add(clusterBufSize); + fCounters->fNRead.Add(onDiskPages.size()); + + auto cluster = std::make_unique(clusterKey.fClusterId); + cluster->Adopt(std::move(pageMap)); + cluster->Adopt(std::move(pageZeroMap)); + for (auto colId : clusterKey.fPhysicalColumnSet) + cluster->SetColumnAvailable(colId); + clusters.emplace_back(std::move(cluster)); + } + + return clusters; +} + +std::unique_ptr ROOT::Experimental::Internal::RPageSourceS3::CloneImpl() const +{ + // The clone opens its own connections, so clones can be read from concurrently. + auto clone = std::unique_ptr(new RPageSourceS3(fNTupleName, fBaseUrl, fOptions, RFromBaseUrl{})); + // Carry the anchor over: an attached clone skips LoadStructureImpl(), so without it MakeObjectUrl() + // would resolve against a default-constructed template. + clone->fAnchor = fAnchor; + return clone; +} + +void ROOT::Experimental::Internal::RPageSourceS3::LoadStreamerInfo() +{ + R__LOG_WARNING(ROOT::Internal::NTupleLog()) << "S3-backed sources have no associated StreamerInfo to load."; +} + +std::unique_ptr +ROOT::Experimental::Internal::RPageSourceS3::OpenWithDifferentAnchor(const ROOT::Internal::RNTupleLink &, + const ROOT::RNTupleReadOptions &) +{ + // An S3 ntuple is self-locating (its anchor is at the base URL), so there is no anchor link to follow + // within the same storage container the way the file backend does. + throw ROOT::RException(R__FAIL("OpenWithDifferentAnchor is not implemented for the S3 backend")); +} diff --git a/tree/ntuple/test/ntuple_storage_s3.cxx b/tree/ntuple/test/ntuple_storage_s3.cxx index 74baff67e6f3b..d924ffabbc12f 100644 --- a/tree/ntuple/test/ntuple_storage_s3.cxx +++ b/tree/ntuple/test/ntuple_storage_s3.cxx @@ -1,7 +1,7 @@ /// \file ntuple_storage_s3.cxx /// \author Jas Mehta /// \date 2026-06-01 -/// \brief Unit tests for the S3 storage backend components (anchor serialization). +/// \brief Unit tests for the S3 storage backend components (anchor serialization, write and read path). #include "ntuple_test.hxx" #include @@ -15,9 +15,12 @@ #include #include +#include #include #include #include +#include +#include #include #include @@ -821,3 +824,729 @@ TEST(RPageSinkS3Wire, CloneAsHiddenWritesUnderClonePrefix) // The clone's anchor is written last, at exactly the clone's base URL. EXPECT_EQ(clonePrefix, paths.back()); } + +// ==================== RPageSourceS3 Wire-Level Tests (mock HTTP server) ==================== + +// The read path needs a mock that answers HEAD and GET, not just PUT, so these tests use a small +// in-memory object store: the sink's PUTs populate it and the source's HEAD/GETs read it back. That +// makes a complete write-then-read round trip possible with no S3 service at all, so it runs in CI. +namespace { + +/// Serve one HTTP request from an accepted socket against the in-memory `store`: PUT stores the body +/// under the request path, HEAD answers with the stored object's size, and GET returns the stored body. +/// Unknown paths get a 404, other methods a 405. +void ServeS3Request(TSocket *sock, std::map &store) +{ + // Read up to and including the end-of-headers marker, byte by byte. + std::string headers; + const char *eof = "\r\n\r\n"; + const std::size_t eofLen = std::strlen(eof); + std::size_t nextInEof = 0; + char c; + while (sock->RecvRaw(&c, 1) > 0) { + headers.push_back(c); + if (c == eof[nextInEof]) { + if (++nextInEof == eofLen) + break; + } else { + nextInEof = 0; + } + } + + // The request line is "METHOD /target HTTP/1.1". + std::string method, path; + if (auto sp1 = headers.find(' '); sp1 != std::string::npos) { + method = headers.substr(0, sp1); + if (auto sp2 = headers.find(' ', sp1 + 1); sp2 != std::string::npos) + path = headers.substr(sp1 + 1, sp2 - sp1 - 1); + } + + std::string lower(headers); + std::transform(lower.begin(), lower.end(), lower.begin(), [](unsigned char ch) { return std::tolower(ch); }); + + // libcurl uploads with "Expect: 100-continue"; acknowledge before reading the body. + if (lower.find("expect: 100-continue") != std::string::npos) { + const char *cont = "HTTP/1.1 100 Continue\r\n\r\n"; + sock->SendRaw(cont, std::strlen(cont)); + } + + std::string body; + if (auto pos = lower.find("content-length: "); pos != std::string::npos) { + auto valStart = pos + std::strlen("content-length: "); + auto valEnd = lower.find("\r\n", valStart); + const auto contentLength = std::stoul(lower.substr(valStart, valEnd - valStart)); + if (contentLength > 0) { + body.resize(contentLength); + sock->RecvRaw(&body[0], contentLength); + } + } + + // Every reply announces "Connection: close" because this mock closes the socket after each request; + // curl then opens a fresh connection per request instead of reusing one we already closed. The + // source's GETs carry a Range header, which this mock ignores: it always returns the full object, + // which RCurlConnection handles as the "server ignored the range" case. + std::string response; + if (method == "PUT") { + store[path] = body; + response = "HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"; + } else if (method == "HEAD" || method == "GET") { + auto it = store.find(path); + if (it == store.end()) { + response = "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"; + } else { + response = + "HTTP/1.1 200 OK\r\nContent-Length: " + std::to_string(it->second.size()) + "\r\nConnection: close\r\n\r\n"; + if (method == "GET") + response += it->second; + } + } else { + response = "HTTP/1.1 405 Method Not Allowed\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"; + } + + sock->SendRaw(response.data(), response.size()); +} + +/// Set the credentials the S3 backend reads from the environment. The values are dummies: requests only +/// ever reach the loopback mock server in these tests, but curl still signs them (SigV4). +void SetDummyS3Env() +{ + gSystem->Setenv("S3_ACCESS_KEY", "dummykey"); + gSystem->Setenv("S3_SECRET_KEY", "dummysecret"); + gSystem->Setenv("S3_REGION", "us-east-1"); +} + +void UnsetS3Env() +{ + gSystem->Unsetenv("S3_ACCESS_KEY"); + gSystem->Unsetenv("S3_SECRET_KEY"); + gSystem->Unsetenv("S3_REGION"); +} + +/// Serve requests until `done` is set. Accept() blocks, so a caller that sets `done` also has to open a +/// throw-away connection (see StopMockServer) to wake this loop up. +void RunMockServer(TServerSocket &server, std::map &store, const std::atomic &done) +{ + while (!done.load()) { + TSocket *sock = server.Accept(); + if (!sock || sock == reinterpret_cast(-1)) + break; + if (done.load()) { + sock->Close(); + break; + } + ServeS3Request(sock, store); + sock->Close(); + } +} + +void StopMockServer(std::atomic &done, std::thread &serverThread, const std::string &host, int port) +{ + done.store(true); + TSocket dummy(host.c_str(), port); + dummy.Close(); + serverThread.join(); +} + +} // anonymous namespace + +TEST(RPageSourceS3Wire, RoundTripViaMock) +{ + TServerSocket server(0, false, TServerSocket::kDefaultBacklog, -1, ESocketBindOption::kInaddrLoopback); + const std::string host = server.GetLocalInetAddress().GetHostAddress(); + const int port = server.GetLocalPort(); + const std::string basePath = "/wirebucket/roundtrip"; + const std::string uri = "ntpl+s3+http://" + host + ":" + std::to_string(port) + basePath; + + SetDummyS3Env(); + + // The object store is only touched by the server thread while that thread runs. + std::map store; + std::atomic done{false}; + std::thread serverThread([&] { RunMockServer(server, store, done); }); + + { + // The sink ctor emits a one-time (std::call_once) experimental warning; allow it (optional + // because it only fires on the first sink construction in the process). + ROOT::TestSupport::CheckDiagsRAII diags; + diags.optionalDiag(kWarning, "[ROOT.NTuple]", "experimental", /*matchFullMessage=*/false); + + // Write phase: header, pages, page list, footer and finally the anchor are PUT into the store. + auto model = ROOT::RNTupleModel::Create(); + auto fldX = model->MakeField("x"); + auto writer = ROOT::RNTupleWriter::Recreate(std::move(model), "wire", uri); + for (int i = 0; i < 20; ++i) { + *fldX = i; + writer->Fill(); + } + } // writer destroyed here -> footer + anchor PUTs + + // Read phase: HEAD + GET for the anchor, then GETs for header, footer, page list and pages. + { + auto reader = ROOT::RNTupleReader::Open("wire", uri); + EXPECT_EQ(20u, reader->GetNEntries()); + + auto viewX = reader->GetView("x"); + for (int i = 0; i < 20; ++i) + EXPECT_EQ(i, viewX(i)); + } + + StopMockServer(done, serverThread, host, port); + UnsetS3Env(); +} + +TEST(RPageSourceS3Wire, RoundTripWithoutClusterCache) +{ + TServerSocket server(0, false, TServerSocket::kDefaultBacklog, -1, ESocketBindOption::kInaddrLoopback); + const std::string host = server.GetLocalInetAddress().GetHostAddress(); + const int port = server.GetLocalPort(); + const std::string basePath = "/wirebucket/nocache"; + const std::string uri = "ntpl+s3+http://" + host + ":" + std::to_string(port) + basePath; + + SetDummyS3Env(); + + std::map store; + std::atomic done{false}; + std::thread serverThread([&] { RunMockServer(server, store, done); }); + + { + ROOT::TestSupport::CheckDiagsRAII diags; + diags.optionalDiag(kWarning, "[ROOT.NTuple]", "experimental", /*matchFullMessage=*/false); + + auto model = ROOT::RNTupleModel::Create(); + auto fldX = model->MakeField("x"); + auto writer = ROOT::RNTupleWriter::Recreate(std::move(model), "nocache", uri); + for (int i = 0; i < 20; ++i) { + *fldX = i; + writer->Fill(); + } + } + + // With the cluster cache off, pages are read one at a time through LoadSealedPageImpl instead of + // being prefetched in bulk by LoadClusters, so this covers the direct single-page read path. + ROOT::RNTupleReadOptions options; + options.SetClusterCache(ROOT::RNTupleReadOptions::EClusterCache::kOff); + { + auto reader = ROOT::RNTupleReader::Open("nocache", uri, options); + EXPECT_EQ(20u, reader->GetNEntries()); + + auto viewX = reader->GetView("x"); + for (int i = 0; i < 20; ++i) + EXPECT_EQ(i, viewX(i)); + } + + StopMockServer(done, serverThread, host, port); + UnsetS3Env(); +} + +TEST(RPageSourceS3Wire, RoundTripManyPagesPerCluster) +{ + TServerSocket server(0, false, TServerSocket::kDefaultBacklog, -1, ESocketBindOption::kInaddrLoopback); + const std::string host = server.GetLocalInetAddress().GetHostAddress(); + const int port = server.GetLocalPort(); + const std::string uri = "ntpl+s3+http://" + host + ":" + std::to_string(port) + "/wirebucket/manypages"; + + SetDummyS3Env(); + + std::map store; + std::atomic done{false}; + std::thread serverThread([&] { RunMockServer(server, store, done); }); + + // Cap pages at 256 bytes (64 ints) but leave the cluster size at its default, so one cluster holds + // many pages from two interleaved columns. LoadClusters() packs them all into a single cluster buffer + // at successive offsets, which is where an off-by-one in the buffer cursor would show up. + ROOT::RNTupleWriteOptions writeOptions; + writeOptions.SetMaxUnzippedPageSize(256); + { + ROOT::TestSupport::CheckDiagsRAII diags; + diags.optionalDiag(kWarning, "[ROOT.NTuple]", "experimental", /*matchFullMessage=*/false); + + auto model = ROOT::RNTupleModel::Create(); + auto fldX = model->MakeField("x"); + auto fldY = model->MakeField("y"); + auto writer = ROOT::RNTupleWriter::Recreate(std::move(model), "manypages", uri, writeOptions); + for (int i = 0; i < 1000; ++i) { + *fldX = i; + *fldY = -i; + writer->Fill(); + } + } + + { + auto reader = ROOT::RNTupleReader::Open("manypages", uri); + EXPECT_EQ(1000u, reader->GetNEntries()); + + // Guard the premise of the test: one cluster, many pages per column. + const auto &desc = reader->GetDescriptor(); + ASSERT_EQ(1u, desc.GetNClusters()); + const auto columnId = desc.FindPhysicalColumnId(desc.FindFieldId("x"), 0, 0); + EXPECT_GT(desc.GetClusterDescriptor(0).GetPageRange(columnId).GetPageInfos().size(), 1u); + + auto viewX = reader->GetView("x"); + auto viewY = reader->GetView("y"); + for (int i = 0; i < 1000; ++i) { + EXPECT_EQ(i, viewX(i)); + EXPECT_EQ(-i, viewY(i)); + } + } + + StopMockServer(done, serverThread, host, port); + UnsetS3Env(); +} + +TEST(RPageSourceS3Wire, RoundTripManyClusters) +{ + TServerSocket server(0, false, TServerSocket::kDefaultBacklog, -1, ESocketBindOption::kInaddrLoopback); + const std::string host = server.GetLocalInetAddress().GetHostAddress(); + const int port = server.GetLocalPort(); + const std::string uri = "ntpl+s3+http://" + host + ":" + std::to_string(port) + "/wirebucket/manyclusters"; + + SetDummyS3Env(); + + std::map store; + std::atomic done{false}; + std::thread serverThread([&] { RunMockServer(server, store, done); }); + + // A tiny target cluster size produces many clusters, so LoadClusters() iterates over its outer loop + // and the base class walks several cluster groups in Attach(). + ROOT::RNTupleWriteOptions writeOptions; + writeOptions.SetApproxZippedClusterSize(1024); + { + ROOT::TestSupport::CheckDiagsRAII diags; + diags.optionalDiag(kWarning, "[ROOT.NTuple]", "experimental", /*matchFullMessage=*/false); + + auto model = ROOT::RNTupleModel::Create(); + auto fldX = model->MakeField("x"); + auto writer = ROOT::RNTupleWriter::Recreate(std::move(model), "manyclusters", uri, writeOptions); + for (int i = 0; i < 5000; ++i) { + *fldX = static_cast(i); + writer->Fill(); + } + } + + { + auto reader = ROOT::RNTupleReader::Open("manyclusters", uri); + EXPECT_EQ(5000u, reader->GetNEntries()); + ASSERT_GT(reader->GetDescriptor().GetNClusters(), 1u); + + auto viewX = reader->GetView("x"); + for (int i = 0; i < 5000; ++i) + EXPECT_FLOAT_EQ(static_cast(i), viewX(i)); + } + + StopMockServer(done, serverThread, host, port); + UnsetS3Env(); +} + +TEST(RPageSourceS3Wire, RoundTripWithoutPageChecksums) +{ + TServerSocket server(0, false, TServerSocket::kDefaultBacklog, -1, ESocketBindOption::kInaddrLoopback); + const std::string host = server.GetLocalInetAddress().GetHostAddress(); + const int port = server.GetLocalPort(); + const std::string uri = "ntpl+s3+http://" + host + ":" + std::to_string(port) + "/wirebucket/nochecksum"; + + SetDummyS3Env(); + + std::map store; + std::atomic done{false}; + std::thread serverThread([&] { RunMockServer(server, store, done); }); + + // Page buffers are sized as payload + HasChecksum() * kNBytesPageChecksum. Every other test covers + // the checksummed case (the default), so this one covers the other branch of that arithmetic. + ROOT::RNTupleWriteOptions writeOptions; + writeOptions.SetEnablePageChecksums(false); + { + ROOT::TestSupport::CheckDiagsRAII diags; + diags.optionalDiag(kWarning, "[ROOT.NTuple]", "experimental", /*matchFullMessage=*/false); + + auto model = ROOT::RNTupleModel::Create(); + auto fldX = model->MakeField("x"); + auto writer = ROOT::RNTupleWriter::Recreate(std::move(model), "nochecksum", uri, writeOptions); + for (int i = 0; i < 100; ++i) { + *fldX = i * 3; + writer->Fill(); + } + } + + { + auto reader = ROOT::RNTupleReader::Open("nochecksum", uri); + EXPECT_EQ(100u, reader->GetNEntries()); + + auto viewX = reader->GetView("x"); + for (int i = 0; i < 100; ++i) + EXPECT_EQ(i * 3, viewX(i)); + } + + StopMockServer(done, serverThread, host, port); + UnsetS3Env(); +} + +TEST(RPageSourceS3Wire, CloneReadsFromItsOwnConnection) +{ + TServerSocket server(0, false, TServerSocket::kDefaultBacklog, -1, ESocketBindOption::kInaddrLoopback); + const std::string host = server.GetLocalInetAddress().GetHostAddress(); + const int port = server.GetLocalPort(); + const std::string uri = "ntpl+s3+http://" + host + ":" + std::to_string(port) + "/wirebucket/clone"; + + SetDummyS3Env(); + + std::map store; + std::atomic done{false}; + std::thread serverThread([&] { RunMockServer(server, store, done); }); + + { + ROOT::TestSupport::CheckDiagsRAII diags; + diags.optionalDiag(kWarning, "[ROOT.NTuple]", "experimental", /*matchFullMessage=*/false); + + auto model = ROOT::RNTupleModel::Create(); + auto fldX = model->MakeField("x"); + auto writer = ROOT::RNTupleWriter::Recreate(std::move(model), "clone", uri); + for (int i = 0; i < 50; ++i) { + *fldX = i; + writer->Fill(); + } + } + + { + ROOT::Experimental::Internal::RPageSourceS3 source("clone", uri, ROOT::RNTupleReadOptions()); + source.Attach(); + ASSERT_EQ(50u, source.GetNEntries()); + + // Clone() copies the descriptor of an attached source, so the clone must come back attached + // without re-running LoadStructureImpl(), yet still be able to do I/O over its own connection. + auto clone = source.Clone(); + ASSERT_EQ(50u, clone->GetNEntries()); + + ROOT::DescriptorId_t columnId; + { + auto descGuard = clone->GetSharedDescriptorGuard(); + columnId = descGuard->FindPhysicalColumnId(descGuard->FindFieldId("x"), 0, 0); + } + + // A null buffer asks only for the size; the second call transfers and verifies the checksum, + // so reaching the end of this block proves the clone read real bytes from the mock. + RPageStorage::RSealedPage sealedPage; + clone->LoadSealedPage(columnId, ROOT::RNTupleLocalIndex(0, 0), sealedPage); + ASSERT_GT(sealedPage.GetBufferSize(), 0u); + + auto buffer = MakeUninitArray(sealedPage.GetBufferSize()); + sealedPage.SetBuffer(buffer.get()); + clone->LoadSealedPage(columnId, ROOT::RNTupleLocalIndex(0, 0), sealedPage); + EXPECT_EQ(50u, sealedPage.GetNElements()); + } + + StopMockServer(done, serverThread, host, port); + UnsetS3Env(); +} + +TEST(RPageSourceS3Wire, HonoursAnchorUrlTemplate) +{ + TServerSocket server(0, false, TServerSocket::kDefaultBacklog, -1, ESocketBindOption::kInaddrLoopback); + const std::string host = server.GetLocalInetAddress().GetHostAddress(); + const int port = server.GetLocalPort(); + const std::string basePath = "/wirebucket/template"; + const std::string uri = "ntpl+s3+http://" + host + ":" + std::to_string(port) + basePath; + + SetDummyS3Env(); + + std::map store; + std::atomic done{false}; + std::thread serverThread([&] { RunMockServer(server, store, done); }); + + { + ROOT::TestSupport::CheckDiagsRAII diags; + diags.optionalDiag(kWarning, "[ROOT.NTuple]", "experimental", /*matchFullMessage=*/false); + + auto model = ROOT::RNTupleModel::Create(); + auto fldX = model->MakeField("x"); + auto writer = ROOT::RNTupleWriter::Recreate(std::move(model), "template", uri); + for (int i = 0; i < 30; ++i) { + *fldX = i * 7; + writer->Fill(); + } + } + + StopMockServer(done, serverThread, host, port); + + // Relocate every data object under a "data/" segment and rewrite the anchor to describe the new + // layout. The writer only ever emits the default template, so this is the only way to prove that the + // source resolves the stored one instead of assuming "/". + std::map relocated; + for (const auto &[key, value] : store) { + if (key == basePath) + continue; // the anchor itself stays at the base path + ASSERT_EQ(0u, key.rfind(basePath + "/", 0)); + relocated[basePath + "/data/" + key.substr(basePath.size() + 1)] = value; + } + ASSERT_FALSE(relocated.empty()); + + auto jsonAnchor = nlohmann::json::parse(store[basePath]); + jsonAnchor.erase("checksum"); + jsonAnchor["urlTemplate"] = "${baseurl}/data/${objid}"; + const auto canonicalJson = jsonAnchor.dump(-1); + jsonAnchor["checksum"] = XXH3_64bits(canonicalJson.data(), canonicalJson.size()); + + relocated[basePath] = jsonAnchor.dump(2); + store = std::move(relocated); + + done.store(false); + std::thread secondServerThread([&] { RunMockServer(server, store, done); }); + + { + auto reader = ROOT::RNTupleReader::Open("template", uri); + EXPECT_EQ(30u, reader->GetNEntries()); + + auto viewX = reader->GetView("x"); + for (int i = 0; i < 30; ++i) + EXPECT_EQ(i * 7, viewX(i)); + } + + StopMockServer(done, secondServerThread, host, port); + UnsetS3Env(); +} + +TEST(RPageSourceS3Wire, ChecksNTupleName) +{ + TServerSocket server(0, false, TServerSocket::kDefaultBacklog, -1, ESocketBindOption::kInaddrLoopback); + const std::string host = server.GetLocalInetAddress().GetHostAddress(); + const int port = server.GetLocalPort(); + const std::string uri = "ntpl+s3+http://" + host + ":" + std::to_string(port) + "/wirebucket/named"; + + SetDummyS3Env(); + + std::map store; + std::atomic done{false}; + std::thread serverThread([&] { RunMockServer(server, store, done); }); + + { + ROOT::TestSupport::CheckDiagsRAII diags; + diags.optionalDiag(kWarning, "[ROOT.NTuple]", "experimental", /*matchFullMessage=*/false); + + auto model = ROOT::RNTupleModel::Create(); + auto fldX = model->MakeField("x"); + auto writer = ROOT::RNTupleWriter::Recreate(std::move(model), "signal", uri); + for (int i = 0; i < 10; ++i) { + *fldX = i; + writer->Fill(); + } + } + + // The name locates nothing in S3, but asking for the wrong one usually means the URL is wrong, so it + // is reported rather than ignored. + EXPECT_THROW(ROOT::RNTupleReader::Open("background", uri), ROOT::RException); + + // The matching name works, and so does an empty one for a caller that does not know it up front. + { + auto reader = ROOT::RNTupleReader::Open("signal", uri); + EXPECT_EQ(10u, reader->GetNEntries()); + } + { + ROOT::Experimental::Internal::RPageSourceS3 source("", uri, ROOT::RNTupleReadOptions()); + source.Attach(); + EXPECT_EQ(10u, source.GetNEntries()); + } + + StopMockServer(done, serverThread, host, port); + UnsetS3Env(); +} + +TEST(RPageSourceS3Wire, MissingPageObjectFails) +{ + TServerSocket server(0, false, TServerSocket::kDefaultBacklog, -1, ESocketBindOption::kInaddrLoopback); + const std::string host = server.GetLocalInetAddress().GetHostAddress(); + const int port = server.GetLocalPort(); + const std::string basePath = "/wirebucket/lostpage"; + const std::string uri = "ntpl+s3+http://" + host + ":" + std::to_string(port) + basePath; + + SetDummyS3Env(); + + std::map store; + std::atomic done{false}; + std::thread serverThread([&] { RunMockServer(server, store, done); }); + + { + ROOT::TestSupport::CheckDiagsRAII diags; + diags.optionalDiag(kWarning, "[ROOT.NTuple]", "experimental", /*matchFullMessage=*/false); + + auto model = ROOT::RNTupleModel::Create(); + auto fldX = model->MakeField("x"); + auto writer = ROOT::RNTupleWriter::Recreate(std::move(model), "lostpage", uri); + for (int i = 0; i < 20; ++i) { + *fldX = i; + writer->Fill(); + } + } + + // Take the server down before touching the store, so the map is not read and written concurrently. + StopMockServer(done, serverThread, host, port); + + // Object 0 is the header and the anchor lives at the base path, so dropping object 1 leaves the + // metadata intact: Attach() succeeds and the failure surfaces later, when a page is actually read. + ASSERT_EQ(1u, store.erase(basePath + "/1")); + + done.store(false); + std::thread secondServerThread([&] { RunMockServer(server, store, done); }); + + // The cluster cache is turned off deliberately. With it on, the failing GET happens inside + // LoadClusters() on the cluster pool's I/O thread, and RClusterPool::ExecReadClusters() does not + // catch exceptions, so the throw would terminate the process instead of reaching the caller. With + // the cache off the page is read synchronously and the exception propagates as it should. + ROOT::RNTupleReadOptions readOptions; + readOptions.SetClusterCache(ROOT::RNTupleReadOptions::EClusterCache::kOff); + { + auto reader = ROOT::RNTupleReader::Open("lostpage", uri, readOptions); + EXPECT_EQ(20u, reader->GetNEntries()); + auto viewX = reader->GetView("x"); + EXPECT_THROW(viewX(0), ROOT::RException); + } + + StopMockServer(done, secondServerThread, host, port); + UnsetS3Env(); +} + +TEST(RPageSourceS3Wire, ReadMissingAnchorFails) +{ + TServerSocket server(0, false, TServerSocket::kDefaultBacklog, -1, ESocketBindOption::kInaddrLoopback); + const std::string host = server.GetLocalInetAddress().GetHostAddress(); + const std::string uri = + "ntpl+s3+http://" + host + ":" + std::to_string(server.GetLocalPort()) + "/wirebucket/missing"; + + SetDummyS3Env(); + + std::map store; // empty: nothing was ever written under this prefix + + // The source's very first request is the HEAD on the anchor; answering it with a 404 is enough. + std::thread serverThread([&] { + TSocket *sock = server.Accept(); + if (sock && sock != reinterpret_cast(-1)) { + ServeS3Request(sock, store); + sock->Close(); + } + }); + + EXPECT_THROW(ROOT::RNTupleReader::Open("test", uri), ROOT::RException); + + serverThread.join(); + UnsetS3Env(); +} + +// ==================== Integration Tests (credential-gated) ==================== + +// These run against a real S3 service over https (CERN Ceph, AWS) and are skipped unless S3_ENDPOINT, +// S3_BUCKET, S3_ACCESS_KEY and S3_SECRET_KEY are all set. S3_ENDPOINT is a bare host, without a scheme. +class RPageS3IntegrationTest : public ::testing::Test { +protected: + std::string fEndpoint; + std::string fBucket; + + void SetUp() override + { + const char *endpoint = gSystem->Getenv("S3_ENDPOINT"); + const char *bucket = gSystem->Getenv("S3_BUCKET"); + const char *accessKey = gSystem->Getenv("S3_ACCESS_KEY"); + const char *secretKey = gSystem->Getenv("S3_SECRET_KEY"); + if (!endpoint || !bucket || !accessKey || !secretKey) + GTEST_SKIP() << "S3 credentials not set; skipping integration test"; + fEndpoint = endpoint; + fBucket = bucket; + } + + std::string MakeUri(const std::string &prefix) const + { + return "ntpl+s3+https://" + fEndpoint + "/" + fBucket + "/" + prefix; + } +}; + +TEST_F(RPageS3IntegrationTest, RoundTripSimple) +{ + const auto uri = MakeUri("roundtrip_simple"); + { + ROOT::TestSupport::CheckDiagsRAII diags; + diags.optionalDiag(kWarning, "[ROOT.NTuple]", "experimental", /*matchFullMessage=*/false); + + auto model = ROOT::RNTupleModel::Create(); + auto fldX = model->MakeField("x"); + auto fldY = model->MakeField("y"); + auto writer = ROOT::RNTupleWriter::Recreate(std::move(model), "test", uri); + for (int i = 0; i < 1000; ++i) { + *fldX = static_cast(i) * 0.5f; + *fldY = i * i; + writer->Fill(); + } + } + + auto reader = ROOT::RNTupleReader::Open("test", uri); + EXPECT_EQ(1000u, reader->GetNEntries()); + auto viewX = reader->GetView("x"); + auto viewY = reader->GetView("y"); + for (int i = 0; i < 1000; ++i) { + EXPECT_FLOAT_EQ(static_cast(i) * 0.5f, viewX(i)); + EXPECT_EQ(i * i, viewY(i)); + } +} + +TEST_F(RPageS3IntegrationTest, RoundTripStrings) +{ + const auto uri = MakeUri("roundtrip_strings"); + { + ROOT::TestSupport::CheckDiagsRAII diags; + diags.optionalDiag(kWarning, "[ROOT.NTuple]", "experimental", /*matchFullMessage=*/false); + + auto model = ROOT::RNTupleModel::Create(); + auto fldName = model->MakeField("name"); + auto writer = ROOT::RNTupleWriter::Recreate(std::move(model), "strings", uri); + for (int i = 0; i < 100; ++i) { + *fldName = "entry_" + std::to_string(i); + writer->Fill(); + } + } + + auto reader = ROOT::RNTupleReader::Open("strings", uri); + EXPECT_EQ(100u, reader->GetNEntries()); + auto viewName = reader->GetView("name"); + for (int i = 0; i < 100; ++i) + EXPECT_EQ("entry_" + std::to_string(i), viewName(i)); +} + +TEST_F(RPageS3IntegrationTest, RoundTripEmpty) +{ + // No entries means no pages and no cluster groups; the read path has to cope with that. + const auto uri = MakeUri("roundtrip_empty"); + { + ROOT::TestSupport::CheckDiagsRAII diags; + diags.optionalDiag(kWarning, "[ROOT.NTuple]", "experimental", /*matchFullMessage=*/false); + + auto model = ROOT::RNTupleModel::Create(); + model->MakeField("x"); + auto writer = ROOT::RNTupleWriter::Recreate(std::move(model), "empty", uri); + } + + auto reader = ROOT::RNTupleReader::Open("empty", uri); + EXPECT_EQ(0u, reader->GetNEntries()); +} + +TEST_F(RPageS3IntegrationTest, RoundTripMultipleClusters) +{ + // A small cluster size forces many clusters, so LoadClusters() runs repeatedly (and from the cluster + // pool's background thread) rather than just once. + const auto uri = MakeUri("roundtrip_clusters"); + ROOT::RNTupleWriteOptions opts; + opts.SetApproxZippedClusterSize(1024); + { + ROOT::TestSupport::CheckDiagsRAII diags; + diags.optionalDiag(kWarning, "[ROOT.NTuple]", "experimental", /*matchFullMessage=*/false); + + auto model = ROOT::RNTupleModel::Create(); + auto fldX = model->MakeField("x"); + auto writer = ROOT::RNTupleWriter::Recreate(std::move(model), "clusters", uri, opts); + for (int i = 0; i < 10000; ++i) { + *fldX = static_cast(i); + writer->Fill(); + } + } + + auto reader = ROOT::RNTupleReader::Open("clusters", uri); + EXPECT_EQ(10000u, reader->GetNEntries()); + auto viewX = reader->GetView("x"); + for (int i = 0; i < 10000; ++i) + EXPECT_FLOAT_EQ(static_cast(i), viewX(i)); +}