Skip to content

Commit

Permalink
Collect data size statistics for Iceberg tables
Browse files Browse the repository at this point in the history 
Previously, the data size statistic was computed by using the
Iceberg data manifests data size field. This is value is misleading
for Presto because it represents the compressed on-disk size.
This change allows ANALYZE to read and write data size statistic
values to puffin files.

This change also updates the hive-statistics-merge-strategy
config value in the Iceberg connector to accept a comma-separated
list of valid values to override from the HMS instead of using
an independent enum. This allows for a wider variety of combinations
using less code.
  • Loading branch information
@ZacBlanco
ZacBlanco committed Apr 1, 2024
1 parent 71bb174 commit 437b48a
Show file tree
Hide file tree
Showing 13 changed files with 421 additions and 211 deletions.
17 changes: 11 additions & 6 deletions presto-docs/src/main/sphinx/connector/iceberg.rst
View file
Original file line number Diff line number Diff line change
Expand Up @@ -225,10 +225,13 @@ Property Name Description

``iceberg.enable-parquet-dereference-pushdown`` Enable parquet dereference pushdown. ``true``

``iceberg.hive-statistics-merge-strategy`` Determines how to merge statistics that are stored in the ``NONE``
Hive Metastore. The available values are ``NONE``,
``USE_NULLS_FRACTION_AND_NDV``, ``USE_NULLS_FRACTIONS``
and, ``USE_NDV``
``iceberg.hive-statistics-merge-strategy`` Comma separated list of statistics to use from the
Hive Metastore to override Iceberg table statistics.
The available values are ``NUMBER_OF_DISTINCT_VALUES``
and ``TOTAL_SIZE_IN_BYTES``.

**Note**: Only valid when the Iceberg connector is
configured with Hive.

``iceberg.statistic-snapshot-record-difference-weight`` The amount that the difference in total record count matters
when calculating the closest snapshot when picking
Expand Down Expand Up @@ -306,6 +309,8 @@ Property Name Description
============================================= ======================================================================
``iceberg.delete_as_join_rewrite_enabled`` Overrides the behavior of the connector property
``iceberg.delete-as-join-rewrite-enabled`` in the current session.
``iceberg.hive_statistics_merge_strategy`` Overrides the behavior of the connector property
``iceberg.hive-statistics-merge-strategy`` in the current session.
============================================= ======================================================================

Caching Support
Expand Down Expand Up @@ -1172,7 +1177,7 @@ each Iceberg data type to the corresponding Presto data type, and from each Pres
The following tables detail the specific type maps between PrestoDB and Iceberg.

Iceberg to PrestoDB type mapping
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Map of Iceberg types to the relevant PrestoDB types:

Expand Down Expand Up @@ -1215,7 +1220,7 @@ Map of Iceberg types to the relevant PrestoDB types:
No other types are supported.

PrestoDB to Iceberg type mapping
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Map of PrestoDB types to the relevant Iceberg types:

Expand Down
17 changes: 10 additions & 7 deletions presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergConfig.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
import com.facebook.airlift.configuration.Config;
import com.facebook.airlift.configuration.ConfigDescription;
import com.facebook.presto.hive.HiveCompressionCodec;
import com.facebook.presto.iceberg.util.HiveStatisticsMergeStrategy;
import com.facebook.presto.spi.statistics.ColumnStatisticType;
import com.google.common.base.Splitter;
import com.google.common.collect.ImmutableList;
import org.apache.iceberg.hadoop.HadoopFileIO;
Expand All @@ -26,11 +26,13 @@
import javax.validation.constraints.Min;
import javax.validation.constraints.NotNull;

import java.util.EnumSet;
import java.util.List;

