Support partition based metadata query optimization for Iceberg

presto-docs/src/main/sphinx/connector/iceberg.rst

@@ -240,6 +240,13 @@ Property Name                                           Description
 
 ``iceberg.pushdown-filter-enabled``                     Experimental: Enable filter pushdown for Iceberg. This is     ``false``
                                                         only supported with Native Worker.
+
+``iceberg.rows-for-metadata-optimization-threshold``    The maximum number of partitions in an Iceberg table to       ``1000``
+                                                        allow optimizing queries of that table using metadata. If
+                                                        an Iceberg table has more partitions than this threshold,
+                                                        metadata optimization is skipped.
+
+                                                        Set to ``0`` to disable metadata optimization.
 ======================================================= ============================================================= ============
 
 Table Properties
@@ -304,14 +311,17 @@ Session Properties
 
 Session properties set behavior changes for queries executed within the given session.
 
-============================================= ======================================================================
-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.
-============================================= ======================================================================
+===================================================== ======================================================================
+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.
+``iceberg.rows_for_metadata_optimization_threshold``  Overrides the behavior of the connector property
+                                                      ``iceberg.rows-for-metadata-optimization-threshold`` in the current
+                                                      session.
+===================================================== ======================================================================
 
 Caching Support
 ----------------

presto-hive-common/src/main/java/com/facebook/presto/hive/MetadataUtils.java

@@ -39,6 +39,7 @@
 import static com.facebook.presto.expressions.LogicalRowExpressions.TRUE_CONSTANT;
 import static com.facebook.presto.expressions.LogicalRowExpressions.binaryExpression;
 import static com.facebook.presto.hive.HiveErrorCode.HIVE_UNKNOWN_ERROR;
+import static com.facebook.presto.hive.HivePartition.UNPARTITIONED_ID;
 import static com.facebook.presto.hive.rule.FilterPushdownUtils.isEntireColumn;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.collect.ImmutableList.toImmutableList;
