// Parse and represent the JSON encoding of the profile information generated by Feldera.

import { fail, enforceNumber, Option, OMap, type Comparable, assert, NumericRange, tuple } from "./util.js";
import { type MirNode, SourcePositionRanges, SourcePositionRange, Sources, type Dataflow } from "./dataflow.js";

type JsonMeasurement = Array<any>;
export type NodeId = string;

export type CircuitMetricCategory = string;

export interface ProfileMetricDescription {
    readonly name: string;
    readonly category: CircuitMetricCategory;
    readonly advanced: boolean;
    readonly description: string;
}

// Base class for all metric representations in the new profile representation
interface JsonMetricBase {
    metric_id: string;
    labels?: Array<[string, string]>;
    value: JsonMetricValue;
}

// A metric value as it appears in the profile, serialized from Rust as JSON
interface MetricValue { };

// Base types
interface PersistentIdMetricValue extends MetricValue {
    type: "string";
    value: string;
}

interface IntMetricValue extends MetricValue {
    type: "int";
    value: number;
}

interface CountMetricValue extends MetricValue {
    type: "count";
    value: number;
}

interface DurationMetricValue {
    type: "duration";
    value: {
        secs: number;
        nanos: number;
    }
}

interface BytesMetricValue {
    type: "bytes";
    value: number;
}

interface PercentMetricValue {
    type: "percent";
    value: {
        numerator: number;
        denominator: number;
    }
}

// Composite types
interface StatsMetricValue extends MetricValue {
    avg_records_count: CountMetricValue;
    batches_count: CountMetricValue;
    max_records_count: CountMetricValue;
    min_records_count: CountMetricValue;
    total_records_count: CountMetricValue;
}

interface MergesMetricValue extends MetricValue {
    avg_step_time: DurationMetricValue;
    batches: CountMetricValue;
    merges: CountMetricValue;
    steps: CountMetricValue;
}

interface SerializedStringValue {
    type: "string";
    value: string;
}

interface RetainmentValue extends MetricValue {
    key: SerializedStringValue | Array<SerializedStringValue>;
    value: SerializedStringValue | Array<SerializedStringValue>;
}

interface DistributionValue {
    values: Array<number>;
}

interface OccupancyValue extends MetricValue {
    max: BytesMetricValue;
    used: BytesMetricValue;
}

interface CacheStatsValue extends MetricValue {
    type: "cachecounts";
    value: {
        count: number;
        bytes: number;
        elapsed: {
            secs: number;
            nanos: number;
        };
    }
}

type JsonMetricValue =
    CountMetricValue |
    BytesMetricValue |
    PercentMetricValue |
    DurationMetricValue |
    StatsMetricValue |
    PersistentIdMetricValue |
    DistributionValue |
    RetainmentValue |
    IntMetricValue |
    MergesMetricValue |
    OccupancyValue |
    CacheStatsValue;

// Serialized representation of the measurements for a node.
type JsonProfileEntries = Array<JsonMetricBase>

interface LegacyJsonProfileEntries {
    entries: Array<JsonMeasurement>;
}

// Serialized representation of a single profile.
interface JsonWorkerProfile {
    // Map node name to measurements.
    readonly metadata: Record<NodeId, JsonProfileEntries>;
}

interface LegacyJsonWorkerProfile {
    // Map node name to measurements.
    readonly metadata: Record<NodeId, LegacyJsonProfileEntries>;
}

// Serialized representation of a simple circuit node.
interface JsonSimpleCircuitNode {
    id: NodeId;
    label: string;
}

interface JsonSimpleNodeWrapper {
    Simple: JsonSimpleCircuitNode;
}

interface JsonClusterWrapper {
    Cluster: JsonCircuitCluster;
}

// Serialized representation of a cluster of nodes.
interface JsonCircuitCluster {
    id: NodeId;
    label: string;
    nodes: Array<JsonSimpleNodeWrapper | JsonClusterWrapper>;
}

// Serialized representation of an edge.
interface JsonCircuitEdge {
    from_node: NodeId;
    to_node: NodeId;
    from_cluster: boolean;
    to_cluster: boolean;
}

// Serialized representation of a circuit graph.
interface JsonGraph {
    readonly nodes: JsonCircuitCluster;
    readonly edges: Array<JsonCircuitEdge>;
}

// Serialized JSON representation for the
// profiles received from the Feldera pipeline manager.
export type JsonProfiles = {
    // The "metrics" property only exists in new profiles
    readonly metrics: Array<ProfileMetricDescription>;
    readonly worker_profiles: Array<JsonWorkerProfile>;
    readonly graph: JsonGraph;
} | {
    readonly worker_profiles: Array<LegacyJsonWorkerProfile>;
    readonly graph: JsonGraph;
}

///////////////// Parsing code to decode measurements

function formatValue(val: number, base: number, prefixes: Array<string>): string {
    let index = 0;
    while ((Math.abs(val) >= base) && (index < (prefixes.length - 1))) {
        val = val / base;
        index++;
    }
    return val.toLocaleString('en-US', { minimumFractionDigits: 0, maximumFractionDigits: 2 }) + prefixes[index];
}

/** Base class for (numeric) property values; representation which abstracts away from the
 * JSON Serialization. */
export abstract class PropertyValue implements Comparable<PropertyValue> {
    abstract getNumericValue(): Option<number>;

    getStringValue(): string {
        return this.toString();
    }

    compareTo(other: PropertyValue): number {
        let v1 = this.getNumericValue();
        let v2 = other.getNumericValue();
        if (v1.isSome() && v2.isSome()) {
            return v1.unwrap() - v2.unwrap();
        } else if (v1.isSome()) {
            return 1;
        } else if (v2.isSome()) {
            return -1;
        } else {
            return 0;
        }
    }

    max(other: PropertyValue): PropertyValue {
        if (this.compareTo(other) >= 0) {
            return this;
        } else {
            return other;
        }
    }

