feat(encodings + SubIntSplit): Extension to SubIntSplit Cost Model Selection

dwio/nimble/encodings/BlockBitPackingEncoding.h

@@ -91,6 +91,55 @@ class BlockBitPackingEncoding final
       std::span<const physicalType> values,
       uint16_t blockSize = kBlockBitPackingBlockSize);
 
+#ifdef NIMBLE_ENABLE_EXPERIMENTAL_ENCODINGS
+  /// Statistics-only overload for general encoding selection (e.g. as a
+  /// SubIntSplit segment candidate), where only `Statistics<physicalType>` --
+  /// not the raw values -- is available. Without raw values, true per-block
+  /// locality can't be observed; this approximates a "typical" per-block bit
+  /// width as the ~50%-coverage point of `statistics.bucketCounts()` (median
+  /// bit width of value - min), analogous to how PFOREncoding's
+  /// `selectBaseBitWidth` picks a 90%-coverage baseline. This is a coarser
+  /// approximation than the span-based overload above.
+  static uint64_t estimateSize(
+      uint64_t rowCount,
+      const Statistics<physicalType>& statistics,
+      uint16_t blockSize = kBlockBitPackingBlockSize) {
+    if (rowCount == 0) {
+      return EncodingPrefix::kFixedPrefixSize;
+    }
+    const auto& bucketCounts = statistics.bucketCounts();
+    const auto fullRange =
+        static_cast<physicalType>(statistics.max() - statistics.min());
+    const uint8_t maxBitWidth =
+        static_cast<uint8_t>(velox::bits::bitsRequired(fullRange));
+
+    constexpr double kCoverageThreshold = 0.5;
+    const uint64_t threshold = static_cast<uint64_t>(
+        static_cast<double>(rowCount) * kCoverageThreshold);
+    uint8_t typicalBitWidth = maxBitWidth;
+    uint64_t cumulative = 0;
+    for (size_t k = 0; k < bucketCounts.size(); ++k) {
+      cumulative += bucketCounts[k];
+      if (cumulative >= threshold) {
+        typicalBitWidth = std::min<uint8_t>(
+            static_cast<uint8_t>(std::min<size_t>((k + 1) * 7, 64)),
+            maxBitWidth);
+        break;
+      }
+    }
+
+    const uint32_t numBlocks =
+        velox::bits::divRoundUp(static_cast<uint32_t>(rowCount), blockSize);
+    const uint64_t packedSize =
+        FixedBitArray::bufferSize(rowCount, typicalBitWidth);
+    const uint64_t metadataSize = kMetadataHeaderSize +
+        TrivialEncoding<physicalType>::estimateSize(numBlocks) +
+        TrivialEncoding<uint8_t>::estimateSize(numBlocks) +
+        TrivialEncoding<uint32_t>::estimateSize(numBlocks);
+    return EncodingPrefix::kFixedPrefixSize + metadataSize + packedSize;
+  }
+#endif
+
   std::string debugString(int offset) const final;
 
  private:

dwio/nimble/encodings/DeltaEncoding.h

@@ -22,6 +22,7 @@
 
 #include "dwio/nimble/common/Buffer.h"
 #include "dwio/nimble/common/Exceptions.h"
+#include "dwio/nimble/common/FixedBitArray.h"
 #include "dwio/nimble/common/Types.h"
 #include "dwio/nimble/common/Vector.h"
 #include "dwio/nimble/encodings/common/Encoding.h"
@@ -98,6 +99,50 @@ class DeltaEncoding final
       Buffer& buffer,
       const Encoding::Options& options = {});
 
