[Iceberg] Use remainingPredicate in filter stats calc

This change combines the TableLayoutHandle's remaining predicate
and the pushed-down predicate when calculating statistics for
the call to the Iceberg connector's `getTableStatistics` in order to
attempt to get more accurate statistic estimates. The logic mimics
the Hive connector.

Explanation:

There were two issues with the previous implementation

1. Non-hive implementations never calculated statistics using the
FilterStatsCalculatorService.
2. The hive implementation did not consider the
`remainingPredicate` field when using the FilterStatsCalculatorService

This commit remedies both issues with some small refactoring and
changes to the arguments the connector passes to the
`FilterStatsCalculatorService`.

The FilterStatsCalculatorService attempts to refine the stats
calculations returned from a connector by considering the
full table statistics and any predicates applied to the scan in
order to reduce the total row count and other statistics like NDVs.

When filter pushdown is enabled on the iceberg connector, only
some parts of the filter can be pushed into the table scan. The rest
is represented by the layout handle's `remainingPredicate` field.

The Hive connector implementation creates a new binary
expression using an AND of the `remainingPredicate` and
`domainPredicate` fields. This change updates the implementation
to do the same.

Author: ZacBlanco

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

@@ -49,6 +49,7 @@
 import com.facebook.presto.spi.connector.ConnectorTableVersion.VersionOperator;
 import com.facebook.presto.spi.connector.ConnectorTableVersion.VersionType;
 import com.facebook.presto.spi.function.StandardFunctionResolution;
+import com.facebook.presto.spi.plan.FilterStatsCalculatorService;
 import com.facebook.presto.spi.relation.RowExpression;
 import com.facebook.presto.spi.relation.RowExpressionService;
 import com.facebook.presto.spi.statistics.ColumnStatisticMetadata;
@@ -154,7 +155,8 @@
 import static com.facebook.presto.iceberg.TypeConverter.toPrestoType;
 import static com.facebook.presto.iceberg.changelog.ChangelogUtil.getRowTypeFromColumnMeta;
 import static com.facebook.presto.iceberg.optimizer.IcebergPlanOptimizer.getEnforcedColumns;
-import static com.facebook.presto.iceberg.util.StatisticsUtil.combineSelectedAndPredicateColumns;
+import static com.facebook.presto.iceberg.util.StatisticsUtil.calculateBaseTableStatistics;
+import static com.facebook.presto.iceberg.util.StatisticsUtil.calculateStatisticsConsideringLayout;
 import static com.facebook.presto.spi.StandardErrorCode.NOT_SUPPORTED;
 import static com.facebook.presto.spi.statistics.TableStatisticType.ROW_COUNT;
 import static com.google.common.base.Verify.verify;
@@ -172,28 +174,32 @@
 public abstract class IcebergAbstractMetadata
         implements ConnectorMetadata
 {
+    private static final Logger log = Logger.get(IcebergAbstractMetadata.class);
+
     protected final TypeManager typeManager;
     protected final JsonCodec<CommitTaskData> commitTaskCodec;
     protected final NodeVersion nodeVersion;
     protected final RowExpressionService rowExpressionService;
+    protected final FilterStatsCalculatorService filterStatsCalculatorService;
     protected Transaction transaction;
 
     private final StandardFunctionResolution functionResolution;
     private final ConcurrentMap<SchemaTableName, Table> icebergTables = new ConcurrentHashMap<>();
-    private static final Logger log = Logger.get(IcebergAbstractMetadata.class);
 
     public IcebergAbstractMetadata(
             TypeManager typeManager,
             StandardFunctionResolution functionResolution,
             RowExpressionService rowExpressionService,
             JsonCodec<CommitTaskData> commitTaskCodec,
-            NodeVersion nodeVersion)
+            NodeVersion nodeVersion,
+            FilterStatsCalculatorService filterStatsCalculatorService)
     {
         this.typeManager = requireNonNull(typeManager, "typeManager is null");
         this.commitTaskCodec = requireNonNull(commitTaskCodec, "commitTaskCodec is null");
         this.functionResolution = requireNonNull(functionResolution, "functionResolution is null");
         this.rowExpressionService = requireNonNull(rowExpressionService, "rowExpressionService is null");
         this.nodeVersion = requireNonNull(nodeVersion, "nodeVersion is null");
+        this.filterStatsCalculatorService = requireNonNull(filterStatsCalculatorService, "filterStatsCalculatorService is null");
     }
 
     protected final Table getIcebergTable(ConnectorSession session, SchemaTableName schemaTableName)