    toString(): string {
        let v = this.getNumericValue();
        if (v.isSome()) {
            let value = v.unwrap();
            let prefixes = ["", "K", "M", "B"];
            return formatValue(value, 1000, prefixes);
        } else {
            return "N/A";
        }
    }

    /** Combine values from multiple operators; the semantics depends on the value kind. */
    abstract combine(other: PropertyValue): PropertyValue;
}

/** A property value represented as a numerator and denominator, which are supposed to represent a percentage. */
class PercentValue extends PropertyValue {
    readonly numerator: number;
    readonly denominator: number;

    constructor(numerator: any, denominator: any) {
        super();
        this.numerator = enforceNumber(numerator);
        this.denominator = enforceNumber(denominator);
    }

    static fromNumbers(numerator: number, denominator: number): PercentValue {
        return new PercentValue(numerator, denominator);
    }

    static fromPercentMetric(s: PercentMetricValue) {
        return new PercentValue(s.value.numerator, s.value.denominator);
    }

    getNumericValue(): Option<number> {
        if (this.denominator === 0) {
            return Option.some(0);
        }
        return Option.some(100 * this.numerator / this.denominator);
    }

    override toString(): string {
        let v = this.getNumericValue();
        if (v.isSome()) {
            return v.unwrap().toFixed(1) + "%";
        } else {
            return "N/A";
        }
    }

    override combine(other: PropertyValue): PropertyValue {
        if (other instanceof MissingValue) {
            return this;
        }
        if (other instanceof PercentValue) {
            if (this.denominator === other.denominator)
                return new PercentValue(this.numerator + other.numerator, this.denominator);
            return new PercentValue(
                this.numerator * other.denominator + other.numerator * this.denominator,
                this.denominator * other.denominator)
        }
        throw new Error("Cannot add PercentValue to " + other);
    }
}

/** A property value that is a simple count. */
class CountValue extends PropertyValue {
    readonly value: number;

    constructor(value: any) {
        super();
        this.value = enforceNumber(value);
    }

    static fromNumber(value: number): CountValue {
        return new CountValue(value);
    }

    static fromCountMetric(value: CountMetricValue): CountValue {
        return new CountValue(value.value);
    }

    getNumericValue(): Option<number> {
        return Option.some(this.value);
    }

    override combine(other: PropertyValue): PropertyValue {
        if (other instanceof MissingValue) {
            return this;
        }
        if (other instanceof CountValue) {
            return new CountValue(this.value + other.value);
        }
        throw new Error("Cannot add CountValue to " + other);
    }
}

class BytesValue extends PropertyValue {
    readonly value: number;

    constructor(value: any) {
        super();
        this.value = enforceNumber(value);
    }

    static fromNumber(value: number): BytesValue {
        return new BytesValue(value);
    }

    static fromBytesMetric(value: BytesMetricValue): BytesValue {
        return new BytesValue(value.value);
    }

    getNumericValue(): Option<number> {
        return Option.some(this.value);
    }

    override combine(other: PropertyValue): PropertyValue {
        if (other instanceof MissingValue) {
            return this;
        }
        if (other instanceof BytesValue) {
            return new BytesValue(this.value + other.value);
        }
        throw new Error("Cannot add BytesValue to " + other);
    }

    override toString(): string {
        let prefixes = ["B", "KiB", "MiB", "GiB", "TiB"];
        return formatValue(this.value, 1024, prefixes);
    }
}

/** A property value that is a Boolean. */
class BooleanValue extends PropertyValue {
    readonly value: boolean;

    constructor(id: any) {
        super();
        if (typeof id !== "boolean") {
            throw new TypeError(`Expected a valid boolean, got, ${id}`);
        }
        this.value = id;
    }

    getNumericValue(): Option<number> {
        return this.value ? Option.some(0) : Option.some(1);
    }

    override compareTo(other: PropertyValue): number {
        let v1 = this.value;
        if (other instanceof BooleanValue) {
            let v2 = other.value;
            if (v1 < v2) return -1;
            if (v1 > v2) return 1;
            return 0;
        }
        return super.compareTo(other);
    }

    override combine(other: PropertyValue): PropertyValue {
        if (other instanceof MissingValue) {
            return this;
        }
        if (other instanceof BooleanValue) {
            return new BooleanValue(this.value || other.value);
        }
        throw new Error("Cannot add BooleanValue to " + other);
    }

    override getStringValue(): string {
        return this.value.toString();
    }

    override toString(): string {
        return this.value.toString();
    }
}

/** A property value that is a string, with no numeric value. */
class StringValue extends PropertyValue {
    readonly value: string;

    constructor(id: any) {
        super();
        if (typeof id !== "string") {
            throw new TypeError(`Expected a valid string, got, ${id}`);
        }
        this.value = id;
    }

    static fromString(value: string): StringValue {
        return new StringValue(value);
    }

    getNumericValue(): Option<number> {
        return Option.none();
    }

    override compareTo(other: PropertyValue): number {
        let v1 = this.value;
        if (other instanceof StringValue) {
            let v2 = other.value;
            if (v1 < v2) return -1;
            if (v1 > v2) return 1;
            return 0;
        }
        return super.compareTo(other);
    }

    override combine(other: PropertyValue): PropertyValue {
        if (other instanceof MissingValue) {
            return this;
        }
        if (other instanceof StringValue) {
            if (this.value === other.value) {
                return this;
            }
            return new StringValue("<multiple values>");
        }
        throw new Error("Cannot add StringValue to " + other);
    }

    override getStringValue(): string {
        return this.value;
    }

    override toString(): string {
        return this.value;
    }
}

/** No value has been provided for a property. */
export class MissingValue extends PropertyValue {
    public static readonly INSTANCE: MissingValue = new MissingValue();

    constructor() {
        super();
    }

    override getNumericValue(): Option<number> {
        return Option.none();
    }

    override combine(other: PropertyValue): PropertyValue {
        return other;
    }
}

/** A property value that represents a time with seconds and nanoseconds. */
class TimeValue extends PropertyValue {
    constructor(readonly seconds: number) {
        super();
    }

