refine

risingwavelabs · Feb 6, 2025 · 0d278ab · 0d278ab
1 parent 4083a47
commit 0d278ab
Show file tree

Hide file tree

Showing 5 changed files with 10 additions and 10 deletions.
diff --git a/src/stream/src/executor/asof_join.rs b/src/stream/src/executor/asof_join.rs
@@ -461,11 +461,11 @@ impl<K: HashKey, S: StateStore, const T: AsOfJoinTypePrimitive> AsOfJoinExecutor
                     if left_post_commit
                         .post_yield_barrier(update_vnode_bitmap)
                         .await?
+                        .unwrap_or(false)
                     {
                         self.watermark_buffers
                             .values_mut()
                             .for_each(|buffers| buffers.clear());
-                        // self.inequality_watermarks.fill(None);
                     }
 
                     // Report metrics of cached join rows/entries

diff --git a/src/stream/src/executor/hash_agg.rs b/src/stream/src/executor/hash_agg.rs
@@ -673,9 +673,8 @@ impl<K: HashKey, S: StateStore> HashAggExecutor<K, S> {
                             post_commit.post_yield_barrier(update_vnode_bitmap.clone())
                         }))
                         .await?
-                        .into_iter()
-                        .flatten()
-                        .next()
+                        .pop()
+                        .expect("should have at least one table")
                         .map(|(_, cache_may_stale)| cache_may_stale)
                     {
                         // Manipulate the cache if necessary.

diff --git a/src/stream/src/executor/hash_join.rs b/src/stream/src/executor/hash_join.rs
@@ -660,6 +660,7 @@ impl<K: HashKey, S: StateStore, const T: JoinTypePrimitive> HashJoinExecutor<K,
                     if left_post_commit
                         .post_yield_barrier(update_vnode_bitmap)
                         .await?
+                        .unwrap_or(false)
                     {
                         self.watermark_buffers
                             .values_mut()

diff --git a/src/stream/src/executor/join/hash_join.rs b/src/stream/src/executor/join/hash_join.rs
@@ -502,15 +502,13 @@ impl<K: HashKey, S: StateStore> JoinHashMapPostCommit<'_, K, S> {
     pub async fn post_yield_barrier(
         self,
         vnode_bitmap: Option<Arc<Bitmap>>,
-    ) -> StreamExecutorResult<bool> {
+    ) -> StreamExecutorResult<Option<bool>> {
         let cache_may_stale = self.state.post_yield_barrier(vnode_bitmap.clone()).await?;
         if let Some(degree_state) = self.degree_state {
             let _ = degree_state.post_yield_barrier(vnode_bitmap).await?;
         }
-        let cache_may_stale = cache_may_stale
-            .map(|(_, cache_may_stale)| cache_may_stale)
-            .unwrap_or(false);
-        if cache_may_stale {
+        let cache_may_stale = cache_may_stale.map(|(_, cache_may_stale)| cache_may_stale);
+        if cache_may_stale.unwrap_or(false) {
             self.inner.clear();
         }
         Ok(cache_may_stale)

diff --git a/src/stream/src/executor/top_n/utils.rs b/src/stream/src/executor/top_n/utils.rs
@@ -40,7 +40,9 @@ pub trait TopNExecutorBase: Send + 'static {
     /// Flush the buffered chunk to the storage backend.
     fn try_flush_data(&mut self) -> impl Future<Output = StreamExecutorResult<()>> + Send;
 
-    fn clear_cache(&mut self) {}
+    fn clear_cache(&mut self) {
+        unreachable!()
+    }
 
     fn evict(&mut self) {}