feat(nimble): Add shared integer dictionary encoding (#18637)

Summary:
Pull Request resolved: #18637

Add shared dictionary encoding for integer Nimble value streams. Encoded chunks store the normal prefix plus dictionary indices, while the alphabet is resolved from a stripe stream, a file catalog, or an external resolver. This lets selected value streams share alphabets even when their physical stream ids vary by stripe, including FlatMap values.

Wire dictionary resolution through tablet readers and bind the resolved alphabet before regular, selective, and index-reader decoding. Writers configure eligible regular columns and flat-map values explicitly, compare shared dictionary encoding with direct encoding for stripe scope on the first chunk, and keep that choice for the rest of the stripe.

File and external dictionaries use prebuilt alphabets and avoid per-chunk scope or dictionary-id payload. Unsupported dictionary lifecycles remain rejected until they are implemented.

Reviewed By: tanjialiang

Differential Revision: D112768733

Author: xiaoxmeng

velox/dwio/common/SelectiveColumnReader.h

@@ -296,15 +296,17 @@ class SelectiveColumnReader {
     numValues_ = size;
   }
 
-  // The number of passing after filtering.
+  // The number of result rows after filtering.
   int32_t numRows() const {
     return outputRows_.size();
   }
 
-  // The number of values copied into the results.
+  // The number of result positions produced so far. This includes null
+  // positions; it is not the count of decoded non-null values.
   int32_t numValues() const {
     return numValues_;
   }
+
   void setNumRows(vector_size_t size) {
     outputRows_.resize(size);
   }

velox/dwio/nimble/encodings/SharedDictionaryEncoding.cpp

@@ -42,6 +42,7 @@ SharedDictionaryAlphabet::SharedDictionaryAlphabet(
     std::shared_ptr<const void> encodedAlphabetOwner,
     velox::memory::MemoryPool* pool)
     : encodedAlphabetOwner_{std::move(encodedAlphabetOwner)},
+      encodedAlphabet_{encoded},
       dataType_{EncodingPrefix::dataType(encoded)},
       encodingType_{EncodingPrefix::encodingType(encoded)},
       entryPayload_{*velox::checkedNotNull(pool)},

velox/dwio/nimble/encodings/SharedDictionaryEncoding.h

@@ -40,6 +40,7 @@
 #include "velox/dwio/nimble/common/Types.h"
 #include "velox/dwio/nimble/common/Vector.h"
 #include "velox/dwio/nimble/encodings/DictionaryEncoding.h"
+#include "velox/dwio/nimble/encodings/NullableEncoding.h"
 #include "velox/dwio/nimble/encodings/common/Encoding.h"
 #include "velox/dwio/nimble/encodings/common/EncodingFactory.h"
 #include "velox/dwio/nimble/encodings/common/EncodingPrefix.h"
@@ -298,6 +299,11 @@ class SharedDictionaryAlphabet {
     return encodingType_;
   }
 
+  /// Returns the encoded alphabet stream passed to create().
+  std::string_view encodedAlphabet() const {
+    return encodedAlphabet_;
+  }
+
   /// Returns the entry stored at index. T must match dataType().
   template <typename T>
   typename TypeTraits<T>::physicalType physicalValueAt(uint32_t index) const {
@@ -432,6 +438,7 @@ class SharedDictionaryAlphabet {
 
   // Keeps borrowed encoded bytes alive when create() receives an owner.
   const std::shared_ptr<const void> encodedAlphabetOwner_;
+  const std::string_view encodedAlphabet_;
   const DataType dataType_;
   const EncodingType encodingType_;
   uint32_t entryCount_{0};
@@ -503,6 +510,13 @@ class SharedDictionaryEncoding
       Buffer& buffer,
       const Encoding::Options& options = {});
 
+  static std::string_view encodeNullable(
+      EncodingSelection<physicalType>&& selection,
+      std::span<const physicalType> values,
+      std::span<const bool> nulls,
+      Buffer& buffer,
+      const Encoding::Options& options = {});
+
   static std::string_view slice(
       std::string_view encoded,
       uint32_t offset,
@@ -666,6 +680,34 @@ std::string_view SharedDictionaryEncoding<T>::encode(
   return {reserved, encodingSize};
 }
 
+template <typename T>
+std::string_view SharedDictionaryEncoding<T>::encodeNullable(
+    EncodingSelection<physicalType>&& selection,
+    std::span<const physicalType> values,
+    std::span<const bool> nulls,
+    Buffer& buffer,
+    const Encoding::Options& options) {
+  static_assert(isIntegralType<T>() && !std::is_same_v<T, bool>);
+
+  auto nullsPolicy = selection.template createNestedPolicy<bool>(
+      EncodingType::Nullable, EncodingIdentifiers::Nullable::Nulls);
+  NIMBLE_CHECK_NOT_NULL(nullsPolicy);
+  auto typedNullsPolicy = std::unique_ptr<EncodingSelectionPolicy<bool>>(
+      static_cast<EncodingSelectionPolicy<bool>*>(nullsPolicy.release()));
+  auto* pool = &buffer.getMemoryPool();
+  ScopedEncodingBuffer scopedBuffer{pool, options.encodingBufferPool};
+  const auto serializedValues = EncodingFactory::encode<T>(
+      std::move(selection), values, scopedBuffer.get(), options);
+  const auto serializedNulls = EncodingFactory::encode<bool>(
+      std::move(typedNullsPolicy), nulls, scopedBuffer.get(), options);
+  return NullableEncoding<T>::encodeNullable(
+      static_cast<uint32_t>(nulls.size()),
+      serializedValues,
+      serializedNulls,
+      buffer,
+      options);
+}
+
 template <typename T>
 std::string_view SharedDictionaryEncoding<T>::encodeIndices(
     std::span<const uint32_t> indices,

velox/dwio/nimble/encodings/common/EncodingFactory.cpp

@@ -284,9 +284,36 @@ std::string_view EncodingFactory::encode(
           selection, castedValues, buffer, options);
     }
     case EncodingType::SharedDictionary: {
+      if constexpr (isIntegralType<T>() && !std::is_same_v<T, bool>) {
+        const auto& sharedDictionaryInput = selection.sharedDictionaryInput();
+        NIMBLE_CHECK(
+            sharedDictionaryInput.has_value(),
+            "SharedDictionary encoding requires writer-provided dictionary "
+            "indices.");
+        NIMBLE_CHECK_EQ(
+            sharedDictionaryInput->indices.size(),
+            castedValues.size(),
+            "SharedDictionary index count differs from value count.");
+        EncodingSelectionPolicyCreator nestedPolicyCreator =
+            [&selection](DataType nestedDataType)
+            -> std::unique_ptr<EncodingSelectionPolicyBase> {
+          NIMBLE_CHECK_EQ(
+              nestedDataType,
+              DataType::Uint32,
+              "SharedDictionary index stream must use Uint32.");
+          return selection.template createNestedPolicy<uint32_t>(
+              EncodingType::SharedDictionary,
+              EncodingIdentifiers::SharedDictionary::Indices);
+        };
+        return SharedDictionaryEncoding<T>::encode(
+            sharedDictionaryInput->indices,
+            nestedPolicyCreator,
+            buffer,
+            options);
+      }
       NIMBLE_INCOMPATIBLE_ENCODING(
-          "SharedDictionary encoding requires writer-provided dictionary "
-          "indices.");
+          "SharedDictionary encoding only supports non-bool integer data "
+          "types.");
     }
     case EncodingType::FixedBitWidth: {
       if constexpr (isNumericType<physicalType>()) {
@@ -429,6 +456,15 @@ std::string_view EncodingFactory::encodeNullable(
       return NullableEncoding<T>::encodeNullable(
           selection, physicalValues, nulls, buffer, options);
     }
+    case EncodingType::SharedDictionary: {
+      if constexpr (isIntegralType<T>() && !std::is_same_v<T, bool>) {
+        return SharedDictionaryEncoding<T>::encodeNullable(
+            std::move(selection), physicalValues, nulls, buffer, options);
+      }
+      NIMBLE_INCOMPATIBLE_ENCODING(
+          "SharedDictionary encoding only supports non-bool integer data "
+          "types.");
+    }
     default: {
       NIMBLE_UNSUPPORTED(
           "Encoding {} is not supported for nullable data.",

velox/dwio/nimble/encodings/common/EncodingFactory.h

@@ -30,6 +30,9 @@ class EncodingSelection;
 template <typename T>
 class EncodingSelectionPolicy;
 
+template <typename T>
+class SharedDictionaryEncoding;
+
 class EncodingFactory {
  public:
   explicit EncodingFactory(Encoding::Options options = {})
@@ -118,6 +121,8 @@ class EncodingFactory {
   friend class EncodingSelection<double>;
   friend class EncodingSelection<bool>;
   friend class EncodingSelection<std::string_view>;
+  template <typename T>
+  friend class SharedDictionaryEncoding;
 };
 
 } // namespace facebook::nimble

velox/dwio/nimble/encodings/selection/EncodingSelection.h

@@ -71,6 +71,12 @@ namespace facebook::nimble {
 
 class EncodingSelectionPolicyBase;
 
+/// Writer-provided input for shared dictionary value-stream encoding.
+struct SharedDictionaryEncodingInput {
+  /// One dictionary index for each non-null input value.
+  std::span<const uint32_t> indices;
+};
+
 /// Type representing a selected encoding.
 /// This is the result type returned from the select() method of an encoding
 /// selection policy. Also provides access to the compression policies for
@@ -85,6 +91,8 @@ struct EncodingSelectionResult {
   std::optional<uint64_t> estimatedSize{};
   std::function<std::unique_ptr<CompressionPolicy>()> compressionPolicyFactory =
       []() { return std::make_unique<NoCompressionPolicy>(); };
+  /// Additional input required when encodingType is SharedDictionary.
+  std::optional<SharedDictionaryEncodingInput> sharedDictionaryInput{};
 };
 
 /// The EncodingSelection class is passed in to the encode() method of each
@@ -114,6 +122,11 @@ class EncodingSelection {
     return policy;
   }
 
+  const std::optional<SharedDictionaryEncodingInput>& sharedDictionaryInput()
+      const noexcept {
+    return selectionResult_.sharedDictionaryInput;
+  }
+
   template <typename NestedT>
   std::unique_ptr<EncodingSelectionPolicyBase> createNestedPolicy(
       EncodingType parentEncodingType,