• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

llnl / dftracer-utils / 29185833209

12 Jul 2026 08:25AM UTC coverage: 51.235% (-1.5%) from 52.754%
29185833209

Pull #96

github

web-flow
Merge 0b9f07110 into 056c79287
Pull Request #96: fix: consolidate stale-index rebuilds across consumers and fix multi-run breaks

33724 of 84486 branches covered (39.92%)

Branch coverage included in aggregate %.

140 of 147 new or added lines in 15 files covered. (95.24%)

5188 existing lines in 197 files now uncovered.

34547 of 48765 relevant lines covered (70.84%)

10791.85 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

55.54
/src/dftracer/utils/python/utilities/comparator.cpp
1
#define PY_SSIZE_T_CLEAN
2
#include <dftracer/utils/core/common/error.h>
3
#include <dftracer/utils/core/common/filesystem.h>
4
#include <dftracer/utils/core/common/platform_compat.h>
5
#include <dftracer/utils/core/coro/channel.h>
6
#include <dftracer/utils/core/coro/task.h>
7
#include <dftracer/utils/core/coro/when_all.h>
8
#include <dftracer/utils/core/pipeline/pipeline.h>
9
#include <dftracer/utils/core/pipeline/pipeline_config.h>
10
#include <dftracer/utils/core/runtime.h>
11
#include <dftracer/utils/core/tasks/coro_scope.h>
12
#include <dftracer/utils/core/tasks/task.h>
13
#include <dftracer/utils/python/arrow_helpers.h>
14
#include <dftracer/utils/python/py_runtime_mixin.h>
15
#include <dftracer/utils/python/py_type_helpers.h>
16
#include <dftracer/utils/python/runtime.h>
17
#include <dftracer/utils/python/utilities/comparator.h>
18
#include <dftracer/utils/utilities/common/query/query.h>
19
#include <dftracer/utils/utilities/composites/dft/aggregators/aggregators.h>
20
#include <dftracer/utils/utilities/composites/dft/comparator/comparison_config.h>
21
#include <dftracer/utils/utilities/composites/dft/comparator/comparison_result.h>
22
#include <dftracer/utils/utilities/composites/dft/comparator/comparison_utility.h>
23
#include <dftracer/utils/utilities/composites/dft/comparator/tree_table_formatter.h>
24
#include <dftracer/utils/utilities/composites/dft/indexing/index_resolver_utility.h>
25
#include <dftracer/utils/utilities/composites/dft/indexing/resolve_and_build.h>
26
#include <dftracer/utils/utilities/composites/dft/internal/utils.h>
27
#include <dftracer/utils/utilities/composites/dft/metadata_collector_utility.h>
28
#include <dftracer/utils/utilities/filesystem/pattern_directory_scanner_utility.h>
29
#include <dftracer/utils/utilities/indexer/index_builder_utility.h>
30
#include <dftracer/utils/utilities/indexer/internal/indexer.h>
31

32
#include <atomic>
33
#include <chrono>
34
#include <cstdio>
35
#include <ctime>
36
#include <stdexcept>
37
#include <string>
38
#include <vector>
39

40
using dftracer::utils::Runtime;
41
using dftracer::utils::coro::CoroTask;
42
using namespace dftracer::utils;
43
using namespace dftracer::utils::utilities;
44
using namespace dftracer::utils::utilities::composites::dft::aggregators;
45
using namespace dftracer::utils::utilities::composites::dft::comparator;
46

47
#include <dftracer/utils/core/common/config.h>
48
#ifdef DFTRACER_UTILS_ENABLE_ARROW
49
using dftracer::utils::python::arrow_result_to_table;
50
#endif
51

52
DFTRACER_UTILS_RUNTIME_BACKED_SLOTS(Comparator, ComparatorObject)
39✔
53

54
// -----------------------------------------------------------------------
55
// Helpers
56
// -----------------------------------------------------------------------
57

58
struct ComparatorArgs {
39!
59
    std::string baseline;
60
    std::string variant;
61
    std::string query;
62
    std::string group_by;
63
    std::string format;
64
    double time_interval_ms = 5000.0;
13✔
65
    double threshold = 0.0;
13✔
66
    std::size_t executor_threads = 0;
13✔
67
    std::string baseline_index_dir;
68
    std::string variant_index_dir;
69
    bool force_rebuild = false;
13✔
70
    std::string config_path;
71
};
72

