DFTracer Aggregation Pipeline

Namespace: dftracer::utils::utilities::composites::dft::aggregators

For usage guide and examples, see DFTracer Aggregation Pipeline.

source

struct AggGlobalConfig

Public Members

std::uint64_t time_interval_us

std::uint32_t config_hash

source

struct AggKeyView

Key view with resolved strings from the intern table. Lifetime: valid as long as aggregation_intern() exists (process lifetime).

Public Members

AggMapType map_type

std::string_view cat

std::string_view name

std::uint64_t pid

std::uint64_t tid

std::string_view hhash

std::string_view fhash

std::uint64_t time_bucket

source

struct AggMetricsFullView

Full metrics view including mean/m2 for stddev computation. Use for iter_aggregation which needs mean and stddev columns.

Public Functions

inline double dur_stddev() const

inline double size_stddev() const

inline double offset_stddev() const

Public Members

std::uint64_t count

std::uint64_t dur_total

std::uint64_t dur_min

std::uint64_t dur_max

double dur_mean

double dur_m2

std::uint64_t size_total

std::uint64_t size_min

std::uint64_t size_max

double size_mean

double size_m2

std::uint64_t offset_total

std::uint64_t offset_min

std::uint64_t offset_max

double offset_mean

double offset_m2

std::uint64_t ts

std::uint64_t te

source

struct AggMetricsView

Lightweight metrics view for Arrow export - only the fields needed.

Public Members

std::uint64_t count

std::uint64_t dur_total

std::uint64_t dur_min

std::uint64_t dur_max

std::uint64_t size_total

std::uint64_t size_min

std::uint64_t size_max

std::uint64_t offset_total

std::uint64_t offset_min

std::uint64_t offset_max

std::uint64_t ts

std::uint64_t te

source

struct AggregationBatch

Public Functions

inline AggregationBatch filter(const common::query::Query &query) const

Filter entries by query, returning a new batch with matching entries.

common::arrow::ArrowExportResult to_arrow() const

common::arrow::ArrowExportResult to_dfanalyzer_arrow(const DfanalyzerContext &ctx) const

Convert to dfanalyzer-compatible Arrow format. Outputs columns matching dfanalyzer schema:

• Events/Profiles: cat, func_name, pid, tid, file_hash, host_hash, file_name, host_name, proc_name, io_cat, acc_pat, count, time, size, time_min, time_max, size_min, size_max, time_range, time_start, time_end

• System: host_hash, time_range, sys_cpu_*, sys_mem_*

Public Members

std::vector<AggregationEntry> entries

AggregationBatchType batch_type

std::size_t total_events_processed

std::size_t total_files_processed

std::size_t total_bytes_processed

bool has_approximated_entries

const std::vector<std::uint32_t> *global_extra_key_ids

const std::vector<std::string> *global_custom_metric_names

source

struct AggregationConfig

Public Functions

inline std::uint32_t compute_hash() const

Public Members

std::uint64_t time_interval_us

bool use_relative_time

std::uint64_t reference_timestamp

bool normalize_time

std::vector<std::string> extra_group_keys

std::vector<std::string> custom_metric_fields

bool track_default_args

bool compute_statistics

bool compute_percentiles

double sketch_accuracy

std::vector<double> percentiles

std::vector<BoundaryEventConfig> boundary_events

bool track_process_parents

std::string output_format

Public Static Functions

static inline bool is_valid_format(const std::string &fmt)

static inline std::string supported_formats_str()

Public Static Attributes

static constexpr const char *FORMAT_JSON

static constexpr const char *FORMAT_ARROW

source

struct AggregationEntry

Public Functions

AggregationEntry() = default

inline AggregationEntry(AggregationKey k, AggregationMetrics m)

inline common::query::ValueMap to_value_map() const

Create a ValueMap from the key and metrics for query evaluation. Includes cat, name, pid, tid, hhash, fhash, time_bucket, extra_keys, and aggregation metrics (count, dur_total, dur_min, dur_max, etc.).

