GH-34698: [C++][Acero] Add node that emits explicit ordering after asserting order

[EnricoMi]

Rationale for this change

An acero node that turns an implicit ordering into an explicit ordering (rows sorted by some columns) is useful to re-use order that already exists in the data.

What changes are included in this PR?

This PR adds the AssertOrderNode that implements this logic. The Scanner employs that node to turn the implicit ordering of the ScanNode into an explicit order as defined by user code via ScanBuilder.Ordering.

Are these changes tested?

There are unit tests for the AssertOrderNode as well as for the ScanNode and ScanBuilder.

Are there any user-facing changes?

The following options has been added:

• the ordering option added to ScanOptions
• the Ordering option added to ScannerBuilder
• the AssertNodeOptions class

• GitHub Issue: [C++] Allow scanner to assert an ordering and/or support implicit ordering #34698

[github-actions]

Thanks for opening a pull request!

If this is not a minor PR. Could you open an issue for this pull request on GitHub? https://github.com/apache/arrow/issues/new/choose

Opening GitHub issues ahead of time contributes to the Openness of the Apache Arrow project.

Then could you also rename the pull request title in the following format?

GH-${GITHUB_ISSUE_ID}: [${COMPONENT}] ${SUMMARY}

or

MINOR: [${COMPONENT}] ${SUMMARY}

In the case of PARQUET issues on JIRA the title also supports:

PARQUET-${JIRA_ISSUE_ID}: [${COMPONENT}] ${SUMMARY}

See also:

• Other pull requests
• Contribution Guidelines - Contributing Overview

[github-actions]

⚠️ GitHub issue #34698 has been automatically assigned in GitHub to PR creator.

[raulcd]

I am probably not qualified to review this thoroughly but the CI failure is related:

[  FAILED  ] TestScannerThreading/TestScanner.ScanOrderingFail/0Serial1d1b1024r, where GetParam() = serial-1d-1b-1024i (4 ms)
[ RUN      ] TestScannerThreading/TestScanner.ScanOrderingFail/1Serial2d16b1024r
/arrow/cpp/src/arrow/dataset/scanner_test.cc:1040: Failure
Expected equality of these values:
  next.status().message()
    Which is: "Data is not ordered\n/arrow/cpp/src/arrow/acero/assert_order_node.cc:190  AssertInBatchOrder(*record_batch, ordering_, comparators_)\n/arrow/cpp/src/arrow/acero/source_node.cc:158  output_->InputReceived(this, std::move(batch))"
  "Data is not ordered"
With diff:
@@ -1,3 @@
 Data is not ordered
-/arrow/cpp/src/arrow/acero/assert_order_node.cc:190  AssertInBatchOrder(*record_batch, ordering_, comparators_)
-/arrow/cpp/src/arrow/acero/source_node.cc:158  output_->InputReceived(this, std::move(batch))

There seems to be some stray output on that specific job.

[EnricoMi]

@raulcd thanks, assertions adjusted

[westonpace]

Thanks! Conceptually this makes sense to me. I like that it verifies the ordering and it's good use of a sequencing queue. I can't speak much to the code in compare.h / compare.cc as I'm not very familiar with it.

It's a brand new node and I don't see much change to existing paths so I'm not too concerned about a regression.

[westonpace]

Is this change related to the PR?

[EnricoMi]

reverted

[westonpace]

Should you use !ordering.is_explicit()?

[EnricoMi]

right, fixed

[EnricoMi]

@westonpace saying: "I can't speak much to the code in compare.h / compare.cc as I'm not very familiar with it."

@pitrou @lidavidm @bkietz @zeroshade you seem to be familiar with compare.cc, can you take a look, please?