Add option to disable crc32c in exchange

Author: yingsu00

velox/core/QueryConfig.h

@@ -716,6 +716,9 @@ class QueryConfig {
   static constexpr const char* kRowSizeTrackingEnabled =
       "row_size_tracking_enabled";
 
+  static constexpr const char* kDisableCrc32ForShuffle =
+      "disable_crc32_for_shuffle";
+
   bool selectiveNimbleReaderEnabled() const {
     return get<bool>(kSelectiveNimbleReaderEnabled, false);
   }
@@ -1291,6 +1294,10 @@ class QueryConfig {
     return get<std::string>(kClientTags, "");
   }
 
+  bool isDisableCrc32ForShuffleEnabled() const {
+    return get<bool>(kDisableCrc32ForShuffle, false);
+  }
+
   template <typename T>
   T get(const std::string& key, const T& defaultValue) const {
     return config_->get<T>(key, defaultValue);

velox/exec/PartitionedOutput.cpp

@@ -30,6 +30,7 @@ std::unique_ptr<VectorSerde::Options> getVectorSerdeOptions(
   options->compressionKind =
       common::stringToCompressionKind(queryConfig.shuffleCompressionKind());
   options->minCompressionRatio = PartitionedOutput::minCompressionRatio();
+  options->disableCrc32c = queryConfig.isDisableCrc32ForShuffleEnabled();
   return options;
 }
 } // namespace
@@ -126,7 +127,12 @@ BlockingReason Destination::flush(
 
   // Upper limit of message size with no columns.
   constexpr int32_t kMinMessageSize = 128;
-  auto listener = bufferManager.newListener();
+
+  std::unique_ptr<OutputStreamListener> listener(nullptr);
+  if (!serdeOptions_->disableCrc32c) {
+    listener = bufferManager.newListener();
+  }
+
   IOBufOutputStream stream(
       *current_->pool(),
       listener.get(),

velox/serializers/PrestoSerializer.cpp

@@ -247,9 +247,14 @@ void PrestoVectorSerde::serializeSingleColumn(
   Scratch scratch;
   serializeColumn(vector, folly::Range(&range, 1), stream.get(), scratch);
 
-  PrestoOutputStreamListener listener;
-  OStreamOutputStream outputStream(output, &listener);
-  stream->flush(&outputStream);
+  if (opts->disableCrc32c) {
+    OStreamOutputStream outputStream(output, nullptr);
+    stream->flush(&outputStream);
+  } else {
+    PrestoOutputStreamListener listener;
+    OStreamOutputStream outputStream(output, &listener);
+    stream->flush(&outputStream);
+  }
 }
 
 // static

velox/serializers/PrestoSerializer.h

@@ -57,8 +57,12 @@ class PrestoVectorSerde : public VectorSerde {
         common::CompressionKind _compressionKind,
         float _minCompressionRatio = 0.8,
         bool _nullsFirst = false,
-        bool _preserveEncodings = false)
-        : VectorSerde::Options(_compressionKind, _minCompressionRatio),
+        bool _preserveEncodings = false,
+        bool _disableCrc32c = false)
+        : VectorSerde::Options(
+              _compressionKind,
+              _minCompressionRatio,
+              _disableCrc32c),
           useLosslessTimestamp(_useLosslessTimestamp),
           nullsFirst(_nullsFirst),
           preserveEncodings(_preserveEncodings) {}

velox/vector/VectorStream.h

@@ -219,9 +219,11 @@ class VectorSerde {
 
     Options(
         common::CompressionKind _compressionKind,
-        float _minCompressionRatio)
+        float _minCompressionRatio,
+        bool _disableCrc32c)
         : compressionKind(_compressionKind),
-          minCompressionRatio(_minCompressionRatio) {}
+          minCompressionRatio(_minCompressionRatio),
+          disableCrc32c(_disableCrc32c) {}
 
     virtual ~Options() = default;
 
@@ -231,6 +233,7 @@ class VectorSerde {
     /// than this causes subsequent compression attempts to be skipped. The more
     /// times compression misses the target the less frequently it is tried.
     float minCompressionRatio{0.8};
+    bool disableCrc32c{false};
   };
 
   Kind kind() const {