diff --git a/velox/dwio/common/ColumnVisitors.h b/velox/dwio/common/ColumnVisitors.h
index 33d7a2d6e32..651cd119719 100644
--- a/velox/dwio/common/ColumnVisitors.h
+++ b/velox/dwio/common/ColumnVisitors.h
@@ -408,6 +408,70 @@ class ColumnVisitor {
   inline void addNull();
   inline void addOutputRow(vector_size_t row);
 
+  /// Bulk variant of process() with SIMD paths for AlwaysTrue and
+  /// deterministic integer filters; otherwise per-row fallback.
+  template <bool hasFilter, bool hasHook, bool scatter>
+  FOLLY_ALWAYS_INLINE void processRun(
+      const T* input,
+      int32_t numInput,
+      const int32_t* scatterRows,
+      int32_t* filterHits,
+      T* values,
+      int32_t& numValues) {
+    DCHECK_EQ(input, values + numValues);
+    if constexpr (
+        !hasFilter && !hasHook && !scatter && isDense &&
+        std::is_same_v<TFilter, velox::common::AlwaysTrue>) {
+      rowIndex_ += numInput;
+      numValues += numInput;
+      return;
+    }
+    if constexpr (
+        hasFilter && !hasHook && !scatter && isDense &&
+        TFilter::deterministic &&
+        (std::is_same_v<T, int32_t> || std::is_same_v<T, int64_t> ||
+         std::is_same_v<T, int16_t>)) {
+      const int32_t firstRow = currentRow();
+      constexpr int32_t kWidth = xsimd::batch<T>::size;
+      int32_t i = 0;
+      while (i + kWidth <= numInput) {
+        auto batch = xsimd::load_unaligned(input + i);
+        processFixedFilter<
+            T,
+            /*filterOnly=*/false,
+            /*scatter=*/false,
+            /*dense=*/true>(
+            batch,
+            kWidth,
+            firstRow + i,
+            filter_,
+            [&](int32_t /*offset*/) {
+              return simd::loadGatherIndices<T>(rows_ + rowIndex_ + i);
+            },
+            values,
+            filterHits,
+            numValues);
+        i += kWidth;
+      }
+      for (; i < numInput; ++i) {
+        if (velox::common::applyFilter(filter_, input[i])) {
+          values[numValues] = input[i];
+          filterHits[numValues] = firstRow + i;
+          ++numValues;
+        }
+      }
+      rowIndex_ += numInput;
+      return;
+    }
+    bool atEnd = false;
+    for (int32_t i = 0; i < numInput; ++i) {
+      process(input[i], atEnd);
+      if (atEnd) {
+        return;
+      }
+    }
+  }
+
   const TFilter& filter() {
     return filter_;
   }
diff --git a/velox/dwio/parquet/reader/DeltaBpDecoder.h b/velox/dwio/parquet/reader/DeltaBpDecoder.h
index 75e846de705..54c191c3d36 100644
--- a/velox/dwio/parquet/reader/DeltaBpDecoder.h
+++ b/velox/dwio/parquet/reader/DeltaBpDecoder.h
@@ -20,6 +20,8 @@
 #include "velox/common/base/BitUtil.h"
 #include "velox/common/base/Exceptions.h"
 #include "velox/common/base/Nulls.h"
+#include "velox/dwio/common/DecoderUtil.h"
+#include "velox/type/Filter.h"
 
 namespace facebook::velox::parquet {
 
@@ -27,6 +29,10 @@ namespace facebook::velox::parquet {
 // https://github.com/apache/arrow/blob/apache-arrow-12.0.0/cpp/src/parquet/encoding.cc#LL2357C18-L2586C3
 class DeltaBpDecoder {
  public:
+  /// Trailing readable bytes the SIMD kernel needs past the last
+  /// page byte; PageReader::kPageReadPadding must be >= this.
+  static constexpr int kRequiredTrailingPadding = 8;
+
   explicit DeltaBpDecoder(const char* start) : bufferStart_(start) {
     initHeader();
   }
@@ -36,7 +42,8 @@ class DeltaBpDecoder {
   }
 
   template <bool hasNulls>
-  inline void skip(int32_t numValues, int32_t current, const uint64_t* nulls) {
+  FOLLY_ALWAYS_INLINE void
+  skip(int32_t numValues, int32_t current, const uint64_t* nulls) {
     if (hasNulls) {
       numValues = bits::countNonNulls(nulls, current, current + numValues);
     }
@@ -49,6 +56,20 @@ class DeltaBpDecoder {
   void readWithVisitor(const uint64_t* nulls, Visitor visitor) {
     int32_t current = visitor.start();
     skip<hasNulls>(current, 0, nulls);
+    if constexpr (
+        Visitor::dense && !hasNulls && Visitor::FilterType::deterministic &&
+        std::is_same_v<typename Visitor::HookType, dwio::common::NoHook> &&
+        std::is_integral_v<typename Visitor::DataType>) {
+      readWithVisitorDenseBatched(visitor);
+      return;
+    }
+    if constexpr (
+        !Visitor::dense && !hasNulls && Visitor::FilterType::deterministic &&
+        std::is_same_v<typename Visitor::HookType, dwio::common::NoHook> &&
+        std::is_integral_v<typename Visitor::DataType>) {
+      readWithVisitorSparseBuffered(visitor);
+      return;
+    }
     int32_t toSkip;
     bool atEnd = false;
     const bool allowNulls = hasNulls && visitor.allowNulls();
@@ -89,14 +110,294 @@ class DeltaBpDecoder {
   }
 
   template <typename T>
-  void readValues(T* values, int32_t numValues) {
+  FOLLY_ALWAYS_INLINE void readValues(T* values, int32_t numValues) {
     VELOX_DCHECK_LE(numValues, totalValuesRemaining_);
-    for (auto i = 0; i < numValues; i++) {
-      values[i] = T(readLong());
+    if constexpr (std::is_integral_v<T>) {
+      decodeLongs(values, numValues);
+    } else {
+      for (auto i = 0; i < numValues; i++) {
+        values[i] = T(readLong());
+      }
     }
   }
 
  private:
+  static constexpr int32_t kBatch = 1024;
+
+  template <typename Visitor>
+  void readWithVisitorDenseBatched(Visitor& visitor) {
+    using DataType = typename Visitor::DataType;
+    constexpr bool kHasFilter =
+        !std::
+            is_same_v<typename Visitor::FilterType, velox::common::AlwaysTrue>;
+    const int32_t total = visitor.numRows();
+    DataType* output = visitor.rawValues(total);
+    int32_t* filterHits = kHasFilter ? visitor.outputRows(total) : nullptr;
+    int32_t numValues = 0;
+    int32_t consumed = 0;
+    while (consumed < total) {
+      const int32_t n = std::min<int32_t>(kBatch, total - consumed);
+      DataType* dst = output + numValues;
+      decodeLongs(dst, n);
+      visitor.template processRun<
+          kHasFilter,
+          /*hasHook=*/false,
+          /*scatter=*/false>(
+          dst,
+          n,
+          /*scatterRows=*/nullptr,
+          filterHits,
+          output,
+          numValues);
+      consumed += n;
+    }
+    visitor.setNumValues(numValues);
+  }
+
+  // DELTA's chain forces every physical row to be decoded; batch the
+  // decode side and run the scalar visitor.process() over a buffer.
+  template <typename Visitor>
+  void readWithVisitorSparseBuffered(Visitor& visitor) {
+    using DataType = typename Visitor::DataType;
+    DataType buf[kBatch];
+
+    const auto* rows = visitor.rows();
+    const int32_t numRows = visitor.numRows();
+    int32_t rowIdx = 0;
+    int32_t currentPhys = (numRows > 0) ? rows[0] : 0;
+    bool atEnd = false;
+    while (!atEnd && rowIdx < numRows) {
+      const int32_t remaining = static_cast<int32_t>(totalValuesRemaining_);
+      if (remaining <= 0) {
+        return;
+      }
+      // Cap n to the visitor's residual span; over-decoding misaligns
+      // the next readWithVisitor call.
+      const int32_t lastRow = rows[numRows - 1];
+      const int32_t maxSpan = lastRow - currentPhys + 1;
+      const int32_t n = std::min<int32_t>({kBatch, remaining, maxSpan});
+      decodeLongs(buf, n);
+      const int32_t batchPhysStart = currentPhys;
+      const int32_t batchPhysEnd = currentPhys + n; // exclusive
+      currentPhys = batchPhysEnd;
+      while (rowIdx < numRows && rows[rowIdx] < batchPhysEnd) {
+        const int32_t i = rows[rowIdx] - batchPhysStart;
+        visitor.process(buf[i], atEnd);
+        ++rowIdx;
+        if (atEnd) {
+          return;
+        }
+      }
+    }
+  }
+
+  template <typename DataType>
+  void decodeLongs(DataType* out, int32_t n) {
+    const char* bufStart = bufferStart_;
+    uint64_t valuesPerMiniBlock = valuesPerMiniBlock_;
+    uint64_t miniBlockRemaining = valuesRemainingCurrentMiniBlock_;
+    uint64_t totalRemaining = totalValuesRemaining_;
+    int64_t lastValue = lastValue_;
+    int64_t minDelta = minDelta_;
+    uint64_t deltaBitWidth = deltaBitWidth_;
+    uint64_t bitOffset =
+        (valuesPerMiniBlock - miniBlockRemaining) * deltaBitWidth;
+
+    int32_t i = 0;
+    while (i < n) {
+      if (miniBlockRemaining == 0) {
+        if (!firstBlockInitialized_) {
+          bufferStart_ = bufStart;
+          valuesRemainingCurrentMiniBlock_ = 0;
+          totalValuesRemaining_ = totalRemaining;
+          lastValue_ = lastValue;
+          int64_t v = readLong();
+          out[i] = static_cast<DataType>(v);
+          bufStart = bufferStart_;
+          miniBlockRemaining = valuesRemainingCurrentMiniBlock_;
+          totalRemaining = totalValuesRemaining_;
+          lastValue = lastValue_;
+          minDelta = minDelta_;
+          deltaBitWidth = deltaBitWidth_;
+          bitOffset = (valuesPerMiniBlock - miniBlockRemaining) * deltaBitWidth;
+          ++i;
+          continue;
+        }
+        bufferStart_ = bufStart;
+        valuesRemainingCurrentMiniBlock_ = 0;
+        totalValuesRemaining_ = totalRemaining;
+        lastValue_ = lastValue;
+        advanceMiniBlock();
+        bufStart = bufferStart_;
+        miniBlockRemaining = valuesRemainingCurrentMiniBlock_;
+        minDelta = minDelta_;
+        deltaBitWidth = deltaBitWidth_;
+        bitOffset = 0;
+      }
+
+      if (miniBlockRemaining == valuesPerMiniBlock &&
+          static_cast<uint64_t>(n - i) >= valuesPerMiniBlock &&
+          totalRemaining >= valuesPerMiniBlock && deltaBitWidth <= 32) {
+        const int32_t miniBlockValues =
+            static_cast<int32_t>(valuesPerMiniBlock);
+        DataType* dst = out + i;
+        const bool dispatched = dispatchSimdMiniBlock<DataType>(
+            deltaBitWidth, bufStart, miniBlockValues, minDelta, lastValue, dst);
+        if (dispatched) {
+          bufStart += bits::nbytes(deltaBitWidth * valuesPerMiniBlock);
+          const uint64_t consumed = valuesPerMiniBlock;
+          miniBlockRemaining = 0;
+          totalRemaining -= consumed;
+          i += static_cast<int32_t>(consumed);
+          bitOffset = 0;
+          continue;
+        }
+      }
+
+      uint64_t value = 0;
+      if (deltaBitWidth) {
+        value = bits::detail::loadBits<uint64_t>(
+            reinterpret_cast<const uint64_t*>(bufStart),
+            bitOffset,
+            deltaBitWidth);
+        value &= (~0ULL >> (64 - deltaBitWidth));
+      }
+      uint64_t result = static_cast<uint64_t>(minDelta) + value +
+          static_cast<uint64_t>(lastValue);
+      lastValue = static_cast<int64_t>(result);
+      out[i] = static_cast<DataType>(result);
+      bitOffset += deltaBitWidth;
+      --miniBlockRemaining;
+      --totalRemaining;
+      if (miniBlockRemaining == 0 || totalRemaining == 0) {
+        bufStart += bits::nbytes(deltaBitWidth * valuesPerMiniBlock);
+        bitOffset = 0;
+      }
+      ++i;
+    }
+
+    bufferStart_ = bufStart;
+    valuesRemainingCurrentMiniBlock_ = miniBlockRemaining;
+    totalValuesRemaining_ = totalRemaining;
+    lastValue_ = lastValue;
+  }
+
+  /// Decode one whole miniblock with prefix-sum fused. Unsigned mod-2^64
+  /// per Parquet spec. Reads up to 7 bytes past the miniblock end;
+  /// safe via kPageReadPadding at page end and adjacent miniblocks
+  /// intra-page.
+  template <typename DataType, uint8_t bitWidth>
+  FOLLY_ALWAYS_INLINE void decodeMiniBlockSimd(
+      const char* src,
+      int32_t numValues,
+      int64_t minDelta,
+      int64_t& lastValue,
+      DataType* out) {
+    static_assert(bitWidth >= 1 && bitWidth <= 32);
+    constexpr uint64_t mask =
+        (bitWidth == 32) ? 0xFFFFFFFFULL : ((1ULL << bitWidth) - 1);
+    const uint8_t* p = reinterpret_cast<const uint8_t*>(src);
+    uint64_t cumulative = static_cast<uint64_t>(lastValue);
+    const uint64_t step = static_cast<uint64_t>(minDelta);
+    if constexpr (bitWidth <= 16) {
+      // 4*bw <= 64; one u64 load per iter.
+      for (int32_t i = 0; i < numValues; i += 4) {
+        const int32_t bitPos = i * bitWidth;
+        const int32_t byteOff = bitPos >> 3;
+        const int32_t bitInByte = bitPos & 7;
+        const uint64_t word =
+            *reinterpret_cast<const uint64_t*>(p + byteOff) >> bitInByte;
+        cumulative += step + (word & mask);
+        out[i + 0] = static_cast<DataType>(cumulative);
+        cumulative += step + ((word >> bitWidth) & mask);
+        out[i + 1] = static_cast<DataType>(cumulative);
+        cumulative += step + ((word >> (2 * bitWidth)) & mask);
+        out[i + 2] = static_cast<DataType>(cumulative);
+        cumulative += step + ((word >> (3 * bitWidth)) & mask);
+        out[i + 3] = static_cast<DataType>(cumulative);
+      }
+    } else {
+      // 2*bw + bitInByte > 64; load via __uint128_t (two u64 + SHRD).
+      // The +8 read is safe: kPageReadPadding covers page end and
+      // adjacent miniblocks cover intra-page.
+      for (int32_t i = 0; i < numValues; i += 2) {
+        const int32_t bitPos = i * bitWidth;
+        const int32_t byteOff = bitPos >> 3;
+        const int32_t bitInByte = bitPos & 7;
+        const __uint128_t window =
+            static_cast<__uint128_t>(
+                *reinterpret_cast<const uint64_t*>(p + byteOff)) |
+            (static_cast<__uint128_t>(
+                 *reinterpret_cast<const uint64_t*>(p + byteOff + 8))
+             << 64);
+        const uint64_t word = static_cast<uint64_t>(window >> bitInByte);
+        cumulative += step + (word & mask);
+        out[i + 0] = static_cast<DataType>(cumulative);
+        cumulative += step + ((word >> bitWidth) & mask);
+        out[i + 1] = static_cast<DataType>(cumulative);
+      }
+    }
+    lastValue = static_cast<int64_t>(cumulative);
+  }
+
+  template <typename DataType>
+  FOLLY_ALWAYS_INLINE void decodeMiniBlockConstantDelta(
+      int32_t numValues,
+      int64_t minDelta,
+      int64_t& lastValue,
+      DataType* out) {
+    uint64_t cumulative = static_cast<uint64_t>(lastValue);
+    const uint64_t step = static_cast<uint64_t>(minDelta);
+    for (int32_t i = 0; i < numValues; ++i) {
+      cumulative += step;
+      out[i] = static_cast<DataType>(cumulative);
+    }
+    lastValue = static_cast<int64_t>(cumulative);
+  }
+
+  /// Dispatches a whole-miniblock SIMD decode by runtime `bitWidth`.
+  /// Returns false for `bitWidth` outside [0, 32] so the caller falls
+  /// back to the scalar inner loop.
+  template <typename DataType>
+  FOLLY_ALWAYS_INLINE bool dispatchSimdMiniBlock(
+      uint64_t bitWidth,
+      const char* src,
+      int32_t numValues,
+      int64_t minDelta,
+      int64_t& lastValue,
+      DataType* out) {
+    if (bitWidth == 0) {
+      decodeMiniBlockConstantDelta(numValues, minDelta, lastValue, out);
+      return true;
+    }
+    return dispatchSimdMiniBlockImpl<DataType>(
+        bitWidth,
+        src,
+        numValues,
+        minDelta,
+        lastValue,
+        out,
+        std::make_index_sequence<32>{});
+  }
+
+  template <typename DataType, std::size_t... Is>
+  FOLLY_ALWAYS_INLINE bool dispatchSimdMiniBlockImpl(
+      uint64_t bitWidth,
+      const char* src,
+      int32_t numValues,
+      int64_t minDelta,
+      int64_t& lastValue,
+      DataType* out,
+      std::index_sequence<Is...>) {
+    bool dispatched = false;
+    (void)((bitWidth == Is + 1 ? (decodeMiniBlockSimd<DataType, Is + 1>(
+                                      src, numValues, minDelta, lastValue, out),
+                                  dispatched = true)
+                               : false) ||
+           ...);
+    return dispatched;
+  }
+
   bool getVlqInt(uint64_t& v) {
     uint64_t tmp = 0;
     for (int i = 0; i < folly::kMaxVarintLength64; i++) {
@@ -177,7 +478,19 @@ class DeltaBpDecoder {
     valuesRemainingCurrentMiniBlock_ = valuesPerMiniBlock_;
   }
 
-  int64_t readLong() {
+  // Advance to the next miniblock without decoding any value.
+  void advanceMiniBlock() {
+    VELOX_DCHECK(firstBlockInitialized_);
+    VELOX_DCHECK_EQ(valuesRemainingCurrentMiniBlock_, 0);
+    ++miniBlockIdx_;
+    if (miniBlockIdx_ < miniBlocksPerBlock_) {
+      initMiniBlock(deltaBitWidths_[miniBlockIdx_]);
+    } else {
+      initBlock();
+    }
+  }
+
+  FOLLY_ALWAYS_INLINE int64_t readLong() {
     int64_t value = 0;
     if (valuesRemainingCurrentMiniBlock_ == 0) {
       if (!firstBlockInitialized_) {
diff --git a/velox/dwio/parquet/reader/DeltaByteArrayDecoder.h b/velox/dwio/parquet/reader/DeltaByteArrayDecoder.h
index b8ad2d18d48..f52d4d30af0 100644
--- a/velox/dwio/parquet/reader/DeltaByteArrayDecoder.h
+++ b/velox/dwio/parquet/reader/DeltaByteArrayDecoder.h
@@ -16,24 +16,44 @@
 
 #pragma once
 
+#include <optional>
+
 #include "velox/common/base/BitUtil.h"
 #include "velox/common/base/Nulls.h"
 #include "velox/dwio/parquet/reader/DeltaBpDecoder.h"
 
 namespace facebook::velox::parquet {
 
-class DeltaByteArrayDecoderBase {
+// DeltaLengthByteArrayDecoder is adapted from Apache Arrow:
+// https://github.com/apache/arrow/blob/apache-arrow-15.0.0/cpp/src/parquet/encoding.cc#L2758-L2889
+class DeltaLengthByteArrayDecoder {
  public:
-  virtual ~DeltaByteArrayDecoderBase() = default;
+  DeltaLengthByteArrayDecoder() = default;
+
+  explicit DeltaLengthByteArrayDecoder(const char* start) {
+    reset(start);
+  }
 
-  virtual std::string_view readString() = 0;
+  void reset(const char* start) {
+    lengthDecoder_.emplace(start);
+    decodeLengths();
+    bufferStart_ = lengthDecoder_->bufferStart();
+  }
+
+  FOLLY_ALWAYS_INLINE std::string_view readString() {
+    const int64_t length = bufferedLength_[lengthIdx_++];
+    VELOX_CHECK_GE(length, 0, "negative string delta length");
+    bufferStart_ += length;
+    return std::string_view(bufferStart_ - length, length);
+  }
 
   void skip(uint64_t numValues) {
     skip<false>(numValues, 0, nullptr);
   }
 
   template <bool hasNulls>
-  inline void skip(int32_t numValues, int32_t current, const uint64_t* nulls) {
+  FOLLY_ALWAYS_INLINE void
+  skip(int32_t numValues, int32_t current, const uint64_t* nulls) {
     if (hasNulls) {
       numValues = bits::countNonNulls(nulls, current, current + numValues);
     }
@@ -76,24 +96,6 @@ class DeltaByteArrayDecoderBase {
       }
     }
   }
-};
-
-// DeltaByteArrayDecoder is adapted from Apache Arrow:
-// https://github.com/apache/arrow/blob/apache-arrow-15.0.0/cpp/src/parquet/encoding.cc#L2758-L2889
-class DeltaLengthByteArrayDecoder : public DeltaByteArrayDecoderBase {
- public:
-  explicit DeltaLengthByteArrayDecoder(const char* start) {
-    lengthDecoder_ = std::make_unique<DeltaBpDecoder>(start);
-    decodeLengths();
-    bufferStart_ = lengthDecoder_->bufferStart();
-  }
-
-  std::string_view readString() override {
-    const int64_t length = bufferedLength_[lengthIdx_++];
-    VELOX_CHECK_GE(length, 0, "negative string delta length");
-    bufferStart_ += length;
-    return std::string_view(bufferStart_ - length, length);
-  }
 
  private:
   void decodeLengths() {
@@ -107,7 +109,7 @@ class DeltaLengthByteArrayDecoder : public DeltaByteArrayDecoderBase {
   }
 
   const char* bufferStart_;
-  std::unique_ptr<DeltaBpDecoder> lengthDecoder_;
+  std::optional<DeltaBpDecoder> lengthDecoder_;
   int32_t numValidValues_{0};
   uint32_t lengthIdx_{0};
   std::vector<uint32_t> bufferedLength_;
@@ -115,74 +117,113 @@ class DeltaLengthByteArrayDecoder : public DeltaByteArrayDecoderBase {
 
 // DeltaByteArrayDecoder is adapted from Apache Arrow:
 // https://github.com/apache/arrow/blob/apache-arrow-15.0.0/cpp/src/parquet/encoding.cc#L3301-L3545
-class DeltaByteArrayDecoder : public DeltaByteArrayDecoderBase {
+class DeltaByteArrayDecoder {
  public:
+  /// Default-constructed instance is uninitialized; call reset() first.
+  DeltaByteArrayDecoder() = default;
+
   explicit DeltaByteArrayDecoder(const char* start) {
-    prefixLenDecoder_ = std::make_unique<DeltaBpDecoder>(start);
+    reset(start);
+  }
+
+  void reset(const char* start) {
+    prefixLenDecoder_.emplace(start);
     int64_t numPrefix = prefixLenDecoder_->validValuesCount();
     bufferedPrefixLength_.resize(numPrefix);
     prefixLenDecoder_->readValues<uint32_t>(
         bufferedPrefixLength_.data(), static_cast<int32_t>(numPrefix));
     prefixLenOffset_ = 0;
     numValidValues_ = static_cast<int32_t>(numPrefix);
+    lastValueLen_ = 0;
 
-    suffixDecoder_ = std::make_unique<DeltaLengthByteArrayDecoder>(
-        prefixLenDecoder_->bufferStart());
+    if (!suffixDecoder_.has_value()) {
+      suffixDecoder_.emplace(prefixLenDecoder_->bufferStart());
+    } else {
+      suffixDecoder_->reset(prefixLenDecoder_->bufferStart());
+    }
   }
 
-  std::string_view readString() override {
+  FOLLY_ALWAYS_INLINE std::string_view readString() {
     auto suffix = suffixDecoder_->readString();
-    bool isFirstRun = (prefixLenOffset_ == 0);
     const int64_t prefixLength = bufferedPrefixLength_[prefixLenOffset_++];
 
     VELOX_CHECK_GE(
         prefixLength, 0, "negative prefix length in DELTA_BYTE_ARRAY");
+    VELOX_CHECK_LE(
+        prefixLength,
+        lastValueLen_,
+        "prefix length too large in DELTA_BYTE_ARRAY");
 
-    buildReadValue(isFirstRun, prefixLength, suffix);
+    const size_t need = prefixLength + suffix.size();
+    if (need > lastValueBuf_.size()) {
+      lastValueBuf_.resize(need);
+    }
+    if (!suffix.empty()) {
+      memcpy(lastValueBuf_.data() + prefixLength, suffix.data(), suffix.size());
+    }
+    lastValueLen_ = static_cast<int32_t>(need);
 
     numValidValues_--;
-    return {lastValue_};
+    return {lastValueBuf_.data(), static_cast<size_t>(lastValueLen_)};
   }
 
- private:
-  void buildReadValue(
-      bool isFirstRun,
-      const int64_t prefixLength,
-      std::string_view suffix) {
-    VELOX_CHECK_LE(
-        prefixLength,
-        lastValue_.size(),
-        "prefix length too large in DELTA_BYTE_ARRAY");
+  void skip(uint64_t numValues) {
+    skip<false>(numValues, 0, nullptr);
+  }
 
-    if (prefixLength == 0) {
-      // prefix is empty.
-      lastValue_ = std::string{suffix};
-      return;
+  template <bool hasNulls>
+  FOLLY_ALWAYS_INLINE void
+  skip(int32_t numValues, int32_t current, const uint64_t* nulls) {
+    if (hasNulls) {
+      numValues = bits::countNonNulls(nulls, current, current + numValues);
     }
+    for (int32_t i = 0; i < numValues; ++i) {
+      readString();
+    }
+  }
+
+  template <bool hasNulls, typename Visitor>
+  void readWithVisitor(const uint64_t* nulls, Visitor visitor) {
+    int32_t current = visitor.start();
+    skip<hasNulls>(current, 0, nulls);
+    int32_t toSkip;
+    bool atEnd = false;
+    const bool allowNulls = hasNulls && visitor.allowNulls();
+    for (;;) {
+      if (hasNulls && allowNulls && bits::isBitNull(nulls, current)) {
+        toSkip = visitor.processNull(atEnd);
+      } else {
+        if (hasNulls && !allowNulls) {
+          toSkip = visitor.checkAndSkipNulls(nulls, current, atEnd);
+          if (!Visitor::dense) {
+            skip<false>(toSkip, current, nullptr);
+          }
+          if (atEnd) {
+            return;
+          }
+        }
 
-    if (!isFirstRun) {
-      if (suffix.empty()) {
-        // suffix is empty: read value can simply point to the prefix
-        // of the lastValue_. This is not possible for the first run since
-        // the prefix would point to the mutable `lastValue_`.
-        lastValue_ = lastValue_.substr(0, prefixLength);
+        // We are at a non-null value on a row to visit.
+        toSkip = visitor.process(readString(), atEnd);
+      }
+      ++current;
+      if (toSkip) {
+        skip<hasNulls>(toSkip, current, nulls);
+        current += toSkip;
+      }
+      if (atEnd) {
         return;
       }
     }
-
-    lastValue_.resize(prefixLength + suffix.size());
-
-    // Both prefix and suffix are non-empty, so we need to decode the string
-    // into read value.
-    // Just keep the prefix in lastValue_, and copy the suffix.
-    memcpy(lastValue_.data() + prefixLength, suffix.data(), suffix.size());
   }
 
-  std::unique_ptr<DeltaBpDecoder> prefixLenDecoder_;
-  std::unique_ptr<DeltaBpDecoder> suffixLenDecoder_;
-  std::unique_ptr<DeltaLengthByteArrayDecoder> suffixDecoder_;
+ private:
+  std::optional<DeltaBpDecoder> prefixLenDecoder_;
+  std::optional<DeltaLengthByteArrayDecoder> suffixDecoder_;
 
-  std::string lastValue_;
+  // The string_view returned by readString() aliases this buffer.
+  std::vector<char> lastValueBuf_;
+  int32_t lastValueLen_{0};
   int32_t numValidValues_{0};
   uint32_t prefixLenOffset_{0};
   std::vector<uint32_t> bufferedPrefixLength_;
diff --git a/velox/dwio/parquet/reader/PageReader.cpp b/velox/dwio/parquet/reader/PageReader.cpp
index 8010c477a06..c3ff7e1cc71 100644
--- a/velox/dwio/parquet/reader/PageReader.cpp
+++ b/velox/dwio/parquet/reader/PageReader.cpp
@@ -33,6 +33,10 @@ using facebook::velox::common::testutil::TestValue;
 
 namespace facebook::velox::parquet {
 
+static_assert(
+    PageReader::kPageReadPadding >= DeltaBpDecoder::kRequiredTrailingPadding,
+    "PageReader::kPageReadPadding must cover DeltaBpDecoder's SIMD over-read");
+
 using thrift::Encoding;
 using thrift::PageHeader;
 
@@ -846,15 +850,20 @@ void PageReader::makeDecoder() {
       break;
     case Encoding::DELTA_BYTE_ARRAY:
       if (parquetType == thrift::Type::BYTE_ARRAY) {
-        deltaByteArrDecoder_ =
-            std::make_unique<DeltaByteArrayDecoder>(pageData_);
+        if (!deltaByteArrDecoder_) {
+          deltaByteArrDecoder_ = std::make_unique<DeltaByteArrayDecoder>();
+        }
+        deltaByteArrDecoder_->reset(pageData_);
         break;
       }
       [[fallthrough]];
     case Encoding::DELTA_LENGTH_BYTE_ARRAY:
       if (parquetType == thrift::Type::BYTE_ARRAY) {
-        deltaLengthByteArrDecoder_ =
-            std::make_unique<DeltaLengthByteArrayDecoder>(pageData_);
+        if (!deltaLengthByteArrDecoder_) {
+          deltaLengthByteArrDecoder_ =
+              std::make_unique<DeltaLengthByteArrayDecoder>();
+        }
+        deltaLengthByteArrDecoder_->reset(pageData_);
         break;
       }
       [[fallthrough]];
diff --git a/velox/dwio/parquet/reader/PageReader.h b/velox/dwio/parquet/reader/PageReader.h
index 1a774613486..5df25fa94b2 100644
--- a/velox/dwio/parquet/reader/PageReader.h
+++ b/velox/dwio/parquet/reader/PageReader.h
@@ -36,6 +36,13 @@ namespace facebook::velox::parquet {
 /// continuous stream accessible via readWithVisitor().
 class PageReader {
  public:
+  /// Trailing readable bytes past readBytes()'s returned size. Sized
+  /// for bits::detail::loadBits<uint64_t>, which touches bytes
+  /// [offset, offset + 9) when the bit field straddles the 8-byte word
+  /// boundary. For any value within a miniblock 'offset < size', so the
+  /// furthest byte is at most 'size + 7' — 8 trailing bytes suffice.
+  static constexpr int kPageReadPadding = 8;
+
   PageReader(
       std::unique_ptr<dwio::common::SeekableInputStream> stream,
       memory::MemoryPool& pool,
@@ -222,13 +229,6 @@ class PageReader {
   // 'copy' as needed.
   const char* readBytes(int32_t size, BufferPtr& copy);
 
-  // Trailing readable bytes past readBytes()'s returned size. Sized
-  // for bits::detail::loadBits<uint64_t>, which touches bytes
-  // [offset, offset + 9) when the bit field straddles the 8-byte word
-  // boundary. For any value within a miniblock 'offset < size', so the
-  // furthest byte is at most 'size + 7' — 8 trailing bytes suffice.
-  static constexpr int kPageReadPadding = 8;
-
   // Decompresses data starting at 'pageData_', consuming 'compressedsize' and
   // producing up to 'uncompressedSize' bytes. The start of the decoding
   // result is returned. an intermediate copy may be made in 'decompresseddata_'
diff --git a/velox/dwio/parquet/tests/reader/ParquetTableScanTest.cpp b/velox/dwio/parquet/tests/reader/ParquetTableScanTest.cpp
index e50311ccce2..6351d48b9e9 100644
--- a/velox/dwio/parquet/tests/reader/ParquetTableScanTest.cpp
+++ b/velox/dwio/parquet/tests/reader/ParquetTableScanTest.cpp
@@ -1620,6 +1620,135 @@ TEST_F(ParquetTableScanTest, deltaBinaryPackedNarrowBitWidth) {
   run.template operator()<int64_t>();
 }
 
+TEST_F(ParquetTableScanTest, deltaBinaryPackedBitWidth32) {
+  WriterOptions options;
+  options.enableDictionary = false;
+  options.encoding =
+      facebook::velox::parquet::arrow::Encoding::kDeltaBinaryPacked;
+
+  constexpr vector_size_t kSize = 1024;
+  constexpr int64_t kStep = (1LL << 32) - 1;
+  auto vector = makeRowVector(
+      {"c"}, {makeFlatVector<int64_t>(kSize, [](auto row) {
+        return static_cast<int64_t>((row / 2) * kStep + (row % 2 ? kStep : 0));
+      })});
+  auto file = TempFilePath::create();
+  writeToParquetFile(file->getPath(), {vector}, options);
+  loadData(vector->rowType(), vector);
+
+  assertSelect({makeSplit(file->getPath())}, {"c"}, "SELECT c FROM tmp");
+}
+
+TEST_F(ParquetTableScanTest, deltaBinaryPackedBitWidthSweep) {
+  auto run = [&]<typename T>(int maxBitWidth) {
+    SCOPED_TRACE(fmt::format("T=int{}", sizeof(T) * 8));
+    constexpr vector_size_t kSize = 256;
+    std::vector<std::string> names;
+    std::vector<VectorPtr> children;
+    for (int bw = 1; bw <= maxBitWidth; ++bw) {
+      const int64_t step = (bw == 32) ? 0xFFFFFFFFLL : ((1LL << bw) - 1);
+      children.push_back(makeFlatVector<T>(kSize, [step](auto row) -> T {
+        return static_cast<T>((row / 2) * step + ((row % 2) ? step : 0));
+      }));
+      names.push_back(fmt::format("c{}", bw));
+    }
+    auto vector = makeRowVector(names, children);
+
+    WriterOptions options;
+    options.enableDictionary = false;
+    options.encoding =
+        facebook::velox::parquet::arrow::Encoding::kDeltaBinaryPacked;
+    auto file = TempFilePath::create();
+    writeToParquetFile(file->getPath(), {vector}, options);
+    loadData(vector->rowType(), vector);
+
+    assertSelect(
+        {makeSplit(file->getPath())}, std::move(names), "SELECT * FROM tmp");
+  };
+
+  run.template operator()<int32_t>(31);
+  run.template operator()<int64_t>(32);
+}
+
+TEST_F(ParquetTableScanTest, deltaBinaryPackedBitWidth33) {
+  WriterOptions options;
+  options.enableDictionary = false;
+  options.encoding =
+      facebook::velox::parquet::arrow::Encoding::kDeltaBinaryPacked;
+
+  constexpr vector_size_t kSize = 1024;
+  constexpr int64_t kStep = 1LL << 32;
+  auto vector = makeRowVector(
+      {"c"}, {makeFlatVector<int64_t>(kSize, [](auto row) {
+        return static_cast<int64_t>((row / 2) * kStep + (row % 2 ? kStep : 0));
+      })});
+  auto file = TempFilePath::create();
+  writeToParquetFile(file->getPath(), {vector}, options);
+  loadData(vector->rowType(), vector);
+
+  assertSelect({makeSplit(file->getPath())}, {"c"}, "SELECT c FROM tmp");
+}
+
+TEST_F(ParquetTableScanTest, deltaBinaryPackedMixedMiniblockWidths) {
+  WriterOptions options;
+  options.enableDictionary = false;
+  options.encoding =
+      facebook::velox::parquet::arrow::Encoding::kDeltaBinaryPacked;
+
+  constexpr vector_size_t kSize = 128;
+  auto vector =
+      makeRowVector({"c"}, {makeFlatVector<int64_t>(kSize, [](auto row) {
+                      auto deltaForRow = [](vector_size_t r) -> int64_t {
+                        const int mb = (r / 32) % 4;
+                        const int parity = r % 2;
+                        if (mb == 0)
+                          return parity;
+                        if (mb == 1)
+                          return parity ? 200LL : 0LL;
+                        if (mb == 2)
+                          return parity ? 50'000LL : 0LL;
+                        return parity ? 12'000'000LL : 0LL;
+                      };
+                      int64_t v = 0;
+                      for (vector_size_t i = 0; i < row; ++i) {
+                        v += deltaForRow(i);
+                      }
+                      return v;
+                    })});
+  auto file = TempFilePath::create();
+  writeToParquetFile(file->getPath(), {vector}, options);
+  loadData(vector->rowType(), vector);
+
+  assertSelect({makeSplit(file->getPath())}, {"c"}, "SELECT c FROM tmp");
+}
+
+TEST_F(ParquetTableScanTest, deltaBinaryPackedFilterScalarTail) {
+  WriterOptions options;
+  options.enableDictionary = false;
+  options.encoding =
+      facebook::velox::parquet::arrow::Encoding::kDeltaBinaryPacked;
+  options.dataPageSize = 8 * 1024;
+  options.batchSize = 1024;
+
+  constexpr vector_size_t kSize = 128 * 1024;
+  auto vector = makeRowVector(
+      {"a", "b"},
+      {makeFlatVector<int64_t>(
+           kSize, [](auto row) { return static_cast<int64_t>(row); }),
+       makeFlatVector<int64_t>(
+           kSize, [](auto row) { return 100'000LL + row * 31LL; })});
+  auto file = TempFilePath::create();
+  writeToParquetFile(file->getPath(), {vector}, options);
+  loadData(vector->rowType(), vector);
+
+  assertSelectWithFilter(
+      {makeSplit(file->getPath())},
+      {"a", "b"},
+      {"a BETWEEN 10000 AND 70000", "b > 200000"},
+      "",
+      "SELECT a, b FROM tmp WHERE a BETWEEN 10000 AND 70000 AND b > 200000");
+}
+
 TEST_F(ParquetTableScanTest, deltaBinaryPackedWideAndNegative) {
   WriterOptions options;
   options.enableDictionary = false;