Add output stats for pos sink operators (prestodb#25915)

Summary:

Adds output row stats for sapphire-velox related sink operators
Properly close write file on broadcast write

Reviewed By: singcha

Differential Revision: D81271224

Author: tanjialiang

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

@@ -82,7 +82,12 @@ void BroadcastFileWriter::collect(const RowVectorPtr& input) {
   write(input);
 }
 
-void BroadcastFileWriter::noMoreData() {}
+void BroadcastFileWriter::noMoreData() {
+  if (writeFile_ != nullptr) {
+    writeFile_->flush();
+    writeFile_->close();
+  }
+}
 
 RowVectorPtr BroadcastFileWriter::fileStats() {
   // No rows written.

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

@@ -70,6 +70,9 @@ class BroadcastWriteOperator : public Operator {
     }
 
     fileBroadcastWriter_->collect(reorderedInput);
+    auto lockedStats = stats_.wlock();
+    lockedStats->addOutputVector(
+        reorderedInput->estimateFlatSize(), reorderedInput->size());
   }
 
   void noMoreInput() override {

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

@@ -104,6 +104,8 @@ class ShuffleWriteOperator : public Operator {
             "collect");
       }
     }
+    auto lockedStats = stats_.wlock();
+    lockedStats->addOutputVector(input->estimateFlatSize(), input->size());
   }
 
   void noMoreInput() override {