4 files changed, 88 insertions, 1 deletions
diff --git a/src/backend/access/common/toast_internals.c b/src/backend/access/common/toast_internals.c
index c036319a0b8..8d2a9964c3f 100644
--- a/src/backend/access/common/toast_internals.c
+++ b/src/backend/access/common/toast_internals.c
@@ -638,8 +638,21 @@ init_toast_snapshot(Snapshot toast_snapshot)
 {
 	Snapshot	snapshot = GetOldestSnapshot();
 
+	/*
+	 * GetOldestSnapshot returns NULL if the session has no active snapshots.
+	 * We can get that if, for example, a procedure fetches a toasted value
+	 * into a local variable, commits, and then tries to detoast the value.
+	 * Such coding is unsafe, because once we commit there is nothing to
+	 * prevent the toast data from being deleted.  Detoasting *must* happen in
+	 * the same transaction that originally fetched the toast pointer.  Hence,
+	 * rather than trying to band-aid over the problem, throw an error.  (This
+	 * is not very much protection, because in many scenarios the procedure
+	 * would have already created a new transaction snapshot, preventing us
+	 * from detecting the problem.  But it's better than nothing, and for sure
+	 * we shouldn't expend code on masking the problem more.)
+	 */
 	if (snapshot == NULL)
-		elog(ERROR, "no known snapshots");
+		elog(ERROR, "cannot fetch toast data without an active snapshot");
 
 	InitToastSnapshot(*toast_snapshot, snapshot->lsn, snapshot->whenTaken);
 }
diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c
index b4c70aaa7fa..27d6ea75517 100644
--- a/src/pl/plpgsql/src/pl_exec.c
+++ b/src/pl/plpgsql/src/pl_exec.c
@@ -5827,6 +5827,17 @@ exec_for_query(PLpgSQL_execstate *estate, PLpgSQL_stmt_forq *stmt,
 	PinPortal(portal);
 
 	/*
+	 * In a non-atomic context, we dare not prefetch, even if it would
+	 * otherwise be safe.  Aside from any semantic hazards that that might
+	 * create, if we prefetch toasted data and then the user commits the
+	 * transaction, the toast references could turn into dangling pointers.
+	 * (Rows we haven't yet fetched from the cursor are safe, because the
+	 * PersistHoldablePortal mechanism handles this scenario.)
+	 */
+	if (!estate->atomic)
+		prefetch_ok = false;
+
+	/*
 	 * Fetch the initial tuple(s).  If prefetching is allowed then we grab a
 	 * few more rows to avoid multiple trips through executor startup
 	 * overhead.
diff --git a/src/test/isolation/expected/plpgsql-toast.out b/src/test/isolation/expected/plpgsql-toast.out
index fc557da5e77..4f216b94b62 100644
--- a/src/test/isolation/expected/plpgsql-toast.out
+++ b/src/test/isolation/expected/plpgsql-toast.out
@@ -192,3 +192,46 @@ pg_advisory_unlock
 t              
 s1: NOTICE:  length(r) = 6002
 step assign5: <... completed>
+
+starting permutation: lock assign6 vacuum unlock
+pg_advisory_unlock_all
+
+               
+pg_advisory_unlock_all
+
+               
+step lock: 
+    SELECT pg_advisory_lock(1);
+
+pg_advisory_lock
+
+               
+step assign6: 
+do $$
+  declare
+    r record;
+  begin
+    insert into test1 values (2, repeat('bar', 3000));
+    insert into test1 values (3, repeat('baz', 4000));
+    for r in select test1.b from test1 loop
+      delete from test1;
+      commit;
+      perform pg_advisory_lock(1);
+      raise notice 'length(r) = %', length(r::text);
+    end loop;
+  end;
+$$;
+ <waiting ...>
+step vacuum: 
+    VACUUM test1;
+
+step unlock: 
+    SELECT pg_advisory_unlock(1);
+
+pg_advisory_unlock
+
+t              
+s1: NOTICE:  length(r) = 6002
+s1: NOTICE:  length(r) = 9002
+s1: NOTICE:  length(r) = 12002
+step assign6: <... completed>
diff --git a/src/test/isolation/specs/plpgsql-toast.spec b/src/test/isolation/specs/plpgsql-toast.spec
index fe7090addbb..d360f8fccbf 100644
--- a/src/test/isolation/specs/plpgsql-toast.spec
+++ b/src/test/isolation/specs/plpgsql-toast.spec
@@ -112,6 +112,25 @@ do $$
 $$;
 }
 
+# FOR loop must not hold any fetched-but-not-detoasted values across commit
+step "assign6"
+{
+do $$
+  declare
+    r record;
+  begin
+    insert into test1 values (2, repeat('bar', 3000));
+    insert into test1 values (3, repeat('baz', 4000));
+    for r in select test1.b from test1 loop
+      delete from test1;
+      commit;
+      perform pg_advisory_lock(1);
+      raise notice 'length(r) = %', length(r::text);
+    end loop;
+  end;
+$$;
+}
+
 session "s2"
 setup
 {
@@ -135,3 +154,4 @@ permutation "lock" "assign2" "vacuum" "unlock"
 permutation "lock" "assign3" "vacuum" "unlock"
 permutation "lock" "assign4" "vacuum" "unlock"
 permutation "lock" "assign5" "vacuum" "unlock"
+permutation "lock" "assign6" "vacuum" "unlock"