Function TestSchemaResolution

cpp/src/arrow/acero/asof_join_node_test.cc:1594–1651

template <typename BatchesMaker>
void TestSchemaResolution(BatchesMaker maker, int num_batches, int batch_size) {
  // GH-39803: The key hasher needs to resolve the types of key columns. All other
  // tests use int32 for all columns, but this test converts the key columns to
  // strings via a projection node to test that the column is correctly resolved
  // to string.
  auto l_schema =
      schema({field("time", int32()), field("key", int32()), field("l_value", int32())});
  auto r_schema =
      schema({field("time", int32()), field("key", int32()), field("r0_value", int32())});

  auto make_shift = [&maker, num_batches, batch_size](
                        const std::shared_ptr<Schema>& schema, int shift) {
    return maker({[](int row) -> int64_t { return row; },
                  [num_batches](int row) -> int64_t { return row / num_batches; },
                  [shift](int row) -> int64_t { return row * 10 + shift; }},
                 schema, num_batches, batch_size);
  };
  ASSERT_OK_AND_ASSIGN(auto l_batches, make_shift(l_schema, 0));
  ASSERT_OK_AND_ASSIGN(auto r_batches, make_shift(r_schema, 1));

  Declaration l_src = {"source",
                       SourceNodeOptions(l_schema, l_batches.gen(false, false))};
  Declaration r_src = {"source",
                       SourceNodeOptions(r_schema, r_batches.gen(false, false))};
  Declaration l_project = {
      "project",
      {std::move(l_src)},
      ProjectNodeOptions({compute::field_ref("time"),
                          compute::call("cast", {compute::field_ref("key")},
                                        compute::CastOptions::Safe(utf8())),
                          compute::field_ref("l_value")},
                         {"time", "key", "l_value"})};
  Declaration r_project = {
      "project",
      {std::move(r_src)},
      ProjectNodeOptions({compute::call("cast", {compute::field_ref("key")},
                                        compute::CastOptions::Safe(utf8())),
                          compute::field_ref("r0_value"), compute::field_ref("time")},
                         {"key", "r0_value", "time"})};

  Declaration asofjoin = {
      "asofjoin", {l_project, r_project}, GetRepeatedOptions(2, "time", {"key"}, 1000)};

  QueryOptions query_options;
  query_options.use_threads = false;
  ASSERT_OK_AND_ASSIGN(auto table, DeclarationToTable(asofjoin, query_options));

  Int32Builder expected_r0_b;
  for (int i = 1; i <= 91; i += 10) {
    ASSERT_OK(expected_r0_b.Append(i));
  }
  ASSERT_OK_AND_ASSIGN(auto expected_r0, expected_r0_b.Finish());

  auto actual_r0 = table->GetColumnByName("r0_value");
  std::vector<std::shared_ptr<arrow::Array>> chunks = {expected_r0};
  auto expected_r0_chunked = std::make_shared<arrow::ChunkedArray>(chunks);
  ASSERT_TRUE(actual_r0->Equals(expected_r0_chunked));
}