rename to MemoryBuffer as a trait name

zuston · zuston · commit 187707a4e9db · 2025-12-26T15:12:18.000+08:00
diff --git a/riffle-server/src/store/hybrid.rs b/riffle-server/src/store/hybrid.rs
@@ -63,7 +63,7 @@ use crate::runtime::manager::RuntimeManager;
 use crate::store::local::LocalfileStoreStat;
 use crate::store::mem::buffer::default_buffer::DefaultMemoryBuffer;
 use crate::store::mem::buffer::unified_buffer::UnifiedBuffer;
-use crate::store::mem::buffer::{BufferOps, BufferType};
+use crate::store::mem::buffer::{BufferType, MemoryBuffer};
 use crate::store::mem::capacity::CapacitySnapshot;
 use crate::store::spill::hierarchy_event_bus::HierarchyEventBus;
 use crate::store::spill::storage_flush_handler::StorageFlushHandler;
diff --git a/riffle-server/src/store/mem/buffer.rs b/riffle-server/src/store/mem/buffer.rs
@@ -69,7 +69,7 @@ impl BufferSpillResult {
     }
 }
 
-pub trait BufferOps {
+pub trait MemoryBuffer {
     /// Creates a new buffer instance
     fn new(options: BufferOptions) -> Self
     where
@@ -125,7 +125,7 @@ pub trait BufferOps {
 mod test {
     use crate::store::mem::buffer::default_buffer::DefaultMemoryBuffer;
     use crate::store::mem::buffer::opt_buffer::OptStagingMemoryBuffer;
-    use crate::store::mem::buffer::BufferOps;
+    use crate::store::mem::buffer::MemoryBuffer;
     use crate::store::Block;
     use hashlink::LinkedHashMap;
     use std::collections::LinkedList;
@@ -158,7 +158,8 @@ mod test {
         }
     }
 
-    fn run_test_with_block_id_zero<B: BufferOps + Send + Sync + 'static>() -> anyhow::Result<()> {
+    fn run_test_with_block_id_zero<B: MemoryBuffer + Send + Sync + 'static>() -> anyhow::Result<()>
+    {
         let mut buffer = B::new(Default::default());
         let block_1 = create_block(10, 100);
         let block_2 = create_block(10, 0);
@@ -192,7 +193,7 @@ mod test {
         Ok(())
     }
 
-    fn run_test_put_get<B: BufferOps + Send + Sync + 'static>() -> anyhow::Result<()> {
+    fn run_test_put_get<B: MemoryBuffer + Send + Sync + 'static>() -> anyhow::Result<()> {
         let mut buffer = B::new(Default::default());
 
         /// case1
@@ -300,7 +301,7 @@ mod test {
         Ok(())
     }
 
-    fn run_test_get_v2_is_end_with_only_staging<B: BufferOps + Send + Sync + 'static>(
+    fn run_test_get_v2_is_end_with_only_staging<B: MemoryBuffer + Send + Sync + 'static>(
     ) -> anyhow::Result<()> {
         let buffer = B::new(Default::default());
         // 0 -> 10 blocks with total 100 bytes
@@ -337,7 +338,7 @@ mod test {
         Ok(())
     }
 
-    fn run_test_get_v2_is_end_across_flight_and_staging<B: BufferOps + Send + Sync + 'static>(
+    fn run_test_get_v2_is_end_across_flight_and_staging<B: MemoryBuffer + Send + Sync + 'static>(
     ) -> anyhow::Result<()> {
         let buffer = B::new(Default::default());
 
diff --git a/riffle-server/src/store/mem/buffer/default_buffer.rs b/riffle-server/src/store/mem/buffer/default_buffer.rs
@@ -1,4 +1,4 @@
-use super::{BufferOps, BufferOptions, BufferSpillResult, MemBlockBatch};
+use super::{BufferOptions, BufferSpillResult, MemBlockBatch, MemoryBuffer};
 use crate::composed_bytes;
 use crate::composed_bytes::ComposedBytes;
 use crate::constant::INVALID_BLOCK_ID;
@@ -43,7 +43,7 @@ impl Inner {
     }
 }
 
-impl BufferOps for DefaultMemoryBuffer {
+impl MemoryBuffer for DefaultMemoryBuffer {
     fn new(opt: BufferOptions) -> DefaultMemoryBuffer {
         DefaultMemoryBuffer {
             buffer: Mutex::new(Inner::new()),
diff --git a/riffle-server/src/store/mem/buffer/opt_buffer.rs b/riffle-server/src/store/mem/buffer/opt_buffer.rs
@@ -1,7 +1,7 @@
 use crate::composed_bytes::ComposedBytes;
 use crate::constant::INVALID_BLOCK_ID;
 use crate::store::mem::buffer::default_buffer::DefaultMemoryBuffer;
-use crate::store::mem::buffer::{BufferOps, BufferOptions, BufferSpillResult, MemBlockBatch};
+use crate::store::mem::buffer::{BufferOptions, BufferSpillResult, MemBlockBatch, MemoryBuffer};
 use crate::store::{Block, DataBytes, DataSegment, PartitionedMemoryData};
 use croaring::Treemap;
 use fastrace::trace;
@@ -44,7 +44,7 @@ pub struct OptStagingMemoryBuffer {
     buffer: Mutex<OptStagingBufferInternal>,
 }
 
-impl BufferOps for OptStagingMemoryBuffer {
+impl MemoryBuffer for OptStagingMemoryBuffer {
     #[trace]
     fn new(opt: BufferOptions) -> Self {
         OptStagingMemoryBuffer {
diff --git a/riffle-server/src/store/mem/buffer/unified_buffer.rs b/riffle-server/src/store/mem/buffer/unified_buffer.rs
@@ -1,7 +1,7 @@
 use crate::store::mem::buffer::default_buffer::DefaultMemoryBuffer;
 use crate::store::mem::buffer::opt_buffer::OptStagingMemoryBuffer;
 use crate::store::mem::buffer::unified_buffer::UnifiedBuffer::{DEFAULT, EXPERIMENTAL};
-use crate::store::mem::buffer::{BufferOps, BufferOptions, BufferSpillResult, BufferType};
+use crate::store::mem::buffer::{BufferOptions, BufferSpillResult, BufferType, MemoryBuffer};
 use crate::store::{Block, PartitionedMemoryData};
 use croaring::Treemap;
 
@@ -11,7 +11,7 @@ pub enum UnifiedBuffer {
     EXPERIMENTAL(OptStagingMemoryBuffer),
 }
 
-impl BufferOps for UnifiedBuffer {
+impl MemoryBuffer for UnifiedBuffer {
     fn new(opts: BufferOptions) -> Self
     where
         Self: Sized,
diff --git a/riffle-server/src/store/memory.rs b/riffle-server/src/store/memory.rs
@@ -39,7 +39,7 @@ use crate::runtime::manager::RuntimeManager;
 use crate::store::mem::budget::MemoryBudget;
 use crate::store::mem::buffer::default_buffer::DefaultMemoryBuffer;
 use crate::store::mem::buffer::opt_buffer::OptStagingMemoryBuffer;
-use crate::store::mem::buffer::{BufferOps, BufferOptions, BufferType};
+use crate::store::mem::buffer::{BufferOptions, BufferType, MemoryBuffer};
 use crate::store::mem::capacity::CapacitySnapshot;
 use crate::store::mem::ticket::TicketManager;
 use crate::store::spill::SpillWritingViewContext;
@@ -52,7 +52,7 @@ use fxhash::{FxBuildHasher, FxHasher};
 use log::{debug, info, warn};
 use std::sync::Arc;
 
-pub struct MemoryStore<B: BufferOps + Send + Sync + 'static = DefaultMemoryBuffer> {
+pub struct MemoryStore<B: MemoryBuffer + Send + Sync + 'static = DefaultMemoryBuffer> {
     memory_capacity: i64,
     state: DDashMap<PartitionUId, Arc<B>>,
     budget: MemoryBudget,
@@ -61,10 +61,10 @@ pub struct MemoryStore<B: BufferOps + Send + Sync + 'static = DefaultMemoryBuffe
     cfg: Option<MemoryStoreConfig>,
 }
 
-unsafe impl<B: BufferOps + Send + Sync> Send for MemoryStore<B> {}
-unsafe impl<B: BufferOps + Send + Sync> Sync for MemoryStore<B> {}
+unsafe impl<B: MemoryBuffer + Send + Sync> Send for MemoryStore<B> {}
+unsafe impl<B: MemoryBuffer + Send + Sync> Sync for MemoryStore<B> {}
 
-impl<B: BufferOps + Send + Sync + 'static> MemoryStore<B> {
+impl<B: MemoryBuffer + Send + Sync + 'static> MemoryStore<B> {
     // only for test cases
     pub fn new(max_memory_size: i64) -> Self {
         let budget = MemoryBudget::new(max_memory_size);
@@ -258,7 +258,7 @@ impl<B: BufferOps + Send + Sync + 'static> MemoryStore<B> {
 }
 
 #[async_trait]
-impl<B: BufferOps + Send + Sync + 'static> Store for MemoryStore<B> {
+impl<B: MemoryBuffer + Send + Sync + 'static> Store for MemoryStore<B> {
     fn start(self: Arc<Self>) {
         // ignore
     }
@@ -433,11 +433,11 @@ mod test {
     use crate::app_manager::purge_event::PurgeReason;
     use crate::store::mem::buffer::default_buffer::DefaultMemoryBuffer;
     use crate::store::mem::buffer::opt_buffer::OptStagingMemoryBuffer;
-    use crate::store::mem::buffer::BufferOps;
+    use crate::store::mem::buffer::MemoryBuffer;
     use anyhow::Result;
     use croaring::Treemap;
 
-    fn run_test_read_buffer_in_flight<B: BufferOps + Send + Sync + 'static>() {
+    fn run_test_read_buffer_in_flight<B: MemoryBuffer + Send + Sync + 'static>() {
         let store: MemoryStore<B> = MemoryStore::new(1024);
         let runtime = store.runtime_manager.clone();
 
@@ -610,7 +610,7 @@ mod test {
         run_test_read_buffer_in_flight::<OptStagingMemoryBuffer>();
     }
 
-    async fn get_data_with_last_block_id<B: BufferOps + Send + Sync + 'static>(
+    async fn get_data_with_last_block_id<B: MemoryBuffer + Send + Sync + 'static>(
         default_single_read_size: i64,
         last_block_id: i64,
         store: &MemoryStore<B>,
@@ -653,7 +653,7 @@ mod test {
         WritingViewContext::create_for_test(uid, data_blocks)
     }
 
-    fn run_test_allocated_and_purge_for_memory<B: BufferOps + Send + Sync + 'static>() {
+    fn run_test_allocated_and_purge_for_memory<B: MemoryBuffer + Send + Sync + 'static>() {
         let store: MemoryStore<B> = MemoryStore::new(1024 * 1024 * 1024);
         let runtime = store.runtime_manager.clone();
 
@@ -685,7 +685,7 @@ mod test {
         run_test_allocated_and_purge_for_memory::<OptStagingMemoryBuffer>();
     }
 
-    fn run_test_purge<B: BufferOps + Send + Sync + 'static>() -> Result<()> {
+    fn run_test_purge<B: MemoryBuffer + Send + Sync + 'static>() -> Result<()> {
         let store: MemoryStore<B> = MemoryStore::new(1024);
         let runtime = store.runtime_manager.clone();
 
@@ -768,7 +768,7 @@ mod test {
         run_test_purge::<OptStagingMemoryBuffer>();
     }
 
-    fn run_test_put_and_get_for_memory<B: BufferOps + Send + Sync + 'static>() {
+    fn run_test_put_and_get_for_memory<B: MemoryBuffer + Send + Sync + 'static>() {
         let store: MemoryStore<B> = MemoryStore::new(1024 * 1024 * 1024);
         let runtime = store.runtime_manager.clone();
 
@@ -817,7 +817,7 @@ mod test {
         run_test_put_and_get_for_memory::<OptStagingMemoryBuffer>();
     }
 
-    fn run_test_block_id_filter_for_memory<B: BufferOps + Send + Sync + 'static>() {
+    fn run_test_block_id_filter_for_memory<B: MemoryBuffer + Send + Sync + 'static>() {
         let store: MemoryStore<B> = MemoryStore::new(1024 * 1024 * 1024);
         let runtime = store.runtime_manager.clone();
 
diff --git a/riffle-server/src/store/spill/spill_test.rs b/riffle-server/src/store/spill/spill_test.rs
@@ -15,7 +15,7 @@ pub mod tests {
     };
     use crate::runtime::manager::RuntimeManager;
     use crate::store::hybrid::{HybridStore, PersistentStore};
-    use crate::store::mem::buffer::BufferOps;
+    use crate::store::mem::buffer::MemoryBuffer;
     use crate::store::spill::spill_test::mock::MockStore;
     use crate::store::spill::storage_flush_handler::StorageFlushHandler;
     use crate::store::spill::storage_select_handler::StorageSelectHandler;

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-use super::{BufferOps, BufferOptions, BufferSpillResult, MemBlockBatch};`
	`1`	`+use super::{BufferOptions, BufferSpillResult, MemBlockBatch, MemoryBuffer};`
`2`	`2`	`use crate::composed_bytes;`
`3`	`3`	`use crate::composed_bytes::ComposedBytes;`
`4`	`4`	`use crate::constant::INVALID_BLOCK_ID;`
`@@ -43,7 +43,7 @@ impl Inner {`
`43`	`43`	`}`
`44`	`44`	`}`
`45`	`45`
`46`		`-impl BufferOps for DefaultMemoryBuffer {`
	`46`	`+impl MemoryBuffer for DefaultMemoryBuffer {`
`47`	`47`	`fn new(opt: BufferOptions) -> DefaultMemoryBuffer {`
`48`	`48`	`DefaultMemoryBuffer {`
`49`	`49`	`buffer: Mutex::new(Inner::new()),`