diff --git a/cpp/src/arrow/dataset/scanner.cc b/cpp/src/arrow/dataset/scanner.cc index 222e1323d4a..ace3e324793 100644 --- a/cpp/src/arrow/dataset/scanner.cc +++ b/cpp/src/arrow/dataset/scanner.cc @@ -1030,14 +1030,14 @@ Result MakeScanNode(acero::ExecPlan* plan, FragmentsToBatches(std::move(fragment_gen), scan_options)); AsyncGenerator 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 a45847b49de..644f8507d86 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 9ddfdfdc9c4..eada682fc5b 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(