Don't capture hard errors for old cagg format

If an error occurs inside the deserialization function for continuous
aggregates, we are using the old numeric format and this was captured
inside `inner_agg_deserialize` and the format was "repaired". However,
this captured hard errors as well, which can cause cascading errors if
execution continues after the error.

This commit fixes that by capturing data exception errors, internal
error (which is the default for elog()) and protocol error (c.f.
`pq_getmsgint64` used by `numeric_deserialize`), but re-throwing all
other errors.

It also adds tests to make sure that the repair is done and not
generates an error.

Author: mkindahl

.github/gh_matrix_builder.py

@@ -61,6 +61,9 @@
     # Not executed as a flake test since it takes a very long time and
     # easily interferes with other tests.
     "bgw_scheduler_restart",
+    # Not executed as a flake test since it generates a lot of cruft
+    # in the log at debug levels.
+    "partialize_finalize",
 }
 
 

.unreleased/pr_7903

@@ -0,0 +1 @@
+Fixes: #7903 Don't capture hard errors for old cagg format

tsl/src/hypercore/attr_capture.c

@@ -240,10 +240,7 @@ capture_ExecutorStart(QueryDesc *queryDesc, int eflags)
 	foreach (cell, queryDesc->plannedstmt->rtable)
 	{
 		RangeTblEntry *rte = lfirst(cell);
-		TS_DEBUG_LOG("rtable #%d: %s (relid: %d)",
-					 foreach_current_index(cell),
-					 get_rel_name(rte->relid),
-					 rte->relid);
+		TS_DEBUG_LOG("rtable #%d: %s", foreach_current_index(cell), get_rel_name(rte->relid));
 	}
 #endif
 

tsl/src/partialize_finalize.c

@@ -16,9 +16,11 @@
 #include <utils/builtins.h>
 #include <utils/datum.h>
 #include <utils/fmgroids.h>
+#include <utils/resowner.h>
 #include <utils/syscache.h>
 
 #include "partialize_finalize.h"
+#include "utils.h"
 
 /*
  * This file implements functions to split the calculation of partial and final
@@ -223,12 +225,15 @@ static bytea *
 sanitize_serialized_partial(Oid deserialfnoid, bytea *serialized_partial)
 {
 	if ((deserialfnoid == F_NUMERIC_DESERIALIZE) || (deserialfnoid == F_NUMERIC_AVG_DESERIALIZE))
+	{
 		/*
 		 * Always add NUMERIC_PARTIAL_MISSING_LENGTH extra bytes because the length is not fixed.
 		 * This is only safe to do when the partial state is known to be short, otherwise an
 		 * exception is thrown if the serialized_partial is not fully consumed by deserialfn().
 		 */
+		TS_DEBUG_LOG("zero-filling serialized numeric type");
 		return zero_fill_bytearray(serialized_partial, NUMERIC_PARTIAL_MISSING_LENGTH);
+	}
 
 	return serialized_partial;
 }
