You are viewing a plain text version of this content. The canonical link for it is here.
Posted to gitbox@hive.apache.org by "Aggarwal-Raghav (via GitHub)" <gi...@apache.org> on 2026/08/02 08:20:54 UTC
[PR] HIVE-29788: Support for rewrite_manifests iceberg procedure [hive]
Aggarwal-Raghav opened a new pull request, #6667:
URL: https://github.com/apache/hive/pull/6667
### What changes were proposed in this pull request?
Add support for the Iceberg `REWRITE_MANIFESTS` procedure in Hive https://iceberg.apache.org/docs/latest/spark-procedures/#rewrite_manifests
The implementation is very similar to Trino's approach https://github.com/trinodb/trino/blob/master/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/procedure/OptimizeManifests.java
### Why are the changes needed?
Optimize manifest files using hive engine rather than running spark sql `CALL spark_catalog.system.rewrite_manifests(...)`. Also, hive `EXPIRE_SNAPSHOTS` or `OPTIMIZE TABLE` doesn't clean up manifest files.
### Does this PR introduce _any_ user-facing change?
Yes. A new SQL command is now available for Hive-Iceberg users:
```ALTER TABLE <table> EXECUTE REWRITE_MANIFESTS```
### How was this patch tested?
Using UT. Added `testRewriteManifests` and `testRewriteManifestsPartitioned` in `TestHiveIcebergSnapshotOperations`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] HIVE-29788: Support for rewrite_manifests iceberg procedure [hive]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #6667:
URL: https://github.com/apache/hive/pull/6667#issuecomment-5245636743
## [](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=6667) **Quality Gate passed**
Issues
 [3 New issues](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=6667&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=6667&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=6667&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=6667&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=6667&metric=new_duplicated_lines_density&view=list)
