Repository: apache/arrow
File: cpp/src/arrow/dataset/dataset_writer.cc (C++)

Code before:
    auto file_finish_task = [self = shared_from_this()] {
      self->writer_state_->open_files_throttle.Release(1);
      return Status::OK();
    };
    auto file_tasks = util::MakeThrottledAsyncTaskGroup(scheduler_, 1, /*queue=*/nullptr,
                                                        std::move(file_finish_task));
    if (init_future_.is_valid()) {
      file_tasks->AddSimpleTask([init_future = init_future_]() { return init_future; },
                                "DatasetWriter::WaitForDirectoryInit"sv);
    }
    latest_open_file_->Start(std::move(file_tasks), filename);
    return Status::OK();
  }

  uint64_t rows_written() const { return rows_written_; }

  void PrepareDirectory() {
    if (directory_.empty() || !write_options_.create_dir) {
      return;
    }
    init_future_ = Future<>::Make();
    auto create_dir_cb = [this] {
      return DeferNotOk(
          write_options_.filesystem->io_context().executor()->Submit([this]() {
            return write_options_.filesystem->CreateDir(directory_,
                                                        write_options_.create_dir);
          }));
    };
    // We need to notify waiters whether the directory succeeded or failed.
    auto notify_waiters_cb = [this] { init_future_.MarkFinished(); };
    auto notify_waiters_on_err_cb = [this](const Status& err) {
      // If there is an error the scheduler will abort but that takes some time
      // and we don't want to start writing in the meantime so we send an error to the
      // file writing queue and return the error.
      init_future_.MarkFinished(err);
      return err;
    };
    std::function<Future<>()> init_task;
    if (write_options_.existing_data_behavior ==
        ExistingDataBehavior::kDeleteMatchingPartitions) {
      init_task = [this, create_dir_cb, notify_waiters_cb, notify_waiters_on_err_cb] {
        return write_options_.filesystem
            ->DeleteDirContentsAsync(directory_,
                                     /*missing_dir_ok=*/true)
            .Then(create_dir_cb)
            .Then(notify_waiters_cb, notify_waiters_on_err_cb);
      };
    } else {
      init_task = [create_dir_cb, notify_waiters_cb, notify_waiters_on_err_cb] {
        return create_dir_cb().Then(notify_waiters_cb, notify_waiters_on_err_cb);
      };

Review comment (suggestion):
This looks like it's passing in the value of `create_dir` as the `recursive` argument.  I'm wondering if it might make more sense to expose `check_directory_existence_before_creation` in R instead of updating the dataset writer?

https://github.com/apache/arrow/blob/b8eed8c9666c1280f92053291c61c59dd7ed19b7/cpp/src/arrow/filesystem/s3fs.h#L173-L180

Code after:
    auto file_finish_task = [self = shared_from_this()] {
      self->writer_state_->open_files_throttle.Release(1);
      return Status::OK();
    };
    auto file_tasks = util::MakeThrottledAsyncTaskGroup(scheduler_, 1, /*queue=*/nullptr,
                                                        std::move(file_finish_task));
    if (init_future_.is_valid()) {
      file_tasks->AddSimpleTask([init_future = init_future_]() { return init_future; },
                                "DatasetWriter::WaitForDirectoryInit"sv);
    }
    latest_open_file_->Start(std::move(file_tasks), filename);
    return Status::OK();
  }

  uint64_t rows_written() const { return rows_written_; }

  void PrepareDirectory() {
    if (directory_.empty() || !write_options_.create_dir) {
      return;
    }
    init_future_ = Future<>::Make();
    auto create_dir_cb = [this] {
      return DeferNotOk(write_options_.filesystem->io_context().executor()->Submit(
          [this]() { return write_options_.filesystem->CreateDir(directory_); }));
    };
    // We need to notify waiters whether the directory succeeded or failed.
    auto notify_waiters_cb = [this] { init_future_.MarkFinished(); };
    auto notify_waiters_on_err_cb = [this](const Status& err) {
      // If there is an error the scheduler will abort but that takes some time
      // and we don't want to start writing in the meantime so we send an error to the
      // file writing queue and return the error.
      init_future_.MarkFinished(err);
      return err;
    };
    std::function<Future<>()> init_task;
    if (write_options_.existing_data_behavior ==
        ExistingDataBehavior::kDeleteMatchingPartitions) {
      init_task = [this, create_dir_cb, notify_waiters_cb, notify_waiters_on_err_cb] {
        return write_options_.filesystem
            ->DeleteDirContentsAsync(directory_,
                                     /*missing_dir_ok=*/true)
            .Then(create_dir_cb)
            .Then(notify_waiters_cb, notify_waiters_on_err_cb);
      };
    } else {
      init_task = [create_dir_cb, notify_waiters_cb, notify_waiters_on_err_cb] {
        return create_dir_cb().Then(notify_waiters_cb, notify_waiters_on_err_cb);
      };
    }
    scheduler_->AddSimpleTask(std::move(init_task),
                              "DatasetWriter::InitializeDirectory"sv);