[CHORE] More details logged to understand compaction.

rust/blockstore/src/arrow/provider.rs

@@ -263,7 +263,7 @@ impl BlockManager {
                     .storage
                     .get(&key)
                     .instrument(
-                        tracing::trace_span!(parent: Span::current(), "BlockManager storage get"),
+                        tracing::trace_span!(parent: Span::current(), "BlockManager storage get {id}"),
                     )
                     .await;
                 match bytes_res {

rust/worker/src/compactor/compaction_manager.rs

@@ -131,11 +131,11 @@ impl CompactionManager {
 
                 match orchestrator.run().await {
                     Ok(result) => {
-                        println!("Compaction Job completed");
+                        tracing::info!("Compaction Job completed: {:?}", result);
                         return Ok(result);
                     }
                     Err(e) => {
-                        println!("Compaction Job failed");
+                        tracing::error!("Compaction Job failed: {:?}", e);
                         return Err(e);
                     }
                 }

rust/worker/src/segment/metadata_segment.rs

@@ -577,7 +577,9 @@ impl<'log_records> SegmentWriter<'log_records> for MetadataSegmentWriter<'_> {
         &self,
         records: Chunk<MaterializedLogRecord<'log_records>>,
     ) -> Result<(), ApplyMaterializedLogError> {
+        let mut count = 0u64;
         for record in records.iter() {
+            count += 1;
             let segment_offset_id = record.0.offset_id;
             match record.0.final_operation {
                 MaterializedLogOperation::AddNew => {
@@ -794,6 +796,7 @@ impl<'log_records> SegmentWriter<'log_records> for MetadataSegmentWriter<'_> {
                 MaterializedLogOperation::Initial => panic!("Not expected mat records in the initial state")
             }
         }
+        tracing::info!("Applied {} records to metadata segment", count,);
         Ok(())
     }
 

rust/worker/src/segment/record_segment.rs

@@ -322,7 +322,9 @@ impl<'a> SegmentWriter<'a> for RecordSegmentWriter {
         &self,
         records: Chunk<MaterializedLogRecord<'a>>,
     ) -> Result<(), ApplyMaterializedLogError> {
+        let mut count = 0u64;
         for (log_record, _) in records.iter() {
+            count += 1;
             match log_record.final_operation {
                 MaterializedLogOperation::AddNew => {
                     // Set all four.
@@ -458,6 +460,7 @@ impl<'a> SegmentWriter<'a> for RecordSegmentWriter {
                 MaterializedLogOperation::Initial => panic!("Invariant violation. Materialized logs should not have any logs in the initial state")
             }
         }
+        tracing::info!("Applied {} records to record segment", count,);
         Ok(())
     }
 

rust/worker/src/segment/types.rs

@@ -493,7 +493,7 @@ impl<'me> LogMaterializer<'me> {
                 Ok(())
             }
             .instrument(
-                tracing::info_span!(parent: Span::current(), "Materialization read from stroage"),
+                tracing::info_span!(parent: Span::current(), "Materialization read from storage"),
             )
             .await?;
         }