| 381 | } |
| 382 | |
| 383 | Status AnalyticEvalNode::AddResultTuple(int64_t stream_idx) { |
| 384 | VLOG_ROW << id() << " AddResultTuple idx=" << stream_idx; |
| 385 | DCHECK(curr_tuple_ != nullptr); |
| 386 | MemPool* curr_tuple_pool = curr_tuple_pool_.get(); |
| 387 | Tuple* result_tuple = Tuple::Create(result_tuple_desc_->byte_size(), curr_tuple_pool); |
| 388 | |
| 389 | AggFnEvaluator::GetValue(analytic_fn_evals_, curr_tuple_, result_tuple); |
| 390 | // Copy any string data in 'result_tuple' into 'curr_tuple_pool'. The var-len data |
| 391 | // returned by GetValue() may be backed by an allocation from |
| 392 | // 'expr_results_pool_' that will be recycled so it must be copied out. |
| 393 | for (const SlotDescriptor* slot_desc : result_tuple_desc_->string_slots()) { |
| 394 | if (result_tuple->IsNull(slot_desc->null_indicator_offset())) continue; |
| 395 | StringValue* sv = result_tuple->GetStringSlot(slot_desc->tuple_offset()); |
| 396 | if (sv->IsSmall()) continue; |
| 397 | StringValue::SimpleString s = sv->ToSimpleString(); |
| 398 | if (s.len == 0) continue; |
| 399 | char* new_ptr = reinterpret_cast<char*>( |
| 400 | curr_tuple_pool->TryAllocateUnaligned(s.len)); |
| 401 | if (UNLIKELY(new_ptr == nullptr)) { |
| 402 | return curr_tuple_pool->mem_tracker()->MemLimitExceeded(nullptr, |
| 403 | "Failed to allocate memory for analytic function's result.", s.len); |
| 404 | } |
| 405 | memcpy(new_ptr, s.ptr, s.len); |
| 406 | sv->SetPtr(new_ptr); |
| 407 | } |
| 408 | |
| 409 | DCHECK_GT(stream_idx, last_result_idx_); |
| 410 | result_tuples_.emplace_back(stream_idx, result_tuple); |
| 411 | last_result_idx_ = stream_idx; |
| 412 | VLOG_ROW << id() << " Added result tuple, final state: " << DebugStateString(true); |
| 413 | return Status::OK(); |
| 414 | } |
| 415 | |
| 416 | inline Status AnalyticEvalNode::TryAddResultTupleForPrevRow( |
| 417 | const TupleRow* child_tuple_cmp_row, bool next_partition, int64_t stream_idx) { |
nothing calls this directly
no test coverage detected