From cf69c13d63a1c67e8589a6a2d29b28ffbca9d3a6 Mon Sep 17 00:00:00 2001
From: egolearner <45122959+egolearner@users.noreply.github.com>
Date: Wed, 3 Sep 2025 16:32:55 +0800
Subject: [PATCH 1/2] RecordBatchReaderSourceNode use Ordering::Implicit

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

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> schema,
                               arrow::AsyncGenerator<std::optional<ExecBatch>> generator)
-      : SourceNode(plan, schema, generator) {}
+      : SourceNode(plan, schema, generator, Ordering::Implicit()) {}
 
   static Result<ExecNode*> Make(ExecPlan* plan, std::vector<ExecNode*> inputs,
                                 const ExecNodeOptions& options) {

From a2cbafd633e2b55312c49ec7a49430bfb00a5c2c Mon Sep 17 00:00:00 2001
From: egolearner <lijiliang@outlook.com>
Date: Fri, 22 May 2026 21:15:10 +0800
Subject: [PATCH 2/2] add ut

---
 cpp/src/arrow/acero/fetch_node_test.cc | 22 ++++++++++++++++++++++
 1 file changed, 22 insertions(+)

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<Table> input = TestTable();
+  for (bool use_threads : {false, true}) {
+    SCOPED_TRACE(use_threads ? "threaded" : "serial");
+    auto reader = std::make_shared<TableBatchReader>(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<Table> actual,
+                         DeclarationToTable(plan, query_options));
+    std::shared_ptr<Table> 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");