iceberg/io/storage/
local_fs.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
18//! Local filesystem storage implementation for testing.
19//!
20//! This module provides a `LocalFsStorage` implementation that uses standard
21//! Rust filesystem operations. It is primarily intended for unit testing
22//! scenarios where tests need to read/write files on the local filesystem.
23
24use std::fs;
25use std::io::{Read, Seek, SeekFrom, Write};
26use std::ops::Range;
27use std::path::PathBuf;
28use std::sync::Arc;
29
30use async_trait::async_trait;
31use bytes::Bytes;
32use futures::StreamExt;
33use futures::stream::BoxStream;
34use serde::{Deserialize, Serialize};
35
36use crate::io::{
37    FileMetadata, FileRead, FileWrite, InputFile, OutputFile, Storage, StorageConfig,
38    StorageFactory,
39};
40use crate::{Error, ErrorKind, Result};
41
42/// Local filesystem storage implementation.
43///
44/// This storage implementation uses standard Rust filesystem operations,
45/// making it suitable for unit tests that need to read/write files on disk.
46///
47/// # Path Normalization
48///
49/// The storage normalizes paths to handle various formats:
50/// - `file:///path/to/file` -> `/path/to/file`
51/// - `file:/path/to/file` -> `/path/to/file`
52/// - `/path/to/file` -> `/path/to/file`
53#[derive(Debug, Clone, Default, Serialize, Deserialize)]
54pub struct LocalFsStorage;
55
56impl LocalFsStorage {
57    /// Create a new `LocalFsStorage` instance.
58    pub fn new() -> Self {
59        Self
60    }
61
62    /// Normalize a path by removing scheme prefixes.
63    ///
64    /// This handles the following formats:
65    /// - `file:///path` -> `/path`
66    /// - `file://path` -> `/path` (treats as absolute)
67    /// - `file:/path` -> `/path`
68    /// - `/path` -> `/path`
69    pub(crate) fn normalize_path(path: &str) -> PathBuf {
70        let path = if let Some(stripped) = path.strip_prefix("file://") {
71            // file:///path -> /path or file://path -> /path
72            if stripped.starts_with('/') {
73                stripped.to_string()
74            } else {
75                format!("/{stripped}")
76            }
77        } else if let Some(stripped) = path.strip_prefix("file:") {
78            // file:/path -> /path
79            if stripped.starts_with('/') {
80                stripped.to_string()
81            } else {
82                format!("/{stripped}")
83            }
84        } else {
85            path.to_string()
86        };
87        PathBuf::from(path)
88    }
89}
90
91#[async_trait]
92#[typetag::serde]
93impl Storage for LocalFsStorage {
94    async fn exists(&self, path: &str) -> Result<bool> {
95        let path = Self::normalize_path(path);
96        Ok(path.exists())
97    }
98
99    async fn metadata(&self, path: &str) -> Result<FileMetadata> {
100        let path = Self::normalize_path(path);
101        let metadata = fs::metadata(&path).map_err(|e| {
102            Error::new(
103                ErrorKind::DataInvalid,
104                format!("Failed to get metadata for {}: {}", path.display(), e),
105            )
106        })?;
107        Ok(FileMetadata {
108            size: metadata.len(),
109        })
110    }
111
112    async fn read(&self, path: &str) -> Result<Bytes> {
113        let path = Self::normalize_path(path);
114        let content = fs::read(&path).map_err(|e| {
115            Error::new(
116                ErrorKind::DataInvalid,
117                format!("Failed to read file {}: {}", path.display(), e),
118            )
119        })?;
120        Ok(Bytes::from(content))
121    }
122
123    async fn reader(&self, path: &str) -> Result<Box<dyn FileRead>> {
124        let path = Self::normalize_path(path);
125        let file = fs::File::open(&path).map_err(|e| {
126            Error::new(
127                ErrorKind::DataInvalid,
128                format!("Failed to open file {}: {}", path.display(), e),
129            )
130        })?;
131        Ok(Box::new(LocalFsFileRead::new(file)))
132    }
133
134    async fn write(&self, path: &str, bs: Bytes) -> Result<()> {
135        let path = Self::normalize_path(path);
136
137        // Create parent directories if they don't exist
138        if let Some(parent) = path.parent() {
139            fs::create_dir_all(parent).map_err(|e| {
140                Error::new(
141                    ErrorKind::Unexpected,
142                    format!("Failed to create directory {}: {}", parent.display(), e),
143                )
144            })?;
145        }
146
147        fs::write(&path, &bs).map_err(|e| {
148            Error::new(
149                ErrorKind::Unexpected,
150                format!("Failed to write file {}: {}", path.display(), e),
151            )
152        })?;
153        Ok(())
154    }
155
156    async fn writer(&self, path: &str) -> Result<Box<dyn FileWrite>> {
157        let path = Self::normalize_path(path);
158
159        // Create parent directories if they don't exist
160        if let Some(parent) = path.parent() {
161            fs::create_dir_all(parent).map_err(|e| {
162                Error::new(
163                    ErrorKind::Unexpected,
164                    format!("Failed to create directory {}: {}", parent.display(), e),
165                )
166            })?;
167        }
168
169        let file = fs::File::create(&path).map_err(|e| {
170            Error::new(
171                ErrorKind::Unexpected,
172                format!("Failed to create file {}: {}", path.display(), e),
173            )
174        })?;
175        Ok(Box::new(LocalFsFileWrite::new(file)))
176    }
177
178    async fn delete(&self, path: &str) -> Result<()> {
179        let path = Self::normalize_path(path);
180        if path.exists() {
181            fs::remove_file(&path).map_err(|e| {
182                Error::new(
183                    ErrorKind::Unexpected,
184                    format!("Failed to delete file {}: {}", path.display(), e),
185                )
186            })?;
187        }
188        Ok(())
189    }
190
191    async fn delete_prefix(&self, path: &str) -> Result<()> {
192        let path = Self::normalize_path(path);
193        if path.is_dir() {
194            fs::remove_dir_all(&path).map_err(|e| {
195                Error::new(
196                    ErrorKind::Unexpected,
197                    format!("Failed to delete directory {}: {}", path.display(), e),
198                )
199            })?;
200        }
201        Ok(())
202    }
203
204    async fn delete_stream(&self, mut paths: BoxStream<'static, String>) -> Result<()> {
205        while let Some(path) = paths.next().await {
206            self.delete(&path).await?;
207        }
208        Ok(())
209    }
210
211    fn new_input(&self, path: &str) -> Result<InputFile> {
212        Ok(InputFile::new(Arc::new(self.clone()), path.to_string()))
213    }
214
215    fn new_output(&self, path: &str) -> Result<OutputFile> {
216        Ok(OutputFile::new(Arc::new(self.clone()), path.to_string()))
217    }
218}
219
220/// File reader for local filesystem storage.
221#[derive(Debug)]
222pub struct LocalFsFileRead {
223    file: std::sync::Mutex<fs::File>,
224}
225
226impl LocalFsFileRead {
227    /// Create a new `LocalFsFileRead` with the given file.
228    pub fn new(file: fs::File) -> Self {
229        Self {
230            file: std::sync::Mutex::new(file),
231        }
232    }
233}
234
235#[async_trait]
236impl FileRead for LocalFsFileRead {
237    async fn read(&self, range: Range<u64>) -> Result<Bytes> {
238        let mut file = self.file.lock().map_err(|e| {
239            Error::new(
240                ErrorKind::Unexpected,
241                format!("Failed to acquire file lock: {e}"),
242            )
243        })?;
244
245        file.seek(SeekFrom::Start(range.start)).map_err(|e| {
246            Error::new(
247                ErrorKind::DataInvalid,
248                format!("Failed to seek to position {}: {}", range.start, e),
249            )
250        })?;
251
252        let len = (range.end - range.start) as usize;
253        let mut buffer = vec![0u8; len];
254        file.read_exact(&mut buffer).map_err(|e| {
255            Error::new(
256                ErrorKind::DataInvalid,
257                format!("Failed to read {len} bytes: {e}"),
258            )
259        })?;
260
261        Ok(Bytes::from(buffer))
262    }
263}
264
265/// File writer for local filesystem storage.
266///
267/// This struct implements `FileWrite` for writing to local files.
268#[derive(Debug)]
269pub struct LocalFsFileWrite {
270    file: Option<fs::File>,
271}
272
273impl LocalFsFileWrite {
274    /// Create a new `LocalFsFileWrite` for the given file.
275    pub fn new(file: fs::File) -> Self {
276        Self { file: Some(file) }
277    }
278}
279
280#[async_trait]
281impl FileWrite for LocalFsFileWrite {
282    async fn write(&mut self, bs: Bytes) -> Result<()> {
283        let file = self
284            .file
285            .as_mut()
286            .ok_or_else(|| Error::new(ErrorKind::DataInvalid, "Cannot write to closed file"))?;
287
288        file.write_all(&bs).map_err(|e| {
289            Error::new(
290                ErrorKind::Unexpected,
291                format!("Failed to write to file: {e}"),
292            )
293        })?;
294
295        Ok(())
296    }
297
298    async fn close(&mut self) -> Result<()> {
299        let file = self
300            .file
301            .take()
302            .ok_or_else(|| Error::new(ErrorKind::DataInvalid, "File already closed"))?;
303
304        file.sync_all()
305            .map_err(|e| Error::new(ErrorKind::Unexpected, format!("Failed to sync file: {e}")))?;
306
307        Ok(())
308    }
309}
310
311/// Factory for creating `LocalFsStorage` instances.
312///
313/// This factory implements `StorageFactory` and creates `LocalFsStorage`
314/// instances for the "file" scheme.
315///
316/// # Example
317///
318/// ```rust,ignore
319/// use iceberg::io::{StorageConfig, StorageFactory, LocalFsStorageFactory};
320///
321/// let factory = LocalFsStorageFactory;
322/// let config = StorageConfig::new();
323/// let storage = factory.build(&config)?;
324/// ```
325#[derive(Clone, Debug, Default, Serialize, Deserialize)]
326pub struct LocalFsStorageFactory;
327
328#[typetag::serde]
329impl StorageFactory for LocalFsStorageFactory {
330    fn build(&self, _config: &StorageConfig) -> Result<Arc<dyn Storage>> {
331        Ok(Arc::new(LocalFsStorage::new()))
332    }
333}
334
335#[cfg(test)]
336mod tests {
337    use tempfile::TempDir;
338
339    use super::*;
340
341    #[test]
342    fn test_normalize_path() {
343        // Test file:/// prefix
344        assert_eq!(
345            LocalFsStorage::normalize_path("file:///path/to/file"),
346            PathBuf::from("/path/to/file")
347        );
348
349        // Test file:// prefix (without leading slash in path)
350        assert_eq!(
351            LocalFsStorage::normalize_path("file://path/to/file"),
352            PathBuf::from("/path/to/file")
353        );
354
355        // Test file:/ prefix
356        assert_eq!(
357            LocalFsStorage::normalize_path("file:/path/to/file"),
358            PathBuf::from("/path/to/file")
359        );
360
361        // Test bare path
362        assert_eq!(
363            LocalFsStorage::normalize_path("/path/to/file"),
364            PathBuf::from("/path/to/file")
365        );
366    }
367
368    #[tokio::test]
369    async fn test_local_fs_storage_write_read() {
370        let tmp_dir = TempDir::new().unwrap();
371        let storage = LocalFsStorage::new();
372        let path = tmp_dir.path().join("test.txt");
373        let path_str = path.to_str().unwrap();
374        let content = Bytes::from("Hello, World!");
375
376        // Write
377        storage.write(path_str, content.clone()).await.unwrap();
378
379        // Read
380        let read_content = storage.read(path_str).await.unwrap();
381        assert_eq!(read_content, content);
382    }
383
384    #[tokio::test]
385    async fn test_local_fs_storage_exists() {
386        let tmp_dir = TempDir::new().unwrap();
387        let storage = LocalFsStorage::new();
388        let path = tmp_dir.path().join("test.txt");
389        let path_str = path.to_str().unwrap();
390
391        // File doesn't exist initially
392        assert!(!storage.exists(path_str).await.unwrap());
393
394        // Write file
395        storage.write(path_str, Bytes::from("test")).await.unwrap();
396
397        // File exists now
398        assert!(storage.exists(path_str).await.unwrap());
399    }
400
401    #[tokio::test]
402    async fn test_local_fs_storage_metadata() {
403        let tmp_dir = TempDir::new().unwrap();
404        let storage = LocalFsStorage::new();
405        let path = tmp_dir.path().join("test.txt");
406        let path_str = path.to_str().unwrap();
407        let content = Bytes::from("Hello, World!");
408
409        storage.write(path_str, content.clone()).await.unwrap();
410
411        let metadata = storage.metadata(path_str).await.unwrap();
412        assert_eq!(metadata.size, content.len() as u64);
413    }
414
415    #[tokio::test]
416    async fn test_local_fs_storage_delete() {
417        let tmp_dir = TempDir::new().unwrap();
418        let storage = LocalFsStorage::new();
419        let path = tmp_dir.path().join("test.txt");
420        let path_str = path.to_str().unwrap();
421
422        storage.write(path_str, Bytes::from("test")).await.unwrap();
423        assert!(storage.exists(path_str).await.unwrap());
424
425        storage.delete(path_str).await.unwrap();
426        assert!(!storage.exists(path_str).await.unwrap());
427    }
428
429    #[tokio::test]
430    async fn test_local_fs_storage_delete_prefix() {
431        let tmp_dir = TempDir::new().unwrap();
432        let storage = LocalFsStorage::new();
433        let dir_path = tmp_dir.path().join("subdir");
434        let file1 = dir_path.join("file1.txt");
435        let file2 = dir_path.join("file2.txt");
436
437        // Create files in subdirectory
438        storage
439            .write(file1.to_str().unwrap(), Bytes::from("1"))
440            .await
441            .unwrap();
442        storage
443            .write(file2.to_str().unwrap(), Bytes::from("2"))
444            .await
445            .unwrap();
446
447        // Delete prefix (directory)
448        storage
449            .delete_prefix(dir_path.to_str().unwrap())
450            .await
451            .unwrap();
452
453        // Directory should be deleted
454        assert!(!dir_path.exists());
455    }
456
457    #[tokio::test]
458    async fn test_local_fs_storage_reader() {
459        let tmp_dir = TempDir::new().unwrap();
460        let storage = LocalFsStorage::new();
461        let path = tmp_dir.path().join("test.txt");
462        let path_str = path.to_str().unwrap();
463        let content = Bytes::from("Hello, World!");
464
465        storage.write(path_str, content.clone()).await.unwrap();
466
467        let reader = storage.reader(path_str).await.unwrap();
468        let read_content = reader.read(0..content.len() as u64).await.unwrap();
469        assert_eq!(read_content, content);
470
471        // Test partial read
472        let partial = reader.read(0..5).await.unwrap();
473        assert_eq!(partial, Bytes::from("Hello"));
474    }
475
476    #[tokio::test]
477    async fn test_local_fs_storage_writer() {
478        let tmp_dir = TempDir::new().unwrap();
479        let storage = LocalFsStorage::new();
480        let path = tmp_dir.path().join("test.txt");
481        let path_str = path.to_str().unwrap();
482
483        let mut writer = storage.writer(path_str).await.unwrap();
484        writer.write(Bytes::from("Hello, ")).await.unwrap();
485        writer.write(Bytes::from("World!")).await.unwrap();
486        writer.close().await.unwrap();
487
488        let content = storage.read(path_str).await.unwrap();
489        assert_eq!(content, Bytes::from("Hello, World!"));
490    }
491
492    #[tokio::test]
493    async fn test_local_fs_file_write_double_close() {
494        let tmp_dir = TempDir::new().unwrap();
495        let storage = LocalFsStorage::new();
496        let path = tmp_dir.path().join("test.txt");
497        let path_str = path.to_str().unwrap();
498
499        let mut writer = storage.writer(path_str).await.unwrap();
500        writer.write(Bytes::from("test")).await.unwrap();
501        writer.close().await.unwrap();
502
503        // Second close should fail
504        let result = writer.close().await;
505        assert!(result.is_err());
506    }
507
508    #[tokio::test]
509    async fn test_local_fs_file_write_after_close() {
510        let tmp_dir = TempDir::new().unwrap();
511        let storage = LocalFsStorage::new();
512        let path = tmp_dir.path().join("test.txt");
513        let path_str = path.to_str().unwrap();
514
515        let mut writer = storage.writer(path_str).await.unwrap();
516        writer.close().await.unwrap();
517
518        // Write after close should fail
519        let result = writer.write(Bytes::from("test")).await;
520        assert!(result.is_err());
521    }
522
523    #[test]
524    fn test_local_fs_storage_factory() {
525        let factory = LocalFsStorageFactory;
526        let config = StorageConfig::new();
527        let storage = factory.build(&config).unwrap();
528
529        // Verify we got a valid storage instance
530        assert!(format!("{storage:?}").contains("LocalFsStorage"));
531    }
532
533    #[tokio::test]
534    async fn test_local_fs_creates_parent_directories() {
535        let tmp_dir = TempDir::new().unwrap();
536        let storage = LocalFsStorage::new();
537        let path = tmp_dir.path().join("a/b/c/test.txt");
538        let path_str = path.to_str().unwrap();
539
540        // Write should create parent directories
541        storage.write(path_str, Bytes::from("test")).await.unwrap();
542
543        assert!(path.exists());
544    }
545
546    #[tokio::test]
547    async fn test_local_fs_storage_delete_stream() {
548        use futures::stream;
549
550        let tmp_dir = TempDir::new().unwrap();
551        let storage = LocalFsStorage::new();
552
553        // Create multiple files
554        let file1 = tmp_dir.path().join("file1.txt");
555        let file2 = tmp_dir.path().join("file2.txt");
556        let file3 = tmp_dir.path().join("file3.txt");
557
558        storage
559            .write(file1.to_str().unwrap(), Bytes::from("1"))
560            .await
561            .unwrap();
562        storage
563            .write(file2.to_str().unwrap(), Bytes::from("2"))
564            .await
565            .unwrap();
566        storage
567            .write(file3.to_str().unwrap(), Bytes::from("3"))
568            .await
569            .unwrap();
570
571        // Verify files exist
572        assert!(storage.exists(file1.to_str().unwrap()).await.unwrap());
573        assert!(storage.exists(file2.to_str().unwrap()).await.unwrap());
574        assert!(storage.exists(file3.to_str().unwrap()).await.unwrap());
575
576        // Delete multiple files using stream
577        let paths = vec![
578            file1.to_str().unwrap().to_string(),
579            file2.to_str().unwrap().to_string(),
580        ];
581        let path_stream = stream::iter(paths).boxed();
582        storage.delete_stream(path_stream).await.unwrap();
583
584        // Verify deleted files no longer exist
585        assert!(!storage.exists(file1.to_str().unwrap()).await.unwrap());
586        assert!(!storage.exists(file2.to_str().unwrap()).await.unwrap());
587
588        // Verify file3 still exists
589        assert!(storage.exists(file3.to_str().unwrap()).await.unwrap());
590    }
591
592    #[tokio::test]
593    async fn test_local_fs_storage_delete_stream_empty() {
594        use futures::stream;
595
596        let storage = LocalFsStorage::new();
597
598        // Delete with empty stream should succeed
599        let path_stream = stream::iter(Vec::<String>::new()).boxed();
600        storage.delete_stream(path_stream).await.unwrap();
601    }
602}
iceberg/io/storage/local_fs.rs

iceberg/io/storage/
local_fs.rs
