Fixed issues discovered during PR review

Author: Enmk

clickhouse/CMakeLists.txt

@@ -1,5 +1,4 @@
 SET ( clickhouse-cpp-lib-src
-    base/coded.cpp
     base/compressed.cpp
     base/input.cpp
     base/output.cpp

clickhouse/base/coded.cpp

@@ -7,13 +7,13 @@
 #include <stdexcept>
 #include <system_error>
 
-#include <iostream>
-
 namespace {
-static const size_t HEADER_SIZE = 9;
-static const size_t EXTRA_PREALLOCATE_COMPRESS_BUFFER = 15;
-static const uint8_t COMPRESSION_METHOD = 0x82;
-#define DBMS_MAX_COMPRESSED_SIZE    0x40000000ULL   // 1GB
+constexpr size_t HEADER_SIZE = 9;
+// see DB::CompressionMethodByte::LZ4 from src/Compression/CompressionInfo.h of ClickHouse project
+constexpr uint8_t COMPRESSION_METHOD = 0x82;
+// Documentation says that compression is faster when output buffer is larger than LZ4_compressBound estimation.
+constexpr size_t EXTRA_COMPRESS_BUFFER_SIZE = 4096;
+constexpr size_t DBMS_MAX_COMPRESSED_SIZE = 0x40000000ULL;   // 1GB
 }
 
 namespace clickhouse {
@@ -30,7 +30,7 @@ CompressedInput::~CompressedInput() {
 #else
         if (!std::uncaught_exceptions()) {
 #endif
-            throw std::runtime_error("some data was not readed");
+            throw std::runtime_error("some data was not read");
         }
     }
 }
@@ -59,8 +59,7 @@ bool CompressedInput::Decompress() {
     }
 
     if (method != COMPRESSION_METHOD) {
-        throw std::runtime_error("unsupported compression method " +
-                                 std::to_string(int(method)));
+        throw std::runtime_error("unsupported compression method " + std::to_string(int(method)));
     } else {
         if (!WireFormat::ReadFixed(input_, &compressed)) {
             return false;
@@ -105,24 +104,27 @@ bool CompressedInput::Decompress() {
 
 
 CompressedOutput::CompressedOutput(OutputStream * destination, size_t max_compressed_chunk_size)
-    : destination_(destination),
-      max_compressed_chunk_size_(max_compressed_chunk_size)
+    : destination_(destination)
+    , max_compressed_chunk_size_(max_compressed_chunk_size)
 {
+    PreallocateCompressBuffer(max_compressed_chunk_size);
 }
 
 CompressedOutput::~CompressedOutput() {
-        Flush();
+    Flush();
 }
 
 size_t CompressedOutput::DoWrite(const void* data, size_t len) {
     const size_t original_len = len;
-    const size_t max_chunk_size = max_compressed_chunk_size_ ? max_compressed_chunk_size_ : len;
+    // what if len > max_compressed_chunk_size_ ?
+    const size_t max_chunk_size = max_compressed_chunk_size_ > 0 ? max_compressed_chunk_size_ : len;
+    if (len > max_compressed_chunk_size_) {
+        PreallocateCompressBuffer(len);
+    }
 
-    while (len > 0)
-    {
+    while (len > 0) {
         auto to_compress = std::min(len, max_chunk_size);
-        if (!Compress(data, to_compress))
-            break;
+        Compress(data, to_compress);
 
         len -= to_compress;
         data = reinterpret_cast<const char*>(data) + to_compress;
@@ -135,16 +137,15 @@ void CompressedOutput::DoFlush() {
     destination_->Flush();
 }
 
-bool CompressedOutput::Compress(const void * data, size_t len) {
-
-    const size_t expected_out_size = LZ4_compressBound(len);
-    compressed_buffer_.resize(std::max(compressed_buffer_.size(), expected_out_size + HEADER_SIZE + EXTRA_PREALLOCATE_COMPRESS_BUFFER));
-
-    const int compressed_size = LZ4_compress_default(
+void CompressedOutput::Compress(const void * data, size_t len) {
+    const auto compressed_size = LZ4_compress_default(
             (const char*)data,
             (char*)compressed_buffer_.data() + HEADER_SIZE,
             len,
             compressed_buffer_.size() - HEADER_SIZE);
+    if (compressed_size <= 0)
+        throw std::runtime_error("Failed to compress chunk of " + std::to_string(len) + " bytes, "
+                "LZ4 error: " + std::to_string(compressed_size));
 
     {
         auto header = compressed_buffer_.data();
@@ -160,7 +161,14 @@ bool CompressedOutput::Compress(const void * data, size_t len) {
     WireFormat::WriteBytes(destination_, compressed_buffer_.data(), compressed_size + HEADER_SIZE);
 
     destination_->Flush();
-    return true;
+}
+
+void CompressedOutput::PreallocateCompressBuffer(size_t input_size) {
+    const auto estimated_compressed_buffer_size = LZ4_compressBound(input_size);
+    if (estimated_compressed_buffer_size <= 0)
+        throw std::runtime_error("Failed to estimate compressed buffer size, LZ4 error: " + std::to_string(estimated_compressed_buffer_size));
+
+    compressed_buffer_.resize(estimated_compressed_buffer_size + HEADER_SIZE + EXTRA_COMPRESS_BUFFER_SIZE);
 }
 
 }

clickhouse/base/coded.h

@@ -8,7 +8,7 @@ namespace clickhouse {
 
 class CompressedInput : public ZeroCopyInput {
 public:
-     CompressedInput(InputStream* input);
+    CompressedInput(InputStream* input);
     ~CompressedInput();
 
 protected:
@@ -31,13 +31,15 @@ class CompressedOutput : public OutputStream {
 protected:
     size_t DoWrite(const void* data, size_t len) override;
     void DoFlush() override;
-    bool Compress(const void * data, size_t len);
 
+private:
+    void Compress(const void * data, size_t len);
+    void PreallocateCompressBuffer(size_t input_size);
 
 private:
     OutputStream * destination_;
+    const size_t max_compressed_chunk_size_;
     Buffer compressed_buffer_;
-    size_t max_compressed_chunk_size_;
 };
 
 }

clickhouse/base/compressed.cpp

@@ -64,6 +64,7 @@ class SSLSocketInput : public InputStream {
     bool Skip(size_t /*bytes*/) override {
         return false;
     }
+
 protected:
     size_t DoRead(void* buf, size_t len) override;
 

clickhouse/base/compressed.h

@@ -6,7 +6,7 @@
 #include <stdexcept>
 
 namespace {
-static const int MAX_VARINT_BYTES = 10;
+constexpr int MAX_VARINT_BYTES = 10;
 }
 
 namespace clickhouse {
@@ -22,12 +22,7 @@ bool WireFormat::ReadAll(InputStream * input, void* buf, size_t len) {
         len -= read_previously;
     }
 
-    // true if all was read successfully
     return !len;
-//    if (len) {
-//        throw std::runtime_error("Failed to read " + std::to_string(original_len)
-//                + " bytes, only read " + std::to_string(original_len - len));
-//    }
 }
 
 void WireFormat::WriteAll(OutputStream* output, const void* buf, size_t len) {
@@ -52,7 +47,7 @@ bool WireFormat::ReadVarint64(InputStream* input, uint64_t* value) {
     *value = 0;
 
     for (size_t i = 0; i < MAX_VARINT_BYTES; ++i) {
-        uint8_t byte;
+        uint8_t byte = 0;
 
         if (!input->ReadByte(&byte)) {
             return false;
@@ -90,7 +85,7 @@ void WireFormat::WriteVarint64(OutputStream* output, uint64_t value) {
 }
 
 bool WireFormat::SkipString(InputStream* input) {
-    uint64_t len;
+    uint64_t len = 0;
 
     if (ReadVarint64(input, &len)) {
         if (len > 0x00FFFFFFULL)

clickhouse/base/sslsocket.h

@@ -35,8 +35,7 @@ inline bool WireFormat::ReadFixed(InputStream* input, T* value) {
 }
 
 inline bool WireFormat::ReadString(InputStream* input, std::string* value) {
-    uint64_t len;
-
+    uint64_t len = 0;
     if (ReadVarint64(input, &len)) {
         if (len > 0x00FFFFFFULL) {
             return false;

clickhouse/base/wire_format.cpp

@@ -16,7 +16,7 @@ TEST(CodedStreamCase, Varint64) {
 
     {
         ArrayInput input(buf.data(), buf.size());
-        uint64_t value;
+        uint64_t value = 0;
         ASSERT_TRUE(WireFormat::ReadVarint64(&input, &value));
         ASSERT_EQ(value, 18446744071965638648ULL);
     }