@@ -53,7 +54,7 @@ private MetadataUtils() {}
     public static Optional<DiscretePredicates> getDiscretePredicates(List<ColumnHandle> partitionColumns, List<HivePartition> partitions)
     {
         Optional<DiscretePredicates> discretePredicates = Optional.empty();
-        if (!partitionColumns.isEmpty()) {
+        if (!partitionColumns.isEmpty() && !(partitions.size() == 1 && partitions.get(0).getPartitionId().equals(UNPARTITIONED_ID))) {
             // Do not create tuple domains for every partition at the same time!
             // There can be a huge number of partitions so use an iterable so
             // all domains do not need to be in memory at the same time.
@@ -107,6 +108,9 @@ public static TupleDomain<ColumnHandle> createPredicate(List<ColumnHandle> parti
         if (partitions.isEmpty()) {
             return TupleDomain.none();
         }
+        if (partitions.size() == 1 && partitions.get(0).getPartitionId().equals(UNPARTITIONED_ID)) {
+            return TupleDomain.all();
+        }
 
         return withColumnDomains(
                 partitionColumns.stream()

presto-iceberg/pom.xml

@@ -500,6 +500,12 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+          <groupId>com.facebook.presto</groupId>
+          <artifactId>presto-parser</artifactId>
+          <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>com.facebook.presto</groupId>
             <artifactId>presto-analyzer</artifactId>

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergAbstractMetadata.java

@@ -129,6 +129,7 @@
 import static com.facebook.presto.iceberg.IcebergUtil.getDeleteMode;
 import static com.facebook.presto.iceberg.IcebergUtil.getFileFormat;
 import static com.facebook.presto.iceberg.IcebergUtil.getPartitionKeyColumnHandles;
+import static com.facebook.presto.iceberg.IcebergUtil.getPartitions;
 import static com.facebook.presto.iceberg.IcebergUtil.getSnapshotIdAsOfTime;
 import static com.facebook.presto.iceberg.IcebergUtil.getTableComment;
 import static com.facebook.presto.iceberg.IcebergUtil.resolveSnapshotIdByName;
@@ -215,10 +216,24 @@ public ConnectorTableLayoutResult getTableLayoutForConstraint(
         IcebergTableHandle handle = (IcebergTableHandle) table;
         Table icebergTable = getIcebergTable(session, handle.getSchemaTableName());
 
-        List<IcebergColumnHandle> partitionColumns = getPartitionKeyColumnHandles(icebergTable, typeManager);
+        List<IcebergColumnHandle> partitionColumns = getPartitionKeyColumnHandles(handle, icebergTable, typeManager);
         TupleDomain<ColumnHandle> partitionColumnPredicate = TupleDomain.withColumnDomains(Maps.filterKeys(constraint.getSummary().getDomains().get(), Predicates.in(partitionColumns)));
         Optional<Set<IcebergColumnHandle>> requestedColumns = desiredColumns.map(columns -> columns.stream().map(column -> (IcebergColumnHandle) column).collect(toImmutableSet()));
 
+        List<HivePartition> partitions;
+        if (handle.getIcebergTableName().getTableType() == CHANGELOG ||
+                handle.getIcebergTableName().getTableType() == EQUALITY_DELETES) {
+            partitions = ImmutableList.of(new HivePartition(handle.getSchemaTableName()));
+        }
+        else {
+            partitions = getPartitions(
+                    typeManager,
+                    handle,
+                    icebergTable,
+                    constraint,
+                    partitionColumns);
+        }
+
         ConnectorTableLayout layout = getTableLayout(
                 session,
                 new IcebergTableLayoutHandle.Builder()
@@ -230,7 +245,7 @@ public ConnectorTableLayoutResult getTableLayoutForConstraint(
                         .setRequestedColumns(requestedColumns)
                         .setPushdownFilterEnabled(isPushdownFilterEnabled(session))
                         .setPartitionColumnPredicate(partitionColumnPredicate)
-                        .setPartitions(Optional.empty())
+                        .setPartitions(Optional.ofNullable(partitions.size() == 0 ? null : partitions))
                         .setTable(handle)
                         .build());
         return new ConnectorTableLayoutResult(layout, constraint.getSummary());
@@ -259,19 +274,19 @@ public ConnectorTableLayout getTableLayout(ConnectorSession session, ConnectorTa
         Table icebergTable = getIcebergTable(session, tableHandle.getSchemaTableName());
         validateTableMode(session, icebergTable);
         List<ColumnHandle> partitionColumns = ImmutableList.copyOf(icebergTableLayoutHandle.getPartitionColumns());
+        Optional<List<HivePartition>> partitions = icebergTableLayoutHandle.getPartitions();
+        Optional<DiscretePredicates> discretePredicates = partitions.flatMap(parts -> getDiscretePredicates(partitionColumns, parts));
         if (!isPushdownFilterEnabled(session)) {
             return new ConnectorTableLayout(
                     icebergTableLayoutHandle,
                     Optional.empty(),
-                    icebergTableLayoutHandle.getPartitionColumnPredicate(),
-                    Optional.empty(),
+                    partitions.isPresent() ? icebergTableLayoutHandle.getPartitionColumnPredicate() : TupleDomain.none(),
                     Optional.empty(),
                     Optional.empty(),
+                    discretePredicates,
                     ImmutableList.of(),
                     Optional.empty());
         }
-        Optional<List<HivePartition>> partitions = icebergTableLayoutHandle.getPartitions();
-        Optional<DiscretePredicates> discretePredicates = partitions.flatMap(parts -> getDiscretePredicates(partitionColumns, parts));
 
         Map<String, ColumnHandle> predicateColumns = icebergTableLayoutHandle.getPredicateColumns().entrySet()
                 .stream().collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergConfig.java

@@ -52,6 +52,7 @@ public class IcebergConfig
     private double statisticSnapshotRecordDifferenceWeight;
     private boolean pushdownFilterEnabled;
     private boolean deleteAsJoinRewriteEnabled = true;
+    private int rowsForMetadataOptimizationThreshold = 1000;
 
     private EnumSet<ColumnStatisticType> hiveStatisticsMergeFlags = EnumSet.noneOf(ColumnStatisticType.class);
     private String fileIOImpl = HadoopFileIO.class.getName();
@@ -251,6 +252,20 @@ public boolean isDeleteAsJoinRewriteEnabled()
         return deleteAsJoinRewriteEnabled;
     }
 
+    @Config("iceberg.rows-for-metadata-optimization-threshold")
+    @ConfigDescription("The max partitions number to utilize metadata optimization. 0 means skip the metadata optimization directly.")
+    public IcebergConfig setRowsForMetadataOptimizationThreshold(int rowsForMetadataOptimizationThreshold)
+    {
+        this.rowsForMetadataOptimizationThreshold = rowsForMetadataOptimizationThreshold;
+        return this;
+    }
+
+    @Min(0)
+    public int getRowsForMetadataOptimizationThreshold()
+    {
+        return rowsForMetadataOptimizationThreshold;
+    }
+
     public boolean getManifestCachingEnabled()
     {
         return manifestCachingEnabled;

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergSessionProperties.java

@@ -62,6 +62,7 @@ public final class IcebergSessionProperties
     public static final String DELETE_AS_JOIN_REWRITE_ENABLED = "delete_as_join_rewrite_enabled";
     public static final String HIVE_METASTORE_STATISTICS_MERGE_STRATEGY = "hive_statistics_merge_strategy";
     public static final String STATISTIC_SNAPSHOT_RECORD_DIFFERENCE_WEIGHT = "statistic_snapshot_record_difference_weight";
+    public static final String ROWS_FOR_METADATA_OPTIMIZATION_THRESHOLD = "rows_for_metadata_optimization_threshold";
 
     private final List<PropertyMetadata<?>> sessionProperties;
 
@@ -185,6 +186,13 @@ public IcebergSessionProperties(
                         DELETE_AS_JOIN_REWRITE_ENABLED,
                         "When enabled equality delete row filtering will be pushed down into a join.",
                         icebergConfig.isDeleteAsJoinRewriteEnabled(),
+                        false),
+                integerProperty(
+                        ROWS_FOR_METADATA_OPTIMIZATION_THRESHOLD,
+                        "The max partitions number to utilize metadata optimization. When partitions number " +
+                                "of an Iceberg table exceeds this threshold, metadata optimization would be skipped for " +
+                                "the table. A value of 0 means skip metadata optimization directly.",
+                        icebergConfig.getRowsForMetadataOptimizationThreshold(),
                         false));
     }
 
@@ -295,4 +303,9 @@ public static boolean isDeleteToJoinPushdownEnabled(ConnectorSession session)
     {
         return session.getProperty(DELETE_AS_JOIN_REWRITE_ENABLED, Boolean.class);
     }
+
+    public static int getRowsForMetadataOptimizationThreshold(ConnectorSession session)
+    {
+        return session.getProperty(ROWS_FOR_METADATA_OPTIMIZATION_THRESHOLD, Integer.class);
+    }
 }

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergUpdateablePageSource.java

@@ -14,11 +14,8 @@
 package com.facebook.presto.iceberg;
 
 import com.facebook.presto.common.Page;
-import com.facebook.presto.common.Utils;
 import com.facebook.presto.common.block.Block;
 import com.facebook.presto.common.block.RunLengthEncodedBlock;
-import com.facebook.presto.common.type.TimeType;
-import com.facebook.presto.common.type.TimestampType;
 import com.facebook.presto.common.type.Type;
 import com.facebook.presto.hive.HivePartitionKey;
 import com.facebook.presto.iceberg.delete.IcebergDeletePageSink;
@@ -38,13 +35,12 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Supplier;
 
+import static com.facebook.presto.common.Utils.nativeValueToBlock;
 import static com.facebook.presto.hive.BaseHiveColumnHandle.ColumnType.PARTITION_KEY;
 import static com.facebook.presto.iceberg.IcebergErrorCode.ICEBERG_BAD_DATA;
 import static com.facebook.presto.iceberg.IcebergUtil.deserializePartitionValue;
 import static com.google.common.base.Throwables.throwIfInstanceOf;
 import static java.util.Objects.requireNonNull;
-import static java.util.concurrent.TimeUnit.MICROSECONDS;
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
 public class IcebergUpdateablePageSource
         implements UpdatablePageSource
@@ -226,12 +222,4 @@ protected void closeWithSuppression(Throwable throwable)
             }
         }
     }
-
-    private Block nativeValueToBlock(Type type, Object prefilledValue)
-    {
-        if (prefilledValue != null && (type instanceof TimestampType && ((TimestampType) type).getPrecision() == MILLISECONDS || type instanceof TimeType)) {
-            return Utils.nativeValueToBlock(type, MICROSECONDS.toMillis((long) prefilledValue));
-        }
-        return Utils.nativeValueToBlock(type, prefilledValue);
-    }
 }