import static com.facebook.presto.hive.HiveCompressionCodec.GZIP;
import static com.facebook.presto.iceberg.CatalogType.HIVE;
import static com.facebook.presto.iceberg.IcebergFileFormat.PARQUET;
import static com.facebook.presto.iceberg.util.StatisticsUtil.decodeMergeFlags;
import static org.apache.iceberg.CatalogProperties.IO_MANIFEST_CACHE_EXPIRATION_INTERVAL_MS_DEFAULT;
import static org.apache.iceberg.CatalogProperties.IO_MANIFEST_CACHE_MAX_CONTENT_LENGTH_DEFAULT;
import static org.apache.iceberg.CatalogProperties.IO_MANIFEST_CACHE_MAX_TOTAL_BYTES_DEFAULT;
Expand All @@ -51,12 +53,13 @@ public class IcebergConfig
private boolean pushdownFilterEnabled;
private boolean deleteAsJoinRewriteEnabled = true;

private HiveStatisticsMergeStrategy hiveStatisticsMergeStrategy = HiveStatisticsMergeStrategy.NONE;
private EnumSet<ColumnStatisticType> hiveStatisticsMergeFlags = EnumSet.noneOf(ColumnStatisticType.class);
private String fileIOImpl = HadoopFileIO.class.getName();
private boolean manifestCachingEnabled;
private long maxManifestCacheSize = IO_MANIFEST_CACHE_MAX_TOTAL_BYTES_DEFAULT;
private long manifestCacheExpireDuration = IO_MANIFEST_CACHE_EXPIRATION_INTERVAL_MS_DEFAULT;
private long manifestCacheMaxContentLength = IO_MANIFEST_CACHE_MAX_CONTENT_LENGTH_DEFAULT;

@NotNull
public FileFormat getFileFormat()
{
Expand Down Expand Up @@ -195,16 +198,16 @@ public boolean isMergeOnReadModeEnabled()
}

@Config("iceberg.hive-statistics-merge-strategy")
@ConfigDescription("determines how to merge statistics that are stored in the Hive Metastore")
public IcebergConfig setHiveStatisticsMergeStrategy(HiveStatisticsMergeStrategy mergeStrategy)
@ConfigDescription("Comma separated list of statistics to use from the Hive metastore to override iceberg table statistics")
public IcebergConfig setHiveStatisticsMergeFlags(String mergeFlags)
{
this.hiveStatisticsMergeStrategy = mergeStrategy;
this.hiveStatisticsMergeFlags = decodeMergeFlags(mergeFlags);
return this;
}

public HiveStatisticsMergeStrategy getHiveStatisticsMergeStrategy()
public EnumSet<ColumnStatisticType> getHiveStatisticsMergeFlags()
{
return hiveStatisticsMergeStrategy;
return hiveStatisticsMergeFlags;
}