73
static int parse_comparator_args(PyObject *args, PyObject *kwds,
13✔
74
                                 ComparatorArgs &out) {
75
    static const char *kwlist[] = {"baseline",
76
                                   "variant",
77
                                   "query",
78
                                   "group_by",
79
                                   "format",
80
                                   "time_interval_ms",
81
                                   "threshold",
82
                                   "executor_threads",
83
                                   "baseline_index_dir",
84
                                   "variant_index_dir",
85
                                   "force_rebuild",
86
                                   "config",
87
                                   NULL};
88

89
    const char *baseline = NULL;
13✔
90
    const char *variant = NULL;
13✔
91
    const char *query = "";
13✔
92
    const char *group_by = "";
13✔
93
    const char *format = "table";
13✔
94
    double time_interval_ms = 5000.0;
13✔
95
    double threshold = 0.0;
13✔
96
    Py_ssize_t executor_threads = 0;
13✔
97
    const char *baseline_index_dir = "";
13✔
98
    const char *variant_index_dir = "";
13✔
99
    int force_rebuild = 0;
13✔
100
    const char *config = "";
13✔
101

102
    if (!PyArg_ParseTupleAndKeywords(
13!
103
            args, kwds, "ss|sssddnssps", (char **)kwlist, &baseline, &variant,
13✔
104
            &query, &group_by, &format, &time_interval_ms, &threshold,
105
            &executor_threads, &baseline_index_dir, &variant_index_dir,
106
            &force_rebuild, &config))
107
        return -1;
×
108

109
    out.baseline = baseline;
13✔
110
    out.variant = variant;
13✔
111
    out.query = query;
13✔
112
    out.group_by = group_by;
13✔
113
    out.format = format;
13✔
114
    out.time_interval_ms = time_interval_ms;
13✔
115
    out.threshold = threshold;
13✔
116
    out.executor_threads = static_cast<std::size_t>(executor_threads);
13✔
117
    out.baseline_index_dir = baseline_index_dir;
13✔
118
    out.variant_index_dir = variant_index_dir;
13✔
119
    out.force_rebuild = force_rebuild != 0;
13✔
120
    out.config_path = config;
13✔
121

122
    return 0;
13✔
123
}
13✔
124

