rtpsw commented on code in PR #14043:
URL: https://github.com/apache/arrow/pull/14043#discussion_r1016297197
##########
cpp/src/arrow/compute/function.cc:
##########
@@ -167,6 +179,109 @@ const Kernel* DispatchExactImpl(const Function* func,
return nullptr;
}
+struct FunctionExecutorImpl : public FunctionExecutor {
+ FunctionExecutorImpl(std::vector<TypeHolder> in_types, const Kernel* kernel,
+ std::unique_ptr<detail::KernelExecutor> executor,
+ const Function& func)
+ : in_types(std::move(in_types)),
+ kernel(kernel),
+ kernel_ctx(default_exec_context(), kernel),
+ executor(std::move(executor)),
+ func(func),
+ state(),
+ options(NULLPTR),
+ inited(false) {}
+ virtual ~FunctionExecutorImpl() {}
+
+ Status KernelInit(const FunctionOptions* options) {
+ RETURN_NOT_OK(CheckOptions(func, options));
+ if (options == NULLPTR) {
+ options = func.default_options();
+ }
+ if (kernel->init) {
+ ARROW_ASSIGN_OR_RAISE(state,
+ kernel->init(&kernel_ctx, {kernel, in_types,
options}));
+ kernel_ctx.SetState(state.get());
+ }
+
+ RETURN_NOT_OK(executor->Init(&kernel_ctx, {kernel, in_types, options}));
+ this->options = options;
+ inited = true;
+ return Status::OK();
+ }
+
+ Status Init(const FunctionOptions* options, ExecContext* exec_ctx) override {
+ kernel_ctx = KernelContext{exec_ctx, kernel};
+ return KernelInit(options);
+ }
+
+ Result<Datum> Execute(const std::vector<Datum>& args, int64_t passed_length)
override {
+ util::tracing::Span span;
+
+ auto func_kind = func.kind();
+ auto func_name = func.name();
+ START_COMPUTE_SPAN(span, func_name,
+ {{"function.name", func_name},
+ {"function.options", options ? options->ToString() :
"<NULLPTR>"},
+ {"function.kind", func_kind}});
+
+ if (!inited) {
+ ARROW_RETURN_NOT_OK(Init(NULLPTR, default_exec_context()));
+ }
+ ExecContext* ctx = kernel_ctx.exec_context();
+ // Cast arguments if necessary
+ std::vector<Datum> args_with_cast;
+ for (size_t i = 0; i != args.size(); ++i) {
+ const auto& in_type = in_types[i];
+ auto arg = args[i];
+ if (in_type != args[i].type()) {
+ ARROW_ASSIGN_OR_RAISE(arg, Cast(args[i], CastOptions::Safe(in_type),
ctx));
+ }
+ args_with_cast.push_back(arg);
+ }
+
+ detail::DatumAccumulator listener;
+
+ ExecBatch input(std::move(args_with_cast), /*length=*/0);
+ if (input.num_values() == 0) {
+ if (passed_length != -1) {
+ input.length = passed_length;
+ }
+ } else {
+ bool all_same_length = false;
+ int64_t inferred_length = detail::InferBatchLength(input.values,
&all_same_length);
+ input.length = inferred_length;
+ if (func_kind == Function::SCALAR) {
+ if (passed_length != -1 && passed_length != inferred_length) {
+ return Status::Invalid(
+ "Passed batch length for execution did not match actual"
+ " length of values for scalar function execution");
+ }
+ } else if (func_kind == Function::VECTOR) {
+ auto vkernel = static_cast<const VectorKernel*>(kernel);
Review Comment:
I tried this and it leads to a compilation error:
```
/arrow/cpp/src/arrow/util/checked_cast.h: In instantiation of 'OutputType
arrow::internal::checked_cast(InputType&&) [with OutputType = const
arrow::compute::VectorKernel*; InputType = const arrow::compute::Kernel*&]':
/arrow/cpp/src/arrow/compute/function.cc:262:64: required from here
/arrow/cpp/src/arrow/util/checked_cast.h:38:10: error: cannot 'dynamic_cast'
'value' (of type 'const struct arrow::compute::Kernel*') to type 'const struct
arrow::compute::VectorKernel*' (source type is not polymorphic)
38 | return dynamic_cast<OutputType>(value);
| ^~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
