Function MakeOrderedSinkNode

cpp/src/arrow/dataset/scanner.cc:1118–1196

}

Result<acero::ExecNode*> MakeOrderedSinkNode(acero::ExecPlan* plan,
                                             std::vector<acero::ExecNode*> inputs,
                                             const acero::ExecNodeOptions& options) {
  if (inputs.size() != 1) {
    return Status::Invalid("Ordered SinkNode requires exactly 1 input, got ",
                           inputs.size());
  }
  auto input = inputs[0];

  AsyncGenerator<std::optional<compute::ExecBatch>> unordered;
  ARROW_ASSIGN_OR_RAISE(auto node,
                        acero::MakeExecNode("sink", plan, std::move(inputs),
                                            acero::SinkNodeOptions{&unordered}));

  const Schema& schema = *input->output_schema();
  ARROW_ASSIGN_OR_RAISE(FieldPath match, FieldRef("__fragment_index").FindOne(schema));
  int i = match[0];
  auto fragment_index = [i](const compute::ExecBatch& batch) {
    return batch.values[i].scalar_as<Int32Scalar>().value;
  };
  compute::ExecBatch before_any{{}, 0};
  before_any.values.resize(i + 1);
  before_any.values.back() = Datum(-1);

  ARROW_ASSIGN_OR_RAISE(match, FieldRef("__batch_index").FindOne(schema));
  i = match[0];
  auto batch_index = [i](const compute::ExecBatch& batch) {
    return batch.values[i].scalar_as<Int32Scalar>().value;
  };

  ARROW_ASSIGN_OR_RAISE(match, FieldRef("__last_in_fragment").FindOne(schema));
  i = match[0];
  auto last_in_fragment = [i](const compute::ExecBatch& batch) {
    return batch.values[i].scalar_as<BooleanScalar>().value;
  };

  auto is_before_any = [=](const compute::ExecBatch& batch) {
    return fragment_index(batch) < 0;
  };

  auto left_after_right = [=](const std::optional<compute::ExecBatch>& left,
                              const std::optional<compute::ExecBatch>& right) {
    // Before any comes first
    if (is_before_any(*left)) {
      return false;
    }
    if (is_before_any(*right)) {
      return true;
    }
    // Compare batches if fragment is the same
    if (fragment_index(*left) == fragment_index(*right)) {
      return batch_index(*left) > batch_index(*right);
    }
    // Otherwise compare fragment
    return fragment_index(*left) > fragment_index(*right);
  };

  auto is_next = [=](const std::optional<compute::ExecBatch>& prev,