Doris 存储层次#

doris 首先有内存的memtable—>rowset->segment->column writer

rowset：rowset 是描述一次写入
segment：一个rowset可能由多个segment组成
tablet：tablet 是包括多个segment，一个segment包括多个rowset
column一行数据有多个column

导入代码#

整个流程第一步：

1
Status FlushToken::_do_flush_memtable(MemTable* memtable, int32_t segment_id, int64_t* flush_size) {
2
    VLOG_CRITICAL << "begin to flush memtable for tablet: " << memtable->tablet_id()
3
                  << ", memsize: " << PrettyPrinter::print_bytes(memtable->memory_usage())
4
                  << ", rows: " << memtable->stat().raw_rows;
5
    memtable->update_mem_type(MemType::FLUSH);
6
    int64_t duration_ns = 0;
7
    {
8
        ...
9
        RETURN_IF_ERROR(_rowset_writer->flush_memtable(block.get(), segment_id, flush_size));
10
        ...
11
    }

我们看看实现

1
Status BetaRowsetWriterV2::flush_memtable(vectorized::Block* block, int32_t segment_id,
2
                                          int64_t* flush_size) {
3
    if (block->rows() == 0) {
4
        return Status::OK();
5
    }
6

7
    {
8
        SCOPED_RAW_TIMER(&_segment_writer_ns);
9
        RETURN_IF_ERROR(_segment_creator.flush_single_block(block, segment_id, flush_size));
10
    }
11
    // delete bitmap and seg compaction are done on the destination BE.
12
    return Status::OK();
13
}

1
void Tablet::add_rowsets(const std::vector<RowsetSharedPtr>& to_add) {
2
    if (to_add.empty()) {
3
        return;
4
    }
5
    std::vector<RowsetMetaSharedPtr> rs_metas;
6
    rs_metas.reserve(to_add.size());
7
    for (auto& rs : to_add) {
8
        _rs_version_map.emplace(rs->version(), rs);
9
        _timestamped_version_tracker.add_version(rs->version());
10
        rs_metas.push_back(rs->rowset_meta());
11
    }
12
    _tablet_meta->modify_rs_metas(rs_metas, {});
13
}

1
Status BetaRowsetWriterV2::flush_memtable(vectorized::Block* block, int32_t segment_id,
2
                                          int64_t* flush_size) {
3
    if (block->rows() == 0) {
4
        return Status::OK();
5
    }
6

7
    {
8
        SCOPED_RAW_TIMER(&_segment_writer_ns);
9
        RETURN_IF_ERROR(_segment_creator.flush_single_block(block, segment_id, flush_size));
10
    }
11
    // delete bitmap and seg compaction are done on the destination BE.
12
    return Status::OK();
13
}

调用：

1
Status SegmentFlusher::flush_single_block(const vectorized::Block* block, int32_t segment_id,
2
                                          int64_t* flush_size) {
3
    if (block->rows() == 0) {
4
        return Status::OK();
5
    }
6
    vectorized::Block flush_block(*block);
7
    if (_context.write_type != DataWriteType::TYPE_COMPACTION &&
8
        _context.tablet_schema->num_variant_columns() > 0) {
9
        RETURN_IF_ERROR(_parse_variant_columns(flush_block));
10
    }
11
    bool no_compression = flush_block.bytes() <= config::segment_compression_threshold_kb * 1024;
12
    if (config::enable_vertical_segment_writer) {
13
        std::unique_ptr<segment_v2::VerticalSegmentWriter> writer;
14
        RETURN_IF_ERROR(_create_segment_writer(writer, segment_id, no_compression));
15
        RETURN_IF_ERROR_OR_CATCH_EXCEPTION(_add_rows(writer, &flush_block, 0, flush_block.rows()));
16
        RETURN_IF_ERROR(_flush_segment_writer(writer, flush_size));
17
    } else {
18
        std::unique_ptr<segment_v2::SegmentWriter> writer;
19
        RETURN_IF_ERROR(_create_segment_writer(writer, segment_id, no_compression));
20
        RETURN_IF_ERROR_OR_CATCH_EXCEPTION(_add_rows(writer, &flush_block, 0, flush_block.rows()));
21
        RETURN_IF_ERROR(_flush_segment_writer(writer, flush_size));
22
    }
23
    return Status::OK();
24
}

后面调用：

1
Status SegmentFlusher::_flush_segment_writer(
2
        std::unique_ptr<segment_v2::VerticalSegmentWriter>& writer, int64_t* flush_size) {
3
    MonotonicStopWatch total_timer;
4
    total_timer.start();
5

6
    uint32_t row_num = writer->num_rows_written();
7
    _num_rows_updated += writer->num_rows_updated();
8
    _num_rows_deleted += writer->num_rows_deleted();
9
    _num_rows_new_added += writer->num_rows_new_added();
10
    _num_rows_filtered += writer->num_rows_filtered();
11

12
    if (row_num == 0) {
13
        return Status::OK();
14
    }
15

16
    MonotonicStopWatch finalize_timer;
17
    finalize_timer.start();
18
    uint64_t segment_file_size;
19
    uint64_t common_index_size;
20
    Status s = writer->finalize(&segment_file_size, &common_index_size);
21
    finalize_timer.stop();
22

23
    if (!s.ok()) {
24
        return Status::Error(s.code(), "failed to finalize segment: {}", s.to_string());
25
    }
26

27
    MonotonicStopWatch inverted_index_timer;
28
    inverted_index_timer.start();
29
    int64_t inverted_index_file_size = 0;
30
    RETURN_IF_ERROR(writer->close_inverted_index(&inverted_index_file_size));
31
    inverted_index_timer.stop();
32

33
    VLOG_DEBUG << "tablet_id:" << _context.tablet_id
34
               << " flushing filename: " << writer->data_dir_path()
35
               << " rowset_id:" << _context.rowset_id;
36

37
    KeyBoundsPB key_bounds;
38
    Slice min_key = writer->min_encoded_key();
39
    Slice max_key = writer->max_encoded_key();
40
    DCHECK_LE(min_key.compare(max_key), 0);
41
    key_bounds.set_min_key(min_key.to_string());
42
    key_bounds.set_max_key(max_key.to_string());
43

44
    uint32_t segment_id = writer->segment_id();
45
    SegmentStatistics segstat;
46
    segstat.row_num = row_num;
47
    segstat.data_size = segment_file_size;
48
    segstat.index_size = inverted_index_file_size;
49
    segstat.key_bounds = key_bounds;
50

51
    writer.reset();
52

53
    MonotonicStopWatch collector_timer;
54
    collector_timer.start();
55
    RETURN_IF_ERROR(_context.segment_collector->add(segment_id, segstat));
56
    collector_timer.stop();
57

58
    total_timer.stop();
59

60
    LOG(INFO) << "tablet_id:" << _context.tablet_id
61
              << ", flushing rowset_dir: " << _context.tablet_path
62
              << ", rowset_id:" << _context.rowset_id
63
              << ", data size:" << PrettyPrinter::print_bytes(segstat.data_size)
64
              << ", index size:" << PrettyPrinter::print_bytes(segstat.index_size)
65
              << ", timing breakdown: total=" << total_timer.elapsed_time_milliseconds() << "ms"
66
              << ", finalize=" << finalize_timer.elapsed_time_milliseconds() << "ms"
67
              << ", inverted_index=" << inverted_index_timer.elapsed_time_milliseconds() << "ms"
68
              << ", collector=" << collector_timer.elapsed_time_milliseconds() << "ms";
69

70
    if (flush_size) {
71
        *flush_size = segment_file_size;
72
    }
73
    return Status::OK();
74
}

核心转换逻辑，将行转换成列：

1
std::pair<Status, IOlapColumnDataAccessor*> OlapBlockDataConvertor::convert_column_data(
2
        size_t cid) {
3
    assert(cid < _convertors.size());
4
    auto convert_func = [&]() -> Status {
5
        RETURN_IF_ERROR_OR_CATCH_EXCEPTION(_convertors[cid]->convert_to_olap());
6
        return Status::OK();
7
    };
8
    auto status = convert_func();
9
    return {status, _convertors[cid].get()};
10
}

1
Table: user_orders
2
├── Partition: p202401 (范围分区)
3
│   ├── Tablet 167 (Hash分桶)
4
│   │   ├── Rowset v1-v5 (基础版本)
5
│   │   │   ├── Segment 0 (256MB, 200万行)
6
│   │   │   │   ├── Page 0-4095: user_id 列
7
│   │   │   │   ├── Page 4096-8191: amount 列
8
│   │   │   │   └── Page 8192-12287: order_time 列
9
│   │   │   └── Segment 1 (256MB, 200万行)
10
│   │   ├── Rowset v6-v9 (增量版本)
11
│   │   └── Rowset v10 (最新版本)
12
│   ├── Tablet 168
13
│   └── ...
14
└── Partition: p202402

1
逻辑层：       表 (Table)
2
                │
3
                ▼
4
物理层：       Tablet (数据分片)
5
                │
6
                ▼
7
版本层：       Rowset (行集，版本管理单元)
8
                │
9
                ▼
10
文件层：       Segment (段，物理文件)
11
                │
12
                ▼
13
存储层：       Page (页，数据块)
14
                │
15
                ▼
16
数据层：       Column (列，存储格式)

1
// 表达式计算得到列数据
2

3
ColumnPtr VExpr::get_const_col(const Block& block, size_t row_idx) {
4
    // 计算这一行这个列的值
5
    // 例如：对于常量值 'Alice'，直接返回包含'Alice'的列
6
}
7

8
// 位置: doris/be/src/vec/columns/column.h
9
// 实际的列数据插入
10

11
template <typename T>
12
void ColumnVector<T>::insert_range_from(const IColumn& src, size_t start, size_t length) {
13
    const auto& src_vec = static_cast<const ColumnVector<T>&>(src);
14

15
    // 将源列的数据追加到当前列
16
    size_t old_size = data.size();
17
    data.resize(old_size + length);
18

19
    // 关键：这里发生了数据复制，从源列到目标列
20
    memcpy(&data[old_size], &src_vec.data[start], length * sizeof(T));
21
}

1
    void insert_data(const char* pos, size_t length) override {
2
        const size_t old_size = chars.size();
3
        const size_t new_size = old_size + length;
4

5
        if (length) {
6
            check_chars_length(new_size, offsets.size() + 1);
7
            chars.resize(new_size);
8
            memcpy(chars.data() + old_size, pos, length);
9
        }
10
        offsets.push_back(new_size);
11
        sanity_check_simple();
12
    }