Fix cache for 0471

src/datanode/src/store.rs

@@ -20,6 +20,7 @@ mod gcs;
 mod oss;
 mod s3;
 
+use std::sync::Arc;
 use std::time::Duration;
 use std::{env, path};
 
@@ -28,7 +29,7 @@ use common_telemetry::info;
 use object_store::layers::{LruCacheLayer, RetryLayer};
 use object_store::services::Fs;
 use object_store::util::{join_dir, normalize_dir, with_instrument_layers};
-use object_store::{HttpClient, ObjectStore};
+use object_store::{HttpClient, ObjectStore, ObjectStoreBuilder};
 use snafu::prelude::*;
 
 use crate::config::{ObjectStoreConfig, DEFAULT_OBJECT_STORE_CACHE_SIZE};
@@ -108,11 +109,9 @@ async fn create_object_store_with_cache(
         clean_temp_dir(&atomic_temp_dir)?;
         let mut builder = Fs::default();
         builder.root(path).atomic_write_dir(&atomic_temp_dir);
-        let cache_store = ObjectStore::new(builder)
-            .context(error::InitBackendSnafu)?
-            .finish();
+        let cache_store = builder.build().context(error::InitBackendSnafu)?;
 
-        let cache_layer = LruCacheLayer::new(cache_store, cache_capacity.0 as usize)
+        let cache_layer = LruCacheLayer::new(Arc::new(cache_store), cache_capacity.0 as usize)
             .await
             .context(error::InitBackendSnafu)?;
 

src/object-store/src/layers/lru_cache.rs

@@ -12,26 +12,28 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+
 use opendal::raw::oio::ReadDyn;
 use opendal::raw::{
     Access, Layer, LayeredAccess, OpDelete, OpList, OpRead, OpWrite, RpDelete, RpList, RpRead,
     RpWrite,
 };
-use opendal::{Operator, Result};
+use opendal::Result;
 mod read_cache;
 use common_telemetry::info;
 use read_cache::ReadCache;
 
 /// An opendal layer with local LRU file cache supporting.
 #[derive(Clone)]
-pub struct LruCacheLayer {
+pub struct LruCacheLayer<C: Access> {
     // The read cache
-    read_cache: ReadCache,
+    read_cache: ReadCache<C>,
 }
 
-impl LruCacheLayer {
+impl<C: Access> LruCacheLayer<C> {
     /// Create a `[LruCacheLayer]` with local file cache and capacity in bytes.
-    pub async fn new(file_cache: Operator, capacity: usize) -> Result<Self> {
+    pub async fn new(file_cache: Arc<C>, capacity: usize) -> Result<Self> {
         let read_cache = ReadCache::new(file_cache, capacity);
         let (entries, bytes) = read_cache.recover_cache().await?;
 
@@ -50,12 +52,12 @@ impl LruCacheLayer {
 
     /// Returns the read cache statistics info `(EntryCount, SizeInBytes)`.
     pub async fn read_cache_stat(&self) -> (u64, u64) {
-        self.read_cache.stat().await
+        self.read_cache.cache_stat().await
     }
 }
 
-impl<I: Access> Layer<I> for LruCacheLayer {
-    type LayeredAccess = LruCacheAccess<I>;
+impl<I: Access, C: Access> Layer<I> for LruCacheLayer<C> {
+    type LayeredAccess = LruCacheAccess<I, C>;
 
     fn layer(&self, inner: I) -> Self::LayeredAccess {
         LruCacheAccess {
@@ -66,12 +68,12 @@ impl<I: Access> Layer<I> for LruCacheLayer {
 }
 
 #[derive(Debug)]
-pub struct LruCacheAccess<I> {
+pub struct LruCacheAccess<I, C> {
     inner: I,
-    read_cache: ReadCache,
+    read_cache: ReadCache<C>,
 }
 
-impl<I: Access> LayeredAccess for LruCacheAccess<I> {
+impl<I: Access, C: Access> LayeredAccess for LruCacheAccess<I, C> {
     type Inner = I;
     type Reader = Box<dyn ReadDyn>;
     type BlockingReader = I::BlockingReader;
@@ -85,7 +87,9 @@ impl<I: Access> LayeredAccess for LruCacheAccess<I> {
     }
 
     async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
-        self.read_cache.read(&self.inner, path, args).await
+        self.read_cache
+            .read_from_cache(&self.inner, path, args)
+            .await
     }
 
     async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {