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