26043728131

Committed 18 May 2026 03:37PM UTC coverage: 51.706% (-0.4%) from 52.076%

Build # 26043728131

Build Type

push

github

Committed by

hariharan-devarajan

Commit Message

feat(perf): performance improvements for parallel reading, indexing, and aggregation

Indexer
- Streaming parse-and-emit worker pipeline with bounded memory usage
- Concurrent SST artifact ingestion with staging support
- Gzip member slicing for parallel indexing
- Lazy decoding for compressed value counts
- Bypass DOM wrapper for indexer hot path (simdjson on_demand)
- Decoupled write workers from parse workers
- --rebuild-summaries flag and optimized root summary rebuild

Aggregator / MPI
- Task-based DAG execution for aggregator pipeline
- Shared staging for multi-node artifact relocation
- Per-node thread scaling to avoid oversubscription
- Unified distributed aggregation tracking, removed manifest consolidation
- Deterministic aggregation and intra-file parallelism

Trace reader / query
- Compiled predicate evaluation for AND-of-EQ queries
- Uniform-match shortcut for AND-of-EQ queries
- Line-range support for work items and checkpoint processing
- Optimized chunk pruning and checkpoint handling

Replay
- Pipelined replay with coroutines and channels
- JsonParser-based trace processing
- Optimized string handling and i/o buffering

Organize / writer / dft
- Parallel slice creation and merging in organize visitor
- Inline indexer in organize
- Gzip member tracking in writer
- Coroutine-based event dispatcher with extracted parse logic
- Batch flushing in organize visitor

Arrow / call_tree
- Optimized arrow conversion
- Arrow IPC support and improved save/load in call_tree

Build / infrastructure
- zlib-ng option, system simdjson fallback
- cgroup v1/v2 memory limit detection
- Auto-computed per-file memory estimates and batch sizes
- CI: perf branch trigger, formatting

Docs
- Rewritten indexer and trace reader API references

Coverage Stats

35907 of 90345 branches covered (39.74%)

Branch coverage included in aggregate %.

16869 of 21880 new or added lines in 137 files covered. (77.1%)

273 existing lines in 39 files now uncovered.

32021 of 41028 relevant lines covered (78.05%)

13164.29 hits per line

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

55.2

/src/dftracer/utils/python/utilities/comparator.cpp

#define PY_SSIZE_T_CLEAN
#include <dftracer/utils/core/common/filesystem.h>
#include <dftracer/utils/core/common/platform_compat.h>
#include <dftracer/utils/core/coro/channel.h>
#include <dftracer/utils/core/coro/task.h>
#include <dftracer/utils/core/coro/when_all.h>
#include <dftracer/utils/core/pipeline/pipeline.h>
#include <dftracer/utils/core/pipeline/pipeline_config.h>
#include <dftracer/utils/core/runtime.h>
#include <dftracer/utils/core/tasks/coro_scope.h>
#include <dftracer/utils/core/tasks/task.h>
#include <dftracer/utils/python/arrow_helpers.h>
#include <dftracer/utils/python/runtime.h>
#include <dftracer/utils/python/utilities/comparator.h>
#include <dftracer/utils/utilities/common/query/query.h>
#include <dftracer/utils/utilities/composites/dft/aggregators/aggregators.h>
#include <dftracer/utils/utilities/composites/dft/comparator/comparison_config.h>
#include <dftracer/utils/utilities/composites/dft/comparator/comparison_result.h>
#include <dftracer/utils/utilities/composites/dft/comparator/comparison_utility.h>
#include <dftracer/utils/utilities/composites/dft/comparator/tree_table_formatter.h>
#include <dftracer/utils/utilities/composites/dft/indexing/index_resolver_utility.h>
#include <dftracer/utils/utilities/composites/dft/internal/utils.h>
#include <dftracer/utils/utilities/composites/dft/metadata_collector_utility.h>
#include <dftracer/utils/utilities/filesystem/pattern_directory_scanner_utility.h>
#include <dftracer/utils/utilities/indexer/index_builder_utility.h>
#include <dftracer/utils/utilities/indexer/internal/indexer.h>