    static fromSecondsNanos(secs: any, nanos: any) {
        return new TimeValue(enforceNumber(secs) + enforceNumber(nanos) / 1_000_000_000);
    }

    static fromDurationMetric(s: DurationMetricValue): TimeValue {
        return this.fromSecondsNanos(s.value.secs, s.value.nanos);
    }

    getNumericValue(): Option<number> {
        return Option.some(this.seconds);
    }

    override combine(other: PropertyValue): PropertyValue {
        if (other instanceof MissingValue) {
            return this;
        }
        if (other instanceof TimeValue) {
            return new TimeValue(this.seconds + other.seconds);
        }
        throw new Error("Cannot add TimeValue to " + other);
    }

    override toString(): string {
        let v = this.getNumericValue();
        if (v.isSome()) {
            let value = v.unwrap();
            if (value === 0) {
                return "0s";
            } else if (value < 0.001) {
                value = value * 1000_000;
                return value.toLocaleString('en-US', { maximumFractionDigits: 2 }) + "us";
            } else if (value < 1) {
                value = value * 1000;
                return value.toLocaleString('en-US', { maximumFractionDigits: 2 }) + "ms";
            } else if (value >= 3600) {
                // Discard sub-second part
                let seconds = Math.floor(value);
                let days = Math.floor(seconds / 86400);
                let inDay = seconds - days * 86400;
                let hours = Math.floor(inDay / 3600);
                let minutes = Math.floor((seconds % 3600) / 60);
                let secs = seconds % 60;
                let result = "";
                if (days > 0) {
                    result += days + "days ";
                }
                result += String(hours).padStart(2, "0") + ":" +
                    String(minutes).padStart(2, "0") + ":" +
                    String(secs).padStart(2, "0");
            }
            return value.toLocaleString('en-US', { maximumFractionDigits: 2 }) + "s";
        } else {
            return "N/A";
        }
    }
}

// This function should be deleted, it is here only to temporarily
// support visualizing profiles generated by old versions of the runtime.
function legacyMeasurementCategory(prop: string): string {
    const map: Map<string, Array<string>> = new Map();
    map.set("CPU", [
        "time%",
        "invocations",
        "steps",
        "wait_time",
        "exchange_wait_time",
        "merge backpressure wait",
        "time",
        "total_idle_time",
        "runtime_elapsed",
        "total_runtime",
        "total rebalancing time",
        "in-progress rebalancing time",
        "integral records to repartition",
        "accumulator records to repartition",
        "local shard size",
        "rebalancings"]);
    map.set("storage", [
        "merge reduction",
        "output redundancy",
        "merging batches",
        "merging size",
        "input batches/batches",
        "input batches/min size",
        "input batches/max size",
        "input batches/avg size",
        "input batches/total records",
        "output batches/batches",
        "output batches/min size",
        "output batches/max size",
        "output batches/avg size",
        "output batches/total records",
        "slot 0 loose",
        "slot 1 loose",
        "slot 2 loose",
        "slot 3 loose",
        "slot 4 loose",
        "slot 0 completed",
        "slot 1 completed",
        "slot 2 completed",
        "slot 3 completed",
        "slot 4 completed",
        "slot 0 merging",
        "slot 1 merging",
        "slot 2 merging",
        "slot 3 merging",
        "slot 4 merging",
        "storage size",
        "batch sizes",
        "bounds",
        "Bloom filter size",
        "Bloom filter bits/key",
        "Bloom filter hits",
        "Bloom filter misses",
        "Bloom filter hit rate"]);
    map.set("cache", [
        "background cache hit",
        "foreground cache hit",
        "background cache miss",
        "foreground cache miss",
        "foreground cache occupancy",
        "background cache occupancy"]);
    map.set("memory", [
        "left inputs",
        "right inputs",
        "computed outputs",
        "inputs",
        "total size",
        "allocated bytes",
        "used bytes",
        "shared bytes",
        "batches",
        "allocations"]);
    for (const [k, v] of map) {
        for (const value of v) {
            if (prop.startsWith(value)) {
                return k;
            }
        }
    }
    return "";
}

export class MetricDescriptions {
    private static KEY_ORDER: Map<string, number> = new Map();
    private static METRIC_DESCRIPTIONS: Map<string, ProfileMetricDescription> = new Map();

    static known(): boolean {
        return MetricDescriptions.METRIC_DESCRIPTIONS.size > 0;
    }

    /** Get the ordinal number of the metric in the metadata array (coming from the profile,
     * where it comes from the metadata.rs file).  Note that the metric string here
     * can be a composite name, e.g., output_batches_stats.record_count; only the part
     * before the dot is the actual metric name. */
    static getMetricIndex(metric: string): number {
        let dot = metric.indexOf(".");
        if (dot >= 0)
            metric = metric.substring(0, dot);
        let result = MetricDescriptions.KEY_ORDER.get(metric);
        if (result !== undefined)
            return result;
        return MetricDescriptions.KEY_ORDER.size;
    }

    static setMetricDescriptions(descriptions: Array<ProfileMetricDescription>) {
        let index = 0;
        for (const pmd of descriptions) {
            MetricDescriptions.KEY_ORDER.set(pmd.name, index);
            MetricDescriptions.METRIC_DESCRIPTIONS.set(pmd.name, pmd);
            index++;
        }
    }

    static getMetricInfo(metric: string): ProfileMetricDescription | undefined {
        return MetricDescriptions.METRIC_DESCRIPTIONS.get(metric);
    }

    static getMetricDescription(metric: string): string | undefined {
        return MetricDescriptions.getMetricInfo(metric)?.description;
    }
}

function stripSuffix(prop: string): string {
    let dot = prop.indexOf(".");
    if (dot >= 0) {
        // Dot is used when appending measurements for a metric with multiple values
        prop = prop.substring(0, dot);
    }
    return prop;
}

/** Which category does a measurement belong to? */
export function measurementCategory(prop: string): string {
    let info = MetricDescriptions.getMetricInfo(stripSuffix(prop));
    if (info !== undefined) {
        return info.category;
    }

    // If we couldn't find the description, treat as if in an old-style profile
    return legacyMeasurementCategory(prop);
}

