Add enable_aggregate_pushdown option to control aggregate push-down.

Considering the complexity of generating the aggregation pipeline, add an option to control the push-down to avoid any issues. This option can be set at the server level or the table level. If any of the tables involved in the join has this option set to false, then the join will not be pushed down. The table-level value of the option takes precedence over the server-level option value. Default is true. FDW-137, Vaibhav Dalvi, reviewed by Suraj Kharage and Jeevan Chalke, tested by Kashif Zeeshan.
EltonARodrigues · May 18, 2022 · 90ea7dc · 90ea7dc
1 parent fd37f1b
commit 90ea7dc
Show file tree

Hide file tree

Showing 8 changed files with 538 additions and 6 deletions.
diff --git a/README.md b/README.md
@@ -274,6 +274,11 @@ The following options are only supported with meta driver:
 	join has set it to false then the join will not be pushed down. The
 	table-level value of the option takes precedence over the server-level
 	option value. Default is `true`.
+  * `enable_aggregate_pushdown`: If `true`, push aggregates to the remote
+	MongoDB server instead of fetching all of the rows and aggregating them
+	locally. This option can also be set for an individual table. The
+	table-level value of the option takes precedence over the server-level
+	option value. Default is `true`.
 
 The following parameters can be set on a MongoDB foreign table object:
 
@@ -283,6 +288,8 @@ The following parameters can be set on a MongoDB foreign table object:
     the foreign table name used in the relevant `CREATE` command.
   * `enable_join_pushdown`: Similar to the server-level option, but can be
     configured at table level as well. Default is `true`.
+  * `enable_aggregate_pushdown`: Similar to the server-level option, but
+    can be configured at table level as well. Default is `true`.
 
 The following parameters can be supplied while creating user mapping:
 

diff --git a/expected/aggregate_pushdown.out b/expected/aggregate_pushdown.out
@@ -1133,6 +1133,155 @@ SELECT c1, count(t1) FROM fprt1 t1 GROUP BY c1 HAVING avg(c2) < 22 ORDER BY 1;
 (8 rows)
 
 SET enable_partitionwise_aggregate TO OFF;
