HIVE-29803: [Iceberg] Support ANCESTORS_OF snapshot lineage command

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

@@ -1361,6 +1361,11 @@ public void executeOperation(org.apache.hadoop.hive.ql.metadata.Table hmsTable,
       case REWRITE_MANIFESTS:
         IcebergTableUtil.rewriteManifests(icebergTable);
         break;
+      case ANCESTORS_OF:
+        AlterTableExecuteSpec.AncestorsOfSpec ancestorsOfSpec =
+            (AlterTableExecuteSpec.AncestorsOfSpec) executeSpec.getOperationParams();
+        IcebergTableUtil.printAncestorsOf(icebergTable, ancestorsOfSpec.snapshotId());
+        break;
       case DELETE_METADATA:
         AlterTableExecuteSpec.DeleteMetadataSpec deleteMetadataSpec =
             (AlterTableExecuteSpec.DeleteMetadataSpec) executeSpec.getOperationParams();

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

@@ -999,4 +999,43 @@
         properties -> IcebergTableUtil.formatVersion(tableProperties) >= 3 &&
             FileFormat.PARQUET == IcebergTableUtil.defaultFileFormat(properties::getOrDefault)).isPresent();
   }
+
+  /**
+   * Returns the ancestors of a given Iceberg snapshot. If snapshotId is null, it defaults to the
+   * current snapshot of the table.
+   */
+  public static Iterable<Snapshot> getAncestorsOf(Table table, Long snapshotId) {
+    long targetSnapshotId = snapshotId != null ? snapshotId : table.currentSnapshot().snapshotId();
+    return SnapshotUtil.ancestorsOf(targetSnapshotId, table::snapshot);
+  }
+
+  /**
+   * Prints the ancestors of a given Iceberg snapshot to the Hive console. If snapshotId is null, it
+   * defaults to the current snapshot of the table.
+   */
+  public static void printAncestorsOf(Table table, Long snapshotId) {
+    long targetSnapshotId = snapshotId != null ? snapshotId : table.currentSnapshot().snapshotId();
+    Iterable<Snapshot> ancestors = getAncestorsOf(table, snapshotId);
+    SessionState.LogHelper console = SessionState.getConsole();
+    if (console != null) {
+      // A width of 25 is used because it fits the column headers. It also safely fits the data,
+      // since a 64-bit long (used for IDs and timestamps) has a maximum of 19 digits.
+      console.printInfo("+---------------------------+---------------------------+");
+      console.printInfo(
+          String.format(
+              "| %s | %s |",
+              StringUtils.center("snapshot_id", 25), StringUtils.center("timestamp_ms", 25)));
+      console.printInfo("+---------------------------+---------------------------+");
+      for (Snapshot snapshot : ancestors) {
+        console.printInfo(
+            String.format("| %-25s | %-25s |", snapshot.snapshotId(), snapshot.timestampMillis()));
+      }
+      console.printInfo("+---------------------------+---------------------------+");
+    } else {
+      LOG.info("Ancestors of snapshot {}:", targetSnapshotId);
+      for (Snapshot snapshot : ancestors) {
+        LOG.info("{} - {}", snapshot.snapshotId(), snapshot.timestampMillis());
+      }
+    }
+  }
 }

iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergSnapshotOperations.java

@@ -238,4 +238,45 @@
     icebergTable.refresh();
     return icebergTable.currentSnapshot().allManifests(icebergTable.io()).size();
   }
