Limit tuple decompression during DML operations

.unreleased/pr_6566

@@ -0,0 +1 @@
+Implements: #6566 Limit tuple decompression during DML operations

src/guc.c

@@ -63,6 +63,7 @@ bool ts_guc_enable_cagg_reorder_groupby = true;
 bool ts_guc_enable_now_constify = true;
 bool ts_guc_enable_osm_reads = true;
 TSDLLEXPORT bool ts_guc_enable_dml_decompression = true;
+TSDLLEXPORT int ts_guc_max_tuples_decompressed_per_dml = 100000;
 TSDLLEXPORT bool ts_guc_enable_transparent_decompression = true;
 TSDLLEXPORT bool ts_guc_enable_decompression_logrep_markers = false;
 TSDLLEXPORT bool ts_guc_enable_decompression_sorted_merge = true;
@@ -339,6 +340,23 @@ _guc_init(void)
 							 NULL,
 							 NULL);
 
+	DefineCustomIntVariable("timescaledb.max_tuples_decompressed_per_dml_transaction",
+							"The max number of tuples that can be decompressed during an "
+							"INSERT, UPDATE, or DELETE.",
+							" If the number of tuples exceeds this value, an error will "
+							"be thrown and transaction rolled back. "
+							"Setting this to 0 sets this value to unlimited number of "
+							"tuples decompressed.",
+							&ts_guc_max_tuples_decompressed_per_dml,
+							100000,
+							0,
+							2147483647,
+							PGC_USERSET,
+							0,
+							NULL,
+							NULL,
+							NULL);
+
 	DefineCustomBoolVariable("timescaledb.enable_transparent_decompression",
 							 "Enable transparent decompression",
 							 "Enable transparent decompression when querying hypertable",

src/guc.h

@@ -27,6 +27,7 @@ extern bool ts_guc_enable_cagg_reorder_groupby;
 extern bool ts_guc_enable_now_constify;
 extern bool ts_guc_enable_osm_reads;
 extern TSDLLEXPORT bool ts_guc_enable_dml_decompression;
+extern TSDLLEXPORT int ts_guc_max_tuples_decompressed_per_dml;
 extern TSDLLEXPORT bool ts_guc_enable_transparent_decompression;
 extern TSDLLEXPORT bool ts_guc_enable_decompression_logrep_markers;
 extern TSDLLEXPORT bool ts_guc_enable_decompression_sorted_merge;

src/nodes/chunk_dispatch/chunk_dispatch.c

@@ -433,6 +433,22 @@ chunk_dispatch_exec(CustomScanState *node)
 												   on_chunk_insert_state_changed,
 												   state);
 
+	if (ts_guc_max_tuples_decompressed_per_dml > 0)
+	{
+		if (cis->cds->tuples_decompressed > ts_guc_max_tuples_decompressed_per_dml)
+		{
+			ereport(ERROR,
+					(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
+					 errmsg("tuple decompression limit exceeded by operation"),
+					 errdetail("current limit: %d, tuples decompressed: %lld",
+							   ts_guc_max_tuples_decompressed_per_dml,
+							   (long long int) cis->cds->tuples_decompressed),
+					 errhint("Consider increasing "
+							 "timescaledb.max_tuples_decompressed_per_dml_transaction or set "
+							 "to 0 (unlimited).")));
+		}
+	}
+
 	/*
 	 * Set the result relation in the executor state to the target chunk.
 	 * This makes sure that the tuple gets inserted into the correct

src/nodes/hypertable_modify.c

@@ -761,6 +761,22 @@ ExecModifyTable(CustomScanState *cs_node, PlanState *pstate)
 			CommandCounterIncrement();
 			/* mark rows visible */
 			estate->es_output_cid = GetCurrentCommandId(true);
