• DORIS 文件读取（研究中）

    前言

    该文章用于记录研究doris 1.1.1版本的后端文件读取流程。目前还只有部分流程。正在完善中。

    文章目录

    先说说Doris的数据存储结构


    1.tablet是真正存储在be的物理结构，有多个副本。参考[1]
    2.rowset是对于tablet下的数据集合进行的一次变更的版本，例如数据导入，删除和更新等。一个tablet下可能有多个rowset，doris会进行compaction对相邻版本的rowset进行合并，如果是group或者unique数据模型会进行相应的聚合操作。
    3.segment是rowset下面的数据分段，是以文件形式存在，里面包含有当前segment的列数据和相关的索引结构。
    4.column 是segment中列式存储的组织结构，包括data column和 index column。
    5.page是列数据的分块，每个列的具体数据存储在page当中。同样分为data page和iddex page。

        1. BE端OlapScanNode文件读取逻辑

      文件目录   doris/be/src/exec/olap_scan_node.cpp 

     开始读取文件的函数Status OlapScanNode::start_scan_thread(RuntimeState* state)

    1. Status OlapScanNode::start_scan_thread(RuntimeState* state) {
    2.     if (_scan_ranges.empty()) {
    3.         _transfer_done = true;
    4.         return Status::OK();
    5.     }
    6.  
    7.     // ranges constructed from scan keys
    8.     std::vector> cond_ranges;
    9.     RETURN_IF_ERROR(_scan_keys.get_key_range(&cond_ranges));
    10.     // if we can't get ranges from conditions, we give it a total range
    11.     if (cond_ranges.empty()) {
    12.         cond_ranges.emplace_back(new OlapScanRange());
    13.     }
    14.  
    15.     bool need_split = true;
    16.     // If we have ranges more than 64, there is no need to call
    17.     // ShowHint to split ranges
    18.     if (limit() != -1 || cond_ranges.size() > 64) {
    19.         need_split = false;
    20.     }
    21.  
    22.     int scanners_per_tablet = std::max(1, 64 / (int)_scan_ranges.size());
    23.     std::unordered_set disk_set;
    24.     for (auto& scan_range : _scan_ranges) {
    25.         auto tablet_id = scan_range->tablet_id;
    26.         int32_t schema_hash = strtoul(scan_range->schema_hash.c_str(), nullptr, 10);
    27.         std::string err;
    28.         TabletSharedPtr tablet = StorageEngine::instance()->tablet_manager()->get_tablet(
    29.                 tablet_id, schema_hash, true, &err);
    30.         if (tablet == nullptr) {
    31.             std::stringstream ss;
    32.             ss << "failed to get tablet: " << tablet_id << " with schema hash: " << schema_hash
    33.                << ", reason: " << err;
    34.             LOG(WARNING) << ss.str();
    35.             return Status::InternalError(ss.str());
    36.         }
    37.         std::vector>* ranges = &cond_ranges;
    38.         std::vector> split_ranges;
    39.         if (need_split && !tablet->all_beta()) {
    40.             auto st = get_hints(tablet, *scan_range, config::doris_scan_range_row_count,
    41.                                 _scan_keys.begin_include(), _scan_keys.end_include(), cond_ranges,
    42.                                 &split_ranges, _runtime_profile.get());
    43.             if (st.ok()) {
    44.                 ranges = &split_ranges;
    45.             }
    46.         }
    47.         // In order to avoid the problem of too many scanners caused by small tablets,
    48.         // in addition to scanRange, we also need to consider the size of the tablet when
    49.         // creating the scanner. One scanner is used for every 1Gb, and the final scanner_per_tablet
    50.         // takes the minimum value calculated by scanrange and size.
    51.         int size_based_scanners_per_tablet = 1;
    52.         if (config::doris_scan_range_max_mb > 0) {
    53.             size_based_scanners_per_tablet = std::max(
    54.                     1, (int)tablet->tablet_footprint() / config::doris_scan_range_max_mb << 20);
    55.         }
    56.         int ranges_per_scanner =
    57.                 std::max(1, (int)ranges->size() /
    58.                                     std::min(scanners_per_tablet, size_based_scanners_per_tablet));
    59.         int num_ranges = ranges->size();
    60.         for (int i = 0; i < num_ranges;) {
    61.             std::vector scanner_ranges;
    62.             scanner_ranges.push_back((*ranges)[i].get());
    63.             ++i;
    64.             for (int j = 1; i < num_ranges && j < ranges_per_scanner &&
    65.                             (*ranges)[i]->end_include == (*ranges)[i - 1]->end_include;
    66.                  ++j, ++i) {
    67.                 scanner_ranges.push_back((*ranges)[i].get());
    68.             }
    69.             OlapScanner* scanner = new OlapScanner(state, this, _olap_scan_node.is_preaggregation,
    70.                                                    _need_agg_finalize, *scan_range);
    71.             scanner->set_batch_size(_batch_size);
    72.             // add scanner to pool before doing prepare.
    73.             // so that scanner can be automatically deconstructed if prepare failed.
    74.             _scanner_pool.add(scanner);
    75.             RETURN_IF_ERROR(scanner->prepare(*scan_range, scanner_ranges, _olap_filter,
    76.                                              _bloom_filters_push_down));
    77.  
    78.             _olap_scanners.push_back(scanner);
    79.             disk_set.insert(scanner->scan_disk());
    80.         }
    81.     }
    82.     COUNTER_SET(_num_disks_accessed_counter, static_cast<int64_t>(disk_set.size()));
    83.     COUNTER_SET(_num_scanners, static_cast<int64_t>(_olap_scanners.size()));
    84.  
    85.     // PAIN_LOG(_olap_scanners.size());
    86.     // init progress
    87.     std::stringstream ss;
    88.     ss << "ScanThread complete (node=" << id() << "):";
    89.     _progress = ProgressUpdater(ss.str(), _olap_scanners.size(), 1);
    90.  
    91.     _transfer_thread = std::make_shared(&OlapScanNode::transfer_thread, this, state);
    92.  
    93.     return Status::OK();
    94. }

    其中以下部分先读取tablet元数据信息，有些tablet的元数据信息存放在Rocksdb中

    1.  TabletSharedPtr tablet = StorageEngine::instance()->tablet_manager()->get_tablet(
    2.                 tablet_id, schema_hash, true, &err);

    再然后创建真正的数据scanner

    1. OlapScanner* scanner = new OlapScanner(state, this, _olap_scan_node.is_preaggregation,
    2.                                                    _need_agg_finalize, *scan_range);

    其中OlapScanner在的prepare函数创建Rowset

    1. Status OlapScanner::prepare(
    2.         const TPaloScanRange& scan_range, const std::vector& key_ranges,
    3.         const std::vector& filters,
    4.         const std::vector>>&
    5.                 bloom_filters)

    其中prepare函数中的下面这么代码会真正的创建Rowset

    1. OLAPStatus acquire_reader_st =
    2.                     _tablet->capture_rs_readers(rd_version, &_tablet_reader_params.rs_readers, _mem_tracker);

    再然后是tablet类中的函数调用

    1. OLAPStatus Tablet::capture_rs_readers(const std::vector& version_path,
    2.                                       std::vector* rs_readers,
    3.                                       std::shared_ptr parent_tracker)

     2. in-memory属性在BE端产生的效果

    该部分用于说明DORIS BE端是如何对in-memory属性进行数据缓存的，原理就是BE端在读取列时，使用LRU缓存。读取的顺序从以下文件开始：

    doris/be/src/olap/rowset/segment_v2/segment_iterator.cpp

    1. Status SegmentIterator::_read_columns(const std::vector& column_ids, RowBlockV2* block,
    2.                                       size_t row_offset, size_t nrows) {
    3.     for (auto cid : column_ids) {
    4.         auto column_block = block->column_block(cid);
    5.         ColumnBlockView dst(&column_block, row_offset);
    6.         size_t rows_read = nrows;
    7.         RETURN_IF_ERROR(_column_iterators[cid]->next_batch(&rows_read, &dst));
    8.         DCHECK_EQ(nrows, rows_read);
    9.     }
    10.     return Status::OK();
    11. }

    读取page数据

    doris/be/src/olap/rowset/segment_v2/column_reader.cpp

    1. Status ColumnReader::read_page(const ColumnIteratorOptions& iter_opts, const PagePointer& pp,
    2.                                PageHandle* handle, Slice* page_body, PageFooterPB* footer,
    3.                                BlockCompressionCodec* codec) {
    4.     iter_opts.sanity_check();
    5.     PageReadOptions opts;
    6.     opts.rblock = iter_opts.rblock;
    7.     opts.page_pointer = pp;
    8.     opts.codec = codec;
    9.     opts.stats = iter_opts.stats;
    10.     opts.verify_checksum = _opts.verify_checksum;
    11.     opts.use_page_cache = iter_opts.use_page_cache;
    12.     opts.kept_in_memory = _opts.kept_in_memory;
    13.     opts.type = iter_opts.type;
    14.     opts.encoding_info = _encoding_info;
    15.  
    16.     return PageIO::read_and_decompress_page(opts, handle, page_body, footer);
    17. }

    其中决定是否启用in-memory从内存读取的关键是，其中的

    if (opts.use_page_cache && cache->is_cache_available(opts.type))，即启用in-memory属性且Cache有空间可用，相关的page在LRU缓存中没被弹出
    1. Status PageIO::read_and_decompress_page(const PageReadOptions& opts, PageHandle* handle,
    2.                                         Slice* body, PageFooterPB* footer) {
    3.     opts.sanity_check();
    4.     opts.stats->total_pages_num++;
    5.  
    6.     auto cache = StoragePageCache::instance();
    7.     PageCacheHandle cache_handle;
    8.     StoragePageCache::CacheKey cache_key(opts.rblock->path_desc().filepath, opts.page_pointer.offset);
    9.     if (opts.use_page_cache && cache->is_cache_available(opts.type) && cache->lookup(cache_key, &cache_handle, opts.type)) {
    10.         // we find page in cache, use it
    11.         *handle = PageHandle(std::move(cache_handle));
    12.         opts.stats->cached_pages_num++;
    13.         // parse body and footer
    14.         Slice page_slice = handle->data();
    15.         uint32_t footer_size = decode_fixed32_le((uint8_t*)page_slice.data + page_slice.size - 4);
    16.         std::string footer_buf(page_slice.data + page_slice.size - 4 - footer_size, footer_size);
    17.         if (!footer->ParseFromString(footer_buf)) {
    18.             return Status::Corruption("Bad page: invalid footer");
    19.         }
    20.         *body = Slice(page_slice.data, page_slice.size - 4 - footer_size);
    21.         return Status::OK();
    22.     }
    23.  
    24.     // every page contains 4 bytes footer length and 4 bytes checksum
    25.     const uint32_t page_size = opts.page_pointer.size;
    26.     if (page_size < 8) {
    27.         return Status::Corruption(strings::Substitute("Bad page: too small size ($0)", page_size));
    28.     }
    29.  
    30.     // hold compressed page at first, reset to decompressed page later
    31.     std::unique_ptr<char[]> page(new char[page_size]);
    32.     Slice page_slice(page.get(), page_size);
    33.     {
    34.         SCOPED_RAW_TIMER(&opts.stats->io_ns);
    35.         RETURN_IF_ERROR(opts.rblock->read(opts.page_pointer.offset, page_slice));
    36.         opts.stats->compressed_bytes_read += page_size;
    37.     }
    38.  
    39.     if (opts.verify_checksum) {
    40.         uint32_t expect = decode_fixed32_le((uint8_t*)page_slice.data + page_slice.size - 4);
    41.         uint32_t actual = crc32c::Value(page_slice.data, page_slice.size - 4);
    42.         if (expect != actual) {
    43.             return Status::Corruption(strings::Substitute(
    44.                     "Bad page: checksum mismatch (actual=$0 vs expect=$1)", actual, expect));
    45.         }
    46.     }
    47.  
    48.     // remove checksum suffix
    49.     page_slice.size -= 4;
    50.     // parse and set footer
    51.     uint32_t footer_size = decode_fixed32_le((uint8_t*)page_slice.data + page_slice.size - 4);
    52.     if (!footer->ParseFromArray(page_slice.data + page_slice.size - 4 - footer_size, footer_size)) {
    53.         return Status::Corruption("Bad page: invalid footer");
    54.     }
    55.  
    56.     uint32_t body_size = page_slice.size - 4 - footer_size;
    57.     if (body_size != footer->uncompressed_size()) { // need decompress body
    58.         if (opts.codec == nullptr) {
    59.             return Status::Corruption("Bad page: page is compressed but codec is NO_COMPRESSION");
    60.         }
    61.         SCOPED_RAW_TIMER(&opts.stats->decompress_ns);
    62.         std::unique_ptr<char[]> decompressed_page(
    63.                 new char[footer->uncompressed_size() + footer_size + 4]);
    64.  
    65.         // decompress page body
    66.         Slice compressed_body(page_slice.data, body_size);
    67.         Slice decompressed_body(decompressed_page.get(), footer->uncompressed_size());
    68.         RETURN_IF_ERROR(opts.codec->decompress(compressed_body, &decompressed_body));
    69.         if (decompressed_body.size != footer->uncompressed_size()) {
    70.             return Status::Corruption(strings::Substitute(
    71.                     "Bad page: record uncompressed size=$0 vs real decompressed size=$1",
    72.                     footer->uncompressed_size(), decompressed_body.size));
    73.         }
    74.         // append footer and footer size
    75.         memcpy(decompressed_body.data + decompressed_body.size, page_slice.data + body_size,
    76.                footer_size + 4);
    77.         // free memory of compressed page
    78.         page = std::move(decompressed_page);
    79.         page_slice = Slice(page.get(), footer->uncompressed_size() + footer_size + 4);
    80.         opts.stats->uncompressed_bytes_read += page_slice.size;
    81.     } else {
    82.         opts.stats->uncompressed_bytes_read += body_size;
    83.     }
    84.  
    85.     if (opts.encoding_info) {
    86.         auto* pre_decoder = opts.encoding_info->get_data_page_pre_decoder();
    87.         if (pre_decoder) {
    88.             RETURN_IF_ERROR(pre_decoder->decode(
    89.                     &page, &page_slice,
    90.                     footer->data_page_footer().nullmap_size() + footer_size + 4));
    91.         }
    92.     }
    93.  
    94.     *body = Slice(page_slice.data, page_slice.size - 4 - footer_size);
    95.     if (opts.use_page_cache && cache->is_cache_available(opts.type)) {
    96.         // insert this page into cache and return the cache handle
    97.         cache->insert(cache_key, page_slice, &cache_handle, opts.type, opts.kept_in_memory);
    98.         *handle = PageHandle(std::move(cache_handle));
    99.     } else {
    100.         *handle = PageHandle(page_slice);
    101.     }
    102.     page.release(); // memory now managed by handle
    103.     return Status::OK();
    104. }

    3 rowset与segment

    SegmentLoader ->  BetaRowsetSharedPtr = std::shared_ptr 
    1. OLAPStatus BetaRowsetReader::init(RowsetReaderContext* read_context) {
    2.  
    3.  // load segments
    4.     RETURN_NOT_OK(SegmentLoader::instance()->load_segments(
    5.             _rowset, &_segment_cache_handle, read_context->reader_type == ReaderType::READER_QUERY));
    6.  
    7. }

    Betarowset -> create_reader(){

    BetaRowsetReader

    }

    参考文档：

    1 https://blog.csdn.net/weixin_44012322/article/details/121163144

    2 https://www.jianshu.com/p/141ad958832d

    3 新一代列式存储格式Parquet_教练_我要踢球的博客-CSDN博客_parquet

    4 【Doris全面解析】Doris SQL 原理解析_ApacheDoris的博客-CSDN博客

    5 doris-查询原理_longlovefilm的博客-CSDN博客

    6 【Doris】Doris存储层设计介绍1——存储结构设计解析_九层之台起于累土的博客-CSDN博客_doris存储

    7 Doris Stream Load原理解析 - 墨天轮

  • 相关阅读:
    第二次提交PR啦
    layui视频上传，文件上传前条件验证，视频大小，视频时长判断
    填矩阵 码蹄集
    计算机网络-速记
    算法与数据结构 --- 遍历二叉树和线索二叉树
    springboot整合百度富文本编辑器ueditor实现图片上传和文件上传以及回显
    uniapp开发短视频系统仿哔哩哔哩
    BC1.2 PD协议
    java对象四大引用案例
    ES6空值合并运算符（??）
  • 原文地址：https://blog.csdn.net/innersense/article/details/126824446