[core] Add opt-in node-shared disk block cache

crates/paimon/src/catalog/factory.rs

@@ -89,7 +89,7 @@ impl CatalogFactory {
             .unwrap_or(METASTORE_FILESYSTEM);
 
         match metastore {
-            METASTORE_FILESYSTEM => Self::create_filesystem_catalog(options),
+            METASTORE_FILESYSTEM => Self::create_filesystem_catalog(options).await,
             METASTORE_REST => Self::create_rest_catalog(options).await,
             _ => ConfigInvalidSnafu {
                 message: format!(
@@ -101,8 +101,13 @@ impl CatalogFactory {
     }
 
     /// Create a filesystem catalog.
-    fn create_filesystem_catalog(options: Options) -> Result<Arc<dyn Catalog>> {
-        let catalog = FileSystemCatalog::new(options)?;
+    async fn create_filesystem_catalog(options: Options) -> Result<Arc<dyn Catalog>> {
+        let catalog = tokio::task::spawn_blocking(move || FileSystemCatalog::new(options))
+            .await
+            .map_err(|error| crate::Error::UnexpectedError {
+                message: "Filesystem catalog initialization task failed".into(),
+                source: Some(Box::new(error)),
+            })??;
         Ok(Arc::new(catalog))
     }
 
@@ -140,6 +145,54 @@ mod tests {
         assert!(result.is_ok());
     }
 
+    #[test]
+    fn test_shared_cache_factory_init_does_not_block_current_thread_runtime() {
+        use std::fs::File;
+        use std::sync::mpsc;
+        use std::time::Duration;
+
+        let directory = tempfile::tempdir().unwrap();
+        let mut options = Options::new();
+        options.set(
+            CatalogOptions::WAREHOUSE,
+            directory.path().join("warehouse").to_str().unwrap(),
+        );
+        options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+        options.set(CatalogOptions::LOCAL_CACHE_SHARED, "true");
+        options.set(CatalogOptions::LOCAL_CACHE_MAX_SIZE, "1 MiB");
+        options.set(
+            CatalogOptions::LOCAL_CACHE_DIR,
+            directory.path().to_str().unwrap(),
+        );
+        drop(FileSystemCatalog::new(options.clone()).unwrap());
+        let gate = File::open(
+            directory
+                .path()
+                .join("paimon-shared-local-cache-v1/directory.lock"),
+        )
+        .unwrap();
+        gate.lock().unwrap();
+
+        let (tx, rx) = mpsc::channel();
+        let catalog_thread = std::thread::spawn(move || {
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .unwrap();
+            runtime.block_on(async move {
+                tokio::spawn(async move {
+                    tokio::time::sleep(Duration::from_millis(50)).await;
+                    tx.send(()).unwrap();
+                });
+                CatalogFactory::create(options).await.unwrap();
+            });
+        });
+        let runtime_progressed = rx.recv_timeout(Duration::from_secs(2)).is_ok();
+        drop(gate);
+        catalog_thread.join().unwrap();
+        assert!(runtime_progressed, "catalog factory blocked the runtime");
+    }
+
     #[tokio::test]
     async fn test_missing_warehouse_option() {
         let options = Options::new();

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

@@ -36,7 +36,7 @@ use crate::catalog::{
 };
 use crate::common::{CatalogOptions, Options};
 use crate::error::Error;
-use crate::io::cache::{create_local_cache_with_namespace, LocalCache};
+use crate::io::cache::{create_local_cache_with_namespace_async, LocalCache};
 use crate::io::FileIOCacheContext;
 use crate::spec::{Partition, PartitionStatistics, Schema, SchemaChange};
 use crate::table::{RESTEnv, Table};
@@ -90,7 +90,7 @@ impl RESTCatalog {
             .unwrap_or(false);
 
         let api_options = api.options().clone();
-        let local_cache = create_local_cache_with_namespace(&options, &api_options)?;
+        let local_cache = create_local_cache_with_namespace_async(&options, &api_options).await?;
         let file_io_cache = FileIOCacheContext::from_props(api_options.to_map())?;
 
         let fuse_resolver = Arc::new(super::FusePathResolver::new(&api_options));
@@ -1229,6 +1229,57 @@ mod tests {
         assert!(catalog.has_local_cache());
     }
 
+    #[test]
+    fn test_shared_cache_catalog_init_does_not_block_current_thread_runtime() {
+        use std::fs::File;
+        use std::sync::mpsc;
+        use std::time::Duration;
+
+        let directory = 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_SHARED, "true");
+        options.set(CatalogOptions::LOCAL_CACHE_MAX_SIZE, "1 MiB");
+        options.set(
+            CatalogOptions::LOCAL_CACHE_DIR,
+            directory.path().to_str().unwrap(),
+        );
+        drop(crate::io::cache::create_local_cache_with_namespace(&options, &options).unwrap());
+        let gate = File::open(
+            directory
+                .path()
+                .join("paimon-shared-local-cache-v1/directory.lock"),
+        )
+        .unwrap();
+        gate.lock().unwrap();
+
+        let (tx, rx) = mpsc::channel();
+        let catalog_thread = std::thread::spawn(move || {
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .unwrap();
+            runtime.block_on(async move {
+                tokio::spawn(async move {
+                    tokio::time::sleep(Duration::from_millis(50)).await;
+                    tx.send(()).unwrap();
+                });
+                RESTCatalog::new(options, false).await.unwrap();
+            });
+        });
+        let runtime_progressed = rx.recv_timeout(Duration::from_secs(2)).is_ok();
+        drop(gate);
+        catalog_thread.join().unwrap();
+        assert!(
+            runtime_progressed,
+            "catalog initialization blocked the runtime"
+        );
+    }
+
     #[test]
     fn test_tag_error_mapping() {
         let identifier = Identifier::new("db", "table");

crates/paimon/src/common/options.rs

@@ -81,6 +81,8 @@ impl CatalogOptions {
 
     /// 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";
+    /// Share a node-local disk cache between cooperating processes.
+    pub const LOCAL_CACHE_SHARED: &'static str = "local-cache.shared";
 
     /// Maximum total size of the local block cache.
     pub const LOCAL_CACHE_MAX_SIZE: &'static str = "local-cache.max-size";

crates/paimon/src/io/cache/disk.rs

@@ -26,6 +26,9 @@ use tokio::io::AsyncWriteExt;
 
 use super::state::{BlockKey, CacheCoordinator, LogicalPath};
 
+#[path = "shared.rs"]
+pub(super) mod shared;
+
 const CACHE_MAGIC: &[u8; 8] = b"PAIMONLC";
 const CACHE_FORMAT_VERSION: u8 = 2;
 const FIXED_HEADER_LEN: usize = CACHE_MAGIC.len() + 1 + 4 + 4 + 8 + 8 + 8;