[AMORO-4268][AMS] Support HDFS owner impersonation for optimizing commits

amoro-ams/src/main/java/org/apache/amoro/server/catalog/InternalCatalogImpl.java

@@ -30,6 +30,7 @@
 import org.apache.amoro.exception.ObjectNotExistsException;
 import org.apache.amoro.formats.iceberg.IcebergTable;
 import org.apache.amoro.io.AuthenticatedFileIO;
+import org.apache.amoro.io.AuthenticatedFileIOs;
 import org.apache.amoro.mixed.InternalMixedIcebergCatalog;
 import org.apache.amoro.server.AmoroManagementConf;
 import org.apache.amoro.server.RestCatalogService;
@@ -56,6 +57,8 @@
 import org.apache.iceberg.rest.RESTCatalog;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
 
+import java.util.Map;
+
 public class InternalCatalogImpl extends InternalCatalog {
 
   private static final String URI = "uri";
@@ -117,23 +120,28 @@ private AmoroTable<?> loadIcebergTable(
       String database, String tableName, InternalTableHandler<TableOperations> handler) {
     TableMetadata tableMetadata = handler.tableMetadata();
     TableOperations ops = handler.newTableOperator();
-
-    BaseTable table =
-        new BaseTable(
-            ops,
-            TableIdentifier.of(
-                    tableMetadata.getTableIdentifier().getDatabase(),
-                    tableMetadata.getTableIdentifier().getTableName())
-                .toString());
+    BaseTable table = newBaseTable(tableMetadata, ops);
+    CatalogMeta catalogMeta = getMetadata();
+    Map<String, String> catalogProperties = catalogMeta.getCatalogProperties();
+    Map<String, String> tableProperties = table.properties();
+    AuthenticatedFileIO fileIO = (AuthenticatedFileIO) ops.io();
+    if (AuthenticatedFileIOs.isHdfsImpersonationEnabledForOptimizingCommit(
+        tableProperties, catalogProperties)) {
+      handler.close();
+      fileIO = InternalTableUtil.newIcebergFileIo(catalogMeta, tableProperties);
+      ops = new InternalIcebergHandler(tableMetadata, fileIO).newTableOperator();
+      table = newBaseTable(tableMetadata, ops);
+    }
     org.apache.amoro.table.TableIdentifier tableIdentifier =
         org.apache.amoro.table.TableIdentifier.of(name(), database, tableName);
     AmoroTable<?> amoroTable =
         IcebergTable.newIcebergTable(
             tableIdentifier,
             table,
-            CatalogUtil.buildMetaStore(getMetadata()),
-            getMetadata().getCatalogProperties());
-    fileIOCloser.put(amoroTable, ops.io());
+            fileIO,
+            CatalogUtil.buildMetaStore(catalogMeta).getConfiguration(),
+            catalogProperties);
+    fileIOCloser.put(amoroTable, fileIO);
     return amoroTable;
   }
 
