Add custom toaster for compression

Compressed rows are often large and get pushed to the toast relation in
~2KB chunks. Core writes each chunk with its own heap_insert() and
index_insert() call. This change adds a custom toasting mechanism using
relevant core function forks: chunks of a value are batched into a
single heap_multi_insert() call instead, cutting the WAL and
buffer-lock overhead.

The forked functions live under tsl/src/import/, the existing home for
code copied from PostgreSQL core, since compression_toast.c is exactly
that.

The new path is gated behind timescaledb.use_custom_toaster, off by
default. Also add a TAP test confirming the WAL this change
generates is correct.

Author: antekresic

.unreleased/pr_10388

@@ -0,0 +1 @@
+Implements: #10388 Add custom toaster for compression

src/guc.c

@@ -85,6 +85,7 @@ TSDLLEXPORT bool ts_guc_enable_direct_compress_auto_segmentby = true;
 int ts_guc_direct_compress_insert_tuple_sort_limit = 30000;
 TSDLLEXPORT int ts_guc_direct_compress_segmentby_min_rows = 5000;
 TSDLLEXPORT int ts_guc_direct_compress_segmentby_batch_size_limit = 500;
+TSDLLEXPORT bool ts_guc_use_custom_toaster = false;
 bool ts_guc_enable_deprecation_warnings = true;
 TSDLLEXPORT bool ts_guc_enable_optimizations = true;
 bool ts_guc_restoring = false;
@@ -631,6 +632,17 @@ _guc_init(void)
 							 NULL,
 							 NULL);
 
+	DefineCustomBoolVariable(MAKE_EXTOPTION("use_custom_toaster"),
+							 "Use a custom TOAST writer for compressed row inserts",
+							 "This setting is only used for compression.",
+							 &ts_guc_use_custom_toaster,
+							 false,
+							 PGC_USERSET,
+							 0,
+							 NULL,
+							 NULL,
+							 NULL);
+
 	DefineCustomIntVariable(MAKE_EXTOPTION("direct_compress_insert_tuple_sort_limit"),
 							"Number of tuples that can be sorted at once in an INSERT operation",
 							"This is mainly used to keep the memory footprint down for "

src/guc.h

@@ -53,6 +53,7 @@ extern TSDLLEXPORT bool ts_guc_enable_direct_compress_auto_segmentby;
 extern int ts_guc_direct_compress_insert_tuple_sort_limit;
 extern TSDLLEXPORT int ts_guc_direct_compress_segmentby_min_rows;
 extern TSDLLEXPORT int ts_guc_direct_compress_segmentby_batch_size_limit;
+extern TSDLLEXPORT bool ts_guc_use_custom_toaster;
 extern TSDLLEXPORT bool ts_guc_enable_compressed_direct_batch_delete;
 extern TSDLLEXPORT int ts_guc_max_tuples_decompressed_per_dml;
 extern TSDLLEXPORT bool ts_guc_enable_compression_wal_markers;

tsl/src/compression/compression.c

@@ -7,6 +7,7 @@
 #include <access/attmap.h>
 #include <access/attnum.h>
 #include <access/detoast.h>
+#include <access/heapam.h>
 #include <access/htup_details.h>
 #include <access/skey.h>
 #include <access/tupdesc.h>
@@ -48,6 +49,7 @@
 #include "debug_assert.h"
 #include "debug_point.h"
 #include "guc.h"
+#include "import/compression_toast.h"
 #include "nodes/modify_hypertable.h"
 #include "ts_catalog/array_utils.h"
 #include "ts_catalog/catalog.h"
@@ -1778,11 +1780,18 @@ row_compressor_flush(RowCompressor *row_compressor, BulkWriter *writer, bool cha
 	}
 
 	Assert(writer->bistate != NULL);
-	heap_insert(writer->out_rel,
-				compressed_tuple,
-				writer->mycid,
-				writer->insert_options /*=options*/,
-				writer->bistate);
+	if (ts_guc_use_custom_toaster)
+	{
+		compression_heap_insert(writer, compressed_tuple);
+	}
+	else
+	{
+		heap_insert(writer->out_rel,
+					compressed_tuple,
+					writer->mycid,
+					writer->insert_options,
+					writer->bistate);
+	}
 	if (writer->indexstate->ri_NumIndices > 0)
 	{
 		ts_catalog_index_insert(writer->indexstate, compressed_tuple);
@@ -2015,6 +2024,7 @@ bulk_writer_close(BulkWriter *writer)
 	{
 		CatalogCloseIndexes(writer->indexstate);
 	}
+	compression_toast_writer_close(writer);
 	FreeExecutorState(writer->estate);
 }
 

tsl/src/compression/compression.h

@@ -155,6 +155,16 @@ typedef struct BulkWriter
 	CommandId mycid;
 	BulkInsertState bistate;
 	int insert_options; /* heap insert options */
+	/*
+	 * Toast relation/indexes for compression_toast_save_datum_multi(),
+	 * opened lazily on first use and closed by
+	 * compression_toast_writer_close(). NULL/0 if never toasted.
+	 */
+	Relation toast_rel;
+	Relation *toast_indexes;
+	int num_toast_indexes;
+	int toast_valid_index;
+	BulkInsertState toast_bistate;
 } BulkWriter;
 
 typedef struct RowDecompressor

tsl/src/import/CMakeLists.txt

@@ -1,4 +1,4 @@
-set(SOURCES)
+set(SOURCES ${CMAKE_CURRENT_SOURCE_DIR}/compression_toast.c)
 
 if(USE_UMASH)
   list(APPEND SOURCES ${CMAKE_CURRENT_SOURCE_DIR}/umash.c)