Restore fault hooks and legacy TOAST chunk reads

contrib/pax_storage/src/test/regress/expected/misc_jiras.out

@@ -39,6 +39,9 @@ select sum(cc) from (
       from misc_jiras.t1
      group by 1, 2
 ) tt;
+NOTICE:  winagg: tuplestore spilled to disk  (seg0 slice1 127.0.0.1:7002 pid=54719)
+NOTICE:  winagg: tuplestore spilled to disk  (seg1 slice1 127.0.0.1:7003 pid=54720)
+NOTICE:  winagg: tuplestore spilled to disk  (seg2 slice1 127.0.0.1:7004 pid=54721)
    sum   
 ---------
  30006.5

contrib/pax_storage/src/test/regress/expected/toast.out

@@ -159,6 +159,14 @@ SELECT DISTINCT gp_inject_fault('decrease_toast_max_chunk_size', 'skip', dbid)
 (1 row)
 
 INSERT INTO toast_chunk_test VALUES (repeat('abcdefghijklmnopqrstuvwxyz', 1000)::bytea);
+SELECT bool_or(gp_inject_fault('decrease_toast_max_chunk_size', 'status', dbid)
+               LIKE '%num times hit:''1''%') AS toast_chunk_fault_fired
+  FROM gp_segment_configuration WHERE role='p';
+ toast_chunk_fault_fired 
+-------------------------
+ t
+(1 row)
+
 SELECT DISTINCT gp_inject_fault('decrease_toast_max_chunk_size', 'reset', dbid)
 	   FROM pg_catalog.gp_segment_configuration
 	   WHERE role = 'p';
@@ -180,3 +188,16 @@ SELECT encode(substring(a from 521*26+1 for 26), 'escape') FROM toast_chunk_test
  abcdefghijklmnopqrstuvwxyz
 (1 row)
 
+-- Check prefixes, chunk boundaries, the final chunk, and slices past the end.
+SELECT bool_and(substring(a FROM pos FOR len) =
+                substring(repeat('abcdefghijklmnopqrstuvwxyz', 1000)::bytea FROM pos FOR len))
+       AS toast_slices_match
+  FROM toast_chunk_test,
+       (VALUES (1, 9000), (10, 16300), (1980, 30), (8130, 30),
+               (13000, 40), (25980, 40), (27000, 2)) AS slices(pos, len);
+ toast_slices_match 
+--------------------
+ t
+(1 row)
+
+

contrib/pax_storage/src/test/regress/sql/toast.sql

@@ -86,6 +86,9 @@ SELECT DISTINCT gp_inject_fault('decrease_toast_max_chunk_size', 'skip', dbid)
 	   FROM pg_catalog.gp_segment_configuration
 	   WHERE role = 'p';
 INSERT INTO toast_chunk_test VALUES (repeat('abcdefghijklmnopqrstuvwxyz', 1000)::bytea);
+SELECT bool_or(gp_inject_fault('decrease_toast_max_chunk_size', 'status', dbid)
+               LIKE '%num times hit:''1''%') AS toast_chunk_fault_fired
+  FROM gp_segment_configuration WHERE role='p';
 SELECT DISTINCT gp_inject_fault('decrease_toast_max_chunk_size', 'reset', dbid)
 	   FROM pg_catalog.gp_segment_configuration
 	   WHERE role = 'p';
