feat: add aggregate pushdown support (COUNT/SUM/AVG/MIN/MAX) (#586)

* feat: add aggregate pushdown support (COUNT/SUM/AVG/MIN/MAX)

Based on PR #549 by JohnCari with critical fix: aggregates and group_by
are now stored in FdwState and passed through output_rel->fdw_private
to the executor. Previously fdw_private was null, so extracted aggregates
were silently discarded.

Changes:
- Add AggregateKind enum, Aggregate struct with deparse() methods
- Add FDW trait methods: supported_aggregates(), supports_group_by(),
  get_aggregate_rel_size(), begin_aggregate_scan()
- Add GetForeignUpperPaths callback in new upper.rs module
- Add aggregates/group_by fields to FdwState in scan.rs
- begin_foreign_scan detects aggregate path and calls begin_aggregate_scan
- EXPLAIN output includes aggregate info when present
- PG version compat: PG13-PG18 (disabled_nodes, fdw_restrictinfo)

* fix: address CodeRabbit review — validate aggregates and GROUP BY

1. Reject pushdown when SUM/AVG/MIN/MAX has no simple column reference
   (e.g. SUM(a+b)) to avoid generating invalid SQL like SUM(*)
2. Abort pushdown when any GROUP BY item is not a plain column reference
   to prevent incomplete GROUP BY clauses
3. Rebuild state.tgts for aggregate paths to reflect the actual output
   shape (group_by columns + aggregate result columns), preventing
   ERRCODE_FDW_INVALID_COLUMN_NUMBER from the executor

* fix: guard upper module import with PG feature cfg

The upper module is conditionally compiled with pg13-pg18 feature flags,
but fdw_routine() imported it unconditionally, causing build failures
without PG features. Wrap the import and GetForeignUpperPaths assignment
with the same cfg guard.

* fix: address CodeRabbit round 2 — aggfilter, aggtype, default warning

1. Reject pushdown for aggregates with FILTER clause (aggfilter check)
2. Add type_oid field to Aggregate struct, populated from Aggref::aggtype
   instead of inferring from the input column type
3. Default begin_aggregate_scan() now emits a warning if called without
   an override, to surface missing implementations early

* fix: fail fast in default begin_aggregate_scan instead of warning

Use report_error to abort the transaction when begin_aggregate_scan is
called without being overridden. This prevents wrong-results failures
when an FDW declares aggregate support but forgets to implement the
scan method.

* feat: implement aggregate pushdown support for ClickHouse FDW

* feat: add aggregate pushdown support for BigQuery and SQL Server FDWs

* fix: update explain statements to use EXPLAIN ANALYZE for pushdown tests

---------

Co-authored-by: Bo Lu <lv.patrick@gmail.com>

Author: wenerme

docs/catalog/bigquery.md

@@ -144,14 +144,46 @@ create foreign table bigquery.my_bigquery_table (
 
 #### Notes
 
-- Supports `where`, `order by` and `limit` clause pushdown
+- Supports `where`, `order by`, `limit` and aggregate clause pushdown
 - When using `rowid_column`, it must be specified for data modification operations
 - Data in the streaming buffer cannot be updated or deleted until the buffer is flushed (up to 90 minutes)
 
 ## Query Pushdown Support
 
 This FDW supports `where`, `order by` and `limit` clause pushdown.
 
+### Aggregate Pushdown
+
+The FDW pushes common aggregate queries down to BigQuery so the aggregation
+runs remotely and only the final result rows are transferred to Postgres. This
+is much faster than fetching every row and aggregating locally, especially
+over large tables — and on BigQuery it also reduces the bytes scanned billed
+to your project.
+
+**Supported aggregates** — `count(*)`, `count(col)`, `count(distinct col)`,
+`sum(col)`, `avg(col)`, `min(col)`, `max(col)`.
+
+**Supported shapes** — scalar aggregates, `group by` over plain columns, with
+or without a `where` clause. Pushdown also works when the foreign `table`
+option is a sub-query.
+
+```sql
+-- All of these run as a single aggregate query on BigQuery:
+select count(*) from bigquery.my_table;
+select id, sum(amount) from bigquery.my_table group by id;
+select count(distinct name) from bigquery.my_table where id = 1;
+```
+
+**Cases that are not pushed down** — the query still returns the correct
+result, but the aggregation happens in Postgres after fetching the rows:
+
+- The query has a `having` clause
+- The aggregate has a `filter (where …)` clause
+- A `distinct` modifier is used on anything other than `count`
+- The aggregate's argument is not a plain column (for example `sum(a + 1)`)
+- A `group by` item is not a plain column (for example `group by id + 1`)
+- The aggregate function is not in the list above (for example `stddev`, `string_agg`)
+
 ## Inserting Rows & the Streaming Buffer
 
 This foreign data wrapper uses BigQuery’s `insertAll` API method to create a `streamingBuffer` with an associated partition time. **Within that partition time, the data cannot be updated, deleted, or fully exported**. Only after the time has elapsed (up to 90 minutes according to [BigQuery’s documentation](https://cloud.google.com/bigquery/docs/streaming-data-into-bigquery)), can you perform operations.
@@ -256,3 +288,67 @@ where id = 1;
 delete from bigquery.people
 where id = 2;
 ```
+
+### Aggregate Query Examples
+
+These examples assume an `orders` table on BigQuery and a matching foreign
+table on Postgres:
+
+```sql
+-- Run on BigQuery
+create table your_project_id.your_dataset_id.orders (
+  id       int64,
+  user_id  int64,
+  amount   numeric,
+  status   string
+);
+
+insert into your_project_id.your_dataset_id.orders values
+  (1, 1, 100.0, 'paid'),
+  (2, 1,  50.0, 'paid'),
+  (3, 2, 200.0, 'pending'),
+  (4, 2,  75.0, 'paid'),
+  (5, 3, 300.0, 'paid');
+```
+
+```sql
+-- Foreign table on Postgres
+create foreign table bigquery.orders (
+  id      bigint,
+  user_id bigint,
+  amount  numeric,
+  status  text
+)
+  server bigquery_server
+  options (
+    table 'orders'
+  );
+```
+
+Each query below runs a single aggregate query against BigQuery and returns
+just the result rows:
+
+```sql
+-- Total order count
+select count(*) from bigquery.orders;
+
+-- Total revenue from paid orders
+select sum(amount) from bigquery.orders where status = 'paid';
+
+-- Per-user order count and revenue
+select user_id, count(*) as orders, sum(amount) as revenue
+from bigquery.orders
+group by user_id
+order by user_id;
+
+-- Smallest and largest order
+select min(amount), max(amount) from bigquery.orders;
+
+-- Number of distinct users who placed an order
+select count(distinct user_id) from bigquery.orders;
+
+-- Average order value per status
+select status, avg(amount) as avg_amount
+from bigquery.orders
+group by status;
+```

docs/catalog/clickhouse.md

@@ -153,14 +153,45 @@ create foreign table clickhouse.my_table (
 
 #### Notes
 
-- Supports `where`, `order by` and `limit` clause pushdown
+- Supports `where`, `order by`, `limit` and aggregate clause pushdown
 - Supports parametrized views in subqueries
 - When using `rowid_column`, it must be specified for data modification operations
 
 ## Query Pushdown Support
 
 This FDW supports `where`, `order by` and `limit` clause pushdown, as well as parametrized view (see above).
 
+### Aggregate Pushdown
+
+The FDW pushes common aggregate queries down to ClickHouse so the aggregation
+runs remotely and only the final result rows are transferred to Postgres. This
+is much faster than fetching every row and aggregating locally, especially
+over large tables.
+
+**Supported aggregates** — `count(*)`, `count(col)`, `count(distinct col)`,
+`sum(col)`, `avg(col)`, `min(col)`, `max(col)`.
+
+**Supported shapes** — scalar aggregates, `group by` over plain columns, with
+or without a `where` clause. Pushdown also works when the foreign `table`
+option is a sub-query or a parametrized view.
+
+```sql
+-- All of these run as a single aggregate query on ClickHouse:
+select count(*) from clickhouse.my_table;
+select id, sum(amount) from clickhouse.my_table group by id;
+select count(distinct name) from clickhouse.my_table where id = 1;
+```
+
+**Cases that are not pushed down** — the query still returns the correct
+result, but the aggregation happens in Postgres after fetching the rows:
+
+- The query has a `having` clause
+- The aggregate has a `filter (where …)` clause
+- A `distinct` modifier is used on anything other than `count`
+- The aggregate's argument is not a plain column (for example `sum(a + 1)`)
+- A `group by` item is not a plain column (for example `group by id + 1`)
+- The aggregate function is not in the list above (for example `stddev`, `string_agg`)
+
 ## Supported Data Types
 
 | Postgres Type      | ClickHouse Type   |
@@ -244,3 +275,69 @@ insert into clickhouse.people values (4, 'Yoda');
 update clickhouse.people set name = 'Princess Leia' where id = 2;
 delete from clickhouse.people where id = 3;
 ```
+
+### Aggregate Query Examples
+
+These examples assume the `clickhouse.people` foreign table from the previous
+section, plus an `orders` table with a row per order:
+
+```sql
+-- Run on ClickHouse
+create table orders (
+  id        Int64,
+  person_id Int64,
+  amount    Float64,
+  status    String
+)
+engine=MergeTree()
+order by id;
+
+insert into orders values
+  (1, 1, 100.0, 'paid'),
+  (2, 1,  50.0, 'paid'),
+  (3, 2, 200.0, 'pending'),
+  (4, 2,  75.0, 'paid'),
+  (5, 3, 300.0, 'paid');
+```
+
+```sql
+-- Foreign table on Postgres
+create foreign table clickhouse.orders (
+  id        bigint,
+  person_id bigint,
+  amount    double precision,
+  status    text
+)
+  server clickhouse_server
+  options (
+    table 'orders'
+  );
+```
+
+Each query below runs a single aggregate query against ClickHouse and returns
+just the result rows:
+
+```sql
+-- Total order count
+select count(*) from clickhouse.orders;
+
+-- Total revenue from paid orders
+select sum(amount) from clickhouse.orders where status = 'paid';
+
+-- Per-person order count and revenue
+select person_id, count(*) as orders, sum(amount) as revenue
+from clickhouse.orders
+group by person_id
+order by person_id;
+
+-- Smallest and largest order
+select min(amount), max(amount) from clickhouse.orders;
+
+-- Number of distinct customers who placed an order
+select count(distinct person_id) from clickhouse.orders;
+
+-- Average order value per status
+select status, avg(amount) as avg_amount
+from clickhouse.orders
+group by status;
+```

docs/catalog/mssql.md

@@ -145,12 +145,50 @@ create foreign table mssql.users (
       - `where` clauses
       - `order by` clauses
       - `limit` clauses
+      - aggregate clauses
 - See Data Types section for type mappings between PostgreSQL and SQL Server
 
 ## Query Pushdown Support
 
 This FDW supports `where`, `order by` and `limit` clause pushdown.
 
+### Aggregate Pushdown
+
+The FDW pushes common aggregate queries down to SQL Server so the aggregation
+runs remotely and only the final result rows are transferred to Postgres. This
+is much faster than fetching every row and aggregating locally, especially
+over large tables.
+
+**Supported aggregates** — `count(*)`, `count(col)`, `count(distinct col)`,
+`sum(col)`, `avg(col)`, `min(col)`, `max(col)`.
+
+**Supported shapes** — scalar aggregates, `group by` over plain columns, with
+or without a `where` clause. Pushdown also works when the foreign `table`
+option is a sub-query.
+
+```sql
+-- All of these run as a single aggregate query on SQL Server:
+select count(*) from mssql.users;
+select id, sum(amount) from mssql.users group by id;
+select count(distinct name) from mssql.users where id = 42;
+```
+
+`count(*)` and `count(col)` are translated to SQL Server's `count_big` so the
+result fits Postgres' `bigint` without overflow. Each aggregate is also wrapped
+in a `cast(... as <sql_server_type>)` so values come back in the exact type
+Postgres expects (for example, `sum` over a `bigint` column is cast to
+`numeric`, matching Postgres' `sum(bigint) → numeric` rule).
+
+**Cases that are not pushed down** — the query still returns the correct
+result, but the aggregation happens in Postgres after fetching the rows:
+
+- The query has a `having` clause
+- The aggregate has a `filter (where …)` clause
+- A `distinct` modifier is used on anything other than `count`
+- The aggregate's argument is not a plain column (for example `sum(a + 1)`)
+- A `group by` item is not a plain column (for example `group by id + 1`)
+- The aggregate function is not in the list above (for example `stddev`, `string_agg`)
+
 ## Supported Data Types
 
 | Postgres Type    | SQL Server Type                  |
@@ -231,3 +269,67 @@ create foreign table mssql.users_subquery (
 
 select * from mssql.users_subquery;
 ```
+
+### Aggregate Query Examples
+
+These examples assume an `orders` table on SQL Server and a matching foreign
+table on Postgres:
+
+```sql
+-- Run on SQL Server
+create table orders (
+  id        bigint,
+  user_id   bigint,
+  amount    numeric(18,2),
+  status    varchar(20)
+);
+
+insert into orders (id, user_id, amount, status) values
+  (1, 42, 100.00, 'paid'),
+  (2, 42,  50.00, 'paid'),
+  (3, 43, 200.00, 'pending'),
+  (4, 43,  75.00, 'paid'),
+  (5, 44, 300.00, 'paid');
+```
+
+```sql
+-- Foreign table on Postgres
+create foreign table mssql.orders (
+  id      bigint,
+  user_id bigint,
+  amount  numeric(18,2),
+  status  text
+)
+  server mssql_server
+  options (
+    table 'orders'
+  );
+```
+
+Each query below runs a single aggregate query against SQL Server and returns
+just the result rows:
+
+```sql
+-- Total order count
+select count(*) from mssql.orders;
+
+-- Total revenue from paid orders
+select sum(amount) from mssql.orders where status = 'paid';
+
+-- Per-user order count and revenue
+select user_id, count(*) as orders, sum(amount) as revenue
+from mssql.orders
+group by user_id
+order by user_id;
+
+-- Smallest and largest order
+select min(amount), max(amount) from mssql.orders;
+
+-- Number of distinct users who placed an order
+select count(distinct user_id) from mssql.orders;
+
+-- Average order value per status
+select status, avg(amount) as avg_amount
+from mssql.orders
+group by status;
+```

docs/contributing/native.md

@@ -25,6 +25,12 @@ pub trait ForeignDataWrapper {
     fn delete(...);
     fn end_modify(...);
 
+    // functions for aggregate pushdown (optional)
+    fn supported_aggregates(...) -> Vec<AggregateKind>;
+    fn supports_group_by(...) -> bool;
+    fn get_aggregate_rel_size(...) -> (i64, i32);
+    fn begin_aggregate_scan(...);
+
     // other optional functions
     ...
 }