流式执行 (Acero)#

创建并运行执行计划#

enum class UnalignedBufferHandling#

如何处理未对齐的缓冲区。

值

enumerator kWarn#

enumerator kIgnore#

enumerator kReallocate#

enumerator kError#

UnalignedBufferHandling GetDefaultUnalignedBufferHandling()#

获取未对齐缓冲区处理的默认行为

这可以通过环境变量 ACERO_ALIGNMENT_HANDLING 进行配置，该变量可以设置为 “warn”（警告）、“ignore”（忽略）、“reallocate”（重新分配）或 “error”（错误）。如果未设置环境变量，或设置为无效值，将返回 kWarn。

Result<std::shared_ptr<Schema>> DeclarationToSchema(const Declaration &declaration, FunctionRegistry *function_registry = NULLPTR)#

计算声明的输出模式 (Schema)。

这不会实际执行计划。如果声明代表一个无效的计划（例如带有多个输入的投影节点），此操作可能会失败。

参数:

declaration – 描述执行计划的声明
function_registry – 用于函数执行的函数注册表。如果为空，则使用默认的函数注册表。

返回：

批处理经过执行计划后所具有的模式。

Result<std::string> DeclarationToString(const Declaration &declaration, FunctionRegistry *function_registry = NULLPTR)#

创建计划的字符串表示形式。

此表示形式仅用于调试目的。

如果声明代表无效计划，转换为字符串可能会失败。

使用 Substrait 进行完整的计划序列化

参数:

declaration – 描述执行计划的声明
function_registry – 用于函数执行的函数注册表。如果为空，则使用默认的函数注册表。

返回：

适合调试输出的计划字符串表示形式

Result<std::shared_ptr<Table>> DeclarationToTable(Declaration declaration, bool use_threads = true, MemoryPool *memory_pool = default_memory_pool(), FunctionRegistry *function_registry = NULLPTR)#

用于运行声明并将结果收集到表中的实用方法。

此方法将向声明中添加一个接收器节点 (sink node) 以将结果收集到表中。然后，它将从该声明创建一个 ExecPlan，启动执行计划，阻塞直到计划完成，并返回创建的表。

参数:

declaration – 描述要运行的计划的声明
use_threads – 如果 use_threads 为 false，则所有 CPU 工作将在调用线程上完成。I/O 任务仍将在 I/O 执行器上发生，并且可能是多线程的（但不应占用显著的 CPU 资源）。
memory_pool – 运行计划时用于分配内存的内存池。
function_registry – 用于函数执行的函数注册表。如果为空，则使用默认的函数注册表。

Result<std::shared_ptr<Table>> DeclarationToTable(Declaration declaration, QueryOptions query_options)#

Future<std::shared_ptr<Table>> DeclarationToTableAsync(Declaration declaration, bool use_threads = true, MemoryPool *memory_pool = default_memory_pool(), FunctionRegistry *function_registry = NULLPTR)#

异步版本的。

另请参阅

DeclarationToTable

参数:

declaration – 描述要运行的计划的声明
use_threads – use_threads 的行为与同步版本略有不同，因为我们不能在调用线程上同步运行。相反，如果 use_threads=false，则会创建一个具有单个线程的新线程池，并将用于所有计算工作。
memory_pool – 运行计划时用于分配内存的内存池。
function_registry – 用于函数执行的函数注册表。如果为空，则使用默认的函数注册表。

Future<std::shared_ptr<Table>> DeclarationToTableAsync(Declaration declaration, ExecContext custom_exec_context)#

的重载。

必须指定执行器（不能为 null），并且必须保持存活直到返回的 future 完成。

另请参阅

接受自定义执行上下文的 DeclarationToTableAsync

Result<BatchesWithCommonSchema> DeclarationToExecBatches(Declaration declaration, bool use_threads = true, MemoryPool *memory_pool = default_memory_pool(), FunctionRegistry *function_registry = NULLPTR)#

用于运行声明并将结果收集到 ExecBatch 向量中的实用方法。

另请参阅

关于线程和执行的详细信息请参阅 DeclarationToTable

Result<BatchesWithCommonSchema> DeclarationToExecBatches(Declaration declaration, QueryOptions query_options)#

Future<BatchesWithCommonSchema> DeclarationToExecBatchesAsync(Declaration declaration, bool use_threads = true, MemoryPool *memory_pool = default_memory_pool(), FunctionRegistry *function_registry = NULLPTR)#

异步版本的。

另请参阅

DeclarationToExecBatches

另请参阅

关于线程和执行的详细信息请参阅 DeclarationToTableAsync

Future<BatchesWithCommonSchema> DeclarationToExecBatchesAsync(Declaration declaration, ExecContext custom_exec_context)#

的重载。

另请参阅

接受自定义执行上下文的 DeclarationToExecBatchesAsync

另请参阅

关于线程和执行的详细信息请参阅 DeclarationToTableAsync

Result<std::vector<std::shared_ptr<RecordBatch>>> DeclarationToBatches(Declaration declaration, bool use_threads = true, MemoryPool *memory_pool = default_memory_pool(), FunctionRegistry *function_registry = NULLPTR)#

用于运行声明并将结果收集到向量中的实用方法。

另请参阅

关于线程和执行的详细信息请参阅 DeclarationToTable

Result<std::vector<std::shared_ptr<RecordBatch>>> DeclarationToBatches(Declaration declaration, QueryOptions query_options)#

Future<std::vector<std::shared_ptr<RecordBatch>>> DeclarationToBatchesAsync(Declaration declaration, bool use_threads = true, MemoryPool *memory_pool = default_memory_pool(), FunctionRegistry *function_registry = NULLPTR)#

异步版本的。

另请参阅

DeclarationToBatches

另请参阅

关于线程和执行的详细信息请参阅 DeclarationToTableAsync

Future<std::vector<std::shared_ptr<RecordBatch>>> DeclarationToBatchesAsync(Declaration declaration, ExecContext exec_context)#

的重载。

另请参阅

接受自定义执行上下文的 DeclarationToBatchesAsync

另请参阅

关于线程和执行的详细信息请参阅 DeclarationToTableAsync

Result<std::unique_ptr<RecordBatchReader>> DeclarationToReader(Declaration declaration, bool use_threads = true, MemoryPool *memory_pool = default_memory_pool(), FunctionRegistry *function_registry = NULLPTR)#

用于运行声明并将结果作为 RecordBatchReader 返回的实用方法。

如果未提供执行上下文，则将根据 use_threads 的值使用默认执行上下文。如果 use_threads 为 false，则 CPU 执行器将是串行执行器，所有 CPU 工作都将在调用线程上完成。I/O 任务仍将在 I/O 执行器上发生，并且可能是多线程的。

