Fix inconsistency of three replicas belongs to one tablet (#523)

There are A, B, C replicas of one tablet.
A has 0 - 10 version.
B has 0 - 5, 6, 7, 9, 10 version.
1. B has missed versions, so it clones 0 - 10 from A, and remove overlapped versions in its header.
2. Coincidentally, 6 is a version for delete predicate (delete where day = 20181221).
   When removing overlapped versions, version 6 is removed but delete predicate is not be removed.
3. Unfortunately, 0-10 cloned from A has data indicated at 20181221.
4. B performs compaction, and data generated by 20181221 is be removed falsely.

be/src/olap/olap_header.cpp

@@ -460,6 +460,48 @@ void OLAPHeader::add_delete_condition(const DeleteConditionMessage& delete_condi
     LOG(INFO) << "add delete condition. version=" << version;
 }
 
+void OLAPHeader::delete_cond_by_version(const Version& version) {
+    DCHECK(version.first == version.second);
+    google::protobuf::RepeatedPtrField<DeleteConditionMessage>* delete_conditions
+            = mutable_delete_data_conditions();
+    int index = 0;
+    for (; index < delete_conditions->size(); ++index) {
+        const DeleteConditionMessage& temp = delete_conditions->Get(index);
+        if (temp.version() == version.first) {
+            // log delete condtion
+            string del_cond_str;
+            const RepeatedPtrField<string>& sub_conditions = temp.sub_conditions();
+
+            for (int i = 0; i != sub_conditions.size(); ++i) {
+                del_cond_str += sub_conditions.Get(i) + ";";
+            }
+
+            LOG(INFO) << "delete one condition. version=" << temp.version()
+                      << ", condition=" << del_cond_str;
+
+            // remove delete condition from PB
+            delete_conditions->SwapElements(index, delete_conditions->size() - 1);
+            delete_conditions->RemoveLast();
+        }
+    }
+}
+
+bool OLAPHeader::is_delete_data_version(Version version) {
+    if (version.first != version.second) {
+        return false;
+    }
+
+    google::protobuf::RepeatedPtrField<DeleteConditionMessage>::const_iterator it;
+    it = delete_data_conditions().begin();
+    for (; it != delete_data_conditions().end(); ++it) {
+        if (it->version() == version.first) {
+            return true;
+        }
+    }
+
+    return false;
+}
+
 const PPendingDelta* OLAPHeader::get_pending_delta(int64_t transaction_id) const {
     for (int i = 0; i < pending_delta_size(); i++) {
         if (pending_delta(i).transaction_id() == transaction_id) {

be/src/olap/olap_header.h

@@ -79,6 +79,8 @@ public:
                                        bool empty, const std::vector<KeyRange>* column_statistics);
 
     void add_delete_condition(const DeleteConditionMessage& delete_condition, int64_t version);
+    void delete_cond_by_version(const Version& version);
+    bool is_delete_data_version(Version version);
 
     const PPendingDelta* get_pending_delta(int64_t transaction_id) const;
     const PPendingSegmentGroup* get_pending_segment_group(int64_t transaction_id, int32_t pending_segment_group_id) const;

be/src/olap/olap_table.cpp

@@ -1307,6 +1307,9 @@ OLAPStatus OLAPTable::clone_data(const OLAPHeader& clone_header,
                              << " version=" << version.first << "-" << version.second << "]";
                 break;
             }
+            if (new_local_header.is_delete_data_version(version)) {
+                new_local_header.delete_cond_by_version(version);
+            }
             LOG(INFO) << "delete version from new local header when clone. [table='" << full_name()
                       << "', version=" << version.first << "-" << version.second << "]";
         }

be/src/olap/olap_table.h

@@ -528,19 +528,7 @@ public:
     }
 
     bool is_delete_data_version(Version version) {
-        if (version.first != version.second) {
-            return false;
-        }
-
-        google::protobuf::RepeatedPtrField<DeleteConditionMessage>::const_iterator it;
-        it = _header->delete_data_conditions().begin();
-        for (; it != _header->delete_data_conditions().end(); ++it) {
-            if (it->version() == version.first) {
-                return true;
-            }
-        }
-
-        return false;
+        return _header->is_delete_data_version(version);
     }
 
     bool is_load_delete_version(Version version);