rok commented on code in PR #14191: URL: https://github.com/apache/arrow/pull/14191#discussion_r1041674331
########## cpp/src/parquet/encoding.cc: ########## @@ -2060,13 +2060,283 @@ class DictByteArrayDecoderImpl : public DictDecoderImpl<ByteArrayType>, } }; +// ---------------------------------------------------------------------- +// DeltaBitPackEncoder + +/// DeltaBitPackEncoder is an encoder for the DeltaBinary Packing format +/// as per the parquet spec. See: +/// https://github.com/apache/parquet-format/blob/master/Encodings.md#delta-encoding-delta_binary_packed--5 +/// +/// Consists of a header followed by blocks of delta encoded values binary packed. +/// +/// Format +/// [header] [block 1] [block 2] ... [block N] +/// +/// Header +/// [block size] [number of mini blocks per block] [total value count] [first value] +/// +/// Block +/// [min delta] [list of bitwidths of the mini blocks] [miniblocks] +/// +/// Sets aside bytes at the start of the internal buffer where the header will be written, +/// and only writes the header when FlushValues is called before returning it. +/// +/// To encode a block, we will: +/// +/// 1. Compute the differences between consecutive elements. For the first element in the +/// block, use the last element in the previous block or, in the case of the first block, +/// use the first value of the whole sequence, stored in the header. +/// +/// 2. Compute the frame of reference (the minimum of the deltas in the block). Subtract +/// this min delta from all deltas in the block. This guarantees that all values are +/// non-negative. +/// +/// 3. Encode the frame of reference (min delta) as a zigzag ULEB128 int followed by the +/// bit widths of the mini blocks and the delta values (minus the min delta) bit packed +/// per mini block. +/// +/// Supports only INT32 and INT64. + +template <typename DType> +class DeltaBitPackEncoder : public EncoderImpl, virtual public TypedEncoder<DType> { + static constexpr uint32_t kMaxPageHeaderWriterSize = 32; + static constexpr uint32_t kValuesPerBlock = 128; + static constexpr uint32_t kMiniBlocksPerBlock = 4; + + public: + using T = typename DType::c_type; + using UT = std::make_unsigned_t<T>; + using TypedEncoder<DType>::Put; + + explicit DeltaBitPackEncoder(const ColumnDescriptor* descr, MemoryPool* pool, + const uint32_t values_per_block = kValuesPerBlock, + const uint32_t mini_blocks_per_block = kMiniBlocksPerBlock) + : EncoderImpl(descr, Encoding::DELTA_BINARY_PACKED, pool), + values_per_block_(values_per_block), + mini_blocks_per_block_(mini_blocks_per_block), + values_per_mini_block_(values_per_block / mini_blocks_per_block), + deltas_(values_per_block, ::arrow::stl::allocator<T>(pool)), + bits_buffer_( + AllocateBuffer(pool, (kMiniBlocksPerBlock + values_per_block) * sizeof(T))), + sink_(pool), + bit_writer_(bits_buffer_->mutable_data(), static_cast<int>(bits_buffer_->size())), + bit_widths_(mini_blocks_per_block, 0) { + if (values_per_block_ % 128 != 0) { + throw ParquetException( + "the number of values in a block must be multiple of 128, but it's " + + std::to_string(values_per_block_)); + } + if (values_per_mini_block_ % 32 != 0) { + throw ParquetException( + "the number of values in a miniblock must be multiple of 32, but it's " + + std::to_string(values_per_mini_block_)); + } + if (values_per_block % mini_blocks_per_block != 0) { + throw ParquetException( + "the number of values per block % number of miniblocks per block must be 0, " + "but it's " + + std::to_string(values_per_block % mini_blocks_per_block)); + } + // Reserve enough space at the beginning of the buffer for largest possible header. + PARQUET_THROW_NOT_OK(sink_.Advance(kMaxPageHeaderWriterSize)); + } + + std::shared_ptr<Buffer> FlushValues() override; + + int64_t EstimatedDataEncodedSize() override { return sink_.length(); } + + void Put(const ::arrow::Array& values) override; + + void Put(const T* buffer, int num_values) override; + + void PutSpaced(const T* src, int num_values, const uint8_t* valid_bits, + int64_t valid_bits_offset) override; + + void FlushBlock(); + + private: + const uint32_t values_per_block_; + const uint32_t mini_blocks_per_block_; + const uint32_t values_per_mini_block_; + uint32_t values_current_block_{0}; + uint32_t total_value_count_{0}; + UT first_value_{0}; + UT current_value_{0}; + ArrowPoolVector<UT> deltas_; + std::shared_ptr<ResizableBuffer> bits_buffer_; + ::arrow::BufferBuilder sink_; + ::arrow::bit_util::BitWriter bit_writer_; + std::vector<uint8_t> bit_widths_; Review Comment: I think we agreed `bit_widths_` was sufficiently small. It was removed now anyway. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