@@ -142,12 +150,22 @@ private AmoroTable<?> loadMixedIcebergTable(
     TableMetadata tableMetadata = handler.tableMetadata();
     org.apache.amoro.table.TableIdentifier tableIdentifier =
         org.apache.amoro.table.TableIdentifier.of(name(), database, tableName);
-    AuthenticatedFileIO fileIO = InternalTableUtil.newIcebergFileIo(getMetadata());
+    CatalogMeta catalogMeta = getMetadata();
+    Map<String, String> catalogProperties = catalogMeta.getCatalogProperties();
+    TableOperations baseOps = handler.newTableOperator();
+    BaseTable baseTable = newBaseTable(tableMetadata, baseOps);
+    Map<String, String> tableProperties = baseTable.properties();
+    AuthenticatedFileIO fileIO = (AuthenticatedFileIO) baseOps.io();
+    if (AuthenticatedFileIOs.isHdfsImpersonationEnabledForOptimizingCommit(
+        tableProperties, catalogProperties)) {
+      handler.close();
+      fileIO = InternalTableUtil.newIcebergFileIo(catalogMeta, tableProperties);
+      baseTable = loadTableStore(tableMetadata, false, fileIO);
+    }
     MixedTable mixedIcebergTable;
 
-    BaseTable baseTable = loadTableStore(tableMetadata, false);
     if (InternalTableUtil.isKeyedMixedTable(tableMetadata)) {
-      BaseTable changeTable = loadTableStore(tableMetadata, true);
+      BaseTable changeTable = loadTableStore(tableMetadata, true, fileIO);
 
       PrimaryKeySpec.Builder keySpecBuilder = PrimaryKeySpec.builderFor(baseTable.schema());
       tableMetadata.buildTableMeta().getKeySpec().getFields().forEach(keySpecBuilder::addColumn);
@@ -158,22 +176,27 @@ private AmoroTable<?> loadMixedIcebergTable(
               tableMetadata.getTableLocation(),
               keySpec,
               new BasicKeyedTable.BaseInternalTable(
-                  tableIdentifier, baseTable, fileIO, getMetadata().getCatalogProperties()),
+                  tableIdentifier, baseTable, fileIO, catalogProperties),
               new BasicKeyedTable.ChangeInternalTable(
-                  tableIdentifier, changeTable, fileIO, getMetadata().getCatalogProperties()));
+                  tableIdentifier, changeTable, fileIO, catalogProperties));
     } else {
       mixedIcebergTable =
-          new BasicUnkeyedTable(
-              tableIdentifier, baseTable, fileIO, getMetadata().getCatalogProperties());
+          new BasicUnkeyedTable(tableIdentifier, baseTable, fileIO, catalogProperties);
     }
     AmoroTable<?> amoroTable =
         new org.apache.amoro.formats.mixed.MixedTable(mixedIcebergTable, TableFormat.MIXED_ICEBERG);
     fileIOCloser.put(amoroTable, fileIO);
     return amoroTable;
   }
 
