feat(stream): add debug logs for join key with high amplification (#1…

…6799) (#16805)

Co-authored-by: Noel Kwan <[email protected]>

ci/workflows/pull-request.yml

@@ -97,7 +97,7 @@ steps:
           config: ci/docker-compose.yml
           mount-buildkite-agent: true
       - ./ci/plugins/upload-failure-logs
-    timeout_in_minutes: 21
+    timeout_in_minutes: 23
     retry: *auto-retry
 
   - label: "end-to-end test (parallel)"

src/stream/src/executor/hash_join.rs

@@ -829,6 +829,19 @@ impl<K: HashKey, S: StateStore, const T: JoinTypePrimitive> HashJoinExecutor<K,
 
             if let Some(rows) = &matched_rows {
                 join_matched_join_keys.observe(rows.len() as _);
+                if rows.len() > 10000 {
+                    let join_key_data_types = side_update.ht.join_key_data_types();
+                    let key = key.deserialize(join_key_data_types)?;
+                    tracing::debug!(target: "hash_join_amplification",
+                        matched_rows_len = rows.len(),
+                        update_table_id = side_update.ht.table_id(),
+                        match_table_id = side_match.ht.table_id(),
+                        join_key = ?key,
+                        actor_id = ctx.id,
+                        fragment_id = ctx.fragment_id,
+                        "large rows matched for join key"
+                    );
+                }
             } else {
                 join_matched_join_keys.observe(0.0)
             }

src/stream/src/executor/join/hash_join.rs

@@ -626,6 +626,10 @@ impl<K: HashKey, S: StateStore> JoinHashMap<K, S> {
     pub fn table_id(&self) -> u32 {
         self.state.table.table_id()
     }
+
+    pub fn join_key_data_types(&self) -> &[DataType] {
+        &self.join_key_data_types
+    }
 }
 
 use risingwave_common_estimate_size::KvSize;