| /* |
| * 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_PLAIN_ENCODER_H |
| #define ENCODING_PLAIN_ENCODER_H |
| |
| #include <cstring> |
| |
| #include "encoder.h" |
| |
| #ifdef ENABLE_SIMD |
| #include "simde/x86/ssse3.h" |
| #endif |
| |
| namespace storage { |
| |
| class PlainEncoder : public Encoder { |
| public: |
| PlainEncoder() {} |
| ~PlainEncoder() { destroy(); } |
| void destroy() override { /* do nothing for PlainEncoder */ |
| } |
| void reset() override { /* do thing for PlainEncoder */ |
| } |
| |
| FORCE_INLINE int encode(bool value, |
| common::ByteStream& out_stream) override { |
| return common::SerializationUtil::write_i8(value ? 1 : 0, out_stream); |
| } |
| |
| FORCE_INLINE int encode(int32_t value, |
| common::ByteStream& out_stream) override { |
| return common::SerializationUtil::write_var_int(value, out_stream); |
| } |
| |
| FORCE_INLINE int encode(int64_t value, |
| common::ByteStream& out_stream) override { |
| return common::SerializationUtil::write_i64(value, out_stream); |
| } |
| |
| FORCE_INLINE int encode(float value, |
| common::ByteStream& out_stream) override { |
| return common::SerializationUtil::write_float(value, out_stream); |
| } |
| |
| FORCE_INLINE int encode(double value, |
| common::ByteStream& out_stream) override { |
| return common::SerializationUtil::write_double(value, out_stream); |
| } |
| |
| FORCE_INLINE int encode(common::String value, |
| common::ByteStream& out_stream) override { |
| return common::SerializationUtil::write_mystring(value, out_stream); |
| } |
| |
| int flush(common::ByteStream& out_stream) override { |
| // do nothing for PlainEncoder |
| return common::E_OK; |
| } |
| |
| int get_max_byte_size() override { return 0; } |
| |
| // Optimized batch encoding: directly byte-swap into ByteStream page buffer. |
| // Avoids per-value write_buf overhead entirely — only calls acquire_buf() |
| // once per page boundary crossing. |
| int encode_batch(const int64_t* values, uint32_t count, |
| common::ByteStream& out_stream) override { |
| if (count == 0) return common::E_OK; |
| uint32_t offset = 0; |
| while (offset < count) { |
| common::ByteStream::Buffer buf = out_stream.acquire_buf(); |
| if (UNLIKELY(buf.buf_ == nullptr)) return common::E_OOM; |
| // How many int64 values fit in the remaining page space? |
| uint32_t capacity = buf.len_ / 8; |
| if (capacity == 0) { |
| // Page has < 8 bytes left, fall back to write_buf for this one |
| return Encoder::encode_batch(values + offset, count - offset, |
| out_stream); |
| } |
| uint32_t batch = std::min(count - offset, capacity); |
| uint8_t* dst = (uint8_t*)buf.buf_; |
| const int64_t* src = values + offset; |
| uint32_t i = 0; |
| #ifdef ENABLE_SIMD |
| // SIMDe: byte-reverse 2 x int64 per iteration |
| const simde__m128i bswap64_shuf = simde_mm_set_epi8( |
| 8, 9, 10, 11, 12, 13, 14, 15, 0, 1, 2, 3, 4, 5, 6, 7); |
| for (; i + 2 <= batch; i += 2) { |
| simde__m128i v = |
| simde_mm_loadu_si128((const simde__m128i*)&src[i]); |
| v = simde_mm_shuffle_epi8(v, bswap64_shuf); |
| simde_mm_storeu_si128((simde__m128i*)dst, v); |
| dst += 16; |
| } |
| #endif |
| // Scalar tail |
| for (; i < batch; i++) { |
| uint64_t v = (uint64_t)src[i]; |
| dst[0] = (uint8_t)(v >> 56); |
| dst[1] = (uint8_t)(v >> 48); |
| dst[2] = (uint8_t)(v >> 40); |
| dst[3] = (uint8_t)(v >> 32); |
| dst[4] = (uint8_t)(v >> 24); |
| dst[5] = (uint8_t)(v >> 16); |
| dst[6] = (uint8_t)(v >> 8); |
| dst[7] = (uint8_t)(v); |
| dst += 8; |
| } |
| out_stream.buffer_used(batch * 8); |
| offset += batch; |
| } |
| return common::E_OK; |
| } |
| |
| int encode_batch(const double* values, uint32_t count, |
| common::ByteStream& out_stream) override { |
| if (count == 0) return common::E_OK; |
| uint32_t offset = 0; |
| while (offset < count) { |
| common::ByteStream::Buffer buf = out_stream.acquire_buf(); |
| if (UNLIKELY(buf.buf_ == nullptr)) return common::E_OOM; |
| uint32_t capacity = buf.len_ / 8; |
| if (capacity == 0) { |
| return Encoder::encode_batch(values + offset, count - offset, |
| out_stream); |
| } |
| uint32_t batch = std::min(count - offset, capacity); |
| uint8_t* dst = (uint8_t*)buf.buf_; |
| const double* src = values + offset; |
| uint32_t i = 0; |
| #ifdef ENABLE_SIMD |
| // SIMDe: byte-reverse 2 x double (64-bit) per iteration |
| const simde__m128i bswap64_shuf = simde_mm_set_epi8( |
| 8, 9, 10, 11, 12, 13, 14, 15, 0, 1, 2, 3, 4, 5, 6, 7); |
| for (; i + 2 <= batch; i += 2) { |
| simde__m128i v = |
| simde_mm_loadu_si128((const simde__m128i*)&src[i]); |
| v = simde_mm_shuffle_epi8(v, bswap64_shuf); |
| simde_mm_storeu_si128((simde__m128i*)dst, v); |
| dst += 16; |
| } |
| #endif |
| // Scalar tail: round-trip the bits via memcpy to avoid the |
| // strict-aliasing violation of reading a double through an |
| // int64_t* (the old reinterpret_cast dispatch). |
| for (; i < batch; i++) { |
| uint64_t v; |
| memcpy(&v, &src[i], sizeof(double)); |
| dst[0] = (uint8_t)(v >> 56); |
| dst[1] = (uint8_t)(v >> 48); |
| dst[2] = (uint8_t)(v >> 40); |
| dst[3] = (uint8_t)(v >> 32); |
| dst[4] = (uint8_t)(v >> 24); |
| dst[5] = (uint8_t)(v >> 16); |
| dst[6] = (uint8_t)(v >> 8); |
| dst[7] = (uint8_t)(v); |
| dst += 8; |
| } |
| out_stream.buffer_used(batch * 8); |
| offset += batch; |
| } |
| return common::E_OK; |
| } |
| |
| int encode_batch(const float* values, uint32_t count, |
| common::ByteStream& out_stream) override { |
| if (count == 0) return common::E_OK; |
| uint32_t offset = 0; |
| while (offset < count) { |
| common::ByteStream::Buffer buf = out_stream.acquire_buf(); |
| if (UNLIKELY(buf.buf_ == nullptr)) return common::E_OOM; |
| uint32_t capacity = buf.len_ / 4; |
| if (capacity == 0) { |
| return Encoder::encode_batch(values + offset, count - offset, |
| out_stream); |
| } |
| uint32_t batch = std::min(count - offset, capacity); |
| uint8_t* dst = (uint8_t*)buf.buf_; |
| const float* src = values + offset; |
| uint32_t i = 0; |
| #ifdef ENABLE_SIMD |
| // SIMDe: byte-reverse 4 x float (32-bit) per iteration |
| const simde__m128i bswap32_shuf = simde_mm_set_epi8( |
| 12, 13, 14, 15, 8, 9, 10, 11, 4, 5, 6, 7, 0, 1, 2, 3); |
| for (; i + 4 <= batch; i += 4) { |
| simde__m128i v = |
| simde_mm_loadu_si128((const simde__m128i*)&src[i]); |
| v = simde_mm_shuffle_epi8(v, bswap32_shuf); |
| simde_mm_storeu_si128((simde__m128i*)dst, v); |
| dst += 16; |
| } |
| #endif |
| for (; i < batch; i++) { |
| uint32_t v; |
| memcpy(&v, &src[i], sizeof(float)); |
| dst[0] = (uint8_t)(v >> 24); |
| dst[1] = (uint8_t)(v >> 16); |
| dst[2] = (uint8_t)(v >> 8); |
| dst[3] = (uint8_t)(v); |
| dst += 4; |
| } |
| out_stream.buffer_used(batch * 4); |
| offset += batch; |
| } |
| return common::E_OK; |
| } |
| |
| // Batch encode strings from Arrow-style offset+buffer layout. |
| // Each string is serialized as: var_int(len) + raw bytes. |
| int encode_string_batch(const char* buffer, const uint32_t* offsets, |
| uint32_t start_idx, uint32_t count, |
| common::ByteStream& out_stream) override { |
| int ret = common::E_OK; |
| for (uint32_t i = 0; i < count; i++) { |
| uint32_t idx = start_idx + i; |
| uint32_t len = offsets[idx + 1] - offsets[idx]; |
| if (RET_FAIL(common::SerializationUtil::write_var_int( |
| (int32_t)len, out_stream))) { |
| return ret; |
| } |
| if (len > 0) { |
| if (RET_FAIL( |
| out_stream.write_buf(buffer + offsets[idx], len))) { |
| return ret; |
| } |
| } |
| } |
| return ret; |
| } |
| }; |
| |
| } // end namespace storage |
| #endif // ENCODING_PLAIN_ENCODER_H |