feat(io): support in-memory local cache

crates/paimon/Cargo.toml

@@ -87,6 +87,7 @@ opendal-service-s3 = { version = "0.58.0", optional = true }
 pretty_assertions = "1"
 apache-avro = { version = "0.21", features = ["snappy", "zstandard"] }
 indexmap = "2.5.0"
+lru = "0.18.2"
 roaring = "0.11"
 crc32fast = "1"
 zstd = "0.13"

crates/paimon/src/catalog/filesystem.rs

@@ -510,17 +510,12 @@ mod tests {
     #[test]
     fn test_filesystem_catalog_builds_local_cache_from_catalog_options() {
         let warehouse = TempDir::new().unwrap();
-        let cache = TempDir::new().unwrap();
         let mut options = Options::new();
         options.set(
             CatalogOptions::WAREHOUSE,
             warehouse.path().to_string_lossy(),
         );
         options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
-        options.set(
-            CatalogOptions::LOCAL_CACHE_DIR,
-            cache.path().to_string_lossy(),
-        );
 
         let catalog = FileSystemCatalog::new(options).unwrap();
 

crates/paimon/src/catalog/rest/rest_catalog.rs

@@ -548,17 +548,12 @@ mod tests {
 
     #[tokio::test]
     async fn test_rest_catalog_builds_local_cache_from_client_options() {
-        let cache = tempfile::tempdir().unwrap();
         let mut options = Options::new();
         options.set(CatalogOptions::URI, "http://localhost:1");
         options.set(CatalogOptions::WAREHOUSE, "test-warehouse");
         options.set("token.provider", "bear");
         options.set("token", "test-token");
         options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
-        options.set(
-            CatalogOptions::LOCAL_CACHE_DIR,
-            cache.path().to_string_lossy(),
-        );
 
         let catalog = RESTCatalog::new(options, false).await.unwrap();
 

crates/paimon/src/catalog/rest/rest_token_file_io.rs

@@ -217,14 +217,9 @@ mod tests {
 
     #[tokio::test]
     async fn test_token_file_io_keeps_catalog_local_cache() {
-        let cache_directory = tempfile::tempdir().unwrap();
         let table_directory = tempfile::tempdir().unwrap();
         let mut options = Options::new();
         options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
-        options.set(
-            CatalogOptions::LOCAL_CACHE_DIR,
-            cache_directory.path().to_string_lossy(),
-        );
         let local_cache = create_local_cache(&options).unwrap();
         let token_file_io = RESTTokenFileIO::new(
             Identifier::new("database", "table"),

crates/paimon/src/common/options.rs

@@ -76,10 +76,10 @@ impl CatalogOptions {
     /// Whether to enable local block caching for file reads.
     pub const LOCAL_CACHE_ENABLED: &'static str = "local-cache.enabled";
 
-    /// Directory for the local disk block cache.
+    /// Directory for the local disk block cache. If unset, an in-memory cache is used.
     pub const LOCAL_CACHE_DIR: &'static str = "local-cache.dir";
 
-    /// Maximum total encoded size of the local block cache.
+    /// Maximum total size of the local block cache.
     pub const LOCAL_CACHE_MAX_SIZE: &'static str = "local-cache.max-size";
 
     /// Block size used by the local cache.
@@ -175,6 +175,40 @@ impl From<HashMap<String, String>> for Options {
     }
 }
 
+#[derive(Debug, Eq, PartialEq)]
+pub(crate) enum ParseMemorySizeError {
+    Invalid,
+    Overflow,
+}
+
+/// Parses Java Paimon memory sizes, plus the binary unit aliases previously
+/// accepted by the Rust local cache.
+pub(crate) fn parse_memory_size(value: &str) -> Result<i64, ParseMemorySizeError> {
+    let value = value.trim();
+    if value.is_empty() {
+        return Err(ParseMemorySizeError::Invalid);
+    }
+
+    let pos = value
+        .find(|character: char| !character.is_ascii_digit())
+        .unwrap_or(value.len());
+    let (number, unit) = value.split_at(pos);
+    let number = number
+        .parse::<i64>()
+        .map_err(|_| ParseMemorySizeError::Invalid)?;
+    let multiplier = match unit.trim().to_ascii_lowercase().as_str() {
+        "" | "b" | "bytes" => 1,
+        "k" | "kb" | "kib" | "kibibytes" => 1024,
+        "m" | "mb" | "mib" | "mebibytes" => 1024 * 1024,
+        "g" | "gb" | "gib" | "gibibytes" => 1024 * 1024 * 1024,
+        "t" | "tb" | "tib" | "tebibytes" => 1024_i64.pow(4),
+        _ => return Err(ParseMemorySizeError::Invalid),
+    };
+    number
+        .checked_mul(multiplier)
+        .ok_or(ParseMemorySizeError::Overflow)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -244,4 +278,23 @@ mod tests {
             "local-cache.whitelist"
         );
     }
+
+    #[test]
+    fn test_parse_memory_size_supports_java_and_rust_units() {
+        for unit in ["k", "kb", "kib", "kibibytes"] {
+            assert_eq!(parse_memory_size(&format!("2 {unit}")), Ok(2 * 1024));
+        }
+        assert_eq!(parse_memory_size("2 bytes"), Ok(2));
+        assert_eq!(parse_memory_size("2 MiB"), Ok(2 * 1024 * 1024));
+        assert_eq!(parse_memory_size("2 gib"), Ok(2 * 1024 * 1024 * 1024));
+        assert_eq!(parse_memory_size("2 TiB"), Ok(2 * 1024_i64.pow(4)));
+        assert_eq!(
+            parse_memory_size("2 unknown"),
+            Err(ParseMemorySizeError::Invalid)
+        );
+        assert_eq!(
+            parse_memory_size("9223372036854775807 tb"),
+            Err(ParseMemorySizeError::Overflow)
+        );
+    }
 }