FDB-319 backport to 5.11.17

src/fdb5/database/Archiver.cc

@@ -47,11 +47,14 @@ void Archiver::archive(const Key &key, BaseArchiveVisitor& visitor) {
 
     dbConfig_.schema().expand(key, visitor);
 
-    if (visitor.rule() == nullptr) { // Make sure we did find a rule that matched
+    const Rule* rule = visitor.rule();
+    if (rule == nullptr) { // Make sure we did find a rule that matched
         std::ostringstream oss;
         oss << "FDB: Could not find a rule to archive " << key;
         throw eckit::SeriousBug(oss.str());
     }
+    // validate metadata
+    rule->check(key);
 }
 
 void Archiver::flush() {

src/fdb5/rules/Rule.cc

@@ -20,7 +20,7 @@
 #include "fdb5/rules/Schema.h"
 #include "fdb5/database/ReadVisitor.h"
 #include "fdb5/database/WriteVisitor.h"
-
+#include "fdb5/types/Type.h"
 
 namespace fdb5 {
 
@@ -494,6 +494,17 @@ const Schema &Rule::schema() const {
     return schema_;
 }
 
+void Rule::check(const Key& key) const {
+    for (const auto& pred : predicates_ ) {
+        auto k = key.find(pred->keyword());
+        if (k != key.end()) {
+            const std::string& value = (*k).second;
+            const Type& type = registry_.lookupType(pred->keyword());
+            ASSERT(value == type.tidy(pred->keyword(), value));
+        }
+    }
+}
+
 std::ostream &operator<<(std::ostream &s, const Rule &x) {
     x.print(s);
     return s;

src/fdb5/rules/Rule.h

@@ -83,6 +83,8 @@ class Rule : public eckit::NonCopyable {
     const Schema &schema() const;
     const TypesRegistry &registry() const;
 
+    void check(const Key& key) const;
+
 private: // methods
 
     void expand(const metkit::mars::MarsRequest &request,