125
namespace {
126

127
void flatten_nodes(const ComparisonNode &node,
13✔
128
                   std::vector<const ComparisonNode *> &out) {
129
    out.push_back(&node);
13✔
130
    for (const auto &child : node.children) {
13!
131
        flatten_nodes(child, out);
×
132
    }
133
}
13✔
134

135
CoroTask<EventAggregatorOutput> run_aggregation(
130!
136
    std::vector<std::string> input_files, AggregationConfig agg_config,
137
    std::optional<common::query::Query> query, std::string index_dir,
138
    std::size_t checkpoint_size, bool force_rebuild,
139
    std::size_t executor_threads) {
26!
140
    constexpr std::size_t CHUNK_SIZE_MB = 4;
26✔
141
    constexpr std::size_t BATCH_SIZE_MB = 4;
26✔
142

143
    auto pipeline_config = PipelineConfig()
52!
144
                               .with_name("DFTracer Comparator Aggregation")
26!
145
                               .with_compute_threads(executor_threads)
26!
146
                               .with_watchdog(false);
26!
147
    Pipeline pipeline(pipeline_config);
26!
148

149
    EventAggregator merger;
26!
150
    std::atomic<int> global_chunk_idx{0};
26✔
151

152
    auto streaming_task = make_task(
52!
153
        [&](CoroScope &ctx) -> CoroTask<void> {
208!
154
            auto chunk_chan = coro::make_channel<ChunkAggregatorInput>(0);
78!
155
            auto result_chan = coro::make_channel<ChunkAggregationOutput>(2);
78!
156

157
            co_await ctx.scope([&](CoroScope &scope) -> CoroTask<void> {
260!
158
                for (const auto &file_path : input_files) {
54✔
159
                    auto *global_chunk_idx_ptr = &global_chunk_idx;
28✔
160
                    scope.spawn([file_path, ch = chunk_chan->producer(),
446!
161
                                 index_dir, checkpoint_size, force_rebuild,
28!
162
                                 agg_config, query, global_chunk_idx_ptr](
28!
163
                                    CoroScope & /*fctx*/) mutable
164
                                    -> CoroTask<void> {
28!
165
                        [[maybe_unused]] auto producer_guard = ch.guard();
82!
166

167
                        std::string index_path =
82✔
168
                            composites::dft::internal::determine_index_path(
82!
169
                                file_path, index_dir);
82✔
170

171
                        auto meta_input =
82✔
172
                            composites::dft::MetadataCollectorUtilityInput::
164!
173
                                from_file(file_path)
82!
174
                                    .with_checkpoint_size(checkpoint_size)
82✔
175
                                    .with_force_rebuild(force_rebuild)
28!
176
                                    .with_index(index_path);
28✔
177
                        auto metadata =
82✔
178
                            co_await composites::dft::MetadataCollectorUtility{}
219!
179
                                .process(meta_input);
82!
180

181
                        if (!metadata.success) {
84!
UNCOV
182
                            co_return;
×
183
                        }
184

185
                        FileChunkMapperUtility file_mapper;
84!
186
                        auto mapper_input =
84✔
187
                            FileChunkMapperInput::from_metadata(metadata)
168!
188
                                .with_config(agg_config)
84✔
189
                                .with_checkpoint_size(checkpoint_size)
28!
190
                                .with_target_chunk_size(CHUNK_SIZE_MB)
28!
191
                                .with_batch_size(BATCH_SIZE_MB * 1024 * 1024);
28!
192
                        mapper_input.query = query;
84!
193
                        auto file_chunks =
84✔
194
                            co_await file_mapper.process(mapper_input);
112!
195

196
                        int start_idx = global_chunk_idx_ptr->fetch_add(
168✔
197
                            static_cast<int>(file_chunks.size()));
84✔
198
                        for (int i = 0;
112✔
199
                             i < static_cast<int>(file_chunks.size()); ++i) {
112✔
200
                            file_chunks[i].chunk_index = start_idx + i;
28✔
201
                        }
28✔
202

203
                        for (auto &chunk : file_chunks) {
112!
204
                            if (!co_await ch.send(std::move(chunk))) {
112!
UNCOV
205
                                co_return;
×
206
                            }
207
                        }
28!
208
                        co_return;
28✔
209
                    });
690✔
210
                }
28✔
211

212
                for (std::size_t w = 0; w < executor_threads; ++w) {
104✔
213
                    (void)w;
214
                    scope.spawn([chunk_chan, rp = result_chan->producer(),
676!
215
                                 result_chan](
78✔
216
                                    CoroScope &wctx) mutable -> CoroTask<void> {
79✔
217
                        [[maybe_unused]] auto producer_guard = rp.guard();
72!
218
                        while (auto input = co_await wctx.receive(chunk_chan)) {
501!
219
                            ChunkAggregatorUtility agg;
84!
220
                            auto output = co_await agg.process(*input);
112!
221
                            if (!co_await result_chan->send(
112!
222
                                    std::move(output))) {
UNCOV
223
                                co_return;
×
224
                            }
225
                        }
216✔
226
                        co_return;
77✔
227
                    });
770✔
228
                }
78✔
229

230
                auto *merger_ptr = &merger;
26✔
231
                scope.spawn([result_chan,
334!
232
                             merger_ptr](CoroScope &mctx) -> CoroTask<void> {
52!
233
                    while (auto output = co_await mctx.receive(result_chan)) {
194!
234
                        merger_ptr->merge_chunk(std::move(*output));
28!
235
                    }
54✔
236
                    co_return;
26✔
237
                });
152✔
238

239
                co_return;
52✔
240
            });
26✔
241

242
            co_return;
26✔
243
        },
78!
244
        "StreamingAggregate");
26!
245

246
    EventAggregatorOutput result;
26✔
247
    auto post_task = make_task(
52!
248
        [&](CoroScope & /*ctx*/) -> CoroTask<bool> {
156!
249
            result = merger.finalize();
26!
250
            co_return result.success;
52!
UNCOV
251
        },
×
252
        "Finalize");
26!
253

254
    post_task->depends_on(streaming_task);
26!
255
    pipeline.set_source(streaming_task);
26!
256
    pipeline.set_destination(post_task);
26!
257
    pipeline.execute();
26!
258

259
    co_return result;
52!
260
}
26✔
261

262
}  // namespace
263

