Acero 使用者指南#

本頁說明如何使用 Acero。建議您先閱讀概述，並熟悉基本概念。

使用 Acero#

Acero 的基本工作流程如下：

首先，建立描述計畫的 Declaration 物件圖
呼叫 DeclarationToXyz 方法之一來執行 Declaration。
1. 從 Declaration 的圖形建立新的 ExecPlan。每個 Declaration 將對應於計畫中的一個 ExecNode。此外，將根據使用的 DeclarationToXyz 方法新增一個 sink 節點。
2. ExecPlan 會被執行。通常這會作為 DeclarationToXyz 呼叫的一部分發生，但在 DeclarationToReader 中，讀取器會在計畫完成執行之前返回。
3. 計畫完成後，它會被銷毀

建立計畫#

使用 Substrait#

Substrait 是建立計畫（Declaration 圖形）的首選機制。原因如下：

Substrait 生產者花費大量時間和精力來建立使用者友善的 API，以便以簡單的方式產生複雜的執行計畫。例如，可以使用一系列複雜的 aggregate 節點來實現 pivot_wider 操作。生產者將為您提供更簡單的 API，而不是手動建立所有這些 aggregate 節點。
如果您使用 Substrait，那麼您可以輕鬆切換到任何其他 Substrait 消費引擎，如果您在某個時候發現它比 Acero 更能滿足您的需求。
我們希望最終會出現基於 Substrait 的最佳化器和規劃器的工具。透過使用 Substrait，您將更容易在未來使用這些工具。

您可以自行建立 Substrait 計畫，但您可能會更容易找到現有的 Substrait 生產者。例如，您可以使用 ibis-substrait 從 python 表達式輕鬆建立 Substrait 計畫。有一些不同的工具能夠從 SQL 建立 Substrait 計畫。最終，我們希望基於 C++ 的 Substrait 生產者將會出現。但是，我們目前尚未得知任何相關資訊。

有關從 Substrait 建立執行計畫的詳細說明，請參閱 Substrait 頁面

程式化計畫建立#

以程式化方式建立執行計畫比從 Substrait 建立計畫更簡單，儘管會失去一些彈性和未來驗證保證。建立 Declaration 的最簡單方法是直接實例化一個。您將需要 declaration 的名稱、輸入向量和選項物件。例如：

/// \brief An example showing a project node
///
/// Scan-Project-Table
/// This example shows how a Scan operation can be used to load the data
/// into the execution plan, how a project operation can be applied on the
/// data stream and how the output is collected into a table
arrow::Status ScanProjectSinkExample() {
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::dataset::Dataset> dataset, GetDataset());

  auto options = std::make_shared<arrow::dataset::ScanOptions>();
  // projection
  cp::Expression a_times_2 = cp::call("multiply", {cp::field_ref("a"), cp::literal(2)});
  options->projection = cp::project({}, {});

  auto scan_node_options = arrow::dataset::ScanNodeOptions{dataset, options};

  ac::Declaration scan{"scan", std::move(scan_node_options)};
  ac::Declaration project{
      "project", {std::move(scan)}, ac::ProjectNodeOptions({a_times_2})};

  return ExecutePlanAndCollectAsTable(std::move(project));
}

上述程式碼建立了一個掃描 declaration（沒有輸入）和一個投影 declaration（使用掃描作為輸入）。這已經夠簡單了，但我們可以讓它稍微容易一些。如果您要建立 declaration 的線性序列（如上述範例），那麼您也可以使用 Declaration::Sequence() 函數。

  // Inputs do not have to be passed to the project node when using Sequence
  ac::Declaration plan =
      ac::Declaration::Sequence({{"scan", std::move(scan_node_options)},
                                 {"project", ac::ProjectNodeOptions({a_times_2})}});

本文檔稍後將提供更多關於程式化計畫建立的範例。

執行計畫#

有許多不同的方法可用於執行 declaration。每種方法都以稍微不同的形式提供資料。由於所有這些方法都以 DeclarationTo... 開頭，因此本指南通常將這些方法稱為 DeclarationToXyz 方法。

DeclarationToTable#

DeclarationToTable() 方法會將所有結果累積到單個 arrow::Table 中。這可能是從 Acero 收集結果的最簡單方法。這種方法的主要缺點是它需要將所有結果累積到記憶體中。

注意

Acero 以小塊處理大型資料集。這在開發者指南中有更詳細的描述。因此，您可能會驚訝地發現，使用 DeclarationToTable 收集的表格與您的輸入相比，是以不同的區塊劃分的。例如，您的輸入可能是一個包含單個區塊（200 萬列）的大型表格。然後，您的輸出表格可能會有 64 個區塊，每個區塊包含 32Ki 列。目前有一個請求，要求在 GH-15155 中指定輸出的區塊大小。

DeclarationToReader#

DeclarationToReader() 方法允許您迭代地使用結果。它將建立一個 arrow::RecordBatchReader，您可以隨時從中讀取。如果您沒有及時從讀取器中讀取，則會套用背壓，並且執行計畫將會暫停。關閉讀取器將取消正在執行的執行計畫，並且讀取器的解構函數將等待執行計畫完成其正在執行的任何操作，因此可能會被封鎖。

DeclarationToStatus#

DeclarationToStatus() 方法在您想要執行計畫但不實際想要使用結果時很有用。例如，這在基準測試或計畫具有副作用（例如資料集寫入節點）時很有用。如果計畫產生任何結果，它們將立即被丟棄。

直接執行計畫#

如果 DeclarationToXyz 方法之一因某種原因不足以滿足需求，則可以直接執行計畫。只有在您正在執行某些獨特的操作時才需要這樣做。例如，如果您建立了一個自訂 sink 節點，或者您需要一個具有多個輸出的計畫。

注意

在學術文獻和許多現有系統中，普遍假設執行計畫最多只有一個輸出。Acero 中的某些東西（例如 DeclarationToXyz 方法）會預期如此。但是，設計中沒有任何東西嚴格禁止擁有多個 sink 節點。

關於如何執行此操作的詳細說明超出了本指南的範圍，但大致步驟如下：

建立新的 ExecPlan 物件。
將 sink 節點新增到您的 Declaration 物件圖形中（這是您需要為 sink 節點建立 declaration 的唯一類型）
使用 Declaration::AddToPlan() 將您的 declaration 新增到您的計畫中（如果您有多個輸出，那麼您將無法使用此方法，並且需要一次新增一個節點）
使用 ExecPlan::Validate() 驗證計畫
使用 ExecPlan::StartProducing() 啟動計畫
等待 ExecPlan::finished() 返回的 future 完成。

提供輸入#

執行計畫的輸入資料可以來自各種來源。它通常從儲存在某種檔案系統上的檔案中讀取。輸入也通常來自記憶體中的資料。記憶體中的資料是典型的，例如，在類似 pandas 的前端中。輸入也可能來自網路串流，例如 Flight 請求。Acero 可以支援所有這些情況，甚至可以支援此處未提及的獨特和自訂情況。

有預先定義的來源節點涵蓋最常見的輸入情境。這些節點列在下方。但是，如果您的來源資料是獨特的，那麼您將需要使用通用的 source 節點。此節點期望您提供批次的非同步串流，更詳細的資訊請參閱此處。

可用的 `ExecNode` 實作#

下表快速總結了可用的運算子。

來源#

這些節點可以用作資料來源

來源節點#
Factory 名稱	選項	簡短描述
`source`	`SourceNodeOptions`	一個通用的來源節點，它包裝了資料的非同步串流 (範例)
`table_source`	`TableSourceNodeOptions`	從 `arrow::Table` 產生資料 (範例)
`record_batch_source`	`RecordBatchSourceNodeOptions`	從 `arrow::RecordBatch` 的迭代器產生資料
`record_batch_reader_source`	`RecordBatchReaderSourceNodeOptions`	從 `arrow::RecordBatchReader` 產生資料
`exec_batch_source`	`ExecBatchSourceNodeOptions`	從 `arrow::compute::ExecBatch` 的迭代器產生資料
`array_vector_source`	`ArrayVectorSourceNodeOptions`	從 `arrow::Array` 向量的迭代器產生資料
`scan`	`arrow::dataset::ScanNodeOptions`	從 `arrow::dataset::Dataset` 產生資料（需要 datasets 模組）(範例)

計算節點#

這些節點對資料執行計算，並且可能會轉換或重塑資料

計算節點#
Factory 名稱	選項	簡短描述
`filter`	`FilterNodeOptions`	移除與給定篩選表達式不符的列 (範例)
`project`	`ProjectNodeOptions`	透過評估計算表達式來建立新欄位。也可以捨棄和重新排序欄位 (範例)
`aggregate`	`AggregateNodeOptions`	計算整個輸入串流或資料群組的摘要統計資訊 (範例)
`pivot_longer`	`PivotLongerNodeOptions`	透過將某些欄位轉換為額外的列來重塑資料

排列節點#

這些節點重新排序、組合或分割資料串流

排列節點#
Factory 名稱	選項	簡短描述
`hash_join`	`HashJoinNodeOptions`	根據通用欄位聯結兩個輸入 (範例)
`asofjoin`	`AsofJoinNodeOptions`	根據通用的排序欄位（通常是時間）將多個輸入聯結到第一個輸入
`union`	不適用	合併具有相同結構描述的兩個輸入 (範例)
`order_by`	`OrderByNodeOptions`	重新排序串流
`fetch`	`FetchNodeOptions`	從串流中分割一系列列

Sink 節點#

這些節點終止計畫。使用者通常不會建立 sink 節點，因為它們是根據用於消耗計畫的 DeclarationToXyz 方法選擇的。但是，此列表對於正在開發新 sink 節點或以進階方式使用 Acero 的人來說可能很有用。

