Preserve cached struct element modification status

It's possible for one update operation to modify the element,
place it back as MODIFIED, then a second operation reads it out
of the cache, ends up being a no-op and puts it back as NOT_MODIFIED.
If we do not preserve the modification status, the changes from the
first operation will be lost when committing.

2 files changed, 52 insertions, 11 deletions

diff --git a/document/src/tests/fieldpathupdatetestcase.cpp b/document/src/tests/fieldpathupdatetestcase.cpp
index 36a1c339ebb..82443f13716 100644
--- a/document/src/tests/fieldpathupdatetestcase.cpp
+++ b/document/src/tests/fieldpathupdatetestcase.cpp
@@ -1124,4 +1124,29 @@ TEST_F(FieldPathUpdateTestCase, array_element_update_for_invalid_index_is_ignore
     EXPECT_EQ(str_array, *new_arr);
 }
 
+TEST_F(FieldPathUpdateTestCase, update_can_have_removes_for_both_existent_and_nonexistent_keys) {
+    DocumentId doc_id("id:ns:foobar::george:costanza");
+    auto doc = std::make_unique<Document>(_foobar_type, doc_id);
+    auto& map_type = dynamic_cast<const MapDataType&>(doc->getType().getField("structmap").getDataType());
+    auto& struct_type = map_type.getValueType();
+    MapFieldValue mfv(map_type);
+
+    StructFieldValue mystruct(struct_type);
+    mystruct.setValue("title", StringFieldValue("sharknado in space, part deux"));
+    mystruct.setValue("rating", IntFieldValue(90));
+    mfv.put(StringFieldValue("coolmovie"), mystruct);
+    doc->setValue("structmap", mfv);
+
+    DocumentUpdate update(*_repo, _foobar_type, doc_id);
+    auto update1 = std::make_unique<RemoveFieldPathUpdate>("structmap{coolmovie}", "");
+    auto update2 = std::make_unique<RemoveFieldPathUpdate>("structmap{no such key}", "");
+    update.addFieldPathUpdate(FieldPathUpdate::CP(std::move(update1)));
+    update.addFieldPathUpdate(FieldPathUpdate::CP(std::move(update2)));
+    update.applyTo(*doc);
+
+    auto new_value = doc->getValue("structmap");
+    auto& map_value = dynamic_cast<MapFieldValue&>(*new_value);
+    EXPECT_EQ(map_value.size(), 0);
+}
+
 }
diff --git a/document/src/vespa/document/fieldvalue/structuredfieldvalue.cpp b/document/src/vespa/document/fieldvalue/structuredfieldvalue.cpp
index 8c2d815e9f7..860151e4bb4 100644
--- a/document/src/vespa/document/fieldvalue/structuredfieldvalue.cpp
+++ b/document/src/vespa/document/fieldvalue/structuredfieldvalue.cpp
@@ -19,7 +19,7 @@ IMPLEMENT_IDENTIFIABLE_ABSTRACT(StructuredFieldValue, FieldValue);
 
 StructuredFieldValue::Iterator::Iterator()
     : _iterator(),
-      _field(0)
+      _field(nullptr)
 {
 }
 
@@ -79,7 +79,7 @@ FieldValue::UP
 StructuredFieldValue::onGetNestedFieldValue(PathRange nested) const
 {
     FieldValue::UP fv = getValue(nested.cur().getFieldRef());
-    if (fv.get() != NULL) {
+    if (fv.get() != nullptr) {
         PathRange next = nested.next();
         if ( ! next.atEnd() ) {
             return fv->getNestedFieldValue(next);
@@ -89,7 +89,18 @@ StructuredFieldValue::onGetNestedFieldValue(PathRange nested) const
 }
 
 namespace {
-    using ValuePair = std::pair<fieldvalue::ModificationStatus, FieldValue::UP>;
+    struct ValuePair {
+        fieldvalue::ModificationStatus status;
+        FieldValue::UP value;
+
+        ValuePair() : status(fieldvalue::ModificationStatus::NOT_MODIFIED), value() {}
+        ValuePair(fieldvalue::ModificationStatus status_,
+                  FieldValue::UP value_)
+            : status(status_),
+              value(std::move(value_))
+        {}
+    };
+
     using Cache = vespalib::hash_map<Field, ValuePair>;
 }
 
@@ -97,16 +108,21 @@ class StructuredCache {
 public:
     void remove(const Field & field) {
         ValuePair & entry = _cache[field];
-        entry.first = ModificationStatus::REMOVED;
-        entry.second.reset();
+        entry.status = ModificationStatus::REMOVED;
+        entry.value.reset();
     }
     Cache::iterator find(const Field & field) {
         return _cache.find(field);
     }
     void set(const Field & field, FieldValue::UP value, ModificationStatus status) {
         ValuePair & entry = _cache[field];
-        entry.first = status;
-        entry.second = std::move(value);
+        // If the entry has previously been tagged modified, the value we're now reinserting
+        // is likely based on those changes. We cannot lose that modification status.
+        entry.status = ((status == ModificationStatus::NOT_MODIFIED) &&
+                        (entry.status == ModificationStatus::MODIFIED))
+                ? ModificationStatus::MODIFIED
+                : status;
+        entry.value = std::move(value);
     }
     Cache::iterator begin() { return _cache.begin(); }
     Cache::iterator end() { return _cache.end(); }
@@ -148,7 +164,7 @@ StructuredFieldValue::getValue(const Field& field, FieldValue::UP container) con
             container = getFieldValue(field);
             _cache->set(field, FieldValue::UP(), ModificationStatus::NOT_MODIFIED);
         } else {
-            container = std::move(found->second.second);
+            container = std::move(found->second.value);
         }
     } else {
         if (container) {
@@ -232,10 +248,10 @@ StructuredFieldValue::beginTransaction() {
 void
 StructuredFieldValue::commitTransaction() {
     for (auto & e : *_cache) {
-        if (e.second.first == ModificationStatus::REMOVED) {
+        if (e.second.status == ModificationStatus::REMOVED) {
             removeFieldValue(e.first);
-        } else if (e.second.first == ModificationStatus ::MODIFIED) {
-            setFieldValue(e.first, std::move(e.second.second));
+        } else if (e.second.status == ModificationStatus::MODIFIED) {
+            setFieldValue(e.first, std::move(e.second.value));
         }
     }
     _cache.reset();