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: [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]