Sink 節點#
Factory 名稱	選項	簡短描述
`sink`	`SinkNodeOptions`	將批次收集到具有可選背壓的 FIFO 佇列中
`write`	`arrow::dataset::WriteNodeOptions`	將批次寫入檔案系統 (範例)
`consuming_sink`	`ConsumingSinkNodeOptions`	使用使用者提供的回呼函數消耗批次
`table_sink`	`TableSinkNodeOptions`	將批次收集到 `arrow::Table` 中
`order_by_sink`	`OrderBySinkNodeOptions`	已棄用
`select_k_sink`	`SelectKSinkNodeOptions`	已棄用

範例#

本文檔的其餘部分包含執行計畫範例。每個範例都突顯了特定執行節點的行為。

`source`#

source 操作可以被視為建立串流執行計畫的入口點。SourceNodeOptions 用於建立 source 操作。source 操作是目前可用的最通用和最靈活的來源類型，但配置起來可能相當棘手。首先，您應該查看其他來源節點類型，以確保沒有更簡單的選擇。

source 節點需要某種函數，可以呼叫該函數來輪詢更多資料。此函數不應接受任何引數，並且應返回 arrow::Future<std::optional<arrow::ExecBatch>>。此函數可能正在讀取檔案、迭代記憶體結構或從網路連線接收資料。arrow 程式庫將這些函數稱為 arrow::AsyncGenerator，並且有許多用於處理這些函數的工具程式。在此範例中，我們使用已儲存在記憶體中的 record batch 向量。此外，資料的結構描述必須事先已知。Acero 必須在執行圖的每個階段都知道資料的結構描述，然後才能開始任何處理。這表示我們必須將來源節點的結構描述與資料本身分開提供。

在這裡，我們定義了一個結構來保存資料產生器定義。這包括記憶體中的批次、結構描述和充當資料產生器的函數

struct BatchesWithSchema {
  std::vector<cp::ExecBatch> batches;
  std::shared_ptr<arrow::Schema> schema;
  // This method uses internal arrow utilities to
  // convert a vector of record batches to an AsyncGenerator of optional batches
  arrow::AsyncGenerator<std::optional<cp::ExecBatch>> gen() const {
    auto opt_batches = ::arrow::internal::MapVector(
        [](cp::ExecBatch batch) { return std::make_optional(std::move(batch)); },
        batches);
    arrow::AsyncGenerator<std::optional<cp::ExecBatch>> gen;
    gen = arrow::MakeVectorGenerator(std::move(opt_batches));
    return gen;
  }
};

產生用於計算的範例批次

arrow::Result<BatchesWithSchema> MakeBasicBatches() {
  BatchesWithSchema out;
  auto field_vector = {arrow::field("a", arrow::int32()),
                       arrow::field("b", arrow::boolean())};
  ARROW_ASSIGN_OR_RAISE(auto b1_int, GetArrayDataSample<arrow::Int32Type>({0, 4}));
  ARROW_ASSIGN_OR_RAISE(auto b2_int, GetArrayDataSample<arrow::Int32Type>({5, 6, 7}));
  ARROW_ASSIGN_OR_RAISE(auto b3_int, GetArrayDataSample<arrow::Int32Type>({8, 9, 10}));

  ARROW_ASSIGN_OR_RAISE(auto b1_bool,
                        GetArrayDataSample<arrow::BooleanType>({false, true}));
  ARROW_ASSIGN_OR_RAISE(auto b2_bool,
                        GetArrayDataSample<arrow::BooleanType>({true, false, true}));
  ARROW_ASSIGN_OR_RAISE(auto b3_bool,
                        GetArrayDataSample<arrow::BooleanType>({false, true, false}));

  ARROW_ASSIGN_OR_RAISE(auto b1,
                        GetExecBatchFromVectors(field_vector, {b1_int, b1_bool}));
  ARROW_ASSIGN_OR_RAISE(auto b2,
                        GetExecBatchFromVectors(field_vector, {b2_int, b2_bool}));
  ARROW_ASSIGN_OR_RAISE(auto b3,
                        GetExecBatchFromVectors(field_vector, {b3_int, b3_bool}));

  out.batches = {b1, b2, b3};
  out.schema = arrow::schema(field_vector);
  return out;
}

使用 source 的範例（sink 的用法在 sink 中詳細說明）

/// \brief An example demonstrating a source and sink node
///
/// Source-Table Example
/// This example shows how a custom source can be used
/// in an execution plan. This includes source node using pregenerated
/// data and collecting it into a table.
///
/// This sort of custom source is often not needed.  In most cases you can
/// use a scan (for a dataset source) or a source like table_source, array_vector_source,
/// exec_batch_source, or record_batch_source (for in-memory data)
arrow::Status SourceSinkExample() {
  ARROW_ASSIGN_OR_RAISE(auto basic_data, MakeBasicBatches());

  auto source_node_options = ac::SourceNodeOptions{basic_data.schema, basic_data.gen()};

  ac::Declaration source{"source", std::move(source_node_options)};

  return ExecutePlanAndCollectAsTable(std::move(source));
}

`table_source`#

在先前的範例 source 節點中，使用 source 節點輸入資料。但在開發應用程式時，如果資料已經以表格形式存在於記憶體中，則使用 TableSourceNodeOptions 會更容易且效能更高。在這裡，輸入資料可以作為 std::shared_ptr<arrow::Table> 連同 max_batch_size 傳遞。max_batch_size 用於分解大型 record batch，以便可以平行處理它們。重要的是要注意，當來源表格的批次大小較小時，表格批次不會合併以形成更大的批次。

使用 table_source 的範例

/// \brief An example showing a table source node
///
/// TableSource-Table Example
/// This example shows how a table_source can be used
/// in an execution plan. This includes a table source node
/// receiving data from a table.  This plan simply collects the
/// data back into a table but nodes could be added that modify
/// or transform the data as well (as is shown in later examples)
arrow::Status TableSourceSinkExample() {
  ARROW_ASSIGN_OR_RAISE(auto table, GetTable());

  arrow::AsyncGenerator<std::optional<cp::ExecBatch>> sink_gen;
  int max_batch_size = 2;
  auto table_source_options = ac::TableSourceNodeOptions{table, max_batch_size};

  ac::Declaration source{"table_source", std::move(table_source_options)};

  return ExecutePlanAndCollectAsTable(std::move(source));
}

`filter`#

filter 操作，顧名思義，提供了定義資料篩選條件的選項。它選擇給定表達式評估為 true 的列。可以使用 arrow::compute::Expression 撰寫篩選器，並且表達式應具有布林值的傳回類型。例如，如果我們希望保留欄 b 的值大於 3 的列，那麼我們可以使用以下表達式。

篩選範例

/// \brief An example showing a filter node
///
/// Source-Filter-Table
/// This example shows how a filter can be used in an execution plan,
/// to filter data from a source. The output from the execution plan
/// is collected into a table.
arrow::Status ScanFilterSinkExample() {
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::dataset::Dataset> dataset, GetDataset());

  auto options = std::make_shared<arrow::dataset::ScanOptions>();
  // specify the filter.  This filter removes all rows where the
  // value of the "a" column is greater than 3.
  cp::Expression filter_expr = cp::greater(cp::field_ref("a"), cp::literal(3));
  // set filter for scanner : on-disk / push-down filtering.
  // This step can be skipped if you are not reading from disk.
  options->filter = filter_expr;
  // empty projection
  options->projection = cp::project({}, {});

  // construct the scan node
  std::cout << "Initialized Scanning Options" << std::endl;

  auto scan_node_options = arrow::dataset::ScanNodeOptions{dataset, options};
  std::cout << "Scan node options created" << std::endl;

  ac::Declaration scan{"scan", std::move(scan_node_options)};

  // pipe the scan node into the filter node
  // Need to set the filter in scan node options and filter node options.
  // At scan node it is used for on-disk / push-down filtering.
  // At filter node it is used for in-memory filtering.
  ac::Declaration filter{
      "filter", {std::move(scan)}, ac::FilterNodeOptions(std::move(filter_expr))};

  return ExecutePlanAndCollectAsTable(std::move(filter));
}

`project`#

project 操作會重新排列、刪除、轉換和建立欄位。每個輸出欄位都是透過針對來源 record batch 評估表達式來計算的。這些必須是純量表達式（由純量文字、欄位參考和純量函數組成的表達式，即元素函數，它們為每個輸入列傳回一個值，而與所有其他列的值無關）。這透過 ProjectNodeOptions 公開，它需要 arrow::compute::Expression 和每個輸出欄位的名稱（如果未提供名稱，則將使用 expr 的字串表示形式）。

投影範例

/// \brief An example showing a project node
///
/// Scan-Project-Table
/// This example shows how a Scan operation can be used to load the data
/// into the execution plan, how a project operation can be applied on the
/// data stream and how the output is collected into a table
arrow::Status ScanProjectSinkExample() {
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::dataset::Dataset> dataset, GetDataset());

  auto options = std::make_shared<arrow::dataset::ScanOptions>();
  // projection
  cp::Expression a_times_2 = cp::call("multiply", {cp::field_ref("a"), cp::literal(2)});
  options->projection = cp::project({}, {});

  auto scan_node_options = arrow::dataset::ScanNodeOptions{dataset, options};

  ac::Declaration scan{"scan", std::move(scan_node_options)};
  ac::Declaration project{
      "project", {std::move(scan)}, ac::ProjectNodeOptions({a_times_2})};

  return ExecutePlanAndCollectAsTable(std::move(project));
}

`aggregate`#

aggregate 節點計算資料的各種聚合。

Arrow 支援兩種聚合類型：「純量」聚合和「雜湊」聚合。純量聚合將陣列或純量輸入縮減為單個純量輸出（例如，計算欄位的平均值）。雜湊聚合的作用類似於 SQL 中的 GROUP BY，並首先根據一個或多個索引鍵欄位分割資料，然後縮減每個分割區中的資料。aggregate 節點同時支援這兩種計算類型，並且可以一次計算任意數量的聚合。

