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

Signed-off-by: Florian Valeye <[email protected]>

Author: fvaleye

crates/core/Cargo.toml

@@ -99,6 +99,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/mod.rs

@@ -936,6 +936,14 @@ fn join_batches_with_add_actions(
 }
 
 /// Determine which files contain a record that satisfies the predicate
+#[tracing::instrument(
+    skip_all,
+    fields(
+        version = snapshot.version(),
+        total_files = tracing::field::Empty,
+        matching_files = tracing::field::Empty
+    )
+)]
 pub(crate) async fn find_files_scan(
     snapshot: &DeltaTableState,
     log_store: LogStoreRef,
@@ -948,6 +956,8 @@ pub(crate) async fn find_files_scan(
         .try_collect()
         .await?;
 
+    tracing::Span::current().record("total_files", candidate_map.len());
+
     let scan_config = DeltaScanConfigBuilder {
         include_file_column: true,
         ..Default::default()
@@ -987,12 +997,20 @@ pub(crate) async fn find_files_scan(
     let task_ctx = Arc::new(TaskContext::from(state));
     let path_batches = datafusion::physical_plan::collect(limit, task_ctx).await?;
 
-    join_batches_with_add_actions(
+    let matching_files = join_batches_with_add_actions(
         path_batches,
         candidate_map,
         config.file_column_name.as_ref().unwrap(),
         true,
-    )
+    )?;
+
+    tracing::Span::current().record("matching_files", matching_files.len());
+    tracing::debug!(
+        matching_files = matching_files.len(),
+        "physical scan completed"
+    );
+
+    Ok(matching_files)
 }
 
 pub(crate) async fn scan_memory_table(
@@ -1051,6 +1069,15 @@ pub(crate) async fn scan_memory_table(
 }
 
 /// Finds files in a snapshot that match the provided predicate.
+#[tracing::instrument(
+    skip_all,
+    fields(
+        version = snapshot.version(),
+        has_predicate = predicate.is_some(),
+        partition_scan = tracing::field::Empty,
+        candidate_count = tracing::field::Empty
+    )
+)]
 pub async fn find_files(
     snapshot: &DeltaTableState,
     log_store: LogStoreRef,
@@ -1072,25 +1099,52 @@ pub async fn find_files(
             expr_properties.result?;
 
             if expr_properties.partition_only {
+                tracing::Span::current().record("partition_scan", true);
+                tracing::debug!("using partition-only scan (memory table)");
                 let candidates = scan_memory_table(&log_store, snapshot, predicate).await?;
+                tracing::Span::current().record("candidate_count", candidates.len());
+                tracing::info!(
+                    candidates = candidates.len(),
+                    scan_type = "partition",
+                    "file pruning completed"
+                );
                 Ok(FindFiles {
                     candidates,
                     partition_scan: true,
                 })
             } else {
+                tracing::Span::current().record("partition_scan", false);
+                tracing::debug!("using physical data scan");
                 let candidates =
                     find_files_scan(snapshot, log_store, state, predicate.to_owned()).await?;
 
+                tracing::Span::current().record("candidate_count", candidates.len());
+                tracing::info!(
+                    candidates = candidates.len(),
+                    scan_type = "physical",
+                    "file pruning completed"
+                );
                 Ok(FindFiles {
                     candidates,
                     partition_scan: false,
                 })
             }
         }
-        None => Ok(FindFiles {
-            candidates: snapshot.file_actions(&log_store).await?,
-            partition_scan: true,
-        }),
+        None => {
+            tracing::Span::current().record("partition_scan", true);
+            tracing::debug!("no predicate, returning all files");
+            let candidates = snapshot.file_actions(&log_store).await?;
+            tracing::Span::current().record("candidate_count", candidates.len());
+            tracing::info!(
+                candidates = candidates.len(),
+                scan_type = "all",
+                "file pruning completed"
+            );
+            Ok(FindFiles {
+                candidates,
+                partition_scan: true,
+            })
+        }
     }
 }
 
@@ -2388,4 +2442,65 @@ mod tests {
             self.inner.rename_if_not_exists(from, to).await
         }
     }
