From f3d31c0a52c74a14465c20037bee745e965c22ef Mon Sep 17 00:00:00 2001
From: Sahil <sahil@bodo.ai>
Date: Mon, 12 Aug 2024 14:19:23 -0400
Subject: [PATCH 01/16] Add a CastingGenerator to Parquet Reader that can apply
 required casts before slices are created

---
 cpp/src/arrow/dataset/file_parquet.cc | 68 ++++++++++++++++++++++++++-
 1 file changed, 67 insertions(+), 1 deletion(-)

diff --git a/cpp/src/arrow/dataset/file_parquet.cc b/cpp/src/arrow/dataset/file_parquet.cc
index 1f8b6cc4882c..4756bb95c565 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"
 #include "arrow/dataset/dataset_internal.h"
 #include "arrow/dataset/parquet_encryption_config.h"
 #include "arrow/dataset/scanner.h"
 #include "arrow/filesystem/path_util.h"
+#include "arrow/memory_pool.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
 #include "arrow/table.h"
+#include "arrow/type.h"
+#include "arrow/type_fwd.h"
 #include "arrow/util/checked_cast.h"
 #include "arrow/util/future.h"
 #include "arrow/util/iterator.h"
 #include "arrow/util/logging.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"
@@ -555,6 +562,58 @@ Future<std::shared_ptr<parquet::arrow::FileReader>> ParquetFileFormat::GetReader
       });
 }
 
