feat(log-store): refine log store trait and add update vnode bitmap i…

…n reader (#11959)

src/connector/src/sink/boxed.rs

@@ -14,6 +14,7 @@
 
 use std::fmt::{Debug, Formatter};
 use std::ops::{Deref, DerefMut};
+use std::sync::Arc;
 
 use async_trait::async_trait;
 use risingwave_common::array::StreamChunk;
@@ -54,7 +55,7 @@ impl<CM: 'static + Send> SinkWriter for BoxWriter<CM> {
         self.deref_mut().abort().await
     }
 
-    async fn update_vnode_bitmap(&mut self, vnode_bitmap: Bitmap) -> crate::sink::Result<()> {
+    async fn update_vnode_bitmap(&mut self, vnode_bitmap: Arc<Bitmap>) -> crate::sink::Result<()> {
         self.deref_mut().update_vnode_bitmap(vnode_bitmap).await
     }
 }

src/connector/src/sink/clickhouse.rs

@@ -19,7 +19,6 @@ use anyhow::anyhow;
 use clickhouse::{Client, Row as ClickHouseRow};
 use itertools::Itertools;
 use risingwave_common::array::{Op, RowRef, StreamChunk};
-use risingwave_common::buffer::Bitmap;
 use risingwave_common::catalog::Schema;
 use risingwave_common::row::Row;
 use risingwave_common::types::{DataType, ScalarRefImpl, Serial};
@@ -439,14 +438,6 @@ impl SinkWriter for ClickHouseSinkWriter {
     async fn barrier(&mut self, _is_checkpoint: bool) -> Result<()> {
         Ok(())
     }
-
-    async fn abort(&mut self) -> Result<()> {
-        Ok(())
-    }
-
-    async fn update_vnode_bitmap(&mut self, _vnode_bitmap: Bitmap) -> Result<()> {
-        Ok(())
-    }
 }
 
 #[derive(ClickHouseRow, Deserialize)]

src/connector/src/sink/coordinate.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+
 use anyhow::anyhow;
 use risingwave_common::array::StreamChunk;
 use risingwave_common::buffer::Bitmap;
@@ -81,7 +83,7 @@ impl<W: SinkWriter<CommitMetadata = Option<SinkMetadata>>> SinkWriter for Coordi
         self.inner.abort().await
     }
 
-    async fn update_vnode_bitmap(&mut self, vnode_bitmap: Bitmap) -> Result<()> {
+    async fn update_vnode_bitmap(&mut self, vnode_bitmap: Arc<Bitmap>) -> Result<()> {
         self.inner.update_vnode_bitmap(vnode_bitmap).await
     }
 }

src/connector/src/sink/iceberg.rs

@@ -411,12 +411,6 @@ impl SinkWriter for IcebergWriter {
         // TODO: abort should clean up all the data written in this epoch.
         Ok(())
     }
-
-    /// Update the vnode bitmap of current sink writer
-    async fn update_vnode_bitmap(&mut self, _vnode_bitmap: Bitmap) -> Result<()> {
-        // Just skip it.
-        Ok(())
-    }
 }
 
 pub struct IcebergSinkCommitter {

src/connector/src/sink/kinesis.rs

@@ -22,7 +22,6 @@ use aws_sdk_kinesis::primitives::Blob;
 use aws_sdk_kinesis::Client as KinesisClient;
 use futures_async_stream::for_await;
 use risingwave_common::array::StreamChunk;
-use risingwave_common::buffer::Bitmap;
 use risingwave_common::catalog::Schema;
 use risingwave_rpc_client::ConnectorClient;
 use serde_derive::Deserialize;
@@ -272,14 +271,6 @@ impl SinkWriter for KinesisSinkWriter {
     async fn barrier(&mut self, _is_checkpoint: bool) -> Result<()> {
         Ok(())
     }
-
-    async fn abort(&mut self) -> Result<()> {
-        Ok(())
-    }
-
-    async fn update_vnode_bitmap(&mut self, _vnode_bitmap: Bitmap) -> Result<()> {
-        Ok(())
-    }
 }
 
 #[macro_export]

src/connector/src/sink/mod.rs

@@ -27,6 +27,7 @@ pub mod test_sink;
 pub mod utils;
 
 use std::collections::HashMap;
+use std::sync::Arc;
 
 use ::clickhouse::error::Error as ClickHouseError;
 use anyhow::anyhow;
@@ -173,10 +174,14 @@ pub trait SinkWriter: Send + 'static {
     async fn barrier(&mut self, is_checkpoint: bool) -> Result<Self::CommitMetadata>;
 
     /// Clean up
-    async fn abort(&mut self) -> Result<()>;
+    async fn abort(&mut self) -> Result<()> {
+        Ok(())
+    }
 
     /// Update the vnode bitmap of current sink writer
-    async fn update_vnode_bitmap(&mut self, vnode_bitmap: Bitmap) -> Result<()>;
+    async fn update_vnode_bitmap(&mut self, _vnode_bitmap: Arc<Bitmap>) -> Result<()> {
+        Ok(())
+    }
 }
 
 #[async_trait]
