diff --git a/be/src/exec/grouping-aggregator.cc b/be/src/exec/grouping-aggregator.cc
index e0d321e..6374f80 100644
--- a/be/src/exec/grouping-aggregator.cc
+++ b/be/src/exec/grouping-aggregator.cc
@@ -399,7 +399,10 @@
   ClosePartitions();
 
   if (tuple_pool_.get() != nullptr) tuple_pool_->FreeAll();
-  if (ht_ctx_.get() != nullptr) ht_ctx_->Close(state);
+  if (ht_ctx_.get() != nullptr) {
+    ht_ctx_->StatsCountersAdd(ht_stats_profile_.get());
+    ht_ctx_->Close(state);
+  }
   ht_ctx_.reset();
   if (serialize_stream_.get() != nullptr) {
     serialize_stream_->Close(nullptr, RowBatch::FlushMode::NO_FLUSH_RESOURCES);
@@ -669,7 +672,7 @@
 }
 
 Status GroupingAggregator::CheckAndResizeHashPartitions(
-    bool partitioning_aggregated_rows, int num_rows, const HashTableCtx* ht_ctx) {
+    bool partitioning_aggregated_rows, int num_rows, HashTableCtx* ht_ctx) {
   DCHECK(!is_streaming_preagg_);
   for (int i = 0; i < PARTITION_FANOUT; ++i) {
     Partition* partition = hash_partitions_[i];
diff --git a/be/src/exec/grouping-aggregator.h b/be/src/exec/grouping-aggregator.h
index 5baac2e..3671327 100644
--- a/be/src/exec/grouping-aggregator.h
+++ b/be/src/exec/grouping-aggregator.h
@@ -583,7 +583,7 @@
   /// resize the hash tables, may spill partitions. 'aggregated_rows' is true if
   /// we're currently partitioning aggregated rows.
   Status CheckAndResizeHashPartitions(
-      bool aggregated_rows, int num_rows, const HashTableCtx* ht_ctx) WARN_UNUSED_RESULT;
+      bool aggregated_rows, int num_rows, HashTableCtx* ht_ctx) WARN_UNUSED_RESULT;
 
   /// Prepares the next partition to return results from. On return, this function
   /// initializes output_iterator_ and output_partition_. This either removes
diff --git a/be/src/exec/hash-table.cc b/be/src/exec/hash-table.cc
index 5bd2b91..bec3c4d 100644
--- a/be/src/exec/hash-table.cc
+++ b/be/src/exec/hash-table.cc
@@ -152,6 +152,7 @@
 }
 
 void HashTableCtx::Close(RuntimeState* state) {
+  if (VLOG_IS_ON(3)) VLOG(3) << PrintStats();
   free(scratch_row_);
   scratch_row_ = NULL;
   expr_values_cache_.Close(expr_perm_pool_->mem_tracker());
@@ -159,6 +160,21 @@
   ScalarExprEvaluator::Close(probe_expr_evals_, state);
 }
 
