be/src/storage/segment/plain_page.h

Source
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#pragma once

#include "common/cast_set.h"
#include "storage/olap_common.h"
#include "storage/segment/options.h"
#include "storage/segment/page_builder.h"
#include "storage/segment/page_decoder.h"
#include "storage/types.h"
#include "util/coding.h"
#include "util/faststring.h"
#include "util/unaligned.h"

namespace doris {
namespace segment_v2 {

static const size_t PLAIN_PAGE_HEADER_SIZE = sizeof(uint32_t);

template <FieldType Type>
class PlainPageBuilder : public PageBuilderHelper<PlainPageBuilder<Type>> {
public:
    using Self = PlainPageBuilder<Type>;
    friend class PageBuilderHelper<Self>;

    Status init() override {
        // Reserve enough space for the page, plus a bit of slop since
        // we often overrun the page by a few values.
        return reset();
    }

    bool is_page_full() override { return _remain_element_capacity == 0; }

    Status add(const uint8_t* vals, size_t* count) override {
        if (is_page_full() || *count == 0) {
            *count = 0;
            return Status::OK();
        }
        size_t old_size = _buffer.size();
        size_t to_add = std::min(_remain_element_capacity, *count);
        // This may need a large memory, should return error if could not allocated
        // successfully, to avoid BE OOM.
        RETURN_IF_CATCH_EXCEPTION(_buffer.resize(old_size + to_add * SIZE_OF_TYPE));
        memcpy(&_buffer[old_size], vals, to_add * SIZE_OF_TYPE);
        _count += to_add;
        _raw_data_size += to_add * SIZE_OF_TYPE;

        *count = to_add;
        _remain_element_capacity -= to_add;
        return Status::OK();
    }

    Status finish(OwnedSlice* slice) override {
        encode_fixed32_le((uint8_t*)&_buffer[0], cast_set<uint32_t>(_count));
        RETURN_IF_CATCH_EXCEPTION({
            if (_count > 0) {
                _first_value.assign_copy(&_buffer[PLAIN_PAGE_HEADER_SIZE], SIZE_OF_TYPE);
                _last_value.assign_copy(
                        &_buffer[PLAIN_PAGE_HEADER_SIZE + (_count - 1) * SIZE_OF_TYPE],
                        SIZE_OF_TYPE);
            }
            *slice = _buffer.build();
        });
        return Status::OK();
    }

    Status reset() override {
        RETURN_IF_CATCH_EXCEPTION({
            _buffer.reserve(_options.data_page_size);
            _count = 0;
            _raw_data_size = 0;
            _buffer.clear();
            _buffer.resize(PLAIN_PAGE_HEADER_SIZE);
            _remain_element_capacity = _options.data_page_size / SIZE_OF_TYPE;
        });
        return Status::OK();
    }

    size_t count() const override { return _count; }

    uint64_t size() const override { return _buffer.size(); }

    uint64_t get_raw_data_size() const override { return _raw_data_size; }

    Status get_first_value(void* value) const override {
        if (_count == 0) {
            return Status::Error<ErrorCode::ENTRY_NOT_FOUND>("page is empty");
        }
        memcpy(value, _first_value.data(), SIZE_OF_TYPE);
        return Status::OK();
    }

    Status get_last_value(void* value) const override {
        if (_count == 0) {
            return Status::Error<ErrorCode::ENTRY_NOT_FOUND>("page is empty");
        }
        memcpy(value, _last_value.data(), SIZE_OF_TYPE);
        return Status::OK();
    }

private:
    PlainPageBuilder(const PageBuilderOptions& options) : _options(options) {}

    faststring _buffer;
    PageBuilderOptions _options;
    size_t _count;
    size_t _remain_element_capacity {0};
    uint64_t _raw_data_size = 0;
    typedef typename TypeTraits<Type>::CppType CppType;
    enum { SIZE_OF_TYPE = TypeTraits<Type>::size };
    faststring _first_value;
    faststring _last_value;
};

template <FieldType Type>
class PlainPageDecoder : public PageDecoder {
public:
    PlainPageDecoder(Slice data, const PageDecoderOptions& options)
            : _data(data), _options(options), _parsed(false), _num_elems(0), _cur_idx(0) {}

