fix(hashjoin): Create new VectorHashers for listNullKeyRows to prevent dangling pointer access (facebookincubator#12106)

Summary:
This PR addresses an issue in the `listNullKeyRows` function that occurs in hash
mode. In this function, `hashers_` from HashTable is used to construct the new HashLookup.
The `joinProbe` function requires accessing `hashers_[0]->decodedVector().base()`
for key comparison. However, this pointer can be dangling, causing the error
described below.

We propose fixing this issue by using separate VectorHashers with properly decoded vectors (1 row with null keys).

```
[ RUN      ] HashJoinTest/MultiThreadedHashJoinTest.hashModeNullAwareAntiJoinWithFilterAndNullKey/1
unknown file: Failure
C++ exception with description "Exception: VeloxUserError
Error Source: USER
Error Code: INVALID_ARGUMENT
Reason: Specified element is not found : -96
Retriable: False
Function: mapTypeKindToName
File: /var/git/velox/velox/type/Type.cpp
Line: 116
Stack trace:
# 0  facebook::velox::VeloxException::VeloxException(char const*, unsigned long, char const*, std::basic_string_view<char, std::char_traits<char> >, std::basic_string_view<char, std::char_traits<char> >, std::basic_string_view<char, std::char_traits<char> >, std::basic_string_view<char, std::char_traits<char> >, bool, facebook::velox::VeloxException::Type, std::basic_string_view<char, std::char_traits<char> >)
# 1  void facebook::velox::detail::veloxCheckFail<facebook::velox::VeloxUserError, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&>(facebook::velox::detail::VeloxCheckFailArgs const&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&)
# 2  facebook::velox::mapTypeKindToName[abi:cxx11](facebook::velox::TypeKind const&)
# 3  facebook::velox::exec::HashTable<false>::compareKeys(char const*, facebook::velox::exec::HashLookup&, int)
# 4  facebook::velox::exec::HashTable<false>::joinProbe(facebook::velox::exec::HashLookup&)
# 5  facebook::velox::exec::HashTable<false>::listNullKeyRows(facebook::velox::exec::BaseHashTable::NullKeyRowsIterator*, int, char**)
# 6  facebook::velox::exec::HashProbe::applyFilterOnTableRowsForNullAwareJoin(facebook::velox::SelectivityVector const&, facebook::velox::SelectivityVector&, std::function<int (char**, int)>) [clone .part.0]
# 7  facebook::velox::exec::HashProbe::evalFilterForNullAwareJoin(int, bool)
# 8  facebook::velox::exec::HashProbe::evalFilter(int)
# 9  facebook::velox::exec::HashProbe::getOutputInternal(bool)
# 10 facebook::velox::exec::HashProbe::getOutput()
# 11 facebook::velox::exec::Driver::runInternal(std::shared_ptr<facebook::velox::exec::Driver>&, std::shared_ptr<facebook::velox::exec::BlockingState>&, std::shared_ptr<facebook::velox::RowVector>&)::{lambda()https://github.com/facebookincubator/velox/issues/5}::operator()() const
# 12 facebook::velox::exec::Driver::runInternal(std::shared_ptr<facebook::velox::exec::Driver>&, std::shared_ptr<facebook::velox::exec::BlockingState>&, std::shared_ptr<facebook::velox::RowVector>&)
# 13 facebook::velox::exec::Driver::run(std::shared_ptr<facebook::velox::exec::Driver>)
# 14 void folly::detail::function::call_<facebook::velox::exec::Driver::enqueue(std::shared_ptr<facebook::velox::exec::Driver>)::{lambda()facebookincubator#1}, true, false, void>(, folly::detail::function::Data&)
# 15 folly::ThreadPoolExecutor::runTask(std::shared_ptr<folly::ThreadPoolExecutor::Thread> const&, folly::ThreadPoolExecutor::Task&&)
# 16 folly::CPUThreadPoolExecutor::threadRun(std::shared_ptr<folly::ThreadPoolExecutor::Thread>)
# 17 void folly::detail::function::call_<std::_Bind<void (folly::ThreadPoolExecutor::*(folly::ThreadPoolExecutor*, std::shared_ptr<folly::ThreadPoolExecutor::Thread>))(std::shared_ptr<folly::ThreadPoolExecutor::Thread>)>, true, false, void>(, folly::detail::function::Data&)
# 18 0x00000000000dc252
# 19 0x0000000000094ac2
# 20 0x000000000012684f
" thrown in the test body.
```

Pull Request resolved: facebookincubator#12106

Reviewed By: Yuhta

Differential Revision: D68544820

Pulled By: xiaoxmeng

fbshipit-source-id: 6c82b61f940ffcb5288902859fc0e83a8199a568

Author: zhli1142015

velox/exec/HashProbe.cpp

@@ -1380,11 +1380,13 @@ SelectivityVector HashProbe::evalFilterForNullAwareJoin(
   }
 
   if (buildSideHasNullKeys_) {
+    prepareNullKeyProbeHashers();
     BaseHashTable::NullKeyRowsIterator iter;
     nullKeyProbeRows.deselect(filterPassedRows);
     applyFilterOnTableRowsForNullAwareJoin(
         nullKeyProbeRows, filterPassedRows, [&](char** data, int32_t maxRows) {
-          return table_->listNullKeyRows(&iter, maxRows, data);
+          return table_->listNullKeyRows(
+              &iter, maxRows, data, nullKeyProbeHashers_);
         });
   }
   BaseHashTable::RowsIterator iter;
@@ -1399,6 +1401,22 @@ SelectivityVector HashProbe::evalFilterForNullAwareJoin(
   return filterPassedRows;
 }
 
+void HashProbe::prepareNullKeyProbeHashers() {
+  if (nullKeyProbeHashers_.empty()) {
+    nullKeyProbeHashers_ =
+        createVectorHashers(probeType_, joinNode_->leftKeys());
+    // Null-aware joins allow only one join key.
+    VELOX_CHECK_EQ(nullKeyProbeHashers_.size(), 1);
+    if (table_->hashMode() == BaseHashTable::HashMode::kHash) {
+      nullKeyProbeInput_ =
+          BaseVector::create(nullKeyProbeHashers_[0]->type(), 1, pool());
+      nullKeyProbeInput_->setNull(0, true);
+      SelectivityVector selectivity(1);
+      nullKeyProbeHashers_[0]->decode(*nullKeyProbeInput_, selectivity);
+    }
+  }
+}
+
 int32_t HashProbe::evalFilter(int32_t numRows) {
   if (!filter_) {
     return numRows;

velox/exec/HashProbe.h

@@ -165,6 +165,12 @@ class HashProbe : public Operator {
       vector_size_t numRows,
       bool filterPropagateNulls);
 
+  // Prepares the hashers for probing with null keys.
+  // Initializes `nullKeyProbeHashers_` if empty, ensuring it has exactly one
+  // hasher. If the table's hash mode is `kHash`, creates and decodes a null
+  // input vector.
+  void prepareNullKeyProbeHashers();
+
   // Combine the selected probe-side rows with all or null-join-key (depending
   // on the iterator) build side rows and evaluate the filter.  Mark probe rows
   // that pass the filter in 'filterPassedRows'. Used in null-aware join
@@ -680,6 +686,12 @@ class HashProbe : public Operator {
 
   // The spilled probe partitions remaining to restore.
   SpillPartitionSet inputSpillPartitionSet_;
+
+  // VectorHashers used for listing rows with null keys.
+  std::vector<std::unique_ptr<VectorHasher>> nullKeyProbeHashers_;
+
+  // Input vector used for listing rows with null keys.
+  VectorPtr nullKeyProbeInput_;
 };
 
 inline std::ostream& operator<<(std::ostream& os, ProbeOperatorState state) {

velox/exec/HashTable.cpp

@@ -2033,11 +2033,14 @@ template <>
 int32_t HashTable<false>::listNullKeyRows(
     NullKeyRowsIterator* iter,
     int32_t maxRows,
-    char** rows) {
+    char** rows,
+    const std::vector<std::unique_ptr<VectorHasher>>& hashers) {
   if (!iter->initialized) {
     VELOX_CHECK_GT(nextOffset_, 0);
+    // Null-aware joins allow only one join key.
     VELOX_CHECK_EQ(hashers_.size(), 1);
-    HashLookup lookup(hashers_);
+    VELOX_CHECK_EQ(hashers_.size(), hashers.size());
+    HashLookup lookup(hashers);
     if (hashMode_ == HashMode::kHash) {
       lookup.hashes.push_back(VectorHasher::kNullHash);
     } else {
@@ -2075,8 +2078,11 @@ int32_t HashTable<false>::listNullKeyRows(
 }
 
 template <>
-int32_t
-HashTable<true>::listNullKeyRows(NullKeyRowsIterator*, int32_t, char**) {
+int32_t HashTable<true>::listNullKeyRows(
+    NullKeyRowsIterator*,
+    int32_t,
+    char**,
+    const std::vector<std::unique_ptr<VectorHasher>>&) {
   VELOX_UNREACHABLE();
 }
 

velox/exec/HashTable.h

@@ -279,8 +279,11 @@ class BaseHashTable {
 
   /// Returns all rows with null keys.  Used by null-aware joins (e.g. anti or
   /// left semi project).
-  virtual int32_t
-  listNullKeyRows(NullKeyRowsIterator* iter, int32_t maxRows, char** rows) = 0;
+  virtual int32_t listNullKeyRows(
+      NullKeyRowsIterator* iter,
+      int32_t maxRows,
+      char** rows,
+      const std::vector<std::unique_ptr<VectorHasher>>& hashers) = 0;
 
   virtual void prepareJoinTable(
       std::vector<std::unique_ptr<BaseHashTable>> tables,
@@ -531,7 +534,8 @@ class HashTable : public BaseHashTable {
   int32_t listNullKeyRows(
       NullKeyRowsIterator* iter,
       int32_t maxRows,
-      char** rows) override;
+      char** rows,
+      const std::vector<std::unique_ptr<VectorHasher>>& hashers) override;
 
   void clear(bool freeTable) override;
 

velox/exec/tests/HashJoinTest.cpp

@@ -1222,6 +1222,51 @@ TEST_P(MultiThreadedHashJoinTest, nullAwareAntiJoinWithFilterAndNullKey) {
   }
 }
 
+TEST_P(
+    MultiThreadedHashJoinTest,
+    hashModeNullAwareAntiJoinWithFilterAndNullKey) {
+  // Use float type keys to trigger hash mode table.
+  auto probeVectors = makeBatches(50, [&](int32_t /*unused*/) {
+    return makeRowVector(
+        {"t0", "t1"},
+        {
+            makeNullableFlatVector<float>({std::nullopt, 1, 2}),
+            makeFlatVector<int32_t>({1, 1, 2}),
+        });
+  });
+  auto buildVectors = makeBatches(5, [&](int32_t /*unused*/) {
+    return makeRowVector(
+        {"u0", "u1"},
+        {
+            makeNullableFlatVector<float>({std::nullopt, 2, 3}),
+            makeFlatVector<int32_t>({0, 2, 3}),
+        });
+  });
+
+  std::vector<std::string> filters({"u1 < t1", "u1 + t1 = 0"});
+  for (const std::string& filter : filters) {
+    const auto referenceSql = fmt::format(
+        "SELECT t.* FROM t WHERE t0 NOT IN (SELECT u0 FROM u WHERE {})",
+        filter);
+
+    auto testProbeVectors = probeVectors;
+    auto testBuildVectors = buildVectors;
+    HashJoinBuilder(*pool_, duckDbQueryRunner_, driverExecutor_.get())
+        .numDrivers(numDrivers_)
+        .probeKeys({"t0"})
+        .probeVectors(std::move(testProbeVectors))
+        .buildKeys({"u0"})
+        .buildVectors(std::move(testBuildVectors))
+        .joinType(core::JoinType::kAnti)
+        .nullAware(true)
+        .joinFilter(filter)
+        .joinOutputLayout({"t0", "t1"})
+        .referenceQuery(referenceSql)
+        .checkSpillStats(false)
+        .run();
+  }
+}
+
 TEST_P(MultiThreadedHashJoinTest, nullAwareAntiJoinWithFilterOnNullableColumn) {
   const std::string referenceSql =
       "SELECT t.* FROM t WHERE t0 NOT IN (SELECT u0 FROM u WHERE t1 <> u1)";

velox/exec/tests/HashTableTest.cpp

@@ -547,7 +547,14 @@ class HashTableTest : public testing::TestWithParam<bool>,
     ASSERT_EQ(table->hashMode(), mode);
     std::vector<char*> rows(nullValues.size());
     BaseHashTable::NullKeyRowsIterator iter;
-    auto numRows = table->listNullKeyRows(&iter, rows.size(), rows.data());
+    std::vector<std::unique_ptr<VectorHasher>> probeHashers;
+    probeHashers.push_back(std::make_unique<VectorHasher>(keys->type(), 0));
+    auto nullKeyProbeInput = BaseVector::create(keys->type(), 1, pool());
+    nullKeyProbeInput->setNull(0, true);
+    SelectivityVector selectivity(1);
+    probeHashers[0]->decode(*nullKeyProbeInput, selectivity);
+    auto numRows =
+        table->listNullKeyRows(&iter, rows.size(), rows.data(), probeHashers);
     ASSERT_EQ(numRows, nullValues.size());
     auto actual =
         BaseVector::create<FlatVector<int64_t>>(BIGINT(), numRows, pool());
@@ -558,7 +565,9 @@ class HashTableTest : public testing::TestWithParam<bool>,
       nullValues.erase(it);
     }
     ASSERT_TRUE(nullValues.empty());
-    ASSERT_EQ(0, table->listNullKeyRows(&iter, rows.size(), rows.data()));
+    ASSERT_EQ(
+        0,
+        table->listNullKeyRows(&iter, rows.size(), rows.data(), probeHashers));
   }
 
   // Bitmap of positions in batches_ that end up in the table.