Avoid footer stats when not reliable

For partition that goes through mutation via commits, the footer stats
may not be reliable to use for optimization. As part of this commit, we
are providing an option to avoid using footer stats.
Avoid footer stats for metalake partition

For metalake partition, the footer stats are not reliable. As part of
this commit, we are avoiding to use footer stats if a particular split
has underwent mutation.

presto-hive/src/main/java/com/facebook/presto/hive/HivePageSourceProvider.java

@@ -328,7 +328,8 @@ private static Optional<ConnectorPageSource> createSelectivePageSource(
                             split.getFileSplit().getFileModifiedTime(),
                             HiveSessionProperties.isVerboseRuntimeStatsEnabled(session)),
                     encryptionInformation,
-                    layout.isAppendRowNumberEnabled());
+                    layout.isAppendRowNumberEnabled(),
+                    layout.isFooterStatsUnreliable());
             if (pageSource.isPresent()) {
                 return Optional.of(pageSource.get());
             }

presto-hive/src/main/java/com/facebook/presto/hive/HiveSelectivePageSourceFactory.java

@@ -43,5 +43,6 @@ Optional<? extends ConnectorPageSource> createPageSource(
             DateTimeZone hiveStorageTimeZone,
             HiveFileContext hiveFileContext,
             Optional<EncryptionInformation> encryptionInformation,
-            boolean appendRowNumberEnabled);
+            boolean appendRowNumberEnabled,
+            boolean footerStatsUnreliable);
 }

presto-hive/src/main/java/com/facebook/presto/hive/HiveTableLayoutHandle.java

@@ -65,6 +65,7 @@ public class HiveTableLayoutHandle
     private final Optional<Set<HiveColumnHandle>> requestedColumns;
     private final boolean partialAggregationsPushedDown;
     private final boolean appendRowNumberEnabled;
+    private final boolean footerStatsUnreliable;
 
     // coordinator-only properties
     private final Optional<List<HivePartition>> partitions;
@@ -86,7 +87,8 @@ public HiveTableLayoutHandle(
             @JsonProperty("layoutString") String layoutString,
             @JsonProperty("requestedColumns") Optional<Set<HiveColumnHandle>> requestedColumns,
             @JsonProperty("partialAggregationsPushedDown") boolean partialAggregationsPushedDown,
-            @JsonProperty("appendRowNumber") boolean appendRowNumberEnabled)
+            @JsonProperty("appendRowNumber") boolean appendRowNumberEnabled,
+            @JsonProperty("footerStatsUnreliable") boolean footerStatsUnreliable)
     {
         this(
                 schemaTableName,
@@ -105,7 +107,8 @@ public HiveTableLayoutHandle(
                 requestedColumns,
                 partialAggregationsPushedDown,
                 appendRowNumberEnabled,
-                Optional.empty());
+                Optional.empty(),
+                footerStatsUnreliable);
     }
 
     protected HiveTableLayoutHandle(
@@ -125,7 +128,8 @@ protected HiveTableLayoutHandle(
             Optional<Set<HiveColumnHandle>> requestedColumns,
             boolean partialAggregationsPushedDown,
             boolean appendRowNumberEnabled,
-            Optional<List<HivePartition>> partitions)
+            Optional<List<HivePartition>> partitions,
+            boolean footerStatsUnreliable)
     {
         this.schemaTableName = requireNonNull(schemaTableName, "table is null");
         this.tablePath = requireNonNull(tablePath, "tablePath is null");
@@ -144,6 +148,7 @@ protected HiveTableLayoutHandle(
         this.partialAggregationsPushedDown = partialAggregationsPushedDown;
         this.appendRowNumberEnabled = appendRowNumberEnabled;
         this.partitions = requireNonNull(partitions, "partitions is null");
+        this.footerStatsUnreliable = footerStatsUnreliable;
     }
 
     @JsonProperty
@@ -259,6 +264,12 @@ public boolean isAppendRowNumberEnabled()
         return appendRowNumberEnabled;
     }
 
+    @JsonProperty
+    public boolean isFooterStatsUnreliable()
+    {
+        return footerStatsUnreliable;
+    }
+
     @Override
     public Object getIdentifier(Optional<ConnectorSplit> split, PlanCanonicalizationStrategy canonicalizationStrategy)
     {
@@ -356,7 +367,8 @@ public Builder builder()
                 .setRequestedColumns(getRequestedColumns())
                 .setPartialAggregationsPushedDown(isPartialAggregationsPushedDown())
                 .setAppendRowNumberEnabled(isAppendRowNumberEnabled())
-                .setPartitions(getPartitions());
+                .setPartitions(getPartitions())
+                .setFooterStatsUnreliable(isFooterStatsUnreliable());
     }
 
     public static class Builder
@@ -377,6 +389,7 @@ public static class Builder
         private Optional<Set<HiveColumnHandle>> requestedColumns;
         private boolean partialAggregationsPushedDown;
         private boolean appendRowNumberEnabled;
