Fix Race Condition in Stats Aggregation During Parallel Encoding with Chunking (facebookincubator#357)

Summary:
Pull Request resolved: facebookincubator#357

This diff addresses a data race condition introduced by a recent refactoring in the handling of column stats objects for individual streams. Previously, the creation and access of column stats objects for individual streams were moved inside a barrier as [part of a refactor.](https://www.internalfb.com/diff/D88053111?entry_point=19) However, that change inadvertently introduced a data race, as multiple threads could concurrently create or access these objects without proper synchronization. This issue was detected by our TSAN (ThreadSanitizer) tests, which reported failures due to the race condition. This diff ensures that the creation and access of column stats objects are properly synchronized within the barrier, eliminating the data race. As a result, TSAN tests now pass, confirming that the concurrency issue has been resolved.x

Reviewed By: xiaoxmeng

Differential Revision: D88445993

fbshipit-source-id: 9cc8ecf39020fbfc2f1d06c077f1a45195e7bfbe

Author: macvincent

dwio/nimble/velox/VeloxWriter.cpp

@@ -993,15 +993,18 @@ bool VeloxWriter::writeChunks(
       velox::dwio::common::ExecutorBarrier barrier{
           context_->options().encodingExecutor};
       for (auto streamIndex : streamIndices) {
-        barrier.add([&, streamData = streams[streamIndex].get()] {
-          const auto offset = streamData->descriptor().offset();
+        auto& streamData = streams[streamIndex];
+        const auto offset = streamData->descriptor().offset();
+        auto& streamSize = context_->columnStats(offset).physicalSize;
+        auto& encodeStream = encodedStreams_[offset];
+        barrier.add([&] {
           if (encodeStreamChunk(
                   *streamData,
                   minChunkSize,
                   maxChunkSize,
                   ensureFullChunks,
-                  encodedStreams_[offset],
-                  context_->columnStats()[offset].physicalSize,
+                  encodeStream,
+                  streamSize,
                   chunkBytes,
                   logicalBytes)) {
             writtenChunk = true;