@@ -95,3 +98,11 @@ SELECT * FROM toast_chunk_test WHERE a <> repeat('abcdefghijklmnopqrstuvwxyz', 1
 
 -- Random access into the toast table should work equally well.
 SELECT encode(substring(a from 521*26+1 for 26), 'escape') FROM toast_chunk_test;
+
+-- Check prefixes, chunk boundaries, the final chunk, and slices past the end.
+SELECT bool_and(substring(a FROM pos FOR len) =
+                substring(repeat('abcdefghijklmnopqrstuvwxyz', 1000)::bytea FROM pos FOR len))
+       AS toast_slices_match
+  FROM toast_chunk_test,
+       (VALUES (1, 9000), (10, 16300), (1980, 30), (8130, 30),
+               (13000, 40), (25980, 40), (27000, 2)) AS slices(pos, len);

gpcontrib/diskquota/src/gp_activetable.c

@@ -428,6 +428,8 @@ diskquota_fetch_table_stat(PG_FUNCTION_ARGS)
 	DiskQuotaSetOFCache       *cache           = NULL;
 	DiskQuotaActiveTableEntry *results_entry   = NULL;
 
+	SIMPLE_FAULT_INJECTOR("diskquota_fetch_table_stat");
+
 #ifdef FAULT_INJECTOR
 	if (SIMPLE_FAULT_INJECTOR("ereport_warning_from_segment") == FaultInjectorTypeSkip)
 	{
@@ -1198,4 +1200,4 @@ pull_active_table_size_from_seg(HTAB *local_table_stats_map, char *active_oid_ar
 	}
 	cdbdisp_clearCdbPgResults(&cdb_pgresults);
 	return;
-}
+}

gpcontrib/diskquota/tests/regress/expected/test_fetch_table_stat.out

@@ -3,8 +3,6 @@
 --    the error message is preserved for us to debug.
 --
 CREATE TABLE t_error_handling (i int) DISTRIBUTED BY (i);
-NOTICE:  Table doesn't have 'DISTRIBUTED BY' clause -- Using column named 'i' as the Apache Cloudberry data distribution key for this table.
-HINT:  The 'DISTRIBUTED BY' clause determines the distribution of data. Make sure column(s) chosen are the optimal data distribution key to minimize skew.
 -- Inject an error to a segment server, since this UDF is only called on segments.
 SELECT gp_inject_fault_infinite('diskquota_fetch_table_stat', 'error', dbid)
     FROM gp_segment_configuration WHERE role='p' AND content=0;
@@ -14,15 +12,19 @@ SELECT gp_inject_fault_infinite('diskquota_fetch_table_stat', 'error', dbid)
 (1 row)
 
 -- Dispatch diskquota_fetch_table_stat to segments.
--- There should be a warning message from segment server saying:
+-- There should be an error message from the segment server saying:
 -- fault triggered, fault name:'diskquota_fetch_table_stat' fault type:'error'
 -- We're not interested in the oid here, we aggregate the result by COUNT(*).
 SELECT COUNT(*)
     FROM (SELECT diskquota.diskquota_fetch_table_stat(1, array[(SELECT oid FROM pg_class WHERE relname='t_error_handling')])
           FROM gp_dist_random('gp_id') WHERE gp_segment_id=0) AS count;
- count 
--------
-     1
+ERROR:  fault triggered, fault name:'diskquota_fetch_table_stat' fault type:'error'  (seg0 slice1 127.0.0.1:7002 pid=1000)
+SELECT bool_and(gp_inject_fault('diskquota_fetch_table_stat', 'status', dbid)
+                ~ 'num times hit:''[1-9][0-9]*''') AS diskquota_fault_fired
+    FROM gp_segment_configuration WHERE role='p' AND content=0;
+ diskquota_fault_fired 
+-----------------------
+ t
 (1 row)
 
 -- Reset the fault injector to prevent future failure.

gpcontrib/diskquota/tests/regress/sql/test_fetch_table_stat.sql

@@ -9,13 +9,17 @@ SELECT gp_inject_fault_infinite('diskquota_fetch_table_stat', 'error', dbid)
     FROM gp_segment_configuration WHERE role='p' AND content=0;
 
 -- Dispatch diskquota_fetch_table_stat to segments.
--- There should be a warning message from segment server saying:
+-- There should be an error message from the segment server saying:
 -- fault triggered, fault name:'diskquota_fetch_table_stat' fault type:'error'
 -- We're not interested in the oid here, we aggregate the result by COUNT(*).
 SELECT COUNT(*)
     FROM (SELECT diskquota.diskquota_fetch_table_stat(1, array[(SELECT oid FROM pg_class WHERE relname='t_error_handling')])
           FROM gp_dist_random('gp_id') WHERE gp_segment_id=0) AS count;
 
