@@ -72,8 +72,8 @@ Do not wait for a deep code read before showing benchmark comparisons or first s
    with the orchestrator or directly:
 
    ```bash
-   cargo build -p datafusion-bench --profile release_debug --features unstable_encodings
-   cargo build -p duckdb-bench --profile release_debug --features unstable_encodings
+   cargo build -p datafusion-bench --profile release_debug
+   cargo build -p duckdb-bench --profile release_debug
    cargo build -p lance-bench --profile release_debug
    ```
 
@@ -85,6 +85,16 @@ Do not wait for a deep code read before showing benchmark comparisons or first s
 
 ## Common Environment
 
+`scripts/bench-v2-drivers.py` compares baseline and candidate binaries for both engines, alternates
+run order, excludes warmup, and reports paired round changes. It uses DuckDB's
+default fresh connection per iteration and clears segment-cache/preload settings. Its `--perf-stat` runs separate
+diagnostic processes with acknowledged perf control FIFOs. The shared runner enables counters only
+around query callbacks after `VORTEX_BENCH_PERF_WARMUP` iterations (default 2), using
+`VORTEX_BENCH_PERF_CONTROL` and `VORTEX_BENCH_PERF_ACK`. Both binaries must include this support.
+DuckDB's callback includes connection reopening unless `--reuse` is specified; perf counts include
+that work, while DuckDB's reported query timer excludes it. Use `--reuse` only for an explicitly
+labeled profiling run, not as the default performance comparison.
+
 - `RUST_LOG` wins over `--verbose`. Without `RUST_LOG`, `--verbose` raises default logging to
   `TRACE`; otherwise the env filter controls output.
 - Useful starting filters:
@@ -119,7 +129,10 @@ Supported diagnostics:
 - `--explain` prints query plans instead of timing;
 - `--show-metrics` prints Vortex execution-plan metrics after a timed run.
 
-Declared but currently not useful unless the source changes: `--threads`, `--emit-plan`, and
+`--threads N` sets both Tokio worker threads and DataFusion execution partitions. Without it,
+Tokio and DataFusion retain their environment-controlled defaults.
+
+Declared but currently not useful unless the source changes: `--emit-plan` and
 `--export-spans` are parsed but not wired into the execution path.
 
 Examples:
@@ -141,6 +154,9 @@ FEATURE_TOGGLE=1 target/release_debug/datafusion-bench tpch \
 
 Source: `benchmarks/duckdb-bench/src/main.rs`.
 
+Uses prepared data, like DataFusion. Prepare missing formats with `vx-bench prepare-data`
+or `cargo run --bin data-gen -- <benchmark> --formats <formats>` before profiling.
+
 Supported diagnostics:
 
 - `--formats parquet,vortex,vortex-compact,duckdb`;
@@ -158,7 +174,7 @@ Example:
 RUST_LOG=duckdb_bench=trace,vortex_duckdb=debug,info \
   target/release_debug/duckdb-bench tpch \
   --display-format gh-json --iterations 5 --hide-progress-bar \
-  --formats <baseline-format>,<candidate-format> --queries <query> --threads 8 --reuse \
+  --formats <baseline-format>,<candidate-format> --queries <query> --threads 8 \
   -o /private/tmp/<label>.jsonl
 ```
 
 
@@ -100,6 +100,15 @@ For long performance sessions, use this cadence:
    samply load --no-open /tmp/<label>.profile.json.gz
    ```
 
+   On Linux, the summary script uses `llvm-symbolizer` for `--symbolicate`. Keep the exact
+   recorded binary available. Check that the profile contains samples before interpreting it.
+   If launching Samply under a CPU affinity mask yields an empty profile, start the benchmark
+   with its desired affinity and attach Samply from a process with the full affinity mask:
+
+   ```bash
+   sudo samply record --save-only --rate 500 --pid <benchmark-pid> --output /tmp/<label>.profile.json.gz
+   ```
+
    Then report the printed local URL.
 
 4. Summarize the profile without opening Firefox Profiler. Run a small summary first and report it
@@ -175,6 +184,11 @@ Samply writes Firefox-profiler JSON, often compressed as `profile.json.gz`.
   `stackTable`. `samples.weight[]` is the number of collapsed samples represented by that row; use
   weight instead of row count when present. `samples.threadCPUDelta[]` is per-thread CPU delta in
   microseconds when present.
