diff --git a/velox/connectors/hive/HiveConnectorUtil.cpp b/velox/connectors/hive/HiveConnectorUtil.cpp
index 09099c37d188..d8cfe275fd71 100644
--- a/velox/connectors/hive/HiveConnectorUtil.cpp
+++ b/velox/connectors/hive/HiveConnectorUtil.cpp
@@ -568,7 +568,10 @@ void configureRowReaderOptions(
   } else {
     cs = std::make_shared<dwio::common::ColumnSelector>(rowType, columnNames);
   }
-  rowReaderOptions.select(cs).range(hiveSplit->start, hiveSplit->length);
+  rowReaderOptions.select(cs);
+  if (hiveSplit) {
+    rowReaderOptions.range(hiveSplit->start, hiveSplit->length);
+  }
 }
 
 namespace {
diff --git a/velox/dwio/common/ColumnVisitors.h b/velox/dwio/common/ColumnVisitors.h
index 3ac3d5e219f7..cc81f4505fee 100644
--- a/velox/dwio/common/ColumnVisitors.h
+++ b/velox/dwio/common/ColumnVisitors.h
@@ -49,29 +49,32 @@ struct DropValues {
   }
 };
 
-template <typename TReader>
 struct ExtractToReader {
   using HookType = dwio::common::NoHook;
   static constexpr bool kSkipNulls = false;
-  explicit ExtractToReader(TReader* readerIn) : reader(readerIn) {}
+  explicit ExtractToReader(SelectiveColumnReader* readerIn)
+      : reader_(readerIn) {}
 
   bool acceptsNulls() const {
     return true;
   }
 
   template <typename T>
-  void addNull(vector_size_t rowIndex);
+  void addNull(vector_size_t /*rowIndex*/) {
+    reader_->template addNull<T>();
+  }
 
   template <typename V>
   void addValue(vector_size_t /*rowIndex*/, V value) {
-    reader->addValue(value);
+    reader_->addValue(value);
   }
 
-  TReader* reader;
-
   dwio::common::NoHook& hook() {
     return noHook();
   }
+
+ private:
+  SelectiveColumnReader* reader_;
 };
 
 template <typename THook>
