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

构建表达式#

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

  • TreeExprBuilder::MakeFunction() 创建函数节点。（您可以调用 GetRegisteredFunctionSignatures() 获取有效函数签名的列表。）

  • TreeExprBuilder::MakeIf() 创建 if-else 逻辑。

  • TreeExprBuilder::MakeAnd() 和 TreeExprBuilder::MakeOr() 创建布尔表达式。（对于“非”，请在 MakeFunction 中使用 not(bool) 函数。）

  • TreeExprBuilder::MakeInExpressionInt32() 和其他“in 表达式”函数创建集合成员资格测试。

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

构建表达式树后，它们将包装在 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，一个与筛选条件匹配的行索引向量。选择向量是围绕箭头整数数组的包装器，由位宽参数化。创建选择向量时（您必须在传递给 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);
本页内容
在 GitHub 上编辑