+- Validate CPU deltas before using CPU-weighted attribution or `profile_activity.py`.
+  If idle threads each accumulate roughly the whole recording duration, the CPU
+  deltas are unsuitable. Use `--weight-mode samples`, select execution workers,
+  and compare with separately recorded perf task-clock. Timestamps may be stored
+  as absolute `samples.time` values or cumulative `samples.timeDeltas`.
 - `stackTable` is a linked list: `stackTable.frame[i]` is the current frame and
   `stackTable.prefix[i]` points to the caller stack. Follow prefixes to `null` and reverse to get
   root-to-leaf order.
 
@@ -1,4 +1,6 @@
 #!/usr/bin/env python3
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright the Vortex contributors
 """Summarize thread activity over time in a Samply / Firefox profiler JSON file."""
 
 from __future__ import annotations
@@ -14,6 +16,8 @@
 
 
 def sample_times(samples: dict[str, Any]) -> list[float]:
+    if isinstance(samples.get("time"), list):
+        return samples["time"]
     total = 0.0
     times = []
     for delta in samples.get("timeDeltas") or []:
 
@@ -1,10 +1,13 @@
 #!/usr/bin/env python3
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright the Vortex contributors
 """Print an inverted call tree from a Samply / Firefox profiler JSON file."""
 
 from __future__ import annotations
 
 import argparse
 import collections
+import platform
 import re
 from collections.abc import Callable
 from pathlib import Path
@@ -20,6 +23,8 @@ def __init__(self) -> None:
 
 
 def sample_times(samples: dict[str, Any]) -> list[float]:
+    if isinstance(samples.get("time"), list):
+        return samples["time"]
     total = 0.0
     times = []
     for delta in samples.get("timeDeltas") or []:
@@ -37,13 +42,13 @@ def thread_matches(thread: dict[str, Any], thread_re: re.Pattern[str] | None, ti
     return bool(thread_re.search(str(thread.get("name", ""))) or thread_re.search(str(thread.get("tid", ""))))
 
 
-def frame_labels(thread: dict[str, Any], symbols: dict[str, str]) -> Callable[[int], str]:
+def frame_labels(thread: dict[str, Any], symbols: dict[str, str], symbol_lib: str | None) -> Callable[[int], str]:
     cache: dict[int, str] = {}
 
     def label(frame_index: int) -> str:
         cached = cache.get(frame_index)
         if cached is None:
-            cached = ps.frame_label(thread, frame_index, symbols)
+            cached = ps.frame_label(thread, frame_index, symbols, symbol_lib)
             cache[frame_index] = cached
         return cached
 
@@ -93,7 +98,7 @@ def build_tree(
         samples = thread.get("samples") or {}
         stacks = samples.get("stack") or []
         times = sample_times(samples)
-        label = frame_labels(thread, symbols)
+        label = frame_labels(thread, symbols, (args.symbol_lib or args.binary.name) if args.symbolicate else None)
         thread_weight = 0
         thread_samples = 0
 
@@ -151,7 +156,11 @@ def main() -> int:
     parser.add_argument("--top", type=int, default=8, help="Max children to print at each tree level")
     parser.add_argument("--depth", type=int, default=12, help="Max inverted tree depth to print")
     parser.add_argument("--binary", type=Path, help="Candidate binary for symbolication")
-    parser.add_argument("--symbolicate", action="store_true", help="Use atos to symbolicate raw app offsets")
+    parser.add_argument(
+        "--symbolicate",
+        action="store_true",
+        help="Symbolicate raw app offsets with llvm-symbolizer on Linux or atos on macOS",
+    )
     parser.add_argument(
         "--symbol-lib",
         help="Library name to symbolicate; defaults to the basename of --binary",
@@ -166,7 +175,11 @@ def main() -> int:
             parser.error("--symbolicate requires --binary")
         symbol_lib = args.symbol_lib or args.binary.name
         addresses = ps.collect_symbol_addresses(profile, symbol_lib)
-        symbols = ps.atos_symbol_map(args.binary, addresses, int(str(args.load_address), 0))
+        symbols = (
+            ps.llvm_symbol_map(args.binary, addresses)
+            if platform.system() == "Linux"
+            else ps.atos_symbol_map(args.binary, addresses, int(str(args.load_address), 0))
+        )
 
     root, matched_threads = build_tree(profile, symbols, args)
     unit = "cpu_us" if args.weight_mode == "cpu" else "samples"
 
@@ -1,4 +1,6 @@
 #!/usr/bin/env python3
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright the Vortex contributors
 """Summarize Samply / Firefox profiler JSON without opening the UI."""
 
 from __future__ import annotations
@@ -8,6 +10,7 @@
 import gzip
 import json
 import os
+import platform
 import re
 import subprocess
 from pathlib import Path
@@ -51,7 +54,12 @@ def resource_name(thread: dict[str, Any], resource_index: object) -> str:
 ADDRESS_RE = re.compile(r"^0x[0-9a-fA-F]+$")
 
 
-def frame_label(thread: dict[str, Any], frame_index: int, symbols: dict[str, str] | None = None) -> str:
+def frame_label(
+    thread: dict[str, Any],
+    frame_index: int,
+    symbols: dict[str, str] | None = None,
+    symbol_lib: str | None = None,
+) -> str:
     frame_table = thread.get("frameTable") or {}
     func_table = thread.get("funcTable") or {}
     func_index = value_at(frame_table, "func", frame_index)
@@ -63,7 +71,8 @@ def frame_label(thread: dict[str, Any], frame_index: int, symbols: dict[str, str
     file_name = decode_string(thread, value_at(func_table, "fileName", func_index))
     line = value_at(func_table, "lineNumber", func_index)
 
-    label = symbols.get(name, name) if symbols else name
+    matching_lib = symbol_lib is None or os.path.basename(resource) == symbol_lib
+    label = symbols.get(name, name) if symbols and matching_lib else name
     label = label or f"<func:{func_index}>"
     if resource and resource not in label:
         label = f"{label} [{resource}]"
@@ -113,6 +122,7 @@ def summarize_thread(
     thread: dict[str, Any],
     symbols: dict[str, str] | None = None,
     weight_mode: str = "samples",
+    symbol_lib: str | None = None,
 ) -> dict[str, Any]:
     samples = thread.get("samples") or {}
     stacks = samples.get("stack") or []
@@ -127,7 +137,7 @@ def summarize_thread(
     def label(frame_index: int) -> str:
         cached = label_cache.get(frame_index)
         if cached is None:
-            cached = frame_label(thread, frame_index, symbols)
+            cached = frame_label(thread, frame_index, symbols, symbol_lib)
             label_cache[frame_index] = cached
         return cached
 
@@ -238,6 +248,24 @@ def atos_symbol_map(binary: Path, addresses: list[str], load_address: int) -> di
     return {raw: symbol.strip() for raw, symbol in zip(addresses, symbols) if symbol.strip() and symbol.strip() != raw}
 
 
+def llvm_symbol_map(binary: Path, addresses: list[str]) -> dict[str, str]:
+    if not addresses:
+        return {}
+    result = subprocess.run(
+        ["llvm-symbolizer", f"--obj={binary}", "--no-inlines", "--demangle"],
+        input="\n".join(addresses) + "\n",
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    blocks = result.stdout.strip().split("\n\n")
+    return {
+        address: block.splitlines()[0]
+        for address, block in zip(addresses, blocks)
+        if block.splitlines() and block.splitlines()[0] != "??"
+    }
+
+
 def main() -> int:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("profile", type=Path)
@@ -247,7 +275,7 @@ def main() -> int:
     parser.add_argument("--stack-depth", type=int, default=14, help="Frames to print from the leaf side")
     parser.add_argument("--libs", action="store_true", help="Print profile library metadata")
     parser.add_argument("--binary", type=Path, help="Print dwarfdump UUID for a candidate binary")
-    parser.add_argument("--symbolicate", action="store_true", help="Use atos to symbolicate raw app offsets")
+    parser.add_argument("--symbolicate", action="store_true", help="Use llvm-symbolizer on Linux or atos on macOS")
     parser.add_argument(
         "--symbol-lib",
         help="Library name to symbolicate; defaults to the basename of --binary",
@@ -275,21 +303,30 @@ def main() -> int:
         print(binary_uuid(args.binary))
 
     symbols: dict[str, str] = {}
+    symbol_lib = None
     if args.symbolicate:
         if not args.binary:
             parser.error("--symbolicate requires --binary")
         symbol_lib = args.symbol_lib or args.binary.name
         addresses = collect_symbol_addresses(profile, symbol_lib)
-        symbols = atos_symbol_map(args.binary, addresses, int(str(args.load_address), 0))
+        if platform.system() == "Linux":
+            symbols = llvm_symbol_map(args.binary, addresses)
+        else:
+            symbols = atos_symbol_map(args.binary, addresses, int(str(args.load_address), 0))
         print(f"Symbolicated {len(symbols)} / {len(addresses)} raw addresses from {symbol_lib}")
 
     if args.libs:
         print_libs(profile)
 
-    summaries = [summarize_thread(thread, symbols, args.weight_mode) for thread in profile.get("threads") or []]
-    summaries.sort(key=lambda s: (s["cpu_ms"], s["weight"]), reverse=True)
+    summaries = [
+        summarize_thread(thread, symbols, args.weight_mode, symbol_lib) for thread in profile.get("threads") or []
+    ]
+    summaries.sort(
+        key=lambda s: (s["weight"], s["cpu_ms"]) if args.weight_mode == "samples" else (s["cpu_ms"], s["weight"]),
+        reverse=True,
+    )
 
-    print("\nThreads by CPU:")
+    print("\nThreads by sample weight:" if args.weight_mode == "samples" else "\nThreads by CPU:")
     for summary in summaries[: max(args.threads, 20)]:
         print(
             f"  cpu_ms={summary['cpu_ms']:10.3f} weight={summary['weight']:8d} "
 
@@ -4,6 +4,7 @@
 pub mod metrics;
 pub mod tracer;
 
+use std::num::NonZeroUsize;
 use std::sync::Arc;
 
 use datafusion::datasource::file_format::FileFormat;
@@ -27,8 +28,9 @@ use vortex_datafusion::VortexFormat;
 use vortex_datafusion::VortexFormatFactory;
 use vortex_datafusion::VortexTableOptions;
 
+/// Creates a benchmark session with the requested execution parallelism.
 #[expect(clippy::expect_used)]
-pub fn get_session_context() -> SessionContext {
+pub fn get_session_context(threads: Option<NonZeroUsize>) -> SessionContext {
     let mut rt_builder = RuntimeEnvBuilder::new();
 
     rt_builder = rt_builder.with_cache_manager(CacheManagerConfig::default());
@@ -39,8 +41,12 @@ pub fn get_session_context() -> SessionContext {
 
     let factory = VortexFormatFactory::new().with_options(vortex_table_options());
 
+    let mut config = SessionConfig::from_env().expect("shouldn't fail");
+    if let Some(threads) = threads {
+        config = config.with_target_partitions(threads.get());
+    }
     let mut session_state_builder = SessionStateBuilder::new()
-        .with_config(SessionConfig::from_env().expect("shouldn't fail"))
+        .with_config(config)
         .with_runtime_env(rt)
         .with_default_features();
 
 
@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
+use std::num::NonZeroUsize;
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::Instant;
@@ -62,8 +63,9 @@ struct Args {
     #[arg(short, long, default_value_t = 5)]
     iterations: usize,
 
+    /// Number of Tokio workers and DataFusion execution partitions.
     #[arg(short, long)]
-    threads: Option<usize>,
+    threads: Option<NonZeroUsize>,
 
     #[arg(short, long)]
     verbose: bool,
@@ -128,9 +130,16 @@ struct Args {
     options: Vec<Opt>,
 }
 
-#[tokio::main]
-async fn main() -> anyhow::Result<()> {
+fn main() -> anyhow::Result<()> {
     let args = Args::parse();
+    let mut runtime = tokio::runtime::Builder::new_multi_thread();
+    if let Some(threads) = args.threads {
+        runtime.worker_threads(threads.get());
+    }
+    runtime.enable_all().build()?.block_on(run(args))
+}
+
+async fn run(args: Args) -> anyhow::Result<()> {
     let opts = Opts::from(args.options);
 
     set_join_set_tracer(get_static_tracer())?;
@@ -187,7 +196,7 @@ async fn main() -> anyhow::Result<()> {
             |format| {
                 let benchmark = &*benchmark;
                 async move {
-                    let session = datafusion_bench::get_session_context();
+                    let session = datafusion_bench::get_session_context(args.threads);
                     for sql in benchmark.engine_init_sql(Engine::DataFusion) {
                         session.sql(&sql).await?.collect().await?;
                     }