Support Adaptive Lightweight Compression Algorithm

dbms/CMakeLists.txt

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

dbms/src/IO/Compression/CompressionCodecDeflateQpl.h

@@ -103,9 +103,6 @@ class CompressionCodecDeflateQpl final : public ICompressionCodec
     UInt8 getMethodByte() const override;
 
 protected:
-    bool isCompression() const override { return true; }
-    bool isGenericCompression() const override { return true; }
-
     UInt32 doCompressData(const char * source, UInt32 source_size, char * dest) const override;
     void doDecompressData(const char * source, UInt32 source_size, char * dest, UInt32 uncompressed_size)
         const override;

dbms/src/IO/Compression/CompressionCodecDeltaFOR.cpp

@@ -17,13 +17,13 @@
 #include <IO/Compression/CompressionCodecDeltaFOR.h>
 #include <IO/Compression/CompressionCodecFOR.h>
 #include <IO/Compression/CompressionInfo.h>
+#include <IO/Compression/CompressionSettings.h>
+#include <IO/Compression/EncodingUtil.h>
 #include <common/likely.h>
 #include <common/unaligned.h>
+#include <lz4.h>
 
-
-#if defined(__AVX2__)
-#include <immintrin.h>
-#endif
+#include <magic_enum.hpp>
 
 namespace DB
 {
@@ -34,8 +34,8 @@ extern const int CANNOT_COMPRESS;
 extern const int CANNOT_DECOMPRESS;
 } // namespace ErrorCodes
 
-CompressionCodecDeltaFOR::CompressionCodecDeltaFOR(UInt8 bytes_size_)
-    : bytes_size(bytes_size_)
+CompressionCodecDeltaFOR::CompressionCodecDeltaFOR(CompressionDataType data_type_)
+    : data_type(data_type_)
 {}
 
 UInt8 CompressionCodecDeltaFOR::getMethodByte() const