/** Return a description of the measurement and a boolean indicating whether the measurements is "advanced" */
export function measurementDescription(prop: string): { description: string, advanced: boolean } {
    let info = MetricDescriptions.getMetricInfo(stripSuffix(prop));
    if (info !== undefined) {
        return { description: info.description, advanced: info.advanced };
    }

    return { description: "", advanced: false };
}

// Decoded measurement value.
export class Measurement {
    constructor(
        readonly property: string,
        readonly value: Option<PropertyValue>) { };

    public static fromJson(prefix: string, json: JsonMeasurement): Measurement {
        let name = prefix + (json[0] as string);
        return new Measurement(
            name,
            Measurement.parseLegacyPropertyValue(name, json.slice(1))
        );
    }

    toString(): string {
        return this.property + ", " + this.value.toString();
    }

    static parseValues(metric: JsonMetricBase): Array<Measurement> {
        let metric_id = metric.metric_id;
        let kind = metric_id.split("_").pop(); // count, bytes, etc
        if (metric.labels) {
           let labels = metric.labels.map(e => e.join(":")).join(".");
           if (metric.labels.length !== 0) {
               metric_id = metric_id + "." + labels;
           }
        }
        switch (kind) {
            case "bytes": {
                let m = metric.value as BytesMetricValue;
                let bytes = BytesValue.fromBytesMetric(m);
                return [new Measurement(metric_id, Option.some(bytes))];
            }
            case "count": {
                let m = metric.value as CountMetricValue;
                let count = CountValue.fromCountMetric(m);
                return [new Measurement(metric_id, Option.some(count))];
            }
            case "hits":
            case "misses": {
                let s = metric.value as CacheStatsValue;
                let count = new CountValue(s.value.count);
                let bytes = new BytesValue(s.value.bytes);
                let elapsed = TimeValue.fromSecondsNanos(s.value.elapsed.secs, s.value.elapsed.nanos);
                let avg_latency = new TimeValue(count.value === 0 ? 0 : elapsed.seconds / count.value);
                return [
                    new Measurement(metric_id + ".count", Option.some(count)),
                    new Measurement(metric_id + ".bytes", Option.some(bytes)),
                    new Measurement(metric_id + ".elapsed", Option.some(elapsed)),
                    new Measurement(metric_id + ".avg_latency", Option.some(avg_latency))
                ];
            }
            case "stats": {
                let s = metric.value as StatsMetricValue;
                let count = CountValue.fromCountMetric(s.batches_count);
                let avg_size = CountValue.fromCountMetric(s.avg_records_count);
                let max_size = CountValue.fromCountMetric(s.max_records_count);
                let min_size = CountValue.fromCountMetric(s.min_records_count);
                let record_count = CountValue.fromCountMetric(s.total_records_count);
                return [
                    new Measurement(metric_id + ".count", Option.some(count)),
                    new Measurement(metric_id + ".avg_size", Option.some(avg_size)),
                    new Measurement(metric_id + ".max_size", Option.some(max_size)),
                    new Measurement(metric_id + ".min_size", Option.some(min_size)),
                    new Measurement(metric_id + ".record_count", Option.some(record_count))
                ];
            }
            case "id": { // persistent_id
                let s = metric.value as PersistentIdMetricValue;
                let str = StringValue.fromString(s.value);
                return [new Measurement(metric_id, Option.some(str))];
            }
            case "percent": {
                let s = metric.value as PercentMetricValue;
                let perc = PercentValue.fromPercentMetric(s);
                return [new Measurement(metric_id, Option.some(perc))];
            }
            case "seconds": {
                let s = metric.value as DurationMetricValue;
                let duration = TimeValue.fromDurationMetric(s);
                return [new Measurement(metric_id, Option.some(duration))];
            }
            case "occupancy": {
                let s = metric.value as OccupancyValue;
                let max = BytesValue.fromBytesMetric(s.max);
                let used = BytesValue.fromBytesMetric(s.used);
                return [
                    new Measurement(metric_id + ".max", Option.some(max)),
                    new Measurement(metric_id + ".used", Option.some(used))
                ];
            }
            case "bounds": {
                // retainment_bounds
                let s = metric.value as RetainmentValue;
                let object: Array<string> = [];
                let result = [];
                if (s.key instanceof Array) {
                    for (const keyVal of s.key) {
                        if (keyVal.value !== "None") {
                            object.push(keyVal.value);
                        }
                    }
                } else {
                    let keyValue = (s.key as SerializedStringValue).value;
                    if (keyValue !== "None") {
                        object.push(keyValue);
                    }
                }
                if (object.length !== 0) {
                    let str = object.join();
                    result.push(new Measurement(metric_id + ".key_bounds", Option.some(new StringValue(str))));
                }

                object = [];
                if (s.value instanceof Array) {
                    for (const valVal of s.value) {
                        if (valVal.value !== "None") {
                            object.push(valVal.value);
                        }
                    }
                } else {
                    let valValue = (s.value as SerializedStringValue).value;
                    if (valValue !== "None") {
                        object.push(valValue);
                    }
                }
                if (object.length !== 0) {
                    let str = object.join();
                    result.push(new Measurement(metric_id + ".value_bounds", Option.some(new StringValue(str))));
                }

                return result;
            }
            case "merges": {
                let s = metric.value as MergesMetricValue;
                let avg_step_time = TimeValue.fromDurationMetric(s.avg_step_time);
                let batches = CountValue.fromCountMetric(s.batches);
                let merges = CountValue.fromCountMetric(s.merges);
                let steps = CountValue.fromCountMetric(s.steps);
                return [
                    new Measurement(metric_id + ".avg_step_time", Option.some(avg_step_time)),
                    new Measurement(metric_id + ".batches", Option.some(batches)),
                    new Measurement(metric_id + ".merges", Option.some(merges)),
                    new Measurement(metric_id + ".steps", Option.some(steps)),
                ];
            }
        }
        return [];
    }

