cpp/src/encoding/rlbe_encoder.h - tsfile - 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.
  */

 #ifndef ENCODING_RLBE_ENCODER_H
 #define ENCODING_RLBE_ENCODER_H

 #include <stdint.h>

 #include "bit_packing_utils.h"
 #include "common/allocator/byte_stream.h"
 #include "common/db_common.h"
 #include "encode_utils.h"
 #include "encoder.h"
 #include "utils/errno_define.h"
 #include "utils/util_define.h"

 namespace storage {

 static constexpr int RLBE_BLOCK_DEFAULT_SIZE = 10000;

 template <typename T>
 struct RLBETraits;

 template <>
 struct RLBETraits<int32_t> {
     using Unsigned = uint32_t;
     static constexpr int VALUE_BITS = 32;
     static constexpr int LENGTH_CODE_BITS = 6;
 };

 template <>
 struct RLBETraits<int64_t> {
     using Unsigned = uint64_t;
     static constexpr int VALUE_BITS = 64;
     static constexpr int LENGTH_CODE_BITS = 7;
 };

 template <typename T>
 class RLBEEncoder : public Encoder {
    public:
     using Traits = RLBETraits<T>;
     using Unsigned = typename Traits::Unsigned;

     RLBEEncoder() { reset(); }
     ~RLBEEncoder() override = default;

     void destroy() override {}

     void reset() override {
         type_ = common::RLBE;
         write_index_ = -1;
         previous_value_ = 0;
         byte_writer_.reset();
         for (int i = 0; i <= RLBE_BLOCK_DEFAULT_SIZE; ++i) {
             diff_value_[i] = 0;
             length_code_[i] = 0;
             length_rle_[i] = 0;
         }
     }

     int flush(common::ByteStream& out) override { return flush_block(out); }

     int get_max_byte_size() override {
         return 5 * 4 * RLBE_BLOCK_DEFAULT_SIZE *
                (Traits::VALUE_BITS == 64 ? 2 : 1);
     }

     int encode(bool, common::ByteStream&) override {
         return common::E_TYPE_NOT_MATCH;
     }
     int encode(int32_t, common::ByteStream&) override {
         return common::E_TYPE_NOT_MATCH;
     }
     int encode(int64_t, common::ByteStream&) override {
         return common::E_TYPE_NOT_MATCH;
     }
     int encode(float, common::ByteStream&) override {
         return common::E_TYPE_NOT_MATCH;
     }
     int encode(double, common::ByteStream&) override {
         return common::E_TYPE_NOT_MATCH;
     }
     int encode(common::String, common::ByteStream&) override {
         return common::E_TYPE_NOT_MATCH;
     }

    protected:
     int encode_value(T value, common::ByteStream& out) {
         if (write_index_ == -1) {
             diff_value_[++write_index_] = value;
             length_code_[write_index_] = binary_length(value);
             previous_value_ = value;
             return common::E_OK;
         }

         diff_value_[++write_index_] =
             static_cast<T>(static_cast<Unsigned>(value) -
                            static_cast<Unsigned>(previous_value_));
         length_code_[write_index_] = binary_length(diff_value_[write_index_]);
         previous_value_ = value;
         if (write_index_ == RLBE_BLOCK_DEFAULT_SIZE - 1) {
             return flush(out);
         }
         return common::E_OK;
     }

    private:
     int binary_length(T value) const {
         if (value == 0) {
             return 1;
         }
         return Traits::VALUE_BITS - number_of_leading_zeros(value);
     }

     Unsigned java_rlbe_delta_bits(T value, int bit_width) const {
         Unsigned bits = static_cast<Unsigned>(value);
         if (bit_width == Traits::VALUE_BITS) {
             bits &= ~(static_cast<Unsigned>(1) << (Traits::VALUE_BITS - 1));
         }
         return bits;
     }

     uint64_t calc_fibonacci(uint64_t value) const {
         uint64_t fib[RLBE_BLOCK_DEFAULT_SIZE * 2 + 1] = {0};
         fib[0] = 1;
         fib[1] = 1;
         int i = 0;
         for (i = 2; fib[i - 1] <= value; ++i) {
             fib[i] = fib[i - 1] + fib[i - 2];
         }
         --i;

         uint64_t encoded = 0;
         while (value > 0) {
             while (fib[i] > value && i >= 1) {
                 --i;
             }
             encoded |= (1ULL << (i - 1));
             value -= fib[i];
         }
         return encoded;
     }

     void rle_on_length_code() {
         int i = 0;
         while (i <= write_index_) {
             int j = i;
             int repeat = 0;
             while (j <= write_index_ && length_code_[j] == length_code_[i]) {
                 ++j;
                 ++repeat;
             }
             length_rle_[i] = repeat;
             i = j;
         }
     }

     int flush_block(common::ByteStream& out) {
         int ret = common::E_OK;
         if (write_index_ == -1) {
             return common::E_OK;
         }
         if (RET_FAIL(byte_writer_.write_bits(
                 static_cast<uint32_t>(write_index_ + 1), 32, out))) {
             return ret;
         }
         rle_on_length_code();
         for (int i = 0; i <= write_index_; ++i) {
             if (length_rle_[i] > 0 && RET_FAIL(flush_segment(i, out))) {
                 return ret;
             }
         }
         if (RET_FAIL(byte_writer_.flush_partial(out))) {
             return ret;
         }
         reset();
         return common::E_OK;
     }

     int flush_segment(int first, common::ByteStream& out) {
         int ret = common::E_OK;
         if (RET_FAIL(byte_writer_.write_bits(
                 static_cast<uint64_t>(length_code_[first]),
                 Traits::LENGTH_CODE_BITS, out))) {
             return ret;
         }

         uint64_t fib = calc_fibonacci(length_rle_[first]);
         int fib_len = binary_length(static_cast<T>(fib));
         for (int i = 0; i < fib_len; ++i) {
             if (RET_FAIL(
                     byte_writer_.write_bit(((fib >> i) & 1ULL) != 0, out))) {
                 return ret;
             }
         }
         if (RET_FAIL(byte_writer_.write_bit(true, out))) {
             return ret;
         }

         int j = first;
         do {
             int diff_len = binary_length(diff_value_[j]);
             if (RET_FAIL(byte_writer_.write_bits(
                     java_rlbe_delta_bits(diff_value_[j], diff_len), diff_len,
                     out))) {
                 return ret;
             }
             ++j;
         } while (j <= write_index_ && length_rle_[j] == 0);
         return common::E_OK;
     }

    protected:
     common::TSEncoding type_;
     T diff_value_[RLBE_BLOCK_DEFAULT_SIZE + 1];
     int length_code_[RLBE_BLOCK_DEFAULT_SIZE + 1];
     int length_rle_[RLBE_BLOCK_DEFAULT_SIZE + 1];
     T previous_value_;
     int write_index_;
     MsbBitWriter byte_writer_;
 };

 class IntRLBEEncoder : public RLBEEncoder<int32_t> {
    public:
     int encode(int32_t value, common::ByteStream& out) override {
         return encode_value(value, out);
     }
 };

 class LongRLBEEncoder : public RLBEEncoder<int64_t> {
    public:
     int encode(int64_t value, common::ByteStream& out) override {
         return encode_value(value, out);
     }
 };

 class FloatRLBEEncoder : public RLBEEncoder<int32_t> {
    public:
     int encode(float value, common::ByteStream& out) override {
         return encode_value(common::float_to_int(value), out);
     }
 };

 class DoubleRLBEEncoder : public RLBEEncoder<int64_t> {
    public:
     int encode(double value, common::ByteStream& out) override {
         return encode_value(common::double_to_long(value), out);
     }
 };

 }  // namespace storage

 #endif  // ENCODING_RLBE_ENCODER_H
	/*
	* 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 ENCODING_RLBE_ENCODER_H
	#define ENCODING_RLBE_ENCODER_H

	#include <stdint.h>

	#include "bit_packing_utils.h"
	#include "common/allocator/byte_stream.h"
	#include "common/db_common.h"
	#include "encode_utils.h"
	#include "encoder.h"
	#include "utils/errno_define.h"
	#include "utils/util_define.h"

	namespace storage {

	static constexpr int RLBE_BLOCK_DEFAULT_SIZE = 10000;

	template <typename T>
	struct RLBETraits;

	template <>
	struct RLBETraits<int32_t> {
	using Unsigned = uint32_t;
	static constexpr int VALUE_BITS = 32;
	static constexpr int LENGTH_CODE_BITS = 6;
	};

	template <>
	struct RLBETraits<int64_t> {
	using Unsigned = uint64_t;
	static constexpr int VALUE_BITS = 64;
	static constexpr int LENGTH_CODE_BITS = 7;
	};

	template <typename T>
	class RLBEEncoder : public Encoder {
	public:
	using Traits = RLBETraits<T>;
	using Unsigned = typename Traits::Unsigned;

	RLBEEncoder() { reset(); }
	~RLBEEncoder() override = default;

	void destroy() override {}

	void reset() override {
	type_ = common::RLBE;
	write_index_ = -1;
	previous_value_ = 0;
	byte_writer_.reset();
	for (int i = 0; i <= RLBE_BLOCK_DEFAULT_SIZE; ++i) {
	diff_value_[i] = 0;
	length_code_[i] = 0;
	length_rle_[i] = 0;
	}
	}

	int flush(common::ByteStream& out) override { return flush_block(out); }

	int get_max_byte_size() override {
	return 5 * 4 * RLBE_BLOCK_DEFAULT_SIZE *
	(Traits::VALUE_BITS == 64 ? 2 : 1);
	}

	int encode(bool, common::ByteStream&) override {
	return common::E_TYPE_NOT_MATCH;
	}
	int encode(int32_t, common::ByteStream&) override {
	return common::E_TYPE_NOT_MATCH;
	}
	int encode(int64_t, common::ByteStream&) override {
	return common::E_TYPE_NOT_MATCH;
	}
	int encode(float, common::ByteStream&) override {
	return common::E_TYPE_NOT_MATCH;
	}
	int encode(double, common::ByteStream&) override {
	return common::E_TYPE_NOT_MATCH;
	}
	int encode(common::String, common::ByteStream&) override {
	return common::E_TYPE_NOT_MATCH;
	}

	protected:
	int encode_value(T value, common::ByteStream& out) {
	if (write_index_ == -1) {
	diff_value_[++write_index_] = value;
	length_code_[write_index_] = binary_length(value);
	previous_value_ = value;
	return common::E_OK;
	}

	diff_value_[++write_index_] =
	static_cast<T>(static_cast<Unsigned>(value) -
	static_cast<Unsigned>(previous_value_));
	length_code_[write_index_] = binary_length(diff_value_[write_index_]);
	previous_value_ = value;
	if (write_index_ == RLBE_BLOCK_DEFAULT_SIZE - 1) {
	return flush(out);
	}
	return common::E_OK;
	}

	private:
	int binary_length(T value) const {
	if (value == 0) {
	return 1;
	}
	return Traits::VALUE_BITS - number_of_leading_zeros(value);
	}

	Unsigned java_rlbe_delta_bits(T value, int bit_width) const {
	Unsigned bits = static_cast<Unsigned>(value);
	if (bit_width == Traits::VALUE_BITS) {
	bits &= ~(static_cast<Unsigned>(1) << (Traits::VALUE_BITS - 1));
	}
	return bits;
	}

	uint64_t calc_fibonacci(uint64_t value) const {
	uint64_t fib[RLBE_BLOCK_DEFAULT_SIZE * 2 + 1] = {0};
	fib[0] = 1;
	fib[1] = 1;
	int i = 0;
	for (i = 2; fib[i - 1] <= value; ++i) {
	fib[i] = fib[i - 1] + fib[i - 2];
	}
	--i;

	uint64_t encoded = 0;
	while (value > 0) {
	while (fib[i] > value && i >= 1) {
	--i;
	}
	encoded \|= (1ULL << (i - 1));
	value -= fib[i];
	}
	return encoded;
	}

	void rle_on_length_code() {
	int i = 0;
	while (i <= write_index_) {
	int j = i;
	int repeat = 0;
	while (j <= write_index_ && length_code_[j] == length_code_[i]) {
	++j;
	++repeat;
	}
	length_rle_[i] = repeat;
	i = j;
	}
	}

	int flush_block(common::ByteStream& out) {
	int ret = common::E_OK;
	if (write_index_ == -1) {
	return common::E_OK;
	}
	if (RET_FAIL(byte_writer_.write_bits(
	static_cast<uint32_t>(write_index_ + 1), 32, out))) {
	return ret;
	}
	rle_on_length_code();
	for (int i = 0; i <= write_index_; ++i) {
	if (length_rle_[i] > 0 && RET_FAIL(flush_segment(i, out))) {
	return ret;
	}
	}
	if (RET_FAIL(byte_writer_.flush_partial(out))) {
	return ret;
	}
	reset();
	return common::E_OK;
	}

	int flush_segment(int first, common::ByteStream& out) {
	int ret = common::E_OK;
	if (RET_FAIL(byte_writer_.write_bits(
	static_cast<uint64_t>(length_code_[first]),
	Traits::LENGTH_CODE_BITS, out))) {
	return ret;
	}

	uint64_t fib = calc_fibonacci(length_rle_[first]);
	int fib_len = binary_length(static_cast<T>(fib));
	for (int i = 0; i < fib_len; ++i) {
	if (RET_FAIL(
	byte_writer_.write_bit(((fib >> i) & 1ULL) != 0, out))) {
	return ret;
	}
	}
	if (RET_FAIL(byte_writer_.write_bit(true, out))) {
	return ret;
	}

	int j = first;
	do {
	int diff_len = binary_length(diff_value_[j]);
	if (RET_FAIL(byte_writer_.write_bits(
	java_rlbe_delta_bits(diff_value_[j], diff_len), diff_len,
	out))) {
	return ret;
	}
	++j;
	} while (j <= write_index_ && length_rle_[j] == 0);
	return common::E_OK;
	}

	protected:
	common::TSEncoding type_;
	T diff_value_[RLBE_BLOCK_DEFAULT_SIZE + 1];
	int length_code_[RLBE_BLOCK_DEFAULT_SIZE + 1];
	int length_rle_[RLBE_BLOCK_DEFAULT_SIZE + 1];
	T previous_value_;
	int write_index_;
	MsbBitWriter byte_writer_;
	};

	class IntRLBEEncoder : public RLBEEncoder<int32_t> {
	public:
	int encode(int32_t value, common::ByteStream& out) override {
	return encode_value(value, out);
	}
	};

	class LongRLBEEncoder : public RLBEEncoder<int64_t> {
	public:
	int encode(int64_t value, common::ByteStream& out) override {
	return encode_value(value, out);
	}
	};

	class FloatRLBEEncoder : public RLBEEncoder<int32_t> {
	public:
	int encode(float value, common::ByteStream& out) override {
	return encode_value(common::float_to_int(value), out);
	}
	};

	class DoubleRLBEEncoder : public RLBEEncoder<int64_t> {
	public:
	int encode(double value, common::ByteStream& out) override {
	return encode_value(common::double_to_long(value), out);
	}
	};

	} // namespace storage

	#endif // ENCODING_RLBE_ENCODER_H