264
static bool run_comparison_pipeline(ComparatorObject *self,
13✔
265
                                    const ComparatorArgs &cargs,
266
                                    ComparisonOutput &output) {
267
    ComparatorArgs args_copy = cargs;
13✔
268
    auto *output_ptr = &output;
13✔
269

270
    return run_blocking([&] {
26!
271
        ComparisonConfig config;
13✔
272
        if (!args_copy.config_path.empty()) {
13!
273
            std::string parse_error;
×
UNCOV
274
            auto parsed = ComparisonConfig::from_json_file(
×
275
                args_copy.config_path, parse_error);
×
276
            if (!parsed) {
×
277
                throw DFTUtilsException(ErrorCode::PARSE,
×
278
                                        "Config error: " + parse_error);
×
279
            }
280
            config = std::move(*parsed);
×
281
        } else {
×
282
            config = ComparisonConfig::from_cli(
13!
283
                args_copy.baseline, args_copy.variant, args_copy.query,
13✔
284
                args_copy.group_by);
13✔
285
        }
286

287
        config.format = args_copy.format;
13!
288
        config.no_color = true;
13✔
289
        if (args_copy.executor_threads > 0)
13!
290
            config.executor_threads = args_copy.executor_threads;
×
291
        if (!args_copy.baseline_index_dir.empty())
13!
292
            config.baseline_index_dir = args_copy.baseline_index_dir;
×
293
        if (!args_copy.variant_index_dir.empty())
13!
294
            config.variant_index_dir = args_copy.variant_index_dir;
×
295
        if (args_copy.force_rebuild)
13!
296
            config.force_rebuild = args_copy.force_rebuild;
×
297
        if (args_copy.threshold > 0.0)
13!
298
            config.defaults.threshold_pct = args_copy.threshold;
×
299
        if (args_copy.time_interval_ms > 0.0)
13!
300
            config.defaults.time_interval_ms = args_copy.time_interval_ms;
13✔
301

302
        config.resolve();
13!
303

304
        if (config.executor_threads == 0) {
13!
305
            config.executor_threads = dftracer_utils_hardware_concurrency();
13!
306
        }
13✔
307
        if (config.checkpoint_size == 0) {
13!
308
            config.checkpoint_size =
13✔
309
                indexer::internal::Indexer::DEFAULT_CHECKPOINT_SIZE;
310
        }
13✔
311

312
        using composites::dft::indexing::IndexResolverUtility;
313
        using composites::dft::indexing::ResolverInput;
314
        using indexer::IndexBatchBuilderUtility;
315
        using indexer::IndexBuildBatchConfig;
316

317
        Runtime *rt = resolve_runtime(self);
13!
318

319
        auto task = [config, output_ptr, rt]() -> CoroTask<void> {
156!
320
            auto resolve_and_build =
39✔
321
                [&config](
104!
322
                    CoroScope &scope, const std::string &path,
323
                    const std::string &index_dir,
324
                    std::vector<std::string> &out_files) -> CoroTask<void> {
13!
325
                std::vector<std::string> ensure_files;
13✔
326
                std::string ensure_dir;
13✔
327
                if (fs::is_regular_file(path)) {
13!
328
                    ensure_files.push_back(path);
12!
329
                } else {
12✔
330
                    ensure_dir = path;
1!
331
                }
332
                co_await composites::dft::indexing::ensure_indexes_fresh(
39!
333
                    &scope, ensure_dir, ensure_files, index_dir,
13!
334
                    config.force_rebuild);
13✔
335

336
                IndexResolverUtility resolver;
13!
337
                ResolverInput resolve_input;
13✔
338
                resolve_input.index_dir = index_dir;
13!
339
                resolve_input.require_checkpoints = !config.force_rebuild;
13✔
340
                if (fs::is_regular_file(path)) {
13!
341
                    resolve_input.files = {path};
12!
342
                } else {
12✔
343
                    resolve_input.directory = path;
1!
344
                }
345

346
                auto result = co_await resolver.process(resolve_input);
26!
347
                out_files = std::move(result.all_files);
13✔
348

349
                if (out_files.empty() || result.needs_checkpoint.empty()) {
13!
350
                    co_return;
13✔
351
                }
352

UNCOV
353
                auto batch_cfg = std::make_shared<IndexBuildBatchConfig>();
×
UNCOV
354
                batch_cfg->file_paths.reserve(result.needs_checkpoint.size());
×
UNCOV
355
                for (const auto &item : result.needs_checkpoint) {
×
UNCOV
356
                    batch_cfg->file_paths.push_back(item.file_path);
×
UNCOV
357
                }
×
UNCOV
358
                batch_cfg->index_dir = index_dir;
×
UNCOV
359
                batch_cfg->checkpoint_size = config.checkpoint_size;
×
UNCOV
360
                batch_cfg->parallelism = config.executor_threads;
×
UNCOV
361
                batch_cfg->force_rebuild = config.force_rebuild;
×
UNCOV
362
                batch_cfg->use_batch_write = true;
×
UNCOV
363
                batch_cfg->rebuild_root_summaries = true;
×
364

UNCOV
365
                co_await IndexBatchBuilderUtility::process(
×
UNCOV
366
                    &scope, std::move(batch_cfg));
×
367
            };
65!
368

369
            std::vector<std::string> baseline_files;
39✔
370
            std::vector<std::string> variant_files;
39✔
371

372
            bool shared_index =
78✔
373
                composites::dft::internal::determine_index_path(
117!
374
                    config.baseline, config.baseline_index_dir) ==
78✔
375
                composites::dft::internal::determine_index_path(
78!
376
                    config.variant, config.variant_index_dir);
39✔
377

378
            co_await run_coro_scope(
52!
379
                rt->executor(), [&](CoroScope &scope) -> CoroTask<void> {
156!
380
                    if (shared_index) {
39!
381
                        co_await resolve_and_build(scope, config.baseline,
104!
382
                                                   config.baseline_index_dir,
39✔
383
                                                   baseline_files);
39✔
384
                        if (config.baseline == config.variant) {
13!
385
                            variant_files = baseline_files;
13!
386
                        } else {
13✔
UNCOV
387
                            co_await resolve_and_build(scope, config.variant,
×
UNCOV
388
                                                       config.variant_index_dir,
×
UNCOV
389
                                                       variant_files);
×
390
                        }
391
                    } else {
13✔
392
                        scope.spawn([&](CoroScope &s) -> CoroTask<void> {
×
UNCOV
393
                            co_await resolve_and_build(
×
UNCOV
394
                                s, config.baseline, config.baseline_index_dir,
×
UNCOV
395
                                baseline_files);
×
396
                        });
×
UNCOV
397
                        co_await resolve_and_build(scope, config.variant,
×
UNCOV
398
                                                   config.variant_index_dir,
×
UNCOV
399
                                                   variant_files);
×
400
                    }
401
                });
26✔
402

403
            if (baseline_files.empty()) {
39!
UNCOV
404
                throw DFTUtilsException(
×
405
                    ErrorCode::NOT_FOUND,
UNCOV
406
                    "No trace files found in baseline: " + config.baseline);
×
407
            }
408
            if (variant_files.empty()) {
39!
UNCOV
409
                throw DFTUtilsException(
×
410
                    ErrorCode::NOT_FOUND,
UNCOV
411
                    "No trace files found in variant: " + config.variant);
×
412
            }
413

414
            output_ptr->baseline_path = config.baseline;
39✔
415
            output_ptr->variant_path = config.variant;
13✔
416

417
            auto start_time = std::chrono::high_resolution_clock::now();
39✔
418

419
            std::size_t b_files_actual = 0;
39✔
420
            std::size_t v_files_actual = 0;
39✔
421
            bool metadata_set = false;
39✔
422

423
            for (auto &node : config.nodes) {
52!
424
                std::vector<const ComparisonNode *> visitors;
39✔
425
                flatten_nodes(node, visitors);
39!
426

427
                std::vector<ComparisonVisitorPair> pairs;
39✔
428
                pairs.reserve(visitors.size());
39!
429

430
                for (const auto *visitor : visitors) {
78!
431
                    std::optional<common::query::Query> query;
39✔
432
                    if (!visitor->composed_query.empty()) {
39✔
433
                        auto result = common::query::Query::from_string(
26!
434
                            visitor->composed_query);
13✔
435
                        if (!result) {
13!
UNCOV
436
                            throw DFTUtilsException(
×
437
                                ErrorCode::QUERY,
UNCOV
438
                                "Invalid query for node '" + visitor->name +
×
UNCOV
439
                                    "': " + result.error().format());
×
440
                        }
441
                        query = std::move(*result);
13!
442
                    }
13✔
443

444
                    AggregationConfig agg_cfg;
39!
445
                    agg_cfg.time_interval_us = static_cast<std::uint64_t>(
39✔
446
                        config.defaults.time_interval_ms * 1000.0);
39✔
447
                    agg_cfg.extra_group_keys = {};
39✔
448
                    agg_cfg.compute_statistics = true;
13✔
449
                    agg_cfg.compute_percentiles = true;
13✔
450
                    agg_cfg.percentiles = visitor->resolved_percentiles;
13✔
451
                    agg_cfg.sketch_accuracy = 0.01;
39✔
452
                    agg_cfg.track_process_parents = false;
39✔
453

454
                    auto [base_result, var_result] = co_await coro::when_all(
91!
455
                        run_aggregation(
78!
456
                            baseline_files, agg_cfg, query,
39!
457
                            config.baseline_index_dir, config.checkpoint_size,
39!
458
                            config.force_rebuild, config.executor_threads),
39✔
459
                        run_aggregation(
78!
460
                            variant_files, agg_cfg, query,
39!
461
                            config.variant_index_dir, config.checkpoint_size,
39!
462
                            config.force_rebuild, config.executor_threads));
39✔
463

464
                    if (!metadata_set) {
13!
465
                        b_files_actual = base_result.total_files_processed;
13✔
466
                        v_files_actual = var_result.total_files_processed;
13✔
467
                        output_ptr->baseline_file_count = b_files_actual;
13✔
468
                        output_ptr->variant_file_count = v_files_actual;
13✔
469
                        output_ptr->baseline_meta = extract_metadata(
26!
470
                            base_result.aggregations, b_files_actual);
13✔
471
                        output_ptr->variant_meta = extract_metadata(
26!
472
                            var_result.aggregations, v_files_actual);
13✔
473
                        metadata_set = true;
13✔
474
                    }
13✔
475

476
                    ComparisonVisitorPair pair;
13✔
477
                    pair.baseline = std::move(base_result);
13✔
478
                    pair.variant = std::move(var_result);
13✔
479
                    pair.node = *visitor;
13!
480
                    pairs.push_back(std::move(pair));
13!
481
                }
13!
482

483
                ComparisonUtilityInput cmp_input;
39✔
484
                cmp_input.visitors = std::move(pairs);
39✔
485
                cmp_input.root_node = node;
39!
486
                cmp_input.baseline_file_count = b_files_actual;
39✔
487
                cmp_input.variant_file_count = v_files_actual;
39✔
488

489
                ComparisonUtility cmp;
39!
490
                auto cmp_output = co_await cmp.process(cmp_input);
52!
491
                output_ptr->nodes.push_back(std::move(cmp_output->result));
13!
492
            }
13!
493

494
            // Inject metadata rows into root SUMMARY.
495
            auto meta_rows = build_metadata_metrics(output_ptr->baseline_meta,
26!
496
                                                    output_ptr->variant_meta);
13✔
497
            for (auto &node : output_ptr->nodes) {
26✔
498
                node.summary.metrics.insert(node.summary.metrics.begin(),
39!
499
                                            meta_rows.begin(), meta_rows.end());
26✔
500
            }
13✔
501

502
            auto end_time = std::chrono::high_resolution_clock::now();
13✔
503
            std::chrono::duration<double, std::milli> duration =
13✔
504
                end_time - start_time;
13!
505
            output_ptr->execution_time_ms = duration.count();
13!
506
        };
299✔
507

508
        rt->submit(task(), "comparator").get();
13!
509
    });
13✔
510
}
13✔
511