    /** Parse a JSON object with the specified label into zero or more measurements.
     * TODO: remove this function once legacy representations are no longer supported. */
    static parseLegacyValues(json: JsonMeasurement): Array<Measurement> {
        let prefix = json[0] as string;
        let value = json.slice(1);
        let result: Array<Measurement> = new Array();
        switch (prefix) {
            case "output batches":
            case "input batches":
                if (Array.isArray(value[0])) {
                    // Empty object was serialized as an empty array
                    break
                }
                let entries = value[0].entries;
                for (const e of entries) {
                    result.push(Measurement.fromJson(prefix + ".", e));
                }
                break;
            case "foreground cache misses":
            case "foreground cache hits":
            case "background cache misses":
            case "background cache hits":
                let count = value[0].count;
                result.push(
                    new Measurement(prefix + ".count", Option.some(new CountValue(count))));
                let bytes = value[0].bytes;
                result.push(
                    new Measurement(prefix + ".bytes", Option.some(new CountValue(bytes))));
                let elapsed = value[0].elapsed;
                result.push(
                    new Measurement(prefix + ".elapsed", Option.some(TimeValue.fromSecondsNanos(
                        elapsed.secs, elapsed.nanos))));
                break;
            default:
                result.push(Measurement.fromJson("", json));
                break;
        }
        return result;
    }

    static unknownProperties: Set<string> = new Set();

    private static parseLegacyPropertyValue(prop: string, value: Array<any>): Option<PropertyValue> {
        switch (prop) {
            case "time%":
            case "merge reduction":
            case "output redundancy":
            case "background cache hit rate":
            case "foreground cache hit rate":
            case "Bloom filter hit rate":
                return Option.some(new PercentValue(value[0][0], value[0][1]));
            case "Bloom filter size":
            case "Bloom filter bits/key":
            case "Bloom filter hits":
            case "Bloom filter misses":
            case "total size":
            case "invocations":
            case "allocated bytes":
            case "used bytes":
            case "shared bytes":
            case "batches":
            case "storage size":
            case "merging batches":
            case "merging size":
            case "allocations":
            case "left inputs":
            case "right inputs":
            case "computed outputs":
            case "inputs":
            case "steps":
            case "input batches/batches":
            case "input batches/min size":
            case "input batches/max size":
            case "input batches/avg size":
            case "input batches/total records":
            case "output batches/batches":
            case "output batches/min size":
            case "output batches/max size":
            case "output batches/avg size":
            case "output batches/total records":
            case "integral records to repartition":
            case "rebalancings":
            case "local shard size":
            case "accumulator records to repartition":
                return Option.some(new CountValue(value[0]));
            case "wait_time":
            case "exchange_wait_time":
            case "merge backpressure wait":
            case "time":
            case "total_idle_time":
            case "runtime_elapsed":
            case "total_runtime":
            case "total rebalancing time":
            case "in-progress rebalancing time":
                return Option.some(TimeValue.fromSecondsNanos(value[0].secs, value[0].nanos));
            case "persistent_id":
            case "foreground cache occupancy":
            case "background cache occupancy":
            case "balancer policy":
            case "slot 0 loose":
            case "slot 1 loose":
            case "slot 2 loose":
            case "slot 3 loose":
            case "slot 4 loose":
            case "slot 0 completed":
            case "slot 1 completed":
            case "slot 2 completed":
            case "slot 3 completed":
            case "slot 4 completed":
            case "slot 0 merging":
            case "slot 1 merging":
            case "slot 2 merging":
            case "slot 3 merging":
            case "slot 4 merging":
            case "balancer.policy":
                return Option.some(new StringValue(value[0]));
            case "rebalancing in progress":
                return Option.some(new BooleanValue(value[0]));
            case "bounds":
            case "mir_node":
            case "key distribution":
                // TODO: currently ignoring some complex properties that we don't know how to convert to scalars
                return Option.none();
            default:
                // We don't know this kind of property, we should probably analyze it.
                if (!Measurement.unknownProperties.has(prop)) {
                    console.log("Unknown property " + prop);
                    Measurement.unknownProperties.add(prop);
                }
                return Option.none();
        }
    }
}

/** A node, its operation, and the (maximum) value of a metric for that node.  The actual metric
 * represented is not part of this data structure */
export class NodeAndMetric {
    constructor(
        public readonly nodeId: string,
        public readonly label: string,
        public readonly operation: string,
        /** Value between 0 and 100% */
        public readonly normalizedValue: number) { }
}

/** A set of measurements for a single circuit graph node and many workers.
 * Only stores numeric values! */
class Measurements {
    readonly measurements: OMap<string, Array<PropertyValue>> = new OMap();

    constructor(private worker_count: number) { }

    add(worker: number, m: Measurement) {
        if (m.value.isNone()) return;  // Ignore unknown properties.
        if (m.property === "persistent_id") {
            return;
        }
        let meas = this.measurements.get(m.property);
        let arr: Array<PropertyValue>;
        if (meas.isNone()) {
            arr = [];
            for (let i = 0; i < this.worker_count; i++)
                arr[i] = new MissingValue();
            this.measurements.set(m.property, arr);
        } else {
            arr = meas.unwrap();
        }
        arr[worker] = m.value.unwrap();
    }

    getMetrics(): Iterable<string> {
        let result = this.measurements.keys();
        if (MetricDescriptions.known()) {
            let data = [...result];
            data.sort((c1, c2) => MetricDescriptions.getMetricIndex(c1) - MetricDescriptions.getMetricIndex(c2));
            return data;
        }
        return result;
    }

    append(other: Measurements) {
        for (const [key, values] of other.measurements.entries()) {
            let existing = this.measurements.get(key);
            if (existing.isNone()) {
                this.measurements.set(key, [...values]);
            } else {
                let arr = existing.unwrap();
                assert(arr.length === values.length, "Mismatched measurement lengths");
                for (let i = 0; i < values.length; i++) {
                    if (key.includes("avg") || key.includes("min") || key.includes("percent") || key.includes("max"))
                        // Heuristic
                        arr[i] = arr[i]!.max(values[i]!);
                    else
                        arr[i] = arr[i]!.combine(values[i]!);
                }
            }
        }
    }

