This is an automated email from the ASF dual-hosted git repository.
okumin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new 4486ba4a3ba HIVE-29788: Support for rewrite_manifests iceberg
procedure (#6667)
4486ba4a3ba is described below
commit 4486ba4a3bae1cd4b1c4856710c1967ef4de7fd0
Author: Raghav Aggarwal <[email protected]>
AuthorDate: Wed Aug 12 14:50:07 2026 +0530
HIVE-29788: Support for rewrite_manifests iceberg procedure (#6667)
* HIVE-29788: Support for rewrite_manifests iceberg procedure
* Review comments
The current implementation targets only data manifests.
---
.../iceberg/mr/hive/HiveIcebergStorageHandler.java | 3 +
.../apache/iceberg/mr/hive/IcebergTableUtil.java | 111 +++++++++++++++++++
.../mr/hive/TestHiveIcebergSnapshotOperations.java | 119 +++++++++++++++++++++
.../hadoop/hive/ql/parse/AlterClauseParser.g | 2 +
.../apache/hadoop/hive/ql/parse/HiveLexerParent.g | 1 +
.../hadoop/hive/ql/parse/IdentifiersParser.g | 1 +
.../table/execute/AlterTableExecuteAnalyzer.java | 4 +
.../hive/ql/parse/AlterTableExecuteSpec.java | 3 +-
8 files changed, 243 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 070bd9b70ae..c6982640d4a 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
@@ -1358,6 +1358,9 @@ public void
executeOperation(org.apache.hadoop.hive.ql.metadata.Table hmsTable,
(AlterTableExecuteSpec.CherryPickSpec)
executeSpec.getOperationParams();
IcebergTableUtil.cherryPick(icebergTable,
cherryPickSpec.getSnapshotId());
break;
+ case REWRITE_MANIFESTS:
+ IcebergTableUtil.rewriteManifests(icebergTable);
+ 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 ad2ff8e4c2f..dd613f452be 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
@@ -33,6 +33,7 @@
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
+import java.util.TreeSet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
@@ -69,11 +70,14 @@
import org.apache.hadoop.hive.ql.session.SessionStateUtil;
import org.apache.hadoop.util.Sets;
import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
import org.apache.iceberg.DeleteFiles;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.ManageSnapshots;
import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
import org.apache.iceberg.MetadataTableType;
import org.apache.iceberg.MetadataTableUtils;
import org.apache.iceberg.PartitionData;
@@ -110,11 +114,13 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.transforms.Transform;
+import org.apache.iceberg.types.Comparators;
import org.apache.iceberg.types.Conversions;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.ByteBuffers;
import org.apache.iceberg.util.Pair;
+import org.apache.iceberg.util.PartitionUtil;
import org.apache.iceberg.util.SnapshotUtil;
import org.apache.iceberg.util.StructProjection;
import org.slf4j.Logger;
@@ -823,10 +829,115 @@ public static ExecutorService newDeleteThreadPool(String
completeName, int numTh
});
}
+ public static void rewriteManifests(Table table) {
+ // Skip empty tables that do not have a snapshot yet
+ if (table.currentSnapshot() == null) {
+ return;
+ }
+
+ // Unpartitioned tables can be safely clustered into a single bucket
+ if (!table.spec().isPartitioned()) {
+ table.rewriteManifests().clusterBy(file -> 0).commit();
+ return;
+ }
+
+ // Determine the target size for each new manifest file (defaults to 8MB)
+ long manifestTargetSizeBytes =
TableProperties.MANIFEST_TARGET_SIZE_BYTES_DEFAULT;
+ if
(table.properties().containsKey(TableProperties.MANIFEST_TARGET_SIZE_BYTES)) {
+ manifestTargetSizeBytes =
+
Long.parseLong(table.properties().get(TableProperties.MANIFEST_TARGET_SIZE_BYTES));
+ }
+
+ List<ManifestFile> dataManifests =
table.currentSnapshot().dataManifests(table.io());
+ if (dataManifests.isEmpty()) {
+ return;
+ }
+
+ // Calculate ideal number of manifest files based on current total
metadata size.
+ // We hard-cap the maximum number of clusters at 200 to prevent the JVM
from
+ // opening too many concurrent file writers and causing OOM or OS ulimit
(Too Many Open Files).
+ long totalManifestsSize =
dataManifests.stream().mapToLong(ManifestFile::length).sum();
+ int targetClusters =
+ (int)
+ Math.min(
+ (totalManifestsSize + manifestTargetSizeBytes - 1) /
manifestTargetSizeBytes, 200);
+
+ if (targetClusters <= 1) {
+ table.rewriteManifests().clusterBy(file -> 0).commit();
+ return;
+ }
+
+ // To cluster files efficiently, we want to group them naturally. We
extract the native
+ // Type of the first partition column (e.g. Timestamp, String) and use
Iceberg's native
+ // Comparators to maintain a sorted TreeSet of all unique partition values.
+ Type.PrimitiveType firstPartitionFieldType =
+
table.spec().partitionType().fields().getFirst().type().asPrimitiveType();
+ Set<Object> uniqueValues =
+ getUniquePartitionValues(table, dataManifests,
firstPartitionFieldType);
+
+ if (uniqueValues.isEmpty()) {
+ table.rewriteManifests().clusterBy(file -> 0).commit();
+ return;
+ }
+
+ // Divide the naturally sorted unique values evenly into our calculated
`targetClusters`
+ Object[] sortedValues = uniqueValues.toArray();
+ Map<Object, Integer> valueToBucket = Maps.newHashMap();
+ for (int i = 0; i < sortedValues.length; i++) {
+ // e.g. If we have 1000 sorted partition values and 200 clusters, this
groups 5 values per
+ // bucket ID
+ valueToBucket.put(sortedValues[i], i * targetClusters /
sortedValues.length);
+ }
+
+ // Rewrite manifests, telling Iceberg to group data files based on our
pre-calculated bucket
+ // mapping
+ table
+ .rewriteManifests()
+ .clusterBy(
+ file -> {
+ StructLike partition =
+ PartitionUtil.coercePartition(
+ table.spec().partitionType(),
+ table.specs().get(file.specId()),
+ file.partition());
+ Object value = partition.get(0, Object.class);
+ return value != null ? valueToBucket.getOrDefault(value, 0) : 0;
+ })
+ .commit();
+ }
+
public static boolean hasUndergonePartitionEvolution(Table table) {
return table.specs().size() > 1;
}
+ private static Set<Object> getUniquePartitionValues(
+ Table table, List<ManifestFile> dataManifests, Type.PrimitiveType
firstPartitionFieldType) {
+ Set<Object> uniqueValues = new
TreeSet<>(Comparators.forType(firstPartitionFieldType));
+
+ for (ManifestFile manifestFile : dataManifests) {
+ try (ManifestReader<DataFile> reader =
+ ManifestFiles.read(manifestFile, table.io(), table.specs())
+ .select(List.of(DataFile.PARTITION_NAME))) {
+ for (DataFile dataFile : reader) {
+ // Coerce partition struct in case of partition evolution
+ StructLike partition =
+ PartitionUtil.coercePartition(
+ table.spec().partitionType(),
+ table.specs().get(dataFile.specId()),
+ dataFile.partition());
+ // Only extract and sort by the FIRST partition column for read
optimization
+ Object value = partition.get(0, Object.class);
+ if (value != null) {
+ uniqueValues.add(value);
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to read manifest file", e);
+ }
+ }
+ return uniqueValues;
+ }
+
public static boolean hasUndergonePartitionEvolution(Snapshot snapshot,
FileIO io) {
if (snapshot == null) {
return false;
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 b483df82e43..23229a3c8b7 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
@@ -119,4 +119,123 @@ public void testReplaceBranchWithSnapshot() {
result = shell.executeStatement("SELECT COUNT(*) FROM
default.testReplaceBranchWithSnapshot.branch_branch1");
assertEquals(6L, result.get(0)[0]);
}
+
+ /**
+ * Tests executing REWRITE_MANIFESTS on an unpartitioned table to verify that
+ * multiple small manifests are successfully compacted into a single
optimized manifest.
+ */
+ @Test
+ public void testRewriteManifests() {
+ TableIdentifier identifier = TableIdentifier.of("default",
"testRewriteManifests");
+ shell.executeStatement(
+ String.format(
+ "CREATE EXTERNAL TABLE %s (id INT, data STRING) STORED BY iceberg
%s %s",
+ identifier.name(),
+ testTables.locationForCreateTableSQL(identifier),
+
testTables.propertiesForCreateTableSQL(ImmutableMap.of("commit.manifest.min-count-to-compact",
"2"))));
+
+ // Create 5 manifests by executing 5 separate INSERT operations
+ for (int i = 1; i <= 5; i++) {
+ shell.executeStatement(
+ String.format("INSERT INTO TABLE %s VALUES(%d, 'val')",
identifier.name(), i));
+ }
+
+ int manifestCountAfterRewrite = generateManifestsAndRewrite(identifier, 5);
+ assertEquals("Should have exactly 1 manifest after REWRITE_MANIFESTS", 1,
manifestCountAfterRewrite);
+ }
+
+ /**
+ * Tests executing REWRITE_MANIFESTS on a partitioned table to verify that
multiple manifests
+ * spread across different partitions are correctly clustered and compacted
into fewer, optimized
+ * manifests.
+ */
+ @Test
+ public void testRewriteManifestsPartitioned() {
+ TableIdentifier identifier = TableIdentifier.of("default",
"testRewriteManifestsPartitioned");
+ shell.executeStatement(
+ String.format(
+ "CREATE EXTERNAL TABLE %s (id INT, data STRING) PARTITIONED BY
(part STRING) STORED BY iceberg %s %s",
+ identifier.name(),
+ testTables.locationForCreateTableSQL(identifier),
+ testTables.propertiesForCreateTableSQL(
+ ImmutableMap.of("commit.manifest.min-count-to-compact",
"2"))));
+
+ // Create 5 manifests by executing 5 separate INSERT operations across 2
partitions
+ for (int i = 1; i <= 5; i++) {
+ String partitionVal = (i % 2 == 0) ? "p2" : "p1";
+ shell.executeStatement(
+ String.format(
+ "INSERT INTO TABLE %s VALUES(%d, 'val', '%s')",
identifier.name(), i, partitionVal));
+ }
+
+ int manifestCountAfterRewrite = generateManifestsAndRewrite(identifier, 5);
+ assertEquals("Should have exactly 1 manifest after REWRITE_MANIFESTS", 1,
manifestCountAfterRewrite);
+ }
+
+ /**
+ * Tests that executing REWRITE_MANIFESTS on an empty table (with no
existing snapshots) safely
+ * acts as a no-op without throwing any exceptions.
+ */
+ @Test
+ public void testRewriteManifestsEmptyTable() {
+ TableIdentifier identifier = TableIdentifier.of("default",
"testRewriteManifestsEmptyTable");
+ shell.executeStatement(
+ String.format(
+ "CREATE EXTERNAL TABLE %s (id INT) STORED BY iceberg %s %s",
+ identifier.name(),
+ testTables.locationForCreateTableSQL(identifier),
+ testTables.propertiesForCreateTableSQL(ImmutableMap.of())));
+
+ // Execute on a table with no snapshot
+ shell.executeStatement(
+ String.format("ALTER TABLE %s EXECUTE REWRITE_MANIFESTS",
identifier.name()));
+
+ org.apache.iceberg.Table icebergTable = testTables.loadTable(identifier);
+ Assert.assertNull(icebergTable.currentSnapshot());
+ }
+
+ /**
+ * Tests REWRITE_MANIFESTS with a very small target size to ensure that
multiple output manifests
+ * are created, proving the bucketing and splitting logic works.
+ */
+ @Test
+ public void testRewriteManifestsMultipleOutputManifests() {
+ TableIdentifier identifier =
+ TableIdentifier.of("default", "testRewriteManifestsMultipleOutputs");
+ shell.executeStatement(
+ String.format(
+ "CREATE EXTERNAL TABLE %s (id INT, data STRING) PARTITIONED BY
(part STRING) STORED BY iceberg %s %s",
+ identifier.name(),
+ testTables.locationForCreateTableSQL(identifier),
+ testTables.propertiesForCreateTableSQL(
+ ImmutableMap.of(
+ "commit.manifest.min-count-to-compact", "100",
+ "commit.manifest.target-size-bytes",
+ "1")))); // Force multiple clusters and splits
+
+ for (int i = 1; i <= 15; i++) {
+ String partitionVal = "p" + (i % 5);
+ shell.executeStatement(
+ String.format(
+ "INSERT INTO TABLE %s VALUES(%d, 'val', '%s')",
identifier.name(), i, partitionVal));
+ }
+
+ int manifestCountAfterRewrite = generateManifestsAndRewrite(identifier,
15);
+ Assert.assertTrue("Should have multiple manifests due to
bucketing/splitting", manifestCountAfterRewrite > 1);
+ }
+
+ private int generateManifestsAndRewrite(TableIdentifier identifier, int
expectedBefore) {
+ org.apache.iceberg.Table icebergTable = testTables.loadTable(identifier);
+ icebergTable.refresh();
+
+ int manifestCountBefore =
icebergTable.currentSnapshot().allManifests(icebergTable.io()).size();
+ assertEquals(
+ "Manifests keep accumulating for each insert", expectedBefore,
manifestCountBefore);
+
+ shell.executeStatement(
+ String.format("ALTER TABLE %s EXECUTE REWRITE_MANIFESTS",
identifier.name()));
+
+ icebergTable.refresh();
+ return
icebergTable.currentSnapshot().allManifests(icebergTable.io()).size();
+ }
}
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 81a3ea6774c..f34c3aec7a0 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
@@ -534,6 +534,8 @@ alterStatementSuffixExecute
-> ^(TOK_ALTERTABLE_EXECUTE KW_ROLLBACK $rollbackParam)
| KW_EXECUTE KW_EXPIRE_SNAPSHOTS (LPAREN (expireParam=expression) RPAREN)?
-> ^(TOK_ALTERTABLE_EXECUTE KW_EXPIRE_SNAPSHOTS $expireParam?)
+ | KW_EXECUTE KW_REWRITE_MANIFESTS
+ -> ^(TOK_ALTERTABLE_EXECUTE KW_REWRITE_MANIFESTS)
| 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 cb3404587d8..a96812d4969 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
@@ -398,6 +398,7 @@ KW_SPEC: 'SPEC';
KW_SYSTEM_TIME: 'SYSTEM_TIME';
KW_SYSTEM_VERSION: 'SYSTEM_VERSION';
KW_EXPIRE_SNAPSHOTS: 'EXPIRE_SNAPSHOTS';
+KW_REWRITE_MANIFESTS: 'REWRITE_MANIFESTS';
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 ac9053cf284..37fd6187d16 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
@@ -1027,6 +1027,7 @@ nonReserved
| KW_SPEC
| KW_SYSTEM_TIME | KW_SYSTEM_VERSION
| KW_EXPIRE_SNAPSHOTS
+ | KW_REWRITE_MANIFESTS
| 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 baba8353430..13bb7aa6f7f 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
@@ -103,6 +103,10 @@ protected void analyzeCommand(TableName tableName,
Map<String, String> partition
case HiveParser.KW_ORPHAN_FILES:
desc = getDeleteOrphanFilesDesc(tableName, partitionSpec,
command.getChildren());
break;
+ case HiveParser.KW_REWRITE_MANIFESTS:
+ desc = new AlterTableExecuteDesc(tableName, partitionSpec,
+ new
AlterTableExecuteSpec(AlterTableExecuteSpec.ExecuteOperationType.REWRITE_MANIFESTS,
null));
+ break;
}
rootTasks.add(TaskFactory.get(new DDLWork(getInputs(), getOutputs(),
desc)));
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 54c8df3573c..17b0b1631d1 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
@@ -43,7 +43,8 @@ public enum ExecuteOperationType {
FAST_FORWARD,
CHERRY_PICK,
DELETE_METADATA,
- DELETE_ORPHAN_FILES;
+ DELETE_ORPHAN_FILES,
+ REWRITE_MANIFESTS;
}
private final ExecuteOperationType operationType;
