rename

Cargo.toml

@@ -93,7 +93,7 @@ parquet = { version = "53", default-features = false, features = [
     "lz4",
     "snap",
 ] }
-parquet-lru = { version = "0.1.0", path = "parquet-lru" }
+parquet-lru = { version = "0.2.0", path = "parquet-lru" }
 pin-project-lite = "0.2"
 regex = "1"
 thiserror = "2.0.3"

bindings/python/Cargo.toml

@@ -18,7 +18,6 @@ fusio-dispatch = { git = "https://github.com/tonbo-io/fusio.git", rev = "8038993
     "tokio",
 ] }
 futures = { version = "0.3" }
-parquet-lru = { version = "0.1.0", path = "../../parquet-lru" }
 pyo3 = { version = "0.21.2", features = [
     "abi3",
     "abi3-py310",

parquet-lru/Cargo.toml

@@ -1,10 +1,13 @@
 [package]
-description = "Implement LRU cache for parquet::arrow::async_reader::AsyncFileReader."
+description = "Implement LRU cache reader for parquet::arrow::async_reader::AsyncFileReader."
 documentation = "https://docs.rs/parquet-lru"
 edition = "2021"
 license = "Apache-2.0"
 name = "parquet-lru"
-version = "0.1.0"
+version = "0.2.0"
+
+[package.metadata.docs.rs]
+all-features = true
 
 [features]
 default = []

src/compaction/mod.rs

@@ -61,7 +61,7 @@ where
 
     pub(crate) async fn check_then_compaction(
         &mut self,
-        parquet_lru_cache: ParquetLru,
+        parquet_lru: ParquetLru,
     ) -> Result<(), CompactionError<R>> {
         let mut guard = self.schema.write().await;
 
@@ -110,7 +110,7 @@ where
                         &mut delete_gens,
                         &guard.record_instance,
                         &self.manager,
-                        parquet_lru_cache,
+                        parquet_lru,
                     )
                     .await?;
                 }
@@ -198,7 +198,7 @@ where
         delete_gens: &mut Vec<(FileId, usize)>,
         instance: &RecordInstance,
         manager: &StoreManager,
-        parquet_cache: ParquetLru,
+        parquet_lru: ParquetLru,
     ) -> Result<(), CompactionError<R>> {
         let mut level = 0;
 
@@ -224,7 +224,7 @@ where
                         .await?;
 
                     streams.push(ScanStream::SsTable {
-                        inner: SsTable::open(parquet_cache.clone(), scope.gen, file)
+                        inner: SsTable::open(parquet_lru.clone(), scope.gen, file)
                             .await?
                             .scan(
                                 (Bound::Unbounded, Bound::Unbounded),
@@ -247,7 +247,7 @@ where
                     None,
                     ProjectionMask::all(),
                     level_fs.clone(),
-                    parquet_cache.clone(),
+                    parquet_lru.clone(),
                 )
                 .ok_or(CompactionError::EmptyLevel)?;
 
@@ -268,7 +268,7 @@ where
                     None,
                     ProjectionMask::all(),
                     level_fs.clone(),
-                    parquet_cache.clone(),
+                    parquet_lru.clone(),
                 )
                 .ok_or(CompactionError::EmptyLevel)?;
 

src/lib.rs

@@ -183,7 +183,7 @@ where
     version_set: VersionSet<R>,
     lock_map: LockMap<R::Key>,
     manager: Arc<StoreManager>,
-    parquet_lru_cache: ParquetLru,
+    parquet_lru: ParquetLru,
     _p: PhantomData<E>,
 }
 
@@ -309,7 +309,7 @@ where
             version_set,
             lock_map: Arc::new(Default::default()),
             manager,
-            parquet_lru_cache: lru_cache,
+            parquet_lru: lru_cache,
             _p: Default::default(),
         })
     }
@@ -324,7 +324,7 @@ where
             self.schema.read().await,
             self.version_set.current().await,
             self.manager.clone(),
-            self.parquet_lru_cache.clone(),
+            self.parquet_lru.clone(),
         )
     }
 
