diff --git a/cpp/src/arrow/acero/fetch_node_test.cc b/cpp/src/arrow/acero/fetch_node_test.cc
index 3ffd48038d23..e1e49354fd8e 100644
--- a/cpp/src/arrow/acero/fetch_node_test.cc
+++ b/cpp/src/arrow/acero/fetch_node_test.cc
@@ -80,6 +80,28 @@ TEST(FetchNode, Basic) {
CheckFetch({0, 0});
}
+TEST(FetchNode, RecordBatchReaderSource) {
+ constexpr random::SeedType kSeed = 42;
+ constexpr int kJitterMod = 4;
+ FetchNodeOptions options{20, 50};
+ RegisterTestNodes();
+ std::shared_ptr
input = TestTable();
+ for (bool use_threads : {false, true}) {
+ SCOPED_TRACE(use_threads ? "threaded" : "serial");
+ auto reader = std::make_shared(input);
+ Declaration plan = Declaration::Sequence(
+ {{"record_batch_reader_source", RecordBatchReaderSourceNodeOptions{reader}},
+ {"jitter", JitterNodeOptions(kSeed, kJitterMod)},
+ {"fetch", options}});
+ QueryOptions query_options;
+ query_options.use_threads = use_threads;
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr actual,
+ DeclarationToTable(plan, query_options));
+ std::shared_ptr expected = input->Slice(options.offset, options.count);
+ AssertTablesEqual(*expected, *actual);
+ }
+}
+
TEST(FetchNode, Invalid) {
CheckFetchInvalid({-1, 10}, "`offset` must be non-negative");
CheckFetchInvalid({10, -1}, "`count` must be non-negative");
diff --git a/cpp/src/arrow/acero/source_node.cc b/cpp/src/arrow/acero/source_node.cc
index 888f6e23c130..ed3723d249da 100644
--- a/cpp/src/arrow/acero/source_node.cc
+++ b/cpp/src/arrow/acero/source_node.cc
@@ -408,7 +408,7 @@ struct SchemaSourceNode : public SourceNode {
struct RecordBatchReaderSourceNode : public SourceNode {
RecordBatchReaderSourceNode(ExecPlan* plan, std::shared_ptr schema,
arrow::AsyncGenerator> generator)
- : SourceNode(plan, schema, generator) {}
+ : SourceNode(plan, schema, generator, Ordering::Implicit()) {}
static Result Make(ExecPlan* plan, std::vector inputs,
const ExecNodeOptions& options) {