+struct CastingGenerator {
+  CastingGenerator(RecordBatchGenerator source, std::shared_ptr<Schema> final_schema,
+                   arrow::MemoryPool* pool = arrow::default_memory_pool())
+      : source_(source),
+        final_schema_(final_schema),
+        exec_ctx(std::make_shared<compute::ExecContext>(pool)) {}
+
+  Future<std::shared_ptr<RecordBatch>> operator()() {
+    return this->source_().Then(
+        [this](const std::shared_ptr<RecordBatch>& next) -> std::shared_ptr<RecordBatch> {
+          if (IsIterationEnd(next)) {
+            return next;
+          }
+          std::vector<std::shared_ptr<::arrow::Array>> out_cols;
+          std::vector<std::shared_ptr<arrow::Field>> out_schema_fields;
+
+          bool changed = false;
+          for (const auto& field : this->final_schema_->fields()) {
+            FieldRef field_ref = FieldRef(field->name());
+            auto column_st = field_ref.GetOneOrNone(*next);
+            std::shared_ptr<Array> column = column_st.ValueUnsafe();
+            if (column) {
+              if (!column->type()->Equals(field->type())) {
+                // Referenced field was present but didn't have the expected type.
+                auto converted_st =
+                    compute::Cast(column, field->type(), compute::CastOptions::Safe(),
+                                  this->exec_ctx.get());
+                auto converted = std::move(converted_st.ValueUnsafe());
+                column = converted.make_array();
+                changed = true;
+              }
+              out_cols.emplace_back(std::move(column));
+              out_schema_fields.emplace_back(field->Copy());
+            }
+          }
+
+          if (changed) {
+            return RecordBatch::Make(
+                std::make_shared<Schema>(std::move(out_schema_fields),
+                                         next->schema()->metadata()),
+                next->num_rows(), std::move(out_cols));
+          } else {
+            return next;
+          }
+        });
+  }
+
+  RecordBatchGenerator source_;
+  std::shared_ptr<Schema> final_schema_;
+  std::shared_ptr<compute::ExecContext> exec_ctx;
+};
+
 struct SlicingGenerator {
   SlicingGenerator(RecordBatchGenerator source, int64_t batch_size)
       : state(std::make_shared<State>(source, batch_size)) {}
@@ -617,6 +676,9 @@ Result<RecordBatchGenerator> ParquetFileFormat::ScanBatchesAsync(
       [this, options, parquet_fragment, pre_filtered,
        row_groups](const std::shared_ptr<parquet::arrow::FileReader>& reader) mutable
       -> Result<RecordBatchGenerator> {
+    // Since we already do the batching through the SlicingGenerator, we don't need the
+    // reader to batch its output.
+    reader->set_batch_size(std::numeric_limits<int64_t>::max());
     // Ensure that parquet_fragment has FileMetaData
     RETURN_NOT_OK(parquet_fragment->EnsureCompleteMetadata(reader.get()));
     if (!pre_filtered) {
@@ -637,8 +699,12 @@ Result<RecordBatchGenerator> ParquetFileFormat::ScanBatchesAsync(
                           reader->GetRecordBatchGenerator(
                               reader, row_groups, column_projection,
                               ::arrow::internal::GetCpuThreadPool(), rows_to_readahead));
+    // Casting before slicing is more efficient. Casts on slices might require wasteful
+    // allocations and computation.
+    RecordBatchGenerator casted =
+        CastingGenerator(std::move(generator), options->dataset_schema, options->pool);
     RecordBatchGenerator sliced =
-        SlicingGenerator(std::move(generator), options->batch_size);
+        SlicingGenerator(std::move(casted), options->batch_size);
     if (batch_readahead == 0) {
       return sliced;
     }

From c47a9ecde673ad23be3809cabf4774895b70c997 Mon Sep 17 00:00:00 2001
From: Sahil <sahil@bodo.ai>
Date: Mon, 12 Aug 2024 18:34:22 -0400
Subject: [PATCH 02/16] add missing error handling in CastingGenerator

---
 cpp/src/arrow/dataset/file_parquet.cc | 13 +++++--------
 1 file changed, 5 insertions(+), 8 deletions(-)

diff --git a/cpp/src/arrow/dataset/file_parquet.cc b/cpp/src/arrow/dataset/file_parquet.cc
index 4756bb95c565..5b3567776d9b 100644
--- a/cpp/src/arrow/dataset/file_parquet.cc
+++ b/cpp/src/arrow/dataset/file_parquet.cc
@@ -571,8 +571,8 @@ struct CastingGenerator {
 
   Future<std::shared_ptr<RecordBatch>> operator()() {
     return this->source_().Then(
-        [this](const std::shared_ptr<RecordBatch>& next) -> std::shared_ptr<RecordBatch> {
-          if (IsIterationEnd(next)) {
+        [this](const std::shared_ptr<RecordBatch>& next) -> Result<std::shared_ptr<RecordBatch>> {
+          if (IsIterationEnd(next) || this->final_schema_.get() == nullptr) {
             return next;
           }
           std::vector<std::shared_ptr<::arrow::Array>> out_cols;
@@ -581,15 +581,12 @@ struct CastingGenerator {
           bool changed = false;
           for (const auto& field : this->final_schema_->fields()) {
             FieldRef field_ref = FieldRef(field->name());
-            auto column_st = field_ref.GetOneOrNone(*next);
-            std::shared_ptr<Array> column = column_st.ValueUnsafe();
+            ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Array> column, field_ref.GetOneOrNone(*next));
             if (column) {
               if (!column->type()->Equals(field->type())) {
                 // Referenced field was present but didn't have the expected type.
-                auto converted_st =
-                    compute::Cast(column, field->type(), compute::CastOptions::Safe(),
-                                  this->exec_ctx.get());
-                auto converted = std::move(converted_st.ValueUnsafe());
+                ARROW_ASSIGN_OR_RAISE(auto converted, compute::Cast(column, field->type(), compute::CastOptions::Safe(),
+                                  this->exec_ctx.get()));
                 column = converted.make_array();
                 changed = true;
               }

From 6fc1aaa574ff6610a96557f4cab4f0980166a97b Mon Sep 17 00:00:00 2001
From: Sahil <sahil@bodo.ai>
Date: Mon, 12 Aug 2024 18:57:57 -0400
Subject: [PATCH 03/16] run formatter

---
 cpp/src/arrow/dataset/file_parquet.cc | 66 ++++++++++++++-------------
 1 file changed, 34 insertions(+), 32 deletions(-)

diff --git a/cpp/src/arrow/dataset/file_parquet.cc b/cpp/src/arrow/dataset/file_parquet.cc
index 5b3567776d9b..664c1e8a64c5 100644
--- a/cpp/src/arrow/dataset/file_parquet.cc
+++ b/cpp/src/arrow/dataset/file_parquet.cc
@@ -570,40 +570,42 @@ struct CastingGenerator {
         exec_ctx(std::make_shared<compute::ExecContext>(pool)) {}
 
   Future<std::shared_ptr<RecordBatch>> operator()() {
-    return this->source_().Then(
-        [this](const std::shared_ptr<RecordBatch>& next) -> Result<std::shared_ptr<RecordBatch>> {
-          if (IsIterationEnd(next) || this->final_schema_.get() == nullptr) {
-            return next;
-          }
-          std::vector<std::shared_ptr<::arrow::Array>> out_cols;
-          std::vector<std::shared_ptr<arrow::Field>> out_schema_fields;
-
-          bool changed = false;
-          for (const auto& field : this->final_schema_->fields()) {
-            FieldRef field_ref = FieldRef(field->name());
-            ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Array> column, field_ref.GetOneOrNone(*next));
-            if (column) {
-              if (!column->type()->Equals(field->type())) {
-                // Referenced field was present but didn't have the expected type.
-                ARROW_ASSIGN_OR_RAISE(auto converted, compute::Cast(column, field->type(), compute::CastOptions::Safe(),
-                                  this->exec_ctx.get()));
-                column = converted.make_array();
-                changed = true;
-              }
-              out_cols.emplace_back(std::move(column));
-              out_schema_fields.emplace_back(field->Copy());
-            }
+    return this->source_().Then([this](const std::shared_ptr<RecordBatch>& next)
+                                    -> Result<std::shared_ptr<RecordBatch>> {
+      if (IsIterationEnd(next) || this->final_schema_.get() == nullptr) {
+        return next;
+      }
+      std::vector<std::shared_ptr<::arrow::Array>> out_cols;
+      std::vector<std::shared_ptr<arrow::Field>> out_schema_fields;
+
+      bool changed = false;
+      for (const auto& field : this->final_schema_->fields()) {
+        FieldRef field_ref = FieldRef(field->name());
+        ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Array> column,
+                              field_ref.GetOneOrNone(*next));
+        if (column) {
+          if (!column->type()->Equals(field->type())) {
+            // Referenced field was present but didn't have the expected type.
+            ARROW_ASSIGN_OR_RAISE(
+                auto converted,
+                compute::Cast(column, field->type(), compute::CastOptions::Safe(),
+                              this->exec_ctx.get()));
+            column = converted.make_array();
+            changed = true;
           }
+          out_cols.emplace_back(std::move(column));
+          out_schema_fields.emplace_back(field->Copy());
+        }
+      }
 
-          if (changed) {
-            return RecordBatch::Make(
-                std::make_shared<Schema>(std::move(out_schema_fields),
-                                         next->schema()->metadata()),
-                next->num_rows(), std::move(out_cols));
-          } else {
-            return next;
-          }
-        });
+      if (changed) {
+        return RecordBatch::Make(std::make_shared<Schema>(std::move(out_schema_fields),
+                                                          next->schema()->metadata()),
+                                 next->num_rows(), std::move(out_cols));
+      } else {
+        return next;
+      }
+    });
   }
 
   RecordBatchGenerator source_;

From 60747bffa7ce22f8627aa4128772f8f3cd21b2e5 Mon Sep 17 00:00:00 2001
From: Sahil <sahil@bodo.ai>
Date: Mon, 12 Aug 2024 19:11:11 -0400
Subject: [PATCH 04/16] remove redundant arrow::s

---
 cpp/src/arrow/dataset/file_parquet.cc | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/cpp/src/arrow/dataset/file_parquet.cc b/cpp/src/arrow/dataset/file_parquet.cc
index 664c1e8a64c5..0de74ba107f2 100644
--- a/cpp/src/arrow/dataset/file_parquet.cc
+++ b/cpp/src/arrow/dataset/file_parquet.cc
@@ -564,7 +564,7 @@ Future<std::shared_ptr<parquet::arrow::FileReader>> ParquetFileFormat::GetReader
 
 struct CastingGenerator {
   CastingGenerator(RecordBatchGenerator source, std::shared_ptr<Schema> final_schema,
-                   arrow::MemoryPool* pool = arrow::default_memory_pool())
+                   MemoryPool* pool = default_memory_pool())
       : source_(source),
         final_schema_(final_schema),
         exec_ctx(std::make_shared<compute::ExecContext>(pool)) {}
@@ -572,11 +572,11 @@ struct CastingGenerator {
   Future<std::shared_ptr<RecordBatch>> operator()() {
     return this->source_().Then([this](const std::shared_ptr<RecordBatch>& next)
                                     -> Result<std::shared_ptr<RecordBatch>> {
-      if (IsIterationEnd(next) || this->final_schema_.get() == nullptr) {
+      if (IsIterationEnd(next) || this->final_schema_ == nullptr) {
         return next;
       }
-      std::vector<std::shared_ptr<::arrow::Array>> out_cols;
-      std::vector<std::shared_ptr<arrow::Field>> out_schema_fields;
+      std::vector<std::shared_ptr<Array>> out_cols;
+      std::vector<std::shared_ptr<Field>> out_schema_fields;
 
       bool changed = false;
       for (const auto& field : this->final_schema_->fields()) {

From 8156cb738d4c37f35e76af8db3aa997d8c6343a2 Mon Sep 17 00:00:00 2001
From: Sahil <sahil@bodo.ai>
Date: Tue, 13 Aug 2024 09:50:19 -0400
Subject: [PATCH 05/16] skip dict-encoded columns

---
 cpp/src/arrow/dataset/file_parquet.cc | 18 ++++++++++++++++--
 1 file changed, 16 insertions(+), 2 deletions(-)

diff --git a/cpp/src/arrow/dataset/file_parquet.cc b/cpp/src/arrow/dataset/file_parquet.cc
index 0de74ba107f2..299ae7693d0e 100644
--- a/cpp/src/arrow/dataset/file_parquet.cc
+++ b/cpp/src/arrow/dataset/file_parquet.cc
@@ -564,9 +564,11 @@ 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,
                    MemoryPool* pool = default_memory_pool())
       : source_(source),
         final_schema_(final_schema),
+        cols_to_skip_(cols_to_skip),
         exec_ctx(std::make_shared<compute::ExecContext>(pool)) {}
 
   Future<std::shared_ptr<RecordBatch>> operator()() {
@@ -584,6 +586,12 @@ struct CastingGenerator {
         ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Array> column,
                               field_ref.GetOneOrNone(*next));
         if (column) {
+          if (this->cols_to_skip_.count(field->name())) {
+            out_cols.emplace_back(std::move(column));
+            // Maintain the original input type.
+            out_schema_fields.emplace_back(field->WithType(column->type()));
+            continue;
+          }
           if (!column->type()->Equals(field->type())) {
             // Referenced field was present but didn't have the expected type.
             ARROW_ASSIGN_OR_RAISE(
@@ -610,6 +618,7 @@ struct CastingGenerator {
 
   RecordBatchGenerator source_;
   std::shared_ptr<Schema> final_schema_;
+  const std::unordered_set<std::string>& cols_to_skip_;
   std::shared_ptr<compute::ExecContext> exec_ctx;
 };
 
@@ -698,10 +707,15 @@ Result<RecordBatchGenerator> ParquetFileFormat::ScanBatchesAsync(
                           reader->GetRecordBatchGenerator(
                               reader, row_groups, column_projection,
                               ::arrow::internal::GetCpuThreadPool(), rows_to_readahead));
+    // We need to skip casting the dictionary columns since the dataset_schema doesn't
+    // have the dictionary-encoding information. Parquet reader will return them with the
+    // dictionary type, which is what we eventually want.
+    const std::unordered_set<std::string>& dict_cols =
+        parquet_fragment->parquet_format_.reader_options.dict_columns;
     // Casting before slicing is more efficient. Casts on slices might require wasteful
     // allocations and computation.
-    RecordBatchGenerator casted =
-        CastingGenerator(std::move(generator), options->dataset_schema, options->pool);
+    RecordBatchGenerator casted = CastingGenerator(
+        std::move(generator), options->dataset_schema, dict_cols, options->pool);
     RecordBatchGenerator sliced =
         SlicingGenerator(std::move(casted), options->batch_size);
     if (batch_readahead == 0) {

From e1d5d7f7d5129ab990d2ac6b83f5cd67e86aae11 Mon Sep 17 00:00:00 2001
From: Sahil <sahil@bodo.ai>
Date: Tue, 13 Aug 2024 11:21:52 -0400
Subject: [PATCH 06/16] get array type before the move

---
 cpp/src/arrow/dataset/file_parquet.cc | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/cpp/src/arrow/dataset/file_parquet.cc b/cpp/src/arrow/dataset/file_parquet.cc
index 299ae7693d0e..a1556373c175 100644
--- a/cpp/src/arrow/dataset/file_parquet.cc
+++ b/cpp/src/arrow/dataset/file_parquet.cc
@@ -587,9 +587,9 @@ struct CastingGenerator {
                               field_ref.GetOneOrNone(*next));
         if (column) {
           if (this->cols_to_skip_.count(field->name())) {
-            out_cols.emplace_back(std::move(column));
             // Maintain the original input type.
             out_schema_fields.emplace_back(field->WithType(column->type()));
+            out_cols.emplace_back(std::move(column));
             continue;
           }
           if (!column->type()->Equals(field->type())) {
@@ -601,8 +601,8 @@ struct CastingGenerator {
             column = converted.make_array();
             changed = true;
           }
-          out_cols.emplace_back(std::move(column));
           out_schema_fields.emplace_back(field->Copy());
+          out_cols.emplace_back(std::move(column));
         }
       }
 

From 019f795b2b8b2f2fabd623186326b12afa78577a Mon Sep 17 00:00:00 2001
From: Sahil <sahil@bodo.ai>
Date: Tue, 13 Aug 2024 12:11:35 -0400
Subject: [PATCH 07/16] [run CI]


From f1a495540a288e3004af14f9be2068bdbaacd198 Mon Sep 17 00:00:00 2001
From: Sahil <sahil@bodo.ai>
Date: Thu, 29 Aug 2024 14:22:23 -0400
Subject: [PATCH 08/16] Use std::move on final_schema

---
 cpp/src/arrow/dataset/file_parquet.cc | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/cpp/src/arrow/dataset/file_parquet.cc b/cpp/src/arrow/dataset/file_parquet.cc
index a1556373c175..6698013fd3e3 100644
--- a/cpp/src/arrow/dataset/file_parquet.cc
+++ b/cpp/src/arrow/dataset/file_parquet.cc
@@ -567,7 +567,7 @@ struct CastingGenerator {
                    const std::unordered_set<std::string>& cols_to_skip,
                    MemoryPool* pool = default_memory_pool())
       : source_(source),
-        final_schema_(final_schema),
+        final_schema_(std::move(final_schema)),
         cols_to_skip_(cols_to_skip),
         exec_ctx(std::make_shared<compute::ExecContext>(pool)) {}
 

From 6444d1f265e14b173c58d054b786f57063cfe810 Mon Sep 17 00:00:00 2001
From: Srinivas Lade <srinulade1@gmail.com>
Date: Tue, 10 Dec 2024 15:16:31 -0500
Subject: [PATCH 09/16] Retrigger CI


From 4862a149105162544c5b22565279627695483bdc Mon Sep 17 00:00:00 2001
From: Scott Routledge <scott@bodo.ai>
Date: Tue, 23 Sep 2025 23:01:56 -0400
Subject: [PATCH 10/16] remove debug prints

---
 cpp/src/arrow/dataset/file_parquet.cc | 2 --
 1 file changed, 2 deletions(-)

diff --git a/cpp/src/arrow/dataset/file_parquet.cc b/cpp/src/arrow/dataset/file_parquet.cc
index 0afeb27bf79c..8c31eba3d3be 100644
--- a/cpp/src/arrow/dataset/file_parquet.cc
+++ b/cpp/src/arrow/dataset/file_parquet.cc
@@ -21,7 +21,6 @@
 #include <mutex>
 #include <unordered_map>
 #include <unordered_set>
-#include <iostream>
 #include <utility>
 #include <vector>
 
@@ -677,7 +676,6 @@ struct SlicingGenerator {
 Result<RecordBatchGenerator> ParquetFileFormat::ScanBatchesAsync(
     const std::shared_ptr<ScanOptions>& options,
     const std::shared_ptr<FileFragment>& file) const {
-  std::cout << "[DEBUG] got here..." << std::endl;
   auto parquet_fragment = checked_pointer_cast<ParquetFileFragment>(file);
   std::vector<int> row_groups;
   bool pre_filtered = false;

From 3d84d8314a183bdc92a2f70be619f8689382c086 Mon Sep 17 00:00:00 2001
From: Scott Routledge <scott@bodo.ai>
Date: Thu, 25 Sep 2025 15:58:44 -0400
Subject: [PATCH 11/16] add benchmark for scanning parquet files with cast

---
 cpp/src/arrow/dataset/CMakeLists.txt          |  4 +
 .../arrow/dataset/parquet_scan_benchmark.cc   | 88 +++++++++++++++++++
 2 files changed, 92 insertions(+)
 create mode 100644 cpp/src/arrow/dataset/parquet_scan_benchmark.cc

diff --git a/cpp/src/arrow/dataset/CMakeLists.txt b/cpp/src/arrow/dataset/CMakeLists.txt
index fa6875527db9..ce5d4784a11b 100644
--- a/cpp/src/arrow/dataset/CMakeLists.txt
+++ b/cpp/src/arrow/dataset/CMakeLists.txt
@@ -241,3 +241,7 @@ endfunction()
 
 add_arrow_dataset_benchmark(file_benchmark)
 add_arrow_dataset_benchmark(scanner_benchmark)
+
+if(ARROW_PARQUET)
+  add_arrow_dataset_benchmark(parquet_scan_benchmark)
+endif()
diff --git a/cpp/src/arrow/dataset/parquet_scan_benchmark.cc b/cpp/src/arrow/dataset/parquet_scan_benchmark.cc
new file mode 100644
index 000000000000..0ed20f7500bb
--- /dev/null
+++ b/cpp/src/arrow/dataset/parquet_scan_benchmark.cc
@@ -0,0 +1,88 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow/testing/gtest_util.h"
+#include "benchmark/benchmark.h"
+
+#include "arrow/api.h"
+#include "arrow/compute/initialize.h"
+#include "arrow/dataset/dataset.h"
+#include "arrow/dataset/file_parquet.h"
+#include "arrow/dataset/scanner.h"
+#include "arrow/io/memory.h"
+#include "parquet/arrow/writer.h"
+
+namespace arrow {
+namespace dataset {
+
+using parquet::arrow::WriteTable;
+
+Result<std::shared_ptr<Buffer>> WriteStringColParquetBuffer(int64_t nrows) {
+  auto schema = arrow::schema({arrow::field("my_string_col", arrow::utf8())});
+
+  arrow::StringBuilder builder;
+  for (int64_t i = 0; i < nrows; i++) {
+    ARROW_RETURN_NOT_OK(builder.Append("row_" + std::to_string(i)));
+  }
+  std::shared_ptr<arrow::Array> arr;
+  ARROW_RETURN_NOT_OK(builder.Finish(&arr));
+  auto table = arrow::Table::Make(schema, {arr});
+
+  ARROW_ASSIGN_OR_RAISE(auto sink, arrow::io::BufferOutputStream::Create());
+  ARROW_RETURN_NOT_OK(WriteTable(*table, arrow::default_memory_pool(), sink));
+  return sink->Finish();
+}
+
+static void ParquetScanToTableCastStrings(benchmark::State& state) {
+  // GH-43660: Scan parquet data including a String column using a dataset object with
+  // LargeString in schema.
+  int64_t nrows = 100'000;
+  int64_t batch_size = 100;
+  bool use_threads = false;
+  auto format = std::make_shared<ParquetFileFormat>();
+
+  // Create a buffer with a single String column and wrap with FileFragment
+  ASSERT_OK_AND_ASSIGN(std::shared_ptr<Buffer> buffer,
+                       WriteStringColParquetBuffer(nrows));
+  auto buffer_reader = std::make_shared<arrow::io::BufferReader>(buffer);
+  FileSource source(buffer_reader, buffer->size());
+  ASSERT_OK_AND_ASSIGN(auto fragment, format->MakeFragment(source));
+  std::vector<std::shared_ptr<FileFragment>> fragments{fragment};
+
+  // Create a dataset from FileFragment and set schema to LargeString (require casting).
+  auto schema = arrow::schema({field("my_string_col", arrow::large_utf8())});
+  ASSERT_OK_AND_ASSIGN(auto dataset, FileSystemDataset::Make(
+                                         schema, compute::literal(true), format,
+                                         /*filesystem=*/nullptr, std::move(fragments)));
+
+  ASSERT_OK_AND_ASSIGN(auto builder, dataset->NewScan());
+  ASSERT_OK(builder->BatchSize(batch_size));
+  ASSERT_OK(builder->UseThreads(use_threads));
+  ASSERT_OK_AND_ASSIGN(auto scanner, builder->Finish());
+
+  for (auto _ : state) {
+    ASSERT_OK_AND_ASSIGN(auto table, scanner->ToTable());
+    benchmark::DoNotOptimize(table);
+  }
+
+  state.SetItemsProcessed(state.iterations() * nrows);
+}
+
+BENCHMARK(ParquetScanToTableCastStrings);
+
+}  // namespace dataset
+}  // namespace arrow

From 9c8044741e17a0b804c43c1144ebca3f2f2759ab Mon Sep 17 00:00:00 2001
From: Scott Routledge <scott@bodo.ai>
Date: Tue, 7 Oct 2025 10:35:47 -0400
Subject: [PATCH 12/16] try different batch sizes

---
 .../arrow/dataset/parquet_scan_benchmark.cc    | 18 +++++++++++++-----
 1 file changed, 13 insertions(+), 5 deletions(-)

diff --git a/cpp/src/arrow/dataset/parquet_scan_benchmark.cc b/cpp/src/arrow/dataset/parquet_scan_benchmark.cc
index 0ed20f7500bb..a733436e264d 100644
--- a/cpp/src/arrow/dataset/parquet_scan_benchmark.cc
+++ b/cpp/src/arrow/dataset/parquet_scan_benchmark.cc
@@ -50,9 +50,9 @@ Result<std::shared_ptr<Buffer>> WriteStringColParquetBuffer(int64_t nrows) {
 static void ParquetScanToTableCastStrings(benchmark::State& state) {
   // GH-43660: Scan parquet data including a String column using a dataset object with
   // LargeString in schema.
-  int64_t nrows = 100'000;
-  int64_t batch_size = 100;
-  bool use_threads = false;
+  size_t num_batches = state.range(0);
+  size_t batch_size = state.range(1);
+  size_t nrows = num_batches * batch_size;
   auto format = std::make_shared<ParquetFileFormat>();
 
   // Create a buffer with a single String column and wrap with FileFragment
@@ -71,7 +71,6 @@ static void ParquetScanToTableCastStrings(benchmark::State& state) {
 
   ASSERT_OK_AND_ASSIGN(auto builder, dataset->NewScan());
   ASSERT_OK(builder->BatchSize(batch_size));
-  ASSERT_OK(builder->UseThreads(use_threads));
   ASSERT_OK_AND_ASSIGN(auto scanner, builder->Finish());
 
   for (auto _ : state) {
@@ -82,7 +81,16 @@ static void ParquetScanToTableCastStrings(benchmark::State& state) {
   state.SetItemsProcessed(state.iterations() * nrows);
 }
 
-BENCHMARK(ParquetScanToTableCastStrings);
+static void ParquetScanBenchmark_Customize(benchmark::internal::Benchmark* b) {
+  for (const int32_t num_batches : {1000}) {
+    for (const int batch_size : {10, 100, 1000}) {
+      b->Args({num_batches, batch_size});
+    }
+  }
+  b->ArgNames({"num_batches", "batch_size"});
+}
+
+BENCHMARK(ParquetScanToTableCastStrings)->Apply(ParquetScanBenchmark_Customize);
 
 }  // namespace dataset
 }  // namespace arrow

From 7e78f2833a509e86d2e9d49ea19a3beb9cb87390 Mon Sep 17 00:00:00 2001
From: Scott Routledge <scott@bodo.ai>
Date: Fri, 24 Oct 2025 12:00:32 -0700
Subject: [PATCH 13/16] try larger batch size

---
 cpp/src/arrow/dataset/parquet_scan_benchmark.cc | 8 ++++----
 testing                                         | 2 +-
 2 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/cpp/src/arrow/dataset/parquet_scan_benchmark.cc b/cpp/src/arrow/dataset/parquet_scan_benchmark.cc
index a733436e264d..b7cda5970522 100644
--- a/cpp/src/arrow/dataset/parquet_scan_benchmark.cc
+++ b/cpp/src/arrow/dataset/parquet_scan_benchmark.cc
@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <tuple>
 #include "arrow/testing/gtest_util.h"
 #include "benchmark/benchmark.h"
 
@@ -82,10 +83,9 @@ static void ParquetScanToTableCastStrings(benchmark::State& state) {
 }
 
 static void ParquetScanBenchmark_Customize(benchmark::internal::Benchmark* b) {
-  for (const int32_t num_batches : {1000}) {
-    for (const int batch_size : {10, 100, 1000}) {
-      b->Args({num_batches, batch_size});
-    }
+  auto params = {std::make_tuple(1000, 1000), std::make_tuple(10000, 100), std::make_tuple(100000, 10)};
+  for (auto param : params) {
+    b->Args({std::get<0>(param), std::get<1>(param)});
   }
   b->ArgNames({"num_batches", "batch_size"});
 }
diff --git a/testing b/testing
index 725fd4a4b12d..6a7b02fac93d 160000
--- a/testing
+++ b/testing
@@ -1 +1 @@
-Subproject commit 725fd4a4b12d01c53c98e80274c0b23aa8397082
+Subproject commit 6a7b02fac93d8addbcdbb213264e58bfdc3068e4

From 4de55874343603872cdfa1c9431fe3edbe0ed626 Mon Sep 17 00:00:00 2001
From: Scott Routledge <scott@bodo.ai>
Date: Sun, 26 Oct 2025 16:54:33 -0400
Subject: [PATCH 14/16] refine batch sizes

---
 cpp/src/arrow/dataset/parquet_scan_benchmark.cc | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/cpp/src/arrow/dataset/parquet_scan_benchmark.cc b/cpp/src/arrow/dataset/parquet_scan_benchmark.cc
index b7cda5970522..3325263a3942 100644
--- a/cpp/src/arrow/dataset/parquet_scan_benchmark.cc
+++ b/cpp/src/arrow/dataset/parquet_scan_benchmark.cc
@@ -83,9 +83,11 @@ static void ParquetScanToTableCastStrings(benchmark::State& state) {
 }
 
 static void ParquetScanBenchmark_Customize(benchmark::internal::Benchmark* b) {
-  auto params = {std::make_tuple(1000, 1000), std::make_tuple(10000, 100), std::make_tuple(100000, 10)};
-  for (auto param : params) {
-    b->Args({std::get<0>(param), std::get<1>(param)});
+  std::vector<int64_t> num_batches = {1000, 100, 10};
+  std::vector<int64_t> batch_sizes = {1000, 10000, 100000};
+
+  for (size_t i = 0; i < batch_sizes.size(); i++) {
+    b->Args({(num_batches[i]), batch_sizes[i]});
   }
   b->ArgNames({"num_batches", "batch_size"});
 }

From cb4286f36a0f264495f4d20a068ad5f3e395564a Mon Sep 17 00:00:00 2001
From: Scott Routledge <scott@bodo.ai>
Date: Sun, 26 Oct 2025 17:13:06 -0400
Subject: [PATCH 15/16] Revert "try larger batch size"

This reverts commit 7e78f2833a509e86d2e9d49ea19a3beb9cb87390.
---
 cpp/src/arrow/dataset/parquet_scan_benchmark.cc | 1 -
 testing                                         | 2 +-
 2 files changed, 1 insertion(+), 2 deletions(-)

diff --git a/cpp/src/arrow/dataset/parquet_scan_benchmark.cc b/cpp/src/arrow/dataset/parquet_scan_benchmark.cc
index 3325263a3942..26db9f0ffde4 100644
--- a/cpp/src/arrow/dataset/parquet_scan_benchmark.cc
+++ b/cpp/src/arrow/dataset/parquet_scan_benchmark.cc
@@ -15,7 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#include <tuple>
 #include "arrow/testing/gtest_util.h"
 #include "benchmark/benchmark.h"
 
diff --git a/testing b/testing
index 6a7b02fac93d..725fd4a4b12d 160000
--- a/testing
+++ b/testing
@@ -1 +1 @@
-Subproject commit 6a7b02fac93d8addbcdbb213264e58bfdc3068e4
+Subproject commit 725fd4a4b12d01c53c98e80274c0b23aa8397082

From 2353bfc4c95c2ad88e542035663ba56697e8344c Mon Sep 17 00:00:00 2001
From: Scott Routledge <scott@bodo.ai>
Date: Tue, 4 Nov 2025 15:42:40 -0500
Subject: [PATCH 16/16] remove unused includes

---
 cpp/src/arrow/dataset/parquet_scan_benchmark.cc | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/cpp/src/arrow/dataset/parquet_scan_benchmark.cc b/cpp/src/arrow/dataset/parquet_scan_benchmark.cc
index 26db9f0ffde4..1006c243c832 100644
--- a/cpp/src/arrow/dataset/parquet_scan_benchmark.cc
+++ b/cpp/src/arrow/dataset/parquet_scan_benchmark.cc
@@ -19,10 +19,7 @@
 #include "benchmark/benchmark.h"
 
 #include "arrow/api.h"
-#include "arrow/compute/initialize.h"
-#include "arrow/dataset/dataset.h"
 #include "arrow/dataset/file_parquet.h"
-#include "arrow/dataset/scanner.h"
 #include "arrow/io/memory.h"
 #include "parquet/arrow/writer.h"