| 57 | namespace aggregate { |
| 58 | |
| 59 | Result<AggregateNodeArgs<ScalarAggregateKernel>> |
| 60 | ScalarAggregateNode::MakeAggregateNodeArgs(const std::shared_ptr<Schema>& input_schema, |
| 61 | const std::vector<FieldRef>& keys, |
| 62 | const std::vector<FieldRef>& segment_keys, |
| 63 | const std::vector<Aggregate>& aggs, |
| 64 | ExecContext* exec_ctx, size_t concurrency, |
| 65 | bool is_cpu_parallel) { |
| 66 | // Copy (need to modify options pointer below) |
| 67 | std::vector<Aggregate> aggregates(aggs); |
| 68 | std::vector<int> segment_field_ids(segment_keys.size()); |
| 69 | std::vector<TypeHolder> segment_key_types(segment_keys.size()); |
| 70 | for (size_t i = 0; i < segment_keys.size(); i++) { |
| 71 | ARROW_ASSIGN_OR_RAISE(FieldPath match, segment_keys[i].FindOne(*input_schema)); |
| 72 | if (match.indices().size() > 1) { |
| 73 | // ARROW-18369: Support nested references as segment ids |
| 74 | return Status::Invalid("Nested references cannot be used as segment ids"); |
| 75 | } |
| 76 | segment_field_ids[i] = match[0]; |
| 77 | segment_key_types[i] = input_schema->field(match[0])->type().get(); |
| 78 | } |
| 79 | |
| 80 | ARROW_ASSIGN_OR_RAISE(auto segmenter, |
| 81 | RowSegmenter::Make(std::move(segment_key_types), |
| 82 | /*nullable_keys=*/false, exec_ctx)); |
| 83 | |
| 84 | std::vector<std::vector<TypeHolder>> kernel_intypes(aggregates.size()); |
| 85 | std::vector<const ScalarAggregateKernel*> kernels(aggregates.size()); |
| 86 | std::vector<std::vector<std::unique_ptr<KernelState>>> states(kernels.size()); |
| 87 | FieldVector fields(kernels.size() + segment_keys.size()); |
| 88 | |
| 89 | // Output the segment keys first, followed by the aggregates |
| 90 | for (size_t i = 0; i < segment_keys.size(); ++i) { |
| 91 | ARROW_ASSIGN_OR_RAISE(fields[i], segment_keys[i].GetOne(*input_schema)); |
| 92 | } |
| 93 | |
| 94 | std::vector<std::vector<int>> target_fieldsets(kernels.size()); |
| 95 | std::size_t base = segment_keys.size(); |
| 96 | for (size_t i = 0; i < kernels.size(); ++i) { |
| 97 | const auto& target_fieldset = aggregates[i].target; |
| 98 | for (const auto& target : target_fieldset) { |
| 99 | ARROW_ASSIGN_OR_RAISE(auto match, FieldRef(target).FindOne(*input_schema)); |
| 100 | target_fieldsets[i].push_back(match[0]); |
| 101 | } |
| 102 | |
| 103 | ARROW_ASSIGN_OR_RAISE(auto function, |
| 104 | exec_ctx->func_registry()->GetFunction(aggregates[i].function)); |
| 105 | |
| 106 | if (function->kind() != Function::SCALAR_AGGREGATE) { |
| 107 | if (function->kind() == Function::HASH_AGGREGATE) { |
| 108 | return Status::Invalid("The provided function (", aggregates[i].function, |
| 109 | ") is a hash aggregate function. Since there are no " |
| 110 | "keys to group by, a scalar aggregate function was " |
| 111 | "expected (normally these do not start with hash_)"); |
| 112 | } |
| 113 | return Status::Invalid("The provided function(", aggregates[i].function, |
| 114 | ") is not an aggregate function"); |
| 115 | } |
| 116 |
nothing calls this directly
no test coverage detected