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

In some business scenarios, business operations need to update different
columns in the same wide table through two or more data streams. For
example, one data stream will write in real time to update some fields
of this table; another data stream will perform imports on demand to
update other columns of this table. During the update process, both data
stream jobs need to ensure the order of replacement; and during queries,
data from all columns should be queryable.

```shell
CREATE TABLE `upsert_test` (
  `a` bigint(20) NULL COMMENT "",
  `b` int(11) NULL COMMENT "",
  `c` int(11) NULL COMMENT "",
  `d` int(11) NULL COMMENT "",
  `e` int(11) NULL COMMENT "",
  `s1` int(11) NULL COMMENT "",
  `s2` int(11) NULL COMMENT ""
) ENGINE=OLAP
UNIQUE KEY(`a`, `b`)
COMMENT "OLAP"
DISTRIBUTED BY HASH(`a`, `b`) BUCKETS 1
PROPERTIES (
"replication_num" = "1",
"sequence_mapping.s1" = "c,d",
"sequence_mapping.s2" = "e"
);

MySQL [test]> insert into upsert_test(a, b, c, d, s1) values (1,1,2,2,2);
Query OK, 1 row affected (0.15 sec)
{'label':'insert_fadb7c3fe6041c6-a51094f635017b12', 'status':'VISIBLE', 'txnId':'2'}

MySQL [test]> insert into upsert_test(a, b, c, d, s1) values (1,1,1,1,1);
Query OK, 1 row affected (0.07 sec)
{'label':'insert_1f3d1d5eb28447fe-889427c9da075c11', 'status':'VISIBLE', 'txnId':'3'}

MySQL [test]>  select * from upsert_test;
+------+------+------+------+------+------+------+
| a    | b    | c    | d    | e    | s1   | s2   |
+------+------+------+------+------+------+------+
|    1 |    1 |    2 |    2 | NULL |    2 | NULL |
+------+------+------+------+------+------+------+


MySQL [test]> insert into upsert_test(a, b, e, s2) values (1,1,2,2);
Query OK, 1 row affected (0.05 sec)
{'label':'insert_3b9614ce9a4e4dc3-97bbbb9b881c24aa', 'status':'VISIBLE', 'txnId':'4'}

MySQL [test]> select * from upsert_test;
+------+------+------+------+------+------+------+
| a    | b    | c    | d    | e    | s1   | s2   |
+------+------+------+------+------+------+------+
|    1 |    1 |    2 |    2 |    2 |    2 |    2 |
+------+------+------+------+------+------+------+
1 row in set (0.01 sec)

``` 

documentation: apache/doris-website#2954

Author: HonestManXin

be/src/cloud/pb_convert.cpp

