feat: add aggregate pushdown support via GetForeignUpperPaths

[JohnCari]

Summary

This PR adds support for pushing aggregate operations (COUNT, SUM, AVG, MIN, MAX) to foreign data sources via the PostgreSQL GetForeignUpperPaths callback. This enables FDWs to execute aggregate queries remotely, significantly reducing data transfer for analytics queries.

Key Changes

• New types in interface.rs:

  • AggregateKind enum: Represents supported aggregate functions (Count, CountColumn, Sum, Avg, Min, Max)
  • Aggregate struct: Contains aggregate operation details including kind, column, distinct flag, and alias
  • Helper methods: sql_name(), deparse(), deparse_with_alias() for SQL generation

• New ForeignDataWrapper trait methods:

  • supported_aggregates(): Declare which aggregates the FDW can push down (default: empty = no pushdown)
  • supports_group_by(): Declare GROUP BY pushdown support (default: false)
  • get_aggregate_rel_size(): Cost estimation for aggregate queries
  • begin_aggregate_scan(): Initialize aggregate query execution

• New module upper.rs:

  • Implements GetForeignUpperPaths callback for aggregate pushdown
  • Extracts aggregate information from GroupPathExtraData
  • Extracts GROUP BY columns from query
  • Creates and registers foreign upper paths for aggregate queries

Backward Compatibility

All new trait methods have default implementations, ensuring existing FDWs continue to work without modification. FDWs opt-in to aggregate pushdown by overriding supported_aggregates() to return a non-empty vector.

Example Usage

impl ForeignDataWrapper<MyError> for MyFdw {
    fn supported_aggregates(&self) -> Vec<AggregateKind> {
        vec![
            AggregateKind::Count,
            AggregateKind::Sum,
            AggregateKind::Avg,
            AggregateKind::Min,
            AggregateKind::Max,
        ]
    }

    fn supports_group_by(&self) -> bool {
        true
    }

    fn begin_aggregate_scan(
        &mut self,
        aggregates: &[Aggregate],
        group_by: &[Column],
        quals: &[Qual],
        options: &HashMap<String, String>,
    ) -> Result<(), MyError> {
        // Build and execute remote aggregate query
        let select_items: Vec<String> = group_by.iter()
            .map(|c| c.name.clone())
            .chain(aggregates.iter().map(|a| a.deparse_with_alias()))
            .collect();
        // Execute query...
        Ok(())
    }
}

Related Issue

Closes #22

Test Plan

• Verify existing FDWs without aggregate support continue to work unchanged
• Test COUNT(*) pushdown with a simple FDW implementation
• Test GROUP BY pushdown with multiple columns
• Test multiple aggregates in single query (e.g., SELECT MIN(x), MAX(x), COUNT(*) FROM t)
• Verify HAVING clause correctly prevents pushdown (not supported)
• Verify unsupported aggregates correctly fall back to local execution
• Test EXPLAIN output shows aggregate pushdown when applicable

[burmecia]

Thanks for the PR! Could you format the code so CI can pass?

[JohnCari]

Yes ofcourse, I will do it at night when I get home and thank you! @burmecia

[JohnCari]

@burmecia working on this now

[JohnCari]

@burmecia I finished formatting the code, it was harder and took me longer than i expected but finally finished, now all seven checks pass, let me know if you need anything else for this PR.