blob: 3db87e173db1d983e1fef120ee7b529948aebf12 [file]
/*
* 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