feat(iceberg): Add memory file IO support

crates/iceberg/Cargo.toml

@@ -29,9 +29,10 @@ license = { workspace = true }
 keywords = ["iceberg"]
 
 [features]
-default = ["storage-fs", "storage-s3", "tokio"]
-storage-all = ["storage-fs", "storage-s3"]
+default = ["storage-memory", "storage-fs", "storage-s3", "tokio"]
+storage-all = ["storage-memory", "storage-fs", "storage-s3"]
 
+storage-memory = ["opendal/services-memory"]
 storage-fs = ["opendal/services-fs"]
 storage-s3 = ["opendal/services-s3"]
 

crates/iceberg/src/io/mod.rs

@@ -52,6 +52,10 @@ mod file_io;
 pub use file_io::*;
 
 mod storage;
+#[cfg(feature = "storage-memory")]
+mod storage_memory;
+#[cfg(feature = "storage-memory")]
+use storage_memory::*;
 #[cfg(feature = "storage-s3")]
 mod storage_s3;
 #[cfg(feature = "storage-s3")]

crates/iceberg/src/io/storage.rs

@@ -18,6 +18,8 @@
 use super::FileIOBuilder;
 #[cfg(feature = "storage-fs")]
 use super::FsConfig;
+#[cfg(feature = "storage-memory")]
+use super::MemoryConfig;
 #[cfg(feature = "storage-s3")]
 use super::S3Config;
 use crate::{Error, ErrorKind};
@@ -26,6 +28,8 @@ use opendal::{Operator, Scheme};
 /// The storage carries all supported storage services in iceberg
 #[derive(Debug)]
 pub(crate) enum Storage {
+    #[cfg(feature = "storage-memory")]
+    Memory { config: MemoryConfig },
     #[cfg(feature = "storage-fs")]
     LocalFs { config: FsConfig },
     #[cfg(feature = "storage-s3")]
@@ -44,6 +48,10 @@ impl Storage {
         let scheme = Self::parse_scheme(&scheme_str)?;
 
         match scheme {
+            #[cfg(feature = "storage-memory")]
+            Scheme::Memory => Ok(Self::Memory {
+                config: MemoryConfig::new(props),
+            }),
             #[cfg(feature = "storage-fs")]
             Scheme::Fs => Ok(Self::LocalFs {
                 config: FsConfig::new(props),
@@ -79,6 +87,16 @@ impl Storage {
     ) -> crate::Result<(Operator, &'a str)> {
         let path = path.as_ref();
         match self {
+            #[cfg(feature = "storage-memory")]
+            Storage::Memory { config } => {
+                let op = config.build(path)?;
+
+                if let Some(stripped) = path.strip_prefix("file:/") {
+                    Ok((op, stripped))
+                } else {
+                    Ok((op, &path[1..]))
+                }
+            }
             #[cfg(feature = "storage-fs")]
             Storage::LocalFs { config } => {
                 let op = config.build(path)?;
@@ -116,6 +134,7 @@ impl Storage {
     /// Parse scheme.
     fn parse_scheme(scheme: &str) -> crate::Result<Scheme> {
         match scheme {
+            "memory" => Ok(Scheme::Memory),
             "file" | "" => Ok(Scheme::Fs),
             "s3" | "s3a" => Ok(Scheme::S3),
             s => Ok(s.parse::<Scheme>()?),

crates/iceberg/src/io/storage_memory.rs

@@ -0,0 +1,43 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::Result;
+use opendal::{Operator, Scheme};
+use std::collections::HashMap;
+use std::fmt::{Debug, Formatter};
+
+#[derive(Default, Clone)]
+pub(crate) struct MemoryConfig {}
+
+impl Debug for MemoryConfig {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("MemoryConfig").finish()
+    }
+}
+
+impl MemoryConfig {
+    /// Decode from iceberg props.
+    pub fn new(_: HashMap<String, String>) -> Self {
+        Self::default()
+    }
+
+    /// Build new opendal operator from give path.
+    pub fn build(&self, _: &str) -> Result<Operator> {
+        let m = HashMap::new();
+        Ok(Operator::via_map(Scheme::Memory, m)?)
+    }
+}