+SELECT bool_and(gp_inject_fault('diskquota_fetch_table_stat', 'status', dbid)
+                ~ 'num times hit:''[1-9][0-9]*''') AS diskquota_fault_fired
+    FROM gp_segment_configuration WHERE role='p' AND content=0;
+
 -- Reset the fault injector to prevent future failure.
 SELECT gp_inject_fault_infinite('diskquota_fetch_table_stat', 'reset', dbid)
     FROM gp_segment_configuration WHERE role='p' AND content=0;

src/backend/access/common/toast_internals.c

@@ -23,6 +23,7 @@
 #include "catalog/catalog.h"
 #include "common/pg_lzcompress.h"
 #include "miscadmin.h"
+#include "utils/faultinjector.h"
 #include "utils/fmgroids.h"
 #include "utils/rel.h"
 #include "utils/snapmgr.h"
@@ -139,6 +140,7 @@ toast_save_datum(Relation rel, Datum value,
 		int32		align_it;
 	}			chunk_data;
 	int32		chunk_size;
+	int32		max_chunk_size = TOAST_MAX_CHUNK_SIZE;
 	int32		chunk_seq = 0;
 	char	   *data_p;
 	int32		data_todo;
@@ -300,6 +302,12 @@ toast_save_datum(Relation rel, Datum value,
 	t_isnull[1] = false;
 	t_isnull[2] = false;
 
+#ifdef FAULT_INJECTOR
+	/* Exercise reading values written with a different chunk size before upgrade. */
+	if (SIMPLE_FAULT_INJECTOR("decrease_toast_max_chunk_size") == FaultInjectorTypeSkip)
+		max_chunk_size--;
+#endif
+
 	/*
 	 * Split up the item into chunks
 	 */
@@ -312,7 +320,7 @@ toast_save_datum(Relation rel, Datum value,
 		/*
 		 * Calculate the size of this chunk
 		 */
-		chunk_size = Min(TOAST_MAX_CHUNK_SIZE, data_todo);
+		chunk_size = Min(max_chunk_size, data_todo);
 
 		/*
 		 * Build a tuple and store it

src/backend/access/heap/heaptoast.c

@@ -689,7 +689,9 @@ heap_fetch_toast_slice(Relation toastrel, Oid valueid, int32 attrsize,
 	SysScanDesc toastscan;
 	HeapTuple	ttup;
 	int32		expectedchunk;
-	int32		totalchunks = ((attrsize - 1) / TOAST_MAX_CHUNK_SIZE) + 1;
+	int32		totalchunks;
+	int32		max_chunk_size = TOAST_MAX_CHUNK_SIZE;
+	bool		fetch_whole_value = sliceoffset == 0 && slicelength == attrsize;
 	int			startchunk;
 	int			endchunk;
 	int			num_indexes;
@@ -702,21 +704,61 @@ heap_fetch_toast_slice(Relation toastrel, Oid valueid, int32 attrsize,
 									&toastidxs,
 									&num_indexes);
 
-	startchunk = sliceoffset / TOAST_MAX_CHUNK_SIZE;
-	endchunk = (sliceoffset + slicelength - 1) / TOAST_MAX_CHUNK_SIZE;
-	Assert(endchunk <= totalchunks);
-
 	/* Set up a scan key to fetch from the index. */
 	ScanKeyInit(&toastkey[0],
 				(AttrNumber) 1,
 				BTEqualStrategyNumber, F_OIDEQ,
 				ObjectIdGetDatum(valueid));
 
+	init_toast_snapshot(&SnapshotToast);
+
 	/*
-	 * No additional condition if fetching all chunks. Otherwise, use an
-	 * equality condition for one chunk, and a range condition otherwise.
+	 * Older clusters can have used a different TOAST_MAX_CHUNK_SIZE.  A slice
+	 * starting in the middle must inspect chunk zero before computing its
+	 * scan bounds.  Reads starting at zero learn the size in the main scan.
 	 */
-	if (startchunk == 0 && endchunk == totalchunks - 1)
+	if (sliceoffset > 0)
+	{
+		Pointer		chunk;
+		int32		chunksize;
+		bool		isnull;
+
+		ScanKeyInit(&toastkey[1], (AttrNumber) 2,
+					BTEqualStrategyNumber, F_INT4EQ, Int32GetDatum(0));
+		toastscan = systable_beginscan_ordered(toastrel, toastidxs[validIndex],
+											   &SnapshotToast, 2, toastkey);
+		ttup = systable_getnext_ordered(toastscan, ForwardScanDirection);
+		if (ttup == NULL)
+			ereport(ERROR,
+					(errcode(ERRCODE_DATA_CORRUPTED),
+					 errmsg_internal("missing chunk number 0 for toast value %u in %s",
+									 valueid, RelationGetRelationName(toastrel))));
+		chunk = DatumGetPointer(fastgetattr(ttup, 3, toasttupDesc, &isnull));
+		Assert(!isnull);
+		if (VARATT_IS_EXTENDED(chunk) && !VARATT_IS_SHORT(chunk))
+			elog(ERROR, "found toasted toast chunk for toast value %u in %s",
+				 valueid, RelationGetRelationName(toastrel));
+		chunksize = VARSIZE_ANY_EXHDR(chunk);
+		if (chunksize <= 0)
+			ereport(ERROR,
+					(errcode(ERRCODE_DATA_CORRUPTED),
+					 errmsg_internal("invalid chunk size %d for toast value %u in %s",
+									 chunksize, valueid, RelationGetRelationName(toastrel))));
+		if (chunksize < attrsize)
+			max_chunk_size = chunksize;
+		systable_endscan_ordered(toastscan);
+	}
+
+	totalchunks = ((attrsize - 1) / max_chunk_size) + 1;
+	startchunk = sliceoffset / max_chunk_size;
+	endchunk = (sliceoffset + slicelength - 1) / max_chunk_size;
+	Assert(endchunk < totalchunks);
+
+	/*
+	 * Reads starting at chunk zero cannot limit the end until its size is
+	 * known. Otherwise, use equality for one chunk, and a range for several.
+	 */
+	if (startchunk == 0)
 		nscankeys = 1;
 	else if (startchunk == endchunk)
 	{
@@ -740,7 +782,6 @@ heap_fetch_toast_slice(Relation toastrel, Oid valueid, int32 attrsize,
 	}
 
 	/* Prepare for scan */
-	init_toast_snapshot(&SnapshotToast);
 	toastscan = systable_beginscan_ordered(toastrel, toastidxs[validIndex],
 										   &SnapshotToast, nscankeys, toastkey);
 
@@ -804,8 +845,14 @@ heap_fetch_toast_slice(Relation toastrel, Oid valueid, int32 attrsize,
 									 curchunk,
 									 startchunk, endchunk, valueid,
 									 RelationGetRelationName(toastrel))));
-		expected_size = curchunk < totalchunks - 1 ? TOAST_MAX_CHUNK_SIZE
-			: attrsize - ((totalchunks - 1) * TOAST_MAX_CHUNK_SIZE);
+		if (curchunk == 0 && chunksize > 0 && chunksize < attrsize)
+		{
+			max_chunk_size = chunksize;
+			totalchunks = ((attrsize - 1) / max_chunk_size) + 1;
+			endchunk = (sliceoffset + slicelength - 1) / max_chunk_size;
+		}
+		expected_size = curchunk < totalchunks - 1 ? max_chunk_size
+			: attrsize - ((totalchunks - 1) * max_chunk_size);
 		if (chunksize != expected_size)
 			ereport(ERROR,
 					(errcode(ERRCODE_DATA_CORRUPTED),
@@ -820,16 +867,18 @@ heap_fetch_toast_slice(Relation toastrel, Oid valueid, int32 attrsize,
 		chcpystrt = 0;
 		chcpyend = chunksize - 1;
 		if (curchunk == startchunk)
-			chcpystrt = sliceoffset % TOAST_MAX_CHUNK_SIZE;
+			chcpystrt = sliceoffset % max_chunk_size;
 		if (curchunk == endchunk)
-			chcpyend = (sliceoffset + slicelength - 1) % TOAST_MAX_CHUNK_SIZE;
+			chcpyend = (sliceoffset + slicelength - 1) % max_chunk_size;
 
 		memcpy(VARDATA(result) +
-			   (curchunk * TOAST_MAX_CHUNK_SIZE - sliceoffset) + chcpystrt,
+			   (curchunk * max_chunk_size - sliceoffset) + chcpystrt,
 			   chunkdata + chcpystrt,
 			   (chcpyend - chcpystrt) + 1);
 
 		expectedchunk++;
+		if (!fetch_whole_value && expectedchunk > endchunk)
+			break;
 	}
 
 	/*

src/backend/executor/nodeWindowAgg.c

@@ -2489,6 +2489,21 @@ ExecWindowAgg(PlanState *pstate)
 		 */
 		spool_tuples(winstate, winstate->currentpos);
 
+#ifdef FAULT_INJECTOR
+		/* Report the completed partition before releasing its tuplestore. */
+		if (winstate->partition_spooled &&
+			winstate->currentpos >= winstate->spooled_rows &&
+			SIMPLE_FAULT_INJECTOR("winagg_after_spool_tuples") == FaultInjectorTypeSkip)
+		{
+			if (winstate->buffer == NULL)
+				ereport(NOTICE, (errmsg("winagg: no input rows")));
+			else if (tuplestore_in_memory(winstate->buffer))
+				ereport(NOTICE, (errmsg("winagg: tuplestore fitted in memory")));
+			else
+				ereport(NOTICE, (errmsg("winagg: tuplestore spilled to disk")));
+		}
+#endif
+
 		/* Move to the next partition if we reached the end of this partition */
 		if (winstate->partition_spooled &&
 			winstate->currentpos >= winstate->spooled_rows)

src/bin/gpfdist/gpfdist.c

@@ -3990,9 +3990,11 @@ int gpfdist_init(int argc, const char* const argv[])
 		errno = 0;
 		val = strtol(wd, &endptr, 10);
 
-		if (errno || endptr == wd || val > INT_MAX)
+		if (errno || endptr == wd || *endptr != '\0' ||
+			val < 0 || val > INT_MAX)
 		{
-			fprintf(stderr, "incorrect watchdog timer: %s\n", strerror(errno));
+			fprintf(stderr, "incorrect watchdog timer \"%s\": expected an integer between 0 and %d\n",
+					wd, INT_MAX);
 			return -1;
 		}
 

src/bin/gpfdist/regress/Makefile

@@ -38,6 +38,7 @@ endif
 	$(top_builddir)/src/test/regress/pg_regress --dbname=gpfdist_regression $(REGRESS) $(REGRESS_OPTS)
 
 watchdog:
+	python3 test_watchdog_options.py
 	sh test_watchdog.sh
 
 ipv4v6_ports:

src/bin/gpfdist/regress/test_watchdog_options.py

@@ -0,0 +1,44 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import os
+import socket
+import subprocess
+import tempfile
+import unittest
+
+
+class WatchdogOptionsTest(unittest.TestCase):
+    def test_invalid_timers(self):
+        with tempfile.TemporaryDirectory() as directory:
+            for value in ("", "abc", "2junk", "2.5", "-1", "-4294967295",
+                          "2147483648", "9223372036854775808"):
+                with self.subTest(value=value):
+                    # Avoid depending on the default port being free.
+                    with socket.socket() as sock:
+                        sock.bind(("127.0.0.1", 0))
+                        port = sock.getsockname()[1]
+                    env = dict(os.environ, GPFDIST_WATCHDOG_TIMER=value)
+                    result = subprocess.run(
+                        ["gpfdist", "-p", str(port), "-d", directory],
+                        env=env, capture_output=True, text=True, timeout=10)
+                    self.assertNotEqual(result.returncode, 0)
+                    self.assertIn("incorrect watchdog timer", result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()

src/test/regress/expected/misc_jiras.out

@@ -39,6 +39,9 @@ select sum(cc) from (
       from misc_jiras.t1
      group by 1, 2
 ) tt;
+NOTICE:  winagg: tuplestore spilled to disk  (seg0 slice1 127.0.0.1:7002 pid=54719)
+NOTICE:  winagg: tuplestore spilled to disk  (seg1 slice1 127.0.0.1:7003 pid=54720)
+NOTICE:  winagg: tuplestore spilled to disk  (seg2 slice1 127.0.0.1:7004 pid=54721)
    sum   
 ---------
  30006.5