This is an automated email from the ASF dual-hosted git repository.
xndai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/orc.git
The following commit(s) were added to refs/heads/master by this push:
new 599a784 ORC-395: Support ZSTD in C++ writer/reader
599a784 is described below
commit 599a78402b8854e09a785260e23d439290a183e3
Author: Gang Wu <[email protected]>
AuthorDate: Tue Nov 6 09:10:10 2018 -0800
ORC-395: Support ZSTD in C++ writer/reader
* ORC-395: Support ZSTD in C++ writer/reader
* add MSVC support for zstd build
* fix MSVC build
* refactor ensureHeader function
* add output check
---
c++/src/CMakeLists.txt | 2 +
c++/src/Compression.cc | 207 ++++++++++++++++++++++++++++++--
c++/test/TestCompression.cc | 30 ++++-
cmake_modules/FindZSTD.cmake | 68 +++++++++++
cmake_modules/ThirdpartyToolchain.cmake | 54 +++++++++
5 files changed, 345 insertions(+), 16 deletions(-)
diff --git a/c++/src/CMakeLists.txt b/c++/src/CMakeLists.txt
index 7358361..72d408d 100644
--- a/c++/src/CMakeLists.txt
+++ b/c++/src/CMakeLists.txt
@@ -169,6 +169,7 @@ include_directories (
${SNAPPY_INCLUDE_DIR}
${LZ4_INCLUDE_DIR}
${LIBHDFSPP_INCLUDE_DIR}
+ ${ZSTD_INCLUDE_DIR}
)
add_custom_command(OUTPUT orc_proto.pb.h orc_proto.pb.cc
@@ -220,6 +221,7 @@ target_link_libraries (orc
zlib
snappy
lz4
+ zstd
${LIBHDFSPP_LIBRARIES}
)
diff --git a/c++/src/Compression.cc b/c++/src/Compression.cc
index 6bd0220..d1f7c4f 100644
--- a/c++/src/Compression.cc
+++ b/c++/src/Compression.cc
@@ -28,6 +28,7 @@
#include <sstream>
#include "zlib.h"
+#include "zstd.h"
#include "wrap/snappy-wrapper.h"
@@ -57,6 +58,9 @@ namespace orc {
buffer[2] = static_cast<char>(compressedSize >> 15);
}
+ // ensure enough room for compression block header
+ void ensureHeader();
+
// Buffer to hold uncompressed data until user calls Next()
DataBuffer<unsigned char> rawInputBuffer;
@@ -117,6 +121,22 @@ namespace orc {
static_cast<uint64_t>(outputSize - outputPosition);
}
+ void CompressionStreamBase::ensureHeader() {
+ // adjust 3 bytes for the compression header
+ if (outputPosition + 3 >= outputSize) {
+ int newPosition = outputPosition + 3 - outputSize;
+ if (!BufferedOutputStream::Next(
+ reinterpret_cast<void **>(&outputBuffer),
+ &outputSize)) {
+ throw std::runtime_error(
+ "Failed to get next output buffer from output stream.");
+ }
+ outputPosition = newPosition;
+ } else {
+ outputPosition += 3;
+ }
+ }
+
/**
* Streaming compression base class
*/
@@ -151,19 +171,7 @@ namespace orc {
bool CompressionStream::Next(void** data, int*size) {
if (bufferSize != 0) {
- // adjust 3 bytes for the compression header
- if (outputPosition + 3 >= outputSize) {
- int newPosition = outputPosition + 3 - outputSize;
- if (!BufferedOutputStream::Next(
- reinterpret_cast<void **>(&outputBuffer),
- &outputSize)) {
- throw std::runtime_error(
- "Failed to get next output buffer from output stream.");
- }
- outputPosition = newPosition;
- } else {
- outputPosition += 3;
- }
+ ensureHeader();
uint64_t totalCompressedSize = doStreamingCompression();
@@ -899,6 +907,169 @@ DIAGNOSTIC_POP
return static_cast<uint64_t>(result);
}
+ /**
+ * Block compression base class
+ */
+ class BlockCompressionStream: public CompressionStreamBase {
+ public:
+ BlockCompressionStream(OutputStream * outStream,
+ int compressionLevel,
+ uint64_t capacity,
+ uint64_t blockSize,
+ MemoryPool& pool)
+ : CompressionStreamBase(outStream,
+ compressionLevel,
+ capacity,
+ blockSize,
+ pool)
+ , compressorBuffer(pool) {
+ // PASS
+ }
+
+ virtual bool Next(void** data, int*size) override;
+ virtual std::string getName() const override = 0;
+
+ protected:
+ // compresses a block and returns the compressed size
+ virtual uint64_t doBlockCompression() = 0;
+
+ // return maximum possible compression size for allocating space for
+ // compressorBuffer below
+ virtual uint64_t estimateMaxCompressionSize() = 0;
+
+ // should allocate max possible compressed size
+ DataBuffer<unsigned char> compressorBuffer;
+ };
+
+ bool BlockCompressionStream::Next(void** data, int*size) {
+ if (bufferSize != 0) {
+ ensureHeader();
+
+ // perform compression
+ size_t totalCompressedSize = doBlockCompression();
+
+ const unsigned char * dataToWrite = nullptr;
+ int totalSizeToWrite = 0;
+ char * header = outputBuffer + outputPosition - 3;
+
+ if (totalCompressedSize >= static_cast<size_t>(bufferSize)) {
+ writeHeader(header, static_cast<size_t>(bufferSize), true);
+ dataToWrite = rawInputBuffer.data();
+ totalSizeToWrite = bufferSize;
+ } else {
+ writeHeader(header, totalCompressedSize, false);
+ dataToWrite = compressorBuffer.data();
+ totalSizeToWrite = static_cast<int>(totalCompressedSize);
+ }
+
+ char * dst = header + 3;
+ while (totalSizeToWrite > 0) {
+ if (outputPosition == outputSize) {
+ if (!BufferedOutputStream::Next(reinterpret_cast<void
**>(&outputBuffer),
+ &outputSize)) {
+ throw std::logic_error(
+ "Failed to get next output buffer from output stream.");
+ }
+ outputPosition = 0;
+ dst = outputBuffer;
+ } else if (outputPosition > outputSize) {
+ // this will unlikely happen, but we have seen a few on zstd v1.1.0
+ throw std::logic_error("Write to an out-of-bound place!");
+ }
+
+ int sizeToWrite = std::min(totalSizeToWrite, outputSize -
outputPosition);
+ std::memcpy(dst, dataToWrite, static_cast<size_t>(sizeToWrite));
+
+ outputPosition += sizeToWrite;
+ dataToWrite += sizeToWrite;
+ totalSizeToWrite -= sizeToWrite;
+ dst += sizeToWrite;
+ }
+ }
+
+ *data = rawInputBuffer.data();
+ *size = static_cast<int>(rawInputBuffer.size());
+ bufferSize = *size;
+ compressorBuffer.resize(estimateMaxCompressionSize());
+
+ return true;
+ }
+
+ /**
+ * ZSTD block compression
+ */
+ class ZSTDCompressionStream: public BlockCompressionStream {
+ public:
+ ZSTDCompressionStream(OutputStream * outStream,
+ int compressionLevel,
+ uint64_t capacity,
+ uint64_t blockSize,
+ MemoryPool& pool)
+ : BlockCompressionStream(outStream,
+ compressionLevel,
+ capacity,
+ blockSize,
+ pool) {
+ // PASS
+ }
+
+ virtual std::string getName() const override {
+ return "ZstdCompressionStream";
+ }
+
+ protected:
+ virtual uint64_t doBlockCompression() override;
+
+ virtual uint64_t estimateMaxCompressionSize() override {
+ return ZSTD_compressBound(static_cast<size_t>(bufferSize));
+ }
+ };
+
+ uint64_t ZSTDCompressionStream::doBlockCompression() {
+ return ZSTD_compress(compressorBuffer.data(),
+ compressorBuffer.size(),
+ rawInputBuffer.data(),
+ static_cast<size_t>(bufferSize),
+ level);
+ }
+
+ /**
+ * ZSTD block decompression
+ */
+ class ZSTDDecompressionStream: public BlockDecompressionStream {
+ public:
+ ZSTDDecompressionStream(std::unique_ptr<SeekableInputStream> inStream,
+ size_t blockSize,
+ MemoryPool& pool)
+ : BlockDecompressionStream(std::move(inStream),
+ blockSize,
+ pool) {
+ // PASS
+ }
+
+ std::string getName() const override {
+ std::ostringstream result;
+ result << "zstd(" << getStreamName() << ")";
+ return result.str();
+ }
+
+ protected:
+ virtual uint64_t decompress(const char *input,
+ uint64_t length,
+ char *output,
+ size_t maxOutputLength) override;
+ };
+
+ uint64_t ZSTDDecompressionStream::decompress(const char *input,
+ uint64_t length,
+ char *output,
+ size_t maxOutputLength) {
+ return static_cast<uint64_t>(ZSTD_decompress(output,
+ maxOutputLength,
+ input,
+ length));
+ }
+
std::unique_ptr<BufferedOutputStream>
createCompressor(
CompressionKind kind,
@@ -920,6 +1091,13 @@ DIAGNOSTIC_POP
(new ZlibCompressionStream(
outStream, level, bufferCapacity, compressionBlockSize, pool));
}
+ case CompressionKind_ZSTD: {
+ int level = (strategy == CompressionStrategy_SPEED) ?
+ 1 : ZSTD_CLEVEL_DEFAULT;
+ return std::unique_ptr<BufferedOutputStream>
+ (new ZSTDCompressionStream(
+ outStream, level, bufferCapacity, compressionBlockSize, pool));
+ }
case CompressionKind_SNAPPY:
case CompressionKind_LZO:
case CompressionKind_LZ4:
@@ -948,6 +1126,9 @@ DIAGNOSTIC_POP
case CompressionKind_LZ4:
return std::unique_ptr<SeekableInputStream>
(new Lz4DecompressionStream(std::move(input), blockSize, pool));
+ case CompressionKind_ZSTD:
+ return std::unique_ptr<SeekableInputStream>
+ (new ZSTDDecompressionStream(std::move(input), blockSize, pool));
default: {
std::ostringstream buffer;
buffer << "Unknown compression codec " << kind;
diff --git a/c++/test/TestCompression.cc b/c++/test/TestCompression.cc
index dcc25e8..9ca2160 100644
--- a/c++/test/TestCompression.cc
+++ b/c++/test/TestCompression.cc
@@ -137,7 +137,7 @@ namespace orc {
compress_original_string(CompressionKind_ZLIB);
}
- void zlib_compress_simple_repeated_string(orc::CompressionKind kind) {
+ void compress_simple_repeated_string(orc::CompressionKind kind) {
MemoryOutputStream memStream(DEFAULT_MEM_STREAM_SIZE);
MemoryPool * pool = getDefaultPool();
@@ -161,8 +161,8 @@ namespace orc {
*pool);
}
- TEST(TestCompression, zlib_compress_simple_repeated_string) {
- zlib_compress_simple_repeated_string(CompressionKind_ZLIB);
+ TEST(TestCompression, compress_simple_repeated_string) {
+ compress_simple_repeated_string(CompressionKind_ZLIB);
}
void compress_two_blocks(orc::CompressionKind kind) {
@@ -312,4 +312,28 @@ namespace orc {
TEST(TestCompression, zlib_protobuff_compression) {
protobuff_compression(CompressionKind_ZLIB, proto::ZLIB);
}
+
+ TEST(Compression, zstd_compress_original_string) {
+ compress_original_string(CompressionKind_ZSTD);
+ }
+
+ TEST(Compression, zstd_compress_simple_repeated_string) {
+ compress_simple_repeated_string(CompressionKind_ZSTD);
+ }
+
+ TEST(Compression, zstd_compress_two_blocks) {
+ compress_two_blocks(CompressionKind_ZSTD);
+ }
+
+ TEST(Compression, zstd_compress_random_letters) {
+ compress_random_letters(CompressionKind_ZSTD);
+ }
+
+ TEST(Compression, zstd_compress_random_bytes) {
+ compress_random_bytes(CompressionKind_ZSTD);
+ }
+
+ TEST(Compression, zstd_protobuff_compression) {
+ protobuff_compression(CompressionKind_ZSTD, proto::ZSTD);
+ }
}
diff --git a/cmake_modules/FindZSTD.cmake b/cmake_modules/FindZSTD.cmake
new file mode 100644
index 0000000..3dba072
--- /dev/null
+++ b/cmake_modules/FindZSTD.cmake
@@ -0,0 +1,68 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# ZSTD_HOME environmental variable is used to check for Zstd headers and
static library
+
+# ZSTD_INCLUDE_DIR: directory containing headers
+# ZSTD_LIBS: directory containing zstd libraries
+# ZSTD_STATIC_LIB: path to libzstd.a
+# ZSTD_FOUND: whether zstd has been found
+
+if( NOT "${ZSTD_HOME}" STREQUAL "")
+ file (TO_CMAKE_PATH "${ZSTD_HOME}" _zstd_path)
+endif()
+
+message (STATUS "ZSTD_HOME: ${ZSTD_HOME}")
+
+find_path (ZSTD_INCLUDE_DIR zstd.h HINTS
+ ${_zstd_path}
+ NO_DEFAULT_PATH
+ PATH_SUFFIXES "include")
+
+find_library (ZSTD_LIBRARIES NAMES zstd HINTS
+ ${_zstd_path}
+ PATH_SUFFIXES "lib")
+
+if (ZSTD_INCLUDE_DIR AND ZSTD_LIBRARIES)
+ set (ZSTD_FOUND TRUE)
+ get_filename_component (ZSTD_LIBS ${ZSTD_LIBRARIES} PATH)
+ set (ZSTD_HEADER_NAME zstd.h)
+ set (ZSTD_HEADER ${ZSTD_INCLUDE_DIR}/${ZSTD_HEADER_NAME})
+ set (ZSTD_LIB_NAME zstd)
+ set (ZSTD_STATIC_LIB
${ZSTD_LIBS}/${CMAKE_STATIC_LIBRARY_PREFIX}${ZSTD_LIB_NAME}${CMAKE_STATIC_LIBRARY_SUFFIX})
+else ()
+ set (ZSTD_FOUND FALSE)
+endif ()
+
+if (ZSTD_FOUND)
+ message (STATUS "Found the zstd header: ${ZSTD_HEADER}")
+ message (STATUS "Found the zstd library: ${ZSTD_STATIC_LIB}")
+else()
+ if (_ZSTD_path)
+ set (ZSTD_ERR_MSG "Could not find zstd. Looked in ${_zstd_path}.")
+ else ()
+ set (ZSTD_ERR_MSG "Could not find zstd in system search paths.")
+ endif()
+
+ if (ZSTD_FIND_REQUIRED)
+ message (FATAL_ERROR "${ZSTD_ERR_MSG}")
+ else ()
+ message (STATUS "${ZSTD_ERR_MSG}")
+ endif ()
+endif()
+
+mark_as_advanced (
+ ZSTD_INCLUDE_DIR
+ ZSTD_STATIC_LIB
+ ZSTD_LIBS
+ ZSTD_LIBRARIES
+)
diff --git a/cmake_modules/ThirdpartyToolchain.cmake
b/cmake_modules/ThirdpartyToolchain.cmake
index 8aabf69..9bf97de 100644
--- a/cmake_modules/ThirdpartyToolchain.cmake
+++ b/cmake_modules/ThirdpartyToolchain.cmake
@@ -15,6 +15,7 @@ set(SNAPPY_VERSION "1.1.7")
set(ZLIB_VERSION "1.2.11")
set(GTEST_VERSION "1.8.0")
set(PROTOBUF_VERSION "3.5.1")
+set(ZSTD_VERSION "1.3.5")
set(THIRDPARTY_DIR "${CMAKE_BINARY_DIR}/c++/libs/thirdparty")
set(THIRDPARTY_LOG_OPTIONS LOG_CONFIGURE 1
@@ -44,6 +45,10 @@ if (DEFINED ENV{PROTOBUF_HOME})
set (PROTOBUF_HOME "$ENV{PROTOBUF_HOME}")
endif ()
+if (DEFINED ENV{ZSTD_HOME})
+ set (ZSTD_HOME "$ENV{ZSTD_HOME}")
+endif ()
+
if (DEFINED ENV{GTEST_HOME})
set (GTEST_HOME "$ENV{GTEST_HOME}")
endif ()
@@ -125,6 +130,55 @@ if (ZLIB_VENDORED)
endif ()
# ----------------------------------------------------------------------
+# Zstd
+
+if (NOT "${ZSTD_HOME}" STREQUAL "")
+ find_package (zstd REQUIRED)
+ set(ZSTD_VENDORED FALSE)
+else ()
+ set(ZSTD_HOME "${THIRDPARTY_DIR}/zstd_ep-install")
+ set(ZSTD_INCLUDE_DIR "${ZSTD_HOME}/include")
+ if (MSVC)
+ set(ZSTD_STATIC_LIB_NAME zstd_static)
+ if (${UPPERCASE_BUILD_TYPE} STREQUAL "DEBUG")
+ set(ZSTD_STATIC_LIB_NAME ${ZSTD_STATIC_LIB_NAME})
+ endif ()
+ else ()
+ set(ZSTD_STATIC_LIB_NAME zstd)
+ endif ()
+ set(ZSTD_STATIC_LIB
"${ZSTD_HOME}/lib/${CMAKE_STATIC_LIBRARY_PREFIX}${ZSTD_STATIC_LIB_NAME}${CMAKE_STATIC_LIBRARY_SUFFIX}")
+ set(ZSTD_CMAKE_ARGS -DCMAKE_INSTALL_PREFIX=${ZSTD_HOME}
+ -DBUILD_SHARED_LIBS=OFF)
+
+ if (CMAKE_VERSION VERSION_GREATER "3.7")
+ set(ZSTD_CONFIGURE SOURCE_SUBDIR "build/cmake" CMAKE_ARGS
${ZSTD_CMAKE_ARGS})
+ else()
+ set(ZSTD_CONFIGURE CONFIGURE_COMMAND "${THIRDPARTY_CONFIGURE_COMMAND}"
${ZSTD_CMAKE_ARGS}
+
"${CMAKE_CURRENT_BINARY_DIR}/zstd_ep-prefix/src/zstd_ep/build/cmake")
+ endif()
+
+ ExternalProject_Add(zstd_ep
+ URL
"https://github.com/facebook/zstd/archive/v${ZSTD_VERSION}.tar.gz";
+ ${ZSTD_CONFIGURE}
+ ${THIRDPARTY_LOG_OPTIONS}
+ BUILD_BYPRODUCTS ${ZSTD_STATIC_LIB})
+
+ set(ZSTD_VENDORED TRUE)
+endif ()
+
+include_directories (SYSTEM ${ZSTD_INCLUDE_DIR})
+add_library (zstd STATIC IMPORTED)
+set_target_properties (zstd PROPERTIES IMPORTED_LOCATION ${ZSTD_STATIC_LIB})
+
+if (ZSTD_VENDORED)
+ add_dependencies (zstd zstd_ep)
+ if (INSTALL_VENDORED_LIBS)
+ install(FILES "${ZSTD_STATIC_LIB}"
+ DESTINATION "lib")
+ endif ()
+endif ()
+
+# ----------------------------------------------------------------------
# LZ4
if (NOT "${LZ4_HOME}" STREQUAL "")
