Compression: support FOR & replace Delta with DeltaFOR (#8983)

ref #8982

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,8 @@ 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/CompressionCodecFOR.cpp
+    src/IO/Compression/CompressionCodecDeltaFOR.cpp
     src/Storages/DeltaMerge/BitmapFilter/BitmapFilter.cpp
     src/Storages/DeltaMerge/DMVersionFilterBlockInputStream.cpp
 )

dbms/src/Common/BitpackingPrimitives.h

@@ -72,6 +72,11 @@ class BitpackingPrimitives
         UInt8 width,
         bool skip_sign_extension = false)
     {
+        if (width == 0)
+        {
+            memset(dst, 0, count * sizeof(T));
+            return;
+        }
         for (size_t i = 0; i < count; i += BITPACKING_ALGORITHM_GROUP_SIZE)
         {
             unPackGroup<T>(dst + i * sizeof(T), src + (i * width) / 8, width, skip_sign_extension);
@@ -125,13 +130,12 @@ class BitpackingPrimitives
     }
 
     // round up to nearest multiple of BITPACKING_ALGORITHM_GROUP_SIZE
-    template <typename T>
-    constexpr static T roundUpToAlgorithmGroupSize(T num_to_round)
+    constexpr static size_t roundUpToAlgorithmGroupSize(size_t num_to_round)
     {
         static_assert(
             (BITPACKING_ALGORITHM_GROUP_SIZE & (BITPACKING_ALGORITHM_GROUP_SIZE - 1)) == 0,
             "BITPACKING_ALGORITHM_GROUP_SIZE must be a power of 2");
-        constexpr T mask = BITPACKING_ALGORITHM_GROUP_SIZE - 1;
+        constexpr size_t mask = BITPACKING_ALGORITHM_GROUP_SIZE - 1;
         return (num_to_round + mask) & ~mask;
     }
 

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

@@ -1,4 +1,4 @@
-// Copyright 2023 PingCAP, Inc.
+// Copyright 2024 PingCAP, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -12,13 +12,15 @@
 // 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/CompressionCodecDeltaFOR.h>
+#include <IO/Compression/CompressionCodecFOR.h>
 #include <IO/Compression/CompressionInfo.h>
 #include <common/likely.h>
 #include <common/unaligned.h>
 
+
 #if defined(__AVX2__)
 #include <immintrin.h>
 #endif
@@ -32,80 +34,76 @@ extern const int CANNOT_COMPRESS;
 extern const int CANNOT_DECOMPRESS;
 } // namespace ErrorCodes
 
-CompressionCodecDelta::CompressionCodecDelta(UInt8 delta_bytes_size_)
-    : delta_bytes_size(delta_bytes_size_)
+CompressionCodecDeltaFOR::CompressionCodecDeltaFOR(UInt8 bytes_size_)
+    : bytes_size(bytes_size_)
 {}
 
-UInt8 CompressionCodecDelta::getMethodByte() const
+UInt8 CompressionCodecDeltaFOR::getMethodByte() const
 {
-    return static_cast<UInt8>(CompressionMethodByte::Delta);
+    return static_cast<UInt8>(CompressionMethodByte::DeltaFOR);
 }
 
-namespace
+UInt32 CompressionCodecDeltaFOR::getMaxCompressedDataSize(UInt32 uncompressed_size) const
 {
-template <typename T>
-void compressData(const char * source, UInt32 source_size, char * dest)
+    /**
+     *|bytes_of_original_type|frame_of_reference|width(bits)  |bitpacked data|
+     *|1 bytes               |bytes_size        |sizeof(UInt8)|required size |
+     */
+    const size_t count = uncompressed_size / bytes_size;
+    return 1 + bytes_size + sizeof(UInt8) + BitpackingPrimitives::getRequiredSize(count, bytes_size * 8);
+}
+
+namespace
 {
-    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));
 
-    T prev_src = 0;
-    const char * const source_end = source + source_size;
-    while (source < source_end)
+template <std::integral T>
+void DeltaEncode(const T * source, UInt32 count, T * dest)
+{
+    T prev = 0;
+    for (UInt32 i = 0; i < count; ++i)
     {
-        T curr_src = unalignedLoad<T>(source);
-        unalignedStore<T>(dest, curr_src - prev_src);
-        prev_src = curr_src;
-
-        source += sizeof(T);
-        dest += sizeof(T);
+        T curr = source[i];
+        dest[i] = curr - prev;
+        prev = curr;
     }
 }
 
