| 84 | } |
| 85 | |
| 86 | void CheckRunOutput(JoinType type, const BatchesWithSchema& l_batches, |
| 87 | const BatchesWithSchema& r_batches, |
| 88 | const std::vector<FieldRef>& left_keys, |
| 89 | const std::vector<FieldRef>& right_keys, |
| 90 | const BatchesWithSchema& exp_batches, bool parallel = false) { |
| 91 | Declaration left{"source", |
| 92 | SourceNodeOptions{l_batches.schema, l_batches.gen(parallel, |
| 93 | /*slow=*/false)}}; |
| 94 | Declaration right{"source", |
| 95 | SourceNodeOptions{r_batches.schema, r_batches.gen(parallel, |
| 96 | /*slow=*/false)}}; |
| 97 | HashJoinNodeOptions join_options{type, left_keys, right_keys}; |
| 98 | Declaration join{"hashjoin", {std::move(left), std::move(right)}, join_options}; |
| 99 | |
| 100 | ASSERT_OK_AND_ASSIGN(auto out_table, DeclarationToTable(std::move(join), parallel)); |
| 101 | |
| 102 | ASSERT_OK_AND_ASSIGN(auto exp_table, |
| 103 | TableFromExecBatches(exp_batches.schema, exp_batches.batches)); |
| 104 | |
| 105 | if (exp_table->num_rows() == 0) { |
| 106 | ASSERT_EQ(exp_table->num_rows(), out_table->num_rows()); |
| 107 | } else { |
| 108 | std::vector<SortKey> sort_keys; |
| 109 | for (auto&& f : exp_batches.schema->fields()) { |
| 110 | sort_keys.emplace_back(f->name()); |
| 111 | } |
| 112 | ASSERT_OK_AND_ASSIGN(auto exp_table_sort_ids, |
| 113 | SortIndices(exp_table, SortOptions(sort_keys))); |
| 114 | ASSERT_OK_AND_ASSIGN(auto exp_table_sorted, Take(exp_table, exp_table_sort_ids)); |
| 115 | ASSERT_OK_AND_ASSIGN(auto out_table_sort_ids, |
| 116 | SortIndices(out_table, SortOptions(sort_keys))); |
| 117 | ASSERT_OK_AND_ASSIGN(auto out_table_sorted, Take(out_table, out_table_sort_ids)); |
| 118 | |
| 119 | AssertTablesEqual(*exp_table_sorted.table(), *out_table_sorted.table(), |
| 120 | /*same_chunk_layout=*/false, /*flatten=*/true); |
| 121 | } |
| 122 | } |
| 123 | |
| 124 | void RunNonEmptyTest(JoinType type, bool parallel) { |
| 125 | auto l_schema = schema({field("l_i32", int32()), field("l_str", utf8())}); |
no test coverage detected