| 398 | } |
| 399 | |
| 400 | Status Exec(compute::KernelContext* ctx, const compute::ExecSpan& batch, |
| 401 | compute::ExecResult* out) { |
| 402 | auto state = arrow::internal::checked_cast<PythonUdfKernelState*>(ctx->state()); |
| 403 | PyObject* function = state->function->obj(); |
| 404 | const int num_args = batch.num_values(); |
| 405 | UdfContext udf_context{ctx->memory_pool(), batch.length}; |
| 406 | |
| 407 | OwnedRef arg_tuple(PyTuple_New(num_args)); |
| 408 | RETURN_NOT_OK(CheckPyError()); |
| 409 | for (int arg_id = 0; arg_id < num_args; arg_id++) { |
| 410 | if (batch[arg_id].is_scalar()) { |
| 411 | std::shared_ptr<Scalar> c_data = batch[arg_id].scalar->GetSharedPtr(); |
| 412 | PyObject* data = wrap_scalar(c_data); |
| 413 | PyTuple_SetItem(arg_tuple.obj(), arg_id, data); |
| 414 | } else { |
| 415 | std::shared_ptr<Array> c_data = batch[arg_id].array.ToArray(); |
| 416 | PyObject* data = wrap_array(c_data); |
| 417 | PyTuple_SetItem(arg_tuple.obj(), arg_id, data); |
| 418 | } |
| 419 | } |
| 420 | |
| 421 | OwnedRef result(cb(function, udf_context, arg_tuple.obj())); |
| 422 | RETURN_NOT_OK(CheckPyError()); |
| 423 | // unwrapping the output for expected output type |
| 424 | if (is_array(result.obj())) { |
| 425 | ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Array> val, unwrap_array(result.obj())); |
| 426 | ARROW_ASSIGN_OR_RAISE(TypeHolder type, ResolveType(ctx, batch.GetTypes())); |
| 427 | if (type.type == NULLPTR) { |
| 428 | return Status::TypeError("expected output datatype is null"); |
| 429 | } |
| 430 | if (*type.type != *val->type()) { |
| 431 | return Status::TypeError("Expected output datatype ", type.type->ToString(), |
| 432 | ", but function returned datatype ", |
| 433 | val->type()->ToString()); |
| 434 | } |
| 435 | out->value = std::move(val->data()); |
| 436 | return Status::OK(); |
| 437 | } else { |
| 438 | return Status::TypeError("Unexpected output type: ", Py_TYPE(result.obj())->tp_name, |
| 439 | " (expected Array)"); |
| 440 | } |
| 441 | return Status::OK(); |
| 442 | } |
| 443 | }; |
| 444 | |
| 445 | Status PythonUdfExec(compute::KernelContext* ctx, const compute::ExecSpan& batch, |
no test coverage detected