| 369 | } |
| 370 | |
| 371 | Status GroupByNode::InputReceived(ExecNode* input, ExecBatch batch) { |
| 372 | auto scope = TraceInputReceived(batch); |
| 373 | |
| 374 | DCHECK_EQ(input, inputs_[0]); |
| 375 | |
| 376 | auto handler = [this](const ExecBatch& full_batch, const Segment& segment) { |
| 377 | if (!segment.extends && segment.offset == 0) |
| 378 | RETURN_NOT_OK(OutputResult(/*is_last=*/false)); |
| 379 | auto exec_batch = full_batch.Slice(segment.offset, segment.length); |
| 380 | auto batch = ExecSpan(exec_batch); |
| 381 | RETURN_NOT_OK(Consume(batch)); |
| 382 | RETURN_NOT_OK( |
| 383 | ExtractSegmenterValues(&segmenter_values_, exec_batch, segment_key_field_ids_)); |
| 384 | if (!segment.is_open) RETURN_NOT_OK(OutputResult(/*is_last=*/false)); |
| 385 | return Status::OK(); |
| 386 | }; |
| 387 | ARROW_RETURN_NOT_OK( |
| 388 | HandleSegments(segmenter_.get(), batch, segment_key_field_ids_, handler)); |
| 389 | |
| 390 | if (input_counter_.Increment()) { |
| 391 | ARROW_RETURN_NOT_OK(OutputResult(/*is_last=*/true)); |
| 392 | } |
| 393 | return Status::OK(); |
| 394 | } |
| 395 | |
| 396 | Status GroupByNode::InputFinished(ExecNode* input, int total_batches) { |
| 397 | auto scope = TraceFinish(); |
no test coverage detected