| 231 | } |
| 232 | |
| 233 | bool GroupingAggregator::TryAddToHashTable(HashTableCtx* __restrict__ ht_ctx, |
| 234 | Partition* __restrict__ partition, HashTable* __restrict__ hash_tbl, |
| 235 | TupleRow* __restrict__ in_row, uint32_t hash, int* __restrict__ remaining_capacity, |
| 236 | Status* status) { |
| 237 | DCHECK(remaining_capacity != nullptr); |
| 238 | DCHECK_EQ(hash_tbl, partition->hash_tbl.get()); |
| 239 | DCHECK_GE(*remaining_capacity, 0); |
| 240 | bool found; |
| 241 | Tuple* intermediate_tuple; |
| 242 | // This is called from ProcessBatchStreaming() so the rows are not aggregated. |
| 243 | HashTable::Iterator it = |
| 244 | hash_tbl->FindBuildRowBucket<BucketType::MATCH_UNSET>(ht_ctx, &found); |
| 245 | if (found) { |
| 246 | intermediate_tuple = it.GetTuple<BucketType::MATCH_UNSET>(); |
| 247 | } else if (*remaining_capacity == 0) { |
| 248 | return false; |
| 249 | } else { |
| 250 | intermediate_tuple = ConstructIntermediateTuple( |
| 251 | partition->agg_fn_evals, partition->aggregated_row_stream.get(), status); |
| 252 | if (LIKELY(intermediate_tuple != nullptr)) { |
| 253 | it.SetTuple(intermediate_tuple, hash); |
| 254 | --(*remaining_capacity); |
| 255 | } else { |
| 256 | // Avoid repeatedly trying to add tuples when under memory pressure. |
| 257 | *remaining_capacity = 0; |
| 258 | return false; |
| 259 | } |
| 260 | } |
| 261 | |
| 262 | UpdateTuple(partition->agg_fn_evals.data(), intermediate_tuple, in_row); |
| 263 | return true; |
| 264 | } |
| 265 | |
| 266 | Tuple* GroupingAggregator::ConstructIntermediateTuple( |
| 267 | const vector<AggFnEvaluator*>& agg_fn_evals, MemPool* pool, Status* status) noexcept { |