| 883 | |
| 884 | template <bool AGGREGATED_ROWS> |
| 885 | Status GroupingAggregator::ProcessStream(BufferedTupleStream* input_stream, |
| 886 | bool has_more_streams) { |
| 887 | DCHECK(!is_streaming_preagg_); |
| 888 | if (input_stream->num_rows() > 0) { |
| 889 | if (!input_stream->is_pinned()) { |
| 890 | // This is the only stream that we are currently reading and it's unpinned. Transfer |
| 891 | // the large read page reservation to the stream by restoring the reservation and |
| 892 | // saving it immediately to the stream. |
| 893 | if (large_read_page_reservation_.GetReservation() > 0) { |
| 894 | RestoreLargeReadPageReservation(); |
| 895 | input_stream->SaveLargeReadPageReservation(); |
| 896 | } else { |
| 897 | DCHECK_EQ(resource_profile_.max_row_buffer_size, |
| 898 | resource_profile_.spillable_buffer_size) |
| 899 | << "Large read page reservation not reclaimed in previous ProcessStream"; |
| 900 | } |
| 901 | } |
| 902 | while (true) { |
| 903 | bool got_buffer = false; |
| 904 | RETURN_IF_ERROR(input_stream->PrepareForRead(/*attach_on_read*/ true, &got_buffer)); |
| 905 | if (got_buffer) break; |
| 906 | // Did not have a buffer to read the input stream. Spill and try again. |
| 907 | RETURN_IF_ERROR(SpillPartition(AGGREGATED_ROWS)); |
| 908 | } |
| 909 | |
| 910 | TPrefetchMode::type prefetch_mode = state_->query_options().prefetch_mode; |
| 911 | bool eos = false; |
| 912 | const RowDescriptor* desc = |
| 913 | AGGREGATED_ROWS ? &intermediate_row_desc_ : &input_row_desc_; |
| 914 | RowBatch batch(desc, state_->batch_size(), mem_tracker_.get()); |
| 915 | int64_t rows_read = 0; |
| 916 | do { |
| 917 | RETURN_IF_ERROR(input_stream->GetNext(&batch, &eos)); |
| 918 | rows_read += batch.num_rows(); |
| 919 | if (rows_read == input_stream->num_rows()) DCHECK(eos); |
| 920 | bool has_more_rows = AGGREGATED_ROWS ? (has_more_streams || !eos) : !eos; |
| 921 | RETURN_IF_ERROR(AddBatchImpl<AGGREGATED_ROWS>(&batch, prefetch_mode, ht_ctx_.get(), |
| 922 | has_more_rows)); |
| 923 | RETURN_IF_ERROR(QueryMaintenance(state_)); |
| 924 | batch.Reset(); |
| 925 | // We are reading in attach_on_read mode, the large read page reservation could be |
| 926 | // used by an attached buffer of the large page. It's only freed after resetting the |
| 927 | // batch. Save back the large read page reservation if we have used it. |
| 928 | input_stream->SaveLargeReadPageReservation(); |
| 929 | } while (!eos); |
| 930 | if (!input_stream->is_pinned() && input_stream->HasLargeReadPageReservation()) { |
| 931 | // Save back the large read page reservation by restoring it from the stream and |
| 932 | // save it immediately. |
| 933 | input_stream->RestoreLargeReadPageReservation(); |
| 934 | SaveLargeReadPageReservation(); |
| 935 | } |
| 936 | } |
| 937 | input_stream->Close(nullptr, RowBatch::FlushMode::NO_FLUSH_RESOURCES); |
| 938 | return Status::OK(); |
| 939 | } |
| 940 | |
| 941 | Status GroupingAggregator::SpillPartition(bool more_aggregate_rows) { |
| 942 | int64_t max_freed_mem = 0; |
nothing calls this directly
no test coverage detected