+-- Support enable_aggregate_pushdown option at server level and table level.
+-- Check only boolean values are accepted.
+ALTER SERVER mongo_server OPTIONS (ADD enable_aggregate_pushdown 'non-bolean');
+ERROR:  enable_aggregate_pushdown requires a Boolean value
+-- Test the option at server level.
+ALTER SERVER mongo_server OPTIONS (ADD enable_aggregate_pushdown 'false');
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT count(*) FROM fdw137_t1 GROUP BY c1 HAVING min(c1) > 500 ORDER BY 1;
+                          QUERY PLAN                          
+--------------------------------------------------------------
+ Sort
+   Output: (count(*)), c1
+   Sort Key: (count(*))
+   ->  HashAggregate
+         Output: count(*), c1
+         Group Key: fdw137_t1.c1
+         Filter: (min(fdw137_t1.c1) > 500)
+         ->  Foreign Scan on public.fdw137_t1
+               Output: _id, c1, c2, c3, c4, c5, c6, c7, c8
+               Foreign Namespace: mongo_fdw_regress.test_tbl1
+(10 rows)
+
+ALTER SERVER mongo_server OPTIONS (SET enable_aggregate_pushdown 'true');
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT count(*) FROM fdw137_t1 GROUP BY c1 HAVING min(c1) > 500 ORDER BY 1;
+                                   QUERY PLAN                                    
+---------------------------------------------------------------------------------
+ Sort
+   Output: (count(*)), c1
+   Sort Key: (count(*))
+   ->  Foreign Scan
+         Output: (count(*)), c1
+         Foreign Namespace: Aggregate on (mongo_fdw_regress.test_tbl1 fdw137_t1)
+(6 rows)
+
+-- Test the option at table level. Setting option at table level does not
+-- affect the setting at server level.
+ALTER FOREIGN TABLE fdw137_t1 OPTIONS (ADD enable_aggregate_pushdown 'false');
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT count(*) FROM fdw137_t1 GROUP BY c1 HAVING min(c1) > 500 ORDER BY 1;
+                          QUERY PLAN                          
+--------------------------------------------------------------
+ Sort
+   Output: (count(*)), c1
+   Sort Key: (count(*))
+   ->  HashAggregate
+         Output: count(*), c1
+         Group Key: fdw137_t1.c1
+         Filter: (min(fdw137_t1.c1) > 500)
+         ->  Foreign Scan on public.fdw137_t1
+               Output: _id, c1, c2, c3, c4, c5, c6, c7, c8
+               Foreign Namespace: mongo_fdw_regress.test_tbl1
+(10 rows)
+
+ALTER SERVER mongo_server OPTIONS (SET enable_aggregate_pushdown 'false');
+ALTER FOREIGN TABLE fdw137_t1 OPTIONS (SET enable_aggregate_pushdown 'true');
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT count(*) FROM fdw137_t1 GROUP BY c1 HAVING min(c1) > 500 ORDER BY 1;
+                                   QUERY PLAN                                    
+---------------------------------------------------------------------------------
+ Sort
+   Output: (count(*)), c1
+   Sort Key: (count(*))
+   ->  Foreign Scan
+         Output: (count(*)), c1
+         Foreign Namespace: Aggregate on (mongo_fdw_regress.test_tbl1 fdw137_t1)
+(6 rows)
+
+-- Test option for aggregation over join. Allow aggregation only if enabled for
+-- both the relations involved in the join.
+ALTER SERVER mongo_server OPTIONS (SET enable_aggregate_pushdown 'true');
+ALTER FOREIGN TABLE fdw137_t1 OPTIONS (SET enable_aggregate_pushdown 'false');
+ALTER FOREIGN TABLE fdw137_t2 OPTIONS (ADD enable_aggregate_pushdown 'false');
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT sum(t2.c1), t1.c8 FROM fdw137_t1 t1 LEFT JOIN fdw137_t2 t2 ON (t1.c8 = t2.c1) GROUP BY t1.c8 ORDER BY 2;
+                                                  QUERY PLAN                                                  
+--------------------------------------------------------------------------------------------------------------
+ Sort
+   Output: (sum(t2.c1)), t1.c8
+   Sort Key: t1.c8
+   ->  HashAggregate
+         Output: sum(t2.c1), t1.c8
+         Group Key: t1.c8
+         ->  Foreign Scan
+               Output: t1.c8, t2.c1
+               Foreign Namespace: (mongo_fdw_regress.test_tbl1 t1) LEFT JOIN (mongo_fdw_regress.test_tbl2 t2)
+(9 rows)
+
+ALTER SERVER mongo_server OPTIONS (SET enable_aggregate_pushdown 'true');
+ALTER FOREIGN TABLE fdw137_t1 OPTIONS (SET enable_aggregate_pushdown 'true');
+ALTER FOREIGN TABLE fdw137_t2 OPTIONS (SET enable_aggregate_pushdown 'false');
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT sum(t2.c1), t1.c8 FROM fdw137_t1 t1 LEFT JOIN fdw137_t2 t2 ON (t1.c8 = t2.c1) GROUP BY t1.c8 ORDER BY 2;
+                                                  QUERY PLAN                                                  
+--------------------------------------------------------------------------------------------------------------
+ Sort
+   Output: (sum(t2.c1)), t1.c8
+   Sort Key: t1.c8
+   ->  HashAggregate
+         Output: sum(t2.c1), t1.c8
+         Group Key: t1.c8
+         ->  Foreign Scan
+               Output: t1.c8, t2.c1
+               Foreign Namespace: (mongo_fdw_regress.test_tbl1 t1) LEFT JOIN (mongo_fdw_regress.test_tbl2 t2)
+(9 rows)
+
+ALTER SERVER mongo_server OPTIONS (SET enable_aggregate_pushdown 'false');
+ALTER FOREIGN TABLE fdw137_t1 OPTIONS (SET enable_aggregate_pushdown 'true');
+ALTER FOREIGN TABLE fdw137_t2 OPTIONS (SET enable_aggregate_pushdown 'true');
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT sum(t2.c1), t1.c8 FROM fdw137_t1 t1 LEFT JOIN fdw137_t2 t2 ON (t1.c8 = t2.c1) GROUP BY t1.c8 ORDER BY 2;
+                                                      QUERY PLAN                                                       
+-----------------------------------------------------------------------------------------------------------------------
+ Sort
+   Output: (sum(t2.c1)), t1.c8
+   Sort Key: t1.c8
+   ->  Foreign Scan
+         Output: (sum(t2.c1)), t1.c8
+         Foreign Namespace: Aggregate on ((mongo_fdw_regress.test_tbl1 t1) LEFT JOIN (mongo_fdw_regress.test_tbl2 t2))
+(6 rows)
+
+-- Check when enable_join_pushdown is OFF and enable_aggregate_pushdown is ON.
+-- Shouldn't push down join as well as aggregation.
+ALTER SERVER mongo_server OPTIONS (ADD enable_join_pushdown 'false');
+ALTER SERVER mongo_server OPTIONS (SET enable_aggregate_pushdown 'true');
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT sum(t2.c1), t1.c8 FROM fdw137_t1 t1 LEFT JOIN fdw137_t2 t2 ON (t1.c8 = t2.c1) GROUP BY t1.c8 ORDER BY 2;
+                             QUERY PLAN                             
+--------------------------------------------------------------------
+ GroupAggregate
+   Output: sum(t2.c1), t1.c8
+   Group Key: t1.c8
+   ->  Merge Left Join
+         Output: t1.c8, t2.c1
+         Merge Cond: (t1.c8 = t2.c1)
+         ->  Sort
+               Output: t1.c8
+               Sort Key: t1.c8
+               ->  Foreign Scan on public.fdw137_t1 t1
+                     Output: t1.c8
+                     Foreign Namespace: mongo_fdw_regress.test_tbl1
+         ->  Sort
+               Output: t2.c1
+               Sort Key: t2.c1
+               ->  Foreign Scan on public.fdw137_t2 t2
+                     Output: t2.c1
+                     Foreign Namespace: mongo_fdw_regress.test_tbl2
+(18 rows)
+
 -- Cleanup
 DELETE FROM fdw137_t1 WHERE c8 IS NULL;
 DELETE FROM fdw137_t1 WHERE c8 = 60;