+  @Test
+  public void testAncestorsOf() throws Exception {
+    TableIdentifier identifier = TableIdentifier.of("default", "testAncestorsOf");
+    shell.executeStatement(
+        String.format(
+            "CREATE EXTERNAL TABLE %s (id INT) STORED BY iceberg %s %s",
+            identifier.name(),
+            testTables.locationForCreateTableSQL(identifier),
+            testTables.propertiesForCreateTableSQL(ImmutableMap.of())));
+
+    // Create 3 snapshots
+    shell.executeStatement(String.format("INSERT INTO TABLE %s VALUES(1)", identifier.name()));
+    shell.executeStatement(String.format("INSERT INTO TABLE %s VALUES(2)", identifier.name()));
+    shell.executeStatement(String.format("INSERT INTO TABLE %s VALUES(3)", identifier.name()));
+
+    org.apache.iceberg.Table icebergTable = testTables.loadTable(identifier);
+    icebergTable.refresh();
+
+    // 1. Positive Test: The command should execute successfully without errors
+    shell.executeStatement(String.format("ALTER TABLE %s EXECUTE ANCESTORS_OF", identifier.name()));
+
+    // 2. Positive Test: Run with a specific valid snapshot ID
+    long currentSnapshotId = icebergTable.currentSnapshot().snapshotId();
+    shell.executeStatement(
+        String.format(
+            "ALTER TABLE %s EXECUTE ANCESTORS_OF(%d)", identifier.name(), currentSnapshotId));
+
+    // 3. Negative Test: Run with a completely fake/invalid snapshot ID
+    long fakeSnapshotId = 99999999999999999L;
+    try {
+      shell.executeStatement(
+          String.format(
+              "ALTER TABLE %s EXECUTE ANCESTORS_OF(%d)", identifier.name(), fakeSnapshotId));
+      Assert.fail("Expected an exception to be thrown for an invalid snapshot ID");
+    } catch (Exception e) {
+      Assert.assertTrue(
+          "Exception message should indicate failure to find snapshot",
+          e.getMessage().contains("Cannot find snapshot") ||
+              e.getMessage().contains("Cannot find"));
+    }
+  }
 }

parser/src/java/org/apache/hadoop/hive/ql/parse/AlterClauseParser.g

@@ -536,6 +536,8 @@ alterStatementSuffixExecute
     -> ^(TOK_ALTERTABLE_EXECUTE KW_EXPIRE_SNAPSHOTS $expireParam?)
     | KW_EXECUTE KW_REWRITE_MANIFESTS
     -> ^(TOK_ALTERTABLE_EXECUTE KW_REWRITE_MANIFESTS)
+    | KW_EXECUTE KW_ANCESTORS_OF (LPAREN (snapshotParam=expression) RPAREN)?
+    -> ^(TOK_ALTERTABLE_EXECUTE KW_ANCESTORS_OF $snapshotParam?)
     | KW_EXECUTE KW_SET_CURRENT_SNAPSHOT LPAREN (snapshotParam=expression) RPAREN
     -> ^(TOK_ALTERTABLE_EXECUTE KW_SET_CURRENT_SNAPSHOT $snapshotParam)
     | KW_EXECUTE KW_FAST_FORWARD sourceBranch=StringLiteral (targetBranch=StringLiteral)?

parser/src/java/org/apache/hadoop/hive/ql/parse/HiveLexerParent.g

@@ -399,6 +399,7 @@ KW_SYSTEM_TIME: 'SYSTEM_TIME';
 KW_SYSTEM_VERSION: 'SYSTEM_VERSION';
 KW_EXPIRE_SNAPSHOTS: 'EXPIRE_SNAPSHOTS';
 KW_REWRITE_MANIFESTS: 'REWRITE_MANIFESTS';
+KW_ANCESTORS_OF: 'ANCESTORS_OF';
 KW_SET_CURRENT_SNAPSHOT: 'SET_CURRENT_SNAPSHOT';
 KW_BRANCH: 'BRANCH';
 KW_SNAPSHOTS: 'SNAPSHOTS';

parser/src/java/org/apache/hadoop/hive/ql/parse/IdentifiersParser.g

@@ -1028,6 +1028,7 @@ nonReserved
     | KW_SYSTEM_TIME | KW_SYSTEM_VERSION
     | KW_EXPIRE_SNAPSHOTS
     | KW_REWRITE_MANIFESTS
+    | KW_ANCESTORS_OF
     | KW_SET_CURRENT_SNAPSHOT
     | KW_BRANCH | KW_SNAPSHOTS | KW_RETAIN | KW_RETENTION
     | KW_TAG