+string HashTableCtx::PrintStats() const {
+  double avg_travel = num_probes_ == 0 ? 0 : (double)travel_length_ / (double)num_probes_;
+  stringstream ss;
+  ss << "Probes: " << num_probes_ << endl;
+  ss << "Travel: " << travel_length_ << " " << avg_travel << endl;
+  ss << "HashCollisions: " << num_hash_collisions_ << endl;
+  return ss.str();
+}
+
+void HashTableCtx::StatsCountersAdd(HashTableStatsProfile* profile) {
+  COUNTER_ADD(profile->num_hash_collisions_, num_hash_collisions_);
+  COUNTER_ADD(profile->num_hash_probes_, num_probes_);
+  COUNTER_ADD(profile->num_hash_travels_, travel_length_);
+}
+
 uint32_t HashTableCtx::Hash(const void* input, int len, uint32_t hash) const {
   /// Use CRC hash at first level for better performance. Switch to murmur hash at
   /// subsequent levels since CRC doesn't randomize well with different seed inputs.
@@ -433,26 +449,21 @@
 }
 
 void HashTable::Close() {
-  // Print statistics only for the large or heavily used hash tables.
-  // TODO: Tweak these numbers/conditions, or print them always?
+  // Print statistics only for the large hash tables or extremely verbose logging is
+  // enabled.
   const int64_t LARGE_HT = 128 * 1024;
-  const int64_t HEAVILY_USED = 1024 * 1024;
-  // TODO: These statistics should go to the runtime profile as well.
-  if ((num_buckets_ > LARGE_HT) || (num_probes_ > HEAVILY_USED)) VLOG(2) << PrintStats();
+  if (num_buckets_ > LARGE_HT || VLOG_IS_ON(3)) VLOG(2) << PrintStats();
   for (auto& data_page : data_pages_) allocator_->Free(move(data_page));
   data_pages_.clear();
   if (bucket_allocation_ != nullptr) allocator_->Free(move(bucket_allocation_));
 }
 
 void HashTable::StatsCountersAdd(HashTableStatsProfile* profile) {
-  COUNTER_ADD(profile->num_hash_collisions_, num_hash_collisions_);
-  COUNTER_ADD(profile->num_hash_probes_, num_probes_);
-  COUNTER_ADD(profile->num_hash_travels_, travel_length_);
-  COUNTER_ADD(profile->num_hash_resizes_, this->num_resizes_);
+  COUNTER_ADD(profile->num_hash_resizes_, num_resizes_);
 }
 
 Status HashTable::CheckAndResize(
-    uint64_t buckets_to_fill, const HashTableCtx* ht_ctx, bool* got_memory) {
+    uint64_t buckets_to_fill, HashTableCtx* __restrict__ ht_ctx, bool* got_memory) {
   uint64_t shift = 0;
   while (num_filled_buckets_ + buckets_to_fill >
          (num_buckets_ << shift) * MAX_FILL_FACTOR) {
@@ -464,7 +475,7 @@
 }
 
 Status HashTable::ResizeBuckets(
-    int64_t num_buckets, const HashTableCtx* ht_ctx, bool* got_memory) {
+    int64_t num_buckets, HashTableCtx* __restrict__ ht_ctx, bool* got_memory) {
   DCHECK_EQ((num_buckets & (num_buckets - 1)), 0)
       << "num_buckets=" << num_buckets << " must be a power of 2";
   DCHECK_GT(num_buckets, num_filled_buckets_)
@@ -502,8 +513,8 @@
        NextFilledBucket(&iter.bucket_idx_, &iter.node_)) {
     Bucket* bucket_to_copy = &buckets_[iter.bucket_idx_];
     bool found = false;
-    int64_t bucket_idx =
-        Probe<true>(new_buckets, num_buckets, NULL, bucket_to_copy->hash, &found);
+    int64_t bucket_idx = Probe<true, false>(
+        new_buckets, num_buckets, ht_ctx, bucket_to_copy->hash, &found);
     DCHECK(!found);
     DCHECK_NE(bucket_idx, Iterator::BUCKET_NOT_FOUND) << " Probe failed even though "
         " there are free buckets. " << num_buckets << " " << num_filled_buckets_;
@@ -581,17 +592,13 @@
 }
 
 string HashTable::PrintStats() const {
-  double curr_fill_factor = (double)num_filled_buckets_/(double)num_buckets_;
-  double avg_travel = (double)travel_length_/(double)num_probes_;
-  double avg_collisions = (double)num_hash_collisions_/(double)num_filled_buckets_;
+  double curr_fill_factor =
+      num_buckets_ == 0 ? 0 : (double)num_filled_buckets_ / (double)num_buckets_;
   stringstream ss;
   ss << "Buckets: " << num_buckets_ << " " << num_filled_buckets_ << " "
      << curr_fill_factor << endl;
   ss << "Duplicates: " << num_buckets_with_duplicates_ << " buckets "
      << num_duplicate_nodes_ << " nodes" << endl;
-  ss << "Probes: " << num_probes_ << endl;
-  ss << "Travel: " << travel_length_ << " " << avg_travel << endl;
-  ss << "HashCollisions: " << num_hash_collisions_ << " " << avg_collisions << endl;
   ss << "Resizes: " << num_resizes_ << endl;
   return ss.str();
 }
diff --git a/be/src/exec/hash-table.h b/be/src/exec/hash-table.h
index c4df52c..03dd850 100644
--- a/be/src/exec/hash-table.h
+++ b/be/src/exec/hash-table.h
@@ -51,6 +51,7 @@
 class Tuple;
 class TupleRow;
 class HashTable;
+struct HashTableStatsProfile;
 
 /// Linear or quadratic probing hash table implementation tailored to the usage pattern
 /// for partitioned hash aggregation and hash joins. The hash table stores TupleRows and
@@ -63,11 +64,16 @@
 /// expr results for the current row being processed when possible into a contiguous
 /// memory buffer. This allows for efficient hash computation.
 //
-/// The hash table does not support removes. The hash table is not thread safe.
-/// The table is optimized for the partition hash aggregation and hash joins and is not
-/// intended to be a generic hash table implementation. The API loosely mimics the
-/// std::hashset API.
-//
+/// The hash table does not support removes. The table is optimized for the partition hash
+/// aggregation and hash joins and is not intended to be a generic hash table
+/// implementation.
+///
+/// Operations that mutate the hash table are not thread-safe. Read-only access to the
+/// hash table, is, however, thread safe: multiple threads, each with their own
+/// HashTableCtx, can safely look up rows in the hash table with FindProbeRow(), etc so
+/// long as no thread is mutating the hash table. Thread-safe methods are explicitly
+/// documented.
+///
 /// The data (rows) are stored in a BufferedTupleStream. The basic data structure of this
 /// hash table is a vector of buckets. The buckets (indexed by the mod of the hash)
 /// contain a pointer to either the slot in the tuple-stream or in case of duplicate
@@ -129,6 +135,15 @@
   /// Call to cleanup any resources allocated by the expression evaluators.
   void Close(RuntimeState* state);
 
+  /// Update and print some statistics that can be used for performance debugging.
+  std::string PrintStats() const;
+
+  /// Add operations stats of this hash table context to the counters in profile.
+  /// This method should only be called once for each context and be called during
+  /// closing the owner object of the context. Not all the counters are added with the
+  /// method, only counters for Probes, travels and collisions are affected.
+  void StatsCountersAdd(HashTableStatsProfile* profile);
+
   void set_level(int level);
 
   int ALWAYS_INLINE level() const { return level_; }
@@ -527,6 +542,19 @@
   /// clearing them when results from the respective expr evaluators are no longer needed.
   MemPool* build_expr_results_pool_;
   MemPool* probe_expr_results_pool_;
+
+  /// The stats below can be used for debugging perf.
+  /// Number of Probe() calls that probe the hash table.
+  int64_t num_probes_ = 0;
+
+  /// Total distance traveled for each probe. That is the sum of the diff between the end
+  /// position of a probe (find/insert) and its start position
+  /// (hash & (num_buckets_ - 1)).
+  int64_t travel_length_ = 0;
+
+  /// The number of cases where we had to compare buckets with the same hash value, but
+  /// the row equality failed.
+  int64_t num_hash_collisions_ = 0;
 };
 
 /// HashTableStatsProfile encapsulates hash tables stats. It tracks the stats of all the
