fix: Do not use hook in DeltaUpdateColumnLoader (facebookincubator#13996)

Summary:
Pull Request resolved: facebookincubator#13996

In `DeltaUpdateColumnLoader`, hooks cannot be used efficiently and only
used as a fallback.  Since hooks are only used in aggregation today, we can add
a new method `VectorLoader::supportsHook` to indicate if hooks should be used,
and avoid using of hook in aggregation if the loader does not support it.  This
allows us to remove some inefficient and error-prone code (some indices mismatch
is causing out of boundary access and crash).
bypass-github-export-checks

Reviewed By: oerling

Differential Revision: D77618589

fbshipit-source-id: 71bef3cc8bef1988b236d336bc9f92d09a020c91

Author: Yuhta

velox/dwio/common/ColumnLoader.cpp

@@ -87,64 +87,6 @@ void scatter(RowSet rows, vector_size_t resultSize, VectorPtr* result) {
   *result = BaseVector::wrapInDictionary(nullptr, indices, resultSize, *result);
 }
 
-template <typename T>
-void addToHookImpl(
-    const DecodedVector& decoded,
-    const RowSet& rows,
-    ValueHook& hook) {
-  if (decoded.isIdentityMapping() && rows.size() == rows.back() + 1) {
-    auto* values = decoded.data<T>();
-    hook.addValues(rows.data(), values, rows.size());
-    return;
-  }
-  // Masked aggregate is not pushed down for now, so the input row set is
-  // guaranteed to be dense.  Just make sure it works with sparse input for
-  // future proof.
-  for (vector_size_t i = 0; i < rows.size(); ++i) {
-    if (!decoded.isNullAt(i)) {
-      hook.addValueTyped(rows[i], decoded.valueAt<T>(i));
-    } else if (hook.acceptsNulls()) {
-      hook.addNull(rows[i]);
-    }
-  }
-}
-
-void addToHook(
-    const DecodedVector& decoded,
-    const RowSet& rows,
-    ValueHook& hook) {
-  switch (decoded.base()->typeKind()) {
-    case TypeKind::BOOLEAN:
-      addToHookImpl<bool>(decoded, rows, hook);
-      break;
-    case TypeKind::TINYINT:
-      addToHookImpl<int8_t>(decoded, rows, hook);
-      break;
-    case TypeKind::SMALLINT:
-      addToHookImpl<int16_t>(decoded, rows, hook);
-      break;
-    case TypeKind::INTEGER:
-      addToHookImpl<int32_t>(decoded, rows, hook);
-      break;
-    case TypeKind::BIGINT:
-      addToHookImpl<int64_t>(decoded, rows, hook);
-      break;
-    case TypeKind::REAL:
-      addToHookImpl<float>(decoded, rows, hook);
-      break;
-    case TypeKind::DOUBLE:
-      addToHookImpl<double>(decoded, rows, hook);
-      break;
-    case TypeKind::VARCHAR:
-    case TypeKind::VARBINARY:
-      addToHookImpl<StringView>(decoded, rows, hook);
-      break;
-    default:
-      VELOX_FAIL(
-          "Unsupported type kind for hook: {}", decoded.base()->typeKind());
-  }
-}
-
 } // namespace
 
 void ColumnLoader::loadInternal(
@@ -198,11 +140,8 @@ void DeltaUpdateColumnLoader::loadInternal(
       read(structReader_, fieldReader_, version_, rows, selectedRows, nullptr);
   fieldReader_->getValues(effectiveRows, result);
   scanSpec->deltaUpdate()->update(effectiveRows, *result);
-  if (hook) {
-    DecodedVector decoded(**result);
-    addToHook(decoded, rows, *hook);
-  } else if (
-      rows.back() + 1 < resultSize ||
+  VELOX_CHECK_NULL(hook);
+  if (rows.back() + 1 < resultSize ||
       rows.size() != structReader_->outputRows().size()) {
     scatter(rows, resultSize, result);
   }

velox/dwio/common/ColumnLoader.h

@@ -30,6 +30,10 @@ class ColumnLoader : public VectorLoader {
         fieldReader_(fieldReader),
         version_(version) {}
 
+  bool supportsHook() const override {
+    return true;
+  }
+
  private:
   void loadInternal(
       RowSet rows,

velox/functions/lib/aggregates/MinMaxAggregateBase.cpp

@@ -114,7 +114,8 @@ class SimpleNumericMaxAggregate : public SimpleNumericMinMaxAggregate<T> {
       bool mayPushdown) override {
     if constexpr (BaseAggregate::template kMayPushdown<T>) {
       if (!args[0]->type()->isDecimal()) {
-        if (mayPushdown && args[0]->isLazy()) {
+        if (mayPushdown && args[0]->isLazy() &&
+            args[0]->asChecked<const LazyVector>()->supportsHook()) {
           BaseAggregate::template pushdown<
               velox::aggregate::MinMaxHook<T, false>>(groups, rows, args[0]);
           return;
@@ -215,7 +216,8 @@ class SimpleNumericMinAggregate : public SimpleNumericMinMaxAggregate<T> {
       bool mayPushdown) override {
     if constexpr (BaseAggregate::template kMayPushdown<T>) {
       if (!args[0]->type()->isDecimal()) {
-        if (mayPushdown && args[0]->isLazy()) {
+        if (mayPushdown && args[0]->isLazy() &&
+            args[0]->asChecked<const LazyVector>()->supportsHook()) {
           BaseAggregate::template pushdown<
               velox::aggregate::MinMaxHook<T, true>>(groups, rows, args[0]);
           return;

velox/functions/lib/aggregates/SimpleNumericAggregate.h

@@ -98,22 +98,24 @@ class SimpleNumericAggregate : public exec::Aggregate {
       UpdateSingleValue updateSingleValue,
       bool mayPushdown) {
     DecodedVector decoded(*arg, rows, !mayPushdown);
-    auto encoding = decoded.base()->encoding();
     if constexpr (kMayPushdown<TData>) {
-      if (encoding == VectorEncoding::Simple::LAZY &&
+      if (mayPushdown &&
+          decoded.base()->encoding() == VectorEncoding::Simple::LAZY &&
           !arg->type()->isDecimal()) {
-        velox::aggregate::SimpleCallableHook<TData, UpdateSingleValue> hook(
-            exec::Aggregate::offset_,
-            exec::Aggregate::nullByte_,
-            exec::Aggregate::nullMask_,
-            groups,
-            &this->exec::Aggregate::numNulls_,
-            updateSingleValue);
-
-        auto indices = decoded.indices();
-        decoded.base()->as<const LazyVector>()->load(
-            RowSet(indices, arg->size()), &hook);
-        return;
+        auto* lazy = decoded.base()->asChecked<const LazyVector>();
+        if (lazy->supportsHook()) {
+          velox::aggregate::SimpleCallableHook<TData, UpdateSingleValue> hook(
+              exec::Aggregate::offset_,
+              exec::Aggregate::nullByte_,
+              exec::Aggregate::nullMask_,
+              groups,
+              &this->exec::Aggregate::numNulls_,
+              updateSingleValue);
+          auto indices = decoded.indices();
+          lazy->load(RowSet(indices, arg->size()), &hook);
+          return;
+        }
+        decoded.decode(*arg, rows);
       }
     }
 
@@ -226,9 +228,9 @@ class SimpleNumericAggregate : public exec::Aggregate {
         numIndices = numSelected;
       }
     }
-
-    decoded.base()->as<const LazyVector>()->load(
-        RowSet(indices, numIndices), &hook);
+    auto* lazy = decoded.base()->asChecked<const LazyVector>();
+    VELOX_CHECK(lazy->supportsHook());
+    lazy->load(RowSet(indices, numIndices), &hook);
   }
 
  private:

velox/functions/lib/aggregates/SumAggregateBase.h

@@ -124,7 +124,8 @@ class SumAggregateBase
       bool mayPushdown) {
     const auto& arg = args[0];
 
-    if (mayPushdown && arg->isLazy()) {
+    if (mayPushdown && arg->isLazy() &&
+        arg->asChecked<const LazyVector>()->supportsHook()) {
       BaseAggregate::template pushdown<
           facebook::velox::aggregate::SumHook<TData, Overflow>>(
           groups, rows, arg);

velox/vector/LazyVector.cpp

@@ -254,6 +254,7 @@ void LazyVector::validate(const VectorValidateOptions& options) const {
 
 void LazyVector::load(RowSet rows, ValueHook* hook) const {
   VELOX_CHECK(!allLoaded_, "A LazyVector can be loaded at most once");
+  VELOX_CHECK(!hook || supportsHook());
 
   allLoaded_ = true;
   if (rows.empty()) {

velox/vector/LazyVector.h

@@ -201,6 +201,10 @@ class VectorLoader {
       vector_size_t resultSize,
       VectorPtr* result);
 
+  virtual bool supportsHook() const {
+    return false;
+  }
+
  protected:
   virtual void loadInternal(
       RowSet rows,
@@ -346,6 +350,10 @@ class LazyVector : public BaseVector {
 
   VectorPtr slice(vector_size_t offset, vector_size_t length) const override;
 
+  bool supportsHook() const {
+    return loader_->supportsHook();
+  }
+
   // Loads 'rows' of 'vector'. 'vector' may be an arbitrary wrapping
   // of a LazyVector. 'rows' are translated through the wrappers. If
   // there is no LazyVector inside 'vector', this has no

velox/vector/tests/VectorTest.cpp

@@ -65,6 +65,10 @@ class TestingLoader : public VectorLoader {
     return rowCounter_;
   }
 
+  bool supportsHook() const override {
+    return true;
+  }
+
  private:
   template <TypeKind Kind>
   void applyHook(RowSet rows, ValueHook* hook) {
@@ -2053,6 +2057,7 @@ class TestingHook : public ValueHook {
       }
     }
   }
+
   int32_t errors() const {
     return errors_;
   }