catch up changes

Signed-off-by: tison <wander4096@gmail.com>

src/common/datasource/src/file_format.rs

@@ -149,7 +149,9 @@ pub fn open_with_decoder<T: ArrowDecoder, F: Fn() -> DataFusionResult<T>>(
             .reader(&path)
             .await
             .map_err(|e| DataFusionError::External(Box::new(e)))?
-            .into_bytes_stream(..);
+            .into_bytes_stream(..)
+            .await
+            .map_err(|e| DataFusionError::External(Box::new(e)))?;
 
         let mut upstream = compression_type.convert_stream(reader).fuse();
 

src/common/datasource/src/file_format/csv.rs

@@ -169,11 +169,14 @@ impl FileFormat for CsvFormat {
             .stat(path)
             .await
             .context(error::ReadObjectSnafu { path })?;
+
         let reader = store
             .reader(path)
             .await
             .context(error::ReadObjectSnafu { path })?
             .into_futures_async_read(0..meta.content_length())
+            .await
+            .context(error::ReadObjectSnafu { path })?
             .compat();
 
         let decoded = self.compression_type.convert_async_read(reader);

src/common/datasource/src/file_format/json.rs

@@ -87,11 +87,14 @@ impl FileFormat for JsonFormat {
             .stat(path)
             .await
             .context(error::ReadObjectSnafu { path })?;
+
         let reader = store
             .reader(path)
             .await
             .context(error::ReadObjectSnafu { path })?
             .into_futures_async_read(0..meta.content_length())
+            .await
+            .context(error::ReadObjectSnafu { path })?
             .compat();
 
         let decoded = self.compression_type.convert_async_read(reader);

src/common/datasource/src/file_format/parquet.rs

@@ -52,11 +52,14 @@ impl FileFormat for ParquetFormat {
             .stat(path)
             .await
             .context(error::ReadObjectSnafu { path })?;
+
         let mut reader = store
             .reader(path)
             .await
             .context(error::ReadObjectSnafu { path })?
             .into_futures_async_read(0..meta.content_length())
+            .await
+            .context(error::ReadObjectSnafu { path })?
             .compat();
 
         let metadata = reader
@@ -129,6 +132,7 @@ impl LazyParquetFileReader {
                 .reader(&self.path)
                 .await?
                 .into_futures_async_read(0..meta.content_length())
+                .await?
                 .compat();
             self.reader = Some(reader);
         }

src/mito2/src/cache/write_cache.rs

@@ -188,7 +188,9 @@ impl WriteCache {
             .reader(&cache_path)
             .await
             .context(error::OpenDalSnafu)?
-            .into_futures_async_read(0..cached_value.content_length());
+            .into_futures_async_read(0..cached_value.content_length())
+            .await
+            .context(error::OpenDalSnafu)?;
 
         let mut writer = remote_store
             .writer_with(upload_path)

src/mito2/src/sst/index/applier.rs

@@ -27,10 +27,7 @@ use snafu::{OptionExt, ResultExt};
 use store_api::storage::RegionId;
 
 use crate::cache::file_cache::{FileCacheRef, FileType, IndexKey};
-use crate::error::{
-    ApplyIndexSnafu, PuffinBlobTypeNotFoundSnafu, PuffinReadBlobSnafu, PuffinReadMetadataSnafu,
-    Result,
-};
+use crate::error::{ApplyIndexSnafu, OpenDalSnafu, PuffinBlobTypeNotFoundSnafu, PuffinReadBlobSnafu, PuffinReadMetadataSnafu, Result};
 use crate::metrics::{
     INDEX_APPLY_ELAPSED, INDEX_APPLY_MEMORY_USAGE, INDEX_PUFFIN_READ_BYTES_TOTAL,
     INDEX_PUFFIN_READ_OP_TOTAL, INDEX_PUFFIN_SEEK_OP_TOTAL,
@@ -128,11 +125,19 @@ impl SstIndexApplier {
             return Ok(None);
         };
 
-        Ok(file_cache
+        let Some(reader) = file_cache
             .reader(IndexKey::new(self.region_id, file_id, FileType::Puffin))
             .await
-            .map(|v| v.into_futures_async_read(0..indexed_value.file_size as u64))
-            .map(PuffinFileReader::new))
+        else {
+            return Ok(None);
+        };
+
+        let reader = reader
+            .into_futures_async_read(0..indexed_value.file_size as u64)
+            .await
+            .context(OpenDalSnafu)?;
+
+        Ok(Some(PuffinFileReader::new(reader)))
     }
 
     /// Helper function to create a [`PuffinFileReader`] from the remote index file.

src/mito2/src/sst/index/store.rs

@@ -67,7 +67,9 @@ impl InstrumentedStore {
             .reader(path)
             .await
             .context(OpenDalSnafu)?
-            .into_futures_async_read(0..meta.content_length());
+            .into_futures_async_read(0..meta.content_length())
+            .await
+            .context(OpenDalSnafu)?;
         Ok(InstrumentedAsyncRead::new(
             reader,
             read_byte_count,

src/object-store/Cargo.toml

@@ -17,7 +17,7 @@ futures.workspace = true
 lazy_static.workspace = true
 md5 = "0.7"
 moka = { workspace = true, features = ["future"] }
-opendal = { version = "0.46", features = [
+opendal = { version = "0.47", features = [
     "layers-tracing",
     "services-azblob",
     "services-fs",

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

@@ -12,8 +12,6 @@
 // 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,
@@ -75,7 +73,7 @@ pub struct LruCacheAccess<I> {
 
 impl<I: Access> LayeredAccess for LruCacheAccess<I> {
     type Inner = I;
-    type Reader = Arc<dyn ReadDyn>;
+    type Reader = Box<dyn ReadDyn>;
     type BlockingReader = I::BlockingReader;
     type Writer = I::Writer;
     type BlockingWriter = I::BlockingWriter;

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

@@ -18,8 +18,8 @@ use common_telemetry::debug;
 use futures::{FutureExt, StreamExt};
 use moka::future::Cache;
 use moka::notification::ListenerFuture;
-use opendal::raw::oio::{Read, ReadDyn, Reader};
-use opendal::raw::{Access, BytesRange, OpRead, OpStat, RpRead};
+use opendal::raw::oio::{Read, Reader};
+use opendal::raw::{Access, OpRead, RpRead};
 use opendal::{Buffer, Error as OpendalError, ErrorKind, Operator, Result};
 
 use crate::metrics::{
@@ -51,9 +51,9 @@ fn can_cache(path: &str) -> bool {
     !path.ends_with("_last_checkpoint")
 }
 
-/// Generate an unique cache key for the read path and range.
-fn read_cache_key(path: &str, range: BytesRange) -> String {
-    format!("{:x}.cache-{}", md5::compute(path), range.to_header())
+/// Generate a unique cache key for the read path and range.
+fn read_cache_key(path: &str) -> String {
+    format!("{:x}.cache", md5::compute(path))
 }
 
 /// Local read cache for files in object storage
@@ -161,32 +161,29 @@ impl ReadCache {
     /// Read from a specific path using the OpRead operation.
     /// It will attempt to retrieve the data from the local cache.
     /// If the data is not found in the local cache,
-    /// it will fallback to retrieving it from remote object storage
+    /// it will fall back to retrieving it from remote object storage
     /// and cache the result locally.
     pub(crate) async fn read<I>(
         &self,
         inner: &I,
         path: &str,
         args: OpRead,
-    ) -> Result<(RpRead, Arc<dyn ReadDyn>)>
+    ) -> Result<(RpRead, Reader)>
     where
         I: Access,
     {
         if !can_cache(path) {
             return inner.read(path, args).await.map(to_output_reader);
         }
 
-        // FIXME: remove this block after opendal v0.47 released.
-        let meta = inner.stat(path, OpStat::new()).await?;
         let (rp, reader) = inner.read(path, args).await?;
         let reader: ReadCacheReader<I> = ReadCacheReader {
             path: Arc::new(path.to_string()),
             inner_reader: reader,
-            size: meta.into_metadata().content_length(),
             file_cache: self.file_cache.clone(),
             mem_cache: self.mem_cache.clone(),
         };
-        Ok((rp, Arc::new(reader)))
+        Ok((rp, Box::new(reader)))
     }
 }
 
@@ -195,14 +192,6 @@ pub struct ReadCacheReader<I: Access> {
     path: Arc<String>,
     /// Remote file reader.
     inner_reader: I::Reader,
-    /// FIXME: remove this field after opendal v0.47 released.
-    ///
-    /// OpenDAL's read_at takes `offset, limit` which means the underlying storage
-    /// services could return less data than limit. We store size here as a workaround.
-    ///
-    /// This API has been refactor into `offset, size` instead. After opendal v0.47 released,
-    /// we don't need this anymore.
-    size: u64,
     /// Local file cache backend
     file_cache: Operator,
     /// Local memory cache to track local cache files
@@ -211,11 +200,11 @@ pub struct ReadCacheReader<I: Access> {
 
 impl<I: Access> ReadCacheReader<I> {
     /// TODO: we can return the Buffer directly to avoid another read from cache.
-    async fn read_remote(&self, offset: u64, limit: usize) -> Result<ReadResult> {
+    async fn read_remote(&mut self) -> Result<ReadResult> {
         OBJECT_STORE_LRU_CACHE_MISS.inc();
 
-        let buf = self.inner_reader.read_at(offset, limit).await?;
-        let result = self.try_write_cache(buf, offset).await;
+        let buf = self.inner_reader.read_all().await?;
+        let result = self.try_write_cache(buf).await;
 
         match result {
             Ok(read_bytes) => {
@@ -243,29 +232,27 @@ impl<I: Access> ReadCacheReader<I> {
         }
     }
 
-    async fn try_write_cache(&self, buf: Buffer, offset: u64) -> Result<usize> {
+    async fn try_write_cache(&self, buf: Buffer) -> Result<usize> {
         let size = buf.len();
-        let read_key = read_cache_key(&self.path, BytesRange::new(offset, Some(size as _)));
+        let read_key = read_cache_key(&self.path);
         self.file_cache.write(&read_key, buf).await?;
         Ok(size)
     }
 }
 
 impl<I: Access> Read for ReadCacheReader<I> {
-    async fn read_at(&self, offset: u64, limit: usize) -> Result<Buffer> {
-        let size = self.size.min(offset + limit as u64) - offset;
-        let read_key = read_cache_key(&self.path, BytesRange::new(offset, Some(size as _)));
-
-        let read_result = self
-            .mem_cache
-            .try_get_with(read_key.clone(), self.read_remote(offset, limit))
+    async fn read(&mut self) -> Result<Buffer> {
+        let read_key = read_cache_key(&self.path);
+        let mem_cache = self.mem_cache.clone();
+        let read_result = mem_cache
+            .try_get_with(read_key.clone(), self.read_remote())
             .await
-            .map_err(|e| OpendalError::new(e.kind(), &e.to_string()))?;
+            .map_err(|e| OpendalError::new(e.kind(), e.to_string()))?;
 
         match read_result {
             ReadResult::Success(_) => {
                 // There is a concurrent issue here, the local cache may be purged
-                // while reading, we have to fallback to remote read
+                // while reading, we have to fall back to remote read
                 match self.file_cache.read(&read_key).await {
                     Ok(ret) => {
                         OBJECT_STORE_LRU_CACHE_HIT
@@ -275,7 +262,7 @@ impl<I: Access> Read for ReadCacheReader<I> {
                     }
                     Err(_) => {
                         OBJECT_STORE_LRU_CACHE_MISS.inc();
-                        self.inner_reader.read_at(offset, limit).await
+                        self.inner_reader.read().await
                     }
                 }
             }
@@ -286,15 +273,15 @@ impl<I: Access> Read for ReadCacheReader<I> {
 
                 Err(OpendalError::new(
                     ErrorKind::NotFound,
-                    &format!("File not found: {}", self.path),
+                    format!("File not found: {}", self.path),
                 ))
             }
         }
     }
 }
 
 fn to_output_reader<R: Read + 'static>(input: (RpRead, R)) -> (RpRead, Reader) {
-    (input.0, Arc::new(input.1))
+    (input.0, Box::new(input.1))
 }
 
 #[cfg(test)]

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

@@ -63,7 +63,7 @@ fn increment_errors_total(op: Operation, kind: ErrorKind) {
     );
 }
 
-/// Please refer to [prometheus](https://docs.rs/prometheus) for every operations.
+/// Please refer to [prometheus](https://docs.rs/prometheus) for every operation.
 ///
 /// # Prometheus Metrics
 ///
@@ -441,17 +441,17 @@ impl<R> PrometheusMetricWrapper<R> {
 }
 
 impl<R: oio::Read> oio::Read for PrometheusMetricWrapper<R> {
-    async fn read_at(&self, offset: u64, limit: usize) -> Result<Buffer> {
-        self.inner.read_at(offset, limit).await.map_err(|err| {
+    async fn read(&mut self) -> Result<Buffer> {
+        self.inner.read().await.map_err(|err| {
             increment_errors_total(self.op, err.kind());
             err
         })
     }
 }
 
 impl<R: oio::BlockingRead> oio::BlockingRead for PrometheusMetricWrapper<R> {
-    fn read_at(&self, offset: u64, limit: usize) -> opendal::Result<Buffer> {
-        self.inner.read_at(offset, limit).map_err(|err| {
+    fn read(&mut self) -> opendal::Result<Buffer> {
+        self.inner.read().map_err(|err| {
             increment_errors_total(self.op, err.kind());
             err
         })

src/operator/src/statement/copy_table_from.rs

@@ -156,6 +156,8 @@ impl StatementExecutor {
                     .await
                     .context(error::ReadObjectSnafu { path: &path })?
                     .into_futures_async_read(0..meta.content_length())
+                    .await
+                    .context(error::ReadObjectSnafu { path: &path })?
                     .compat();
                 let metadata = ArrowReaderMetadata::load_async(&mut reader, Default::default())
                     .await
@@ -301,6 +303,8 @@ impl StatementExecutor {
                     .await
                     .context(error::ReadObjectSnafu { path })?
                     .into_futures_async_read(0..meta.content_length())
+                    .await
+                    .context(error::ReadObjectSnafu { path })?
                     .compat();
                 let builder =
                     ParquetRecordBatchStreamBuilder::new_with_metadata(reader, metadata.clone());