feat(streaming): call may_exist when insert cache miss in join executor

grafana/risingwave-dashboard.dashboard.py

@@ -1329,7 +1329,11 @@ def section_streaming_actors(outer_panels):
                         ),
                         panels.target(
                             f"rate({metric('stream_join_insert_cache_miss_count')}[$__rate_interval])",
-                            "cache miss when insert{{actor_id}} {{side}}",
+                            "cache miss when insert {{actor_id}} {{side}}",
+                        ),
+                        panels.target(
+                            f"rate({metric('stream_join_may_exist_true_count')}[$__rate_interval])",
+                            "may_exist true when insert {{actor_id}} {{side}}",
                         ),
                     ],
                 ),

src/stream/src/common/table/state_table.rs

@@ -307,6 +307,31 @@ impl<S: StateStore, W: WatermarkBufferStrategy> StateTable<S, W> {
             Distribution::fallback(),
             None,
             false,
+            0,
+        )
+        .await
+    }
+
+    /// Create a state table without distribution, with given `prefix_hint_len`, used for unit
+    /// tests.
+    pub async fn new_without_distribution_with_prefix_hint_len(
+        store: S,
+        table_id: TableId,
+        columns: Vec<ColumnDesc>,
+        order_types: Vec<OrderType>,
+        pk_indices: Vec<usize>,
+        prefix_hint_len: usize,
+    ) -> Self {
+        Self::new_with_distribution_inner(
+            store,
+            table_id,
+            columns,
+            order_types,
+            pk_indices,
+            Distribution::fallback(),
+            None,
+            true,
+            prefix_hint_len,
         )
         .await
     }
@@ -331,6 +356,7 @@ impl<S: StateStore, W: WatermarkBufferStrategy> StateTable<S, W> {
             distribution,
             value_indices,
             true,
+            0,
         )
         .await
     }
@@ -353,6 +379,7 @@ impl<S: StateStore, W: WatermarkBufferStrategy> StateTable<S, W> {
             distribution,
             value_indices,
             false,
+            0,
         )
         .await
     }
@@ -370,6 +397,7 @@ impl<S: StateStore, W: WatermarkBufferStrategy> StateTable<S, W> {
         }: Distribution,
         value_indices: Option<Vec<usize>>,
         is_consistent_op: bool,