inline bool matches(const common::query::Query &query) const

Check if this entry matches a query.

Public Members

AggregationKey key

AggregationMetrics metrics

bool is_approximated

CountConfidenceInterval count_ci

source

struct AggregationKey

Public Functions

inline bool operator==(const AggregationKey &other) const

AggregationKey() = default

inline AggregationKey(const AggregationKey &other)

inline AggregationKey &operator=(const AggregationKey &other)

AggregationKey(AggregationKey&&) = default

AggregationKey &operator=(AggregationKey&&) = default

inline std::string_view cat() const

inline std::string_view name() const

inline std::string_view hhash() const

inline std::string_view fhash() const

Public Members

std::uint32_t cat_id

std::uint32_t name_id

std::uint64_t pid

std::uint64_t tid

std::uint32_t hhash_id

std::uint32_t fhash_id

std::uint64_t time_bucket

std::unique_ptr<std::vector<std::pair<std::uint32_t, std::uint32_t>>> extra_keys

source

struct AggregationKeyEqual

Public Functions

inline bool operator()(const AggregationKey &a, const AggregationKey &b) const

source

struct AggregationKeyHash

Public Types

using is_avalanching = void

Public Functions

inline std::size_t operator()(const AggregationKey &key) const

source

class AggregationMergeOperator : public rocksdb::MergeOperator

Public Functions

bool FullMergeV2(const MergeOperationInput &merge_in, MergeOperationOutput *merge_out) const override

bool PartialMerge(const ::rocksdb::Slice &key, const ::rocksdb::Slice &left_operand, const ::rocksdb::Slice &right_operand, std::string *new_value, ::rocksdb::Logger *logger) const override

inline const char *Name() const override

source

struct AggregationMetrics

Public Functions

inline explicit AggregationMetrics(double relative_accuracy = 0.01)

inline AggregationMetrics(const AggregationMetrics &other)

inline AggregationMetrics &operator=(const AggregationMetrics &other)

AggregationMetrics(AggregationMetrics&&) = default

AggregationMetrics &operator=(AggregationMetrics&&) = default

void update_duration(std::uint64_t dur, bool compute_percentiles = false)

void update_size(std::uint64_t sz, bool compute_percentiles = false)

void update_offset(std::uint64_t off, bool compute_percentiles = false)

void update_timestamp(std::uint64_t event_ts, std::uint64_t dur)

void update_timestamp_clamped(std::uint64_t event_ts, std::uint64_t dur, std::uint64_t bucket_start, std::uint64_t bucket_size)

void update_custom_metric(std::string_view name, std::uint64_t value, bool compute_percentiles = false)

void merge_from(const AggregationMetrics &other)

Public Members

std::uint64_t count

MetricStats duration

MetricStats size

MetricStats offset

std::uint64_t ts

std::uint64_t te

std::unique_ptr<std::unordered_map<std::string, std::string>> boundary_associations

std::uint64_t parent_pid

std::unique_ptr<CustomMetricsMap> custom_metrics

double sketch_accuracy

source

struct AggregationRunInput

Public Members

std::string log_dir

std::string index_dir

AggregationConfig agg_config

::dftracer::utils::PipelineConfig pipeline_config

std::optional<std::string> output_file

std::string output_format

PerfettoEventFormat event_format

bool compress_output

int compression_level

bool force_rebuild

std::size_t checkpoint_size

::dftracer::utils::Timer *stages

bool verbose

source

struct AggregationRunResult

Public Members

std::string index_path

std::size_t total_keys

std::size_t input_file_count

std::size_t processed_file_count

std::size_t cached_file_count

double elapsed_ms

source

class AggregationVisitor : public dftracer::utils::utilities::composites::dft::DftEventVisitor

Public Functions

AggregationVisitor(std::shared_ptr<rocksdb::RocksDatabase> db, std::uint32_t config_hash, AggregationConfig config, std::string file_path)