如果 use_threads 为 false，则所有 CPU 工作都将在调用 RecordBatchReader::Next 期间发生，并且后台不会发生 CPU 工作。如果 use_threads 为 true，则 CPU 工作将在 CPU 线程池上发生，并且任务可能会在对 RecordBatchReader::Next 的调用之间运行。如果返回的读取器没有被足够快地消费，随着反压队列填满，计划最终会暂停。

如果提供了自定义执行上下文，则 use_threads 的值将被忽略。

可以提前关闭返回的 RecordBatchReader 以取消记录批次的计算。在这种情况下，可能只会报告计算中遇到的错误。特别是，不会报告取消错误。

Result<std::unique_ptr<RecordBatchReader>> DeclarationToReader(Declaration declaration, QueryOptions query_options)#

Status DeclarationToStatus(Declaration declaration, bool use_threads = true, MemoryPool *memory_pool = default_memory_pool(), FunctionRegistry *function_registry = NULLPTR)#

用于运行声明并忽略结果的实用方法。

当数据作为计划本身的一部分被消费时（例如当计划以写入节点结束时），这非常有用。

另请参阅

关于线程和执行的详细信息请参阅 DeclarationToTable

Status DeclarationToStatus(Declaration declaration, QueryOptions query_options)#

Future DeclarationToStatusAsync(Declaration declaration, bool use_threads = true, MemoryPool *memory_pool = default_memory_pool(), FunctionRegistry *function_registry = NULLPTR)#

异步版本的。

当数据作为计划本身的一部分被消费时（例如当计划以写入节点结束时），这非常有用。

另请参阅

DeclarationToStatus

另请参阅

关于线程和执行的详细信息请参阅 DeclarationToTableAsync

Future DeclarationToStatusAsync(Declaration declaration, ExecContext exec_context)#

的重载。

另请参阅

接受自定义执行上下文的 DeclarationToStatusAsync

另请参阅

关于线程和执行的详细信息请参阅 DeclarationToTableAsync

struct Declaration#

#include <arrow/acero/exec_plan.h>

用于声明执行节点的辅助类。

Declaration 表示一个未构造的 ExecNode（也可能是一个完整的图，因为它的输入也可能是 Declaration）。

Declaration 可以转换为计划，并使用 DeclarationToXyz 方法之一执行。

为了更直接的控制，可以使用 Declaration::AddToPlan 将 Declaration 添加到现有的执行计划中，这将根据需要递归地构造任何输入。

公共类型

using Input = std::variant<ExecNode*, Declaration>#

公共函数

inline Declaration()#

inline Declaration(std::string factory_name, std::vector<Input> inputs, std::shared_ptr<ExecNodeOptions> options, std::string label)#

构造声明

参数:

factory_name – 要构造的执行节点的名称。该节点必须已以此名称添加到执行节点注册表中。
inputs – 节点的输入，它们应该是其他声明。
options – 控制节点行为的选项。必须使用适当的子类。例如，如果 factory_name 是 “project”，则 options 应该是 ProjectNodeOptions。
label – 给节点的标签。可用于在计划中将其与其他相同类型的节点区分开来。

template<typename Options> inline Declaration(std::string factory_name, std::vector<Input> inputs, Options options, std::string label)#

template<typename Options> inline Declaration(std::string factory_name, std::vector<Input> inputs, Options options)#

template<typename Options> inline Declaration(std::string factory_name, Options options)#

template<typename Options> inline Declaration(std::string factory_name, Options options, std::string label)#

Result<ExecNode*> AddToPlan(ExecPlan *plan, ExecFactoryRegistry *registry = default_exec_factory_registry()) const#

将声明添加到已创建的执行计划中

此方法将递归地对声明的所有输入调用 AddToPlan。此方法仅适用于当 DeclarationToXyz 方法不足时的进阶使用场景。

参数:

plan – 要添加节点的计划
registry – 用于查找节点工厂的注册表

返回：

实例化的执行节点

bool IsValid(ExecFactoryRegistry *registry = default_exec_factory_registry()) const#

公共成员

std::string factory_name#: 创建节点时使用的工厂名称

std::vector<Input> inputs#: 声明的输入

std::shared_ptr<ExecNodeOptions> options#: 控制节点行为的选项

std::string label#: 在计划中给节点的标签

公共静态函数

static Declaration Sequence(std::vector<Declaration> decls)#

针对简单节点序列常见情况的便捷工厂。

decls 中的每一个都将被追加到后续声明的输入中，并返回最终修改后的声明。

没有这个便捷工厂，构建序列将需要显式且难以阅读的嵌套。

Declaration{"n3",
              {
                  Declaration{"n2",
                              {
                                  Declaration{"n1",
                                              {
                                                  Declaration{"n0", N0Opts{}},
                                              },
                                              N1Opts{}},
                              },
                              N2Opts{}},
              },
              N3Opts{}};

可以使用 Sequence 更简洁地构建等效的 Declaration。

Declaration::Sequence({
    {"n0", N0Opts{}},
    {"n1", N1Opts{}},
    {"n2", N2Opts{}},
    {"n3", N3Opts{}},
});

struct QueryOptions#

#include <arrow/acero/exec_plan.h>

执行执行计划时可以指定的计划范围选项

公共成员

bool use_legacy_batching = false#

计划是否应使用遗留批处理策略。

这目前仅用于支持 Scanner::ToTable 方法。该方法依赖于扫描仪的批索引保持一致。这在 ExecPlan 中是不切实际的，因为它可能会根据需要对批次进行切片（例如用于连接）。

然而，它对于简单的计划仍然有效，这是我们目前维持隐含顺序的唯一方法。

std::optional<bool> sequence_output = std::nullopt#

如果输出具有有意义的顺序，则对计划的输出进行排序。

默认行为 (std::nullopt) 是，如果最终节点中存在有意义的顺序，则对输出批次进行排序，否则立即发出批次。

如果显式设置为 true，则如果没有有意义的顺序，计划执行将失败。这对于验证应该发出有序结果的查询很有用。

如果显式设置为 false，则即使存在有意义的顺序，批次也会立即发出。这可能会导致批次乱序发出，但可能会略微降低延迟。

bool use_threads = true#

计划是否应使用多个后台线程进行 CPU 密集型工作

如果为 false，则所有 CPU 工作将在调用线程上完成。I/O 任务仍将在 I/O 执行器上发生，并且可能是多线程的（但不应占用显著的 CPU 资源）。

如果设置了 custom_cpu_executor，则此项将被忽略。

