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

paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/HadoopCompliantFileIO.java

@@ -26,6 +26,7 @@
 import org.apache.paimon.fs.RemoteIterator;
 import org.apache.paimon.fs.SeekableInputStream;
 import org.apache.paimon.fs.VectoredReadable;
+import org.apache.paimon.options.Options;
 import org.apache.paimon.utils.Pair;
 
 import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
@@ -37,6 +38,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
+
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.net.URI;
@@ -58,6 +61,7 @@ public abstract class HadoopCompliantFileIO implements FileIO {
     private static final Logger LOG = LoggerFactory.getLogger(HadoopCompliantFileIO.class);
 
     private static final long serialVersionUID = 1L;
+    private static final String JINDO_CACHE_RPC_ADDRESS = "fs.jindocache.namespace.rpc.address";
 
     /// Detailed cache strategies are retrieved from REST server.
     private static final String META_CACHE_ENABLED_TAG = "meta";
@@ -68,14 +72,21 @@ public abstract class HadoopCompliantFileIO implements FileIO {
     protected boolean metaCacheEnabled = false;
     protected boolean readCacheEnabled = false;
     protected boolean writeCacheEnabled = false;
+    protected boolean existsCacheEnabled = false;
 
     protected transient volatile Map<String, Pair<JindoHadoopSystem, String>> fsMap;
     protected transient volatile Map<String, Pair<JindoHadoopSystem, String>> jindoCacheFsMap;
 
+    // Non-null when endpoint routing is configured instead of JindoCache RPC.
+    @Nullable IoCacheRouting cacheRouting;
+
     // Only enable cache for path which is generated with uuid
     private List<String> cacheWhitelistPaths = new ArrayList<>();
 
     boolean shouldCache(Path path) {
+        if (cacheRouting != null) {
+            return cacheRouting.targetOf(path) != null;
+        }
         if (cacheWhitelistPaths.isEmpty()) {
             return true;
         }
@@ -90,28 +101,44 @@ boolean shouldCache(Path path) {
 
     @Override
     public void configure(CatalogContext context) {
-        // Process file io cache configuration
-        if (!context.options().get(IO_CACHE_ENABLED)
-                || context.options().get(IO_CACHE_POLICY) == null
-                || context.options().get(IO_CACHE_POLICY).contains(DISABLE_CACHE_TAG)) {
+        Options options = context.options();
+        // Configure cache for the selected backend.
+        if (!options.get(IO_CACHE_ENABLED) || options.get(IO_CACHE_POLICY) == null) {
             LOG.debug(
                     "Cache is disabled with io-cache.enabled={}, io-cache.policy={}",
-                    context.options().get(IO_CACHE_ENABLED),
-                    context.options().get(IO_CACHE_POLICY));
+                    options.get(IO_CACHE_ENABLED),
+                    options.get(IO_CACHE_POLICY));
             return;
         }
-        // Enable file io cache
-        if (context.options().get("fs.jindocache.namespace.rpc.address") == null) {
+        if (options.get(JINDO_CACHE_RPC_ADDRESS) == null) {
+            cacheRouting = IoCacheRouting.create(options);
+            if (cacheRouting == null) {
+                LOG.debug(
+                        "FileIO cache endpoint routing is not configured for io-cache.policy={}",
+                        options.get(IO_CACHE_POLICY));
+                return;
+            }
+            metaCacheEnabled = cacheRouting.metaCacheEnabled();
+            readCacheEnabled = cacheRouting.readCacheEnabled();
+            writeCacheEnabled = cacheRouting.writeCacheEnabled();
+            existsCacheEnabled = cacheRouting.existsCacheEnabled();
             LOG.info(
-                    "FileIO cache is enabled but JindoCache RPC address is not set, fallback to no-cache");
+                    "Cache endpoints enabled: meta {}, read {}, write {}, exists {}, {}",
+                    metaCacheEnabled,
+                    readCacheEnabled,
+                    writeCacheEnabled,
+                    existsCacheEnabled,
+                    cacheRouting);
         } else {
-            metaCacheEnabled =
-                    context.options().get(IO_CACHE_POLICY).contains(META_CACHE_ENABLED_TAG);
-            readCacheEnabled =
-                    context.options().get(IO_CACHE_POLICY).contains(READ_CACHE_ENABLED_TAG);
-            writeCacheEnabled =
-                    context.options().get(IO_CACHE_POLICY).contains(WRITE_CACHE_ENABLED_TAG);
-            String whitelist = context.options().get(IO_CACHE_WHITELIST_PATH);
+            // Keep legacy JindoCache policy matching unchanged; endpoint routing uses exact tokens.
+            if (options.get(IO_CACHE_POLICY).contains(DISABLE_CACHE_TAG)) {
+                return;
+            }
+            metaCacheEnabled = options.get(IO_CACHE_POLICY).contains(META_CACHE_ENABLED_TAG);
+            readCacheEnabled = options.get(IO_CACHE_POLICY).contains(READ_CACHE_ENABLED_TAG);
+            writeCacheEnabled = options.get(IO_CACHE_POLICY).contains(WRITE_CACHE_ENABLED_TAG);
+            existsCacheEnabled = metaCacheEnabled;
+            String whitelist = options.get(IO_CACHE_WHITELIST_PATH);
             if (!whitelist.equals("*")) {
                 cacheWhitelistPaths = Lists.newArrayList(whitelist.split(","));
             }
@@ -194,7 +221,7 @@ public FileStatus next() throws IOException {
     @Override
     public boolean exists(Path path) throws IOException {
         org.apache.hadoop.fs.Path hadoopPath = path(path);
-        boolean shouldCache = metaCacheEnabled && shouldCache(path);
+        boolean shouldCache = existsCacheEnabled && shouldCache(path);
         LOG.debug("Exists should cache {} for path {}", shouldCache, path);
         return getFileSystem(hadoopPath, shouldCache).exists(hadoopPath);
     }