@@ -45,181 +45,74 @@ UInt8 CompressionCodecDeltaFOR::getMethodByte() const
 
 UInt32 CompressionCodecDeltaFOR::getMaxCompressedDataSize(UInt32 uncompressed_size) const
 {
-    /**
-     *|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
-{
-
-template <std::integral T>
-void DeltaEncode(const T * source, UInt32 count, T * dest)
-{
-    T prev = 0;
-    for (UInt32 i = 0; i < count; ++i)
+    switch (data_type)
     {
-        T curr = source[i];
-        dest[i] = curr - prev;
-        prev = curr;
-    }
-}
-
-template <std::integral T>
-UInt32 compressData(const char * source, UInt32 source_size, char * dest)
-{
-    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);
-}
-
-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);
-        unalignedStore<T>(dest, accumulator);
-
-        source += sizeof(T);
-        dest += sizeof(T);
-    }
-}
-
-template <std::integral T>
-void DeltaDecode(const char * source, UInt32 source_size, char * dest)
-{
-    ordinaryDeltaDecode<T>(source, source_size, dest);
-}
-
-#if defined(__AVX2__)
-// Note: using SIMD to rewrite compress does not improve performance.
-
-template <>
-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);
-    auto * dest = reinterpret_cast<UInt32 *>(raw_dest);
-    __m128i prev = _mm_setzero_si128();
-    size_t i = 0;
-    for (; i < source_size / 4; i++)
+    case CompressionDataType::Int8:
+    case CompressionDataType::Int16:
+    case CompressionDataType::Int32:
+    case CompressionDataType::Int64:
     {
-        auto curr = _mm_lddqu_si128(reinterpret_cast<const __m128i *>(source) + i);
-        const auto tmp1 = _mm_add_epi32(_mm_slli_si128(curr, 8), curr);
-        const auto tmp2 = _mm_add_epi32(_mm_slli_si128(tmp1, 4), tmp1);
-        prev = _mm_add_epi32(tmp2, _mm_shuffle_epi32(prev, 0xff));
-        _mm_storeu_si128(reinterpret_cast<__m128i *>(dest) + i, prev);
+        // |bytes_of_original_type|first_value|frame_of_reference|width(bits)  |bitpacked data|
+        // |1 bytes               |bytes_size |bytes_size        |sizeof(UInt8)|required size |
+        auto bytes_size = magic_enum::enum_integer(data_type);
+        const size_t deltas_count = uncompressed_size / bytes_size - 1;
+        return 1 + bytes_size * 2 + sizeof(UInt8) + BitpackingPrimitives::getRequiredSize(deltas_count, bytes_size * 8);
     }
-    uint32_t lastprev = _mm_extract_epi32(prev, 3);
-    for (i = 4 * i; i < source_size; ++i)
-    {
-        lastprev = lastprev + source[i];
-        dest[i] = lastprev;
+    default:
+        return 1 + LZ4_COMPRESSBOUND(uncompressed_size);
     }
 }
 
-template <>
-void DeltaDecode<UInt64>(const char * __restrict__ raw_source, UInt32 raw_source_size, char * __restrict__ raw_dest)
+namespace
 {
-    const auto * source = reinterpret_cast<const UInt64 *>(raw_source);
-    auto source_size = raw_source_size / sizeof(UInt64);
-    auto * dest = reinterpret_cast<UInt64 *>(raw_dest);
-    // AVX2 does not support shffule across 128-bit lanes, so we need to use permute.
-    __m256i prev = _mm256_setzero_si256();
-    __m256i zero = _mm256_setzero_si256();
-    size_t i = 0;
-    for (; i < source_size / 4; ++i)
-    {
-        // curr = {a0, a1, a2, a3}
-        auto curr = _mm256_loadu_si256(reinterpret_cast<const __m256i *>(source) + i);
-        // x0 = {0, a0, a1, a2}
-        auto x0 = _mm256_blend_epi32(_mm256_permute4x64_epi64(curr, 0b10010011), zero, 0b00000011);
-        // x1 = {a0, a01, a12, a23}
-        auto x1 = _mm256_add_epi64(curr, x0);
-        // x2 = {0, 0, a0, a01}
-        auto x2 = _mm256_permute2f128_si256(x1, x1, 0b00101000);
-        // prev = prev + {a0, a01, a012, a0123}
-        prev = _mm256_add_epi64(prev, _mm256_add_epi64(x1, x2));
-        _mm256_storeu_si256(reinterpret_cast<__m256i *>(dest) + i, prev);
-        // prev = {prev[3], prev[3], prev[3], prev[3]}
-        prev = _mm256_permute4x64_epi64(prev, 0b11111111);
-    }
-    UInt64 lastprev = _mm256_extract_epi64(prev, 3);
-    for (i = 4 * i; i < source_size; ++i)
-    {
-        lastprev += source[i];
-        dest[i] = lastprev;
-    }
-}
-
-#endif
 
 template <std::integral T>
-void ordinaryDecompressData(const char * source, UInt32 source_size, char * dest, UInt32 output_size)
+UInt32 compressData(const char * source, UInt32 source_size, char * dest)
 {
+    constexpr auto bytes_size = sizeof(T);
+    if unlikely (source_size % bytes_size != 0)
+        throw Exception(ErrorCodes::CANNOT_DECOMPRESS, "source size {} is not aligned to {}", source_size, bytes_size);
+    const auto count = source_size / bytes_size;
+    DB::Compression::deltaEncoding<T>(reinterpret_cast<const T *>(source), count, reinterpret_cast<T *>(dest));
+    if (unlikely(count == 1))
+        return bytes_size;
+    // Cast deltas to signed type to better compress negative values.
+    // For example, if we have a sequence of UInt8 values [3, 2, 1, 0], the deltas will be [3, -1, -1, -1]
+    // If we compress them as UInt8, we will get [3, 255, 255, 255], which is not optimal.
     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);
+    auto for_size = DB::CompressionCodecFOR::compressData<TS>(
+        reinterpret_cast<TS *>(dest + bytes_size),
+        source_size - bytes_size,
+        dest + bytes_size);
+    return bytes_size + for_size;
 }
 
 } // namespace
 
 UInt32 CompressionCodecDeltaFOR::doCompressData(const char * source, UInt32 source_size, char * dest) const
 {
-    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 (bytes_size)
+    dest[0] = magic_enum::enum_integer(data_type);
+    dest += 1;
+    switch (data_type)
     {
-    case 1:
-        return 1 + compressData<UInt8>(source, source_size, &dest[start_pos]);
-    case 2:
-        return 1 + compressData<UInt16>(source, source_size, &dest[start_pos]);
-    case 4:
-        return 1 + compressData<UInt32>(source, source_size, &dest[start_pos]);
-    case 8:
-        return 1 + compressData<UInt64>(source, source_size, &dest[start_pos]);
+    case CompressionDataType::Int8:
+        return 1 + compressData<UInt8>(source, source_size, dest);
+    case CompressionDataType::Int16:
+        return 1 + compressData<UInt16>(source, source_size, dest);
+    case CompressionDataType::Int32:
+        return 1 + compressData<UInt32>(source, source_size, dest);
+    case CompressionDataType::Int64:
+        return 1 + compressData<UInt64>(source, source_size, dest);
     default:
-        throw Exception(ErrorCodes::CANNOT_COMPRESS, "Cannot compress DeltaFor-encoded data. Unsupported bytes size");
+        auto success = LZ4_compress_fast(
+            source,
+            dest,
+            source_size,
+            LZ4_COMPRESSBOUND(source_size),
+            CompressionSetting::getDefaultLevel(CompressionMethod::LZ4));
+        if (unlikely(!success))
+            throw Exception("Cannot LZ4_compress_fast", ErrorCodes::CANNOT_COMPRESS);
+        return 1 + success;
     }
 }
 
@@ -234,80 +127,72 @@ void CompressionCodecDeltaFOR::doDecompressData(
             ErrorCodes::CANNOT_DECOMPRESS,
             "Cannot decompress DeltaFor-encoded data. File has wrong header");
 
-    if (uncompressed_size == 0)
+    if (unlikely(uncompressed_size == 0))
         return;
 
     UInt8 bytes_size = source[0];
-    if unlikely (uncompressed_size % bytes_size != 0)
-        throw Exception(
-            ErrorCodes::CANNOT_DECOMPRESS,
-            "uncompressed size {} is not aligned to {}",
-            uncompressed_size,
-            bytes_size);
+    auto data_type = magic_enum::enum_cast<CompressionDataType>(bytes_size);
+    RUNTIME_CHECK(data_type.has_value());
 
     UInt32 source_size_no_header = source_size - 1;
-    switch (bytes_size)
+    switch (data_type.value())
     {
-    case 1:
-        decompressData<UInt8>(&source[1], source_size_no_header, dest, uncompressed_size);
+    case CompressionDataType::Int8:
+        DB::Compression::deltaFORDecoding<UInt8>(&source[1], source_size_no_header, dest, uncompressed_size);
         break;
-    case 2:
-        decompressData<UInt16>(&source[1], source_size_no_header, dest, uncompressed_size);
+    case CompressionDataType::Int16:
+        DB::Compression::deltaFORDecoding<UInt16>(&source[1], source_size_no_header, dest, uncompressed_size);
         break;
-    case 4:
-        decompressData<UInt32>(&source[1], source_size_no_header, dest, uncompressed_size);
+    case CompressionDataType::Int32:
+        DB::Compression::deltaFORDecoding<UInt32>(&source[1], source_size_no_header, dest, uncompressed_size);
         break;
-    case 8:
-        decompressData<UInt64>(&source[1], source_size_no_header, dest, uncompressed_size);
+    case CompressionDataType::Int64:
+        DB::Compression::deltaFORDecoding<UInt64>(&source[1], source_size_no_header, dest, uncompressed_size);
         break;
     default:
-        throw Exception(
-            ErrorCodes::CANNOT_DECOMPRESS,
-            "Cannot decompress DeltaFor-encoded data. Unsupported bytes size");
+        if (unlikely(LZ4_decompress_safe(&source[1], dest, source_size_no_header, uncompressed_size) < 0))
+            throw Exception("Cannot LZ4_decompress_safe", ErrorCodes::CANNOT_DECOMPRESS);
+        break;
     }
 }
 
 void CompressionCodecDeltaFOR::ordinaryDecompress(
     const char * source,
     UInt32 source_size,
     char * dest,
-    UInt32 dest_size)
+    UInt32 uncompressed_size)
 {
     if unlikely (source_size < 2)
         throw Exception(
             ErrorCodes::CANNOT_DECOMPRESS,
             "Cannot decompress DeltaFor-encoded data. File has wrong header");
 
-    if (dest_size == 0)
+    if (unlikely(uncompressed_size == 0))
         return;
 
     UInt8 bytes_size = source[0];
-    if unlikely (dest_size % bytes_size != 0)
-        throw Exception(
-            ErrorCodes::CANNOT_DECOMPRESS,
-            "uncompressed size {} is not aligned to {}",
-            dest_size,
-            bytes_size);
+    auto data_type = magic_enum::enum_cast<CompressionDataType>(bytes_size);
+    RUNTIME_CHECK(data_type.has_value());
 
     UInt32 source_size_no_header = source_size - 1;
-    switch (bytes_size)
+    switch (data_type.value())
     {
-    case 1:
-        ordinaryDecompressData<UInt8>(&source[1], source_size_no_header, dest, dest_size);
+    case CompressionDataType::Int8:
+        DB::Compression::ordinaryDeltaFORDecoding<UInt8>(&source[1], source_size_no_header, dest, uncompressed_size);
         break;
-    case 2:
-        ordinaryDecompressData<UInt16>(&source[1], source_size_no_header, dest, dest_size);
+    case CompressionDataType::Int16:
+        DB::Compression::ordinaryDeltaFORDecoding<UInt16>(&source[1], source_size_no_header, dest, uncompressed_size);
         break;
-    case 4:
-        ordinaryDecompressData<UInt32>(&source[1], source_size_no_header, dest, dest_size);
+    case CompressionDataType::Int32:
+        DB::Compression::ordinaryDeltaFORDecoding<UInt32>(&source[1], source_size_no_header, dest, uncompressed_size);
         break;
-    case 8:
-        ordinaryDecompressData<UInt64>(&source[1], source_size_no_header, dest, dest_size);
+    case CompressionDataType::Int64:
+        DB::Compression::ordinaryDeltaFORDecoding<UInt64>(&source[1], source_size_no_header, dest, uncompressed_size);
         break;
     default:
-        throw Exception(
-            ErrorCodes::CANNOT_DECOMPRESS,
-            "Cannot decompress DeltaFor-encoded data. Unsupported bytes size");
+        if (unlikely(LZ4_decompress_safe(&source[1], dest, source_size_no_header, uncompressed_size) < 0))
+            throw Exception("Cannot LZ4_decompress_safe", ErrorCodes::CANNOT_DECOMPRESS);
+        break;
     }
 }