ecmwf · ChrisspyB · Jan 10, 2025 · Jan 13, 2025 · Jan 15, 2025 · Jan 24, 2025
diff --git a/src/dummy_daos/dummy_daos.cc b/src/dummy_daos/dummy_daos.cc
@@ -33,6 +33,7 @@ const PathName& dummy_daos_root() {
 
     static PathName tmpdir    = eckit::Resource<PathName>("$TMPDIR", "/tmp");
     static PathName daos_root = eckit::Resource<PathName>("$DUMMY_DAOS_DATA_ROOT", tmpdir / "fdb5_dummy_daos");
+    daos_root                 = daos_root.realName();
     return daos_root;
 }
 

diff --git a/src/fdb5/CMakeLists.txt b/src/fdb5/CMakeLists.txt
@@ -46,6 +46,7 @@ list( APPEND fdb5_srcs
     api/helpers/LockIterator.h
     api/helpers/MoveIterator.h
     api/helpers/StatusIterator.h
+    api/helpers/WipeIterator.cc
     api/helpers/WipeIterator.h
     api/helpers/PurgeIterator.h
     api/helpers/StatsIterator.cc
@@ -112,14 +113,17 @@ list( APPEND fdb5_srcs
     database/IndexStats.h
     database/Inspector.cc
     database/Inspector.h
+    database/Signature.h
     database/StatsReportVisitor.cc
     database/StatsReportVisitor.h
     database/Store.cc
     database/Store.h
     database/PurgeVisitor.cc
     database/PurgeVisitor.h
-    database/WipeVisitor.cc
-    database/WipeVisitor.h
+    database/WipeCoordinator.cc
+    database/WipeCoordinator.h
+    database/WipeState.cc
+    database/WipeState.h
     database/MoveVisitor.cc
     database/MoveVisitor.h
     database/IndexAxis.cc
@@ -322,8 +326,6 @@ if( HAVE_TOCFDB )
         toc/TocPurgeVisitor.h
         toc/TocSerialisationVersion.cc
         toc/TocSerialisationVersion.h
-        toc/TocWipeVisitor.cc
-        toc/TocWipeVisitor.h
         toc/TocMoveVisitor.cc
         toc/TocMoveVisitor.h
         toc/TocRecord.cc
@@ -393,8 +395,6 @@ if( HAVE_DAOSFDB )
         daos/DaosIndexLocation.cc
         daos/DaosCommon.h
         daos/DaosCommon.cc
-        daos/DaosWipeVisitor.h
-        daos/DaosWipeVisitor.cc
         daos/DaosArrayPartHandle.h
         daos/DaosArrayPartHandle.cc
         daos/DaosLazyFieldLocation.h

diff --git a/src/fdb5/api/DistFDB.cc b/src/fdb5/api/DistFDB.cc
@@ -30,6 +30,7 @@
 #include "fdb5/api/helpers/FDBToolRequest.h"
 #include "fdb5/api/helpers/ListIterator.h"
 #include "fdb5/database/Notifier.h"
+#include "fdb5/database/WipeState.h"
 #include "fdb5/io/HandleGatherer.h"
 
 using eckit::Log;
@@ -190,11 +191,10 @@ StatusIterator DistFDB::status(const FDBToolRequest& request) {
     return queryInternal(request, [](FDB& fdb, const FDBToolRequest& request) { return fdb.status(request); });
 }
 
-WipeIterator DistFDB::wipe(const FDBToolRequest& request, bool doit, bool porcelain, bool unsafeWipeAll) {
+// DIST FDB WILL BE REMOVED IN A FUTURE UPDATE
+WipeStateIterator DistFDB::wipe(const FDBToolRequest& request, bool doit, bool porcelain, bool unsafeWipeAll) {
     LOG_DEBUG_LIB(LibFdb5) << "DistFDB::wipe() : " << request << std::endl;
-    return queryInternal(request, [doit, porcelain, unsafeWipeAll](FDB& fdb, const FDBToolRequest& request) {
-        return fdb.wipe(request, doit, porcelain, unsafeWipeAll);
-    });
+    NOTIMP;
 }
 
 PurgeIterator DistFDB::purge(const FDBToolRequest& request, bool doit, bool porcelain) {

diff --git a/src/fdb5/api/DistFDB.h b/src/fdb5/api/DistFDB.h
@@ -53,7 +53,7 @@ class DistFDB : public FDBBase {
 
     StatusIterator status(const FDBToolRequest& request) override;
 
-    WipeIterator wipe(const FDBToolRequest& request, bool doit, bool porcelain, bool unsafeWipeAll) override;
+    WipeStateIterator wipe(const FDBToolRequest& request, bool doit, bool porcelain, bool unsafeWipeAll) override;
 
     PurgeIterator purge(const FDBToolRequest& request, bool doit, bool porcelain) override;
 

diff --git a/src/fdb5/api/FDB.cc b/src/fdb5/api/FDB.cc
@@ -35,8 +35,11 @@
 #include "fdb5/api/helpers/FDBToolRequest.h"
 #include "fdb5/api/helpers/ListElement.h"
 #include "fdb5/api/helpers/ListIterator.h"
+#include "fdb5/api/helpers/WipeIterator.h"
 #include "fdb5/database/FieldLocation.h"
 #include "fdb5/database/Key.h"
+#include "fdb5/database/WipeCoordinator.h"
+#include "fdb5/database/WipeState.h"
 #include "fdb5/io/FieldHandle.h"
 #include "fdb5/io/HandleGatherer.h"
 #include "fdb5/message/MessageDecoder.h"
@@ -254,7 +257,24 @@ StatusIterator FDB::status(const FDBToolRequest& request) {
 }
 
 WipeIterator FDB::wipe(const FDBToolRequest& request, bool doit, bool porcelain, bool unsafeWipeAll) {
-    return internal_->wipe(request, doit, porcelain, unsafeWipeAll);
+
+    auto async = [this, request, doit, porcelain, unsafeWipeAll](eckit::Queue<WipeElement>& queue) {
+        // Visit the catalogues to determine what they would wipe
+        WipeStateIterator it = internal_->wipe(request, doit, porcelain, unsafeWipeAll);
+
+        // Coordinate the wipe across catalogues and stores
+        WipeCoordinator coordinator{internal_->config()};
+        CatalogueWipeState catalogueWipeState;
+        while (it.next(catalogueWipeState)) {
+
+            auto elements = coordinator.wipe(catalogueWipeState, doit, unsafeWipeAll);
+            for (auto& el : elements) {
+                queue.emplace(el);
+            }
+        }
+    };
+
+    return WipeIterator(new APIAsyncIterator<WipeElement>(internal_->shared(), async));
 }
 
 PurgeIterator FDB::purge(const FDBToolRequest& request, bool doit, bool porcelain) {

diff --git a/src/fdb5/api/FDB.h b/src/fdb5/api/FDB.h
@@ -51,6 +51,10 @@ class MarsRequest;
 
 namespace fdb5 {
 
+namespace remote {
+struct WipeHelper;
+}
+
 class FDBBase;
 class FDBToolRequest;
 class Key;
@@ -291,6 +295,10 @@ class FDB {
 
     bool sorted(const metkit::mars::MarsRequest& request);
 
+private:
+
+    friend struct remote::WipeHelper;
+
 private:  // members
 
     /// @brief The FDBBase instance is held in a shared_ptr so that it can be kept alive by any Iterator instances

diff --git a/src/fdb5/api/FDBFactory.h b/src/fdb5/api/FDBFactory.h
@@ -30,8 +30,8 @@
 #include "fdb5/api/helpers/PurgeIterator.h"
 #include "fdb5/api/helpers/StatsIterator.h"
 #include "fdb5/api/helpers/StatusIterator.h"
-#include "fdb5/api/helpers/WipeIterator.h"
 #include "fdb5/config/Config.h"
+#include "fdb5/database/WipeState.h"
 
 namespace eckit::message {
 
@@ -68,6 +68,8 @@ class FDBBase : public std::enable_shared_from_this<FDBBase>, public CallbackReg
 
     virtual ~FDBBase() = default;
 
+    std::shared_ptr<FDBBase> shared() { return shared_from_this(); }
+
     // -------------- Primary API functions ----------------------------
 
     virtual void archive(const Key& key, const void* data, size_t length) = 0;
@@ -84,7 +86,7 @@ class FDBBase : public std::enable_shared_from_this<FDBBase>, public CallbackReg
 
     virtual StatusIterator status(const FDBToolRequest& request) = 0;
 
-    virtual WipeIterator wipe(const FDBToolRequest& request, bool doit, bool porcelain, bool unsafeWipeAll) = 0;
+    virtual WipeStateIterator wipe(const FDBToolRequest& request, bool doit, bool porcelain, bool unsafeWipeAll) = 0;
 
     virtual PurgeIterator purge(const FDBToolRequest& request, bool doit, bool porcelain) = 0;
 

diff --git a/src/fdb5/api/LocalFDB.cc b/src/fdb5/api/LocalFDB.cc
@@ -31,6 +31,7 @@
 #include "fdb5/api/local/StatusVisitor.h"
 #include "fdb5/api/local/WipeVisitor.h"
 #include "fdb5/database/Archiver.h"
+#include "fdb5/database/Catalogue.h"
 #include "fdb5/database/EntryVisitMechanism.h"
 #include "fdb5/database/Inspector.h"
 #include "fdb5/database/Key.h"
@@ -103,9 +104,9 @@ StatusIterator LocalFDB::status(const FDBToolRequest& request) {
     return queryInternal<StatusVisitor>(request);
 }
 
-WipeIterator LocalFDB::wipe(const FDBToolRequest& request, bool doit, bool porcelain, bool unsafeWipeAll) {
+WipeStateIterator LocalFDB::wipe(const FDBToolRequest& request, bool doit, bool porcelain, bool unsafeWipeAll) {
     LOG_DEBUG_LIB(LibFdb5) << "LocalFDB::wipe() : " << request << std::endl;
-    return queryInternal<fdb5::api::local::WipeVisitor>(request, doit, porcelain, unsafeWipeAll);
+    return queryInternal<WipeCatalogueVisitor>(request, doit);
 }
 
 MoveIterator LocalFDB::move(const FDBToolRequest& request, const eckit::URI& dest) {

diff --git a/src/fdb5/api/LocalFDB.h b/src/fdb5/api/LocalFDB.h
@@ -49,7 +49,7 @@ class LocalFDB : public FDBBase {
 
     StatusIterator status(const FDBToolRequest& request) override;
 
-    WipeIterator wipe(const FDBToolRequest& request, bool doit, bool porcelain, bool unsafeWipeAll) override;
+    WipeStateIterator wipe(const FDBToolRequest& request, bool doit, bool porcelain, bool unsafeWipeAll) override;
 
     PurgeIterator purge(const FDBToolRequest& request, bool doit, bool porcelain) override;
 

diff --git a/src/fdb5/api/RemoteFDB.cc b/src/fdb5/api/RemoteFDB.cc
@@ -13,6 +13,7 @@
 #include "fdb5/api/helpers/ListElement.h"
 #include "fdb5/database/Archiver.h"
 #include "fdb5/database/Inspector.h"
+#include "fdb5/database/WipeState.h"
 
 #include "fdb5/remote/RemoteFieldLocation.h"
 #include "fdb5/remote/client/ClientConnectionRouter.h"
@@ -49,9 +50,11 @@ struct ListHelper : BaseAPIHelper<fdb5::ListElement, fdb5::remote::Message::List
 
         if (elem.hasLocation()) {
 
-            eckit::Log::debug<fdb5::LibFdb5>() << "ListHelper::valueFromStream - original location: ";
-            elem.location().dump(eckit::Log::debug<fdb5::LibFdb5>());
-            eckit::Log::debug<fdb5::LibFdb5>() << std::endl;
+            if (fdb5::LibFdb5::instance().debug()) {
+                eckit::Log::debug<fdb5::LibFdb5>() << "ListHelper::valueFromStream - original location: ";
+                elem.location().dump(eckit::Log::debug<fdb5::LibFdb5>());
+                eckit::Log::debug<fdb5::LibFdb5>() << std::endl;
+            }
 
             // TODO move the endpoint replacement to the server side ()
             if (elem.location().uri().scheme() == "fdb") {
@@ -92,9 +95,11 @@ struct InspectHelper : BaseAPIHelper<fdb5::ListElement, fdb5::remote::Message::I
     static fdb5::ListElement valueFromStream(eckit::Stream& s, fdb5::RemoteFDB* fdb) {
         fdb5::ListElement elem(s);
 
-        eckit::Log::debug<fdb5::LibFdb5>() << "InspectHelper::valueFromStream - original location: ";
-        elem.location().dump(eckit::Log::debug<fdb5::LibFdb5>());
-        eckit::Log::debug<fdb5::LibFdb5>() << std::endl;
+        if (fdb5::LibFdb5::instance().debug()) {
+            eckit::Log::debug<fdb5::LibFdb5>() << "InspectHelper::valueFromStream - original location: ";
+            elem.location().dump(eckit::Log::debug<fdb5::LibFdb5>());
+            eckit::Log::debug<fdb5::LibFdb5>() << std::endl;
+        }
 
         if (elem.location().uri().scheme() == "fdb") {
             eckit::net::Endpoint fieldLocationEndpoint{elem.location().uri().host(), elem.location().uri().port()};
@@ -111,6 +116,25 @@ struct InspectHelper : BaseAPIHelper<fdb5::ListElement, fdb5::remote::Message::I
     }
 };
 
+struct WipeHelper : BaseAPIHelper<fdb5::CatalogueWipeState, fdb5::remote::Message::Wipe> {
+
+    WipeHelper(bool doit, bool porcelain, bool unsafeWipeAll) : doit_(doit), unsafeWipeAll_(unsafeWipeAll) {}
+
+    void encodeExtra(eckit::Stream& s) const {
+        s << doit_;
+        s << unsafeWipeAll_;
+    }
+
+    static fdb5::CatalogueWipeState valueFromStream(eckit::Stream& s, fdb5::RemoteFDB* fdb) {
+        return fdb5::CatalogueWipeState(s);
+    }
+
+private:
+
+    bool doit_;
+    bool unsafeWipeAll_;
+};
+
 }  // namespace
 
 namespace fdb5 {
@@ -281,6 +305,10 @@ StatsIterator RemoteFDB::stats(const FDBToolRequest& request) {
     return forwardApiCall(StatsHelper(), request);
 }
 
+WipeStateIterator RemoteFDB::wipe(const FDBToolRequest& request, bool doit, bool porcelain, bool unsafeWipeAll) {
+    return forwardApiCall(WipeHelper(doit, porcelain, unsafeWipeAll), request);
+}
+
 void RemoteFDB::print(std::ostream& s) const {
     s << "RemoteFDB(...)";
 }

diff --git a/src/fdb5/api/RemoteFDB.h b/src/fdb5/api/RemoteFDB.h
@@ -52,7 +52,7 @@ class RemoteFDB : public LocalFDB, public remote::Client {
 
     StatusIterator status(const FDBToolRequest& request) override { NOTIMP; }
 
-    WipeIterator wipe(const FDBToolRequest& request, bool doit, bool porcelain, bool unsafeWipeAll) override { NOTIMP; }
+    WipeStateIterator wipe(const FDBToolRequest& request, bool doit, bool porcelain, bool unsafeWipeAll) override;
 
     PurgeIterator purge(const FDBToolRequest& request, bool doit, bool porcelain) override { NOTIMP; }
 

diff --git a/src/fdb5/api/SelectFDB.cc b/src/fdb5/api/SelectFDB.cc
@@ -14,11 +14,18 @@
  */
 
 #include "fdb5/api/SelectFDB.h"
+#include <sstream>
+#include <vector>
+#include "eckit/log/CodeLocation.h"
 #include "eckit/log/Log.h"
 #include "fdb5/LibFdb5.h"
+#include "fdb5/api/FDB.h"
+#include "fdb5/api/FDBFactory.h"
 #include "fdb5/api/helpers/FDBToolRequest.h"
 #include "fdb5/api/helpers/ListIterator.h"
+#include "fdb5/api/helpers/WipeIterator.h"
 #include "fdb5/database/Key.h"
+#include "fdb5/database/WipeState.h"
 #include "fdb5/rules/SelectMatcher.h"
 
 using namespace eckit;
@@ -138,11 +145,30 @@ StatusIterator SelectFDB::status(const FDBToolRequest& request) {
     return queryInternal(request, [](FDBBase& fdb, const FDBToolRequest& request) { return fdb.status(request); });
 }
 
-WipeIterator SelectFDB::wipe(const FDBToolRequest& request, bool doit, bool porcelain, bool unsafeWipeAll) {
+WipeStateIterator SelectFDB::wipe(const FDBToolRequest& request, bool doit, bool porcelain, bool unsafeWipeAll) {
     LOG_DEBUG_LIB(LibFdb5) << "SelectFDB::wipe() >> " << request << std::endl;
-    return queryInternal(request, [doit, porcelain, unsafeWipeAll](FDBBase& fdb, const FDBToolRequest& request) {
-        return fdb.wipe(request, doit, porcelain, unsafeWipeAll);
-    });
+
+    FDBLane* matchingLane = nullptr;
+    for (auto& lane : subFdbs_) {
+        if (lane.matches(request.request(), Matcher::MatchOnMissing)) {
+            if (matchingLane != nullptr) {
+                std::stringstream ss;
+                ss << "Multiple matching lanes for request " << request.request();
+                ss << " - wipe request must not match multiple SelectFDB lanes.";
+                throw eckit::UserError(ss.str(), Here());
+            }
+
+            matchingLane = &lane;
+        }
+    }
+
+    if (matchingLane == nullptr) {
+        std::stringstream ss;
+        ss << "No matching lane for request " << request.request();
+        throw eckit::UserError(ss.str(), Here());
+    }
+
+    return matchingLane->get().wipe(request, doit, porcelain, unsafeWipeAll);
 }
 
 PurgeIterator SelectFDB::purge(const FDBToolRequest& request, bool doit, bool porcelain) {

diff --git a/src/fdb5/api/SelectFDB.h b/src/fdb5/api/SelectFDB.h
@@ -70,7 +70,7 @@ class SelectFDB : public FDBBase {
 
     StatusIterator status(const FDBToolRequest& request) override;
 
-    WipeIterator wipe(const FDBToolRequest& request, bool doit, bool porcelain, bool unsafeWipeAll) override;
+    WipeStateIterator wipe(const FDBToolRequest& request, bool doit, bool porcelain, bool unsafeWipeAll) override;
 
     PurgeIterator purge(const FDBToolRequest& request, bool doit, bool porcelain) override;
 

diff --git a/src/fdb5/api/fdb_c.cc b/src/fdb5/api/fdb_c.cc
@@ -226,6 +226,7 @@ struct fdb_datareader_t {
 };
 
 // Wipe iterator
+// To be removed / replaced by pybind impl.
 struct fdb_wipe_iterator_t {
 
     fdb_wipe_iterator_t(WipeIterator&& iter) : iter_(std::move(iter)) {}
@@ -241,7 +242,7 @@ struct fdb_wipe_element_t {
 
     fdb_wipe_element_t(WipeElement&& e) : element_(std::move(e)) {}
 
-    const char* c_str() const { return element_.c_str(); }
+    const char* c_str() const { return element_.msg().c_str(); }
 
 private:
 

diff --git a/src/fdb5/api/helpers/ListIterator.h b/src/fdb5/api/helpers/ListIterator.h
@@ -19,6 +19,7 @@
 #ifndef fdb5_ListIterator_H
 #define fdb5_ListIterator_H
 
+#include <unordered_map>
 #include <unordered_set>
 #include <utility>