Integrate Max Stream Size Chunking in Velox Writer (facebookincubator#249)

Summary:
Pull Request resolved: facebookincubator#249

This is the last feature of the new chunking policy described in this [doc](https://fburl.com/gdoc/gkdwwju1). Here, we break down large streams into multiple chunks of size up to `maxStreamChunkRawSize`. This protects the reader from attempting to materialize huge chunks. We included StreamData support for this in the previous diff. In this diff, we integrate with the VeloxWriter. With this change, while memory pressure is detected, we:

1. Chunk large streams above `maxStreamChunkRawSize`, retaining stream data below the limit.
2. If there is still memory pressure after the first step, chunk streams with size above `minStreamChunkRawSize`.

During stripe flush, we chunk all remaining data, breaking down streams above `maxStreamChunkRawSize` into smaller chunks.

 ---

The general chunking policy has two phases:

## **Phase 1 - Memory Pressure Management (shouldChunk)**
The policy monitors total in-memory data size:
- When memory usage exceeds the maximum threshold, it initiates chunking to reduce memory footprint while continuing data ingestion.
- When previous chunking attempts succeeded and memory remains above the minimum threshold, it continues chunking to further reduce memory usage.
- When chunking fails to reduce memory usage effectively and memory stays above the minimum threshold, it forces a full stripe flush to guarantee memory relief.

## **Phase 2 - Storage Size Optimization (shouldFlush)**
Implements compression-aware stripe size prediction:
- Calculates the anticipated final compressed stripe size by applying the estimated compression ratio to unencoded data.
- Triggers stripe flush when the predicted compressed size reaches the target stripe size threshold.

Differential Revision: D82175496

Author: macvincent

dwio/nimble/velox/StreamChunker.cpp

@@ -8,26 +8,27 @@ std::unique_ptr<StreamChunker> encodeStreamTyped(
     StreamData& streamData,
     uint64_t maxChunkSize,
     uint64_t minChunkSize,
-    bool lastChunk) {
+    bool emptyStreamContent,
+    bool isNullStream) {
   const auto& streamDataType = streamData.type();
   if (streamDataType == ContentStreamData<T>::TYPE_NAME) {
     return std::make_unique<ContentStreamChunker<T>>(
         static_cast<ContentStreamData<T>&>(streamData),
         maxChunkSize,
-        minChunkSize,
-        lastChunk);
+        minChunkSize);
   } else if (streamDataType == NullsStreamData::TYPE_NAME) {
     return std::make_unique<NullsStreamChunker>(
         static_cast<NullsStreamData&>(streamData),
         maxChunkSize,
         minChunkSize,
-        lastChunk);
+        emptyStreamContent);
   } else if (streamDataType == NullableContentStreamData<T>::TYPE_NAME) {
     return std::make_unique<NullableContentStreamChunker<T>>(
         static_cast<NullableContentStreamData<T>&>(streamData),
         maxChunkSize,
         minChunkSize,
-        lastChunk);
+        emptyStreamContent,
+        isNullStream);
   }
   NIMBLE_UNREACHABLE(
       fmt::format("Unsupported streamData type {}", streamDataType))
@@ -37,37 +38,74 @@ std::unique_ptr<StreamChunker> getStreamChunker(
     StreamData& streamData,
     uint64_t maxChunkSize,
     uint64_t minChunkSize,
-    bool lastChunk) {
+    bool emptyStreamContent,
+    bool isNullStream) {
   const auto scalarKind = streamData.descriptor().scalarKind();
   switch (scalarKind) {
     case ScalarKind::Bool:
       return encodeStreamTyped<bool>(
-          streamData, maxChunkSize, minChunkSize, lastChunk);
+          streamData,
+          maxChunkSize,
+          minChunkSize,
+          emptyStreamContent,
+          isNullStream);
     case ScalarKind::Int8:
       return encodeStreamTyped<int8_t>(
-          streamData, maxChunkSize, minChunkSize, lastChunk);
+          streamData,
+          maxChunkSize,
+          minChunkSize,
+          emptyStreamContent,
+          isNullStream);
     case ScalarKind::Int16:
       return encodeStreamTyped<int16_t>(
-          streamData, maxChunkSize, minChunkSize, lastChunk);
+          streamData,
+          maxChunkSize,
+          minChunkSize,
+          emptyStreamContent,
+          isNullStream);
     case ScalarKind::Int32:
       return encodeStreamTyped<int32_t>(
-          streamData, maxChunkSize, minChunkSize, lastChunk);
+          streamData,
+          maxChunkSize,
+          minChunkSize,
+          emptyStreamContent,
+          isNullStream);
     case ScalarKind::UInt32:
       return encodeStreamTyped<uint32_t>(
-          streamData, maxChunkSize, minChunkSize, lastChunk);
+          streamData,
+          maxChunkSize,
+          minChunkSize,
+          emptyStreamContent,
+          isNullStream);
     case ScalarKind::Int64:
       return encodeStreamTyped<int64_t>(
-          streamData, maxChunkSize, minChunkSize, lastChunk);
+          streamData,
+          maxChunkSize,
+          minChunkSize,
+          emptyStreamContent,
+          isNullStream);
     case ScalarKind::Float:
       return encodeStreamTyped<float>(
-          streamData, maxChunkSize, minChunkSize, lastChunk);
+          streamData,
+          maxChunkSize,
+          minChunkSize,
+          emptyStreamContent,
+          isNullStream);
     case ScalarKind::Double:
       return encodeStreamTyped<double>(
-          streamData, maxChunkSize, minChunkSize, lastChunk);
+          streamData,
+          maxChunkSize,
+          minChunkSize,
+          emptyStreamContent,
+          isNullStream);
     case ScalarKind::String:
     case ScalarKind::Binary:
       return encodeStreamTyped<std::string_view>(
-          streamData, maxChunkSize, minChunkSize, lastChunk);
+          streamData,
+          maxChunkSize,
+          minChunkSize,
+          emptyStreamContent,
+          isNullStream);
     default:
       NIMBLE_UNREACHABLE(
           fmt::format("Unsupported scalar kind {}", toString(scalarKind)));