[feat](seq mapping) uniq table supports multi-stream merging through sequence mapping

Author: HonestManXin

be/src/olap/compaction.cpp

@@ -202,7 +202,7 @@ Status Compaction::merge_input_rowsets() {
     {
         SCOPED_TIMER(_merge_rowsets_latency_timer);
         // 1. Merge segment files and write bkd inverted index
-        if (_is_vertical) {
+        if (_is_vertical && !_tablet->tablet_schema()->has_seq_map()) {
             if (!_tablet->tablet_schema()->cluster_key_uids().empty()) {
                 RETURN_IF_ERROR(update_delete_bitmap());
             }

be/src/olap/memtable.cpp

@@ -244,6 +244,35 @@ Status MemTable::insert(const vectorized::Block* input_block,
     return Status::OK();
 }
 
+void MemTable::_aggregate_two_row_with_sequence_map(vectorized::MutableBlock& mutable_block,
+                                                 RowInBlock* src_row, RowInBlock* dst_row) {
+    const auto& seq_map = _tablet_schema->seq_map();
+    for (const auto& it : seq_map) {
+        auto sequence = it.first;
+        auto* sequence_col_ptr = mutable_block.mutable_columns()[sequence].get();
+        auto res = sequence_col_ptr->compare_at(dst_row->_row_pos, src_row->_row_pos,
+                                                *sequence_col_ptr, -1);
+        if (res > 0) {
+            continue;
+        }
+        for (auto cid : it.second) {
+            if (cid < _num_columns) {
+                auto* col_ptr = mutable_block.mutable_columns()[cid].get();
+                _agg_functions[cid]->add(dst_row->agg_places(cid),
+                                         const_cast<const doris::vectorized::IColumn**>(&col_ptr),
+                                         src_row->_row_pos, _arena);
+            }
+        }
+        if (sequence < _num_columns) {
+            _agg_functions[sequence]->add(
+                    dst_row->agg_places(sequence),
+                    const_cast<const doris::vectorized::IColumn**>(&sequence_col_ptr),
+                    src_row->_row_pos, _arena);
+            sequence_col_ptr->replace_column_data(*sequence_col_ptr, src_row->_row_pos, dst_row->_row_pos);
+        }
+    }
+}
+
 template <bool has_skip_bitmap_col>
 void MemTable::_aggregate_two_row_in_block(vectorized::MutableBlock& mutable_block,
                                            RowInBlock* src_row, RowInBlock* dst_row) {
@@ -504,7 +533,11 @@ void MemTable::_aggregate() {
                     _init_row_for_agg(prev_row, mutable_block);
                 }
                 _stat.merged_rows++;
-                _aggregate_two_row_in_block<has_skip_bitmap_col>(mutable_block, cur_row, prev_row);
+                if (_tablet_schema->has_seq_map()) {
+                    _aggregate_two_row_with_sequence_map(mutable_block, cur_row, prev_row);
+                } else {
+                   _aggregate_two_row_in_block<has_skip_bitmap_col>(mutable_block, cur_row, prev_row);
+                }
             } else {
                 prev_row = cur_row;
                 if (!temp_row_in_blocks.empty()) {

be/src/olap/memtable.h

@@ -210,6 +210,9 @@ class MemTable {
     void _aggregate_two_row_in_block(vectorized::MutableBlock& mutable_block, RowInBlock* new_row,
                                      RowInBlock* row_in_skiplist);
 
+    void _aggregate_two_row_with_sequence_map(vectorized::MutableBlock& mutable_block, RowInBlock* new_row,
+                                     RowInBlock* row_in_skiplist);
+
     // Used to wrapped by to_block to do exception handle logic
     Status _to_block(std::unique_ptr<vectorized::Block>* res);
 

be/src/olap/rowset/beta_rowset_reader.cpp

@@ -176,7 +176,8 @@ Status BetaRowsetReader::get_segment_iterators(RowsetReaderContext* read_context
     }
 
     if (_should_push_down_value_predicates()) {
-        if (_read_context->value_predicates != nullptr) {
+        if (_read_context->value_predicates != nullptr &&
+            !read_context->tablet_schema->has_seq_map()) {
             _read_options.column_predicates.insert(_read_options.column_predicates.end(),
                                                    _read_context->value_predicates->begin(),
                                                    _read_context->value_predicates->end());

be/src/olap/schema.h

@@ -78,7 +78,7 @@ class Schema {
             columns.push_back(std::make_shared<TabletColumn>(column));
         }
         _delete_sign_idx = tablet_schema->delete_sign_idx();
-        if (tablet_schema->has_sequence_col()) {
+        if (tablet_schema->has_sequence_col() || tablet_schema->has_seq_map()) {
             _has_sequence_col = true;
         }
         _init(columns, col_ids, num_key_columns);

be/src/olap/tablet_meta.cpp

@@ -164,6 +164,16 @@ TabletMeta::TabletMeta(int64_t table_id, int64_t partition_id, int64_t tablet_id
     schema->set_num_short_key_columns(tablet_schema.short_key_column_count);
     schema->set_num_rows_per_row_block(config::default_num_rows_per_column_file_block);
     schema->set_sequence_col_idx(tablet_schema.sequence_col_idx);
+    auto p_seq_map = schema->mutable_seq_map();  // ColumnGroupsPB
+
+    for (auto& it : tablet_schema.seq_map) {  // std::vector< ::doris::TColumnGroup>
+        uint32_t key = it.sequence_column;
+        ColumnGroupPB* cg_pb = p_seq_map->add_cg();  // ColumnGroupPB {key: {v1, v2, v3}}
+        cg_pb->set_sequence_column(key);
+        for (auto v : it.columns_in_group) {
+            cg_pb->add_columns_in_group(v);
+        }
+    }
     switch (tablet_schema.keys_type) {
     case TKeysType::DUP_KEYS:
         schema->set_keys_type(KeysType::DUP_KEYS);

be/src/olap/tablet_reader.cpp

@@ -336,6 +336,47 @@ Status TabletReader::_init_params(const ReaderParams& read_params) {
         }
     }
 
+    if (_tablet_schema->has_seq_map()) {
+        if (_sequence_col_idx != -1) {
+            auto msg = "sequence columns conflict, both seq_col and seq_map are true!";
+            LOG(WARNING) << msg;
+            return Status::InternalError(msg);
+        }
+        _val_to_seq = _tablet_schema->value_to_seq();
+        if (_val_to_seq.size() <= 0) {
+            auto msg = "seq_map is empty!";
+            LOG(WARNING) << msg;
+            return Status::InternalError(msg);
+        }
+    }
+
+    /*
+     * |** KEY **|        ** VALUE **     |
+        ------------------------------------
+        |** KEY **|  CDE is value| sequence|
+        |----|----|----|----|----|----|----|
+        A    B    C    D    E   S1   S2
+        0    1    2    3    4    5    6
+
+        for example SQL is select C,C,D,E from table;
+        _value_cids is {2,2,3,4}
+        _seq_map is {5:{2,3}, 6:{4}}
+        _value_to_seq is {2:5,3:5,5:5,4:6,6:6}
+        _return_seq_map is {5:{2,2,3}, 6:{4}}
+    */
+    _return_seq_map.clear();
+    for (const auto &val : _value_cids) {
+        auto seq = _val_to_seq[val];
+
+        if (_return_seq_map.find(seq) == _return_seq_map.end()) {
+            _return_seq_map[seq] = std::vector<uint32_t>();
+        }
+
+        if (val != seq) {
+            _return_seq_map[seq].push_back(val);
+        }
+    }
+
     return res;
 }
 

be/src/olap/tablet_reader.h

@@ -318,6 +318,9 @@ class TabletReader {
     std::vector<uint32_t> _key_cids;
     std::vector<uint32_t> _value_cids;
 
+    std::unordered_map<uint32_t /* seq cid */, std::vector<uint32_t> /* value cids */> _return_seq_map;
+    std::unordered_map<uint32_t /* cid */, uint32_t /* sequence */> _val_to_seq;
+
     uint64_t _merged_rows = 0;
     OlapReaderStatistics _stats;
 };

be/src/olap/tablet_schema.cpp

@@ -1193,6 +1193,41 @@ void TabletSchema::init_from_pb(const TabletSchemaPB& schema, bool ignore_extrac
     _storage_page_size = schema.storage_page_size();
     _storage_dict_page_size = schema.storage_dict_page_size();
     _schema_version = schema.schema_version();
+    auto column_groups_pb = schema.seq_map();
+    _seq_map.clear();
+    _value_to_seq.clear();
+    /*
+     * ColumnGroupsPB is a list of cg_pb, and
+     * ColumnGroupsPB do not have begin() or end() method.
+     * we must use for(i=0;i<xx;i++) loop
+     */
+    for (int i = 0; i < column_groups_pb.cg_size(); i++) {
+        ColumnGroupPB cg_pb = column_groups_pb.cg(i);
+        uint32_t key = cg_pb.sequence_column();
+        for (auto j : cg_pb.columns_in_group()) {
+            _seq_map[key].push_back(j);
+        }
+    }
+
+    if (_seq_map.size() > 0) {
+        /*
+            |** KEY **|        ** VALUE **     |
+            ------------------------------------
+            |** KEY **|  CDE is value| sequence|
+            |----|----|----|----|----|----|----|
+            A    B    C    D    E   S1   S2
+            0    1    2    3    4    5    6
+            for example: _seq_map is {5:{2,3}, 6:{4}}
+            then, _value_to_seq = {2:5,3:5,5:5,4:6,6:6}
+        */
+        for (auto it = _seq_map.cbegin(); it != _seq_map.cend(); it++) {
+            auto k = it->first;
+            for (auto v : it->second) {
+                _value_to_seq[v] = k;
+            }
+            _value_to_seq[k] = k;
+        }
+    }
     // Default to V1 inverted index storage format for backward compatibility if not specified in schema.
     if (!schema.has_inverted_index_storage_format()) {
         _inverted_index_storage_format = InvertedIndexStorageFormatPB::V1;
@@ -1456,6 +1491,15 @@ void TabletSchema::to_schema_pb(TabletSchemaPB* tablet_schema_pb) const {
     tablet_schema_pb->mutable_row_store_column_unique_ids()->Assign(
             _row_store_column_unique_ids.begin(), _row_store_column_unique_ids.end());
     tablet_schema_pb->set_enable_variant_flatten_nested(_enable_variant_flatten_nested);
+    auto column_groups_pb = tablet_schema_pb->mutable_seq_map();
+    for (const auto& it : _seq_map) {
+        uint32_t key = it.first;
+        ColumnGroupPB* cg_pb = column_groups_pb->add_cg();  // ColumnGroupPB {key: {v1, v2, v3}}
+        cg_pb->set_sequence_column(key);
+        for (auto v : it.second) {
+            cg_pb->add_columns_in_group(v);
+        }
+    }
 }
 
 size_t TabletSchema::row_size() const {

be/src/olap/tablet_schema.h

@@ -637,6 +637,9 @@ class TabletSchema : public MetadataAdder<TabletSchema> {
         }
         return 0;
     }
+    inline const std::unordered_map<uint32_t, std::vector<uint32_t>>& seq_map() const { return _seq_map; }
+    inline bool has_seq_map() const { return _seq_map.size() > 0;}
+    inline const std::unordered_map<uint32_t, uint32_t>& value_to_seq() const { return _value_to_seq; }
 
 private:
     friend bool operator==(const TabletSchema& a, const TabletSchema& b);
@@ -716,6 +719,8 @@ class TabletSchema : public MetadataAdder<TabletSchema> {
     // value: indexes
     using PatternToIndex = std::unordered_map<std::string, std::vector<TabletIndexPtr>>;
     std::unordered_map<int32_t, PatternToIndex> _index_by_unique_id_with_pattern;
+    std::unordered_map<uint32_t /* seq cid */, std::vector<uint32_t> /* value cids */> _seq_map;
+    std::unordered_map<uint32_t /* cid */, uint32_t /* sequence */> _value_to_seq;
 };
 
 bool operator==(const TabletSchema& a, const TabletSchema& b);