void FunnelInit(FunctionContext* context, StringVal* val) {
EventLogs* eventLogs = new EventLogs();
val->ptr = (uint8_t*) eventLogs;
// Exit on failed allocation. Impala will fail the query after some time.
if (val->ptr == NULL) {
*val = StringVal::null();
return;
}
val->is_null = false;
val->len = sizeof(EventLogs);
} I did another scan and the memory management in the above function is also slightly problematic - the memory attached to the intermediate StringVal would be better allocated from the Impala UDF interface so that Impala can track the memory consumption. E.g. see https://github.com/cloudera/impala-udf-samples/blob/bc70833/uda-sample.cc#L76 . I think the real issue though is the EventLogs data structure and lack of a Serialize() function. It's a somewhat complex nested structure with the string and vector. In order for the UDA to work, you need to have a Serialize() function that flattens out the intermediate result into a single StringVal. This is pretty unavoidable since Impala needs to be able to send the intermediate values over the network and/or write it to disk, and Impala doesn't know enough about your data structure to do it automatically. Our docs do mention this here https://www.cloudera.com/documentation/enterprise/latest/topics/impala_udf.html#udafs. Putting it into practice is a bit tricky. One working example is the implementation of reservoir sampling in Impala itself. Unfortunately I think it's a little over-complicated:https://github.com/apache/impala/blob/df53ec/be/src/exprs/aggregate-functions-ir.cc#L1067 The general pattern for complex intermediate values is to have a "header" that lets your determine whether the intermediate value is currently serialized, then either the deserialized representation, or the serialized representation after the "header" using a flexible array member or similar - https://en.wikipedia.org/wiki/Flexible_array_member. The Serialize() function will convert the representation by packing any nested structures into a single StringVal() with the header in front. Then other functions can switch back to the deserialized representation. Or in some cases, you can be clever and avoid the conversion in some case (that's what the reservoir sample function above is doing, and part of why it's overly complex). Anyway, a really rough illustration of the idea is as follows: struct DeserializedValue {
...
}
struct IntermediateValue {
bool serialized;
union {
DeserializedValue val;
char buf[0];
};
StringVal Serialize() {
if (serialized) {
// Just copy serialized representation to output StringVal
} else {
// Flatten val into an output StringVal
}
}
void DeserializeIfNeeded() {
if (serialized) {
// Unpack buf into val.
}
}
}; Just as a side note, the use of C++ builtin vector and string in the intermediate value can be problematic if they're large, since Impala doesn't account for the memory involved. But that's very much a second-order problem compared to it not working at all.
... View more
