fix: Delta update corner cases (facebookincubator#11682)

Summary: Pull Request resolved: facebookincubator#11682

Reviewed By: darrenfu

Differential Revision: D66557832

fbshipit-source-id: 6d50fc13cfa7ed92fc01f152d6068d099d400169

velox/dwio/common/ScanSpec.cpp

@@ -133,6 +133,19 @@ bool ScanSpec::hasFilter() const {
   return false;
 }
 
+bool ScanSpec::hasFilterApplicableToConstant() const {
+  if (filter_) {
+    return true;
+  }
+  for (auto& child : children_) {
+    if (!child->isArrayElementOrMapEntry_ &&
+        child->hasFilterApplicableToConstant()) {
+      return true;
+    }
+  }
+  return false;
+}
+
 bool ScanSpec::testNull() const {
   auto* filter = this->filter();
   if (filter && !filter->testNull()) {
@@ -368,10 +381,16 @@ std::string ScanSpec::toString() const {
     out << fieldName_;
     if (filter_) {
       out << " filter " << filter_->toString();
+      if (filterDisabled_) {
+        out << " disabled";
+      }
     }
     if (isConstant()) {
       out << " constant";
     }
+    if (deltaUpdate_) {
+      out << " deltaUpdate_=" << deltaUpdate_;
+    }
     if (!metadataFilters_.empty()) {
       out << " metadata_filters(" << metadataFilters_.size() << ")";
     }
@@ -471,9 +490,9 @@ void filterSimpleVectorRows(
     Filter& filter,
     vector_size_t size,
     uint64_t* result) {
+  VELOX_CHECK(size == 0 || result);
   using T = typename TypeTraits<kKind>::NativeType;
   auto* simpleVector = vector.asChecked<SimpleVector<T>>();
-  VELOX_CHECK_NOT_NULL(simpleVector);
   bits::forEachSetBit(result, 0, size, [&](auto i) {
     if (simpleVector->isNullAt(i)) {
       if (!filter.testNull()) {
@@ -521,11 +540,8 @@ void filterRows(
 } // namespace
 
 void ScanSpec::applyFilter(const BaseVector& vector, uint64_t* result) const {
-  if (!hasFilter()) {
-    return;
-  }
-  if (auto* filter = this->filter()) {
-    filterRows(vector, *filter, vector.size(), result);
+  if (filter_) {
+    filterRows(vector, *filter_, vector.size(), result);
   }
   if (!vector.type()->isRow()) {
     // Filter on MAP or ARRAY children are pruning, and won't affect correctness

velox/dwio/common/ScanSpec.h

@@ -258,6 +258,12 @@ class ScanSpec {
   // This may change as a result of runtime adaptation.
   bool hasFilter() const;
 
+  /// Similar as hasFilter() but also return true even there is a filter on
+  /// constant.  Used by delta updated columns because these columns will have
+  /// delta update on constants which makes them no longer constant.  This
+  /// method also ignores filterDisabled_.
+  bool hasFilterApplicableToConstant() const;
+
   /// Assume this field is read as null constant vector (usually due to missing
   /// field), check if any filter in the struct subtree would make the whole
   /// vector to be filtered out.  Return false when the whole vector should be
@@ -351,6 +357,9 @@ class ScanSpec {
     }
   }
 
+  /// Apply filter to the input `vector' and set the passed bits in `result'.
+  /// This method is used by non-selective reader and delta update, so it
+  /// ignores the filterDisabled_ state.
   void applyFilter(const BaseVector& vector, uint64_t* result) const;
 
   bool isFlatMapAsStruct() const {