@Config("iceberg.statistic-snapshot-record-difference-weight")
Expand Down
35 changes: 24 additions & 11 deletions presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergHiveMetadata.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,6 @@
import com.facebook.presto.hive.metastore.PrestoTableType;
import com.facebook.presto.hive.metastore.PrincipalPrivileges;
import com.facebook.presto.hive.metastore.Table;
import com.facebook.presto.iceberg.util.HiveStatisticsMergeStrategy;
import com.facebook.presto.spi.ColumnHandle;
import com.facebook.presto.spi.ConnectorNewTableLayout;
import com.facebook.presto.spi.ConnectorOutputTableHandle;
Expand All @@ -57,6 +56,7 @@
import com.facebook.presto.spi.relation.VariableReferenceExpression;
import com.facebook.presto.spi.security.PrestoPrincipal;
import com.facebook.presto.spi.statistics.ColumnStatisticMetadata;
import com.facebook.presto.spi.statistics.ColumnStatisticType;
import com.facebook.presto.spi.statistics.ColumnStatistics;
import com.facebook.presto.spi.statistics.ComputedStatistics;
import com.facebook.presto.spi.statistics.Estimate;
Expand All @@ -82,6 +82,7 @@
import java.io.IOException;
import java.time.ZoneId;
import java.util.Collection;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
Expand Down Expand Up @@ -442,7 +443,7 @@ public TableStatistics getTableStatistics(ConnectorSession session, ConnectorTab
IcebergTableHandle handle = (IcebergTableHandle) tableHandle;
org.apache.iceberg.Table icebergTable = getIcebergTable(session, handle.getSchemaTableName());
TableStatistics icebergStatistics = TableStatisticsMaker.getTableStatistics(session, typeManager, constraint, handle, icebergTable, columnHandles.stream().map(IcebergColumnHandle.class::cast).collect(Collectors.toList()));
HiveStatisticsMergeStrategy mergeStrategy = getHiveStatisticsMergeStrategy(session);
EnumSet<ColumnStatisticType> mergeFlags = getHiveStatisticsMergeStrategy(session);
return tableLayoutHandle.map(IcebergTableLayoutHandle.class::cast).map(layoutHandle -> {
TupleDomain<ColumnHandle> domainPredicate = layoutHandle.getDomainPredicate()
.transform(subfield -> isEntireColumn(subfield) ? subfield.getRootName() : null)
Expand All @@ -453,9 +454,13 @@ public TableStatistics getTableStatistics(ConnectorSession session, ConnectorTab
IcebergColumnHandle columnHandle = (IcebergColumnHandle) icebergLayout;
return new VariableReferenceExpression(Optional.empty(), columnHandle.getName(), columnHandle.getType());
});
RowExpression translatedPredicate = rowExpressionService.getDomainTranslator().toPredicate(predicate);
PartitionStatistics hiveStatistics = metastore.getTableStatistics(getMetastoreContext(session), handle.getSchemaName(), handle.getIcebergTableName().getTableName());
TableStatistics mergedStatistics = mergeHiveStatistics(icebergStatistics, hiveStatistics, mergeStrategy, icebergTable.spec());
TableStatistics mergedStatistics = Optional.of(mergeFlags)
.filter(set -> !set.isEmpty())
.map(flags -> {
PartitionStatistics hiveStatistics = metastore.getTableStatistics(getMetastoreContext(session), handle.getSchemaName(), handle.getIcebergTableName().getTableName());
return mergeHiveStatistics(icebergStatistics, hiveStatistics, mergeFlags, icebergTable.spec());
})
.orElse(icebergStatistics);
TableStatistics.Builder filteredStatsBuilder = TableStatistics.builder()
.setRowCount(mergedStatistics.getRowCount());
double totalSize = 0;
Expand All @@ -469,6 +474,7 @@ public TableStatistics getTableStatistics(ConnectorSession session, ConnectorTab
}
}
}
RowExpression translatedPredicate = rowExpressionService.getDomainTranslator().toPredicate(predicate);
filteredStatsBuilder.setTotalSize(Estimate.of(totalSize));
return filterStatsCalculatorService.filterStats(
filteredStatsBuilder.build(),
Expand All @@ -481,9 +487,9 @@ public TableStatistics getTableStatistics(ConnectorSession session, ConnectorTab
IcebergColumnHandle::getName,
IcebergColumnHandle::getType)));
}).orElseGet(() -> {
if (!mergeStrategy.equals(HiveStatisticsMergeStrategy.NONE)) {
PartitionStatistics hiveStats = metastore.getTableStatistics(getMetastoreContext(session), handle.getSchemaName(), handle.getIcebergTableName().getTableName());
return mergeHiveStatistics(icebergStatistics, hiveStats, mergeStrategy, icebergTable.spec());
PartitionStatistics hiveStats = metastore.getTableStatistics(getMetastoreContext(session), handle.getSchemaName(), handle.getIcebergTableName().getTableName());
if (!mergeFlags.isEmpty()) {
return mergeHiveStatistics(icebergStatistics, hiveStats, mergeFlags, icebergTable.spec());
}
return icebergStatistics;
});
Expand All @@ -500,9 +506,16 @@ public TableStatisticsMetadata getStatisticsCollectionMetadata(ConnectorSession
{
Set<ColumnStatisticMetadata> columnStatistics = tableMetadata.getColumns().stream()
.filter(column -> !column.isHidden())
.flatMap(meta -> metastore.getSupportedColumnStatistics(getMetastoreContext(session), meta.getType())
.stream()
.map(statType -> statType.getColumnStatisticMetadata(meta.getName())))
.flatMap(meta -> {
try {
return metastore.getSupportedColumnStatistics(getMetastoreContext(session), meta.getType())
.stream()
.map(statType -> statType.getColumnStatisticMetadata(meta.getName()));
}
catch (IllegalArgumentException e) {
return ImmutableSet.<ColumnStatisticMetadata>of().stream();
}
})
.collect(toImmutableSet());

Set<TableStatisticType> tableStatistics = ImmutableSet.of(ROW_COUNT);
Expand Down
23 changes: 14 additions & 9 deletions presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergSessionProperties.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,24 +18,29 @@
import com.facebook.presto.hive.OrcFileWriterConfig;
import com.facebook.presto.hive.ParquetFileWriterConfig;
import com.facebook.presto.iceberg.nessie.NessieConfig;
import com.facebook.presto.iceberg.util.HiveStatisticsMergeStrategy;
import com.facebook.presto.iceberg.util.StatisticsUtil;
import com.facebook.presto.spi.ConnectorSession;
import com.facebook.presto.spi.session.PropertyMetadata;
import com.facebook.presto.spi.statistics.ColumnStatisticType;
import com.google.common.base.Joiner;
import com.google.common.collect.ImmutableList;
import io.airlift.units.DataSize;
import org.apache.parquet.column.ParquetProperties;

import javax.inject.Inject;

import java.util.EnumSet;
import java.util.List;

import static com.facebook.presto.common.type.VarcharType.VARCHAR;
import static com.facebook.presto.common.type.VarcharType.createUnboundedVarcharType;
import static com.facebook.presto.iceberg.util.StatisticsUtil.decodeMergeFlags;
import static com.facebook.presto.spi.session.PropertyMetadata.booleanProperty;
import static com.facebook.presto.spi.session.PropertyMetadata.doubleProperty;
import static com.facebook.presto.spi.session.PropertyMetadata.integerProperty;
import static com.facebook.presto.spi.session.PropertyMetadata.stringProperty;
import static com.facebook.presto.spi.statistics.ColumnStatisticType.NUMBER_OF_DISTINCT_VALUES;
import static com.facebook.presto.spi.statistics.ColumnStatisticType.TOTAL_SIZE_IN_BYTES;

public final class IcebergSessionProperties
{
Expand Down Expand Up @@ -156,14 +161,14 @@ public IcebergSessionProperties(
false),
new PropertyMetadata<>(
HIVE_METASTORE_STATISTICS_MERGE_STRATEGY,
"choose how to include statistics from the Hive Metastore when calculating table stats. Valid values are: "
+ Joiner.on(", ").join(HiveStatisticsMergeStrategy.values()),
"Flags to choose which statistics from the Hive Metastore are used when calculating table stats. Valid values are: "
+ Joiner.on(", ").join(NUMBER_OF_DISTINCT_VALUES, TOTAL_SIZE_IN_BYTES),
VARCHAR,
HiveStatisticsMergeStrategy.class,
icebergConfig.getHiveStatisticsMergeStrategy(),
EnumSet.class,
icebergConfig.getHiveStatisticsMergeFlags(),
false,
val -> HiveStatisticsMergeStrategy.valueOf((String) val),
HiveStatisticsMergeStrategy::name),
val -> decodeMergeFlags((String) val),
StatisticsUtil::encodeMergeFlags),
booleanProperty(
PUSHDOWN_FILTER_ENABLED,
"Experimental: Enable Filter Pushdown for Iceberg. This is only supported with Native Worker.",
Expand Down Expand Up @@ -272,9 +277,9 @@ public static boolean isMergeOnReadModeEnabled(ConnectorSession session)
return session.getProperty(MERGE_ON_READ_MODE_ENABLED, Boolean.class);
}

public static HiveStatisticsMergeStrategy getHiveStatisticsMergeStrategy(ConnectorSession session)
public static EnumSet<ColumnStatisticType> getHiveStatisticsMergeStrategy(ConnectorSession session)
{
return session.getProperty(HIVE_METASTORE_STATISTICS_MERGE_STRATEGY, HiveStatisticsMergeStrategy.class);
return session.getProperty(HIVE_METASTORE_STATISTICS_MERGE_STRATEGY, EnumSet.class);
}

public static boolean isPushdownFilterEnabled(ConnectorSession session)
Expand Down
Loading

0 comments on commit 437b48a

Please sign in to comment.