AggregateNodeOptions 用於定義聚合條件。它採用聚合函數及其選項的列表；要聚合的目標欄位列表（每個函數一個）；以及輸出欄位的名稱列表（每個函數一個）。或者，它可以採用用於分割資料的欄位列表（在雜湊聚合的情況下）。可以從此聚合函數列表中選擇聚合函數。

注意

此節點是「管線中斷器」，將完全實現記憶體中的資料集。未來將新增溢出機制，這應該可以減輕此限制。

聚合可以將結果作為群組或純量提供。例如，像 hash_count 這樣的操作會提供每個唯一記錄的計數作為分組結果，而像 sum 這樣的操作會提供單個記錄。

純量聚合範例

/// \brief An example showing an aggregation node to aggregate an entire table
///
/// Source-Aggregation-Table
/// This example shows how an aggregation operation can be applied on a
/// execution plan resulting in a scalar output. The source node loads the
/// data and the aggregation (counting unique types in column 'a')
/// is applied on this data. The output is collected into a table (that will
/// have exactly one row)
arrow::Status SourceScalarAggregateSinkExample() {
  ARROW_ASSIGN_OR_RAISE(auto basic_data, MakeBasicBatches());

  auto source_node_options = ac::SourceNodeOptions{basic_data.schema, basic_data.gen()};

  ac::Declaration source{"source", std::move(source_node_options)};
  auto aggregate_options =
      ac::AggregateNodeOptions{/*aggregates=*/{{"sum", nullptr, "a", "sum(a)"}}};
  ac::Declaration aggregate{
      "aggregate", {std::move(source)}, std::move(aggregate_options)};

  return ExecutePlanAndCollectAsTable(std::move(aggregate));
}

群組聚合範例

/// \brief An example showing an aggregation node to perform a group-by operation
///
/// Source-Aggregation-Table
/// This example shows how an aggregation operation can be applied on a
/// execution plan resulting in grouped output. The source node loads the
/// data and the aggregation (counting unique types in column 'a') is
/// applied on this data. The output is collected into a table that will contain
/// one row for each unique combination of group keys.
arrow::Status SourceGroupAggregateSinkExample() {
  ARROW_ASSIGN_OR_RAISE(auto basic_data, MakeBasicBatches());

  arrow::AsyncGenerator<std::optional<cp::ExecBatch>> sink_gen;

  auto source_node_options = ac::SourceNodeOptions{basic_data.schema, basic_data.gen()};

  ac::Declaration source{"source", std::move(source_node_options)};
  auto options = std::make_shared<cp::CountOptions>(cp::CountOptions::ONLY_VALID);
  auto aggregate_options =
      ac::AggregateNodeOptions{/*aggregates=*/{{"hash_count", options, "a", "count(a)"}},
                               /*keys=*/{"b"}};
  ac::Declaration aggregate{
      "aggregate", {std::move(source)}, std::move(aggregate_options)};

  return ExecutePlanAndCollectAsTable(std::move(aggregate));
}

`sink`#

sink 操作提供輸出，並且是串流執行定義的最終節點。SinkNodeOptions 介面用於傳遞所需的選項。與 source 運算子類似，sink 運算子透過每次呼叫時傳回 record batch future 的函數來公開輸出。預期呼叫者會重複呼叫此函數，直到產生器函數耗盡（傳回 std::optional::nullopt）。如果沒有經常呼叫此函數，則 record batch 將累積在記憶體中。執行計畫應僅有一個「終端」節點（一個 sink 節點）。ExecPlan 可能會由於取消或錯誤而提前終止，然後再完全消耗輸出。但是，可以安全地獨立於 sink 銷毀計畫，sink 將透過 exec_plan->finished() 保留未消耗的批次。

作為 Source 範例的一部分，也包含 Sink 操作；

/// \brief An example demonstrating a source and sink node
///
/// Source-Table Example
/// This example shows how a custom source can be used
/// in an execution plan. This includes source node using pregenerated
/// data and collecting it into a table.
///
/// This sort of custom source is often not needed.  In most cases you can
/// use a scan (for a dataset source) or a source like table_source, array_vector_source,
/// exec_batch_source, or record_batch_source (for in-memory data)
arrow::Status SourceSinkExample() {
  ARROW_ASSIGN_OR_RAISE(auto basic_data, MakeBasicBatches());

  auto source_node_options = ac::SourceNodeOptions{basic_data.schema, basic_data.gen()};

  ac::Declaration source{"source", std::move(source_node_options)};

  return ExecutePlanAndCollectAsTable(std::move(source));
}

`consuming_sink`#

consuming_sink 運算子是一個 sink 操作，其中包含執行計畫內的消耗操作（即，執行計畫在消耗完成之前不應完成）。與 sink 節點不同，此節點採用預期會消耗批次的回呼函數。一旦此回呼完成，執行計畫將不再保留對批次的任何參考。可以在先前的調用完成之前呼叫消耗函數。如果消耗函數執行速度不夠快，則可能會堆積許多並行執行，從而封鎖 CPU 執行緒池。在所有消耗函數回呼完成之前，執行計畫將不會被標記為已完成。一旦交付了所有批次，執行計畫將等待 finish future 完成，然後再將執行計畫標記為已完成。這允許工作流程，其中消耗函數將批次轉換為非同步任務（目前在內部為資料集寫入節點完成）。

範例

// define a Custom SinkNodeConsumer
std::atomic<uint32_t> batches_seen{0};
arrow::Future<> finish = arrow::Future<>::Make();
struct CustomSinkNodeConsumer : public cp::SinkNodeConsumer {

    CustomSinkNodeConsumer(std::atomic<uint32_t> *batches_seen, arrow::Future<>finish):
    batches_seen(batches_seen), finish(std::move(finish)) {}
    // Consumption logic can be written here
    arrow::Status Consume(cp::ExecBatch batch) override {
    // data can be consumed in the expected way
    // transfer to another system or just do some work
    // and write to disk
    (*batches_seen)++;
    return arrow::Status::OK();
    }

    arrow::Future<> Finish() override { return finish; }

    std::atomic<uint32_t> *batches_seen;
    arrow::Future<> finish;

};

std::shared_ptr<CustomSinkNodeConsumer> consumer =
        std::make_shared<CustomSinkNodeConsumer>(&batches_seen, finish);

arrow::acero::ExecNode *consuming_sink;

ARROW_ASSIGN_OR_RAISE(consuming_sink, MakeExecNode("consuming_sink", plan.get(),
    {source}, cp::ConsumingSinkNodeOptions(consumer)));

消耗 Sink 範例

/// \brief An example showing a consuming sink node
///
/// Source-Consuming-Sink
/// This example shows how the data can be consumed within the execution plan
/// by using a ConsumingSink node. There is no data output from this execution plan.
arrow::Status SourceConsumingSinkExample() {
  ARROW_ASSIGN_OR_RAISE(auto basic_data, MakeBasicBatches());

  auto source_node_options = ac::SourceNodeOptions{basic_data.schema, basic_data.gen()};

  ac::Declaration source{"source", std::move(source_node_options)};

  std::atomic<uint32_t> batches_seen{0};
  arrow::Future<> finish = arrow::Future<>::Make();
  struct CustomSinkNodeConsumer : public ac::SinkNodeConsumer {
    CustomSinkNodeConsumer(std::atomic<uint32_t>* batches_seen, arrow::Future<> finish)
        : batches_seen(batches_seen), finish(std::move(finish)) {}

    arrow::Status Init(const std::shared_ptr<arrow::Schema>& schema,
                       ac::BackpressureControl* backpressure_control,
                       ac::ExecPlan* plan) override {
      // This will be called as the plan is started (before the first call to Consume)
      // and provides the schema of the data coming into the node, controls for pausing /
      // resuming input, and a pointer to the plan itself which can be used to access
      // other utilities such as the thread indexer or async task scheduler.
      return arrow::Status::OK();
    }

    arrow::Status Consume(cp::ExecBatch batch) override {
      (*batches_seen)++;
      return arrow::Status::OK();
    }

    arrow::Future<> Finish() override {
      // Here you can perform whatever (possibly async) cleanup is needed, e.g. closing
      // output file handles and flushing remaining work
      return arrow::Future<>::MakeFinished();
    }

    std::atomic<uint32_t>* batches_seen;
    arrow::Future<> finish;
  };
  std::shared_ptr<CustomSinkNodeConsumer> consumer =
      std::make_shared<CustomSinkNodeConsumer>(&batches_seen, finish);

  ac::Declaration consuming_sink{"consuming_sink",
                                 {std::move(source)},
                                 ac::ConsumingSinkNodeOptions(std::move(consumer))};

  // Since we are consuming the data within the plan there is no output and we simply
  // run the plan to completion instead of collecting into a table.
  ARROW_RETURN_NOT_OK(ac::DeclarationToStatus(std::move(consuming_sink)));

  std::cout << "The consuming sink node saw " << batches_seen.load() << " batches"
            << std::endl;
  return arrow::Status::OK();
}

`order_by_sink`#

order_by_sink 操作是 sink 操作的擴充。此操作透過提供 OrderBySinkNodeOptions 來提供保證串流順序的能力。在這裡，提供 arrow::compute::SortOptions 以定義哪些欄位用於排序，以及是否按升序或降序值排序。

注意

此節點是「管線中斷器」，將完全實現記憶體中的資料集。未來將新增溢出機制，這應該可以減輕此限制。

排序 Sink 範例

