| 1161 | |
| 1162 | private: |
| 1163 | Status Consume(const ExecSpan& span) { |
| 1164 | // TODO(wesm): this is odd and should be examined soon -- only one state |
| 1165 | // "should" be needed per thread of execution |
| 1166 | |
| 1167 | // FIXME(ARROW-11840) don't merge *any* aggregates for every batch |
| 1168 | ARROW_ASSIGN_OR_RAISE(auto batch_state, |
| 1169 | kernel_->init(kernel_ctx_, {kernel_, *input_types_, options_})); |
| 1170 | |
| 1171 | if (batch_state == nullptr) { |
| 1172 | return Status::Invalid("ScalarAggregation requires non-null kernel state"); |
| 1173 | } |
| 1174 | |
| 1175 | KernelContext batch_ctx(exec_context()); |
| 1176 | batch_ctx.SetState(batch_state.get()); |
| 1177 | |
| 1178 | RETURN_NOT_OK(kernel_->consume(&batch_ctx, span)); |
| 1179 | RETURN_NOT_OK(kernel_->merge(kernel_ctx_, std::move(*batch_state), state())); |
| 1180 | return Status::OK(); |
| 1181 | } |
| 1182 | |
| 1183 | ExecSpanIterator span_iterator_; |
| 1184 | const std::vector<TypeHolder>* input_types_; |