::arrow::internal::Executor *custom_cpu_executor = NULLPTR#

用于 CPU 密集型工作的自定义执行器

必须为 null 或在计划期间保持有效。如果为 null，则将选择一个默认线程池，其行为将由 use_threads 选项控制。

::arrow::internal::Executor *custom_io_executor = NULLPTR#

用于 IO 工作的自定义执行器

必须为 null 或在计划期间保持有效。如果为 null，则将选择全局 io 线程池，其行为将由 “ARROW_IO_THREADS” 环境变量控制。

MemoryPool *memory_pool = default_memory_pool()#

用于分配的内存池

必须在计划期间保持有效。

FunctionRegistry *function_registry = GetFunctionRegistry()#

用于计划的函数注册表

必须在计划期间保持有效。

std::vector<std::string> field_names#

输出列的名称

如果为空，则名称将根据输入列生成

如果设置了名称，则名称数量必须等于输出列的数量

std::optional<UnalignedBufferHandling> unaligned_buffer_handling#

源数据中未对齐缓冲区的策略。

各种计算函数和 Acero 内部组件会将数组缓冲区从 uint8_t* 类型转换为某种值类型（例如，我们可能会转换为 int32_t* 以相加两个 int32 数组）。

如果缓冲区对齐不当（例如 int32 数组未在 4 字节边界上对齐），那么这在 C++ 中属于未定义行为。然而，大多数现代编译器和 CPU 对此行为相当宽容，除了轻微的性能损失外，不太可能发生糟糕的情况。

请注意，这仅适用于源缓冲区。Acero 内部分配的所有缓冲区都将是适当对齐的。

如果此字段设置为 kWarn，则 Acero 将检查是否有未对齐的缓冲区，如果有，将发出警告。

如果此字段设置为 kReallocate，则 Acero 将分配一个新的适当对齐的缓冲区，并将内容从旧缓冲区复制到新缓冲区中。

如果此字段设置为 kError，则 Acero 将改为平稳地中止该计划。

如果此字段设置为 kIgnore，则 Acero 甚至不会检查缓冲区是否未对齐。

如果未设置此字段，则除非被 ACERO_ALIGNMENT_HANDLING 环境变量覆盖，否则它将被视为 kWarn。

struct BatchesWithCommonSchema#

#include <arrow/acero/exec_plan.h>

具有共同模式的执行批次集合

公共成员

std::vector<ExecBatch> batches#

std::shared_ptr<Schema> schema#

执行节点的配置#

enum class JoinType#

值

enumerator LEFT_SEMI#

enumerator RIGHT_SEMI#

enumerator LEFT_ANTI#

enumerator RIGHT_ANTI#

enumerator INNER#

enumerator LEFT_OUTER#

enumerator RIGHT_OUTER#

enumerator FULL_OUTER#

enum class JoinKeyCmp#

值

enumerator EQ#

enumerator IS#

using ArrayVectorIteratorMaker = std::function<Iterator<std::shared_ptr<ArrayVector>>()>#: 从数组向量迭代器读取的源节点

using ExecBatchIteratorMaker = std::function<Iterator<std::shared_ptr<ExecBatch>>()>#: 从 ExecBatch 迭代器读取的源节点

using RecordBatchIteratorMaker = std::function<Iterator<std::shared_ptr<RecordBatch>>()>#

constexpr int32_t kDefaultBackpressureHighBytes = 1 << 30#: 应用背压的默认阈值（字节数）

constexpr int32_t kDefaultBackpressureLowBytes = 1 << 28#: 解除背压的默认阈值（字节数）

std::string ToString(JoinType t)#

class ExecNodeOptions#

#include <arrow/acero/options.h>

所有选项对象的基类。

仅当节点没有配置时才会直接使用此类

公共函数

virtual ~ExecNodeOptions() = default#

公共成员

std::shared_ptr<DebugOptions> debug_opts#: 此选项不得在发布模式（release-mode）下使用。

class SourceNodeOptions : public arrow::acero::ExecNodeOptions #

#include <arrow/acero/options.h>

表示 Acero 数据源的通用节点。

源节点将在 StartProducing 时开始调用 generator。它会创建一个初始任务来调用 generator。它不会重入地调用 generator。如果数据源可以并行读取，则这些细节应封装在 generator 中。

对于接收到的每个批次，都会创建一个新任务将该批次推送到下游。此任务将从父批次中切分出大小为 ExecPlan::kMaxBatchSize 的较小单元，并调用 InputReceived。因此，如果 generator 产生了大型批次，可能会导致多次调用 InputReceived。

默认情况下，SourceNode 会为输出批次分配隐含的顺序。只要生成器以确定性方式生成批次，这就是有效的。目前，覆盖此行为的唯一方法是继承 SourceNode。

该节点通常不直接使用，但可以作为各种特殊节点的基础。

公共函数

inline SourceNodeOptions(std::shared_ptr<Schema> output_schema, std::function<Future<std::optional<ExecBatch>>()> generator, Ordering ordering = Ordering::Unordered())#: 从值创建实例。

公共成员

std::shared_ptr<Schema> output_schema#: 由此源生成的批次的 schema

std::function<Future<std::optional<ExecBatch>>()> generator#: 异步批次流，以 std::nullopt 结束

Ordering ordering#: 数据顺序，默认为 Ordering::Unordered

class TableSourceNodeOptions : public arrow::acero::ExecNodeOptions #

#include <arrow/acero/options.h>

一个从已加载到内存中的表生成数据的节点

Table source 节点将根据 max_batch_size 进行切分，以便并行处理。Table source 节点扩展了 source node，因此这些块将以小批次的形式迭代处理。

另请参阅

有关详细信息，请参阅 SourceNodeOptions。

公共函数

inline TableSourceNodeOptions(std::shared_ptr<Table> table, int64_t max_batch_size = kDefaultMaxBatchSize)#: 从值创建实例。

公共成员

std::shared_ptr<Table> table#: 作为数据源的表

int64_t max_batch_size#: 此节点发出的批次大小。如果表更大，节点将从表中发出多个批次以进行并行处理。

公共静态属性

static constexpr int64_t kDefaultMaxBatchSize = 1 << 20#

class NamedTableNodeOptions : public arrow::acero::ExecNodeOptions #

#include <arrow/acero/options.h>

定义一个延迟解析的 Arrow 表。

由名称唯一标识的表通常可以在消费计划时进行解析。

此节点仅用于序列化目的，不可执行。

公共函数

inline NamedTableNodeOptions(std::vector<std::string> names, std::shared_ptr<Schema> schema)#: 从值创建实例。