arrow::Status ExecutePlanAndCollectAsTableWithCustomSink(
    std::shared_ptr<ac::ExecPlan> plan, std::shared_ptr<arrow::Schema> schema,
    arrow::AsyncGenerator<std::optional<cp::ExecBatch>> sink_gen) {
  // translate sink_gen (async) to sink_reader (sync)
  std::shared_ptr<arrow::RecordBatchReader> sink_reader =
      ac::MakeGeneratorReader(schema, std::move(sink_gen), arrow::default_memory_pool());

  // validate the ExecPlan
  ARROW_RETURN_NOT_OK(plan->Validate());
  std::cout << "ExecPlan created : " << plan->ToString() << std::endl;
  // start the ExecPlan
  plan->StartProducing();

  // collect sink_reader into a Table
  std::shared_ptr<arrow::Table> response_table;

  ARROW_ASSIGN_OR_RAISE(response_table,
                        arrow::Table::FromRecordBatchReader(sink_reader.get()));

  std::cout << "Results : " << response_table->ToString() << std::endl;

  // stop producing
  plan->StopProducing();
  // plan mark finished
  auto future = plan->finished();
  return future.status();
}

/// \brief An example showing an order-by node
///
/// Source-OrderBy-Sink
/// In this example, the data enters through the source node
/// and the data is ordered in the sink node. The order can be
/// ASCENDING or DESCENDING and it is configurable. The output
/// is obtained as a table from the sink node.
arrow::Status SourceOrderBySinkExample() {
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<ac::ExecPlan> plan,
                        ac::ExecPlan::Make(*cp::threaded_exec_context()));

  ARROW_ASSIGN_OR_RAISE(auto basic_data, MakeSortTestBasicBatches());

  arrow::AsyncGenerator<std::optional<cp::ExecBatch>> sink_gen;

  auto source_node_options = ac::SourceNodeOptions{basic_data.schema, basic_data.gen()};
  ARROW_ASSIGN_OR_RAISE(ac::ExecNode * source,
                        ac::MakeExecNode("source", plan.get(), {}, source_node_options));

  ARROW_RETURN_NOT_OK(ac::MakeExecNode(
      "order_by_sink", plan.get(), {source},
      ac::OrderBySinkNodeOptions{
          cp::SortOptions{{cp::SortKey{"a", cp::SortOrder::Descending}}}, &sink_gen}));

  return ExecutePlanAndCollectAsTableWithCustomSink(plan, basic_data.schema, sink_gen);
}

`select_k_sink`#

select_k_sink 選項啟用選擇前/後 K 個元素，類似於 SQL ORDER BY ... LIMIT K 子句。SelectKOptions 是透過使用 OrderBySinkNode 定義來定義的。此選項傳回一個 sink 節點，該節點接收輸入，然後計算 top_k/bottom_k。

注意

此節點是「管線中斷器」，將完全實現記憶體中的輸入。未來將新增溢出機制，這應該可以減輕此限制。

SelectK 範例

/// \brief An example showing a select-k node
///
/// Source-KSelect
/// This example shows how K number of elements can be selected
/// either from the top or bottom. The output node is a modified
/// sink node where output can be obtained as a table.
arrow::Status SourceKSelectExample() {
  ARROW_ASSIGN_OR_RAISE(auto input, MakeGroupableBatches());
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<ac::ExecPlan> plan,
                        ac::ExecPlan::Make(*cp::threaded_exec_context()));
  arrow::AsyncGenerator<std::optional<cp::ExecBatch>> sink_gen;

  ARROW_ASSIGN_OR_RAISE(
      ac::ExecNode * source,
      ac::MakeExecNode("source", plan.get(), {},
                       ac::SourceNodeOptions{input.schema, input.gen()}));

  cp::SelectKOptions options = cp::SelectKOptions::TopKDefault(/*k=*/2, {"i32"});

  ARROW_RETURN_NOT_OK(ac::MakeExecNode("select_k_sink", plan.get(), {source},
                                       ac::SelectKSinkNodeOptions{options, &sink_gen}));

  auto schema = arrow::schema(
      {arrow::field("i32", arrow::int32()), arrow::field("str", arrow::utf8())});

  return ExecutePlanAndCollectAsTableWithCustomSink(plan, schema, sink_gen);
}

`table_sink`#

table_sink 節點提供將輸出作為記憶體中表格接收的能力。這比串流執行引擎提供的其他 sink 節點更易於使用，但僅在輸出可以輕鬆放入記憶體時才有意義。該節點是使用 TableSinkNodeOptions 建立的。

使用 table_sink 的範例

/// \brief An example showing a table sink node
///
/// TableSink Example
/// This example shows how a table_sink can be used
/// in an execution plan. This includes a source node
/// receiving data as batches and the table sink node
/// which emits the output as a table.
arrow::Status TableSinkExample() {
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<ac::ExecPlan> plan,
                        ac::ExecPlan::Make(*cp::threaded_exec_context()));

  ARROW_ASSIGN_OR_RAISE(auto basic_data, MakeBasicBatches());

  auto source_node_options = ac::SourceNodeOptions{basic_data.schema, basic_data.gen()};

  ARROW_ASSIGN_OR_RAISE(ac::ExecNode * source,
                        ac::MakeExecNode("source", plan.get(), {}, source_node_options));

  std::shared_ptr<arrow::Table> output_table;
  auto table_sink_options = ac::TableSinkNodeOptions{&output_table};

  ARROW_RETURN_NOT_OK(
      ac::MakeExecNode("table_sink", plan.get(), {source}, table_sink_options));
  // validate the ExecPlan
  ARROW_RETURN_NOT_OK(plan->Validate());
  std::cout << "ExecPlan created : " << plan->ToString() << std::endl;
  // start the ExecPlan
  plan->StartProducing();

  // Wait for the plan to finish
  auto finished = plan->finished();
  RETURN_NOT_OK(finished.status());
  std::cout << "Results : " << output_table->ToString() << std::endl;
  return arrow::Status::OK();
}

`scan`#

scan 是一種用於載入和處理資料集的操作。當您的輸入是資料集時，應優先於更通用的 source 節點。行為是使用 arrow::dataset::ScanNodeOptions 定義的。有關資料集和各種掃描選項的更多資訊，請參閱表格資料集。

此節點能夠將下推篩選器套用於檔案讀取器，從而減少需要讀取的資料量。這表示您可以將相同的篩選表達式提供給 scan 節點，您也可以將其提供給 FilterNode，因為篩選是在兩個不同的位置完成的。

掃描範例

/// \brief An example demonstrating a scan and sink node
///
/// Scan-Table
/// This example shows how scan operation can be applied on a dataset.
/// There are operations that can be applied on the scan (project, filter)
/// and the input data can be processed. The output is obtained as a table
arrow::Status ScanSinkExample() {
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::dataset::Dataset> dataset, GetDataset());

  auto options = std::make_shared<arrow::dataset::ScanOptions>();
  options->projection = cp::project({}, {});  // create empty projection

  // construct the scan node
  auto scan_node_options = arrow::dataset::ScanNodeOptions{dataset, options};

  ac::Declaration scan{"scan", std::move(scan_node_options)};

  return ExecutePlanAndCollectAsTable(std::move(scan));
}

`write`#

write 節點以 Parquet、Feather、CSV 等格式，將查詢結果儲存為檔案資料集，此功能使用 Arrow 中的表格資料集功能。arrow::dataset::WriteNodeOptions 提供寫入選項，而 arrow::dataset::FileSystemDatasetWriteOptions 則包含在其中。arrow::dataset::FileSystemDatasetWriteOptions 可控制寫入的資料集，包含輸出目錄、檔案命名方案等選項。

寫入範例

/// \brief An example showing a write node
/// \param file_path The destination to write to
///
/// Scan-Filter-Write
/// This example shows how scan node can be used to load the data
/// and after processing how it can be written to disk.
arrow::Status ScanFilterWriteExample(const std::string& file_path) {
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::dataset::Dataset> dataset, GetDataset());

  auto options = std::make_shared<arrow::dataset::ScanOptions>();
  // empty projection
  options->projection = cp::project({}, {});

  auto scan_node_options = arrow::dataset::ScanNodeOptions{dataset, options};

  ac::Declaration scan{"scan", std::move(scan_node_options)};

  arrow::AsyncGenerator<std::optional<cp::ExecBatch>> sink_gen;

  std::string root_path = "";
  std::string uri = "file://" + file_path;
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::fs::FileSystem> filesystem,
                        arrow::fs::FileSystemFromUri(uri, &root_path));

  auto base_path = root_path + "/parquet_dataset";
  // Uncomment the following line, if run repeatedly
  // ARROW_RETURN_NOT_OK(filesystem->DeleteDirContents(base_path));
  ARROW_RETURN_NOT_OK(filesystem->CreateDir(base_path));

  // The partition schema determines which fields are part of the partitioning.
  auto partition_schema = arrow::schema({arrow::field("a", arrow::int32())});
  // We'll use Hive-style partitioning,
  // which creates directories with "key=value" pairs.

  auto partitioning =
      std::make_shared<arrow::dataset::HivePartitioning>(partition_schema);
  // We'll write Parquet files.
  auto format = std::make_shared<arrow::dataset::ParquetFileFormat>();

  arrow::dataset::FileSystemDatasetWriteOptions write_options;
  write_options.file_write_options = format->DefaultWriteOptions();
  write_options.filesystem = filesystem;
  write_options.base_dir = base_path;
  write_options.partitioning = partitioning;
  write_options.basename_template = "part{i}.parquet";

  arrow::dataset::WriteNodeOptions write_node_options{write_options};

  ac::Declaration write{"write", {std::move(scan)}, std::move(write_node_options)};

  // Since the write node has no output we simply run the plan to completion and the
  // data should be written
  ARROW_RETURN_NOT_OK(ac::DeclarationToStatus(std::move(write)));

  std::cout << "Dataset written to " << base_path << std::endl;
  return arrow::Status::OK();
}

`union`#

union 將多個具有相同結構描述的資料流合併為一個，類似於 SQL 的 UNION ALL 子句。

以下範例示範如何使用兩個資料來源達成此目的。

Union 範例