-template <typename T>
-void ordinaryDecompressData(const char * source, UInt32 source_size, char * dest, UInt32 output_size)
+template <std::integral T>
+UInt32 compressData(const char * source, UInt32 source_size, char * dest)
 {
-    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));
+    const auto count = source_size / sizeof(T);
+    DeltaEncode<T>(reinterpret_cast<const T *>(source), count, reinterpret_cast<T *>(dest));
+    // Cast deltas to signed type to better compress negative values.
+    using TS = typename std::make_signed<T>::type;
+    return CompressionCodecFOR::compressData<TS>(reinterpret_cast<TS *>(dest), count, dest);
+}
 
-    const char * const output_end = dest + output_size;
+template <std::integral 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);
         dest += sizeof(T);
     }
 }
 
-template <typename T>
-void decompressData(const char * source, UInt32 source_size, char * dest, UInt32 output_size)
+template <std::integral T>
+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 +127,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,46 +162,77 @@ void decompressData<UInt64>(
 
 #endif
 
+template <std::integral T>
+void ordinaryDecompressData(const char * source, UInt32 source_size, char * dest, UInt32 output_size)
+{
+    using TS = typename std::make_signed<T>::type;
+    CompressionCodecFOR::decompressData<TS>(source, source_size, dest, output_size);
+    ordinaryDeltaDecode<T>(dest, output_size, dest);
+}
+
+template <std::integral 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);
+    // Reserve enough space for the temporary buffer.
+    const auto required_size = round_size * sizeof(UInt32);
+    char tmp_buffer[required_size];
+    CompressionCodecFOR::decompressData<Int32>(source, source_size, tmp_buffer, required_size);
+    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);
+    const auto round_size = BitpackingPrimitives::roundUpToAlgorithmGroupSize(count);
+    // Reserve enough space for the temporary buffer.
+    const auto required_size = round_size * sizeof(UInt64);
+    char tmp_buffer[required_size];
+    CompressionCodecFOR::decompressData<Int64>(source, source_size, tmp_buffer, required_size);
+    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 CompressionCodecDeltaFOR::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");
+        throw Exception(ErrorCodes::CANNOT_COMPRESS, "Cannot compress DeltaFor-encoded data. Unsupported bytes size");
     }
-    return 1 + source_size;
 }
 
-void CompressionCodecDelta::doDecompressData(
+void CompressionCodecDeltaFOR::doDecompressData(
     const char * source,
     UInt32 source_size,
     char * dest,
     UInt32 uncompressed_size) const
 {
     if unlikely (source_size < 2)
-        throw Exception(ErrorCodes::CANNOT_DECOMPRESS, "Cannot decompress delta-encoded data. File has wrong header");
+        throw Exception(
+            ErrorCodes::CANNOT_DECOMPRESS,
+            "Cannot decompress DeltaFor-encoded data. File has wrong header");
 
     if (uncompressed_size == 0)
         return;
@@ -219,32 +244,39 @@ 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");
+        throw Exception(
+            ErrorCodes::CANNOT_DECOMPRESS,
+            "Cannot decompress DeltaFor-encoded data. Unsupported bytes size");
     }
 }
 
-void CompressionCodecDelta::ordinaryDecompress(const char * source, UInt32 source_size, char * dest, UInt32 dest_size)
+void CompressionCodecDeltaFOR::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");
+        throw Exception(
+            ErrorCodes::CANNOT_DECOMPRESS,
+            "Cannot decompress DeltaFor-encoded data. File has wrong header");
 
     if (dest_size == 0)
         return;
@@ -273,7 +305,9 @@ void CompressionCodecDelta::ordinaryDecompress(const char * source, UInt32 sourc
         ordinaryDecompressData<UInt64>(&source[1], source_size_no_header, dest, dest_size);
         break;
     default:
-        throw Exception(ErrorCodes::CANNOT_DECOMPRESS, "Cannot decompress Delta-encoded data. Unsupported bytes size");
+        throw Exception(
+            ErrorCodes::CANNOT_DECOMPRESS,
+            "Cannot decompress DeltaFor-encoded data. Unsupported bytes size");
     }
 }
 

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

@@ -1,4 +1,4 @@
-// Copyright 2023 PingCAP, Inc.
+// Copyright 2024 PingCAP, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -19,10 +19,10 @@
 namespace DB
 {
 
-class CompressionCodecDelta : public ICompressionCodec
+class CompressionCodecDeltaFOR : public ICompressionCodec
 {
 public:
-    explicit CompressionCodecDelta(UInt8 delta_bytes_size_);
+    explicit CompressionCodecDeltaFOR(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