512
// -----------------------------------------------------------------------
513
// compare() -- returns ArrowTable
514
// -----------------------------------------------------------------------
515

516
static PyObject *Comparator_compare(ComparatorObject *self, PyObject *args,
6✔
517
                                    PyObject *kwds) {
518
    ComparatorArgs cargs;
6✔
519
    if (parse_comparator_args(args, kwds, cargs) < 0) return NULL;
6!
520

521
    ComparisonOutput output;
6✔
522
    if (!run_comparison_pipeline(self, cargs, output)) {
6!
523
        return NULL;
×
524
    }
525

526
#ifdef DFTRACER_UTILS_ENABLE_ARROW
527
    auto arrow_result = output.to_arrow();
6!
528
    if (!arrow_result.valid()) {
6!
529
        PyErr_SetString(PyExc_RuntimeError,
×
530
                        "Failed to convert comparison output "
531
                        "to Arrow");
532
        return NULL;
×
533
    }
534
    return arrow_result_to_table(std::move(arrow_result));
6!
535
#else
536
    PyErr_SetString(PyExc_RuntimeError,
537
                    "dftracer-utils was built without Arrow support");
538
    return NULL;
539
#endif
540
}
6✔
541

542
// -----------------------------------------------------------------------
543
// compare_json() -- returns JSON string
544
// -----------------------------------------------------------------------
545

