ARROW-13650: [C++] Create dataset writer to encapsulate dataset writer logic

c_glib/arrow-dataset-glib/scanner.cpp

@@ -244,6 +244,25 @@ gadataset_scanner_builder_new_record_batch_reader(
   return gadataset_scanner_builder_new_raw(&arrow_scanner_builder);
 }
 
+/**
+ * gadataset_scanner_builder_use_async:
+ * @builder: A #GADatasetScannerBuilder.
+ * @use_async: Use the asynchronous scanner
+ * @error: (nullable): Return location for a #GError or %NULL.
+ *
+ * Returns: void
+ *
+ * Since: 6.0.0
+ */
+void
+gadataset_scanner_builder_use_async(GADatasetScannerBuilder *builder, gboolean use_async,
+                                 GError **error)
+{
+  auto arrow_builder = gadataset_scanner_builder_get_raw(builder);
+  auto use_async_result = arrow_builder->UseAsync(use_async);
+  garrow::check(error, use_async_result, "[scanner-builder][use_async]");
+}
+
 /**
  * gadataset_scanner_builder_finish:
  * @builder: A #GADatasetScannerBuilder.

c_glib/arrow-dataset-glib/scanner.h

@@ -59,6 +59,12 @@ GARROW_AVAILABLE_IN_6_0
 GADatasetScannerBuilder *
 gadataset_scanner_builder_new_record_batch_reader(
   GArrowRecordBatchReader *reader);
+
+GARROW_AVAILABLE_IN_6_0
+void
+gadataset_scanner_builder_use_async(
+  GADatasetScannerBuilder *builder, gboolean use_async, GError **error);				    
+
 GARROW_AVAILABLE_IN_5_0
 GADatasetScanner *
 gadataset_scanner_builder_finish(GADatasetScannerBuilder *builder,

c_glib/test/dataset/test-file-system-dataset.rb

@@ -60,6 +60,7 @@ def test_read_write
                         count: build_int32_array([1, 10, 2, 3]))
     table_reader = Arrow::TableBatchReader.new(table)
     scanner_builder = ArrowDataset::ScannerBuilder.new(table_reader)
+    scanner_builder.use_async(true)
     scanner = scanner_builder.finish
     options = ArrowDataset::FileSystemDatasetWriteOptions.new
     options.file_write_options = @format.default_write_options

cpp/src/arrow/CMakeLists.txt

@@ -201,6 +201,7 @@ set(ARROW_SRCS
     util/bpacking.cc
     util/cancel.cc
     util/compression.cc
+    util/counting_semaphore.cc
     util/cpu_info.cc
     util/decimal.cc
     util/delimiting.cc

cpp/src/arrow/dataset/CMakeLists.txt

@@ -21,6 +21,7 @@ arrow_install_all_headers("arrow/dataset")
 
 set(ARROW_DATASET_SRCS
     dataset.cc
+    dataset_writer.cc
     discovery.cc
     file_base.cc
     file_ipc.cc
@@ -110,6 +111,7 @@ function(ADD_ARROW_DATASET_TEST REL_TEST_NAME)
 endfunction()
 
 add_arrow_dataset_test(dataset_test)
+add_arrow_dataset_test(dataset_writer_test)
 add_arrow_dataset_test(discovery_test)
 add_arrow_dataset_test(file_ipc_test)
 add_arrow_dataset_test(file_test)