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

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

Author: fvaleye

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,
+            })
+        }
     }
 }
 

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() {
+                tracing::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 = tracing::info_span!(
+                "commit_with_retries",
+                base_version = read_snapshot.version(),
+                max_retries = this.max_retries,
+                attempt = tracing::field::Empty,
+                target_version = tracing::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 {
+                        tracing::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);
+                    tracing::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) => {
+                                tracing::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);
+                    tracing::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(()) => {
+                        tracing::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!");
+                        tracing::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) => {
+                        tracing::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> {
                 }
             }
 
+            tracing::error!(
+                max_retries = this.max_retries,
+                "exceeded maximum commit attempts"
+            );
             Err(TransactionError::MaxCommitAttempts(this.max_retries as i32).into())
         })
     }

crates/core/src/logstore/mod.rs

@@ -680,19 +680,31 @@ pub async fn get_latest_version(
 }
 
 /// Read delta log for a specific version
+#[tracing::instrument(skip(storage), fields(version = version, path = %commit_uri_from_version(version)))]
 pub async fn read_commit_entry(
     storage: &dyn ObjectStore,
     version: i64,
 ) -> DeltaResult<Option<Bytes>> {
     let commit_uri = commit_uri_from_version(version);
     match storage.get(&commit_uri).await {
-        Ok(res) => Ok(Some(res.bytes().await?)),
-        Err(ObjectStoreError::NotFound { .. }) => Ok(None),
-        Err(err) => Err(err.into()),
+        Ok(res) => {
+            let bytes = res.bytes().await?;
+            tracing::debug!(size = bytes.len(), "commit entry read successfully");
+            Ok(Some(bytes))
+        }
+        Err(ObjectStoreError::NotFound { .. }) => {
+            tracing::debug!("commit entry not found");
+            Ok(None)
+        }
+        Err(err) => {
+            tracing::error!(error = %err, "failed to read commit entry");
+            Err(err.into())
+        }
     }
 }
 
 /// Default implementation for writing a commit entry
+#[tracing::instrument(skip(storage), fields(version = version, tmp_path = %tmp_commit, commit_path = %commit_uri_from_version(version)))]
 pub async fn write_commit_entry(
     storage: &dyn ObjectStore,
     version: i64,
@@ -706,21 +718,28 @@ pub async fn write_commit_entry(
         .map_err(|err| -> TransactionError {
             match err {
                 ObjectStoreError::AlreadyExists { .. } => {
+                    tracing::warn!("commit entry already exists");
                     TransactionError::VersionAlreadyExists(version)
                 }
-                _ => TransactionError::from(err),
+                _ => {
+                    tracing::error!(error = %err, "failed to write commit entry");
+                    TransactionError::from(err)
+                }
             }
         })?;
+    tracing::debug!("commit entry written successfully");
     Ok(())
 }
 
 /// Default implementation for aborting a commit entry
+#[tracing::instrument(skip(storage), fields(version = _version, tmp_path = %tmp_commit))]
 pub async fn abort_commit_entry(
     storage: &dyn ObjectStore,
     _version: i64,
     tmp_commit: &Path,
 ) -> Result<(), TransactionError> {
     storage.delete_with_retries(tmp_commit, 15).await?;
+    tracing::debug!("commit entry aborted successfully");
     Ok(())
 }
 

crates/core/src/logstore/storage/retry_ext.rs

@@ -31,6 +31,7 @@ pub trait ObjectStoreRetryExt: ObjectStore {
     /// bytes to location, or fail. No clients should be able to observe a partially written object
     ///
     /// Note that `put_with_opts` may have precondition semantics, and thus may not be retriable.
+    #[tracing::instrument(skip(self, bytes), fields(path = %location, size = bytes.content_length()))]
     async fn put_with_retries(
         &self,
         location: &Path,
@@ -40,15 +41,20 @@ pub trait ObjectStoreRetryExt: ObjectStore {
         let mut attempt_number = 1;
         while attempt_number <= max_retries {
             match self.put(location, bytes.clone()).await {
-                Ok(result) => return Ok(result),
+                Ok(result) => {
+                    tracing::debug!(attempt = attempt_number, "put operation succeeded");
+                    return Ok(result);
+                }
                 Err(err) if attempt_number == max_retries => {
+                    tracing::warn!(attempt = attempt_number, error = %err, "put operation failed after max retries");
                     return Err(err);
                 }
                 Err(Error::Generic { store, source }) => {
                     debug!("put_with_retries attempt {attempt_number} failed: {store} {source}");
                     attempt_number += 1;
                 }
                 Err(err) => {
+                    tracing::error!(attempt = attempt_number, error = %err, "put operation failed with non-retryable error");
                     return Err(err);
                 }
             }
@@ -57,19 +63,25 @@ pub trait ObjectStoreRetryExt: ObjectStore {
     }
 
     /// Delete the object at the specified location
+    #[tracing::instrument(skip(self), fields(path = %location))]
     async fn delete_with_retries(&self, location: &Path, max_retries: usize) -> Result<()> {
         let mut attempt_number = 1;
         while attempt_number <= max_retries {
             match self.delete(location).await {
-                Ok(()) | Err(Error::NotFound { .. }) => return Ok(()),
+                Ok(()) | Err(Error::NotFound { .. }) => {
+                    tracing::debug!(attempt = attempt_number, "delete operation succeeded");
+                    return Ok(());
+                }
                 Err(err) if attempt_number == max_retries => {
+                    tracing::warn!(attempt = attempt_number, error = %err, "delete operation failed after max retries");
                     return Err(err);
                 }
                 Err(Error::Generic { store, source }) => {
                     debug!("delete_with_retries attempt {attempt_number} failed: {store} {source}");
                     attempt_number += 1;
                 }
                 Err(err) => {
+                    tracing::error!(attempt = attempt_number, error = %err, "delete operation failed with non-retryable error");
                     return Err(err);
                 }
             }