    Status init() override {
        CHECK(!_parsed);

        if (_data.size < PLAIN_PAGE_HEADER_SIZE) {
            return Status::InternalError(
                    "file corruption: not enough bytes for header in PlainPageDecoder ."
                    "invalid data size:{}, header size:{}",
                    _data.size, PLAIN_PAGE_HEADER_SIZE);
        }

        _num_elems = decode_fixed32_le((const uint8_t*)&_data[0]);

        if (_data.size != PLAIN_PAGE_HEADER_SIZE + _num_elems * SIZE_OF_TYPE) {
            return Status::InternalError("file corruption: unexpected data size.");
        }

        _parsed = true;

        RETURN_IF_ERROR(seek_to_position_in_page(0));
        return Status::OK();
    }

    Status seek_to_position_in_page(size_t pos) override {
        CHECK(_parsed) << "Must call init()";
        if (_num_elems == 0) [[unlikely]] {
            if (pos != 0) {
                return Status::Error<ErrorCode::INTERNAL_ERROR, false>(
                        "seek pos {} is larger than total elements  {}", pos, _num_elems);
            }
        }

        DCHECK_LE(pos, _num_elems);

        _cur_idx = cast_set<uint32_t>(pos);
        return Status::OK();
    }

    Status seek_at_or_after_value(const void* value, bool* exact_match) override {
        DCHECK(_parsed) << "Must call init() firstly";

        if (_num_elems == 0) {
            return Status::Error<ErrorCode::ENTRY_NOT_FOUND>("page is empty");
        }

        uint32_t left = 0;
        uint32_t right = _num_elems;

        const void* mid_value = nullptr;

        // find the first value >= target. after loop,
        // - left == index of first value >= target when found
        // - left == _num_elems when not found (all values < target)
        while (left < right) {
            uint32_t mid = left + (right - left) / 2;
            mid_value = &_data[PLAIN_PAGE_HEADER_SIZE + mid * SIZE_OF_TYPE];
            if (TypeTraits<Type>::cmp(mid_value, value) < 0) {
                left = mid + 1;
            } else {
                right = mid;
            }
        }
        if (left >= _num_elems) {
            return Status::Error<ErrorCode::ENTRY_NOT_FOUND>("all value small than the value");
        }
        const void* find_value = &_data[PLAIN_PAGE_HEADER_SIZE + left * SIZE_OF_TYPE];
        if (TypeTraits<Type>::cmp(find_value, value) == 0) {
            *exact_match = true;
        } else {
            *exact_match = false;
        }

        _cur_idx = left;
        return Status::OK();
    }

    Status next_batch(size_t* n, MutableColumnPtr& dst) override {
        DCHECK(_parsed);
        if (*n == 0 || _cur_idx >= _num_elems) [[unlikely]] {
            return Status::OK();
        }

        size_t max_fetch = std::min(*n, static_cast<size_t>(_num_elems - _cur_idx));
        const void* src_data = &_data[PLAIN_PAGE_HEADER_SIZE + _cur_idx * SIZE_OF_TYPE];

        dst->insert_many_fix_len_data((const char*)src_data, max_fetch);

        *n = max_fetch;
        _cur_idx += max_fetch;

        return Status::OK();
    }

    Status read_by_rowids(const rowid_t* rowids, ordinal_t page_first_ordinal, size_t* n,
                          MutableColumnPtr& dst) override {
        DCHECK(_parsed);
        if (*n == 0) [[unlikely]] {
            return Status::OK();
        }

        auto total = *n;
        auto read_count = 0;
        _buffer.resize(total);
        for (size_t i = 0; i < total; ++i) {
            ordinal_t ord = rowids[i] - page_first_ordinal;
            if (UNLIKELY(ord >= _num_elems)) {
                break;
            }

            _buffer[read_count++] =
                    unaligned_load<CppType>(&_data[PLAIN_PAGE_HEADER_SIZE + ord * SIZE_OF_TYPE]);
        }

        if (LIKELY(read_count > 0)) {
            dst->insert_many_fix_len_data((char*)_buffer.data(), read_count);
        }

        *n = read_count;
        return Status::OK();
    }

    size_t count() const override {
        DCHECK(_parsed);
        return _num_elems;
    }

    size_t current_index() const override {
        DCHECK(_parsed);
        return _cur_idx;
    }

private:
    Slice _data;
    PageDecoderOptions _options;
    bool _parsed;
    uint32_t _num_elems;
    uint32_t _cur_idx;
    typedef typename TypeTraits<Type>::CppType CppType;
    enum { SIZE_OF_TYPE = TypeTraits<Type>::size };

    std::vector<std::conditional_t<std::is_same_v<CppType, bool>, uint8_t, CppType>> _buffer;
};

} // namespace segment_v2
} // namespace doris

Coverage Report

Created: 2026-04-20 07:27