    toString(): string {
        let result = "";
        for (const [key, value] of this.measurements.entries()) {
            if (result !== "")
                result += "\n";
            result += key + " => " + value.toString();
        }
        return result;
    }
}

/** A simple circuit node corresponds to a primitive DBSP operator.
 * ComplexNodes are really regions in the profile comprising multiple other nodes. */
export class SimpleNode implements JsonSimpleCircuitNode {
    // One measurement set per worker
    readonly measurements: Measurements;
    operation: string;
    // Source position of the Rust code in the DBSP library.
    readonly dbspSourcePosition: Option<string>;
    sourcePositions: SourcePositionRanges;
    persistentId: Option<string>;

    constructor(readonly id: NodeId, readonly label: string, worker_count: number) {
        this.measurements = new Measurements(worker_count);
        let parts = label.split('@');
        this.operation = parts[0]!;
        this.dbspSourcePosition = parts[1] ? Option.some(parts[1]) : Option.none();
        this.sourcePositions = SourcePositionRanges.empty();
        this.persistentId = Option.none();
    }

    getChildren(): Array<NodeId> {
        return [];
    }

    setPersistentId(id: string) {
        this.persistentId = Option.some(id);
    }

    addMeasurement(m: Measurement, worker: number) {
        this.measurements.add(worker, m);
    }

    getMeasurements(property: string): Array<PropertyValue> {
        return this.measurements.measurements.get(property).unwrapOr([]);
    }

    // Add all the values from `measurements` into this node's measurements.
    append(measurements: Measurements) {
        this.measurements.append(measurements);
    }

    setSourcePositions(positions: SourcePositionRanges) {
        this.sourcePositions = positions;
    }
}

/** A complex node has a list of children nodes.  Note that the class extends SimpleNode. */
export class ComplexNode extends SimpleNode {
    children: Array<NodeId>;
    depth: number = 0;

    constructor(id: NodeId, label: string, worker_count: number) {
        super(id, label, worker_count);
        this.children = [];
    }

    addChild(node: SimpleNode) {
        this.children.push(node.id);
    }

    appendMeasurements(node: SimpleNode) {
        this.measurements.append(node.measurements);
    }

    setDepth(depth: number) {
        this.depth = depth;
    }

    override getChildren(): Array<NodeId> {
        return this.children;
    }
}

/** An edge in the profile graph. */
export class ProfileEdge {
    constructor(readonly source: NodeId, readonly target: NodeId, readonly back: boolean) { }

    toString(): string {
        return this.source + "->" + this.target;
    }
}

export class CircuitProfile {
    public readonly simpleNodes: OMap<NodeId, SimpleNode> = new OMap();
    public readonly edges: Array<ProfileEdge> = new Array();
    public readonly complexNodes: OMap<NodeId, ComplexNode> = new OMap();
    // Maps each node id to it's immediate parent id.
    public readonly parents: OMap<NodeId, NodeId> = new OMap();
    // Set of all metrics found in the profile.
    private allMetrics: Set<string> = new Set();
    // Names of all worker threads
    private workerNames: Array<number> = new Array();
    // For each measurement the aggregate range of values across all nodes.
    readonly dataRange: OMap<string, NumericRange> = new OMap();
    // Index nodes by their persistent IDs
    readonly byPersistentId: OMap<string, SimpleNode> = new OMap();
    // Source information; only available if the dataflow information is provided.
    sources: Option<Sources> = Option.none();

    /** True if this id corresponds to the toplevel profile graph node.
     * Profile graphs are always a single node containing everything else inside.
     * That node is pretty much ignored everywhere else in this code after parsing. */
    isTop(node: NodeId): boolean {
        return node === "n";
    }

    addNode(n: JsonSimpleNodeWrapper | JsonClusterWrapper, parent: Option<NodeId>) {
        if ('Simple' in n) {
            let node = n.Simple;
            this.simpleNodes.set(node.id, new SimpleNode(node.id, node.label, this.worker_count));
            if (parent.isSome() && !this.isTop(parent.unwrap())) {
                this.parents.set(node.id, parent.unwrap());
            }
        } else {
            let node = n.Cluster;
            this.complexNodes.set(node.id, new ComplexNode(node.id, node.label, this.worker_count));
            for (const child of node.nodes) {
                this.addNode(child, Option.some(node.id));
            }
            if (parent.isSome() && !this.isTop(parent.unwrap())) {
                this.parents.set(node.id, parent.unwrap());
            }
        }
    }

    // The profile graph lies about some nodes, by reverting the edges between them
    // These are the nodes we are looking for and "correcting" while parsing the graph.
    static readonly Z1_TRACE = "Z1 (trace)";
    static readonly Z1_TRACE_OUTPUT = "Z1 (trace) (output)";
    static readonly Z1 = "Z^-1";
    static readonly Z1_OUTPUT = "Z^-1 (output)";
    // Name of the property that holds the persistent ID.
    // In the profile graph this is represented in all workers (if present)
    public static readonly PERSISTENT_ID = "persistent_id";

    // Parse the MIR node of a dataflow graph representation.
    processMirNode(mir: MirNode) {
        let persistent_id = mir.persistent_id;
        if (!persistent_id) return;
        let profileNode = this.byPersistentId.get(persistent_id);
        // This can happen for some Z nodes in recursive components
        if (profileNode.isNone()) return;
        let n = profileNode.unwrap();
        if (mir.table !== null) {
            n.operation += " " + mir.table;
        } else if (mir.view !== null) {
            n.operation += " " + mir.view;
        }

        n.setSourcePositions(new SourcePositionRanges(
            mir.positions.map(p => new SourcePositionRange(p))));
    }

