perf(nimble): Add bulk-scan path to legacy fixed-width reads (#18612)

Summary:

This diff is created by PerfAICT to optimize `facebook::nimble::legacy::callReadWithVisitor` in "fbcode/velox/dwio/nimble/encodings/legacy/EncodingUtils.h", by reducing CPU cycles spent in this function.

### Optimization Details

The inclusive cost of `legacy::callReadWithVisitor` was dominated by `FixedBitWidthEncoding<T>::readWithVisitor`, which always decoded values one at a time via the scalar per-value slow path (`fixedBitArray_.get(row_++) + baseline_` followed by per-value `ColumnVisitor::process`). This change adds a bulk-scan fast path — mirroring the already-landed canonical implementation in the non-legacy `encodings/FixedBitWidthEncoding.h` — that handles the dominant dense, no-null, no-filter/hook, 4/8-byte integral ExtractToReader case with a single `bulkGetWithBaseline` decode plus `processFixedWidthRun`, falling back to the unchanged slow path for all other cases. All supporting helpers (`readWithVisitorFast`, `useFastPath`, `processFixedWidthRun`, `FixedBitArray`) are shared, not duplicated, so behavior is preserved.

Differential Revision: D116602815

Author: MatzeB

velox/dwio/nimble/encodings/legacy/FixedBitWidthEncoding.h

@@ -15,6 +15,7 @@
  */
 #pragma once
 
+#include <cstring>
 #include <span>
 #include <type_traits>
 
@@ -64,6 +65,14 @@ class FixedBitWidthEncoding final
   template <typename DecoderVisitor>
   void readWithVisitor(DecoderVisitor& visitor, ReadWithVisitorParams& params);
 
+  template <bool kScatter, typename Visitor>
+  void bulkScan(
+      Visitor& visitor,
+      vector_size_t currentRow,
+      const vector_size_t* selectedRows,
+      vector_size_t numSelected,
+      const vector_size_t* scatterRows);
+
   static std::string_view encode(
       EncodingSelection<physicalType>& selection,
       std::span<const physicalType> values,
@@ -131,6 +140,25 @@ template <typename V>
 void FixedBitWidthEncoding<T>::readWithVisitor(
     V& visitor,
     ReadWithVisitorParams& params) {
+  using OutputType = detail::ValueType<typename V::DataType>;
+  constexpr bool kIsSuitableWidth =
+      (isFourByteIntegralType<physicalType>() ||
+       isEightByteIntegralType<physicalType>());
+  constexpr bool kIsFluidCast = sizeof(OutputType) >= sizeof(physicalType) &&
+      std::is_integral_v<OutputType> && std::is_integral_v<physicalType>;
+  // Limit bulk decoding to supported integral conversions.
+  if constexpr (
+      kIsSuitableWidth &&
+      std::is_same_v<
+          typename V::Extract,
+          velox::dwio::common::ExtractToReader> &&
+      kIsFluidCast) {
+    auto* nulls = visitor.reader().rawNullsInReadRange();
+    if (velox::dwio::common::useFastPath(visitor, nulls)) {
+      detail::readWithVisitorFast(*this, visitor, params, nulls);
+      return;
+    }
+  }
   detail::readWithVisitorSlow(
       visitor,
       params,
@@ -141,6 +169,106 @@ void FixedBitWidthEncoding<T>::readWithVisitor(
       });
 }
 
+template <typename T>
+template <bool kScatter, typename V>
+void FixedBitWidthEncoding<T>::bulkScan(
+    V& visitor,
+    vector_size_t currentRow,
+    const vector_size_t* selectedRows,
+    vector_size_t numSelected,
+    const vector_size_t* scatterRows) {
+  using OutputType = detail::ValueType<typename V::DataType>;
+  static_assert(
+      isFourByteIntegralType<physicalType>() ||
+          isEightByteIntegralType<physicalType>(),
+      "bulkScan only supports 4-byte or 8-byte integral types");
+
+  if (numSelected == 0) {
+    return;
+  }
+
+  const auto numRows = visitor.numRows() - visitor.rowIndex();
+
+  // Translate logical rows to the encoding's non-null row space.
+  const auto offset =
+      static_cast<int32_t>(row_) - static_cast<int32_t>(currentRow);
+
+  auto* values = detail::mutableValues<OutputType>(visitor, numRows);
+
+  constexpr bool kSameSize = sizeof(physicalType) == sizeof(OutputType);
+  constexpr bool kIsUpcast = sizeof(OutputType) > sizeof(physicalType) &&
+      std::is_integral_v<OutputType> && std::is_integral_v<physicalType>;
+
+  if constexpr (V::dense) {
+    if constexpr (isFourByteIntegralType<physicalType>()) {
+      if constexpr (kSameSize) {
+        buffer_.resize(numSelected);
+        fixedBitArray_.bulkGetWithBaseline(
+            selectedRows[0] + offset, numSelected, buffer_.data(), baseline_);
+        std::memcpy(values, buffer_.data(), numSelected * sizeof(physicalType));
+      } else if constexpr (kIsUpcast) {
+        static_assert(isEightByteIntegralType<OutputType>());
+        // Decode directly into the widened output.
+        fixedBitArray_.bulkGetWithBaseline32Into64(
+            selectedRows[0] + offset,
+            numSelected,
+            reinterpret_cast<uint64_t*>(values),
+            baseline_);
+      }
+    } else {
+      static_assert(isEightByteIntegralType<physicalType>());
+      static_assert(kSameSize, "8-byte bulkScan requires same-size output");
+      fixedBitArray_.bulkGetWithBaseline(
+          selectedRows[0] + offset,
+          numSelected,
+          reinterpret_cast<physicalType*>(values),
+          baseline_);
+    }
+  } else {
+    for (vector_size_t i = 0; i < numSelected; ++i) {
+      values[i] = static_cast<OutputType>(
+          fixedBitArray_.get(selectedRows[i] + offset) + baseline_);
+    }
+  }
+
+  row_ += selectedRows[numSelected - 1] - currentRow + 1;
+
+  if constexpr (!kScatter && !V::kHasFilter && !V::kHasHook) {
+    visitor.addNumValues(numRows);
+    visitor.setRowIndex(visitor.numRows());
+    return;
+  }
+
+  // Apply scattering, filtering, or hooks after decoding.
+  if constexpr (!V::kHasHook) {
+    values = reinterpret_cast<OutputType*>(visitor.reader().rawValues());
+  }
+
+  auto numValues = visitor.reader().numValues();
+  int32_t* filterHits = nullptr;
+  if constexpr (V::kHasFilter) {
+    filterHits = visitor.outputRows(numSelected) - numValues;
+  }
+
+  velox::dwio::common::
+      processFixedWidthRun<OutputType, V::kFilterOnly, kScatter, V::dense>(
+          velox::RowSet(selectedRows, numSelected),
+          0,
+          numSelected,
+          scatterRows,
+          values,
+          filterHits,
+          numValues,
+          visitor.filter(),
+          visitor.hook());
+
+  if constexpr (!V::kHasHook) {
+    visitor.addNumValues(
+        V::kHasFilter ? numValues - visitor.reader().numValues() : numRows);
+  }
+  visitor.setRowIndex(visitor.numRows());
+}
+
 template <typename T>
 std::string_view FixedBitWidthEncoding<T>::encode(
     EncodingSelection<physicalType>& selection,