bytedance · lizhen-0710 · Jul 25, 2026 · Jul 28, 2026
diff --git a/bolt/common/memory/RawVector.cpp b/bolt/common/memory/RawVector.cpp
@@ -43,13 +43,14 @@ bool initializeIota() {
 }
 } // namespace
 
-const int32_t* iota(int32_t size, raw_vector<int32_t>& storage) {
-  if (iotaData.size() < size) {
+const int32_t*
+iota(int32_t size, raw_vector<int32_t>& storage, int32_t offset) {
+  if (iotaData.size() < offset + size) {
     storage.resize(size);
-    std::iota(&storage[0], &storage[storage.size()], 0);
+    std::iota(storage.begin(), storage.end(), offset);
     return storage.data();
   }
-  return iotaData.data();
+  return iotaData.data() + offset;
 }
 
 static bool FB_ANONYMOUS_VARIABLE(g_iotaConstants) = initializeIota();

diff --git a/bolt/common/memory/RawVector.h b/bolt/common/memory/RawVector.h
@@ -266,11 +266,12 @@ class raw_vector {
 };
 
 // Returns a pointer to 'size' int32_t's with consecutive values
-// starting at 0. There are at least kPadding / sizeof(int32_t) values
+// starting at 'offset'. There are at least kPadding / sizeof(int32_t) values
 // past 'size', so that it is safe to access the returned pointer at maximum
 // SIMD width. Typically returns preallocated memory but if this is
 // not large enough,resizes and initializes 'storage' to the requested
 // size and returns storage.data().
-const int32_t* iota(int32_t size, raw_vector<int32_t>& storage);
+const int32_t*
+iota(int32_t size, raw_vector<int32_t>& storage, int32_t offset = 0);
 
 } // namespace bytedance::bolt
diff --git a/bolt/common/memory/tests/RawVectorTest.cpp b/bolt/common/memory/tests/RawVectorTest.cpp
@@ -134,10 +134,15 @@ TEST_P(RawVectorTest, iota) {
   raw_vector<int32_t> storage(pool_.get());
   // Small sizes are preallocated.
   EXPECT_EQ(11, iota(12, storage)[11]);
+  EXPECT_EQ(10, iota(12, storage, 10)[0]);
+  EXPECT_EQ(21, iota(12, storage, 10)[11]);
   EXPECT_TRUE(storage.empty());
   EXPECT_EQ(110000, iota(110001, storage)[110000]);
   // Larger sizes are allocated in 'storage'.
   EXPECT_FALSE(storage.empty());
+  EXPECT_EQ(20000, iota(3, storage, 20000)[0]);
+  EXPECT_EQ(20002, iota(3, storage, 20000)[2]);
+  EXPECT_EQ(3, storage.size());
 }
 
 TEST_P(RawVectorTest, iterator) {

diff --git a/bolt/connectors/hive/HiveDataSource.cpp b/bolt/connectors/hive/HiveDataSource.cpp
@@ -157,10 +157,14 @@ HiveDataSource::HiveDataSource(
   if (remainingFilter) {
     remainingFilterExprSet_ = expressionEvaluator_->compile(remainingFilter);
     auto& remainingFilterExpr = remainingFilterExprSet_->expr(0);
-    folly::F14FastSet<std::string> columnNames(
-        readerRowNames.begin(), readerRowNames.end());
+    folly::F14FastMap<std::string, column_index_t> columnNames;
+    for (column_index_t i = 0; i < readerRowNames.size(); ++i) {
+      columnNames[readerRowNames[i]] = i;
+    }
     for (auto& input : remainingFilterExpr->distinctFields()) {
-      if (columnNames.count(input->field()) > 0) {
+      auto it = columnNames.find(input->field());
+      if (it != columnNames.end()) {
+        multiReferencedFields_.push_back(it->second);
         continue;
       }
       // Remaining filter may reference columns that are not used otherwise,
@@ -849,9 +853,16 @@ int64_t HiveDataSource::estimatedRowSize() {
 }
 
 vector_size_t HiveDataSource::evaluateRemainingFilter(RowVectorPtr& rowVector) {
-  auto filterStartMicros = getCurrentTimeMicro();
   filterRows_.resize(output_->size());
+  for (auto fieldIndex : multiReferencedFields_) {
+    LazyVector::ensureLoadedRows(
+        rowVector->childAt(fieldIndex),
+        filterRows_,
+        filterLazyDecoded_,
+        filterLazyBaseRows_);
+  }
 
+  auto filterStartMicros = getCurrentTimeMicro();
   expressionEvaluator_->evaluate(
       remainingFilterExprSet_.get(),
       filterRows_,

diff --git a/bolt/connectors/hive/HiveDataSource.h b/bolt/connectors/hive/HiveDataSource.h
@@ -45,6 +45,8 @@
 #include "bolt/dwio/common/Statistics.h"
 #include "bolt/exec/OperatorUtils.h"
 #include "bolt/expression/Expr.h"
+#include "bolt/vector/DecodedVector.h"
+#include "bolt/vector/LazyVector.h"
 namespace bytedance::bolt::connector::hive {
 
 class HiveConfig;
@@ -210,9 +212,15 @@ class HiveDataSource : public DataSource {
   std::atomic<uint64_t> totalRemainingFilterTime_{0};
   uint64_t completedRows_ = 0;
 
+  // Field indices referenced in both remaining filter and output type. These
+  // columns need to be materialized eagerly to avoid missing values in output.
+  std::vector<column_index_t> multiReferencedFields_;
+
   // Reusable memory for remaining filter evaluation.
   VectorPtr filterResult_;
   SelectivityVector filterRows_;
+  DecodedVector filterLazyDecoded_;
+  SelectivityVector filterLazyBaseRows_;
   exec::FilterEvalCtx filterEvalCtx_;
 
   RowVectorPtr emptyResult_;

diff --git a/bolt/connectors/hive/PaimonSplitReader.cpp b/bolt/connectors/hive/PaimonSplitReader.cpp
@@ -113,6 +113,7 @@ PaimonRowIteratorPtr PaimonSplitReader::getIterator(SplitReader* rowReader) {
 
   if (rowReader->next(paimon::kMAX_BATCH_SIZE, result)) {
     RowVectorPtr resultAsRowVect = std::static_pointer_cast<RowVector>(result);
+    resultAsRowVect->loadedVector();
     auto primaryKeys = projectVector(resultAsRowVect, primaryKeyIndices_);
 
     auto sequenceFields =

diff --git a/bolt/dwio/common/ColumnVisitors.h b/bolt/dwio/common/ColumnVisitors.h
@@ -168,6 +168,10 @@ class ColumnVisitor {
   using DataType = T;
   static constexpr bool dense = isDense;
   static constexpr bool kHasBulkPath = hasBulkPath;
+  static constexpr bool kHasFilter =
+      !std::is_same_v<TFilter, bolt::common::AlwaysTrue>;
+  static constexpr bool kHasHook =
+      !std::is_same_v<HookType, dwio::common::NoHook>;
   ColumnVisitor(
       TFilter& filter,
       SelectiveColumnReader* reader,
@@ -433,6 +437,10 @@ class ColumnVisitor {
     numValuesBias_ = bias;
   }
 
+  int32_t numValuesBias() const {
+    return numValuesBias_;
+  }
+
   void setNumValues(int32_t size) {
     reader_->setNumValues(numValuesBias_ + size);
     if constexpr (!std::is_same_v<TFilter, bolt::common::AlwaysTrue>) {
@@ -528,7 +536,7 @@ template <
 inline void
 ColumnVisitor<T, TFilter, ExtractValues, isDense, hasBulkPath>::addResult(
     T value) {
-  values_.addValue(rowIndex_, value);
+  values_.addValue(rowIndex_ + numValuesBias_, value);
 }
 
 template <
@@ -539,7 +547,7 @@ template <
     bool hasBulkPath>
 inline void
 ColumnVisitor<T, TFilter, ExtractValues, isDense, hasBulkPath>::addNull() {
-  values_.template addNull<T>(rowIndex_);
+  values_.template addNull<T>(rowIndex_ + numValuesBias_);
 }
 
 template <
@@ -857,7 +865,10 @@ class DictionaryColumnVisitor
         translateByDict(input, numInput, values);
         super::values_.hook().addValues(
             scatter ? scatterRows + super::rowIndex_
-                    : bolt::iota(super::numRows_, super::innerNonNullRows()) +
+                    : bolt::iota(
+                          super::numRows_,
+                          super::innerNonNullRows(),
+                          super::numValuesBias_) +
                     super::rowIndex_,
             values,
             numInput,
@@ -1139,6 +1150,12 @@ class DictionaryColumnVisitor
     return state_.dictionary.numValues;
   }
 
+  static constexpr bool hasFilter() {
+    // Dictionary values cannot be null.
+    return !std::is_same_v<TFilter, bolt::common::AlwaysTrue> &&
+        !std::is_same_v<TFilter, bolt::common::IsNotNull>;
+  }
+
   uint8_t* filterCache() const {
     return state_.filterCache;
   }
@@ -1214,8 +1231,13 @@ class StringDictionaryColumnVisitor
     }
     vector_size_t previous =
         isDense && TFilter::deterministic ? 0 : super::currentRow();
-    if constexpr (std::is_same_v<TFilter, bolt::common::AlwaysTrue>) {
-      super::filterPassed(index);
+    if constexpr (!DictSuper::hasFilter()) {
+      if constexpr (super::kHasHook) {
+        super::values_.addValue(
+            super::rowIndex_ + super::numValuesBias_, valueInDictionary(index));
+      } else {
+        super::filterPassed(index);
+      }
     } else {
       // check the dictionary cache
       if (TFilter::deterministic &&
@@ -1227,7 +1249,7 @@ class StringDictionaryColumnVisitor
         super::filterFailed();
       } else {
         if (bolt::common::applyFilter(
-                super::filter_, valueInDictionary(value, inStrideDict))) {
+                super::filter_, valueInDictionary(index))) {
           super::filterPassed(index);
           if constexpr (TFilter::deterministic) {
             DictSuper::filterCache()[index] = FilterResult::kSuccess;
@@ -1270,25 +1292,31 @@ class StringDictionaryColumnVisitor
       int32_t& numValues) {
     DCHECK(input == values + numValues);
     setByInDict(values + numValues, numInput);
-    if (!hasFilter) {
+    if constexpr (!DictSuper::hasFilter()) {
       if (hasHook) {
         for (auto i = 0; i < numInput; ++i) {
-          auto value = input[i];
           super::values_.addValue(
               scatterRows ? scatterRows[super::rowIndex_ + i]
-                          : super::rowIndex_ + i,
-              value);
+                          : super::rowIndex_ + super::numValuesBias_ + i,
+              valueInDictionary(input[i]));
         }
       }
-      DCHECK_EQ(input, values + numValues);
-      if (scatter) {
+      if constexpr (std::is_same_v<TFilter, bolt::common::IsNotNull>) {
+        auto* begin = (scatter ? scatterRows : super::rows_) + super::rowIndex_;
+        std::copy(begin, begin + numInput, filterHits + numValues);
+        numValues += numInput;
+      } else if (scatter) {
         dwio::common::scatterDense(
             input, scatterRows + super::rowIndex_, numInput, values);
+        numValues = scatterRows[super::rowIndex_ + numInput - 1] + 1;
+      } else {
+        DCHECK_EQ(input, values + numValues);
+        numValues += numInput;
       }
-      numValues = scatter ? scatterRows[super::rowIndex_ + numInput - 1] + 1
-                          : numValues + numInput;
       super::rowIndex_ += numInput;
       return;
+    } else {
+      static_assert(hasFilter);
     }
     constexpr bool filterOnly =
         std::is_same_v<typename super::Extract, DropValues>;
@@ -1325,16 +1353,7 @@ class StringDictionaryColumnVisitor
         while (bits) {
           int index = bits::getAndClearLastSetBit(bits);
           int32_t value = input[i + index];
-          bool result;
-          if (value >= DictSuper::dictionarySize()) {
-            result = applyFilter(
-                super::filter_,
-                valueInDictionary(value - DictSuper::dictionarySize(), true));
-          } else {
-            result =
-                applyFilter(super::filter_, valueInDictionary(value, false));
-          }
-          if (result) {
+          if (applyFilter(super::filter_, valueInDictionary(value))) {
             DictSuper::filterCache()[value] = FilterResult::kSuccess;
             passed |= 1 << index;
           } else {
@@ -1414,67 +1433,17 @@ class StringDictionaryColumnVisitor
     }
   }
 
-  folly::StringPiece valueInDictionary(int64_t index, bool inStrideDict) {
-    if (inStrideDict) {
-      return folly::StringPiece(reinterpret_cast<const StringView*>(
-          DictSuper::state_.dictionary2.values)[index]);
+  folly::StringPiece valueInDictionary(int64_t index) {
+    auto stripeDictSize = DictSuper::state_.dictionary.numValues;
+    if (index < stripeDictSize) {
+      return reinterpret_cast<const StringView*>(
+          DictSuper::state_.dictionary.values)[index];
     }
-    return folly::StringPiece(reinterpret_cast<const StringView*>(
-        DictSuper::state_.dictionary.values)[index]);
+    return reinterpret_cast<const StringView*>(
+        DictSuper::state_.dictionary2.values)[index - stripeDictSize];
   }
 };
 
-class ExtractStringDictionaryToGenericHook {
- public:
-  static constexpr bool kSkipNulls = true;
-  using HookType = ValueHook;
-
-  ExtractStringDictionaryToGenericHook(
-      ValueHook* hook,
-      RowSet rows,
-      RawScanState state)
-
-      : hook_(hook), rows_(rows), state_(state) {}
-
-  bool acceptsNulls() {
-    return hook_->acceptsNulls();
-  }
-
-  template <typename T>
-  void addNull(vector_size_t rowIndex) {
-    hook_->addNull(rowIndex);
-  }
-
-  void addValue(vector_size_t rowIndex, int32_t value) {
-    // We take the string from the stripe or stride dictionary
-    // according to the index. Stride dictionary indices are offset up
-    // by the stripe dict size.
-    if (value < dictionarySize()) {
-      auto view = folly::StringPiece(
-          reinterpret_cast<const StringView*>(state_.dictionary.values)[value]);
-      hook_->addValue(rowIndex, &view);
-    } else {
-      BOLT_DCHECK(state_.inDictionary);
-      auto view = folly::StringPiece(reinterpret_cast<const StringView*>(
-          state_.dictionary2.values)[value - dictionarySize()]);
-      hook_->addValue(rowIndex, &view);
-    }
-  }
-
-  ValueHook& hook() {
-    return *hook_;
-  }
-
- private:
-  int32_t dictionarySize() const {
-    return state_.dictionary.numValues;
-  }
-
-  ValueHook* const hook_;
-  RowSet const rows_;
-  RawScanState state_;
-};
-
 template <typename T, typename TFilter, typename ExtractValues, bool isDense>
 class DirectRleColumnVisitor
     : public ColumnVisitor<T, TFilter, ExtractValues, isDense> {

diff --git a/bolt/dwio/common/DirectDecoder.h b/bolt/dwio/common/DirectDecoder.h
@@ -221,6 +221,11 @@ class DirectDecoder : public IntDecoder<isSigned> {
           return;
         }
       }
+      if (hasHook && visitor.numValuesBias() > 0) {
+        for (auto& row : *outerVector) {
+          row += visitor.numValuesBias();
+        }
+      }
       if (super::useVInts) {
         if (Visitor::dense) {
           super::bulkRead(numNonNull, data);
@@ -271,7 +276,10 @@ class DirectDecoder : public IntDecoder<isSigned> {
                 rowsAsRange,
                 0,
                 rowsAsRange.size(),
-                hasHook ? bolt::iota(numRows, visitor.innerNonNullRows())
+                hasHook ? bolt::iota(
+                              numRows,
+                              visitor.innerNonNullRows(),
+                              visitor.numValuesBias())
                         : nullptr,
                 visitor.rawValues(numRows),
                 hasFilter ? visitor.outputRows(numRows) : nullptr,
@@ -281,7 +289,11 @@ class DirectDecoder : public IntDecoder<isSigned> {
       } else {
         dwio::common::fixedWidthScan<T, filterOnly, false>(
             rowsAsRange,
-            hasHook ? bolt::iota(numRows, visitor.innerNonNullRows()) : nullptr,
+            hasHook ? bolt::iota(
+                          numRows,
+                          visitor.innerNonNullRows(),
+                          visitor.numValuesBias())
+                    : nullptr,
             visitor.rawValues(numRows),
             hasFilter ? visitor.outputRows(numRows) : nullptr,
             numValues,