#include <atomic>
#include <chrono>
#include <cstdio>
#include <ctime>
#include <string>
#include <vector>

using dftracer::utils::Runtime;
using dftracer::utils::coro::CoroTask;
using namespace dftracer::utils;
using namespace dftracer::utils::utilities;
using namespace dftracer::utils::utilities::composites::dft::aggregators;
using namespace dftracer::utils::utilities::composites::dft::comparator;

#include <dftracer/utils/core/common/config.h>
#ifdef DFTRACER_UTILS_ENABLE_ARROW
using dftracer::utils::python::arrow_result_to_table;
using dftracer::utils::utilities::common::arrow::ArrowExportResult;
#endif

static Runtime *get_runtime(ComparatorObject *self) {
    if (self->runtime_obj)
        return ((RuntimeObject *)self->runtime_obj)->runtime.get();
    return get_default_runtime();
}

static void Comparator_dealloc(ComparatorObject *self) {
    Py_XDECREF(self->runtime_obj);
    Py_TYPE(self)->tp_free((PyObject *)self);
}

static PyObject *Comparator_new(PyTypeObject *type, PyObject *args,
                                PyObject *kwds) {
    ComparatorObject *self = (ComparatorObject *)type->tp_alloc(type, 0);
    if (self) {
        self->runtime_obj = NULL;
    }
    return (PyObject *)self;
}

static int Comparator_init(ComparatorObject *self, PyObject *args,
                           PyObject *kwds) {
    static const char *kwlist[] = {"runtime", NULL};
    PyObject *runtime_arg = NULL;

    if (!PyArg_ParseTupleAndKeywords(args, kwds, "|O", (char **)kwlist,
                                     &runtime_arg)) {
        return -1;
    }

    if (runtime_arg && runtime_arg != Py_None) {
        if (PyObject_TypeCheck(runtime_arg, &RuntimeType)) {
            Py_INCREF(runtime_arg);
            self->runtime_obj = runtime_arg;
        } else {
            PyObject *native = PyObject_GetAttrString(runtime_arg, "_native");
            if (native && PyObject_TypeCheck(native, &RuntimeType)) {
                self->runtime_obj = native;
            } else {
                Py_XDECREF(native);
                PyErr_SetString(PyExc_TypeError,
                                "runtime must be a Runtime instance or None");
                return -1;
            }
        }
    }

    return 0;
}

// -----------------------------------------------------------------------
// Helpers
// -----------------------------------------------------------------------

struct ComparatorArgs {
    std::string baseline;
    std::string variant;
    std::string query;
    std::string group_by;
    std::string format;
    double time_interval_ms = 5000.0;
    double threshold = 0.0;
    std::size_t executor_threads = 0;
    std::string baseline_index_dir;
    std::string variant_index_dir;
    bool force_rebuild = false;
    std::string config_path;
};

static int parse_comparator_args(PyObject *args, PyObject *kwds,
                                 ComparatorArgs &out) {
    static const char *kwlist[] = {"baseline",
                                   "variant",
                                   "query",
                                   "group_by",
                                   "format",
                                   "time_interval_ms",
                                   "threshold",
                                   "executor_threads",
                                   "baseline_index_dir",
                                   "variant_index_dir",
                                   "force_rebuild",
                                   "config",
                                   NULL};

    const char *baseline = NULL;
    const char *variant = NULL;
    const char *query = "";
    const char *group_by = "";
    const char *format = "table";
    double time_interval_ms = 5000.0;
    double threshold = 0.0;
    Py_ssize_t executor_threads = 0;
    const char *baseline_index_dir = "";
    const char *variant_index_dir = "";
    int force_rebuild = 0;
    const char *config = "";

    if (!PyArg_ParseTupleAndKeywords(
            args, kwds, "ss|sssddnssps", (char **)kwlist, &baseline, &variant,
            &query, &group_by, &format, &time_interval_ms, &threshold,
            &executor_threads, &baseline_index_dir, &variant_index_dir,
            &force_rebuild, &config))
        return -1;

    out.baseline = baseline;
    out.variant = variant;
    out.query = query;
    out.group_by = group_by;
    out.format = format;
    out.time_interval_ms = time_interval_ms;
    out.threshold = threshold;
    out.executor_threads = static_cast<std::size_t>(executor_threads);
    out.baseline_index_dir = baseline_index_dir;
    out.variant_index_dir = variant_index_dir;
    out.force_rebuild = force_rebuild != 0;
    out.config_path = config;

    return 0;
}

