| 101 | } |
| 102 | |
| 103 | Status GroupingAggregator::Partition::SerializeStreamForSpilling() { |
| 104 | DCHECK(!parent->is_streaming_preagg_); |
| 105 | if (parent->needs_serialize_) { |
| 106 | // We need to do a lot more work in this case. This step effectively does a merge |
| 107 | // aggregation in this node. We need to serialize the intermediates, spill the |
| 108 | // intermediates and then feed them into the aggregate function's merge step. |
| 109 | // This is often used when the intermediate is a string type, meaning the current |
| 110 | // (before serialization) in-memory layout is not the on-disk block layout. |
| 111 | // The disk layout does not support mutable rows. We need to rewrite the stream |
| 112 | // into the on disk format. |
| 113 | // TODO: if it happens to not be a string, we could serialize in place. This is |
| 114 | // a future optimization since it is very unlikely to have a serialize phase |
| 115 | // for those UDAs. |
| 116 | DCHECK(parent->serialize_stream_.get() != nullptr); |
| 117 | DCHECK(!parent->serialize_stream_->is_pinned()); |
| 118 | |
| 119 | // Serialize and copy the spilled partition's stream into the new stream. |
| 120 | Status status; |
| 121 | BufferedTupleStream* new_stream = parent->serialize_stream_.get(); |
| 122 | HashTable::Iterator it = hash_tbl->Begin(parent->ht_ctx_.get()); |
| 123 | // Marks if we have used the large write page reservation. We only reclaim it after we |
| 124 | // finish writing to 'new_stream', because there are no other works interleaving that |
| 125 | // could occupy it. |
| 126 | bool used_large_page_reservation = false; |
| 127 | while (!it.AtEnd()) { |
| 128 | Tuple* tuple = it.GetTuple<BucketType::MATCH_UNSET>(); |
| 129 | it.Next(); |
| 130 | AggFnEvaluator::Serialize(agg_fn_evals, tuple); |
| 131 | TupleRow* row = reinterpret_cast<TupleRow*>(&tuple); |
| 132 | if (UNLIKELY(!new_stream->AddRow(row, &status))) { |
| 133 | bool row_is_added = false; |
| 134 | if (status.ok()) { |
| 135 | // Don't get enough unused reservation to add the large row. Restore the saved |
| 136 | // reservation for a large write page and try again. |
| 137 | DCHECK(!used_large_page_reservation |
| 138 | && parent->large_write_page_reservation_.GetReservation() > 0) |
| 139 | << "Run out of large page reservation in spilling " << DebugString() |
| 140 | << "\nused_large_page_reservation=" << used_large_page_reservation << "\n" |
| 141 | << parent->DebugString() << "\n" |
| 142 | << parent->buffer_pool_client()->DebugString() << "\n" |
| 143 | << new_stream->DebugString() << "\n" |
| 144 | << this->aggregated_row_stream->DebugString(); |
| 145 | used_large_page_reservation = true; |
| 146 | parent->RestoreLargeWritePageReservation(); |
| 147 | row_is_added = new_stream->AddRow(row, &status); |
| 148 | } |
| 149 | if (UNLIKELY(!row_is_added)) { |
| 150 | if (status.ok()) { |
| 151 | // Still fail to write the large row after restoring all the extra reservation |
| 152 | // for the large page. We can't spill anything else to free some reservation |
| 153 | // since we are currently spilling a partition. This indicates a bug that some |
| 154 | // of the min reservation are used incorrectly. |
| 155 | status = Status(TErrorCode::INTERNAL_ERROR, strings::Substitute( |
| 156 | "Internal error: couldn't serialize a large row in $0 of $1, only had $2 " |
| 157 | "bytes of unused reservation:\n$3", DebugString(), parent->DebugString(), |
| 158 | parent->buffer_pool_client()->GetUnusedReservation(), |
| 159 | parent->buffer_pool_client()->DebugString())); |
| 160 | } |
nothing calls this directly
no test coverage detected