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

velox/exec/HashProbe.cpp

@@ -1373,11 +1373,24 @@ SelectivityVector HashProbe::evalFilterForNullAwareJoin(
   }
 
   if (buildSideHasNullKeys_) {
+    if (nullKeyProbeHashers_.empty()) {
+      nullKeyProbeHashers_ =
+          createVectorHashers(probeType_, joinNode_->leftKeys());
+      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);
+      }
+    }
     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;

velox/exec/HashProbe.h

@@ -680,6 +680,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

@@ -2020,11 +2020,11 @@ 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);
-    VELOX_CHECK_EQ(hashers_.size(), 1);
-    HashLookup lookup(hashers_);
+    HashLookup lookup(hashers);
     if (hashMode_ == HashMode::kHash) {
       lookup.hashes.push_back(VectorHasher::kNullHash);
     } else {
@@ -2062,8 +2062,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

@@ -2138,6 +2138,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.