@@ -381,7 +381,7 @@ where
                 key,
                 self.version_set.load_ts(),
                 Projection::All,
-                self.parquet_lru_cache.clone(),
+                self.parquet_lru.clone(),
             )
             .await?
             .and_then(|entry| {
@@ -409,7 +409,7 @@ where
                 self.version_set.load_ts(),
                 &*current,
                 Box::new(|_| None),
-                self.parquet_lru_cache.clone(),
+                self.parquet_lru.clone(),
             ).take().await?;
 
             while let Some(record) = scan.next().await {
@@ -596,7 +596,7 @@ where
         key: &'get R::Key,
         ts: Timestamp,
         projection: Projection,
-        parquet_cache: ParquetLru,
+        parquet_lru: ParquetLru,
     ) -> Result<Option<Entry<'get, R>>, DbError<R>> {
         if let Some(entry) = self.mutable.get(key, ts) {
             return Ok(Some(Entry::Mutable(entry)));
@@ -630,7 +630,7 @@ where
                 manager,
                 TimestampedRef::new(key, ts),
                 projection,
-                parquet_cache,
+                parquet_lru,
             )
             .await?
             .map(|entry| Entry::RecordBatch(entry)))
@@ -671,7 +671,7 @@ where
     projection_indices: Option<Vec<usize>>,
     projection: ProjectionMask,
 
-    parquet_cache: ParquetLru,
+    parquet_lru: ParquetLru,
 }
 
 impl<'scan, 'range, R> Scan<'scan, 'range, R>
@@ -687,7 +687,7 @@ where
         fn_pre_stream: Box<
             dyn FnOnce(Option<ProjectionMask>) -> Option<ScanStream<'scan, R>> + Send + 'scan,
         >,
-        parquet_cache: ParquetLru,
+        parquet_lru: ParquetLru,
     ) -> Self {
         Self {
             schema,
@@ -700,7 +700,7 @@ where
             limit: None,
             projection_indices: None,
             projection: ProjectionMask::all(),
-            parquet_cache,
+            parquet_lru,
         }
     }
 
@@ -776,7 +776,7 @@ where
                 self.ts,
                 self.limit,
                 self.projection,
-                self.parquet_cache,
+                self.parquet_lru,
             )
             .await?;
 
@@ -829,7 +829,7 @@ where
                 self.ts,
                 self.limit,
                 self.projection,
-                self.parquet_cache,
+                self.parquet_lru,
             )
             .await?;
         let merge_stream = MergeStream::from_vec(streams, self.ts).await?;
@@ -1350,7 +1350,7 @@ pub(crate) mod tests {
             version_set,
             lock_map: Arc::new(Default::default()),
             manager,
-            parquet_lru_cache: Arc::new(NoCache::default()),
+            parquet_lru: Arc::new(NoCache::default()),
             _p: Default::default(),
         })
     }

src/snapshot.rs

@@ -21,7 +21,7 @@ where
     share: RwLockReadGuard<'s, Schema<R>>,
     version: VersionRef<R>,
     manager: Arc<StoreManager>,
-    parquet_cache: ParquetLru,
+    parquet_lru: ParquetLru,
 }
 
 impl<'s, R> Snapshot<'s, R>
@@ -41,7 +41,7 @@ where
                 key,
                 self.ts,
                 projection,
-                self.parquet_cache.clone(),
+                self.parquet_lru.clone(),
             )
             .await?
             .and_then(|entry| {
@@ -64,22 +64,22 @@ where
             self.ts,
             &self.version,
             Box::new(move |_: Option<ProjectionMask>| None),
-            self.parquet_cache.clone(),
+            self.parquet_lru.clone(),
         )
     }
 
     pub(crate) fn new(
         share: RwLockReadGuard<'s, Schema<R>>,
         version: VersionRef<R>,
         manager: Arc<StoreManager>,
-        parquet_cache: ParquetLru,
+        parquet_lru: ParquetLru,
     ) -> Self {
         Self {
             ts: version.load_ts(),
             share,
             version,
             manager,
-            parquet_cache,
+            parquet_lru,
         }
     }
 