@@ -727,19 +733,8 @@ public Map<String, ColumnHandle> getColumnHandles(ConnectorSession session, Conn
     @Override
     public TableStatistics getTableStatistics(ConnectorSession session, ConnectorTableHandle tableHandle, Optional<ConnectorTableLayoutHandle> tableLayoutHandle, List<ColumnHandle> columnHandles, Constraint<ColumnHandle> constraint)
     {
-        IcebergTableHandle handle = (IcebergTableHandle) tableHandle;
-        Table icebergTable = getIcebergTable(session, handle.getSchemaTableName());
-
-        List<IcebergColumnHandle> handles = combineSelectedAndPredicateColumns(
-                columnHandles.stream()
-                        .map(IcebergColumnHandle.class::cast)
-                        .collect(toImmutableList()),
-                tableLayoutHandle.map(IcebergTableLayoutHandle.class::cast));
-        return TableStatisticsMaker.getTableStatistics(session, typeManager,
-                tableLayoutHandle
-                        .map(IcebergTableLayoutHandle.class::cast)
-                        .map(IcebergTableLayoutHandle::getValidPredicate),
-                constraint, handle, icebergTable, handles);
+        TableStatistics baseStatistics = calculateBaseTableStatistics(this, typeManager, session, (IcebergTableHandle) tableHandle, tableLayoutHandle, columnHandles, constraint);
+        return calculateStatisticsConsideringLayout(filterStatsCalculatorService, rowExpressionService, baseStatistics, session, tableLayoutHandle);
     }
 
     @Override

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

@@ -14,7 +14,6 @@
 package com.facebook.presto.iceberg;
 
 import com.facebook.airlift.json.JsonCodec;
-import com.facebook.presto.common.predicate.TupleDomain;
 import com.facebook.presto.common.type.Type;
 import com.facebook.presto.common.type.TypeManager;
 import com.facebook.presto.hive.HdfsContext;
@@ -51,13 +50,10 @@
 import com.facebook.presto.spi.ViewNotFoundException;
 import com.facebook.presto.spi.function.StandardFunctionResolution;
 import com.facebook.presto.spi.plan.FilterStatsCalculatorService;
-import com.facebook.presto.spi.relation.RowExpression;
 import com.facebook.presto.spi.relation.RowExpressionService;
-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.TableStatisticType;
 import com.facebook.presto.spi.statistics.TableStatistics;
@@ -126,13 +122,13 @@
 import static com.facebook.presto.iceberg.PartitionFields.parsePartitionFields;
 import static com.facebook.presto.iceberg.PartitionSpecConverter.toPrestoPartitionSpec;
 import static com.facebook.presto.iceberg.SchemaConverter.toPrestoSchema;
-import static com.facebook.presto.iceberg.util.StatisticsUtil.calculateAndSetTableSize;
+import static com.facebook.presto.iceberg.util.StatisticsUtil.calculateBaseTableStatistics;
+import static com.facebook.presto.iceberg.util.StatisticsUtil.calculateStatisticsConsideringLayout;
 import static com.facebook.presto.iceberg.util.StatisticsUtil.mergeHiveStatistics;
 import static com.facebook.presto.spi.StandardErrorCode.INVALID_SCHEMA_PROPERTY;
 import static com.facebook.presto.spi.StandardErrorCode.NOT_SUPPORTED;
 import static com.facebook.presto.spi.StandardErrorCode.SCHEMA_NOT_EMPTY;
 import static com.facebook.presto.spi.security.PrincipalType.USER;
-import static com.facebook.presto.spi.statistics.SourceInfo.ConfidenceLevel.LOW;
 import static com.facebook.presto.spi.statistics.TableStatisticType.ROW_COUNT;
 import static com.google.common.base.Verify.verify;
 import static com.google.common.collect.ImmutableList.toImmutableList;
@@ -141,7 +137,6 @@
 import static java.util.Collections.emptyList;
 import static java.util.Locale.ENGLISH;
 import static java.util.Objects.requireNonNull;
-import static java.util.function.Function.identity;
 import static org.apache.iceberg.TableMetadata.newTableMetadata;
 import static org.apache.iceberg.TableProperties.OBJECT_STORE_PATH;
 import static org.apache.iceberg.TableProperties.WRITE_DATA_LOCATION;
@@ -154,7 +149,6 @@ public class IcebergHiveMetadata
     private final ExtendedHiveMetastore metastore;
     private final HdfsEnvironment hdfsEnvironment;
     private final DateTimeZone timeZone = DateTimeZone.forTimeZone(TimeZone.getTimeZone(ZoneId.of(TimeZone.getDefault().getID())));
-    private final FilterStatsCalculatorService filterStatsCalculatorService;
     private final IcebergHiveTableOperationsConfig hiveTableOeprationsConfig;
 
     public IcebergHiveMetadata(
@@ -168,10 +162,9 @@ public IcebergHiveMetadata(
             FilterStatsCalculatorService filterStatsCalculatorService,
             IcebergHiveTableOperationsConfig hiveTableOeprationsConfig)
     {
-        super(typeManager, functionResolution, rowExpressionService, commitTaskCodec, nodeVersion);
+        super(typeManager, functionResolution, rowExpressionService, commitTaskCodec, nodeVersion, filterStatsCalculatorService);
         this.metastore = requireNonNull(metastore, "metastore is null");
         this.hdfsEnvironment = requireNonNull(hdfsEnvironment, "hdfsEnvironment is null");
-        this.filterStatsCalculatorService = requireNonNull(filterStatsCalculatorService, "filterStatsCalculatorService is null");
         this.hiveTableOeprationsConfig = requireNonNull(hiveTableOeprationsConfig, "hiveTableOperationsConfig is null");
     }
 
@@ -455,48 +448,16 @@ public TableStatistics getTableStatistics(ConnectorSession session, ConnectorTab
     {
         IcebergTableHandle handle = (IcebergTableHandle) tableHandle;
         org.apache.iceberg.Table icebergTable = getIcebergTable(session, handle.getSchemaTableName());
-        TableStatistics icebergStatistics = super.getTableStatistics(session, tableHandle, tableLayoutHandle, columnHandles, constraint);
+        TableStatistics icebergStatistics = calculateBaseTableStatistics(this, typeManager, session, (IcebergTableHandle) tableHandle, tableLayoutHandle, columnHandles, constraint);
         EnumSet<ColumnStatisticType> mergeFlags = getHiveStatisticsMergeStrategy(session);
-        return tableLayoutHandle.map(IcebergTableLayoutHandle.class::cast).map(layoutHandle -> {
-            TupleDomain<VariableReferenceExpression> predicate = layoutHandle.getValidPredicate()
-                    .transform(columnHandle -> new VariableReferenceExpression(Optional.empty(), columnHandle.getName(), columnHandle.getType()));
-            RowExpression translatedPredicate = rowExpressionService.getDomainTranslator().toPredicate(predicate);
-            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());
-            Set<ColumnHandle> fullColumnSet = icebergStatistics.getColumnStatistics().keySet();
-            for (ColumnHandle colHandle : fullColumnSet) {
-                IcebergColumnHandle icebergHandle = (IcebergColumnHandle) colHandle;
-                if (mergedStatistics.getColumnStatistics().containsKey(icebergHandle)) {
-                    ColumnStatistics stats = mergedStatistics.getColumnStatistics().get(icebergHandle);
-                    filteredStatsBuilder.setColumnStatistics(icebergHandle, stats);
-                }
-            }
-            return filterStatsCalculatorService.filterStats(
-                    calculateAndSetTableSize(filteredStatsBuilder).setConfidenceLevel(LOW).build(),
-                    translatedPredicate,
-                    session,
-                    fullColumnSet.stream()
-                            .map(IcebergColumnHandle.class::cast).collect(toImmutableMap(
-                                    identity(),
-                                    IcebergColumnHandle::getName)),
-                    fullColumnSet
-                            .stream().map(IcebergColumnHandle.class::cast).collect(toImmutableMap(
-                                    IcebergColumnHandle::getName,
-                                    IcebergColumnHandle::getType)));
-        }).orElseGet(() -> {
-            if (!mergeFlags.isEmpty()) {
-                PartitionStatistics hiveStats = metastore.getTableStatistics(getMetastoreContext(session), handle.getSchemaName(), handle.getIcebergTableName().getTableName());
-                return mergeHiveStatistics(icebergStatistics, hiveStats, mergeFlags, icebergTable.spec());
-            }
-            return icebergStatistics;
-        });
+        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);
+        return calculateStatisticsConsideringLayout(filterStatsCalculatorService, rowExpressionService, mergedStatistics, session, tableLayoutHandle);
     }
 
     @Override

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

@@ -27,6 +27,7 @@
 import com.facebook.presto.spi.PrestoException;
 import com.facebook.presto.spi.SchemaTableName;
 import com.facebook.presto.spi.function.StandardFunctionResolution;
+import com.facebook.presto.spi.plan.FilterStatsCalculatorService;
 import com.facebook.presto.spi.relation.RowExpressionService;
 import org.apache.hadoop.fs.Path;
 import org.apache.iceberg.PartitionSpec;
@@ -80,9 +81,10 @@ public IcebergNativeMetadata(
             RowExpressionService rowExpressionService,
             JsonCodec<CommitTaskData> commitTaskCodec,
             CatalogType catalogType,
-            NodeVersion nodeVersion)
+            NodeVersion nodeVersion,
+            FilterStatsCalculatorService filterStatsCalculatorService)
     {
-        super(typeManager, functionResolution, rowExpressionService, commitTaskCodec, nodeVersion);
+        super(typeManager, functionResolution, rowExpressionService, commitTaskCodec, nodeVersion, filterStatsCalculatorService);
         this.catalogFactory = requireNonNull(catalogFactory, "catalogFactory is null");
         this.catalogType = requireNonNull(catalogType, "catalogType is null");
     }

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