+
+    #[test]
+    fn test_find_files_tracing_span() {
+        let span = tracing::info_span!(
+            "find_files",
+            version = 10,
+            has_predicate = true,
+            partition_scan = tracing::field::Empty,
+            candidate_count = tracing::field::Empty
+        );
+
+        let metadata = span.metadata().expect("span should have metadata");
+        assert_eq!(metadata.name(), "find_files");
+        assert_eq!(metadata.level(), &tracing::Level::INFO);
+        assert!(metadata.is_span());
+
+        span.record("partition_scan", true);
+        span.record("candidate_count", 45);
+    }
+
+    #[test]
+    fn test_find_files_scan_tracing_span() {
+        let span = tracing::info_span!(
+            "find_files_scan",
+            version = 10,
+            total_files = tracing::field::Empty,
+            matching_files = tracing::field::Empty
+        );
+
+        let metadata = span.metadata().expect("span should have metadata");
+        assert_eq!(metadata.name(), "find_files_scan");
+        assert_eq!(metadata.level(), &tracing::Level::INFO);
+
+        span.record("total_files", 1000);
+        span.record("matching_files", 12);
+    }
+
+    #[test]
+    fn test_find_files_result_structure() {
+        use crate::kernel::Add;
+
+        let find_result = FindFiles {
+            candidates: vec![Add {
+                path: "test.parquet".to_string(),
+                size: 1000,
+                modification_time: 0,
+                data_change: true,
+                stats: None,
+                partition_values: Default::default(),
+                tags: None,
+                deletion_vector: None,
+                base_row_id: None,
+                default_row_commit_version: None,
+                clustering_provider: None,
+            }],
+            partition_scan: true,
+        };
+
+        assert_eq!(find_result.candidates.len(), 1);
+        assert!(find_result.partition_scan);
+    }
 }

crates/core/src/kernel/transaction/mod.rs