@@ -150,6 +153,7 @@ class ColumnVisitor {
   using DataType = T;
   static constexpr bool dense = isDense;
   static constexpr bool kHasBulkPath = true;
+
   ColumnVisitor(
       TFilter& filter,
       SelectiveColumnReader* reader,
@@ -163,6 +167,20 @@ class ColumnVisitor {
         rowIndex_(0),
         values_(values) {}
 
+  template <bool isDense2 = isDense, std::enable_if_t<isDense2, int> = 0>
+  ColumnVisitor(
+      TFilter& filter,
+      SelectiveColumnReader* reader,
+      vector_size_t numRows,
+      ExtractValues values)
+      : filter_(filter),
+        reader_(reader),
+        allowNulls_(!TFilter::deterministic || filter.testNull()),
+        rows_(nullptr),
+        numRows_(numRows),
+        rowIndex_(0),
+        values_(values) {}
+
   bool allowNulls() {
     if (ExtractValues::kSkipNulls && TFilter::deterministic) {
       return false;
@@ -269,7 +287,7 @@ class ColumnVisitor {
     }
     if (++rowIndex_ >= numRows_) {
       atEnd = true;
-      return rows_[numRows_ - 1] - previous;
+      return rowAt(numRows_ - 1) - previous;
     }
     if (TFilter::deterministic && isDense) {
       return 0;
@@ -301,7 +319,7 @@ class ColumnVisitor {
     if (isDense) {
       return 0;
     }
-    return currentRow() - rows_[rowIndex_ - 1] - 1;
+    return currentRow() - rowAt(rowIndex_ - 1) - 1;
   }
 
   FOLLY_ALWAYS_INLINE vector_size_t process(T value, bool& atEnd) {
@@ -314,7 +332,7 @@ class ColumnVisitor {
       }
       if (++rowIndex_ >= numRows_) {
         atEnd = true;
-        return rows_[numRows_ - 1] - previous;
+        return rowAt(numRows_ - 1) - previous;
       }
       return currentRow() - previous - 1;
     }
@@ -331,7 +349,7 @@ class ColumnVisitor {
     if (isDense) {
       return 0;
     }
-    return currentRow() - rows_[rowIndex_ - 1] - 1;
+    return currentRow() - rowAt(rowIndex_ - 1) - 1;
   }
 
   // Returns space for 'size' items of T for a scan to fill. The scan
@@ -341,26 +359,30 @@ class ColumnVisitor {
     return reader_->mutableValues<T>(size);
   }
 
-  int32_t numRows() const {
-    return reader_->numRows();
-  }
-
   SelectiveColumnReader& reader() const {
     return *reader_;
   }
 
-  inline vector_size_t rowAt(vector_size_t index) {
+  inline vector_size_t rowAt(vector_size_t index) const {
     if (isDense) {
       return index;
     }
     return rows_[index];
   }
 
-  bool atEnd() {
+  vector_size_t rowIndex() const {
+    return rowIndex_;
+  }
+
+  void setRowIndex(vector_size_t index) {
+    rowIndex_ = index;
+  }
+
+  bool atEnd() const {
     return rowIndex_ >= numRows_;
   }
 
-  vector_size_t currentRow() {
+  vector_size_t currentRow() const {
     if (isDense) {
       return rowIndex_;
     }
@@ -371,7 +393,7 @@ class ColumnVisitor {
     return rows_;
   }
 
-  vector_size_t numRows() {
+  vector_size_t numRows() const {
     return numRows_;
   }
 
@@ -504,12 +526,6 @@ inline void ColumnVisitor<T, TFilter, ExtractValues, isDense>::addOutputRow(
   reader_->addOutputRow(row);
 }
 
-template <typename TReader>
-template <typename T>
-void ExtractToReader<TReader>::addNull(vector_size_t /*rowIndex*/) {
-  reader->template addNull<T>();
-}
-
 enum FilterResult { kUnknown = 0x40, kSuccess = 0x80, kFailure = 0 };
 
 namespace detail {
@@ -1390,13 +1406,6 @@ class DirectRleColumnVisitor
             rows,
             values) {}
 
-  // Use for replacing all rows with non-null rows for fast path with
-  // processRun and processRle.
-  void setRows(folly::Range<const int32_t*> newRows) {
-    super::rows_ = newRows.data();
-    super::numRows_ = newRows.size();
-  }
-
   // Processes 'numInput' T's in 'input'. Sets 'values' and
   // 'numValues'' to the resulting values. 'scatterRows' may be
   // non-null if there is no filter and the decoded values should be
@@ -1479,4 +1488,113 @@ class DirectRleColumnVisitor
   }
 };
 
+template <bool kEncodingHasNulls>
+class StringColumnReadWithVisitorHelper {
+ public:
+  StringColumnReadWithVisitorHelper(SelectiveColumnReader& reader, RowSet rows)
+      : reader_(reader), rows_(rows) {}
+
+  template <typename F>
+  auto operator()(F&& readWithVisitor) {
+    const bool isDense = rows_.back() == rows_.size() - 1;
+    if (reader_.scanSpec()->keepValues()) {
+      if (auto* hook = reader_.scanSpec()->valueHook()) {
+        if (isDense) {
+          readHelper<velox::common::AlwaysTrue, true>(
+              &alwaysTrue(),
+              ExtractToGenericHook(hook),
+              std::forward<F>(readWithVisitor));
+        } else {
+          readHelper<velox::common::AlwaysTrue, false>(
+              &alwaysTrue(),
+              ExtractToGenericHook(hook),
+              std::forward<F>(readWithVisitor));
+        }
+      } else {
+        if (isDense) {
+          processFilter<true>(
+              ExtractToReader(&reader_), std::forward<F>(readWithVisitor));
+        } else {
+          processFilter<false>(
+              ExtractToReader(&reader_), std::forward<F>(readWithVisitor));
+        }
+      }
+    } else {
+      if (isDense) {
+        processFilter<true>(DropValues(), std::forward<F>(readWithVisitor));
+      } else {
+        processFilter<false>(DropValues(), std::forward<F>(readWithVisitor));
+      }
+    }
+  }
+
+ private:
+  template <typename TFilter, bool kIsDense, typename ExtractValues, typename F>
+  void readHelper(
+      velox::common::Filter* filter,
+      ExtractValues extractValues,
+      F readWithVisitor) {
+    readWithVisitor(
+        ColumnVisitor<folly::StringPiece, TFilter, ExtractValues, kIsDense>(
+            *static_cast<TFilter*>(filter), &reader_, rows_, extractValues));
+  }
+
+  template <bool kIsDense, typename ExtractValues, typename F>
+  void processFilter(ExtractValues extractValues, F&& readWithVisitor) {
+    auto* filter = reader_.scanSpec()->filter();
+    if (filter == nullptr) {
+      readHelper<velox::common::AlwaysTrue, kIsDense>(
+          &alwaysTrue(), extractValues, std::forward<F>(readWithVisitor));
+      return;
+    }
+    switch (filter->kind()) {
+      case velox::common::FilterKind::kAlwaysTrue:
+        readHelper<velox::common::AlwaysTrue, kIsDense>(
+            filter, extractValues, std::forward<F>(readWithVisitor));
+        break;
+      case velox::common::FilterKind::kIsNull:
+        if constexpr (kEncodingHasNulls) {
+          reader_.filterNulls<StringView>(
+              rows_, true, !std::is_same_v<ExtractValues, DropValues>);
+        } else {
+          readHelper<velox::common::IsNull, kIsDense>(
+              filter, extractValues, std::forward<F>(readWithVisitor));
+        }
+        break;
+      case velox::common::FilterKind::kIsNotNull:
+        if constexpr (
+            kEncodingHasNulls && std::is_same_v<ExtractValues, DropValues>) {
+          reader_.filterNulls<StringView>(rows_, false, false);
+        } else {
+          readHelper<velox::common::IsNotNull, kIsDense>(
+              filter, extractValues, std::forward<F>(readWithVisitor));
+        }
+        break;
+      case velox::common::FilterKind::kBytesRange:
+        readHelper<velox::common::BytesRange, kIsDense>(
+            filter, extractValues, std::forward<F>(readWithVisitor));
+        break;
+      case velox::common::FilterKind::kNegatedBytesRange:
+        readHelper<velox::common::NegatedBytesRange, kIsDense>(
+            filter, extractValues, std::forward<F>(readWithVisitor));
+        break;
+      case velox::common::FilterKind::kBytesValues:
+        readHelper<velox::common::BytesValues, kIsDense>(
+            filter, extractValues, std::forward<F>(readWithVisitor));
+        break;
+      case velox::common::FilterKind::kNegatedBytesValues:
+        readHelper<velox::common::NegatedBytesValues, kIsDense>(
+            filter, extractValues, std::forward<F>(readWithVisitor));
+        break;
+      default:
+        readHelper<velox::common::Filter, kIsDense>(
+            filter, extractValues, std::forward<F>(readWithVisitor));
+        break;
+    }
+  }
+
+  SelectiveColumnReader& reader_;
+  const RowSet rows_;
+};
+
 } // namespace facebook::velox::dwio::common
diff --git a/velox/dwio/common/FormatData.h b/velox/dwio/common/FormatData.h
index 034860460646..1f0b5d4426bb 100644
--- a/velox/dwio/common/FormatData.h
+++ b/velox/dwio/common/FormatData.h
@@ -34,7 +34,7 @@ class FormatData {
 
   template <typename T>
   T& as() {
-    return *reinterpret_cast<T*>(this);
+    return *static_cast<T*>(this);
   }
 
   /// Reads nulls if the format has nulls separate from the encoded
diff --git a/velox/dwio/common/SelectiveByteRleColumnReader.h b/velox/dwio/common/SelectiveByteRleColumnReader.h
index 06aae1c4986b..67537ea8d8b8 100644
--- a/velox/dwio/common/SelectiveByteRleColumnReader.h
+++ b/velox/dwio/common/SelectiveByteRleColumnReader.h
@@ -39,7 +39,11 @@ class SelectiveByteRleColumnReader : public SelectiveColumnReader {
 
   void getValues(RowSet rows, VectorPtr* result) override;
 
-  template <typename Reader, bool isDense, typename ExtractValues>
+  template <
+      typename Reader,
+      bool isDense,
+      bool kEncodingHasNulls,
+      typename ExtractValues>
   void processFilter(
       velox::common::Filter* filter,
       ExtractValues extractValues,
@@ -58,7 +62,7 @@ class SelectiveByteRleColumnReader : public SelectiveColumnReader {
       RowSet rows,
       ExtractValues extractValues);
 
-  template <typename Reader>
+  template <typename Reader, bool kEncodingHasNulls>
   void
   readCommon(vector_size_t offset, RowSet rows, const uint64_t* incomingNulls);
 };
@@ -78,7 +82,11 @@ void SelectiveByteRleColumnReader::readHelper(
           *reinterpret_cast<TFilter*>(filter), this, rows, extractValues));
 }
 
-template <typename Reader, bool isDense, typename ExtractValues>
+template <
+    typename Reader,
+    bool isDense,
+    bool kEncodingHasNulls,
+    typename ExtractValues>
 void SelectiveByteRleColumnReader::processFilter(
     velox::common::Filter* filter,
     ExtractValues extractValues,
@@ -90,13 +98,20 @@ void SelectiveByteRleColumnReader::processFilter(
           filter, rows, extractValues);
       break;
     case FilterKind::kIsNull:
-      filterNulls<int8_t>(
-          rows,
-          true,
-          !std::is_same_v<decltype(extractValues), dwio::common::DropValues>);
+      if constexpr (kEncodingHasNulls) {
+        filterNulls<int8_t>(
+            rows,
+            true,
+            !std::is_same_v<decltype(extractValues), dwio::common::DropValues>);
+      } else {
+        readHelper<Reader, velox::common::IsNull, isDense>(
+            filter, rows, extractValues);
+      }
       break;
     case FilterKind::kIsNotNull:
-      if (std::is_same_v<decltype(extractValues), dwio::common::DropValues>) {
+      if constexpr (
+          kEncodingHasNulls &&
+          std::is_same_v<decltype(extractValues), dwio::common::DropValues>) {
         filterNulls<int8_t>(rows, false, false);
       } else {
         readHelper<Reader, velox::common::IsNotNull, isDense>(
@@ -148,7 +163,7 @@ void SelectiveByteRleColumnReader::processValueHook(
   }
 }
 
-template <typename Reader>
+template <typename Reader, bool kEncodingHasNulls>
 void SelectiveByteRleColumnReader::readCommon(
     vector_size_t offset,
     RowSet rows,
@@ -167,17 +182,19 @@ void SelectiveByteRleColumnReader::readCommon(
       return;
     }
     if (isDense) {
-      processFilter<Reader, true>(
+      processFilter<Reader, true, kEncodingHasNulls>(
           filter, dwio::common::ExtractToReader(this), rows);
     } else {
-      processFilter<Reader, false>(
+      processFilter<Reader, false, kEncodingHasNulls>(
           filter, dwio::common::ExtractToReader(this), rows);
     }
   } else {
     if (isDense) {
-      processFilter<Reader, true>(filter, dwio::common::DropValues(), rows);
+      processFilter<Reader, true, kEncodingHasNulls>(
+          filter, dwio::common::DropValues(), rows);
     } else {
-      processFilter<Reader, false>(filter, dwio::common::DropValues(), rows);
+      processFilter<Reader, false, kEncodingHasNulls>(
+          filter, dwio::common::DropValues(), rows);
     }
   }
 }
diff --git a/velox/dwio/common/SelectiveColumnReader.h b/velox/dwio/common/SelectiveColumnReader.h
index 4b97f4c6f465..1fdcbc2e2a76 100644
--- a/velox/dwio/common/SelectiveColumnReader.h
+++ b/velox/dwio/common/SelectiveColumnReader.h
@@ -294,9 +294,8 @@ class SelectiveColumnReader {
   template <typename T>
   inline void addNull() {
     VELOX_DCHECK_NE(valueSize_, kNoValueSize);
-    VELOX_DCHECK_LE(
-        rawResultNulls_ && rawValues_ && (numValues_ + 1) * valueSize_,
-        values_->capacity());
+    VELOX_DCHECK(rawResultNulls_ && rawValues_);
+    VELOX_DCHECK_LE((numValues_ + 1) * valueSize_, values_->capacity());
 
     anyNulls_ = true;
     bits::setNull(rawResultNulls_, numValues_);
@@ -441,12 +440,12 @@ class SelectiveColumnReader {
     isFlatMapValue_ = value;
   }
 
- protected:
   // Filters 'rows' according to 'is_null'. Only applies to cases where
   // scanSpec_->readsNullsOnly() is true.
   template <typename T>
   void filterNulls(RowSet rows, bool isNull, bool extractValues);
 
+ protected:
   template <typename T>
   void
   prepareRead(vector_size_t offset, RowSet rows, const uint64_t* incomingNulls);
@@ -670,6 +669,8 @@ inline void SelectiveColumnReader::addValue(const folly::StringPiece value) {
   addStringValue(value);
 }
 
+velox::common::AlwaysTrue& alwaysTrue();
+
 } // namespace facebook::velox::dwio::common
 
 namespace facebook::velox::dwio::common {
diff --git a/velox/dwio/common/SelectiveColumnReaderInternal.h b/velox/dwio/common/SelectiveColumnReaderInternal.h
index 61bcf2b5befa..ed38c6551fd5 100644
--- a/velox/dwio/common/SelectiveColumnReaderInternal.h
+++ b/velox/dwio/common/SelectiveColumnReaderInternal.h
@@ -31,8 +31,6 @@
 
 namespace facebook::velox::dwio::common {
 
-velox::common::AlwaysTrue& alwaysTrue();
-
 class Timer {
  public:
   Timer() : startClocks_{folly::hardware_timestamp()} {}
diff --git a/velox/dwio/common/SelectiveFloatingPointColumnReader.h b/velox/dwio/common/SelectiveFloatingPointColumnReader.h
index 61ccd7e4d8b6..ea2455afa0c8 100644
--- a/velox/dwio/common/SelectiveFloatingPointColumnReader.h
+++ b/velox/dwio/common/SelectiveFloatingPointColumnReader.h
@@ -40,7 +40,7 @@ class SelectiveFloatingPointColumnReader : public SelectiveColumnReader {
     return std::is_same_v<TData, TRequested>;
   }
 
-  template <typename Reader>
+  template <typename Reader, bool kEncodingHasNulls>
   void
   readCommon(vector_size_t offset, RowSet rows, const uint64_t* incomingNulls);
 
@@ -57,7 +57,11 @@ class SelectiveFloatingPointColumnReader : public SelectiveColumnReader {
   void
   readHelper(velox::common::Filter* filter, RowSet rows, ExtractValues values);
 
-  template <typename Reader, bool isDense, typename ExtractValues>
+  template <
+      typename Reader,
+      bool isDense,
+      bool kEncodingHasNulls,
+      typename ExtractValues>
   void processFilter(
       velox::common::Filter* filter,
       RowSet rows,
@@ -84,7 +88,11 @@ void SelectiveFloatingPointColumnReader<TData, TRequested>::readHelper(
 }
 
 template <typename TData, typename TRequested>
-template <typename Reader, bool isDense, typename ExtractValues>
+template <
+    typename Reader,
+    bool isDense,
+    bool kEncodingHasNulls,
+    typename ExtractValues>
 void SelectiveFloatingPointColumnReader<TData, TRequested>::processFilter(
     velox::common::Filter* filter,
     RowSet rows,
@@ -101,11 +109,18 @@ void SelectiveFloatingPointColumnReader<TData, TRequested>::processFilter(
           filter, rows, extractValues);
       break;
     case velox::common::FilterKind::kIsNull:
-      filterNulls<TRequested>(
-          rows, true, !std::is_same_v<decltype(extractValues), DropValues>);
+      if constexpr (kEncodingHasNulls) {
+        filterNulls<TRequested>(
+            rows, true, !std::is_same_v<decltype(extractValues), DropValues>);
+      } else {
+        readHelper<Reader, velox::common::IsNull, isDense>(
+            filter, rows, extractValues);
+      }
       break;
     case velox::common::FilterKind::kIsNotNull:
-      if (std::is_same_v<decltype(extractValues), DropValues>) {
+      if constexpr (
+          kEncodingHasNulls &&
+          std::is_same_v<decltype(extractValues), DropValues>) {
         filterNulls<TRequested>(rows, false, false);
       } else {
         readHelper<Reader, velox::common::IsNotNull, isDense>(
@@ -163,7 +178,7 @@ void SelectiveFloatingPointColumnReader<TData, TRequested>::processValueHook(
 }
 
 template <typename TData, typename TRequested>
-template <typename Reader>
+template <typename Reader, bool kEncodingHasNulls>
 void SelectiveFloatingPointColumnReader<TData, TRequested>::readCommon(
     vector_size_t offset,
     RowSet rows,
@@ -179,18 +194,20 @@ void SelectiveFloatingPointColumnReader<TData, TRequested>::readCommon(
       }
     } else {
       if (isDense) {
-        processFilter<Reader, true>(
+        processFilter<Reader, true, kEncodingHasNulls>(
             scanSpec_->filter(), rows, ExtractToReader(this));
       } else {
-        processFilter<Reader, false>(
+        processFilter<Reader, false, kEncodingHasNulls>(
             scanSpec_->filter(), rows, ExtractToReader(this));
       }
     }
   } else {
     if (isDense) {
-      processFilter<Reader, true>(scanSpec_->filter(), rows, DropValues());
+      processFilter<Reader, true, kEncodingHasNulls>(
+          scanSpec_->filter(), rows, DropValues());
     } else {
-      processFilter<Reader, false>(scanSpec_->filter(), rows, DropValues());
+      processFilter<Reader, false, kEncodingHasNulls>(
+          scanSpec_->filter(), rows, DropValues());
     }
   }
 }
diff --git a/velox/dwio/common/SelectiveIntegerColumnReader.h b/velox/dwio/common/SelectiveIntegerColumnReader.h
index ba4b63e168ca..444f341e06b9 100644
--- a/velox/dwio/common/SelectiveIntegerColumnReader.h
+++ b/velox/dwio/common/SelectiveIntegerColumnReader.h
@@ -41,7 +41,11 @@ class SelectiveIntegerColumnReader : public SelectiveColumnReader {
 
  protected:
   // Switches based on filter type between different readHelper instantiations.
-  template <typename Reader, bool isDense, typename ExtractValues>
+  template <
+      typename Reader,
+      bool isDense,
+      bool kEncodingHasNulls,
+      typename ExtractValues>
   void processFilter(
       velox::common::Filter* filter,
       ExtractValues extractValues,
@@ -66,7 +70,7 @@ class SelectiveIntegerColumnReader : public SelectiveColumnReader {
   // The common part of integer reading. calls the appropriate
   // instantiation of processValueHook or processFilter based on
   // possible value hook, filter and denseness.
-  template <typename Reader>
+  template <typename Reader, bool kEncodingHasNulls>
   void readCommon(RowSet rows);
 };
 
@@ -113,7 +117,11 @@ void SelectiveIntegerColumnReader::readHelper(
   }
 }
 
-template <typename Reader, bool isDense, typename ExtractValues>
+template <
+    typename Reader,
+    bool isDense,
+    bool kEncodingHasNulls,
+    typename ExtractValues>
 void SelectiveIntegerColumnReader::processFilter(
     velox::common::Filter* filter,
     ExtractValues extractValues,
@@ -130,11 +138,18 @@ void SelectiveIntegerColumnReader::processFilter(
           filter, rows, extractValues);
       break;
     case velox::common::FilterKind::kIsNull:
-      filterNulls<int64_t>(
-          rows, true, !std::is_same_v<decltype(extractValues), DropValues>);
+      if constexpr (kEncodingHasNulls) {
+        filterNulls<int64_t>(
+            rows, true, !std::is_same_v<decltype(extractValues), DropValues>);
+      } else {
+        readHelper<Reader, velox::common::IsNull, isDense>(
+            filter, rows, extractValues);
+      }
       break;
     case velox::common::FilterKind::kIsNotNull:
-      if (std::is_same_v<decltype(extractValues), DropValues>) {
+      if constexpr (
+          kEncodingHasNulls &&
+          std::is_same_v<decltype(extractValues), DropValues>) {
         filterNulls<int64_t>(rows, false, false);
       } else {
         readHelper<Reader, velox::common::IsNotNull, isDense>(
@@ -211,7 +226,7 @@ void SelectiveIntegerColumnReader::processValueHook(
   }
 }
 
-template <typename Reader>
+template <typename Reader, bool kEncodingHasNulls>
 void SelectiveIntegerColumnReader::readCommon(RowSet rows) {
   bool isDense = rows.back() == rows.size() - 1;
   velox::common::Filter* filter =
@@ -225,16 +240,20 @@ void SelectiveIntegerColumnReader::readCommon(RowSet rows) {
       }
     } else {
       if (isDense) {
-        processFilter<Reader, true>(filter, ExtractToReader(this), rows);
+        processFilter<Reader, true, kEncodingHasNulls>(
+            filter, ExtractToReader(this), rows);
       } else {
-        processFilter<Reader, false>(filter, ExtractToReader(this), rows);
+        processFilter<Reader, false, kEncodingHasNulls>(
+            filter, ExtractToReader(this), rows);
       }
     }
   } else {
     if (isDense) {
-      processFilter<Reader, true>(filter, DropValues(), rows);
+      processFilter<Reader, true, kEncodingHasNulls>(
+          filter, DropValues(), rows);
     } else {
-      processFilter<Reader, false>(filter, DropValues(), rows);
+      processFilter<Reader, false, kEncodingHasNulls>(
+          filter, DropValues(), rows);
     }
   }
 }
diff --git a/velox/dwio/common/TypeWithId.cpp b/velox/dwio/common/TypeWithId.cpp
index 6b803a6249da..c4c9b4f8d36f 100644
--- a/velox/dwio/common/TypeWithId.cpp
+++ b/velox/dwio/common/TypeWithId.cpp
@@ -86,4 +86,37 @@ std::unique_ptr<TypeWithId> TypeWithId::create(
       type, std::move(children), myId, maxId, column);
 }
 
+std::string TypeWithId::fullName() const {
+  std::vector<std::string> path;
+  auto* child = this;
+  while (child->parent_) {
+    switch (child->parent_->type()->kind()) {
+      case TypeKind::ROW:
+        VELOX_CHECK(
+            child == child->parent_->children_.at(child->column_).get());
+        path.push_back(
+            '.' + child->parent_->type()->asRow().nameOf(child->column_));
+        break;
+      case TypeKind::ARRAY:
+        break;
+      case TypeKind::MAP:
+        if (child == child->parent_->children_.at(0).get()) {
+          path.push_back(".<keys>");
+        } else {
+          VELOX_CHECK(child == child->children_.at(1).get());
+          path.push_back(".<values>");
+        }
+        break;
+      default:
+        VELOX_UNREACHABLE();
+    }
+    child = parent_;
+  }
+  std::string ans = "<root>";
+  for (int i = path.size() - 1; i >= 0; --i) {
+    ans += path[i];
+  }
+  return ans;
+}
+
 } // namespace facebook::velox::dwio::common
diff --git a/velox/dwio/common/TypeWithId.h b/velox/dwio/common/TypeWithId.h
index 39a988f9936a..5c5fbc5d070c 100644
--- a/velox/dwio/common/TypeWithId.h
+++ b/velox/dwio/common/TypeWithId.h
@@ -73,6 +73,8 @@ class TypeWithId : public velox::Tree<std::shared_ptr<const TypeWithId>> {
     return children_;
   }
 
+  std::string fullName() const;
+
  private:
   static std::unique_ptr<TypeWithId> create(
       const std::shared_ptr<const velox::Type>& type,
diff --git a/velox/dwio/dwrf/reader/DwrfReader.cpp b/velox/dwio/dwrf/reader/DwrfReader.cpp
index 483bde0be4a0..5ca3c2b041df 100644
--- a/velox/dwio/dwrf/reader/DwrfReader.cpp
+++ b/velox/dwio/dwrf/reader/DwrfReader.cpp
@@ -316,6 +316,7 @@ uint64_t DwrfRowReader::seekToRow(uint64_t rowNumber) {
   if (isEmptyFile()) {
     return 0;
   }
+  nextRowNumber_.reset();
 
   // If we are reading only a portion of the file
   // (bounded by firstStripe_ and stripeCeiling_),
@@ -586,6 +587,9 @@ void DwrfRowReader::readWithRowNumber(
 }
 
 int64_t DwrfRowReader::nextRowNumber() {
+  if (nextRowNumber_.has_value()) {
+    return *nextRowNumber_;
+  }
   auto strideSize = getReader().getFooter().rowIndexStride();
   while (currentStripe_ < stripeCeiling_) {
     if (currentRowInStripe_ == 0) {
@@ -604,20 +608,21 @@ int64_t DwrfRowReader::nextRowNumber() {
     }
     checkSkipStrides(strideSize);
     if (currentRowInStripe_ < rowsInCurrentStripe_) {
-      return firstRowOfStripe_[currentStripe_] + currentRowInStripe_;
+      nextRowNumber_ = firstRowOfStripe_[currentStripe_] + currentRowInStripe_;
+      return *nextRowNumber_;
     }
   advanceToNextStripe:
     ++currentStripe_;
     currentRowInStripe_ = 0;
     currentUnit_ = nullptr;
   }
-  atEnd_ = true;
+  nextRowNumber_ = kAtEnd;
   return kAtEnd;
 }
 
 int64_t DwrfRowReader::nextReadSize(uint64_t size) {
   VELOX_DCHECK_GT(size, 0);
-  if (atEnd_) {
+  if (nextRowNumber() == kAtEnd) {
     return kAtEnd;
   }
   auto rowsToRead = std::min(size, rowsInCurrentStripe_ - currentRowInStripe_);
@@ -646,6 +651,7 @@ uint64_t DwrfRowReader::next(
     return 0;
   }
   auto rowsToRead = nextReadSize(size);
+  nextRowNumber_.reset();
   previousRow_ = nextRow;
   // Record strideIndex for use by the columnReader_ which may delay actual
   // reading of the data.
diff --git a/velox/dwio/dwrf/reader/DwrfReader.h b/velox/dwio/dwrf/reader/DwrfReader.h
index acd496e414de..549746776fa1 100644
--- a/velox/dwio/dwrf/reader/DwrfReader.h
+++ b/velox/dwio/dwrf/reader/DwrfReader.h
@@ -160,7 +160,7 @@ class DwrfRowReader : public StrideIndexProvider,
 
   dwio::common::ColumnReaderStatistics columnReaderStatistics_;
 
-  bool atEnd_{false};
+  std::optional<int64_t> nextRowNumber_;
 
   std::unique_ptr<dwio::common::UnitLoader> unitLoader_;
   DwrfUnit* currentUnit_;
diff --git a/velox/dwio/dwrf/reader/SelectiveByteRleColumnReader.h b/velox/dwio/dwrf/reader/SelectiveByteRleColumnReader.h
index 363082eccd64..8684842c6fa2 100644
--- a/velox/dwio/dwrf/reader/SelectiveByteRleColumnReader.h
+++ b/velox/dwio/dwrf/reader/SelectiveByteRleColumnReader.h
@@ -79,7 +79,7 @@ class SelectiveByteRleColumnReader
 
   void read(vector_size_t offset, RowSet rows, const uint64_t* incomingNulls)
       override {
-    readCommon<SelectiveByteRleColumnReader>(offset, rows, incomingNulls);
+    readCommon<SelectiveByteRleColumnReader, true>(offset, rows, incomingNulls);
     readOffset_ += rows.back() + 1;
   }
 
diff --git a/velox/dwio/dwrf/reader/SelectiveFloatingPointColumnReader.h b/velox/dwio/dwrf/reader/SelectiveFloatingPointColumnReader.h
index 5498ea77e8c5..6cce3ad3ec75 100644
--- a/velox/dwio/dwrf/reader/SelectiveFloatingPointColumnReader.h
+++ b/velox/dwio/dwrf/reader/SelectiveFloatingPointColumnReader.h
@@ -49,7 +49,7 @@ class SelectiveFloatingPointColumnReader
   void read(vector_size_t offset, RowSet rows, const uint64_t* incomingNulls)
       override {
     using T = SelectiveFloatingPointColumnReader<TFile, TRequested>;
-    this->template readCommon<T>(offset, rows, incomingNulls);
+    this->template readCommon<T, true>(offset, rows, incomingNulls);
     this->readOffset_ += rows.back() + 1;
   }
 
diff --git a/velox/dwio/dwrf/reader/SelectiveIntegerDictionaryColumnReader.cpp b/velox/dwio/dwrf/reader/SelectiveIntegerDictionaryColumnReader.cpp
index 649319a6f9da..eaf99cafb937 100644
--- a/velox/dwio/dwrf/reader/SelectiveIntegerDictionaryColumnReader.cpp
+++ b/velox/dwio/dwrf/reader/SelectiveIntegerDictionaryColumnReader.cpp
@@ -105,7 +105,7 @@ void SelectiveIntegerDictionaryColumnReader::read(
 
   // lazy load dictionary only when it's needed
   ensureInitialized();
-  readCommon<SelectiveIntegerDictionaryColumnReader>(rows);
+  readCommon<SelectiveIntegerDictionaryColumnReader, true>(rows);
 
   readOffset_ += rows.back() + 1;
 }
diff --git a/velox/dwio/dwrf/reader/SelectiveIntegerDirectColumnReader.cpp b/velox/dwio/dwrf/reader/SelectiveIntegerDirectColumnReader.cpp
index 57dc53090953..c7bd41bda3be 100644
--- a/velox/dwio/dwrf/reader/SelectiveIntegerDirectColumnReader.cpp
+++ b/velox/dwio/dwrf/reader/SelectiveIntegerDirectColumnReader.cpp
@@ -34,7 +34,7 @@ void SelectiveIntegerDirectColumnReader::read(
       offset,
       rows,
       incomingNulls);
-  readCommon<SelectiveIntegerDirectColumnReader>(rows);
+  readCommon<SelectiveIntegerDirectColumnReader, true>(rows);
   readOffset_ += rows.back() + 1;
 }
 
diff --git a/velox/dwio/dwrf/reader/SelectiveStringDirectColumnReader.cpp b/velox/dwio/dwrf/reader/SelectiveStringDirectColumnReader.cpp
index 76f17361e6a6..adc775d2d101 100644
--- a/velox/dwio/dwrf/reader/SelectiveStringDirectColumnReader.cpp
+++ b/velox/dwio/dwrf/reader/SelectiveStringDirectColumnReader.cpp
@@ -422,9 +422,7 @@ void SelectiveStringDirectColumnReader::readWithVisitor(
   int32_t current = visitor.start();
   constexpr bool isExtract =
       std::is_same_v<typename TVisitor::FilterType, common::AlwaysTrue> &&
-      std::is_same_v<
-          typename TVisitor::Extract,
-          dwio::common::ExtractToReader<SelectiveStringDirectColumnReader>>;
+      std::is_same_v<typename TVisitor::Extract, dwio::common::ExtractToReader>;
   auto nulls = nullsInReadRange_ ? nullsInReadRange_->as<uint64_t>() : nullptr;
 
   if (process::hasAvx2() && isExtract) {
@@ -465,73 +463,11 @@ void SelectiveStringDirectColumnReader::readWithVisitor(
   }
 }
 
-template <typename TFilter, bool isDense, typename ExtractValues>
-void SelectiveStringDirectColumnReader::readHelper(
-    common::Filter* filter,
-    RowSet rows,
-    ExtractValues extractValues) {
-  readWithVisitor(
-      rows,
-      dwio::common::
-          ColumnVisitor<folly::StringPiece, TFilter, ExtractValues, isDense>(
-              *reinterpret_cast<TFilter*>(filter), this, rows, extractValues));
-}
-
-template <bool isDense, typename ExtractValues>
-void SelectiveStringDirectColumnReader::processFilter(
-    common::Filter* filter,
-    RowSet rows,
-    ExtractValues extractValues) {
-  if (filter == nullptr) {
-    readHelper<common::AlwaysTrue, isDense>(
-        &dwio::common::alwaysTrue(), rows, extractValues);
-    return;
-  }
-
-  switch (filter->kind()) {
-    case common::FilterKind::kAlwaysTrue:
-      readHelper<common::AlwaysTrue, isDense>(filter, rows, extractValues);
-      break;
-    case common::FilterKind::kIsNull:
-      filterNulls<StringView>(
-          rows,
-          true,
-          !std::is_same_v<decltype(extractValues), dwio::common::DropValues>);
-      break;
-    case common::FilterKind::kIsNotNull:
-      if (std::is_same_v<decltype(extractValues), dwio::common::DropValues>) {
-        filterNulls<StringView>(rows, false, false);
-      } else {
-        readHelper<common::IsNotNull, isDense>(filter, rows, extractValues);
-      }
-      break;
-    case common::FilterKind::kBytesRange:
-      readHelper<common::BytesRange, isDense>(filter, rows, extractValues);
-      break;
-    case common::FilterKind::kNegatedBytesRange:
-      readHelper<common::NegatedBytesRange, isDense>(
-          filter, rows, extractValues);
-      break;
-    case common::FilterKind::kBytesValues:
-      readHelper<common::BytesValues, isDense>(filter, rows, extractValues);
-      break;
-    case common::FilterKind::kNegatedBytesValues:
-      readHelper<common::NegatedBytesValues, isDense>(
-          filter, rows, extractValues);
-      break;
-    default:
-      readHelper<common::Filter, isDense>(filter, rows, extractValues);
-      break;
-  }
-}
-
 void SelectiveStringDirectColumnReader::read(
     vector_size_t offset,
     RowSet rows,
     const uint64_t* incomingNulls) {
   prepareRead<folly::StringPiece>(offset, rows, incomingNulls);
-  bool isDense = rows.back() == rows.size() - 1;
-
   auto numRows = rows.back() + 1;
   auto numNulls = nullsInReadRange_
       ? BaseVector::countNulls(nullsInReadRange_, 0, numRows)
@@ -542,38 +478,8 @@ void SelectiveStringDirectColumnReader::read(
       lengths_->asMutable<int32_t>(), numRows - numNulls);
   rawLengths_ = lengths_->as<uint32_t>();
   lengthIndex_ = 0;
-  if (scanSpec_->keepValues()) {
-    if (scanSpec_->valueHook()) {
-      if (isDense) {
-        readHelper<common::AlwaysTrue, true>(
-            &dwio::common::alwaysTrue(),
-            rows,
-            dwio::common::ExtractToGenericHook(scanSpec_->valueHook()));
-      } else {
-        readHelper<common::AlwaysTrue, false>(
-            &dwio::common::alwaysTrue(),
-            rows,
-            dwio::common::ExtractToGenericHook(scanSpec_->valueHook()));
-      }
-    } else {
-      if (isDense) {
-        processFilter<true>(
-            scanSpec_->filter(), rows, dwio::common::ExtractToReader(this));
-      } else {
-        processFilter<false>(
-            scanSpec_->filter(), rows, dwio::common::ExtractToReader(this));
-      }
-    }
-  } else {
-    if (isDense) {
-      processFilter<true>(
-          scanSpec_->filter(), rows, dwio::common::DropValues());
-    } else {
-      processFilter<false>(
-          scanSpec_->filter(), rows, dwio::common::DropValues());
-    }
-  }
-
+  dwio::common::StringColumnReadWithVisitorHelper<true>(
+      *this, rows)([&](auto visitor) { readWithVisitor(rows, visitor); });
   readOffset_ += numRows;
 }
 
diff --git a/velox/dwio/dwrf/reader/SelectiveStringDirectColumnReader.h b/velox/dwio/dwrf/reader/SelectiveStringDirectColumnReader.h
index 21fe4a3a25e5..cfa8a7350136 100644
--- a/velox/dwio/dwrf/reader/SelectiveStringDirectColumnReader.h
+++ b/velox/dwio/dwrf/reader/SelectiveStringDirectColumnReader.h
@@ -66,15 +66,6 @@ class SelectiveStringDirectColumnReader
   template <typename TVisitor>
   void readWithVisitor(RowSet rows, TVisitor visitor);
 
-  template <typename TFilter, bool isDense, typename ExtractValues>
-  void readHelper(common::Filter* filter, RowSet rows, ExtractValues values);
-
-  template <bool isDense, typename ExtractValues>
-  void processFilter(
-      common::Filter* filter,
-      RowSet rows,
-      ExtractValues extractValues);
-
   void extractCrossBuffers(
       const int32_t* lengths,
       const int32_t* starts,
diff --git a/velox/dwio/parquet/reader/BooleanColumnReader.h b/velox/dwio/parquet/reader/BooleanColumnReader.h
index 41d3405abd54..73126f467988 100644
--- a/velox/dwio/parquet/reader/BooleanColumnReader.h
+++ b/velox/dwio/parquet/reader/BooleanColumnReader.h
@@ -49,7 +49,7 @@ class BooleanColumnReader : public dwio::common::SelectiveByteRleColumnReader {
 
   void read(vector_size_t offset, RowSet rows, const uint64_t* incomingNulls)
       override {
-    readCommon<BooleanColumnReader>(offset, rows, incomingNulls);
+    readCommon<BooleanColumnReader, true>(offset, rows, incomingNulls);
     readOffset_ += rows.back() + 1;
   }
 
diff --git a/velox/dwio/parquet/reader/FloatingPointColumnReader.h b/velox/dwio/parquet/reader/FloatingPointColumnReader.h
index be4c2cd84363..ed91e67a739f 100644
--- a/velox/dwio/parquet/reader/FloatingPointColumnReader.h
+++ b/velox/dwio/parquet/reader/FloatingPointColumnReader.h
@@ -48,7 +48,7 @@ class FloatingPointColumnReader
   void read(vector_size_t offset, RowSet rows, const uint64_t* incomingNulls)
       override {
     using T = FloatingPointColumnReader<TData, TRequested>;
-    this->template readCommon<T>(offset, rows, incomingNulls);
+    this->template readCommon<T, true>(offset, rows, incomingNulls);
     this->readOffset_ += rows.back() + 1;
   }
 
diff --git a/velox/dwio/parquet/reader/IntegerColumnReader.h b/velox/dwio/parquet/reader/IntegerColumnReader.h
index 59a9fc12bf91..d7b458c73953 100644
--- a/velox/dwio/parquet/reader/IntegerColumnReader.h
+++ b/velox/dwio/parquet/reader/IntegerColumnReader.h
@@ -75,7 +75,7 @@ class IntegerColumnReader : public dwio::common::SelectiveIntegerColumnReader {
         offset,
         rows,
         nullptr);
-    readCommon<IntegerColumnReader>(rows);
+    readCommon<IntegerColumnReader, true>(rows);
     readOffset_ += rows.back() + 1;
   }
 
diff --git a/velox/dwio/parquet/reader/StringColumnReader.cpp b/velox/dwio/parquet/reader/StringColumnReader.cpp
index 334c3c02a7e2..2dd0250159c3 100644
--- a/velox/dwio/parquet/reader/StringColumnReader.cpp
+++ b/velox/dwio/parquet/reader/StringColumnReader.cpp
@@ -31,104 +31,15 @@ uint64_t StringColumnReader::skip(uint64_t numValues) {
   return numValues;
 }
 
-template <typename TFilter, bool isDense, typename ExtractValues>
-void StringColumnReader::readHelper(
-    common::Filter* filter,
-    RowSet rows,
-    ExtractValues extractValues) {
-  formatData_->as<ParquetData>().readWithVisitor(
-      dwio::common::
-          ColumnVisitor<folly::StringPiece, TFilter, ExtractValues, isDense>(
-              *reinterpret_cast<TFilter*>(filter), this, rows, extractValues));
-}
-
-template <bool isDense, typename ExtractValues>
-void StringColumnReader::processFilter(
-    common::Filter* filter,
-    RowSet rows,
-    ExtractValues extractValues) {
-  if (filter == nullptr) {
-    readHelper<common::AlwaysTrue, isDense>(
-        &dwio::common::alwaysTrue(), rows, extractValues);
-    return;
-  }
-
-  switch (filter->kind()) {
-    case common::FilterKind::kAlwaysTrue:
-      readHelper<common::AlwaysTrue, isDense>(filter, rows, extractValues);
-      break;
-    case common::FilterKind::kIsNull:
-      filterNulls<StringView>(
-          rows,
-          true,
-          !std::is_same<decltype(extractValues), dwio::common::DropValues>::
-              value);
-      break;
-    case common::FilterKind::kIsNotNull:
-      if (std::is_same<decltype(extractValues), dwio::common::DropValues>::
-              value) {
-        filterNulls<StringView>(rows, false, false);
-      } else {
-        readHelper<common::IsNotNull, isDense>(filter, rows, extractValues);
-      }
-      break;
-    case common::FilterKind::kBytesRange:
-      readHelper<common::BytesRange, isDense>(filter, rows, extractValues);
-      break;
-    case common::FilterKind::kNegatedBytesRange:
-      readHelper<common::NegatedBytesRange, isDense>(
-          filter, rows, extractValues);
-      break;
-    case common::FilterKind::kBytesValues:
-      readHelper<common::BytesValues, isDense>(filter, rows, extractValues);
-      break;
-    case common::FilterKind::kNegatedBytesValues:
-      readHelper<common::NegatedBytesValues, isDense>(
-          filter, rows, extractValues);
-      break;
-    default:
-      readHelper<common::Filter, isDense>(filter, rows, extractValues);
-      break;
-  }
-}
-
 void StringColumnReader::read(
     vector_size_t offset,
     RowSet rows,
     const uint64_t* incomingNulls) {
   prepareRead<folly::StringPiece>(offset, rows, incomingNulls);
-  bool isDense = rows.back() == rows.size() - 1;
-  if (scanSpec_->keepValues()) {
-    if (scanSpec_->valueHook()) {
-      if (isDense) {
-        readHelper<common::AlwaysTrue, true>(
-            &dwio::common::alwaysTrue(),
-            rows,
-            dwio::common::ExtractToGenericHook(scanSpec_->valueHook()));
-      } else {
-        readHelper<common::AlwaysTrue, false>(
-            &dwio::common::alwaysTrue(),
-            rows,
-            dwio::common::ExtractToGenericHook(scanSpec_->valueHook()));
-      }
-      return;
-    }
-    if (isDense) {
-      processFilter<true>(
-          scanSpec_->filter(), rows, dwio::common::ExtractToReader(this));
-    } else {
-      processFilter<false>(
-          scanSpec_->filter(), rows, dwio::common::ExtractToReader(this));
-    }
-  } else {
-    if (isDense) {
-      processFilter<true>(
-          scanSpec_->filter(), rows, dwio::common::DropValues());
-    } else {
-      processFilter<false>(
-          scanSpec_->filter(), rows, dwio::common::DropValues());
-    }
-  }
+  dwio::common::StringColumnReadWithVisitorHelper<true>(
+      *this, rows)([&](auto visitor) {
+    formatData_->as<ParquetData>().readWithVisitor(visitor);
+  });
   readOffset_ += rows.back() + 1;
 }
 
diff --git a/velox/dwio/parquet/reader/StringColumnReader.h b/velox/dwio/parquet/reader/StringColumnReader.h
index 23269fda8446..e9bedc2365fc 100644
--- a/velox/dwio/parquet/reader/StringColumnReader.h
+++ b/velox/dwio/parquet/reader/StringColumnReader.h
@@ -49,27 +49,6 @@ class StringColumnReader : public dwio::common::SelectiveColumnReader {
   void getValues(RowSet rows, VectorPtr* result) override;
 
   void dedictionarize() override;
-
- private:
-  template <bool hasNulls>
-  void skipInDecode(int32_t numValues, int32_t current, const uint64_t* nulls);
-
-  folly::StringPiece readValue(int32_t length);
-
-  template <bool hasNulls, typename Visitor>
-  void decode(const uint64_t* nulls, Visitor visitor);
-
-  template <typename TVisitor>
-  void readWithVisitor(RowSet rows, TVisitor visitor);
-
-  template <typename TFilter, bool isDense, typename ExtractValues>
-  void readHelper(common::Filter* filter, RowSet rows, ExtractValues values);
-
-  template <bool isDense, typename ExtractValues>
-  void processFilter(
-      common::Filter* filter,
-      RowSet rows,
-      ExtractValues extractValues);
 };
 
 } // namespace facebook::velox::parquet