@@ -18,6 +18,7 @@
 import com.facebook.presto.hive.NodeVersion;
 import com.facebook.presto.spi.connector.ConnectorMetadata;
 import com.facebook.presto.spi.function.StandardFunctionResolution;
+import com.facebook.presto.spi.plan.FilterStatsCalculatorService;
 import com.facebook.presto.spi.relation.RowExpressionService;
 
 import javax.inject.Inject;
@@ -34,6 +35,7 @@ public class IcebergNativeMetadataFactory
     final StandardFunctionResolution functionResolution;
     final RowExpressionService rowExpressionService;
     final NodeVersion nodeVersion;
+    final FilterStatsCalculatorService filterStatsCalculatorService;
 
     @Inject
     public IcebergNativeMetadataFactory(
@@ -43,7 +45,8 @@ public IcebergNativeMetadataFactory(
             StandardFunctionResolution functionResolution,
             RowExpressionService rowExpressionService,
             JsonCodec<CommitTaskData> commitTaskCodec,
-            NodeVersion nodeVersion)
+            NodeVersion nodeVersion,
+            FilterStatsCalculatorService filterStatsCalculatorService)
     {
         this.catalogFactory = requireNonNull(catalogFactory, "catalogFactory is null");
         this.typeManager = requireNonNull(typeManager, "typeManager is null");
@@ -53,10 +56,11 @@ public IcebergNativeMetadataFactory(
         this.nodeVersion = requireNonNull(nodeVersion, "nodeVersion is null");
         requireNonNull(config, "config is null");
         this.catalogType = config.getCatalogType();
+        this.filterStatsCalculatorService = requireNonNull(filterStatsCalculatorService, "filterStatsCalculatorService is null");
     }
 
     public ConnectorMetadata create()
     {
-        return new IcebergNativeMetadata(catalogFactory, typeManager, functionResolution, rowExpressionService, commitTaskCodec, catalogType, nodeVersion);
+        return new IcebergNativeMetadata(catalogFactory, typeManager, functionResolution, rowExpressionService, commitTaskCodec, catalogType, nodeVersion, filterStatsCalculatorService);
     }
 }