Fix creating rollup bug from multi SegmentGroups generated streaming … (#1287)

In streaming load, one version will generate multi SegmentGroups.
Upon creating rollup, the previous code only check version exists or not.
Instead, every SegmentGroup should be checked independently.

Author: lichaoyong

be/src/olap/olap_table.cpp

@@ -1867,8 +1867,19 @@ void OLAPTable::_list_files_with_suffix(const string& file_suffix, set<string>*
     }
 }
 
-bool OLAPTable::has_version(const Version& version) const {
-    return (_data_sources.find(version) != _data_sources.end());
+bool OLAPTable::has_segment_group(const Version& version, const SegmentGroup* new_segment_group) const {
+    auto it = _data_sources.find(version);
+    if (it == _data_sources.end()) {
+        return false;
+    }
+    bool exist = false;
+    for (auto segment_group : it->second) {
+        if (segment_group->segment_group_id() == new_segment_group->segment_group_id()) {
+            exist = true;
+            break;
+        }
+    }
+    return exist;
 }
 
 void OLAPTable::list_versions(vector<Version>* versions) const {

be/src/olap/olap_table.h

@@ -234,7 +234,7 @@ class OLAPTable : public std::enable_shared_from_this<OLAPTable> {
 
     void list_index_files(std::set<std::string>* filenames) const;
 
-    bool has_version(const Version& version) const;
+    bool has_segment_group(const Version& version, const SegmentGroup* new_segment_group) const;
 
     void list_versions(std::vector<Version>* versions) const;
 

be/src/olap/schema_change.cpp

@@ -2106,7 +2106,7 @@ OLAPStatus SchemaChangeHandler::_alter_table(SchemaChangeParams* sc_params) {
         sc_params->ref_olap_table->obtain_header_wrlock();
         sc_params->new_olap_table->obtain_header_wrlock();
 
-        if (!sc_params->new_olap_table->has_version((*it)->version())) {
+        if (!sc_params->new_olap_table->has_segment_group((*it)->version(), new_segment_group)) {
             // register version
             std::vector<SegmentGroup*> segment_group_vec;
             segment_group_vec.push_back(new_segment_group);