Gandiva 表达式、投影器和过滤器#

构建表达式#

Gandiva 提供了一种通用的表达式表示，其中表达式由节点树表示。表达式树使用 TreeExprBuilder 构建。表达式树的叶子通常是字段引用（由 TreeExprBuilder::MakeField() 创建）和字面值（由 TreeExprBuilder::MakeLiteral() 创建）。节点可以使用以下方法组合成更复杂的表达式树：

TreeExprBuilder::MakeFunction() 用于创建函数节点。（您可以调用 GetRegisteredFunctionSignatures() 获取有效函数签名的列表。）
TreeExprBuilder::MakeIf() 用于创建 if-else 逻辑。
TreeExprBuilder::MakeAnd() 和 TreeExprBuilder::MakeOr() 用于创建布尔表达式。（对于“not”，在 MakeFunction 中使用 not(bool) 函数。）
TreeExprBuilder::MakeInExpressionInt32() 和其他“in expression”函数用于创建集合成员测试。

这些函数中的每一个都创建新的复合节点，这些节点包含叶节点（字面值和字段引用）或其他复合节点作为子节点。通过组合这些，您可以创建任意复杂的表达式树。

一旦表达式树构建完成，它们将被封装在 Expression 或 Condition 中，具体取决于它们的用法。Expression 用于投影，而 Condition 用于筛选。

例如，以下是如何创建表示 x + 3 的 Expression 和表示 x < 3 的 Condition

std::shared_ptr<arrow::Field> field_x_raw = arrow::field("x", arrow::int32());
std::shared_ptr<Node> field_x = TreeExprBuilder::MakeField(field_x_raw);
std::shared_ptr<Node> literal_3 = TreeExprBuilder::MakeLiteral(3);
std::shared_ptr<arrow::Field> field_result = arrow::field("result", arrow::int32());

std::shared_ptr<Node> add_node =
    TreeExprBuilder::MakeFunction("add", {field_x, literal_3}, arrow::int32());
std::shared_ptr<Expression> expression =
    TreeExprBuilder::MakeExpression(add_node, field_result);

std::shared_ptr<Node> less_than_node =
    TreeExprBuilder::MakeFunction("less_than", {field_x, literal_3}, arrow::boolean());
std::shared_ptr<Condition> condition = TreeExprBuilder::MakeCondition(less_than_node);

投影器和过滤器#

Gandiva 的两个执行内核是 Projector 和 Filter。Projector 接收一个记录批次并将其投影到一个新的记录批次。Filter 接收一个记录批次并生成一个 SelectionVector，其中包含符合条件的索引。

对于 Projector 和 Filter，表达式 IR 的优化在创建实例时发生。它们是针对静态模式编译的，因此在此刻必须知道记录批次的模式。

继续使用上一节中创建的 expression 和 condition，以下是创建 Projector 和 Filter 的示例

std::shared_ptr<arrow::Schema> input_schema = arrow::schema({field_x_raw});
std::shared_ptr<arrow::Schema> output_schema = arrow::schema({field_result});
std::shared_ptr<Projector> projector;
Status status;
std::vector<std::shared_ptr<Expression>> expressions = {expression};
status = Projector::Make(input_schema, expressions, &projector);
ARROW_RETURN_NOT_OK(status);

std::shared_ptr<Filter> filter;
status = Filter::Make(input_schema, condition, &filter);
ARROW_RETURN_NOT_OK(status);

一旦创建了 Projector 或 Filter，就可以在 Arrow 记录批次上对其进行评估。这些执行内核本身是单线程的，但设计为可重复使用以并行处理不同的记录批次。

评估投影#

执行通过 Projector::Evaluate() 进行。它输出一个数组向量，可以与输出模式一起传递给 arrow::RecordBatch::Make()。

auto pool = arrow::default_memory_pool();
int num_records = 4;
arrow::Int32Builder builder;
int32_t values[4] = {1, 2, 3, 4};
ARROW_RETURN_NOT_OK(builder.AppendValues(values, 4));
ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> array, builder.Finish());
auto in_batch = arrow::RecordBatch::Make(input_schema, num_records, {array});

arrow::ArrayVector outputs;
status = projector->Evaluate(*in_batch, pool, &outputs);
ARROW_RETURN_NOT_OK(status);
std::shared_ptr<arrow::RecordBatch> result =
    arrow::RecordBatch::Make(output_schema, outputs[0]->length(), outputs);

评估过滤器#

Filter::Evaluate() 生成 SelectionVector，这是一个符合过滤条件的行索引向量。选择向量是 Arrow 整数数组的包装器，按位宽参数化。创建选择向量时（您必须在传递给 Evaluate() 之前对其进行初始化），您必须选择位宽，它决定了它能容纳的最大索引值，以及最大槽数，它决定了它可能包含多少个索引。通常，最大槽数应设置为您的批次大小，位宽应设置为能够表示小于批次大小的所有整数的最小整数大小。例如，如果您的批次大小为 100k，则将最大槽数设置为 100k，位宽设置为 32（因为 2^16 = 64k 会太小）。

一旦 Evaluate() 运行并且 SelectionVector 被填充，使用 SelectionVector::ToArray() 方法获取底层数组，然后使用 ::arrow::compute::Take() 实现输出记录批次。

std::shared_ptr<gandiva::SelectionVector> result_indices;
// Use 16-bit integers for indices. Result can be no longer than input size,
// so use batch num_rows as max_slots.
status = gandiva::SelectionVector::MakeInt16(/*max_slots=*/in_batch->num_rows(), pool,
                                             &result_indices);
ARROW_RETURN_NOT_OK(status);
status = filter->Evaluate(*in_batch, result_indices);
ARROW_RETURN_NOT_OK(status);
std::shared_ptr<arrow::Array> take_indices = result_indices->ToArray();
Datum maybe_batch;
ARROW_ASSIGN_OR_RAISE(maybe_batch,
                      arrow::compute::Take(Datum(in_batch), Datum(take_indices),
                                           TakeOptions::NoBoundsCheck()));
result = maybe_batch.record_batch();

评估投影和过滤器#

最后，您还可以在应用选择向量时进行投影，使用 Projector::Evaluate()。为此，首先确保使用 SelectionVector::GetMode() 初始化 Projector，以便投影器使用正确的位宽进行编译。然后您可以将 SelectionVector 传递给 Projector::Evaluate() 方法。

// Make sure the projector is compiled for the appropriate selection vector mode
status = Projector::Make(input_schema, expressions, result_indices->GetMode(),
                         ConfigurationBuilder::DefaultConfiguration(), &projector);
ARROW_RETURN_NOT_OK(status);

arrow::ArrayVector outputs_filtered;
status = projector->Evaluate(*in_batch, result_indices.get(), pool, &outputs_filtered);
ARROW_RETURN_NOT_OK(status);

result =
    arrow::RecordBatch::Make(output_schema, outputs[0]->length(), outputs_filtered);