misc: Add config and session property for exchange checksum

Add exchange.checksum-enabled config property and
native_exchange_checksum session property. The default value is false,
meaning crc32 checksum computation is disabled by default.

Author: yingsu00

velox/core/QueryConfig.h

@@ -721,6 +721,9 @@ class QueryConfig {
   static constexpr const char* kRowSizeTrackingEnabled =
       "row_size_tracking_enabled";
 
+  static constexpr const char* kExchangeChecksum =
+      "exchange_checksum";
+
   bool selectiveNimbleReaderEnabled() const {
     return get<bool>(kSelectiveNimbleReaderEnabled, false);
   }
@@ -1296,6 +1299,10 @@ class QueryConfig {
     return get<std::string>(kClientTags, "");
   }
 
+  bool isExchangeChecksumEnabled() const {
+    return get<bool>(kExchangeChecksum, 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->exchangeChecksum = queryConfig.isExchangeChecksumEnabled();
   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_->exchangeChecksum) {
+    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->exchangeChecksum) {
+    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 _exchangeChecksum = false)
+        : VectorSerde::Options(
+              _compressionKind,
+              _minCompressionRatio,
+              _exchangeChecksum),
           useLosslessTimestamp(_useLosslessTimestamp),
           nullsFirst(_nullsFirst),
           preserveEncodings(_preserveEncodings) {}

velox/serializers/tests/CompactRowSerializerTest.cpp

@@ -77,7 +77,7 @@ class CompactRowSerializerTest : public ::testing::Test,
     appendRow_ = GetParam().appendRow;
     compressionKind_ = GetParam().compressionKind;
     microBatchDeserialize_ = GetParam().microBatchDeserialize;
-    options_ = std::make_unique<VectorSerde::Options>(compressionKind_, 0.8);
+    options_ = std::make_unique<VectorSerde::Options>(compressionKind_, 0.8, false);
   }
 
   void TearDown() override {

velox/serializers/tests/SerializedPageFileTest.cpp

@@ -101,7 +101,8 @@ class SerializedPageFileTest : public ::testing::TestWithParam<TestParams>,
             false); // preserveEncodings
       case SerdeType::kCompactRow:
       case SerdeType::kUnsafeRow:
-        return std::make_unique<VectorSerde::Options>(compressionKind_, 0.8);
+        return std::make_unique<VectorSerde::Options>(
+            compressionKind_, 0.8, false);
     }
     return nullptr;
   }

velox/serializers/tests/UnsafeRowSerializerTest.cpp

@@ -76,7 +76,8 @@ class UnsafeRowSerializerTest : public ::testing::Test,
     appendRow_ = GetParam().appendRow;
     compressionKind_ = GetParam().compressionKind;
     microBatchDeserialize_ = GetParam().microBatchDeserialize;
-    options_ = std::make_unique<VectorSerde::Options>(compressionKind_, 0.8);
+    options_ =
+        std::make_unique<VectorSerde::Options>(compressionKind_, 0.8, false);
   }
 
   void TearDown() override {

velox/vector/VectorStream.h

@@ -219,9 +219,11 @@ class VectorSerde {
 
     Options(
         common::CompressionKind _compressionKind,
-        float _minCompressionRatio)
+        float _minCompressionRatio,
+        bool _exchangeChecksum)
         : compressionKind(_compressionKind),
-          minCompressionRatio(_minCompressionRatio) {}
+          minCompressionRatio(_minCompressionRatio),
+          exchangeChecksum(_exchangeChecksum) {}
 
     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 exchangeChecksum{false};
   };
 
   Kind kind() const {