Deplete stream and print batches
| 485 | |
| 486 | // Deplete stream and print batches |
| 487 | void ReadStream(ReceiverInfo* info) { |
| 488 | RowBatch* batch; |
| 489 | VLOG_QUERY << "start reading"; |
| 490 | while (!(info->status = info->stream_recvr->GetBatch(&batch)).IsCancelled() && |
| 491 | (batch != nullptr)) { |
| 492 | VLOG_QUERY << "read batch #rows=" << batch->num_rows(); |
| 493 | for (int i = 0; i < batch->num_rows(); ++i) { |
| 494 | TupleRow* row = batch->GetRow(i); |
| 495 | info->data_values.insert(*static_cast<int64_t*>(row->GetTuple(0)->GetSlot(0))); |
| 496 | } |
| 497 | SleepForMs(100); // slow down receiver to exercise buffering logic |
| 498 | } |
| 499 | if (info->status.IsCancelled()) VLOG_QUERY << "reader is cancelled"; |
| 500 | VLOG_QUERY << "done reading"; |
| 501 | } |
| 502 | |
| 503 | void ReadStreamMerging(ReceiverInfo* info, RuntimeProfile* profile, |
| 504 | TupleRowComparator* less_than_comparator) { |