[adriangb]

Note

Draft prototype stacked on #27022. Only the last two commits are new; review those.

Which issue does this PR close?

• Part of [EPIC] Improve I/O patterns for Parquet scans #26682
• Prototype for the suggestion in test(parquet): add I/O access tests for parquet pruning, filter pushdown and TopK #27022 (review)

Rationale for this change

The I/O tests in #27022 only see object store requests. They cannot say which scan step caused a request. The review suggested tracing::debug! events at important places, so tests can snapshot planning and scan operations as well as I/O.

What changes are included in this PR?

1. datafusion-datasource-parquet depends on tracing (no default features). The scan emits spans and events at these places:
   • opening a file
   • loading metadata, with metadata cache hits and misses
   • loading the page index and the Bloom filters, with one event per Bloom filter read
   • row group pruning with statistics and with Bloom filters, and page index pruning
   • column chunk fetches and row filter evaluation
2. Each call site records its own fields. The library formats no strings for tests.
3. The test harness (core/tests/parquet/io/trace_capture.rs) is a tracing_subscriber layer. It renders spans and events as an indented tree and prints every recorded field. The test object store emits a span for every operation and names the Parquet structure that each byte range covers.

The tests look like this:

let test = filter_pushdown_test(true).await;
let traces = test.capture_traces();
test.query("SELECT * FROM t WHERE val = 7").await.unwrap();
assert_snapshot!(traces.render(), @"
Query (sql=SELECT * FROM t WHERE val = 7, rows=1)
    ...
    Fetch column chunk data (ranges=1)
        ObjectStore::get_ranges (location=table.parquet, ranges=1)
            Read byte range (range=145..192, parquet=dictionary page and data page 0 for column 'val' (rg 1))
    Evaluate row filter (predicate=val@0 = 7, rows=3, rows_matched=1)
    ...
");

Two findings came out of this:

• In test(parquet): add I/O access tests for parquet pruning, filter pushdown and TopK #27022, "Request 1: Read Parquet footer (offsets 0..678)" is really the listing table's HEAD request. The store reports a GET with head: true as a read of the whole file.
• Planning loads the page index into the cached metadata before the scan opens the file.

What is the testing strategy for this PR?

The 14 snapshot tests in datafusion/core/tests/parquet/io/ were converted to the new API and regenerated with cargo insta accept.

The capture uses tracing::subscriber::set_default, which only applies to the current thread. This works because #[tokio::test] runs every task on the test thread.

Are there any user-facing changes?

There is a new tracing dependency in datafusion-datasource-parquet. Events are at debug level and cost almost nothing when no subscriber is installed. The API is unchanged.

🤖 Generated with Claude Code

[alamb]

Thank you @adriangb -- I really like this idea of instrumenting the parquet opener with key events to understand it better

However, I don't like the use of the tracing crate as I don't think everyone uses tracing (some use otel, for example, and we have our own thing at InfluxData)

I wonder if we could instead add our own tracing/span API in DataFusion , maybe something like

/// Receives scan events; default impl is a no-op
pub trait ParquetScanObserver: Debug + Send + Sync {
    /// Called at a point of interest with structured fields
    fn event(&self, event: ParquetScanEvent<'_>) {}
    /// Begin a nested region; dropping the guard ends it
    fn span(&self, span: ParquetScanSpan<'_>) -> Option<Box<dyn Any + Send>> { None }
}

pub enum ParquetScanEvent<'a> {
    OpenFile { path: &'a str, size: u64 },
    RowGroupsPruned { stage: PruneStage, before: usize, after: usize },
    BloomFilterRead { column: &'a str, row_group: usize, range: Range<i64> },
    RowFilterEvaluated { predicate: &'a dyn PhysicalExpr, rows: usize, matched: usize },
}

[alamb]

A nit is I would probably combine this into one line

            ObjectStore::get (location=table.parquet)
                Read byte range (range=192..213, parquet=OffsetIndex for column 'a' (rg 0))

into

            ObjectStore::get (location=table.parquet) (range=192..213, parquet=OffsetIndex for column 'a' (rg 0))

[alamb]

this is cool to show the metadata cache hit

[alamb]

this is my biggest hesitation -- to add this new tracing dependency (I know it is just a dev-dependency, but still I feel like it would be easy to accidentally turn into a real dependnecy, and not everyone uses the tracing crate

[alamb]

I really like the idea of instrumenting the parquet opener, but I don't like tying the instrumentation to the tracing crate

[alamb]

I think this is a bad idea to pick up the tracing dependency

[alamb]

As an alternative, we could maybe even add a more general API for tracing to cover other important events (like the sql planning, analyzer, optimizer passes, etc)

Something like

#[non_exhaustive]
pub enum TraceEvent<'a> {
    // ---- planning ----
    /// Span: parsing SQL text into a statement
    ParseSql { sql: &'a str },
    /// Event: one analyzer rule ran
    AnalyzerRule { rule: &'a str, plan: &'a dyn Display },
...
    // ---- parquet scan ----
    /// Event: the scan started opening a file
    ParquetOpenFile { path: &'a str, size: u64 },
    /// Span: footer metadata load; `cached` says whether the metadata cache hit
    ParquetLoadMetadata { path: &'a str, cached: bool },
...
}

/// Registered once on the session; every subsystem reports to it
pub trait TraceObserver: Debug + Send + Sync {
...
}

// ---- registration: one place ----
let ctx = SessionContext::new_with_state(
    SessionStateBuilder::new()
        .with_trace_observer(Arc::new(MyObserver))   // stored on SessionState
        .build(),
);
// SessionState::task_ctx() copies it onto TaskContext, like the function registry.

Then we could instrument call sites like

Planner

// EXPLAIN VERBOSE and tracing consume the same points.
self.optimizer().optimize(plan, self, |plan, rule| {
    observer.event(&TraceEvent::OptimizerRule { rule: rule.name(), pass, transformed, plan });
})?;

or in the Parquet opener

observer.event(&TraceEvent::ParquetOpenFile { path: &file_name, size });
...
let _guard = observer.span(&TraceEvent::ParquetRowGroupsPruned { .. }); // sync regio

[adriangb]

I understand the hesitation to bring in tracing. As you already pointed our right now this is a dev only dep so there is zero impact / direct decision being made about what event / tracing API Datafusion exposes more generally. I think we both agree that it is somewhat of a slippery slope, but only in the sense that it's easy to say "hey all of these operations are instrumented, why not just expose them to users!".

That said, if we do want to discuss what should Datafusion offer as it's event / tracing implementation I see 3 options:

1. Pick a single crate (which IMO should be tracing or opentelemetry given https://lib.rs/crates/ numbers). You can bridge between them, e.g. we use opentelemetry but have plenty of crates that offer only tracing instrumentation. For example using https://github.com/tokio-rs/tracing-opentelemetry.
2. Offer multiple instrumentation. We offer feature flags for various instrumentations. You can turn on one or many. It'd be pretty straightforward to tell an LLM to add a new instrumentation. This is mechanical code that would not be hard to maintain and doesn't need to be perfect. Realistically I think this is 2-3 instrumenations. Companies that use custom instrumentations would need to build a bridge between a standard open source instrumenation and their internal implementation (which IMO is a reasonable ask).
3. Create our own events API. IMO this requires design work, compatibility guarantees, etc. And every single consumer needs to build some sort of bridge into whatever they use for tracing. I find this to be the worst of all worlds: more work for us, more work for users.

Thus my vote would be to:

1. If we find this approach useful for testing (and maybe we don't, that should be discussed more on it's own merits) make changes to it / touch it up and adopt it.
2. Gradually expand to planning, other areas of the codebase like you indicated & add utilities like filtering events, selecting which attributes to show, other views of the data that we can snapshot in tests.
3. If it makes sense we can later decide to make this a feature flagged feature exposed to users.
4. If users request other integrations e.g. opentelemetry instead of tracing (and there are good arguments for not just using a bridge) we can add a second feature flag.