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::ops::Range;
19use std::sync::{Arc, OnceLock};
20
21use bytes::Bytes;
22use futures::{Stream, StreamExt};
23
24use super::storage::{
25    LocalFsStorageFactory, MemoryStorageFactory, Storage, StorageConfig, StorageFactory,
26};
27use crate::Result;
28
29/// FileIO implementation, used to manipulate files in underlying storage.
30///
31/// FileIO wraps a `dyn Storage` with lazy initialization via `StorageFactory`.
32/// The storage is created on first use and cached for subsequent operations.
33///
34/// # Note
35///
36/// All paths passed to `FileIO` must be absolute paths starting with the scheme string
37/// appropriate for the storage backend being used.
38///
39/// This crate provides native support for local filesystem (`file://`) and
40/// memory (`memory://`) storage. For extensive storage backend support (S3, GCS,
41/// OSS, Azure, etc.), use the
42/// [`iceberg-storage-opendal`](https://crates.io/crates/iceberg-storage-opendal) crate.
43///
44/// # Example
45///
46/// ```rust,ignore
47/// use iceberg::io::{FileIO, FileIOBuilder};
48/// use iceberg::io::{LocalFsStorageFactory, MemoryStorageFactory};
49/// use std::sync::Arc;
50///
51/// // Create FileIO with memory storage for testing
52/// let file_io = FileIO::new_with_memory();
53///
54/// // Create FileIO with local filesystem storage
55/// let file_io = FileIO::new_with_fs();
56///
57/// // Create FileIO with custom factory
58/// let file_io = FileIOBuilder::new(Arc::new(LocalFsStorageFactory))
59///     .with_prop("key", "value")
60///     .build();
61/// ```
62#[derive(Clone, Debug)]
63pub struct FileIO {
64    /// Storage configuration containing properties
65    config: StorageConfig,
66    /// Factory for creating storage instances
67    factory: Arc<dyn StorageFactory>,
68    /// Cached storage instance (lazily initialized)
69    storage: Arc<OnceLock<Arc<dyn Storage>>>,
70}
71
72mod _serde {
73    use std::sync::Arc;
74
75    use serde::{Deserialize, Serialize};
76
77    use super::{StorageConfig, StorageFactory};
78
79    #[derive(Serialize)]
80    pub(super) struct SerializableFileIO<'a> {
81        pub(super) config: &'a StorageConfig,
82        pub(super) factory: &'a Arc<dyn StorageFactory>,
83    }
84
85    #[derive(Deserialize)]
86    pub(super) struct DeserializedFileIO {
87        pub(super) config: StorageConfig,
88        pub(super) factory: Arc<dyn StorageFactory>,
89    }
90}
91
92impl FileIO {
93    /// Create a new FileIO backed by in-memory storage.
94    ///
95    /// This is useful for testing scenarios where persistent storage is not needed.
96    pub fn new_with_memory() -> Self {
97        Self {
98            config: StorageConfig::new(),
99            factory: Arc::new(MemoryStorageFactory),
100            storage: Arc::new(OnceLock::new()),
101        }
102    }
103
104    /// Create a new FileIO backed by local filesystem storage.
105    ///
106    /// This is useful for local development and testing with real files.
107    pub fn new_with_fs() -> Self {
108        Self {
109            config: StorageConfig::new(),
110            factory: Arc::new(LocalFsStorageFactory),
111            storage: Arc::new(OnceLock::new()),
112        }
113    }
114
115    /// Serializes all portable state of this `FileIO` into a byte vector.
116    ///
117    /// This includes the storage configuration and factory, but not the cached storage instance.
118    /// The storage cache is rebuilt lazily on first use after calling [`FileIO::deserialize_all`].
119    ///
120    /// The serialized representation is not a stable format and may change between crate versions.
121    /// Applications should not rely on it for long-term storage or exchange it between incompatible
122    /// versions of this crate.
123    ///
124    /// All storage configuration properties are included in the serialized representation. These
125    /// properties may contain credentials or other sensitive values, so the returned bytes must be
126    /// protected in transit and at rest by the application embedding this crate.
127    ///
128    /// Storage factories are serialized through [`typetag`](https://docs.rs/typetag). Third-party
129    /// factories must use `#[typetag::serde]` on their [`StorageFactory`] implementation.
130    pub fn serialize_all(&self) -> Result<Vec<u8>> {
131        Ok(serde_json::to_vec(&_serde::SerializableFileIO {
132            config: &self.config,
133            factory: &self.factory,
134        })?)
135    }
136
137    /// Deserializes a `FileIO` previously produced by [`FileIO::serialize_all`].
138    ///
139    /// The receiving binary must use a compatible crate version and link the concrete factory
140    /// implementation so it is registered with `typetag`. Backend-specific requirements are
141    /// documented by each storage factory implementation.
142    pub fn deserialize_all(bytes: &[u8]) -> Result<Self> {
143        let _serde::DeserializedFileIO { config, factory } = serde_json::from_slice(bytes)?;
144        Ok(Self {
145            config,
146            factory,
147            storage: Arc::new(OnceLock::new()),
148        })
149    }
150
151    /// Get the storage configuration.
152    pub fn config(&self) -> &StorageConfig {
153        &self.config
154    }
155
156    /// Get or create the storage instance.
157    ///
158    /// The factory is invoked on first access and the result is cached
159    /// for all subsequent operations.
160    fn get_storage(&self) -> Result<Arc<dyn Storage>> {
161        // Check if already initialized
162        if let Some(storage) = self.storage.get() {
163            return Ok(storage.clone());
164        }
165
166        // Build the storage
167        let storage = self.factory.build(&self.config)?;
168
169        // Try to set it (another thread might have set it first)
170        let _ = self.storage.set(storage.clone());
171
172        // Return whatever is in the cell (either ours or another thread's)
173        Ok(self.storage.get().unwrap().clone())
174    }
175
176    /// Deletes file.
177    ///
178    /// # Arguments
179    ///
180    /// * path: It should be *absolute* path starting with scheme string used to construct [`FileIO`].
181    pub async fn delete(&self, path: impl AsRef<str>) -> Result<()> {
182        self.get_storage()?.delete(path.as_ref()).await
183    }
184
185    /// Remove the path and all nested dirs and files recursively.
186    ///
187    /// # Arguments
188    ///
189    /// * path: It should be *absolute* path starting with scheme string used to construct [`FileIO`].
190    ///
191    /// # Behavior
192    ///
193    /// - If the path is a file or not exist, this function will be no-op.
194    /// - If the path is a empty directory, this function will remove the directory itself.
195    /// - If the path is a non-empty directory, this function will remove the directory and all nested files and directories.
196    pub async fn delete_prefix(&self, path: impl AsRef<str>) -> Result<()> {
197        self.get_storage()?.delete_prefix(path.as_ref()).await
198    }
199
200    /// Delete multiple files from a stream of paths.
201    ///
202    /// # Arguments
203    ///
204    /// * paths: A stream of absolute paths starting with the scheme string used to construct [`FileIO`].
205    pub async fn delete_stream(
206        &self,
207        paths: impl Stream<Item = String> + Send + 'static,
208    ) -> Result<()> {
209        self.get_storage()?.delete_stream(paths.boxed()).await
210    }
211
212    /// Check file exists.
213    ///
214    /// # Arguments
215    ///
216    /// * path: It should be *absolute* path starting with scheme string used to construct [`FileIO`].
217    pub async fn exists(&self, path: impl AsRef<str>) -> Result<bool> {
218        self.get_storage()?.exists(path.as_ref()).await
219    }
220
221    /// Creates input file.
222    ///
223    /// # Arguments
224    ///
225    /// * path: It should be *absolute* path starting with scheme string used to construct [`FileIO`].
226    pub fn new_input(&self, path: impl AsRef<str>) -> Result<InputFile> {
227        self.get_storage()?.new_input(path.as_ref())
228    }
229
230    /// Creates output file.
231    ///
232    /// # Arguments
233    ///
234    /// * path: It should be *absolute* path starting with scheme string used to construct [`FileIO`].
235    pub fn new_output(&self, path: impl AsRef<str>) -> Result<OutputFile> {
236        self.get_storage()?.new_output(path.as_ref())
237    }
238}
239
240/// Builder for [`FileIO`].
241///
242/// The builder accepts an explicit `StorageFactory` and configuration properties.
243/// Storage is lazily initialized on first use.
244#[derive(Clone, Debug)]
245pub struct FileIOBuilder {
246    /// Factory for creating storage instances
247    factory: Arc<dyn StorageFactory>,
248    /// Storage configuration
249    config: StorageConfig,
250}
251
252impl FileIOBuilder {
253    /// Creates a new builder with the given storage factory.
254    pub fn new(factory: Arc<dyn StorageFactory>) -> Self {
255        Self {
256            factory,
257            config: StorageConfig::new(),
258        }
259    }
260
261    /// Add a configuration property.
262    pub fn with_prop(mut self, key: impl ToString, value: impl ToString) -> Self {
263        self.config = self.config.with_prop(key.to_string(), value.to_string());
264        self
265    }
266
267    /// Add multiple configuration properties.
268    pub fn with_props(
269        mut self,
270        args: impl IntoIterator<Item = (impl ToString, impl ToString)>,
271    ) -> Self {
272        self.config = self
273            .config
274            .with_props(args.into_iter().map(|e| (e.0.to_string(), e.1.to_string())));
275        self
276    }
277
278    /// Get the storage configuration.
279    pub fn config(&self) -> &StorageConfig {
280        &self.config
281    }
282
283    /// Builds [`FileIO`].
284    pub fn build(self) -> FileIO {
285        FileIO {
286            config: self.config,
287            factory: self.factory,
288            storage: Arc::new(OnceLock::new()),
289        }
290    }
291}
292
293/// The struct the represents the metadata of a file.
294///
295/// TODO: we can add last modified time, content type, etc. in the future.
296pub struct FileMetadata {
297    /// The size of the file.
298    pub size: u64,
299}
300
301/// Trait for reading file.
302///
303/// # TODO
304/// It's possible for us to remove the async_trait, but we need to figure
305/// out how to handle the object safety.
306#[async_trait::async_trait]
307pub trait FileRead: Send + Sync + Unpin + 'static {
308    /// Read file content with given range.
309    ///
310    /// TODO: we can support reading non-contiguous bytes in the future.
311    async fn read(&self, range: Range<u64>) -> Result<Bytes>;
312}
313
314#[async_trait::async_trait]
315impl<T: AsRef<dyn FileRead> + Send + Sync + Unpin + 'static> FileRead for T {
316    async fn read(&self, range: Range<u64>) -> Result<Bytes> {
317        self.as_ref().read(range).await
318    }
319}
320
321/// Input file is used for reading from files.
322#[derive(Debug)]
323pub struct InputFile {
324    storage: Arc<dyn Storage>,
325    // Absolute path of file.
326    path: String,
327}
328
329impl InputFile {
330    /// Creates a new input file.
331    pub fn new(storage: Arc<dyn Storage>, path: String) -> Self {
332        Self { storage, path }
333    }
334
335    /// Absolute path to root uri.
336    pub fn location(&self) -> &str {
337        &self.path
338    }
339
340    /// Check if file exists.
341    pub async fn exists(&self) -> Result<bool> {
342        self.storage.exists(&self.path).await
343    }
344
345    /// Fetch and returns metadata of file.
346    pub async fn metadata(&self) -> Result<FileMetadata> {
347        self.storage.metadata(&self.path).await
348    }
349
350    /// Read and returns whole content of file.
351    ///
352    /// For continuous reading, use [`Self::reader`] instead.
353    pub async fn read(&self) -> Result<Bytes> {
354        self.storage.read(&self.path).await
355    }
356
357    /// Creates [`FileRead`] for continuous reading.
358    ///
359    /// For one-time reading, use [`Self::read`] instead.
360    pub async fn reader(&self) -> Result<Box<dyn FileRead>> {
361        self.storage.reader(&self.path).await
362    }
363}
364
365/// Trait for writing file.
366///
367/// # TODO
368///
369/// It's possible for us to remove the async_trait, but we need to figure
370/// out how to handle the object safety.
371#[async_trait::async_trait]
372pub trait FileWrite: Send + Unpin + 'static {
373    /// Write bytes to file.
374    ///
375    /// TODO: we can support writing non-contiguous bytes in the future.
376    async fn write(&mut self, bs: Bytes) -> Result<()>;
377
378    /// Close file.
379    ///
380    /// Calling close on closed file will generate an error.
381    async fn close(&mut self) -> Result<()>;
382}
383
384/// Output file is used for writing to files..
385#[derive(Debug)]
386pub struct OutputFile {
387    storage: Arc<dyn Storage>,
388    // Absolute path of file.
389    path: String,
390}
391
392impl OutputFile {
393    /// Creates a new output file.
394    pub fn new(storage: Arc<dyn Storage>, path: String) -> Self {
395        Self { storage, path }
396    }
397
398    /// Relative path to root uri.
399    pub fn location(&self) -> &str {
400        &self.path
401    }
402
403    /// Checks if file exists.
404    pub async fn exists(&self) -> Result<bool> {
405        self.storage.exists(&self.path).await
406    }
407
408    /// Deletes file.
409    ///
410    /// If the file does not exist, it will not return error.
411    pub async fn delete(&self) -> Result<()> {
412        self.storage.delete(&self.path).await
413    }
414
415    /// Converts into [`InputFile`].
416    pub fn to_input_file(self) -> InputFile {
417        InputFile {
418            storage: self.storage,
419            path: self.path,
420        }
421    }
422
423    /// Create a new output file with given bytes.
424    ///
425    /// # Notes
426    ///
427    /// Calling `write` will overwrite the file if it exists.
428    /// For continuous writing, use [`Self::writer`].
429    pub async fn write(&self, bs: Bytes) -> Result<()> {
430        self.storage.write(&self.path, bs).await
431    }
432
433    /// Creates output file for continuous writing.
434    ///
435    /// # Notes
436    ///
437    /// For one-time writing, use [`Self::write`] instead.
438    pub async fn writer(&self) -> Result<Box<dyn FileWrite>> {
439        self.storage.writer(&self.path).await
440    }
441}
442
443#[cfg(test)]
444mod tests {
445    use std::fs::{File, create_dir_all};
446    use std::io::Write;
447    use std::path::Path;
448    use std::sync::Arc;
449
450    use bytes::Bytes;
451    use futures::AsyncReadExt;
452    use futures::io::AllowStdIo;
453    use tempfile::TempDir;
454
455    use super::{FileIO, FileIOBuilder};
456    use crate::io::{LocalFsStorageFactory, MemoryStorageFactory};
457
458    fn create_local_file_io() -> FileIO {
459        FileIO::new_with_fs()
460    }
461
462    fn write_to_file<P: AsRef<Path>>(s: &str, path: P) {
463        create_dir_all(path.as_ref().parent().unwrap()).unwrap();
464        let mut f = File::create(path).unwrap();
465        write!(f, "{s}").unwrap();
466    }
467
468    async fn read_from_file<P: AsRef<Path>>(path: P) -> String {
469        let mut f = AllowStdIo::new(File::open(path).unwrap());
470        let mut s = String::new();
471        f.read_to_string(&mut s).await.unwrap();
472        s
473    }
474
475    #[tokio::test]
476    async fn test_local_input_file() {
477        let tmp_dir = TempDir::new().unwrap();
478
479        let file_name = "a.txt";
480        let content = "Iceberg loves rust.";
481
482        let full_path = format!("{}/{}", tmp_dir.path().to_str().unwrap(), file_name);
483        write_to_file(content, &full_path);
484
485        let file_io = create_local_file_io();
486        let input_file = file_io.new_input(&full_path).unwrap();
487
488        assert!(input_file.exists().await.unwrap());
489        assert_eq!(&full_path, input_file.location());
490        let read_content = read_from_file(full_path).await;
491
492        assert_eq!(content, &read_content);
493    }
494
495    #[tokio::test]
496    async fn test_delete_local_file() {
497        let tmp_dir = TempDir::new().unwrap();
498
499        let a_path = format!("{}/{}", tmp_dir.path().to_str().unwrap(), "a.txt");
500        let sub_dir_path = format!("{}/sub", tmp_dir.path().to_str().unwrap());
501        let b_path = format!("{}/{}", sub_dir_path, "b.txt");
502        let c_path = format!("{}/{}", sub_dir_path, "c.txt");
503        write_to_file("Iceberg loves rust.", &a_path);
504        write_to_file("Iceberg loves rust.", &b_path);
505        write_to_file("Iceberg loves rust.", &c_path);
506
507        let file_io = create_local_file_io();
508        assert!(file_io.exists(&a_path).await.unwrap());
509
510        // Remove a file should be no-op.
511        file_io.delete_prefix(&a_path).await.unwrap();
512        assert!(file_io.exists(&a_path).await.unwrap());
513
514        // Remove a not exist dir should be no-op.
515        file_io.delete_prefix("not_exists/").await.unwrap();
516
517        // Remove a dir should remove all files in it.
518        file_io.delete_prefix(&sub_dir_path).await.unwrap();
519        assert!(!file_io.exists(&b_path).await.unwrap());
520        assert!(!file_io.exists(&c_path).await.unwrap());
521        assert!(file_io.exists(&a_path).await.unwrap());
522
523        file_io.delete(&a_path).await.unwrap();
524        assert!(!file_io.exists(&a_path).await.unwrap());
525    }
526
527    #[tokio::test]
528    async fn test_delete_non_exist_file() {
529        let tmp_dir = TempDir::new().unwrap();
530
531        let file_name = "a.txt";
532        let full_path = format!("{}/{}", tmp_dir.path().to_str().unwrap(), file_name);
533
534        let file_io = create_local_file_io();
535        assert!(!file_io.exists(&full_path).await.unwrap());
536        assert!(file_io.delete(&full_path).await.is_ok());
537        assert!(file_io.delete_prefix(&full_path).await.is_ok());
538    }
539
540    #[tokio::test]
541    async fn test_local_output_file() {
542        let tmp_dir = TempDir::new().unwrap();
543
544        let file_name = "a.txt";
545        let content = "Iceberg loves rust.";
546
547        let full_path = format!("{}/{}", tmp_dir.path().to_str().unwrap(), file_name);
548
549        let file_io = create_local_file_io();
550        let output_file = file_io.new_output(&full_path).unwrap();
551
552        assert!(!output_file.exists().await.unwrap());
553        {
554            output_file.write(content.into()).await.unwrap();
555        }
556
557        assert_eq!(&full_path, output_file.location());
558
559        let read_content = read_from_file(full_path).await;
560
561        assert_eq!(content, &read_content);
562    }
563
564    #[tokio::test]
565    async fn test_memory_io() {
566        let io = FileIO::new_with_memory();
567
568        let path = format!("{}/1.txt", TempDir::new().unwrap().path().to_str().unwrap());
569
570        let output_file = io.new_output(&path).unwrap();
571        output_file.write("test".into()).await.unwrap();
572
573        assert!(io.exists(&path.clone()).await.unwrap());
574        let input_file = io.new_input(&path).unwrap();
575        let content = input_file.read().await.unwrap();
576        assert_eq!(content, Bytes::from("test"));
577
578        io.delete(&path).await.unwrap();
579        assert!(!io.exists(&path).await.unwrap());
580    }
581
582    #[tokio::test]
583    async fn test_file_io_builder_with_props() {
584        let factory = Arc::new(MemoryStorageFactory);
585        let file_io = FileIOBuilder::new(factory)
586            .with_prop("key1", "value1")
587            .with_prop("key2", "value2")
588            .build();
589
590        assert_eq!(file_io.config().get("key1"), Some(&"value1".to_string()));
591        assert_eq!(file_io.config().get("key2"), Some(&"value2".to_string()));
592    }
593
594    #[tokio::test]
595    async fn test_file_io_builder_with_multiple_props() {
596        let factory = Arc::new(LocalFsStorageFactory);
597        let props = vec![("key1", "value1"), ("key2", "value2")];
598        let file_io = FileIOBuilder::new(factory).with_props(props).build();
599
600        assert_eq!(file_io.config().get("key1"), Some(&"value1".to_string()));
601        assert_eq!(file_io.config().get("key2"), Some(&"value2".to_string()));
602    }
603
604    #[tokio::test]
605    async fn test_memory_file_io_serialization_roundtrip() {
606        let file_io = FileIOBuilder::new(Arc::new(MemoryStorageFactory))
607            .with_prop("test-property", "test-value")
608            .with_prop("s3.session-token", "test-token")
609            .build();
610
611        file_io
612            .new_output("memory://test/file.txt")
613            .unwrap()
614            .write("test".into())
615            .await
616            .unwrap();
617        assert!(file_io.storage.get().is_some());
618
619        let serialized = file_io.serialize_all().unwrap();
620        let deserialized = FileIO::deserialize_all(&serialized).unwrap();
621        assert!(deserialized.storage.get().is_none());
622        assert_eq!(
623            deserialized.config().get("test-property"),
624            Some(&"test-value".to_string())
625        );
626        assert_eq!(
627            deserialized.config().get("s3.session-token"),
628            Some(&"test-token".to_string())
629        );
630
631        deserialized
632            .new_output("memory://test/roundtrip.txt")
633            .unwrap()
634            .write("roundtrip".into())
635            .await
636            .unwrap();
637        assert_eq!(
638            deserialized
639                .new_input("memory://test/roundtrip.txt")
640                .unwrap()
641                .read()
642                .await
643                .unwrap(),
644            Bytes::from("roundtrip")
645        );
646        assert!(deserialized.storage.get().is_some());
647    }
648
649    #[tokio::test]
650    async fn test_local_fs_file_io_serialization_roundtrip() {
651        let tmp_dir = TempDir::new().unwrap();
652        let path = tmp_dir.path().join("roundtrip.txt");
653        let path = path.to_str().unwrap();
654        let file_io = FileIOBuilder::new(Arc::new(LocalFsStorageFactory))
655            .with_prop("test-property", "test-value")
656            .build();
657
658        file_io
659            .new_output(path)
660            .unwrap()
661            .write("roundtrip".into())
662            .await
663            .unwrap();
664        assert!(file_io.storage.get().is_some());
665
666        let serialized = file_io.serialize_all().unwrap();
667        let deserialized = FileIO::deserialize_all(&serialized).unwrap();
668        assert!(deserialized.storage.get().is_none());
669        assert_eq!(
670            deserialized.config().get("test-property"),
671            Some(&"test-value".to_string())
672        );
673        assert!(deserialized.exists(path).await.unwrap());
674        assert_eq!(
675            deserialized.new_input(path).unwrap().read().await.unwrap(),
676            Bytes::from("roundtrip")
677        );
678        assert!(deserialized.storage.get().is_some());
679    }
680}
iceberg/io/file_io.rs

iceberg/io/
file_io.rs