@@ -655,7 +683,7 @@
   /// Add operations stats of this hash table to the counters in profile.
   /// This method should only be called once for each HashTable and be called during
   /// closing the owner object of the HashTable. Not all the counters are added with the
-  /// method, only counters for Probes, travels, collisions and resizes are affected.
+  /// method, only counters for resizes are affected.
   void StatsCountersAdd(HashTableStatsProfile* profile);
 
   /// Inserts the row to the hash table. The caller is responsible for ensuring that the
@@ -669,11 +697,12 @@
   /// only one tuple, a pointer to that tuple is stored. Otherwise the 'flat_row' pointer
   /// is stored. The 'row' is not copied by the hash table and the caller must guarantee
   /// it stays in memory. This will not grow the hash table.
-  bool IR_ALWAYS_INLINE Insert(HashTableCtx* ht_ctx,
+  bool IR_ALWAYS_INLINE Insert(HashTableCtx* __restrict__ ht_ctx,
       BufferedTupleStream::FlatRowPtr flat_row, TupleRow* row,
       Status* status) WARN_UNUSED_RESULT;
 
   /// Prefetch the hash table bucket which the given hash value 'hash' maps to.
+  /// Thread-safe for read-only hash tables.
   template <const bool READ>
   void IR_ALWAYS_INLINE PrefetchBucket(uint32_t hash);
 
@@ -685,26 +714,33 @@
   /// row. The matching rows do not need to be evaluated since all the nodes of a bucket
   /// are duplicates. One scan can be in progress for each 'ht_ctx'. Used in the probe
   /// phase of hash joins.
-  Iterator IR_ALWAYS_INLINE FindProbeRow(HashTableCtx* ht_ctx);
+  /// Thread-safe for read-only hash tables.
+  Iterator IR_ALWAYS_INLINE FindProbeRow(HashTableCtx* __restrict__ ht_ctx);
 
   /// If a match is found in the table, return an iterator as in FindProbeRow(). If a
   /// match was not present, return an iterator pointing to the empty bucket where the key
   /// should be inserted. Returns End() if the table is full. The caller can set the data
   /// in the bucket using a Set*() method on the iterator.
-  Iterator IR_ALWAYS_INLINE FindBuildRowBucket(HashTableCtx* ht_ctx, bool* found);
+  /// Thread-safe for read-only hash tables.
+  Iterator IR_ALWAYS_INLINE FindBuildRowBucket(
+      HashTableCtx* __restrict__ ht_ctx, bool* found);
 
   /// Returns number of elements inserted in the hash table
+  /// Thread-safe for read-only hash tables.
   int64_t size() const {
     return num_filled_buckets_ - num_buckets_with_duplicates_ + num_duplicate_nodes_;
   }
 
   /// Returns the number of empty buckets.
+  /// Thread-safe for read-only hash tables.
   int64_t EmptyBuckets() const { return num_buckets_ - num_filled_buckets_; }
 
   /// Returns the number of buckets
+  /// Thread-safe for read-only hash tables.
   int64_t num_buckets() const { return num_buckets_; }
 
   /// Returns the load factor (the number of non-empty buckets)
+  /// Thread-safe for read-only hash tables.
   double load_factor() const {
     return static_cast<double>(num_filled_buckets_) / num_buckets_;
   }
@@ -726,6 +762,7 @@
 
   /// Returns the memory occupied by the hash table, takes into account the number of
   /// duplicates.
+  /// Thread-safe for read-only hash tables.
   int64_t CurrentMemSize() const;
 
   /// Returns the number of inserts that can be performed before resizing the table.
@@ -738,7 +775,7 @@
   /// inserted without need to resize. If there is not enough memory available to
   /// resize the hash table, Status::OK() is returned and 'got_memory' is false. If a
   /// another error occurs, an error status may be returned.
-  Status CheckAndResize(uint64_t buckets_to_fill, const HashTableCtx* ht_ctx,
+  Status CheckAndResize(uint64_t buckets_to_fill, HashTableCtx* __restrict__ ht_ctx,
       bool* got_memory) WARN_UNUSED_RESULT;
 
   /// Returns the number of bytes allocated to the hash table from the block manager.
@@ -748,23 +785,28 @@
 
   /// Returns an iterator at the beginning of the hash table.  Advancing this iterator
   /// will traverse all elements.
+  /// Thread-safe for read-only hash tables.
   Iterator Begin(const HashTableCtx* ht_ctx);
 
   /// Return an iterator pointing to the first element (Bucket or DuplicateNode, if the
   /// bucket has duplicates) in the hash table that does not have its matched flag set.
   /// Used in right joins and full-outer joins.
+  /// Thread-safe for read-only hash tables.
   Iterator FirstUnmatched(HashTableCtx* ctx);
 
   /// Return true if there was a least one match.
+  /// Thread-safe for read-only hash tables.
   bool HasMatches() const { return has_matches_; }
 
   /// Return end marker.
+  /// Thread-safe for read-only hash tables.
   Iterator End() { return Iterator(); }
 
   /// Dump out the entire hash table to string.  If 'skip_empty', empty buckets are
   /// skipped.  If 'show_match', it also prints the matched flag of each node. If
   /// 'build_desc' is non-null, the build rows will be printed. Otherwise, only the
   /// the addresses of the build rows will be printed.
+  /// Thread-safe for read-only hash tables.
   std::string DebugString(bool skip_empty, bool show_match,
       const RowDescriptor* build_desc);
 
@@ -772,6 +814,7 @@
   void DebugStringTuple(std::stringstream& ss, HtData& htdata, const RowDescriptor* desc);
 
   /// Update and print some statistics that can be used for performance debugging.
+  /// Thread-safe for read-only hash tables.
   std::string PrintStats() const;
 
   /// stl-like iterator interface.
@@ -788,6 +831,7 @@
         node_(NULL) { }
 
     /// Iterates to the next element. It should be called only if !AtEnd().
+    /// Thread-safe for read-only hash tables.
     void IR_ALWAYS_INLINE Next();
 
     /// Iterates to the next duplicate node. If the bucket does not have duplicates or
@@ -795,39 +839,48 @@
     /// Used when we want to iterate over all the duplicate nodes bypassing the Next()
     /// interface (e.g. in semi/outer joins without other_join_conjuncts, in order to
     /// iterate over all nodes of an unmatched bucket).
+    /// Thread-safe for read-only hash tables.
     void IR_ALWAYS_INLINE NextDuplicate();
 
     /// Iterates to the next element that does not have its matched flag set. Used in
     /// right-outer and full-outer joins.
+    /// Thread-safe for read-only hash tables.
     void IR_ALWAYS_INLINE NextUnmatched();
 
     /// Return the current row or tuple. Callers must check the iterator is not AtEnd()
     /// before calling them.  The returned row is owned by the iterator and valid until
     /// the next call to GetRow(). It is safe to advance the iterator.
+    /// Thread-safe for read-only hash tables.
     TupleRow* IR_ALWAYS_INLINE GetRow() const;
     Tuple* IR_ALWAYS_INLINE GetTuple() const;
 
     /// Set the current tuple for an empty bucket. Designed to be used with the iterator
     /// returned from FindBuildRowBucket() in the case when the value is not found.  It is
     /// not valid to call this function if the bucket already has an entry.
+    /// Not thread-safe.
     void SetTuple(Tuple* tuple, uint32_t hash);
 
     /// Sets as matched the Bucket or DuplicateNode currently pointed by the iterator,
     /// depending on whether the bucket has duplicates or not. The iterator cannot be
     /// AtEnd().
+    /// Not thread-safe.
     void SetMatched();
 
     /// Returns the 'matched' flag of the current Bucket or DuplicateNode, depending on
     /// whether the bucket has duplicates or not. It should be called only if !AtEnd().
+    /// Thread-safe for read-only hash tables.
     bool IsMatched() const;
 
     /// Resets everything but the pointer to the hash table.
+    /// Not thread-safe.
     void SetAtEnd();
 
     /// Returns true if this iterator is at the end, i.e. GetRow() cannot be called.
+    /// Thread-safe for read-only hash tables.
     bool ALWAYS_INLINE AtEnd() const { return bucket_idx_ == BUCKET_NOT_FOUND; }
 
     /// Prefetch the hash table bucket which the iterator is pointing to now.
+    /// Thread-safe for read-only hash tables.
     template<const bool READ>
     void IR_ALWAYS_INLINE PrefetchBucket();
 
@@ -868,9 +921,10 @@
 
   /// Performs the probing operation according to the probing algorithm (linear or
   /// quadratic. Returns one of the following:
-  /// (a) the index of the bucket that contains the entry that matches with the last row
-  ///     evaluated in 'ht_ctx'. If 'ht_ctx' is NULL then it does not check for row
-  ///     equality and returns the index of the first empty bucket.
+  /// (a) the index of the bucket that contains the entry matching 'hash' and, if
+  ///     COMPARE_ROW is true, also equals the last row evaluated in 'ht_ctx'.
+  ///     If COMPARE_ROW is false, returns the index of the first bucket with
+  ///     matching hash.
   /// (b) the index of the first empty bucket according to the probing algorithm (linear
   ///     or quadratic), if the entry is not in the hash table or 'ht_ctx' is NULL.
   /// (c) Iterator::BUCKET_NOT_FOUND if the probe was not successful, i.e. the maximum
@@ -889,15 +943,16 @@
   /// 'found' indicates that a bucket that contains an equal row is found.
   ///
   /// There are wrappers of this function that perform the Find and Insert logic.
-  template <bool INCLUSIVE_EQUALITY>
+  template <bool INCLUSIVE_EQUALITY, bool COMPARE_ROW>
   int64_t IR_ALWAYS_INLINE Probe(Bucket* buckets, int64_t num_buckets,
-      HashTableCtx* ht_ctx, uint32_t hash, bool* found);
+      HashTableCtx* __restrict__ ht_ctx, uint32_t hash, bool* found);
 
   /// Performs the insert logic. Returns the HtData* of the bucket or duplicate node
   /// where the data should be inserted. Returns NULL if the insert was not successful
   /// and either sets 'status' to OK if it failed because not enough reservation was
   /// available or the error if an error was encountered.
-  HtData* IR_ALWAYS_INLINE InsertInternal(HashTableCtx* ht_ctx, Status* status);
+  HtData* IR_ALWAYS_INLINE InsertInternal(
+      HashTableCtx* __restrict__ ht_ctx, Status* status);
 
   /// Updates 'bucket_idx' to the index of the next non-empty bucket. If the bucket has
   /// duplicates, 'node' will be pointing to the head of the linked list of duplicates.
@@ -906,7 +961,8 @@
   void NextFilledBucket(int64_t* bucket_idx, DuplicateNode** node);
 
   /// Resize the hash table to 'num_buckets'. 'got_memory' is false on OOM.
-  Status ResizeBuckets(int64_t num_buckets, const HashTableCtx* ht_ctx, bool* got_memory);
+  Status ResizeBuckets(
+      int64_t num_buckets, HashTableCtx* __restrict__ ht_ctx, bool* got_memory);
 
   /// Appends the DuplicateNode pointed by next_node_ to 'bucket' and moves the next_node_
   /// pointer to the next DuplicateNode in the page, updating the remaining node counter.
@@ -1016,19 +1072,6 @@
   /// (IMPALA-1488).
   bool has_matches_ = false;
 
-  /// The stats below can be used for debugging perf.
-  /// Number of Probe() calls that probe the hash table.
-  int64_t num_probes_ = 0;
-
-  /// Total distance traveled for each probe. That is the sum of the diff between the end
-  /// position of a probe (find/insert) and its start position
-  /// (hash & (num_buckets_ - 1)).
-  int64_t travel_length_ = 0;
-
-  /// The number of cases where we had to compare buckets with the same hash value, but
-  /// the row equality failed.
-  int64_t num_hash_collisions_ = 0;
-
   /// How many times this table has resized so far.
   int64_t num_resizes_ = 0;
 };
