Frequency partitioning and FOR encoding rebased & synced with nimble/main

CMakeLists.txt

@@ -119,16 +119,24 @@ add_link_options(-Wl,--gc-sections)
 
 include(CTest) # include after project() but before add_subdirectory()
 
-find_package(flatbuffers)
-set(NIMBLE_NEED_BUNDLED_BUILD_FLAT_BUFFERS FLASE)
-if(flatbuffers_FOUND)
-  if(flatbuffers_VERSION VERSION_LESS "22.9.4")
+# Try to find FlatBuffers with uppercase name first (newer versions)
+find_package(FlatBuffers QUIET)
+set(NIMBLE_NEED_BUNDLED_BUILD_FLAT_BUFFERS FALSE)
+if(FlatBuffers_FOUND)
+  if(FlatBuffers_VERSION VERSION_LESS "22.9.4")
     set(NIMBLE_NEED_BUNDLED_BUILD_FLAT_BUFFERS TRUE)
   endif()
 else()
-  # Fallback to old FlatBuffers (< 2.0.0).
-  find_package(Flatbuffers REQUIRED)
-  set(NIMBLE_NEED_BUNDLED_BUILD_FLAT_BUFFERS TRUE)
+  # Try lowercase variant
+  find_package(flatbuffers QUIET)
+  if(flatbuffers_FOUND)
+    if(flatbuffers_VERSION VERSION_LESS "22.9.4")
+      set(NIMBLE_NEED_BUNDLED_BUILD_FLAT_BUFFERS TRUE)
+    endif()
+  else()
+    # No system FlatBuffers found, use bundled
+    set(NIMBLE_NEED_BUNDLED_BUILD_FLAT_BUFFERS TRUE)
+  endif()
 endif()
 if(NIMBLE_NEED_BUNDLED_BUILD_FLAT_BUFFERS)
   # Old FlatBuffers (< 22.9.4) doesn't provide build_flatbuffers(). So

dwio/nimble/common/Types.cpp

