iceberg/io/
file_io.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::collections::HashMap;
19use std::ops::Range;
20use std::sync::{Arc, OnceLock};
21
22use bytes::Bytes;
23use futures::{Stream, StreamExt, stream};
24
25use super::storage::{
26    LocalFsStorageFactory, MemoryStorageFactory, Storage, StorageConfig, StorageFactory,
27};
28use crate::Result;
29
30/// FileIO implementation, used to manipulate files in underlying storage.
31///
32/// FileIO wraps a `dyn Storage` with lazy initialization via `StorageFactory`.
33/// The storage is created on first use and cached for subsequent operations.
34///
35/// # Note
36///
37/// All paths passed to `FileIO` must be absolute paths starting with the scheme string
38/// appropriate for the storage backend being used.
39///
40/// This crate provides native support for local filesystem (`file://`) and
41/// memory (`memory://`) storage. For extensive storage backend support (S3, GCS,
42/// OSS, Azure, etc.), use the
43/// [`iceberg-storage-opendal`](https://crates.io/crates/iceberg-storage-opendal) crate.
44///
45/// # Example
46///
47/// ```rust,ignore
48/// use iceberg::io::{FileIO, FileIOBuilder};
49/// use iceberg::io::{LocalFsStorageFactory, MemoryStorageFactory};
50/// use std::sync::Arc;
51///
52/// // Create FileIO with memory storage for testing
53/// let file_io = FileIO::new_with_memory();
54///
55/// // Create FileIO with local filesystem storage
56/// let file_io = FileIO::new_with_fs();
57///
58/// // Create FileIO with custom factory
59/// let file_io = FileIOBuilder::new(Arc::new(LocalFsStorageFactory))
60///     .with_prop("key", "value")
61///     .build();
62/// ```
63#[derive(Clone, Debug)]
64pub struct FileIO {
65    /// Storage configuration containing properties
66    config: StorageConfig,
67    /// Factory for creating storage instances
68    factory: Arc<dyn StorageFactory>,
69    /// Cached storage instance (lazily initialized)
70    storage: Arc<OnceLock<Arc<dyn Storage>>>,
71    /// Per-prefix storages (longest prefix first) for tables that vend distinct
72    /// credentials per location prefix. Paths matching none use `storage` above.
73    prefixed: Arc<Vec<PrefixedStorage>>,
74}
75
76/// A storage scoped to a location `prefix`, lazily built from its own config.
77#[derive(Debug)]
78struct PrefixedStorage {
79    prefix: String,
80    config: StorageConfig,
81    storage: OnceLock<Arc<dyn Storage>>,
82}
83
84impl FileIO {
85    /// Create a new FileIO backed by in-memory storage.
86    ///
87    /// This is useful for testing scenarios where persistent storage is not needed.
88    pub fn new_with_memory() -> Self {
89        Self {
90            config: StorageConfig::new(),
91            factory: Arc::new(MemoryStorageFactory),
92            storage: Arc::new(OnceLock::new()),
93            prefixed: Arc::new(Vec::new()),
94        }
95    }
96
97    /// Create a new FileIO backed by local filesystem storage.
98    ///
99    /// This is useful for local development and testing with real files.
100    pub fn new_with_fs() -> Self {
101        Self {
102            config: StorageConfig::new(),
103            factory: Arc::new(LocalFsStorageFactory),
104            storage: Arc::new(OnceLock::new()),
105            prefixed: Arc::new(Vec::new()),
106        }
107    }
108
109    /// Get the storage configuration.
110    pub fn config(&self) -> &StorageConfig {
111        &self.config
112    }
113
114    /// Get or create the storage for `path`, routing to the longest-matching
115    /// prefix storage if any, else the default. Built once, then cached.
116    fn get_storage(&self, path: &str) -> Result<Arc<dyn Storage>> {
117        // `prefixed` is sorted longest-first, so the first match is most specific.
118        // Selection is by longest matching string prefix, per the Iceberg REST
119        // spec's storage-credentials semantics (and Java's `S3FileIO`).
120        for ps in self.prefixed.iter() {
121            if path.starts_with(&ps.prefix) {
122                return Self::get_or_build(&ps.storage, &self.factory, &ps.config);
123            }
124        }
125        Self::get_or_build(&self.storage, &self.factory, &self.config)
126    }
127
128    /// Get a cached storage from `cell`, building it from `config` on first use.
129    fn get_or_build(
130        cell: &OnceLock<Arc<dyn Storage>>,
131        factory: &Arc<dyn StorageFactory>,
132        config: &StorageConfig,
133    ) -> Result<Arc<dyn Storage>> {
134        if let Some(storage) = cell.get() {
135            return Ok(storage.clone());
136        }
137        let storage = factory.build(config)?;
138        // Another thread might have set it first; keep whatever ends up in the cell.
139        let _ = cell.set(storage);
140        Ok(cell.get().unwrap().clone())
141    }
142
143    /// Deletes file.
144    ///
145    /// # Arguments
146    ///
147    /// * path: It should be *absolute* path starting with scheme string used to construct [`FileIO`].
148    pub async fn delete(&self, path: impl AsRef<str>) -> Result<()> {
149        self.get_storage(path.as_ref())?.delete(path.as_ref()).await
150    }
151
152    /// Remove the path and all nested dirs and files recursively.
153    ///
154    /// # Arguments
155    ///
156    /// * path: It should be *absolute* path starting with scheme string used to construct [`FileIO`].
157    ///
158    /// # Behavior
159    ///
160    /// - If the path is a file or not exist, this function will be no-op.
161    /// - If the path is a empty directory, this function will remove the directory itself.
162    /// - If the path is a non-empty directory, this function will remove the directory and all nested files and directories.
163    pub async fn delete_prefix(&self, path: impl AsRef<str>) -> Result<()> {
164        self.get_storage(path.as_ref())?
165            .delete_prefix(path.as_ref())
166            .await
167    }
168
169    /// Delete multiple files from a stream of paths.
170    ///
171    /// # Arguments
172    ///
173    /// * paths: A stream of absolute paths starting with the scheme string used to construct [`FileIO`].
174    pub async fn delete_stream(
175        &self,
176        paths: impl Stream<Item = String> + Send + 'static,
177    ) -> Result<()> {
178        // No per-prefix storages: delete the whole batch on the default storage.
179        if self.prefixed.is_empty() {
180            return self.get_storage("")?.delete_stream(paths.boxed()).await;
181        }
182
183        // Route by prefix, flushing bounded batches as we iterate so memory stays
184        // bounded on large streams (like Java's `S3FileIO.deleteFiles`).
185        const DELETE_BATCH_SIZE: usize = 1000;
186        let mut groups: HashMap<String, Vec<String>> = HashMap::new();
187        let mut paths = paths.boxed();
188        while let Some(path) = paths.next().await {
189            let key = self
190                .prefixed
191                .iter()
192                .find(|ps| path.starts_with(&ps.prefix))
193                .map(|ps| ps.prefix.clone())
194                .unwrap_or_default();
195            let buf = groups.entry(key).or_default();
196            buf.push(path);
197            if buf.len() >= DELETE_BATCH_SIZE {
198                let full = std::mem::take(buf);
199                self.get_storage(&full[0])?
200                    .delete_stream(stream::iter(full).boxed())
201                    .await?;
202            }
203        }
204
205        // Flush remainders.
206        for batch in groups.into_values() {
207            if batch.is_empty() {
208                continue;
209            }
210            self.get_storage(&batch[0])?
211                .delete_stream(stream::iter(batch).boxed())
212                .await?;
213        }
214        Ok(())
215    }
216
217    /// Check file exists.
218    ///
219    /// # Arguments
220    ///
221    /// * path: It should be *absolute* path starting with scheme string used to construct [`FileIO`].
222    pub async fn exists(&self, path: impl AsRef<str>) -> Result<bool> {
223        self.get_storage(path.as_ref())?.exists(path.as_ref()).await
224    }
225
226    /// Creates input file.
227    ///
228    /// # Arguments
229    ///
230    /// * path: It should be *absolute* path starting with scheme string used to construct [`FileIO`].
231    pub fn new_input(&self, path: impl AsRef<str>) -> Result<InputFile> {
232        self.get_storage(path.as_ref())?.new_input(path.as_ref())
233    }
234
235    /// Creates output file.
236    ///
237    /// # Arguments
238    ///
239    /// * path: It should be *absolute* path starting with scheme string used to construct [`FileIO`].
240    pub fn new_output(&self, path: impl AsRef<str>) -> Result<OutputFile> {
241        self.get_storage(path.as_ref())?.new_output(path.as_ref())
242    }
243}
244
245/// Builder for [`FileIO`].
246///
247/// The builder accepts an explicit `StorageFactory` and configuration properties.
248/// Storage is lazily initialized on first use.
249#[derive(Clone, Debug)]
250pub struct FileIOBuilder {
251    /// Factory for creating storage instances
252    factory: Arc<dyn StorageFactory>,
253    /// Storage configuration
254    config: StorageConfig,
255    /// Per-location-prefix configs (prefix, config).
256    prefixed: Vec<(String, StorageConfig)>,
257}
258
259impl FileIOBuilder {
260    /// Creates a new builder with the given storage factory.
261    pub fn new(factory: Arc<dyn StorageFactory>) -> Self {
262        Self {
263            factory,
264            config: StorageConfig::new(),
265            prefixed: Vec::new(),
266        }
267    }
268
269    /// Add a configuration property.
270    pub fn with_prop(mut self, key: impl ToString, value: impl ToString) -> Self {
271        self.config = self.config.with_prop(key.to_string(), value.to_string());
272        self
273    }
274
275    /// Add multiple configuration properties.
276    pub fn with_props(
277        mut self,
278        args: impl IntoIterator<Item = (impl ToString, impl ToString)>,
279    ) -> Self {
280        self.config = self
281            .config
282            .with_props(args.into_iter().map(|e| (e.0.to_string(), e.1.to_string())));
283        self
284    }
285
286    /// Add a per-prefix storage config. Paths starting with `prefix` (longest
287    /// match wins) use these props instead of the default config.
288    pub fn with_prefixed_props(
289        mut self,
290        prefix: impl Into<String>,
291        props: impl IntoIterator<Item = (impl ToString, impl ToString)>,
292    ) -> Self {
293        let config = StorageConfig::from_props(
294            props
295                .into_iter()
296                .map(|(k, v)| (k.to_string(), v.to_string()))
297                .collect(),
298        );
299        self.prefixed.push((prefix.into(), config));
300        self
301    }
302
303    /// Get the storage configuration.
304    pub fn config(&self) -> &StorageConfig {
305        &self.config
306    }
307
308    /// Builds [`FileIO`].
309    pub fn build(self) -> FileIO {
310        let mut prefixed: Vec<PrefixedStorage> = self
311            .prefixed
312            .into_iter()
313            .map(|(prefix, config)| PrefixedStorage {
314                prefix,
315                config,
316                storage: OnceLock::new(),
317            })
318            .collect();
319        // Longest prefix first so routing picks the most specific match.
320        prefixed.sort_by_key(|item| std::cmp::Reverse(item.prefix.len()));
321        FileIO {
322            config: self.config,
323            factory: self.factory,
324            storage: Arc::new(OnceLock::new()),
325            prefixed: Arc::new(prefixed),
326        }
327    }
328}
329
330/// The struct the represents the metadata of a file.
331///
332/// TODO: we can add last modified time, content type, etc. in the future.
333pub struct FileMetadata {
334    /// The size of the file.
335    pub size: u64,
336}
337
338/// Trait for reading file.
339///
340/// # TODO
341/// It's possible for us to remove the async_trait, but we need to figure
342/// out how to handle the object safety.
343#[async_trait::async_trait]
344pub trait FileRead: Send + Sync + Unpin + 'static {
345    /// Read file content with given range.
346    ///
347    /// TODO: we can support reading non-contiguous bytes in the future.
348    async fn read(&self, range: Range<u64>) -> crate::Result<Bytes>;
349}
350
351#[async_trait::async_trait]
352impl<T: AsRef<dyn FileRead> + Send + Sync + Unpin + 'static> FileRead for T {
353    async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
354        self.as_ref().read(range).await
355    }
356}
357
358/// Input file is used for reading from files.
359#[derive(Debug)]
360pub struct InputFile {
361    storage: Arc<dyn Storage>,
362    // Absolute path of file.
363    path: String,
364}
365
366impl InputFile {
367    /// Creates a new input file.
368    pub fn new(storage: Arc<dyn Storage>, path: String) -> Self {
369        Self { storage, path }
370    }
371
372    /// Absolute path to root uri.
373    pub fn location(&self) -> &str {
374        &self.path
375    }
376
377    /// Check if file exists.
378    pub async fn exists(&self) -> crate::Result<bool> {
379        self.storage.exists(&self.path).await
380    }
381
382    /// Fetch and returns metadata of file.
383    pub async fn metadata(&self) -> crate::Result<FileMetadata> {
384        self.storage.metadata(&self.path).await
385    }
386
387    /// Read and returns whole content of file.
388    ///
389    /// For continuous reading, use [`Self::reader`] instead.
390    pub async fn read(&self) -> crate::Result<Bytes> {
391        self.storage.read(&self.path).await
392    }
393
394    /// Creates [`FileRead`] for continuous reading.
395    ///
396    /// For one-time reading, use [`Self::read`] instead.
397    pub async fn reader(&self) -> crate::Result<Box<dyn FileRead>> {
398        self.storage.reader(&self.path).await
399    }
400}
401
402/// Trait for writing file.
403///
404/// # TODO
405///
406/// It's possible for us to remove the async_trait, but we need to figure
407/// out how to handle the object safety.
408#[async_trait::async_trait]
409pub trait FileWrite: Send + Unpin + 'static {
410    /// Write bytes to file.
411    ///
412    /// TODO: we can support writing non-contiguous bytes in the future.
413    async fn write(&mut self, bs: Bytes) -> crate::Result<()>;
414
415    /// Close file.
416    ///
417    /// Calling close on closed file will generate an error.
418    async fn close(&mut self) -> crate::Result<()>;
419}
420
421/// Output file is used for writing to files..
422#[derive(Debug)]
423pub struct OutputFile {
424    storage: Arc<dyn Storage>,
425    // Absolute path of file.
426    path: String,
427}
428
429impl OutputFile {
430    /// Creates a new output file.
431    pub fn new(storage: Arc<dyn Storage>, path: String) -> Self {
432        Self { storage, path }
433    }
434
435    /// Relative path to root uri.
436    pub fn location(&self) -> &str {
437        &self.path
438    }
439
440    /// Checks if file exists.
441    pub async fn exists(&self) -> Result<bool> {
442        self.storage.exists(&self.path).await
443    }
444
445    /// Deletes file.
446    ///
447    /// If the file does not exist, it will not return error.
448    pub async fn delete(&self) -> Result<()> {
449        self.storage.delete(&self.path).await
450    }
451
452    /// Converts into [`InputFile`].
453    pub fn to_input_file(self) -> InputFile {
454        InputFile {
455            storage: self.storage,
456            path: self.path,
457        }
458    }
459
460    /// Create a new output file with given bytes.
461    ///
462    /// # Notes
463    ///
464    /// Calling `write` will overwrite the file if it exists.
465    /// For continuous writing, use [`Self::writer`].
466    pub async fn write(&self, bs: Bytes) -> crate::Result<()> {
467        self.storage.write(&self.path, bs).await
468    }
469
470    /// Creates output file for continuous writing.
471    ///
472    /// # Notes
473    ///
474    /// For one-time writing, use [`Self::write`] instead.
475    pub async fn writer(&self) -> crate::Result<Box<dyn FileWrite>> {
476        self.storage.writer(&self.path).await
477    }
478}
479
480#[cfg(test)]
481mod tests {
482    use std::fs::{File, create_dir_all};
483    use std::io::Write;
484    use std::path::Path;
485    use std::sync::Arc;
486
487    use bytes::Bytes;
488    use futures::AsyncReadExt;
489    use futures::io::AllowStdIo;
490    use tempfile::TempDir;
491
492    use super::{FileIO, FileIOBuilder};
493    use crate::io::{LocalFsStorageFactory, MemoryStorageFactory};
494
495    fn create_local_file_io() -> FileIO {
496        FileIO::new_with_fs()
497    }
498
499    fn write_to_file<P: AsRef<Path>>(s: &str, path: P) {
500        create_dir_all(path.as_ref().parent().unwrap()).unwrap();
501        let mut f = File::create(path).unwrap();
502        write!(f, "{s}").unwrap();
503    }
504
505    async fn read_from_file<P: AsRef<Path>>(path: P) -> String {
506        let mut f = AllowStdIo::new(File::open(path).unwrap());
507        let mut s = String::new();
508        f.read_to_string(&mut s).await.unwrap();
509        s
510    }
511
512    #[tokio::test]
513    async fn test_local_input_file() {
514        let tmp_dir = TempDir::new().unwrap();
515
516        let file_name = "a.txt";
517        let content = "Iceberg loves rust.";
518
519        let full_path = format!("{}/{}", tmp_dir.path().to_str().unwrap(), file_name);
520        write_to_file(content, &full_path);
521
522        let file_io = create_local_file_io();
523        let input_file = file_io.new_input(&full_path).unwrap();
524
525        assert!(input_file.exists().await.unwrap());
526        assert_eq!(&full_path, input_file.location());
527        let read_content = read_from_file(full_path).await;
528
529        assert_eq!(content, &read_content);
530    }
531
532    #[tokio::test]
533    async fn test_delete_local_file() {
534        let tmp_dir = TempDir::new().unwrap();
535
536        let a_path = format!("{}/{}", tmp_dir.path().to_str().unwrap(), "a.txt");
537        let sub_dir_path = format!("{}/sub", tmp_dir.path().to_str().unwrap());
538        let b_path = format!("{}/{}", sub_dir_path, "b.txt");
539        let c_path = format!("{}/{}", sub_dir_path, "c.txt");
540        write_to_file("Iceberg loves rust.", &a_path);
541        write_to_file("Iceberg loves rust.", &b_path);
542        write_to_file("Iceberg loves rust.", &c_path);
543
544        let file_io = create_local_file_io();
545        assert!(file_io.exists(&a_path).await.unwrap());
546
547        // Remove a file should be no-op.
548        file_io.delete_prefix(&a_path).await.unwrap();
549        assert!(file_io.exists(&a_path).await.unwrap());
550
551        // Remove a not exist dir should be no-op.
552        file_io.delete_prefix("not_exists/").await.unwrap();
553
554        // Remove a dir should remove all files in it.
555        file_io.delete_prefix(&sub_dir_path).await.unwrap();
556        assert!(!file_io.exists(&b_path).await.unwrap());
557        assert!(!file_io.exists(&c_path).await.unwrap());
558        assert!(file_io.exists(&a_path).await.unwrap());
559
560        file_io.delete(&a_path).await.unwrap();
561        assert!(!file_io.exists(&a_path).await.unwrap());
562    }
563
564    #[tokio::test]
565    async fn test_delete_non_exist_file() {
566        let tmp_dir = TempDir::new().unwrap();
567
568        let file_name = "a.txt";
569        let full_path = format!("{}/{}", tmp_dir.path().to_str().unwrap(), file_name);
570
571        let file_io = create_local_file_io();
572        assert!(!file_io.exists(&full_path).await.unwrap());
573        assert!(file_io.delete(&full_path).await.is_ok());
574        assert!(file_io.delete_prefix(&full_path).await.is_ok());
575    }
576
577    #[tokio::test]
578    async fn test_local_output_file() {
579        let tmp_dir = TempDir::new().unwrap();
580
581        let file_name = "a.txt";
582        let content = "Iceberg loves rust.";
583
584        let full_path = format!("{}/{}", tmp_dir.path().to_str().unwrap(), file_name);
585
586        let file_io = create_local_file_io();
587        let output_file = file_io.new_output(&full_path).unwrap();
588
589        assert!(!output_file.exists().await.unwrap());
590        {
591            output_file.write(content.into()).await.unwrap();
592        }
593
594        assert_eq!(&full_path, output_file.location());
595
596        let read_content = read_from_file(full_path).await;
597
598        assert_eq!(content, &read_content);
599    }
600
601    #[tokio::test]
602    async fn test_memory_io() {
603        let io = FileIO::new_with_memory();
604
605        let path = format!("{}/1.txt", TempDir::new().unwrap().path().to_str().unwrap());
606
607        let output_file = io.new_output(&path).unwrap();
608        output_file.write("test".into()).await.unwrap();
609
610        assert!(io.exists(&path.clone()).await.unwrap());
611        let input_file = io.new_input(&path).unwrap();
612        let content = input_file.read().await.unwrap();
613        assert_eq!(content, Bytes::from("test"));
614
615        io.delete(&path).await.unwrap();
616        assert!(!io.exists(&path).await.unwrap());
617    }
618
619    #[tokio::test]
620    async fn test_file_io_builder_with_props() {
621        let factory = Arc::new(MemoryStorageFactory);
622        let file_io = FileIOBuilder::new(factory)
623            .with_prop("key1", "value1")
624            .with_prop("key2", "value2")
625            .build();
626
627        assert_eq!(file_io.config().get("key1"), Some(&"value1".to_string()));
628        assert_eq!(file_io.config().get("key2"), Some(&"value2".to_string()));
629    }
630
631    #[tokio::test]
632    async fn test_file_io_builder_with_multiple_props() {
633        let factory = Arc::new(LocalFsStorageFactory);
634        let props = vec![("key1", "value1"), ("key2", "value2")];
635        let file_io = FileIOBuilder::new(factory).with_props(props).build();
636
637        assert_eq!(file_io.config().get("key1"), Some(&"value1".to_string()));
638        assert_eq!(file_io.config().get("key2"), Some(&"value2".to_string()));
639    }
640
641    #[tokio::test]
642    async fn test_prefixed_props_sorted_by_descending_prefix_length() {
643        let factory = Arc::new(MemoryStorageFactory);
644        let file_io = FileIOBuilder::new(factory)
645            .with_prefixed_props("memory://a/", [("k", "short")])
646            .with_prefixed_props("memory://a/longer/", [("k", "long")])
647            .build();
648
649        // Longest prefix first so the most specific match wins at routing time.
650        let prefixes: Vec<&str> = file_io.prefixed.iter().map(|p| p.prefix.as_str()).collect();
651        assert_eq!(prefixes, vec!["memory://a/longer/", "memory://a/"]);
652    }
653
654    #[tokio::test]
655    async fn test_prefixed_config_carries_credential_values() {
656        // Prefix config gets the vended credentials; default config keeps only base props.
657        let factory = Arc::new(MemoryStorageFactory);
658        let file_io = FileIOBuilder::new(factory)
659            .with_prop("s3.region", "us-east-1")
660            .with_prefixed_props("s3://bucket/table", [
661                ("s3.region", "us-east-1"),
662                ("s3.access-key-id", "vended-key"),
663                ("s3.secret-access-key", "vended-secret"),
664            ])
665            .build();
666
667        // Default: base props, no credentials.
668        assert_eq!(
669            file_io.config().get("s3.region"),
670            Some(&"us-east-1".to_string())
671        );
672        assert_eq!(file_io.config().get("s3.access-key-id"), None);
673
674        // Prefix: base props + vended credentials.
675        let prefixed = &file_io.prefixed[0].config;
676        assert_eq!(prefixed.get("s3.region"), Some(&"us-east-1".to_string()));
677        assert_eq!(
678            prefixed.get("s3.access-key-id"),
679            Some(&"vended-key".to_string())
680        );
681        assert_eq!(
682            prefixed.get("s3.secret-access-key"),
683            Some(&"vended-secret".to_string())
684        );
685    }
686
687    #[tokio::test]
688    async fn test_get_storage_routes_by_prefix() {
689        let factory = Arc::new(MemoryStorageFactory);
690        let file_io = FileIOBuilder::new(factory)
691            .with_prop("scope", "default")
692            .with_prefixed_props("memory://creds/", [("scope", "prefixed")])
693            .build();
694
695        let default_a = file_io.get_storage("memory://other/x").unwrap();
696        let default_b = file_io.get_storage("memory://other/y").unwrap();
697        let prefixed_a = file_io.get_storage("memory://creds/x").unwrap();
698        let prefixed_b = file_io.get_storage("memory://creds/y").unwrap();
699
700        // Repeated routing to the same bucket returns the memoized storage...
701        assert!(Arc::ptr_eq(&default_a, &default_b));
702        assert!(Arc::ptr_eq(&prefixed_a, &prefixed_b));
703        // ...and a prefix-matching path resolves to a distinct storage from the default.
704        assert!(!Arc::ptr_eq(&default_a, &prefixed_a));
705    }
706
707    #[tokio::test]
708    async fn test_delete_stream_routes_by_prefix() {
709        let factory = Arc::new(MemoryStorageFactory);
710        let file_io = FileIOBuilder::new(factory)
711            .with_prefixed_props("memory:/creds/", [("k", "v")])
712            .build();
713
714        // One file under each routing bucket (default vs prefixed storage).
715        let default_path = "memory:/other/a.txt";
716        let prefixed_path = "memory:/creds/b.txt";
717        for path in [default_path, prefixed_path] {
718            file_io
719                .new_output(path)
720                .unwrap()
721                .write("x".into())
722                .await
723                .unwrap();
724            assert!(file_io.exists(path).await.unwrap());
725        }
726
727        // delete_stream must route each path to the storage that holds it.
728        file_io
729            .delete_stream(futures::stream::iter(vec![
730                default_path.to_string(),
731                prefixed_path.to_string(),
732            ]))
733            .await
734            .unwrap();
735
736        assert!(!file_io.exists(default_path).await.unwrap());
737        assert!(!file_io.exists(prefixed_path).await.unwrap());
738    }
739
740    #[tokio::test]
741    async fn test_delete_stream_flushes_across_batches() {
742        // More than the flush threshold (1000): exercises mid-stream flush + remainder.
743        let factory = Arc::new(MemoryStorageFactory);
744        let file_io = FileIOBuilder::new(factory)
745            .with_prefixed_props("memory:/creds/", [("k", "v")])
746            .build();
747
748        let n = 1050;
749        let mut paths = Vec::with_capacity(n);
750        for i in 0..n {
751            let p = format!("memory:/creds/f{i}.txt");
752            file_io
753                .new_output(&p)
754                .unwrap()
755                .write("x".into())
756                .await
757                .unwrap();
758            paths.push(p);
759        }
760
761        file_io
762            .delete_stream(futures::stream::iter(paths.clone()))
763            .await
764            .unwrap();
765
766        for p in &paths {
767            assert!(!file_io.exists(p).await.unwrap());
768        }
769    }
770}
iceberg/io/file_io.rs

iceberg/io/
file_io.rs
