Compression: replace Delta with Delta+PFor

Signed-off-by: Lloyd-Pottiger <[email protected]>

contrib/CMakeLists.txt

@@ -179,6 +179,8 @@ target_include_directories(cpptoml INTERFACE
 SET (BENCHMARK_ENABLE_TESTING OFF CACHE BOOL "Disable google-benchmark testing" FORCE)
 SET (BENCHMARK_ENABLE_GTEST_TESTS OFF CACHE BOOL "Disable google-benchmark testing" FORCE)
 add_subdirectory(benchmark)
+target_compile_options(benchmark PRIVATE "-Wno-error=thread-safety-analysis")
+target_no_warning(benchmark thread-safety-analysis)
 
 set (BUILD_TESTING OFF CACHE BOOL "Disable cpu-features testing" FORCE)
 if (NOT (OS_DARWIN AND ARCH_AARCH64))

dbms/CMakeLists.txt

@@ -104,7 +104,7 @@ check_then_add_sources_compile_flag (
     src/Columns/ColumnVector.cpp
     src/DataTypes/DataTypeString.cpp
     src/Interpreters/Join.cpp
-    src/IO/Compression/CompressionCodecDelta.cpp
+    src/IO/Compression/CompressionCodecDeltaPFor.cpp
     src/Storages/DeltaMerge/BitmapFilter/BitmapFilter.cpp
     src/Storages/DeltaMerge/DMVersionFilterBlockInputStream.cpp
 )

dbms/src/IO/Compression/CompressionCodecDelta.cpp → dbms/src/IO/Compression/CompressionCodecDeltaPFor.cpp

@@ -12,9 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include <Common/BitpackingPrimitives.h>
 #include <Common/Exception.h>
 #include <DataTypes/IDataType.h>
-#include <IO/Compression/CompressionCodecDelta.h>
+#include <IO/Compression/CompressionCodecDeltaPFor.h>
 #include <IO/Compression/CompressionInfo.h>
 #include <common/likely.h>
 #include <common/unaligned.h>
@@ -32,58 +33,118 @@ extern const int CANNOT_COMPRESS;
 extern const int CANNOT_DECOMPRESS;
 } // namespace ErrorCodes
 
-CompressionCodecDelta::CompressionCodecDelta(UInt8 delta_bytes_size_)
-    : delta_bytes_size(delta_bytes_size_)
+CompressionCodecDeltaPFor::CompressionCodecDeltaPFor(UInt8 bytes_size_)
+    : bytes_size(bytes_size_)
 {}
 
