feat(selective): SIMD filterCache evaluation for dictionary index reading with filters (#17954)

Summary:
X-link: facebookincubator/nimble#916

Pull Request resolved: #17954

Dictionary-encoded string columns in the selective Nimble reader emit a `DictionaryVector` backed by a single alphabet merged across all chunks of a read range. A filter pushed down on such a column cannot be evaluated inside the encoding's per-row decode: the merged alphabet (and therefore the per-alphabet-index `filterCache`) is not known until every chunk has been read, and the inner encoding's `bulkScan` would apply a string filter directly to raw `int32` dictionary indices. The filter is therefore suppressed during the bulk `materializeIndices` read and applied post-hoc on the merged alphabet.

This diff replaces the previous in-encoding (`populateScanState` / framework `StringDictionaryColumnVisitor`) filter evaluation with that post-hoc path, makes it SIMD-accelerated, and removes the now-dead in-encoding code:

- Shared SIMD kernel. Add a free function `filterDictionaryRunSimd<kFilterOnly>(...)` in `velox/dwio/common/ColumnVisitors.h` that gathers each index's cached verdict from `filterCache`, resolves cache misses through the filter (recording the result back), and SIMD-compacts the passing rows (and, unless `kFilterOnly`, the passing indices). Nimble's `filterByCache` calls it directly. DWRF's `StringDictionaryColumnVisitor::processRun` is left unchanged; it still holds an equivalent inline loop, marked with a TODO to adopt the shared kernel in a follow-up diff (kept separate to isolate the DWRF change from this Nimble feature).
- Post-hoc filter in `StringColumnReader`. `readWithDictionary` saves and clears the scan-spec filter so the bulk index read runs, then `filterDictionaryIndices` applies the filter on the merged alphabet via `filterByCache`, compacting `rawValues_`/`outputRows_` to the passing rows. `ensureFilterCache` lazily sizes the per-alphabet-index cache, which grows and clears together with the alphabet.
- Output-layout null realignment. Pre-compacting `rawValues_`/`outputRows_` makes the framework's `compactScalarValues` a no-op (`rows.size() == numValues_`), which skips its null move, so `filterDictionaryIndices` realigns the result nulls itself in three cases: (1) no nulls — filter in place; (2) value filter rejects nulls — filter in place and mark the compacted output all-non-null; (3) IS NULL accepts nulls — merge null rows with the passing non-null rows in row order. `ensureWritableResultNulls` mirrors the framework's `shouldMoveNulls` by switching off the dense `returnReaderNulls_` fast path and allocating an output-indexed null bitmap.
- Delete dead code. With `StringColumnReader` the sole owner of dictionary filtering, the per-encoding `kHasFilter` branches and the `prepareResultNullsForDenseFilter` helper in `encodings/common/Encoding.h` (and their call sites in the `Constant`, `Dictionary`, `MainlyConstant`, and `RLE` encodings) are unreachable and removed, along with `StringColumnReader::populateScanState`.

Performance. On a simulated filter workload (`parallel_reader` over a dictionary-encoded string column, `l_returnflag='R'`, opt mode, `batch_size=1024`, concurrency 16, `read_count=500`, P50), the SIMD `filterCache` path reads at ~52 ms, versus ~70 ms for the parent diff's scalar per-row filter (−26%) and 130 ms for a no-dictionary flat read (−60%). The gain comes from evaluating the byte filter once per distinct dictionary value (`filterCache` memoization) and SIMD-gathering/compacting survivors, instead of testing the filter per row. (Parent/flat figures are from the 2026-06-05 A/B; the SIMD figure was reconfirmed on the current stack on 2026-06-13: P50 wall 52/52/54 ms across 3 runs.)

No behavior change for non-dictionary columns or for dictionary columns read without a pushed-down filter.

Reviewed By: Yuhta

Differential Revision: D102273283

fbshipit-source-id: 6ff65058a06af4be9f04056e6508062b3297793b

Author: Huameng (Michael) Jiang

velox/dwio/common/ColumnVisitors.h

@@ -617,8 +617,6 @@ ColumnVisitor<T, TFilter, ExtractValues, isDense, hasBulkPath>::addOutputRow(
   reader_->addOutputRow(row);
 }
 
-enum FilterResult { kUnknown = 0x40, kSuccess = 0x80, kFailure = 0 };
-
 namespace detail {
 
 template <typename T, typename A>
@@ -1387,6 +1385,11 @@ class StringDictionaryColumnVisitor
     }
     constexpr bool filterOnly =
         std::is_same_v<typename super::Extract, DropValues>;
+    // TODO: This inline SIMD loop duplicates the shared
+    // filterDictionaryRunSimd() free function in DecoderUtil.h; rewire
+    // processRun to call it in a follow-up diff. The dedup is split out to keep
+    // this DWRF change separate from the Nimble post-hoc dictionary filter
+    // feature.
     constexpr int32_t kWidth = xsimd::batch<int32_t>::size;
     for (auto i = 0; i < numInput; i += kWidth) {
       auto indices = xsimd::load_unaligned(input + i);

velox/dwio/common/DecoderUtil.h

@@ -17,6 +17,7 @@
 #pragma once
 
 #include <algorithm>
+#include "velox/common/base/Portability.h"
 #include "velox/common/memory/RawVector.h"
 #include "velox/common/process/ProcessBase.h"
 #include "velox/dwio/common/StreamUtil.h"
@@ -33,6 +34,115 @@ static bool applyFilter(TFilter& filter, T value);
 
 namespace facebook::velox::dwio::common {
 
+// Per-dictionary-index verdict cached by the dictionary filter path. The byte
+// values are chosen so the SIMD kernel below tests them from the gathered
+// int32: bit 7 (kSuccess) is the sign bit and bit 6 (kUnknown) selects
+// unknowns.
+enum FilterResult { kUnknown = 0x40, kSuccess = 0x80, kFailure = 0 };
+
+/// SIMD dictionary-filter kernel shared by the DWRF framework
+/// StringDictionaryColumnVisitor and the Nimble dictionary filter path. Reads a
+/// run of materialized dictionary indices, consults the per-index 'filterCache'
+/// to skip indices already known to pass or fail, resolves cache-unknown
+/// indices by invoking 'testIndex' (which the kernel then records into
+/// 'filterCache' as kSuccess/kFailure), and compacts the passing entries. For
+/// each passing entry the corresponding 'rows' value is appended to
+/// 'filterHits'; unless 'kFilterOnly' is set, the passing dictionary index is
+/// also appended to 'values'. The caller supplies 'rows' already offset to the
+/// current position. Returns the updated number of output values.
+///
+/// Caller-side buffer contracts -- the kernel reads and writes a full SIMD
+/// width at the tail and gathers the cache through a misaligned load, none of
+/// which is bounds-checked:
+///  - 'input' and 'rows' must each have at least kWidth
+///    (= xsimd::batch<int32_t>::size) readable elements beyond 'numInput'.
+///  - 'filterHits', and unless 'kFilterOnly' also 'values', must have room for
+///    at least 'numValues' + 'numInput' + kWidth entries.
+///  - 'filterCache' must have >= 3 bytes of valid leading padding, because each
+///    verdict is gathered via 'filterCache + index - 3'. The Nimble and Velox
+///    callers satisfy this with raw_vector<uint8_t>, whose data() is offset by
+///    velox::simd::kPadding (>= 16).
+template <bool kFilterOnly, typename TestIndex>
+int32_t filterDictionaryRunSimd(
+    const int32_t* input,
+    int32_t numInput,
+    const int32_t* rows,
+    uint8_t* filterCache,
+    int32_t* filterHits,
+    int32_t* values,
+    int32_t numValues,
+    TestIndex&& testIndex) {
+  constexpr int32_t kWidth = xsimd::batch<int32_t>::size;
+  for (auto i = 0; i < numInput; i += kWidth) {
+    auto indices = xsimd::load_unaligned(input + i);
+    xsimd::batch<int32_t> cache;
+    // The gather reads each cache byte as the high byte of an int32 via a
+    // misaligned load at 'filterCache + index - 3', so for index 0 it touches
+    // up to 3 bytes before 'filterCache'. The param is a raw uint8_t*, so this
+    // is a caller contract: 'filterCache' must point into a buffer with >= 3
+    // bytes of valid leading padding. Both callers satisfy it with
+    // raw_vector<uint8_t> (Nimble's DictionaryState::filterCache and velox's
+    // ScanState::filterCache), whose data() is always offset by
+    // velox::simd::kPadding (>= 16). Mirrors the StringDictionaryColumnVisitor
+    // filterCache() - 3 idiom in ColumnVisitors.h.
+    auto base = reinterpret_cast<const int32_t*>(filterCache - 3);
+    if (i + kWidth > numInput) {
+      cache = simd::maskGather<int32_t, int32_t, 1>(
+          xsimd::broadcast<int32_t>(0),
+          simd::leadingMask<int32_t>(numInput - i),
+          base,
+          indices);
+    } else {
+      cache = simd::gather<int32_t, int32_t, 1>(base, indices);
+    }
+#ifdef SVE_BITS
+    auto unknowns = simd::toBitMask(
+        simd::reinterpretBatch<uint32_t>((cache & (kUnknown << 24)) << 1) !=
+        xsimd::batch<uint32_t>(0));
+    auto passed = simd::toBitMask(
+        (simd::reinterpretBatch<uint32_t>(cache) & xsimd::batch<uint32_t>(1)) !=
+        xsimd::batch<uint32_t>(0));
+#else
+    auto unknowns = simd::toBitMask(
+        xsimd::batch_bool<int32_t>(
+            simd::reinterpretBatch<uint32_t>((cache & (kUnknown << 24)) << 1)));
+    auto passed = simd::toBitMask(
+        xsimd::batch_bool<int32_t>(simd::reinterpretBatch<uint32_t>(cache)));
+#endif
+    if (UNLIKELY(unknowns)) {
+      uint16_t bits = unknowns;
+      while (bits) {
+        int index = bits::getAndClearLastSetBit(bits);
+        int32_t value = input[i + index];
+        if (testIndex(value)) {
+          filterCache[value] = FilterResult::kSuccess;
+          passed |= 1 << index;
+        } else {
+          filterCache[value] = FilterResult::kFailure;
+        }
+      }
+    }
+    if (!passed) {
+      continue;
+    } else if (passed == (1 << kWidth) - 1) {
+      xsimd::load_unaligned(rows + i).store_unaligned(filterHits + numValues);
+      if (!kFilterOnly) {
+        indices.store_unaligned(values + numValues);
+      }
+      numValues += kWidth;
+    } else {
+      const int32_t numBits = __builtin_popcount(passed);
+      simd::filter(xsimd::load_unaligned(rows + i), passed)
+          .store_unaligned(filterHits + numValues);
+      if (!kFilterOnly) {
+        simd::filter(indices, passed).store_unaligned(values + numValues);
+      }
+      numValues += numBits;
+    }
+  }
+  return numValues;
+}
+
 inline int32_t firstNullIndex(const uint64_t* nulls, int32_t numRows) {
   int32_t first = -1;
   bits::testUnsetBits(nulls, 0, numRows, [&](int32_t row) {