公共成员

std::vector<std::string> names#: 放入序列化计划中的名称

std::shared_ptr<Schema> schema#: 表的输出 schema

template<typename ItMaker> class SchemaSourceNodeOptions : public arrow::acero::ExecNodeOptions #

#include <arrow/acero/options.h>

一个源节点，它从表格数据的同步迭代器馈送数据

ItMaker 是表格数据迭代器的生成器。

该节点可配置为使用 I/O 执行器。如果设置了此项，则每次轮询迭代器时，都会创建一个新的 I/O 线程任务来执行轮询。这允许阻塞迭代器不占用 CPU 线程池。

公共函数

inline SchemaSourceNodeOptions(std::shared_ptr<Schema> schema, ItMaker it_maker, arrow::internal::Executor *io_executor)#: 创建一个实例，该实例将为每次迭代在 io_executor 上创建一个新任务。

inline SchemaSourceNodeOptions(std::shared_ptr<Schema> schema, ItMaker it_maker, bool requires_io = false)#: 创建一个将同步迭代或使用默认 I/O 执行器的实例。

公共成员

std::shared_ptr<Schema> schema#: 迭代器中记录批次的 schema。

ItMaker it_maker#: 作为数据源的迭代器生成器。

arrow::internal::Executor *io_executor#

用于扫描迭代器的执行器。

默认为默认 I/O 执行器。仅在 requires_io 为 true 时使用。如果 requires_io 为 false，则此项必须为 nullptr。

bool requires_io#: 如果为 true，则将从专用 I/O 线程上的迭代器获取项目，以使 I/O 远离 CPU 线程。

class RecordBatchReaderSourceNodeOptions : public arrow::acero::ExecNodeOptions #

#include <arrow/acero/options.h>

从 RecordBatchReader 读取的源节点

RecordBatchReader 的每次迭代都将在 I/O 线程池上创建的新线程任务上运行。

公共函数

inline RecordBatchReaderSourceNodeOptions(std::shared_ptr<RecordBatchReader> reader, arrow::internal::Executor *io_executor = NULLPTR)#: 从值创建实例。

公共成员

std::shared_ptr<RecordBatchReader> reader#: 作为数据源的 RecordBatchReader。

arrow::internal::Executor *io_executor#

用于读取器的执行器。

默认为默认 I/O 执行器。

class ArrayVectorSourceNodeOptions : public arrow::acero::SchemaSourceNodeOptions<ArrayVectorIteratorMaker>#: #include <arrow/acero/options.h>

一个扩展的 Source 节点，它接受一个 schema 和数组向量。

class ExecBatchSourceNodeOptions : public arrow::acero::SchemaSourceNodeOptions<ExecBatchIteratorMaker>#

#include <arrow/acero/options.h>

一个扩展的 Source 节点，它接受一个 schema 和执行批次。

公共函数

ExecBatchSourceNodeOptions(std::shared_ptr<Schema> schema, std::vector<ExecBatch> batches, ::arrow::internal::Executor *io_executor)#

ExecBatchSourceNodeOptions(std::shared_ptr<Schema> schema, std::vector<ExecBatch> batches, bool requires_io = false)#

inline SchemaSourceNodeOptions(std::shared_ptr<Schema> schema, ItMaker it_maker, arrow::internal::Executor *io_executor)#: 创建一个实例，该实例将为每次迭代在 io_executor 上创建一个新任务。

inline SchemaSourceNodeOptions(std::shared_ptr<Schema> schema, ItMaker it_maker, bool requires_io = false)#: 创建一个将同步迭代或使用默认 I/O 执行器的实例。

class RecordBatchSourceNodeOptions : public arrow::acero::SchemaSourceNodeOptions<RecordBatchIteratorMaker>#: #include <arrow/acero/options.h>

从 RecordBatch 迭代器读取的源节点

class FilterNodeOptions : public arrow::acero::ExecNodeOptions #

#include <arrow/acero/options.h>

一个节点，用于排除通过它的批次中的某些行

filter_expression 将针对推送到此节点的每个批次进行评估。对于 filter_expression 评估结果不为 true 的任何行，都将在此节点发出的批次中被排除。

如果所有行都被排除，此节点将发出空批次。这样做是为了避免顺序出现中断。

公共函数

inline explicit FilterNodeOptions(Expression filter_expression)#: 从值创建实例

公共成员

Expression filter_expression#

用于过滤批次的表达式

此表达式的返回类型必须是布尔值

class FetchNodeOptions : public arrow::acero::ExecNodeOptions #

#include <arrow/acero/options.h>

一个节点，用于从输入中选择指定的子集

公共函数

inline FetchNodeOptions(int64_t offset, int64_t count)#: 从值创建实例

公共成员

int64_t offset#: 要跳过的行数

int64_t count#: 要保留的行数（不计算跳过的行）

公共静态属性

static constexpr std::string_view kName = "fetch"#

class ProjectNodeOptions : public arrow::acero::ExecNodeOptions #

#include <arrow/acero/options.h>

一个节点，用于在输入批次上执行表达式，产生长度相同且带有新列的批次。

每个表达式将针对推送到此节点的每个批次进行评估，以产生相应的输出列。

如果未提供名称，将使用表达式的字符串表示形式。

公共函数

inline explicit ProjectNodeOptions(std::vector<Expression> expressions, std::vector<std::string> names = {})#: 从值创建实例

公共成员

std::vector<Expression> expressions#

在批次上运行的表达式

输出将为每个表达式包含一列。如果您希望保留来自输入中的任何列，则应为该列创建一个简单的 field_ref 表达式。

std::vector<std::string> names#

输出列的名称

如果未指定，则将使用在表达式上调用 ToString 的结果

此列表应为空，或者与 expressions 长度相同

class AggregateNodeOptions : public arrow::acero::ExecNodeOptions #

#include <arrow/acero/options.h>

一个节点，用于聚合输入批次并计算汇总统计信息

该节点可以汇总整个输入，也可以使用分组键和段键对输入进行分组。

默认情况下，聚合节点是一个流水线断点（pipeline breaker）。它必须在产生任何输出之前累积所有输入。段键是一种性能优化。如果您知道输入已经按一个或多个列进行了分区，则可以将其指定为段键。在段键的每次更改时，节点将为目前为止看到的所有数据发出值。

段键目前仅限于单线程模式。

键和段键都决定了组。然而，段键也用于确定分组段（这应该很大），并允许在处理每个段后流式传输部分聚合结果。段键的一个常见用例是有序聚合，其中段键属性指定具有非递减值的列或此类列的字典序集合。

