[orc] branch main updated: ORC-377: [C++] Add SnappyCompressionStream

[gangwu]

This is an automated email from the ASF dual-hosted git repository.

gangwu pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/orc.git


The following commit(s) were added to refs/heads/main by this push:
     new 39629d6  ORC-377: [C++] Add SnappyCompressionStream
39629d6 is described below

commit 39629d6e7a3816a4b1d978527563a3bfbe3ac80f
Author: noirello <noire...@gmail.com>
AuthorDate: Thu Jun 10 04:34:48 2021 +0200

    ORC-377: [C++] Add SnappyCompressionStream
    
    This close #713
---
 c++/src/Compression.cc      | 50 ++++++++++++++++++++++++++++++++++++++++++++-
 c++/test/TestCompression.cc | 25 +++++++++++++++++++++++
 2 files changed, 74 insertions(+), 1 deletion(-)

diff --git a/c++/src/Compression.cc b/c++/src/Compression.cc
index 96d07d5..fa94e91 100644
--- a/c++/src/Compression.cc
+++ b/c++/src/Compression.cc
@@ -1031,6 +1031,49 @@ DIAGNOSTIC_POP
   }
 
   /**
+   * Snappy block compression
+   */
+  class SnappyCompressionStream: public BlockCompressionStream {
+  public:
+    SnappyCompressionStream(OutputStream * outStream,
+                        int compressionLevel,
+                        uint64_t capacity,
+                        uint64_t blockSize,
+                        MemoryPool& pool)
+                        : BlockCompressionStream(outStream,
+                                                 compressionLevel,
+                                                 capacity,
+                                                 blockSize,
+                                                 pool) {
+    }
+
+    virtual std::string getName() const override {
+      return "SnappyCompressionStream";
+    }
+    
+    virtual ~SnappyCompressionStream() override {
+      // PASS
+    }
+
+  protected:
+    virtual uint64_t doBlockCompression() override;
+
+    virtual uint64_t estimateMaxCompressionSize() override {
+      return static_cast<uint64_t>
+        (snappy::MaxCompressedLength(static_cast<size_t>(bufferSize)));
+    }
+  };
+
+  uint64_t SnappyCompressionStream::doBlockCompression() {
+    size_t compressedLength;
+    snappy::RawCompress(reinterpret_cast<const char*>(rawInputBuffer.data()),
+                        static_cast<size_t>(bufferSize),
+                        reinterpret_cast<char*>(compressorBuffer.data()),
+                        &compressedLength);
+    return static_cast<uint64_t>(compressedLength);
+  }
+
+  /**
    * ZSTD block compression
    */
   class ZSTDCompressionStream: public BlockCompressionStream {
@@ -1204,7 +1247,12 @@ DIAGNOSTIC_PUSH
         (new Lz4CompressionSteam(
           outStream, level, bufferCapacity, compressionBlockSize, pool));
     }
-    case CompressionKind_SNAPPY:
+    case CompressionKind_SNAPPY: {
+      int level = 0;
+      return std::unique_ptr<BufferedOutputStream>
+        (new SnappyCompressionStream(
+          outStream, level, bufferCapacity, compressionBlockSize, pool));
+    }
     case CompressionKind_LZO:
     default:
       throw NotImplementedYet("compression codec");
diff --git a/c++/test/TestCompression.cc b/c++/test/TestCompression.cc
index f7d3b09..e6fc997 100644
--- a/c++/test/TestCompression.cc
+++ b/c++/test/TestCompression.cc
@@ -362,6 +362,30 @@ namespace orc {
     protobuff_compression(CompressionKind_LZ4, proto::LZ4);
   }
 
+  TEST(Compression, snappy_compress_original_string) {
+    compress_original_string(CompressionKind_SNAPPY);
+  }
+
+  TEST(Compression, snappy_compress_simple_repeated_string) {
+    compress_simple_repeated_string(CompressionKind_SNAPPY);
+  }
+
+  TEST(Compression, snappy_compress_two_blocks) {
+    compress_two_blocks(CompressionKind_SNAPPY);
+  }
+
+  TEST(Compression, snappy_compress_random_letters) {
+    compress_random_letters(CompressionKind_SNAPPY);
+  }
+
+  TEST(Compression, snappy_compress_random_bytes) {
+    compress_random_bytes(CompressionKind_SNAPPY);
+  }
+
+  TEST(Compression, snappy_protobuff_compression) {
+    protobuff_compression(CompressionKind_SNAPPY, proto::SNAPPY);
+  }
+
   void testSeekDecompressionStream(CompressionKind kind) {
     MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
     MemoryPool * pool = getDefaultPool();
@@ -427,5 +451,6 @@ namespace orc {
     testSeekDecompressionStream(CompressionKind_ZSTD);
     testSeekDecompressionStream(CompressionKind_ZLIB);
     testSeekDecompressionStream(CompressionKind_LZ4);
+    testSeekDecompressionStream(CompressionKind_SNAPPY);
   }
 }