546
static PyObject *Comparator_compare_json(ComparatorObject *self, PyObject *args,
4✔
547
                                         PyObject *kwds) {
548
    ComparatorArgs cargs;
4✔
549
    if (parse_comparator_args(args, kwds, cargs) < 0) return NULL;
4!
550

551
    ComparisonOutput output;
4✔
552
    if (!run_comparison_pipeline(self, cargs, output)) {
4!
553
        return NULL;
×
554
    }
555

556
    TreeTableFormatter formatter;
4!
557
    std::string json = formatter.render_json(output);
4!
558
    return PyUnicode_FromStringAndSize(json.data(), (Py_ssize_t)json.size());
4!
559
}
4✔
560

561
// -----------------------------------------------------------------------
562
// compare_table() -- returns formatted table string
563
// -----------------------------------------------------------------------
564

565
static PyObject *Comparator_compare_table(ComparatorObject *self,
3✔
566
                                          PyObject *args, PyObject *kwds) {
567
    ComparatorArgs cargs;
3✔
568
    if (parse_comparator_args(args, kwds, cargs) < 0) return NULL;
3!
569

570
    ComparisonOutput output;
3✔
571
    if (!run_comparison_pipeline(self, cargs, output)) {
3!
572
        return NULL;
×
573
    }
574

575
    FormatterOptions fmt_opts;
3✔
576
    fmt_opts.use_color = false;
3✔
577
    fmt_opts.use_unicode = false;
3✔
578
    TreeTableFormatter formatter(fmt_opts);
3!
579

580
    // Render to a temporary file and read back as string
581
    char *buf = NULL;
3✔
582
    std::size_t buf_size = 0;
3✔
583
    FILE *memstream = open_memstream(&buf, &buf_size);
3!
584
    if (!memstream) {
3!
585
        PyErr_SetString(PyExc_RuntimeError, "Failed to create memory stream");
×
586
        return NULL;
×
587
    }
588

589
    formatter.render(memstream, output);
3!
590
    fflush(memstream);
3!
591
    fclose(memstream);
3!
592

593
    PyObject *result = PyUnicode_FromStringAndSize(buf, (Py_ssize_t)buf_size);
3!
594
    free(buf);
3!
595
    return result;
3✔
596
}
3✔
597