/// \brief An example showing a union node
///
/// Source-Union-Table
/// This example shows how a union operation can be applied on two
/// data sources. The output is collected into a table.
arrow::Status SourceUnionSinkExample() {
  ARROW_ASSIGN_OR_RAISE(auto basic_data, MakeBasicBatches());

  ac::Declaration lhs{"source",
                      ac::SourceNodeOptions{basic_data.schema, basic_data.gen()}};
  lhs.label = "lhs";
  ac::Declaration rhs{"source",
                      ac::SourceNodeOptions{basic_data.schema, basic_data.gen()}};
  rhs.label = "rhs";
  ac::Declaration union_plan{
      "union", {std::move(lhs), std::move(rhs)}, ac::ExecNodeOptions{}};

  return ExecutePlanAndCollectAsTable(std::move(union_plan));
}

`hash_join`#

hash_join 運算提供關係代數運算，使用基於雜湊的演算法進行 join。HashJoinNodeOptions 包含定義 join 時所需的選項。hash_join 支援 left/right/full semi/anti/outerjoins。此外，也可以透過 join 選項設定 join 鍵 (即要 join 的欄位) 和後綴 (例如 "_x" 這樣的後綴詞，可以附加為左右關係中重複欄位名稱的後綴)。深入了解 hash-join。

Hash-Join 範例

/// \brief An example showing a hash join node
///
/// Source-HashJoin-Table
/// This example shows how source node gets the data and how a self-join
/// is applied on the data. The join options are configurable. The output
/// is collected into a table.
arrow::Status SourceHashJoinSinkExample() {
  ARROW_ASSIGN_OR_RAISE(auto input, MakeGroupableBatches());

  ac::Declaration left{"source", ac::SourceNodeOptions{input.schema, input.gen()}};
  ac::Declaration right{"source", ac::SourceNodeOptions{input.schema, input.gen()}};

  ac::HashJoinNodeOptions join_opts{
      ac::JoinType::INNER,
      /*left_keys=*/{"str"},
      /*right_keys=*/{"str"}, cp::literal(true), "l_", "r_"};

  ac::Declaration hashjoin{
      "hashjoin", {std::move(left), std::move(right)}, std::move(join_opts)};

  return ExecutePlanAndCollectAsTable(std::move(hashjoin));
}

摘要#

這些節點的範例可以在 Arrow 原始碼的 cpp/examples/arrow/execution_plan_documentation_examples.cc 中找到。

完整範例

#include <arrow/array.h>
#include <arrow/builder.h>

#include <arrow/acero/exec_plan.h>
#include <arrow/compute/api.h>
#include <arrow/compute/api_vector.h>
#include <arrow/compute/cast.h>

#include <arrow/csv/api.h>

#include <arrow/dataset/dataset.h>
#include <arrow/dataset/file_base.h>
#include <arrow/dataset/file_parquet.h>
#include <arrow/dataset/plan.h>
#include <arrow/dataset/scanner.h>

#include <arrow/io/interfaces.h>
#include <arrow/io/memory.h>

#include <arrow/result.h>
#include <arrow/status.h>
#include <arrow/table.h>

#include <arrow/ipc/api.h>

#include <arrow/util/future.h>
#include <arrow/util/range.h>
#include <arrow/util/thread_pool.h>
#include <arrow/util/vector.h>

#include <iostream>
#include <memory>
#include <utility>

// Demonstrate various operators in Arrow Streaming Execution Engine

namespace cp = ::arrow::compute;
namespace ac = ::arrow::acero;

constexpr char kSep[] = "******";

void PrintBlock(const std::string& msg) {
  std::cout << "\n\t" << kSep << " " << msg << " " << kSep << "\n" << std::endl;
}

template <typename TYPE,
          typename = typename std::enable_if<arrow::is_number_type<TYPE>::value |
                                             arrow::is_boolean_type<TYPE>::value |
                                             arrow::is_temporal_type<TYPE>::value>::type>
arrow::Result<std::shared_ptr<arrow::Array>> GetArrayDataSample(
    const std::vector<typename TYPE::c_type>& values) {
  using ArrowBuilderType = typename arrow::TypeTraits<TYPE>::BuilderType;
  ArrowBuilderType builder;
  ARROW_RETURN_NOT_OK(builder.Reserve(values.size()));
  ARROW_RETURN_NOT_OK(builder.AppendValues(values));
  return builder.Finish();
}

template <class TYPE>
arrow::Result<std::shared_ptr<arrow::Array>> GetBinaryArrayDataSample(
    const std::vector<std::string>& values) {
  using ArrowBuilderType = typename arrow::TypeTraits<TYPE>::BuilderType;
  ArrowBuilderType builder;
  ARROW_RETURN_NOT_OK(builder.Reserve(values.size()));
  ARROW_RETURN_NOT_OK(builder.AppendValues(values));
  return builder.Finish();
}

arrow::Result<std::shared_ptr<arrow::RecordBatch>> GetSampleRecordBatch(
    const arrow::ArrayVector array_vector, const arrow::FieldVector& field_vector) {
  std::shared_ptr<arrow::RecordBatch> record_batch;
  ARROW_ASSIGN_OR_RAISE(auto struct_result,
                        arrow::StructArray::Make(array_vector, field_vector));
  return record_batch->FromStructArray(struct_result);
}

/// \brief Create a sample table
/// The table's contents will be:
/// a,b
/// 1,null
/// 2,true
/// null,true
/// 3,false
/// null,true
/// 4,false
/// 5,null
/// 6,false
/// 7,false
/// 8,true
/// \return The created table

arrow::Result<std::shared_ptr<arrow::Table>> GetTable() {
  auto null_long = std::numeric_limits<int64_t>::quiet_NaN();
  ARROW_ASSIGN_OR_RAISE(auto int64_array,
                        GetArrayDataSample<arrow::Int64Type>(
                            {1, 2, null_long, 3, null_long, 4, 5, 6, 7, 8}));

  arrow::BooleanBuilder boolean_builder;
  std::shared_ptr<arrow::BooleanArray> bool_array;

  std::vector<uint8_t> bool_values = {false, true,  true,  false, true,
                                      false, false, false, false, true};
  std::vector<bool> is_valid = {false, true,  true, true, true,
                                true,  false, true, true, true};

  ARROW_RETURN_NOT_OK(boolean_builder.Reserve(10));

  ARROW_RETURN_NOT_OK(boolean_builder.AppendValues(bool_values, is_valid));

  ARROW_RETURN_NOT_OK(boolean_builder.Finish(&bool_array));

  auto record_batch =
      arrow::RecordBatch::Make(arrow::schema({arrow::field("a", arrow::int64()),
                                              arrow::field("b", arrow::boolean())}),
                               10, {int64_array, bool_array});
  ARROW_ASSIGN_OR_RAISE(auto table, arrow::Table::FromRecordBatches({record_batch}));
  return table;
}

/// \brief Create a sample dataset
/// \return An in-memory dataset based on GetTable()
arrow::Result<std::shared_ptr<arrow::dataset::Dataset>> GetDataset() {
  ARROW_ASSIGN_OR_RAISE(auto table, GetTable());
  auto ds = std::make_shared<arrow::dataset::InMemoryDataset>(table);
  return ds;
}

arrow::Result<cp::ExecBatch> GetExecBatchFromVectors(
    const arrow::FieldVector& field_vector, const arrow::ArrayVector& array_vector) {
  std::shared_ptr<arrow::RecordBatch> record_batch;
  ARROW_ASSIGN_OR_RAISE(auto res_batch, GetSampleRecordBatch(array_vector, field_vector));
  cp::ExecBatch batch{*res_batch};
  return batch;
}

// (Doc section: BatchesWithSchema Definition)
struct BatchesWithSchema {
  std::vector<cp::ExecBatch> batches;
  std::shared_ptr<arrow::Schema> schema;
  // This method uses internal arrow utilities to
  // convert a vector of record batches to an AsyncGenerator of optional batches
  arrow::AsyncGenerator<std::optional<cp::ExecBatch>> gen() const {
    auto opt_batches = ::arrow::internal::MapVector(
        [](cp::ExecBatch batch) { return std::make_optional(std::move(batch)); },
        batches);
    arrow::AsyncGenerator<std::optional<cp::ExecBatch>> gen;
    gen = arrow::MakeVectorGenerator(std::move(opt_batches));
    return gen;
  }
};
// (Doc section: BatchesWithSchema Definition)

// (Doc section: MakeBasicBatches Definition)
arrow::Result<BatchesWithSchema> MakeBasicBatches() {
  BatchesWithSchema out;
  auto field_vector = {arrow::field("a", arrow::int32()),
                       arrow::field("b", arrow::boolean())};
  ARROW_ASSIGN_OR_RAISE(auto b1_int, GetArrayDataSample<arrow::Int32Type>({0, 4}));
  ARROW_ASSIGN_OR_RAISE(auto b2_int, GetArrayDataSample<arrow::Int32Type>({5, 6, 7}));
  ARROW_ASSIGN_OR_RAISE(auto b3_int, GetArrayDataSample<arrow::Int32Type>({8, 9, 10}));

  ARROW_ASSIGN_OR_RAISE(auto b1_bool,
                        GetArrayDataSample<arrow::BooleanType>({false, true}));
  ARROW_ASSIGN_OR_RAISE(auto b2_bool,
                        GetArrayDataSample<arrow::BooleanType>({true, false, true}));
  ARROW_ASSIGN_OR_RAISE(auto b3_bool,
                        GetArrayDataSample<arrow::BooleanType>({false, true, false}));

  ARROW_ASSIGN_OR_RAISE(auto b1,
                        GetExecBatchFromVectors(field_vector, {b1_int, b1_bool}));
  ARROW_ASSIGN_OR_RAISE(auto b2,
                        GetExecBatchFromVectors(field_vector, {b2_int, b2_bool}));
  ARROW_ASSIGN_OR_RAISE(auto b3,
                        GetExecBatchFromVectors(field_vector, {b3_int, b3_bool}));

  out.batches = {b1, b2, b3};
  out.schema = arrow::schema(field_vector);
  return out;
}
// (Doc section: MakeBasicBatches Definition)

