| 50 | } |
| 51 | |
| 52 | static void TableJoinOverhead(benchmark::State& state, |
| 53 | TableGenerationProperties left_table_properties, |
| 54 | TableGenerationProperties right_table_properties, |
| 55 | int batch_size, int num_right_tables, |
| 56 | std::string factory_name, |
| 57 | std::shared_ptr<ExecNodeOptions> options) { |
| 58 | left_table_properties.column_prefix = "lt"; |
| 59 | left_table_properties.seed = 0; |
| 60 | ASSERT_OK_AND_ASSIGN(TableStats left_table_stats, MakeTable(left_table_properties)); |
| 61 | |
| 62 | size_t right_hand_rows = 0; |
| 63 | size_t right_hand_bytes = 0; |
| 64 | std::vector<TableStats> right_input_tables; |
| 65 | right_input_tables.reserve(num_right_tables); |
| 66 | |
| 67 | for (int i = 0; i < num_right_tables; i++) { |
| 68 | right_table_properties.column_prefix = "rt" + std::to_string(i); |
| 69 | right_table_properties.seed = i + 1; |
| 70 | ASSERT_OK_AND_ASSIGN(TableStats right_table_stats, MakeTable(right_table_properties)); |
| 71 | right_hand_rows += right_table_stats.rows; |
| 72 | right_hand_bytes += right_table_stats.bytes; |
| 73 | right_input_tables.push_back(std::move(right_table_stats)); |
| 74 | } |
| 75 | |
| 76 | for (auto _ : state) { |
| 77 | state.PauseTiming(); |
| 78 | std::vector<Declaration::Input> input_nodes = {Declaration( |
| 79 | "table_source", |
| 80 | arrow::acero::TableSourceNodeOptions(left_table_stats.table, batch_size))}; |
| 81 | input_nodes.reserve(right_input_tables.size() + 1); |
| 82 | for (TableStats table_stats : right_input_tables) { |
| 83 | input_nodes.push_back(Declaration( |
| 84 | "table_source", |
| 85 | arrow::acero::TableSourceNodeOptions(table_stats.table, batch_size))); |
| 86 | } |
| 87 | Declaration join_node{factory_name, {input_nodes}, options}; |
| 88 | state.ResumeTiming(); |
| 89 | // asof-join must currently be run synchronously as it relies on data arriving |
| 90 | // in-order |
| 91 | ASSERT_OK(DeclarationToStatus(std::move(join_node), /*use_threads=*/false)); |
| 92 | } |
| 93 | |
| 94 | state.counters["rows_per_second"] = benchmark::Counter( |
| 95 | static_cast<double>(state.iterations() * (left_table_stats.rows + right_hand_rows)), |
| 96 | benchmark::Counter::kIsRate); |
| 97 | |
| 98 | state.counters["bytes_per_second"] = |
| 99 | benchmark::Counter(static_cast<double>(state.iterations() * |
| 100 | (left_table_stats.bytes + right_hand_bytes)), |
| 101 | benchmark::Counter::kIsRate); |
| 102 | |
| 103 | state.counters["maximum_peak_memory"] = |
| 104 | benchmark::Counter(static_cast<double>(default_memory_pool()->max_memory())); |
| 105 | } |
| 106 | |
| 107 | AsofJoinNodeOptions GetRepeatedOptions(size_t repeat, FieldRef on_key, |
| 108 | std::vector<FieldRef> by_key, int64_t tolerance) { |
no test coverage detected