Update patchfile from arrow 21 tag

Author: IsaacWarren

recipe/patches/0004-Bodo-Changes.patch

@@ -1,9 +1,9 @@
 diff --git a/cpp/src/arrow/dataset/file_parquet.cc b/cpp/src/arrow/dataset/file_parquet.cc
-index e6ac389..60561d2 100644
+index 62b8f57ba1..a108fb51eb 100644
 --- a/cpp/src/arrow/dataset/file_parquet.cc
 +++ b/cpp/src/arrow/dataset/file_parquet.cc
 @@ -26,16 +26,23 @@
-
+ 
  #include "arrow/compute/cast.h"
  #include "arrow/compute/exec.h"
 +#include "arrow/dataset/dataset.h"
@@ -20,16 +20,16 @@ index e6ac389..60561d2 100644
  #include "arrow/util/checked_cast.h"
  #include "arrow/util/future.h"
  #include "arrow/util/iterator.h"
- #include "arrow/util/logging.h"
+ #include "arrow/util/logging_internal.h"
  #include "arrow/util/range.h"
 +#include "arrow/util/thread_pool.h"
  #include "arrow/util/tracing_internal.h"
  #include "parquet/arrow/reader.h"
  #include "parquet/arrow/schema.h"
-@@ -558,6 +565,68 @@ Future<std::shared_ptr<parquet::arrow::FileReader>> ParquetFileFormat::GetReader
+@@ -564,6 +571,68 @@ Future<std::shared_ptr<parquet::arrow::FileReader>> ParquetFileFormat::GetReader
        });
  }