@@ -260,25 +265,56 @@ inner_agg_deserialize(FACombineFnMeta *combine_meta, bytea *volatile serialized_
 		deser_fcinfo->isnull = false;
 
 		/*
-		 * When an exception is thrown and longjmp() is called, CurrentMemoryContext is potentially
-		 * different than what it was inside the PG_TRY() block below.
+		 * When an exception is thrown and longjmp() is called,
+		 * CurrentMemoryContext is potentially different than what it was
+		 * inside the PG_TRY() block below.
+		 *
+		 * Restore it to the old value so that the code in the subsequent
+		 * PG_CATCH() block does not corrupt the memory.
 		 *
-		 * Restore it to the old value so that the code in the subsequent PG_CATCH() block does not
-		 * corrupt the memory.
+		 * No need for volatile variables since we don't modify any of this
+		 * function's stack frame inside the PG_TRY() block.
 		 *
-		 * No need for volatile variables since we don't modify any of this function's stack frame
-		 * inside the PG_TRY() block.
+		 * We create a subtransaction to run the functions below, in case they
+		 * abort.
 		 */
 		MemoryContext oldcontext = CurrentMemoryContext;
+		ResourceOwner oldowner = CurrentResourceOwner;
+		BeginInternalSubTransaction(NULL);
 		PG_TRY();
 		{
 			deserialized = FunctionCallInvoke(deser_fcinfo);
+			ReleaseCurrentSubTransaction();
+			MemoryContextSwitchTo(oldcontext);
+			CurrentResourceOwner = oldowner;
 		}
 		PG_CATCH();
 		{
-			CurrentMemoryContext = oldcontext;
+			const int sqlerrcode = geterrcode();
+
+			/*
+			 * If we get a data exception category error or a protocol
+			 * violation or an internal error (which is generated by default
+			 * when using elog()), the format is wrong and we can attempt to
+			 * repair it by switching to a different function for
+			 * deserializing.
+			 *
+			 * If the error is an insufficient resources category we should
+			 * *not* continue executing.
+			 *
+			 * Errors in other categories are not expected, but we cannot do
+			 * anything with them anyway, so just re-throw them too.
+			 */
+			if (ERRCODE_TO_CATEGORY(sqlerrcode) != ERRCODE_DATA_EXCEPTION &&
+				sqlerrcode != ERRCODE_PROTOCOL_VIOLATION && sqlerrcode != ERRCODE_INTERNAL_ERROR)
+			{
+				PG_RE_THROW();
+			}
 			FlushErrorState();
-			/* attempt to repair the serialized partial */
+			RollbackAndReleaseCurrentSubTransaction();
+			MemoryContextSwitchTo(oldcontext);
+			CurrentResourceOwner = oldowner;
+			TS_DEBUG_LOG("attempting repair of serialized partial");
 			serialized_partial =
 				sanitize_serialized_partial(combine_meta->deserialfnoid, serialized_partial);
 			FC_ARG(deser_fcinfo, 0) = PointerGetDatum(serialized_partial);

tsl/test/expected/partialize_finalize.out

@@ -166,6 +166,7 @@ as
  , _timescaledb_functions.partialize_agg(max(d))
  , _timescaledb_functions.partialize_agg(stddev(b))
  , _timescaledb_functions.partialize_agg(stddev(e)) from foo group by a, b ;
+create table saved_partials as select * from v1;
 create table t1 as select * from v1;
 --sum 2114, collid 0, min(text) 2145, collid 100, max(ts) 2127
 insert into foo values(12, 10, 'hello', '2010-01-02 09:00:00-05', 10);
@@ -1102,3 +1103,45 @@ FROM issue4922_partials_parallel;
  5001129 | 50.0112900000000000 |   0 | 100 | 100000
 (1 row)
 
+-- Test that a "soft" error causes an attempt to repair the row
+-- We don't care about the result here (but it is the sum of the b
+-- column where a = 1 in table "foo" above), just that it should
+-- attempt a repair, do the repair, and then not fail.
+--
+-- Do this by reading the partial numeric format and drop bytes from
+-- the end of it (16 bytes, meaning 32 characters). This is what we
+-- know how to repair.  Next, attempt to decode the partial and check
+-- that it attempt a repair and actually does the repair (zero-filling
+-- the end).
+SET client_min_messages TO debug2;
+WITH
+  sample AS (SELECT encode(partialb, 'hex') part FROM saved_partials WHERE a = 1),
+  broken AS (SELECT decode(left(part, -32), 'hex') partialb FROM sample)
+SELECT _timescaledb_functions.finalize_agg('sum(numeric)', null, null, null, broken.partialb, null::numeric) sumb
+  FROM broken;
+LOG:  statement: WITH
+  sample AS (SELECT encode(partialb, 'hex') part FROM saved_partials WHERE a = 1),
+  broken AS (SELECT decode(left(part, -32), 'hex') partialb FROM sample)
+SELECT _timescaledb_functions.finalize_agg('sum(numeric)', null, null, null, broken.partialb, null::numeric) sumb
+  FROM broken;
+DEBUG:  capture_ExecutorStart - rtable #0: (null)
+DEBUG:  capture_ExecutorStart - rtable #1: (null)
+DEBUG:  capture_ExecutorStart - rtable #2: saved_partials
+DEBUG:  inner_agg_deserialize - attempting repair of serialized partial
+DEBUG:  sanitize_serialized_partial - zero-filling serialized numeric type
+DEBUG:  inner_agg_deserialize - attempting repair of serialized partial
+DEBUG:  sanitize_serialized_partial - zero-filling serialized numeric type
+DEBUG:  inner_agg_deserialize - attempting repair of serialized partial
+DEBUG:  sanitize_serialized_partial - zero-filling serialized numeric type
+DEBUG:  inner_agg_deserialize - attempting repair of serialized partial
+DEBUG:  sanitize_serialized_partial - zero-filling serialized numeric type
+DEBUG:  inner_agg_deserialize - attempting repair of serialized partial
+DEBUG:  sanitize_serialized_partial - zero-filling serialized numeric type
+ sumb 
+------
+  150
+(1 row)
+
+RESET client_min_messages;
+LOG:  statement: RESET client_min_messages;
+DROP TABLE saved_partials;

tsl/test/sql/CMakeLists.txt

@@ -67,7 +67,6 @@ set(TEST_FILES
     merge_compress.sql
     modify_exclusion.sql
     move.sql
-    partialize_finalize.sql
     policy_generalization.sql
     reorder.sql
     size_utils_tsl.sql
@@ -149,6 +148,7 @@ if(CMAKE_BUILD_TYPE MATCHES Debug)
     tsl_tables.sql
     license_tsl.sql
     fixed_schedules.sql
+    partialize_finalize.sql
     recompress_chunk_segmentwise.sql
     feature_flags.sql
     vector_agg_default.sql

tsl/test/sql/partialize_finalize.sql

@@ -137,6 +137,7 @@ as
  , _timescaledb_functions.partialize_agg(stddev(b))
  , _timescaledb_functions.partialize_agg(stddev(e)) from foo group by a, b ;
 
+create table saved_partials as select * from v1;
 create table t1 as select * from v1;
 
 --sum 2114, collid 0, min(text) 2145, collid 100, max(ts) 2127
@@ -370,3 +371,23 @@ SELECT
   _timescaledb_functions.finalize_agg('pg_catalog.max(integer)'::text, NULL::name, NULL::name, '{{pg_catalog,int4}}'::name[], partial_max, NULL::integer) AS max,
   _timescaledb_functions.finalize_agg('pg_catalog.count()'::text, NULL::name, NULL::name, '{}'::name[], partial_count, NULL::bigint) AS count
 FROM issue4922_partials_parallel;
+
+-- Test that a "soft" error causes an attempt to repair the row
+
+-- We don't care about the result here (but it is the sum of the b
+-- column where a = 1 in table "foo" above), just that it should
+-- attempt a repair, do the repair, and then not fail.
+--
+-- Do this by reading the partial numeric format and drop bytes from
+-- the end of it (16 bytes, meaning 32 characters). This is what we
+-- know how to repair.  Next, attempt to decode the partial and check
+-- that it attempt a repair and actually does the repair (zero-filling
+-- the end).
+SET client_min_messages TO debug2;
+WITH
+  sample AS (SELECT encode(partialb, 'hex') part FROM saved_partials WHERE a = 1),
+  broken AS (SELECT decode(left(part, -32), 'hex') partialb FROM sample)
+SELECT _timescaledb_functions.finalize_agg('sum(numeric)', null, null, null, broken.partialb, null::numeric) sumb
+  FROM broken;
+RESET client_min_messages;
+DROP TABLE saved_partials;