Merge pull request JanKaul#249 from JanKaul/add-logging-information

Add logging information

Author: JanKaul

Cargo.lock

@@ -26,6 +26,7 @@ regex = "1.11.1"
 serde_json = { workspace = true }
 thiserror = { workspace = true }
 tokio = { version = "1.43", features = ["rt-multi-thread"] }
+tracing = { workspace = true }
 url = { workspace = true }
 uuid = { workspace = true }
 

datafusion_iceberg/Cargo.toml

@@ -33,6 +33,7 @@ use tokio::sync::{
     mpsc::{self},
     RwLock, RwLockWriteGuard,
 };
+use tracing::instrument;
 
 use datafusion::{
     arrow::datatypes::{DataType, Field, Schema as ArrowSchema, SchemaBuilder, SchemaRef},
@@ -355,6 +356,13 @@ fn fake_object_store_url(table_location_url: &str) -> Option<ObjectStoreUrl> {
 }
 
 #[allow(clippy::too_many_arguments)]
+#[instrument(level = "debug", skip(arrow_schema, statistics, session, filters), fields(
+    table_location = %table.metadata().location,
+    snapshot_range = ?snapshot_range,
+    projection = ?projection,
+    filter_count = filters.len(),
+    limit = ?limit
+))]
 async fn table_scan(
     table: &Table,
     snapshot_range: &(Option<i64>, Option<i64>),

datafusion_iceberg/src/table.rs

@@ -16,6 +16,7 @@
 //! * Managing snapshots and branches
 
 use std::collections::HashMap;
+use tracing::debug;
 
 use iceberg_rust_spec::spec::{manifest::DataFile, schema::Schema, snapshot::SnapshotReference};
 
@@ -398,6 +399,14 @@ impl<'table> TableTransaction<'table> {
             return Ok(());
         }
 
+        debug!(
+            "Committing {} updates to table {}: requirements={:?}, updates={:?}",
+            updates.len(),
+            identifier,
+            requirements,
+            updates
+        );
+
         let new_table = catalog
             .clone()
             .update_table(CommitTable {

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

@@ -23,6 +23,7 @@ use iceberg_rust_spec::util::strip_prefix;
 use object_store::ObjectStore;
 use smallvec::SmallVec;
 use tokio::task::JoinHandle;
+use tracing::{debug, instrument};
 
 use crate::table::manifest::ManifestWriter;
 use crate::table::manifest_list::ManifestListWriter;
@@ -90,6 +91,7 @@ pub enum Operation {
 }
 
 impl Operation {
+    #[instrument(level = "debug", skip(object_store))]
     pub async fn execute(
         self,
         table_metadata: &TableMetadata,
@@ -102,6 +104,10 @@ impl Operation {
                 delete_files,
                 additional_summary,
             } => {
+                debug!(
+                    "Executing Append operation: branch={:?}, data_files={}, delete_files={}, additional_summary={:?}",
+                    branch, data_files.len(), delete_files.len(), additional_summary
+                );
                 let old_snapshot = table_metadata.current_snapshot(branch.as_deref())?;
 
                 let manifest_list_schema = match table_metadata.format_version {
@@ -220,6 +226,12 @@ impl Operation {
                 files,
                 additional_summary,
             } => {
+                debug!(
+                    "Executing Replace operation: branch={:?}, files={}, additional_summary={:?}",
+                    branch,
+                    files.len(),
+                    additional_summary
+                );
                 let partition_fields =
                     table_metadata.current_partition_fields(branch.as_deref())?;
                 let old_snapshot = table_metadata.current_snapshot(branch.as_deref())?;
@@ -367,6 +379,10 @@ impl Operation {
                 files_to_overwrite,
                 additional_summary,
             } => {
+                debug!(
+                    "Executing Overwrite operation: branch={:?}, data_files={}, files_to_overwrite={:?}, additional_summary={:?}",
+                    branch, data_files.len(), files_to_overwrite, additional_summary
+                );
                 let old_snapshot = table_metadata
                     .current_snapshot(branch.as_deref())?
                     .ok_or(Error::InvalidFormat("Snapshot to overwrite".to_owned()))?;
@@ -500,26 +516,42 @@ impl Operation {
                     ],
                 ))
             }
-            Operation::UpdateProperties(entries) => Ok((
-                None,
-                vec![TableUpdate::SetProperties {
-                    updates: HashMap::from_iter(entries),
-                }],
-            )),
-            Operation::SetSnapshotRef((key, value)) => Ok((
-                table_metadata
-                    .refs
-                    .get(&key)
-                    .map(|x| TableRequirement::AssertRefSnapshotId {
-                        r#ref: key.clone(),
-                        snapshot_id: x.snapshot_id,
-                    }),
-                vec![TableUpdate::SetSnapshotRef {
-                    ref_name: key,
-                    snapshot_reference: value,
-                }],
-            )),
+            Operation::UpdateProperties(entries) => {
+                debug!(
+                    "Executing UpdateProperties operation: entries={:?}",
+                    entries
+                );
+                Ok((
+                    None,
+                    vec![TableUpdate::SetProperties {
+                        updates: HashMap::from_iter(entries),
+                    }],
+                ))
+            }
+            Operation::SetSnapshotRef((key, value)) => {
+                debug!(
+                    "Executing SetSnapshotRef operation: key={}, value={:?}",
+                    key, value
+                );
+                Ok((
+                    table_metadata
+                        .refs
+                        .get(&key)
+                        .map(|x| TableRequirement::AssertRefSnapshotId {
+                            r#ref: key.clone(),
+                            snapshot_id: x.snapshot_id,
+                        }),
+                    vec![TableUpdate::SetSnapshotRef {
+                        ref_name: key,
+                        snapshot_reference: value,
+                    }],
+                ))
+            }
             Operation::AddSchema(schema) => {
+                debug!(
+                    "Executing AddSchema operation: schema_id={:?}",
+                    schema.schema_id()
+                );
                 let last_column_id = schema.fields().iter().map(|x| x.id).max();
                 Ok((
                     None,
@@ -530,6 +562,7 @@ impl Operation {
                 ))
             }
             Operation::SetDefaultSpec(spec_id) => {
+                debug!("Executing SetDefaultSpec operation: spec_id={}", spec_id);
                 Ok((None, vec![TableUpdate::SetDefaultSpec { spec_id }]))
             }
         }

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

@@ -4,6 +4,7 @@
 
 pub mod operation;
 use iceberg_rust_spec::spec::{types::StructType, view_metadata::ViewRepresentation};
+use tracing::debug;
 
 use crate::{catalog::commit::CommitView, error::Error};
 
@@ -61,6 +62,15 @@ impl<'view> Transaction<'view> {
             }
             updates.extend(update);
         }
+
+        debug!(
+            "Committing {} updates to view {}: requirements={:?}, updates={:?}",
+            updates.len(),
+            identifier,
+            requirements,
+            updates
+        );
+
         let new_view = catalog
             .clone()
             .update_view(CommitView {

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

@@ -12,14 +12,17 @@ use iceberg_rust_spec::{
 };
 use std::{
     collections::HashMap,
+    fmt::Debug,
     time::{SystemTime, UNIX_EPOCH},
 };
+use tracing::{debug, instrument};
 
 use crate::{
     catalog::commit::{ViewRequirement, ViewUpdate},
     error::Error,
 };
 
+#[derive(Debug)]
 /// View operation
 pub enum Operation {
     /// Update vresion
@@ -77,7 +80,8 @@ fn upsert_representations(
 
 impl Operation {
     /// Execute operation
-    pub async fn execute<T: Materialization>(
+    #[instrument(level = "debug")]
+    pub async fn execute<T: Materialization + Debug>(
         self,
         metadata: &GeneralViewMetadata<T>,
     ) -> Result<(Option<ViewRequirement>, Vec<ViewUpdate<T>>), Error> {
@@ -87,6 +91,12 @@ impl Operation {
                 schema,
                 branch,
             } => {
+                debug!(
+                    "Executing UpdateRepresentations operation: representations={}, schema_fields={}, branch={:?}",
+                    representations.len(),
+                    schema.len(),
+                    branch
+                );
                 let schema_changed = metadata
                     .current_schema(branch.as_deref())
                     .map(|s| schema != *s.fields())
@@ -158,12 +168,18 @@ impl Operation {
                     view_updates,
                 ))
             }
-            Operation::UpdateProperties(entries) => Ok((
-                None,
-                vec![ViewUpdate::SetProperties {
-                    updates: HashMap::from_iter(entries),
-                }],
-            )),
+            Operation::UpdateProperties(entries) => {
+                debug!(
+                    "Executing UpdateProperties operation: entries={:?}",
+                    entries
+                );
+                Ok((
+                    None,
+                    vec![ViewUpdate::SetProperties {
+                        updates: HashMap::from_iter(entries),
+                    }],
+                ))
+            }
         }
     }
 }