+
+			if (ts_guc_max_tuples_decompressed_per_dml > 0)
+			{
+				if (ht_state->tuples_decompressed > ts_guc_max_tuples_decompressed_per_dml)
+				{
+					ereport(ERROR,
+							(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
+							 errmsg("tuple decompression limit exceeded by operation"),
+							 errdetail("current limit: %d, tuples decompressed: %lld",
+									   ts_guc_max_tuples_decompressed_per_dml,
+									   (long long int) ht_state->tuples_decompressed),
+							 errhint("Consider increasing "
+									 "timescaledb.max_tuples_decompressed_per_dml_transaction or "
+									 "set to 0 (unlimited).")));
+				}
+			}
 		}
 	}
 	/*

tsl/test/expected/compression_insert.out

@@ -1042,3 +1042,40 @@ SELECT count(compress_chunk(ch)) FROM show_chunks('test_copy') ch;
 
 \copy test_copy FROM data/copy_data.csv WITH CSV HEADER;
 DROP TABLE test_copy;
+-- Text limitting decompressed tuple during an insert
+CREATE TABLE test_limit (
+    timestamp int not null,
+    id bigint
+);
+SELECT * FROM create_hypertable('test_limit', 'timestamp', chunk_time_interval=>1000);
+ hypertable_id | schema_name | table_name | created 
+---------------+-------------+------------+---------
+            24 | public      | test_limit | t
+(1 row)
+
+INSERT INTO test_limit SELECT t, i FROM generate_series(1,10000,1) t CROSS JOIN generate_series(1,3,1) i;
+CREATE UNIQUE INDEX timestamp_id_idx ON test_limit(timestamp, id);
+ALTER TABLE test_limit SET (
+    timescaledb.compress,
+    timescaledb.compress_orderby = 'timestamp'
+);
+WARNING:  column "id" should be used for segmenting or ordering
+SELECT count(compress_chunk(ch)) FROM show_chunks('test_limit') ch;
+ count 
+-------
+    11
+(1 row)
+
+SET timescaledb.max_tuples_decompressed_per_dml_transaction = 5000;
+\set VERBOSITY default
+\set ON_ERROR_STOP 0
+-- Inserting in the same period should decompress tuples
+INSERT INTO test_limit SELECT t, 11 FROM generate_series(1,6000,1000) t;
+ERROR:  tuple decompression limit exceeded by operation
+DETAIL:  current limit: 5000, tuples decompressed: 6000
+HINT:  Consider increasing timescaledb.max_tuples_decompressed_per_dml_transaction or set to 0 (unlimited).
+-- Setting to 0 should remove the limit.
+SET timescaledb.max_tuples_decompressed_per_dml_transaction = 0;
+INSERT INTO test_limit SELECT t, 11 FROM generate_series(1,6000,1000) t;
+\set ON_ERROR_STOP 1
+DROP TABLE test_limit;

tsl/test/expected/compression_update_delete.out

@@ -2660,3 +2660,43 @@ RESET timescaledb.debug_compression_path_info;
 DROP TABLE t6367;
 \c :TEST_DBNAME :ROLE_SUPERUSER
 DROP DATABASE test6367;
+-- Text limitting decompressed tuple during an UPDATE or DELETE
+CREATE TABLE test_limit (
+    timestamp int not null,
+    id bigint
+);
+SELECT * FROM create_hypertable('test_limit', 'timestamp', chunk_time_interval=>10000);
+ hypertable_id | schema_name | table_name | created 
+---------------+-------------+------------+---------
+            33 | public      | test_limit | t
+(1 row)
+
+INSERT INTO test_limit SELECT t, i FROM generate_series(1,10000,1) t CROSS JOIN generate_series(1,3,1) i;
+ALTER TABLE test_limit SET (
+    timescaledb.compress,
+    timescaledb.compress_orderby = 'timestamp'
+);
+SELECT count(compress_chunk(ch)) FROM show_chunks('test_limit') ch;
+ count 
+-------
+     2
+(1 row)
+
+SET timescaledb.max_tuples_decompressed_per_dml_transaction = 5000;
+\set VERBOSITY default
+\set ON_ERROR_STOP 0
+-- Updating or deleting everything will break the set limit.
+UPDATE test_limit SET id = 0;
+ERROR:  tuple decompression limit exceeded by operation
+DETAIL:  current limit: 5000, tuples decompressed: 30000
+HINT:  Consider increasing timescaledb.max_tuples_decompressed_per_dml_transaction or set to 0 (unlimited).
+DELETE FROM test_limit WHERE id > 0;
+ERROR:  tuple decompression limit exceeded by operation
+DETAIL:  current limit: 5000, tuples decompressed: 30000
+HINT:  Consider increasing timescaledb.max_tuples_decompressed_per_dml_transaction or set to 0 (unlimited).
+-- Setting to 0 should remove the limit.
+SET timescaledb.max_tuples_decompressed_per_dml_transaction = 0;
+UPDATE test_limit SET id = 0;
+DELETE FROM test_limit WHERE id > 0;
+\set ON_ERROR_STOP 1
+DROP TABLE test_limit;

tsl/test/sql/compression_insert.sql

@@ -688,3 +688,30 @@ SELECT count(compress_chunk(ch)) FROM show_chunks('test_copy') ch;
 \copy test_copy FROM data/copy_data.csv WITH CSV HEADER;
 
 DROP TABLE test_copy;
+
+-- Text limitting decompressed tuple during an insert
+CREATE TABLE test_limit (
+    timestamp int not null,
+    id bigint
+);
+SELECT * FROM create_hypertable('test_limit', 'timestamp', chunk_time_interval=>1000);
+INSERT INTO test_limit SELECT t, i FROM generate_series(1,10000,1) t CROSS JOIN generate_series(1,3,1) i;
+CREATE UNIQUE INDEX timestamp_id_idx ON test_limit(timestamp, id);
+
+ALTER TABLE test_limit SET (
+    timescaledb.compress,
+    timescaledb.compress_orderby = 'timestamp'
+);
+SELECT count(compress_chunk(ch)) FROM show_chunks('test_limit') ch;
+
+SET timescaledb.max_tuples_decompressed_per_dml_transaction = 5000;
+\set VERBOSITY default
+\set ON_ERROR_STOP 0
+-- Inserting in the same period should decompress tuples
+INSERT INTO test_limit SELECT t, 11 FROM generate_series(1,6000,1000) t;
+-- Setting to 0 should remove the limit.
+SET timescaledb.max_tuples_decompressed_per_dml_transaction = 0;
+INSERT INTO test_limit SELECT t, 11 FROM generate_series(1,6000,1000) t;
+\set ON_ERROR_STOP 1
+
+DROP TABLE test_limit;

tsl/test/sql/compression_update_delete.sql

@@ -1418,3 +1418,31 @@ RESET timescaledb.debug_compression_path_info;
 DROP TABLE t6367;
 \c :TEST_DBNAME :ROLE_SUPERUSER
 DROP DATABASE test6367;
+
+-- Text limitting decompressed tuple during an UPDATE or DELETE
+CREATE TABLE test_limit (
+    timestamp int not null,
+    id bigint
+);
+SELECT * FROM create_hypertable('test_limit', 'timestamp', chunk_time_interval=>10000);
+INSERT INTO test_limit SELECT t, i FROM generate_series(1,10000,1) t CROSS JOIN generate_series(1,3,1) i;
+
+ALTER TABLE test_limit SET (
+    timescaledb.compress,
+    timescaledb.compress_orderby = 'timestamp'
+);
+SELECT count(compress_chunk(ch)) FROM show_chunks('test_limit') ch;
+
+SET timescaledb.max_tuples_decompressed_per_dml_transaction = 5000;
+\set VERBOSITY default
+\set ON_ERROR_STOP 0
+-- Updating or deleting everything will break the set limit.
+UPDATE test_limit SET id = 0;
+DELETE FROM test_limit WHERE id > 0;
+-- Setting to 0 should remove the limit.
+SET timescaledb.max_tuples_decompressed_per_dml_transaction = 0;
+UPDATE test_limit SET id = 0;
+DELETE FROM test_limit WHERE id > 0;
+\set ON_ERROR_STOP 1
+
+DROP TABLE test_limit;