feat: update scan plan instrumentation. Add o11y integ test. Add Jaeger to integ test docker stack

Author: sdd

Cargo.lock

@@ -43,7 +43,7 @@ impl ExpressionEvaluator {
     /// the provided [`DataFile`]'s partition [`Struct`]. Used by [`TableScan`]
     /// to see if this [`DataFile`] could possibly contain data that matches
     /// the scan's filter.
-    #[tracing::instrument(skip_all)]
+    #[tracing::instrument(skip_all, level = "trace")]
     pub(crate) fn eval(&self, data_file: &DataFile) -> Result<bool> {
         let mut visitor = ExpressionEvaluatorVisitor::new(data_file.partition());
 

crates/iceberg/src/expr/visitors/expression_evaluator.rs

@@ -39,7 +39,7 @@ impl<'a> InclusiveMetricsEvaluator<'a> {
     /// provided [`DataFile`]'s metrics. Used by [`TableScan`] to
     /// see if this `DataFile` contains data that could match
     /// the scan's filter.
-    #[tracing::instrument(skip_all)]
+    #[tracing::instrument(skip_all, level = "trace")]
     pub(crate) fn eval(
         filter: &'a BoundPredicate,
         data_file: &'a DataFile,

crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs

@@ -85,7 +85,12 @@ impl ObjectCache {
 
     /// Retrieves an Arc [`Manifest`] from the cache
     /// or retrieves one from FileIO and parses it if not present
-    #[tracing::instrument(skip_all)]
+    #[tracing::instrument(
+        skip_all,
+        name = "iceberg.object_cache.get_manifest"
+        level = "debug",
+        fields(iceberg.object_cache.manifest.cache_hit)
+    )]
     pub(crate) async fn get_manifest(&self, manifest_file: &ManifestFile) -> Result<Arc<Manifest>> {
         if self.cache_disabled {
             return manifest_file
@@ -102,13 +107,23 @@ impl ObjectCache {
             .or_try_insert_with(self.fetch_and_parse_manifest(manifest_file))
             .await
             .map_err(|err| {
+                tracing::error!(
+                    iceberg.error_msg = %err.to_string(),
+                    "failed to load manifest from cache"
+                );
                 Error::new(
                     ErrorKind::Unexpected,
                     format!("Failed to load manifest {}", manifest_file.manifest_path),
                 )
                 .with_source(err)
-            })?
-            .into_value();
+            })?;
+
+        tracing::Span::current().record(
+            "iceberg.object_cache.manifest.cache_hit",
+            cache_entry.is_fresh(),
+        );
+
+        let cache_entry = cache_entry.into_value();
 
         match cache_entry {
             CachedItem::Manifest(arc_manifest) => Ok(arc_manifest),

crates/iceberg/src/io/object_cache.rs

@@ -19,6 +19,8 @@ use std::sync::Arc;
 
 use futures::StreamExt;
 use futures::stream::BoxStream;
+use tracing::Instrument;
+use tracing::field::Empty;
 
 use crate::delete_file_index::DeleteFileIndex;
 use crate::expr::{Bind, BoundPredicate, Predicate};
@@ -31,6 +33,7 @@ use crate::spec::{
     ManifestContentType, ManifestEntryRef, ManifestFile, ManifestList, SchemaRef, SnapshotRef,
     TableMetadataRef,
 };
+use crate::traced_stream::TracedStream;
 use crate::{Error, ErrorKind, Result};
 
 /// Wraps a [`ManifestFile`] alongside the objects that are needed
@@ -53,14 +56,25 @@ pub(crate) struct ManifestEntryContext {
     pub bound_predicates: Option<Arc<BoundPredicates>>,
     pub partition_spec_id: i32,
     pub snapshot_schema: SchemaRef,
+    pub(crate) span: tracing::Span,
 }
 
 impl ManifestFileContext {
     /// Consumes this [`ManifestFileContext`], fetching its Manifest from FileIO and then
     /// streaming its constituent [`ManifestEntries`]
     pub(crate) async fn fetch_manifest_and_stream_entries(
         self,
+        parent_span: tracing::Span,
     ) -> Result<BoxStream<'static, Result<ManifestEntryContext>>> {
+        let manifest_span = tracing::debug_span!(
+            parent: &parent_span,
+            "iceberg.scan.plan.process_manifest",
+            iceberg.scan.plan.manifest.file_path = self.manifest_file.manifest_path,
+            iceberg.scan.plan.manifest.entries_count = Empty,
+        );
+
+        let span = manifest_span.clone();
+
         let ManifestFileContext {
             object_cache,
             manifest_file,
@@ -71,21 +85,46 @@ impl ManifestFileContext {
             ..
         } = self;
 
-        let manifest = object_cache.get_manifest(&manifest_file).await?;
+        let (manifest, manifest_file) = async move {
+            let manifest = object_cache.get_manifest(&manifest_file).await;
+            (manifest, manifest_file)
+        }
+        .instrument(manifest_span.clone())
+        .await;
+        let manifest = manifest?;
+
+        span.record(
+            "iceberg.scan.plan.manifest.entries_count",
+            manifest.entries().len(),
+        );
 
-        Ok(async_stream::stream! {
+        let stream = async_stream::stream! {
             for manifest_entry in manifest.entries() {
+                let manifest_entry_span = tracing::debug_span!(
+                    parent: span.clone(),
+                    "iceberg.scan.plan.process_data_file",
+                    iceberg.scam.plan.data_file.file_path = manifest_entry.file_path(),
+                    "iceberg.scan.plan_data_file.type" = Empty,
+                    iceberg.scan.plan.data_file.skipped = Empty,
+                    iceberg.scan.plan.data_file.skipped_reason = Empty,
+                );
+
                 yield Ok(ManifestEntryContext {
                     manifest_entry: manifest_entry.clone(),
                     expression_evaluator_cache: expression_evaluator_cache.clone(),
                     field_ids: field_ids.clone(),
                     partition_spec_id: manifest_file.partition_spec_id,
                     bound_predicates: bound_predicates.clone(),
                     snapshot_schema: snapshot_schema.clone(),
+                    span: manifest_entry_span,
                 });
             }
         }
-        .boxed())
+        .boxed();
+
+        Ok(Box::pin(TracedStream::new(stream, vec![
+            manifest_span.clone(),
+        ])))
     }
 
     pub(crate) fn is_delete(&self) -> bool {
@@ -144,7 +183,11 @@ pub(crate) struct PlanContext {
 }
 
 impl PlanContext {
-    #[tracing::instrument(skip_all)]
+    #[tracing::instrument(
+        skip_all,
+        level = "debug",
+        fields(iceberg.scan.plan.manifest_list.file_path = ?self.snapshot.manifest_list()),
+    )]
     pub(crate) async fn get_manifest_list(&self) -> Result<Arc<ManifestList>> {
         self.object_cache
             .as_ref()
@@ -175,14 +218,19 @@ impl PlanContext {
 
     #[tracing::instrument(
         skip_all,
+        level = "debug",
+        name = "iceberg.scan.plan.process_manifest_list",
         fields(
-            manifest_list.len = manifest_list.entries().len(),
+            iceberg.scan.plan.manifest_list.entries_count = manifest_list.entries().len(),
         )
     )]
-    pub(crate) fn build_manifest_file_context_iter(
+    pub(crate) fn build_manifest_file_contexts(
         &self,
         manifest_list: Arc<ManifestList>,
-    ) -> impl Iterator<Item = Result<ManifestFileContext>> {
+    ) -> (
+        Vec<Result<ManifestFileContext>>,
+        Vec<Result<ManifestFileContext>>,
+    ) {
         let has_predicate = self.predicate.is_some();
 
         (0..manifest_list.entries().len())
@@ -198,24 +246,28 @@ impl PlanContext {
                             .get(manifest_file.partition_spec_id, predicate.clone())
                             .eval(&manifest_file)?
                         {
-                            tracing::trace!(file_path = manifest_file.manifest_path, "iceberg.scan.manifest_file.skipped");
-                            metrics::counter!("iceberg.scan.manifest_file.skipped", "reason" => "partition").increment(1);
+                            tracing::debug!(
+                                iceberg.scan.plan.manifest.file_path = manifest_file.manifest_path,
+                                iceberg.scan.plan.manifest.skip_reason = "partition",
+                                "iceberg.scan.plan.manifest_file.skipped"
+                            );
+                            metrics::counter!("iceberg.scan.plan.manifest_file.skipped", "reason" => "partition").increment(1);
                             return Ok(None); // Skip this file.
                         }
                         Some(predicate)
                     } else {
                         None
                     };
 
-                    tracing::trace!(file_path = manifest_file.manifest_path, "iceberg.scan.manifest_file.included");
-                    metrics::counter!("iceberg.scan.manifest_file.included").increment(1);
+                    metrics::counter!("iceberg.scan.plan.manifest_file.included").increment(1);
 
                     let context = self
                         .create_manifest_file_context(manifest_file, partition_bound_predicate)?;
                     Ok(Some(context))
                 })()
                 .transpose()
             })
+            .partition(|ctx| ctx.as_ref().map_or(true, |ctx| ctx.is_delete()))
     }
 
     fn create_manifest_file_context(