| /* |
| * 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. |
| */ |
| #ifndef WRITER_PAGE_WRITER_H |
| #define WRITER_PAGE_WRITER_H |
| |
| #include "common/allocator/byte_stream.h" |
| #include "common/allocator/my_string.h" |
| #include "common/statistic.h" |
| #include "compress/compressor.h" |
| #include "encoding/encoder.h" |
| #include "utils/db_utils.h" |
| |
| namespace storage { |
| |
| /* ================ PageData ================ */ |
| /* |
| * PageData consists of: |
| * 1. time_buf_size |
| * 2. time_buf |
| * 3. value_buf |
| * |
| * Since page header consists compressed_size_ and uncompressed_size_, |
| * so we can deduce the value_buf_size_. |
| * |
| * We abstract this struct because chunk writer will delay writer |
| * the first page data. So we have to save the first page data. |
| */ |
| struct PageData { |
| uint32_t time_buf_size_; |
| uint32_t value_buf_size_; |
| uint32_t uncompressed_size_; |
| uint32_t compressed_size_; |
| char* uncompressed_buf_; |
| char* compressed_buf_; |
| Compressor* compressor_; |
| |
| PageData() |
| : time_buf_size_(0), |
| value_buf_size_(0), |
| uncompressed_size_(0), |
| compressed_size_(0), |
| uncompressed_buf_(nullptr), |
| compressed_buf_(nullptr), |
| compressor_(nullptr) {} |
| int init(common::ByteStream& time_bs, common::ByteStream& value_bs, |
| Compressor* compressor); |
| void destroy() { |
| // Be careful about the memory |
| if (uncompressed_buf_ != nullptr) { |
| common::mem_free(uncompressed_buf_); |
| uncompressed_buf_ = nullptr; |
| } |
| if (compressed_buf_ != nullptr && compressor_ != nullptr) { |
| compressor_->after_compress(compressed_buf_); |
| compressed_buf_ = nullptr; |
| } |
| } |
| }; |
| |
| /* ================ PageWriter ================ */ |
| #define PW_DO_WRITE_FOR_TYPE(TSDATATYPE) \ |
| { \ |
| int ret = common::E_OK; \ |
| /* std::cout << "page_writer writer: time=" << timestamp << ", value=" \ |
| * << value << std::endl; */ \ |
| if (RET_FAIL(time_encoder_->encode(timestamp, time_out_stream_))) { \ |
| } else if (RET_FAIL( \ |
| value_encoder_->encode(value, value_out_stream_))) { \ |
| } else { \ |
| statistic_->update(timestamp, value); \ |
| } \ |
| return ret; \ |
| } |
| |
| class PageWriter { |
| public: |
| PageWriter() |
| : data_type_(common::VECTOR), |
| time_encoder_(nullptr), |
| value_encoder_(nullptr), |
| statistic_(nullptr), |
| time_out_stream_(OUT_STREAM_PAGE_SIZE, |
| common::MOD_PAGE_WRITER_OUTPUT_STREAM), |
| value_out_stream_(OUT_STREAM_PAGE_SIZE, |
| common::MOD_PAGE_WRITER_OUTPUT_STREAM), |
| cur_page_data_(), |
| compressor_(nullptr), |
| is_inited_(false) {} |
| int init(common::TSDataType data_type, common::TSEncoding encoding, |
| common::CompressionType compression); |
| ~PageWriter() { destroy(); } |
| // reset statistic_, time_out_stream_, value_out_stream_ |
| void reset(); |
| void destroy(); |
| |
| FORCE_INLINE int write(int64_t timestamp, bool value) { |
| if (UNLIKELY(data_type_ != common::BOOLEAN)) { |
| return common::E_TYPE_NOT_MATCH; |
| } |
| PW_DO_WRITE_FOR_TYPE(); |
| } |
| FORCE_INLINE int write(int64_t timestamp, int32_t value) { |
| if (UNLIKELY(data_type_ != common::INT32 && |
| data_type_ != common::DATE)) { |
| return common::E_TYPE_NOT_MATCH; |
| } |
| PW_DO_WRITE_FOR_TYPE(); |
| } |
| FORCE_INLINE int write(int64_t timestamp, int64_t value) { |
| if (UNLIKELY(data_type_ != common::INT64 && |
| data_type_ != common::TIMESTAMP)) { |
| return common::E_TYPE_NOT_MATCH; |
| } |
| PW_DO_WRITE_FOR_TYPE(); |
| } |
| FORCE_INLINE int write(int64_t timestamp, float value) { |
| if (UNLIKELY(data_type_ != common::FLOAT)) { |
| return common::E_TYPE_NOT_MATCH; |
| } |
| PW_DO_WRITE_FOR_TYPE(); |
| } |
| FORCE_INLINE int write(int64_t timestamp, double value) { |
| if (UNLIKELY(data_type_ != common::DOUBLE)) { |
| return common::E_TYPE_NOT_MATCH; |
| } |
| PW_DO_WRITE_FOR_TYPE(); |
| } |
| FORCE_INLINE int write(int64_t timestamp, common::String value) { |
| if (UNLIKELY(data_type_ != common::STRING && |
| data_type_ != common::TEXT && |
| data_type_ != common::BLOB)) { |
| return common::E_TYPE_NOT_MATCH; |
| } |
| PW_DO_WRITE_FOR_TYPE(); |
| } |
| |
| template <typename T> |
| FORCE_INLINE int write_batch(const int64_t* timestamps, const T* values, |
| uint32_t count) { |
| int ret = common::E_OK; |
| if (count == 0) return ret; |
| if (UNLIKELY(partial_failure_)) return common::E_DATA_INCONSISTENCY; |
| if (RET_FAIL(time_encoder_->encode_batch(timestamps, count, |
| time_out_stream_))) { |
| // Time stream wasn't advanced (encode_batch is atomic w.r.t. the |
| // stream cursor on failure for these encoders) — leave the page |
| // intact so the caller can retry. |
| } else if (RET_FAIL(value_encoder_->encode_batch(values, count, |
| value_out_stream_))) { |
| // Time stream already advanced; we can't roll it back here. |
| // Mark the page poisoned so write_to_chunk() refuses to seal a |
| // page where time and value rows are out of sync. |
| partial_failure_ = true; |
| } else { |
| statistic_->update_batch(timestamps, values, count); |
| } |
| return ret; |
| } |
| |
| // Batch write strings from Arrow-style offset+buffer layout. |
| FORCE_INLINE int write_string_batch(const int64_t* timestamps, |
| const char* buffer, |
| const uint32_t* offsets, |
| uint32_t start_idx, uint32_t count) { |
| int ret = common::E_OK; |
| if (count == 0) return ret; |
| if (UNLIKELY(partial_failure_)) return common::E_DATA_INCONSISTENCY; |
| if (RET_FAIL(time_encoder_->encode_batch(timestamps, count, |
| time_out_stream_))) { |
| } else if (RET_FAIL(value_encoder_->encode_string_batch( |
| buffer, offsets, start_idx, count, value_out_stream_))) { |
| partial_failure_ = true; |
| } else { |
| for (uint32_t i = 0; i < count; i++) { |
| uint32_t idx = start_idx + i; |
| uint32_t len = offsets[idx + 1] - offsets[idx]; |
| common::String val(buffer + offsets[idx], len); |
| statistic_->update(timestamps[i], val); |
| } |
| } |
| return ret; |
| } |
| |
| FORCE_INLINE bool has_partial_failure() const { return partial_failure_; } |
| |
| FORCE_INLINE uint32_t get_point_numer() const { return statistic_->count_; } |
| FORCE_INLINE uint32_t get_time_out_stream_size() const { |
| return time_out_stream_.total_size(); |
| } |
| // Logical bytes written — used by the page-seal-when-full heuristic. |
| // Memory-pressure accounting should use estimate_max_mem_size() below, |
| // which reflects the real 64 KiB-page footprint of the underlying |
| // ByteStreams. |
| FORCE_INLINE uint32_t get_page_memory_size() const { |
| return time_out_stream_.total_size() + value_out_stream_.total_size(); |
| } |
| /** |
| * calculate max possible memory size it occupies, including time |
| * outputStream and value outputStream, because size outputStream is never |
| * used until flushing. |
| * |
| * Reports the *allocated* stream footprint (sum of backing 64 KiB pages) |
| * rather than the logical bytes written. Sparse workloads with many |
| * measurements would otherwise look like they hold ~0 memory while |
| * actually pinning a full 64 KiB page per stream, so chunk-group memory |
| * thresholds couldn't keep peak memory under the configured cap. |
| * |
| * @return allocated size in time, value and outputStream |
| */ |
| FORCE_INLINE uint32_t estimate_max_mem_size() const { |
| return static_cast<uint32_t>(time_out_stream_.allocated_bytes() + |
| value_out_stream_.allocated_bytes()) + |
| time_encoder_->get_max_byte_size() + |
| value_encoder_->get_max_byte_size(); |
| } |
| int write_to_chunk(common::ByteStream& pages_data, bool write_header, |
| bool write_statistic, bool write_data_to_chunk_data); |
| FORCE_INLINE common::ByteStream& get_time_data() { |
| return time_out_stream_; |
| } |
| FORCE_INLINE common::ByteStream& get_value_data() { |
| return value_out_stream_; |
| } |
| FORCE_INLINE Statistic* get_statistic() { return statistic_; } |
| PageData get_cur_page_data() { return cur_page_data_; } |
| // See ValuePageWriter::release_cur_page_data for rationale. |
| void release_cur_page_data() { |
| cur_page_data_.uncompressed_buf_ = nullptr; |
| cur_page_data_.compressed_buf_ = nullptr; |
| } |
| void destroy_page_data() { cur_page_data_.destroy(); } |
| |
| private: |
| FORCE_INLINE int prepare_end_page() { |
| int ret = common::E_OK; |
| if (RET_FAIL(time_encoder_->flush(time_out_stream_))) { |
| } else if (RET_FAIL(value_encoder_->flush(value_out_stream_))) { |
| } |
| return ret; |
| } |
| int copy_page_data_to(common::ByteStream& my_page_data, |
| common::ByteStream& pages_data); |
| |
| private: |
| static const uint32_t OUT_STREAM_PAGE_SIZE = 1024; |
| |
| private: |
| common::TSDataType data_type_; |
| Encoder* time_encoder_; |
| Encoder* value_encoder_; |
| Statistic* statistic_; |
| common::ByteStream time_out_stream_{common::MOD_PAGE_WRITER_OUTPUT_STREAM}; |
| common::ByteStream value_out_stream_{common::MOD_PAGE_WRITER_OUTPUT_STREAM}; |
| PageData cur_page_data_; |
| Compressor* compressor_; |
| bool is_inited_; |
| // Set when write_batch advanced the time stream but value encoding |
| // failed. We can't unwind the partial time write, so refuse further |
| // writes and surface the poisoning to the higher layer via |
| // write_to_chunk(). |
| bool partial_failure_ = false; |
| }; |
| |
| } // end namespace storage |
| |
| #endif // WRITER_PAGE_WRITER_H |