598
// -----------------------------------------------------------------------
599
// __call__ delegates to compare()
600
// -----------------------------------------------------------------------
601

602
static PyObject *Comparator_call(PyObject *self, PyObject *args,
1✔
603
                                 PyObject *kwds) {
604
    return Comparator_compare((ComparatorObject *)self, args, kwds);
1✔
605
}
606

607
// -----------------------------------------------------------------------
608
// Method table
609
// -----------------------------------------------------------------------
610

611
static const char *COMPARE_DOC =
612
    "compare(baseline, variant, query='', group_by='',\n"
613
    "        format='table', time_interval_ms=5000.0,\n"
614
    "        threshold=0.0, executor_threads=0,\n"
615
    "        index_dir='', force_rebuild=False, config='')\n"
616
    "--\n"
617
    "\n"
618
    "Run comparison pipeline, return materialized ArrowTable.\n"
619
    "\n"
620
    "Args:\n"
621
    "    baseline (str): Baseline trace file or directory.\n"
622
    "    variant (str): Variant trace file or directory.\n"
623
    "    query (str): Query filter (default: all events).\n"
624
    "    group_by (str): Comma-separated group keys.\n"
625
    "    format (str): Output format (default 'table').\n"
626
    "    time_interval_ms (float): Time bucket in ms "
627
    "(default 5000).\n"
628
    "    threshold (float): Hide changes below this pct.\n"
629
    "    executor_threads (int): Parallel threads (0=auto).\n"
630
    "    index_dir (str): Directory for .dftindex stores.\n"
631
    "    force_rebuild (bool): Force index rebuild.\n"
632
    "    config (str): JSON config file path.\n"
633
    "\n"
634
    "Returns:\n"
635
    "    ArrowTable: Comparison results.\n";
636

637
static const char *COMPARE_JSON_DOC =
638
    "compare_json(baseline, variant, query='', group_by='',\n"
639
    "             format='table', time_interval_ms=5000.0,\n"
640
    "             threshold=0.0, executor_threads=0,\n"
641
    "             index_dir='', force_rebuild=False, "
642
    "config='')\n"
643
    "--\n"
644
    "\n"
645
    "Run comparison pipeline, return JSON string.\n"
646
    "\n"
647
    "Args:\n"
648
    "    baseline (str): Baseline trace file or directory.\n"
649
    "    variant (str): Variant trace file or directory.\n"
650
    "    query (str): Query filter (default: all events).\n"
651
    "    group_by (str): Comma-separated group keys.\n"
652
    "    format (str): Output format (default 'table').\n"
653
    "    time_interval_ms (float): Time bucket in ms "
654
    "(default 5000).\n"
655
    "    threshold (float): Hide changes below this pct.\n"
656
    "    executor_threads (int): Parallel threads (0=auto).\n"
657
    "    index_dir (str): Directory for .dftindex stores.\n"
658
    "    force_rebuild (bool): Force index rebuild.\n"
659
    "    config (str): JSON config file path.\n"
660
    "\n"
661
    "Returns:\n"
662
    "    str: JSON representation of comparison results.\n";
663

664
static const char *COMPARE_TABLE_DOC =
665
    "compare_table(baseline, variant, query='', group_by='',\n"
666
    "              format='table', time_interval_ms=5000.0,\n"
667
    "              threshold=0.0, executor_threads=0,\n"
668
    "              index_dir='', force_rebuild=False, "