Legacy mode: flush directly to a live RocksDatabase via Merge/Put during parse. FLUSH_THRESHOLD commits the visitor’s own batch to bound memory. Used by aggregator_utility.cpp, dftracer_aggregator.cpp, dftracer_organize.cpp.

AggregationVisitor(std::string staging_dir, std::string batch_id_prefix, std::uint32_t config_hash, AggregationConfig config, std::string file_path)

Distributed mode: flush to a per-visitor SstWriterContext rooted at staging_dir. FLUSH_THRESHOLD emits partial SSTs (mixed Put+Merge) so the in-memory map never exceeds the threshold. At on_file_complete, the writer context is committed and its Artifacts are embedded in the ChunkAggregationOutput so the worker / coordinator can forward them to the main SstArtifactRegistry.

staging_dir is typically the same node-local dir the rest of the SST pipeline uses. batch_id_prefix is joined with a per-file suffix to form a unique SstWriterContext root (so concurrent per-file visitors never collide).

virtual void begin(std::size_t num_checkpoints) override

virtual void on_checkpoint(std::size_t checkpoint_idx) override

virtual void on_event(const EventRecord &record) override

virtual coro::CoroTask<void> on_file_complete() override

inline virtual bool needs_args_map() const override

ChunkAggregationOutput take_output()

void flush_to_batch(rocksdb::RocksDatabase::Batch &batch)

inline const std::unordered_set<std::string> &observed_extra_keys() const

inline const std::unordered_set<std::string> &observed_custom_metrics() const

inline std::vector<indexer::IndexDatabaseSstWriterContext::Artifacts> &aggregation_artifacts() noexcept

Distributed mode only: one or more per-flush SST artifact sets produced by this visitor after on_file_complete. Each flush emits its own SST(s) because SstFileWriter requires strictly ascending keys and merge operands for the same key across flushes would violate that invariant. Empty in legacy mode.

source

struct AggregatorInput

Public Functions

AggregatorInput &with_directory(const std::string &dir)

AggregatorInput &with_config(const AggregationConfig &cfg)

AggregatorInput &with_checkpoint_size(std::size_t sz)

AggregatorInput &with_index_dir(const std::string &dir)

AggregatorInput &with_force_rebuild(bool force)

AggregatorInput &with_parallelism(std::size_t n)

AggregatorInput &with_event_batch_size(std::size_t sz)

Public Members

std::string directory

AggregationConfig config

std::optional<common::query::Query> query

std::size_t checkpoint_size

std::string index_dir

bool force_rebuild

std::size_t parallelism

std::size_t event_batch_size

source

class AggregatorUtility : public dftracer::utils::utilities::StreamingUtility<AggregatorInput, AggregationBatch, tags::NeedsContext>

Public Functions

coro::AsyncGenerator<AggregationBatch> process(const AggregatorInput &input) override

source

struct AssociationResolverInput

Public Members

EventAggregatorOutput aggregations

std::vector<std::shared_ptr<AssociationTracker>> trackers

AggregationConfig config

source

struct AssociationResolverOutput

Public Members

EventAggregatorOutput aggregations

std::unordered_set<std::uint64_t> root_pids

std::uint64_t trace_duration

BoundaryTimeRangesMap boundary_ranges

bool success

source

class AssociationResolverUtility : public dftracer::utils::utilities::Utility<AssociationResolverInput, AssociationResolverOutput>

Public Functions

coro::CoroTask<AssociationResolverOutput> process(const AssociationResolverInput &input) override

source

class AssociationTracker

Public Functions

AssociationTracker() = default

void extract_from_event(std::string_view name, std::uint64_t pid, std::uint64_t ts, std::uint64_t dur, const ArgsMap &args, const AggregationConfig &config)

void finalize()

std::uint64_t get_parent_pid(std::uint64_t pid) const

std::unordered_map<std::string, std::string> get_boundary_associations(std::uint64_t pid, std::uint64_t ts) const

