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,7 @@ 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, level = "debug")]
     pub(crate) async fn get_manifest(&self, manifest_file: &ManifestFile) -> Result<Arc<Manifest>> {
         if self.cache_disabled {
             return manifest_file

crates/iceberg/src/io/object_cache.rs

@@ -19,6 +19,7 @@ use std::sync::Arc;
 
 use futures::StreamExt;
 use futures::stream::BoxStream;
+use tracing::field::Empty;
 
 use crate::delete_file_index::DeleteFileIndex;
 use crate::expr::{Bind, BoundPredicate, Predicate};
@@ -53,14 +54,26 @@ 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 _guard = manifest_span.enter();
+
         let ManifestFileContext {
             object_cache,
             manifest_file,
@@ -73,15 +86,30 @@ impl ManifestFileContext {
 
         let manifest = object_cache.get_manifest(&manifest_file).await?;
 
+        span.record(
+            "iceberg.scan.plan.manifest.entries_count",
+            manifest.entries().len(),
+        );
+
         Ok(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,
                 });
             }
         }
@@ -144,7 +172,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 +207,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 +235,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(