[core] Support per-request endpoint routing for io-cache

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

@@ -31,7 +31,7 @@ use crate::api::rest_util::RESTUtil;
 use crate::catalog::Identifier;
 use crate::common::{CatalogOptions, Options};
 use crate::io::cache::LocalCache;
-use crate::io::{FileIO, FileIOCacheContext, FileIOProvider};
+use crate::io::{FileIO, FileIOCacheContext, FileIOProvider, RoutedOperator, IO_CACHE_ENABLED};
 use crate::Result;
 
 use super::rest_token::RESTToken;
@@ -182,6 +182,10 @@ impl RESTTokenFileIO {
                 merged.insert(OSS_ENDPOINT.to_string(), dlf_oss_endpoint.clone());
             }
         }
+        // An explicit client value wins over the token, so users can turn io-cache routing off.
+        if let Some(enabled) = self.catalog_options.get(IO_CACHE_ENABLED) {
+            merged.insert(IO_CACHE_ENABLED.to_string(), enabled.clone());
+        }
         merged
     }
 }
@@ -201,6 +205,21 @@ impl FileIOProvider for RESTTokenFileIO {
         let (op, relative_path) = file_io.create_static(path)?;
         Ok((op, relative_path, Some(cache_namespace)))
     }
+
+    async fn create_routed(
+        &self,
+        path: &str,
+        file_prefixes: &[String],
+    ) -> Result<(RoutedOperator, String, Option<String>)> {
+        let file_io = self.current_file_io().await?;
+        let namespace = file_io.cache_namespace_for_path(path)?;
+        let (op, relative_path) = file_io.create_routed_static(path, file_prefixes)?;
+        Ok((op, relative_path, Some(namespace)))
+    }
+
+    async fn routes_to_io_cache(&self) -> Result<bool> {
+        self.current_file_io().await?.routes_to_io_cache().await
+    }
 }
 
 #[cfg(test)]
@@ -441,4 +460,196 @@ mod tests {
         assert_eq!(requests.load(Ordering::SeqCst), 2);
         server.abort();
     }