如果 keys 属性是非空向量，则 aggregates 中的每个聚合都被预期为 HashAggregate 函数。如果 keys 属性是空向量，则每个聚合都被假定为 ScalarAggregate 函数。

如果 segment_keys 属性是非空向量，则适用上述的分段聚合。

keys 和 segment_keys 向量必须是不相交的。

如果不提供度量（measures），则只会得到唯一键的列表。

此节点首先输出段键，然后是常规键，最后是每个聚合的一列。

公共函数

inline explicit AggregateNodeOptions(std::vector<Aggregate> aggregates, std::vector<FieldRef> keys = {}, std::vector<FieldRef> segment_keys = {})#: 从值创建实例

公共成员

std::vector<Aggregate> aggregates#

std::vector<FieldRef> keys#

std::vector<FieldRef> segment_keys#

class BackpressureMonitor#

#include <arrow/acero/options.h>

可查询背压统计信息的接口

公共函数

virtual ~BackpressureMonitor() = default#

virtual uint64_t bytes_in_use() = 0#: 获取当前排队的字节数

virtual bool is_paused() = 0#: 检查是否当前应用了背压

struct BackpressureOptions#

#include <arrow/acero/options.h>

控制背压行为的选项。

公共函数

inline BackpressureOptions()#: 创建不执行背压的默认选项。

inline BackpressureOptions(uint64_t resume_if_below, uint64_t pause_if_above)#

创建执行背压的选项。

参数:

resume_if_below – 如果背压队列中的项目少于 resume_if_below，生产者应恢复生产。
pause_if_above – 如果背压队列中的项目多于 pause_if_above，生产者应暂停生产。

inline bool should_apply_backpressure() const#

确定背压是否已禁用的辅助方法

返回：: 如果 pause_if_above 大于零，则返回 true，否则返回 false

公共成员

uint64_t resume_if_below#: 生产者应恢复生产的字节数

uint64_t pause_if_above#

生产者应暂停生产的字节数

如果此值 <= 0，则背压将被禁用

公共静态函数

static inline BackpressureOptions DefaultBackpressure()#: 使用背压限制的默认值创建实例

class SinkNodeOptions : public arrow::acero::ExecNodeOptions #

#include <arrow/acero/options.h>

一个在队列中收集结果的 Sink 节点

仅在存在有意义的顺序且 sequence_output 未设置为 false 时，发出的批次才会是有序的。

子类包括 arrow::acero::OrderBySinkNodeOptions, arrow::acero::SelectKSinkNodeOptions

公共函数

inline explicit SinkNodeOptions(std::function<Future<std::optional<ExecBatch>>()> *generator, std::shared_ptr<Schema> *schema, BackpressureOptions backpressure = {}, BackpressureMonitor **backpressure_monitor = NULLPTR, std::optional<bool> sequence_output = std::nullopt)#

inline explicit SinkNodeOptions(std::function<Future<std::optional<ExecBatch>>()> *generator, BackpressureOptions backpressure = {}, BackpressureMonitor **backpressure_monitor = NULLPTR, std::optional<bool> sequence_output = std::nullopt)#

公共成员

std::function<Future<std::optional<ExecBatch>>()> *generator#

指向批处理生成器的指针。

此指针将在节点添加到执行计划时被设置，并应被用于从计划中消费数据。如果此函数调用不够频繁，Sink 节点将开始堆积数据，并可能会应用反压（Backpressure）。

std::shared_ptr<Schema> *schema#

一个将被设置为所生成批处理模式（Schema）的指针。

这是可选的，如果传入 nullptr 则会被忽略。此项将在节点添加到计划时，在调用 StartProducing 之前被设置。

BackpressureOptions backpressure#

控制何时应用反压的选项。

这是可选的，默认是不应用反压。如果计划消费速度不够快，系统最终可能会耗尽内存。

BackpressureMonitor **backpressure_monitor#

指向反压监控器的指针。

此指针将在节点添加到计划时被设置。可用于检查当前在 Sink 节点中排队的数据量。这是一个可选的实用工具，即使不使用它，也可以应用反压。

std::optional<bool> sequence_output#

控制批处理是应该立即发出还是按顺序排列。

另请参阅

有关详细信息，请参阅 QueryOptions。

class BackpressureControl#

#include <arrow/acero/options.h>

SinkNodeConsumer 用于暂停和恢复的控制器。

调用者应确保不会同时调用 Pause 和 Resume，并应按顺序进行操作，以便对 Pause() 的调用始终伴随最终对 Resume() 的调用。

公共函数

virtual ~BackpressureControl() = default#

virtual void Pause() = 0#

请求输入端暂停。

这是一种尽力而为的操作，批处理可能会继续到达。必须最终跟随对 Resume() 的调用，否则会导致死锁。

virtual void Resume() = 0#: 请求输入端恢复。

class SinkNodeConsumer#

#include <arrow/acero/options.h>

一种在执行计划中通过回调消费数据的 Sink 节点。

由 arrow::acero::NullSinkNodeConsumer, arrow::acero::TableSinkNodeConsumer 子类化。

公共函数

virtual ~SinkNodeConsumer() = default#

virtual Status Init(const std::shared_ptr<Schema> &schema, BackpressureControl *backpressure_control, ExecPlan *plan) = 0#

准备消费者状态。

此函数将在计划开始且模式最终确定后，在任何 Consume 调用之前运行。常见用途是保存模式，以便能够解读批处理。

virtual Status Consume(ExecBatch batch) = 0#: 消费数据批处理。

virtual Future Finish() = 0#

向消费者发出最后一个批处理已交付的信号。

返回的 future 仅在所有未完成的任务完成后才应完成。

如果计划提前结束或因错误而中止，则不会调用此函数。

class ConsumingSinkNodeOptions : public arrow::acero::ExecNodeOptions #

#include <arrow/acero/options.h>

添加一个在执行计划运行期间消费数据的 Sink 节点。

公共函数

inline explicit ConsumingSinkNodeOptions(std::shared_ptr<SinkNodeConsumer> consumer, std::vector<std::string> names = {}, std::optional<bool> sequence_output = std::nullopt)#

公共成员

std::shared_ptr<SinkNodeConsumer> consumer#

std::vector<std::string> names#

用于重命名 Sink 模式字段的名称。

如果指定了名称，则必须为所有字段提供名称。目前仅支持平面模式 (参见 GH-31875)。

如果未指定，名称将基于源数据生成。

std::optional<bool> sequence_output#

控制批处理是应该立即发出还是按顺序排列。

另请参阅

有关详细信息，请参阅 QueryOptions。

class OrderBySinkNodeOptions : public arrow::acero::SinkNodeOptions #

