feat: Support stripe level batched index read (facebookincubator#480)

Summary:
X-link: facebookincubator/velox#16360


This diff implements stripe-level batched index read support for Nimble files in Velox. The key motivation is to improve index lookup performance by processing multiple lookup requests in batches at the stripe level, rather than processing each request individually.

**New `SelectiveNimbleIndexReader` class**: A format-specific index reader that handles:
   - Encoding index bounds into Nimble-specific encoded keys
   - Looking up stripes and row ranges using the tablet index
   - Managing stripe iteration and data reading with batched processing
   - Returning results in request order via an iterator pattern (`startLookup`/`hasNext`/`next`)

**Batched stripe processing**: Instead of loading stripes per-request, the reader:
   - Maps all lookup requests to their matching stripes upfront
   - Merges overlapping row ranges within stripes for efficient reading (without filters)
   - Splits overlapping ranges into non-overlapping segments to preserve filter semantics (with filters)
   - Tracks output references with ref-counting to share read data across requests

**HiveIndexReader refactoring**: Simplified to focus on index bounds creation and result assembly, delegating format-specific control logic to `SelectiveNimbleIndexReader`.

**KeyEncoder enhancements**: Added support for encoding index bounds with constant values for efficient multi-row range queries, and extended test coverage for edge cases.

**New runtime stats**: Added metrics for tracking index lookup performance:
   - `kNumIndexLookupRequests`: Total lookup requests processed
   - `kNumIndexLookupStripes`: Number of stripes accessed
   - `kNumIndexLookupReadSegments`: Number of read segments processed

Reviewed By: HuamengJiang, tanjialiang

Differential Revision: D92848948

Author: xiaoxmeng

dwio/nimble/index/TabletIndex.cpp

@@ -20,7 +20,9 @@
 #include "dwio/nimble/common/Exceptions.h"
 #include "dwio/nimble/tablet/IndexGenerated.h"
 #include "dwio/nimble/tablet/MetadataBuffer.h"
+#include "folly/String.h"
 #include "folly/json/json.h"
+#include "folly/logging/xlog.h"
 
 namespace facebook::nimble::index {
 
@@ -122,7 +124,6 @@ std::optional<StripeLocation> TabletIndex::lookup(
 
   // Calculate which stripe contains the key
   const uint32_t targetStripe = (it - stripeKeys_.begin()) - 1;
-
   // Check if the key is beyond all stripes
   if (targetStripe >= numStripes_) {
     return std::nullopt;

dwio/nimble/index/tests/IndexTestUtils.cpp

@@ -60,12 +60,16 @@ void writeFile(
     const std::string& filePath,
     const std::vector<velox::RowVectorPtr>& data,
     IndexConfig indexConfig,
-    velox::memory::MemoryPool& pool) {
+    velox::memory::MemoryPool& pool,
+    std::function<std::unique_ptr<FlushPolicy>()> flushPolicyFactory) {
   NIMBLE_CHECK(!data.empty(), "Data must not be empty");
 
   VeloxWriterOptions options;
   options.enableChunking = true;
   options.indexConfig = std::move(indexConfig);
+  if (flushPolicyFactory) {
+    options.flushPolicyFactory = std::move(flushPolicyFactory);
+  }
 
   auto fs = velox::filesystems::getFileSystem(filePath, {});
   auto writeFile = fs->openFileForWrite(

dwio/nimble/index/tests/IndexTestUtils.h

@@ -65,11 +65,15 @@ std::vector<velox::RowVectorPtr> generateData(
 /// @param data Vector of RowVectorPtr to write.
 /// @param indexConfig Index configuration specifying columns, sort orders, etc.
 /// @param pool Memory pool for writing.
+/// @param flushPolicyFactory Optional factory for custom flush policy to
+/// control
+///        stripe sizes. If not provided, uses default flush policy.
 void writeFile(
     const std::string& filePath,
     const std::vector<velox::RowVectorPtr>& data,
     IndexConfig indexConfig,
-    velox::memory::MemoryPool& pool);
+    velox::memory::MemoryPool& pool,
+    std::function<std::unique_ptr<FlushPolicy>()> flushPolicyFactory = nullptr);
 
 /// Simple StreamLoader implementation for testing that holds data in memory.
 class TestStreamLoader : public StreamLoader {