From 93b0d5b666ccc39570495b35aada6e5e1dc74b47 Mon Sep 17 00:00:00 2001 From: Raghav Aggarwal Date: Wed, 5 Aug 2026 13:56:04 +0530 Subject: [PATCH] HIVE-29803: [Iceberg] Support ANCESTORS_OF snapshot lineage command --- .../mr/hive/HiveIcebergStorageHandler.java | 5 +++ .../iceberg/mr/hive/IcebergTableUtil.java | 39 ++++++++++++++++++ .../TestHiveIcebergSnapshotOperations.java | 41 +++++++++++++++++++ .../hadoop/hive/ql/parse/AlterClauseParser.g | 2 + .../hadoop/hive/ql/parse/HiveLexerParent.g | 1 + .../hadoop/hive/ql/parse/IdentifiersParser.g | 1 + .../execute/AlterTableExecuteAnalyzer.java | 18 ++++++++ .../hive/ql/parse/AlterTableExecuteSpec.java | 19 ++++++++- 8 files changed, 125 insertions(+), 1 deletion(-) diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java index c6982640d4a7..1656d14e1358 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java +++ b/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(); diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java index dd613f452bef..2b85f95a1255 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java @@ -999,4 +999,43 @@ public static boolean supportsRowLineage(Map tableProperties) { 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 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 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()); + } + } + } } diff --git a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergSnapshotOperations.java b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergSnapshotOperations.java index 23229a3c8b79..b036685a0095 100644 --- a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergSnapshotOperations.java +++ b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergSnapshotOperations.java @@ -238,4 +238,45 @@ private int generateManifestsAndRewrite(TableIdentifier identifier, int expected 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")); + } + } } diff --git a/parser/src/java/org/apache/hadoop/hive/ql/parse/AlterClauseParser.g b/parser/src/java/org/apache/hadoop/hive/ql/parse/AlterClauseParser.g index f34c3aec7a0e..9fab8cc89f2c 100644 --- a/parser/src/java/org/apache/hadoop/hive/ql/parse/AlterClauseParser.g +++ b/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)? diff --git a/parser/src/java/org/apache/hadoop/hive/ql/parse/HiveLexerParent.g b/parser/src/java/org/apache/hadoop/hive/ql/parse/HiveLexerParent.g index a96812d49698..0afbef2d2f6d 100644 --- a/parser/src/java/org/apache/hadoop/hive/ql/parse/HiveLexerParent.g +++ b/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'; diff --git a/parser/src/java/org/apache/hadoop/hive/ql/parse/IdentifiersParser.g b/parser/src/java/org/apache/hadoop/hive/ql/parse/IdentifiersParser.g index 37fd6187d16d..f3594fc940c0 100644 --- a/parser/src/java/org/apache/hadoop/hive/ql/parse/IdentifiersParser.g +++ b/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 diff --git a/ql/src/java/org/apache/hadoop/hive/ql/ddl/table/execute/AlterTableExecuteAnalyzer.java b/ql/src/java/org/apache/hadoop/hive/ql/ddl/table/execute/AlterTableExecuteAnalyzer.java index 13bb7aa6f7f7..207ccfb1d645 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/ddl/table/execute/AlterTableExecuteAnalyzer.java +++ b/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; @@ -107,6 +108,9 @@ protected void analyzeCommand(TableName tableName, Map partition 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 @@ private AlterTableExecuteDesc getDeleteOrphanFilesDesc(TableName tableName, Map< return new AlterTableExecuteDesc(tableName, partitionSpec, spec); } + private AlterTableExecuteDesc getAncestorsOfDesc( + TableName tableName, Map 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(); diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/AlterTableExecuteSpec.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/AlterTableExecuteSpec.java index 17b0b1631d1f..916f159e8c21 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/parse/AlterTableExecuteSpec.java +++ b/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. + * + *
    + *
  • snapshotId: the snapshotId to find ancestors of (optional) + *
+ */ + 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;