RecordBatchReader allow implicit ordering

Author: Rafał Hibner

cpp/src/arrow/acero/options.h

@@ -198,8 +198,11 @@ class ARROW_ACERO_EXPORT RecordBatchReaderSourceNodeOptions : public ExecNodeOpt
  public:
   /// Create an instance from values
   RecordBatchReaderSourceNodeOptions(std::shared_ptr<RecordBatchReader> reader,
-                                     arrow::internal::Executor* io_executor = NULLPTR)
-      : reader(std::move(reader)), io_executor(io_executor) {}
+                                     arrow::internal::Executor* io_executor = NULLPTR,
+                                     bool implicit_ordering = false)
+      : reader(std::move(reader)),
+        io_executor(io_executor),
+        implicit_ordering(implicit_ordering) {}
 
   /// \brief The RecordBatchReader which acts as the data source
   std::shared_ptr<RecordBatchReader> reader;
@@ -208,6 +211,8 @@ class ARROW_ACERO_EXPORT RecordBatchReaderSourceNodeOptions : public ExecNodeOpt
   ///
   /// Defaults to the default I/O executor.
   arrow::internal::Executor* io_executor;
+
+  bool implicit_ordering{false};
 };
 
 /// a source node that reads from an iterator of array vectors

cpp/src/arrow/acero/source_node.cc

@@ -407,15 +407,18 @@ 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) {}
+                              arrow::AsyncGenerator<std::optional<ExecBatch>> generator,
+                              Ordering ordering = Ordering::Unordered())
+      : SourceNode(plan, schema, generator, ordering) {}
 
   static Result<ExecNode*> Make(ExecPlan* plan, std::vector<ExecNode*> inputs,
                                 const ExecNodeOptions& options) {
     RETURN_NOT_OK(ValidateExecNodeInputs(plan, inputs, 0, kKindName));
     const auto& cast_options =
         checked_cast<const RecordBatchReaderSourceNodeOptions&>(options);
     auto& reader = cast_options.reader;
+    Ordering ordering =
+        cast_options.implicit_ordering ? Ordering::Implicit() : Ordering::Unordered();
     auto io_executor = cast_options.io_executor;
 
     if (reader == nullptr) {
@@ -428,7 +431,7 @@ struct RecordBatchReaderSourceNode : public SourceNode {
 
     ARROW_ASSIGN_OR_RAISE(auto generator, MakeGenerator(reader, io_executor));
     return plan->EmplaceNode<RecordBatchReaderSourceNode>(plan, reader->schema(),
-                                                          generator);
+                                                          generator, ordering);
   }
 
   static Result<arrow::AsyncGenerator<std::optional<ExecBatch>>> MakeGenerator(