From 314859eb2202bfbc2336737ee9834ee439eeb18a Mon Sep 17 00:00:00 2001
From: 555aaditya <107977627+555aaditya@users.noreply.github.com>
Date: Wed, 28 Oct 2026 14:20:52 +0530
Subject: [PATCH] GH-48111: [C++][Dataset] Avoid hang when fragment_readahead
 is 0

Unordered scans called MakeMergedGenerator with fragment_readahead,
which never subscribes to any fragment when it is 0. Use the
concatenated generator for fragment_readahead <= 1 on both the
sequenced and unsequenced paths, as GH-15264 already did for the
sequenced one.
---
 cpp/src/arrow/dataset/scanner.cc      | 16 ++++++++--------
 cpp/src/arrow/dataset/scanner_test.cc | 12 ++++++++++++
 python/pyarrow/tests/test_dataset.py  |  8 ++++++++
 3 files changed, 28 insertions(+), 8 deletions(-)

diff --git a/cpp/src/arrow/dataset/scanner.cc b/cpp/src/arrow/dataset/scanner.cc
index 222e1323d4a8..ace3e3247936 100644
--- a/cpp/src/arrow/dataset/scanner.cc
+++ b/cpp/src/arrow/dataset/scanner.cc
@@ -1030,14 +1030,14 @@ Result<acero::ExecNode*> MakeScanNode(acero::ExecPlan* plan,
                         FragmentsToBatches(std::move(fragment_gen), scan_options));
 
   AsyncGenerator<EnumeratedRecordBatch> merged_batch_gen;
-  if (require_sequenced_output) {
-    if (scan_options->fragment_readahead > 1) {
-      ARROW_ASSIGN_OR_RAISE(merged_batch_gen, MakeSequencedMergedGenerator(
-                                                  std::move(batch_gen_gen),
-                                                  scan_options->fragment_readahead));
-    } else {
-      merged_batch_gen = MakeConcatenatedGenerator(std::move(batch_gen_gen));
-    }
+  if (scan_options->fragment_readahead <= 1) {
+    // No fragment readahead: scan one fragment at a time.  MakeMergedGenerator
+    // would never subscribe to any fragment if fragment_readahead is 0.
+    merged_batch_gen = MakeConcatenatedGenerator(std::move(batch_gen_gen));
+  } else if (require_sequenced_output) {
+    ARROW_ASSIGN_OR_RAISE(merged_batch_gen,
+                          MakeSequencedMergedGenerator(std::move(batch_gen_gen),
+                                                       scan_options->fragment_readahead));
   } else {
     merged_batch_gen =
         MakeMergedGenerator(std::move(batch_gen_gen), scan_options->fragment_readahead);
diff --git a/cpp/src/arrow/dataset/scanner_test.cc b/cpp/src/arrow/dataset/scanner_test.cc
index a45847b49def..644f8507d867 100644
--- a/cpp/src/arrow/dataset/scanner_test.cc
+++ b/cpp/src/arrow/dataset/scanner_test.cc
@@ -1049,6 +1049,18 @@ TEST_P(TestScanner, ScanBatchesUnordered) {
   AssertScanBatchesUnorderedEqualRepetitionsOf(MakeScanner(batch), batch);
 }
 
+TEST_P(TestScanner, ScanWithoutFragmentReadahead) {
+  // GH-48111: unordered scans used to hang with fragment_readahead=0
+  SetSchema({field("i32", int32()), field("f64", float64())});
+  auto batch = ConstantArrayGenerator::Zeroes(GetParam().items_per_batch, schema_);
+  for (int fragment_readahead : {0, 1}) {
+    ARROW_SCOPED_TRACE("fragment_readahead = ", fragment_readahead);
+    options_->fragment_readahead = fragment_readahead;
+    AssertScanBatchesEqualRepetitionsOf(MakeScanner(batch), batch);
+    AssertScanBatchesUnorderedEqualRepetitionsOf(MakeScanner(batch), batch);
+  }
+}
+
 TEST_P(TestScanner, ScanWithCappedBatchSize) {
   SetSchema({field("i32", int32()), field("f64", float64())});
   auto batch = ConstantArrayGenerator::Zeroes(GetParam().items_per_batch, schema_);
diff --git a/python/pyarrow/tests/test_dataset.py b/python/pyarrow/tests/test_dataset.py
index 9ddfdfdc9c49..eada682fc5b2 100644
--- a/python/pyarrow/tests/test_dataset.py
+++ b/python/pyarrow/tests/test_dataset.py
@@ -565,6 +565,14 @@ def test_scanner_options(dataset):
     assert batch.num_columns == 7
 
 
+@pytest.mark.parquet
+@pytest.mark.parametrize("fragment_readahead", [0, 1])
+def test_scanner_no_fragment_readahead(dataset, fragment_readahead):
+    # GH-48111: this used to hang with fragment_readahead=0
+    table = dataset.scanner(fragment_readahead=fragment_readahead).to_table()
+    assert table.equals(dataset.to_table())
+
+
 @pytest.mark.parquet
 def test_scanner(dataset, dataset_reader):
     scanner = dataset_reader.scanner(