#include <arrow/acero/options.h>

创建一个对流经它的行进行排序的节点。

所有推送到此节点的批处理都将被累积，然后按给定的字段进行排序。随后，排序后的批处理将按顺序转发给生成器。

公共函数

inline explicit OrderBySinkNodeOptions(SortOptions sort_options, std::function<Future<std::optional<ExecBatch>>()> *generator)#: 从值创建实例

公共成员

SortOptions sort_options#: 描述要排序的列和方向的选项。

class OrderByNodeOptions : public arrow::acero::ExecNodeOptions #

#include <arrow/acero/options.h>

应用新的数据排序。

目前，此节点的工作方式是累积所有数据，进行排序，然后发出带有更新批处理索引的新数据。

目前不支持超出内存的排序。

公共函数

inline explicit OrderByNodeOptions(Ordering ordering)#

公共成员

Ordering ordering#: 应用于传出数据的新排序。

公共静态属性

static constexpr std::string_view kName = "order_by"#

class HashJoinNodeOptions : public arrow::acero::ExecNodeOptions #

#include <arrow/acero/options.h>

使用哈希表实现连接操作的节点。

公共函数

inline HashJoinNodeOptions(JoinType in_join_type, std::vector<FieldRef> in_left_keys, std::vector<FieldRef> in_right_keys, Expression filter = literal(true), std::string output_suffix_for_left = default_output_suffix_for_left, std::string output_suffix_for_right = default_output_suffix_for_right, bool disable_bloom_filter = false)#: 从输出所有列的值创建一个实例。

inline HashJoinNodeOptions(std::vector<FieldRef> in_left_keys, std::vector<FieldRef> in_right_keys)#

从键创建一个实例。

这将创建一个输出所有列且没有连接后过滤器的内连接（inner join）。

in_left_keys 应与 in_right_keys 具有相同的长度和类型。

参数:

in_left_keys – 左侧输入的键。
in_right_keys – 右侧输入的键。

inline HashJoinNodeOptions(JoinType join_type, std::vector<FieldRef> left_keys, std::vector<FieldRef> right_keys, std::vector<FieldRef> left_output, std::vector<FieldRef> right_output, Expression filter = literal(true), std::string output_suffix_for_left = default_output_suffix_for_left, std::string output_suffix_for_right = default_output_suffix_for_right, bool disable_bloom_filter = false)#: 创建一个使用 JoinKeyCmp::EQ 进行所有比较的值的实例。

inline HashJoinNodeOptions(JoinType join_type, std::vector<FieldRef> left_keys, std::vector<FieldRef> right_keys, std::vector<FieldRef> left_output, std::vector<FieldRef> right_output, std::vector<JoinKeyCmp> key_cmp, Expression filter = literal(true), std::string output_suffix_for_left = default_output_suffix_for_left, std::string output_suffix_for_right = default_output_suffix_for_right, bool disable_bloom_filter = false)#: 从值创建实例

HashJoinNodeOptions() = default#

公共成员

JoinType join_type = JoinType::INNER #

std::vector<FieldRef> left_keys#

std::vector<FieldRef> right_keys#

bool output_all = false#

std::vector<FieldRef> left_output#

std::vector<FieldRef> right_output#

std::vector<JoinKeyCmp> key_cmp#

std::string output_suffix_for_left#

std::string output_suffix_for_right#

Expression filter = literal(true)#

bool disable_bloom_filter = false#

公共静态属性

static constexpr const char *default_output_suffix_for_left = ""#

static constexpr const char *default_output_suffix_for_right = ""#

class AsofJoinNodeOptions : public arrow::acero::ExecNodeOptions #

#include <arrow/acero/options.h>

实现 Asof Join 操作的节点。

注意，此 API 是实验性的，将来会发生变化。

此节点接收一个左表和任意数量的右表，并将它们进行 Asof 连接。每个输入产生的批处理必须按“on”键排序。此节点将为左表中的每一行输出一行。

公共函数

inline AsofJoinNodeOptions(std::vector<Keys> input_keys, int64_t tolerance)#

公共成员

std::vector<Keys> input_keys#

每个输入表的 AsofJoin 键。

必须至少提供两个键。第一个键对应左表，所有其他键对应 Asof-Join 的右表。

另请参阅

有关详细信息，请参阅 Keys。

int64_t tolerance#

非精确“on”键匹配的容差。

如果 right.on - left.on 在 [min(0, tolerance), max(0, tolerance)] 范围内，则右侧行被视为与左侧行匹配。tolerance 可以是：

负数，此时发生 Past-As-Of-Join（匹配条件为 tolerance <= right.on - left.on <= 0）；
或正数，此时发生 Future-As-Of-Join（匹配条件为 0 <= right.on - left.on <= tolerance）；
或零，此时发生精确 As-Of-Join（匹配条件为 right.on == left.on）。

容差的单位与“on”键相同。

struct Keys#

#include <arrow/acero/options.h>

AsofJoin 操作中一个输入表的键。

键必须在各输入表之间保持一致：每个“on”键必须引用各表中具有相同类型和单位的字段。每个“by”键必须引用各表中具有相同类型的一组字段。

公共成员

FieldRef on_key#

连接的“on”键。

输入表必须按“on”键排序。必须是具有公共类型的单个字段。“on”键使用非精确匹配，即当且仅当 right.on - left.on 在 [min(0, tolerance), max(0, tolerance)] 范围内时，行才被视为匹配。目前，“on”键必须是整数、日期或时间戳类型。

std::vector<FieldRef> by_key#

连接的“by”键。

每个输入表必须包含“by”键的每个字段。“by”键的每个字段使用精确相等比较。目前，“by”键的每个字段必须是整数、日期、时间戳或基础二进制类型。

class SelectKSinkNodeOptions : public arrow::acero::SinkNodeOptions #

#include <arrow/acero/options.h>

一种选择流经它的 top_k/bottom_k 行的节点。

所有推送到此节点的批处理都将被累积，然后按给定的字段进行选择。随后，排序后的批处理将按顺序转发给生成器。

公共函数

inline explicit SelectKSinkNodeOptions(SelectKOptions select_k_options, std::function<Future<std::optional<ExecBatch>>()> *generator)#

公共成员

SelectKOptions select_k_options#: SelectK 选项。

class TableSinkNodeOptions : public arrow::acero::ExecNodeOptions #

#include <arrow/acero/options.h>

将所有输出累积到表中的 Sink 节点。

公共函数

inline explicit TableSinkNodeOptions(std::shared_ptr<Table> *output_table, std::optional<bool> sequence_output = std::nullopt)#: 从值创建实例