-  private BaseTable loadTableStore(TableMetadata tableMetadata, boolean isChangeStore) {
-    TableOperations ops = newTableStoreHandler(tableMetadata, isChangeStore).newTableOperator();
+  private BaseTable loadTableStore(
+      TableMetadata tableMetadata, boolean isChangeStore, AuthenticatedFileIO fileIO) {
+    TableOperations ops =
+        newTableStoreHandler(tableMetadata, isChangeStore, fileIO).newTableOperator();
+    return newBaseTable(tableMetadata, ops);
+  }
+
+  private BaseTable newBaseTable(TableMetadata tableMetadata, TableOperations ops) {
     return new BaseTable(
         ops,
         TableIdentifier.of(
@@ -248,6 +271,11 @@ private InternalTableHandler<TableOperations> newTableStoreHandler(
     return new InternalMixedIcebergHandler(getMetadata(), metadata, isChangeStore);
   }
 
+  private InternalTableHandler<TableOperations> newTableStoreHandler(
+      TableMetadata metadata, boolean isChangeStore, AuthenticatedFileIO fileIO) {
+    return new InternalMixedIcebergHandler(getMetadata(), metadata, isChangeStore, fileIO);
+  }
+
   private Cache<AmoroTable<?>, FileIO> newFileIOCloser() {
     return Caffeine.newBuilder()
         .weakKeys()

amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java

@@ -27,6 +27,7 @@
 import org.apache.amoro.exception.OptimizingClosedException;
 import org.apache.amoro.exception.PersistenceException;
 import org.apache.amoro.exception.TaskNotFoundException;
+import org.apache.amoro.io.AuthenticatedFileIOs;
 import org.apache.amoro.optimizing.MetricsSummary;
 import org.apache.amoro.optimizing.OptimizingType;
 import org.apache.amoro.optimizing.RewriteFilesInput;
@@ -923,10 +924,12 @@ public MetricsSummary getSummary() {
 
     private UnKeyedTableCommit buildCommit() {
       MixedTable table =
-          (MixedTable)
-              catalogManager
-                  .loadTable(tableRuntime.getTableIdentifier().getIdentifier())
-                  .originalTable();
+          AuthenticatedFileIOs.withOptimizingCommitImpersonation(
+              () ->
+                  (MixedTable)
+                      catalogManager
+                          .loadTable(tableRuntime.getTableIdentifier().getIdentifier())
+                          .originalTable());
       if (table.isUnkeyedTable()) {
         return new UnKeyedTableCommit(targetSnapshotId, table, taskMap.values());
       } else {

amoro-ams/src/main/java/org/apache/amoro/server/table/internal/InternalIcebergHandler.java

@@ -37,8 +37,12 @@ public class InternalIcebergHandler implements InternalTableHandler<TableOperati
   private final TableMetadata tableMetadata;
 
   public InternalIcebergHandler(CatalogMeta catalogMeta, TableMetadata metadata) {
+    this(metadata, InternalTableUtil.newIcebergFileIo(catalogMeta));
+  }
+
+  public InternalIcebergHandler(TableMetadata metadata, AuthenticatedFileIO io) {
     this.tableMetadata = metadata;
-    this.io = InternalTableUtil.newIcebergFileIo(catalogMeta);
+    this.io = io;
   }
 
   @Override

amoro-ams/src/main/java/org/apache/amoro/server/table/internal/InternalMixedIcebergHandler.java

@@ -21,6 +21,7 @@
 import org.apache.amoro.TableFormat;
 import org.apache.amoro.api.CatalogMeta;
 import org.apache.amoro.api.TableMeta;
+import org.apache.amoro.io.AuthenticatedFileIO;
 import org.apache.amoro.op.MixedHadoopTableOperations;
 import org.apache.amoro.server.table.TableMetadata;
 import org.apache.amoro.server.utils.InternalTableUtil;
@@ -43,7 +44,15 @@ public class InternalMixedIcebergHandler extends InternalIcebergHandler {
 
   public InternalMixedIcebergHandler(
       CatalogMeta catalogMeta, TableMetadata metadata, boolean changeStore) {
-    super(catalogMeta, metadata);
+    this(catalogMeta, metadata, changeStore, InternalTableUtil.newIcebergFileIo(catalogMeta));
+  }
+
+  public InternalMixedIcebergHandler(
+      CatalogMeta catalogMeta,
+      TableMetadata metadata,
+      boolean changeStore,
+      AuthenticatedFileIO io) {
+    super(metadata, io);
     this.changeStore = changeStore;
     this.catalogMeta = catalogMeta;
   }

amoro-ams/src/main/java/org/apache/amoro/server/utils/InternalTableUtil.java

@@ -83,6 +83,18 @@ public static boolean isKeyedMixedTable(
    * @return iceberg file io
    */
   public static AuthenticatedFileIO newIcebergFileIo(CatalogMeta meta) {
+    return newIcebergFileIo(meta, null);
+  }
+
+  /**
+   * Create an Iceberg FileIO that honors owner impersonation while loading an optimizing commit.
+   *
+   * @param meta catalog meta
+   * @param tableProperties table properties used to resolve the owner and table-level override
+   * @return iceberg file io
+   */
+  public static AuthenticatedFileIO newIcebergFileIo(
+      CatalogMeta meta, Map<String, String> tableProperties) {
     Map<String, String> catalogProperties = meta.getCatalogProperties();
     TableMetaStore store = CatalogUtil.buildMetaStore(meta);
     Configuration conf = store.getConfiguration();
@@ -95,7 +107,11 @@ public static AuthenticatedFileIO newIcebergFileIo(CatalogMeta meta) {
     }
     String ioImpl = catalogProperties.getOrDefault(CatalogProperties.FILE_IO_IMPL, defaultImpl);
     FileIO fileIO = org.apache.iceberg.CatalogUtil.loadFileIO(ioImpl, catalogProperties, conf);
-    return AuthenticatedFileIOs.buildAdaptIcebergFileIO(store, fileIO);
+    if (tableProperties == null) {
+      return AuthenticatedFileIOs.buildAdaptIcebergFileIO(store, fileIO);
+    }
+    return AuthenticatedFileIOs.buildAdaptIcebergFileIO(
+        store, fileIO, tableProperties, catalogProperties);
   }
 
   /**

amoro-ams/src/test/java/org/apache/amoro/server/table/internal/TestInternalIcebergHandlerImpersonation.java

@@ -0,0 +1,87 @@
+/*
+ * 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.amoro.server.table.internal;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+
+import org.apache.amoro.api.CatalogMeta;
+import org.apache.amoro.io.AuthenticatedFileIO;
+import org.apache.amoro.io.AuthenticatedFileIOs;
+import org.apache.amoro.properties.CatalogMetaProperties;
+import org.apache.amoro.server.table.TableMetadata;
+import org.apache.amoro.server.utils.InternalTableUtil;
+import org.apache.amoro.table.TableProperties;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.Map;
+
+public class TestInternalIcebergHandlerImpersonation {
+
+  @Test
+  public void testInternalHandlerUsesTableOwner() {
+    String catalogUser = "amoro-service";
+    String tableOwner = "table-owner";
+    TableMetadata tableMetadata = mock(TableMetadata.class);
+    CatalogMeta catalogMeta = newCatalogMeta(catalogUser);
+    AuthenticatedFileIO fileIO =
+        AuthenticatedFileIOs.withOptimizingCommitImpersonation(
+            () ->
+                InternalTableUtil.newIcebergFileIo(
+                    catalogMeta, Map.of(TableProperties.OWNER, tableOwner)));
+    InternalIcebergHandler handler = new InternalIcebergHandler(tableMetadata, fileIO);
+    try {
+      AuthenticatedFileIO handlerFileIO = (AuthenticatedFileIO) handler.newTableOperator().io();
+      assertEquals(tableOwner, handlerFileIO.doAs(this::currentUser));
+    } finally {
+      handler.close();
+    }
+  }
+
+  private static CatalogMeta newCatalogMeta(String catalogUser) {
+    String emptyConfiguration =
+        Base64.getEncoder().encodeToString("<configuration/>".getBytes(StandardCharsets.UTF_8));
+    Map<String, String> storageConfigs = new HashMap<>();
+    storageConfigs.put(
+        CatalogMetaProperties.STORAGE_CONFIGS_KEY_TYPE,
+        CatalogMetaProperties.STORAGE_CONFIGS_VALUE_TYPE_HADOOP);
+    storageConfigs.put(CatalogMetaProperties.STORAGE_CONFIGS_KEY_CORE_SITE, emptyConfiguration);
+    storageConfigs.put(CatalogMetaProperties.STORAGE_CONFIGS_KEY_HDFS_SITE, emptyConfiguration);
+
+    Map<String, String> authConfigs = new HashMap<>();
+    authConfigs.put(
+        CatalogMetaProperties.AUTH_CONFIGS_KEY_TYPE,
+        CatalogMetaProperties.AUTH_CONFIGS_VALUE_TYPE_SIMPLE);
+    authConfigs.put(CatalogMetaProperties.AUTH_CONFIGS_KEY_HADOOP_USERNAME, catalogUser);
+
+    Map<String, String> catalogProperties = new HashMap<>();
+    catalogProperties.put(CatalogMetaProperties.KEY_WAREHOUSE, "file:///tmp/amoro");
+    catalogProperties.put(CatalogMetaProperties.HDFS_IMPERSONATION_ENABLED, "true");
+    return new CatalogMeta("test", "ams", storageConfigs, authConfigs, catalogProperties);
+  }
+
+  private String currentUser() throws IOException {
+    return UserGroupInformation.getCurrentUser().getShortUserName();
+  }
+}

amoro-common/src/main/java/org/apache/amoro/properties/CatalogMetaProperties.java

@@ -98,6 +98,8 @@ public class CatalogMetaProperties {
   public static final String DATA_EXPIRATION_PREFIX = "data-expire.";
   public static final String TABLE_TRASH_PREFIX = "table-trash.";
   public static final String AUTO_CREATE_TAG_PREFIX = "tag.auto-create.";
+  public static final String HDFS_IMPERSONATION_PREFIX = "hdfs.impersonation.";
+  public static final String HDFS_IMPERSONATION_ENABLED = HDFS_IMPERSONATION_PREFIX + "enabled";
 
   // mixed-format properties
   public static final String MIXED_FORMAT_TABLE_STORE_SEPARATOR =

amoro-common/src/main/java/org/apache/amoro/table/TableMetaStore.java

@@ -231,6 +231,12 @@ public boolean isKerberosAuthMethod() {
     return AUTH_METHOD_KERBEROS.equalsIgnoreCase(authMethod);
   }
 
+  /** Returns whether this meta store uses an authentication mode that supports proxy users. */
+  public boolean supportsHadoopImpersonation() {
+    return AUTH_METHOD_SIMPLE.equalsIgnoreCase(authMethod)
+        || AUTH_METHOD_KERBEROS.equalsIgnoreCase(authMethod);
+  }
+
   public String getHadoopUsername() {
     return hadoopUsername;
   }

amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/IcebergCatalog.java

@@ -20,6 +20,7 @@
 
 import org.apache.amoro.AmoroTable;
 import org.apache.amoro.FormatCatalog;
+import org.apache.amoro.io.TableOwnerResolver;
 import org.apache.amoro.table.TableMetaStore;
 import org.apache.amoro.utils.MixedFormatCatalogUtil;
 import org.apache.iceberg.Table;
@@ -39,14 +40,24 @@ public class IcebergCatalog implements FormatCatalog {
   private final Catalog icebergCatalog;
   private final TableMetaStore metaStore;
   private final Map<String, String> properties;
+  private final String metastoreType;
 
   public IcebergCatalog(Catalog catalog, Map<String, String> properties, TableMetaStore metaStore) {
+    this(catalog, null, properties, metaStore);
+  }
+
+  public IcebergCatalog(
+      Catalog catalog,
+      String metastoreType,
+      Map<String, String> properties,
+      TableMetaStore metaStore) {
     this.icebergCatalog = MixedFormatCatalogUtil.buildCacheCatalog(catalog, properties);
     if (catalog instanceof SupportsNamespaces) {
       this.asNamespaceCatalog = (SupportsNamespaces) catalog;
     }
     this.metaStore = metaStore;
     this.properties = properties;
+    this.metastoreType = metastoreType;
   }
 
   @Override
@@ -102,11 +113,13 @@ public AmoroTable<?> loadTable(String database, String table) {
         () -> {
           try {
             Table icebergTable = icebergCatalog.loadTable(TableIdentifier.of(database, table));
+            org.apache.amoro.table.TableIdentifier identifier =
+                org.apache.amoro.table.TableIdentifier.of(icebergCatalog.name(), database, table);
+            String tableOwner =
+                TableOwnerResolver.resolve(
+                    metastoreType, identifier, icebergTable, properties, metaStore);
             return IcebergTable.newIcebergTable(
-                org.apache.amoro.table.TableIdentifier.of(icebergCatalog.name(), database, table),
-                icebergTable,
-                metaStore,
-                properties);
+                identifier, icebergTable, metaStore, properties, tableOwner);
           } catch (NoSuchTableException e) {
             throw new org.apache.amoro.NoSuchTableException(e);
           }

amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/IcebergCatalogFactory.java

@@ -42,7 +42,7 @@ public FormatCatalog create(
 
     Catalog icebergCatalog =
         CatalogUtil.buildIcebergCatalog(name, properties, metaStore.getConfiguration());
-    return new IcebergCatalog(icebergCatalog, properties, metaStore);
+    return new IcebergCatalog(icebergCatalog, metastoreType, properties, metaStore);
   }
 
   @Override