From 4b01f79734bd4a1a8acc986f82275255f942f883 Mon Sep 17 00:00:00 2001
From: Noel Kwan <noelkwan1998@gmail.com>
Date: Fri, 17 May 2024 12:24:42 +0800
Subject: [PATCH 1/3] add logs

---
 src/stream/src/executor/hash_join.rs | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/src/stream/src/executor/hash_join.rs b/src/stream/src/executor/hash_join.rs
index 60e2e532eb1fa..3fcf0032e4975 100644
--- a/src/stream/src/executor/hash_join.rs
+++ b/src/stream/src/executor/hash_join.rs
@@ -829,6 +829,17 @@ 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 {
+                    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)
             }

From ace0230a7e918b4e3fa20a5fc406e879a7cb2ac1 Mon Sep 17 00:00:00 2001
From: Noel Kwan <noelkwan1998@gmail.com>
Date: Fri, 17 May 2024 13:54:18 +0800
Subject: [PATCH 2/3] bump e2e test timeout

---
 ci/workflows/pull-request.yml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/ci/workflows/pull-request.yml b/ci/workflows/pull-request.yml
index 475b981a18b07..4c7308289ae73 100644
--- a/ci/workflows/pull-request.yml
+++ b/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)"

From 977d15df8eae09f827df0619124a357314ce2940 Mon Sep 17 00:00:00 2001
From: Noel Kwan <noelkwan1998@gmail.com>
Date: Fri, 17 May 2024 14:15:05 +0800
Subject: [PATCH 3/3] deserialize join key directly

---
 src/stream/src/executor/hash_join.rs      | 2 ++
 src/stream/src/executor/join/hash_join.rs | 4 ++++
 2 files changed, 6 insertions(+)

diff --git a/src/stream/src/executor/hash_join.rs b/src/stream/src/executor/hash_join.rs
index 3fcf0032e4975..427b036fe7fe1 100644
--- a/src/stream/src/executor/hash_join.rs
+++ b/src/stream/src/executor/hash_join.rs
@@ -830,6 +830,8 @@ 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(),
diff --git a/src/stream/src/executor/join/hash_join.rs b/src/stream/src/executor/join/hash_join.rs
index 8805ea3cc3120..0c6a12b1e68f5 100644
--- a/src/stream/src/executor/join/hash_join.rs
+++ b/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;