[#4888] improvement(core): Add the implementations for getFileLocation interface in core module

core/src/main/java/org/apache/gravitino/catalog/FilesetNormalizeDispatcher.java

@@ -96,7 +96,9 @@ public boolean dropFileset(NameIdentifier ident) {
 
   @Override
   public String getFileLocation(NameIdentifier ident, String subPath) {
-    throw new UnsupportedOperationException("Not implemented");
+    // The constraints of the name spec may be more strict than underlying catalog,
+    // and for compatibility reasons, we only apply case-sensitive capabilities here.
+    return dispatcher.getFileLocation(normalizeCaseSensitive(ident), subPath);
   }
 
   private NameIdentifier normalizeNameIdentifier(NameIdentifier ident) {

core/src/main/java/org/apache/gravitino/catalog/FilesetOperationDispatcher.java

@@ -208,6 +208,9 @@ public boolean dropFileset(NameIdentifier ident) {
   @Override
   public String getFileLocation(NameIdentifier ident, String subPath)
       throws NoSuchFilesetException {
-    throw new UnsupportedOperationException("Not implemented");
+    return doWithCatalog(
+        getCatalogIdentifier(ident),
+        c -> c.doWithFilesetOps(f -> f.getFileLocation(ident, subPath)),
+        NonEmptyEntityException.class);
   }
 }

core/src/main/java/org/apache/gravitino/hook/FilesetHookDispatcher.java

@@ -97,6 +97,6 @@ public boolean filesetExists(NameIdentifier ident) {
   @Override
   public String getFileLocation(NameIdentifier ident, String subPath)
       throws NoSuchFilesetException {
-    throw new UnsupportedOperationException("Not implemented");
+    return dispatcher.getFileLocation(ident, subPath);
   }
 }

core/src/main/java/org/apache/gravitino/listener/FilesetEventDispatcher.java

@@ -19,9 +19,11 @@
 
 package org.apache.gravitino.listener;
 
+import com.google.common.collect.ImmutableMap;
 import java.util.Map;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.Namespace;
+import org.apache.gravitino.audit.CallerContext;
 import org.apache.gravitino.catalog.FilesetDispatcher;
 import org.apache.gravitino.exceptions.FilesetAlreadyExistsException;
 import org.apache.gravitino.exceptions.NoSuchFilesetException;
@@ -34,6 +36,8 @@
 import org.apache.gravitino.listener.api.event.CreateFilesetFailureEvent;
 import org.apache.gravitino.listener.api.event.DropFilesetEvent;
 import org.apache.gravitino.listener.api.event.DropFilesetFailureEvent;
+import org.apache.gravitino.listener.api.event.GetFileLocationEvent;
+import org.apache.gravitino.listener.api.event.GetFileLocationFailureEvent;
 import org.apache.gravitino.listener.api.event.ListFilesetEvent;
 import org.apache.gravitino.listener.api.event.ListFilesetFailureEvent;
 import org.apache.gravitino.listener.api.event.LoadFilesetEvent;
@@ -142,6 +146,24 @@ public boolean dropFileset(NameIdentifier ident) {
   @Override
   public String getFileLocation(NameIdentifier ident, String subPath)
       throws NoSuchFilesetException {
-    throw new UnsupportedOperationException("Not implemented");
+    try {
+      String actualFileLocation = dispatcher.getFileLocation(ident, subPath);
+      // get the audit info from the thread local context
+      CallerContext callerContext = CallerContext.CallerContextHolder.get();
+      ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+      builder.putAll(callerContext.context());
+      eventBus.dispatchEvent(
+          new GetFileLocationEvent(
+              PrincipalUtils.getCurrentUserName(),
+              ident,
+              actualFileLocation,
+              subPath,
+              builder.build()));
+      return actualFileLocation;
+    } catch (Exception e) {
+      eventBus.dispatchEvent(
+          new GetFileLocationFailureEvent(PrincipalUtils.getCurrentUserName(), ident, subPath, e));
+      throw e;
+    }
   }
 }

core/src/main/java/org/apache/gravitino/listener/api/event/GetFileLocationEvent.java

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ *  regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ *  under the License.
+ */
+
+package org.apache.gravitino.listener.api.event;
+
+import java.util.Map;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.annotation.DeveloperApi;
+
+/** Represents an event that occurs when getting an actual file location. */
+@DeveloperApi
+public final class GetFileLocationEvent extends FilesetEvent {
+  private final String actualFileLocation;
+  private final String subPath;
+  private final Map<String, String> context;
+
+  /**
+   * Constructs a new {@code GetFileLocationEvent}, recording the attempt to get a file location.
+   *
+   * @param user The user who initiated the get file location.
+   * @param identifier The identifier of the file location that was attempted to be got.
+   * @param actualFileLocation The actual file location which want to get.
+   * @param subPath The accessing sub path of the get file location operation.
+   * @param context The audit context, this param can be null.
+   */
+  public GetFileLocationEvent(
+      String user,
+      NameIdentifier identifier,
+      String actualFileLocation,
+      String subPath,
+      Map<String, String> context) {
+    super(user, identifier);
+    this.actualFileLocation = actualFileLocation;
+    this.subPath = subPath;
+    this.context = context;
+  }
+
+  /**
+   * Get the actual file location after processing of the get file location operation.
+   *
+   * @return The actual file location.
+   */
+  public String actualFileLocation() {
+    return actualFileLocation;
+  }
+
+  /**
+   * Get the accessing sub path of the get file location operation.
+   *
+   * @return The accessing sub path.
+   */
+  public String subPath() {
+    return subPath;
+  }
+
+  /**
+   * Get the audit context map of the get file location operation.
+   *
+   * @return The audit context map.
+   */
+  public Map<String, String> context() {
+    return context;
+  }
+}

core/src/main/java/org/apache/gravitino/listener/api/event/GetFileLocationFailureEvent.java

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ *  regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ *  under the License.
+ */
+
+package org.apache.gravitino.listener.api.event;
+
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.annotation.DeveloperApi;
+
+/**
+ * Represents an event that is generated when an attempt to get a file location from the system
+ * fails.
+ */
+@DeveloperApi
+public final class GetFileLocationFailureEvent extends FilesetFailureEvent {
+  private final String subPath;
+
+  /**
+   * Constructs a new {@code GetFileLocationFailureEvent}.
+   *
+   * @param user The user who initiated the get a file location.
+   * @param identifier The identifier of the file location that was attempted to be got.
+   * @param subPath The sub path of the actual file location which want to get.
+   * @param exception The exception that was thrown during the get a file location. This exception
+   *     is key to diagnosing the failure, providing insights into what went wrong during the
+   *     operation.
+   */
+  public GetFileLocationFailureEvent(
+      String user, NameIdentifier identifier, String subPath, Exception exception) {
+    super(user, identifier, exception);
+    this.subPath = subPath;
+  }
+
+  /**
+   * Get the audit context map of the get file location operation.
+   *
+   * @return The audit context map.
+   */
+  public String subPath() {
+    return subPath;
+  }
+}

core/src/test/java/org/apache/gravitino/catalog/TestFilesetOperationDispatcher.java

@@ -21,8 +21,10 @@
 import static org.apache.gravitino.StringIdentifier.ID_KEY;
 
 import com.google.common.collect.ImmutableMap;
+import java.io.File;
 import java.io.IOException;
 import java.util.Map;
+import java.util.UUID;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.Namespace;
 import org.apache.gravitino.file.Fileset;
@@ -169,4 +171,53 @@ public void testCreateAndDropFileset() {
     Assertions.assertTrue(dropped);
     Assertions.assertFalse(filesetOperationDispatcher.dropFileset(filesetIdent1));
   }
+
+  @Test
+  public void testCreateAndGetFileLocation() {
+    String tmpDir = "/tmp/test_get_file_location_" + UUID.randomUUID();
+    try {
+      Namespace filesetNs = Namespace.of(metalake, catalog, "schema1024");
+      Map<String, String> props = ImmutableMap.of("k1", "v1", "location", "schema1024");
+      schemaOperationDispatcher.createSchema(
+          NameIdentifier.of(filesetNs.levels()), "comment", props);
+
+      NameIdentifier filesetIdent1 = NameIdentifier.of(filesetNs, "fileset1024");
+      Fileset fileset1 =
+          filesetOperationDispatcher.createFileset(
+              filesetIdent1, "comment", Fileset.Type.MANAGED, tmpDir, props);
+      Assertions.assertEquals("fileset1024", fileset1.name());
+      Assertions.assertEquals("comment", fileset1.comment());
+      testProperties(props, fileset1.properties());
+      Assertions.assertEquals(Fileset.Type.MANAGED, fileset1.type());
+      Assertions.assertNotNull(fileset1.storageLocation());
+
+      // test sub path starts with "/"
+      String subPath1 = "/test/test.parquet";
+      String fileLocation1 = filesetOperationDispatcher.getFileLocation(filesetIdent1, subPath1);
+      Assertions.assertEquals(
+          String.format("%s%s", fileset1.storageLocation(), subPath1), fileLocation1);
+
+      // test sub path not starts with "/"
+      String subPath2 = "test/test.parquet";
+      String fileLocation2 = filesetOperationDispatcher.getFileLocation(filesetIdent1, subPath2);
+      Assertions.assertEquals(
+          String.format("%s/%s", fileset1.storageLocation(), subPath2), fileLocation2);
+
+      // test sub path is null
+      String subPath3 = null;
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () -> filesetOperationDispatcher.getFileLocation(filesetIdent1, subPath3));
+
+      // test sub path is blank but not null
+      String subPath4 = "";
+      String fileLocation3 = filesetOperationDispatcher.getFileLocation(filesetIdent1, subPath4);
+      Assertions.assertEquals(fileset1.storageLocation(), fileLocation3);
+    } finally {
+      File path = new File(tmpDir);
+      if (path.exists()) {
+        path.delete();
+      }
+    }
+  }
 }

core/src/test/java/org/apache/gravitino/connector/TestCatalogOperations.java

@@ -18,11 +18,13 @@
  */
 package org.apache.gravitino.connector;
 
+import com.google.common.base.Preconditions;
 import com.google.common.collect.Maps;
 import java.io.IOException;
 import java.time.Instant;
 import java.util.HashMap;
 import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.gravitino.Catalog;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.Namespace;
@@ -75,6 +77,8 @@ public class TestCatalogOperations
 
   public static final String FAIL_TEST = "need-fail";
 
+  private static final String SLASH = "/";
+
   public TestCatalogOperations(Map<String, String> config) {
     tables = Maps.newHashMap();
     schemas = Maps.newHashMap();
@@ -432,7 +436,28 @@ public boolean dropFileset(NameIdentifier ident) {
 
   @Override
   public String getFileLocation(NameIdentifier ident, String subPath) {
-    throw new UnsupportedOperationException("Not implemented");
+    Preconditions.checkArgument(subPath != null, "subPath must not be null");
+    String processedSubPath;
+    if (!subPath.trim().isEmpty() && !subPath.trim().startsWith(SLASH)) {
+      processedSubPath = SLASH + subPath.trim();
+    } else {
+      processedSubPath = subPath.trim();
+    }
+
+    Fileset fileset = loadFileset(ident);
+
+    String fileLocation;
+    // subPath cannot be null, so we only need check if it is blank
+    if (StringUtils.isBlank(processedSubPath)) {
+      fileLocation = fileset.storageLocation();
+    } else {
+      String storageLocation =
+          fileset.storageLocation().endsWith(SLASH)
+              ? fileset.storageLocation().substring(0, fileset.storageLocation().length() - 1)
+              : fileset.storageLocation();
+      fileLocation = String.format("%s%s", storageLocation, processedSubPath);
+    }
+    return fileLocation;
   }
 
   @Override