Merge pull request JanKaul#259 from JanKaul/add-traces

add traces to common functions

Author: JanKaul

datafusion_iceberg/src/table.rs

@@ -350,8 +350,8 @@ fn fake_object_store_url(table_location_url: &str) -> ObjectStoreUrl {
 }
 
 #[allow(clippy::too_many_arguments)]
-#[instrument(level = "debug", skip(arrow_schema, statistics, session, filters), fields(
-    table_location = %table.metadata().location,
+#[instrument(name = "datafusion_iceberg::table_scan", level = "debug", skip(arrow_schema, statistics, session, filters), fields(
+    table_identifier = %table.identifier(),
     snapshot_range = ?snapshot_range,
     projection = ?projection,
     filter_count = filters.len(),
@@ -1112,6 +1112,10 @@ pub async fn write_parquet_equality_delete_files(
     write_parquet_files(table, batches, context, Some(equality_ids), branch).await
 }
 
+#[instrument(name = "datafusion_iceberg::write_parquet_files", level = "debug", skip(table, batches, context), fields(
+    table_identifier = %table.identifier(),
+    branch = ?branch
+))]
 async fn write_parquet_files(
     table: &Table,
     batches: SendableRecordBatchStream,

iceberg-rust/src/file_format/parquet.rs

@@ -23,10 +23,17 @@ use parquet::{
     schema::types::{from_thrift, SchemaDescriptor},
 };
 use thrift::protocol::{TCompactOutputProtocol, TSerializable};
+use tracing::instrument;
 
 use crate::error::Error;
 
 /// Read datafile statistics from parquetfile
+#[instrument(name = "iceberg_rust::file_format::parquet::parquet_to_datafile", level = "debug", skip(file_metadata, schema, partition_fields), fields(
+    location = location,
+    file_size = file_size,
+    partition_field_count = partition_fields.len(),
+    has_equality_ids = equality_ids.is_some()
+))]
 pub fn parquet_to_datafile(
     location: &str,
     file_size: u64,

iceberg-rust/src/table/mod.rs

@@ -34,6 +34,8 @@ use iceberg_rust_spec::{
     },
 };
 
+use tracing::instrument;
+
 use crate::{
     catalog::{create::CreateTableBuilder, identifier::Identifier, Catalog},
     error::Error,
@@ -199,6 +201,11 @@ impl Table {
     /// Returns an error if:
     /// * The end snapshot ID is invalid
     /// * Reading the manifest list fails
+    #[instrument(name = "iceberg_rust::table::manifests", level = "debug", skip(self), fields(
+        table_identifier = %self.identifier,
+        start = ?start,
+        end = ?end
+    ))]
     pub async fn manifests(
         &self,
         start: Option<i64>,
@@ -295,6 +302,11 @@ impl Table {
 /// Path of a Manifest file
 pub type ManifestPath = String;
 
+#[instrument(name = "iceberg_rust::table::datafiles", level = "debug", skip(object_store, manifests), fields(
+    manifest_count = manifests.len(),
+    filter_provided = filter.is_some(),
+    sequence_range = ?sequence_number_range
+))]
 async fn datafiles(
     object_store: Arc<dyn ObjectStore>,
     manifests: &'_ [ManifestListEntry],

iceberg-rust/src/table/transaction/mod.rs

@@ -16,7 +16,7 @@
 //! * Managing snapshots and branches
 
 use std::collections::HashMap;
-use tracing::debug;
+use tracing::{debug, instrument};
 
 use iceberg_rust_spec::spec::{manifest::DataFile, schema::Schema, snapshot::SnapshotReference};
 
@@ -419,6 +419,10 @@ impl<'table> TableTransaction<'table> {
     ///     .commit()
     ///     .await?;
     /// ```
+    #[instrument(name = "iceberg_rust::table::transaction::commit", level = "debug", skip(self), fields(
+        table_identifier = %self.table.identifier,
+        branch = ?self.branch
+    ))]
     pub async fn commit(self) -> Result<(), Error> {
         let catalog = self.table.catalog();
         let identifier = self.table.identifier.clone();

iceberg-rust/src/table/transaction/operation.rs

@@ -110,7 +110,11 @@ pub enum Operation {
 }
 
 impl Operation {
-    #[instrument(level = "debug", skip(object_store))]
+    #[instrument(
+        name = "iceberg_rust::table::transaction::operation::execute",
+        level = "debug",
+        skip(object_store)
+    )]
     pub async fn execute(
         self,
         table_metadata: &TableMetadata,

iceberg-rust/src/view/transaction/operation.rs

@@ -80,7 +80,10 @@ fn upsert_representations(
 
 impl Operation {
     /// Execute operation
-    #[instrument(level = "debug")]
+    #[instrument(
+        name = "iceberg_rust::view::transaction::operation::execute",
+        level = "debug"
+    )]
     pub async fn execute<T: Materialization + Debug>(
         self,
         metadata: &GeneralViewMetadata<T>,