Skip to content

Commit 7c9d880

Browse files
GH-47481: [C++][Acero] record_batch_reader_source specify Ordering::Implicit to support select * limit k (#47482)
### Rationale for this change record_batch_reader_source does not support `select * limit k` as described in #47481 ### What changes are included in this PR? record_batch_reader_source specify Ordering::Implicit ### Are these changes tested? yes ### Are there any user-facing changes? no * GitHub Issue: #47481 Lead-authored-by: egolearner <lijiliang@outlook.com> Co-authored-by: egolearner <45122959+egolearner@users.noreply.github.com> Co-authored-by: Rossi Sun <zanmato1984@gmail.com> Signed-off-by: Rossi Sun <zanmato1984@gmail.com>
1 parent d40c02e commit 7c9d880

2 files changed

Lines changed: 23 additions & 1 deletion

File tree

cpp/src/arrow/acero/fetch_node_test.cc

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,28 @@ TEST(FetchNode, Basic) {
8080
CheckFetch({0, 0});
8181
}
8282

83+
TEST(FetchNode, RecordBatchReaderSource) {
84+
constexpr random::SeedType kSeed = 42;
85+
constexpr int kJitterMod = 4;
86+
FetchNodeOptions options{20, 50};
87+
RegisterTestNodes();
88+
std::shared_ptr<Table> input = TestTable();
89+
for (bool use_threads : {false, true}) {
90+
SCOPED_TRACE(use_threads ? "threaded" : "serial");
91+
auto reader = std::make_shared<TableBatchReader>(input);
92+
Declaration plan = Declaration::Sequence(
93+
{{"record_batch_reader_source", RecordBatchReaderSourceNodeOptions{reader}},
94+
{"jitter", JitterNodeOptions(kSeed, kJitterMod)},
95+
{"fetch", options}});
96+
QueryOptions query_options;
97+
query_options.use_threads = use_threads;
98+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<Table> actual,
99+
DeclarationToTable(plan, query_options));
100+
std::shared_ptr<Table> expected = input->Slice(options.offset, options.count);
101+
AssertTablesEqual(*expected, *actual);
102+
}
103+
}
104+
83105
TEST(FetchNode, Invalid) {
84106
CheckFetchInvalid({-1, 10}, "`offset` must be non-negative");
85107
CheckFetchInvalid({10, -1}, "`count` must be non-negative");

cpp/src/arrow/acero/source_node.cc

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -408,7 +408,7 @@ struct SchemaSourceNode : public SourceNode {
408408
struct RecordBatchReaderSourceNode : public SourceNode {
409409
RecordBatchReaderSourceNode(ExecPlan* plan, std::shared_ptr<Schema> schema,
410410
arrow::AsyncGenerator<std::optional<ExecBatch>> generator)
411-
: SourceNode(plan, schema, generator) {}
411+
: SourceNode(plan, schema, generator, Ordering::Implicit()) {}
412412

413413
static Result<ExecNode*> Make(ExecPlan* plan, std::vector<ExecNode*> inputs,
414414
const ExecNodeOptions& options) {

0 commit comments

Comments
 (0)