diff --git a/datafusion_iceberg/src/table/mod.rs b/datafusion_iceberg/src/table/mod.rs index 591448d0..e45f057d 100644 --- a/datafusion_iceberg/src/table/mod.rs +++ b/datafusion_iceberg/src/table/mod.rs @@ -810,9 +810,21 @@ async fn table_scan( .iter() .map(|index| scan_schema.index_of(arrow_schema.field(*index).name())) .collect::, _>>()?; + // Row-filter pushdown (predicate evaluated inside the parquet decoder, late materialization + // of the remaining columns, TopK / join dynamic filters reaching the scan) pays off only when + // the scan is wide and the predicate narrow: `SELECT * ... WHERE url LIKE ... ORDER BY t LIMIT n` + // decodes 100+ columns for the few surviving rows. On narrow scans the same machinery costs + // more than the vectorized FilterExec it replaces, so it stays off there. + let filter_columns: std::collections::HashSet<_> = filters + .iter() + .flat_map(|f| f.column_refs().into_iter().cloned()) + .collect(); + let pushdown_filters = + requested_projection.len() >= 8 && !filter_columns.is_empty() && filter_columns.len() <= 2; let file_source = Arc::new( ParquetSource::new(table_schema) - .with_parquet_file_reader_factory(parquet_reader_factory.clone()), + .with_parquet_file_reader_factory(parquet_reader_factory.clone()) + .with_pushdown_filters(pushdown_filters), ); // Create plan for every partition with delete files