diff --git a/be/src/exec/hash-table.inline.h b/be/src/exec/hash-table.inline.h
index 2bcecef..74f2470 100644
--- a/be/src/exec/hash-table.inline.h
+++ b/be/src/exec/hash-table.inline.h
@@ -48,13 +48,14 @@
   DCHECK_LE(cur_expr_values_hash_ - expr_values_hash_array_.get(), capacity_);
 }
 
-template <bool INCLUSIVE_EQUALITY>
+template <bool INCLUSIVE_EQUALITY, bool COMPARE_ROW>
 inline int64_t HashTable::Probe(Bucket* buckets, int64_t num_buckets,
-    HashTableCtx* ht_ctx, uint32_t hash, bool* found) {
-  DCHECK(buckets != NULL);
+    HashTableCtx* __restrict__ ht_ctx, uint32_t hash, bool* found) {
+  DCHECK(ht_ctx != nullptr);
+  DCHECK(buckets != nullptr);
   DCHECK_GT(num_buckets, 0);
   *found = false;
-  ++num_probes_;
+  ++ht_ctx->num_probes_;
   int64_t bucket_idx = hash & (num_buckets - 1);
 
   // In case of linear probing it counts the total number of steps for statistics and
@@ -65,18 +66,18 @@
     Bucket* bucket = &buckets[bucket_idx];
     if (LIKELY(!bucket->filled)) return bucket_idx;
     if (hash == bucket->hash) {
-      if (ht_ctx != NULL &&
-          ht_ctx->Equals<INCLUSIVE_EQUALITY>(GetRow(bucket, ht_ctx->scratch_row_))) {
+      if (COMPARE_ROW
+          && ht_ctx->Equals<INCLUSIVE_EQUALITY>(GetRow(bucket, ht_ctx->scratch_row_))) {
         *found = true;
         return bucket_idx;
       }
       // Row equality failed, or not performed. This is a hash collision. Continue
       // searching.
-      ++num_hash_collisions_;
+      ++ht_ctx->num_hash_collisions_;
     }
     // Move to the next bucket.
     ++step;
-    ++travel_length_;
+    ++ht_ctx->travel_length_;
     if (quadratic_probing()) {
       // The i-th probe location is idx = (hash + (step * (step + 1)) / 2) mod num_buckets.
       // This gives num_buckets unique idxs (between 0 and N-1) when num_buckets is a power
@@ -86,16 +87,17 @@
       bucket_idx = (bucket_idx + 1) & (num_buckets - 1);
     }
   } while (LIKELY(step < num_buckets));
+
   DCHECK_EQ(num_filled_buckets_, num_buckets) << "Probing of a non-full table "
       << "failed: " << quadratic_probing() << " " << hash;
   return Iterator::BUCKET_NOT_FOUND;
 }
 
 inline HashTable::HtData* HashTable::InsertInternal(
-    HashTableCtx* ht_ctx, Status* status) {
+    HashTableCtx* __restrict__ ht_ctx, Status* status) {
   bool found = false;
   uint32_t hash = ht_ctx->expr_values_cache()->CurExprValuesHash();
-  int64_t bucket_idx = Probe<true>(buckets_, num_buckets_, ht_ctx, hash, &found);
+  int64_t bucket_idx = Probe<true, true>(buckets_, num_buckets_, ht_ctx, hash, &found);
   DCHECK_NE(bucket_idx, Iterator::BUCKET_NOT_FOUND);
   if (found) {
     // We need to insert a duplicate node, note that this may fail to allocate memory.
@@ -108,7 +110,7 @@
   }
 }
 