公共成员

std::shared_ptr<Table> *output_table#

一个指定将要创建的表的“输出参数”。

不得为空，并且在整个计划执行期间必须保持有效。在计划完成后，该参数将被设置为指向结果表。

std::optional<bool> sequence_output#

控制批处理是应该立即发出还是按顺序排列。

另请参阅

有关详细信息，请参阅 QueryOptions。

std::vector<std::string> names#

用于列的自定义名称。

如果指定了名称，则必须为所有字段提供名称。目前仅支持平面模式 (参见 GH-31875)。

如果未指定，名称将基于源数据生成。

struct PivotLongerRowTemplate#

#include <arrow/acero/options.h>

一种行模板，描述了将为每个输入行生成的行。

公共函数

inline PivotLongerRowTemplate(std::vector<std::string> feature_values, std::vector<std::optional<FieldRef>> measurement_values)#

公共成员

std::vector<std::string> feature_values#

模板的一组（通常是唯一的）特征值，通常派生自列名。

这些值将用于填充特征列。

std::vector<std::optional<FieldRef>> measurement_values#

包含此行所使用的度量值的字段。

这些字段将用于填充度量列。如果为 nullopt，则给定值将插入空值 (null)。

class PivotLongerNodeOptions : public arrow::acero::ExecNodeOptions #

#include <arrow/acero/options.h>

通过将某些列转换为额外的行来重塑表格。

此操作有时也称为 UNPIVOT（逆透视）。

这通常用于当每一行中存在多个观测值时，以便将其转换为每行仅包含一个观测值的表格。

例如：

time	left_temp	right_temp
1	10	20
2	15	18

上表每行包含两个观测值。存在一个隐含的特征“位置”（左侧与右侧）和一个度量“temp”。我们真正想要的是

time	location	temp
1	left	10
1	right	20
2	left	15
2	right	18

对于一个更复杂的示例，请考虑

time	ax1	ay1	bx1	ay2
0	1	2	3	4

我们可以假定 a 与 b 以及 x 与 y 是特征，而 1 和 2 是两种不同的度量。因此，我们想要透视到

a/b	x/y	f1	f2
a	x	1	空 (null)
a	y	2	4
b	x	3	空 (null)

为此，我们为每种特征组合创建一个行模板。仅通过查看列名应该就能做到这一点。例如，给定上述列“ax1”、“ay1”、“bx1”和“ay2”，我们知道我们有三种特征组合 (a, x)、(a, y) 和 (b, x)。同样，我们知道我们有两种可能的度量，“1”和“2”。

对于每种特征组合，我们创建一个行模板。在每个行模板中，我们描述该组合，然后列出用于度量的列。如果某种组合不存在度量，则使用 nullopt。

因此，对于上面的例子，我们有

(a, x): names={“a”, “x”}, values={“ax1”, nullopt} (a, y): names={“a”, “y”}, values={“ay1”, “ay2”} (b, x): names={“b”, “x”}, values={“bx1”, nullopt}

最后，我们将新列命名为：feature_field_names={“a/b”,”x/y”} measurement_field_names={“f1”, “f2”}

公共成员

std::vector<PivotLongerRowTemplate> row_templates#

用于创建新输出行的一个或多个行模板。

通常至少有两个行模板。输出的行数将是输入行数 * 行模板数量。

std::vector<std::string> feature_field_names#: 描述新特征的列名。

std::vector<std::string> measurement_field_names#: 代表度量的列名。

公共静态属性

static constexpr std::string_view kName = "pivot_longer"#

创建自定义节点的内部机制#

ExecFactoryRegistry *default_exec_factory_registry()#: 默认注册表，包含内置工厂。

inline Result<ExecNode*> MakeExecNode(const std::string &factory_name, ExecPlan *plan, std::vector<ExecNode*> inputs, const ExecNodeOptions &options, ExecFactoryRegistry *registry = default_exec_factory_registry())#: 使用指定的工厂构建 ExecNode。

inline bool operator==(const ExecBatch &l, const ExecBatch &r)#

inline bool operator!=(const ExecBatch &l, const ExecBatch &r)#

void PrintTo(const ExecBatch&, std::ostream*)#

class ExecPlan : public std::enable_shared_from_this<ExecPlan>#

#include <arrow/acero/exec_plan.h>

公共类型

using NodeVector = std::vector<ExecNode*>#

公共函数

virtual ~ExecPlan() = default#

QueryContext *query_context()#

const NodeVector &nodes() const#: 检索计划中的节点

ExecNode *AddNode(std::unique_ptr<ExecNode> node)#

template<typename Node, typename ...Args> inline Node *EmplaceNode(Args&&... args)#

Status Validate()#

void StartProducing()#

开始在所有节点上生产。

节点以反向拓扑顺序启动，确保任何节点在其所有输入启动之前启动。

void StopProducing()#

停止在所有节点上生产。

触发所有源停止生产新数据。为了干净地停止，计划将继续运行已在处理中的任何任务。调用者在销毁计划之前仍应等待 finished 完成。

Future finished()#: 一个将在所有任务完成时被标记为完成的 future。

bool HasMetadata() const#: 返回计划是否具有非空元数据。

std::shared_ptr<const KeyValueMetadata> metadata() const#: 返回计划附加的元数据。

std::string ToString() const#

公共静态函数

static Result<std::shared_ptr<ExecPlan>> Make(QueryOptions options, ExecContext exec_context = *threaded_exec_context(), std::shared_ptr<const KeyValueMetadata> metadata = NULLPTR)#: 创建一个空的执行计划。

static Result<std::shared_ptr<ExecPlan>> Make(ExecContext exec_context = *threaded_exec_context(), std::shared_ptr<const KeyValueMetadata> metadata = NULLPTR)#

static Result<std::shared_ptr<ExecPlan>> Make(QueryOptions options, ExecContext *exec_context, std::shared_ptr<const KeyValueMetadata> metadata = NULLPTR)#

static Result<std::shared_ptr<ExecPlan>> Make(ExecContext *exec_context, std::shared_ptr<const KeyValueMetadata> metadata = NULLPTR)#

公共静态属性

static const uint32_t kMaxBatchSize = 1 << 15#

class ExecNode#

#include <arrow/acero/exec_plan.h>

被 arrow::acero::MapNode 等子类化

公共类型

using NodeVector = std::vector<ExecNode*>#

公共函数

virtual ~ExecNode() = default#

virtual const char *kind_name() const = 0#

inline int num_inputs() const#

inline const NodeVector &inputs() const#: 此节点在执行计划中的前驱节点。

