add zstd

CMakeLists.txt

@@ -141,6 +141,7 @@ option(VELOX_ENABLE_ARROW "Enable Arrow support" OFF)
 option(VELOX_ENABLE_REMOTE_FUNCTIONS "Enable remote function support" OFF)
 option(VELOX_ENABLE_CCACHE "Use ccache if installed." ON)
 option(VELOX_ENABLE_COMPRESSION_LZ4 "Enable Lz4 compression support." OFF)
+option(VELOX_ENABLE_COMPRESSION_ZSTD "Enable Zstd compression support." OFF)
 
 option(VELOX_BUILD_TEST_UTILS "Builds Velox test utilities" OFF)
 option(VELOX_BUILD_VECTOR_TEST_UTILS "Builds Velox vector test utilities" OFF)
@@ -176,6 +177,7 @@ endif()
 
 if(${VELOX_BUILD_MINIMAL_WITH_DWIO} OR ${VELOX_ENABLE_HIVE_CONNECTOR})
   set(VELOX_ENABLE_COMPRESSION_LZ4 ON)
+  set(VELOX_ENABLE_COMPRESSION_ZSTD ON)
 endif()
 
 if(${VELOX_BUILD_TESTING})
@@ -190,6 +192,7 @@ if(${VELOX_BUILD_TESTING})
   set(VELOX_ENABLE_EXAMPLES ON)
   set(VELOX_ENABLE_PARQUET ON)
   set(VELOX_ENABLE_COMPRESSION_LZ4 ON)
+  set(VELOX_ENABLE_COMPRESSION_ZSTD ON)
 endif()
 
 if(${VELOX_ENABLE_BENCHMARKS})
@@ -479,14 +482,8 @@ if(VELOX_ENABLE_COMPRESSION_LZ4)
   find_package(lz4 REQUIRED)
 endif()
 
-if(${VELOX_BUILD_MINIMAL_WITH_DWIO} OR ${VELOX_ENABLE_HIVE_CONNECTOR})
-  # DWIO needs all sorts of stream compression libraries.
-  #
-  # TODO: make these optional and pluggable.
-  find_package(ZLIB REQUIRED)
-  find_package(lzo2 REQUIRED)
+if(VELOX_ENABLE_COMPRESSION_ZSTD)
   find_package(zstd REQUIRED)
-  find_package(Snappy REQUIRED)
   if(NOT TARGET zstd::zstd)
     if(TARGET zstd::libzstd_static)
       set(ZSTD_TYPE static)
@@ -497,6 +494,15 @@ if(${VELOX_BUILD_MINIMAL_WITH_DWIO} OR ${VELOX_ENABLE_HIVE_CONNECTOR})
   endif()
 endif()
 
+if(${VELOX_BUILD_MINIMAL_WITH_DWIO} OR ${VELOX_ENABLE_HIVE_CONNECTOR})
+  # DWIO needs all sorts of stream compression libraries.
+  #
+  # TODO: make these optional and pluggable.
+  find_package(ZLIB REQUIRED)
+  find_package(lzo2 REQUIRED)
+  find_package(Snappy REQUIRED)
+endif()
+
 velox_set_source(re2)
 velox_resolve_dependency(re2)
 

velox/common/compression/CMakeLists.txt

@@ -23,9 +23,16 @@ velox_link_libraries(
   PRIVATE velox_exception)
 
 if(VELOX_ENABLE_COMPRESSION_LZ4)
-  velox_sources(velox_common_compression PRIVATE Lz4Compression.cpp
-                HadoopCompressionFormat.cpp)
+  velox_sources(velox_common_compression PRIVATE Lz4Compression.cpp)
   velox_link_libraries(velox_common_compression PUBLIC lz4::lz4)
   velox_compile_definitions(velox_common_compression
                             PRIVATE VELOX_ENABLE_COMPRESSION_LZ4)
 endif()
+
+if(VELOX_ENABLE_COMPRESSION_ZSTD)
+  velox_sources(velox_common_compression PRIVATE ZstdCompression.cpp
+                HadoopCompressionFormat.cpp)
+  velox_link_libraries(velox_common_compression PUBLIC zstd::zstd)
+  velox_compile_definitions(velox_common_compression
+                            PRIVATE VELOX_ENABLE_COMPRESSION_ZSTD)
+endif()

velox/common/compression/Compression.cpp

@@ -19,6 +19,9 @@
 #ifdef VELOX_ENABLE_COMPRESSION_LZ4
 #include "velox/common/compression/Lz4Compression.h"
 #endif
+#ifdef VELOX_ENABLE_COMPRESSION_ZSTD
+#include "velox/common/compression/ZstdCompression.h"
+#endif
 
 #include <folly/Conv.h>
 
@@ -107,24 +110,40 @@ Status Codec::init() {
 }
 
 bool Codec::supportsGetUncompressedLength(CompressionKind kind) {
-  // TODO: Return true if it's supported by compression kind.
-  return false;
+  switch (kind) {
+#ifdef VELOX_ENABLE_COMPRESSION_ZSTD
+    case CompressionKind::CompressionKind_ZSTD:
+      return true;
+#endif
+    default:
+      return false;
+  }
 }
 
 bool Codec::supportsStreamingCompression(CompressionKind kind) {
   switch (kind) {
 #ifdef VELOX_ENABLE_COMPRESSION_LZ4
     case CompressionKind::CompressionKind_LZ4:
       return true;
+#endif
+#ifdef VELOX_ENABLE_COMPRESSION_ZSTD
+    case CompressionKind::CompressionKind_ZSTD:
+      return true;
 #endif
     default:
       return false;
   }
 }
 
 bool Codec::supportsCompressFixedLength(CompressionKind kind) {
-  // TODO: Return true if it's supported by compression kind.
-  return false;
+  switch (kind) {
+#ifdef VELOX_ENABLE_COMPRESSION_ZSTD
+    case CompressionKind::CompressionKind_ZSTD:
+      return true;
+#endif
+    default:
+      return false;
+  }
 }
 
 Expected<std::unique_ptr<Codec>> Codec::create(
@@ -162,6 +181,11 @@ Expected<std::unique_ptr<Codec>> Codec::create(
       // By default, create LZ4 Frame codec.
       codec = makeLz4FrameCodec(compressionLevel);
       break;
+#endif
+#ifdef VELOX_ENABLE_COMPRESSION_ZSTD
+    case CompressionKind::CompressionKind_ZSTD:
+      codec = makeZstdCodec(compressionLevel);
+      break;
 #endif
     default:
       break;
@@ -191,6 +215,10 @@ bool Codec::isAvailable(CompressionKind kind) {
 #ifdef VELOX_ENABLE_COMPRESSION_LZ4
     case CompressionKind::CompressionKind_LZ4:
       return true;
+#endif
+#ifdef VELOX_ENABLE_COMPRESSION_ZSTD
+    case CompressionKind::CompressionKind_ZSTD:
+      return true;
 #endif
     default:
       return false;