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

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

@@ -687,6 +687,20 @@ void readWithVisitorFast(
     auto numNonNulls = velox::simd::indicesOfSetBits(
         nulls, visitor.rowIndex(), visitor.numRows(), outerRows.data());
     outerRows.resize(numNonNulls);
+    if constexpr (kOutputNulls) {
+      if (numNonNulls != numRows) {
+        if (!visitor.reader().returnReaderNulls()) {
+          params.prepareResultNulls();
+          velox::bits::copyBits(
+              nulls,
+              visitor.rowIndex(),
+              visitor.reader().rawResultNulls(),
+              visitor.rowIndex(),
+              numRows);
+        }
+        visitor.setHasNulls();
+      }
+    }
     if (outerRows.empty()) {
       if constexpr (kOutputNulls) {
         visitor.addNumValues(numRows);

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,

velox/dwio/nimble/encodings/tests/ReadWithVisitorTest.cpp

@@ -748,6 +748,14 @@ TEST_P(ReadWithVisitorTest, denseNoFilterWithNulls) {
   // Every row is "output" since there is no filter.
   EXPECT_EQ(child->numValues(), kRows);
   EXPECT_TRUE(child->hasNulls());
+
+  std::vector<vector_size_t> rowNumbers(kRows);
+  std::iota(rowNumbers.begin(), rowNumbers.end(), 0);
+  VectorPtr result;
+  child->getValues(RowSet(rowNumbers.data(), rowNumbers.size()), &result);
+  for (int i = 0; i < kRows; ++i) {
+    EXPECT_EQ(result->isNullAt(i), i % 7 == 0) << "row " << i;
+  }
 }
 
 // ===========================================================================