    // Get the topmost parent of a node which is not the toplevel graph node.
    getTopParent(node: NodeId): NodeId {
        while (this.parents.has(node)) {
            node = this.parents.get(node).unwrap();
        }
        return node;
    }

    // TODO: memoize depths for faster computation
    computeDepth(nodeId: NodeId): number {
        let node = this.getNode(nodeId);
        let depth = 0;
        if (!node.isNone()) {
            for (const child of node.unwrap().getChildren()) {
                depth = Math.max(depth, this.computeDepth(child) + 1);
            }
        }
        return depth;
    }

    // Set the dataflow graph information of a profile graph.
    setDataflow(dataflow?: Dataflow, programCode?: string[]) {
        // Add information from the program dataflow generated by the SQL compiler.
        const sources = dataflow?.sources ?? programCode;
        if (sources) {
            this.sources = Option.some(new Sources(sources));
        }

        if (!dataflow) {
            return;
        }

        let mir = dataflow.mir;
        for (const [_, info] of Object.entries(mir)) {
            if (info.operation === "nested") {
                for (const [key, mir] of Object.entries(info)) {
                    if (["calcite", "table", "view", "inputs", "outputs", "persistent_id", "positions"].includes(key)) {
                        continue;
                    }
                    this.processMirNode(mir);
                }
            } else {
                let mir: MirNode = info as MirNode;
                this.processMirNode(mir);
            }
        }

        // Adjust source positions
        for (const node of this.simpleNodes.values()) {
            const parent = this.parents.get(node.id);
            if (parent.isSome()) {
                // If parent is toplevel graph, it's not in the graph
                const complex = this.complexNodes.get(parent.unwrap()).unwrap();
                complex.sourcePositions.append(node.sourcePositions);
            }
        }
        // And adjust the parents of the parents, etc.
        for (const node of this.getSortedComplexNodes()) {
            const parent = this.parents.get(node.id);
            if (parent.isSome()) {
                const complex = this.complexNodes.get(parent.unwrap()).unwrap();
                complex.sourcePositions.append(node.sourcePositions);
            }
        }
    }

    // Return the complex nodes in order of increasing depth
    getSortedComplexNodes(): Array<ComplexNode> {
        let result = [... this.complexNodes.values()];
        result.sort((n1, n2) => n1.depth - n2.depth);
        return result;
    }

    addEdge(e: JsonCircuitEdge) {
        // Some edges are reversed in the profile.
        // For example, between Z1 (trace) and Z1 (trace) (output)
        let from = e.from_node;
        let to = e.to_node;
        let source = this.getNode(from);
        let target = this.getNode(to);
        let back = false;
        if (source.isSome() && target.isSome()) {
            let sourceLabel = source.unwrap().label;
            let targetLabel = target.unwrap().label;
            // Reverse the edge if it's between Z1 (trace) and Z1 (trace) (output)
            // or between Z1_OUTPUT and Z1
            if ((sourceLabel.includes(CircuitProfile.Z1_TRACE_OUTPUT) && targetLabel.includes(CircuitProfile.Z1_TRACE)) ||
                (sourceLabel.includes(CircuitProfile.Z1_OUTPUT) && targetLabel === CircuitProfile.Z1)) {
                let tmp = from;
                from = to;
                to = tmp;
                back = true;
            }
        }

        this.edges.push(new ProfileEdge(from, to, back));
    }

    getWorkerNames(): Array<number> {
        return this.workerNames;
    }

    getNode(id: NodeId): Option<SimpleNode> {
        if (this.simpleNodes.has(id))
            return this.simpleNodes.get(id);
        return this.complexNodes.get(id);
    }

    /**
     * Get all source code position ranges for a given node ID.
     *
     * A node can have multiple source position ranges when a single DBSP operator
     * corresponds to multiple SQL fragments (e.g., a join referencing tables defined
     * in different parts of the SQL).
     *
     * @param id - The node ID to look up
     * @returns Array of SourcePositionRange objects, or empty array if,
     *          - Node doesn't exist
     *          - Node has no source position information
     */
    // This method is in CircuitProfile rather than Visualizer because CircuitProfile
    // owns both the node data (simpleNodes/complexNodes) and the source data (sources).
    getSourceRanges(id: NodeId): Array<SourcePositionRange> {
        const node = this.getNode(id);
        if (node.isNone()) {
            return [];
        }

        return node.unwrap().sourcePositions.positions;
    }

    /**
     * Get the first source code position range for a given node ID.
     *
     * Convenience method that returns the first (primary) range from getSourceRanges(),
     * typically used for simple navigation purposes.
     *
     * @param id - The node ID to look up
     * @returns Option containing the first SourcePositionRange, or none if no ranges exist
     */
    getFirstSourceRange(id: NodeId): Option<SourcePositionRange> {
        const ranges = this.getSourceRanges(id);
        if (ranges.length === 0) {
            return Option.none();
        }
        return Option.some(ranges[0]!);
    }

    constructor(readonly worker_count: number) { }

    // Scan the nodes and compute the range of each property
    computePropertyRanges() {
        for (const metric of this.allMetrics) {
            let range = NumericRange.empty();
            for (const node of this.simpleNodes.values()) {
                let m = node.getMeasurements(metric);
                let values = m.map(v => v.getNumericValue()).filter(v => v.isSome()).map(v => v.unwrap());
                range = range.union(NumericRange.getRange(values));
            }
            this.dataRange.set(metric, range);
        }
    }

    // Given a property, compute the range of values across all nodes.
    propertyRange(property: string): NumericRange {
        return this.dataRange.get(property).unwrap();
    }

    // Decode the data in a worker profile; return a map from nodeId to an array of measurements for that node
    private decodeWorkerProfile(json: JsonWorkerProfile['metadata']): Map<NodeId, Array<Measurement>> {
        let metadata = new Map<NodeId, Array<Measurement>>();
        for (const [nodeId, data] of Object.entries(json)) {
            let parsed: Array<Measurement> = [];
            for (const datum of data) {
                let measurements = Measurement.parseValues(datum);
                for (const m of measurements) {
                    parsed.push(m);
                    this.allMetrics.add(m.property);
                }
            }
            metadata.set(nodeId, parsed);
        }
        return metadata;
    }