+#ifdef NIMBLE_ENABLE_EXPERIMENTAL_ENCODINGS
+  /// Statistics-only size estimate for general encoding selection (e.g. as a
+  /// SubIntSplit segment candidate), where only `Statistics<physicalType>` --
+  /// not the raw values -- is available. Without raw values, the true
+  /// monotonic-step pattern can't be observed; this assumes a single leading
+  /// restatement and an average step size of range / (rowCount - 1) -- the
+  /// typical step for a column whose values are roughly evenly spread across
+  /// [min, max] in row order. This is a coarse approximation: non-monotonic
+  /// columns (which need many restatements) will be underestimated.
+  static uint64_t estimateSize(
+      uint64_t rowCount,
+      const Statistics<physicalType>& statistics) {
+    if (rowCount == 0) {
+      return EncodingPrefix::kFixedPrefixSize;
+    }
+    const auto fullRange =
+        static_cast<uint64_t>(statistics.max() - statistics.min());
+    const double avgAbsDelta = rowCount > 1
+        ? static_cast<double>(fullRange) / static_cast<double>(rowCount - 1)
+        : 0.0;
+    const uint8_t deltaBitWidth = avgAbsDelta < 1.0
+        ? uint8_t{0}
+        : static_cast<uint8_t>(
+              velox::bits::bitsRequired(static_cast<uint64_t>(avgAbsDelta)));
+
+    const uint64_t deltasSize =
+        FixedBitArray::bufferSize(rowCount, deltaBitWidth);
+    // Assume a single leading restatement (best case; non-monotonic columns
+    // need more, but Statistics<T> doesn't expose monotonicity).
+    const uint64_t restatementsSize = sizeof(physicalType);
+    const uint64_t isRestatementsSize = velox::bits::nbytes(rowCount);
+
+    // Each of the three nested sub-streams has its own ~7-byte header
+    // (prefix(6) + compressionType(1)).
+    constexpr uint64_t kNestedHeaderSize = 7;
+    // Outer prefix(6) + two 4-byte relative offsets.
+    constexpr uint64_t kOuterHeaderSize = EncodingPrefix::kFixedPrefixSize + 8;
+
+    return kOuterHeaderSize + (kNestedHeaderSize + deltasSize) +
+        (kNestedHeaderSize + restatementsSize) +
+        (kNestedHeaderSize + isRestatementsSize);
+  }
+#endif
+
  private:
   // Ensures isRestatementsBitmap_ has capacity for rowCount bits and
   // returns a mutable pointer to the underlying uint64_t words.

dwio/nimble/encodings/ForEncoding.h

@@ -24,6 +24,7 @@
 #include "dwio/nimble/common/FixedBitArray.h"
 #include "dwio/nimble/common/Vector.h"
 #include "dwio/nimble/compression/Compression.h"
+#include "dwio/nimble/encodings/TrivialEncoding.h"
 #include "dwio/nimble/encodings/common/Encoding.h"
 #include "dwio/nimble/encodings/common/EncodingFactory.h"
 #include "dwio/nimble/encodings/common/EncodingPrimitives.h"
@@ -83,6 +84,55 @@ class ForEncoding final
       Buffer& buffer,
       const Encoding::Options& options = {});
 