namespace {

void flatten_nodes(const ComparisonNode &node,
                   std::vector<const ComparisonNode *> &out) {
    out.push_back(&node);
    for (const auto &child : node.children) {
        flatten_nodes(child, out);
    }
}

CoroTask<EventAggregatorOutput> run_aggregation(
    std::vector<std::string> input_files, AggregationConfig agg_config,
    std::optional<common::query::Query> query, std::string index_dir,
    std::size_t checkpoint_size, bool force_rebuild,
    std::size_t executor_threads) {
    constexpr std::size_t CHUNK_SIZE_MB = 4;
    constexpr std::size_t BATCH_SIZE_MB = 4;

    auto pipeline_config = PipelineConfig()
                               .with_name("DFTracer Comparator Aggregation")
                               .with_compute_threads(executor_threads)
                               .with_watchdog(false);
    Pipeline pipeline(pipeline_config);

    EventAggregator merger;
    std::atomic<int> global_chunk_idx{0};

    auto streaming_task = make_task(
        [&](CoroScope &ctx) -> CoroTask<void> {
            auto chunk_chan = coro::make_channel<ChunkAggregatorInput>(0);
            auto result_chan = coro::make_channel<ChunkAggregationOutput>(2);

            co_await ctx.scope([&](CoroScope &scope) -> CoroTask<void> {
                for (const auto &file_path : input_files) {
                    auto *global_chunk_idx_ptr = &global_chunk_idx;
                    scope.spawn([file_path, ch = chunk_chan->producer(),
                                 index_dir, checkpoint_size, force_rebuild,
                                 agg_config, query, global_chunk_idx_ptr](
                                    CoroScope & /*fctx*/) mutable
                                    -> CoroTask<void> {
                        [[maybe_unused]] auto producer_guard = ch.guard();

                        std::string index_path =
                            composites::dft::internal::determine_index_path(
                                file_path, index_dir);

                        auto meta_input =
                            composites::dft::MetadataCollectorUtilityInput::
                                from_file(file_path)
                                    .with_checkpoint_size(checkpoint_size)
                                    .with_force_rebuild(force_rebuild)
                                    .with_index(index_path);
                        auto metadata =
                            co_await composites::dft::MetadataCollectorUtility{}
                                .process(meta_input);

                        if (!metadata.success) {
                            co_return;
                        }

                        FileChunkMapperUtility file_mapper;
                        auto mapper_input =
                            FileChunkMapperInput::from_metadata(metadata)
                                .with_config(agg_config)
                                .with_checkpoint_size(checkpoint_size)
                                .with_target_chunk_size(CHUNK_SIZE_MB)
                                .with_batch_size(BATCH_SIZE_MB * 1024 * 1024);
                        mapper_input.query = query;
                        auto file_chunks =
                            co_await file_mapper.process(mapper_input);

                        int start_idx = global_chunk_idx_ptr->fetch_add(
                            static_cast<int>(file_chunks.size()));
                        for (int i = 0;
                             i < static_cast<int>(file_chunks.size()); ++i) {
                            file_chunks[i].chunk_index = start_idx + i;
                        }

                        for (auto &chunk : file_chunks) {
                            if (!co_await ch.send(std::move(chunk))) {
                                co_return;
                            }
                        }
                        co_return;
                    });
                }

                for (std::size_t w = 0; w < executor_threads; ++w) {
                    (void)w;
                    scope.spawn([chunk_chan, rp = result_chan->producer(),
                                 result_chan](
                                    CoroScope &wctx) mutable -> CoroTask<void> {
                        [[maybe_unused]] auto producer_guard = rp.guard();
                        while (auto input = co_await wctx.receive(chunk_chan)) {
                            ChunkAggregatorUtility agg;
                            auto output = co_await agg.process(*input);
                            if (!co_await result_chan->send(
                                    std::move(output))) {
                                co_return;
                            }
                        }
                        co_return;
                    });
                }

                auto *merger_ptr = &merger;
                scope.spawn([result_chan,
                             merger_ptr](CoroScope &mctx) -> CoroTask<void> {
                    while (auto output = co_await mctx.receive(result_chan)) {
                        merger_ptr->merge_chunk(std::move(*output));
                    }
                    co_return;
                });

                co_return;
            });

            co_return;
        },
        "StreamingAggregate");

    EventAggregatorOutput result;
    auto post_task = make_task(
        [&](CoroScope & /*ctx*/) -> CoroTask<bool> {
            result = merger.finalize();
            co_return result.success;
        },
        "Finalize");

    post_task->depends_on(streaming_task);
    pipeline.set_source(streaming_task);
    pipeline.set_destination(post_task);
    pipeline.execute();

    co_return result;
}

}  // namespace

