diff --git a/be/src/exec/partitioned-hash-join-node-ir.cc b/be/src/exec/partitioned-hash-join-node-ir.cc
index a2a63c1..f316928 100644
--- a/be/src/exec/partitioned-hash-join-node-ir.cc
+++ b/be/src/exec/partitioned-hash-join-node-ir.cc
@@ -365,8 +365,9 @@
 template <int const JoinOp>
 int PartitionedHashJoinNode::ProcessProbeBatch(TPrefetchMode::type prefetch_mode,
     RowBatch* out_batch, HashTableCtx* __restrict__ ht_ctx, Status* __restrict__ status) {
-  DCHECK(state_ == PARTITIONING_PROBE || state_ == PROBING_SPILLED_PARTITION
-      || state_ == REPARTITIONING_PROBE);
+  DCHECK(state_ == HashJoinState::PARTITIONING_PROBE
+      || state_ == HashJoinState::PROBING_SPILLED_PARTITION
+      || state_ == HashJoinState::REPARTITIONING_PROBE);
   ScalarExprEvaluator* const* other_join_conjunct_evals =
       other_join_conjunct_evals_.data();
   const int num_other_join_conjuncts = other_join_conjunct_evals_.size();
diff --git a/be/src/exec/partitioned-hash-join-node.cc b/be/src/exec/partitioned-hash-join-node.cc
index fd9703f..880a528 100644
--- a/be/src/exec/partitioned-hash-join-node.cc
+++ b/be/src/exec/partitioned-hash-join-node.cc
@@ -51,14 +51,7 @@
 PartitionedHashJoinNode::PartitionedHashJoinNode(
     ObjectPool* pool, const TPlanNode& tnode, const DescriptorTbl& descs)
   : BlockingJoinNode(
-        "PartitionedHashJoinNode", tnode.hash_join_node.join_op, pool, tnode, descs),
-    num_probe_rows_partitioned_(NULL),
-    null_aware_eval_timer_(NULL),
-    state_(PARTITIONING_BUILD),
-    output_null_aware_probe_rows_running_(false),
-    null_probe_output_idx_(-1),
-    process_probe_batch_fn_(NULL),
-    process_probe_batch_fn_level0_(NULL) {
+        "PartitionedHashJoinNode", tnode.hash_join_node.join_op, pool, tnode, descs) {
   memset(hash_tbls_, 0, sizeof(HashTable*) * PARTITION_FANOUT);
 }
 
@@ -177,9 +170,10 @@
   RETURN_IF_ERROR(BlockingJoinNode::ProcessBuildInputAndOpenProbe(state, builder_.get()));
   RETURN_IF_ERROR(PrepareForProbe());
 
-  UpdateState(PARTITIONING_PROBE);
+  UpdateState(HashJoinState::PARTITIONING_PROBE);
   RETURN_IF_ERROR(BlockingJoinNode::GetFirstProbeRow(state));
   ResetForProbe();
+  probe_state_ = ProbeState::PROBING_IN_BATCH;
   DCHECK(null_aware_probe_partition_ == NULL
       || join_op_ == TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN);
   return Status::OK();
@@ -203,9 +197,8 @@
   if (join_op_ == TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN) {
     null_probe_output_idx_ = -1;
     matched_null_probe_.clear();
-    output_null_aware_probe_rows_running_ = false;
   }
-  state_ = PARTITIONING_BUILD;
+  state_ = HashJoinState::PARTITIONING_BUILD;
   ht_ctx_->set_level(0);
   CloseAndDeletePartitions(row_batch);
   builder_->Reset(IsLeftSemiJoin(join_op_) ? nullptr : row_batch);
@@ -250,7 +243,6 @@
   if (is_closed()) return;
   if (ht_ctx_ != nullptr) ht_ctx_->Close(state);
   ht_ctx_.reset();
-  output_null_aware_probe_rows_running_ = false;
   output_unmatched_batch_.reset();
   output_unmatched_batch_iter_.reset();
   CloseAndDeletePartitions(nullptr);
@@ -293,7 +285,30 @@
 
 Status PartitionedHashJoinNode::NextProbeRowBatch(
     RuntimeState* state, RowBatch* out_batch) {
-  DCHECK_EQ(state_, PARTITIONING_PROBE);
+  DCHECK_ENUM_EQ(probe_state_, ProbeState::PROBING_END_BATCH);
+  DCHECK(probe_batch_pos_ == probe_batch_->num_rows() || probe_batch_pos_ == -1);
+  if (state_ == HashJoinState::PARTITIONING_PROBE) {
+    DCHECK(input_partition_ == nullptr);
+    RETURN_IF_ERROR(NextProbeRowBatchFromChild(state, out_batch));
+  } else {
+    DCHECK(state_ == HashJoinState::REPARTITIONING_PROBE
+        || state_ == HashJoinState::PROBING_SPILLED_PARTITION)
+        << PrintState();
+    DCHECK(probe_side_eos_);
+    DCHECK(input_partition_ != nullptr);
+    RETURN_IF_ERROR(NextSpilledProbeRowBatch(state, out_batch));
+  }
+  // Free expr result allocations of the probe side expressions only after
+  // ExprValuesCache has been reset.
+  DCHECK(ht_ctx_->expr_values_cache()->AtEnd());
+  probe_expr_results_pool_->Clear();
+  return Status::OK();
+}
+
+Status PartitionedHashJoinNode::NextProbeRowBatchFromChild(
+    RuntimeState* state, RowBatch* out_batch) {
+  DCHECK_ENUM_EQ(state_, HashJoinState::PARTITIONING_PROBE);
+  DCHECK_ENUM_EQ(probe_state_, ProbeState::PROBING_END_BATCH);
   DCHECK(probe_batch_pos_ == probe_batch_->num_rows() || probe_batch_pos_ == -1);
   do {
     // Loop until we find a non-empty row batch.
@@ -319,7 +334,9 @@
 Status PartitionedHashJoinNode::NextSpilledProbeRowBatch(
     RuntimeState* state, RowBatch* out_batch) {
   DCHECK(input_partition_ != NULL);
-  DCHECK(state_ == PROBING_SPILLED_PARTITION || state_ == REPARTITIONING_PROBE);
+  DCHECK(state_ == HashJoinState::PROBING_SPILLED_PARTITION
+      || state_ == HashJoinState::REPARTITIONING_PROBE);
+  DCHECK_ENUM_EQ(probe_state_, ProbeState::PROBING_END_BATCH);
   probe_batch_->TransferResourceOwnership(out_batch);
   if (out_batch->AtCapacity()) {
     // The out_batch has resources associated with it that will be recycled on the
@@ -336,7 +353,7 @@
     ResetForProbe();
   } else {
     // Finished processing spilled probe rows from this partition.
-    if (state_ == PROBING_SPILLED_PARTITION
+    if (state_ == HashJoinState::PROBING_SPILLED_PARTITION
         && NeedToProcessUnmatchedBuildRows(join_op_)) {
       // If the build partition was in memory, we are done probing this partition.
       // In case of right-outer, right-anti and full-outer joins, we move this partition
@@ -368,16 +385,12 @@
 
 // TODO: refactor this method to better separate the logic operating on the builder
 // vs probe data structures.
-Status PartitionedHashJoinNode::PrepareSpilledPartitionForProbe(
-    RuntimeState* state, bool* got_partition) {
+Status PartitionedHashJoinNode::PrepareSpilledPartitionForProbe() {
   VLOG(2) << "PrepareSpilledPartitionForProbe\n" << NodeDebugString();
   DCHECK(input_partition_ == NULL);
   DCHECK_EQ(builder_->num_hash_partitions(), 0);
   DCHECK(probe_hash_partitions_.empty());
-  if (spilled_partitions_.empty()) {
-    *got_partition = false;
-    return Status::OK();
-  }
+  DCHECK(!spilled_partitions_.empty());
 
   // TODO: the builder should choose the spilled partition to process.
   input_partition_ = std::move(spilled_partitions_.front());
@@ -397,8 +410,7 @@
           runtime_state_, Substitute(PREPARE_FOR_READ_FAILED_ERROR_MSG, id_));
     }
 
-    *got_partition = true;
-    UpdateState(PROBING_SPILLED_PARTITION);
+    UpdateState(HashJoinState::PROBING_SPILLED_PARTITION);
     COUNTER_ADD(num_hash_table_builds_skipped_, 1);
     return Status::OK();
   }
@@ -415,7 +427,7 @@
 
   if (!built) {
     // This build partition still does not fit in memory, repartition.
-    UpdateState(REPARTITIONING_BUILD);
+    UpdateState(HashJoinState::REPARTITIONING_BUILD);
 
     int next_partition_level = build_partition->level() + 1;
     if (UNLIKELY(next_partition_level >= MAX_PARTITION_DEPTH)) {
@@ -446,7 +458,7 @@
     }
 
     RETURN_IF_ERROR(PrepareForProbe());
-    UpdateState(REPARTITIONING_PROBE);
+    UpdateState(HashJoinState::REPARTITIONING_PROBE);
   } else {
     DCHECK(!input_partition_->build_partition()->is_spilled());
     DCHECK(input_partition_->build_partition()->hash_tbl() != NULL);
@@ -455,11 +467,41 @@
     for (int i = 0; i < PARTITION_FANOUT; ++i) {
       hash_tbls_[i] = input_partition_->build_partition()->hash_tbl();
     }
-    UpdateState(PROBING_SPILLED_PARTITION);
+    UpdateState(HashJoinState::PROBING_SPILLED_PARTITION);
   }
 
   COUNTER_ADD(num_probe_rows_partitioned_, input_partition_->probe_rows()->num_rows());
-  *got_partition = true;
+  return Status::OK();
+}
+
+Status PartitionedHashJoinNode::ProcessProbeBatch(RowBatch* out_batch) {
+  DCHECK_ENUM_EQ(probe_state_, ProbeState::PROBING_IN_BATCH);
+  DCHECK_NE(probe_batch_pos_, -1);
+  // Putting SCOPED_TIMER in the IR version of ProcessProbeBatch() causes weird exception
+  // handling IR in the xcompiled function, so call it here instead.
+  int rows_added = 0;
+  Status status;
+  TPrefetchMode::type prefetch_mode = runtime_state_->query_options().prefetch_mode;
+  SCOPED_TIMER(probe_timer_);
+  if (process_probe_batch_fn_ == NULL) {
+    rows_added = ProcessProbeBatch(
+        join_op_, prefetch_mode, out_batch, ht_ctx_.get(), &status);
+  } else {
+    DCHECK(process_probe_batch_fn_level0_ != NULL);
+    if (ht_ctx_->level() == 0) {
+      rows_added = process_probe_batch_fn_level0_(
+          this, prefetch_mode, out_batch, ht_ctx_.get(), &status);
+    } else {
+      rows_added = process_probe_batch_fn_(
+          this, prefetch_mode, out_batch, ht_ctx_.get(), &status);
+    }
+  }
+  if (UNLIKELY(rows_added < 0)) {
+    DCHECK(!status.ok());
+    return status;
+  }
+  DCHECK(status.ok());
+  out_batch->CommitRows(rows_added);
   return Status::OK();
 }
 
@@ -513,133 +555,113 @@
   // which can happen in a subplan.
   int num_rows_before = out_batch->num_rows();
 
-  while (true) {
-    DCHECK(!*eos);
+  // This loop executes the 'probe_state_' state machine until either a full batch is
+  // produced, resources are attached to 'out_batch' that require flushing, or eos
+  // is reached (i.e. all rows are returned). The next call into GetNext() will resume
+  // execution of the state machine where the current call into GetNext() left off.
+  // See the definition of ProbeState for description of the state machine and states.
+  do {
     DCHECK(status.ok());
-    DCHECK_NE(state_, PARTITIONING_BUILD) << "Should not be in GetNext()";
+    DCHECK(state_ != HashJoinState::PARTITIONING_BUILD)
+        << "Should not be in GetNext() " << static_cast<int>(state_);
     RETURN_IF_CANCELLED(state);
     RETURN_IF_ERROR(QueryMaintenance(state));
+    switch (probe_state_) {
+      case ProbeState::PROBING_IN_BATCH: {
+        // Finish processing rows in the current probe batch.
+        RETURN_IF_ERROR(ProcessProbeBatch(out_batch));
+        DCHECK(out_batch->AtCapacity() || probe_batch_pos_ == probe_batch_->num_rows());
+        if (probe_batch_pos_ == probe_batch_->num_rows()
+            && current_probe_row_ == nullptr) {
+          probe_state_ = ProbeState::PROBING_END_BATCH;
+        }
+        break;
+      }
+      case ProbeState::PROBING_END_BATCH: {
+        // Try to get the next row batch from the current probe input.
+        RETURN_IF_ERROR(NextProbeRowBatch(state, out_batch));
 
-    if (!output_build_partitions_.empty()) {
-      DCHECK(NeedToProcessUnmatchedBuildRows(join_op_));
-
-      // Flush the remaining unmatched build rows of any partitions we are done
-      // processing before moving onto the next partition.
-      RETURN_IF_ERROR(OutputUnmatchedBuild(out_batch));
-      if (out_batch->AtCapacity()) break;
-
-      // Stopped before batch was at capacity: - we must have finished outputting
-      // unmatched build rows.
-      DCHECK(output_build_partitions_.empty());
-      DCHECK_EQ(builder_->num_hash_partitions(), 0);
-      DCHECK(probe_hash_partitions_.empty());
-    }
-
-    if (join_op_ == TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN) {
-      // In this case, we want to output rows from the null aware partition.
-      if (builder_->null_aware_partition() == NULL) {
-        DCHECK(null_aware_probe_partition_ == NULL);
+        if (probe_batch_pos_ == 0) {
+          // Got a batch, need to process it.
+          probe_state_ = ProbeState::PROBING_IN_BATCH;
+        } else if (probe_side_eos_ && input_partition_ == nullptr) {
+          DCHECK_EQ(probe_batch_pos_, -1);
+          // Finished processing all the probe rows for the current hash partitions.
+          // There may be some partitions that need to outpt their unmatched build rows.
+          RETURN_IF_ERROR(DoneProbing(state, out_batch));
+          probe_state_ = output_build_partitions_.empty() ?
+              ProbeState::PROBE_COMPLETE :
+              ProbeState::OUTPUTTING_UNMATCHED;
+        } else {
+          // Got an empty batch with resources that we need to flush before getting the
+          // next batch.
+          DCHECK_EQ(probe_batch_pos_, -1);
+        }
+        break;
+      }
+      case ProbeState::OUTPUTTING_UNMATCHED: {
+        DCHECK(!output_build_partitions_.empty());
+        DCHECK_EQ(builder_->num_hash_partitions(), 0);
+        DCHECK(probe_hash_partitions_.empty());
+        DCHECK(NeedToProcessUnmatchedBuildRows(join_op_));
+        // Output the remaining batch of build rows from the current partition.
+        RETURN_IF_ERROR(OutputUnmatchedBuild(out_batch));
+        DCHECK(out_batch->AtCapacity() || output_build_partitions_.empty());
+        if (output_build_partitions_.empty()) probe_state_ = ProbeState::PROBE_COMPLETE;
+        break;
+      }
+      case ProbeState::PROBE_COMPLETE: {
+        if (!spilled_partitions_.empty()) {
+          // Move to the next spilled partition.
+          RETURN_IF_ERROR(PrepareSpilledPartitionForProbe());
+          probe_state_ = ProbeState::PROBING_END_BATCH;
+        } else if (join_op_ == TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN
+            && builder_->null_aware_partition() != nullptr) {
+          // Null aware anti join outputs additional rows after all the probe input,
+          // including spilled partitions, is processed.
+          bool has_null_aware_rows;
+          RETURN_IF_ERROR(PrepareNullAwarePartition(&has_null_aware_rows));
+          probe_state_ = has_null_aware_rows ? ProbeState::OUTPUTTING_NULL_AWARE :
+                                               ProbeState::OUTPUTTING_NULL_PROBE;
+        } else {
+          // No more rows to output from GetNext().
+          probe_state_ = ProbeState::EOS;
+        }
+        break;
+      }
+      case ProbeState::OUTPUTTING_NULL_AWARE: {
+        DCHECK_ENUM_EQ(join_op_, TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN);
+        DCHECK(builder_->null_aware_partition() != nullptr);
+        DCHECK(null_aware_probe_partition_ != nullptr);
+        bool napr_eos;
+        RETURN_IF_ERROR(OutputNullAwareProbeRows(state, out_batch, &napr_eos));
+        if (napr_eos) probe_state_ = ProbeState::OUTPUTTING_NULL_PROBE;
+        break;
+      }
+      case ProbeState::OUTPUTTING_NULL_PROBE: {
+        DCHECK_ENUM_EQ(join_op_, TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN);
+        DCHECK(builder_->null_aware_partition() != nullptr);
+        DCHECK_GE(null_probe_output_idx_, 0);
+        bool nanp_done;
+        RETURN_IF_ERROR(OutputNullAwareNullProbe(state, out_batch, &nanp_done));
+        if (nanp_done) probe_state_ = ProbeState::EOS;
+        break;
+      }
+      case ProbeState::EOS: {
+        // Ensure that all potential sources of output rows are exhausted.
+        DCHECK(probe_side_eos_);
+        DCHECK(output_build_partitions_.empty());
+        DCHECK(spilled_partitions_.empty());
+        DCHECK(builder_->null_aware_partition() == nullptr);
+        DCHECK(null_aware_probe_partition_ == nullptr);
         *eos = true;
         break;
       }
-
-      if (null_probe_output_idx_ >= 0) {
-        RETURN_IF_ERROR(OutputNullAwareNullProbe(state, out_batch));
-        if (out_batch->AtCapacity()) break;
-        continue;
-      }
-
-      if (output_null_aware_probe_rows_running_) {
-        RETURN_IF_ERROR(OutputNullAwareProbeRows(state, out_batch));
-        if (out_batch->AtCapacity()) break;
-        continue;
-      }
+      default:
+        DCHECK(false) << "invalid probe_state_" << static_cast<int>(probe_state_);
+        break;
     }
-
-    // Finish processing rows in the current probe batch.
-    if (probe_batch_pos_ != -1) {
-      // Putting SCOPED_TIMER in ProcessProbeBatch() causes weird exception handling IR
-      // in the xcompiled function, so call it here instead.
-      int rows_added = 0;
-      TPrefetchMode::type prefetch_mode = state->query_options().prefetch_mode;
-      SCOPED_TIMER(probe_timer_);
-      if (process_probe_batch_fn_ == NULL) {
-        rows_added = ProcessProbeBatch(join_op_, prefetch_mode, out_batch, ht_ctx_.get(),
-            &status);
-      } else {
-        DCHECK(process_probe_batch_fn_level0_ != NULL);
-        if (ht_ctx_->level() == 0) {
-          rows_added = process_probe_batch_fn_level0_(this, prefetch_mode, out_batch,
-              ht_ctx_.get(), &status);
-        } else {
-          rows_added = process_probe_batch_fn_(this, prefetch_mode, out_batch,
-              ht_ctx_.get(), &status);
-        }
-      }
-      if (UNLIKELY(rows_added < 0)) {
-        DCHECK(!status.ok());
-        return status;
-      }
-      DCHECK(status.ok());
-      out_batch->CommitRows(rows_added);
-      if (out_batch->AtCapacity()) break;
-
-      DCHECK(current_probe_row_ == NULL);
-    }
-
-    // Try to continue from the current probe side input.
-    if (state_ == PARTITIONING_PROBE) {
-      DCHECK(input_partition_ == NULL);
-      RETURN_IF_ERROR(NextProbeRowBatch(state, out_batch));
-    } else {
-      DCHECK(state_ == REPARTITIONING_PROBE || state_ == PROBING_SPILLED_PARTITION)
-          << state_;
-      DCHECK(probe_side_eos_);
-      if (input_partition_ != NULL) {
-        RETURN_IF_ERROR(NextSpilledProbeRowBatch(state, out_batch));
-      }
-    }
-    // Free expr result allocations of the probe side expressions only after
-    // ExprValuesCache has been reset.
-    DCHECK(ht_ctx_->expr_values_cache()->AtEnd());
-    probe_expr_results_pool_->Clear();
-
-    // We want to return as soon as we have attached a tuple stream to the out_batch
-    // (before preparing a new partition). The attached tuple stream will be recycled
-    // by the caller, freeing up more memory when we prepare the next partition.
-    if (out_batch->AtCapacity()) break;
-
-    // Got a batch, just keep going.
-    if (probe_batch_pos_ == 0) continue;
-    DCHECK_EQ(probe_batch_pos_, -1);
-
-    // Finished up all probe rows for 'hash_partitions_'. We may have already cleaned up
-    // the hash partitions, e.g. if we had to output some unmatched build rows below.
-    if (builder_->num_hash_partitions() != 0) {
-      RETURN_IF_ERROR(DoneProbing(state, out_batch));
-      if (out_batch->AtCapacity()) break;
-    }
-
-    // There are some partitions that need to flush their unmatched build rows.
-    if (!output_build_partitions_.empty()) continue;
-
-    // We get this far if there is nothing left to return from the current partition.
-    // Move to the next spilled partition.
-    bool got_partition;
-    RETURN_IF_ERROR(PrepareSpilledPartitionForProbe(state, &got_partition));
-    if (got_partition) continue; // Probe the spilled partition.
-
-    if (join_op_ == TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN) {
-      // Prepare the null-aware partitions, then resume at the top of the loop to output
-      // the rows.
-      RETURN_IF_ERROR(PrepareNullAwarePartition());
-      continue;
-    }
-    DCHECK(builder_->null_aware_partition() == NULL);
-
-    *eos = true;
-    break;
-  }
+  } while (!out_batch->AtCapacity() && !*eos);
 
   int num_rows_added = out_batch->num_rows() - num_rows_before;
   DCHECK_GE(num_rows_added, 0);
@@ -662,6 +684,7 @@
   SCOPED_TIMER(probe_timer_);
   DCHECK(NeedToProcessUnmatchedBuildRows(join_op_));
   DCHECK(!output_build_partitions_.empty());
+  DCHECK_ENUM_EQ(probe_state_, ProbeState::OUTPUTTING_UNMATCHED);
 
   if (output_unmatched_batch_iter_.get() != NULL) {
     // There were no probe rows so we skipped building the hash table. In this case, all
@@ -676,6 +699,7 @@
 }
 
 Status PartitionedHashJoinNode::OutputAllBuild(RowBatch* out_batch) {
+  DCHECK_ENUM_EQ(probe_state_, ProbeState::OUTPUTTING_UNMATCHED);
   // This will only be called for partitions that are added to 'output_build_partitions_'
   // in NextSpilledProbeRowBatch(), which adds one partition that is then processed until
   // it is done by the loop in GetNext(). So, there must be exactly one partition in
@@ -727,6 +751,7 @@
 }
 
 void PartitionedHashJoinNode::OutputUnmatchedBuildFromHashTable(RowBatch* out_batch) {
+  DCHECK_ENUM_EQ(probe_state_, ProbeState::OUTPUTTING_UNMATCHED);
   ScalarExprEvaluator* const* conjunct_evals = conjunct_evals_.data();
   const int num_conjuncts = conjuncts_.size();
   RowBatch::Iterator out_batch_iterator(out_batch, out_batch->num_rows());
@@ -786,12 +811,13 @@
   return Status::OK();
 }
 
-Status PartitionedHashJoinNode::OutputNullAwareNullProbe(RuntimeState* state,
-    RowBatch* out_batch) {
+Status PartitionedHashJoinNode::OutputNullAwareNullProbe(
+    RuntimeState* state, RowBatch* out_batch, bool* done) {
+  DCHECK_ENUM_EQ(probe_state_, ProbeState::OUTPUTTING_NULL_PROBE);
   DCHECK(builder_->null_aware_partition() != NULL);
   DCHECK(null_aware_probe_partition_ != NULL);
-  DCHECK(!output_null_aware_probe_rows_running_);
   DCHECK_NE(probe_batch_pos_, -1);
+  *done = false;
 
   if (probe_batch_pos_ == probe_batch_->num_rows()) {
     probe_batch_pos_ = 0;
@@ -807,6 +833,7 @@
       // Flush out the resources to free up memory.
       null_probe_rows_->Close(out_batch, RowBatch::FlushMode::FLUSH_RESOURCES);
       null_probe_rows_.reset();
+      *done = true;
       return Status::OK();
     }
   }
@@ -876,10 +903,10 @@
   return Status::OK();
 }
 
-Status PartitionedHashJoinNode::PrepareNullAwarePartition() {
+Status PartitionedHashJoinNode::PrepareNullAwarePartition(bool* has_null_aware_rows) {
+  DCHECK_ENUM_EQ(probe_state_, ProbeState::PROBE_COMPLETE);
   DCHECK(builder_->null_aware_partition() != NULL);
   DCHECK(null_aware_probe_partition_ != NULL);
-  DCHECK(!output_null_aware_probe_rows_running_);
   DCHECK_EQ(probe_batch_pos_, -1);
   DCHECK_EQ(probe_batch_->num_rows(), 0);
 
@@ -890,12 +917,11 @@
     // There were no build rows. Nothing to do. Just prepare to output the null
     // probe rows.
     DCHECK_EQ(probe_stream->num_rows(), 0);
-    output_null_aware_probe_rows_running_ = false;
     RETURN_IF_ERROR(PrepareNullAwareNullProbe());
+    *has_null_aware_rows = false;
     return Status::OK();
   }
 
-  output_null_aware_probe_rows_running_ = true;
   bool pinned;
   RETURN_IF_ERROR(build_stream->PinStream(&pinned));
   if (!pinned) return NullAwareAntiJoinError(build_stream);
@@ -908,15 +934,16 @@
         runtime_state_, Substitute(PREPARE_FOR_READ_FAILED_ERROR_MSG, id_));
   }
   probe_batch_pos_ = 0;
+  *has_null_aware_rows = true;
   return Status::OK();
 }
 
-Status PartitionedHashJoinNode::OutputNullAwareProbeRows(RuntimeState* state,
-    RowBatch* out_batch) {
+Status PartitionedHashJoinNode::OutputNullAwareProbeRows(
+    RuntimeState* state, RowBatch* out_batch, bool* done) {
+  DCHECK_ENUM_EQ(probe_state_, ProbeState::OUTPUTTING_NULL_AWARE);
   DCHECK(builder_->null_aware_partition() != NULL);
   DCHECK(null_aware_probe_partition_ != NULL);
-  DCHECK(output_null_aware_probe_rows_running_);
-
+  *done = false;
   ScalarExprEvaluator* const* join_conjunct_evals = other_join_conjunct_evals_.data();
   int num_join_conjuncts = other_join_conjuncts_.size();
   DCHECK(probe_batch_ != NULL);
@@ -931,11 +958,11 @@
     bool eos;
     RETURN_IF_ERROR(probe_stream->GetNext(probe_batch_.get(), &eos));
 
-    if (probe_batch_->num_rows() == 0) {
-      RETURN_IF_ERROR(EvaluateNullProbe(
-            state, builder_->null_aware_partition()->build_rows()));
-      output_null_aware_probe_rows_running_ = false;
+    if (probe_batch_->num_rows() == 0 && eos) {
+      RETURN_IF_ERROR(
+          EvaluateNullProbe(state, builder_->null_aware_partition()->build_rows()));
       RETURN_IF_ERROR(PrepareNullAwareNullProbe());
+      *done = true;
       return Status::OK();
     }
   }
@@ -1101,10 +1128,17 @@
 }
 
 Status PartitionedHashJoinNode::DoneProbing(RuntimeState* state, RowBatch* batch) {
+  DCHECK_ENUM_EQ(probe_state_, ProbeState::PROBING_END_BATCH);
   DCHECK_EQ(probe_batch_pos_, -1);
   // At this point all the rows have been read from the probe side for all partitions in
   // hash_partitions_.
   VLOG(2) << "Probe Side Consumed\n" << NodeDebugString();
+  if (builder_->num_hash_partitions() == 0) {
+    // No hash partitions, so no cleanup required. This can only happen when we are
+    // processing a single spilled partition.
+    DCHECK_ENUM_EQ(state_, HashJoinState::PROBING_SPILLED_PARTITION);
+    return Status::OK();
+  }
 
   // Walk the partitions that had hash tables built for the probe phase and close them.
   // In the case of right outer and full outer joins, instead of closing those partitions,
@@ -1170,15 +1204,20 @@
 void PartitionedHashJoinNode::UpdateState(HashJoinState next_state) {
   // Validate the state transition.
   switch (state_) {
-    case PARTITIONING_BUILD: DCHECK_EQ(next_state, PARTITIONING_PROBE); break;
-    case PARTITIONING_PROBE:
-    case REPARTITIONING_PROBE:
-    case PROBING_SPILLED_PARTITION:
-      DCHECK(
-          next_state == REPARTITIONING_BUILD || next_state == PROBING_SPILLED_PARTITION);
+    case HashJoinState::PARTITIONING_BUILD:
+      DCHECK_ENUM_EQ(next_state, HashJoinState::PARTITIONING_PROBE);
       break;
-    case REPARTITIONING_BUILD: DCHECK_EQ(next_state, REPARTITIONING_PROBE); break;
-    default: DCHECK(false) << "Invalid state " << state_;
+    case HashJoinState::PARTITIONING_PROBE:
+    case HashJoinState::REPARTITIONING_PROBE:
+    case HashJoinState::PROBING_SPILLED_PARTITION:
+      DCHECK(next_state == HashJoinState::REPARTITIONING_BUILD
+          || next_state == HashJoinState::PROBING_SPILLED_PARTITION);
+      break;
+    case HashJoinState::REPARTITIONING_BUILD:
+      DCHECK_ENUM_EQ(next_state, HashJoinState::REPARTITIONING_PROBE);
+      break;
+    default:
+      DCHECK(false) << "Invalid state " << static_cast<int>(state_);
   }
   state_ = next_state;
   VLOG(2) << "Transitioned State:" << endl << NodeDebugString();
@@ -1186,11 +1225,16 @@
 
 string PartitionedHashJoinNode::PrintState() const {
   switch (state_) {
-    case PARTITIONING_BUILD: return "PartitioningBuild";
-    case PARTITIONING_PROBE: return "PartitioningProbe";
-    case PROBING_SPILLED_PARTITION: return "ProbingSpilledPartition";
-    case REPARTITIONING_BUILD: return "RepartitioningBuild";
-    case REPARTITIONING_PROBE: return "RepartitioningProbe";
+    case HashJoinState::PARTITIONING_BUILD:
+      return "PartitioningBuild";
+    case HashJoinState::PARTITIONING_PROBE:
+      return "PartitioningProbe";
+    case HashJoinState::PROBING_SPILLED_PARTITION:
+      return "ProbingSpilledPartition";
+    case HashJoinState::REPARTITIONING_BUILD:
+      return "RepartitioningBuild";
+    case HashJoinState::REPARTITIONING_PROBE:
+      return "RepartitioningProbe";
     default: DCHECK(false);
   }
   return "";
diff --git a/be/src/exec/partitioned-hash-join-node.h b/be/src/exec/partitioned-hash-join-node.h
index b771e32..126beb1 100644
--- a/be/src/exec/partitioned-hash-join-node.h
+++ b/be/src/exec/partitioned-hash-join-node.h
@@ -83,6 +83,9 @@
 ///      either PROBING_SPILLED_PARTITION or REPARTITIONING_BUILD, depending on whether
 ///      the spilled partition's hash table fits in memory or not.
 ///
+///      This phase has sub-states (see ProbeState) that are used in GetNext() to drive
+///      progress.
+///
 ///   3. [PROBING_SPILLED_PARTITION] Read the probe rows from a spilled partition that
 ///      was brought back into memory and probe the partition's hash table. Finally,
 ///      output unmatched build rows for join modes that require it.
@@ -92,6 +95,9 @@
 ///      either PROBING_SPILLED_PARTITION or REPARTITIONING_BUILD, depending on whether
 ///      the spilled partition's hash table fits in memory or not.
 ///
+///      This phase has sub-states (see ProbeState) that are used in GetNext() to drive
+///      progress.
+///
 /// Null aware anti-join (NAAJ) extends the above algorithm by accumulating rows with
 /// NULLs into several different streams, which are processed in a separate step to
 /// produce additional output rows. The NAAJ algorithm is documented in more detail in
@@ -121,7 +127,7 @@
  private:
   class ProbePartition;
 
-  enum HashJoinState {
+  enum class HashJoinState {
     /// Partitioning the build (right) child's input into the builder's hash partitions.
     PARTITIONING_BUILD,
 
@@ -144,6 +150,73 @@
     REPARTITIONING_PROBE,
   };
 
+  // This enum represents a sub-state of the PARTITIONING_PROBE,
+  // PROBING_SPILLED_PARTITION and REPARTITIONING_PROBE states.
+  // This drives the state machine in GetNext() that processes probe batches and generates
+  // output rows. This state machine executes within a HashJoinState state, starting with
+  // PROBING_IN_BATCH and ending with PROBE_COMPLETE.
+  //
+  // The state transition diagram is below. The state machine handles iterating through
+  // probe batches (PROBING_IN_BATCH <-> PROBING_END_BATCH), with each input probe batch
+  // producing a variable number of output rows. Once the last probe batch is processed,
+  // additional rows may need to be emitted from the build side per-partition, e.g.
+  // for right outer join (OUTPUTTING_UNMATCHED). Then PROBE_COMPLETE is entered,
+  // indicating that probing is complete. Then the top-level state machine (described by
+  // HashJoinState) takes over and either the next spilled partition is processed, final
+  // null-aware anti join processing is done, or eos can be returned from GetNext().
+  //
+  // start                     if hash tables
+  //     +------------------+  store matches  +----------------------+
+  //---->+ PROBING_IN_BATCH |  +------------->+ OUTPUTTING_UNMATCHED |
+  //     +-----+-----+------+  |              +------+---------------+
+  //           ^     |         |                     |
+  //           |     |         |                     |
+  //           |     v         |                     v
+  //     +-----+-----+-------+ | otherwise  +--------+-------+
+  // +-->+ PROBING_END_BATCH +-+----------->+ PROBE_COMPLETE |
+  // |   +-------------------+              +--+------+------+
+  // |                                       | |    | if NAAJ
+  // +---------------------------------------+ |    | and no spilled
+  //      if spilled partitions left           |    | partitions left
+  //                                           |    |
+  //                           if not NAAJ     |    |
+  //                           and no spilled  |    +---------------+
+  //                +-------+  partitions left |    | if null-aware | otherwise
+  //                |  EOS  +<-----------------+    | partition     |
+  //                +---+---+                       v has rows      |
+  //                    ^              +------------+----------+    |
+  //                    |              | OUTPUTTING_NULL_AWARE |    |
+  //                    |              +------------+----------+    |
+  //                    |                           |               |
+  //                    |                           v               |
+  //                    |              +------------+----------+    |
+  //                    +--------------+ OUTPUTTING_NULL_PROBE +<---+
+  //                                   +-----------------------+
+  enum class ProbeState {
+    // Processing probe batches and more rows in the current probe batch must be
+    // processed.
+    PROBING_IN_BATCH,
+    // Processing probe batches and no more rows in the current probe batch to process.
+    PROBING_END_BATCH,
+    // All probe batches have been processed, unmatched build rows need to be outputted
+    // from 'output_build_partitions_'.
+    // This state is only used if NeedToProcessUnmatchedBuildRows(join_op_) is true.
+    OUTPUTTING_UNMATCHED,
+    // All input probe rows from the child ExecNode or the current spilled partition have
+    // been processed, and all unmatched rows from the build have been output.
+    PROBE_COMPLETE,
+    // All input has been processed. We need to process builder->null_aware_partition()
+    // and output any rows from it.
+    // This state is only used if join_op_ is NULL_AWARE_ANTI_JOIN.
+    OUTPUTTING_NULL_AWARE,
+    // All input has been processed. We need to process null_probe_rows_ and output any
+    // rows from it.
+    // This state is only used if join_op_ is NULL_AWARE_ANTI_JOIN.
+    OUTPUTTING_NULL_PROBE,
+    // All output rows have been produced - GetNext() should return eos.
+    EOS,
+  };
+
   /// Constants from PhjBuilder, added to this node for convenience.
   static const int PARTITION_FANOUT = PhjBuilder::PARTITION_FANOUT;
   static const int NUM_PARTITIONING_BITS = PhjBuilder::NUM_PARTITIONING_BITS;
@@ -288,6 +361,10 @@
   int ProcessProbeBatch(TPrefetchMode::type, RowBatch* out_batch, HashTableCtx* ht_ctx,
       Status* status);
 
+  /// Wrapper that calls either the interpreted or codegen'd version of
+  /// ProcessProbeBatch() and commits the rows to 'out_batch' on success.
+  Status ProcessProbeBatch(RowBatch* out_batch);
+
   /// Wrapper that calls the templated version of ProcessProbeBatch() based on 'join_op'.
   int ProcessProbeBatch(const TJoinOp::type join_op, TPrefetchMode::type,
       RowBatch* out_batch, HashTableCtx* ht_ctx, Status* status);
@@ -323,12 +400,17 @@
   Status InitNullProbeRows() WARN_UNUSED_RESULT;
 
   /// Initializes null_aware_partition_ and nulls_build_batch_ to output rows.
-  Status PrepareNullAwarePartition() WARN_UNUSED_RESULT;
+  /// *has_null_aware_rows is set to true if the null-aware partition has rows that need
+  /// to be processed by calling OutputNullAwareProbeRows(), false otherwise. In both
+  /// cases, null probe rows need to be processed with OutputNullAwareNullProbe().
+  Status PrepareNullAwarePartition(bool* has_null_aware_rows) WARN_UNUSED_RESULT;
 
-  /// Continues processing from null_aware_partition_. Called after we have finished
-  /// processing all build and probe input (including repartitioning them).
+  /// Output rows from builder_->null_aware_partition(). Called when 'probe_state_'
+  /// is OUTPUTTING_NULL_AWARE - after all input is processed, including spilled
+  /// partitions. Sets *done = true if there are no more rows to output from this
+  /// function, false otherwise.
   Status OutputNullAwareProbeRows(
-      RuntimeState* state, RowBatch* out_batch) WARN_UNUSED_RESULT;
+      RuntimeState* state, RowBatch* out_batch, bool* done) WARN_UNUSED_RESULT;
 
   /// Evaluates all other_join_conjuncts against null_probe_rows_ with all the
   /// rows in build. This updates matched_null_probe_, short-circuiting if one of the
@@ -337,17 +419,21 @@
   Status EvaluateNullProbe(
       RuntimeState* state, BufferedTupleStream* build) WARN_UNUSED_RESULT;
 
-  /// Prepares to output NULLs on the probe side for NAAJ. Before calling this,
-  /// matched_null_probe_ should have been fully evaluated.
+  /// Prepares to output NULLs on the probe side for NAAJ, when transitioning
+  /// 'probe_state_' to OUTPUTTING_NULL_PROBE. Before calling this, 'matched_null_probe_'
+  /// must be fully evaluated.
   Status PrepareNullAwareNullProbe() WARN_UNUSED_RESULT;
 
   /// Outputs NULLs on the probe side, returning rows where matched_null_probe_[i] is
-  /// false. Used for NAAJ.
+  /// false. Called repeatedly after PrepareNullAwareNullProbe(), when 'probe_state_'
+  /// is OUTPUTTING_NULL_PROBE for NAAJ. Sets *done = true if there are no more rows to
+  /// output from this function, false otherwise.
   Status OutputNullAwareNullProbe(
-      RuntimeState* state, RowBatch* out_batch) WARN_UNUSED_RESULT;
+      RuntimeState* state, RowBatch* out_batch, bool* done) WARN_UNUSED_RESULT;
 
-  /// Call at the end of consuming the probe rows. Cleans up the build and probe hash
-  /// partitions and:
+  /// Call at the end of consuming the probe rows, when 'probe_state_' is
+  /// PROBING_END_BATCH, before transitioning to PROBE_COMPLETE or OUTPUTTING_UNMATCHED.
+  /// Cleans up the build and probe hash partitions, if needed, and:
   ///  - If the build partition had a hash table, close it. The build and probe
   ///    partitions are fully processed. The streams are transferred to 'batch'.
   ///    In the case of right-outer and full-outer joins, instead of closing this
@@ -357,26 +443,37 @@
   ///    rows were spilled, move the partition to 'spilled_partitions_'.
   Status DoneProbing(RuntimeState* state, RowBatch* batch) WARN_UNUSED_RESULT;
 
+  /// Get the next row batch from the probe (left) side (child(0)), if we are still
+  /// doing the first pass over the input (i.e. state_ is PARTITIONING_PROBE) or
+  /// from the spilled 'input_partition_' if state_ is REPARTITIONING_PROBE.
+  //. If we are done consuming the input, sets 'probe_batch_pos_' to -1, otherwise,
+  /// sets it to 0.  'probe_state_' must be PROBING_END_BATCH.
+  Status NextProbeRowBatch(
+      RuntimeState* state, RowBatch* out_batch) WARN_UNUSED_RESULT;
+
   /// Get the next row batch from the probe (left) side (child(0)). If we are done
   /// consuming the input, sets 'probe_batch_pos_' to -1, otherwise, sets it to 0.
-  Status NextProbeRowBatch(RuntimeState* state, RowBatch* out_batch) WARN_UNUSED_RESULT;
+  /// 'probe_state_' must be PROBING_END_BATCH.
+  Status NextProbeRowBatchFromChild(
+      RuntimeState* state, RowBatch* out_batch) WARN_UNUSED_RESULT;
 
   /// Get the next probe row batch from 'input_partition_'. If we are done consuming the
   /// input, sets 'probe_batch_pos_' to -1, otherwise, sets it to 0.
+  /// 'probe_state_' must be PROBING_END_BATCH.
   Status NextSpilledProbeRowBatch(
       RuntimeState* state, RowBatch* out_batch) WARN_UNUSED_RESULT;
 
-  /// Moves onto the next spilled partition and initializes 'input_partition_'. This
-  /// function processes the entire build side of 'input_partition_' and when this
-  /// function returns, we are ready to consume the probe side of 'input_partition_'.
+  /// Called when 'probe_state_' is PROBE_COMPLETE to start processing the next spilled
+  /// partition. This function sets 'input_partition_' to the chosen partition, then
+  /// processes the entire build side of 'input_partition_'. When this function returns
+  /// function returns, we are ready to consume probe rows in 'input_partition_'.
   /// If the build side's hash table fits in memory and there are probe rows, we will
   /// construct input_partition_'s hash table. If it does not fit, meaning we need to
   /// repartition, this function will repartition the build rows into
   /// 'builder->hash_partitions_' and prepare for repartitioning the partition's probe
   /// rows. If there are no probe rows, we just prepare the build side to be read by
   /// OutputUnmatchedBuild().
-  Status PrepareSpilledPartitionForProbe(
-      RuntimeState* state, bool* got_partition) WARN_UNUSED_RESULT;
+  Status PrepareSpilledPartitionForProbe() WARN_UNUSED_RESULT;
 
   /// Construct an error status for the null-aware anti-join when it could not fit 'rows'
   /// from the build side in memory.
@@ -387,7 +484,8 @@
   /// 'row_batch' is not NULL, transfers ownership of all row-backing resources to it.
   void CloseAndDeletePartitions(RowBatch* row_batch);
 
-  /// Prepares for probing the next batch.
+  /// Prepares for probing the next batch. Called after populating 'probe_batch_'
+  /// with rows and entering 'probe_state_' PROBING_IN_BATCH.
   void ResetForProbe();
 
   /// Codegen function to create output row. Assumes that the probe row is non-NULL.
@@ -432,20 +530,24 @@
   HashTable::Iterator hash_tbl_iterator_;
 
   /// Number of probe rows that have been partitioned.
-  RuntimeProfile::Counter* num_probe_rows_partitioned_;
+  RuntimeProfile::Counter* num_probe_rows_partitioned_ = nullptr;
 
   /// Time spent evaluating other_join_conjuncts for NAAJ.
-  RuntimeProfile::Counter* null_aware_eval_timer_;
+  RuntimeProfile::Counter* null_aware_eval_timer_ = nullptr;
 
   /// Number of partitions which had zero probe rows and we therefore didn't build the
   /// hash table.
-  RuntimeProfile::Counter* num_hash_table_builds_skipped_;
+  RuntimeProfile::Counter* num_hash_table_builds_skipped_ = nullptr;
 
   /////////////////////////////////////////
   /// BEGIN: Members that must be Reset()
 
-  /// State of the partitioned hash join algorithm. Used just for debugging.
-  HashJoinState state_;
+  /// State of the partitioned hash join algorithm. See HashJoinState for more
+  /// information.
+  HashJoinState state_ = HashJoinState::PARTITIONING_BUILD;
+
+  /// State of the probing algorithm. Used to drive the state machine in GetNext().
+  ProbeState probe_state_ = ProbeState::PROBE_COMPLETE;
 
   /// The build-side of the join. Initialized in Init().
   boost::scoped_ptr<PhjBuilder> builder_;
@@ -475,13 +577,10 @@
   std::unique_ptr<ProbePartition> input_partition_;
 
   /// In the case of right-outer and full-outer joins, this is the list of the partitions
-  /// for which we need to output their unmatched build rows.
-  /// This list is populated at DoneProbing().
+  /// for which we need to output their unmatched build rows. This list is populated at
+  /// DoneProbing(). If this is non-empty, probe_state_ must be OUTPUTTING_UNMATCHED.
   std::list<PhjBuilder::Partition*> output_build_partitions_;
 
-  /// Whether this join is in a state outputting rows from OutputNullAwareProbeRows().
-  bool output_null_aware_probe_rows_running_;
-
   /// Partition used if 'null_aware_' is set. During probing, rows from the probe
   /// side that did not have a match in the hash table are appended to this partition.
   /// At the very end, we then iterate over the partition's probe rows. For each probe
@@ -495,24 +594,31 @@
   /// conjuncts. Must be unique_ptr so we can release it and transfer to output batches.
   /// The stream starts off in memory but is unpinned if there is memory pressure,
   /// specifically if any partitions spilled or appending to the pinned stream failed.
+  /// Populated during the first pass over the probe input (i.e. while state_ is
+  /// PARTITIONING_PROBE) and then output at the end after all input data is processed.
   std::unique_ptr<BufferedTupleStream> null_probe_rows_;
 
   /// For each row in null_probe_rows_, true if this row has matched any build row
-  /// (i.e. the resulting joined row passes other_join_conjuncts).
-  /// TODO: remove this. We need to be able to put these bits inside the tuple itself.
+  /// (i.e. the resulting joined row passes other_join_conjuncts). Populated
+  /// during the first pass over the probe input (i.e. while state_ is PARTITIONING_PROBE)
+  /// and then evaluated for each build partition.
+  /// TODO: ideally we would store the bits inside the tuple data of 'null_probe_rows_'
+  /// instead of in this untracked auxiliary memory.
   std::vector<bool> matched_null_probe_;
 
   /// The current index into null_probe_rows_/matched_null_probe_ that we are
-  /// outputting.
-  int64_t null_probe_output_idx_;
+  /// outputting. -1 means invalid. Only has a valid index when probe_state_ is
+  /// OUTPUTTING_NULL_PROBE.
+  int64_t null_probe_output_idx_ = -1;
 
   /// Used by OutputAllBuild() to iterate over the entire build side tuple stream of the
-  /// current partition.
+  /// current partition. Only used when probe_state_ is OUTPUTTING_UNMATCHED.
   std::unique_ptr<RowBatch> output_unmatched_batch_;
 
   /// Stores an iterator into 'output_unmatched_batch_' to start from on the next call to
   /// OutputAllBuild(), or NULL if there are no partitions without hash tables needing to
-  /// be processed by OutputUnmatchedBuild().
+  /// be processed by OutputUnmatchedBuild(). Only used when probe_state_ is
+  /// OUTPUTTING_UNMATCHED.
   std::unique_ptr<RowBatch::Iterator> output_unmatched_batch_iter_;
 
   /// END: Members that must be Reset()
@@ -562,9 +668,8 @@
   typedef int (*ProcessProbeBatchFn)(PartitionedHashJoinNode*,
       TPrefetchMode::type, RowBatch*, HashTableCtx*, Status*);
   /// Jitted ProcessProbeBatch function pointers.  NULL if codegen is disabled.
-  ProcessProbeBatchFn process_probe_batch_fn_;
-  ProcessProbeBatchFn process_probe_batch_fn_level0_;
-
+  ProcessProbeBatchFn process_probe_batch_fn_ = nullptr;
+  ProcessProbeBatchFn process_probe_batch_fn_level0_ = nullptr;
 };
 
 }