inline const std::vector<BoundaryInterval> &get_all_intervals() const

inline bool has_boundary_events() const

inline bool has_process_tree() const

std::unordered_set<std::uint64_t> get_root_pids() const

void merge(const AssociationTracker &other)

std::string serialize() const

Public Static Functions

static AssociationTracker deserialize(std::string_view data)

source

struct AugmentationConfig

Public Members

std::uint64_t source_interval_us

std::uint64_t target_interval_us

source

struct BoundaryEventConfig

Public Members

std::string event_name

std::string value_field

std::string output_name

source

struct BoundaryInterval

Public Members

std::string name

std::string value

std::uint64_t start_ts

std::uint64_t end_ts

source

struct BoundaryTimeRange

Public Members

std::uint64_t ts

std::uint64_t te

source

struct ChunkAggregationOutput

Public Members

int chunk_index

AggregationMap aggregations

AggregationMap profile_aggregations

AggregationMap system_aggregations

std::size_t events_processed

std::size_t bytes_processed

std::string file_path

bool success

std::shared_ptr<AssociationTracker> local_tracker

std::uint64_t min_time_bucket

std::uint64_t max_time_bucket

source

struct ChunkAggregatorInput

Public Functions

inline ChunkAggregatorInput &with_file_path(const std::string &path)

inline ChunkAggregatorInput &with_index_path(const std::string &path)

inline ChunkAggregatorInput &with_byte_range(std::size_t start, std::size_t end)

inline ChunkAggregatorInput &with_line_range(std::size_t start, std::size_t end)

inline ChunkAggregatorInput &with_chunk_index(int index)

inline ChunkAggregatorInput &with_config(const AggregationConfig &cfg)

inline ChunkAggregatorInput &with_checkpoint_size(std::size_t size)

inline ChunkAggregatorInput &with_batch_size(std::size_t size)

Public Members

std::string file_path

std::string index_path

std::size_t start_byte

std::size_t end_byte

std::size_t start_line

std::size_t end_line

AggregationConfig config

std::optional<common::query::Query> query

std::size_t checkpoint_size

int chunk_index

std::size_t batch_size

source

class ChunkAggregatorUtility : public dftracer::utils::utilities::Utility<ChunkAggregatorInput, ChunkAggregationOutput>

Public Functions

ChunkAggregatorUtility() = default

coro::CoroTask<ChunkAggregationOutput> process(const ChunkAggregatorInput &input) override

source

struct CountConfidenceInterval

Public Members

double lower

double upper

source

struct DeserializedAggKey

Public Members

std::uint32_t config_hash

AggMapType map_type

AggregationKey key

source

struct DeserializedSystemKey

Public Members

SystemMetricKey key

source

struct DfanalyzerContext

Context for converting aggregation data to dfanalyzer-compatible Arrow format.

Public Members

const std::unordered_map<std::string, std::string> *file_hashes

Hash tables for resolving fhash/hhash to file_name/host_name.

const std::unordered_map<std::string, std::string> *host_hashes

const common::query::Query *query_filter

std::uint64_t time_origin

Time origin (minimum time_bucket) for normalization.

double time_resolution

Time resolution (microseconds per output unit, default 1e6 = seconds).

double time_granularity

Time granularity in seconds (bucket width for time_range computation).

source

class EventAggregator

Public Types

using ScanCallback = std::function<bool(AggMapType, const AggregationKey&, AggregationMetrics&)>

using RawScanCallbackFn = bool (*)(void *ctx, std::string_view key_bytes, std::string_view value_bytes)

Type-erased raw scan. Use the templated overload below for zero- allocation calls.

Public Functions

EventAggregator()

EventAggregator(std::shared_ptr<rocksdb::RocksDatabase> db, std::uint32_t config_hash)

void merge_chunk(ChunkAggregationOutput &&chunk_output)

EventAggregatorOutput finalize()

std::size_t scan(ScanCallback callback) const