static int run_comparison_pipeline(ComparatorObject *self,
                                   const ComparatorArgs &cargs,
                                   ComparisonOutput &output,
                                   std::string &error_msg) {
    ComparatorArgs args_copy = cargs;
    auto *output_ptr = &output;

    Py_BEGIN_ALLOW_THREADS try {
        ComparisonConfig config;
        if (!args_copy.config_path.empty()) {
            std::string parse_error;
            auto parsed = ComparisonConfig::from_json_file(
                args_copy.config_path, parse_error);
            if (!parsed) {
                error_msg = "Config error: " + parse_error;
                goto done;
            }
            config = std::move(*parsed);
        } else {
            config = ComparisonConfig::from_cli(
                args_copy.baseline, args_copy.variant, args_copy.query,
                args_copy.group_by);
        }

        config.format = args_copy.format;
        config.no_color = true;
        if (args_copy.executor_threads > 0)
            config.executor_threads = args_copy.executor_threads;
        if (!args_copy.baseline_index_dir.empty())
            config.baseline_index_dir = args_copy.baseline_index_dir;
        if (!args_copy.variant_index_dir.empty())
            config.variant_index_dir = args_copy.variant_index_dir;
        if (args_copy.force_rebuild)
            config.force_rebuild = args_copy.force_rebuild;
        if (args_copy.threshold > 0.0)
            config.defaults.threshold_pct = args_copy.threshold;
        if (args_copy.time_interval_ms > 0.0)
            config.defaults.time_interval_ms = args_copy.time_interval_ms;

        config.resolve();

        if (config.executor_threads == 0) {
            config.executor_threads = dftracer_utils_hardware_concurrency();
        }
        if (config.checkpoint_size == 0) {
            config.checkpoint_size =
                indexer::internal::Indexer::DEFAULT_CHECKPOINT_SIZE;
        }

        using composites::dft::indexing::IndexResolverUtility;
        using composites::dft::indexing::ResolverInput;
        using indexer::IndexBatchBuilderUtility;
        using indexer::IndexBuildBatchConfig;

        Runtime *rt = get_runtime(self);

        auto *error_msg_ptr = &error_msg;
        auto task = [config, output_ptr, error_msg_ptr,
                     rt]() -> CoroTask<void> {
            auto resolve_and_build =
                [&config](
                    CoroScope &scope, const std::string &path,
                    const std::string &index_dir,
                    std::vector<std::string> &out_files) -> CoroTask<void> {
                IndexResolverUtility resolver;
                ResolverInput resolve_input;
                resolve_input.index_dir = index_dir;
                resolve_input.require_checkpoints = !config.force_rebuild;
                if (fs::is_regular_file(path)) {
                    resolve_input.files = {path};
                } else {
                    resolve_input.directory = path;
                }

                auto result = co_await resolver.process(resolve_input);
                out_files = std::move(result.all_files);

                if (out_files.empty() || result.needs_checkpoint.empty()) {
                    co_return;
                }

                auto batch_cfg = std::make_shared<IndexBuildBatchConfig>();
                batch_cfg->file_paths.reserve(result.needs_checkpoint.size());
                for (const auto &item : result.needs_checkpoint) {
                    batch_cfg->file_paths.push_back(item.file_path);
                }
                batch_cfg->index_dir = index_dir;
                batch_cfg->checkpoint_size = config.checkpoint_size;
                batch_cfg->parallelism = config.executor_threads;
                batch_cfg->force_rebuild = config.force_rebuild;
                batch_cfg->use_batch_write = true;
                batch_cfg->rebuild_root_summaries = true;

                co_await IndexBatchBuilderUtility::process(
                    &scope, std::move(batch_cfg));
            };

            std::vector<std::string> baseline_files;
            std::vector<std::string> variant_files;

            bool shared_index =
                composites::dft::internal::determine_index_path(
                    config.baseline, config.baseline_index_dir) ==
                composites::dft::internal::determine_index_path(
                    config.variant, config.variant_index_dir);

            co_await run_coro_scope(
                rt->executor(), [&](CoroScope &scope) -> CoroTask<void> {
                    if (shared_index) {
                        co_await resolve_and_build(scope, config.baseline,
                                                   config.baseline_index_dir,
                                                   baseline_files);
                        if (config.baseline == config.variant) {
                            variant_files = baseline_files;
                        } else {
                            co_await resolve_and_build(scope, config.variant,
                                                       config.variant_index_dir,
                                                       variant_files);
                        }
                    } else {
                        scope.spawn([&](CoroScope &s) -> CoroTask<void> {
                            co_await resolve_and_build(
                                s, config.baseline, config.baseline_index_dir,
                                baseline_files);
                        });
                        co_await resolve_and_build(scope, config.variant,
                                                   config.variant_index_dir,
                                                   variant_files);
                    }
                });

            if (baseline_files.empty()) {
                *error_msg_ptr =
                    "No trace files found in baseline: " + config.baseline;
                co_return;
            }
            if (variant_files.empty()) {
                *error_msg_ptr =
                    "No trace files found in variant: " + config.variant;
                co_return;
            }

            output_ptr->baseline_path = config.baseline;
            output_ptr->variant_path = config.variant;
            output_ptr->baseline_file_count = baseline_files.size();
            output_ptr->variant_file_count = variant_files.size();

            auto start_time = std::chrono::high_resolution_clock::now();

            for (auto &node : config.nodes) {
                std::vector<const ComparisonNode *> visitors;
                flatten_nodes(node, visitors);

                std::vector<ComparisonVisitorPair> pairs;
                pairs.reserve(visitors.size());

                for (const auto *visitor : visitors) {
                    std::optional<common::query::Query> query;
                    if (!visitor->composed_query.empty()) {
                        auto result = common::query::Query::from_string(
                            visitor->composed_query);
                        if (!result) {
                            *error_msg_ptr = "Invalid query for node '" +
                                             visitor->name +
                                             "': " + result.error().format();
                            co_return;
                        }
                        query = std::move(*result);
                    }

                    AggregationConfig agg_cfg;
                    agg_cfg.time_interval_us = static_cast<std::uint64_t>(
                        config.defaults.time_interval_ms * 1000.0);
                    agg_cfg.extra_group_keys = {};
                    agg_cfg.compute_statistics = true;
                    agg_cfg.compute_percentiles = true;
                    agg_cfg.percentiles = visitor->resolved_percentiles;
                    agg_cfg.sketch_accuracy = 0.01;
                    agg_cfg.track_process_parents = false;

                    auto [base_result, var_result] = co_await coro::when_all(
                        run_aggregation(
                            baseline_files, agg_cfg, query,
                            config.baseline_index_dir, config.checkpoint_size,
                            config.force_rebuild, config.executor_threads),
                        run_aggregation(
                            variant_files, agg_cfg, query,
                            config.variant_index_dir, config.checkpoint_size,
                            config.force_rebuild, config.executor_threads));

                    if (pairs.empty()) {
                        output_ptr->baseline_meta = extract_metadata(
                            base_result.aggregations, baseline_files.size());
                        output_ptr->variant_meta = extract_metadata(
                            var_result.aggregations, variant_files.size());
                    }

                    ComparisonVisitorPair pair;
                    pair.baseline = std::move(base_result);
                    pair.variant = std::move(var_result);
                    pair.node = *visitor;
                    pairs.push_back(std::move(pair));
                }

                ComparisonUtilityInput cmp_input;
                cmp_input.visitors = std::move(pairs);
                cmp_input.root_node = node;
                cmp_input.baseline_file_count = baseline_files.size();
                cmp_input.variant_file_count = variant_files.size();

                ComparisonUtility cmp;
                auto cmp_output = co_await cmp.process(cmp_input);
                output_ptr->nodes.push_back(std::move(cmp_output.result));
            }

            // Inject metadata rows into root SUMMARY.
            auto meta_rows = build_metadata_metrics(output_ptr->baseline_meta,
                                                    output_ptr->variant_meta);
            for (auto &node : output_ptr->nodes) {
                node.summary.metrics.insert(node.summary.metrics.begin(),
                                            meta_rows.begin(), meta_rows.end());
            }

            auto end_time = std::chrono::high_resolution_clock::now();
            std::chrono::duration<double, std::milli> duration =
                end_time - start_time;
            output_ptr->execution_time_ms = duration.count();
        };

        rt->submit(task(), "comparator").get();
    } catch (const std::exception &e) {
        error_msg = e.what();
    }
done:
    Py_END_ALLOW_THREADS

        return error_msg.empty()
        ? 0
        : -1;
}

