Method EnqueueBatchWithTimeout

be/src/exec/scan-node.cc:356–370

}

bool ScanNode::ScannerThreadState::EnqueueBatchWithTimeout(
    unique_ptr<RowBatch>* row_batch, int64_t timeout_micros) {
  // Only need to count tuple_data_pool() bytes since after IMPALA-5307, no buffers are
  // returned from the scan node.
  int64_t bytes = (*row_batch)->tuple_data_pool()->total_reserved_bytes();
  // Transfer memory ownership before enqueueing. If the caller retries, this transfer
  // is idempotent.
  (*row_batch)->SetMemTracker(row_batches_mem_tracker_);
  if (!batch_queue_->AddBatchWithTimeout(move(*row_batch), timeout_micros)) {
    return false;
  }
  COUNTER_ADD(row_batches_enqueued_, 1);
  COUNTER_ADD(row_batch_bytes_enqueued_, bytes);
  return true;
}

void ScanNode::ScannerThreadState::Shutdown() {
  if (batch_queue_ != nullptr) batch_queue_->Shutdown();