HIVE-28930: Implement a metastore service that expires iceberg table snapshots periodically

common/src/java/org/apache/hadoop/hive/conf/HiveConf.java

@@ -2240,7 +2240,7 @@ public static enum ConfVars {
         "Use stats from iceberg table snapshot for query planning. This has two values metastore and iceberg"),
     HIVE_ICEBERG_EXPIRE_SNAPSHOT_NUMTHREADS("hive.iceberg.expire.snapshot.numthreads", 4,
         "The number of threads to be used for deleting files during expire snapshot. If set to 0 or below it uses the" +
-            " defult DirectExecutorService"),
+            " default DirectExecutorService"),
 
     HIVE_ICEBERG_MASK_DEFAULT_LOCATION("hive.iceberg.mask.default.location", false,
         "If this is set to true the URI for auth will have the default location masked with DEFAULT_TABLE_LOCATION"),

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java

@@ -39,8 +39,6 @@
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.commons.collections.MapUtils;
@@ -1133,7 +1131,8 @@ private void deleteOrphanFiles(Table icebergTable, long timestampMillis, int num
     try {
       if (numThreads > 0) {
         LOG.info("Executing delete orphan files on iceberg table {} with {} threads", icebergTable.name(), numThreads);
-        deleteExecutorService = getDeleteExecutorService(icebergTable.name(), numThreads);
+        deleteExecutorService = IcebergTableUtil.newDeleteThreadPool(icebergTable.name(),
+            numThreads);
       }
 
       HiveIcebergDeleteOrphanFiles deleteOrphanFiles = new HiveIcebergDeleteOrphanFiles(conf, icebergTable);
@@ -1156,7 +1155,7 @@ private void expireSnapshot(Table icebergTable, AlterTableExecuteSpec.ExpireSnap
     try {
       if (numThreads > 0) {
         LOG.info("Executing expire snapshots on iceberg table {} with {} threads", icebergTable.name(), numThreads);
-        deleteExecutorService = getDeleteExecutorService(icebergTable.name(), numThreads);
+        deleteExecutorService = IcebergTableUtil.newDeleteThreadPool(icebergTable.name(), numThreads);
       }
       if (expireSnapshotsSpec == null) {
         expireSnapshotWithDefaultParams(icebergTable, deleteExecutorService);
@@ -1235,15 +1234,6 @@ private void expireSnapshotByIds(Table icebergTable, String[] idsToExpire,
     }
   }
 
-  private ExecutorService getDeleteExecutorService(String completeName, int numThreads) {
-    AtomicInteger deleteThreadsIndex = new AtomicInteger(0);
-    return Executors.newFixedThreadPool(numThreads, runnable -> {
-      Thread thread = new Thread(runnable);
-      thread.setName("remove-snapshot-" + completeName + "-" + deleteThreadsIndex.getAndIncrement());
-      return thread;
-    });
-  }
-
   @Override
   public void alterTableSnapshotRefOperation(org.apache.hadoop.hive.ql.metadata.Table hmsTable,
       AlterTableSnapshotRefSpec alterTableSnapshotRefSpec) {

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java

@@ -27,6 +27,9 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BinaryOperator;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -551,4 +554,12 @@ public static TransformSpec getTransformSpec(Table table, String transformName,
     return spec;
   }
 
+  public static ExecutorService newDeleteThreadPool(String completeName, int numThreads) {
+    AtomicInteger deleteThreadsIndex = new AtomicInteger(0);
+    return Executors.newFixedThreadPool(numThreads, runnable -> {
+      Thread thread = new Thread(runnable);
+      thread.setName("remove-snapshot-" + completeName + "-" + deleteThreadsIndex.getAndIncrement());
+      return thread;
+    });
+  }
 }

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/metastore/task/IcebergHouseKeeperService.java

@@ -0,0 +1,162 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.iceberg.mr.hive.metastore.task;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.common.TableName;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.MetastoreTaskThread;
+import org.apache.hadoop.hive.metastore.api.GetTableRequest;
+import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
+import org.apache.hadoop.hive.metastore.txn.NoMutex;
+import org.apache.hadoop.hive.metastore.txn.TxnStore;
+import org.apache.hadoop.hive.metastore.txn.TxnUtils;
+import org.apache.hadoop.hive.metastore.utils.TableFetcher;
+import org.apache.iceberg.ExpireSnapshots;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.mr.hive.IcebergTableUtil;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class IcebergHouseKeeperService implements MetastoreTaskThread {
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergHouseKeeperService.class);
+
+  private Configuration conf;
+  private TxnStore txnHandler;
+  private boolean shouldUseMutex;
+  private ExecutorService deleteExecutorService = null;
+
+  // table cache to avoid making repeated requests for the same Iceberg tables more than once per day
+  private final Cache<TableName, Table> tableCache = Caffeine.newBuilder()
+      .maximumSize(1000)
+      .expireAfterWrite(1, TimeUnit.DAYS)
+      .build();
+
+  @Override
+  public long runFrequency(TimeUnit unit) {
+    return MetastoreConf.getTimeVar(conf, MetastoreConf.ConfVars.ICEBERG_TABLE_EXPIRY_INTERVAL, unit);
+  }
+
+  @Override
+  public void run() {
+    LOG.debug("Running IcebergHouseKeeperService...");
+
+    String catalogName = MetastoreConf.getVar(conf, MetastoreConf.ConfVars.ICEBERG_TABLE_EXPIRY_CATALOG_NAME);
+    String dbPattern = MetastoreConf.getVar(conf, MetastoreConf.ConfVars.ICEBERG_TABLE_EXPIRY_DATABASE_PATTERN);
+    String tablePattern = MetastoreConf.getVar(conf, MetastoreConf.ConfVars.ICEBERG_TABLE_EXPIRY_TABLE_PATTERN);
+
+    TxnStore.MutexAPI mutex = shouldUseMutex ? txnHandler.getMutexAPI() : new NoMutex();
+
+    try (AutoCloseable closeable = mutex.acquireLock(TxnStore.MUTEX_KEY.IcebergHouseKeeper.name())) {
+      expireTables(catalogName, dbPattern, tablePattern);
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private void expireTables(String catalogName, String dbPattern, String tablePattern) {
+    try (IMetaStoreClient msc = new HiveMetaStoreClient(conf)) {
+      // TODO: HIVE-28952 – modify TableFetcher to return HMS Table API objects directly,
+      // avoiding the need for subsequent msc.getTable calls to fetch each matched table individually
+      List<TableName> tables = getTableFetcher(msc, catalogName, dbPattern, tablePattern).getTables();
+
+      LOG.debug("{} candidate tables found", tables.size());
+
+      for (TableName table : tables) {
+        try {
+          expireSnapshotsForTable(getIcebergTable(table, msc));
+        } catch (Exception e) {
+          LOG.error("Exception while running iceberg expiry service on catalog/db/table: {}/{}/{}",
+              catalogName, dbPattern, tablePattern, e);
+        }
+      }
+    } catch (Exception e) {
+      throw new RuntimeException("Error while getting tables from metastore", e);
+    }
+  }
+
+  @VisibleForTesting
+  TableFetcher getTableFetcher(IMetaStoreClient msc, String catalogName, String dbPattern, String tablePattern) {
+    return new TableFetcher.Builder(msc, catalogName, dbPattern, tablePattern).tableTypes(
+            "EXTERNAL_TABLE")
+        .tableCondition(
+            hive_metastoreConstants.HIVE_FILTER_FIELD_PARAMS + "table_type like \"ICEBERG\" ")
+        .build();
+  }
+
+  private Table getIcebergTable(TableName tableName, IMetaStoreClient msc) {
+    return tableCache.get(tableName, key -> {
+      LOG.debug("Getting iceberg table from metastore as it's not present in table cache: {}", tableName);
+      GetTableRequest request = new GetTableRequest(tableName.getDb(), tableName.getTable());
+      try {
+        return IcebergTableUtil.getTable(conf, msc.getTable(request));
+      } catch (TException e) {
+        throw new RuntimeException(e);
+      }
+    });
+  }
+
+  /**
+   * Deletes snapshots of an Iceberg table, using the number of threads defined by the
+   * Hive config HIVE_ICEBERG_EXPIRE_SNAPSHOT_NUMTHREADS.
+   * This is largely equivalent to the HiveIcebergStorageHandler.expireSnapshotWithDefaultParams method.
+   *
+   * @param icebergTable the iceberg Table reference
+   */
+  private void expireSnapshotsForTable(Table icebergTable) {
+    ExpireSnapshots expireSnapshots = icebergTable.expireSnapshots();
+    if (deleteExecutorService != null) {
+      expireSnapshots.executeDeleteWith(deleteExecutorService);
+    }
+    expireSnapshots.commit();
+  }
+
+  @Override
+  public void enforceMutex(boolean enableMutex) {
+    this.shouldUseMutex = enableMutex;
+  }
+
+  @Override
+  public Configuration getConf() {
+    return conf;
+  }
+
+  @Override
+  public void setConf(Configuration configuration) {
+    conf = configuration;
+    txnHandler = TxnUtils.getTxnStore(conf);
+
+    int numThreads = conf.getInt(HiveConf.ConfVars.HIVE_ICEBERG_EXPIRE_SNAPSHOT_NUMTHREADS.varname,
+        HiveConf.ConfVars.HIVE_ICEBERG_EXPIRE_SNAPSHOT_NUMTHREADS.defaultIntVal);
+    if (numThreads > 0) {
+      LOG.info("Will expire Iceberg snapshots using an executor service with {} threads", numThreads);
+      deleteExecutorService = IcebergTableUtil.newDeleteThreadPool("iceberg-housekeeper-service", numThreads);
+    }
+  }
+}

iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/metastore/task/TestIcebergHouseKeeperService.java

@@ -0,0 +1,143 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.iceberg.mr.hive.metastore.task;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import org.apache.hadoop.hive.common.TableName;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.TableType;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.GetTableRequest;
+import org.apache.hadoop.hive.metastore.utils.TableFetcher;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.ql.metadata.Table;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.mr.hive.IcebergTableUtil;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TestIcebergHouseKeeperService {
+  private static final Logger LOG = LoggerFactory.getLogger(TestIcebergHouseKeeperService.class);
+
+  private static final HiveConf conf = new HiveConf(TestIcebergHouseKeeperService.class);
+  private static Hive db;
+
+  @BeforeClass
+  public static void beforeClass() throws Exception {
+    conf.set("hive.security.authorization.enabled", "false");
+    conf.set("hive.security.authorization.manager",
+        "org.apache.hadoop.hive.ql.security.authorization.plugin.sqlstd.SQLStdConfOnlyAuthorizerFactory");
+    conf.set("iceberg.engine.hive.lock-enabled", "false");
+
+    db = Hive.get(conf);
+  }
+
+  @AfterClass
+  public static void afterClass() {
+    db.close(true);
+  }
+
+  @Test
+  public void testIcebergTableFetched() throws Exception {
+    createIcebergTable("iceberg_table");
+
+    IcebergHouseKeeperService service = new IcebergHouseKeeperService();
+    TableFetcher tableFetcher = service.getTableFetcher(db.getMSC(), null, "default", "*");
+
+    List<TableName> tables = tableFetcher.getTables();
+    Assert.assertEquals(new TableName("hive", "default", "iceberg_table"), tables.get(0));
+  }
+
+  @Test
+  public void testExpireSnapshotsByServiceRun() throws Exception {
+    String tableName = "iceberg_table_snapshot_expiry_e2e_test";
+    createIcebergTable(tableName);
+    IcebergHouseKeeperService service = getServiceForTable("default", tableName);
+
+    GetTableRequest request = new GetTableRequest("default", tableName);
+    org.apache.iceberg.Table icebergTable = IcebergTableUtil.getTable(conf, db.getMSC().getTable(request));
+
+    String metadataDirectory = icebergTable.location().replaceAll("^[a-zA-Z]+:", "") + "/metadata";
+
+    DataFile datafile = DataFiles.builder(icebergTable.spec())
+        .withRecordCount(3)
+        .withPath("/tmp/file.parquet")
+        .withFileSizeInBytes(10)
+        .build();
+
+    icebergTable.newAppend().appendFile(datafile).commit();
+    assertSnapshotFiles(metadataDirectory, 1);
+    icebergTable.newAppend().appendFile(datafile).commit();
+    assertSnapshotFiles(metadataDirectory, 2);
+
+    Thread.sleep(1000); // allow snapshots that are 1000ms old to become eligible for snapshot expiry
+    service.run();
+
+    assertSnapshotFiles(metadataDirectory, 1);
+    db.dropTable("default", "iceberg_table_snapshot_expiry_e2e_test");
+  }
+
+  private void createIcebergTable(String name) throws Exception {
+    Table table = new Table("default", name);
+    List<FieldSchema> columns = Lists.newArrayList();
+    columns.add(new FieldSchema("col", "string", "First column"));
+    table.setFields(columns); // Set columns
+
+    table.setProperty("EXTERNAL", "TRUE");
+    table.setTableType(TableType.EXTERNAL_TABLE);
+    table.setProperty("table_type", "ICEBERG");
+
+    table.setProperty("history.expire.max-snapshot-age-ms", "500");
+
+    db.createTable(table);
+  }
+
+  /**
+   * Creates IcebergHouseKeeperService that's configured to clean up a table by database and table name.
+   *
+   * @param tableName to be cleaned up
+   * @return IcebergHouseKeeperService
+   */
+  private IcebergHouseKeeperService getServiceForTable(String dbName, String tableName) {
+    IcebergHouseKeeperService service = new IcebergHouseKeeperService();
+    HiveConf serviceConf = new HiveConf(conf);
+    serviceConf.set("hive.metastore.iceberg.table.expiry.database.pattern", dbName);
+    serviceConf.set("hive.metastore.iceberg.table.expiry.table.pattern", tableName);
+    service.setConf(serviceConf);
+    return service;
+  }
+
+  private void assertSnapshotFiles(String metadataDirectory, int numberForSnapshotFiles) {
+    File[] matchingFiles = new File(metadataDirectory).listFiles((dir, name) -> name.startsWith("snap-"));
+    List<File> files = Optional.ofNullable(matchingFiles).map(Arrays::asList).orElse(Collections.emptyList());
+    LOG.debug("Snapshot files found in directory({}): {}", metadataDirectory, files);
+    Assert.assertEquals(String.format("Unexpected no. of snapshot files in metadata directory: %s",
+        metadataDirectory), numberForSnapshotFiles, files.size());
+  }
+}