std::size_t scan_shard_range(std::uint16_t shard_begin, std::uint16_t shard_end, ScanCallback callback) const

std::size_t scan_shard_range_raw_fn(std::uint16_t shard_begin, std::uint16_t shard_end, RawScanCallbackFn fn, void *ctx) const

std::size_t scan_system_metrics_raw_fn(RawScanCallbackFn fn, void *ctx) const

Sequential scan of the SYSTEM_METRICS CF (no shard prefix in keys).

template<typename F>
inline std::size_t scan_system_metrics_raw(F &&callback) const

template<typename F>
inline std::size_t scan_shard_range_raw(std::uint16_t shard_begin, std::uint16_t shard_end, F &&callback) const

Template wrapper: forwards any callable (sv, sv) -> bool into the raw scan with zero heap allocations. The adapter lambda is a captureless +[] so it decays to a plain function pointer.

std::vector<std::shared_ptr<AssociationTracker>> take_trackers()

Move trackers out without materializing the full aggregation map.

std::unique_ptr<AssociationTracker> build_global_tracker()

Merge fresh trackers with any persisted tracker from the DB, persist the result, and return the merged tracker.

ObservedColumns observed_columns()

void add_observed_extra_key(const std::string &key)

void add_observed_custom_metric(const std::string &name)

inline std::size_t total_events() const

inline std::size_t total_bytes() const

inline std::size_t total_files() const

void update_time_bounds(std::uint64_t time_bucket)

std::uint64_t min_time_bucket() const

std::uint64_t max_time_bucket() const

TimeBoundsResult query_time_bounds() const

void persist_time_bounds()

Persist the in-memory min/max time bucket to the AGGREGATION CF so a later read-only reopen can recover the trace origin. finalize() does this too; the SST build path needs it called explicitly after merge_chunk().

inline bool is_rocksdb_mode() const

inline std::shared_ptr<rocksdb::RocksDatabase> db() const

inline std::uint32_t config_hash() const

Public Static Functions

static std::shared_ptr<rocksdb::RocksDatabase> open_with_merge_operator(const std::string &index_path)

static std::shared_ptr<rocksdb::RocksDatabase> open_read_only_with_merge_operator(const std::string &index_path)

Read-only variant for multi-process concurrent scan (e.g. MPI ranks covering disjoint shard-prefix ranges of a shared unified index). Multi-process RocksDB writes are forbidden; read-only opens do not hold the exclusive LOCK, so N ranks can open the same DB at once.

struct ObservedColumns

Public Members

std::vector<std::uint32_t> extra_key_ids

std::vector<std::string> custom_metric_names

struct TimeBoundsResult

Public Members

std::uint64_t min_time_bucket

std::uint64_t max_time_bucket

bool valid

source

struct EventAggregatorOutput

Public Members

AggregationMap aggregations

AggregationMap profile_aggregations

AggregationMap system_aggregations

std::size_t total_events_processed

std::size_t total_files_processed

std::size_t total_bytes_processed

std::vector<std::shared_ptr<AssociationTracker>> trackers

bool success

source

struct FileChunkMapperInput

Public Functions

inline FileChunkMapperInput &with_config(const AggregationConfig &cfg)

inline FileChunkMapperInput &with_checkpoint_size(std::size_t size)

inline FileChunkMapperInput &with_target_chunk_size(std::size_t size_mb)

inline FileChunkMapperInput &with_batch_size(std::size_t size_bytes)

inline FileChunkMapperInput &with_start_chunk_index(int index)

Public Members

utilities::composites::dft::MetadataCollectorUtilityOutput metadata

AggregationConfig config

std::optional<common::query::Query> query

std::size_t checkpoint_size

std::size_t target_chunk_size_mb

std::size_t batch_size

int start_chunk_index

Public Static Functions

static inline FileChunkMapperInput from_metadata(const utilities::composites::dft::MetadataCollectorUtilityOutput &meta)

source