@@ -45,6 +45,10 @@ std::string toString(EncodingType encodingType) {
       return "Constant";
     case EncodingType::MainlyConstant:
       return "MainlyConstant";
+    case EncodingType::FrequencyPartition:
+      return "FrequencyPartition";
+    case EncodingType::FOR:
+      return "FOR";
     case EncodingType::Sentinel:
       return "Sentinel";
     case EncodingType::Prefix:

dwio/nimble/common/Types.h

@@ -104,6 +104,12 @@ enum class EncodingType {
   // shared across consecutive entries to reduce storage. Supports seek
   // operations for efficient random access.
   Prefix = 11,
+  // Partitions data by value frequency. Frequent values get shorter bit-width
+  // codes. Rows are reordered to group values with same code length.
+  FrequencyPartition = 12,
+  // Frame of Reference: stores offsets from per-frame minimum values.
+  // Supports O(1) random access. Preserves row order.
+  FOR = 13,
 };
 std::string toString(EncodingType encodingType);
 std::ostream& operator<<(std::ostream& out, EncodingType encodingType);

dwio/nimble/encodings/Compression.cpp

@@ -43,13 +43,60 @@ struct CompressorRegistry {
 
 ICompressor& getCompressor(CompressionType compressionType) {
   static CompressorRegistry registry;
+
+#ifdef DISABLE_META_INTERNAL_COMPRESSOR
+  // When MetaInternal is not available, redirect to Zstd
+  if (compressionType == CompressionType::MetaInternal) {
+    compressionType = CompressionType::Zstd;
+  }
+#endif
+
   auto it = registry.compressors.find(compressionType);
   NIMBLE_CHECK(
       it != registry.compressors.end(),
       "Compressor for type {} is not registered.",
       toString(compressionType));
   return *it->second;
 }
+
+#ifdef DISABLE_META_INTERNAL_COMPRESSOR
+// Wrapper to redirect MetaInternal compression to Zstd
+class MetaInternalToZstdPolicy : public CompressionPolicy {
+ public:
+  explicit MetaInternalToZstdPolicy(const CompressionPolicy& base)
+      : base_(base) {}
+
+  CompressionInformation compression() const override {
+    auto info = base_.compression();
+    if (info.compressionType == CompressionType::MetaInternal) {
+      // Convert MetaInternal parameters to Zstd
+      CompressionInformation zstdInfo{
+          .compressionType = CompressionType::Zstd,
+          .minCompressionSize = info.minCompressionSize};
+      // Map MetaInternal compression level to Zstd compression level
+      zstdInfo.parameters.zstd.compressionLevel = 
+          info.parameters.metaInternal.compressionLevel;
+      return zstdInfo;
+    }
+    return info;
+  }
+
+  bool shouldAccept(
+      CompressionType compressionType,
+      uint64_t uncompressedSize,
+      uint64_t compressedSize) const override {
+    // Redirect MetaInternal to Zstd for acceptance check
+    if (compressionType == CompressionType::MetaInternal) {
+      compressionType = CompressionType::Zstd;
+    }
+    return base_.shouldAccept(compressionType, uncompressedSize, compressedSize);
+  }
+
+ private:
+  const CompressionPolicy& base_;
+};
+#endif
+
 } // namespace
 
 /* static */ CompressionResult Compression::compress(
@@ -60,6 +107,16 @@ ICompressor& getCompressor(CompressionType compressionType) {
     const CompressionPolicy& compressionPolicy) {
   auto compression = compressionPolicy.compression();
 
+#ifdef DISABLE_META_INTERNAL_COMPRESSOR
+  // Wrap the policy to redirect MetaInternal to Zstd
+  if (compression.compressionType == CompressionType::MetaInternal) {
+    MetaInternalToZstdPolicy wrapper(compressionPolicy);
+    compression = wrapper.compression();
+    return getCompressor(compression.compressionType)
+        .compress(memoryPool, data, dataType, bitWidth, wrapper);
+  }
+#endif
+
   return getCompressor(compression.compressionType)
       .compress(memoryPool, data, dataType, bitWidth, compressionPolicy);
 }

dwio/nimble/encodings/Encoding.h

@@ -100,6 +100,10 @@ class Encoding {
     /// fixed 4-byte uint32. Determined from file format version or serializer
     /// version.
     bool useVarintRowCount;
+    /// FrequencyPartitionEncoding index type (cast to FreqPartIndexType).
+    /// 0 = NoIndex (default, backward-compatible), 1 = PerTierBitmaps,
+    /// 2 = TierTagArray, 3 = EliasFano.
+    uint8_t frequencyPartitionIndex;
   };
 
   /// The binary layout for each Encoding begins with the same prefix:
@@ -249,16 +253,15 @@ class Encoding {
 
  protected:
   static void serializePrefix(
-      EncodingType encodingType,
-      DataType dataType,
-      uint32_t rowCount,
-      bool useVarint,
-      char*& pos);
+    EncodingType encodingType,
+    DataType dataType,
+    uint32_t rowCount,
+    bool useVarint,
+    char*& pos);
 
   // Compute the prefix size for serialization. Returns kPrefixSize for fixed
   // format, or 2 + varintSize(rowCount) for varint format.
   static uint32_t serializePrefixSize(uint32_t rowCount, bool useVarint);
-
   // Static helpers for initializer list computation.
   static DataType readDataType(std::string_view data);
   static uint32_t readRowCount(std::string_view data, bool useVarint);

dwio/nimble/encodings/EncodingFactory.cpp

@@ -19,6 +19,8 @@
 #include "dwio/nimble/encodings/DictionaryEncoding.h"
 #include "dwio/nimble/encodings/EncodingSelection.h"
 #include "dwio/nimble/encodings/FixedBitWidthEncoding.h"
+#include "dwio/nimble/encodings/ForEncoding.h"
+#include "dwio/nimble/encodings/FrequencyPartitionEncoding.h"
 #include "dwio/nimble/encodings/MainlyConstantEncoding.h"
 #include "dwio/nimble/encodings/NullableEncoding.h"
 #include "dwio/nimble/encodings/PrefixEncoding.h"
@@ -199,6 +201,30 @@ std::unique_ptr<Encoding> EncodingFactory::create(
           toString(dataType));                              \
   }
 
+#define RETURN_ENCODING_BY_INTEGER_TYPE(Encoding, dataType)           \
+  switch (dataType) {                                                 \
+    case DataType::Int8:                                              \
+      return std::make_unique<Encoding<int8_t>>(memoryPool, data);    \
+    case DataType::Uint8:                                             \
+      return std::make_unique<Encoding<uint8_t>>(memoryPool, data);   \
+    case DataType::Int16:                                             \
+      return std::make_unique<Encoding<int16_t>>(memoryPool, data);   \
+    case DataType::Uint16:                                            \
+      return std::make_unique<Encoding<uint16_t>>(memoryPool, data);  \
+    case DataType::Int32:                                             \
+      return std::make_unique<Encoding<int32_t>>(memoryPool, data);   \
+    case DataType::Uint32:                                            \
+      return std::make_unique<Encoding<uint32_t>>(memoryPool, data);  \
+    case DataType::Int64:                                             \
+      return std::make_unique<Encoding<int64_t>>(memoryPool, data);   \
+    case DataType::Uint64:                                            \
+      return std::make_unique<Encoding<uint64_t>>(memoryPool, data);  \
+    default:                                                          \
+      NIMBLE_UNREACHABLE(                                             \
+          "ForEncoding only supports integer types, got {}.",         \
+          toString(dataType));                                        \
+  }
+
   switch (encodingType) {
     case EncodingType::Trivial: {
       RETURN_ENCODING_BY_LEAF_TYPE(TrivialEncoding, dataType);
@@ -243,6 +269,12 @@ std::unique_ptr<Encoding> EncodingFactory::create(
     case EncodingType::Delta: {
       RETURN_ENCODING_BY_NUMERIC_TYPE(DeltaEncoding, dataType);
     }
+    case EncodingType::FOR: {
+      RETURN_ENCODING_BY_NUMERIC_TYPE(ForEncoding, dataType);
+    }
+    case EncodingType::FrequencyPartition: {
+      RETURN_ENCODING_BY_NUMERIC_TYPE(FrequencyPartitionEncoding, dataType);
+    }
     default: {
       NIMBLE_UNREACHABLE(
           "Trying to deserialize invalid EncodingType:{} -- garbage input?",
@@ -349,6 +381,23 @@ std::string_view EncodingFactory::encode(
             selection, castedValues, buffer, options);
       }
     }
+    case EncodingType::FrequencyPartition: {
+      if constexpr (std::is_same<T, bool>::value) {
+        NIMBLE_INCOMPATIBLE_ENCODING(
+            "FrequencyPartition encoding should not be selected for bool data types.");
+      } else {
+        return FrequencyPartitionEncoding<T>::encode(
+            selection, castedValues, buffer, options);
+      }
+    }
+    case EncodingType::FOR: {
+      if constexpr (std::is_integral<physicalType>::value && !std::is_same<T, bool>::value) {
+        return ForEncoding<T>::encode(selection, castedValues, buffer, options);
+      } else {
+        NIMBLE_INCOMPATIBLE_ENCODING(
+            "For encoding can only be selected for integral data types (not bool).");
+      }
+    }
     case EncodingType::SparseBool: {
       if constexpr (!std::is_same<T, bool>::value) {
         NIMBLE_INCOMPATIBLE_ENCODING(

dwio/nimble/encodings/EncodingIdentifier.h

@@ -57,6 +57,28 @@ struct EncodingIdentifiers {
     static constexpr NestedEncodingIdentifier Restatements = 1;
     static constexpr NestedEncodingIdentifier IsRestatements = 2;
   };
+
+  struct FrequencyPartition {
+    // Partition metadata
+    static constexpr NestedEncodingIdentifier PartitionOffsets = 0;
+    static constexpr NestedEncodingIdentifier PartitionSizes = 1;
+    // Per-tier dictionaries (1-bit, 2-bit, 4-bit, 8-bit, etc.)
+    static constexpr NestedEncodingIdentifier Dict1Bit = 2;
+    static constexpr NestedEncodingIdentifier Dict2Bit = 3;
+    static constexpr NestedEncodingIdentifier Dict4Bit = 4;
+    static constexpr NestedEncodingIdentifier Dict8Bit = 5;
+    static constexpr NestedEncodingIdentifier Dict16Bit = 6;
+    static constexpr NestedEncodingIdentifier Dict32Bit = 7;
+    // Per-tier encoded keys
+    static constexpr NestedEncodingIdentifier Keys1Bit = 8;
+    static constexpr NestedEncodingIdentifier Keys2Bit = 9;
+    static constexpr NestedEncodingIdentifier Keys4Bit = 10;
+    static constexpr NestedEncodingIdentifier Keys8Bit = 11;
+    static constexpr NestedEncodingIdentifier Keys16Bit = 12;
+    static constexpr NestedEncodingIdentifier Keys32Bit = 13;
+    // Unencoded partition (raw values that don't fit in any tier)
+    static constexpr NestedEncodingIdentifier UnencodedValues = 14;
+  };
 };
 
 } // namespace facebook::nimble

dwio/nimble/encodings/EncodingLayout.cpp

@@ -96,7 +96,15 @@ std::pair<EncodingLayout, uint32_t> EncodingLayout::create(
 
   auto pos = encoding.data();
   const auto encodingType = encoding::read<uint8_t, EncodingType>(pos);
-  const auto compressionType = encoding::read<uint8_t, CompressionType>(pos);
+  auto compressionType = encoding::read<uint8_t, CompressionType>(pos);
+
+#ifdef DISABLE_META_INTERNAL_COMPRESSOR
+  // When MetaInternal is not available, map it to Zstd
+  if (compressionType == CompressionType::MetaInternal) {
+    compressionType = CompressionType::Zstd;
+  }
+#endif
+
   const auto childrenCount = encoding::read<uint8_t>(pos);
   [[maybe_unused]] const auto extraDataSize = encoding::read<uint16_t>(pos);
 

dwio/nimble/encodings/EncodingUtils.h

@@ -19,6 +19,7 @@
 #include "dwio/nimble/encodings/DeltaEncoding.h"
 #include "dwio/nimble/encodings/DictionaryEncoding.h"
 #include "dwio/nimble/encodings/FixedBitWidthEncoding.h"
+#include "dwio/nimble/encodings/ForEncoding.h"
 #include "dwio/nimble/encodings/MainlyConstantEncoding.h"
 #include "dwio/nimble/encodings/NullableEncoding.h"
 #include "dwio/nimble/encodings/PrefixEncoding.h"
@@ -130,6 +131,12 @@ auto encodingTypeDispatchNonString(Encoding& encoding, F&& f) {
       return f(static_cast<MainlyConstantEncoding<T>&>(encoding));
     case EncodingType::Delta:
       return f(static_cast<DeltaEncoding<T>&>(encoding));
+    case EncodingType::FOR:
+      if constexpr (std::is_integral_v<T> && !std::is_same_v<T, bool>) {
+        return f(static_cast<ForEncoding<T>&>(encoding));
+      } else {
+        NIMBLE_UNREACHABLE(toString(encoding.dataType()));
+      }
     default:
       NIMBLE_UNSUPPORTED(toString(encoding.encodingType()));
   }