diff --git a/cpp/cmake_modules/SetupCxxFlags.cmake b/cpp/cmake_modules/SetupCxxFlags.cmake index 21341167fe9..8887a1ed929 100644 --- a/cpp/cmake_modules/SetupCxxFlags.cmake +++ b/cpp/cmake_modules/SetupCxxFlags.cmake @@ -633,21 +633,12 @@ endif() if(NOT MSVC) set(C_RELEASE_FLAGS "") - if(CMAKE_C_FLAGS_RELEASE MATCHES "-O3") - string(APPEND C_RELEASE_FLAGS " -O2") - endif() + # Local override: keep -O3 from CMake's default Release flags for the + # pfor benchmark — the bench is sensitive to inlining/unrolling that + # -O3 enables. Upstream Arrow downgrades to -O2 here; we skip that. set(CXX_RELEASE_FLAGS "") - if(CMAKE_CXX_FLAGS_RELEASE MATCHES "-O3") - string(APPEND CXX_RELEASE_FLAGS " -O2") - endif() set(C_RELWITHDEBINFO_FLAGS "") - if(CMAKE_C_FLAGS_RELWITHDEBINFO MATCHES "-O3") - string(APPEND C_RELWITHDEBINFO_FLAGS " -O2") - endif() set(CXX_RELWITHDEBINFO_FLAGS "") - if(CMAKE_CXX_FLAGS_RELWITHDEBINFO MATCHES "-O3") - string(APPEND CXX_RELWITHDEBINFO_FLAGS " -O2") - endif() if(CMAKE_CXX_COMPILER_ID STREQUAL "GNU") string(APPEND C_RELEASE_FLAGS " -ftree-vectorize") string(APPEND CXX_RELEASE_FLAGS " -ftree-vectorize") diff --git a/cpp/src/arrow/util/CMakeLists.txt b/cpp/src/arrow/util/CMakeLists.txt index 628e9a4d1c7..5603eb2411e 100644 --- a/cpp/src/arrow/util/CMakeLists.txt +++ b/cpp/src/arrow/util/CMakeLists.txt @@ -118,6 +118,22 @@ add_arrow_test(threading-utility-test test_common.cc thread_pool_test.cc) +add_arrow_test(pfor-test + SOURCES + pfor/pfor_test.cc + pfor/pfor.cc + pfor/pfor_wrapper.cc) + +add_arrow_benchmark(pfor/pfor_benchmark + EXTRA_SOURCES + pfor/pfor.cc + pfor/pfor_wrapper.cc) + +add_arrow_test(fastlanes-for-test + SOURCES + fastlanes/fastlanes_for_test.cc + fastlanes/fastlanes_for.cc) + add_arrow_benchmark(bit_block_counter_benchmark) add_arrow_benchmark(bit_util_benchmark) add_arrow_benchmark(bitmap_reader_benchmark) diff --git a/cpp/src/arrow/util/fastlanes/fastlanes_for.cc b/cpp/src/arrow/util/fastlanes/fastlanes_for.cc new file mode 100644 index 00000000000..8ac078a6a16 --- /dev/null +++ b/cpp/src/arrow/util/fastlanes/fastlanes_for.cc @@ -0,0 +1,245 @@ +// 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. + +#include "arrow/util/fastlanes/fastlanes_for.h" + +#include +#include +#include + +#include "arrow/util/fastlanes/fastlanes_kernels.h" + +namespace arrow { +namespace util { +namespace fastlanes { + +namespace { + +// Dispatch by runtime bit_width into the templated PackBlock. +inline void PackBlockDispatch(uint32_t bit_width, const uint32_t* in, + uint32_t* out) { + switch (bit_width) { +#define CASE_W(N) \ + case N: \ + PackBlock(in, out); \ + break + CASE_W(1); CASE_W(2); CASE_W(3); CASE_W(4); CASE_W(5); CASE_W(6); + CASE_W(7); CASE_W(8); CASE_W(9); CASE_W(10); CASE_W(11); CASE_W(12); + CASE_W(13); CASE_W(14); CASE_W(15); CASE_W(16); CASE_W(17); CASE_W(18); + CASE_W(19); CASE_W(20); CASE_W(21); CASE_W(22); CASE_W(23); CASE_W(24); + CASE_W(25); CASE_W(26); CASE_W(27); CASE_W(28); CASE_W(29); CASE_W(30); + CASE_W(31); CASE_W(32); +#undef CASE_W + default: break; + } +} + +inline void UnpackBlockDispatch(uint32_t bit_width, const uint32_t* packed, + uint32_t* out) { + switch (bit_width) { +#define CASE_W(N) \ + case N: \ + UnpackBlock(packed, out); \ + break + CASE_W(1); CASE_W(2); CASE_W(3); CASE_W(4); CASE_W(5); CASE_W(6); + CASE_W(7); CASE_W(8); CASE_W(9); CASE_W(10); CASE_W(11); CASE_W(12); + CASE_W(13); CASE_W(14); CASE_W(15); CASE_W(16); CASE_W(17); CASE_W(18); + CASE_W(19); CASE_W(20); CASE_W(21); CASE_W(22); CASE_W(23); CASE_W(24); + CASE_W(25); CASE_W(26); CASE_W(27); CASE_W(28); CASE_W(29); CASE_W(30); + CASE_W(31); CASE_W(32); +#undef CASE_W + default: break; + } +} + +inline uint8_t BitsRequired(uint32_t range) { + if (range == 0) return 0; + return static_cast(32 - __builtin_clz(range)); +} + +} // namespace + +Result FastLanesForCodec::Encode(const int32_t* input, + int64_t num_values, uint8_t* output) { + if (num_values % kChunkSize != 0) { + return Status::Invalid("FastLanesForCodec::Encode: num_values (", num_values, + ") must be a multiple of ", kChunkSize); + } + const int64_t num_chunks = num_values / kChunkSize; + uint8_t* out_ptr = output; + + // Scratch buffers: one chunk's worth of FOR-shifted values (uint32_t) in + // transposed FL_ORDER, ready to feed PackBlock twice (one per 1024-block). + alignas(64) uint32_t transposed[kChunkSize]; + + for (int64_t chunk = 0; chunk < num_chunks; ++chunk) { + const int32_t* in_chunk = input + chunk * kChunkSize; + + // Compute min over the 2048 values. + int32_t min_val = std::numeric_limits::max(); + int32_t max_val = std::numeric_limits::min(); + for (int64_t i = 0; i < kChunkSize; ++i) { + const int32_t v = in_chunk[i]; + if (v < min_val) min_val = v; + if (v > max_val) max_val = v; + } + const uint32_t range = static_cast(max_val - min_val); + const uint8_t bit_width = BitsRequired(range); + + // Write 5-byte header: [min (4B LE int32_t)] [bit_width (1B)]. + std::memcpy(out_ptr, &min_val, sizeof(int32_t)); + out_ptr[4] = bit_width; + out_ptr += kChunkHeaderSize; + + if (bit_width == 0) { + // All values equal min; no payload bytes needed. + continue; + } + + // Build the transposed FOR-shifted scratch: per 1024-block, gather + // input[fromTransposed32(t)] - min into transposed[t]. + for (int64_t block = 0; block < 2; ++block) { + const int32_t* block_in = in_chunk + block * static_cast(kBlockSize); + uint32_t* block_t = transposed + block * kBlockSize; + for (size_t t = 0; t < kBlockSize; ++t) { + block_t[t] = static_cast(block_in[fromTransposed32(t)] - min_val); + } + } + + // Pack each 1024-block; payload is 2 * 128 * bit_width bytes total. + const int64_t bytes_per_block = 128 * bit_width; + for (int64_t block = 0; block < 2; ++block) { + PackBlockDispatch(bit_width, transposed + block * kBlockSize, + reinterpret_cast(out_ptr + block * bytes_per_block)); + } + out_ptr += 2 * bytes_per_block; + } + + return out_ptr - output; +} + +Status FastLanesForCodec::Decode(int32_t* output, int64_t num_values, + const uint8_t* input, int64_t input_size) { + if (num_values % kChunkSize != 0) { + return Status::Invalid("FastLanesForCodec::Decode: num_values (", num_values, + ") must be a multiple of ", kChunkSize); + } + const int64_t num_chunks = num_values / kChunkSize; + const uint8_t* in_ptr = input; + const uint8_t* in_end = input + input_size; + + alignas(64) uint32_t transposed[kChunkSize]; + + for (int64_t chunk = 0; chunk < num_chunks; ++chunk) { + if (in_ptr + kChunkHeaderSize > in_end) { + return Status::Invalid("FastLanesForCodec::Decode: truncated input header"); + } + int32_t min_val; + std::memcpy(&min_val, in_ptr, sizeof(int32_t)); + const uint8_t bit_width = in_ptr[4]; + in_ptr += kChunkHeaderSize; + + int32_t* out_chunk = output + chunk * kChunkSize; + + if (bit_width == 0) { + // All values equal min; emit min in transposed order (just fill). + for (int64_t i = 0; i < kChunkSize; ++i) out_chunk[i] = min_val; + continue; + } + + const int64_t bytes_per_block = 128 * bit_width; + if (in_ptr + 2 * bytes_per_block > in_end) { + return Status::Invalid("FastLanesForCodec::Decode: truncated input body"); + } + + // Unpack each 1024-block into the transposed scratch. + for (int64_t block = 0; block < 2; ++block) { + UnpackBlockDispatch( + bit_width, + reinterpret_cast(in_ptr + block * bytes_per_block), + transposed + block * kBlockSize); + } + in_ptr += 2 * bytes_per_block; + + // Add min back. Output stays in transposed FL_ORDER (no scatter). + for (int64_t i = 0; i < kChunkSize; ++i) { + out_chunk[i] = static_cast(transposed[i]) + min_val; + } + } + + return Status::OK(); +} + +Status FastLanesForCodec::DecodeFlat(int32_t* output, int64_t num_values, + const uint8_t* input, int64_t input_size) { + if (num_values % kChunkSize != 0) { + return Status::Invalid("FastLanesForCodec::DecodeFlat: num_values (", + num_values, ") must be a multiple of ", kChunkSize); + } + const int64_t num_chunks = num_values / kChunkSize; + const uint8_t* in_ptr = input; + const uint8_t* in_end = input + input_size; + + alignas(64) uint32_t transposed[kChunkSize]; + + for (int64_t chunk = 0; chunk < num_chunks; ++chunk) { + if (in_ptr + kChunkHeaderSize > in_end) { + return Status::Invalid("FastLanesForCodec::DecodeFlat: truncated header"); + } + int32_t min_val; + std::memcpy(&min_val, in_ptr, sizeof(int32_t)); + const uint8_t bit_width = in_ptr[4]; + in_ptr += kChunkHeaderSize; + + int32_t* out_chunk = output + chunk * kChunkSize; + + if (bit_width == 0) { + for (int64_t i = 0; i < kChunkSize; ++i) out_chunk[i] = min_val; + continue; + } + + const int64_t bytes_per_block = 128 * bit_width; + if (in_ptr + 2 * bytes_per_block > in_end) { + return Status::Invalid("FastLanesForCodec::DecodeFlat: truncated body"); + } + + for (int64_t block = 0; block < 2; ++block) { + UnpackBlockDispatch( + bit_width, + reinterpret_cast(in_ptr + block * bytes_per_block), + transposed + block * kBlockSize); + } + in_ptr += 2 * bytes_per_block; + + // Scatter via fromTransposed32 per 1024-block to restore original + // input order: out_chunk[block*1024 + fromTransposed32(t)] = transposed + + // min. This is the FL_ORDER inverse of the encode-time gather. + for (int64_t block = 0; block < 2; ++block) { + const uint32_t* block_t = transposed + block * kBlockSize; + int32_t* block_out = out_chunk + block * static_cast(kBlockSize); + for (size_t t = 0; t < kBlockSize; ++t) { + block_out[fromTransposed32(t)] = static_cast(block_t[t]) + min_val; + } + } + } + + return Status::OK(); +} + +} // namespace fastlanes +} // namespace util +} // namespace arrow diff --git a/cpp/src/arrow/util/fastlanes/fastlanes_for.h b/cpp/src/arrow/util/fastlanes/fastlanes_for.h new file mode 100644 index 00000000000..55bcf412bca --- /dev/null +++ b/cpp/src/arrow/util/fastlanes/fastlanes_for.h @@ -0,0 +1,83 @@ +// 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. + +// FastLanes + Frame-of-Reference encoder/decoder. +// +// Splits the input into 2048-value chunks. Per chunk: +// 1. Compute min over the 2048 values. +// 2. Subtract min from every value -> all values in [0, max-min]. +// 3. Compute bit_width = ceil(log2(max-min+1)). +// 4. Apply the FastLanes FL_ORDER permutation within each 1024-block +// (gather input[fromTransposed32(t)]) and pack with FastLanes +// lane-interleaved bit-packing at bit_width. +// +// The decoded output is in the FastLanes transposed order (NO scatter +// step on decode), i.e. output[chunk*2048 + block*1024 + t] equals +// input[chunk*2048 + block*1024 + fromTransposed32(t)] + min. +// +// Per-chunk header (5 bytes): [min (4B int32_t little-endian)] [bit_width (1B)] +// followed by 2 * 128 * bit_width bytes of packed data (zero if bit_width=0). + +#pragma once + +#include + +#include "arrow/result.h" +#include "arrow/status.h" + +namespace arrow { +namespace util { +namespace fastlanes { + +class FastLanesForCodec { + public: + // Number of values per encoded chunk (= 2 FastLanes 1024-blocks). + static constexpr int64_t kChunkSize = 2048; + + // Per-chunk header size: 4 bytes (int32 min) + 1 byte (bit_width). + static constexpr int64_t kChunkHeaderSize = 5; + + // Upper-bound size of the encoded output for `num_values` (must be a + // multiple of kChunkSize). + static int64_t MaxEncodedSize(int64_t num_values) { + const int64_t num_chunks = (num_values + kChunkSize - 1) / kChunkSize; + // Max bit_width = 32 → 2 * 128 * 32 = 8192 bytes per chunk packed body. + return num_chunks * (kChunkHeaderSize + 2 * 128 * 32); + } + + // Encode `num_values` int32_t values. `num_values` must be a multiple + // of kChunkSize. Returns the number of bytes written to `output`. + // `output` must be at least MaxEncodedSize(num_values) bytes. + static Result Encode(const int32_t* input, int64_t num_values, + uint8_t* output); + + // Decode `num_values` int32_t values. Output is in FastLanes + // transposed order within each 1024-block (see header comment). + static Status Decode(int32_t* output, int64_t num_values, + const uint8_t* input, int64_t input_size); + + // Decode and scatter back to ORIGINAL input order: output[i] == input[i] + // for the encoded input. Pays the FL_ORDER scatter cost on top of the + // kernel — slower than Decode(), included for apples-to-apples + // comparison against codecs that produce flat output. + static Status DecodeFlat(int32_t* output, int64_t num_values, + const uint8_t* input, int64_t input_size); +}; + +} // namespace fastlanes +} // namespace util +} // namespace arrow diff --git a/cpp/src/arrow/util/fastlanes/fastlanes_for_test.cc b/cpp/src/arrow/util/fastlanes/fastlanes_for_test.cc new file mode 100644 index 00000000000..0d65d10791a --- /dev/null +++ b/cpp/src/arrow/util/fastlanes/fastlanes_for_test.cc @@ -0,0 +1,128 @@ +// 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. + +#include "arrow/util/fastlanes/fastlanes_for.h" + +#include +#include +#include + +#include + +#include "arrow/util/fastlanes/fastlanes_kernels.h" + +namespace arrow { +namespace util { +namespace fastlanes { + +// Round-trip checker. Decoder output is in transposed order: per +// 1024-block, output[t] should equal input[fromTransposed32(t)]. +void CheckRoundTrip(const std::vector& input) { + ASSERT_EQ(input.size() % FastLanesForCodec::kChunkSize, 0u); + std::vector encoded(FastLanesForCodec::MaxEncodedSize(input.size())); + auto encoded_size = FastLanesForCodec::Encode(input.data(), input.size(), + encoded.data()); + ASSERT_TRUE(encoded_size.ok()); + + std::vector decoded(input.size(), 0xdeadbeef); + auto status = FastLanesForCodec::Decode(decoded.data(), input.size(), + encoded.data(), *encoded_size); + ASSERT_TRUE(status.ok()) << status.ToString(); + + const int64_t num_chunks = + static_cast(input.size()) / FastLanesForCodec::kChunkSize; + for (int64_t chunk = 0; chunk < num_chunks; ++chunk) { + for (int64_t block = 0; block < 2; ++block) { + const int64_t base = chunk * FastLanesForCodec::kChunkSize + block * 1024; + for (size_t t = 0; t < 1024; ++t) { + ASSERT_EQ(decoded[base + t], input[base + fromTransposed32(t)]) + << "chunk=" << chunk << " block=" << block << " t=" << t; + } + } + } +} + +TEST(FastLanesForRoundTrip, NarrowRange) { + std::mt19937 rng(42); + std::uniform_int_distribution dist(1000, 1500); + std::vector in(2048); + for (auto& v : in) v = dist(rng); + CheckRoundTrip(in); +} + +TEST(FastLanesForRoundTrip, SingleValue) { + std::vector in(2048, 12345); // bit_width = 0 path + CheckRoundTrip(in); +} + +TEST(FastLanesForRoundTrip, FullRange) { + std::mt19937 rng(7); + std::uniform_int_distribution dist(std::numeric_limits::min(), + std::numeric_limits::max()); + std::vector in(2048); + for (auto& v : in) v = dist(rng); + CheckRoundTrip(in); +} + +TEST(FastLanesForRoundTrip, MultipleChunks) { + std::mt19937 rng(99); + std::uniform_int_distribution dist(-100, 100); + std::vector in(8 * 2048); // 8 chunks + for (auto& v : in) v = dist(rng); + CheckRoundTrip(in); +} + +TEST(FastLanesForRoundTrip, BoundaryValues) { + for (int32_t mn : {-1000, 0, 1000}) { + for (int32_t span : {0, 1, 127, 128, 65535, 1 << 20}) { + std::vector in(2048); + std::mt19937 rng(static_cast(mn + span)); + std::uniform_int_distribution dist(mn, mn + span); + for (auto& v : in) v = dist(rng); + SCOPED_TRACE(testing::Message() << "min=" << mn << " span=" << span); + CheckRoundTrip(in); + } + } +} + +// DecodeFlat produces output in ORIGINAL input order (output[i] == input[i]). +// Confirms the FL_ORDER scatter on decode inverts the FL_ORDER gather on +// encode. +TEST(FastLanesForRoundTrip, DecodeFlatIsIdentity) { + std::mt19937 rng(12345); + std::uniform_int_distribution dist(-1000, 1000); + std::vector in(4 * 2048); + for (auto& v : in) v = dist(rng); + + std::vector encoded(FastLanesForCodec::MaxEncodedSize(in.size())); + auto encoded_size = FastLanesForCodec::Encode(in.data(), in.size(), + encoded.data()); + ASSERT_TRUE(encoded_size.ok()); + + std::vector decoded(in.size(), 0xdeadbeef); + auto status = FastLanesForCodec::DecodeFlat(decoded.data(), in.size(), + encoded.data(), *encoded_size); + ASSERT_TRUE(status.ok()) << status.ToString(); + + for (size_t i = 0; i < in.size(); ++i) { + ASSERT_EQ(decoded[i], in[i]) << "i=" << i; + } +} + +} // namespace fastlanes +} // namespace util +} // namespace arrow diff --git a/cpp/src/arrow/util/fastlanes/fastlanes_kernels.h b/cpp/src/arrow/util/fastlanes/fastlanes_kernels.h new file mode 100644 index 00000000000..a8f5b3c7638 --- /dev/null +++ b/cpp/src/arrow/util/fastlanes/fastlanes_kernels.h @@ -0,0 +1,163 @@ +// 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. + +// FastLanes auto-vectorized bit-packing kernels (header-only). +// +// Portable C++ port of the lane-interleaved 1024-bit format from +// FastLanes (Afroozeh & Boncz, VLDB '23). No SIMD intrinsics — the +// inner lane loop is written so the compiler auto-vectorizes it to +// 4-wide NEON / 8-wide AVX2 / 16-wide AVX512 without source changes. +// +// Within a 1024-value block (32 lanes × 32 rows for uint32_t), the +// packed buffer holds w u32 rows of 32 u32 words, where w is the bit +// width. Row r, lane l contributes to packed[r * 32 + l] at bit shift +// (r*w) % 32, possibly straddling into packed[r * 32 + 32 + l]. +// +// FL_ORDER (the 8x16 -> 16x8 within-sub-block transpose plus the +// 3-bit-reversal sub-block reorder) is applied outside these kernels: +// callers gather input[fromTransposed32(t)] before packing; the kernel +// produces output in the same transposed order (no scatter on decode). + +#pragma once + +#include +#include +#include + +namespace arrow { +namespace util { +namespace fastlanes { + +constexpr size_t kBlockSize = 1024; +constexpr size_t kLanes = 32; // 1024 / sizeof(uint32_t) / 8 +constexpr size_t kRowsPerBlock = 32; // 1024 / kLanes + +// Sub-block reorder permutation (3-bit reversal). Used by toTransposed32 / +// fromTransposed32. Self-inverse, but note that toTransposed32 and +// fromTransposed32 themselves are NOT self-inverse — they are mutual +// inverses (the 8x16 -> 16x8 within-sub-block transpose is not involutive). +inline constexpr size_t kBlockReorder[8] = {0, 4, 2, 6, 1, 5, 3, 7}; + +// Convert an original (flat) input index to its position in the transposed +// (stream) order within a 1024-value block. +inline size_t toTransposed32(size_t origIdx) { + const size_t block = origIdx >> 7; + const size_t withinBlock = origIdx & 0x7F; + const size_t row = withinBlock >> 4; // row in 8x16 + const size_t col = withinBlock & 0xF; // col in 8x16 + const size_t transposedWithin = (col << 3) | row; // col * 8 + row (16x8) + const size_t outputBlock = kBlockReorder[block]; + return (outputBlock << 7) | transposedWithin; +} + +// Convert a transposed (stream) index back to its original input index +// within a 1024-value block. Inverse of toTransposed32 (the 8x16 -> 16x8 +// transpose is mutual-inverse with the 16x8 -> 8x16 transpose). +inline size_t fromTransposed32(size_t t) { + const size_t outputBlock = t >> 7; // t / 128 + const size_t transposedWithin = t & 0x7F; // t % 128 + const size_t originalBlock = kBlockReorder[outputBlock]; + const size_t row = transposedWithin >> 3; // / 8 + const size_t col = transposedWithin & 0x7; // % 8 + const size_t withinBlock = (col << 4) | row; // col * 16 + row + return (originalBlock << 7) | withinBlock; +} + +// --------------------------------------------------------------------------- +// Pack: 1024 transposed-order u32 inputs (already gathered via +// fromTransposed32) → w*32 u32 packed words. +// --------------------------------------------------------------------------- +template +inline void PackBlock(const uint32_t* __restrict__ in, uint32_t* __restrict__ out) { + static_assert(w >= 1 && w <= 32); + constexpr uint32_t kMask = (w == 32) ? 0xFFFFFFFFu : ((1u << w) - 1); + + if constexpr (w == 32) { + std::memcpy(out, in, kBlockSize * sizeof(uint32_t)); + return; + } + + std::memset(out, 0, w * kLanes * sizeof(uint32_t)); + +#pragma GCC unroll 32 + for (uint32_t row = 0; row < kRowsPerBlock; ++row) { + constexpr uint32_t kT = 32; + const uint32_t startBit = row * w; + const uint32_t word = startBit / kT; + const uint32_t shift = startBit % kT; + const uint32_t endBit = startBit + w; + const uint32_t endWord = (endBit - 1) / kT; + + if (word == endWord) { + for (uint32_t lane = 0; lane < kLanes; ++lane) { + const uint32_t v = in[row * kLanes + lane] & kMask; + out[word * kLanes + lane] |= (v << shift); + } + } else { + const uint32_t lowBits = kT - shift; + for (uint32_t lane = 0; lane < kLanes; ++lane) { + const uint32_t v = in[row * kLanes + lane] & kMask; + out[word * kLanes + lane] |= (v << shift); + out[endWord * kLanes + lane] |= (v >> lowBits); + } + } + } +} + +// --------------------------------------------------------------------------- +// Unpack: w*32 packed u32 words → 1024 u32 outputs in transposed order. +// (No scatter: output[t] is the value at stream-position t = row*32+lane.) +// --------------------------------------------------------------------------- +template +inline void UnpackBlock(const uint32_t* __restrict__ packed, + uint32_t* __restrict__ out) { + static_assert(w >= 1 && w <= 32); + constexpr uint32_t kMask = (w == 32) ? 0xFFFFFFFFu : ((1u << w) - 1); + + if constexpr (w == 32) { + std::memcpy(out, packed, kBlockSize * sizeof(uint32_t)); + return; + } + +#pragma GCC unroll 32 + for (uint32_t row = 0; row < kRowsPerBlock; ++row) { + constexpr uint32_t kT = 32; + const uint32_t startBit = row * w; + const uint32_t word = startBit / kT; + const uint32_t shift = startBit % kT; + const uint32_t endBit = startBit + w; + const uint32_t endWord = (endBit - 1) / kT; + + if (word == endWord) { + for (uint32_t lane = 0; lane < kLanes; ++lane) { + out[row * kLanes + lane] = + (packed[word * kLanes + lane] >> shift) & kMask; + } + } else { + const uint32_t lowBits = kT - shift; + for (uint32_t lane = 0; lane < kLanes; ++lane) { + const uint32_t lo = packed[word * kLanes + lane] >> shift; + const uint32_t hi = packed[endWord * kLanes + lane] << lowBits; + out[row * kLanes + lane] = (lo | hi) & kMask; + } + } + } +} + +} // namespace fastlanes +} // namespace util +} // namespace arrow diff --git a/cpp/src/arrow/util/pfor/pfor.cc b/cpp/src/arrow/util/pfor/pfor.cc new file mode 100644 index 00000000000..0a185d3512e --- /dev/null +++ b/cpp/src/arrow/util/pfor/pfor.cc @@ -0,0 +1,430 @@ +// 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. + +// Core PFOR (Patched Frame of Reference) compression implementation +// +// Implementation notes: +// - Vector size: 1024 +// - Max exceptions: int16 +// - Exception values: original integers (not FOR offsets) +// - Bit packing: Arrow's BitWriter/unpack + +#include "arrow/util/pfor/pfor.h" + +#include +#include +#include +#include + +#include "arrow/util/bit_stream_utils_internal.h" +#include "arrow/util/bit_util.h" +#include "arrow/util/bpacking_internal.h" +#include "arrow/util/endian.h" +#include "arrow/util/fastlanes/fastlanes_kernels.h" +#include "arrow/util/logging.h" +#include "arrow/util/macros.h" +#include "arrow/util/span.h" +#include "arrow/util/ubsan.h" + +namespace arrow { +namespace util { +namespace pfor { + +static_assert(ARROW_LITTLE_ENDIAN, + "PFOR serialization assumes little-endian byte order"); + +// ---------------------------------------------------------------------- +// FindOptimalBitWidth: histogram-based cost model + +template +BitWidthResult PforCompression::FindOptimalBitWidth(const UnsignedT* deltas, + int32_t num_elements) { + constexpr uint8_t max_bits = PforTypeTraits::kMaxBitWidth; + constexpr int32_t position_bits = 16; + constexpr int32_t value_bits = sizeof(T) * 8; + + // Build histogram: histogram[b] = count of deltas requiring exactly b bits + std::array histogram{}; + for (int32_t i = 0; i < num_elements; ++i) { + uint8_t bits = PforTypeTraits::BitsRequired(deltas[i]); + histogram[bits]++; + } + + // Evaluate each candidate bit width + int64_t best_cost = std::numeric_limits::max(); + uint8_t best_bit_width = max_bits; + int16_t best_num_exceptions = 0; + + int64_t exceptions_above = num_elements; + + for (uint8_t b = 0; b <= max_bits; ++b) { + exceptions_above -= histogram[b]; + + if (exceptions_above > std::numeric_limits::max()) { + continue; + } + + int64_t packing_cost = static_cast(num_elements) * b; + int64_t exception_cost = exceptions_above * (position_bits + value_bits); + int64_t total_cost = packing_cost + exception_cost; + + if (total_cost < best_cost) { + best_cost = total_cost; + best_bit_width = b; + best_num_exceptions = static_cast(exceptions_above); + } + } + + return {best_bit_width, best_num_exceptions}; +} + +namespace { + +// Dispatch by runtime bit_width into the templated FastLanes pack/unpack +// kernels. Only used when use_fastlanes && num_elements == kPforVectorSize +// (the FastLanes block size matches PFOR's vector size of 1024). +inline void FastLanesPackBlockDispatch(uint8_t bit_width, const uint32_t* in, + uint32_t* out) { + switch (bit_width) { +#define CASE_W(N) \ + case N: \ + fastlanes::PackBlock(in, out); \ + break + CASE_W(1); CASE_W(2); CASE_W(3); CASE_W(4); CASE_W(5); CASE_W(6); + CASE_W(7); CASE_W(8); CASE_W(9); CASE_W(10); CASE_W(11); CASE_W(12); + CASE_W(13); CASE_W(14); CASE_W(15); CASE_W(16); CASE_W(17); CASE_W(18); + CASE_W(19); CASE_W(20); CASE_W(21); CASE_W(22); CASE_W(23); CASE_W(24); + CASE_W(25); CASE_W(26); CASE_W(27); CASE_W(28); CASE_W(29); CASE_W(30); + CASE_W(31); CASE_W(32); +#undef CASE_W + default: break; + } +} + +inline void FastLanesUnpackBlockDispatch(uint8_t bit_width, const uint32_t* packed, + uint32_t* out) { + switch (bit_width) { +#define CASE_W(N) \ + case N: \ + fastlanes::UnpackBlock(packed, out); \ + break + CASE_W(1); CASE_W(2); CASE_W(3); CASE_W(4); CASE_W(5); CASE_W(6); + CASE_W(7); CASE_W(8); CASE_W(9); CASE_W(10); CASE_W(11); CASE_W(12); + CASE_W(13); CASE_W(14); CASE_W(15); CASE_W(16); CASE_W(17); CASE_W(18); + CASE_W(19); CASE_W(20); CASE_W(21); CASE_W(22); CASE_W(23); CASE_W(24); + CASE_W(25); CASE_W(26); CASE_W(27); CASE_W(28); CASE_W(29); CASE_W(30); + CASE_W(31); CASE_W(32); +#undef CASE_W + default: break; + } +} + +} // namespace + +// ---------------------------------------------------------------------- +// EncodeVector + +template +PforEncodedVector PforCompression::EncodeVector(const T* values, + int32_t num_elements, + PackingMode mode) { + ARROW_DCHECK(num_elements > 0); + + // Step 1: Find min (frame of reference) + T min_val = values[0]; + for (int32_t i = 1; i < num_elements; ++i) { + if (values[i] < min_val) min_val = values[i]; + } + + // Step 2: Compute unsigned deltas + const auto unsigned_min = static_cast(min_val); + std::vector deltas(num_elements); + for (int32_t i = 0; i < num_elements; ++i) { + deltas[i] = static_cast(values[i]) - unsigned_min; + } + + // Step 3: Find optimal bit width + auto [bit_width, num_exceptions] = + FindOptimalBitWidth(deltas.data(), num_elements); + + // FastLanes only applies when the vector matches the FastLanes block size + // (1024) and the type is 32-bit. For shorter tails or 64-bit values, fall + // back to PackingMode::BitPack. + const PackingMode effective_mode = + (mode == PackingMode::FastLanes && + num_elements == static_cast(fastlanes::kBlockSize) && + sizeof(T) == 4) + ? PackingMode::FastLanes + : PackingMode::BitPack; + + // Step 4: Collect exceptions and replace with placeholder (0) + PforEncodedVector result; + result.set_info( + PforVectorInfo(min_val, bit_width, num_exceptions, effective_mode)); + + if (num_exceptions > 0) { + result.mutable_exception_positions().reserve(num_exceptions); + result.mutable_exception_values().reserve(num_exceptions); + + UnsignedT mask = (bit_width >= PforTypeTraits::kMaxBitWidth) + ? static_cast(-1) + : (static_cast(1) << bit_width) - 1; + + for (int32_t i = 0; i < num_elements; ++i) { + if (deltas[i] > mask) { + result.mutable_exception_positions().push_back(static_cast(i)); + result.mutable_exception_values().push_back(values[i]); + deltas[i] = 0; + } + } + } + + // Step 5: Bit-pack the deltas + if (bit_width > 0) { + int64_t packed_size = + bit_util::BytesForBits(static_cast(num_elements) * bit_width); + result.mutable_packed_values().resize(static_cast(packed_size), 0); + + if (effective_mode == PackingMode::FastLanes) { + // Gather deltas via FL_ORDER into transposed scratch, then pack with + // FastLanes' lane-interleaved kernel. The packed payload is exactly + // 128 * bit_width bytes — same as the BitPack-encoded size. + alignas(64) uint32_t transposed[fastlanes::kBlockSize]; + for (size_t t = 0; t < fastlanes::kBlockSize; ++t) { + transposed[t] = static_cast(deltas[fastlanes::fromTransposed32(t)]); + } + FastLanesPackBlockDispatch( + bit_width, transposed, + reinterpret_cast(result.mutable_packed_values().data())); + } else { + bit_util::BitWriter writer(result.mutable_packed_values().data(), + static_cast(packed_size)); + for (int32_t i = 0; i < num_elements; ++i) { + writer.PutValue(static_cast(deltas[i]), bit_width); + } + writer.Flush(); + } + } + + return result; +} + +// ---------------------------------------------------------------------- +// DecodeVector + +template +Result PforCompression::DecodeVector(T* values, + arrow::util::span data, + int32_t num_elements, + OutputOrder order) { + // Step 1: Read vector info + ARROW_ASSIGN_OR_RAISE(auto info, PforVectorInfo::Load(data)); + const uint8_t* read_ptr = data.data() + PforVectorInfo::kStoredSize; + + // OutputOrder::Transposed is only meaningful for FastLanes-encoded vectors. + // BitPack vectors have no transposition to skip, so they always emit flat + // output regardless of `order`. + const bool emit_transposed = + (order == OutputOrder::Transposed) && + (info.packing_mode() == PackingMode::FastLanes); + + // Step 2: Handle constant data (bit_width == 0, no exceptions) + if (info.bit_width() == 0 && info.num_exceptions() == 0) { + std::fill(values, values + num_elements, info.frame_of_reference()); + return PforVectorInfo::kStoredSize; + } + + // Step 3: Unpack bit-packed deltas and add FOR + if (info.bit_width() > 0) { + const auto unsigned_for = static_cast(info.frame_of_reference()); + + if (info.packing_mode() == PackingMode::FastLanes) { + // FastLanes-packed payload: 128 * bit_width bytes per 1024-block. + // Unpack into transposed scratch. + ARROW_DCHECK(num_elements == + static_cast(fastlanes::kBlockSize)); + alignas(64) uint32_t transposed[fastlanes::kBlockSize]; + FastLanesUnpackBlockDispatch( + info.bit_width(), + reinterpret_cast(read_ptr), transposed); + + if (emit_transposed) { + // No FL_ORDER inverse: write `values[t] = transposed[t] + FOR` + // sequentially. The output is in FastLanes stream order, i.e. + // values[t] corresponds to the original input at fromTransposed32(t). + // Sequential read + sequential write is auto-vec friendly. + for (size_t t = 0; t < fastlanes::kBlockSize; ++t) { + values[t] = util::SafeCopy( + static_cast(transposed[t]) + unsigned_for); + } + } else { + // Fused FL_ORDER inverse + FOR-add + SafeCopy. The gather index is + // toTransposed32(i) (the inverse of the encode-side fromTransposed32 + // gather — the two are mutual inverses, not self-inverse). + for (size_t i = 0; i < fastlanes::kBlockSize; ++i) { + const UnsignedT v = + static_cast(transposed[fastlanes::toTransposed32(i)]); + values[i] = util::SafeCopy(v + unsigned_for); + } + } + } else { + std::vector unsigned_values(num_elements); + // Arrow's unpack handles arbitrary sizes: SIMD for complete batches, + // then unpack_exact for the remainder. + arrow::internal::unpack(read_ptr, unsigned_values.data(), + static_cast(num_elements), info.bit_width()); + + // Add FOR and convert to signed output via SafeCopy +#pragma GCC unroll PforConstants::kLoopUnrolls +#pragma GCC ivdep + for (int32_t i = 0; i < num_elements; ++i) { + unsigned_values[i] += unsigned_for; + values[i] = util::SafeCopy(unsigned_values[i]); + } + } + + int64_t packed_size = + bit_util::BytesForBits(static_cast(num_elements) * info.bit_width()); + read_ptr += packed_size; + } else { + // bit_width == 0 but has exceptions - fill with FOR + std::fill(values, values + num_elements, info.frame_of_reference()); + } + + // Step 4: Patch exceptions (stored as original values at FLAT positions). + // When emitting transposed output, redirect each patch to toTransposed32(pos). + const int16_t num_exceptions = info.num_exceptions(); + if (num_exceptions > 0) { + const uint8_t* positions_ptr = read_ptr; + read_ptr += num_exceptions * sizeof(int16_t); + + const uint8_t* values_ptr = read_ptr; + read_ptr += num_exceptions * sizeof(T); + +#pragma GCC unroll PforConstants::kLoopUnrolls +#pragma GCC ivdep + for (int16_t i = 0; i < num_exceptions; ++i) { + int16_t pos = util::SafeLoadAs(positions_ptr + i * sizeof(int16_t)); + T value = util::SafeLoadAs(values_ptr + i * sizeof(T)); + const size_t out_pos = + emit_transposed ? fastlanes::toTransposed32(static_cast(pos)) + : static_cast(pos); + values[out_pos] = value; + } + } + + return static_cast(read_ptr - data.data()); +} + +// ---------------------------------------------------------------------- +// Serialization helpers + +// ---------------------------------------------------------------------- +// PforEncodedVectorView::LoadView + +template +Result> PforEncodedVectorView::LoadView( + arrow::util::span data, int32_t num_elements) { + ARROW_ASSIGN_OR_RAISE(auto info, PforVectorInfo::Load(data)); + + PforEncodedVectorView view; + view.set_info(info); + view.set_num_elements(num_elements); + + const uint8_t* ptr = data.data() + PforVectorInfo::kStoredSize; + + // packed_values: zero-copy span into the buffer + int64_t packed_size = 0; + if (info.bit_width() > 0) { + packed_size = + bit_util::BytesForBits(static_cast(num_elements) * info.bit_width()); + view.set_packed_values(arrow::util::span(ptr, packed_size)); + ptr += packed_size; + } + + // Exception positions and values: copy into aligned storage + if (info.num_exceptions() > 0) { + view.mutable_exception_positions().resize(info.num_exceptions()); + std::memcpy(view.mutable_exception_positions().data(), ptr, + info.num_exceptions() * sizeof(int16_t)); + ptr += info.num_exceptions() * sizeof(int16_t); + + view.mutable_exception_values().resize(info.num_exceptions()); + std::memcpy(view.mutable_exception_values().data(), ptr, + info.num_exceptions() * sizeof(T)); + } + + return view; +} + +template class PforEncodedVectorView; +template class PforEncodedVectorView; + +// ---------------------------------------------------------------------- +// Serialization helpers + +template +int64_t PforCompression::SerializedVectorSize(const PforEncodedVector& vec, + int32_t num_elements) { + int64_t size = PforVectorInfo::kStoredSize; + if (vec.info().bit_width() > 0) { + size += bit_util::BytesForBits( + static_cast(num_elements) * vec.info().bit_width()); + } + size += vec.info().num_exceptions() * static_cast(sizeof(int16_t)); + size += vec.info().num_exceptions() * static_cast(sizeof(T)); + return size; +} + +template +int64_t PforCompression::SerializeVector(const PforEncodedVector& vec, + int32_t num_elements, + arrow::util::span dest) { + uint8_t* write_ptr = dest.data(); + + // Write vector info + vec.info().Store(arrow::util::span(write_ptr, PforVectorInfo::kStoredSize)); + write_ptr += PforVectorInfo::kStoredSize; + + // Write packed values + if (vec.info().bit_width() > 0 && !vec.packed_values().empty()) { + std::memcpy(write_ptr, vec.packed_values().data(), vec.packed_values().size()); + write_ptr += vec.packed_values().size(); + } + + // Write exception positions + if (vec.info().num_exceptions() > 0) { + std::memcpy(write_ptr, vec.exception_positions().data(), + vec.info().num_exceptions() * sizeof(int16_t)); + write_ptr += vec.info().num_exceptions() * sizeof(int16_t); + + // Write exception values (original integers) + std::memcpy(write_ptr, vec.exception_values().data(), + vec.info().num_exceptions() * sizeof(T)); + write_ptr += vec.info().num_exceptions() * sizeof(T); + } + + return static_cast(write_ptr - dest.data()); +} + +// Explicit template instantiations +template class PforCompression; +template class PforCompression; + +} // namespace pfor +} // namespace util +} // namespace arrow diff --git a/cpp/src/arrow/util/pfor/pfor.h b/cpp/src/arrow/util/pfor/pfor.h new file mode 100644 index 00000000000..f5bbfe2c826 --- /dev/null +++ b/cpp/src/arrow/util/pfor/pfor.h @@ -0,0 +1,285 @@ +// 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. + +// Core PFOR (Patched Frame of Reference) compression algorithm +// +// PFOR compresses integer columns by: +// 1. Subtracting the minimum value (Frame of Reference) +// 2. Choosing an optimal bit width via a cost model +// 3. Bit-packing the deltas at the chosen width +// 4. Storing outlier values (exceptions) separately + +#pragma once + +#include +#include +#include + +#include "arrow/result.h" +#include "arrow/status.h" +#include "arrow/util/pfor/pfor_constants.h" +#include "arrow/util/span.h" +#include "arrow/util/ubsan.h" + +namespace arrow { +namespace util { +namespace pfor { + +// ---------------------------------------------------------------------- +// Per-vector metadata + +/// \brief PFOR vector metadata stored at the start of each compressed vector. +/// +/// For INT32 (7 bytes): [frame_of_reference(4B)] [bit_width(1B)] [num_exceptions(2B)] +/// For INT64 (11 bytes): [frame_of_reference(8B)] [bit_width(1B)] [num_exceptions(2B)] +/// +/// The bit_width byte packs two fields: +/// bits 0..5 — the actual bit width (range 0..32, fits in 6 bits) +/// bit 7 — packing-mode flag (0 = PackingMode::BitPack, 1 = PackingMode::FastLanes) +/// bit 6 — reserved (zero) +/// Legacy encoders (which only wrote the bit width) produce vectors with the +/// high bits clear, so they round-trip through the new Load as BitPack. +template +class PforVectorInfo { + public: + static constexpr uint8_t kPackingModeFlagMask = 0x80; + static constexpr uint8_t kBitWidthMask = 0x3F; + + PforVectorInfo() = default; + PforVectorInfo(T frame_of_reference, uint8_t bit_width, int16_t num_exceptions, + PackingMode packing_mode = PackingMode::BitPack) + : frame_of_reference_(frame_of_reference), + bit_width_(bit_width), + num_exceptions_(num_exceptions), + packing_mode_(packing_mode) {} + + T frame_of_reference() const { return frame_of_reference_; } + uint8_t bit_width() const { return bit_width_; } + int16_t num_exceptions() const { return num_exceptions_; } + PackingMode packing_mode() const { return packing_mode_; } + + void set_frame_of_reference(T frame_of_reference) { + frame_of_reference_ = frame_of_reference; + } + void set_bit_width(uint8_t bit_width) { bit_width_ = bit_width; } + void set_num_exceptions(int16_t num_exceptions) { num_exceptions_ = num_exceptions; } + void set_packing_mode(PackingMode mode) { packing_mode_ = mode; } + + /// \brief Store this info to a byte buffer (little-endian) + void Store(arrow::util::span dest) const { + uint8_t* ptr = dest.data(); + util::SafeStore(ptr, frame_of_reference_); + const uint8_t mode_bit = + (packing_mode_ == PackingMode::FastLanes) ? kPackingModeFlagMask : 0; + ptr[sizeof(T)] = static_cast((bit_width_ & kBitWidthMask) | mode_bit); + util::SafeStore(ptr + sizeof(T) + 1, num_exceptions_); + } + + /// \brief Load this info from a byte buffer (little-endian) + static Result Load(arrow::util::span src) { + if (src.size() < static_cast(kStoredSize)) { + return Status::Invalid("PFOR vector info buffer too small: ", src.size(), + " < ", kStoredSize); + } + PforVectorInfo info; + const uint8_t* ptr = src.data(); + info.frame_of_reference_ = util::SafeLoadAs(ptr); + const uint8_t packed_bw = ptr[sizeof(T)]; + info.bit_width_ = static_cast(packed_bw & kBitWidthMask); + info.packing_mode_ = (packed_bw & kPackingModeFlagMask) != 0 + ? PackingMode::FastLanes + : PackingMode::BitPack; + info.num_exceptions_ = util::SafeLoadAs(ptr + sizeof(T) + 1); + if (info.bit_width_ > PforTypeTraits::kMaxBitWidth) { + return Status::Invalid("PFOR bit_width out of range: ", + static_cast(info.bit_width_)); + } + if (info.num_exceptions_ < 0) { + return Status::Invalid("PFOR num_exceptions negative: ", + info.num_exceptions_); + } + return info; + } + + /// \brief Serialized size in bytes + static constexpr int64_t kStoredSize = PforTypeTraits::kVectorInfoSize; + + private: + T frame_of_reference_ = 0; + uint8_t bit_width_ = 0; + int16_t num_exceptions_ = 0; + PackingMode packing_mode_ = PackingMode::BitPack; +}; + +// ---------------------------------------------------------------------- +// Encoded vector representation + +/// \brief A PFOR-encoded vector with all its data sections +template +class PforEncodedVector { + public: + PforEncodedVector() = default; + + const PforVectorInfo& info() const { return info_; } + PforVectorInfo& mutable_info() { return info_; } + void set_info(const PforVectorInfo& info) { info_ = info; } + + const std::vector& packed_values() const { return packed_values_; } + std::vector& mutable_packed_values() { return packed_values_; } + void set_packed_values(std::vector v) { packed_values_ = std::move(v); } + + const std::vector& exception_positions() const { return exception_positions_; } + std::vector& mutable_exception_positions() { return exception_positions_; } + void set_exception_positions(std::vector v) { + exception_positions_ = std::move(v); + } + + const std::vector& exception_values() const { return exception_values_; } + std::vector& mutable_exception_values() { return exception_values_; } + void set_exception_values(std::vector v) { exception_values_ = std::move(v); } + + private: + PforVectorInfo info_; + std::vector packed_values_; + std::vector exception_positions_; + std::vector exception_values_; +}; + +// ---------------------------------------------------------------------- +// Zero-copy encoded vector view + +/// \brief A zero-copy view over a serialized PFOR vector +/// +/// The packed_values span points directly into the compressed buffer. +/// Exception positions and values are copied into aligned storage. +template +class PforEncodedVectorView { + public: + PforEncodedVectorView() = default; + + const PforVectorInfo& info() const { return info_; } + PforVectorInfo& mutable_info() { return info_; } + void set_info(const PforVectorInfo& info) { info_ = info; } + + int32_t num_elements() const { return num_elements_; } + void set_num_elements(int32_t n) { num_elements_ = n; } + + arrow::util::span packed_values() const { return packed_values_; } + void set_packed_values(arrow::util::span v) { packed_values_ = v; } + + const std::vector& exception_positions() const { return exception_positions_; } + std::vector& mutable_exception_positions() { return exception_positions_; } + void set_exception_positions(std::vector v) { + exception_positions_ = std::move(v); + } + + const std::vector& exception_values() const { return exception_values_; } + std::vector& mutable_exception_values() { return exception_values_; } + void set_exception_values(std::vector v) { exception_values_ = std::move(v); } + + /// \brief Create a zero-copy view from a serialized vector buffer + /// + /// \param[in] data span over the serialized vector data + /// \param[in] num_elements number of elements in this vector + /// \return the view, or an error if the buffer is too small + static Result LoadView( + arrow::util::span data, int32_t num_elements); + + private: + PforVectorInfo info_; + int32_t num_elements_ = 0; + arrow::util::span packed_values_; + std::vector exception_positions_; + std::vector exception_values_; +}; + +// ---------------------------------------------------------------------- +// Cost model result + +/// \brief Result of the optimal bit width search +struct BitWidthResult { + uint8_t bit_width = 0; + int16_t num_exceptions = 0; +}; + +// ---------------------------------------------------------------------- +// Core compression/decompression + +/// \brief PFOR compression and decompression algorithms +/// +/// \tparam T the integer type (int32_t or int64_t) +template +class PforCompression { + public: + using UnsignedT = typename PforTypeTraits::UnsignedType; + + /// \brief Find the optimal bit width using the cost model + /// + /// Evaluates every candidate bit width and selects the one that + /// minimizes total encoded size (packing cost + exception cost). + /// + /// \param[in] deltas unsigned deltas after FOR subtraction + /// \param[in] num_elements number of elements + /// \return the optimal bit width and exception count + static BitWidthResult FindOptimalBitWidth(const UnsignedT* deltas, + int32_t num_elements); + + /// \brief Encode a single vector of integers + /// + /// \param[in] values input integer values + /// \param[in] num_elements number of elements (up to vector_size) + /// \param[in] mode bit-packing layout for the payload. PackingMode::FastLanes + /// requires num_elements == kPforVectorSize (1024); otherwise + /// falls back to PackingMode::BitPack for this vector. 64-bit + /// types always fall back to BitPack (FastLanes is u32-only). + /// \return the encoded vector with all sections + static PforEncodedVector EncodeVector(const T* values, int32_t num_elements, + PackingMode mode = PackingMode::BitPack); + + /// \brief Decode a single vector from compressed data + /// + /// \param[out] values output buffer for decoded integers + /// \param[in] data span over the compressed vector data + /// \param[in] num_elements number of elements in this vector + /// \param[in] order output value order. Default OutputOrder::Flat returns + /// values in their original input positions. OutputOrder:: + /// Transposed only affects FastLanes-encoded vectors (skips the + /// FL_ORDER gather on decode); BitPack vectors always return + /// flat output regardless of `order`. + /// \return number of bytes consumed from data, or error + static Result DecodeVector(T* values, arrow::util::span data, + int32_t num_elements, + OutputOrder order = OutputOrder::Flat); + + /// \brief Calculate the serialized size of an encoded vector + static int64_t SerializedVectorSize(const PforEncodedVector& vec, + int32_t num_elements); + + /// \brief Serialize an encoded vector to a byte buffer + /// + /// \param[in] vec the encoded vector + /// \param[in] num_elements number of elements + /// \param[out] dest output buffer (must be large enough) + /// \return number of bytes written + static int64_t SerializeVector(const PforEncodedVector& vec, + int32_t num_elements, + arrow::util::span dest); +}; + +} // namespace pfor +} // namespace util +} // namespace arrow diff --git a/cpp/src/arrow/util/pfor/pfor_benchmark.cc b/cpp/src/arrow/util/pfor/pfor_benchmark.cc new file mode 100644 index 00000000000..9368278a262 --- /dev/null +++ b/cpp/src/arrow/util/pfor/pfor_benchmark.cc @@ -0,0 +1,327 @@ +// 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. + +// PFOR encoding/decoding benchmarks. +// +// Data distributions cover the key archetypes that exercise PFOR's cost model +// differently: +// - Constant: bit_width=0, best case +// - Sequential: small range, ideal FOR +// - SmallRange: clustered random, good FOR compression +// - HighBaseSmallRange: high absolute values, small delta range (timestamps) +// - WithOutliers: tests exception handling path +// - Random: worst case, full bit-width +// - TPC-DS DateSk/StoreSk/ItemSk/Quantity: realistic surrogate key distributions + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "benchmark/benchmark.h" + +#include "arrow/util/pfor/pfor.h" +#include "arrow/util/pfor/pfor_wrapper.h" + +namespace arrow::util::pfor { +namespace { + +// ====================================================================== +// Data Generators + +using Int32Gen = std::vector (*)(int64_t); +using Int64Gen = std::vector (*)(int64_t); + +template +std::vector GenConstant(int64_t n) { + return std::vector(n, static_cast(42)); +} + +template +std::vector GenSequential(int64_t n) { + std::vector v(n); + std::iota(v.begin(), v.end(), static_cast(0)); + return v; +} + +template +std::vector GenSmallRange(int64_t n) { + std::vector v(n); + std::mt19937_64 rng(12345); + std::uniform_int_distribution dist(100000, 200000); + for (auto& x : v) x = dist(rng); + return v; +} + +template +std::vector GenHighBaseSmallRange(int64_t n) { + std::vector v(n); + const T kBase = static_cast(1704067200); + std::mt19937_64 rng(12345); + std::uniform_int_distribution dist(0, 1000); + for (auto& x : v) x = kBase + dist(rng); + return v; +} + +template +std::vector GenWithOutliers(int64_t n) { + std::vector v(n); + std::mt19937_64 rng(42); + std::uniform_int_distribution small_dist(1000, 1255); + for (auto& x : v) x = small_dist(rng); + std::uniform_int_distribution pos_dist(0, n - 1); + int num_outliers = std::max(static_cast(1), n / 100); + for (int i = 0; i < num_outliers; ++i) { + v[pos_dist(rng)] = static_cast(std::numeric_limits::max() / 2 + i); + } + return v; +} + +template +std::vector GenRandom(int64_t n) { + std::vector v(n); + std::mt19937_64 rng(99); + std::uniform_int_distribution dist(std::numeric_limits::min(), + std::numeric_limits::max()); + for (auto& x : v) x = dist(rng); + return v; +} + +template +std::vector GenTpcdsSoldDateSk(int64_t n) { + std::vector v(n); + const T kBase = 2450815; + std::mt19937_64 rng(12345); + std::uniform_int_distribution dist(0, 1820); + for (auto& x : v) x = kBase + dist(rng); + return v; +} + +template +std::vector GenTpcdsStoreSk(int64_t n) { + std::vector v(n); + std::mt19937_64 rng(12345); + std::uniform_int_distribution dist(1, 1000); + for (auto& x : v) x = dist(rng); + return v; +} + +template +std::vector GenTpcdsItemSk(int64_t n) { + std::vector v(n); + const T kMax = 100000; + std::mt19937_64 rng(12345); + std::exponential_distribution exp_dist(0.00005); + for (auto& x : v) { + T val = static_cast(exp_dist(rng)); + x = std::min(static_cast(val + 1), kMax); + } + return v; +} + +template +std::vector GenTpcdsQuantity(int64_t n) { + std::vector v(n); + std::mt19937_64 rng(12345); + std::uniform_int_distribution small_dist(1, 10); + std::uniform_int_distribution large_dist(11, 100); + std::uniform_int_distribution chance(0, 99); + for (auto& x : v) { + x = (chance(rng) < 90) ? small_dist(rng) : large_dist(rng); + } + return v; +} + +// ====================================================================== +// Benchmark Core + +template +void BM_PforEncodeImpl(benchmark::State& state, + std::vector (*generator)(int64_t)) { + const int64_t num_values = state.range(0); + auto values = generator(num_values); + + int64_t max_size = PforWrapper::GetMaxCompressedSize(num_values); + std::vector compressed(max_size); + + for (auto _ : state) { + int64_t comp_size = max_size; + PforWrapper::Encode(values.data(), num_values, + compressed.data(), &comp_size); + benchmark::DoNotOptimize(comp_size); + benchmark::ClobberMemory(); + } + + state.SetBytesProcessed(state.iterations() * num_values * + static_cast(sizeof(T))); + state.SetItemsProcessed(state.iterations() * num_values); + + // Report compression ratio + int64_t comp_size = max_size; + PforWrapper::Encode(values.data(), num_values, compressed.data(), &comp_size); + state.counters["CompRatio%"] = benchmark::Counter( + 100.0 * static_cast(comp_size) / + static_cast(num_values * sizeof(T))); +} + +template +void BM_PforDecodeImpl(benchmark::State& state, + std::vector (*generator)(int64_t)) { + const int64_t num_values = state.range(0); + auto values = generator(num_values); + + int64_t max_size = PforWrapper::GetMaxCompressedSize(num_values); + std::vector compressed(max_size); + int64_t comp_size = max_size; + PforWrapper::Encode(values.data(), num_values, compressed.data(), &comp_size); + + std::vector decoded(num_values); + + for (auto _ : state) { + PforWrapper::Decode(decoded.data(), num_values, + compressed.data(), comp_size); + benchmark::ClobberMemory(); + } + + state.SetBytesProcessed(state.iterations() * num_values * + static_cast(sizeof(T))); + state.SetItemsProcessed(state.iterations() * num_values); +} + +// ====================================================================== +// Non-template wrappers to avoid comma-in-macro issues with BENCHMARK_CAPTURE + +void BM_PforEncodeInt32(benchmark::State& state, Int32Gen gen) { + BM_PforEncodeImpl(state, gen); +} +void BM_PforDecodeInt32(benchmark::State& state, Int32Gen gen) { + BM_PforDecodeImpl(state, gen); +} +void BM_PforEncodeInt64(benchmark::State& state, Int64Gen gen) { + BM_PforEncodeImpl(state, gen); +} +void BM_PforDecodeInt64(benchmark::State& state, Int64Gen gen) { + BM_PforDecodeImpl(state, gen); +} + +// ====================================================================== +// Benchmark sizes: 1K, 10K, 100K, 1M + +static void CustomArgs(benchmark::internal::Benchmark* b) { + for (int64_t n : {1024, 10240, 102400, 1048576}) { + b->Arg(n); + } +} + +// ====================================================================== +// INT32 Encode + +BENCHMARK_CAPTURE(BM_PforEncodeInt32, Constant, &GenConstant) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforEncodeInt32, Sequential, &GenSequential) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforEncodeInt32, SmallRange, &GenSmallRange) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforEncodeInt32, HighBaseSmallRange, + &GenHighBaseSmallRange) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforEncodeInt32, WithOutliers, &GenWithOutliers) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforEncodeInt32, Random, &GenRandom) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforEncodeInt32, TpcdsSoldDateSk, + &GenTpcdsSoldDateSk) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforEncodeInt32, TpcdsStoreSk, &GenTpcdsStoreSk) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforEncodeInt32, TpcdsItemSk, &GenTpcdsItemSk) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforEncodeInt32, TpcdsQuantity, + &GenTpcdsQuantity) + ->Apply(CustomArgs); + +// INT32 Decode + +BENCHMARK_CAPTURE(BM_PforDecodeInt32, Constant, &GenConstant) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforDecodeInt32, Sequential, &GenSequential) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforDecodeInt32, SmallRange, &GenSmallRange) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforDecodeInt32, HighBaseSmallRange, + &GenHighBaseSmallRange) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforDecodeInt32, WithOutliers, &GenWithOutliers) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforDecodeInt32, Random, &GenRandom) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforDecodeInt32, TpcdsSoldDateSk, + &GenTpcdsSoldDateSk) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforDecodeInt32, TpcdsStoreSk, &GenTpcdsStoreSk) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforDecodeInt32, TpcdsItemSk, &GenTpcdsItemSk) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforDecodeInt32, TpcdsQuantity, + &GenTpcdsQuantity) + ->Apply(CustomArgs); + +// ====================================================================== +// INT64 Encode + +BENCHMARK_CAPTURE(BM_PforEncodeInt64, Constant, &GenConstant) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforEncodeInt64, Sequential, &GenSequential) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforEncodeInt64, SmallRange, &GenSmallRange) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforEncodeInt64, HighBaseSmallRange, + &GenHighBaseSmallRange) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforEncodeInt64, WithOutliers, &GenWithOutliers) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforEncodeInt64, Random, &GenRandom) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforEncodeInt64, TpcdsSoldDateSk, + &GenTpcdsSoldDateSk) + ->Apply(CustomArgs); + +// INT64 Decode + +BENCHMARK_CAPTURE(BM_PforDecodeInt64, Constant, &GenConstant) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforDecodeInt64, Sequential, &GenSequential) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforDecodeInt64, SmallRange, &GenSmallRange) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforDecodeInt64, HighBaseSmallRange, + &GenHighBaseSmallRange) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforDecodeInt64, WithOutliers, &GenWithOutliers) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforDecodeInt64, Random, &GenRandom) + ->Apply(CustomArgs); +BENCHMARK_CAPTURE(BM_PforDecodeInt64, TpcdsSoldDateSk, + &GenTpcdsSoldDateSk) + ->Apply(CustomArgs); + +} // namespace +} // namespace arrow::util::pfor diff --git a/cpp/src/arrow/util/pfor/pfor_constants.h b/cpp/src/arrow/util/pfor/pfor_constants.h new file mode 100644 index 00000000000..40164e436e1 --- /dev/null +++ b/cpp/src/arrow/util/pfor/pfor_constants.h @@ -0,0 +1,132 @@ +// 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. + +// Constants for PFOR (Patched Frame of Reference) compression + +#pragma once + +#include +#include + +namespace arrow { +namespace util { +namespace pfor { + +/// \brief Constants used throughout PFOR compression +class PforConstants { + public: + /// Number of elements compressed together as a unit. + static constexpr int64_t kPforVectorSize = 1024; + + /// log2(kPforVectorSize) + static constexpr uint8_t kDefaultLogVectorSize = 10; + + /// Minimum allowed log vector size + static constexpr uint8_t kMinLogVectorSize = 3; + + /// Maximum allowed log vector size + static constexpr uint8_t kMaxLogVectorSize = 15; + + /// Type used to store vector data offsets (supports pages up to 4GB) + using OffsetType = uint32_t; + + /// Type used to store exception positions within a compressed vector. + using PositionType = int16_t; + + /// Page header size in bytes. + static constexpr int64_t kHeaderSize = 7; + + /// Packing mode: FOR + bit-packing (currently the only mode). + static constexpr uint8_t kPackingModeForBitPack = 0; + + /// Loop unroll factor for compiler hints in decode loops. + static constexpr int64_t kLoopUnrolls = 4; +}; + +/// \brief Per-vector packing mode for PforCompression::EncodeVector. +/// +/// Stored in the high bit of PforVectorInfo::bit_width (bit 7). BitPack = 0 +/// preserves the prior on-disk layout so existing buffers continue to +/// round-trip; new values are added with higher numeric codes. +enum class PackingMode : uint8_t { + /// Sequential little-endian bit-packed stream — the original PFOR layout. + /// Decoded with arrow::internal::unpack. + BitPack = 0, + + /// FastLanes lane-interleaved 1024-bit format (Afroozeh & Boncz, VLDB '23). + /// Auto-vectorizable kernel; only valid when num_elements equals the + /// FastLanes block size (1024). Falls back to BitPack for shorter vectors. + FastLanes = 1, +}; + +/// \brief Output value order requested by the decoder. +/// +/// Affects only FastLanes-encoded vectors (BitPack always returns flat). +/// Selecting `Transposed` skips the per-vector FL_ORDER gather and is the +/// only way to take full advantage of the FastLanes layout: the gather is +/// scalar and dominates end-to-end decode cost. Downstream operators must +/// be permutation-aware (work on values in stream order, i.e. apply +/// fromTransposed32(t) when they need the original index). +enum class OutputOrder : uint8_t { + /// Default. Decoded values appear in their original input order: + /// output[i] == input[i] for every i, for both PackingMode variants. + Flat = 0, + + /// FastLanes transposed (stream) order. For FastLanes-encoded vectors, + /// output[t] == input[fromTransposed32(t)] (per 1024-block, no scatter + /// on decode). For BitPack-encoded vectors there is no permutation, so + /// they still produce flat output. + Transposed = 1, +}; + +/// \brief Type traits for PFOR integer types +template +struct PforTypeTraits {}; + +template <> +struct PforTypeTraits { + using UnsignedType = uint32_t; + static constexpr uint8_t kMaxBitWidth = 32; + static constexpr uint8_t kValueByteWidth = 4; + + /// PforVectorInfo size: 4B FOR + 1B bitWidth + 2B numExceptions = 7 bytes + static constexpr int64_t kVectorInfoSize = 7; + + static uint8_t BitsRequired(uint32_t value) { + if (value == 0) return 0; + return static_cast(32 - __builtin_clz(value)); + } +}; + +template <> +struct PforTypeTraits { + using UnsignedType = uint64_t; + static constexpr uint8_t kMaxBitWidth = 64; + static constexpr uint8_t kValueByteWidth = 8; + + /// PforVectorInfo size: 8B FOR + 1B bitWidth + 2B numExceptions = 11 bytes + static constexpr int64_t kVectorInfoSize = 11; + + static uint8_t BitsRequired(uint64_t value) { + if (value == 0) return 0; + return static_cast(64 - __builtin_clzll(value)); + } +}; + +} // namespace pfor +} // namespace util +} // namespace arrow diff --git a/cpp/src/arrow/util/pfor/pfor_test.cc b/cpp/src/arrow/util/pfor/pfor_test.cc new file mode 100644 index 00000000000..40f173f7082 --- /dev/null +++ b/cpp/src/arrow/util/pfor/pfor_test.cc @@ -0,0 +1,665 @@ +// 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. + +#include + +#include +#include +#include +#include +#include +#include + +#include "arrow/testing/gtest_util.h" +#include "arrow/util/fastlanes/fastlanes_kernels.h" +#include "arrow/util/pfor/pfor.h" +#include "arrow/util/pfor/pfor_wrapper.h" +#include "arrow/util/span.h" + +namespace arrow::util::pfor { + +// ====================================================================== +// Constants Tests + +TEST(PforConstantsTest, VectorSizeIsPowerOfTwo) { + EXPECT_EQ(PforConstants::kPforVectorSize, 1024); + EXPECT_EQ(1 << PforConstants::kDefaultLogVectorSize, + PforConstants::kPforVectorSize); +} + +TEST(PforConstantsTest, VectorInfoSizes) { + EXPECT_EQ(PforTypeTraits::kVectorInfoSize, 7); + EXPECT_EQ(PforTypeTraits::kVectorInfoSize, 11); +} + +// ====================================================================== +// BitsRequired Tests + +TEST(PforBitsRequiredTest, Int32) { + EXPECT_EQ(PforTypeTraits::BitsRequired(0), 0); + EXPECT_EQ(PforTypeTraits::BitsRequired(1), 1); + EXPECT_EQ(PforTypeTraits::BitsRequired(2), 2); + EXPECT_EQ(PforTypeTraits::BitsRequired(3), 2); + EXPECT_EQ(PforTypeTraits::BitsRequired(255), 8); + EXPECT_EQ(PforTypeTraits::BitsRequired(256), 9); + EXPECT_EQ(PforTypeTraits::BitsRequired(0xFFFFFFFF), 32); +} + +TEST(PforBitsRequiredTest, Int64) { + EXPECT_EQ(PforTypeTraits::BitsRequired(0), 0); + EXPECT_EQ(PforTypeTraits::BitsRequired(1), 1); + EXPECT_EQ(PforTypeTraits::BitsRequired(0xFFFFFFFFFFFFFFFFULL), 64); +} + +// ====================================================================== +// VectorInfo Serialization Tests + +TEST(PforVectorInfoTest, Int32RoundTrip) { + PforVectorInfo info; + info.set_frame_of_reference(-42); + info.set_bit_width(17); + info.set_num_exceptions(300); + + uint8_t buf[7]; + info.Store(arrow::util::span(buf, 7)); + ASSERT_OK_AND_ASSIGN(auto loaded, + PforVectorInfo::Load(arrow::util::span(buf, 7))); + + EXPECT_EQ(loaded.frame_of_reference(), -42); + EXPECT_EQ(loaded.bit_width(), 17); + EXPECT_EQ(loaded.num_exceptions(), 300); +} + +TEST(PforVectorInfoTest, Int64RoundTrip) { + PforVectorInfo info; + info.set_frame_of_reference(-123456789012345LL); + info.set_bit_width(48); + info.set_num_exceptions(30000); + + uint8_t buf[11]; + info.Store(arrow::util::span(buf, 11)); + ASSERT_OK_AND_ASSIGN(auto loaded, + PforVectorInfo::Load(arrow::util::span(buf, 11))); + + EXPECT_EQ(loaded.frame_of_reference(), -123456789012345LL); + EXPECT_EQ(loaded.bit_width(), 48); + EXPECT_EQ(loaded.num_exceptions(), 30000); +} + +// ====================================================================== +// Cost Model Tests + +TEST(PforCostModelTest, AllIdentical) { + // All deltas are 0 => bit_width should be 0, no exceptions + std::vector deltas(100, 0); + auto result = PforCompression::FindOptimalBitWidth(deltas.data(), 100); // NOLINT + EXPECT_EQ(result.bit_width, 0); + EXPECT_EQ(result.num_exceptions, 0); +} + +TEST(PforCostModelTest, SingleOutlier) { + // 99 values fit in 3 bits, 1 outlier needs 16 bits + std::vector deltas(100, 5); // all fit in 3 bits + deltas[50] = 50000; // outlier: 16 bits + auto result = PforCompression::FindOptimalBitWidth(deltas.data(), 100); + // Cost at bit_width=3: 100*3 + 1*(16+32) = 300 + 48 = 348 + // Cost at bit_width=16: 100*16 + 0 = 1600 + // 348 < 1600, so should pick 3 with 1 exception + EXPECT_EQ(result.bit_width, 3); + EXPECT_EQ(result.num_exceptions, 1); +} + +TEST(PforCostModelTest, NoOutliers) { + // All values fit in 8 bits + std::vector deltas(100); + for (int32_t i = 0; i < 100; ++i) deltas[i] = i * 2; + auto result = PforCompression::FindOptimalBitWidth(deltas.data(), 100); + EXPECT_EQ(result.num_exceptions, 0); + EXPECT_LE(result.bit_width, 8); +} + +// ====================================================================== +// Vector Encode/Decode Round-Trip Tests + +TEST(PforVectorTest, Int32SimpleSequence) { + std::vector values(64); + std::iota(values.begin(), values.end(), 100); + + auto encoded = PforCompression::EncodeVector(values.data(), 64); + EXPECT_EQ(encoded.info().frame_of_reference(), 100); + EXPECT_EQ(encoded.info().num_exceptions(), 0); + + // Serialize then decode + size_t serialized_size = + PforCompression::SerializedVectorSize(encoded, 64); + std::vector buffer(serialized_size); + PforCompression::SerializeVector(encoded, 64, buffer); + + std::vector decoded(64); + ASSERT_OK(PforCompression::DecodeVector(decoded.data(), buffer, 64)); + + EXPECT_EQ(values, decoded); +} + +TEST(PforVectorTest, Int32WithOutlier) { + std::vector values = {100, 102, 101, 103, 100, 99, 50000, 104}; + + auto encoded = PforCompression::EncodeVector(values.data(), 8); + EXPECT_EQ(encoded.info().frame_of_reference(), 99); + EXPECT_GT(encoded.info().num_exceptions(), 0); + + size_t serialized_size = + PforCompression::SerializedVectorSize(encoded, 8); + std::vector buffer(serialized_size); + PforCompression::SerializeVector(encoded, 8, buffer); + + std::vector decoded(8); + ASSERT_OK(PforCompression::DecodeVector(decoded.data(), buffer, 8)); + + EXPECT_EQ(values, decoded); +} + +TEST(PforVectorTest, Int32AllIdentical) { + std::vector values(100, 42); + + auto encoded = PforCompression::EncodeVector(values.data(), 100); + EXPECT_EQ(encoded.info().bit_width(), 0); + EXPECT_EQ(encoded.info().num_exceptions(), 0); + + size_t serialized_size = + PforCompression::SerializedVectorSize(encoded, 100); + std::vector buffer(serialized_size); + PforCompression::SerializeVector(encoded, 100, buffer); + + std::vector decoded(100); + ASSERT_OK(PforCompression::DecodeVector(decoded.data(), buffer, 100)); + + EXPECT_EQ(values, decoded); +} + +TEST(PforVectorTest, Int32NegativeValues) { + std::vector values = {-100, -50, -200, -1, -150}; + + auto encoded = PforCompression::EncodeVector(values.data(), 5); + EXPECT_EQ(encoded.info().frame_of_reference(), -200); + + size_t serialized_size = + PforCompression::SerializedVectorSize(encoded, 5); + std::vector buffer(serialized_size); + PforCompression::SerializeVector(encoded, 5, buffer); + + std::vector decoded(5); + ASSERT_OK(PforCompression::DecodeVector(decoded.data(), buffer, 5)); + + EXPECT_EQ(values, decoded); +} + +TEST(PforVectorTest, Int32MinMaxEdge) { + std::vector values = {std::numeric_limits::min(), + std::numeric_limits::max(), 0, -1, 1}; + + auto encoded = PforCompression::EncodeVector(values.data(), 5); + + size_t serialized_size = + PforCompression::SerializedVectorSize(encoded, 5); + std::vector buffer(serialized_size); + PforCompression::SerializeVector(encoded, 5, buffer); + + std::vector decoded(5); + ASSERT_OK(PforCompression::DecodeVector(decoded.data(), buffer, 5)); + + EXPECT_EQ(values, decoded); +} + +TEST(PforVectorTest, Int64SimpleSequence) { + std::vector values(64); + std::iota(values.begin(), values.end(), 1000000000LL); + + auto encoded = PforCompression::EncodeVector(values.data(), 64); + + size_t serialized_size = + PforCompression::SerializedVectorSize(encoded, 64); + std::vector buffer(serialized_size); + PforCompression::SerializeVector(encoded, 64, buffer); + + std::vector decoded(64); + ASSERT_OK(PforCompression::DecodeVector(decoded.data(), buffer, 64)); + + EXPECT_EQ(values, decoded); +} + +TEST(PforVectorTest, Int64WithOutlier) { + std::vector values(100, 5000000LL); + values[42] = 999999999999LL; // Outlier + + auto encoded = PforCompression::EncodeVector(values.data(), 100); + EXPECT_GT(encoded.info().num_exceptions(), 0); + + size_t serialized_size = + PforCompression::SerializedVectorSize(encoded, 100); + std::vector buffer(serialized_size); + PforCompression::SerializeVector(encoded, 100, buffer); + + std::vector decoded(100); + ASSERT_OK(PforCompression::DecodeVector(decoded.data(), buffer, 100)); + + EXPECT_EQ(values, decoded); +} + +// ====================================================================== +// Page-Level Wrapper Tests + +TEST(PforWrapperTest, Int32SmallPage) { + std::vector values = {10, 20, 30, 40, 50}; + + int64_t max_size = PforWrapper::GetMaxCompressedSize(5); + std::vector compressed(max_size); + int64_t comp_size = max_size; + + PforWrapper::Encode(values.data(), 5, compressed.data(), &comp_size); + EXPECT_GT(comp_size, 0); + + std::vector decoded(5); + ASSERT_OK(PforWrapper::Decode(decoded.data(), 5, compressed.data(), comp_size)); + + EXPECT_EQ(values, decoded); +} + +TEST(PforWrapperTest, Int32ExactOneVector) { + std::vector values(1024); + std::iota(values.begin(), values.end(), 0); + + int64_t max_size = PforWrapper::GetMaxCompressedSize(1024); + std::vector compressed(max_size); + int64_t comp_size = max_size; + + PforWrapper::Encode(values.data(), 1024, compressed.data(), &comp_size); + + std::vector decoded(1024); + ASSERT_OK(PforWrapper::Decode(decoded.data(), 1024, compressed.data(), comp_size)); + + EXPECT_EQ(values, decoded); +} + +TEST(PforWrapperTest, Int32MultipleVectors) { + // 2.5 vectors worth of data + const int32_t n = 2560; + std::vector values(n); + std::mt19937 rng(42); + std::uniform_int_distribution dist(0, 1000); + for (auto& v : values) v = dist(rng); + + int64_t max_size = PforWrapper::GetMaxCompressedSize(n); + std::vector compressed(max_size); + int64_t comp_size = max_size; + + PforWrapper::Encode(values.data(), n, compressed.data(), &comp_size); + + std::vector decoded(n); + ASSERT_OK(PforWrapper::Decode(decoded.data(), n, compressed.data(), comp_size)); + + EXPECT_EQ(values, decoded); +} + +TEST(PforWrapperTest, Int32WithOutliers) { + std::vector values(1024, 100); + // Sprinkle outliers + values[0] = -999999; + values[100] = 888888; + values[500] = 777777; + values[1023] = -123456; + + int64_t max_size = PforWrapper::GetMaxCompressedSize(1024); + std::vector compressed(max_size); + int64_t comp_size = max_size; + + PforWrapper::Encode(values.data(), 1024, compressed.data(), &comp_size); + + std::vector decoded(1024); + ASSERT_OK(PforWrapper::Decode(decoded.data(), 1024, compressed.data(), comp_size)); + + EXPECT_EQ(values, decoded); +} + +TEST(PforWrapperTest, Int64MultipleVectors) { + const int32_t n = 3000; + std::vector values(n); + std::mt19937 rng(123); + std::uniform_int_distribution dist(0, 100000); + for (auto& v : values) v = dist(rng); + // Add outliers + values[0] = 9999999999999LL; + values[1500] = -9999999999999LL; + + int64_t max_size = PforWrapper::GetMaxCompressedSize(n); + std::vector compressed(max_size); + int64_t comp_size = max_size; + + PforWrapper::Encode(values.data(), n, compressed.data(), &comp_size); + + std::vector decoded(n); + ASSERT_OK(PforWrapper::Decode(decoded.data(), n, compressed.data(), comp_size)); + + EXPECT_EQ(values, decoded); +} + +TEST(PforWrapperTest, Int32SingleElement) { + std::vector values = {42}; + + int64_t max_size = PforWrapper::GetMaxCompressedSize(1); + std::vector compressed(max_size); + int64_t comp_size = max_size; + + PforWrapper::Encode(values.data(), 1, compressed.data(), &comp_size); + + std::vector decoded(1); + ASSERT_OK(PforWrapper::Decode(decoded.data(), 1, compressed.data(), comp_size)); + + EXPECT_EQ(values, decoded); +} + +TEST(PforWrapperTest, Int32AllZeros) { + std::vector values(1024, 0); + + int64_t max_size = PforWrapper::GetMaxCompressedSize(1024); + std::vector compressed(max_size); + int64_t comp_size = max_size; + + PforWrapper::Encode(values.data(), 1024, compressed.data(), &comp_size); + + // Should compress very well (bit_width = 0) + EXPECT_LT(comp_size, 100); + + std::vector decoded(1024); + ASSERT_OK(PforWrapper::Decode(decoded.data(), 1024, compressed.data(), comp_size)); + + EXPECT_EQ(values, decoded); +} + +TEST(PforWrapperTest, Int32LargeRandom) { + const int32_t n = 10000; + std::vector values(n); + std::mt19937 rng(99); + std::uniform_int_distribution dist( + std::numeric_limits::min(), + std::numeric_limits::max()); + for (auto& v : values) v = dist(rng); + + int64_t max_size = PforWrapper::GetMaxCompressedSize(n); + std::vector compressed(max_size); + int64_t comp_size = max_size; + + PforWrapper::Encode(values.data(), n, compressed.data(), &comp_size); + + std::vector decoded(n); + ASSERT_OK(PforWrapper::Decode(decoded.data(), n, compressed.data(), comp_size)); + + EXPECT_EQ(values, decoded); +} + +// ====================================================================== +// Compression Ratio Test + +TEST(PforCompressionRatioTest, ClusteredDataCompresses) { + // Data clustered around 1000 with one outlier + std::vector values(1024); + std::mt19937 rng(42); + std::uniform_int_distribution dist(1000, 1255); + for (auto& v : values) v = dist(rng); + values[500] = 999999; // One outlier + + int64_t max_size = PforWrapper::GetMaxCompressedSize(1024); + std::vector compressed(max_size); + int64_t comp_size = max_size; + + PforWrapper::Encode(values.data(), 1024, compressed.data(), &comp_size); + + // PLAIN would be 4096 bytes. PFOR should be much smaller. + size_t plain_size = 1024 * sizeof(int32_t); + EXPECT_LT(comp_size, plain_size / 2); +} + +// ============================================================================ +// FastLanes packing-mode tests +// ============================================================================ +// +// All round-trips must produce flat output identical to the input — the +// FastLanes scatter on decode is the inverse of the gather on encode, so the +// PFOR contract (flat output, no permutation visible to the caller) holds. + +TEST(PforPackingModeTest, VectorRoundTripIdentityForBothModes) { + std::vector values(1024); + std::mt19937 rng(7); + std::uniform_int_distribution dist(1000, 1500); + for (auto& v : values) v = dist(rng); + values[123] = 99999; // outlier — exercises the exception path + values[800] = -50; // another, below FOR + + for (PackingMode mode : {PackingMode::BitPack, PackingMode::FastLanes}) { + SCOPED_TRACE(testing::Message() << "mode=" << static_cast(mode)); + auto encoded = + PforCompression::EncodeVector(values.data(), 1024, mode); + EXPECT_EQ(encoded.info().packing_mode(), mode); + + int64_t sz = PforCompression::SerializedVectorSize(encoded, 1024); + std::vector buffer(sz); + PforCompression::SerializeVector(encoded, 1024, buffer); + + std::vector decoded(1024); + ASSERT_OK(PforCompression::DecodeVector(decoded.data(), buffer, 1024)); + + EXPECT_EQ(values, decoded); + } +} + +TEST(PforPackingModeTest, FastLanesFallsBackOnPartialTail) { + // 700-element vector — smaller than the FastLanes block size. Encoder + // must fall back to BitPack regardless of the requested mode so the + // legacy bit-packer handles the partial tail. + std::vector values(700); + std::mt19937 rng(99); + std::uniform_int_distribution dist(0, 100); + for (auto& v : values) v = dist(rng); + + auto encoded = + PforCompression::EncodeVector(values.data(), 700, PackingMode::FastLanes); + EXPECT_EQ(encoded.info().packing_mode(), PackingMode::BitPack); + + int64_t sz = PforCompression::SerializedVectorSize(encoded, 700); + std::vector buffer(sz); + PforCompression::SerializeVector(encoded, 700, buffer); + + std::vector decoded(700); + ASSERT_OK(PforCompression::DecodeVector(decoded.data(), buffer, 700)); + EXPECT_EQ(values, decoded); +} + +TEST(PforPackingModeTest, WrapperRoundTripFastLanes) { + // PforWrapper covers multiple vectors plus the offset array. + constexpr int32_t kN = 5 * 1024; // 5 full FastLanes vectors + std::vector values(kN); + std::mt19937 rng(2026); + std::uniform_int_distribution dist(-100, 100); + for (auto& v : values) v = dist(rng); + + int64_t max_size = PforWrapper::GetMaxCompressedSize(kN); + std::vector compressed(max_size); + int64_t comp_size = max_size; + + PforWrapper::Encode(values.data(), kN, compressed.data(), &comp_size, + PackingMode::FastLanes); + + std::vector decoded(kN); + ASSERT_OK(PforWrapper::Decode(decoded.data(), kN, compressed.data(), + comp_size)); + EXPECT_EQ(values, decoded); +} + +TEST(PforPackingModeTest, WrapperRoundTripMixedTail) { + // 5*1024 + 700 elements: 5 full vectors (FastLanes-packed when requested) + // + 1 tail vector (falls back to BitPack inside EncodeVector). Both modes + // round-trip via the per-vector flag. + constexpr int32_t kN = 5 * 1024 + 700; + std::vector values(kN); + std::mt19937 rng(123); + std::uniform_int_distribution dist(50000, 60000); + for (auto& v : values) v = dist(rng); + + int64_t max_size = PforWrapper::GetMaxCompressedSize(kN); + std::vector compressed(max_size); + int64_t comp_size = max_size; + + PforWrapper::Encode(values.data(), kN, compressed.data(), &comp_size, + PackingMode::FastLanes); + + std::vector decoded(kN); + ASSERT_OK(PforWrapper::Decode(decoded.data(), kN, compressed.data(), + comp_size)); + EXPECT_EQ(values, decoded); +} + +TEST(PforPackingModeTest, ConstantVectorBothModes) { + // bit_width = 0 path: FOR captures everything, no packed payload. Should + // round-trip identically under either mode (the FastLanes kernel is never + // entered because bit_width == 0). + std::vector values(1024, 42); + for (PackingMode mode : {PackingMode::BitPack, PackingMode::FastLanes}) { + SCOPED_TRACE(testing::Message() << "mode=" << static_cast(mode)); + auto encoded = + PforCompression::EncodeVector(values.data(), 1024, mode); + EXPECT_EQ(encoded.info().bit_width(), 0); + + int64_t sz = PforCompression::SerializedVectorSize(encoded, 1024); + std::vector buffer(sz); + PforCompression::SerializeVector(encoded, 1024, buffer); + + std::vector decoded(1024); + ASSERT_OK(PforCompression::DecodeVector(decoded.data(), buffer, 1024)); + EXPECT_EQ(values, decoded); + } +} + +// ============================================================================ +// OutputOrder::Transposed tests — verify the transposed-output decode path +// matches the FL_ORDER permutation of the input for FastLanes vectors. +// ============================================================================ + +TEST(PforOutputOrderTest, TransposedMatchesFLOrder) { + std::vector values(1024); + std::mt19937 rng(11); + std::uniform_int_distribution dist(1000, 1500); + for (auto& v : values) v = dist(rng); + values[7] = 99999; // outlier — exercises the exception path under transposed + values[412] = -50; + + auto encoded = PforCompression::EncodeVector( + values.data(), 1024, PackingMode::FastLanes); + ASSERT_EQ(encoded.info().packing_mode(), PackingMode::FastLanes); + + int64_t sz = PforCompression::SerializedVectorSize(encoded, 1024); + std::vector buffer(sz); + PforCompression::SerializeVector(encoded, 1024, buffer); + + std::vector decoded(1024); + ASSERT_OK(PforCompression::DecodeVector(decoded.data(), buffer, 1024, + OutputOrder::Transposed)); + + // out[t] == input[fromTransposed32(t)] for every stream position t. + for (size_t t = 0; t < 1024; ++t) { + ASSERT_EQ(decoded[t], + values[fastlanes::fromTransposed32(t)]) + << "t=" << t; + } +} + +TEST(PforOutputOrderTest, TransposedThenInvertEqualsInput) { + // Belt-and-suspenders: the user code path is "decode transposed, then + // apply fromTransposed32 myself when I need original-order positions". + // Confirm that inverting the permutation manually produces the input. + std::vector values(1024); + std::mt19937 rng(31); + std::uniform_int_distribution dist(-200, 200); + for (auto& v : values) v = dist(rng); + + auto encoded = PforCompression::EncodeVector( + values.data(), 1024, PackingMode::FastLanes); + + int64_t sz = PforCompression::SerializedVectorSize(encoded, 1024); + std::vector buffer(sz); + PforCompression::SerializeVector(encoded, 1024, buffer); + + std::vector transposed(1024); + ASSERT_OK(PforCompression::DecodeVector(transposed.data(), buffer, 1024, + OutputOrder::Transposed)); + + std::vector flat(1024); + for (size_t t = 0; t < 1024; ++t) { + flat[fastlanes::fromTransposed32(t)] = transposed[t]; + } + EXPECT_EQ(values, flat); +} + +TEST(PforOutputOrderTest, BitPackIgnoresTransposedRequest) { + // BitPack mode has no transposition; OutputOrder::Transposed must be + // ignored (output is still flat / identity). + std::vector values(1024); + std::iota(values.begin(), values.end(), 5000); + + auto encoded = PforCompression::EncodeVector( + values.data(), 1024, PackingMode::BitPack); + ASSERT_EQ(encoded.info().packing_mode(), PackingMode::BitPack); + + int64_t sz = PforCompression::SerializedVectorSize(encoded, 1024); + std::vector buffer(sz); + PforCompression::SerializeVector(encoded, 1024, buffer); + + std::vector decoded(1024); + ASSERT_OK(PforCompression::DecodeVector(decoded.data(), buffer, 1024, + OutputOrder::Transposed)); + EXPECT_EQ(values, decoded); +} + +TEST(PforOutputOrderTest, WrapperTransposedAcrossManyVectors) { + constexpr int32_t kN = 5 * 1024; + std::vector values(kN); + std::mt19937 rng(909); + std::uniform_int_distribution dist(-1000, 1000); + for (auto& v : values) v = dist(rng); + + int64_t max_size = PforWrapper::GetMaxCompressedSize(kN); + std::vector compressed(max_size); + int64_t comp_size = max_size; + + PforWrapper::Encode(values.data(), kN, compressed.data(), &comp_size, + PackingMode::FastLanes); + + std::vector decoded(kN); + ASSERT_OK(PforWrapper::Decode(decoded.data(), kN, compressed.data(), + comp_size, OutputOrder::Transposed)); + + // Verify per-block: decoded[block*1024 + t] == values[block*1024 + fromTransposed32(t)] + for (int32_t block = 0; block < 5; ++block) { + for (size_t t = 0; t < 1024; ++t) { + const int32_t base = block * 1024; + ASSERT_EQ(decoded[base + t], + values[base + fastlanes::fromTransposed32(t)]) + << "block=" << block << " t=" << t; + } + } +} + +} // namespace arrow::util::pfor diff --git a/cpp/src/arrow/util/pfor/pfor_wrapper.cc b/cpp/src/arrow/util/pfor/pfor_wrapper.cc new file mode 100644 index 00000000000..0774dfe74b3 --- /dev/null +++ b/cpp/src/arrow/util/pfor/pfor_wrapper.cc @@ -0,0 +1,230 @@ +// 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. + +// PFOR page-level wrapper implementation +// +// Page layout: +// [Header 7B] [Offset Array: numVectors * 4B] [Vector 0] [Vector 1] ... +// +// Each vector: +// [PforVectorInfo] [PackedValues] [ExceptionPositions] [ExceptionValues] + +#include "arrow/util/pfor/pfor_wrapper.h" + +#include +#include +#include + +#include "arrow/util/bit_util.h" +#include "arrow/util/logging.h" +#include "arrow/util/span.h" +#include "arrow/util/ubsan.h" + +namespace arrow { +namespace util { +namespace pfor { + +// ---------------------------------------------------------------------- +// Header serialization + +template +void PforWrapper::StoreHeader(arrow::util::span dest, + const PforHeader& header) { + uint8_t* ptr = dest.data(); + util::SafeStore(ptr + 0, header.packing_mode); + util::SafeStore(ptr + 1, header.log_vector_size); + util::SafeStore(ptr + 2, header.value_byte_width); + util::SafeStore(ptr + 3, header.num_elements); +} + +template +Result::PforHeader> PforWrapper::LoadHeader( + arrow::util::span src) { + if (src.size() < static_cast(PforConstants::kHeaderSize)) { + return Status::Invalid("PFOR compressed buffer too small for header: ", + src.size(), " < ", PforConstants::kHeaderSize); + } + PforHeader header; + const uint8_t* ptr = src.data(); + header.packing_mode = util::SafeLoadAs(ptr + 0); + header.log_vector_size = util::SafeLoadAs(ptr + 1); + header.value_byte_width = util::SafeLoadAs(ptr + 2); + header.num_elements = util::SafeLoadAs(ptr + 3); + + if (header.packing_mode != PforConstants::kPackingModeForBitPack) { + return Status::Invalid("PFOR unsupported packing mode: ", + static_cast(header.packing_mode)); + } + if (header.value_byte_width != sizeof(T)) { + return Status::Invalid("PFOR value_byte_width mismatch: ", + static_cast(header.value_byte_width), + " vs expected ", sizeof(T)); + } + if (header.log_vector_size < PforConstants::kMinLogVectorSize || + header.log_vector_size > PforConstants::kMaxLogVectorSize) { + return Status::Invalid("PFOR invalid log_vector_size: ", + static_cast(header.log_vector_size)); + } + if (header.num_elements < 0) { + return Status::Invalid("PFOR invalid num_elements: ", header.num_elements); + } + return header; +} + +// ---------------------------------------------------------------------- +// Encode + +template +void PforWrapper::Encode(const T* values, int32_t num_values, int32_t vector_size, + uint8_t* comp, int64_t* comp_size, PackingMode mode) { + ARROW_DCHECK(num_values > 0); + ARROW_DCHECK(comp != nullptr); + ARROW_DCHECK(comp_size != nullptr); + ARROW_DCHECK(bit_util::IsPowerOf2(static_cast(vector_size))); + + const int32_t num_vectors = + static_cast(bit_util::CeilDiv(num_values, vector_size)); + + // Compute log2(vector_size) + uint8_t log_vector_size = 0; + for (int32_t v = vector_size; v > 1; v >>= 1) ++log_vector_size; + + uint8_t* dest = comp; + + // Step 1: Write header + PforHeader header; + header.packing_mode = PforConstants::kPackingModeForBitPack; + header.log_vector_size = log_vector_size; + header.value_byte_width = sizeof(T); + header.num_elements = num_values; + StoreHeader(arrow::util::span(dest, PforConstants::kHeaderSize), header); + uint8_t* write_ptr = dest + PforConstants::kHeaderSize; + + // Step 2: Reserve space for offset array + uint8_t* offset_array_start = write_ptr; + write_ptr += num_vectors * sizeof(uint32_t); + + // Step 3: Encode each vector and build offset array + const uint8_t* data_start = offset_array_start; + + for (int32_t v = 0; v < num_vectors; ++v) { + // Record offset (from start of offset array) + uint32_t offset = static_cast(write_ptr - data_start); + util::SafeStore(offset_array_start + v * sizeof(uint32_t), offset); + + // Determine elements in this vector + int32_t start_idx = v * vector_size; + int32_t elements_in_vector = + std::min(vector_size, num_values - start_idx); + + // Encode vector (EncodeVector itself falls back to BitPack for partial + // tails or 64-bit T, so it's safe to pass `mode` unconditionally). + auto encoded = PforCompression::EncodeVector( + values + start_idx, elements_in_vector, mode); + + // Serialize to output + int64_t bytes_written = PforCompression::SerializeVector( + encoded, elements_in_vector, + arrow::util::span(write_ptr, dest + *comp_size - write_ptr)); + write_ptr += bytes_written; + } + + *comp_size = static_cast(write_ptr - dest); +} + +template +void PforWrapper::Encode(const T* values, int32_t num_values, uint8_t* comp, + int64_t* comp_size, PackingMode mode) { + Encode(values, num_values, kVectorSize, comp, comp_size, mode); +} + +// ---------------------------------------------------------------------- +// Decode + +template +Status PforWrapper::Decode(T* values, int32_t num_values, const uint8_t* comp, + int64_t comp_size, OutputOrder order) { + if (num_values <= 0) { + return Status::Invalid("PFOR num_values must be positive: ", num_values); + } + if (comp == nullptr) { + return Status::Invalid("PFOR compressed data pointer is null"); + } + + const uint8_t* src = comp; + + // Step 1: Read header + ARROW_ASSIGN_OR_RAISE( + PforHeader header, + LoadHeader(arrow::util::span(src, comp_size))); + + const int32_t vector_size = 1 << header.log_vector_size; + const int32_t num_vectors = + static_cast(bit_util::CeilDiv(header.num_elements, vector_size)); + + // Step 2: Read offset array + const uint8_t* offset_array_start = src + PforConstants::kHeaderSize; + + // Step 3: Decode each vector + for (int32_t v = 0; v < num_vectors; ++v) { + uint32_t offset = + util::SafeLoadAs(offset_array_start + v * sizeof(uint32_t)); + + const uint8_t* vector_data = offset_array_start + offset; + + int32_t start_idx = v * vector_size; + int32_t elements_in_vector = + std::min(vector_size, header.num_elements - start_idx); + + ARROW_RETURN_NOT_OK(PforCompression::DecodeVector( + values + start_idx, + arrow::util::span(vector_data, src + comp_size - vector_data), + elements_in_vector, order)); + } + + return Status::OK(); +} + +// ---------------------------------------------------------------------- +// GetMaxCompressedSize + +template +int64_t PforWrapper::GetMaxCompressedSize(int32_t num_values, int32_t vector_size) { + const int32_t num_vectors = + static_cast(bit_util::CeilDiv(num_values, vector_size)); + + // Header + offset array + int64_t size = PforConstants::kHeaderSize + + num_vectors * static_cast(sizeof(uint32_t)); + + // Worst case per vector: full bit width + all exceptions + int64_t max_vector_size = PforVectorInfo::kStoredSize + + vector_size * static_cast(sizeof(T)) // packed at full width + + vector_size * static_cast(sizeof(int16_t)) // exception positions + + vector_size * static_cast(sizeof(T)); // exception values + + size += num_vectors * max_vector_size; + return size; +} + +// Explicit template instantiations +template class PforWrapper; +template class PforWrapper; + +} // namespace pfor +} // namespace util +} // namespace arrow diff --git a/cpp/src/arrow/util/pfor/pfor_wrapper.h b/cpp/src/arrow/util/pfor/pfor_wrapper.h new file mode 100644 index 00000000000..d092a7624d9 --- /dev/null +++ b/cpp/src/arrow/util/pfor/pfor_wrapper.h @@ -0,0 +1,110 @@ +// 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. + +// High-level wrapper interface for PFOR compression +// +// Handles page-level serialization: header, offset array, and vectors. + +#pragma once + +#include +#include + +#include "arrow/result.h" +#include "arrow/status.h" +#include "arrow/util/pfor/pfor.h" +#include "arrow/util/span.h" + +namespace arrow { +namespace util { +namespace pfor { + +/// \class PforWrapper +/// \brief High-level interface for PFOR page-level compression +/// +/// Manages the page layout: [Header 7B] [Offset Array] [Vector 0] [Vector 1] ... +/// +/// \tparam T the integer type (int32_t or int64_t) +template +class PforWrapper { + public: + /// \brief Encode integer values into a PFOR-compressed page + /// + /// \param[in] values pointer to input integers + /// \param[in] num_values total number of values + /// \param[in] vector_size number of elements per vector (must be a power of 2, + /// in [2^kMinLogVectorSize, 2^kMaxLogVectorSize]) + /// \param[out] comp pointer to output buffer (caller must ensure sufficient size) + /// \param[in,out] comp_size input: available buffer size; output: bytes written + /// \param[in] mode per-vector bit-packing layout. PackingMode::FastLanes is + /// applied only to full-size 32-bit vectors; tails and 64-bit + /// values fall back to PackingMode::BitPack per vector. + static void Encode(const T* values, int32_t num_values, int32_t vector_size, + uint8_t* comp, int64_t* comp_size, + PackingMode mode = PackingMode::BitPack); + + /// Convenience overload with default vector_size = kPforVectorSize + static void Encode(const T* values, int32_t num_values, uint8_t* comp, + int64_t* comp_size, + PackingMode mode = PackingMode::BitPack); + + /// \brief Decode a PFOR-compressed page + /// + /// \param[out] values pointer to output buffer + /// \param[in] num_values number of values to decode (from page context) + /// \param[in] comp pointer to compressed data + /// \param[in] comp_size size of compressed data + /// \param[in] order output value order. Default OutputOrder::Flat returns + /// values in their original input positions. OutputOrder:: + /// Transposed only affects FastLanes-encoded vectors (skips + /// their FL_ORDER gather on decode); BitPack vectors always + /// produce flat output regardless of `order`, so a mixed page + /// will have BitPack tails in flat order and FastLanes vectors + /// in transposed order. + /// \return Status::OK on success, or an error if the data is malformed + static Status Decode(T* values, int32_t num_values, const uint8_t* comp, + int64_t comp_size, + OutputOrder order = OutputOrder::Flat); + + /// \brief Get the maximum compressed size for a given number of values + /// + /// \param[in] num_values number of integer values + /// \param[in] vector_size number of elements per vector + /// \return maximum possible compressed page size in bytes + static int64_t GetMaxCompressedSize( + int32_t num_values, + int32_t vector_size = static_cast(PforConstants::kPforVectorSize)); + + private: + /// \brief Page header structure (7 bytes) + struct PforHeader { + uint8_t packing_mode; // 0 = FOR + bit-packing + uint8_t log_vector_size; // log2(vector_size) + uint8_t value_byte_width; // sizeof(T): 4 or 8 + int32_t num_elements; // total element count + }; + + static constexpr int32_t kVectorSize = + static_cast(PforConstants::kPforVectorSize); + + static void StoreHeader(arrow::util::span dest, const PforHeader& header); + static Result LoadHeader(arrow::util::span src); +}; + +} // namespace pfor +} // namespace util +} // namespace arrow diff --git a/cpp/src/parquet/CMakeLists.txt b/cpp/src/parquet/CMakeLists.txt index f8a42b5b96b..12194796dc5 100644 --- a/cpp/src/parquet/CMakeLists.txt +++ b/cpp/src/parquet/CMakeLists.txt @@ -191,7 +191,9 @@ set(PARQUET_SRCS statistics.cc stream_reader.cc stream_writer.cc - types.cc) + types.cc + ${CMAKE_SOURCE_DIR}/src/arrow/util/pfor/pfor.cc + ${CMAKE_SOURCE_DIR}/src/arrow/util/pfor/pfor_wrapper.cc) if(ARROW_HAVE_RUNTIME_AVX2) # AVX2 is used as a proxy for BMI2. @@ -452,5 +454,14 @@ add_parquet_benchmark(level_conversion_benchmark) add_parquet_benchmark(metadata_benchmark) add_parquet_benchmark(page_index_benchmark SOURCES page_index_benchmark.cc benchmark_util.cc) +if(ARROW_BUILD_BENCHMARKS) +add_executable(parquet-pfor-comparison-benchmark + pfor_comparison_benchmark.cc + ${CMAKE_SOURCE_DIR}/src/arrow/util/pfor/pfor.cc + ${CMAKE_SOURCE_DIR}/src/arrow/util/pfor/pfor_wrapper.cc + ${CMAKE_SOURCE_DIR}/src/arrow/util/fastlanes/fastlanes_for.cc) +target_link_libraries(parquet-pfor-comparison-benchmark + PRIVATE parquet_static benchmark::benchmark_main) +endif() add_parquet_benchmark(arrow/reader_writer_benchmark PREFIX "parquet-arrow") add_parquet_benchmark(arrow/size_stats_benchmark PREFIX "parquet-arrow") diff --git a/cpp/src/parquet/column_reader.cc b/cpp/src/parquet/column_reader.cc index 162a72bd157..b2d5c18630c 100644 --- a/cpp/src/parquet/column_reader.cc +++ b/cpp/src/parquet/column_reader.cc @@ -971,7 +971,8 @@ class ColumnReaderImplBase { case Encoding::RLE: case Encoding::DELTA_BINARY_PACKED: case Encoding::DELTA_BYTE_ARRAY: - case Encoding::DELTA_LENGTH_BYTE_ARRAY: { + case Encoding::DELTA_LENGTH_BYTE_ARRAY: + case Encoding::PFOR: { auto decoder = MakeTypedDecoder(encoding, descr_, pool_); current_decoder_.SetDecoder(decoder.get()); decoders_[static_cast(encoding)] = std::move(decoder); diff --git a/cpp/src/parquet/decoder.cc b/cpp/src/parquet/decoder.cc index c4d3fe5a8a5..53e8124341d 100644 --- a/cpp/src/parquet/decoder.cc +++ b/cpp/src/parquet/decoder.cc @@ -46,6 +46,7 @@ #include "arrow/util/logging_internal.h" #include "arrow/util/rle_encoding_internal.h" #include "arrow/util/spaced_internal.h" +#include "arrow/util/pfor/pfor_wrapper.h" #include "arrow/util/ubsan.h" #include "arrow/visit_data_inline.h" @@ -2372,6 +2373,83 @@ class ByteStreamSplitDecoder : public ByteStreamSplitDecoderBase +class PforDecoder : public TypedDecoderImpl { + public: + using T = typename DType::c_type; + + explicit PforDecoder(const ColumnDescriptor* descr, + MemoryPool* pool = ::arrow::default_memory_pool()) + : TypedDecoderImpl(descr, Encoding::PFOR), pool_(pool) {} + + void SetData(int num_values, const uint8_t* data, int len) override { + this->num_values_ = num_values; + data_ = data; + data_len_ = len; + values_decoded_ = 0; + } + + int Decode(T* buffer, int max_values) override { + int values_to_decode = std::min(max_values, this->num_values_); + if (values_to_decode == 0) return 0; + + // Decode all values on first call, cache for subsequent calls + if (decoded_values_.empty() && this->num_values_ > 0) { + decoded_values_.resize(this->num_values_); + ::arrow::util::pfor::PforWrapper::Decode( + decoded_values_.data(), static_cast(this->num_values_), + data_, static_cast(data_len_)); + } + + std::memcpy(buffer, decoded_values_.data() + values_decoded_, + values_to_decode * sizeof(T)); + values_decoded_ += values_to_decode; + this->num_values_ -= values_to_decode; + return values_to_decode; + } + + int DecodeArrow(int num_values, int null_count, const uint8_t* valid_bits, + int64_t valid_bits_offset, + typename EncodingTraits::Accumulator* out) override { + if (null_count != 0) { + ParquetException::NYI("PFOR DecodeArrow with null slots"); + } + std::vector values(num_values); + int decoded_count = Decode(values.data(), num_values); + PARQUET_THROW_NOT_OK(out->AppendValues(values.data(), decoded_count)); + return decoded_count; + } + + int DecodeArrow(int num_values, int null_count, const uint8_t* valid_bits, + int64_t valid_bits_offset, + typename EncodingTraits::DictAccumulator* out) override { + if (null_count != 0) { + ParquetException::NYI("PFOR DecodeArrow with null slots"); + } + std::vector values(num_values); + int decoded_count = Decode(values.data(), num_values); + PARQUET_THROW_NOT_OK(out->Reserve(decoded_count)); + for (int i = 0; i < decoded_count; ++i) { + PARQUET_THROW_NOT_OK(out->Append(values[i])); + } + return decoded_count; + } + + private: + MemoryPool* pool_; + const uint8_t* data_ = nullptr; + int data_len_ = 0; + int values_decoded_ = 0; + std::vector decoded_values_; +}; + } // namespace // ---------------------------------------------------------------------- @@ -2448,6 +2526,15 @@ std::unique_ptr MakeDecoder(Type::type type_num, Encoding::type encodin return std::make_unique(descr); } throw ParquetException("RLE encoding only supports BOOLEAN"); + } else if (encoding == Encoding::PFOR) { + switch (type_num) { + case Type::INT32: + return std::make_unique>(descr, pool); + case Type::INT64: + return std::make_unique>(descr, pool); + default: + throw ParquetException("PFOR decoder only supports INT32 and INT64"); + } } else { ParquetException::NYI("Selected encoding is not supported"); } diff --git a/cpp/src/parquet/encoder.cc b/cpp/src/parquet/encoder.cc index 3e469df277b..4bd5f9baa4f 100644 --- a/cpp/src/parquet/encoder.cc +++ b/cpp/src/parquet/encoder.cc @@ -42,6 +42,7 @@ #include "arrow/util/logging_internal.h" #include "arrow/util/rle_encoding_internal.h" #include "arrow/util/spaced_internal.h" +#include "arrow/util/pfor/pfor_wrapper.h" #include "arrow/util/ubsan.h" #include "arrow/visit_data_inline.h" @@ -1763,6 +1764,88 @@ std::shared_ptr RleBooleanEncoder::FlushValues() { } // namespace +// ---------------------------------------------------------------------- +// PFOR Encoder + +// TODO: support incremental encoding. Today `Put` only appends raw input +// to `values_`, and `FlushValues` runs the entire PFOR pipeline on the +// whole buffer in one shot. A future revision should encode complete +// vectors as `Put` calls fill them, holding only a partial-vector tail +// across calls, so the encoder can produce output progressively and use +// bounded memory. +template +class PforEncoder : public EncoderImpl, virtual public TypedEncoder { + public: + using T = typename DType::c_type; + using TypedEncoder::Put; + + explicit PforEncoder(const ColumnDescriptor* descr, MemoryPool* pool) + : EncoderImpl(descr, Encoding::PFOR, pool), pool_(pool) {} + + std::shared_ptr FlushValues() override { + if (values_.empty()) { + PARQUET_ASSIGN_OR_THROW(auto empty_buf, ::arrow::AllocateBuffer(0, pool_)); + return std::move(empty_buf); + } + + const uint32_t num_values = static_cast(values_.size()); + int64_t max_size = + ::arrow::util::pfor::PforWrapper::GetMaxCompressedSize(num_values); + PARQUET_ASSIGN_OR_THROW(auto buffer, ::arrow::AllocateResizableBuffer( + max_size, pool_)); + + int64_t comp_size = max_size; + ::arrow::util::pfor::PforWrapper::Encode( + values_.data(), num_values, + buffer->mutable_data(), &comp_size); + + PARQUET_THROW_NOT_OK(buffer->Resize(comp_size)); + values_.clear(); + return std::move(buffer); + } + + int64_t EstimatedDataEncodedSize() override { + return static_cast(values_.size() * sizeof(T)); + } + + void Put(const ::arrow::Array& values) override { + const auto& data = *values.data(); + if (data.length > std::numeric_limits::max()) { + throw ParquetException("Array cannot be longer than ", + std::numeric_limits::max()); + } + if (values.null_count() == 0) { + Put(data.template GetValues(1), static_cast(data.length)); + } else { + PutSpaced(data.template GetValues(1), static_cast(data.length), + data.GetValues(0, 0), data.offset); + } + } + + void Put(const T* buffer, int num_values) override { + values_.insert(values_.end(), buffer, buffer + num_values); + } + + void PutSpaced(const T* src, int num_values, const uint8_t* valid_bits, + int64_t valid_bits_offset) override { + if (valid_bits != NULLPTR) { + PARQUET_ASSIGN_OR_THROW(auto buffer, + ::arrow::AllocateBuffer(num_values * sizeof(T), pool_)); + T* dest = reinterpret_cast(buffer->mutable_data()); + int num_valid = + ::arrow::util::internal::SpacedCompress(src, num_values, valid_bits, + valid_bits_offset, dest); + Put(dest, num_valid); + } else { + Put(src, num_values); + } + } + + private: + MemoryPool* pool_; + std::vector values_; +}; + // ---------------------------------------------------------------------- // Factory function @@ -1862,6 +1945,15 @@ std::unique_ptr MakeEncoder(Type::type type_num, Encoding::type encodin throw ParquetException( "DELTA_BYTE_ARRAY only supports BYTE_ARRAY and FIXED_LEN_BYTE_ARRAY"); } + } else if (encoding == Encoding::PFOR) { + switch (type_num) { + case Type::INT32: + return std::make_unique>(descr, pool); + case Type::INT64: + return std::make_unique>(descr, pool); + default: + throw ParquetException("PFOR encoder only supports INT32 and INT64"); + } } else { ParquetException::NYI("Selected encoding is not supported"); } diff --git a/cpp/src/parquet/pfor_comparison_benchmark.cc b/cpp/src/parquet/pfor_comparison_benchmark.cc new file mode 100644 index 00000000000..d5420ab65b6 --- /dev/null +++ b/cpp/src/parquet/pfor_comparison_benchmark.cc @@ -0,0 +1,926 @@ +// 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. + +// Comparison benchmark: PFOR vs DeltaBitPack vs ZSTD vs RleBitPackHybrid +// vs ByteStreamSplit+ZSTD vs ByteStreamSplit+LZ4 +// +// All throughput is reported as uncompressed_size / time (MB/s). +// Data generators mimic ClickBench and TPC-DS column distributions. + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "benchmark/benchmark.h" + +#include "arrow/util/compression.h" +#include "arrow/util/fastlanes/fastlanes_for.h" +#include "arrow/util/logging.h" +#include "arrow/util/pfor/pfor_wrapper.h" +#include "arrow/util/rle_encoding_internal.h" + +#include "parquet/encoding.h" +#include "parquet/platform.h" +#include "parquet/schema.h" +#include "parquet/types.h" + +using ::arrow::Compression; +using ::arrow::util::Codec; + +namespace parquet { +namespace { + +// ============================================================================ +// Data Generators — ClickBench-inspired +// ============================================================================ + +using Gen32 = std::vector (*)(int64_t); + +std::vector GenClientIP(int64_t n) { + std::vector v(n); + std::mt19937 rng(101); + std::uniform_int_distribution dist(0x0A000000, 0xDFFFFFFF); + for (auto& x : v) x = static_cast(dist(rng)); + return v; +} + +std::vector GenUrlRegionID(int64_t n) { + std::vector v(n); + std::mt19937 rng(102); + // Zipf-like over ~1000 values + std::uniform_real_distribution uni(0.0, 1.0); + for (auto& x : v) { + double u = uni(rng); + x = static_cast(std::pow(u, 2.0) * 1000) + 1; + } + return v; +} + +std::vector GenCounterID(int64_t n) { + std::vector v(n); + std::mt19937 rng(103); + std::uniform_int_distribution jitter(0, 3); + int32_t counter = 100000; + for (auto& x : v) { + counter += 1 + jitter(rng); + x = counter; + } + return v; +} + +std::vector GenEventDate(int64_t n) { + std::vector v(n); + std::mt19937 rng(104); + const int32_t dates[] = {19691, 19692, 19693, 19694, 19695}; + std::uniform_int_distribution idx(0, 4); + for (auto& x : v) x = dates[idx(rng)]; + return v; +} + +std::vector GenEventTime(int64_t n) { + std::vector v(n); + std::mt19937 rng(105); + const int32_t base = 1704067200; // 2024-01-01 + std::uniform_int_distribution offset(0, 86399); + for (auto& x : v) x = base + offset(rng); + return v; +} + +std::vector GenGoodEvent(int64_t n) { + std::vector v(n); + std::mt19937 rng(106); + std::uniform_int_distribution dist(0, 99); + for (auto& x : v) x = (dist(rng) < 95) ? 1 : 0; + return v; +} + +std::vector GenHID(int64_t n) { + std::vector v(n); + std::mt19937 rng(107); + std::uniform_int_distribution dist(std::numeric_limits::min(), + std::numeric_limits::max()); + for (auto& x : v) x = dist(rng); + return v; +} + +std::vector GenHitColor(int64_t n) { + std::vector v(n); + std::mt19937 rng(108); + const int32_t colors[] = {1, 2, 3, 4, 5}; + std::uniform_int_distribution idx(0, 4); + for (auto& x : v) x = colors[idx(rng)]; + return v; +} + +std::vector GenIPNetworkID(int64_t n) { + std::vector v(n); + std::mt19937 rng(109); + std::uniform_int_distribution dist(1, 10000); + for (auto& x : v) x = dist(rng); + return v; +} + +std::vector GenJavaEnable(int64_t n) { + std::vector v(n); + std::mt19937 rng(110); + std::uniform_int_distribution dist(0, 99); + for (auto& x : v) x = (dist(rng) < 85) ? 1 : 0; + return v; +} + +std::vector GenOS(int64_t n) { + std::vector v(n); + std::mt19937 rng(111); + std::uniform_int_distribution dist(1, 20); + for (auto& x : v) x = dist(rng); + return v; +} + +std::vector GenResolution(int64_t n) { + std::vector v(n); + std::mt19937 rng(112); + const int32_t resolutions[] = {360, 480, 600, 720, 768, 800, 900, + 1024, 1050, 1080, 1200, 1440, 1600, 2160}; + std::uniform_int_distribution idx(0, 13); + for (auto& x : v) x = resolutions[idx(rng)]; + return v; +} + +std::vector GenTrafficSourceID(int64_t n) { + std::vector v(n); + std::mt19937 rng(113); + std::uniform_int_distribution dist(0, 10); + for (auto& x : v) x = dist(rng); + return v; +} + +std::vector GenUserAgent(int64_t n) { + std::vector v(n); + std::mt19937 rng(114); + // Zipf-like over ~100 user agents + std::uniform_real_distribution uni(0.0, 1.0); + for (auto& x : v) { + double u = uni(rng); + x = static_cast(std::pow(u, 1.5) * 100) + 1; + } + return v; +} + +// ============================================================================ +// Data Generators — TPC-DS (4 most queried columns from store_sales) +// ============================================================================ + +std::vector GenTpcdsSoldDateSk(int64_t n) { + std::vector v(n); + const int32_t kBase = 2450815; + std::mt19937 rng(201); + std::uniform_int_distribution dist(0, 1820); + for (auto& x : v) x = kBase + dist(rng); + return v; +} + +std::vector GenTpcdsStoreSk(int64_t n) { + std::vector v(n); + std::mt19937 rng(202); + std::uniform_int_distribution dist(1, 1000); + for (auto& x : v) x = dist(rng); + return v; +} + +std::vector GenTpcdsItemSk(int64_t n) { + std::vector v(n); + const int32_t kMax = 100000; + std::mt19937 rng(203); + std::exponential_distribution exp_dist(0.00005); + for (auto& x : v) { + int32_t val = static_cast(exp_dist(rng)); + x = std::min(val + 1, kMax); + } + return v; +} + +std::vector GenTpcdsQuantity(int64_t n) { + std::vector v(n); + std::mt19937 rng(204); + std::uniform_int_distribution small_dist(1, 10); + std::uniform_int_distribution large_dist(11, 100); + std::uniform_int_distribution chance(0, 99); + for (auto& x : v) { + x = (chance(rng) < 90) ? small_dist(rng) : large_dist(rng); + } + return v; +} + +// ============================================================================ +// Helpers +// ============================================================================ + +static int32_t ComputeBitWidth(const std::vector& values) { + uint32_t max_val = 0; + for (int32_t v : values) { + max_val = std::max(max_val, static_cast(v)); + } + if (max_val == 0) return 1; + return static_cast(32 - __builtin_clz(max_val)); +} + +static std::shared_ptr MakeInt32Descriptor() { + auto node = + schema::PrimitiveNode::Make("col", Repetition::REQUIRED, Type::INT32); + return std::make_shared(node, /*max_def_level=*/0, + /*max_rep_level=*/0); +} + +// ============================================================================ +// PFOR Encode/Decode +// ============================================================================ + +static void BM_PforEncode(benchmark::State& state, Gen32 gen) { + const int64_t num_values = state.range(0); + auto values = gen(num_values); + const int64_t uncompressed_size = num_values * sizeof(int32_t); + + int64_t max_size = + ::arrow::util::pfor::PforWrapper::GetMaxCompressedSize( + static_cast(num_values)); + std::vector compressed(max_size); + + // Compute comp_size once for the counter + int64_t comp_size = max_size; + ::arrow::util::pfor::PforWrapper::Encode( + values.data(), static_cast(num_values), compressed.data(), &comp_size); + + for (auto _ : state) { + int64_t sz = max_size; + ::arrow::util::pfor::PforWrapper::Encode( + values.data(), static_cast(num_values), compressed.data(), &sz); + benchmark::DoNotOptimize(sz); + benchmark::ClobberMemory(); + } + + state.SetBytesProcessed(state.iterations() * uncompressed_size); + state.SetItemsProcessed(state.iterations() * num_values); + state.counters["compression_ratio"] = + static_cast(uncompressed_size) / static_cast(comp_size); +} + +static void BM_PforDecode(benchmark::State& state, Gen32 gen) { + const int64_t num_values = state.range(0); + auto values = gen(num_values); + const int64_t uncompressed_size = num_values * sizeof(int32_t); + + int64_t max_size = + ::arrow::util::pfor::PforWrapper::GetMaxCompressedSize( + static_cast(num_values)); + std::vector compressed(max_size); + int64_t comp_size = max_size; + ::arrow::util::pfor::PforWrapper::Encode( + values.data(), static_cast(num_values), compressed.data(), + &comp_size); + + std::vector decoded(num_values); + for (auto _ : state) { + auto status = ::arrow::util::pfor::PforWrapper::Decode( + decoded.data(), static_cast(num_values), compressed.data(), + comp_size); + ARROW_CHECK_OK(status); + benchmark::ClobberMemory(); + } + + state.SetBytesProcessed(state.iterations() * uncompressed_size); + state.SetItemsProcessed(state.iterations() * num_values); + state.counters["compression_ratio"] = + static_cast(uncompressed_size) / static_cast(comp_size); +} + +// ============================================================================ +// PFOR with FastLanes-mode bit packing (per-vector flag in PFOR header) +// ============================================================================ +// +// Same PFOR pipeline (FOR + exceptions) but the bit-packing payload uses +// FastLanes lane-interleaved layout instead of the legacy sequential bit +// stream. Output is still flat (PFOR contract preserved); the decoder +// scatters via FL_ORDER inside DecodeVector. + +static void BM_PforFastLanesEncode(benchmark::State& state, Gen32 gen) { + const int64_t num_values = state.range(0); + auto values = gen(num_values); + const int64_t uncompressed_size = num_values * sizeof(int32_t); + + int64_t max_size = + ::arrow::util::pfor::PforWrapper::GetMaxCompressedSize( + static_cast(num_values)); + std::vector compressed(max_size); + + int64_t comp_size = max_size; + ::arrow::util::pfor::PforWrapper::Encode( + values.data(), static_cast(num_values), compressed.data(), &comp_size, + ::arrow::util::pfor::PackingMode::FastLanes); + + for (auto _ : state) { + int64_t sz = max_size; + ::arrow::util::pfor::PforWrapper::Encode( + values.data(), static_cast(num_values), compressed.data(), &sz, + ::arrow::util::pfor::PackingMode::FastLanes); + benchmark::DoNotOptimize(sz); + benchmark::ClobberMemory(); + } + + state.SetBytesProcessed(state.iterations() * uncompressed_size); + state.SetItemsProcessed(state.iterations() * num_values); + state.counters["compression_ratio"] = + static_cast(uncompressed_size) / static_cast(comp_size); +} + +static void BM_PforFastLanesDecode(benchmark::State& state, Gen32 gen) { + const int64_t num_values = state.range(0); + auto values = gen(num_values); + const int64_t uncompressed_size = num_values * sizeof(int32_t); + + int64_t max_size = + ::arrow::util::pfor::PforWrapper::GetMaxCompressedSize( + static_cast(num_values)); + std::vector compressed(max_size); + int64_t comp_size = max_size; + ::arrow::util::pfor::PforWrapper::Encode( + values.data(), static_cast(num_values), compressed.data(), + &comp_size, ::arrow::util::pfor::PackingMode::FastLanes); + + std::vector decoded(num_values); + for (auto _ : state) { + auto status = ::arrow::util::pfor::PforWrapper::Decode( + decoded.data(), static_cast(num_values), compressed.data(), + comp_size); + ARROW_CHECK_OK(status); + benchmark::ClobberMemory(); + } + + state.SetBytesProcessed(state.iterations() * uncompressed_size); + state.SetItemsProcessed(state.iterations() * num_values); + state.counters["compression_ratio"] = + static_cast(uncompressed_size) / static_cast(comp_size); +} + +// PFOR with FastLanes bit-packing, decoded in TRANSPOSED order: skips the +// per-vector FL_ORDER gather, output is in FastLanes stream order. The +// downstream consumer must be permutation-aware. Apples-to-apples this +// against BM_PforFastLanesDecode (flat-order, with the scatter cost) to see +// how much of the gap to pfor+bitpack closes. +static void BM_PforFastLanesDecodeTransposed(benchmark::State& state, Gen32 gen) { + const int64_t num_values = state.range(0); + auto values = gen(num_values); + const int64_t uncompressed_size = num_values * sizeof(int32_t); + + int64_t max_size = + ::arrow::util::pfor::PforWrapper::GetMaxCompressedSize( + static_cast(num_values)); + std::vector compressed(max_size); + int64_t comp_size = max_size; + ::arrow::util::pfor::PforWrapper::Encode( + values.data(), static_cast(num_values), compressed.data(), + &comp_size, ::arrow::util::pfor::PackingMode::FastLanes); + + std::vector decoded(num_values); + for (auto _ : state) { + auto status = ::arrow::util::pfor::PforWrapper::Decode( + decoded.data(), static_cast(num_values), compressed.data(), + comp_size, ::arrow::util::pfor::OutputOrder::Transposed); + ARROW_CHECK_OK(status); + benchmark::ClobberMemory(); + } + + state.SetBytesProcessed(state.iterations() * uncompressed_size); + state.SetItemsProcessed(state.iterations() * num_values); + state.counters["compression_ratio"] = + static_cast(uncompressed_size) / static_cast(comp_size); +} + +// ============================================================================ +// DeltaBitPack Encode/Decode +// ============================================================================ + +static void BM_DeltaBitPackEncode(benchmark::State& state, Gen32 gen) { + const int64_t num_values = state.range(0); + auto values = gen(num_values); + const int64_t uncompressed_size = num_values * sizeof(int32_t); + + auto encoder = MakeTypedEncoder(Encoding::DELTA_BINARY_PACKED); + + // Compute comp_size once for the counter + encoder->Put(values.data(), static_cast(num_values)); + auto pre_buf = encoder->FlushValues(); + int64_t comp_size = pre_buf->size(); + + for (auto _ : state) { + encoder->Put(values.data(), static_cast(num_values)); + auto buf = encoder->FlushValues(); + benchmark::DoNotOptimize(buf); + } + + state.SetBytesProcessed(state.iterations() * uncompressed_size); + state.SetItemsProcessed(state.iterations() * num_values); + state.counters["compression_ratio"] = + static_cast(uncompressed_size) / static_cast(comp_size); +} + +static void BM_DeltaBitPackDecode(benchmark::State& state, Gen32 gen) { + const int64_t num_values = state.range(0); + auto values = gen(num_values); + const int64_t uncompressed_size = num_values * sizeof(int32_t); + + auto encoder = MakeTypedEncoder(Encoding::DELTA_BINARY_PACKED); + encoder->Put(values.data(), static_cast(num_values)); + auto buf = encoder->FlushValues(); + int64_t comp_size = buf->size(); + + std::vector decoded(num_values); + auto decoder = MakeTypedDecoder(Encoding::DELTA_BINARY_PACKED); + + for (auto _ : state) { + decoder->SetData(static_cast(num_values), buf->data(), + static_cast(buf->size())); + decoder->Decode(decoded.data(), static_cast(num_values)); + benchmark::ClobberMemory(); + } + + state.SetBytesProcessed(state.iterations() * uncompressed_size); + state.SetItemsProcessed(state.iterations() * num_values); + state.counters["compression_ratio"] = + static_cast(uncompressed_size) / static_cast(comp_size); +} + +// ============================================================================ +// FastLanes + Frame-of-Reference Encode/Decode +// ============================================================================ +// +// Round-trip is NOT flat: decoder produces output in transposed FL_ORDER +// (per 1024-block, output[t] == input[fromTransposed32(t)] + min). Chunked +// at 2048 values; per-chunk header stores [min, bit_width]. + +static void BM_FastLanesEncode(benchmark::State& state, Gen32 gen) { + using ::arrow::util::fastlanes::FastLanesForCodec; + // Round num_values up to a multiple of kChunkSize (2048) for this codec. + int64_t num_values = state.range(0); + num_values -= num_values % FastLanesForCodec::kChunkSize; + if (num_values == 0) num_values = FastLanesForCodec::kChunkSize; + + auto values = gen(num_values); + const int64_t uncompressed_size = num_values * sizeof(int32_t); + + std::vector compressed(FastLanesForCodec::MaxEncodedSize(num_values)); + + auto first = FastLanesForCodec::Encode(values.data(), num_values, compressed.data()); + ARROW_CHECK_OK(first.status()); + const int64_t comp_size = *first; + + for (auto _ : state) { + auto sz = FastLanesForCodec::Encode(values.data(), num_values, compressed.data()); + benchmark::DoNotOptimize(sz); + benchmark::ClobberMemory(); + } + + state.SetBytesProcessed(state.iterations() * uncompressed_size); + state.SetItemsProcessed(state.iterations() * num_values); + state.counters["compression_ratio"] = + static_cast(uncompressed_size) / static_cast(comp_size); +} + +static void BM_FastLanesDecode(benchmark::State& state, Gen32 gen) { + using ::arrow::util::fastlanes::FastLanesForCodec; + int64_t num_values = state.range(0); + num_values -= num_values % FastLanesForCodec::kChunkSize; + if (num_values == 0) num_values = FastLanesForCodec::kChunkSize; + + auto values = gen(num_values); + const int64_t uncompressed_size = num_values * sizeof(int32_t); + + std::vector compressed(FastLanesForCodec::MaxEncodedSize(num_values)); + auto comp = FastLanesForCodec::Encode(values.data(), num_values, compressed.data()); + ARROW_CHECK_OK(comp.status()); + const int64_t comp_size = *comp; + + std::vector decoded(num_values); + for (auto _ : state) { + auto status = FastLanesForCodec::Decode(decoded.data(), num_values, + compressed.data(), comp_size); + ARROW_CHECK_OK(status); + benchmark::ClobberMemory(); + } + + state.SetBytesProcessed(state.iterations() * uncompressed_size); + state.SetItemsProcessed(state.iterations() * num_values); + state.counters["compression_ratio"] = + static_cast(uncompressed_size) / static_cast(comp_size); +} + +// Decode + FL_ORDER scatter: produces flat output in original input +// order (apples-to-apples vs PFOR/DeltaBitPack which also produce flat). +static void BM_FastLanesDecodeFlat(benchmark::State& state, Gen32 gen) { + using ::arrow::util::fastlanes::FastLanesForCodec; + int64_t num_values = state.range(0); + num_values -= num_values % FastLanesForCodec::kChunkSize; + if (num_values == 0) num_values = FastLanesForCodec::kChunkSize; + + auto values = gen(num_values); + const int64_t uncompressed_size = num_values * sizeof(int32_t); + + std::vector compressed(FastLanesForCodec::MaxEncodedSize(num_values)); + auto comp = FastLanesForCodec::Encode(values.data(), num_values, compressed.data()); + ARROW_CHECK_OK(comp.status()); + const int64_t comp_size = *comp; + + std::vector decoded(num_values); + for (auto _ : state) { + auto status = FastLanesForCodec::DecodeFlat(decoded.data(), num_values, + compressed.data(), comp_size); + ARROW_CHECK_OK(status); + benchmark::ClobberMemory(); + } + + state.SetBytesProcessed(state.iterations() * uncompressed_size); + state.SetItemsProcessed(state.iterations() * num_values); + state.counters["compression_ratio"] = + static_cast(uncompressed_size) / static_cast(comp_size); +} + +// ============================================================================ +// Plain + ZSTD Encode/Decode +// ============================================================================ + +static void BM_PlainZstdEncode(benchmark::State& state, Gen32 gen) { + const int64_t num_values = state.range(0); + auto values = gen(num_values); + const int64_t uncompressed_size = num_values * sizeof(int32_t); + const uint8_t* raw = reinterpret_cast(values.data()); + + auto codec = *Codec::Create(Compression::ZSTD); + int64_t max_comp = codec->MaxCompressedLen(uncompressed_size, raw); + std::vector compressed(max_comp); + + // Compute comp_size once for the counter + int64_t comp_size = + *codec->Compress(uncompressed_size, raw, max_comp, compressed.data()); + + for (auto _ : state) { + auto sz = *codec->Compress(uncompressed_size, raw, max_comp, compressed.data()); + benchmark::DoNotOptimize(sz); + } + + state.SetBytesProcessed(state.iterations() * uncompressed_size); + state.SetItemsProcessed(state.iterations() * num_values); + state.counters["compression_ratio"] = + static_cast(uncompressed_size) / static_cast(comp_size); +} + +static void BM_PlainZstdDecode(benchmark::State& state, Gen32 gen) { + const int64_t num_values = state.range(0); + auto values = gen(num_values); + const int64_t uncompressed_size = num_values * sizeof(int32_t); + const uint8_t* raw = reinterpret_cast(values.data()); + + auto codec = *Codec::Create(Compression::ZSTD); + int64_t max_comp = codec->MaxCompressedLen(uncompressed_size, raw); + std::vector compressed(max_comp); + int64_t comp_size = + *codec->Compress(uncompressed_size, raw, max_comp, compressed.data()); + + std::vector decompressed(uncompressed_size); + for (auto _ : state) { + auto result = codec->Decompress(comp_size, compressed.data(), uncompressed_size, + decompressed.data()); + ARROW_CHECK_OK(result.status()); + benchmark::ClobberMemory(); + } + + state.SetBytesProcessed(state.iterations() * uncompressed_size); + state.SetItemsProcessed(state.iterations() * num_values); + state.counters["compression_ratio"] = + static_cast(uncompressed_size) / static_cast(comp_size); +} + +// ============================================================================ +// Plain + LZ4 Encode/Decode +// ============================================================================ + +static void BM_PlainLz4Encode(benchmark::State& state, Gen32 gen) { + const int64_t num_values = state.range(0); + auto values = gen(num_values); + const int64_t uncompressed_size = num_values * sizeof(int32_t); + const uint8_t* raw = reinterpret_cast(values.data()); + + auto codec = *Codec::Create(Compression::LZ4_FRAME); + int64_t max_comp = codec->MaxCompressedLen(uncompressed_size, raw); + std::vector compressed(max_comp); + + int64_t comp_size = + *codec->Compress(uncompressed_size, raw, max_comp, compressed.data()); + + for (auto _ : state) { + auto sz = *codec->Compress(uncompressed_size, raw, max_comp, compressed.data()); + benchmark::DoNotOptimize(sz); + } + + state.SetBytesProcessed(state.iterations() * uncompressed_size); + state.SetItemsProcessed(state.iterations() * num_values); + state.counters["compression_ratio"] = + static_cast(uncompressed_size) / static_cast(comp_size); +} + +static void BM_PlainLz4Decode(benchmark::State& state, Gen32 gen) { + const int64_t num_values = state.range(0); + auto values = gen(num_values); + const int64_t uncompressed_size = num_values * sizeof(int32_t); + const uint8_t* raw = reinterpret_cast(values.data()); + + auto codec = *Codec::Create(Compression::LZ4_FRAME); + int64_t max_comp = codec->MaxCompressedLen(uncompressed_size, raw); + std::vector compressed(max_comp); + int64_t comp_size = + *codec->Compress(uncompressed_size, raw, max_comp, compressed.data()); + + std::vector decompressed(uncompressed_size); + for (auto _ : state) { + auto result = codec->Decompress(comp_size, compressed.data(), uncompressed_size, + decompressed.data()); + ARROW_CHECK_OK(result.status()); + benchmark::ClobberMemory(); + } + + state.SetBytesProcessed(state.iterations() * uncompressed_size); + state.SetItemsProcessed(state.iterations() * num_values); + state.counters["compression_ratio"] = + static_cast(uncompressed_size) / static_cast(comp_size); +} + +// ============================================================================ +// RleBitPackHybrid Encode/Decode +// ============================================================================ + +static void BM_RleBitPackEncode(benchmark::State& state, Gen32 gen) { + const int64_t num_values = state.range(0); + auto values = gen(num_values); + const int64_t uncompressed_size = num_values * sizeof(int32_t); + + int32_t bit_width = ComputeBitWidth(values); + int64_t max_buf = + ::arrow::util::RleBitPackedEncoder::MaxBufferSize(bit_width, num_values) + + ::arrow::util::RleBitPackedEncoder::MinBufferSize(bit_width); + std::vector buffer(max_buf); + + // Compute comp_size once for the counter + int64_t comp_size; + { + ::arrow::util::RleBitPackedEncoder enc(buffer.data(), + static_cast(max_buf), bit_width); + for (int64_t i = 0; i < num_values; ++i) { + enc.Put(static_cast(static_cast(values[i]))); + } + comp_size = enc.Flush(); + } + + for (auto _ : state) { + ::arrow::util::RleBitPackedEncoder encoder(buffer.data(), + static_cast(max_buf), bit_width); + for (int64_t i = 0; i < num_values; ++i) { + encoder.Put(static_cast(static_cast(values[i]))); + } + auto sz = encoder.Flush(); + benchmark::DoNotOptimize(sz); + } + + state.SetBytesProcessed(state.iterations() * uncompressed_size); + state.SetItemsProcessed(state.iterations() * num_values); + state.counters["compression_ratio"] = + static_cast(uncompressed_size) / static_cast(comp_size); +} + +static void BM_RleBitPackDecode(benchmark::State& state, Gen32 gen) { + const int64_t num_values = state.range(0); + auto values = gen(num_values); + const int64_t uncompressed_size = num_values * sizeof(int32_t); + + int32_t bit_width = ComputeBitWidth(values); + int64_t max_buf = + ::arrow::util::RleBitPackedEncoder::MaxBufferSize(bit_width, num_values) + + ::arrow::util::RleBitPackedEncoder::MinBufferSize(bit_width); + std::vector buffer(max_buf); + + ::arrow::util::RleBitPackedEncoder encoder(buffer.data(), + static_cast(max_buf), bit_width); + for (int64_t i = 0; i < num_values; ++i) { + encoder.Put(static_cast(static_cast(values[i]))); + } + int comp_size = encoder.Flush(); + + std::vector decoded(num_values); + for (auto _ : state) { + ::arrow::util::RleBitPackedParser parser(buffer.data(), comp_size, bit_width); + int64_t out_idx = 0; + struct Handler { + int32_t* output; + int64_t* idx; + int64_t max_values; + int32_t bw; + ::arrow::util::RleBitPackedParser::ControlFlow OnRleRun( + ::arrow::util::RleRun run) { + ::arrow::util::RleRunDecoder dec(run, bw); + auto want = static_cast( + std::min(static_cast(run.values_count()), max_values - *idx)); + auto count = dec.GetBatch(output + *idx, want, bw); + *idx += count; + return *idx >= max_values + ? ::arrow::util::RleBitPackedParser::ControlFlow::Break + : ::arrow::util::RleBitPackedParser::ControlFlow::Continue; + } + ::arrow::util::RleBitPackedParser::ControlFlow OnBitPackedRun( + ::arrow::util::BitPackedRun run) { + ::arrow::util::BitPackedRunDecoder dec(run, bw); + auto want = static_cast( + std::min(static_cast(dec.remaining()), max_values - *idx)); + auto count = dec.GetBatch(output + *idx, want, bw); + *idx += count; + return *idx >= max_values + ? ::arrow::util::RleBitPackedParser::ControlFlow::Break + : ::arrow::util::RleBitPackedParser::ControlFlow::Continue; + } + }; + Handler handler{decoded.data(), &out_idx, num_values, bit_width}; + parser.Parse(handler); + benchmark::ClobberMemory(); + } + + state.SetBytesProcessed(state.iterations() * uncompressed_size); + state.SetItemsProcessed(state.iterations() * num_values); + state.counters["compression_ratio"] = + static_cast(uncompressed_size) / static_cast(comp_size); +} + +// ============================================================================ +// ByteStreamSplit + Codec (ZSTD or LZ4) +// ============================================================================ + +static void BM_BssCodecEncode(benchmark::State& state, Gen32 gen, + Compression::type codec_type) { + const int64_t num_values = state.range(0); + auto values = gen(num_values); + const int64_t uncompressed_size = num_values * sizeof(int32_t); + + auto descr = MakeInt32Descriptor(); + auto encoder = MakeTypedEncoder(Encoding::BYTE_STREAM_SPLIT, + /*use_dictionary=*/false, descr.get()); + auto codec = *Codec::Create(codec_type); + + encoder->Put(values.data(), static_cast(num_values)); + auto encoded_buf = encoder->FlushValues(); + int64_t encoded_size = encoded_buf->size(); + + int64_t max_comp = codec->MaxCompressedLen(encoded_size, encoded_buf->data()); + std::vector compressed(max_comp); + + // Compute comp_size once for the counter + int64_t comp_size = + *codec->Compress(encoded_size, encoded_buf->data(), max_comp, compressed.data()); + + for (auto _ : state) { + encoder->Put(values.data(), static_cast(num_values)); + auto buf = encoder->FlushValues(); + auto sz = + *codec->Compress(buf->size(), buf->data(), max_comp, compressed.data()); + benchmark::DoNotOptimize(sz); + } + + state.SetBytesProcessed(state.iterations() * uncompressed_size); + state.SetItemsProcessed(state.iterations() * num_values); + state.counters["compression_ratio"] = + static_cast(uncompressed_size) / static_cast(comp_size); +} + +static void BM_BssCodecDecode(benchmark::State& state, Gen32 gen, + Compression::type codec_type) { + const int64_t num_values = state.range(0); + auto values = gen(num_values); + const int64_t uncompressed_size = num_values * sizeof(int32_t); + + auto descr = MakeInt32Descriptor(); + auto encoder = MakeTypedEncoder(Encoding::BYTE_STREAM_SPLIT, + /*use_dictionary=*/false, descr.get()); + auto codec = *Codec::Create(codec_type); + + encoder->Put(values.data(), static_cast(num_values)); + auto encoded_buf = encoder->FlushValues(); + int64_t encoded_size = encoded_buf->size(); + + int64_t max_comp = codec->MaxCompressedLen(encoded_size, encoded_buf->data()); + std::vector compressed(max_comp); + int64_t comp_size = + *codec->Compress(encoded_size, encoded_buf->data(), max_comp, compressed.data()); + + std::vector decompressed(encoded_size); + std::vector decoded(num_values); + auto decoder = MakeTypedDecoder(Encoding::BYTE_STREAM_SPLIT, descr.get()); + + for (auto _ : state) { + auto result = codec->Decompress(comp_size, compressed.data(), encoded_size, + decompressed.data()); + ARROW_CHECK_OK(result.status()); + decoder->SetData(static_cast(num_values), decompressed.data(), + static_cast(encoded_size)); + decoder->Decode(decoded.data(), static_cast(num_values)); + benchmark::ClobberMemory(); + } + + state.SetBytesProcessed(state.iterations() * uncompressed_size); + state.SetItemsProcessed(state.iterations() * num_values); + state.counters["compression_ratio"] = + static_cast(uncompressed_size) / static_cast(comp_size); +} + +// Wrappers for BSS+ZSTD +static void BM_BssZstdEncode(benchmark::State& state, Gen32 gen) { + BM_BssCodecEncode(state, gen, Compression::ZSTD); +} +static void BM_BssZstdDecode(benchmark::State& state, Gen32 gen) { + BM_BssCodecDecode(state, gen, Compression::ZSTD); +} + +// Wrappers for BSS+LZ4 +static void BM_BssLz4Encode(benchmark::State& state, Gen32 gen) { + BM_BssCodecEncode(state, gen, Compression::LZ4_FRAME); +} +static void BM_BssLz4Decode(benchmark::State& state, Gen32 gen) { + BM_BssCodecDecode(state, gen, Compression::LZ4_FRAME); +} + +// ============================================================================ +// Benchmark Registration +// ============================================================================ + +static void CustomArgs(benchmark::internal::Benchmark* b) { b->Arg(102400); } + +// Macro to register all algorithms for a given dataset +#define REGISTER_DATASET(Name, GenFunc) \ + BENCHMARK_CAPTURE(BM_PforEncode, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_PforDecode, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_PforFastLanesEncode, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_PforFastLanesDecode, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_PforFastLanesDecodeTransposed, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_DeltaBitPackEncode, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_DeltaBitPackDecode, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_FastLanesEncode, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_FastLanesDecode, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_FastLanesDecodeFlat, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_PlainZstdEncode, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_PlainZstdDecode, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_PlainLz4Encode, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_PlainLz4Decode, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_RleBitPackEncode, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_RleBitPackDecode, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_BssZstdEncode, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_BssZstdDecode, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_BssLz4Encode, Name, &GenFunc)->Apply(CustomArgs); \ + BENCHMARK_CAPTURE(BM_BssLz4Decode, Name, &GenFunc)->Apply(CustomArgs); + +// ClickBench datasets +REGISTER_DATASET(ClientIP, GenClientIP) +REGISTER_DATASET(UrlRegionID, GenUrlRegionID) +REGISTER_DATASET(CounterID, GenCounterID) +REGISTER_DATASET(EventDate, GenEventDate) +REGISTER_DATASET(EventTime, GenEventTime) +REGISTER_DATASET(GoodEvent, GenGoodEvent) +REGISTER_DATASET(HID, GenHID) +REGISTER_DATASET(HitColor, GenHitColor) +REGISTER_DATASET(IPNetworkID, GenIPNetworkID) +REGISTER_DATASET(JavaEnable, GenJavaEnable) +REGISTER_DATASET(OS, GenOS) +REGISTER_DATASET(Resolution, GenResolution) +REGISTER_DATASET(TrafficSourceID, GenTrafficSourceID) +REGISTER_DATASET(UserAgent, GenUserAgent) + +// TPC-DS datasets +REGISTER_DATASET(TpcdsSoldDateSk, GenTpcdsSoldDateSk) +REGISTER_DATASET(TpcdsStoreSk, GenTpcdsStoreSk) +REGISTER_DATASET(TpcdsItemSk, GenTpcdsItemSk) +REGISTER_DATASET(TpcdsQuantity, GenTpcdsQuantity) + +} // namespace +} // namespace parquet diff --git a/cpp/src/parquet/types.cc b/cpp/src/parquet/types.cc index 9d7604faec3..1567d2b559e 100644 --- a/cpp/src/parquet/types.cc +++ b/cpp/src/parquet/types.cc @@ -270,6 +270,8 @@ std::string EncodingToString(Encoding::type t) { return "RLE_DICTIONARY"; case Encoding::BYTE_STREAM_SPLIT: return "BYTE_STREAM_SPLIT"; + case Encoding::PFOR: + return "PFOR"; default: return "UNKNOWN"; } diff --git a/cpp/src/parquet/types.h b/cpp/src/parquet/types.h index 687353aa9bc..acfedb45d79 100644 --- a/cpp/src/parquet/types.h +++ b/cpp/src/parquet/types.h @@ -539,8 +539,9 @@ struct Encoding { DELTA_BYTE_ARRAY = 7, RLE_DICTIONARY = 8, BYTE_STREAM_SPLIT = 9, + PFOR = 11, // Should always be last element (except UNKNOWN) - UNDEFINED = 10, + UNDEFINED = 12, UNKNOWN = 999 }; };