+        prefix_hint_len: usize,
     ) -> Self {
         let local_state_store = store.new_local(table_id).await;
 
@@ -396,7 +424,7 @@ impl<S: StateStore, W: WatermarkBufferStrategy> StateTable<S, W> {
             pk_indices,
             dist_key_indices,
             dist_key_in_pk_indices,
-            prefix_hint_len: 0,
+            prefix_hint_len,
             vnodes,
             table_option: Default::default(),
             is_consistent_op,

src/stream/src/executor/hash_join.rs

@@ -1026,18 +1026,20 @@ mod tests {
         order_types: &[OrderType],
         pk_indices: &[usize],
         table_id: u32,
+        prefix_hint_len: usize,
     ) -> (StateTable<MemoryStateStore>, StateTable<MemoryStateStore>) {
         let column_descs = data_types
             .iter()
             .enumerate()
             .map(|(id, data_type)| ColumnDesc::unnamed(ColumnId::new(id as i32), data_type.clone()))
             .collect_vec();
-        let state_table = StateTable::new_without_distribution(
+        let state_table = StateTable::new_without_distribution_with_prefix_hint_len(
             mem_state.clone(),
             TableId::new(table_id),
             column_descs,
             order_types.to_vec(),
             pk_indices.to_vec(),
+            prefix_hint_len,
         )
         .await;
 
@@ -1088,8 +1090,9 @@ mod tests {
         };
         let (tx_l, source_l) = MockSource::channel(schema.clone(), vec![1]);
         let (tx_r, source_r) = MockSource::channel(schema, vec![1]);
-        let params_l = JoinParams::new(vec![0], vec![1]);
-        let params_r = JoinParams::new(vec![0], vec![1]);
+        let join_key_indices = vec![0];
+        let params_l = JoinParams::new(join_key_indices.clone(), vec![1]);
+        let params_r = JoinParams::new(join_key_indices.clone(), vec![1]);
         let cond = with_condition.then(create_cond);
 
         let mem_state = MemoryStateStore::new();
@@ -1100,6 +1103,7 @@ mod tests {
             &[OrderType::Ascending, OrderType::Ascending],
             &[0, 1],
             0,
+            join_key_indices.len(),
         )
         .await;
 
@@ -1109,6 +1113,7 @@ mod tests {
             &[OrderType::Ascending, OrderType::Ascending],
             &[0, 1],
             2,
+            join_key_indices.len(),
         )
         .await;
 
@@ -1154,8 +1159,9 @@ mod tests {
         };
         let (tx_l, source_l) = MockSource::channel(schema.clone(), vec![0]);
         let (tx_r, source_r) = MockSource::channel(schema, vec![0]);
-        let params_l = JoinParams::new(vec![0, 1], vec![]);
-        let params_r = JoinParams::new(vec![0, 1], vec![]);
+        let join_key_indices = vec![0, 1];
+        let params_l = JoinParams::new(join_key_indices.clone(), vec![]);
+        let params_r = JoinParams::new(join_key_indices.clone(), vec![]);
         let cond = with_condition.then(create_cond);
 
         let mem_state = MemoryStateStore::new();
@@ -1170,6 +1176,7 @@ mod tests {
             ],
             &[0, 1, 0],
             0,
+            join_key_indices.len(),
         )
         .await;
 
@@ -1183,6 +1190,7 @@ mod tests {
             ],
             &[0, 1, 1],
             0,
+            join_key_indices.len(),
         )
         .await;
         let schema_len = match T {

src/stream/src/executor/managed_state/join/mod.rs

@@ -160,6 +160,7 @@ pub struct JoinHashMapMetrics {
     total_lookup_count: usize,
     /// How many times have we miss the cache when insert row
     insert_cache_miss_count: usize,
+    may_exist_true_count: usize,
 }
 
 impl JoinHashMapMetrics {
@@ -171,6 +172,7 @@ impl JoinHashMapMetrics {
             lookup_miss_count: 0,
             total_lookup_count: 0,
             insert_cache_miss_count: 0,
+            may_exist_true_count: 0,
         }
     }
 
@@ -187,9 +189,14 @@ impl JoinHashMapMetrics {
             .join_insert_cache_miss_count
             .with_label_values(&[&self.actor_id, self.side])
             .inc_by(self.insert_cache_miss_count as u64);
+        self.metrics
+            .join_may_exist_true_count
+            .with_label_values(&[&self.actor_id, self.side])
+            .inc_by(self.may_exist_true_count as u64);
         self.total_lookup_count = 0;
         self.lookup_miss_count = 0;
         self.insert_cache_miss_count = 0;
+        self.may_exist_true_count = 0;
     }
 }
 
@@ -417,11 +424,22 @@ impl<K: HashKey, S: StateStore> JoinHashMap<K, S> {
             // Update cache
             entry.insert(pk, value.encode());
         } else if self.pk_contained_in_jk {
-            // Refill cache when the join key exist in neither cache or storage.
+            // Refill cache when the join key contains primary key.
             self.metrics.insert_cache_miss_count += 1;
             let mut state = JoinEntryState::default();
             state.insert(pk, value.encode());
             self.update_state(key, state.into());
+        } else {
+            let prefix = key.deserialize(&self.join_key_data_types)?;
+            self.metrics.insert_cache_miss_count += 1;
+            // Refill cache when the join key exists in neither cache or storage.
+            if !self.state.table.may_exist(&prefix).await? {
+                let mut state = JoinEntryState::default();
+                state.insert(pk, value.encode());
+                self.update_state(key, state.into());
+            } else {
+                self.metrics.may_exist_true_count += 1;
+            }
         }
 
         // Update the flush buffer.
@@ -433,7 +451,6 @@ impl<K: HashKey, S: StateStore> JoinHashMap<K, S> {
 
     /// Insert a row.
     /// Used when the side does not need to update degree.
-    #[allow(clippy::unused_async)]
     pub async fn insert_row(&mut self, key: &K, value: impl Row) -> StreamExecutorResult<()> {
         let join_row = JoinRow::new(&value, 0);
         let pk = (&value)
@@ -443,11 +460,22 @@ impl<K: HashKey, S: StateStore> JoinHashMap<K, S> {
             // Update cache
             entry.insert(pk, join_row.encode());
         } else if self.pk_contained_in_jk {
-            // Refill cache when the join key exist in neither cache or storage.
+            // Refill cache when the join key contains primary key.
             self.metrics.insert_cache_miss_count += 1;
             let mut state = JoinEntryState::default();
             state.insert(pk, join_row.encode());
             self.update_state(key, state.into());
+        } else {
+            let prefix = key.deserialize(&self.join_key_data_types)?;
+            self.metrics.insert_cache_miss_count += 1;
+            // Refill cache when the join key exists in neither cache or storage.
+            if !self.state.table.may_exist(&prefix).await? {
+                let mut state = JoinEntryState::default();
+                state.insert(pk, join_row.encode());
+                self.update_state(key, state.into());
+            } else {
+                self.metrics.may_exist_true_count += 1;
+            }
         }
 
         // Update the flush buffer.

src/stream/src/executor/monitor/streaming_stats.rs

@@ -52,6 +52,7 @@ pub struct StreamingMetrics {
     pub join_lookup_miss_count: GenericCounterVec<AtomicU64>,
     pub join_total_lookup_count: GenericCounterVec<AtomicU64>,
     pub join_insert_cache_miss_count: GenericCounterVec<AtomicU64>,
+    pub join_may_exist_true_count: GenericCounterVec<AtomicU64>,
     pub join_actor_input_waiting_duration_ns: GenericCounterVec<AtomicU64>,
     pub join_match_duration_ns: GenericCounterVec<AtomicU64>,
     pub join_barrier_align_duration: HistogramVec,
@@ -266,23 +267,31 @@ impl StreamingMetrics {
 
         let join_lookup_miss_count = register_int_counter_vec_with_registry!(
             "stream_join_lookup_miss_count",
-            "Join executor lookup miss duration",
+            "Join executor lookup miss count",
             &["actor_id", "side"],
             registry
         )
         .unwrap();
 
         let join_total_lookup_count = register_int_counter_vec_with_registry!(
             "stream_join_lookup_total_count",
-            "Join executor lookup total operation",
+            "Join executor lookup total count",
             &["actor_id", "side"],
             registry
         )
         .unwrap();
 
         let join_insert_cache_miss_count = register_int_counter_vec_with_registry!(
             "stream_join_insert_cache_miss_count",
-            "Join executor cache miss when insert operation",
+            "Count of cache miss when insert rows in join executor",
+            &["actor_id", "side"],
+            registry
+        )
+        .unwrap();
+
+        let join_may_exist_true_count = register_int_counter_vec_with_registry!(
+            "stream_join_may_exist_true_count",
+            "Count of may_exist's true returns of when insert rows in join executor",
             &["actor_id", "side"],
             registry
         )
@@ -468,6 +477,7 @@ impl StreamingMetrics {
             join_lookup_miss_count,
             join_total_lookup_count,
             join_insert_cache_miss_count,
+            join_may_exist_true_count,
             join_actor_input_waiting_duration_ns,
             join_match_duration_ns,
             join_barrier_align_duration,