Function ExecPlan_Write

r/src/compute-exec.cpp:312–354

[[dataset::export]]

// [[dataset::export]]
void ExecPlan_Write(
    const std::shared_ptr<acero::ExecPlan>& plan,
    const std::shared_ptr<acero::ExecNode>& final_node,
    const std::shared_ptr<arrow::Schema>& schema,
    const std::shared_ptr<ds::FileWriteOptions>& file_write_options,
    const std::shared_ptr<fs::FileSystem>& filesystem, std::string base_dir,
    const std::shared_ptr<ds::Partitioning>& partitioning, std::string basename_template,
    arrow::dataset::ExistingDataBehavior existing_data_behavior, int max_partitions,
    uint32_t max_open_files, uint64_t max_rows_per_file, uint64_t min_rows_per_group,
    uint64_t max_rows_per_group, bool create_directory, bool preserve_order) {
  arrow::dataset::internal::Initialize();

  // TODO(ARROW-16200): expose FileSystemDatasetWriteOptions in R
  // and encapsulate this logic better
  ds::FileSystemDatasetWriteOptions opts;
  opts.file_write_options = file_write_options;
  opts.existing_data_behavior = existing_data_behavior;
  opts.filesystem = filesystem;
  opts.base_dir = base_dir;
  opts.partitioning = partitioning;
  opts.basename_template = basename_template;
  opts.max_partitions = max_partitions;
  opts.max_open_files = max_open_files;
  opts.max_rows_per_file = max_rows_per_file;
  opts.min_rows_per_group = min_rows_per_group;
  opts.max_rows_per_group = max_rows_per_group;
  opts.create_dir = create_directory;
  opts.preserve_order = preserve_order;

  ds::WriteNodeOptions options(std::move(opts));
  options.custom_schema = std::move(schema);

  MakeExecNodeOrStop("write", final_node->plan(), {final_node.get()}, std::move(options));

  StopIfNotOk(plan->Validate());

  arrow::Status result = RunWithCapturedRIfPossibleVoid([&]() {
    plan->StartProducing();
    return plan->finished().status();
  });

  StopIfNotOk(result);
}

#endif