@@ -240,10 +245,6 @@ impl<W: SinkWriterV1> SinkWriter for SinkWriterV1Adapter<W> {
     async fn abort(&mut self) -> Result<()> {
         self.inner.abort().await
     }
-
-    async fn update_vnode_bitmap(&mut self, _vnode_bitmap: Bitmap) -> Result<()> {
-        Ok(())
-    }
 }
 
 #[async_trait]
@@ -314,17 +315,9 @@ impl SinkWriter for BlackHoleSink {
         Ok(())
     }
 
-    async fn abort(&mut self) -> Result<()> {
-        Ok(())
-    }
-
     async fn barrier(&mut self, _is_checkpoint: bool) -> Result<()> {
         Ok(())
     }
-
-    async fn update_vnode_bitmap(&mut self, _vnode_bitmap: Bitmap) -> Result<()> {
-        Ok(())
-    }
 }
 
 impl SinkConfig {

src/connector/src/sink/nats.rs

@@ -17,7 +17,6 @@ use std::collections::HashMap;
 use anyhow::anyhow;
 use async_nats::jetstream::context::Context;
 use risingwave_common::array::StreamChunk;
-use risingwave_common::buffer::Bitmap;
 use risingwave_common::catalog::Schema;
 use risingwave_common::error::anyhow_error;
 use risingwave_rpc_client::ConnectorClient;
@@ -155,12 +154,4 @@ impl SinkWriter for NatsSinkWriter {
     async fn barrier(&mut self, _is_checkpoint: bool) -> Result<()> {
         Ok(())
     }
-
-    async fn abort(&mut self) -> Result<()> {
-        Ok(())
-    }
-
-    async fn update_vnode_bitmap(&mut self, _vnode_bitmap: Bitmap) -> Result<()> {
-        Ok(())
-    }
 }

src/connector/src/sink/redis.rs

@@ -14,7 +14,6 @@
 
 use async_trait::async_trait;
 use risingwave_common::array::StreamChunk;
-use risingwave_common::buffer::Bitmap;
 use risingwave_common::catalog::Schema;
 use risingwave_rpc_client::ConnectorClient;
 
@@ -58,15 +57,7 @@ impl SinkWriter for RedisSinkWriter {
         todo!()
     }
 
-    async fn abort(&mut self) -> Result<()> {
-        todo!()
-    }
-
     async fn barrier(&mut self, _is_checkpoint: bool) -> Result<()> {
         todo!()
     }
-
-    async fn update_vnode_bitmap(&mut self, _vnode_bitmap: Bitmap) -> Result<()> {
-        todo!()
-    }
 }

src/connector/src/sink/remote.rs

@@ -20,7 +20,6 @@ use async_trait::async_trait;
 use itertools::Itertools;
 use prost::Message;
 use risingwave_common::array::StreamChunk;
-use risingwave_common::buffer::Bitmap;
 use risingwave_common::catalog::Schema;
 use risingwave_common::error::anyhow_error;
 use risingwave_common::types::DataType;
@@ -409,15 +408,6 @@ where
             Ok(<Self as HandleBarrierResponse>::non_checkpoint_return_value())
         }
     }
-
-    async fn abort(&mut self) -> Result<()> {
-        Ok(())
-    }
-
-    async fn update_vnode_bitmap(&mut self, _vnode_bitmap: Bitmap) -> Result<()> {
-        // TODO: handle scaling
-        Ok(())
-    }
 }
 
 pub struct RemoteCoordinator {