-
+ 
 +struct CastingGenerator {
 +  CastingGenerator(RecordBatchGenerator source, std::shared_ptr<Schema> final_schema,
 +                   const std::unordered_set<std::string>& cols_to_skip,
@@ -95,7 +95,7 @@ index e6ac389..60561d2 100644
  struct SlicingGenerator {
    SlicingGenerator(RecordBatchGenerator source, int64_t batch_size)
        : state(std::make_shared<State>(source, batch_size)) {}
-@@ -620,6 +689,9 @@ Result<RecordBatchGenerator> ParquetFileFormat::ScanBatchesAsync(
+@@ -626,6 +695,9 @@ Result<RecordBatchGenerator> ParquetFileFormat::ScanBatchesAsync(
        [this, options, parquet_fragment, pre_filtered,
         row_groups](const std::shared_ptr<parquet::arrow::FileReader>& reader) mutable
        -> Result<RecordBatchGenerator> {
@@ -105,7 +105,7 @@ index e6ac389..60561d2 100644
      // Ensure that parquet_fragment has FileMetaData
      RETURN_NOT_OK(parquet_fragment->EnsureCompleteMetadata(reader.get()));
      if (!pre_filtered) {
-@@ -636,12 +708,24 @@ Result<RecordBatchGenerator> ParquetFileFormat::ScanBatchesAsync(
+@@ -642,12 +714,24 @@ Result<RecordBatchGenerator> ParquetFileFormat::ScanBatchesAsync(
              kParquetTypeName, options.get(), default_fragment_scan_options));
      int batch_readahead = options->batch_readahead;
      int64_t rows_to_readahead = batch_readahead * options->batch_size;
@@ -136,7 +136,7 @@ index e6ac389..60561d2 100644
        return sliced;
      }
 diff --git a/cpp/src/arrow/dataset/scanner.cc b/cpp/src/arrow/dataset/scanner.cc
-index 8fa45ac..7cd0b73 100644
+index a8c8c6bde6..93d0d1934e 100644
 --- a/cpp/src/arrow/dataset/scanner.cc
 +++ b/cpp/src/arrow/dataset/scanner.cc
 @@ -360,8 +360,10 @@ class OneShotFragment : public Fragment {
@@ -151,15 +151,15 @@ index 8fa45ac..7cd0b73 100644
 +    return MakeTransferredGenerator(std::move(background_gen), cpu_executor);
    }
    std::string type_name() const override { return "one-shot"; }
-
+ 
 @@ -387,7 +389,7 @@ Result<TaggedRecordBatchIterator> AsyncScanner::ScanBatches() {
        [this](::arrow::internal::Executor* executor) {
          return ScanBatchesAsync(executor);
        },
 -      scan_options_->use_threads);
 +      scan_options_->use_threads, this->async_cpu_executor());
  }
-
+ 
  Result<EnumeratedRecordBatchIterator> AsyncScanner::ScanBatchesUnordered() {
 @@ -395,7 +397,7 @@ Result<EnumeratedRecordBatchIterator> AsyncScanner::ScanBatchesUnordered() {
        [this](::arrow::internal::Executor* executor) {
@@ -168,51 +168,51 @@ index 8fa45ac..7cd0b73 100644
 -      scan_options_->use_threads);
 +      scan_options_->use_threads, this->async_cpu_executor());
  }
-
+ 
  Result<std::shared_ptr<Table>> AsyncScanner::ToTable() {
 @@ -405,7 +407,7 @@ Result<std::shared_ptr<Table>> AsyncScanner::ToTable() {
  }
-
+ 
  Result<EnumeratedRecordBatchGenerator> AsyncScanner::ScanBatchesUnorderedAsync() {
 -  return ScanBatchesUnorderedAsync(::arrow::internal::GetCpuThreadPool(),
 +  return ScanBatchesUnorderedAsync(this->async_cpu_executor(),
                                     /*sequence_fragments=*/false);
  }
-
+ 
 @@ -606,7 +608,7 @@ Result<std::shared_ptr<Table>> AsyncScanner::Head(int64_t num_rows) {
  }
-
+ 
  Result<TaggedRecordBatchGenerator> AsyncScanner::ScanBatchesAsync() {
 -  return ScanBatchesAsync(::arrow::internal::GetCpuThreadPool());
 +  return ScanBatchesAsync(this->async_cpu_executor());
  }
-
+ 
  Result<TaggedRecordBatchGenerator> AsyncScanner::ScanBatchesAsync(
 @@ -783,7 +785,7 @@ Future<int64_t> AsyncScanner::CountRowsAsync(Executor* executor) {
  }
-
+ 
  Future<int64_t> AsyncScanner::CountRowsAsync() {
 -  return CountRowsAsync(::arrow::internal::GetCpuThreadPool());
 +  return CountRowsAsync(this->async_cpu_executor());
  }
-
+ 
  Result<int64_t> AsyncScanner::CountRows() {
 diff --git a/cpp/src/arrow/dataset/scanner.h b/cpp/src/arrow/dataset/scanner.h
-index 5031057..c867ece 100644
+index 50310577f1..0b5162f25f 100644
 --- a/cpp/src/arrow/dataset/scanner.h
 +++ b/cpp/src/arrow/dataset/scanner.h
-@@ -35,6 +35,7 @@
- #include "arrow/type_fwd.h"
+@@ -36,6 +36,7 @@
  #include "arrow/util/async_generator_fwd.h"
  #include "arrow/util/iterator.h"
-+#include "arrow/util/thread_pool.h"
  #include "arrow/util/type_fwd.h"
-
++#include "arrow/util/thread_pool.h"
+ 
  namespace arrow {
+ 
 @@ -104,6 +105,11 @@ struct ARROW_DS_EXPORT ScanOptions {
    /// Note: The IOContext executor will be ignored if use_threads is set to false
    io::IOContext io_context;
-
+ 
 +  /// ExecContext for any CPU tasks
 +  ///
 +  /// Note: The ExecContext executor will be ignored if use_threads is set to false
@@ -223,64 +223,13 @@ index 5031057..c867ece 100644
    /// Note: If true, this will use threads from both the cpu_executor and the
 @@ -459,6 +465,11 @@ class ARROW_DS_EXPORT Scanner {
        TaggedRecordBatchIterator scan);
-
+ 
    const std::shared_ptr<ScanOptions> scan_options_;
 +
 +  ::arrow::internal::Executor* async_cpu_executor() const {
 +    return scan_options_->exec_context.executor() ? scan_options_->exec_context.executor()
 +                                                  : ::arrow::internal::GetCpuThreadPool();
 +  }
  };
-
+ 
  /// \brief ScannerBuilder is a factory class to construct a Scanner. It is used
-diff --git a/cpp/src/arrow/util/thread_pool.h b/cpp/src/arrow/util/thread_pool.h
-index cd32781..a8935d7 100644
---- a/cpp/src/arrow/util/thread_pool.h
-+++ b/cpp/src/arrow/util/thread_pool.h
-@@ -20,6 +20,7 @@
- #include <cstdint>
- #include <memory>
- #include <queue>
-+#include <thread>
- #include <type_traits>
- #include <unordered_set>
- #include <utility>
-@@ -591,6 +592,21 @@ typename Fut::SyncType RunSynchronously(FnOnce<Fut(Executor*)> get_future,
-   }
- }
-
-+template <typename T>
-+Iterator<T> IterateSynchronously(
-+    FnOnce<Result<std::function<Future<T>()>>(Executor*)> get_gen, bool use_threads,
-+    Executor* executor) {
-+  if (use_threads) {
-+    auto maybe_gen = std::move(get_gen)(executor);
-+    if (!maybe_gen.ok()) {
-+      return MakeErrorIterator<T>(maybe_gen.status());
-+    }
-+    return MakeGeneratorIterator(*maybe_gen);
-+  } else {
-+    return SerialExecutor::IterateGenerator(std::move(get_gen));
-+  }
-+}
-+
- /// \brief Potentially iterate an async generator serially (if use_threads is false)
- /// \see IterateGenerator
- ///
-@@ -605,15 +621,7 @@ typename Fut::SyncType RunSynchronously(FnOnce<Fut(Executor*)> get_future,
- template <typename T>
- Iterator<T> IterateSynchronously(
-     FnOnce<Result<std::function<Future<T>()>>(Executor*)> get_gen, bool use_threads) {
--  if (use_threads) {
--    auto maybe_gen = std::move(get_gen)(GetCpuThreadPool());
--    if (!maybe_gen.ok()) {
--      return MakeErrorIterator<T>(maybe_gen.status());
--    }
--    return MakeGeneratorIterator(*maybe_gen);
--  } else {
--    return SerialExecutor::IterateGenerator(std::move(get_gen));
--  }
-+  return IterateSynchronously(std::move(get_gen), use_threads, GetCpuThreadPool());
- }
-
- }  // namespace internal