[#12457] feat(secret): Add load*WithResolvedProperties for catalog/schema/fileset

api/src/main/java/org/apache/gravitino/SupportsCatalogs.java

@@ -59,6 +59,26 @@ public interface SupportsCatalogs {
    */
   Catalog loadCatalog(String catalogName) throws NoSuchCatalogException;
 
+  /**
+   * Loads a catalog with secret URNs resolved to plaintext in {@link Catalog#properties()}.
+   *
+   * <p>Credential-vending keys are omitted (use the credentials API). Legacy hidden plaintext
+   * secrets are omitted. Default {@link #loadCatalog(String)} property omit behavior is unchanged.
+   *
+   * <p>The returned catalog is intended for reading resolved properties. Server-side / in-process
+   * callers may receive a properties-focused view that does not support {@code asSchemas()}, {@code
+   * asTableCatalog()}, etc. HTTP clients rebuild an operable catalog from the response DTO.
+   *
+   * @param catalogName the name of the catalog
+   * @return the catalog with resolved plaintext properties
+   * @throws NoSuchCatalogException If the catalog does not exist.
+   */
+  default Catalog loadCatalogWithResolvedProperties(String catalogName)
+      throws NoSuchCatalogException {
+    throw new UnsupportedOperationException(
+        "Loading catalog with resolved properties is not supported");
+  }
+
   /**
    * Check if a catalog exists.
    *

api/src/main/java/org/apache/gravitino/SupportsSchemas.java

@@ -112,6 +112,20 @@ Schema createSchema(String schemaName, String comment, Map<String, String> prope
    */
   Schema loadSchema(String schemaName) throws NoSuchSchemaException;
 
+  /**
+   * Loads a schema with secret URNs resolved to plaintext in {@link Schema#properties()}.
+   *
+   * <p>The returned schema is intended for reading resolved properties.
+   *
+   * @param schemaName The name of the schema.
+   * @return The schema with resolved plaintext properties.
+   * @throws NoSuchSchemaException If the schema does not exist.
+   */
+  default Schema loadSchemaWithResolvedProperties(String schemaName) throws NoSuchSchemaException {
+    throw new UnsupportedOperationException(
+        "Loading schema with resolved properties is not supported by this catalog");
+  }
+
   /**
    * Apply the metadata change to a schema in the catalog.
    *

api/src/main/java/org/apache/gravitino/file/FilesetCatalog.java

@@ -59,6 +59,19 @@ public interface FilesetCatalog {
    */
   Fileset loadFileset(NameIdentifier ident) throws NoSuchFilesetException;
 
+  /**
+   * Loads a fileset with secret URNs resolved to plaintext in {@link Fileset#properties()}.
+   *
+   * @param ident A fileset identifier.
+   * @return The fileset with resolved plaintext properties.
+   * @throws NoSuchFilesetException If the fileset does not exist.
+   */
+  default Fileset loadFilesetWithResolvedProperties(NameIdentifier ident)
+      throws NoSuchFilesetException {
+    throw new UnsupportedOperationException(
+        "Loading fileset with resolved properties is not supported by this catalog");
+  }
+
   /**
    * Check if a fileset exists using an {@link NameIdentifier} from the catalog.
    *

clients/client-java/src/main/java/org/apache/gravitino/client/BaseSchemaCatalog.java

@@ -217,6 +217,27 @@ public Schema loadSchema(String schemaName) throws NoSuchSchemaException {
     return new GenericSchema(resp.getSchema(), restClient, catalogNamespace.level(0), this.name());
   }
 
+  /**
+   * Load the schema with secret URNs resolved to plaintext in {@link Schema#properties()}.
+   *
+   * @param schemaName The name of the schema.
+   * @return The schema with resolved plaintext properties.
+   * @throws NoSuchSchemaException if the schema with specified identifier does not exist.
+   */
+  @Override
+  public Schema loadSchemaWithResolvedProperties(String schemaName) throws NoSuchSchemaException {
+    SchemaResponse resp =
+        restClient.get(
+            formatSchemaRequestPath(schemaNamespace()) + "/" + RESTUtils.encodeString(schemaName),
+            Collections.singletonMap("view", "resolved"),
+            SchemaResponse.class,
+            Collections.emptyMap(),
+            ErrorHandlers.schemaErrorHandler());
+    resp.validate();
+
+    return new GenericSchema(resp.getSchema(), restClient, catalogNamespace.level(0), this.name());
+  }
+
   /**
    * Alter the schema with specified identifier by applying the changes.
    *

clients/client-java/src/main/java/org/apache/gravitino/client/FilesetCatalog.java

@@ -129,6 +129,30 @@ public Fileset loadFileset(NameIdentifier ident) throws NoSuchFilesetException {
     return new GenericFileset(resp.getFileset(), restClient, fullNamespace);
   }
 
+  /**
+   * Load the fileset with secret URNs resolved to plaintext in {@link Fileset#properties()}.
+   *
+   * @param ident A fileset identifier in {@code schema.fileset} format.
+   * @return The fileset with resolved plaintext properties.
+   * @throws NoSuchFilesetException If the fileset does not exist.
+   */
+  @Override
+  public Fileset loadFilesetWithResolvedProperties(NameIdentifier ident)
+      throws NoSuchFilesetException {
+    checkFilesetNameIdentifier(ident);
+
+    Namespace fullNamespace = getFilesetFullNamespace(ident.namespace());
+    FilesetResponse resp =
+        restClient.get(
+            formatFilesetRequestPath(fullNamespace) + "/" + RESTUtils.encodeString(ident.name()),
+            Collections.singletonMap("view", "resolved"),
+            FilesetResponse.class,
+            Collections.emptyMap(),
+            ErrorHandlers.filesetErrorHandler());
+    resp.validate();
+    return new GenericFileset(resp.getFileset(), restClient, fullNamespace);
+  }
+
   /**
    * Create a fileset metadata with multiple storage locations in the catalog.
    *

clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClient.java

@@ -132,6 +132,12 @@ public Catalog loadCatalog(String catalogName) throws NoSuchCatalogException {
     return getMetalake().loadCatalog(catalogName);
   }
 
+  @Override
+  public Catalog loadCatalogWithResolvedProperties(String catalogName)
+      throws NoSuchCatalogException {
+    return getMetalake().loadCatalogWithResolvedProperties(catalogName);
+  }
+
   @Override
   public Catalog createCatalog(
       String catalogName,

clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoMetalake.java

@@ -236,6 +236,31 @@ public Catalog loadCatalog(String catalogName) throws NoSuchCatalogException {
     return DTOConverters.toCatalog(this.name(), resp.getCatalog(), restClient);
   }
 
+  /**
+   * Load the catalog with secret URNs resolved to plaintext in {@link Catalog#properties()}.
+   *
+   * @param catalogName The name of the catalog.
+   * @return The catalog with resolved plaintext properties.
+   * @throws NoSuchCatalogException if the catalog with specified name does not exist.
+   */
+  @Override
+  public Catalog loadCatalogWithResolvedProperties(String catalogName)
+      throws NoSuchCatalogException {
+    CatalogResponse resp =
+        restClient.get(
+            String.format(
+                API_METALAKES_CATALOGS_PATH,
+                RESTUtils.encodeString(this.name()),
+                RESTUtils.encodeString(catalogName)),
+            ImmutableMap.of("view", "resolved"),
+            CatalogResponse.class,
+            Collections.emptyMap(),
+            ErrorHandlers.catalogErrorHandler());
+    resp.validate();
+
+    return DTOConverters.toCatalog(this.name(), resp.getCatalog(), restClient);
+  }
+
   /**
    * Create a new catalog with specified identifier, type, comment and properties.
    *

clients/client-python/gravitino/client/base_schema_catalog.py

@@ -220,6 +220,34 @@ def load_schema(self, schema_name: str) -> Schema:
             self._name,
         )
 
+    def load_schema_with_resolved_properties(self, schema_name: str) -> Schema:
+        """Load schema with secret URNs resolved to plaintext in properties.
+
+        Args:
+            schema_name: The name of the schema.
+
+        Raises:
+            NoSuchSchemaException if the schema with specified identifier does not exist.
+
+        Returns:
+            The schema with resolved plaintext properties.
+        """
+        resp = self.rest_client.get(
+            BaseSchemaCatalog.format_schema_request_path(self._schema_namespace())
+            + "/"
+            + encode_string(schema_name),
+            params={"view": "resolved"},
+            error_handler=SCHEMA_ERROR_HANDLER,
+        )
+        schema_resp = SchemaResponse.from_json(resp.body, infer_missing=True)
+        schema_resp.validate()
+        return GenericSchema(
+            schema_resp.schema(),
+            self.rest_client,
+            self._catalog_namespace.level(0),
+            self._name,
+        )
+
     def alter_schema(self, schema_name: str, *changes: SchemaChange) -> Schema:
         """Alter the schema with specified identifier by applying the changes.
 

clients/client-python/gravitino/client/fileset_catalog.py

@@ -136,6 +136,32 @@ def load_fileset(self, ident: NameIdentifier) -> Fileset:
 
         return GenericFileset(fileset_resp.fileset(), self.rest_client, full_namespace)
 
+    def load_fileset_with_resolved_properties(self, ident: NameIdentifier) -> Fileset:
+        """Load fileset with secret URNs resolved to plaintext in properties.
+
+        Args:
+            ident: A fileset identifier in schema.fileset format.
+
+        Raises:
+            NoSuchFilesetException If the fileset does not exist.
+
+        Returns:
+            The fileset with resolved plaintext properties.
+        """
+        self.check_fileset_name_identifier(ident)
+
+        full_namespace = self._get_fileset_full_namespace(ident.namespace())
+
+        resp = self.rest_client.get(
+            f"{self.format_fileset_request_path(full_namespace)}/"
+            f"{encode_string(ident.name())}",
+            params={"view": "resolved"},
+            error_handler=FILESET_ERROR_HANDLER,
+        )
+        fileset_resp = FilesetResponse.from_json(resp.body, infer_missing=True)
+        fileset_resp.validate()
+        return GenericFileset(fileset_resp.fileset(), self.rest_client, full_namespace)
+
     def create_fileset(
         self,
         ident: NameIdentifier,

clients/client-python/gravitino/client/gravitino_client.py

@@ -96,6 +96,9 @@ def list_catalogs_info(self) -> List[Catalog]:
     def load_catalog(self, name: str) -> Catalog:
         return self.get_metalake().load_catalog(name)
 
+    def load_catalog_with_resolved_properties(self, name: str) -> Catalog:
+        return self.get_metalake().load_catalog_with_resolved_properties(name)
+
     def create_catalog(
         self,
         name: str,

clients/client-python/gravitino/client/gravitino_metalake.py

@@ -210,6 +210,32 @@ def load_catalog(self, name: str) -> Catalog:
             self.name(), catalog_resp.catalog(), self.rest_client
         )
 
+    def load_catalog_with_resolved_properties(self, name: str) -> Catalog:
+        """Load catalog with secret URNs resolved to plaintext in properties.
+
+        Args:
+            name: The name of the catalog.
+
+        Raises:
+            NoSuchCatalogException if the catalog with specified name does not exist.
+
+        Returns:
+            The catalog with resolved plaintext properties.
+        """
+        url = self.API_METALAKES_CATALOGS_PATH.format(
+            encode_string(self.name()), encode_string(name)
+        )
+        response = self.rest_client.get(
+            url,
+            params={"view": "resolved"},
+            error_handler=CATALOG_ERROR_HANDLER,
+        )
+        catalog_resp = CatalogResponse.from_json(response.body, infer_missing=True)
+        catalog_resp.validate()
+        return DTOConverters.to_catalog(
+            self.name(), catalog_resp.catalog(), self.rest_client
+        )
+
     def create_catalog(
         self,
         name: str,

clients/client-python/gravitino/filesystem/gvfs_base_operations.py

@@ -495,18 +495,27 @@ def _get_actual_filesystem(
 
     def _merge_fileset_properties(
         self,
-        catalog: FilesetCatalog,
-        schema: Schema,
-        fileset: Fileset,
+        fileset_ident: NameIdentifier,
         actual_location: str,
     ) -> Dict[str, str]:
-        """Merge properties from catalog, schema, fileset, options, and user-defined configs.
-        :param catalog: The fileset catalog
-        :param schema: The schema
-        :param fileset: The fileset
+        """Merge resolved properties from catalog, schema, fileset, options, and configs.
+
+        Uses load_*_with_resolved_properties so secret URNs become plaintext for FS access.
+        Credential-vending keys remain omitted from those APIs.
+
+        :param fileset_ident: The fileset identifier
         :param actual_location: The actual storage location
         :return: Merged properties dictionary
         """
+        catalog_name = fileset_ident.namespace().level(1)
+        schema_name = fileset_ident.namespace().level(2)
+        catalog = self._get_gravitino_client().load_catalog_with_resolved_properties(
+            catalog_name
+        )
+        schema = catalog.as_schemas().load_schema_with_resolved_properties(schema_name)
+        fileset = catalog.as_fileset_catalog().load_fileset_with_resolved_properties(
+            NameIdentifier.of(schema_name, fileset_ident.name())
+        )
         fileset_props = dict(catalog.properties() or {})
         fileset_props.update(schema.properties() or {})
         fileset_props.update(fileset.properties() or {})
@@ -530,13 +539,6 @@ def _get_actual_filesystem_by_location_name(
         :param location_name: The location name, None means the default location
         :return: The actual filesystem
         """
-        catalog_ident: NameIdentifier = NameIdentifier.of(
-            self._metalake, fileset_ident.namespace().level(1)
-        )
-        catalog = self._get_fileset_catalog(catalog_ident)
-        schema = self._get_fileset_schema(
-            NameIdentifier.parse(str(fileset_ident.namespace()))
-        )
         fileset = self._get_fileset(fileset_ident)
 
         # Determine target location name
@@ -553,9 +555,7 @@ def _get_actual_filesystem_by_location_name(
                 f"Cannot find the location: {target_location_name} in fileset: {fileset_ident}"
             )
 
-        fileset_props = self._merge_fileset_properties(
-            catalog, schema, fileset, actual_location
-        )
+        fileset_props = self._merge_fileset_properties(fileset_ident, actual_location)
 
         # Set caller context for credential vending
         if location_name:

clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/BaseGVFSOperations.java

@@ -726,7 +726,7 @@ protected FileSystem getActualFileSystemByLocationName(
           filesetIdent);
 
       Path targetLocation = new Path(fileset.storageLocations().get(targetLocationName));
-      Map<String, String> allProperties = getAllProperties(filesetIdent, fileset.properties());
+      Map<String, String> allProperties = getAllProperties(filesetIdent);
       allProperties.putAll(
           FilesetUtil.getUserDefinedFileSystemConfigs(
               targetLocation.toUri(), allProperties, FS_GRAVITINO_PATH_CONFIG_PREFIX));
@@ -961,19 +961,25 @@ private Cache<FileSystemCacheKey, FileSystem> newFileSystemCache(Configuration c
     return cacheBuilder.build();
   }
 
-  private Map<String, String> getAllProperties(
-      NameIdentifier filesetIdent, Map<String, String> filesetProperties) {
+  private Map<String, String> getAllProperties(NameIdentifier filesetIdent) {
     Map<String, String> allProperties = new HashMap<>();
-    Catalog catalog =
-        (Catalog)
-            getFilesetCatalog(
-                NameIdentifier.of(
-                    filesetIdent.namespace().level(0), filesetIdent.namespace().level(1)));
-    allProperties.putAll(catalog.properties());
-
-    Schema schema = getSchema(NameIdentifier.parse(filesetIdent.namespace().toString()));
-    allProperties.putAll(schema.properties());
-    allProperties.putAll(filesetProperties);
+    String catalogName = filesetIdent.namespace().level(1);
+    String schemaName = filesetIdent.namespace().level(2);
+    Catalog catalog = getGravitinoClient().loadCatalogWithResolvedProperties(catalogName);
+    if (catalog.properties() != null) {
+      allProperties.putAll(catalog.properties());
+    }
+    Schema schema = catalog.asSchemas().loadSchemaWithResolvedProperties(schemaName);
+    if (schema.properties() != null) {
+      allProperties.putAll(schema.properties());
+    }
+    Fileset resolvedFileset =
+        catalog
+            .asFilesetCatalog()
+            .loadFilesetWithResolvedProperties(NameIdentifier.of(schemaName, filesetIdent.name()));
+    if (resolvedFileset.properties() != null) {
+      allProperties.putAll(resolvedFileset.properties());
+    }
     allProperties.putAll(extractNonDefaultConfig(conf));
     return allProperties;
   }