class FileChunkMapperUtility : public dftracer::utils::utilities::Utility<FileChunkMapperInput, FileChunkMapperOutput>

Public Functions

coro::CoroTask<FileChunkMapperOutput> process(const FileChunkMapperInput &input) override

source

struct FloatMetricStats

Public Functions

inline explicit FloatMetricStats(double relative_accuracy = 0.01)

inline FloatMetricStats(const FloatMetricStats &other)

inline FloatMetricStats &operator=(const FloatMetricStats &other)

FloatMetricStats(FloatMetricStats&&) = default

FloatMetricStats &operator=(FloatMetricStats&&) = default

inline void update(double value, bool compute_percentiles = false)

inline void merge_from(const FloatMetricStats &other)

inline double get_stddev() const

Public Members

std::uint64_t count

double total

double min

double max

double mean

double m2

std::unique_ptr<DDSketch> sketch

double sketch_accuracy_

source

struct MetricStats

Public Functions

inline explicit MetricStats(double relative_accuracy = 0.01)

inline MetricStats(const MetricStats &other)

inline MetricStats &operator=(const MetricStats &other)

MetricStats(MetricStats&&) = default

MetricStats &operator=(MetricStats&&) = default

void update(std::uint64_t value, bool compute_percentiles = false)

void merge_from(const MetricStats &other)

double get_stddev() const

double get_skewness() const

double get_kurtosis() const

Public Members

std::uint64_t count

std::uint64_t total

std::uint64_t min

std::uint64_t max

double mean

double m2

double m3

double m4

std::unique_ptr<DDSketch> sketch

double sketch_accuracy_

source

struct PerfettoTraceWriterInput

Public Members

std::string output_path

const EventAggregator *aggregator

const AssociationTracker *tracker

const AggregationConfig *agg_config

std::unique_ptr<AssociationTracker> owned_tracker

std::unordered_set<std::uint64_t> root_pids

std::uint64_t trace_duration

BoundaryTimeRangesMap boundary_ranges

bool compute_statistics

bool compute_percentiles

std::vector<double> percentiles

bool compress

int compression_level

PerfettoEventFormat format

std::atomic<std::size_t> *keys_written

Workers add their per-shard key count here if non-null.

bool merge_on_sharded

Concatenate shards into output_path and unlink them on SHARDED layouts (typically NFS). Callers that read shards directly leave false.

std::uint16_t shard_begin

Total shard-prefix range (half-open) this invocation is responsible for. Defaults cover the whole key space. MPI drivers set a disjoint range per rank so N ranks collectively cover [0, AGG_KEY_NUM_SHARDS) without overlap. Local coroutine workers within a single process further subdivide this range.

std::uint16_t shard_end

bool emit_header

Emit the JSON array prologue ([\n + trace_metadata + root_process markers) to this invocation’s output. MPI drivers set this on rank 0 only so concatenated rank outputs produce exactly one array open.

bool emit_footer

Emit the JSON array epilogue (]). MPI drivers set this on the last rank only.

source

class PerfettoTraceWriterUtility : public dftracer::utils::utilities::Utility<PerfettoTraceWriterInput, PerfettoTraceWriterOutput, utilities::tags::NeedsContext>

Public Functions

coro::CoroTask<bool> process(const PerfettoTraceWriterInput &input) override

source

struct SystemAggregationMetrics

Public Functions

inline explicit SystemAggregationMetrics(double relative_accuracy = 0.01)

inline SystemAggregationMetrics(const SystemAggregationMetrics &other)

inline SystemAggregationMetrics &operator=(const SystemAggregationMetrics &other)

SystemAggregationMetrics(SystemAggregationMetrics&&) = default

SystemAggregationMetrics &operator=(SystemAggregationMetrics&&) = default

inline void update_metric(std::string_view name, double value, bool compute_percentiles = false)

inline void update_timestamp(std::uint64_t event_ts)

inline void merge_from(const SystemAggregationMetrics &other)

Public Members