diff --git a/expected/aggregate_pushdown_1.out b/expected/aggregate_pushdown_1.out
@@ -1133,6 +1133,155 @@ SELECT c1, count(t1) FROM fprt1 t1 GROUP BY c1 HAVING avg(c2) < 22 ORDER BY 1;
 (8 rows)
 
 SET enable_partitionwise_aggregate TO OFF;
+-- Support enable_aggregate_pushdown option at server level and table level.
+-- Check only boolean values are accepted.
+ALTER SERVER mongo_server OPTIONS (ADD enable_aggregate_pushdown 'non-bolean');
+ERROR:  enable_aggregate_pushdown requires a Boolean value
+-- Test the option at server level.
+ALTER SERVER mongo_server OPTIONS (ADD enable_aggregate_pushdown 'false');
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT count(*) FROM fdw137_t1 GROUP BY c1 HAVING min(c1) > 500 ORDER BY 1;
+                          QUERY PLAN                          
+--------------------------------------------------------------
+ Sort
+   Output: (count(*)), c1
+   Sort Key: (count(*))
+   ->  HashAggregate
+         Output: count(*), c1
+         Group Key: fdw137_t1.c1
+         Filter: (min(fdw137_t1.c1) > 500)
+         ->  Foreign Scan on public.fdw137_t1
+               Output: _id, c1, c2, c3, c4, c5, c6, c7, c8
+               Foreign Namespace: mongo_fdw_regress.test_tbl1
+(10 rows)
+
+ALTER SERVER mongo_server OPTIONS (SET enable_aggregate_pushdown 'true');
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT count(*) FROM fdw137_t1 GROUP BY c1 HAVING min(c1) > 500 ORDER BY 1;
+                                   QUERY PLAN                                    
+---------------------------------------------------------------------------------
+ Sort
+   Output: (count(*)), c1
+   Sort Key: (count(*))
+   ->  Foreign Scan
+         Output: (count(*)), c1
+         Foreign Namespace: Aggregate on (mongo_fdw_regress.test_tbl1 fdw137_t1)
+(6 rows)
+
+-- Test the option at table level. Setting option at table level does not
+-- affect the setting at server level.
+ALTER FOREIGN TABLE fdw137_t1 OPTIONS (ADD enable_aggregate_pushdown 'false');
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT count(*) FROM fdw137_t1 GROUP BY c1 HAVING min(c1) > 500 ORDER BY 1;
+                          QUERY PLAN                          
+--------------------------------------------------------------
+ Sort
+   Output: (count(*)), c1
+   Sort Key: (count(*))
+   ->  HashAggregate
+         Output: count(*), c1
+         Group Key: fdw137_t1.c1
+         Filter: (min(fdw137_t1.c1) > 500)
+         ->  Foreign Scan on public.fdw137_t1
+               Output: _id, c1, c2, c3, c4, c5, c6, c7, c8
+               Foreign Namespace: mongo_fdw_regress.test_tbl1
+(10 rows)
+
+ALTER SERVER mongo_server OPTIONS (SET enable_aggregate_pushdown 'false');
+ALTER FOREIGN TABLE fdw137_t1 OPTIONS (SET enable_aggregate_pushdown 'true');
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT count(*) FROM fdw137_t1 GROUP BY c1 HAVING min(c1) > 500 ORDER BY 1;
+                                   QUERY PLAN                                    
+---------------------------------------------------------------------------------
+ Sort
+   Output: (count(*)), c1
+   Sort Key: (count(*))
+   ->  Foreign Scan
+         Output: (count(*)), c1
+         Foreign Namespace: Aggregate on (mongo_fdw_regress.test_tbl1 fdw137_t1)
+(6 rows)
+
+-- Test option for aggregation over join. Allow aggregation only if enabled for
+-- both the relations involved in the join.
+ALTER SERVER mongo_server OPTIONS (SET enable_aggregate_pushdown 'true');
+ALTER FOREIGN TABLE fdw137_t1 OPTIONS (SET enable_aggregate_pushdown 'false');
+ALTER FOREIGN TABLE fdw137_t2 OPTIONS (ADD enable_aggregate_pushdown 'false');
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT sum(t2.c1), t1.c8 FROM fdw137_t1 t1 LEFT JOIN fdw137_t2 t2 ON (t1.c8 = t2.c1) GROUP BY t1.c8 ORDER BY 2;
+                                                  QUERY PLAN                                                  
+--------------------------------------------------------------------------------------------------------------
+ Sort
+   Output: (sum(t2.c1)), t1.c8
+   Sort Key: t1.c8
+   ->  HashAggregate
+         Output: sum(t2.c1), t1.c8
+         Group Key: t1.c8
+         ->  Foreign Scan
+               Output: t1.c8, t2.c1
+               Foreign Namespace: (mongo_fdw_regress.test_tbl1 t1) LEFT JOIN (mongo_fdw_regress.test_tbl2 t2)
+(9 rows)
+
+ALTER SERVER mongo_server OPTIONS (SET enable_aggregate_pushdown 'true');
+ALTER FOREIGN TABLE fdw137_t1 OPTIONS (SET enable_aggregate_pushdown 'true');
+ALTER FOREIGN TABLE fdw137_t2 OPTIONS (SET enable_aggregate_pushdown 'false');
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT sum(t2.c1), t1.c8 FROM fdw137_t1 t1 LEFT JOIN fdw137_t2 t2 ON (t1.c8 = t2.c1) GROUP BY t1.c8 ORDER BY 2;
+                                                  QUERY PLAN                                                  
+--------------------------------------------------------------------------------------------------------------
+ Sort
+   Output: (sum(t2.c1)), t1.c8
+   Sort Key: t1.c8
+   ->  HashAggregate
+         Output: sum(t2.c1), t1.c8
+         Group Key: t1.c8
+         ->  Foreign Scan
+               Output: t1.c8, t2.c1
+               Foreign Namespace: (mongo_fdw_regress.test_tbl1 t1) LEFT JOIN (mongo_fdw_regress.test_tbl2 t2)
+(9 rows)
+
+ALTER SERVER mongo_server OPTIONS (SET enable_aggregate_pushdown 'false');
+ALTER FOREIGN TABLE fdw137_t1 OPTIONS (SET enable_aggregate_pushdown 'true');
+ALTER FOREIGN TABLE fdw137_t2 OPTIONS (SET enable_aggregate_pushdown 'true');
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT sum(t2.c1), t1.c8 FROM fdw137_t1 t1 LEFT JOIN fdw137_t2 t2 ON (t1.c8 = t2.c1) GROUP BY t1.c8 ORDER BY 2;
+                                                      QUERY PLAN                                                       
+-----------------------------------------------------------------------------------------------------------------------
+ Sort
+   Output: (sum(t2.c1)), t1.c8
+   Sort Key: t1.c8
+   ->  Foreign Scan
+         Output: (sum(t2.c1)), t1.c8
+         Foreign Namespace: Aggregate on ((mongo_fdw_regress.test_tbl1 t1) LEFT JOIN (mongo_fdw_regress.test_tbl2 t2))
+(6 rows)
+
+-- Check when enable_join_pushdown is OFF and enable_aggregate_pushdown is ON.
+-- Shouldn't push down join as well as aggregation.
+ALTER SERVER mongo_server OPTIONS (ADD enable_join_pushdown 'false');
+ALTER SERVER mongo_server OPTIONS (SET enable_aggregate_pushdown 'true');
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT sum(t2.c1), t1.c8 FROM fdw137_t1 t1 LEFT JOIN fdw137_t2 t2 ON (t1.c8 = t2.c1) GROUP BY t1.c8 ORDER BY 2;
+                             QUERY PLAN                             
+--------------------------------------------------------------------
+ GroupAggregate
+   Output: sum(t2.c1), t1.c8
+   Group Key: t1.c8
+   ->  Merge Left Join
+         Output: t1.c8, t2.c1
+         Merge Cond: (t1.c8 = t2.c1)
+         ->  Sort
+               Output: t1.c8
+               Sort Key: t1.c8
+               ->  Foreign Scan on public.fdw137_t1 t1
+                     Output: t1.c8
+                     Foreign Namespace: mongo_fdw_regress.test_tbl1
+         ->  Sort
+               Output: t2.c1
+               Sort Key: t2.c1
+               ->  Foreign Scan on public.fdw137_t2 t2
+                     Output: t2.c1
+                     Foreign Namespace: mongo_fdw_regress.test_tbl2
+(18 rows)
+
 -- Cleanup
 DELETE FROM fdw137_t1 WHERE c8 IS NULL;
 DELETE FROM fdw137_t1 WHERE c8 = 60;