refactor(streaming): better traces for development (risingwavelabs#12024)

Signed-off-by: Bugen Zhao <[email protected]>

Author: BugenZhao

docs/developer-guide.md

@@ -272,7 +272,10 @@ The Rust components use `tokio-tracing` to handle both logging and tracing. The
 * Third-party libraries: warn
 * Other libraries: debug
 
-If you need to adjust log levels, change the logging filters in `src/utils/runtime/src/lib.rs`.
+If you need to override the default log levels, launch RisingWave with the environment variable `RUST_LOG` set as described [here](https://docs.rs/tracing-subscriber/0.3/tracing_subscriber/filter/struct.EnvFilter.html).
+
+There're also some logs designated for debugging purposes with target names starting with `events::`.
+For example, by setting `RUST_LOG=events::stream::message::chunk=trace`, all chunk messages will be logged as it passes through the executors in the streaming engine. Search in the codebase to find more of them.
 
 
 ## Test your code changes

src/common/Cargo.toml

@@ -99,7 +99,7 @@ tracing-opentelemetry = "0.21"
 tracing-subscriber = "0.3.17"
 twox-hash = "1"
 url = "2"
-uuid = "1.4.1"
+uuid = { version = "1", features = ["v4"] }
 
 [target.'cfg(not(madsim))'.dependencies]
 workspace-hack = { path = "../workspace-hack" }

src/common/src/array/data_chunk.rs

@@ -12,11 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::fmt::Display;
 use std::hash::BuildHasher;
 use std::sync::Arc;
 use std::{fmt, usize};
 
 use bytes::Bytes;
+use either::Either;
 use itertools::Itertools;
 use rand::rngs::SmallRng;
 use rand::{Rng, SeedableRng};
@@ -65,6 +67,8 @@ pub struct DataChunk {
 }
 
 impl DataChunk {
+    pub(crate) const PRETTY_TABLE_PRESET: &str = "||--+-++|    ++++++";
+
     /// Create a `DataChunk` with `columns` and visibility. The visibility can either be a `Bitmap`
     /// or a simple cardinality number.
     pub fn new<V: Into<Vis>>(columns: Vec<ArrayRef>, vis: V) -> Self {
@@ -392,24 +396,31 @@ impl DataChunk {
         RowRef::new(self, pos)
     }
 
-    /// `to_pretty_string` returns a table-like text representation of the `DataChunk`.
-    pub fn to_pretty_string(&self) -> String {
+    /// Returns a table-like text representation of the `DataChunk`.
+    pub fn to_pretty(&self) -> impl Display {
         use comfy_table::Table;
+
+        if self.cardinality() == 0 {
+            return Either::Left("(empty)");
+        }
+
         let mut table = Table::new();
-        table.load_preset("||--+-++|    ++++++\n");
+        table.load_preset(Self::PRETTY_TABLE_PRESET);
+
         for row in self.rows() {
             let cells: Vec<_> = row
                 .iter()
                 .map(|v| {
                     match v {
-                        None => "".to_owned(), // null
+                        None => "".to_owned(), // NULL
                         Some(scalar) => scalar.to_text(),
                     }
                 })
                 .collect();
             table.add_row(cells);
         }
-        table.to_string()
+
+        Either::Right(table)
     }
 
     /// Keep the specified columns and set the rest elements to null.
@@ -626,7 +637,7 @@ impl fmt::Debug for DataChunk {
             "DataChunk {{ cardinality = {}, capacity = {}, data = \n{} }}",
             self.cardinality(),
             self.capacity(),
-            self.to_pretty_string()
+            self.to_pretty()
         )
     }
 }
@@ -1007,7 +1018,7 @@ mod tests {
             4,
         );
         assert_eq!(
-            chunk.to_pretty_string(),
+            chunk.to_pretty().to_string(),
             "\
 +---+---+
 | 1 | 6 |

src/common/src/array/stream_chunk.rs

@@ -12,11 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::fmt::Display;
 use std::mem::size_of;
 use std::ops::{Deref, DerefMut};
 use std::sync::Arc;
 use std::{fmt, mem};
 
+use either::Either;
 use itertools::Itertools;
 use rand::prelude::SmallRng;
 use rand::{Rng, SeedableRng};
@@ -26,6 +28,7 @@ use super::vis::VisMut;
 use super::{ArrayImpl, ArrayRef, ArrayResult, DataChunkTestExt};
 use crate::array::{DataChunk, Vis};
 use crate::buffer::Bitmap;
+use crate::catalog::Schema;
 use crate::estimate_size::EstimateSize;
 use crate::field_generator::VarcharProperty;
 use crate::row::Row;
@@ -199,16 +202,34 @@ impl StreamChunk {
         &self.ops
     }
 
-    /// `to_pretty_string` returns a table-like text representation of the `StreamChunk`.
-    pub fn to_pretty_string(&self) -> String {
+    /// Returns a table-like text representation of the `StreamChunk`.
+    pub fn to_pretty(&self) -> impl Display {
+        self.to_pretty_inner(None)
+    }
+
+    /// Returns a table-like text representation of the `StreamChunk` with a header of column names
+    /// from the given `schema`.
+    pub fn to_pretty_with_schema(&self, schema: &Schema) -> impl Display {
+        self.to_pretty_inner(Some(schema))
+    }
+
+    fn to_pretty_inner(&self, schema: Option<&Schema>) -> impl Display {
         use comfy_table::{Cell, CellAlignment, Table};
 
         if self.cardinality() == 0 {
-            return "(empty)".to_owned();
+            return Either::Left("(empty)");
         }
 
         let mut table = Table::new();
-        table.load_preset("||--+-++|    ++++++");
+        table.load_preset(DataChunk::PRETTY_TABLE_PRESET);
+
+        if let Some(schema) = schema {
+            assert_eq!(self.dimension(), schema.len());
+            let cells = std::iter::once(String::new())
+                .chain(schema.fields().iter().map(|f| f.name.clone()));
+            table.set_header(cells);
+        }
+
         for (op, row_ref) in self.rows() {
             let mut cells = Vec::with_capacity(row_ref.len() + 1);
             cells.push(
@@ -229,7 +250,8 @@ impl StreamChunk {
             }
             table.add_row(cells);
         }
-        table.to_string()
+
+        Either::Right(table)
     }
 
     /// Reorder (and possibly remove) columns.
@@ -290,7 +312,7 @@ impl fmt::Debug for StreamChunk {
                 "StreamChunk {{ cardinality: {}, capacity: {}, data: \n{}\n }}",
                 self.cardinality(),
                 self.capacity(),
-                self.to_pretty_string()
+                self.to_pretty()
             )
         } else {
             f.debug_struct("StreamChunk")
@@ -636,7 +658,7 @@ mod tests {
             U+ 4 .",
         );
         assert_eq!(
-            chunk.to_pretty_string(),
+            chunk.to_pretty().to_string(),
             "\
 +----+---+---+
 |  + | 1 | 6 |

src/stream/src/executor/dispatch.rs

@@ -624,7 +624,7 @@ impl Dispatcher for HashDataDispatcher {
             // get hash value of every line by its key
             let vnodes = VirtualNode::compute_chunk(chunk.data_chunk(), &self.keys);
 
-            tracing::trace!(target: "events::stream::dispatch::hash", "\n{}\n keys {:?} => {:?}", chunk.to_pretty_string(), self.keys, vnodes);
+            tracing::trace!(target: "events::stream::dispatch::hash", "\n{}\n keys {:?} => {:?}", chunk.to_pretty(), self.keys, vnodes);
 
             let mut vis_maps = repeat_with(|| BitmapBuilder::with_capacity(chunk.capacity()))
                 .take(num_outputs)

src/stream/src/executor/merge.rs

@@ -138,7 +138,7 @@ impl MergeExecutor {
                 }
                 Message::Barrier(barrier) => {
                     tracing::trace!(
-                        target: "events::barrier::path",
+                        target: "events::stream::barrier::path",
                         actor_id = actor_id,
                         "receiver receives barrier from path: {:?}",
                         barrier.passed_actors

src/stream/src/executor/receiver.rs

@@ -140,7 +140,7 @@ impl Executor for ReceiverExecutor {
                     }
                     Message::Barrier(barrier) => {
                         tracing::trace!(
-                            target: "events::barrier::path",
+                            target: "events::stream::barrier::path",
                             actor_id = actor_id,
                             "receiver receives barrier from path: {:?}",
                             barrier.passed_actors

src/stream/src/executor/wrapper/trace.rs

@@ -47,20 +47,45 @@ pub async fn trace(
     pin_mut!(input);
 
     while let Some(message) = input.next().instrument(span.clone()).await.transpose()? {
-        if let Message::Chunk(chunk) = &message {
-            if chunk.cardinality() > 0 && (enable_executor_row_count || is_sink_or_mv) {
-                metrics
-                    .executor_row_count
-                    .with_label_values(&[&actor_id_string, &span_name])
-                    .inc_by(chunk.cardinality() as u64);
-                tracing::trace!(?chunk, "chunk");
+        // Trace the message in the span's scope.
+        span.in_scope(|| match &message {
+            Message::Chunk(chunk) => {
+                if chunk.cardinality() > 0 {
+                    if enable_executor_row_count || is_sink_or_mv {
+                        metrics
+                            .executor_row_count
+                            .with_label_values(&[&actor_id_string, &span_name])
+                            .inc_by(chunk.cardinality() as u64);
+                    }
+                    tracing::trace!(
+                        target: "events::stream::message::chunk",
+                        cardinality = chunk.cardinality(),
+                        capacity = chunk.capacity(),
+                        "\n{}\n", chunk.to_pretty_with_schema(&info.schema),
+                    );
+                }
             }
-        }
+            Message::Watermark(watermark) => {
+                tracing::trace!(
+                    target: "events::stream::message::watermark",
+                    value = ?watermark.val,
+                    col_idx = watermark.col_idx,
+                );
+            }
+            Message::Barrier(barrier) => {
+                tracing::trace!(
+                    target: "events::stream::message::barrier",
+                    prev_epoch = barrier.epoch.prev,
+                    curr_epoch = barrier.epoch.curr,
+                    kind = ?barrier.kind,
+                );
+            }
+        });
 
+        // Yield the message and update the span.
         match &message {
             Message::Chunk(_) | Message::Watermark(_) => yield message,
-
-            Message::Barrier(_barrier) => {
+            Message::Barrier(_) => {
                 // Drop the span as the inner executor has finished processing the barrier (then all
                 // data from the previous epoch).
                 let _ = std::mem::replace(&mut span, Span::none());

src/stream/src/task/barrier_manager.rs

@@ -101,7 +101,11 @@ impl LocalBarrierManager {
 
     /// Register sender for source actors, used to send barriers.
     pub fn register_sender(&mut self, actor_id: ActorId, sender: UnboundedSender<Barrier>) {
-        tracing::trace!(actor_id = actor_id, "register sender");
+        tracing::trace!(
+            target: "events::stream::barrier::manager",
+            actor_id = actor_id,
+            "register sender"
+        );
         self.senders.entry(actor_id).or_default().push(sender);
     }
 
@@ -129,6 +133,7 @@ impl LocalBarrierManager {
         };
         let to_collect: HashSet<ActorId> = actor_ids_to_collect.into_iter().collect();
         trace!(
+            target: "events::stream::barrier::manager::send",
             "send barrier {:?}, senders = {:?}, actor_ids_to_collect = {:?}",
             barrier,
             to_send,
@@ -167,7 +172,11 @@ impl LocalBarrierManager {
 
         // Actors to stop should still accept this barrier, but won't get sent to in next times.
         if let Some(actors) = barrier.all_stop_actors() {
-            trace!("remove actors {:?} from senders", actors);
+            trace!(
+                target: "events::stream::barrier::manager",
+                "remove actors {:?} from senders",
+                actors
+            );
             for actor in actors {
                 self.senders.remove(actor);
             }

src/stream/src/task/barrier_manager/managed_state.rs

@@ -205,7 +205,7 @@ impl ManagedBarrierState {
     /// Collect a `barrier` from the actor with `actor_id`.
     pub(super) fn collect(&mut self, actor_id: ActorId, barrier: &Barrier) {
         tracing::trace!(
-            target: "events::stream::barrier::collect_barrier",
+            target: "events::stream::barrier::manager::collect",
             "collect_barrier: epoch = {}, actor_id = {}, state = {:#?}",
             barrier.epoch.curr,
             actor_id,

src/stream/tests/integration_tests/snapshot.rs

@@ -154,7 +154,7 @@ where
             }
             SnapshotEvent::Chunk(chunk_str) => {
                 let chunk = StreamChunk::from_pretty(chunk_str);
-                *chunk_str = chunk.to_pretty_string();
+                *chunk_str = chunk.to_pretty().to_string();
                 tx.push_chunk(chunk);
             }
             SnapshotEvent::Watermark { col_idx, val } => tx.push_watermark(
@@ -191,10 +191,10 @@ fn run_until_pending(
                 if options.sort_chunk {
                     chunk = chunk.sort_rows();
                 }
-                let mut output = chunk.to_pretty_string();
+                let mut output = chunk.to_pretty().to_string();
                 if options.include_applied_result {
                     let applied = store.apply_chunk(&chunk);
-                    output += &format!("\napplied result:\n{}", applied.to_pretty_string());
+                    output += &format!("\napplied result:\n{}", applied.to_pretty());
                 }
                 SnapshotEvent::Chunk(output)
             }

src/utils/runtime/src/logger.rs

@@ -48,8 +48,6 @@ fn configure_risingwave_targets_fmt(targets: filter::Targets) -> filter::Targets
         .with_target("foyer_memory", Level::WARN)
         .with_target("foyer_storage", Level::WARN)
         // disable events that are too verbose
-        // if you want to enable any of them, find the target name and set it to `TRACE`
-        // .with_target("events::stream::mview::scan", Level::TRACE)
         .with_target("events", Level::ERROR)
 }