-UInt8 CompressionCodecDelta::getMethodByte() const
+UInt8 CompressionCodecDeltaPFor::getMethodByte() const
 {
-    return static_cast<UInt8>(CompressionMethodByte::Delta);
+    return static_cast<UInt8>(CompressionMethodByte::DeltaPFor);
+}
+
+UInt32 CompressionCodecDeltaPFor::getMaxCompressedDataSize(UInt32 uncompressed_size) const
+{
+    // 1 byte for bytes_size, x bytes for frame of reference, 1 byte for width.
+    return 1 + bytes_size + sizeof(UInt8)
+        + BitpackingPrimitives::getRequiredSize(uncompressed_size / bytes_size, bytes_size * 8);
 }
 
 namespace
 {
 template <typename T>
-void compressData(const char * source, UInt32 source_size, char * dest)
+UInt32 compressData(const char * source, UInt32 source_size, char * dest)
 {
-    if (source_size % sizeof(T) != 0)
-        throw Exception(
-            ErrorCodes::CANNOT_COMPRESS,
-            "Cannot compress with Delta codec, data size {} is not aligned to {}",
-            source_size,
-            sizeof(T));
-
+    const auto count = source_size / sizeof(T);
+    using ST = typename std::make_signed<T>::type;
+    std::vector<ST> deltas;
+    deltas.reserve(count);
     T prev_src = 0;
     const char * const source_end = source + source_size;
     while (source < source_end)
     {
         T curr_src = unalignedLoad<T>(source);
-        unalignedStore<T>(dest, curr_src - prev_src);
+        deltas.push_back(static_cast<ST>(curr_src - prev_src));
         prev_src = curr_src;
-
         source += sizeof(T);
-        dest += sizeof(T);
     }
+    ST frame_of_reference = *std::min_element(deltas.cbegin(), deltas.cend());
+    // store frame of reference
+    unalignedStore<ST>(dest, frame_of_reference);
+    dest += sizeof(ST);
+    if (frame_of_reference != 0)
+    {
+        for (auto & delta : deltas)
+            delta -= frame_of_reference;
+    }
+    ST max_value = *std::max_element(deltas.cbegin(), deltas.cend());
+    UInt8 width = BitpackingPrimitives::minimumBitWidth(max_value);
+    // store width
+    unalignedStore<UInt8>(dest, width);
+    dest += sizeof(UInt8);
+    // if width == 0, skip bitpacking
+    if (width == 0)
+        return sizeof(ST) + sizeof(UInt8);
+    auto required_size = BitpackingPrimitives::getRequiredSize(count, width);
+    // after applying frame of reference, all deltas are bigger than 0.
+    BitpackingPrimitives::packBuffer(
+        reinterpret_cast<unsigned char *>(dest),
+        reinterpret_cast<const T *>(deltas.data()),
+        count,
+        width);
+    return sizeof(ST) + sizeof(UInt8) + required_size;
 }
 
-template <typename T>
-void ordinaryDecompressData(const char * source, UInt32 source_size, char * dest, UInt32 output_size)
+template <class T>
+void ApplyFrameOfReference(T * dst, T frame_of_reference, UInt32 count)
 {
-    if (source_size % sizeof(T) != 0)
-        throw Exception(
-            ErrorCodes::CANNOT_DECOMPRESS,
-            "Cannot decompress Delta-encoded data, data size {} is not aligned to {}",
-            source_size,
-            sizeof(T));
+    if (!frame_of_reference)
+        return;
 
-    const char * const output_end = dest + output_size;
+    UInt32 i = 0;
+    UInt32 misaligned_count = count;
+#if defined(__AVX2__)
+    misaligned_count = count % (sizeof(__m256i) / sizeof(T));
+#endif
+    for (; i < misaligned_count; ++i)
+    {
+        dst[i] += frame_of_reference;
+    }
+#if defined(__AVX2__)
+    for (; i < count; i += (sizeof(__m256i) / sizeof(T)))
+    {
+        // Load the data using SIMD
+        __m256i delta = _mm256_loadu_si256(reinterpret_cast<__m256i *>(dst + i));
+        // Perform vectorized addition
+        if constexpr (sizeof(T) == 1)
+        {
+            delta = _mm256_add_epi8(delta, _mm256_set1_epi8(frame_of_reference));
+        }
+        else if constexpr (sizeof(T) == 2)
+        {
+            delta = _mm256_add_epi16(delta, _mm256_set1_epi16(frame_of_reference));
+        }
+        else if constexpr (sizeof(T) == 4)
+        {
+            delta = _mm256_add_epi32(delta, _mm256_set1_epi32(frame_of_reference));
+        }
+        else if constexpr (sizeof(T) == 8)
+        {
+            delta = _mm256_add_epi64(delta, _mm256_set1_epi64x(frame_of_reference));
+        }
+        // Store the result back to memory
+        _mm256_storeu_si256(reinterpret_cast<__m256i *>(dst + i), delta);
+    }
+#endif
+}
+
+template <typename T>
+void ordinaryDeltaDecode(const char * source, UInt32 source_size, char * dest)
+{
     T accumulator{};
     const char * const source_end = source + source_size;
     while (source < source_end)
     {
         accumulator += unalignedLoad<T>(source);
-        if unlikely (dest + sizeof(accumulator) > output_end)
-            throw Exception(ErrorCodes::CANNOT_DECOMPRESS, "Cannot decompress delta-encoded data");
         unalignedStore<T>(dest, accumulator);
 
         source += sizeof(T);
@@ -92,20 +153,16 @@ void ordinaryDecompressData(const char * source, UInt32 source_size, char * dest
 }
 
 template <typename T>
-void decompressData(const char * source, UInt32 source_size, char * dest, UInt32 output_size)
+void DeltaDecode(const char * source, UInt32 source_size, char * dest)
 {
-    ordinaryDecompressData<T>(source, source_size, dest, output_size);
+    ordinaryDeltaDecode<T>(source, source_size, dest);
 }
 
 #if defined(__AVX2__)
 // Note: using SIMD to rewrite compress does not improve performance.
 
 template <>
-void decompressData<UInt32>(
-    const char * __restrict__ raw_source,
-    UInt32 raw_source_size,
-    char * __restrict__ raw_dest,
-    UInt32 /*raw_output_size*/)
+void DeltaDecode<UInt32>(const char * __restrict__ raw_source, UInt32 raw_source_size, char * __restrict__ raw_dest)
 {
     const auto * source = reinterpret_cast<const UInt32 *>(raw_source);
     auto source_size = raw_source_size / sizeof(UInt32);
@@ -129,11 +186,7 @@ void decompressData<UInt32>(
 }
 
 template <>
-void decompressData<UInt64>(
-    const char * __restrict__ raw_source,
-    UInt32 raw_source_size,
-    char * __restrict__ raw_dest,
-    UInt32 /*raw_output_size*/)
+void DeltaDecode<UInt64>(const char * __restrict__ raw_source, UInt32 raw_source_size, char * __restrict__ raw_dest)
 {
     const auto * source = reinterpret_cast<const UInt64 *>(raw_source);
     auto source_size = raw_source_size / sizeof(UInt64);
@@ -168,39 +221,81 @@ void decompressData<UInt64>(
 
 #endif
 
+template <typename T>
+void PForDecode(const char * source, UInt32 source_size, unsigned char * dest, UInt32 count)
+{
+    using ST = typename std::make_signed<T>::type;
+    ST frame_of_reference = unalignedLoad<ST>(source);
+    source += sizeof(ST);
+    auto width = unalignedLoad<UInt8>(source);
+    source += sizeof(UInt8);
+    const auto required_size = source_size - sizeof(ST) - sizeof(UInt8);
+    RUNTIME_CHECK(BitpackingPrimitives::getRequiredSize(count, width) == required_size);
+    if (width != 0)
+        BitpackingPrimitives::unPackBuffer<T>(dest, reinterpret_cast<const unsigned char *>(source), count, width);
+    ApplyFrameOfReference(reinterpret_cast<ST *>(dest), frame_of_reference, count);
+}
+
+template <typename T>
+void ordinaryDecompressData(const char * source, UInt32 source_size, char * dest, UInt32 output_size)
+{
+    const auto count = output_size / sizeof(T);
+    auto round_size = BitpackingPrimitives::roundUpToAlgorithmGroupSize(count);
+    unsigned char tmp_buffer[round_size * sizeof(T)];
+    PForDecode<T>(source, source_size, tmp_buffer, count);
+    ordinaryDeltaDecode<T>(reinterpret_cast<const char *>(tmp_buffer), output_size, dest);
+}
+
+template <typename T>
+void decompressData(const char * source, UInt32 source_size, char * dest, UInt32 output_size)
+{
+    ordinaryDecompressData<T>(source, source_size, dest, output_size);
+}
+
+template <>
+void decompressData<UInt32>(const char * source, UInt32 source_size, char * dest, UInt32 output_size)
+{
+    const auto count = output_size / sizeof(UInt32);
+    auto round_size = BitpackingPrimitives::roundUpToAlgorithmGroupSize(count);
+    unsigned char tmp_buffer[round_size * sizeof(UInt32)];
+    PForDecode<UInt32>(source, source_size, tmp_buffer, count);
+    DeltaDecode<UInt32>(reinterpret_cast<const char *>(tmp_buffer), output_size, dest);
+}
+
+template <>
+void decompressData<UInt64>(const char * source, UInt32 source_size, char * dest, UInt32 output_size)
+{
+    const auto count = output_size / sizeof(UInt64);
+    auto round_size = BitpackingPrimitives::roundUpToAlgorithmGroupSize(count);
+    unsigned char tmp_buffer[round_size * sizeof(UInt64)];
+    PForDecode<UInt64>(source, source_size, tmp_buffer, count);
+    DeltaDecode<UInt64>(reinterpret_cast<const char *>(tmp_buffer), output_size, dest);
+}
+
 } // namespace
 
-UInt32 CompressionCodecDelta::doCompressData(const char * source, UInt32 source_size, char * dest) const
+UInt32 CompressionCodecDeltaPFor::doCompressData(const char * source, UInt32 source_size, char * dest) const
 {
-    if unlikely (source_size % delta_bytes_size != 0)
-        throw Exception(
-            ErrorCodes::CANNOT_DECOMPRESS,
-            "source size {} is not aligned to {}",
-            source_size,
-            delta_bytes_size);
-    dest[0] = delta_bytes_size;
+    if unlikely (source_size % bytes_size != 0)
+        throw Exception(ErrorCodes::CANNOT_DECOMPRESS, "source size {} is not aligned to {}", source_size, bytes_size);
+    dest[0] = bytes_size;
     size_t start_pos = 1;
-    switch (delta_bytes_size)
+    switch (bytes_size)
     {
     case 1:
-        compressData<UInt8>(source, source_size, &dest[start_pos]);
-        break;
+        return 1 + compressData<UInt8>(source, source_size, &dest[start_pos]);
     case 2:
-        compressData<UInt16>(source, source_size, &dest[start_pos]);
-        break;
+        return 1 + compressData<UInt16>(source, source_size, &dest[start_pos]);
     case 4:
-        compressData<UInt32>(source, source_size, &dest[start_pos]);
-        break;
+        return 1 + compressData<UInt32>(source, source_size, &dest[start_pos]);
     case 8:
-        compressData<UInt64>(source, source_size, &dest[start_pos]);
-        break;
+        return 1 + compressData<UInt64>(source, source_size, &dest[start_pos]);
     default:
         throw Exception(ErrorCodes::CANNOT_COMPRESS, "Cannot compress Delta-encoded data. Unsupported bytes size");
     }
-    return 1 + source_size;
 }
 
-void CompressionCodecDelta::doDecompressData(
+void CompressionCodecDeltaPFor::doDecompressData(
     const char * source,
     UInt32 source_size,
     char * dest,
@@ -219,29 +314,32 @@ void CompressionCodecDelta::doDecompressData(
             "uncompressed size {} is not aligned to {}",
             uncompressed_size,
             bytes_size);
-    UInt32 output_size = uncompressed_size;
 
     UInt32 source_size_no_header = source_size - 1;
     switch (bytes_size)
     {
     case 1:
-        decompressData<UInt8>(&source[1], source_size_no_header, dest, output_size);
+        decompressData<UInt8>(&source[1], source_size_no_header, dest, uncompressed_size);
         break;
     case 2:
-        decompressData<UInt16>(&source[1], source_size_no_header, dest, output_size);
+        decompressData<UInt16>(&source[1], source_size_no_header, dest, uncompressed_size);
         break;
     case 4:
-        decompressData<UInt32>(&source[1], source_size_no_header, dest, output_size);
+        decompressData<UInt32>(&source[1], source_size_no_header, dest, uncompressed_size);
         break;
     case 8:
-        decompressData<UInt64>(&source[1], source_size_no_header, dest, output_size);
+        decompressData<UInt64>(&source[1], source_size_no_header, dest, uncompressed_size);
         break;
     default:
         throw Exception(ErrorCodes::CANNOT_DECOMPRESS, "Cannot decompress Delta-encoded data. Unsupported bytes size");
     }
 }
 
-void CompressionCodecDelta::ordinaryDecompress(const char * source, UInt32 source_size, char * dest, UInt32 dest_size)
+void CompressionCodecDeltaPFor::ordinaryDecompress(
+    const char * source,
+    UInt32 source_size,
+    char * dest,
+    UInt32 dest_size)
 {
     if unlikely (source_size < 2)
         throw Exception(ErrorCodes::CANNOT_DECOMPRESS, "Cannot decompress delta-encoded data. File has wrong header");

dbms/src/IO/Compression/CompressionCodecDelta.h → dbms/src/IO/Compression/CompressionCodecDeltaPFor.h

@@ -19,10 +19,10 @@
 namespace DB
 {
 
-class CompressionCodecDelta : public ICompressionCodec
+class CompressionCodecDeltaPFor : public ICompressionCodec
 {
 public:
-    explicit CompressionCodecDelta(UInt8 delta_bytes_size_);
+    explicit CompressionCodecDeltaPFor(UInt8 bytes_size_);
 
     UInt8 getMethodByte() const override;
 
@@ -36,13 +36,13 @@ class CompressionCodecDelta : public ICompressionCodec
     void doDecompressData(const char * source, UInt32 source_size, char * dest, UInt32 uncompressed_size)
         const override;
 
-    UInt32 getMaxCompressedDataSize(UInt32 uncompressed_size) const override { return uncompressed_size + 1; }
+    UInt32 getMaxCompressedDataSize(UInt32 uncompressed_size) const override;
 
-    bool isCompression() const override { return false; }
+    bool isCompression() const override { return true; }
     bool isGenericCompression() const override { return false; }
 
 private:
-    const UInt8 delta_bytes_size;
+    const UInt8 bytes_size;
 };
 
 } // namespace DB