+        private boolean footerStatsUnreliable;
 
         private Optional<List<HivePartition>> partitions;
 
@@ -489,6 +502,12 @@ public Builder setPartitions(Optional<List<HivePartition>> partitions)
             return this;
         }
 
+        public Builder setFooterStatsUnreliable(boolean footerStatsUnreliable)
+        {
+            this.footerStatsUnreliable = footerStatsUnreliable;
+            return this;
+        }
+
         public HiveTableLayoutHandle build()
         {
             return new HiveTableLayoutHandle(
@@ -508,7 +527,8 @@ public HiveTableLayoutHandle build()
                     requestedColumns,
                     partialAggregationsPushedDown,
                     appendRowNumberEnabled,
-                    partitions);
+                    partitions,
+                    footerStatsUnreliable);
         }
     }
 }

presto-hive/src/main/java/com/facebook/presto/hive/orc/DwrfSelectivePageSourceFactory.java

@@ -107,7 +107,8 @@ public Optional<? extends ConnectorPageSource> createPageSource(
             DateTimeZone hiveStorageTimeZone,
             HiveFileContext hiveFileContext,
             Optional<EncryptionInformation> encryptionInformation,
-            boolean appendRowNumberEnabled)
+            boolean appendRowNumberEnabled,
+            boolean footerStatsUnreliable)
     {
         if (!OrcSerde.class.getName().equals(storage.getStorageFormat().getSerDe())) {
             return Optional.empty();
@@ -144,6 +145,7 @@ public Optional<? extends ConnectorPageSource> createPageSource(
                 tupleDomainFilterCache,
                 encryptionInformation,
                 dwrfEncryptionProvider,
-                appendRowNumberEnabled));
+                appendRowNumberEnabled,
+                footerStatsUnreliable));
     }
 }

presto-hive/src/main/java/com/facebook/presto/hive/orc/OrcSelectivePageSourceFactory.java

@@ -213,7 +213,8 @@ public Optional<? extends ConnectorPageSource> createPageSource(
             DateTimeZone hiveStorageTimeZone,
             HiveFileContext hiveFileContext,
             Optional<EncryptionInformation> encryptionInformation,
-            boolean appendRowNumberEnabled)
+            boolean appendRowNumberEnabled,
+            boolean footerStatsUnreliable)
     {
         if (!OrcSerde.class.getName().equals(storage.getStorageFormat().getSerDe())) {
             return Optional.empty();
@@ -251,7 +252,8 @@ public Optional<? extends ConnectorPageSource> createPageSource(
                 tupleDomainFilterCache,
                 encryptionInformation,
                 NO_ENCRYPTION,
-                appendRowNumberEnabled));
+                appendRowNumberEnabled,
+                footerStatsUnreliable));
     }
 
     public static ConnectorPageSource createOrcPageSource(
@@ -281,7 +283,8 @@ public static ConnectorPageSource createOrcPageSource(
             TupleDomainFilterCache tupleDomainFilterCache,
             Optional<EncryptionInformation> encryptionInformation,
             DwrfEncryptionProvider dwrfEncryptionProvider,
-            boolean appendRowNumberEnabled)
+            boolean appendRowNumberEnabled,
+            boolean footerStatsUnreliable)
     {
         checkArgument(domainCompactionThreshold >= 1, "domainCompactionThreshold must be at least 1");
 
@@ -340,7 +343,7 @@ public static ConnectorPageSource createOrcPageSource(
 
             List<HiveColumnHandle> physicalColumns = getPhysicalHiveColumnHandles(columns, useOrcColumnNames, reader.getTypes(), path);
 
-            if (!physicalColumns.isEmpty() && physicalColumns.stream().allMatch(hiveColumnHandle -> hiveColumnHandle.getColumnType() == AGGREGATED)) {
+            if (!footerStatsUnreliable && !physicalColumns.isEmpty() && physicalColumns.stream().allMatch(hiveColumnHandle -> hiveColumnHandle.getColumnType() == AGGREGATED)) {
                 return new AggregatedOrcPageSource(physicalColumns, reader.getFooter(), typeManager, functionResolution);
             }
 

presto-hive/src/main/java/com/facebook/presto/hive/parquet/ParquetSelectivePageSourceFactory.java

@@ -70,7 +70,8 @@ public Optional<? extends ConnectorPageSource> createPageSource(
             DateTimeZone hiveStorageTimeZone,
             HiveFileContext hiveFileContext,
             Optional<EncryptionInformation> encryptionInformation,
-            boolean appendRowNumberEnabled)
+            boolean appendRowNumberEnabled,
+            boolean footerStatsUnreliable)
     {
         if (!PARQUET_SERDE_CLASS_NAMES.contains(storage.getStorageFormat().getSerDe())) {
             return Optional.empty();