Merge upstream Vortex 0.73.0 into spiceai-52

.agents/skills/bench-performance/agents/openai.yaml

@@ -0,0 +1,6 @@
+interface:
+  display_name: "Bench Performance"
+  short_description: "Iterate on Vortex benchmark performance"
+  default_prompt: "Use $bench-performance to investigate a vx-bench query performance issue with comparisons, logs, metrics, and profiling."
+policy:
+  allow_implicit_invocation: true

.agents/skills/bench-performance/scripts/compare_gh_json.py

@@ -0,0 +1,101 @@
+#!/usr/bin/env python3
+"""Summarize Vortex benchmark gh-json / JSONL output with target ratios."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import statistics
+from collections import defaultdict
+from pathlib import Path
+from typing import Any
+
+
+def load_records(paths: list[Path]) -> list[dict[str, Any]]:
+    records: list[dict[str, Any]] = []
+    for path in paths:
+        with path.open("r", encoding="utf-8") as f:
+            for line in f:
+                line = line.strip()
+                if not line or not line.startswith("{"):
+                    continue
+                try:
+                    record = json.loads(line)
+                except json.JSONDecodeError:
+                    continue
+                if "target" in record and "value" in record:
+                    record["_source"] = str(path)
+                    records.append(record)
+    return records
+
+
+def target_name(record: dict[str, Any]) -> str:
+    target = record.get("target") or {}
+    engine = target.get("engine", "?")
+    fmt = target.get("format", "?")
+    return f"{engine}:{fmt}"
+
+
+def query_name(record: dict[str, Any]) -> str:
+    name = str(record.get("name", ""))
+    if "/" in name:
+        return name.split("/", 1)[0]
+    return name or str(record.get("_source", "unknown"))
+
+
+def ns_to_ms(value: float) -> float:
+    return value / 1_000_000.0
+
+
+def runtime_summary(record: dict[str, Any]) -> str:
+    runtimes = record.get("all_runtimes")
+    if not isinstance(runtimes, list) or not runtimes:
+        value = float(record["value"])
+        return f"{ns_to_ms(value):.3f}/{ns_to_ms(value):.3f}/{ns_to_ms(value):.3f}"
+    values = sorted(float(v) for v in runtimes if isinstance(v, (int, float)))
+    if not values:
+        value = float(record["value"])
+        return f"{ns_to_ms(value):.3f}/{ns_to_ms(value):.3f}/{ns_to_ms(value):.3f}"
+    return f"{ns_to_ms(values[0]):.3f}/{ns_to_ms(statistics.median(values)):.3f}/{ns_to_ms(values[-1]):.3f}"
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("jsonl", nargs="+", type=Path, help="gh-json JSONL files")
+    parser.add_argument(
+        "--baseline",
+        help="Baseline target such as datafusion:parquet. Defaults to first target per query.",
+    )
+    args = parser.parse_args()
+
+    records = load_records(args.jsonl)
+    if not records:
+        print("No benchmark records found.")
+        return 1
+
+    groups: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    for record in records:
+        groups[query_name(record)].append(record)
+
+    print("query\ttarget\tvalue_ms\tratio\tmin/median/max_ms\tsource")
+    for query in sorted(groups):
+        rows = sorted(groups[query], key=target_name)
+        baseline = None
+        if args.baseline:
+            baseline = next((r for r in rows if target_name(r) == args.baseline), None)
+        if baseline is None:
+            baseline = rows[0]
+        baseline_value = float(baseline["value"])
+
+        for record in rows:
+            value = float(record["value"])
+            ratio = value / baseline_value if baseline_value else float("nan")
+            print(
+                f"{query}\t{target_name(record)}\t{ns_to_ms(value):.3f}\t"
+                f"{ratio:.2f}x\t{runtime_summary(record)}\t{record['_source']}"
+            )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

.agents/skills/bench-performance/scripts/compare_metrics.py

@@ -0,0 +1,114 @@
+#!/usr/bin/env python3
+"""Compare Vortex benchmark --show-metrics text logs."""
+
+from __future__ import annotations
+
+import argparse
+import re
+from pathlib import Path
+
+METRIC_RE = re.compile(r"^\s*([A-Za-z0-9_.\[\]-]+)=(.+?)\s*$")
+NUMBER_UNIT_RE = re.compile(r"^\s*([-+]?[0-9]*\.?[0-9]+)\s*([A-Za-z/]+)?\s*$")
+
+UNIT_SCALE = {
+    "": 1.0,
+    "K": 1_000.0,
+    "M": 1_000_000.0,
+    "B": 1.0,
+    "KB": 1_000.0,
+    "MB": 1_000_000.0,
+    "GB": 1_000_000_000.0,
+    "ns": 1e-9,
+    "us": 1e-6,
+    "µs": 1e-6,
+    "ms": 1e-3,
+    "s": 1.0,
+}
+
+
+def parse_value(raw: str) -> tuple[float | None, str]:
+    normalized = raw.strip().replace("\u00b5", "µ")
+    match = NUMBER_UNIT_RE.match(normalized)
+    if not match:
+        return None, raw.strip()
+    value = float(match.group(1))
+    unit = match.group(2) or ""
+    scale = UNIT_SCALE.get(unit)
+    if scale is None:
+        return value, unit
+    return value * scale, unit
+
+
+def parse_metrics(path: Path) -> dict[str, tuple[float | None, str, str]]:
+    metrics: dict[str, tuple[float | None, str, str]] = {}
+    with path.open("r", encoding="utf-8") as f:
+        for line in f:
+            match = METRIC_RE.match(line)
+            if not match:
+                continue
+            name, raw = match.groups()
+            parsed, unit = parse_value(raw)
+            metrics[name] = (parsed, raw.strip(), unit)
+    return metrics
+
+
+def default_metrics(all_metrics: list[dict[str, tuple[float | None, str, str]]]) -> list[str]:
+    preferred = [
+        "vortex.io.read.duration_count",
+        "vortex.io.read.total_size",
+        "vortex.io.read.duration_max",
+        "vortex.io.read.size_max",
+        "vortex.file.segments.cache.misses",
+        "vortex.file.segments.cache.hits",
+        "io.requests.individual",
+        "io.requests.coalesced",
+        "time_elapsed_opening",
+        "time_elapsed_processing",
+        "time_elapsed_scanning_total",
+        "time_elapsed_scanning_until_data",
+        "output_rows",
+        "output_bytes",
+    ]
+    present = set().union(*(m.keys() for m in all_metrics))
+    return [metric for metric in preferred if metric in present]
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("logs", nargs="+", type=Path)
+    parser.add_argument(
+        "--metrics",
+        help="Comma-separated metric names. Defaults to common Vortex scan metrics.",
+    )
+    args = parser.parse_args()
+
+    parsed = [(path, parse_metrics(path)) for path in args.logs]
+    metric_names = (
+        [m.strip() for m in args.metrics.split(",") if m.strip()]
+        if args.metrics
+        else default_metrics([metrics for _, metrics in parsed])
+    )
+
+    if not metric_names:
+        print("No metrics found.")
+        return 1
+
+    baseline_metrics = parsed[0][1]
+    print("metric\t" + "\t".join(str(path) for path, _ in parsed))
+    for metric in metric_names:
+        cells = []
+        baseline_value = baseline_metrics.get(metric, (None, "", ""))[0]
+        for _, metrics in parsed:
+            value, raw, _unit = metrics.get(metric, (None, "", ""))
+            if not raw:
+                cells.append("-")
+            elif baseline_value and value is not None:
+                cells.append(f"{raw} ({value / baseline_value:.2f}x)")
+            else:
+                cells.append(raw)
+        print(metric + "\t" + "\t".join(cells))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

.agents/skills/bench-performance/scripts/summarize_conjunct_debug.py

@@ -0,0 +1,141 @@
+#!/usr/bin/env python3
+"""Summarize Vortex conjunct evaluation tracing lines."""
+
+from __future__ import annotations
+
+import argparse
+import collections
+import re
+import statistics
+from pathlib import Path
+
+FIELD_RE = re.compile(
+    r"(?P<key>[A-Za-z_][A-Za-z0-9_]*)="
+    r"(?P<value>\"(?:[^\"\\]|\\.)*\"|Some\([^)]+\)|None|[^\s]+)"
+)
+FIRST_FIELD_RE = re.compile(r" [A-Za-z_][A-Za-z0-9_]*=")
+
+
+def parse_value(raw: str) -> str:
+    if raw.startswith('"') and raw.endswith('"'):
+        return raw[1:-1]
+    if raw.startswith("Some(") and raw.endswith(")"):
+        return raw[5:-1]
+    return raw
+
+
+def as_int(fields: dict[str, str], key: str) -> int:
+    value = fields.get(key)
+    if value in (None, "None"):
+        return 0
+    return int(value)
+
+
+def as_float(fields: dict[str, str], key: str) -> float:
+    value = fields.get(key)
+    if value in (None, "None"):
+        return 0.0
+    return float(value)
+
+
+def message_for(line: str) -> str | None:
+    rest_match = re.search(r":\d+: (?P<rest>.*)$", line.rstrip())
+    rest = rest_match.group("rest") if rest_match else line.rstrip()
+    first_field = FIRST_FIELD_RE.search(rest)
+    message = rest[: first_field.start() if first_field else len(rest)].strip()
+    if "conjunct" in message and "evaluated" in message:
+        return message
+    return None
+
+
+def quantile(values: list[int], fraction: float) -> float:
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    index = min(len(ordered) - 1, max(0, round((len(ordered) - 1) * fraction)))
+    return float(ordered[index])
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("logs", nargs="+", type=Path)
+    parser.add_argument("--top-orders", type=int, default=12)
+    parser.add_argument("--message-regex", help="Only include messages matching this regex")
+    args = parser.parse_args()
+
+    message_re = re.compile(args.message_regex) if args.message_regex else None
+    groups: dict[tuple[str, str, str], list[dict[str, str]]] = collections.defaultdict(list)
+    order_counts: collections.Counter[tuple[str, ...]] = collections.Counter()
+    order_input_rows: collections.defaultdict[tuple[str, ...], int] = collections.defaultdict(int)
+    current_windows: dict[tuple[str, str, str, str], list[str]] = {}
+    rows = 0
+
+    for path in args.logs:
+        with path.open("r", encoding="utf-8", errors="replace") as f:
+            for line in f:
+                message = message_for(line)
+                if message is None or (message_re is not None and not message_re.search(message)):
+                    continue
+                fields = {match.group("key"): parse_value(match.group("value")) for match in FIELD_RE.finditer(line)}
+                original_idx = (
+                    fields.get("original_idx") or fields.get("conjunct_idx") or fields.get("child_idx") or "?"
+                )
+                conjunct = fields.get("conjunct", "")
+                groups[(message, original_idx, conjunct)].append(fields)
+                rows += 1
+
+                window_key = (
+                    fields.get("scan_label", ""),
+                    fields.get("coord_start", ""),
+                    fields.get("coord_end", ""),
+                    fields.get("output_coord_hash", ""),
+                )
+                order = current_windows.setdefault(window_key, [])
+                if not order:
+                    order_input_rows[tuple(order)] += 0
+                order.append(original_idx)
+                if as_int(fields, "output_rows") == 0:
+                    order_tuple = tuple(order)
+                    order_counts[order_tuple] += 1
+                    order_input_rows[order_tuple] += as_int(fields, "input_rows")
+                    current_windows.pop(window_key, None)
+
+    for order in current_windows.values():
+        if order:
+            order_tuple = tuple(order)
+            order_counts[order_tuple] += 1
+
+    if rows == 0:
+        print("No conjunct debug rows found.")
+        return 1
+
+    print(f"rows={rows:,}")
+    print(
+        "message\tconjunct\tindex\tevents\tinput_rows\toutput_rows\tcompute_input_rows\tcompute_output_rows\telapsed_ms\tcompute_per_input"
+    )
+    for (message, idx, conjunct), entries in sorted(groups.items(), key=lambda item: item[0]):
+        input_rows = sum(as_int(e, "input_rows") for e in entries)
+        output_rows = sum(as_int(e, "output_rows") for e in entries)
+        compute_input = sum(as_int(e, "compute_input_rows") or as_int(e, "input_rows") for e in entries)
+        compute_output = sum(as_int(e, "compute_output_rows") or as_int(e, "output_rows") for e in entries)
+        elapsed = sum(as_float(e, "elapsed_ms") for e in entries)
+        ratio = compute_input / input_rows if input_rows else 0.0
+        print(
+            f"{message}\t{conjunct}\t{idx}\t{len(entries):,}\t{input_rows:,}\t{output_rows:,}\t"
+            f"{compute_input:,}\t{compute_output:,}\t{elapsed:,.3f}\t{ratio:.3f}"
+        )
+
+    print("\norders:")
+    for order, count in order_counts.most_common(args.top_orders):
+        print(f"  {order}: count={count:,} input_rows={order_input_rows[order]:,}")
+
+    print("\nevents_per_window:")
+    lengths = [len(order) for order, count in order_counts.items() for _ in range(count)]
+    if lengths:
+        print(f"  median={statistics.median(lengths):.0f} p90={quantile(lengths, 0.90):.0f} max={max(lengths)}")
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())