+#ifdef NIMBLE_ENABLE_EXPERIMENTAL_ENCODINGS
+  /// Statistics-only size estimate for general encoding selection (e.g. as a
+  /// SubIntSplit segment candidate), where only `Statistics<physicalType>` --
+  /// not the raw values -- is available. The local (per-frame) bit width is
+  /// estimated from the average step size scaled to the frame size -- a
+  /// random-walk heuristic where the local range over a frame of
+  /// `kFrameSize` steps grows roughly with avgAbsDelta -- capped by the
+  /// overall range. Per-frame metadata streams are estimated as Trivial,
+  /// matching encode()'s frame size of 128.
+  static uint64_t estimateSize(
+      uint64_t rowCount,
+      const Statistics<physicalType>& statistics) {
+    if (rowCount == 0) {
+      return EncodingPrefix::kFixedPrefixSize;
+    }
+    constexpr uint32_t kFrameSize = 128;
+    const auto numFrames =
+        static_cast<uint32_t>(velox::bits::divRoundUp(rowCount, kFrameSize));
+
+    const auto fullRange =
+        static_cast<uint64_t>(statistics.max() - statistics.min());
+    const double avgAbsDelta = rowCount > 1
+        ? static_cast<double>(fullRange) / static_cast<double>(rowCount - 1)
+        : 0.0;
+    const double localRange = std::min(
+        static_cast<double>(fullRange),
+        avgAbsDelta * static_cast<double>(kFrameSize) / 2.0);
+    const uint8_t localBits = localRange < 1.0
+        ? uint8_t{0}
+        : static_cast<uint8_t>(
+              velox::bits::bitsRequired(static_cast<uint64_t>(localRange)));
+
+    const uint64_t packedSize = FixedBitArray::bufferSize(rowCount, localBits);
+    const uint64_t bitWidthsSize =
+        TrivialEncoding<uint8_t>::estimateSize(numFrames);
+    const uint64_t referencesSize =
+        TrivialEncoding<physicalType>::estimateSize(numFrames);
+    const uint64_t bitOffsetsSize =
+        TrivialEncoding<uint64_t>::estimateSize(numFrames);
+
+    // EncodingPrefix::kFixedPrefixSize(6) + FOR-specific fixed fields
+    // (compressionType + frameSize + numFrames + enableBitOffsets = 10),
+    // plus each of the four sub-streams' 4-byte size prefix.
+    return EncodingPrefix::kFixedPrefixSize +
+        (kPrefixSize - Encoding::kPrefixSize) + 4 + bitWidthsSize + 4 +
+        referencesSize + 4 + bitOffsetsSize + 4 + packedSize;
+  }
+#endif
+
   std::string debugString(int offset) const final;
 
  private:

dwio/nimble/encodings/FrequencyPartitionEncoding.h

@@ -122,6 +122,39 @@ class FrequencyPartitionEncoding
       Buffer& buffer,
       const Encoding::Options& options = {});
 
+#ifdef NIMBLE_ENABLE_EXPERIMENTAL_ENCODINGS
+  // Statistics-only size estimate for encoding selection. Uses
+  // consecutiveRepeatCount as a proxy for top-tier coverage (high repetition
+  // → values concentrate in the 1-bit tier → cheaper FPE). Assumes
+  // PerTierBitmaps index (2 index bits/value) since that is what
+  // SubIntSplitEncoding overrides for segment children.
+  static uint64_t estimateSize(
+      uint64_t rowCount,
+      const Statistics<physicalType>& statistics) {
+    if (rowCount == 0) {
+      return Encoding::kPrefixSize;
+    }
+    const double n = static_cast<double>(rowCount);
+    const double repeatFraction = (rowCount > 1)
+        ? static_cast<double>(statistics.consecutiveRepeatCount()) /
+              static_cast<double>(rowCount - 1)
+        : 0.0;
+    constexpr double kFallbackBitsPerValue =
+        static_cast<double>(sizeof(physicalType) * 8u);
+    const double tier0Frac = repeatFraction;
+    const double fallbackFrac = 1.0 - tier0Frac;
+    // Key cost: top-tier values get 1-bit codes, remainder get full-width.
+    const double keyCostBytes =
+        (tier0Frac * n * 1.0 + fallbackFrac * n * kFallbackBitsPerValue) /
+        8.0;
+    // PerTierBitmaps index: 2 tiers × 1 bit/value/tier.
+    const double indexBytes = 2.0 * n / 8.0;
+    // Tier sub-stream headers + outer encoding prefix.
+    constexpr double kOverheadBytes = 6.0 + 4.0 + 4.0 + 4.0 + 4.0 * 7.0;
+    return static_cast<uint64_t>(kOverheadBytes + keyCostBytes + indexBytes);
+  }
+#endif // NIMBLE_ENABLE_EXPERIMENTAL_ENCODINGS
+
   std::string debugString(int offset) const final;
   uint32_t getTierForRow(uint32_t rowIndex) const;