// -----------------------------------------------------------------------
// compare() -- returns ArrowTable
// -----------------------------------------------------------------------

static PyObject *Comparator_compare(ComparatorObject *self, PyObject *args,
                                    PyObject *kwds) {
    ComparatorArgs cargs;
    if (parse_comparator_args(args, kwds, cargs) < 0) return NULL;

    ComparisonOutput output;
    std::string error_msg;
    if (run_comparison_pipeline(self, cargs, output, error_msg) < 0) {
        PyErr_SetString(PyExc_RuntimeError, error_msg.c_str());
        return NULL;
    }

#ifdef DFTRACER_UTILS_ENABLE_ARROW
    auto arrow_result = output.to_arrow();
    if (!arrow_result.valid()) {
        PyErr_SetString(PyExc_RuntimeError,
                        "Failed to convert comparison output "
                        "to Arrow");
        return NULL;
    }
    return arrow_result_to_table(std::move(arrow_result));
#else
    PyErr_SetString(PyExc_RuntimeError,
                    "dftracer-utils was built without Arrow support");
    return NULL;
#endif
}

// -----------------------------------------------------------------------
// compare_json() -- returns JSON string
// -----------------------------------------------------------------------

static PyObject *Comparator_compare_json(ComparatorObject *self, PyObject *args,
                                         PyObject *kwds) {
    ComparatorArgs cargs;
    if (parse_comparator_args(args, kwds, cargs) < 0) return NULL;

    ComparisonOutput output;
    std::string error_msg;
    if (run_comparison_pipeline(self, cargs, output, error_msg) < 0) {
        PyErr_SetString(PyExc_RuntimeError, error_msg.c_str());
        return NULL;
    }

    TreeTableFormatter formatter;
    std::string json = formatter.render_json(output);
    return PyUnicode_FromStringAndSize(json.data(), (Py_ssize_t)json.size());
}

