risingwave_object_store/object/opendal_engine/
opendal_object_store.rs

1// Copyright 2023 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::sync::Arc;
16use std::time::Duration;
17
18use bytes::{Bytes, BytesMut};
19use fail::fail_point;
20use futures::StreamExt;
21use opendal::layers::{RetryLayer, TimeoutLayer};
22use opendal::raw::BoxedStaticFuture;
23use opendal::services::Memory;
24use opendal::{Execute, Executor, Operator, Writer};
25use risingwave_common::config::ObjectStoreConfig;
26use risingwave_common::range::RangeBoundsExt;
27use thiserror_ext::AsReport;
28
29use crate::object::object_metrics::ObjectStoreMetrics;
30use crate::object::{
31    MonitoredObjectStore, ObjectDataStream, ObjectError, ObjectMetadata, ObjectMetadataIter,
32    ObjectRangeBounds, ObjectResult, ObjectStore, OperationType, StreamingUploader, prefix,
33};
34
35/// Opendal object storage.
36#[derive(Clone)]
37pub struct OpendalObjectStore {
38    pub(crate) op: Operator,
39    pub(crate) media_type: MediaType,
40    pub(crate) config: Arc<ObjectStoreConfig>,
41    pub(crate) metrics: Arc<ObjectStoreMetrics>,
42}
43
44#[derive(Clone)]
45pub enum MediaType {
46    Memory,
47    Hdfs,
48    Gcs,
49    Minio,
50    S3,
51    Obs,
52    Oss,
53    Webhdfs,
54    Azblob,
55    Fs,
56}
57
58impl MediaType {
59    pub fn as_str(&self) -> &'static str {
60        match self {
61            MediaType::Memory => "Memory",
62            MediaType::Hdfs => "Hdfs",
63            MediaType::Gcs => "Gcs",
64            MediaType::Minio => "Minio",
65            MediaType::S3 => "S3",
66            MediaType::Obs => "Obs",
67            MediaType::Oss => "Oss",
68            MediaType::Webhdfs => "Webhdfs",
69            MediaType::Azblob => "Azblob",
70            MediaType::Fs => "Fs",
71        }
72    }
73}
74
75impl OpendalObjectStore {
76    /// create opendal memory engine, used for unit tests.
77    pub fn test_new_memory_engine() -> ObjectResult<Self> {
78        // Create memory backend builder.
79        let builder = Memory::default();
80        let op: Operator = Operator::new(builder)?;
81
82        Ok(Self {
83            op,
84            media_type: MediaType::Memory,
85            config: Arc::new(ObjectStoreConfig::default()),
86            metrics: Arc::new(ObjectStoreMetrics::unused()),
87        })
88    }
89}
90
91#[inline]
92fn timestamp_to_secs(t: opendal::raw::Timestamp) -> f64 {
93    t.into_inner().as_second() as f64
94}
95
96#[async_trait::async_trait]
97impl ObjectStore for OpendalObjectStore {
98    type StreamingUploader = OpendalStreamingUploader;
99
100    fn get_object_prefix(&self, obj_id: u64, use_new_object_prefix_strategy: bool) -> String {
101        match self.media_type {
102            MediaType::S3 => prefix::s3::get_object_prefix(obj_id),
103            MediaType::Minio => prefix::s3::get_object_prefix(obj_id),
104            MediaType::Memory => String::default(),
105            MediaType::Hdfs
106            | MediaType::Gcs
107            | MediaType::Obs
108            | MediaType::Oss
109            | MediaType::Webhdfs
110            | MediaType::Azblob
111            | MediaType::Fs => {
112                prefix::opendal_engine::get_object_prefix(obj_id, use_new_object_prefix_strategy)
113            }
114        }
115    }
116
117    async fn upload(&self, path: &str, obj: Bytes) -> ObjectResult<()> {
118        if obj.is_empty() {
119            Err(ObjectError::internal("upload empty object"))
120        } else {
121            self.op.write(path, obj).await?;
122            Ok(())
123        }
124    }
125
126    async fn streaming_upload(&self, path: &str) -> ObjectResult<Self::StreamingUploader> {
127        if self.op.info().capability().write_can_multi {
128            Ok(OpendalStreamingUploader::Native(Box::new(
129                OpendalNativeStreamingUploader::new(
130                    self.op.clone(),
131                    path.to_owned(),
132                    self.config.clone(),
133                    self.metrics.clone(),
134                    self.store_media_type(),
135                )
136                .await?,
137            )))
138        } else {
139            Ok(OpendalStreamingUploader::Buffered(
140                OpendalBufferedStreamingUploader::new(
141                    self.clone()
142                        .monitored(self.metrics.clone(), self.config.clone()),
143                    path.to_owned(),
144                ),
145            ))
146        }
147    }
148
149    async fn read(&self, path: &str, range: impl ObjectRangeBounds) -> ObjectResult<Bytes> {
150        let data = if range.is_full() {
151            self.op.read(path).await?
152        } else {
153            self.op
154                .read_with(path)
155                .range(range.map(|v| *v as u64))
156                .await?
157        };
158
159        if let Some(len) = range.len()
160            && len != data.len()
161        {
162            return Err(ObjectError::internal(format!(
163                "mismatched size: expected {}, found {} when reading {} at {:?}",
164                len,
165                data.len(),
166                path,
167                range,
168            )));
169        }
170
171        Ok(data.to_bytes())
172    }
173
174    /// Returns a stream reading the object specified in `path`. If given, the stream starts at the
175    /// byte with index `start_pos` (0-based). As far as possible, the stream only loads the amount
176    /// of data into memory that is read from the stream.
177    async fn streaming_read(
178        &self,
179        path: &str,
180        range: impl ObjectRangeBounds,
181    ) -> ObjectResult<ObjectDataStream> {
182        fail_point!("opendal_streaming_read_err", |_| Err(
183            ObjectError::internal("opendal streaming read error")
184        ));
185        let range = range.map(|v| *v as u64);
186
187        // The layer specified first will be executed first.
188        // `TimeoutLayer` must be specified before `RetryLayer`.
189        // Otherwise, it will lead to bad state inside OpenDAL and panic.
190        // See https://docs.rs/opendal/latest/opendal/layers/struct.RetryLayer.html#panics
191        let reader = self
192            .op
193            .clone()
194            .layer(TimeoutLayer::new().with_io_timeout(Duration::from_millis(
195                self.config.retry.streaming_read_attempt_timeout_ms,
196            )))
197            .layer(
198                RetryLayer::new()
199                    .with_min_delay(Duration::from_millis(
200                        self.config.retry.req_backoff_interval_ms,
201                    ))
202                    .with_max_delay(Duration::from_millis(
203                        self.config.retry.req_backoff_max_delay_ms,
204                    ))
205                    .with_max_times(self.config.retry.streaming_read_retry_attempts)
206                    .with_factor(self.config.retry.req_backoff_factor as f32)
207                    .with_jitter(),
208            )
209            .reader_with(path)
210            .await?;
211        let stream = reader.into_bytes_stream(range).await?.map(|item| {
212            item.map(|b| Bytes::copy_from_slice(b.as_ref()))
213                .map_err(|e| {
214                    ObjectError::internal(format!("reader into_stream fail {}", e.as_report()))
215                })
216        });
217
218        Ok(Box::pin(stream))
219    }
220
221    async fn metadata(&self, path: &str) -> ObjectResult<ObjectMetadata> {
222        let opendal_metadata = self.op.stat(path).await?;
223        let key = path.to_owned();
224        let last_modified = match opendal_metadata.last_modified() {
225            Some(t) => timestamp_to_secs(t),
226            None => 0_f64,
227        };
228
229        let total_size = opendal_metadata.content_length() as usize;
230        let metadata = ObjectMetadata {
231            key,
232            last_modified,
233            total_size,
234        };
235        Ok(metadata)
236    }
237
238    async fn delete(&self, path: &str) -> ObjectResult<()> {
239        self.op.delete(path).await?;
240        Ok(())
241    }
242
243    /// Deletes the objects with the given paths permanently from the storage. If an object
244    /// specified in the request is not found, it will be considered as successfully deleted.
245    async fn delete_objects(&self, paths: &[String]) -> ObjectResult<()> {
246        self.op.delete_iter(paths.to_vec()).await?;
247        Ok(())
248    }
249
250    async fn list(
251        &self,
252        prefix: &str,
253        start_after: Option<String>,
254        limit: Option<usize>,
255    ) -> ObjectResult<ObjectMetadataIter> {
256        let mut object_lister = self.op.lister_with(prefix).recursive(true);
257        if let Some(start_after) = start_after {
258            object_lister = object_lister.start_after(&start_after);
259        }
260        let object_lister = object_lister.await?;
261
262        let op = self.op.clone();
263        let stream = object_lister.then(move |object| {
264            let op = op.clone();
265            async move { Self::listed_object_metadata(&op, object?).await }
266        });
267
268        Ok(stream.take(limit.unwrap_or(usize::MAX)).boxed())
269    }
270
271    fn store_media_type(&self) -> &'static str {
272        self.media_type.as_str()
273    }
274}
275
276impl OpendalObjectStore {
277    async fn listed_object_metadata(
278        op: &Operator,
279        object: opendal::Entry,
280    ) -> ObjectResult<ObjectMetadata> {
281        let key = object.path().to_owned();
282        // OpenDAL 0.55 removed list metadata capability flags and reports unknown sizes as 0.
283        let meta = object.metadata();
284        let mut last_modified = meta.last_modified().map(timestamp_to_secs);
285        let mut total_size = meta.content_length() as usize;
286        // FS listing includes the actual size, even for empty files. Avoid a redundant stat
287        // that can fail if an atomic upload renames the listed temporary file.
288        if last_modified.is_none()
289            || (total_size == 0 && op.info().scheme() != opendal::services::FS_SCHEME)
290        {
291            // Propagate stat failures; treating one as EOF can make recovery trust a partial scan.
292            let stat_meta = op.stat(&key).await?;
293            last_modified = stat_meta.last_modified().map(timestamp_to_secs);
294            total_size = stat_meta.content_length() as usize;
295        }
296        Ok(ObjectMetadata {
297            key,
298            last_modified: last_modified.unwrap_or(0_f64),
299            total_size,
300        })
301    }
302
303    pub async fn copy(&self, from_path: &str, to_path: &str) -> ObjectResult<()> {
304        self.op.copy(from_path, to_path).await?;
305        Ok(())
306    }
307}
308
309struct OpendalStreamingUploaderExecute {
310    /// To record metrics for uploading part.
311    metrics: Arc<ObjectStoreMetrics>,
312    media_type: &'static str,
313}
314
315impl OpendalStreamingUploaderExecute {
316    const STREAMING_UPLOAD_TYPE: OperationType = OperationType::StreamingUpload;
317
318    fn new(metrics: Arc<ObjectStoreMetrics>, media_type: &'static str) -> Self {
319        Self {
320            metrics,
321            media_type,
322        }
323    }
324}
325
326impl Execute for OpendalStreamingUploaderExecute {
327    fn execute(&self, f: BoxedStaticFuture<()>) {
328        let operation_type_str = Self::STREAMING_UPLOAD_TYPE.as_str();
329        let media_type = self.media_type;
330
331        let metrics = self.metrics.clone();
332        let _handle = tokio::spawn(async move {
333            let _timer = metrics
334                .operation_latency
335                .with_label_values(&[media_type, operation_type_str])
336                .start_timer();
337
338            f.await
339        });
340    }
341}
342
343pub enum OpendalStreamingUploader {
344    Native(Box<OpendalNativeStreamingUploader>),
345    Buffered(OpendalBufferedStreamingUploader),
346}
347
348/// Store multiple parts in a map, and concatenate them on finish.
349pub struct OpendalNativeStreamingUploader {
350    writer: Writer,
351    /// Buffer for data. It will store at least `UPLOAD_BUFFER_SIZE` bytes of data before wrapping itself
352    /// into a stream and upload to object store as a part.
353    buf: Vec<Bytes>,
354    /// Length of the data that have not been uploaded to object store.
355    not_uploaded_len: usize,
356    /// Whether the writer is valid. The writer is invalid after abort/close.
357    is_valid: bool,
358
359    abort_on_err: bool,
360
361    upload_part_size: usize,
362}
363
364impl OpendalNativeStreamingUploader {
365    pub async fn new(
366        op: Operator,
367        path: String,
368        config: Arc<ObjectStoreConfig>,
369        metrics: Arc<ObjectStoreMetrics>,
370        media_type: &'static str,
371    ) -> ObjectResult<Self> {
372        let monitored_execute = OpendalStreamingUploaderExecute::new(metrics, media_type);
373        let executor = Executor::with(monitored_execute);
374        // Start from the base context so existing layers are replayed exactly once by
375        // `with_context`. Using the composed context here would double-wrap resources such as
376        // the HTTP concurrency semaphore.
377        let ctx = op.base_context().with_executor(executor);
378        let op = op.with_context(ctx);
379
380        let writer = op
381            .clone()
382            .layer(TimeoutLayer::new().with_io_timeout(Duration::from_millis(
383                config.retry.streaming_upload_attempt_timeout_ms,
384            )))
385            .layer(
386                RetryLayer::new()
387                    .with_min_delay(Duration::from_millis(config.retry.req_backoff_interval_ms))
388                    .with_max_delay(Duration::from_millis(config.retry.req_backoff_max_delay_ms))
389                    .with_max_times(config.retry.streaming_upload_retry_attempts)
390                    .with_factor(config.retry.req_backoff_factor as f32)
391                    .with_jitter(),
392            )
393            .writer_with(&path)
394            .concurrent(config.opendal_upload_concurrency)
395            .await?;
396        Ok(Self {
397            writer,
398            buf: vec![],
399            not_uploaded_len: 0,
400            is_valid: true,
401            abort_on_err: config.opendal_writer_abort_on_err,
402            upload_part_size: config.upload_part_size,
403        })
404    }
405
406    async fn flush(&mut self) -> ObjectResult<()> {
407        let data: Vec<Bytes> = self.buf.drain(..).collect();
408        debug_assert_eq!(
409            data.iter().map(|b| b.len()).sum::<usize>(),
410            self.not_uploaded_len
411        );
412        if let Err(err) = self.writer.write(data).await {
413            self.is_valid = false;
414            if self.abort_on_err {
415                self.writer.abort().await?;
416            }
417            return Err(err.into());
418        }
419        self.not_uploaded_len = 0;
420        Ok(())
421    }
422}
423
424impl StreamingUploader for OpendalNativeStreamingUploader {
425    async fn write_bytes(&mut self, data: Bytes) -> ObjectResult<()> {
426        assert!(self.is_valid);
427        self.not_uploaded_len += data.len();
428        self.buf.push(data);
429        if self.not_uploaded_len >= self.upload_part_size {
430            self.flush().await?;
431        }
432        Ok(())
433    }
434
435    async fn finish(mut self) -> ObjectResult<()> {
436        assert!(self.is_valid);
437        if self.not_uploaded_len > 0 {
438            self.flush().await?;
439        }
440
441        assert!(self.buf.is_empty());
442        assert_eq!(self.not_uploaded_len, 0);
443
444        self.is_valid = false;
445        match self.writer.close().await {
446            Ok(_) => (),
447            Err(err) => {
448                if self.abort_on_err {
449                    self.writer.abort().await?;
450                }
451                return Err(err.into());
452            }
453        };
454
455        Ok(())
456    }
457
458    // Not absolutely accurate. Some bytes may be in the infight request.
459    fn get_memory_usage(&self) -> u64 {
460        self.not_uploaded_len as u64
461    }
462}
463
464pub struct OpendalBufferedStreamingUploader {
465    store: MonitoredObjectStore<OpendalObjectStore>,
466    path: String,
467    buf: BytesMut,
468}
469
470impl OpendalBufferedStreamingUploader {
471    fn new(store: MonitoredObjectStore<OpendalObjectStore>, path: String) -> Self {
472        Self {
473            store,
474            path,
475            buf: BytesMut::new(),
476        }
477    }
478}
479
480impl StreamingUploader for OpendalBufferedStreamingUploader {
481    async fn write_bytes(&mut self, data: Bytes) -> ObjectResult<()> {
482        self.buf.extend_from_slice(&data);
483        Ok(())
484    }
485
486    async fn finish(self) -> ObjectResult<()> {
487        self.store.upload(&self.path, self.buf.freeze()).await
488    }
489
490    fn get_memory_usage(&self) -> u64 {
491        self.buf.len() as u64
492    }
493}
494
495impl StreamingUploader for OpendalStreamingUploader {
496    async fn write_bytes(&mut self, data: Bytes) -> ObjectResult<()> {
497        match self {
498            Self::Native(uploader) => uploader.write_bytes(data).await,
499            Self::Buffered(uploader) => uploader.write_bytes(data).await,
500        }
501    }
502
503    async fn finish(self) -> ObjectResult<()> {
504        match self {
505            Self::Native(uploader) => (*uploader).finish().await,
506            Self::Buffered(uploader) => uploader.finish().await,
507        }
508    }
509
510    fn get_memory_usage(&self) -> u64 {
511        match self {
512            Self::Native(uploader) => uploader.get_memory_usage(),
513            Self::Buffered(uploader) => uploader.get_memory_usage(),
514        }
515    }
516}
517
518#[cfg(test)]
519mod tests {
520    use futures::TryStreamExt;
521
522    use super::*;
523
524    async fn list_all(prefix: &str, store: &OpendalObjectStore) -> Vec<ObjectMetadata> {
525        store
526            .list(prefix, None, None)
527            .await
528            .unwrap()
529            .try_collect::<Vec<_>>()
530            .await
531            .unwrap()
532    }
533
534    #[tokio::test]
535    async fn test_memory_upload() {
536        let block = Bytes::from("123456");
537        let store = OpendalObjectStore::test_new_memory_engine().unwrap();
538        store.upload("/abc", block).await.unwrap();
539
540        // No such object.
541        store.read("/ab", 0..3).await.unwrap_err();
542
543        let bytes = store.read("/abc", 4..6).await.unwrap();
544        assert_eq!(String::from_utf8(bytes.to_vec()).unwrap(), "56".to_owned());
545
546        store.delete("/abc").await.unwrap();
547
548        // No such object.
549        store.read("/abc", 0..3).await.unwrap_err();
550    }
551
552    #[tokio::test]
553    async fn test_memory_read_out_of_range_returns_error() {
554        let block = Bytes::from("123456");
555        let store = OpendalObjectStore::test_new_memory_engine().unwrap();
556        store.upload("/abc", block).await.unwrap();
557
558        // OpenDAL 0.58 reports an out-of-range read instead of panicking.
559        store.read("/abc", 4..44).await.unwrap_err();
560    }
561
562    #[tokio::test]
563    async fn test_memory_metadata() {
564        let block = Bytes::from("123456");
565        let path = "/abc".to_owned();
566        let obj_store = OpendalObjectStore::test_new_memory_engine().unwrap();
567        obj_store.upload("/abc", block).await.unwrap();
568
569        let err = obj_store.metadata("/not_exist").await.unwrap_err();
570        assert!(err.is_object_not_found_error());
571        let metadata = obj_store.metadata("/abc").await.unwrap();
572        assert_eq!(metadata.total_size, 6);
573        obj_store.delete(&path).await.unwrap();
574    }
575
576    #[tokio::test]
577    async fn test_buffered_streaming_upload() {
578        let store = OpendalObjectStore::test_new_memory_engine().unwrap();
579        let mut uploader = OpendalBufferedStreamingUploader::new(
580            store
581                .clone()
582                .monitored(store.metrics.clone(), store.config.clone()),
583            "/abc".to_owned(),
584        );
585
586        uploader.write_bytes(Bytes::from("123")).await.unwrap();
587        assert_eq!(uploader.get_memory_usage(), 3);
588        uploader.write_bytes(Bytes::from("456")).await.unwrap();
589        assert_eq!(uploader.get_memory_usage(), 6);
590        uploader.finish().await.unwrap();
591
592        let read_obj = store.read("/abc", ..).await.unwrap();
593        assert_eq!(read_obj, Bytes::from("123456"));
594    }
595
596    #[tokio::test]
597    async fn test_empty_buffered_streaming_upload() {
598        let store = OpendalObjectStore::test_new_memory_engine().unwrap();
599        let uploader = OpendalBufferedStreamingUploader::new(
600            store
601                .clone()
602                .monitored(store.metrics.clone(), store.config.clone()),
603            "/abc".to_owned(),
604        );
605
606        uploader.finish().await.unwrap_err();
607    }
608
609    #[cfg(not(madsim))] // OpenDAL FS uses real filesystem I/O.
610    #[tokio::test]
611    async fn test_listed_empty_fs_file_survives_rename() {
612        let dir = tempfile::tempdir().unwrap();
613        let op = Operator::new(opendal::services::Fs::default().root(dir.path().to_str().unwrap()))
614            .unwrap();
615        tokio::fs::write(dir.path().join("sst.tmp"), b"")
616            .await
617            .unwrap();
618        let mut objects = op.lister_with("").recursive(true).await.unwrap();
619        let object = objects.next().await.unwrap().unwrap();
620        assert_eq!(object.path(), "sst.tmp");
621        assert_eq!(object.metadata().content_length(), 0);
622        let last_modified = timestamp_to_secs(object.metadata().last_modified().unwrap());
623
624        // Model an atomic upload finishing after LIST has captured the temporary file.
625        tokio::fs::rename(dir.path().join("sst.tmp"), dir.path().join("sst"))
626            .await
627            .unwrap();
628        let metadata = OpendalObjectStore::listed_object_metadata(&op, object)
629            .await
630            .unwrap();
631        assert_eq!(metadata.key, "sst.tmp");
632        assert_eq!(metadata.total_size, 0);
633        assert_eq!(metadata.last_modified, last_modified);
634    }
635
636    #[tokio::test]
637    async fn test_listed_object_metadata_propagates_stat_failure() {
638        let store = OpendalObjectStore::test_new_memory_engine().unwrap();
639        store
640            .upload("sst", Bytes::from_static(b"data"))
641            .await
642            .unwrap();
643        // Capture the listed entry before deletion so the subsequent metadata lookup fails.
644        // Memory entries have no last_modified, so our adapter must stat even a listed file.
645        let mut objects = store.op.lister_with("").recursive(true).await.unwrap();
646        let object = objects.next().await.unwrap().unwrap();
647        store.delete("sst").await.unwrap();
648        let error = OpendalObjectStore::listed_object_metadata(&store.op, object)
649            .await
650            .unwrap_err();
651        assert!(error.is_object_not_found_error());
652    }
653
654    #[tokio::test]
655    async fn test_memory_delete_objects_and_list_object() {
656        let block1 = Bytes::from("123456");
657        let block2 = Bytes::from("987654");
658        let store = OpendalObjectStore::test_new_memory_engine().unwrap();
659        store.upload("abc", Bytes::from("123456")).await.unwrap();
660        store.upload("prefix/abc", block1).await.unwrap();
661
662        store.upload("prefix/xyz", block2).await.unwrap();
663
664        assert_eq!(list_all("", &store).await.len(), 3);
665        assert_eq!(list_all("prefix/", &store).await.len(), 2);
666        let str_list = [String::from("prefix/abc"), String::from("prefix/xyz")];
667
668        store.delete_objects(&str_list).await.unwrap();
669
670        assert!(store.read("prefix/abc/", ..).await.is_err());
671        assert!(store.read("prefix/xyz/", ..).await.is_err());
672        assert_eq!(list_all("", &store).await.len(), 1);
673        assert_eq!(list_all("prefix/", &store).await.len(), 0);
674    }
675}
risingwave_object_store/object/opendal_engine/opendal_object_store.rs

risingwave_object_store/object/opendal_engine/
opendal_object_store.rs