<!-- sqra-placement-anchor -->
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=6667)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] HIVE-29788: Support for rewrite_manifests iceberg procedure [hive]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin commented on code in PR #6667:
URL: https://github.com/apache/hive/pull/6667#discussion_r3755977949
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java:
##########
@@ -887,99 +887,114 @@ 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 -> "").commit();
- } else {
- // 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));
- }
+ table.rewriteManifests().clusterBy(file -> 0).commit();
+ return;
+ }
- List<ManifestFile> dataManifests = table.currentSnapshot().dataManifests(table.io());
- if (dataManifests.isEmpty()) {
- 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));
+ }
- // 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;
- }
+ List<ManifestFile> dataManifests = table.currentSnapshot().dataManifests(table.io());
+ if (dataManifests.isEmpty()) {
+ 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 = 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);
- }
- }
+ // 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;
+ }
- if (uniqueValues.isEmpty()) {
- 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);
- // 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);
- }
+ if (uniqueValues.isEmpty()) {
+ table.rewriteManifests().clusterBy(file -> 0).commit();
+ return;
+ }
- // 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();
+ // 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));
Review Comment:
Thanks for reducing the cognitive complexity per method 👍
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java:
##########
@@ -880,6 +886,96 @@ public static ExecutorService newDeleteThreadPool(String completeName, int numTh
});
}
+ public static void rewriteManifests(Table table) {
+ if (!table.spec().isPartitioned()) {
+ table.rewriteManifests().clusterBy(file -> "").commit();
+ } else {
+ // 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 = 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);
+ }
+ }
+
+ 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();
+ }
+ }
Review Comment:
I am feeling Spark's procedure can optimize delete manifests as well. I don't know how it matters or how to test the three types of deletion files. Therefore, let's make it out of scope of HIVE-29788.
https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java#L203-L225
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] HIVE-29788: Support for rewrite_manifests iceberg procedure [hive]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin commented on PR #6667:
URL: https://github.com/apache/hive/pull/6667#issuecomment-5264823452
Merged. @Aggarwal-Raghav Thanks for implementing the missing feature! We are heading for the perfect Iceberg platform!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] HIVE-29788: Support for rewrite_manifests iceberg procedure [hive]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin commented on code in PR #6667:
URL: https://github.com/apache/hive/pull/6667#discussion_r3751231681
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java:
##########
@@ -880,6 +886,96 @@ public static ExecutorService newDeleteThreadPool(String completeName, int numTh
});
}
+ public static void rewriteManifests(Table table) {
+ if (!table.spec().isPartitioned()) {
+ table.rewriteManifests().clusterBy(file -> "").commit();
Review Comment:
Trino uses zero. I guess any value is OK.
https://github.com/trinodb/trino/blob/67f588f0c81b21b425e1a43b05d70f9cf8798d6c/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/procedure/OptimizeManifests.java#L100
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java:
##########
@@ -880,6 +886,96 @@ public static ExecutorService newDeleteThreadPool(String completeName, int numTh
});
}
+ public static void rewriteManifests(Table table) {
+ if (!table.spec().isPartitioned()) {
+ table.rewriteManifests().clusterBy(file -> "").commit();
+ } else {
+ // 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());
Review Comment:
`table.currentSnapshot()` can return null.
##########
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 (LPAREN (rewriteManifestsParam=expression) RPAREN)?
+ -> ^(TOK_ALTERTABLE_EXECUTE KW_REWRITE_MANIFESTS $rewriteManifestsParam?)
Review Comment:
How do we use the param?
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java:
##########
@@ -880,6 +886,96 @@ public static ExecutorService newDeleteThreadPool(String completeName, int numTh
});
}
+ public static void rewriteManifests(Table table) {
+ if (!table.spec().isPartitioned()) {
+ table.rewriteManifests().clusterBy(file -> "").commit();
Review Comment:
As the other line uses not `""` but `0`, I feel we want to align L891 with 0.
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java:
##########
@@ -880,6 +886,96 @@ public static ExecutorService newDeleteThreadPool(String completeName, int numTh
});
}
+ public static void rewriteManifests(Table table) {
+ if (!table.spec().isPartitioned()) {
+ table.rewriteManifests().clusterBy(file -> "").commit();
Review Comment:
Also, though this is my preference, I'd like to add an early return here, so that we can unnest the `else` block
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java:
##########
@@ -880,6 +886,96 @@ public static ExecutorService newDeleteThreadPool(String completeName, int numTh
});
}
+ public static void rewriteManifests(Table table) {
+ if (!table.spec().isPartitioned()) {
+ table.rewriteManifests().clusterBy(file -> "").commit();
+ } else {
+ // 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 = 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);
+ }
+ }
+
+ 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();
+ }
+ }
Review Comment:
Does this work for delete files?
##########
iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergSnapshotOperations.java:
##########
@@ -119,4 +119,76 @@ public void testReplaceBranchWithSnapshot() {
result = shell.executeStatement("SELECT COUNT(*) FROM default.testReplaceBranchWithSnapshot.branch_branch1");
assertEquals(6L, result.get(0)[0]);
}
+
+ @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));
+ }
+
+ org.apache.iceberg.Table icebergTable = testTables.loadTable(identifier);
+ icebergTable.refresh();
+
+ // After 5 inserts, Iceberg will have generated 5 separate manifest files
+ int manifestCountBefore = icebergTable.currentSnapshot().allManifests(icebergTable.io()).size();
+ assertEquals("Manifests keep accumulating for each insert", 5, manifestCountBefore);
+
+ // Execute REWRITE_MANIFESTS procedure
+ shell.executeStatement(
+ String.format("ALTER TABLE %s EXECUTE REWRITE_MANIFESTS", identifier.name()));
+
+ icebergTable.refresh();
+ int manifestCountAfterRewrite =
+ icebergTable.currentSnapshot().allManifests(icebergTable.io()).size();
+
+ // Rewrite manifests should confidently compact all of them into exactly 1 manifest
+ assertEquals(
+ "Should have exactly 1 manifest after REWRITE_MANIFESTS", 1, manifestCountAfterRewrite);
+ }
+
+ @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));
+ }
+
+ 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", 5, manifestCountBefore);
+
+ shell.executeStatement(
+ String.format("ALTER TABLE %s EXECUTE REWRITE_MANIFESTS", identifier.name()));
+
+ icebergTable.refresh();
+ int manifestCountAfterRewrite =
+ icebergTable.currentSnapshot().allManifests(icebergTable.io()).size();
+
+ assertEquals(
+ "Should have exactly 1 manifest after REWRITE_MANIFESTS", 1, manifestCountAfterRewrite);
+ }
Review Comment:
We may want some more cases.
- No snapshot
- The number of data files or manifests is zero
- A large case
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] HIVE-29788: Support for rewrite_manifests iceberg procedure [hive]
Posted by "Aggarwal-Raghav (via GitHub)" <gi...@apache.org>.
Aggarwal-Raghav commented on code in PR #6667:
URL: https://github.com/apache/hive/pull/6667#discussion_r3752090090
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java:
##########
@@ -880,6 +886,96 @@ public static ExecutorService newDeleteThreadPool(String completeName, int numTh
});
}
+ public static void rewriteManifests(Table table) {
+ if (!table.spec().isPartitioned()) {
+ table.rewriteManifests().clusterBy(file -> "").commit();
+ } else {
+ // 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 = 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);
+ }
+ }
+
+ 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();
+ }
+ }
Review Comment:
Iceberg's `table.rewriteManifests()` API only targets Data Manifests.
https://github.com/apache/iceberg/blob/89e2f887491c1b5fa9f8b9de81b3aa8b31fa6974/api/src/main/java/org/apache/iceberg/RewriteManifests.java#L49
The arg takes `DataFile` lambda
```java
RewriteManifests clusterBy(Function<DataFile, Object> func);
```
Delete Manifests are completely ignored during this rewrite. If there are large number of delete files in a table then compaction should be preferred choice.
##########
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 (LPAREN (rewriteManifestsParam=expression) RPAREN)?
+ -> ^(TOK_ALTERTABLE_EXECUTE KW_REWRITE_MANIFESTS $rewriteManifestsParam?)
Review Comment:
Mistake from me, initially I copied from expire snapshots i.e. 2 lines above and forgot to update later 😬
Will address it. Thanks for catching this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] HIVE-29788: Support for rewrite_manifests iceberg procedure [hive]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #6667:
URL: https://github.com/apache/hive/pull/6667#issuecomment-5156842667
## [](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=6667) **Quality Gate passed**
Issues
 [4 New issues](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=6667&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=6667&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=6667&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=6667&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=6667&metric=new_duplicated_lines_density&view=list)
<!-- sqra-placement-anchor -->
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=6667)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] HIVE-29788: Support for rewrite_manifests iceberg procedure [hive]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin merged PR #6667:
URL: https://github.com/apache/hive/pull/6667
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
Re: [PR] HIVE-29788: Support for rewrite_manifests iceberg procedure [hive]
Posted by "Aggarwal-Raghav (via GitHub)" <gi...@apache.org>.
Aggarwal-Raghav commented on PR #6667:
URL: https://github.com/apache/hive/pull/6667#issuecomment-5256207740
@okumin , **thanks a lot for review.** I agree that spark is doing the rewrite of delelte manifest as well. Keeping delete manifests out of scope for this PR makes perfect sense.
Maybe during merge you can edit the commit message or add a line "Applicable only for data manifests" or something along the lines.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org