@@ -619,14 +619,21 @@ impl<'a> std::future::IntoFuture for PreparedCommit<'a> {
             let commit_or_bytes = this.commit_or_bytes;
 
             if this.table_data.is_none() {
+                debug!("committing initial table version 0");
                 this.log_store
                     .write_commit_entry(0, commit_or_bytes.clone(), this.operation_id)
                     .await?;
                 return Ok(PostCommit {
                     version: 0,
                     data: this.data,
-                    create_checkpoint: false,
-                    cleanup_expired_logs: None,
+                    create_checkpoint: this
+                        .post_commit
+                        .map(|v| v.create_checkpoint)
+                        .unwrap_or_default(),
+                    cleanup_expired_logs: this
+                        .post_commit
+                        .map(|v| v.cleanup_expired_logs)
+                        .unwrap_or_default(),
                     log_store: this.log_store,
                     table_data: None,
                     custom_execute_handler: this.post_commit_hook_handler,
@@ -637,9 +644,20 @@ impl<'a> std::future::IntoFuture for PreparedCommit<'a> {
             // unwrap() is safe here due to the above check
             let mut read_snapshot = this.table_data.unwrap().eager_snapshot().clone();
 
+            let commit_span = info_span!(
+                "commit_with_retries",
+                base_version = read_snapshot.version(),
+                max_retries = this.max_retries,
+                attempt = field::Empty,
+                target_version = field::Empty,
+                conflicts_checked = 0
+            );
+            let _enter = commit_span.enter();
+
             let mut attempt_number = 1;
             let total_retries = this.max_retries + 1;
             while attempt_number <= total_retries {
+                commit_span.record("attempt", attempt_number);
                 let latest_version = this
                     .log_store
                     .get_latest_version(read_snapshot.version())
@@ -649,16 +667,28 @@ impl<'a> std::future::IntoFuture for PreparedCommit<'a> {
                     // If max_retries are set to 0, do not try to use the conflict checker to resolve the conflict
                     // and throw immediately
                     if this.max_retries == 0 {
+                        warn!(
+                            base_version = read_snapshot.version(),
+                            latest_version = latest_version,
+                            "table updated but max_retries is 0, failing immediately"
+                        );
                         return Err(
                             TransactionError::MaxCommitAttempts(this.max_retries as i32).into()
                         );
                     }
-                    warn!("Attempting to write a transaction {} but the underlying table has been updated to {latest_version}\n{:?}", read_snapshot.version() + 1, this.log_store);
+                    warn!(
+                        base_version = read_snapshot.version(),
+                        latest_version = latest_version,
+                        versions_behind = latest_version - read_snapshot.version(),
+                        "table updated during transaction, checking for conflicts"
+                    );
                     let mut steps = latest_version - read_snapshot.version();
+                    let mut conflicts_checked = 0;
 
                     // Need to check for conflicts with each version between the read_snapshot and
                     // the latest!
                     while steps != 0 {
+                        conflicts_checked += 1;
                         let summary = WinningCommitSummary::try_new(
                             this.log_store.as_ref(),
                             latest_version - steps,
@@ -680,24 +710,40 @@ impl<'a> std::future::IntoFuture for PreparedCommit<'a> {
                         match conflict_checker.check_conflicts() {
                             Ok(_) => {}
                             Err(err) => {
+                                error!(
+                                    conflicts_checked = conflicts_checked,
+                                    error = %err,
+                                    "conflict detected, aborting transaction"
+                                );
                                 return Err(TransactionError::CommitConflict(err).into());
                             }
                         }
                         steps -= 1;
                     }
+                    commit_span.record("conflicts_checked", conflicts_checked);
+                    debug!(
+                        conflicts_checked = conflicts_checked,
+                        "all conflicts resolved, updating snapshot"
+                    );
                     // Update snapshot to latest version after successful conflict check
                     read_snapshot
                         .update(&this.log_store, Some(latest_version as u64))
                         .await?;
                 }
                 let version: i64 = latest_version + 1;
+                commit_span.record("target_version", version);
 
                 match this
                     .log_store
                     .write_commit_entry(version, commit_or_bytes.clone(), this.operation_id)
                     .await
                 {
                     Ok(()) => {
+                        info!(
+                            version = version,
+                            num_retries = attempt_number as u64 - 1,
+                            "transaction committed successfully"
+                        );
                         return Ok(PostCommit {
                             version,
                             data: this.data,
@@ -718,12 +764,21 @@ impl<'a> std::future::IntoFuture for PreparedCommit<'a> {
                         });
                     }
                     Err(TransactionError::VersionAlreadyExists(version)) => {
-                        error!("The transaction {version} already exists, will retry!");
+                        warn!(
+                            version = version,
+                            attempt = attempt_number,
+                            "version already exists, will retry"
+                        );
                         // If the version already exists, loop through again and re-check
                         // conflicts
                         attempt_number += 1;
                     }
                     Err(err) => {
+                        error!(
+                            version = version,
+                            error = %err,
+                            "commit failed, aborting"
+                        );
                         this.log_store
                             .abort_commit_entry(version, commit_or_bytes, this.operation_id)
                             .await?;
@@ -732,6 +787,10 @@ impl<'a> std::future::IntoFuture for PreparedCommit<'a> {
                 }
             }
 
+            error!(
+                max_retries = this.max_retries,
+                "exceeded maximum commit attempts"
+            );
             Err(TransactionError::MaxCommitAttempts(this.max_retries as i32).into())
         })
     }
@@ -967,4 +1026,41 @@ mod tests {
             .await
             .unwrap();
     }
+
+    #[test]
+    fn test_commit_with_retries_tracing_span() {
+        let span = info_span!(
+            "commit_with_retries",
+            base_version = 5,
+            max_retries = 10,
+            attempt = field::Empty,
+            target_version = field::Empty,
+            conflicts_checked = 0
+        );
+
+        let metadata = span.metadata().expect("span should have metadata");
+        assert_eq!(metadata.name(), "commit_with_retries");
+        assert_eq!(metadata.level(), &Level::INFO);
+        assert!(metadata.is_span());
+
+        span.record("attempt", 1);
+        span.record("target_version", 6);
+        span.record("conflicts_checked", 2);
+    }
+
+    #[test]
+    fn test_commit_properties_with_retries() {
+        let props = CommitProperties::default()
+            .with_max_retries(5)
+            .with_create_checkpoint(false);
+
+        assert_eq!(props.max_retries, 5);
+        assert!(!props.create_checkpoint);
+    }
+
+    #[test]
+    fn test_commit_metrics() {
+        let metrics = CommitMetrics { num_retries: 3 };
+        assert_eq!(metrics.num_retries, 3);
+    }
 }