inline bool is_sink() const#: 如果计划没有输出模式（即接收器），则为 True

inline const std::vector<std::string> &input_labels() const#: 标识每个输入功能的标签。

inline const ExecNode *output() const#: 此节点在执行计划中的后继节点。

inline const std::shared_ptr<Schema> &output_schema() const#: 此节点产生的批处理的数据类型。

inline ExecPlan *plan()#: 此节点的执行计划。

inline const std::string &label() const#

用于显示和调试的可选标签。

不保证此值是非空或唯一的。

inline void SetLabel(std::string label)#

virtual Status Validate() const#

virtual const Ordering &ordering() const#

输出批处理的排序

这并不保证批处理将按顺序由该节点发出。相反，它保证批处理将以符合此排序的方式设置其 ExecBatch::index 属性。

换句话说，给定排序 {{“x”, SortOrder::Ascending}}，我们知道批处理索引为 N 的批处理中 x 的所有值将小于或等于批处理索引为 N+k（假设 k > 0）的批处理中 x 的所有值。此外，我们也知道批处理内部的值将是排好序的。任何行 N 的 x 值都将小于任何行 N+k 的 x 值。

注意，排序可以是 Ordering::Unordered 也可以是 Ordering::Implicit。如果顺序是非确定性的，节点的输出应标记为 Ordering::Unordered。例如，哈希连接没有可预测的输出顺序。

如果排序是 Ordering::Implicit，

则存在有意义的顺序，但该顺序并未由数据中的任何列表示。最常见的情况是从内存中的表格读取数据时。数据具有隐含的“行

顺序”，而不一定在数据集中表示。

过滤器或投影节点不会修改排序。除了确保分配给输出批处理的索引与映射的输入批处理相同之外，无需执行任何操作。

其他节点可能会引入顺序。例如，order-by 节点将发出一种完全独立于输入排序的新排序。

最后，如上所述，诸如哈希连接或聚合之类的操作可能会破坏排序（尽管这些节点也可以选择根据哈希键建立新的排序）。

某些节点需要排序。例如，fetch 节点或 asof join 节点仅在输入数据已排序时才有效（对于 fetch，隐式排序就足够了。对于 asof join，排序必须是显式的且与 on 键兼容。）

维护排序的节点应小心避免在批处理索引中引入间隙。这可能需要发出空的批处理以保持连续性。

virtual Status InputReceived(ExecNode *input, ExecBatch batch) = 0#

上游 API：这些函数由希望通知此节点有关更新条件（新输入批处理或即将结束的流）的输入节点调用。

实现规则

这些可以在 StartProducing() 成功后随时调用（甚至在 StopProducing() 期间或之后）
这些可以并发调用
这些允许回调到 PauseProducing()、ResumeProducing() 和 StopProducing() 将输入批处理传输到 ExecNode

节点通常会对批处理执行某种操作，然后调用其输出节点的 InputReceived，并将结果传递过去。

其他节点可能需要在产生任何输出之前累积一定数量的输入。这些节点将把批处理添加到某种内存累积队列中，然后返回。

virtual Status InputFinished(ExecNode *input, int total_batches) = 0#

在给定批次数后标记输入完成。

这可以在接收到所有输入之前调用。这仅固定了输入端传入的批次总数，以便 ExecNode 知道何时接收到了所有输入，而无论顺序如何。

virtual Status Init()#

执行任何必要的初始化。

此挂钩在创建 ExecPlan 和调用 StartProducing 之间执行任何操作。例如 Bloom 过滤器下推。执行此方法的 ExecNodes 的顺序是不确定的，但调用是同步进行的。

此时，节点可以依赖于所有输入和输出（以及输入模式）的良好定义。

virtual Status StartProducing() = 0#

生命周期 API

启动/停止以初始化和终止生产
暂停/恢复以应用背压

实现规则

StartProducing() 不应递归到输入中，因为它由 ExecPlan::StartProducing() 处理
PauseProducing()、ResumeProducing()、StopProducing() 可能会并发调用，甚至可能在 StartProducing 调用完成之前。
PauseProducing()、ResumeProducing()、StopProducing() 可能会被下游节点的 InputReceived()、InputFinished() 方法调用

StopProducing 可能因错误、用户（例如取消）或节点确定不再需要更多输入（例如限制节点）而调用。这意味着该方法可能会被多次调用，并且我们有以下附加规则

StopProducing() 必须是幂等的
StopProducing() 必须转发到输入（这对于限制/top-k 情况是必需的，因为我们可能不会停止整个计划）

开始生产

这必须只调用一次。

这通常由 ExecPlan::StartProducing() 自动调用。

virtual void PauseProducing(ExecNode *output, int32_t counter) = 0#

暂时暂停生产。

此调用提示输出节点当前不愿意接收数据。

此调用可以调用任意次数。但是，节点仍然可以自由地生产数据（如果数据是使用多个线程生产的，无论如何可能很难防止这种情况）。

参数:

output – 指向已满的输出的指针
counter – 用于对暂停/恢复调用进行排序的计数器

virtual void ResumeProducing(ExecNode *output, int32_t counter) = 0#

在暂时暂停后恢复生产。

此调用提示输出节点再次愿意接收数据。

此调用可以调用任意次数。

参数:

output – 指向现在空闲的输出的指针
counter – 用于对暂停/恢复调用进行排序的计数器

virtual Status StopProducing()#

停止生产新数据。

如果此节点是源，则源应尽快停止生成数据。如果此节点不是源，则通常无需执行任何操作，尽管节点可以选择开始忽略传入的数据。

当计划中发生错误时，将调用此方法。如果用户希望提前结束计划，也可以调用此方法。最后，如果节点确定不再需要任何输入（例如，限制节点），则可能会调用此方法。

此方法可能会被多次调用。

这不是暂停。调用此方法后，将无法再次启动源。

std::string ToString(int indent = 0) const#

class ExecFactoryRegistry#

#include <arrow/acero/exec_plan.h>

ExecNodes 工厂的可扩展注册表。

公共类型

using Factory = std::function<Result<ExecNode*>(ExecPlan*, std::vector<ExecNode*>, const ExecNodeOptions&)>#

公共函数

virtual ~ExecFactoryRegistry() = default#

virtual Result<Factory> GetFactory(const std::string &factory_name) = 0#

从该注册表获取指定的工厂。

如果未找到 factory_name，将引发错误

virtual Status AddFactory(std::string factory_name, Factory factory) = 0#

将具有给定名称的工厂添加到此注册表中。

如果 factory_name 已在注册表中，将引发错误

struct ExecBatch#: #include <arrow/compute/exec.h>