// -----------------------------------------------------------------------
// compare_table() -- returns formatted table string
// -----------------------------------------------------------------------

static PyObject *Comparator_compare_table(ComparatorObject *self,
                                          PyObject *args, PyObject *kwds) {
    ComparatorArgs cargs;
    if (parse_comparator_args(args, kwds, cargs) < 0) return NULL;

    ComparisonOutput output;
    std::string error_msg;
    if (run_comparison_pipeline(self, cargs, output, error_msg) < 0) {
        PyErr_SetString(PyExc_RuntimeError, error_msg.c_str());
        return NULL;
    }

    FormatterOptions fmt_opts;
    fmt_opts.use_color = false;
    fmt_opts.use_unicode = false;
    TreeTableFormatter formatter(fmt_opts);

    // Render to a temporary file and read back as string
    char *buf = NULL;
    std::size_t buf_size = 0;
    FILE *memstream = open_memstream(&buf, &buf_size);
    if (!memstream) {
        PyErr_SetString(PyExc_RuntimeError, "Failed to create memory stream");
        return NULL;
    }

    formatter.render(memstream, output);
    fflush(memstream);
    fclose(memstream);

    PyObject *result = PyUnicode_FromStringAndSize(buf, (Py_ssize_t)buf_size);
    free(buf);
    return result;
}