ql/src/java/org/apache/hadoop/hive/ql/ddl/table/execute/AlterTableExecuteAnalyzer.java

@@ -59,6 +59,7 @@
 import static org.apache.hadoop.hive.ql.parse.AlterTableExecuteSpec.ExecuteOperationType.FAST_FORWARD;
 import static org.apache.hadoop.hive.ql.parse.AlterTableExecuteSpec.ExecuteOperationType.ROLLBACK;
 import static org.apache.hadoop.hive.ql.parse.AlterTableExecuteSpec.ExecuteOperationType.SET_CURRENT_SNAPSHOT;
+import static org.apache.hadoop.hive.ql.parse.AlterTableExecuteSpec.ExecuteOperationType.ANCESTORS_OF;
 import static org.apache.hadoop.hive.ql.parse.AlterTableExecuteSpec.RollbackSpec.RollbackType.TIME;
 import static org.apache.hadoop.hive.ql.parse.AlterTableExecuteSpec.RollbackSpec.RollbackType.VERSION;
 import static org.apache.hadoop.hive.ql.parse.HiveLexer.KW_RETAIN;
@@ -106,7 +107,10 @@
       case HiveParser.KW_REWRITE_MANIFESTS:
         desc = new AlterTableExecuteDesc(tableName, partitionSpec,
             new AlterTableExecuteSpec(AlterTableExecuteSpec.ExecuteOperationType.REWRITE_MANIFESTS, null));
         break;
+      case HiveParser.KW_ANCESTORS_OF:
+        desc = getAncestorsOfDesc(tableName, partitionSpec, command);
+        break;
     }
 
     rootTasks.add(TaskFactory.get(new DDLWork(getInputs(), getOutputs(), desc)));
@@ -216,6 +220,20 @@
     return new AlterTableExecuteDesc(tableName, partitionSpec, spec);
   }
 
+  private AlterTableExecuteDesc getAncestorsOfDesc(
+      TableName tableName, Map<String, String> partitionSpec, ASTNode command)
+      throws SemanticException {
+    Long snapshotId = null;
+    if (command.getChildCount() == 2) {
+      ASTNode childNode = (ASTNode) command.getChild(1);
+      snapshotId = Long.parseLong(childNode.getText());
+    }
+    AlterTableExecuteSpec spec =
+        new AlterTableExecuteSpec(
+            ANCESTORS_OF, new AlterTableExecuteSpec.AncestorsOfSpec(snapshotId));
+    return new AlterTableExecuteDesc(tableName, partitionSpec, spec);
+  }
+
   private long getTimeStampMillis(ASTNode childNode) {
     String childNodeText = PlanUtils.stripQuotes(childNode.getText());
     ZoneId timeZone = conf.getLocalTimeZone();

ql/src/java/org/apache/hadoop/hive/ql/parse/AlterTableExecuteSpec.java

@@ -24,6 +24,7 @@
 import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
 
 import java.util.Arrays;
+import org.jetbrains.annotations.NotNull;
 
 /**
  * Execute operation specification. It stores the type of the operation and its parameters.
@@ -44,7 +45,8 @@ public enum ExecuteOperationType {
     CHERRY_PICK,
     DELETE_METADATA,
     DELETE_ORPHAN_FILES,
-    REWRITE_MANIFESTS;
+    REWRITE_MANIFESTS,
+    ANCESTORS_OF;
   }
 
   private final ExecuteOperationType operationType;
@@ -257,6 +259,21 @@ public String toString() {
     }
   }
 
+  /**
+   * Value object class, that stores the ancestors of operation specific parameters.
+   *
+   * <ul>
+   *   <li>snapshotId: the snapshotId to find ancestors of (optional)
+   * </ul>
+   */
+  public record AncestorsOfSpec(Long snapshotId) {
+
+    @Override
+    public @NotNull String toString() {
+      return MoreObjects.toStringHelper(this).add("snapshotId", snapshotId).toString();
+    }
+  }
+
   public static class DeleteMetadataSpec {
     private final String branchName;
     private final SearchArgument sarg;