Rebuild the query for recalculating min/max after search_path change

Cached plans for recalculating min/max values are built using
pg_ivm_get_viewdef() that returns the view definition query text.
Therefore, if the search_path is changed, the query text is analyzed
again by SPI, and tables or functions in a wrong schema could be
referenced in the plan.

To fix this, we check whether the search_path is still the same
as when we made the cached plan and, if it isn't, we rebuild the
query text.

CVE-2023-23554

expected/pg_ivm.out

@@ -464,6 +464,40 @@ SELECT * FROM mv_ivm_min_max;
        |       |                   0 |                   0 |             0
 (1 row)
 
+ROLLBACK;
+-- Test MIN/MAX after search_path change
+BEGIN;
+SELECT create_immv('mv_ivm_min', 'SELECT MIN(j) FROM mv_base_a');
+ create_immv 
+-------------
+           1
+(1 row)
+
+SELECT * FROM mv_ivm_min ORDER BY 1,2,3;
+ min | __ivm_count_min__ | __ivm_count__ 
+-----+-------------------+---------------
+  10 |                 5 |             5
+(1 row)
+
+CREATE SCHEMA myschema;
+GRANT ALL ON SCHEMA myschema TO public;
+CREATE TABLE myschema.mv_base_a (j int);
+INSERT INTO myschema.mv_base_a VALUES (1);
+DELETE FROM mv_base_a WHERE (i,j) = (1,10);
+SELECT * FROM mv_ivm_min ORDER BY 1,2,3;
+ min | __ivm_count_min__ | __ivm_count__ 
+-----+-------------------+---------------
+  20 |                 4 |             4
+(1 row)
+
+SET search_path TO myschema,public,pg_catalog;
+DELETE FROM public.mv_base_a WHERE (i,j) = (2,20);
+SELECT * FROM mv_ivm_min ORDER BY 1,2,3;
+ min | __ivm_count_min__ | __ivm_count__ 
+-----+-------------------+---------------
+  30 |                 3 |             3
+(1 row)
+
 ROLLBACK;
 -- aggregate views with column names specified
 BEGIN;

matview.c

@@ -78,6 +78,9 @@ typedef struct MV_QueryHashEntry
 {
 	MV_QueryKey key;
 	SPIPlanPtr	plan;
+	OverrideSearchPath *search_path;	/* search_path used for parsing
+										 * and planning */
+
 } MV_QueryHashEntry;
 
 /*
@@ -2807,18 +2810,27 @@ mv_FetchPreparedPlan(MV_QueryKey *key)
 	 *
 	 * CAUTION: this check is only trustworthy if the caller has already
 	 * locked both materialized views and base tables.
+	 *
+	 * Also, check whether the search_path is still the same as when we made it.
+	 * If it isn't, we need to rebuild the query text because the result of
+	 * pg_ivm_get_viewdef() will change.
 	 */
 	plan = entry->plan;
-	if (plan && SPI_plan_is_valid(plan))
+	if (plan && SPI_plan_is_valid(plan) &&
+		OverrideSearchPathMatchesCurrent(entry->search_path))
 		return plan;
 
 	/*
 	 * Otherwise we might as well flush the cached plan now, to free a little
 	 * memory space before we make a new one.
 	 */
-	entry->plan = NULL;
 	if (plan)
 		SPI_freeplan(plan);
+	if (entry->search_path)
+		pfree(entry->search_path);
+
+	entry->plan = NULL;
+	entry->search_path = NULL;
 
 	return NULL;
 }
@@ -2849,6 +2861,7 @@ mv_HashPreparedPlan(MV_QueryKey *key, SPIPlanPtr plan)
 											  HASH_ENTER, &found);
 	Assert(!found || entry->plan == NULL);
 	entry->plan = plan;
+	entry->search_path = GetOverrideSearchPath(TopMemoryContext);
 }
 
 /*

sql/pg_ivm.sql

@@ -150,6 +150,24 @@ DELETE FROM mv_base_a;
 SELECT * FROM mv_ivm_min_max;
 ROLLBACK;
 
+-- Test MIN/MAX after search_path change
+BEGIN;
+SELECT create_immv('mv_ivm_min', 'SELECT MIN(j) FROM mv_base_a');
+SELECT * FROM mv_ivm_min ORDER BY 1,2,3;
+
+CREATE SCHEMA myschema;
+GRANT ALL ON SCHEMA myschema TO public;
+CREATE TABLE myschema.mv_base_a (j int);
+INSERT INTO myschema.mv_base_a VALUES (1);
+
+DELETE FROM mv_base_a WHERE (i,j) = (1,10);
+SELECT * FROM mv_ivm_min ORDER BY 1,2,3;
+
+SET search_path TO myschema,public,pg_catalog;
+DELETE FROM public.mv_base_a WHERE (i,j) = (2,20);
+SELECT * FROM mv_ivm_min ORDER BY 1,2,3;
+ROLLBACK;
+
 -- aggregate views with column names specified
 BEGIN;
 SELECT create_immv('mv_ivm_agg(a)', 'SELECT i, SUM(j) FROM mv_base_a GROUP BY i');