| 501 | } |
| 502 | |
| 503 | void ReadStreamMerging(ReceiverInfo* info, RuntimeProfile* profile, |
| 504 | TupleRowComparator* less_than_comparator) { |
| 505 | /// Note that codegend_heapify_helper_fn currently stores a nullptr. |
| 506 | /// The codegened case of this function is covered in end-to-end tests. |
| 507 | CodegenFnPtr<SortedRunMerger::HeapifyHelperFn> codegend_heapify_helper_fn; |
| 508 | info->status = info->stream_recvr->CreateMerger( |
| 509 | *less_than_comparator, codegend_heapify_helper_fn); |
| 510 | if (info->status.IsCancelled()) return; |
| 511 | RowBatch batch(row_desc_, 1024, &tracker_); |
| 512 | VLOG_QUERY << "start reading merging"; |
| 513 | bool eos; |
| 514 | while (!(info->status = info->stream_recvr->GetNext(&batch, &eos)).IsCancelled()) { |
| 515 | VLOG_QUERY << "read batch #rows=" << batch.num_rows(); |
| 516 | for (int i = 0; i < batch.num_rows(); ++i) { |
| 517 | TupleRow* row = batch.GetRow(i); |
| 518 | info->data_values.insert(*static_cast<int64_t*>(row->GetTuple(0)->GetSlot(0))); |
| 519 | } |
| 520 | SleepForMs(100); |
| 521 | batch.Reset(); |
| 522 | if (eos) break; |
| 523 | } |
| 524 | if (info->status.IsCancelled()) VLOG_QUERY << "reader is cancelled"; |
| 525 | VLOG_QUERY << "done reading"; |
| 526 | } |
| 527 | |
| 528 | // Verify correctness of receivers' data values. |
| 529 | void CheckReceivers(TPartitionType::type stream_type, int num_senders, |
nothing calls this directly
no test coverage detected