feat(tracing): add tracing spans to all I/O sections (#3795)

# Description
To better understand performance in the `delta-rs` crate, I added
additional tracing to capture more detailed debug-level performance
information.

Python now uses `OpenTelemetry` to collect tracing data emitted from
Rust.
With this change, we gain true end-to-end visibility: Python spans can
serve as parents of Rust spans (and vice versa), ensuring a continuous
trace across both runtimes.

# Related Issue(s)
-  close #3641

# Documentation
- https://docs.rs/tracing/latest/tracing/

---------

Signed-off-by: Florian Valeye <[email protected]>
Co-authored-by: Ion Koutsouris <[email protected]>

Author: fvaleye

Cargo.toml

@@ -72,6 +72,13 @@ uuid = { version = "1" }
 async-trait = { version = "0.1" }
 futures = { version = "0.3" }
 tokio = { version = "1" }
+
+# opentelemetry
+tracing-opentelemetry = { version = "0.32" }
+opentelemetry = { version = "0.31" }
+opentelemetry-otlp = { version = "0.31", features = ["http-proto"] }
+opentelemetry_sdk = { version = "0.31", features = ["rt-tokio"] }
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 num_cpus = { version = "1" }
 
 [workspace.metadata.typos]

crates/core/Cargo.toml

@@ -100,6 +100,7 @@ rstest = { version = "0.26.1" }
 serial_test = "3"
 tempfile = "3"
 tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 
 [features]
 default = ["rustls"]

crates/core/src/delta_datafusion/find_files.rs

@@ -13,6 +13,7 @@ use datafusion::physical_plan::filter::FilterExec;
 use datafusion::physical_plan::limit::LocalLimitExec;
 use datafusion::physical_plan::ExecutionPlan;
 use itertools::Itertools;
+use tracing::*;
 
 use crate::delta_datafusion::{
     df_logical_schema, get_path_column, DeltaScanBuilder, DeltaScanConfigBuilder, PATH_COLUMN,
@@ -31,6 +32,15 @@ pub(crate) struct FindFiles {
 }
 
 /// Finds files in a snapshot that match the provided predicate.
+#[instrument(
+    skip_all,
+    fields(
+        version = snapshot.version(),
+        has_predicate = predicate.is_some(),
+        partition_scan = field::Empty,
+        candidate_count = field::Empty
+    )
+)]
 pub(crate) async fn find_files(
     snapshot: &EagerSnapshot,
     log_store: LogStoreRef,
@@ -53,24 +63,35 @@ pub(crate) async fn find_files(
 
             if expr_properties.partition_only {
                 let candidates = scan_memory_table(snapshot, predicate).await?;
-                Ok(FindFiles {
+                let result = FindFiles {
                     candidates,
                     partition_scan: true,
-                })
+                };
+                Span::current().record("partition_scan", result.partition_scan);
+                Span::current().record("candidate_count", result.candidates.len());
+                Ok(result)
             } else {
                 let candidates =
                     find_files_scan(snapshot, log_store, session, predicate.to_owned()).await?;
 
-                Ok(FindFiles {
+                let result = FindFiles {
                     candidates,
                     partition_scan: false,
-                })
+                };
+                Span::current().record("partition_scan", result.partition_scan);
+                Span::current().record("candidate_count", result.candidates.len());
+                Ok(result)
             }
         }
-        None => Ok(FindFiles {
-            candidates: snapshot.log_data().iter().map(|f| f.add_action()).collect(),
-            partition_scan: true,
-        }),
+        None => {
+            let result = FindFiles {
+                candidates: snapshot.log_data().iter().map(|f| f.add_action()).collect(),
+                partition_scan: true,
+            };
+            Span::current().record("partition_scan", result.partition_scan);
+            Span::current().record("candidate_count", result.candidates.len());
+            Ok(result)
+        }
     }
 }
 
@@ -188,6 +209,14 @@ fn join_batches_with_add_actions(
 }
 
 /// Determine which files contain a record that satisfies the predicate
+#[instrument(
+    skip_all,
+    fields(
+        version = snapshot.version(),
+        total_files = field::Empty,
+        matching_files = field::Empty
+    )
+)]
 async fn find_files_scan(
     snapshot: &EagerSnapshot,
     log_store: LogStoreRef,
@@ -204,6 +233,8 @@ async fn find_files_scan(
         })
         .collect();
 
+    Span::current().record("total_files", candidate_map.len());
+
     let scan_config = DeltaScanConfigBuilder::default()
         .with_file_column(true)
         .build(snapshot)?;
@@ -240,12 +271,15 @@ async fn find_files_scan(
     let task_ctx = Arc::new(TaskContext::from(session));
     let path_batches = datafusion::physical_plan::collect(limit, task_ctx).await?;
 
-    join_batches_with_add_actions(
+    let result = join_batches_with_add_actions(
         path_batches,
         candidate_map,
         config.file_column_name.as_ref().unwrap(),
         true,
-    )
+    )?;
+
+    Span::current().record("matching_files", result.len());
+    Ok(result)
 }
 
 async fn scan_memory_table(snapshot: &EagerSnapshot, predicate: &Expr) -> DeltaResult<Vec<Add>> {