arrow::Result<BatchesWithSchema> MakeSortTestBasicBatches() {
  BatchesWithSchema out;
  auto field = arrow::field("a", arrow::int32());
  ARROW_ASSIGN_OR_RAISE(auto b1_int, GetArrayDataSample<arrow::Int32Type>({1, 3, 0, 2}));
  ARROW_ASSIGN_OR_RAISE(auto b2_int,
                        GetArrayDataSample<arrow::Int32Type>({121, 101, 120, 12}));
  ARROW_ASSIGN_OR_RAISE(auto b3_int,
                        GetArrayDataSample<arrow::Int32Type>({10, 110, 210, 121}));
  ARROW_ASSIGN_OR_RAISE(auto b4_int,
                        GetArrayDataSample<arrow::Int32Type>({51, 101, 2, 34}));
  ARROW_ASSIGN_OR_RAISE(auto b5_int,
                        GetArrayDataSample<arrow::Int32Type>({11, 31, 1, 12}));
  ARROW_ASSIGN_OR_RAISE(auto b6_int,
                        GetArrayDataSample<arrow::Int32Type>({12, 101, 120, 12}));
  ARROW_ASSIGN_OR_RAISE(auto b7_int,
                        GetArrayDataSample<arrow::Int32Type>({0, 110, 210, 11}));
  ARROW_ASSIGN_OR_RAISE(auto b8_int,
                        GetArrayDataSample<arrow::Int32Type>({51, 10, 2, 3}));

  ARROW_ASSIGN_OR_RAISE(auto b1, GetExecBatchFromVectors({field}, {b1_int}));
  ARROW_ASSIGN_OR_RAISE(auto b2, GetExecBatchFromVectors({field}, {b2_int}));
  ARROW_ASSIGN_OR_RAISE(auto b3,
                        GetExecBatchFromVectors({field, field}, {b3_int, b8_int}));
  ARROW_ASSIGN_OR_RAISE(auto b4,
                        GetExecBatchFromVectors({field, field, field, field},
                                                {b4_int, b5_int, b6_int, b7_int}));
  out.batches = {b1, b2, b3, b4};
  out.schema = arrow::schema({field});
  return out;
}

arrow::Result<BatchesWithSchema> MakeGroupableBatches(int multiplicity = 1) {
  BatchesWithSchema out;
  auto fields = {arrow::field("i32", arrow::int32()), arrow::field("str", arrow::utf8())};
  ARROW_ASSIGN_OR_RAISE(auto b1_int, GetArrayDataSample<arrow::Int32Type>({12, 7, 3}));
  ARROW_ASSIGN_OR_RAISE(auto b2_int, GetArrayDataSample<arrow::Int32Type>({-2, -1, 3}));
  ARROW_ASSIGN_OR_RAISE(auto b3_int, GetArrayDataSample<arrow::Int32Type>({5, 3, -8}));
  ARROW_ASSIGN_OR_RAISE(auto b1_str, GetBinaryArrayDataSample<arrow::StringType>(
                                         {"alpha", "beta", "alpha"}));
  ARROW_ASSIGN_OR_RAISE(auto b2_str, GetBinaryArrayDataSample<arrow::StringType>(
                                         {"alpha", "gamma", "alpha"}));
  ARROW_ASSIGN_OR_RAISE(auto b3_str, GetBinaryArrayDataSample<arrow::StringType>(
                                         {"gamma", "beta", "alpha"}));
  ARROW_ASSIGN_OR_RAISE(auto b1, GetExecBatchFromVectors(fields, {b1_int, b1_str}));
  ARROW_ASSIGN_OR_RAISE(auto b2, GetExecBatchFromVectors(fields, {b2_int, b2_str}));
  ARROW_ASSIGN_OR_RAISE(auto b3, GetExecBatchFromVectors(fields, {b3_int, b3_str}));
  out.batches = {b1, b2, b3};

  size_t batch_count = out.batches.size();
  for (int repeat = 1; repeat < multiplicity; ++repeat) {
    for (size_t i = 0; i < batch_count; ++i) {
      out.batches.push_back(out.batches[i]);
    }
  }

  out.schema = arrow::schema(fields);
  return out;
}

arrow::Status ExecutePlanAndCollectAsTable(ac::Declaration plan) {
  // collect sink_reader into a Table
  std::shared_ptr<arrow::Table> response_table;
  ARROW_ASSIGN_OR_RAISE(response_table, ac::DeclarationToTable(std::move(plan)));

  std::cout << "Results : " << response_table->ToString() << std::endl;

  return arrow::Status::OK();
}

// (Doc section: Scan Example)

/// \brief An example demonstrating a scan and sink node
///
/// Scan-Table
/// This example shows how scan operation can be applied on a dataset.
/// There are operations that can be applied on the scan (project, filter)
/// and the input data can be processed. The output is obtained as a table
arrow::Status ScanSinkExample() {
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::dataset::Dataset> dataset, GetDataset());

  auto options = std::make_shared<arrow::dataset::ScanOptions>();
  options->projection = cp::project({}, {});  // create empty projection

  // construct the scan node
  auto scan_node_options = arrow::dataset::ScanNodeOptions{dataset, options};

  ac::Declaration scan{"scan", std::move(scan_node_options)};

  return ExecutePlanAndCollectAsTable(std::move(scan));
}
// (Doc section: Scan Example)

// (Doc section: Source Example)

/// \brief An example demonstrating a source and sink node
///
/// Source-Table Example
/// This example shows how a custom source can be used
/// in an execution plan. This includes source node using pregenerated
/// data and collecting it into a table.
///
/// This sort of custom source is often not needed.  In most cases you can
/// use a scan (for a dataset source) or a source like table_source, array_vector_source,
/// exec_batch_source, or record_batch_source (for in-memory data)
arrow::Status SourceSinkExample() {
  ARROW_ASSIGN_OR_RAISE(auto basic_data, MakeBasicBatches());

  auto source_node_options = ac::SourceNodeOptions{basic_data.schema, basic_data.gen()};

  ac::Declaration source{"source", std::move(source_node_options)};

  return ExecutePlanAndCollectAsTable(std::move(source));
}
// (Doc section: Source Example)

// (Doc section: Table Source Example)

/// \brief An example showing a table source node
///
/// TableSource-Table Example
/// This example shows how a table_source can be used
/// in an execution plan. This includes a table source node
/// receiving data from a table.  This plan simply collects the
/// data back into a table but nodes could be added that modify
/// or transform the data as well (as is shown in later examples)
arrow::Status TableSourceSinkExample() {
  ARROW_ASSIGN_OR_RAISE(auto table, GetTable());

  arrow::AsyncGenerator<std::optional<cp::ExecBatch>> sink_gen;
  int max_batch_size = 2;
  auto table_source_options = ac::TableSourceNodeOptions{table, max_batch_size};

  ac::Declaration source{"table_source", std::move(table_source_options)};

  return ExecutePlanAndCollectAsTable(std::move(source));
}
// (Doc section: Table Source Example)

// (Doc section: Filter Example)

/// \brief An example showing a filter node
///
/// Source-Filter-Table
/// This example shows how a filter can be used in an execution plan,
/// to filter data from a source. The output from the execution plan
/// is collected into a table.
arrow::Status ScanFilterSinkExample() {
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::dataset::Dataset> dataset, GetDataset());

  auto options = std::make_shared<arrow::dataset::ScanOptions>();
  // specify the filter.  This filter removes all rows where the
  // value of the "a" column is greater than 3.
  cp::Expression filter_expr = cp::greater(cp::field_ref("a"), cp::literal(3));
  // set filter for scanner : on-disk / push-down filtering.
  // This step can be skipped if you are not reading from disk.
  options->filter = filter_expr;
  // empty projection
  options->projection = cp::project({}, {});

  // construct the scan node
  std::cout << "Initialized Scanning Options" << std::endl;

  auto scan_node_options = arrow::dataset::ScanNodeOptions{dataset, options};
  std::cout << "Scan node options created" << std::endl;

  ac::Declaration scan{"scan", std::move(scan_node_options)};

  // pipe the scan node into the filter node
  // Need to set the filter in scan node options and filter node options.
  // At scan node it is used for on-disk / push-down filtering.
  // At filter node it is used for in-memory filtering.
  ac::Declaration filter{
      "filter", {std::move(scan)}, ac::FilterNodeOptions(std::move(filter_expr))};

  return ExecutePlanAndCollectAsTable(std::move(filter));
}

// (Doc section: Filter Example)

// (Doc section: Project Example)

/// \brief An example showing a project node
///
/// Scan-Project-Table
/// This example shows how a Scan operation can be used to load the data
/// into the execution plan, how a project operation can be applied on the
/// data stream and how the output is collected into a table
arrow::Status ScanProjectSinkExample() {
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::dataset::Dataset> dataset, GetDataset());

  auto options = std::make_shared<arrow::dataset::ScanOptions>();
  // projection
  cp::Expression a_times_2 = cp::call("multiply", {cp::field_ref("a"), cp::literal(2)});
  options->projection = cp::project({}, {});

  auto scan_node_options = arrow::dataset::ScanNodeOptions{dataset, options};

  ac::Declaration scan{"scan", std::move(scan_node_options)};
  ac::Declaration project{
      "project", {std::move(scan)}, ac::ProjectNodeOptions({a_times_2})};

  return ExecutePlanAndCollectAsTable(std::move(project));
}

// (Doc section: Project Example)

