icechunk_storage/
storage.rs

1//! Core storage trait and shared types.
2
3use chrono::{DateTime, Utc};
4use core::fmt;
5use futures::{
6    Stream, StreamExt as _, TryStreamExt as _,
7    stream::{self, BoxStream, FuturesOrdered},
8};
9use itertools::Itertools as _;
10use serde::{Deserialize, Serialize};
11use std::{
12    cmp::{max, min},
13    ffi::OsString,
14    fmt::Display,
15    iter,
16    num::{NonZeroU16, NonZeroU64},
17    ops::Range,
18    pin::Pin,
19    sync::{Arc, Mutex, OnceLock},
20};
21use tokio::io::AsyncBufRead;
22use tokio_util::io::StreamReader;
23use tracing::{instrument, warn};
24
25use async_trait::async_trait;
26use bytes::Bytes;
27use thiserror::Error;
28
29use crate::ICError;
30use crate::sealed;
31
32/// Storage operation error types.
33#[derive(Debug, Error)]
34pub enum StorageErrorKind {
35    #[error("object not found")]
36    ObjectNotFound,
37    #[error("bad object store prefix {0:?}")]
38    BadPrefix(OsString),
39    #[error("object store error {0}")]
40    ObjectStore(#[from] Box<dyn std::error::Error + Send + Sync + 'static>),
41    #[error("I/O error: {0}")]
42    IOError(#[from] std::io::Error),
43    #[error("storage configuration error: {0}")]
44    R2ConfigurationError(String),
45    #[error("error parsing URL: {url:?}")]
46    CannotParseUrl {
47        #[source]
48        cause: url::ParseError,
49        url: String,
50    },
51    #[error("Redirect Storage error: {0}")]
52    BadRedirect(String),
53    #[error("storage error: {0}")]
54    Other(String),
55}
56pub type StorageError = ICError<StorageErrorKind>;
57
58pub type StorageResult<A> = Result<A, StorageError>;
59
60pub fn obj_store_error_res<T>(
61    err: impl std::error::Error + Send + Sync + 'static,
62) -> StorageResult<T> {
63    Err(obj_store_error(err))
64}
65
66pub fn obj_store_error(
67    err: impl std::error::Error + Send + Sync + 'static,
68) -> StorageError {
69    StorageError::capture(StorageErrorKind::ObjectStore(Box::new(err)))
70}
71
72pub fn obj_not_found_res<T>() -> StorageResult<T> {
73    Err(StorageError::capture(StorageErrorKind::ObjectNotFound))
74}
75
76pub fn other_error(s: impl Into<String>) -> StorageError {
77    StorageError::capture(StorageErrorKind::Other(s.into()))
78}
79
80#[derive(Debug)]
81pub struct ListInfo<A> {
82    pub id: A,
83    pub created_at: DateTime<Utc>,
84    pub size_bytes: u64,
85}
86
87pub use icechunk_types::ETag;
88#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone, Default)]
89pub struct Generation(pub String);
90
91#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone)]
92pub struct VersionInfo {
93    pub etag: Option<ETag>,
94    pub generation: Option<Generation>,
95}
96
97impl VersionInfo {
98    pub fn for_creation() -> Self {
99        Self { etag: None, generation: None }
100    }
101
102    pub fn from_etag_only(etag: String) -> Self {
103        Self { etag: Some(ETag(etag)), generation: None }
104    }
105
106    pub fn is_create(&self) -> bool {
107        self.etag.is_none() && self.generation.is_none()
108    }
109
110    pub fn etag(&self) -> Option<&String> {
111        self.etag.as_ref().map(|e| &e.0)
112    }
113
114    pub fn generation(&self) -> Option<&String> {
115        self.generation.as_ref().map(|e| &e.0)
116    }
117}
118
119impl Display for VersionInfo {
120    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
121        match (&self.etag, &self.generation) {
122            (Some(etag), Some(generation)) => {
123                write!(f, "etag={}, generation={}", etag.0, generation.0)
124            }
125            (Some(etag), None) => write!(f, "etag={}", etag.0),
126            (None, Some(generation)) => write!(f, "generation={}", generation.0),
127            (None, None) => write!(f, "new"),
128        }
129    }
130}
131
132#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone, Copy, Default)]
133pub struct RetriesSettings {
134    pub max_tries: Option<NonZeroU16>,
135    pub initial_backoff_ms: Option<u32>,
136    pub max_backoff_ms: Option<u32>,
137}
138
139impl RetriesSettings {
140    pub fn max_tries(&self) -> NonZeroU16 {
141        self.max_tries.unwrap_or_else(|| NonZeroU16::new(10).unwrap_or(NonZeroU16::MIN))
142    }
143
144    pub fn initial_backoff_ms(&self) -> u32 {
145        self.initial_backoff_ms.unwrap_or(100)
146    }
147
148    pub fn max_backoff_ms(&self) -> u32 {
149        self.max_backoff_ms.unwrap_or(3 * 60 * 1000)
150    }
151
152    pub fn merge(&self, other: Self) -> Self {
153        Self {
154            max_tries: other.max_tries.or(self.max_tries),
155            initial_backoff_ms: other.initial_backoff_ms.or(self.initial_backoff_ms),
156            max_backoff_ms: other.max_backoff_ms.or(self.max_backoff_ms),
157        }
158    }
159}
160
161#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone, Copy, Default)]
162pub struct TimeoutSettings {
163    pub connect_timeout_ms: Option<u32>,
164    pub read_timeout_ms: Option<u32>,
165    pub operation_timeout_ms: Option<u32>,
166    pub operation_attempt_timeout_ms: Option<u32>,
167}
168
169impl TimeoutSettings {
170    pub fn merge(&self, other: Self) -> Self {
171        Self {
172            connect_timeout_ms: other.connect_timeout_ms.or(self.connect_timeout_ms),
173            read_timeout_ms: other.read_timeout_ms.or(self.read_timeout_ms),
174            operation_timeout_ms: other
175                .operation_timeout_ms
176                .or(self.operation_timeout_ms),
177            operation_attempt_timeout_ms: other
178                .operation_attempt_timeout_ms
179                .or(self.operation_attempt_timeout_ms),
180        }
181    }
182}
183
184#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone, Copy, Default)]
185pub struct ConcurrencySettings {
186    pub max_concurrent_requests_for_object: Option<NonZeroU16>,
187    pub ideal_concurrent_request_size: Option<NonZeroU64>,
188}
189
190impl ConcurrencySettings {
191    // AWS recommendations: https://docs.aws.amazon.com/whitepapers/latest/s3-optimizing-performance-best-practices/horizontal-scaling-and-request-parallelization-for-high-throughput.html
192    // 8-16 MB requests
193    // 85-90 MB/s per request
194    // these numbers would saturate a 12.5 Gbps network
195
196    pub fn max_concurrent_requests_for_object(&self) -> NonZeroU16 {
197        self.max_concurrent_requests_for_object
198            .unwrap_or_else(|| NonZeroU16::new(18).unwrap_or(NonZeroU16::MIN))
199    }
200    pub fn ideal_concurrent_request_size(&self) -> NonZeroU64 {
201        self.ideal_concurrent_request_size.unwrap_or_else(|| {
202            NonZeroU64::new(12 * 1024 * 1024).unwrap_or(NonZeroU64::MIN)
203        })
204    }
205
206    pub fn merge(&self, other: Self) -> Self {
207        Self {
208            max_concurrent_requests_for_object: other
209                .max_concurrent_requests_for_object
210                .or(self.max_concurrent_requests_for_object),
211            ideal_concurrent_request_size: other
212                .ideal_concurrent_request_size
213                .or(self.ideal_concurrent_request_size),
214        }
215    }
216}
217
218/// Configuration for storage operations (retries, concurrency, storage classes).
219#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone, Default)]
220pub struct Settings {
221    #[serde(default)]
222    pub concurrency: Option<ConcurrencySettings>,
223
224    #[serde(default)]
225    pub retries: Option<RetriesSettings>,
226
227    #[serde(default)]
228    pub timeouts: Option<TimeoutSettings>,
229
230    #[serde(default)]
231    pub unsafe_use_conditional_update: Option<bool>,
232
233    #[serde(default)]
234    pub unsafe_use_conditional_create: Option<bool>,
235
236    #[serde(default)]
237    pub unsafe_use_metadata: Option<bool>,
238
239    #[serde(default)]
240    pub storage_class: Option<String>,
241
242    #[serde(default)]
243    pub metadata_storage_class: Option<String>,
244
245    #[serde(default)]
246    pub chunks_storage_class: Option<String>,
247
248    #[serde(default)]
249    pub minimum_size_for_multipart_upload: Option<u64>,
250}
251
252static DEFAULT_CONCURRENCY: OnceLock<ConcurrencySettings> = OnceLock::new();
253static DEFAULT_RETRIES: OnceLock<RetriesSettings> = OnceLock::new();
254
255impl Settings {
256    pub fn concurrency(&self) -> &ConcurrencySettings {
257        self.concurrency
258            .as_ref()
259            .unwrap_or_else(|| DEFAULT_CONCURRENCY.get_or_init(Default::default))
260    }
261
262    pub fn retries(&self) -> &RetriesSettings {
263        self.retries
264            .as_ref()
265            .unwrap_or_else(|| DEFAULT_RETRIES.get_or_init(Default::default))
266    }
267
268    pub fn timeouts(&self) -> Option<&TimeoutSettings> {
269        self.timeouts.as_ref()
270    }
271
272    pub fn unsafe_use_conditional_create(&self) -> bool {
273        self.unsafe_use_conditional_create.unwrap_or(true)
274    }
275
276    pub fn unsafe_use_conditional_update(&self) -> bool {
277        self.unsafe_use_conditional_update.unwrap_or(true)
278    }
279
280    pub fn unsafe_use_metadata(&self) -> bool {
281        self.unsafe_use_metadata.unwrap_or(true)
282    }
283
284    pub fn metadata_storage_class(&self) -> Option<&String> {
285        self.metadata_storage_class.as_ref().or(self.storage_class.as_ref())
286    }
287
288    pub fn storage_class(&self) -> Option<&String> {
289        self.storage_class.as_ref()
290    }
291
292    pub fn chunks_storage_class(&self) -> Option<&String> {
293        self.chunks_storage_class.as_ref().or(self.storage_class.as_ref())
294    }
295
296    pub fn minimum_size_for_multipart_upload(&self) -> u64 {
297        // per AWS  recommendation: 100 MB
298        self.minimum_size_for_multipart_upload.unwrap_or(100 * 1024 * 1024)
299    }
300
301    pub fn merge(&self, other: Self) -> Self {
302        Self {
303            concurrency: match (&self.concurrency, other.concurrency) {
304                (None, None) => None,
305                (None, Some(c)) => Some(c),
306                (Some(c), None) => Some(*c),
307                (Some(mine), Some(theirs)) => Some(mine.merge(theirs)),
308            },
309            retries: match (&self.retries, other.retries) {
310                (None, None) => None,
311                (None, Some(c)) => Some(c),
312                (Some(c), None) => Some(*c),
313                (Some(mine), Some(theirs)) => Some(mine.merge(theirs)),
314            },
315            timeouts: match (&self.timeouts, other.timeouts) {
316                (None, None) => None,
317                (None, Some(c)) => Some(c),
318                (Some(c), None) => Some(*c),
319                (Some(mine), Some(theirs)) => Some(mine.merge(theirs)),
320            },
321            unsafe_use_conditional_create: match (
322                &self.unsafe_use_conditional_create,
323                other.unsafe_use_conditional_create,
324            ) {
325                (None, None) => None,
326                (None, Some(c)) => Some(c),
327                (Some(c), None) => Some(*c),
328                (Some(_), Some(theirs)) => Some(theirs),
329            },
330            unsafe_use_conditional_update: match (
331                &self.unsafe_use_conditional_update,
332                other.unsafe_use_conditional_update,
333            ) {
334                (None, None) => None,
335                (None, Some(c)) => Some(c),
336                (Some(c), None) => Some(*c),
337                (Some(_), Some(theirs)) => Some(theirs),
338            },
339            unsafe_use_metadata: match (
340                &self.unsafe_use_metadata,
341                other.unsafe_use_metadata,
342            ) {
343                (None, None) => None,
344                (None, Some(c)) => Some(c),
345                (Some(c), None) => Some(*c),
346                (Some(_), Some(theirs)) => Some(theirs),
347            },
348            storage_class: match (&self.storage_class, other.storage_class) {
349                (None, None) => None,
350                (None, Some(c)) => Some(c),
351                (Some(c), None) => Some(c.clone()),
352                (Some(_), Some(theirs)) => Some(theirs),
353            },
354            metadata_storage_class: match (
355                &self.metadata_storage_class,
356                other.metadata_storage_class,
357            ) {
358                (None, None) => None,
359                (None, Some(c)) => Some(c),
360                (Some(c), None) => Some(c.clone()),
361                (Some(_), Some(theirs)) => Some(theirs),
362            },
363            chunks_storage_class: match (
364                &self.chunks_storage_class,
365                other.chunks_storage_class,
366            ) {
367                (None, None) => None,
368                (None, Some(c)) => Some(c),
369                (Some(c), None) => Some(c.clone()),
370                (Some(_), Some(theirs)) => Some(theirs),
371            },
372            minimum_size_for_multipart_upload: match (
373                &self.minimum_size_for_multipart_upload,
374                other.minimum_size_for_multipart_upload,
375            ) {
376                (None, None) => None,
377                (None, Some(c)) => Some(c),
378                (Some(c), None) => Some(*c),
379                (Some(_), Some(theirs)) => Some(theirs),
380            },
381        }
382    }
383}
384
385#[derive(Debug, Clone, PartialEq, Eq)]
386pub enum VersionedUpdateResult {
387    Updated { new_version: VersionInfo },
388    NotOnLatestVersion,
389}
390
391impl VersionedUpdateResult {
392    pub fn must_write(self) -> StorageResult<VersionInfo> {
393        match self {
394            VersionedUpdateResult::Updated { new_version } => Ok(new_version),
395            VersionedUpdateResult::NotOnLatestVersion => {
396                Err(StorageError::capture(StorageErrorKind::ObjectNotFound))
397            }
398        }
399    }
400}
401
402#[derive(Debug, Clone, PartialEq, Eq, Default)]
403pub struct DeleteObjectsResult {
404    pub deleted_objects: u64,
405    pub deleted_bytes: u64,
406}
407
408impl DeleteObjectsResult {
409    pub fn merge(&mut self, other: &Self) {
410        self.deleted_objects += other.deleted_objects;
411        self.deleted_bytes += other.deleted_bytes;
412    }
413}
414
415pub enum GetModifiedResult {
416    Modified { data: Pin<Box<dyn AsyncBufRead + Send>>, new_version: VersionInfo },
417    OnLatestVersion,
418}
419
420impl fmt::Debug for GetModifiedResult {
421    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
422        match self {
423            Self::Modified { new_version, .. } => {
424                f.debug_struct("Modified").field("new_version", new_version).finish()
425            }
426            Self::OnLatestVersion => write!(f, "OnLatestVersion"),
427        }
428    }
429}
430
431impl Display for GetModifiedResult {
432    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
433        match self {
434            Self::Modified { new_version, .. } => {
435                write!(f, "Modified(new_version={new_version})")
436            }
437            Self::OnLatestVersion => write!(f, "OnLatestVersion"),
438        }
439    }
440}
441
442/// Fetch and write the parquet files that represent the repository in object store
443///
444/// Different implementation can cache the files differently, or not at all.
445/// Structured metadata about a storage backend, for display/repr purposes.
446#[derive(Debug, Clone)]
447pub struct StorageInfo {
448    /// Human-readable backend type, e.g. "S3", "GCS", "in-memory".
449    pub backend_type: &'static str,
450    /// Key-value pairs of relevant configuration (bucket, prefix, path, etc.).
451    pub fields: Vec<(&'static str, String)>,
452}
453
454/// Implementations are free to assume files are never overwritten.
455#[async_trait]
456#[typetag::serde(tag = "type")]
457pub trait Storage: fmt::Debug + Display + sealed::Sealed + Sync + Send {
458    /// Return structured metadata about this storage backend for display/repr.
459    fn storage_info(&self) -> StorageInfo;
460
461    async fn default_settings(&self) -> StorageResult<Settings> {
462        Ok(Default::default())
463    }
464
465    async fn can_write(&self) -> StorageResult<bool>;
466
467    async fn get_object(
468        &self,
469        settings: &Settings,
470        path: &str,
471        range: Option<&Range<u64>>,
472    ) -> StorageResult<(Pin<Box<dyn AsyncBufRead + Send>>, VersionInfo)> {
473        if let Some(range) = range {
474            self.get_object_concurrently(settings, path, range).await
475        } else {
476            self.get_object_range_read(settings, path, range).await
477        }
478    }
479
480    async fn get_object_range_read(
481        &self,
482        settings: &Settings,
483        path: &str,
484        range: Option<&Range<u64>>,
485    ) -> StorageResult<(Pin<Box<dyn AsyncBufRead + Send>>, VersionInfo)> {
486        let (stream, version) = self.get_object_range(settings, path, range).await?;
487        let reader = StreamReader::new(stream.map_err(std::io::Error::other));
488        Ok((Box::pin(reader), version))
489    }
490
491    async fn get_object_range(
492        &self,
493        settings: &Settings,
494        path: &str,
495        range: Option<&Range<u64>>,
496    ) -> StorageResult<(
497        Pin<Box<dyn Stream<Item = Result<Bytes, StorageError>> + Send>>,
498        VersionInfo,
499    )>;
500
501    async fn put_object(
502        &self,
503        settings: &Settings,
504        path: &str,
505        bytes: Bytes,
506        content_type: Option<&str>,
507        metadata: Vec<(String, String)>,
508        previous_version: Option<&VersionInfo>,
509    ) -> StorageResult<VersionedUpdateResult>;
510
511    async fn copy_object(
512        &self,
513        settings: &Settings,
514        from: &str,
515        to: &str,
516        content_type: Option<&str>,
517        version: &VersionInfo,
518    ) -> StorageResult<VersionedUpdateResult>;
519
520    /// List objects in storage whose keys start with the given prefix.
521    ///
522    /// Returns a stream of [`ListInfo`] entries, each containing the object's key and size in bytes.
523    /// Pass an empty prefix to list all objects in the repository's storage root.
524    async fn list_objects<'a>(
525        &'a self,
526        settings: &Settings,
527        prefix: &str,
528    ) -> StorageResult<BoxStream<'a, StorageResult<ListInfo<String>>>>;
529
530    async fn delete_batch(
531        &self,
532        settings: &Settings,
533        prefix: &str,
534        batch: Vec<(String, u64)>,
535    ) -> StorageResult<DeleteObjectsResult>;
536
537    async fn get_object_last_modified(
538        &self,
539        path: &str,
540        settings: &Settings,
541    ) -> StorageResult<DateTime<Utc>>;
542
543    async fn get_object_conditional(
544        &self,
545        settings: &Settings,
546        path: &str,
547        previous_version: Option<&VersionInfo>,
548    ) -> StorageResult<GetModifiedResult>;
549
550    /// Delete a stream of objects, by their id string representations
551    /// Input stream includes sizes to get as result the total number of bytes deleted
552    #[instrument(skip(self, settings, ids))]
553    async fn delete_objects(
554        &self,
555        settings: &Settings,
556        prefix: &str,
557        ids: BoxStream<'_, (String, u64)>,
558    ) -> StorageResult<DeleteObjectsResult> {
559        let res = Arc::new(Mutex::new(DeleteObjectsResult::default()));
560        ids.chunks(1_000)
561            // FIXME: configurable concurrency
562            .for_each_concurrent(10, |batch| {
563                let res = Arc::clone(&res);
564                async move {
565                    let new_deletes = self
566                        .delete_batch(settings, prefix, batch)
567                        .await
568                        .unwrap_or_else(|_| {
569                            // FIXME: handle error instead of skipping
570                            warn!("ignoring error in Storage::delete_batch");
571                            Default::default()
572                        });
573                    #[expect(clippy::expect_used)]
574                    res.lock().expect("Bug in delete objects").merge(&new_deletes);
575                }
576            })
577            .await;
578        #[expect(clippy::expect_used)]
579        let res = res.lock().expect("Bug in delete objects");
580        Ok(res.clone())
581    }
582
583    async fn root_is_clean(&self, settings: &Settings) -> StorageResult<bool> {
584        match self.list_objects(settings, "").await?.next().await {
585            None => Ok(true),
586            Some(Ok(_)) => Ok(false),
587            Some(Err(err)) => Err(err),
588        }
589    }
590
591    async fn get_object_concurrently_multiple(
592        &self,
593        settings: &Settings,
594        key: &str,
595        parts: Vec<Range<u64>>,
596    ) -> StorageResult<(Pin<Box<dyn AsyncBufRead + Send>>, VersionInfo)> {
597        let settings2 = settings.clone();
598        let key2 = key.to_string();
599        let results = parts
600            .into_iter()
601            .map(move |range| {
602                let key = key2.clone();
603                let settings = settings2.clone();
604                async move {
605                    let (stream, version) = self
606                        .get_object_range(&settings, key.as_ref(), Some(&range))
607                        .await?;
608                    let all_bytes: Vec<_> = stream.try_collect().await?;
609                    Ok::<_, StorageError>((all_bytes, version))
610                }
611            })
612            .collect::<FuturesOrdered<_>>();
613
614        let results = results.peekable();
615        tokio::pin!(results);
616        let version = match results.as_mut().peek().await {
617            Some(Ok((_, version))) => version.clone(),
618            _ => VersionInfo::for_creation(),
619        };
620        let all_bytes = results
621            .map_ok(|(all_bytes, _)| stream::iter(all_bytes).map(Ok::<_, StorageError>))
622            .try_flatten()
623            .try_collect::<Vec<_>>()
624            .await?;
625
626        let res = StreamReader::new(stream::iter(all_bytes).map(Ok::<_, std::io::Error>));
627        Ok((Box::pin(res), version))
628    }
629
630    async fn get_object_concurrently(
631        &self,
632        settings: &Settings,
633        key: &str,
634        range: &Range<u64>,
635    ) -> StorageResult<(Pin<Box<dyn AsyncBufRead + Send>>, VersionInfo)> {
636        let parts = split_in_multiple_requests(
637            range,
638            settings.concurrency().ideal_concurrent_request_size().get(),
639            settings.concurrency().max_concurrent_requests_for_object().get(),
640        )
641        .collect::<Vec<_>>();
642
643        let res: (Pin<Box<dyn AsyncBufRead + Send>>, VersionInfo) = match parts.len() {
644            0 => (Box::pin(tokio::io::empty()), VersionInfo::for_creation()),
645            1 => self.get_object_range_read(settings, key, Some(range)).await?,
646            _ => self.get_object_concurrently_multiple(settings, key, parts).await?,
647        };
648        Ok(res)
649    }
650}
651
652/// Split an object request into multiple byte range requests
653///
654/// Returns tuples of Range for each request.
655///
656/// It generates requests that are as similar as possible in size, this means no more than 1 byte
657/// difference between the requests.
658///
659/// It tries to generate `ceil(size/ideal_req_size)` requests, but never exceeds `max_requests`.
660///
661/// `ideal_req_size` and `max_requests` must be > 0
662pub fn split_in_multiple_requests(
663    range: &Range<u64>,
664    ideal_req_size: u64,
665    max_requests: u16,
666) -> impl Iterator<Item = Range<u64>> + use<> {
667    let size = max(0, range.end - range.start);
668    // we do a ceiling division, rounding always up
669    let num_parts = size.div_ceil(ideal_req_size);
670    // no more than max_parts, so we limit
671    let num_parts = max(1, min(num_parts, max_requests as u64));
672
673    // we split the total size into request that are as similar as possible in size
674    // this means, we are going to have a few requests that are 1 byte larger
675    let big_parts = size % num_parts;
676    let small_parts_size = size / num_parts;
677    let big_parts_size = small_parts_size + 1;
678
679    iter::successors(Some((1, range.start..range.start)), move |(index, prev_range)| {
680        let size = if *index <= big_parts { big_parts_size } else { small_parts_size };
681        Some((index + 1, prev_range.end..prev_range.end + size))
682    })
683    .dropping(1)
684    .take(num_parts as usize)
685    .map(|(_, range)| range)
686}
687
688/// Split an object request into multiple byte range requests ensuring only the last request is
689/// smaller
690///
691/// Returns tuples of Range for each request.
692///
693/// It tries to generate `ceil(size/ideal_req_size)` requests, but never exceeds `max_requests`.
694///
695/// `ideal_req_size` and `max_requests` must be > 0
696pub fn split_in_multiple_equal_requests(
697    range: &Range<u64>,
698    ideal_req_size: u64,
699    max_requests: u16,
700) -> impl Iterator<Item = Range<u64>> + use<> {
701    let size = max(0, range.end - range.start);
702    // we do a ceiling division, rounding always up
703    let num_parts = size.div_ceil(ideal_req_size);
704    // no more than max_parts, so we limit
705    let num_parts = max(1, min(num_parts, max_requests as u64));
706
707    let big_parts = num_parts - 1;
708    let big_parts_size = size / max(1, big_parts);
709    let small_part_size = size - big_parts_size * big_parts;
710
711    iter::successors(Some((1, range.start..range.start)), move |(index, prev_range)| {
712        let size = if *index <= big_parts { big_parts_size } else { small_part_size };
713        Some((index + 1, prev_range.end..prev_range.end + size))
714    })
715    .dropping(1)
716    .take(num_parts as usize)
717    .map(|(_, range)| range)
718}
719
720pub fn strip_quotes(s: &str) -> &str {
721    s.strip_prefix('"').and_then(|s| s.strip_suffix('"')).unwrap_or(s)
722}
icechunk_storage/storage.rs

icechunk_storage/
storage.rs
