be/src/storage/segment/plain_page.h - doris - Git at Google

 // 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({ *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; }

 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 };
 };

 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
	// 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({ *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; }

	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 };
	};

	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