+
+    #[cfg(feature = "storage-oss")]
+    mod io_cache {
+        use super::*;
+        use crate::io::oss_test_server::TestOss;
+
+        async fn fixed_token(
+            State(token): State<Arc<HashMap<String, String>>>,
+        ) -> Json<GetTableTokenResponse> {
+            let now = std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap()
+                .as_millis() as i64;
+            Json(GetTableTokenResponse {
+                token: token.as_ref().clone(),
+                expires_at_millis: Some(now + TOKEN_EXPIRATION_SAFE_TIME_MILLIS * 2),
+            })
+        }
+
+        /// A token FileIO for `oss://bkt/db.db/t` whose token vends io-cache routing.
+        async fn io_cache_file_io(
+            origin: &TestOss,
+            cache: &TestOss,
+            catalog_options: &[(&str, &str)],
+        ) -> (FileIO, tokio::task::JoinHandle<()>) {
+            let token = HashMap::from(
+                [
+                    ("fs.oss.endpoint", cache.endpoint()),
+                    ("fs.oss.accessKeyId", "ak"),
+                    ("fs.oss.accessKeySecret", "sk"),
+                    ("io-cache.enabled", "true"),
+                    ("io-cache.endpoint", cache.endpoint()),
+                    ("io-cache.target.default.path-style-access", "true"),
+                    ("io-cache.origin.endpoint", origin.endpoint()),
+                    ("io-cache.policy", "meta,read"),
+                ]
+                .map(|(k, v)| (k.to_string(), v.to_string())),
+            );
+            let app = Router::new()
+                .route("/v1/databases/db/tables/t/token", get(fixed_token))
+                .with_state(Arc::new(token));
+            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+            let address = listener.local_addr().unwrap();
+            let server = tokio::spawn(async move {
+                axum::serve(listener, app).await.unwrap();
+            });
+
+            let mut options = Options::new();
+            options.set(CatalogOptions::URI, format!("http://{address}"));
+            options.set(CatalogOptions::TOKEN_PROVIDER, "bear");
+            options.set(CatalogOptions::TOKEN, "test-token");
+            options.set("fs.oss.path-style-access", "true");
+            for (key, value) in catalog_options {
+                options.set(*key, *value);
+            }
+            let api = Arc::new(RESTApi::new(options.clone(), false).await.unwrap());
+            let file_io_cache = file_io_cache(&options);
+            let token_file_io = Arc::new(RESTTokenFileIO::new(
+                Identifier::new("db", "t"),
+                "oss://bkt/db.db/t".to_string(),
+                options,
+                api,
+                None,
+                file_io_cache,
+            ));
+            (token_file_io.build_file_io().await.unwrap(), server)
+        }
+
+        #[tokio::test]
+        async fn test_token_file_io_routes_table_files_to_io_cache() {
+            let origin = TestOss::start().await;
+            let cache = TestOss::start_sharing(&origin).await;
+            let (file_io, server) = io_cache_file_io(&origin, &cache, &[]).await;
+
+            let data =
+                "oss://bkt/db.db/t/bucket-0/data-123e4567-e89b-12d3-a456-426614174000-1.parquet";
+            file_io
+                .new_output(data)
+                .unwrap()
+                .write(Bytes::from_static(b"data"))
+                .await
+                .unwrap();
+            assert!(file_io.exists(data).await.unwrap());
+            assert_eq!(
+                file_io.new_input(data).unwrap().read().await.unwrap(),
+                "data"
+            );
+            assert_eq!(
+                cache.take_requests(),
+                ["GET bkt/db.db/t/bucket-0/data-123e4567-e89b-12d3-a456-426614174000-1.parquet"]
+            );
+            // The token's policy has no exists, so the existence check stays on origin.
+            assert_eq!(
+                origin.take_requests(),
+                [
+                    "PUT bkt/db.db/t/bucket-0/data-123e4567-e89b-12d3-a456-426614174000-1.parquet",
+                    "HEAD bkt/db.db/t/bucket-0/data-123e4567-e89b-12d3-a456-426614174000-1.parquet"
+                ]
+            );
+
+            server.abort();
+        }
+
+        #[tokio::test]
+        async fn test_token_file_io_routes_files_named_with_table_prefixes() {
+            let origin = TestOss::start().await;
+            let cache = TestOss::start_sharing(&origin).await;
+            let (file_io, server) = io_cache_file_io(&origin, &cache, &[]).await;
+
+            let data =
+                "oss://bkt/db.db/t/bucket-0/custom-123e4567-e89b-12d3-a456-426614174000-1.parquet";
+            file_io
+                .new_output(data)
+                .unwrap()
+                .write(Bytes::from_static(b"data"))
+                .await
+                .unwrap();
+            // Without the table's prefix the name is not known as a data file.
+            file_io.new_input(data).unwrap().read().await.unwrap();
+            assert!(cache.take_requests().is_empty());
+
+            let table_file_io = file_io.with_file_prefixes(["custom-"]);
+            assert_eq!(
+                table_file_io.new_input(data).unwrap().read().await.unwrap(),
+                "data"
+            );
+            assert_eq!(
+                cache.take_requests(),
+                ["GET bkt/db.db/t/bucket-0/custom-123e4567-e89b-12d3-a456-426614174000-1.parquet"]
+            );
+            // An origin-only view, as a Format Table gets, keeps the prefix but not the target.
+            table_file_io
+                .origin_only()
+                .new_input(data)
+                .unwrap()
+                .read()
+                .await
+                .unwrap();
+            assert!(cache.take_requests().is_empty());
+            server.abort();
+        }
+
+        #[tokio::test]
+        async fn test_dlf_oss_endpoint_turns_off_io_cache_routing() {
+            let origin = TestOss::start().await;
+            let cache = TestOss::start_sharing(&origin).await;
+            let (file_io, server) = io_cache_file_io(
+                &origin,
+                &cache,
+                &[(CatalogOptions::DLF_OSS_ENDPOINT, origin.endpoint())],
+            )
+            .await;
+
+            let data =
+                "oss://bkt/db.db/t/bucket-0/data-123e4567-e89b-12d3-a456-426614174000-1.parquet";
+            file_io
+                .new_output(data)
+                .unwrap()
+                .write(Bytes::from_static(b"data"))
+                .await
+                .unwrap();
+            assert!(file_io.exists(data).await.unwrap());
+            assert!(cache.take_requests().is_empty());
+            assert_eq!(origin.take_requests().len(), 2);
+            server.abort();
+        }
+
+        #[tokio::test]
+        async fn test_client_io_cache_enabled_wins_over_token() {
+            let origin = TestOss::start().await;
+            let cache = TestOss::start_sharing(&origin).await;
+            let (file_io, server) =
+                io_cache_file_io(&origin, &cache, &[("io-cache.enabled", "false")]).await;
+
+            // Like clients without routing, every request uses fs.oss.endpoint.
+            let data =
+                "oss://bkt/db.db/t/bucket-0/data-123e4567-e89b-12d3-a456-426614174000-1.parquet";
+            file_io
+                .new_output(data)
+                .unwrap()
+                .write(Bytes::from_static(b"data"))
+                .await
+                .unwrap();
+            assert_eq!(
+                file_io.new_input(data).unwrap().read().await.unwrap(),
+                "data"
+            );
+            assert_eq!(cache.take_requests().len(), 2);
+            assert!(origin.take_requests().is_empty());
+            server.abort();
+        }
+    }
 }

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