// This is a variation of ScanProjectSinkExample introducing how to use the
// Declaration::Sequence function
arrow::Status ScanProjectSequenceSinkExample() {
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::dataset::Dataset> dataset, GetDataset());

  auto options = std::make_shared<arrow::dataset::ScanOptions>();
  // projection
  cp::Expression a_times_2 = cp::call("multiply", {cp::field_ref("a"), cp::literal(2)});
  options->projection = cp::project({}, {});

  auto scan_node_options = arrow::dataset::ScanNodeOptions{dataset, options};

  // (Doc section: Project Sequence Example)
  // Inputs do not have to be passed to the project node when using Sequence
  ac::Declaration plan =
      ac::Declaration::Sequence({{"scan", std::move(scan_node_options)},
                                 {"project", ac::ProjectNodeOptions({a_times_2})}});
  // (Doc section: Project Sequence Example)

  return ExecutePlanAndCollectAsTable(std::move(plan));
}

// (Doc section: Scalar Aggregate Example)

/// \brief An example showing an aggregation node to aggregate an entire table
///
/// Source-Aggregation-Table
/// This example shows how an aggregation operation can be applied on a
/// execution plan resulting in a scalar output. The source node loads the
/// data and the aggregation (counting unique types in column 'a')
/// is applied on this data. The output is collected into a table (that will
/// have exactly one row)
arrow::Status SourceScalarAggregateSinkExample() {
  ARROW_ASSIGN_OR_RAISE(auto basic_data, MakeBasicBatches());

  auto source_node_options = ac::SourceNodeOptions{basic_data.schema, basic_data.gen()};

  ac::Declaration source{"source", std::move(source_node_options)};
  auto aggregate_options =
      ac::AggregateNodeOptions{/*aggregates=*/{{"sum", nullptr, "a", "sum(a)"}}};
  ac::Declaration aggregate{
      "aggregate", {std::move(source)}, std::move(aggregate_options)};

  return ExecutePlanAndCollectAsTable(std::move(aggregate));
}
// (Doc section: Scalar Aggregate Example)

// (Doc section: Group Aggregate Example)

/// \brief An example showing an aggregation node to perform a group-by operation
///
/// Source-Aggregation-Table
/// This example shows how an aggregation operation can be applied on a
/// execution plan resulting in grouped output. The source node loads the
/// data and the aggregation (counting unique types in column 'a') is
/// applied on this data. The output is collected into a table that will contain
/// one row for each unique combination of group keys.
arrow::Status SourceGroupAggregateSinkExample() {
  ARROW_ASSIGN_OR_RAISE(auto basic_data, MakeBasicBatches());

  arrow::AsyncGenerator<std::optional<cp::ExecBatch>> sink_gen;

  auto source_node_options = ac::SourceNodeOptions{basic_data.schema, basic_data.gen()};

  ac::Declaration source{"source", std::move(source_node_options)};
  auto options = std::make_shared<cp::CountOptions>(cp::CountOptions::ONLY_VALID);
  auto aggregate_options =
      ac::AggregateNodeOptions{/*aggregates=*/{{"hash_count", options, "a", "count(a)"}},
                               /*keys=*/{"b"}};
  ac::Declaration aggregate{
      "aggregate", {std::move(source)}, std::move(aggregate_options)};

  return ExecutePlanAndCollectAsTable(std::move(aggregate));
}
// (Doc section: Group Aggregate Example)

// (Doc section: ConsumingSink Example)

/// \brief An example showing a consuming sink node
///
/// Source-Consuming-Sink
/// This example shows how the data can be consumed within the execution plan
/// by using a ConsumingSink node. There is no data output from this execution plan.
arrow::Status SourceConsumingSinkExample() {
  ARROW_ASSIGN_OR_RAISE(auto basic_data, MakeBasicBatches());

  auto source_node_options = ac::SourceNodeOptions{basic_data.schema, basic_data.gen()};

  ac::Declaration source{"source", std::move(source_node_options)};

  std::atomic<uint32_t> batches_seen{0};
  arrow::Future<> finish = arrow::Future<>::Make();
  struct CustomSinkNodeConsumer : public ac::SinkNodeConsumer {
    CustomSinkNodeConsumer(std::atomic<uint32_t>* batches_seen, arrow::Future<> finish)
        : batches_seen(batches_seen), finish(std::move(finish)) {}

    arrow::Status Init(const std::shared_ptr<arrow::Schema>& schema,
                       ac::BackpressureControl* backpressure_control,
                       ac::ExecPlan* plan) override {
      // This will be called as the plan is started (before the first call to Consume)
      // and provides the schema of the data coming into the node, controls for pausing /
      // resuming input, and a pointer to the plan itself which can be used to access
      // other utilities such as the thread indexer or async task scheduler.
      return arrow::Status::OK();
    }

    arrow::Status Consume(cp::ExecBatch batch) override {
      (*batches_seen)++;
      return arrow::Status::OK();
    }

    arrow::Future<> Finish() override {
      // Here you can perform whatever (possibly async) cleanup is needed, e.g. closing
      // output file handles and flushing remaining work
      return arrow::Future<>::MakeFinished();
    }

    std::atomic<uint32_t>* batches_seen;
    arrow::Future<> finish;
  };
  std::shared_ptr<CustomSinkNodeConsumer> consumer =
      std::make_shared<CustomSinkNodeConsumer>(&batches_seen, finish);

  ac::Declaration consuming_sink{"consuming_sink",
                                 {std::move(source)},
                                 ac::ConsumingSinkNodeOptions(std::move(consumer))};

  // Since we are consuming the data within the plan there is no output and we simply
  // run the plan to completion instead of collecting into a table.
  ARROW_RETURN_NOT_OK(ac::DeclarationToStatus(std::move(consuming_sink)));

  std::cout << "The consuming sink node saw " << batches_seen.load() << " batches"
            << std::endl;
  return arrow::Status::OK();
}
// (Doc section: ConsumingSink Example)

// (Doc section: OrderBySink Example)

arrow::Status ExecutePlanAndCollectAsTableWithCustomSink(
    std::shared_ptr<ac::ExecPlan> plan, std::shared_ptr<arrow::Schema> schema,
    arrow::AsyncGenerator<std::optional<cp::ExecBatch>> sink_gen) {
  // translate sink_gen (async) to sink_reader (sync)
  std::shared_ptr<arrow::RecordBatchReader> sink_reader =
      ac::MakeGeneratorReader(schema, std::move(sink_gen), arrow::default_memory_pool());

  // validate the ExecPlan
  ARROW_RETURN_NOT_OK(plan->Validate());
  std::cout << "ExecPlan created : " << plan->ToString() << std::endl;
  // start the ExecPlan
  plan->StartProducing();

  // collect sink_reader into a Table
  std::shared_ptr<arrow::Table> response_table;

  ARROW_ASSIGN_OR_RAISE(response_table,
                        arrow::Table::FromRecordBatchReader(sink_reader.get()));

  std::cout << "Results : " << response_table->ToString() << std::endl;

  // stop producing
  plan->StopProducing();
  // plan mark finished
  auto future = plan->finished();
  return future.status();
}

/// \brief An example showing an order-by node
///
/// Source-OrderBy-Sink
/// In this example, the data enters through the source node
/// and the data is ordered in the sink node. The order can be
/// ASCENDING or DESCENDING and it is configurable. The output
/// is obtained as a table from the sink node.
arrow::Status SourceOrderBySinkExample() {
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<ac::ExecPlan> plan,
                        ac::ExecPlan::Make(*cp::threaded_exec_context()));

  ARROW_ASSIGN_OR_RAISE(auto basic_data, MakeSortTestBasicBatches());

  arrow::AsyncGenerator<std::optional<cp::ExecBatch>> sink_gen;

  auto source_node_options = ac::SourceNodeOptions{basic_data.schema, basic_data.gen()};
  ARROW_ASSIGN_OR_RAISE(ac::ExecNode * source,
                        ac::MakeExecNode("source", plan.get(), {}, source_node_options));

  ARROW_RETURN_NOT_OK(ac::MakeExecNode(
      "order_by_sink", plan.get(), {source},
      ac::OrderBySinkNodeOptions{
          cp::SortOptions{{cp::SortKey{"a", cp::SortOrder::Descending}}}, &sink_gen}));

  return ExecutePlanAndCollectAsTableWithCustomSink(plan, basic_data.schema, sink_gen);
}

// (Doc section: OrderBySink Example)

// (Doc section: HashJoin Example)

/// \brief An example showing a hash join node
///
/// Source-HashJoin-Table
/// This example shows how source node gets the data and how a self-join
/// is applied on the data. The join options are configurable. The output
/// is collected into a table.
arrow::Status SourceHashJoinSinkExample() {
  ARROW_ASSIGN_OR_RAISE(auto input, MakeGroupableBatches());

  ac::Declaration left{"source", ac::SourceNodeOptions{input.schema, input.gen()}};
  ac::Declaration right{"source", ac::SourceNodeOptions{input.schema, input.gen()}};

  ac::HashJoinNodeOptions join_opts{
      ac::JoinType::INNER,
      /*left_keys=*/{"str"},
      /*right_keys=*/{"str"}, cp::literal(true), "l_", "r_"};

  ac::Declaration hashjoin{
      "hashjoin", {std::move(left), std::move(right)}, std::move(join_opts)};

  return ExecutePlanAndCollectAsTable(std::move(hashjoin));
}

// (Doc section: HashJoin Example)

// (Doc section: KSelect Example)

/// \brief An example showing a select-k node
///
/// Source-KSelect
/// This example shows how K number of elements can be selected
/// either from the top or bottom. The output node is a modified
/// sink node where output can be obtained as a table.
arrow::Status SourceKSelectExample() {
  ARROW_ASSIGN_OR_RAISE(auto input, MakeGroupableBatches());
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<ac::ExecPlan> plan,
                        ac::ExecPlan::Make(*cp::threaded_exec_context()));
  arrow::AsyncGenerator<std::optional<cp::ExecBatch>> sink_gen;

  ARROW_ASSIGN_OR_RAISE(
      ac::ExecNode * source,
      ac::MakeExecNode("source", plan.get(), {},
                       ac::SourceNodeOptions{input.schema, input.gen()}));

  cp::SelectKOptions options = cp::SelectKOptions::TopKDefault(/*k=*/2, {"i32"});

  ARROW_RETURN_NOT_OK(ac::MakeExecNode("select_k_sink", plan.get(), {source},
                                       ac::SelectKSinkNodeOptions{options, &sink_gen}));

  auto schema = arrow::schema(
      {arrow::field("i32", arrow::int32()), arrow::field("str", arrow::utf8())});

  return ExecutePlanAndCollectAsTableWithCustomSink(plan, schema, sink_gen);
}