    // Decoder for legacy profile format, to be removed
    private decodeLegacyWorkerProfile(json: LegacyJsonWorkerProfile['metadata']): Map<NodeId, Array<Measurement>> {
        let metadata = new Map<NodeId, Array<Measurement>>();
        for (const [nodeId, data] of Object.entries(json)) {
            let parsed: Array<Measurement> = [];
            for (const datum of data.entries) {
                let measurements = Measurement.parseLegacyValues(datum);
                for (const m of measurements) {
                    parsed.push(m);
                    this.allMetrics.add(m.property);
                }
            }
            metadata.set(nodeId, parsed);
        }
        return metadata;
    }

    /** Create a CircuitProfile from the JSON serialization */
    static fromJson(json: JsonProfiles) {
        // For some reason, the profiles generated in the new format sometimes do not include `metrics` field.
        // With this simple heuristic we can detect and still parse those profiles, but we should investigate why this happens and fix it.
        // Here, we check if the first profile measurement contains a metric_id field, which is a required field in the new format and is absent in the old format.
        // TODO: remove this block once legacy representations are no longer supported.
        if (!('metrics' in json) && (Object.values(json.worker_profiles[0]?.metadata ?? {}) as any)?.[0]?.[0]?.['metric_id']) {
            (json as any).metrics = [];
        }

        const isNewProfile = 'metrics' in json;
        if (isNewProfile) {
            MetricDescriptions.setMetricDescriptions(json.metrics);
        }

        let worker_count = json.worker_profiles.length;
        // Decode the graph structure and create the nodes.
        // The graph itself is always a complex node.
        let rootNodeId = json.graph.nodes.id;
        let result = new CircuitProfile(worker_count);
        result.complexNodes.set(rootNodeId,
            new ComplexNode(rootNodeId, json.graph.nodes.label, worker_count));
        for (const nodeWrapper of json.graph.nodes.nodes) {
            // Ignore top-level graph region
            result.addNode(nodeWrapper, Option.none());
        }
        for (const edge of json.graph.edges) {
            result.addEdge(edge);
        }

        const decodedProfiles = isNewProfile ?
            json.worker_profiles.entries().map(([index, workerProfile]) => tuple(index, result.decodeWorkerProfile(workerProfile.metadata))) :
            json.worker_profiles.entries().map(([index, workerProfile]) => tuple(index, result.decodeLegacyWorkerProfile(workerProfile.metadata)))

        // Decode the profile information and attach to the nodes.
        for (const [index, decoded] of decodedProfiles
        ) {
            for (const [node, measurements] of decoded.entries()) {
                let n: SimpleNode;
                if (result.simpleNodes.has(node)) {
                    n = result.simpleNodes.get(node).unwrap();
                    for (const m of measurements) {
                        n.addMeasurement(m, index);
                        if (m.property === CircuitProfile.PERSISTENT_ID && m.value.isSome()) {
                            let pid = m.value.unwrap().getStringValue();
                            result.byPersistentId.set(pid, n);
                            n.setPersistentId(pid);
                        }
                    }
                } else if (result.complexNodes.has(node)) {
                    // Ignore measurements for complex nodes, except the root node, which has some special attributes
                    if (node === rootNodeId) {
                        n = result.complexNodes.get(node).unwrap();
                        for (const m of measurements) {
                            n.addMeasurement(m, index);
                        }
                    }
                } else {
                    fail("Node not found " + node);
                }
            }
        }

        result.workerNames = Array.from({ length: json.worker_profiles.length }, (_, i) => i);
        // Merge Z1 (trace) nodes; they are artificially split in the profile into two halves each.
        result.fixZ1Nodes();

        // Register children
        for (const node of [...result.simpleNodes.values(), ...result.complexNodes.values()]) {
            const parent = result.parents.get(node.id);
            if (parent.isSome()) {
                // If parent is toplevel graph, it is None
                const complex = result.complexNodes.get(parent.unwrap()).unwrap();
                complex.addChild(node);
            }
        }

        // Compute and save depths
        for (const node of result.complexNodes.values()) {
            node.setDepth(result.computeDepth(node.id));
        }

        // Assign to every complex node the "sum" of the children's measurements.
        // Process in order of increasing depth.
        for (const complex of result.getSortedComplexNodes()) {
            for (const child of complex.children) {
                let node = result.getNode(child);
                complex.appendMeasurements(node.unwrap());
            }
        }

        result.computePropertyRanges();
        return {
            profile: result,
            rootNodeId
        };
    }

    fixZ1Nodes() {
        // A Z1 node is represented as a "Z1 (trace)" node followed by a "Z1 (trace) (output)" node.
        // We have already flipped the edges while parsing so that there's a back-edge from Z1 to Z1 (output).
        // Now we remove the Z1 node, redirect all its incoming edges to Z1 (output),
        // and merge the measurements into the Z1 (output).
        const edges = new Array<ProfileEdge>();

        let backEdges = this.edges.filter(e => e.back);
        let toReplace = new Map<string, string>(backEdges.map(e => [e.source, e.target]));

        for (const edge of this.edges) {
            if (edge.back) {
                let sourceNode = this.simpleNodes.get(edge.source).unwrap();
                let targetNode = this.simpleNodes.get(edge.target).unwrap();
                targetNode.append(sourceNode.measurements);
                assert(toReplace.has(edge.source), "Expected to delete node " + edge.source);
                this.simpleNodes.delete(edge.source);
            } else {
                if (toReplace.has(edge.target)) {
                    // Redirect edge to the target of the back edge.
                    let redirect = new ProfileEdge(edge.source, toReplace.get(edge.target)!, true);
                    edges.push(redirect);
                } else {
                    edges.push(edge);
                }
            }
        }
        this.edges.length = 0;
        for (const edge of edges) {
            this.edges.push(edge);
        }
    }
}

