IcebergFilterPushdown disables COUNT(*)/MIN/MAX metadata optimization, causing full TableScan

Summary

When iceberg.pushdown-filter-enabled=true, Iceberg's file-statistics-based aggregate optimization (which answers COUNT(*), MIN, and MAX from Iceberg manifest metadata without reading data files) is explicitly disabled. Queries that would otherwise be answered in milliseconds fall back to a full TableScan over all data files.

Observed behavior

-- With pushdown-filter-enabled=false
SELECT COUNT(*) FROM large_iceberg_table;
-- Plan: Values(1234567)  ← answered from manifest file stats, very fast

-- With pushdown-filter-enabled=true  
SELECT COUNT(*) FROM large_iceberg_table;
-- Plan: TableScan → AggregationNode  ← workers read all data files (or at minimum all Parquet footers), slow

The query with pushdown-filter-enabled=true can be orders of magnitude slower for large tables with many files.

Root cause

The Iceberg connector has an optimizer (IcebergAggregationOptimizer) that converts COUNT(*), MIN(col), and MAX(col) into a Values node by reading per-file statistics from Iceberg's manifest files — without accessing any data files. This is correct and safe because Iceberg maintains precise row counts and column-level min/max bounds in its manifest metadata.

However, this optimizer contains an explicit guard:

public PlanNode optimize(PlanNode maxSubplan, ConnectorSession session, ...) {
    if (!IcebergSessionProperties.isAggregatePushDownEnabled(session)
            || IcebergSessionProperties.isPushdownFilterEnabled(session)) {  // ← skips when filter pushdown is on
        return maxSubplan;
    }
    // ... aggregate optimization
}

The || isPushdownFilterEnabled(session) condition completely disables the aggregate optimization whenever filter pushdown is enabled, even for queries with no WHERE clause where the optimization would be perfectly safe.

Why the guard exists (and why it's overly conservative)

IcebergAggregationOptimizer requires the TableScanNode to have an IcebergTableLayoutHandle already attached (it calls tableScan.getTable().getLayout().get()). This layout is set in two ways:

1. By the Presto core PickTableLayout rule (when filter pushdown is disabled)
2. By IcebergFilterPushdown (when filter pushdown is enabled)

In the logical plan optimizer registration order, IcebergFilterPushdown runs before IcebergAggregationOptimizer. So when filter pushdown is enabled, IcebergFilterPushdown has already populated the IcebergTableLayoutHandle by the time IcebergAggregationOptimizer runs. The layout is present and correctly typed — the guard is unnecessary.

Fix

Remove the || isPushdownFilterEnabled(session) condition:

public PlanNode optimize(PlanNode maxSubplan, ConnectorSession session, ...) {
    if (!IcebergSessionProperties.isAggregatePushDownEnabled(session)) {
        return maxSubplan;
    }
    // ... aggregate optimization
}

IcebergAggregationOptimizer already handles the case where the layout has predicates (it extracts them via IcebergUtil.getNonMetadataColumnConstraints and applies them as an Iceberg scan filter). It also correctly falls back to returning the unmodified AggregationNode when the optimization cannot be applied (e.g., tables with row-level deletes, columns without statistics, or when AggregateEvaluator reports incomplete results).

Impact

• SELECT COUNT(*) FROM iceberg_table with pushdown-filter-enabled=true goes from TableScan (seconds/minutes for large tables) to Values (milliseconds).
• Same benefit for MIN/MAX queries on columns with file-level statistics.
• No correctness impact: IcebergAggregationOptimizer already has conservative fallback paths.

Related

• PR fixing ClassNotFoundException: PrestoS3FileSystem (also triggered by planning-thread S3 access with pushdown-filter-enabled=true): #28479
• Note: IcebergAggregationOptimizer reads Iceberg manifest files during planning (via table.newScan().planFiles()), which also goes through HiveCachingHdfsConfiguration. The ClassLoader fix in #28479 is a prerequisite for this fix to work correctly for newly registered tables.