@@ -109,7 +109,7 @@ where
             self.ts,
             &self.version,
             fn_pre_stream,
-            self.parquet_cache.clone(),
+            self.parquet_lru.clone(),
         )
     }
 }

src/stream/level.rs

@@ -58,7 +58,7 @@ where
     status: FutureStatus<'level, R>,
     fs: Arc<dyn DynFs>,
     path: Option<Path>,
-    parquet_cache: Arc<dyn DynLruCache<Ulid> + Send + Sync>,
+    parquet_lru: Arc<dyn DynLruCache<Ulid> + Send + Sync>,
 }
 
 impl<'level, R> LevelStream<'level, R>
@@ -77,7 +77,7 @@ where
         limit: Option<usize>,
         projection_mask: ProjectionMask,
         fs: Arc<dyn DynFs>,
-        parquet_cache: Arc<dyn DynLruCache<Ulid> + Send + Sync>,
+        parquet_lru: Arc<dyn DynLruCache<Ulid> + Send + Sync>,
     ) -> Option<Self> {
         let (lower, upper) = range;
         let mut gens: VecDeque<FileId> = version.level_slice[level][start..end + 1]
@@ -99,7 +99,7 @@ where
             status,
             fs,
             path: None,
-            parquet_cache,
+            parquet_lru,
         })
     }
 }
@@ -175,7 +175,7 @@ where
                     Poll::Ready(Ok(file)) => {
                         let id = *id;
                         self.status = FutureStatus::OpenSst(Box::pin(SsTable::open(
-                            self.parquet_cache.clone(),
+                            self.parquet_lru.clone(),
                             id,
                             file,
                         )));

src/version/mod.rs

@@ -121,7 +121,7 @@ where
         manager: &StoreManager,
         key: &TimestampedRef<R::Key>,
         projection_mask: ProjectionMask,
-        parquet_cache: ParquetLru,
+        parquet_lru: ParquetLru,
     ) -> Result<Option<RecordBatchEntry<R>>, VersionError<R>> {
         let level_0_path = self
             .option
@@ -139,7 +139,7 @@ where
                     0,
                     scope.gen,
                     projection_mask.clone(),
-                    parquet_cache.clone(),
+                    parquet_lru.clone(),
                 )
                 .await?
             {
@@ -167,7 +167,7 @@ where
                     leve,
                     sort_runs[index].gen,
                     projection_mask.clone(),
-                    parquet_cache.clone(),
+                    parquet_lru.clone(),
                 )
                 .await?
             {
@@ -185,7 +185,7 @@ where
         level: usize,
         gen: FileId,
         projection_mask: ProjectionMask,
-        parquet_cache: ParquetLru,
+        parquet_lru: ParquetLru,
     ) -> Result<Option<RecordBatchEntry<R>>, VersionError<R>> {
         let file = store
             .open_options(
@@ -194,7 +194,7 @@ where
             )
             .await
             .map_err(VersionError::Fusio)?;
-        SsTable::<R>::open(parquet_cache, gen, file)
+        SsTable::<R>::open(parquet_lru, gen, file)
             .await?
             .get(key, projection_mask)
             .await
@@ -220,7 +220,7 @@ where
         ts: Timestamp,
         limit: Option<usize>,
         projection_mask: ProjectionMask,
-        parquet_cache: ParquetLru,
+        parquet_lru: ParquetLru,
     ) -> Result<(), VersionError<R>> {
         let level_0_path = self
             .option
@@ -238,7 +238,7 @@ where
                 )
                 .await
                 .map_err(VersionError::Fusio)?;
-            let table = SsTable::open(parquet_cache.clone(), scope.gen, file).await?;
+            let table = SsTable::open(parquet_lru.clone(), scope.gen, file).await?;
 
             streams.push(ScanStream::SsTable {
                 inner: table
@@ -283,7 +283,7 @@ where
                     limit,
                     projection_mask.clone(),
                     level_fs.clone(),
-                    parquet_cache.clone(),
+                    parquet_lru.clone(),
                 )
                 .unwrap(),
             });