669
    "config='')\n"
670
    "--\n"
671
    "\n"
672
    "Run comparison pipeline, return formatted table string.\n"
673
    "\n"
674
    "Args:\n"
675
    "    baseline (str): Baseline trace file or directory.\n"
676
    "    variant (str): Variant trace file or directory.\n"
677
    "    query (str): Query filter (default: all events).\n"
678
    "    group_by (str): Comma-separated group keys.\n"
679
    "    format (str): Output format (default 'table').\n"
680
    "    time_interval_ms (float): Time bucket in ms "
681
    "(default 5000).\n"
682
    "    threshold (float): Hide changes below this pct.\n"
683
    "    executor_threads (int): Parallel threads (0=auto).\n"
684
    "    index_dir (str): Directory for .dftindex stores.\n"
685
    "    force_rebuild (bool): Force index rebuild.\n"
686
    "    config (str): JSON config file path.\n"
687
    "\n"
688
    "Returns:\n"
689
    "    str: Formatted ASCII table of comparison results.\n";
690

691
static PyMethodDef Comparator_methods[] = {
1✔
692
    {"compare", (PyCFunction)Comparator_compare, METH_VARARGS | METH_KEYWORDS,
2✔
693
     COMPARE_DOC},
1✔
694
    {"compare_json", (PyCFunction)Comparator_compare_json,
2✔
695
     METH_VARARGS | METH_KEYWORDS, COMPARE_JSON_DOC},
1✔
696
    {"compare_table", (PyCFunction)Comparator_compare_table,
2✔
697
     METH_VARARGS | METH_KEYWORDS, COMPARE_TABLE_DOC},
1✔
698
    {NULL}};
1✔
699

700
PyTypeObject ComparatorType = {
701
    PyVarObject_HEAD_INIT(
702
        NULL, 0) "dftracer_utils_ext.ComparatorUtility", /* tp_name */
703
    sizeof(ComparatorObject),                            /* tp_basicsize */
704
    0,                                                   /* tp_itemsize */
705
    (destructor)Comparator_dealloc,                      /* tp_dealloc */
706
    0,                                        /* tp_vectorcall_offset */
707
    0,                                        /* tp_getattr */
708
    0,                                        /* tp_setattr */
709
    0,                                        /* tp_as_async */
710
    0,                                        /* tp_repr */
711
    0,                                        /* tp_as_number */
712
    0,                                        /* tp_as_sequence */
713
    0,                                        /* tp_as_mapping */
714
    0,                                        /* tp_hash */
715
    Comparator_call,                          /* tp_call */
716
    0,                                        /* tp_str */
717
    0,                                        /* tp_getattro */
718
    0,                                        /* tp_setattro */
719
    0,                                        /* tp_as_buffer */
720
    Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE, /* tp_flags */
721
    "ComparatorUtility(runtime: Runtime | None = None)\n"
722
    "--\n\n"
723
    "Compare DFTracer trace metrics between baseline and "
724
    "variant.\n\n"
725
    "Args:\n"
726
    "    runtime (Runtime or None): Runtime for thread pool "
727
    "control.\n"
728
    "        If None, uses the default global Runtime.\n\n"
729
    "compare(baseline, variant, ...) -> ArrowTable\n"
730
    "    Run comparison and return a materialized Arrow "
731
    "table.\n\n"
732
    "compare_json(baseline, variant, ...) -> str\n"
733
    "    Run comparison and return JSON string.\n\n"
734
    "compare_table(baseline, variant, ...) -> str\n"
735
    "    Run comparison and return formatted table "
736
    "string.\n",               /* tp_doc */
737
    0,                         /* tp_traverse */
738
    0,                         /* tp_clear */
739
    0,                         /* tp_richcompare */
740
    0,                         /* tp_weaklistoffset */
741
    0,                         /* tp_iter */
742
    0,                         /* tp_iternext */
743
    Comparator_methods,        /* tp_methods */
744
    0,                         /* tp_members */
745
    0,                         /* tp_getset */
746
    0,                         /* tp_base */
747
    0,                         /* tp_dict */
748
    0,                         /* tp_descr_get */
749
    0,                         /* tp_descr_set */
750
    0,                         /* tp_dictoffset */
751
    (initproc)Comparator_init, /* tp_init */
752
    0,                         /* tp_alloc */
753
    Comparator_new,            /* tp_new */
754
};
755

756
int init_comparator(PyObject *m) {
1✔
757
    if (register_type(m, &ComparatorType, "ComparatorUtility") < 0) return -1;
1!
758

759
    return 0;
1✔
760
}
1✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE TRIAL · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc