Add tracing support to the compressor (#7385)

## Summary

Tracking issue: #7216

We have very little observability into the compressor. When we are
debugging, we don't really have any idea of what schemes the compressor
is trying, how good or how bad estimates are, how reliable sampling is,
how the cascading paths look, etc.

This change adds structured `tracing` support to `vortex-compressor`.
The compressor now emits a top-level `compress` span and decision/debug
events on the `vortex_compressor::encode` target, so a normal tracing
subscriber can see what the compressor sampled, selected,
accepted/rejected, and where nested failures happened.

The `scheme.compress_result` event reports `scheme`, `before_nbytes`,
`after_nbytes`, `estimated_ratio` when available, `actual_ratio` when
available, and `accepted`. Sampling is recorded through `sample.result`;
compression failures are recorded through `scheme.compress_failed` /
`sample.compress_failed` with `cascade_path` and `cascade_depth`.
Zero-byte outputs intentionally omit ratio fields instead of logging
infinities.

This also adds JSON formatting to the benchmark logging setup via
`--log-format json`, which makes `data-gen` / `compress-bench` output
usable as JSONL. One useful workflow is to generate TPC-H data with
compressor logs enabled and use `jq` to find over-optimistic estimates
that were rejected.

<details>
<summary>Example jq query for rejected over-estimates</summary>

```fish
set LOG data-gen.jsonl

jq -R -s -r '
  def rows: split("\n") | map(fromjson? // empty);
  def r:
    if type == "number" then ((. * 1000 | round) / 1000)
    else .
    end;

  ([
    "estimated_over_actual",
    "scheme",
    "estimated_ratio",
    "actual_ratio",
    "before_nbytes",
    "after_nbytes",
    "extra_bytes",
    "span_dtype",
    "span_len"
  ] | @TSV),
  (
    rows
    | map(select(.target == "vortex_compressor::encode"))
    | map(select(.fields.message == "scheme.compress_result"))
    | map(select(.fields.accepted == false))
    | map(select(.fields.estimated_ratio != null and .fields.actual_ratio != null))
    | map(select(.fields.estimated_ratio > .fields.actual_ratio))
    | map(.fields as $f | {
        scheme: $f.scheme,
        estimated_ratio: $f.estimated_ratio,
        actual_ratio: $f.actual_ratio,
        estimated_over_actual: ($f.estimated_ratio / $f.actual_ratio),
        before_nbytes: $f.before_nbytes,
        after_nbytes: $f.after_nbytes,
        extra_bytes: ($f.after_nbytes - $f.before_nbytes),
        span_dtype: (.span.dtype // ""),
        span_len: (.span.len // "")
      })
    | sort_by(.estimated_over_actual) | reverse
    | .[:50][]
    | [
        (.estimated_over_actual | r),
        .scheme,
        (.estimated_ratio | r),
        (.actual_ratio | r),
        .before_nbytes,
        .after_nbytes,
        .extra_bytes,
        .span_dtype,
        .span_len
      ]
    | @TSV
  )
' $LOG
```

```
estimated_over_actual	scheme	estimated_ratio	actual_ratio	before_nbytes	after_nbytes	extra_bytes	span_dtype	span_len
512	vortex.int.for	1.6	0.003	2	640	638	utf8?	2
512	vortex.int.for	2.667	0.005	2	384	382	utf8?	2
512	vortex.int.for	5.333	0.01	8	768	760	decimal(15,2)?	2
512	vortex.int.for	4.571	0.009	8	896	888	decimal(15,2)?	2
512	vortex.int.for	4	0.008	8	1024	1016	decimal(15,2)?	2
512	vortex.int.for	1.6	0.003	2	640	638	utf8?	2
512	vortex.int.for	8	0.016	2	128	126	utf8?	2
512	vortex.int.for	2.56	0.005	16	3200	3184	i64?	2
512	vortex.int.for	5.818	0.011	16	1408	1392	i64?	2
256	vortex.int.for	2	0.008	4	512	508	utf8	4
256	vortex.int.for	2	0.008	4	512	508	utf8	4
256	vortex.int.for	2	0.008	4	512	508	utf8	8192
256	vortex.int.for	2	0.008	4	512	508	utf8	8192
256	vortex.int.for	2	0.008	4	512	508	utf8	8192
256	vortex.int.for	2	0.008	4	512	508	utf8	8192
256	vortex.int.for	2	0.008	4	512	508	utf8	8192
256	vortex.int.for	2	0.008	4	512	508	utf8	8192
204.8	vortex.int.for	4	0.02	5	256	251	utf8	5
204.8	vortex.int.for	4	0.02	5	256	251	utf8	8192
204.8	vortex.int.for	4	0.02	5	256	251	utf8	8192
204.8	vortex.int.for	2.667	0.013	5	384	379	utf8	5
204.8	vortex.int.for	2.667	0.013	5	384	379	utf8	5
146.286	vortex.int.for	1.333	0.009	7	768	761	utf8?	19
146.286	vortex.int.for	1.333	0.009	7	768	761	utf8?	19
128	vortex.int.for	2	0.016	8	512	504	utf8?	98
128	vortex.int.for	2	0.016	8	512	504	utf8?	98
113.778	vortex.int.for	1.6	0.014	18	1280	1262	i32?	184
113.778	vortex.int.for	1.6	0.014	18	1280	1262	i32?	184
113.778	vortex.int.for	32	0.281	36	128	92	i32?	184
113.778	vortex.int.for	32	0.281	36	128	92	i32?	184
64	vortex.int.for	2	0.031	16	512	496	utf8?	98
64	vortex.int.for	2	0.031	16	512	496	utf8?	98
64	vortex.int.for	2	0.031	16	512	496	utf8?	98
64	vortex.int.for	2	0.031	16	512	496	utf8?	98
53.895	vortex.int.for	1.6	0.03	19	640	621	utf8?	19
53.895	vortex.int.for	3.2	0.059	76	1280	1204	decimal(15,2)?	19
53.895	vortex.int.for	2.909	0.054	76	1408	1332	decimal(15,2)?	19
53.895	vortex.int.for	1.6	0.03	19	640	621	utf8?	19
53.895	vortex.int.for	1.6	0.03	19	640	621	utf8?	19
53.895	vortex.int.for	32	0.594	76	128	52	i32?	184
53.895	vortex.int.for	32	0.594	76	128	52	i32?	184
53.895	vortex.int.for	4	0.074	76	1024	948	decimal(15,2)?	19
51.2	vortex.int.for	2	0.039	20	512	492	utf8?	98
51.2	vortex.int.for	2	0.039	20	512	492	utf8?	98
51.2	vortex.int.for	2	0.039	20	512	492	utf8?	98
51.2	vortex.int.for	2	0.039	20	512	492	utf8?	98
51.2	vortex.int.for	2	0.039	20	512	492	utf8?	98
51.2	vortex.int.for	2	0.039	20	512	492	utf8?	98
40.96	vortex.int.for	2	0.049	25	512	487	utf8?	25
40.96	vortex.int.for	2	0.049	25	512	487	utf8?	25
```

</details>

## Testing

Some basic tracing tests (that was claude-generated).

---------

Signed-off-by: Connor Tsui <connor.tsui20@gmail.com>

Author: connortsui20

Cargo.lock

@@ -15,6 +15,7 @@ use regex::Regex;
 use vortex::utils::aliases::hash_map::HashMap;
 use vortex_bench::Engine;
 use vortex_bench::Format;
+use vortex_bench::LogFormat;
 use vortex_bench::Target;
 use vortex_bench::compress::CompressMeasurements;
 use vortex_bench::compress::CompressOp;
@@ -39,7 +40,7 @@ use vortex_bench::public_bi::PBIDataset::CMSprovider;
 use vortex_bench::public_bi::PBIDataset::Euro2016;
 use vortex_bench::public_bi::PBIDataset::Food;
 use vortex_bench::public_bi::PBIDataset::HashTags;
-use vortex_bench::setup_logging_and_tracing;
+use vortex_bench::setup_logging_and_tracing_with_format;
 
 #[derive(Parser, Debug)]
 #[command(version, about, long_about = None)]
@@ -69,13 +70,17 @@ struct Args {
     output_path: Option<PathBuf>,
     #[arg(long)]
     tracing: bool,
+    /// Format for the primary stderr log sink. `text` is the default human-readable format;
+    /// `json` emits one JSON object per event, suitable for piping into `jq`.
+    #[arg(long, value_enum, default_value_t = LogFormat::Text)]
+    log_format: LogFormat,
 }
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     let args = Args::parse();
 
-    setup_logging_and_tracing(args.verbose, args.tracing)?;
+    setup_logging_and_tracing_with_format(args.verbose, args.tracing, args.log_format)?;
 
     run_compress(
         args.iterations,

benchmarks/compress-bench/src/main.rs

@@ -61,6 +61,7 @@ tracing = { workspace = true }
 tracing-perfetto = { workspace = true }
 tracing-subscriber = { workspace = true, features = [
     "env-filter",
+    "json",
     "tracing-log",
 ] }
 url = { workspace = true }

vortex-bench/Cargo.toml

@@ -17,12 +17,13 @@ use vortex_bench::Benchmark;
 use vortex_bench::BenchmarkArg;
 use vortex_bench::CompactionStrategy;
 use vortex_bench::Format;
+use vortex_bench::LogFormat;
 use vortex_bench::Opt;
 use vortex_bench::Opts;
 use vortex_bench::conversions::convert_parquet_directory_to_vortex;
 use vortex_bench::create_benchmark;
 use vortex_bench::generate_duckdb_registration_sql;
-use vortex_bench::setup_logging_and_tracing;
+use vortex_bench::setup_logging_and_tracing_with_format;
 
 #[derive(Parser)]
 #[command(name = "bench-data-gen")]
@@ -37,6 +38,11 @@ struct Args {
     #[arg(long)]
     tracing: bool,
 
+    /// Format for the primary stderr log sink. `text` is the default human-readable format;
+    /// `json` emits one JSON object per event, suitable for piping into `jq`.
+    #[arg(long, value_enum, default_value_t = LogFormat::Text)]
+    log_format: LogFormat,
+
     #[arg(long, value_delimiter = ',', value_parser = value_parser!(Format))]
     formats: Vec<Format>,
 
@@ -49,7 +55,7 @@ async fn main() -> anyhow::Result<()> {
     let args = Args::parse();
     let opts = Opts::from(args.options);
 
-    setup_logging_and_tracing(args.verbose, args.tracing)?;
+    setup_logging_and_tracing_with_format(args.verbose, args.tracing, args.log_format)?;
 
     let benchmark = create_benchmark(args.benchmark, &opts)?;
 

vortex-bench/src/bin/data-gen.rs

@@ -4,34 +4,77 @@
 use std::fs::File;
 use std::io::IsTerminal;
 
+use clap::ValueEnum;
 use tracing::level_filters::LevelFilter;
 use tracing_perfetto::PerfettoLayer;
 use tracing_subscriber::EnvFilter;
+use tracing_subscriber::Layer;
 use tracing_subscriber::prelude::*;
 
-/// Initialize logging/tracing for a benchmark
-pub fn setup_logging_and_tracing(verbose: bool, tracing: bool) -> anyhow::Result<()> {
+/// Format for the primary stderr log sink.
+///
+/// `Text` is the default human-readable formatter matching the historical behavior of this crate.
+/// `Json` emits one newline-delimited JSON object per event, suitable for piping into `jq` or a log
+/// aggregator.
+#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, ValueEnum)]
+pub enum LogFormat {
+    #[default]
+    Text,
+    Json,
+}
+
+/// Initialize logging/tracing for a benchmark, hardcoding [`LogFormat::Text`].
+///
+/// See [`setup_logging_and_tracing_with_format`] if you want to select JSON
+/// output from a CLI flag.
+pub fn setup_logging_and_tracing(verbose: bool, perfetto: bool) -> anyhow::Result<()> {
+    setup_logging_and_tracing_with_format(verbose, perfetto, LogFormat::Text)
+}
+
+/// Initialize logging/tracing for a benchmark with an explicit stderr format.
+///
+/// - `verbose`: when `RUST_LOG` is unset, raises the default filter from `INFO` to `TRACE`. Has no
+///   effect when `RUST_LOG` is set (the env var wins).
+/// - `perfetto`: when `true`, additionally attaches a [`tracing_perfetto::PerfettoLayer`] that
+///   writes span begin/end events to `trace.json` in the current directory. Intended to be loaded
+///   into the Perfetto UI for flamegraph visualization.
+/// - `format`: controls the primary stderr sink's formatting. See [`LogFormat`].
+pub fn setup_logging_and_tracing_with_format(
+    verbose: bool,
+    perfetto: bool,
+    format: LogFormat,
+) -> anyhow::Result<()> {
     let filter = default_env_filter(verbose);
 
-    let fmt_layer = tracing_subscriber::fmt::layer()
-        .with_writer(std::io::stderr)
-        .with_level(true)
-        .with_file(true)
-        .with_line_number(true)
-        .with_ansi(std::io::stderr().is_terminal());
+    let perfetto_layer = perfetto
+        .then(|| {
+            Ok::<_, anyhow::Error>(
+                PerfettoLayer::new(File::create("trace.json")?).with_debug_annotations(true),
+            )
+        })
+        .transpose()?;
+
+    // `fmt::layer()` and `fmt::layer().json()` produce different concrete types,
+    // so erase each to a `dyn Layer` via `.boxed()` and keep the registry uniform.
+    let fmt_layer: Box<dyn Layer<_> + Send + Sync> = match format {
+        LogFormat::Text => tracing_subscriber::fmt::layer()
+            .with_writer(std::io::stderr)
+            .with_level(true)
+            .with_file(true)
+            .with_line_number(true)
+            .with_ansi(std::io::stderr().is_terminal())
+            .boxed(),
+        LogFormat::Json => tracing_subscriber::fmt::layer()
+            .json()
+            .with_writer(std::io::stderr)
+            .with_current_span(true)
+            .with_span_list(true)
+            .boxed(),
+    };
 
     tracing_subscriber::registry()
         .with(filter)
-        .with(
-            tracing
-                .then(|| {
-                    Ok::<_, anyhow::Error>(
-                        PerfettoLayer::new(File::create("trace.json")?)
-                            .with_debug_annotations(true),
-                    )
-                })
-                .transpose()?,
-        )
+        .with(perfetto_layer)
         .with(fmt_layer)
         .init();
 

vortex-bench/src/utils/logging.rs

@@ -307,8 +307,6 @@ impl Scheme for ZigZagScheme {
 
         let compressed = compressor.compress_child(&encoded.into_array(), &ctx, self.id(), 0)?;
 
-        tracing::debug!("zigzag output: {}", compressed.encoding_id());
-
         Ok(ZigZag::try_new(compressed)?.into_array())
     }
 }

vortex-btrblocks/src/schemes/integer.rs

@@ -19,7 +19,7 @@ num-traits = { workspace = true }
 parking_lot = { workspace = true }
 rand = { workspace = true }
 rustc-hash = { workspace = true }
-tracing = { workspace = true }
+tracing = { workspace = true, features = ["std", "attributes"] }
 vortex-array = { workspace = true }
 vortex-buffer = { workspace = true }
 vortex-error = { workspace = true }
@@ -29,6 +29,7 @@ vortex-utils = { workspace = true }
 [dev-dependencies]
 divan = { workspace = true }
 rstest = { workspace = true }
+tracing-subscriber = { workspace = true, features = ["env-filter"] }
 vortex-array = { workspace = true, features = ["_test-harness"] }
 
 [lints]