std::uint64_t count

std::uint64_t ts

std::uint64_t te

std::unique_ptr<FloatMetricsMap> metrics

double sketch_accuracy

source

struct SystemMetricKey

Public Members

std::string hhash

std::string name

std::uint64_t time_bucket

source

class SystemMetricsMergeOperator : public rocksdb::MergeOperator

Public Functions

bool FullMergeV2(const MergeOperationInput &merge_in, MergeOperationOutput *merge_out) const override

bool PartialMerge(const ::rocksdb::Slice &key, const ::rocksdb::Slice &left_operand, const ::rocksdb::Slice &right_operand, std::string *new_value, ::rocksdb::Logger *logger) const override

inline const char *Name() const override

Free Functions

source

inline StringIntern &dftracer::utils::utilities::composites::dft::aggregators::aggregation_intern()

source

AggregationBatch dftracer::utils::utilities::composites::dft::aggregators::augment_batch(const AggregationBatch &input, const AugmentationConfig &config)

source

AggregationKey dftracer::utils::utilities::composites::dft::aggregators::build_aggregation_key(const DFTracerEvent &ev, const AggregationConfig &config)

source

inline CountConfidenceInterval dftracer::utils::utilities::composites::dft::aggregators::compute_poisson_ci(double count, double confidence = 1.96)

source

std::uint64_t dftracer::utils::utilities::composites::dft::aggregators::compute_time_bucket(std::uint64_t timestamp, std::uint64_t duration, const AggregationConfig &config)

source

inline std::uint64_t dftracer::utils::utilities::composites::dft::aggregators::decode_varint(const std::uint8_t *&p, const std::uint8_t *end)

Decode a LEB128 varint, advancing p (bounded by end).

source

inline AggGlobalConfig dftracer::utils::utilities::composites::dft::aggregators::deserialize_agg_global_config(std::string_view data)

source

DeserializedAggKey dftracer::utils::utilities::composites::dft::aggregators::deserialize_agg_key(std::string_view data)

source

AggregationMetrics dftracer::utils::utilities::composites::dft::aggregators::deserialize_agg_value(std::string_view data)

source

DeserializedSystemKey dftracer::utils::utilities::composites::dft::aggregators::deserialize_system_key(std::string_view data)

source

SystemAggregationMetrics dftracer::utils::utilities::composites::dft::aggregators::deserialize_system_value(std::string_view data)

source

template<typename Map>
Map::mapped_type &dftracer::utils::utilities::composites::dft::aggregators::find_or_create(Map &map, std::string_view name, double accuracy)

source

void dftracer::utils::utilities::composites::dft::aggregators::flush_intern_dictionary(dftracer::utils::rocksdb::RocksDatabase &db, dftracer::utils::rocksdb::RocksDatabase::Batch &batch)

Flush any new intern entries to RocksDB as 0xFFFD keys.

source

void dftracer::utils::utilities::composites::dft::aggregators::flush_intern_dictionary(dftracer::utils::utilities::indexer::IndexBatchSink &sink)

Sink-backed overload: flushes new intern entries via IndexBatchSink::insert_aggregation_put. Used by the distributed SST pipeline where the visitor writes to an SST instead of a live DB.

source

template<typename Metrics, typename Deserialize, typename Serialize>
bool dftracer::utils::utilities::composites::dft::aggregators::full_merge_metrics(const ::rocksdb::MergeOperator::MergeOperationInput &merge_in, ::rocksdb::MergeOperator::MergeOperationOutput *merge_out, Deserialize &&deserialize, Serialize &&serialize)

source

inline bool dftracer::utils::utilities::composites::dft::aggregators::is_preagg_suffix(std::string_view k)

source

inline bool dftracer::utils::utilities::composites::dft::aggregators::is_reserved_arg(std::string_view k)

source

void dftracer::utils::utilities::composites::dft::aggregators::load_intern_dictionary(dftracer::utils::rocksdb::RocksDatabase &db)

