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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