@@ -22,7 +22,7 @@ use std::collections::HashSet;
 const TEMP_FILE_SUFFIX_LEN: usize = 41;
 
 #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
-pub(super) enum FileType {
+pub(crate) enum FileType {
     Meta,
     DedicatedData,
     StructuredData,
@@ -42,7 +42,7 @@ impl FileType {
     ];
 
     /// Keeps metadata/index classification ahead of dedicated data detection.
-    pub(super) fn classify(path: &str) -> Self {
+    pub(crate) fn classify(path: &str) -> Self {
         let mut segments = path.rsplit('/');
         let name = unwrap_temp_file_name(segments.next().unwrap_or(path));
 
@@ -88,12 +88,12 @@ impl FileType {
     }
 
     /// Matches Java's BLOB/video and vector-store file naming rules.
-    fn is_dedicated_file(name: &str) -> bool {
+    pub(crate) fn is_dedicated_file(name: &str) -> bool {
         let name = name.to_ascii_lowercase();
         name.ends_with(".blob") || name.ends_with(".video") || name.contains(".vector.")
     }
 
-    pub(super) fn is_mutable(path: &str) -> bool {
+    pub(crate) fn is_mutable(path: &str) -> bool {
         let name = path.rsplit('/').next().unwrap_or(path);
         // Iceberg-compatible `version-hint.text`, `retire-pending` and `v{N}.metadata.json`
         // (on tag changes) are rewritten in place.
@@ -110,7 +110,7 @@ impl FileType {
             || name.contains(".tmp.")
     }
 
-    pub(super) fn parse_whitelist(value: &str) -> HashSet<Self> {
+    pub(crate) fn parse_whitelist(value: &str) -> HashSet<Self> {
         value
             .split(',')
             .flat_map(|name| -> &'static [Self] {

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

@@ -22,7 +22,7 @@ mod priority_lru;
 mod reader;
 mod state;
 
-use self::file_type::FileType;
+pub(crate) use self::file_type::FileType;
 use self::memory::MemoryCache;
 use self::priority_lru::DEFAULT_HIGH_PRIORITY_POOL_RATIO;
 use self::state::{BlockKey, CacheCoordinator, CacheReadToken};