[#10559] feat(trino-connector): Forward Trino session credentials to Gravitino server

docs/trino-connector/authentication.md

@@ -31,7 +31,7 @@ gravitino.user=admin
 
 | Property                          | Description                                                          | Default value    | Required                               | Since version |
 |-----------------------------------|----------------------------------------------------------------------|------------------|----------------------------------------|---------------|
-| `gravitino.client.authType`       | Authentication type: `simple`, `oauth2`, or `kerberos`              | (none)           | No                                     | 1.3.0         |
+| `gravitino.client.authType`       | Authentication type: `simple`, `oauth2`, or `kerberos`                 | (none)        | No                                     | 1.3.0         |
 | `gravitino.user`                  | Username for simple authentication                                   | (none)           | No (uses system user if not specified) | 1.3.0         |
 
 ### OAuth2 Authentication
@@ -57,7 +57,7 @@ gravitino.client.oauth2.scope=gravitino
 
 | Property                             | Description                                            | Default value | Required                    | Since version |
 |--------------------------------------|--------------------------------------------------------|---------------|-----------------------------|---------------|
-| `gravitino.client.authType`          | Authentication type: `simple`, `oauth2`, or `kerberos` | (none)        | Yes (to enable OAuth2)      | 1.3.0         |
+| `gravitino.client.authType`          | Authentication type: `simple`, `oauth2`, or `kerberos` | (none)        | Yes (to enable OAuth2)           | 1.3.0         |
 | `gravitino.client.oauth2.serverUri`  | OAuth2 server URI                                      | (none)        | Yes if authType is `oauth2` | 1.3.0         |
 | `gravitino.client.oauth2.credential` | OAuth2 credentials in format `client_id:client_secret` | (none)        | Yes if authType is `oauth2` | 1.3.0         |
 | `gravitino.client.oauth2.path`       | OAuth2 token endpoint path                             | (none)        | Yes if authType is `oauth2` | 1.3.0         |
@@ -88,6 +88,7 @@ gravitino.client.kerberos.keytabFilePath=/path/to/user.keytab
 | `gravitino.client.kerberos.principal`      | Kerberos principal  | (none)        | Yes if authType is `kerberos`           | 1.3.0         |
 | `gravitino.client.kerberos.keytabFilePath` | Path to keytab file | (none)        | No (uses ticket cache if not specified) | 1.3.0         |
 
+
 ### Example: Connecting to OAuth-protected Gravitino Server
 
 This example shows how to configure the Trino connector to connect to a Gravitino server protected by OAuth authentication.
@@ -123,6 +124,29 @@ gravitino.client.oauth2.scope=test
 SHOW CATALOGS;
 ```
 
+### Session Credential Forwarding
+
+Setting `gravitino.client.session.forwardUser=true` with `authType=simple` creates a dedicated Gravitino client per Trino session user, so each user is visible in the Gravitino audit log instead of the shared `gravitino.user`.
+
+**Configuration:**
+
+```properties
+connector.name=gravitino
+gravitino.metalake=metalake
+gravitino.uri=http://localhost:8090
+
+gravitino.client.authType=simple
+gravitino.client.session.forwardUser=true
+```
+
+**Configuration properties:**
+
+| Property                                                   | Description                                                                                | Default value | Required | Since version |
+|------------------------------------------------------------|--------------------------------------------------------------------------------------------|---------------|----------|---------------|
+| `gravitino.client.session.forwardUser`                     | When `true` with `authType=simple`, forwards the Trino session user to Gravitino per-query | `false`       | No       | 1.3.0         |
+| `gravitino.client.session.cache.maxSize`                   | Maximum number of per-user sessions to keep in the cache                                   | `500`         | No       | 1.3.0         |
+| `gravitino.client.session.cache.expireAfterAccessSeconds`  | Seconds before an idle per-user session is evicted from the cache                          | `3600`        | No       | 1.3.0         |
+
 ### Notes
 
 - The Gravitino server must be configured with the corresponding authentication mechanism enabled.

integration-test-common/docker-script/init/trino/config/catalog/gravitino.properties

@@ -21,3 +21,5 @@ connector.name = gravitino
 gravitino.uri = http://GRAVITINO_HOST_IP:GRAVITINO_HOST_PORT
 gravitino.metalake = GRAVITINO_METALAKE_NAME
 gravitino.trino.skip-version-validation=true
+gravitino.client.authType = simple
+gravitino.client.session.forwardUser = true

trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TrinoQueryTestTool.java

@@ -20,6 +20,7 @@
 
 import java.io.File;
 import java.util.Arrays;
+import java.util.concurrent.CountDownLatch;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.CommandLineParser;
 import org.apache.commons.cli.HelpFormatter;
@@ -102,6 +103,11 @@ public static void main(String[] args) throws Exception {
               + "The JAR file under ${trino_connector_dir} will be copied into the test image, "
               + "the default value is ${project_root}/trino-connector/trino-connector/build/libs.");
 
+      options.addOption(
+          "env_only",
+          false,
+          "Start the environment (Gravitino + Trino) and keep it running for manual testing. Press Ctrl+C to shutdown.");
+
       options.addOption("help", false, "Print this help message");
 
       CommandLineParser parser = new PosixParser();
@@ -254,6 +260,26 @@ public static void main(String[] args) throws Exception {
         return;
       }
 
+      if (commandLine.hasOption("env_only")) {
+        CountDownLatch shutdownLatch = new CountDownLatch(1);
+        Runtime.getRuntime().addShutdownHook(new Thread(shutdownLatch::countDown));
+
+        System.out.println("=======================================================");
+        System.out.println("Environment is ready for manual testing.");
+        System.out.println("  Gravitino URI : " + TrinoQueryITBase.gravitinoUri);
+        System.out.println("  Trino URI     : " + TrinoQueryITBase.trinoUri);
+        System.out.println("Connect to Trino CLI:");
+        System.out.println("  docker exec -it trino-ci-trino trino");
+        System.out.println("Press Ctrl+C to shutdown the environment.");
+        System.out.println("=======================================================");
+        try {
+          shutdownLatch.await();
+        } catch (InterruptedException e) {
+          Thread.currentThread().interrupt();
+        }
+        return;
+      }
+
       if (testSet == null) {
         testerRunner.testSql();
       } else {

trino-connector/integration-test/trino-test-tools/trino_integration_test.sh

@@ -26,6 +26,35 @@ export HADOOP_USER_NAME=anonymous
 echo $GRAVITINO_ROOT_DIR
 cd $GRAVITINO_ROOT_DIR
 
+PID_FILE="$GRAVITINO_ROOT_DIR/integration-test-common/build/trino-test-env.pid"
+
+if [ "$1" = "--stop" ]; then
+  if [ -f "$PID_FILE" ]; then
+    GRADLE_PID=$(cat "$PID_FILE")
+    if kill -0 "$GRADLE_PID" 2>/dev/null; then
+      PGID=$(ps -o pgid= -p "$GRADLE_PID" 2>/dev/null | tr -d ' ')
+      if [[ -n "$PGID" && "$PGID" =~ ^[0-9]+$ ]]; then
+        echo "Stopping environment (PGID: $PGID)..."
+        kill -TERM -- "-$PGID" 2>/dev/null
+        for i in $(seq 1 60); do
+          kill -0 "$GRADLE_PID" 2>/dev/null || break
+          sleep 1
+        done
+      else
+        echo "Could not determine process group for PID $GRADLE_PID, stopping containers directly..."
+      fi
+    else
+      echo "Environment process ($GRADLE_PID) is no longer running, stopping containers directly..."
+    fi
+    "$GRAVITINO_ROOT_DIR/integration-test-common/docker-script/shutdown.sh"
+    rm -f "$PID_FILE"
+  else
+    echo "No running environment found, stopping containers directly..."
+    "$GRAVITINO_ROOT_DIR/integration-test-common/docker-script/shutdown.sh"
+  fi
+  exit 0
+fi
+
 # Parse --auto_patch and --trino_version from arguments.
 # --auto_patch is consumed here and not forwarded to Gradle.
 # --trino_version is forwarded to Gradle and also used for patch selection.
@@ -106,4 +135,34 @@ if [ "$auto_patch" = true ]; then
     fi
 fi
 
-./gradlew :trino-connector:integration-test:TrinoTest -PappArgs="\"$args\""
+if echo "$args" | grep -q -- '--env_only'; then
+  LOG_FILE="$GRAVITINO_ROOT_DIR/integration-test-common/build/trino-test-env.log"
+  mkdir -p "$(dirname "$PID_FILE")" "$(dirname "$LOG_FILE")"
+
+  ./gradlew :trino-connector:integration-test:TrinoTest -PappArgs="\"$args\"" > "$LOG_FILE" 2>&1 &
+  GRADLE_PID=$!
+  echo $GRADLE_PID > "$PID_FILE"
+
+  echo "Starting environment, please wait..."
+  while kill -0 "$GRADLE_PID" 2>/dev/null; do
+    if grep -q "Press Ctrl+C to shutdown" "$LOG_FILE" 2>/dev/null; then
+      break
+    fi
+    sleep 1
+  done
+
+  if ! kill -0 "$GRADLE_PID" 2>/dev/null; then
+    echo "Environment failed to start. Check logs at: $LOG_FILE"
+    rm -f "$PID_FILE"
+    exit 1
+  fi
+
+  grep -A 6 "=======================================================" "$LOG_FILE" | head -8
+  echo ""
+  echo "Environment is running in background (PID: $GRADLE_PID)"
+  echo "Logs  : $LOG_FILE"
+  echo "Stop  : ./trino-connector/integration-test/trino-test-tools/trino_integration_test.sh --stop"
+  exit 0
+else
+  ./gradlew :trino-connector:integration-test:TrinoTest -PappArgs="\"$args\""
+fi

trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConfig.java

@@ -33,6 +33,7 @@
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.gravitino.trino.connector.security.GravitinoAuthProvider;
 
 /** Gravitino config. */
 public class GravitinoConfig {
@@ -151,6 +152,20 @@ public class GravitinoConfig {
           "",
           false);
 
+  private static final ConfigEntry GRAVITINO_SESSION_CACHE_MAX_SIZE =
+      new ConfigEntry(
+          GravitinoAuthProvider.SESSION_CACHE_MAX_SIZE_KEY,
+          "Maximum number of per-user sessions to keep in the cache when session.forwardUser=true",
+          "500",
+          false);
+
+  private static final ConfigEntry GRAVITINO_SESSION_CACHE_EXPIRE_AFTER_ACCESS_SECONDS =
+      new ConfigEntry(
+          GravitinoAuthProvider.SESSION_CACHE_EXPIRE_AFTER_ACCESS_SECONDS_KEY,
+          "Seconds before an idle per-user session is evicted from the cache when session.forwardUser=true",
+          "3600",
+          false);
+
   /**
    * Constructs a new GravitinoConfig with the specified configuration.
    *
@@ -389,6 +404,46 @@ private List<Pattern> initSkipCatalogPatterns() {
         .collect(Collectors.toList());
   }
 
+  /**
+   * Returns whether Trino session user forwarding is enabled.
+   *
+   * @return true if forwardUser is set to true
+   */
+  public boolean isForwardUser() {
+    return Boolean.parseBoolean(
+        config.getOrDefault(GravitinoAuthProvider.FORWARD_SESSION_USER_KEY, "false"));
+  }
+
+  /**
+   * Retrieves the maximum number of per-user sessions to keep in the cache.
+   *
+   * @return the session cache maximum size
+   */
+  public long getSessionCacheMaxSize() {
+    return parseLongConfigEntry(GRAVITINO_SESSION_CACHE_MAX_SIZE);
+  }
+
+  /**
+   * Retrieves the expiry (in seconds) for idle per-user sessions in the cache.
+   *
+   * @return the session cache expiry in seconds
+   */
+  public long getSessionCacheExpireAfterAccessSeconds() {
+    return parseLongConfigEntry(GRAVITINO_SESSION_CACHE_EXPIRE_AFTER_ACCESS_SECONDS);
+  }
+
+  private long parseLongConfigEntry(ConfigEntry entry) {
+    String value = config.getOrDefault(entry.key, entry.defaultValue);
+    try {
+      return Long.parseLong(value);
+    } catch (NumberFormatException e) {
+      throw new TrinoException(
+          GravitinoErrorCode.GRAVITINO_ILLEGAL_ARGUMENT,
+          "Invalid value for config '" + entry.key + "': expected a number, got: " + value,
+          e);
+    }
+  }
+
   /**
    * Retrieves a comma-separated list of catalog name regex patterns that should be excluded from
    * loading