-inline bool HashTable::Insert(HashTableCtx* ht_ctx,
+inline bool HashTable::Insert(HashTableCtx* __restrict__ ht_ctx,
     BufferedTupleStream::FlatRowPtr flat_row, TupleRow* row, Status* status) {
   HtData* htdata = InsertInternal(ht_ctx, status);
   // If successful insert, update the contents of the newly inserted entry with 'idx'.
@@ -134,10 +136,10 @@
   __builtin_prefetch(&buckets_[bucket_idx], READ ? 0 : 1, 1);
 }
 
-inline HashTable::Iterator HashTable::FindProbeRow(HashTableCtx* ht_ctx) {
+inline HashTable::Iterator HashTable::FindProbeRow(HashTableCtx* __restrict__ ht_ctx) {
   bool found = false;
   uint32_t hash = ht_ctx->expr_values_cache()->CurExprValuesHash();
-  int64_t bucket_idx = Probe<false>(buckets_, num_buckets_, ht_ctx, hash, &found);
+  int64_t bucket_idx = Probe<false, true>(buckets_, num_buckets_, ht_ctx, hash, &found);
   if (found) {
     return Iterator(this, ht_ctx->scratch_row(), bucket_idx,
         stores_duplicates() ? buckets_[bucket_idx].bucketData.duplicates : NULL);
@@ -147,9 +149,9 @@
 
 // TODO: support lazy evaluation like HashTable::Insert().
 inline HashTable::Iterator HashTable::FindBuildRowBucket(
-    HashTableCtx* ht_ctx, bool* found) {
+    HashTableCtx* __restrict__ ht_ctx, bool* found) {
   uint32_t hash = ht_ctx->expr_values_cache()->CurExprValuesHash();
-  int64_t bucket_idx = Probe<true>(buckets_, num_buckets_, ht_ctx, hash, found);
+  int64_t bucket_idx = Probe<true, true>(buckets_, num_buckets_, ht_ctx, hash, found);
   DuplicateNode* duplicates = NULL;
   if (stores_duplicates() && LIKELY(bucket_idx != Iterator::BUCKET_NOT_FOUND)) {
     duplicates = buckets_[bucket_idx].bucketData.duplicates;
diff --git a/be/src/exec/partitioned-hash-join-builder.cc b/be/src/exec/partitioned-hash-join-builder.cc
index 6df580d..c6752d6 100644
--- a/be/src/exec/partitioned-hash-join-builder.cc
+++ b/be/src/exec/partitioned-hash-join-builder.cc
@@ -288,8 +288,11 @@
 void PhjBuilder::Close(RuntimeState* state) {
   if (closed_) return;
   CloseAndDeletePartitions(nullptr);
-  if (ht_ctx_ != nullptr) ht_ctx_->Close(state);
-  ht_ctx_.reset();
+  if (ht_ctx_ != nullptr) {
+    ht_ctx_->StatsCountersAdd(ht_stats_profile_.get());
+    ht_ctx_->Close(state);
+    ht_ctx_.reset();
+  }
   for (const FilterContext& ctx : filter_ctxs_) {
     if (ctx.expr_eval != nullptr) ctx.expr_eval->Close(state);
   }
diff --git a/be/src/exec/partitioned-hash-join-builder.h b/be/src/exec/partitioned-hash-join-builder.h
index 0dec0b8..fc8d075 100644
--- a/be/src/exec/partitioned-hash-join-builder.h
+++ b/be/src/exec/partitioned-hash-join-builder.h
@@ -316,6 +316,10 @@
   /// TODO: IMPALA-9176: improve the encapsulation of the null-aware partition.
   inline Partition* null_aware_partition() const { return null_aware_partition_.get(); }
 
+  /// TODO: IMPALA-9156: document thread safety for accessing this from
+  /// multiple PartitionedHashJoinNodes.
+  HashTableStatsProfile* ht_stats_profile() const { return ht_stats_profile_.get(); }
+
   std::string DebugString() const;
 
   /// A partition containing a subset of build rows.
diff --git a/be/src/exec/partitioned-hash-join-node.cc b/be/src/exec/partitioned-hash-join-node.cc
index a9fe4e9..32b4cd5 100644
--- a/be/src/exec/partitioned-hash-join-node.cc
+++ b/be/src/exec/partitioned-hash-join-node.cc
@@ -253,8 +253,11 @@
 
 void PartitionedHashJoinNode::Close(RuntimeState* state) {
   if (is_closed()) return;
-  if (ht_ctx_ != nullptr) ht_ctx_->Close(state);
-  ht_ctx_.reset();
+  if (ht_ctx_ != nullptr) {
+    ht_ctx_->StatsCountersAdd(builder_->ht_stats_profile());
+    ht_ctx_->Close(state);
+    ht_ctx_.reset();
+  }
   output_unmatched_batch_.reset();
   output_unmatched_batch_iter_.reset();
   CloseAndDeletePartitions(nullptr);