// -----------------------------------------------------------------------
// __call__ delegates to compare()
// -----------------------------------------------------------------------

static PyObject *Comparator_call(PyObject *self, PyObject *args,
                                 PyObject *kwds) {
    return Comparator_compare((ComparatorObject *)self, args, kwds);
}

// -----------------------------------------------------------------------
// Method table
// -----------------------------------------------------------------------

static const char *COMPARE_DOC =
    "compare(baseline, variant, query='', group_by='',\n"
    "        format='table', time_interval_ms=5000.0,\n"
    "        threshold=0.0, executor_threads=0,\n"
    "        index_dir='', force_rebuild=False, config='')\n"
    "--\n"
    "\n"
    "Run comparison pipeline, return materialized ArrowTable.\n"
    "\n"
    "Args:\n"
    "    baseline (str): Baseline trace file or directory.\n"
    "    variant (str): Variant trace file or directory.\n"
    "    query (str): Query filter (default: all events).\n"
    "    group_by (str): Comma-separated group keys.\n"
    "    format (str): Output format (default 'table').\n"
    "    time_interval_ms (float): Time bucket in ms "
    "(default 5000).\n"
    "    threshold (float): Hide changes below this pct.\n"
    "    executor_threads (int): Parallel threads (0=auto).\n"
    "    index_dir (str): Directory for .dftindex stores.\n"
    "    force_rebuild (bool): Force index rebuild.\n"
    "    config (str): JSON config file path.\n"
    "\n"
    "Returns:\n"
    "    ArrowTable: Comparison results.\n";

static const char *COMPARE_JSON_DOC =
    "compare_json(baseline, variant, query='', group_by='',\n"
    "             format='table', time_interval_ms=5000.0,\n"
    "             threshold=0.0, executor_threads=0,\n"
    "             index_dir='', force_rebuild=False, "
    "config='')\n"
    "--\n"
    "\n"
    "Run comparison pipeline, return JSON string.\n"
    "\n"
    "Args:\n"
    "    baseline (str): Baseline trace file or directory.\n"
    "    variant (str): Variant trace file or directory.\n"
    "    query (str): Query filter (default: all events).\n"
    "    group_by (str): Comma-separated group keys.\n"
    "    format (str): Output format (default 'table').\n"
    "    time_interval_ms (float): Time bucket in ms "
    "(default 5000).\n"
    "    threshold (float): Hide changes below this pct.\n"
    "    executor_threads (int): Parallel threads (0=auto).\n"
    "    index_dir (str): Directory for .dftindex stores.\n"
    "    force_rebuild (bool): Force index rebuild.\n"
    "    config (str): JSON config file path.\n"
    "\n"
    "Returns:\n"
    "    str: JSON representation of comparison results.\n";

