refactor(native): Use getVectorSerdeOptions() from OperatorUtils.h (#26358)

Summary: Use the centralized method for convenient getting vector serde
options.

Differential Revision: D84941606

Author: tanjialiang

presto-native-execution/presto_cpp/main/operators/BroadcastWrite.cpp

@@ -17,6 +17,7 @@
 #include <boost/uuid/uuid_io.hpp>
 #include "presto_cpp/main/operators/BroadcastFile.h"
 #include "velox/common/file/FileSystems.h"
+#include "velox/exec/OperatorUtils.h"
 
 using namespace facebook::velox::exec;
 using namespace facebook::velox;
@@ -31,22 +32,7 @@ velox::core::PlanNodeId deserializePlanNodeId(const folly::dynamic& obj) {
   return obj["id"].asString();
 }
 
-// TODO: This is a copy from Exchange.cpp. We should refactor
-// such that this method is globally accessible from a single location. This is
-// to prevent diverges of serde options during write and read.
-std::unique_ptr<VectorSerde::Options> getVectorSerdeOptions(
-    const core::QueryConfig& queryConfig,
-    VectorSerde::Kind kind) {
-  std::unique_ptr<VectorSerde::Options> options =
-      kind == VectorSerde::Kind::kPresto
-      ? std::make_unique<serializer::presto::PrestoVectorSerde::PrestoOptions>()
-      : std::make_unique<VectorSerde::Options>();
-  options->compressionKind =
-      common::stringToCompressionKind(queryConfig.shuffleCompressionKind());
-  return options;
-}
-
-// BroadcastWriteOperator writes input RowVectors to specified file.
+/// BroadcastWriteOperator writes input RowVectors to specified file.
 class BroadcastWriteOperator : public Operator {
  public:
   BroadcastWriteOperator(
@@ -73,7 +59,10 @@ class BroadcastWriteOperator : public Operator {
         fmt::format("{}/file_broadcast_{}", basePath, makeUuid()),
         planNode->maxBroadcastBytes(),
         8 << 20,
-        getVectorSerdeOptions(ctx->queryConfig(), VectorSerde::Kind::kPresto),
+        getVectorSerdeOptions(
+            common::stringToCompressionKind(
+                ctx->queryConfig().shuffleCompressionKind()),
+            VectorSerde::Kind::kPresto),
         operatorCtx_->pool());
   }