Load intern dictionary from RocksDB into aggregation_intern().

source

inline std::string dftracer::utils::utilities::composites::dft::aggregators::make_agg_file_key(std::int32_t file_id)

source

std::vector<std::string> dftracer::utils::utilities::composites::dft::aggregators::merge_aggregation_visitors(std::vector<std::vector<std::unique_ptr<DftEventVisitor>>> &extra_visitors, EventAggregator *merger)

source

inline bool dftracer::utils::utilities::composites::dft::aggregators::parse_agg_key_view(std::string_view data, AggKeyView &out)

Parse aggregation key: reads varint intern IDs and resolves to strings. Returns false if parsing fails.

source

inline bool dftracer::utils::utilities::composites::dft::aggregators::parse_agg_value_full_view(std::string_view data, AggMetricsFullView &out)

Full value parser for iter_aggregation - includes mean/m2 for stddev.

source

inline bool dftracer::utils::utilities::composites::dft::aggregators::parse_agg_value_view(std::string_view data, AggMetricsView &out)

Fast value parser for Arrow export - skips mean/m2/m3/m4/sketch.

source

template<typename Metrics, typename Deserialize, typename Serialize>
bool dftracer::utils::utilities::composites::dft::aggregators::partial_merge_metrics(const ::rocksdb::Slice &left_operand, const ::rocksdb::Slice &right_operand, std::string *new_value, Deserialize &&deserialize, Serialize &&serialize)

source

coro::CoroTask<Result<AggregationRunResult>> dftracer::utils::utilities::composites::dft::aggregators::run_aggregation(AggregationRunInput input)

source

inline std::string dftracer::utils::utilities::composites::dft::aggregators::serialize_agg_global_config(const AggGlobalConfig &cfg)

source

std::string dftracer::utils::utilities::composites::dft::aggregators::serialize_agg_key(std::uint32_t config_hash, AggMapType map_type, const AggregationKey &key)

source

void dftracer::utils::utilities::composites::dft::aggregators::serialize_agg_key_into(std::string &out, std::uint32_t config_hash, AggMapType map_type, const AggregationKey &key)

source

void dftracer::utils::utilities::composites::dft::aggregators::serialize_agg_key_into(std::string &out, std::uint32_t config_hash, AggMapType map_type, std::string_view cat, std::string_view name, std::uint64_t pid, std::uint64_t tid, std::string_view hhash, std::string_view fhash, std::uint64_t time_bucket, const std::vector<std::pair<std::string_view, std::string_view>> *extra_keys = nullptr)

source

std::string dftracer::utils::utilities::composites::dft::aggregators::serialize_agg_value(const AggregationMetrics &metrics)

source

void dftracer::utils::utilities::composites::dft::aggregators::serialize_agg_value_into(std::string &out, const AggregationMetrics &metrics)

source

std::string dftracer::utils::utilities::composites::dft::aggregators::serialize_system_key(std::string_view hhash, std::string_view name, std::uint64_t time_bucket)

source

void dftracer::utils::utilities::composites::dft::aggregators::serialize_system_key_into(std::string &out, std::string_view hhash, std::string_view name, std::uint64_t time_bucket)

source

std::string dftracer::utils::utilities::composites::dft::aggregators::serialize_system_value(const SystemAggregationMetrics &metrics)

source

void dftracer::utils::utilities::composites::dft::aggregators::serialize_system_value_into(std::string &out, const SystemAggregationMetrics &metrics)

source

void dftracer::utils::utilities::composites::dft::aggregators::update_aggregation_entry(const DFTracerEvent &ev, const AggregationConfig &config, AggregationMap &aggregations, const AggregationKey &key)

source

void dftracer::utils::utilities::composites::dft::aggregators::write_aggregation_tracking(::dftracer::utils::rocksdb::RocksDatabase *db, const AggregationConfig &config, const std::vector<std::string> &processed_files, const std::string &index_path, std::uint32_t config_hash)