static const char *COMPARE_TABLE_DOC =
    "compare_table(baseline, variant, query='', group_by='',\n"
    "              format='table', time_interval_ms=5000.0,\n"
    "              threshold=0.0, executor_threads=0,\n"
    "              index_dir='', force_rebuild=False, "
    "config='')\n"
    "--\n"
    "\n"
    "Run comparison pipeline, return formatted table string.\n"
    "\n"
    "Args:\n"
    "    baseline (str): Baseline trace file or directory.\n"
    "    variant (str): Variant trace file or directory.\n"
    "    query (str): Query filter (default: all events).\n"
    "    group_by (str): Comma-separated group keys.\n"
    "    format (str): Output format (default 'table').\n"
    "    time_interval_ms (float): Time bucket in ms "
    "(default 5000).\n"
    "    threshold (float): Hide changes below this pct.\n"
    "    executor_threads (int): Parallel threads (0=auto).\n"
    "    index_dir (str): Directory for .dftindex stores.\n"
    "    force_rebuild (bool): Force index rebuild.\n"
    "    config (str): JSON config file path.\n"
    "\n"
    "Returns:\n"
    "    str: Formatted ASCII table of comparison results.\n";

static PyMethodDef Comparator_methods[] = {
    {"compare", (PyCFunction)Comparator_compare, METH_VARARGS | METH_KEYWORDS,
     COMPARE_DOC},
    {"compare_json", (PyCFunction)Comparator_compare_json,
     METH_VARARGS | METH_KEYWORDS, COMPARE_JSON_DOC},
    {"compare_table", (PyCFunction)Comparator_compare_table,
     METH_VARARGS | METH_KEYWORDS, COMPARE_TABLE_DOC},
    {NULL}};

PyTypeObject ComparatorType = {
    PyVarObject_HEAD_INIT(
        NULL, 0) "dftracer_utils_ext.ComparatorUtility", /* tp_name */
    sizeof(ComparatorObject),                            /* tp_basicsize */
    0,                                                   /* tp_itemsize */
    (destructor)Comparator_dealloc,                      /* tp_dealloc */
    0,                                        /* tp_vectorcall_offset */
    0,                                        /* tp_getattr */
    0,                                        /* tp_setattr */
    0,                                        /* tp_as_async */
    0,                                        /* tp_repr */
    0,                                        /* tp_as_number */
    0,                                        /* tp_as_sequence */
    0,                                        /* tp_as_mapping */
    0,                                        /* tp_hash */
    Comparator_call,                          /* tp_call */
    0,                                        /* tp_str */
    0,                                        /* tp_getattro */
    0,                                        /* tp_setattro */
    0,                                        /* tp_as_buffer */
    Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE, /* tp_flags */
    "ComparatorUtility(runtime: Runtime | None = None)\n"
    "--\n\n"
    "Compare DFTracer trace metrics between baseline and "
    "variant.\n\n"
    "Args:\n"
    "    runtime (Runtime or None): Runtime for thread pool "
    "control.\n"
    "        If None, uses the default global Runtime.\n\n"
    "compare(baseline, variant, ...) -> ArrowTable\n"
    "    Run comparison and return a materialized Arrow "
    "table.\n\n"
    "compare_json(baseline, variant, ...) -> str\n"
    "    Run comparison and return JSON string.\n\n"
    "compare_table(baseline, variant, ...) -> str\n"
    "    Run comparison and return formatted table "
    "string.\n",               /* tp_doc */
    0,                         /* tp_traverse */
    0,                         /* tp_clear */
    0,                         /* tp_richcompare */
    0,                         /* tp_weaklistoffset */
    0,                         /* tp_iter */
    0,                         /* tp_iternext */
    Comparator_methods,        /* tp_methods */
    0,                         /* tp_members */
    0,                         /* tp_getset */
    0,                         /* tp_base */
    0,                         /* tp_dict */
    0,                         /* tp_descr_get */
    0,                         /* tp_descr_set */
    0,                         /* tp_dictoffset */
    (initproc)Comparator_init, /* tp_init */
    0,                         /* tp_alloc */
    Comparator_new,            /* tp_new */
};

int init_comparator(PyObject *m) {
    if (PyType_Ready(&ComparatorType) < 0) return -1;

    Py_INCREF(&ComparatorType);
    if (PyModule_AddObject(m, "ComparatorUtility",
                           (PyObject *)&ComparatorType) < 0) {
        Py_DECREF(&ComparatorType);
        Py_DECREF(m);
        return -1;
    }

    return 0;
}

llnl / dftracer-utils / 26043728131

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

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