// (Doc section: KSelect Example)

// (Doc section: Write Example)

/// \brief An example showing a write node
/// \param file_path The destination to write to
///
/// Scan-Filter-Write
/// This example shows how scan node can be used to load the data
/// and after processing how it can be written to disk.
arrow::Status ScanFilterWriteExample(const std::string& file_path) {
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::dataset::Dataset> dataset, GetDataset());

  auto options = std::make_shared<arrow::dataset::ScanOptions>();
  // empty projection
  options->projection = cp::project({}, {});

  auto scan_node_options = arrow::dataset::ScanNodeOptions{dataset, options};

  ac::Declaration scan{"scan", std::move(scan_node_options)};

  arrow::AsyncGenerator<std::optional<cp::ExecBatch>> sink_gen;

  std::string root_path = "";
  std::string uri = "file://" + file_path;
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::fs::FileSystem> filesystem,
                        arrow::fs::FileSystemFromUri(uri, &root_path));

  auto base_path = root_path + "/parquet_dataset";
  // Uncomment the following line, if run repeatedly
  // ARROW_RETURN_NOT_OK(filesystem->DeleteDirContents(base_path));
  ARROW_RETURN_NOT_OK(filesystem->CreateDir(base_path));

  // The partition schema determines which fields are part of the partitioning.
  auto partition_schema = arrow::schema({arrow::field("a", arrow::int32())});
  // We'll use Hive-style partitioning,
  // which creates directories with "key=value" pairs.

  auto partitioning =
      std::make_shared<arrow::dataset::HivePartitioning>(partition_schema);
  // We'll write Parquet files.
  auto format = std::make_shared<arrow::dataset::ParquetFileFormat>();

  arrow::dataset::FileSystemDatasetWriteOptions write_options;
  write_options.file_write_options = format->DefaultWriteOptions();
  write_options.filesystem = filesystem;
  write_options.base_dir = base_path;
  write_options.partitioning = partitioning;
  write_options.basename_template = "part{i}.parquet";

  arrow::dataset::WriteNodeOptions write_node_options{write_options};

  ac::Declaration write{"write", {std::move(scan)}, std::move(write_node_options)};

  // Since the write node has no output we simply run the plan to completion and the
  // data should be written
  ARROW_RETURN_NOT_OK(ac::DeclarationToStatus(std::move(write)));

  std::cout << "Dataset written to " << base_path << std::endl;
  return arrow::Status::OK();
}

// (Doc section: Write Example)

// (Doc section: Union Example)

/// \brief An example showing a union node
///
/// Source-Union-Table
/// This example shows how a union operation can be applied on two
/// data sources. The output is collected into a table.
arrow::Status SourceUnionSinkExample() {
  ARROW_ASSIGN_OR_RAISE(auto basic_data, MakeBasicBatches());

  ac::Declaration lhs{"source",
                      ac::SourceNodeOptions{basic_data.schema, basic_data.gen()}};
  lhs.label = "lhs";
  ac::Declaration rhs{"source",
                      ac::SourceNodeOptions{basic_data.schema, basic_data.gen()}};
  rhs.label = "rhs";
  ac::Declaration union_plan{
      "union", {std::move(lhs), std::move(rhs)}, ac::ExecNodeOptions{}};

  return ExecutePlanAndCollectAsTable(std::move(union_plan));
}

// (Doc section: Union Example)

// (Doc section: Table Sink Example)

/// \brief An example showing a table sink node
///
/// TableSink Example
/// This example shows how a table_sink can be used
/// in an execution plan. This includes a source node
/// receiving data as batches and the table sink node
/// which emits the output as a table.
arrow::Status TableSinkExample() {
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<ac::ExecPlan> plan,
                        ac::ExecPlan::Make(*cp::threaded_exec_context()));

  ARROW_ASSIGN_OR_RAISE(auto basic_data, MakeBasicBatches());

  auto source_node_options = ac::SourceNodeOptions{basic_data.schema, basic_data.gen()};

  ARROW_ASSIGN_OR_RAISE(ac::ExecNode * source,
                        ac::MakeExecNode("source", plan.get(), {}, source_node_options));

  std::shared_ptr<arrow::Table> output_table;
  auto table_sink_options = ac::TableSinkNodeOptions{&output_table};

  ARROW_RETURN_NOT_OK(
      ac::MakeExecNode("table_sink", plan.get(), {source}, table_sink_options));
  // validate the ExecPlan
  ARROW_RETURN_NOT_OK(plan->Validate());
  std::cout << "ExecPlan created : " << plan->ToString() << std::endl;
  // start the ExecPlan
  plan->StartProducing();

  // Wait for the plan to finish
  auto finished = plan->finished();
  RETURN_NOT_OK(finished.status());
  std::cout << "Results : " << output_table->ToString() << std::endl;
  return arrow::Status::OK();
}

// (Doc section: Table Sink Example)

// (Doc section: RecordBatchReaderSource Example)

/// \brief An example showing the usage of a RecordBatchReader as the data source.
///
/// RecordBatchReaderSourceSink Example
/// This example shows how a record_batch_reader_source can be used
/// in an execution plan. This includes the source node
/// receiving data from a TableRecordBatchReader.

arrow::Status RecordBatchReaderSourceSinkExample() {
  ARROW_ASSIGN_OR_RAISE(auto table, GetTable());
  std::shared_ptr<arrow::RecordBatchReader> reader =
      std::make_shared<arrow::TableBatchReader>(table);
  ac::Declaration reader_source{"record_batch_reader_source",
                                ac::RecordBatchReaderSourceNodeOptions{reader}};
  return ExecutePlanAndCollectAsTable(std::move(reader_source));
}

// (Doc section: RecordBatchReaderSource Example)

enum ExampleMode {
  SOURCE_SINK = 0,
  TABLE_SOURCE_SINK = 1,
  SCAN = 2,
  FILTER = 3,
  PROJECT = 4,
  SCALAR_AGGREGATION = 5,
  GROUP_AGGREGATION = 6,
  CONSUMING_SINK = 7,
  ORDER_BY_SINK = 8,
  HASHJOIN = 9,
  KSELECT = 10,
  WRITE = 11,
  UNION = 12,
  TABLE_SOURCE_TABLE_SINK = 13,
  RECORD_BATCH_READER_SOURCE = 14,
  PROJECT_SEQUENCE = 15
};

int main(int argc, char** argv) {
  if (argc < 3) {
    // Fake success for CI purposes.
    return EXIT_SUCCESS;
  }

  std::string base_save_path = argv[1];
  int mode = std::atoi(argv[2]);
  arrow::Status status;
  // ensure arrow::dataset node factories are in the registry
  arrow::dataset::internal::Initialize();
  switch (mode) {
    case SOURCE_SINK:
      PrintBlock("Source Sink Example");
      status = SourceSinkExample();
      break;
    case TABLE_SOURCE_SINK:
      PrintBlock("Table Source Sink Example");
      status = TableSourceSinkExample();
      break;
    case SCAN:
      PrintBlock("Scan Example");
      status = ScanSinkExample();
      break;
    case FILTER:
      PrintBlock("Filter Example");
      status = ScanFilterSinkExample();
      break;
    case PROJECT:
      PrintBlock("Project Example");
      status = ScanProjectSinkExample();
      break;
    case PROJECT_SEQUENCE:
      PrintBlock("Project Example (using Declaration::Sequence)");
      status = ScanProjectSequenceSinkExample();
      break;
    case GROUP_AGGREGATION:
      PrintBlock("Aggregate Example");
      status = SourceGroupAggregateSinkExample();
      break;
    case SCALAR_AGGREGATION:
      PrintBlock("Aggregate Example");
      status = SourceScalarAggregateSinkExample();
      break;
    case CONSUMING_SINK:
      PrintBlock("Consuming-Sink Example");
      status = SourceConsumingSinkExample();
      break;
    case ORDER_BY_SINK:
      PrintBlock("OrderBy Example");
      status = SourceOrderBySinkExample();
      break;
    case HASHJOIN:
      PrintBlock("HashJoin Example");
      status = SourceHashJoinSinkExample();
      break;
    case KSELECT:
      PrintBlock("KSelect Example");
      status = SourceKSelectExample();
      break;
    case WRITE:
      PrintBlock("Write Example");
      status = ScanFilterWriteExample(base_save_path);
      break;
    case UNION:
      PrintBlock("Union Example");
      status = SourceUnionSinkExample();
      break;
    case TABLE_SOURCE_TABLE_SINK:
      PrintBlock("TableSink Example");
      status = TableSinkExample();
      break;
    case RECORD_BATCH_READER_SOURCE:
      PrintBlock("RecordBatchReaderSource Example");
      status = RecordBatchReaderSourceSinkExample();
      break;
    default:
      break;
  }

  if (status.ok()) {
    return EXIT_SUCCESS;
  } else {
    std::cout << "Error occurred: " << status.message() << std::endl;
    return EXIT_FAILURE;
  }
}

Acero 使用者指南#

使用 Acero#

建立計畫#

使用 Substrait#

程式化計畫建立#

執行計畫#

DeclarationToTable#

DeclarationToReader#

DeclarationToStatus#

直接執行計畫#

提供輸入#

可用的 ExecNode 實作#

來源#

計算節點#

排列節點#

Sink 節點#

範例#

source#

table_source#

filter#

project#

aggregate#

sink#

consuming_sink#

order_by_sink#

select_k_sink#

table_sink#

scan#

write#

union#

hash_join#

摘要#