@@ -384,6 +384,9 @@ void doris_tablet_schema_to_cloud(TabletSchemaCloudPB* out, const TabletSchemaPB
     if (in.has_binary_plain_encoding_default_impl()) {
         out->set_binary_plain_encoding_default_impl(in.binary_plain_encoding_default_impl());
     }
+    if (in.has_seq_map()) {
+        out->mutable_seq_map()->CopyFrom(in.seq_map());
+    }
 
     if (in.has___split_schema()) {
         out->mutable___split_schema()->CopyFrom(in.__split_schema());
@@ -428,6 +431,9 @@ void doris_tablet_schema_to_cloud(TabletSchemaCloudPB* out, TabletSchemaPB&& in)
     if (in.has_binary_plain_encoding_default_impl()) {
         out->set_binary_plain_encoding_default_impl(in.binary_plain_encoding_default_impl());
     }
+    if (in.has_seq_map()) {
+        out->mutable_seq_map()->CopyFrom(in.seq_map());
+    }
 
     if (in.has___split_schema()) {
         out->mutable___split_schema()->CopyFrom(in.__split_schema());
@@ -485,6 +491,9 @@ void cloud_tablet_schema_to_doris(TabletSchemaPB* out, const TabletSchemaCloudPB
     if (in.has_binary_plain_encoding_default_impl()) {
         out->set_binary_plain_encoding_default_impl(in.binary_plain_encoding_default_impl());
     }
+    if (in.has_seq_map()) {
+        out->mutable_seq_map()->CopyFrom(in.seq_map());
+    }
 
     if (in.has___split_schema()) {
         out->mutable___split_schema()->CopyFrom(in.__split_schema());
@@ -530,6 +539,9 @@ void cloud_tablet_schema_to_doris(TabletSchemaPB* out, TabletSchemaCloudPB&& in)
     if (in.has_binary_plain_encoding_default_impl()) {
         out->set_binary_plain_encoding_default_impl(in.binary_plain_encoding_default_impl());
     }
+    if (in.has_seq_map()) {
+        out->mutable_seq_map()->CopyFrom(in.seq_map());
+    }
 
     if (in.has___split_schema()) {
         out->mutable___split_schema()->CopyFrom(in.__split_schema());

be/src/olap/compaction.cpp

@@ -204,7 +204,8 @@ Status Compaction::merge_input_rowsets() {
     {
         SCOPED_TIMER(_merge_rowsets_latency_timer);
         // 1. Merge segment files and write bkd inverted index
-        if (_is_vertical) {
+        // TODO implement vertical compaction for seq map
+        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

@@ -249,6 +249,45 @@ 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) {
+    // for each mapping replace value columns according to the sequence column compare result
+    // for example: a b c d s1 s2  (key:a , s1=>[b,c], s2=>[d])
+    // src row:     1 4 5 6  8 9
+    // dst row:     1 2 3 4  7 10
+    // after aggregate
+    // dst row:     1 4 5 4  8 10  (b,c,s1 will be replaced, d,s2)
+    const auto& seq_map = _tablet_schema->seq_col_idx_to_value_cols_idx();
+    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);
+            // must use replace column instead of update row_pos
+            // because one row may have multi sequence column
+            // and agg function add method won't change the real column value
+            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) {
@@ -509,7 +548,13 @@ 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);
+                }
+
                 // Clean up aggregation state of the merged row to avoid memory leak
                 if (cur_row) {
                     _clear_row_agg(cur_row);

be/src/olap/memtable.h

@@ -213,6 +213,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

@@ -186,7 +186,9 @@ Status BetaRowsetReader::get_segment_iterators(RowsetReaderContext* read_context
     }
 
     if (_should_push_down_value_predicates()) {
-        if (_read_context->value_predicates != nullptr) {
+        // sequence mapping currently only support merge on read, so can not push down value predicates
+        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

@@ -80,7 +80,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

@@ -191,6 +191,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_schema.cpp

@@ -999,6 +999,35 @@ void TabletSchema::append_column(TabletColumn column, ColumnType col_type) {
     }
     _num_columns++;
     _num_virtual_columns = _vir_col_idx_to_unique_id.size();
+    // generate column index mapping for seq map
+    if (_seq_col_uid_to_value_cols_uid.contains(column.unique_id())) {
+        const auto seq_idx = _field_uniqueid_to_index[column.unique_id()];
+        if (!_seq_col_idx_to_value_cols_idx.contains(seq_idx)) {
+            _seq_col_idx_to_value_cols_idx[seq_idx] = {};
+        }
+    }
+    if (_value_col_uid_to_seq_col_uid.contains(column.unique_id())) {
+        const auto seq_uid = _value_col_uid_to_seq_col_uid[column.unique_id()];
+        if (_field_uniqueid_to_index.contains(seq_uid)) {
+            bool all_uid_index_found = true;
+            std::vector<int32_t> value_cols_index;
+            for (const auto value_col_uid : _seq_col_uid_to_value_cols_uid[seq_uid]) {
+                if (!_field_uniqueid_to_index.contains(value_col_uid)) {
+                    all_uid_index_found = false;
+                    break;
+                }
+                value_cols_index.push_back(_field_uniqueid_to_index[value_col_uid]);
+            }
+            if (all_uid_index_found) {
+                const auto seq_idx = _field_uniqueid_to_index[seq_uid];
+                for (const auto col_idx : value_cols_index) {
+                    _seq_col_idx_to_value_cols_idx[seq_idx].push_back(col_idx);
+                    _value_col_idx_to_seq_col_idx[col_idx] = seq_idx;
+                }
+                _value_col_idx_to_seq_col_idx[seq_idx] = seq_idx;
+            }
+        }
+    }
 }
 
 void TabletSchema::append_index(TabletIndex&& index) {
@@ -1060,6 +1089,8 @@ void TabletSchema::clear_columns() {
     _num_variant_columns = 0;
     _num_null_columns = 0;
     _num_key_columns = 0;
+    _seq_col_idx_to_value_cols_idx.clear();
+    _value_col_idx_to_seq_col_idx.clear();
     _cols.clear();
 }
 
@@ -1167,6 +1198,64 @@ 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();
+    if (schema.has_seq_map()) {
+        auto column_groups_pb = schema.seq_map();
+        _seq_col_uid_to_value_cols_uid.clear();
+        _value_col_uid_to_seq_col_uid.clear();
+        _seq_col_idx_to_value_cols_idx.clear();
+        _value_col_idx_to_seq_col_idx.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_uid = cg_pb.sequence_column();
+            auto found = _field_uniqueid_to_index.find(key_uid);
+            DCHECK(found != _field_uniqueid_to_index.end())
+                    << "could not find sequence col with unique id = " << key_uid
+                    << " table_id=" << _table_id;
+            int32_t seq_index = found->second;
+            _seq_col_uid_to_value_cols_uid[key_uid] = {};
+            _seq_col_idx_to_value_cols_idx[seq_index] = {};
+            for (auto val_uid : cg_pb.columns_in_group()) {
+                _seq_col_uid_to_value_cols_uid[key_uid].push_back(val_uid);
+                found = _field_uniqueid_to_index.find(val_uid);
+                DCHECK(found != _field_uniqueid_to_index.end())
+                        << "could not find value col with unique id = " << key_uid
+                        << " table_id=" << _table_id;
+                int32_t val_index = found->second;
+                _seq_col_idx_to_value_cols_idx[seq_index].push_back(val_index);
+            }
+        }
+
+        if (!_seq_col_uid_to_value_cols_uid.empty()) {
+            /*
+                |** 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& [seq_uid, cols_uid] : _seq_col_uid_to_value_cols_uid) {
+                for (auto col_uid : cols_uid) {
+                    _value_col_uid_to_seq_col_uid[col_uid] = seq_uid;
+                }
+                _value_col_uid_to_seq_col_uid[seq_uid] = seq_uid;
+            }
+
+            for (auto& [seq_idx, value_cols_idx] : _seq_col_idx_to_value_cols_idx) {
+                for (auto col_idx : value_cols_idx) {
+                    _value_col_idx_to_seq_col_idx[col_idx] = seq_idx;
+                }
+                _value_col_idx_to_seq_col_idx[seq_idx] = seq_idx;
+            }
+        }
+    }
     // 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;
@@ -1463,6 +1552,15 @@ void TabletSchema::to_schema_pb(TabletSchemaPB* tablet_schema_pb) const {
     tablet_schema_pb->set_integer_type_default_use_plain_encoding(
             _integer_type_default_use_plain_encoding);
     tablet_schema_pb->set_binary_plain_encoding_default_impl(_binary_plain_encoding_default_impl);
+    auto column_groups_pb = tablet_schema_pb->mutable_seq_map();
+    for (const auto& it : _seq_col_uid_to_value_cols_uid) {
+        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

@@ -678,6 +678,16 @@ class TabletSchema : public MetadataAdder<TabletSchema> {
         }
         return 0;
     }
+    const std::unordered_map<uint32_t, std::vector<uint32_t>>& seq_col_idx_to_value_cols_idx()
+            const {
+        return _seq_col_idx_to_value_cols_idx;
+    }
+
+    bool has_seq_map() const { return !_seq_col_idx_to_value_cols_idx.empty(); }
+
+    const std::unordered_map<uint32_t, uint32_t>& value_col_idx_to_seq_col_idx() const {
+        return _value_col_idx_to_seq_col_idx;
+    }
 
     void add_pruned_columns_data_type(int32_t col_unique_id, vectorized::DataTypePtr data_type) {
         _pruned_columns_data_type[col_unique_id] = std::move(data_type);
@@ -798,6 +808,14 @@ class TabletSchema : public MetadataAdder<TabletSchema> {
     bool _integer_type_default_use_plain_encoding {false};
     BinaryPlainEncodingTypePB _binary_plain_encoding_default_impl {
             BinaryPlainEncodingTypePB::BINARY_PLAIN_ENCODING_V1};
+    // Sequence column unique id mapping to value columns unique id
+    std::unordered_map<uint32_t, std::vector<uint32_t>> _seq_col_uid_to_value_cols_uid;
+    // Value column unique id mapping to sequence column unique id(also map sequence column it self)
+    std::unordered_map<uint32_t, uint32_t> _value_col_uid_to_seq_col_uid;
+    // Sequence column index mapping to value column index
+    std::unordered_map<uint32_t, std::vector<uint32_t>> _seq_col_idx_to_value_cols_idx;
+    // Value column index mapping to sequence column index(also map sequence column it self)
+    std::unordered_map<uint32_t, uint32_t> _value_col_idx_to_seq_col_idx;
 };
 
 bool operator==(const TabletSchema& a, const TabletSchema& b);

be/src/vec/exec/scan/olap_scanner.cpp

@@ -424,7 +424,7 @@ Status OlapScanner::_init_tablet_reader_params(
             _tablet_reader_params.return_columns.push_back(index);
         }
         // expand the sequence column
-        if (tablet_schema->has_sequence_col()) {
+        if (tablet_schema->has_sequence_col() || tablet_schema->has_seq_map()) {
             bool has_replace_col = false;
             for (auto col : _return_columns) {
                 if (tablet_schema->column(col).aggregation() ==
@@ -434,10 +434,32 @@ Status OlapScanner::_init_tablet_reader_params(
                 }
             }
             if (auto sequence_col_idx = tablet_schema->sequence_col_idx();
-                has_replace_col && std::find(_return_columns.begin(), _return_columns.end(),
-                                             sequence_col_idx) == _return_columns.end()) {
+                has_replace_col && tablet_schema->has_sequence_col() &&
+                std::find(_return_columns.begin(), _return_columns.end(), sequence_col_idx) ==
+                        _return_columns.end()) {
                 _tablet_reader_params.return_columns.push_back(sequence_col_idx);
             }
+            if (has_replace_col) {
+                const auto& val_to_seq = tablet_schema->value_col_idx_to_seq_col_idx();
+                std::set<uint32_t> return_seq_columns;
+
+                for (auto col : _tablet_reader_params.return_columns) {
+                    // we need to add the necessary sequence column in _return_columns, and
+                    // Avoid adding the same seq column twice
+                    const auto val_iter = val_to_seq.find(col);
+                    if (val_iter != val_to_seq.end()) {
+                        auto seq = val_iter->second;
+                        if (std::find(_tablet_reader_params.return_columns.begin(),
+                                      _tablet_reader_params.return_columns.end(),
+                                      seq) == _tablet_reader_params.return_columns.end()) {
+                            return_seq_columns.insert(seq);
+                        }
+                    }
+                }
+                _tablet_reader_params.return_columns.insert(
+                        std::end(_tablet_reader_params.return_columns),
+                        std::begin(return_seq_columns), std::end(return_seq_columns));
+            }
         }
     }