Add support for ON CONFLICT DO UPDATE for compressed hypertables

This patch fixes execution of INSERT with ON CONFLICT DO UPDATE by
removing error and allowing UPDATE do happen on the given compressed
hypertable.

CHANGELOG.md

@@ -12,6 +12,7 @@ accidentally triggering the load of a previous DB version.**
 * #5361 Add parallel support for partialize_agg()
 * #5252 Improve unique constraint support on compressed hypertables
 * #5312 Add timeout support to ping_data_node()
+* #5454 Add support for ON CONFLICT DO UPDATE for compressed hypertables
 
 **Bugfixes**
 * #5396 Fix SEGMENTBY columns predicates to be pushed down

src/nodes/chunk_dispatch/chunk_dispatch.c

@@ -4,6 +4,7 @@
  * LICENSE-APACHE for a copy of the license.
  */
 #include <postgres.h>
+#include <access/xact.h>
 #include <nodes/nodes.h>
 #include <nodes/extensible.h>
 #include <nodes/makefuncs.h>
@@ -145,7 +146,14 @@ ts_chunk_dispatch_get_chunk_insert_state(ChunkDispatch *dispatch, Point *point,
 			 * postgres can do proper constraint checking.
 			 */
 			if (ts_cm_functions->decompress_batches_for_insert)
+			{
 				ts_cm_functions->decompress_batches_for_insert(cis, chunk, slot);
+				OnConflictAction onconflict_action =
+					chunk_dispatch_get_on_conflict_action(dispatch);
+				/* mark rows visible */
+				if (onconflict_action == ONCONFLICT_UPDATE)
+					dispatch->estate->es_output_cid = GetCurrentCommandId(true);
+			}
 			else
 				ereport(ERROR,
 						(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),

src/nodes/chunk_dispatch/chunk_insert_state.c

@@ -83,7 +83,7 @@ chunk_dispatch_get_returning_clauses(const ChunkDispatch *dispatch)
 #endif
 }
 
-static OnConflictAction
+OnConflictAction
 chunk_dispatch_get_on_conflict_action(const ChunkDispatch *dispatch)
 {
 	if (!dispatch->dispatch_state)
@@ -583,7 +583,6 @@ ts_chunk_insert_state_create(const Chunk *chunk, ChunkDispatch *dispatch)
 													  ALLOCSET_DEFAULT_SIZES);
 	OnConflictAction onconflict_action = chunk_dispatch_get_on_conflict_action(dispatch);
 	ResultRelInfo *relinfo;
-	bool has_compressed_chunk = (chunk->fd.compressed_chunk_id != 0);
 
 	/* permissions NOT checked here; were checked at hypertable level */
 	if (check_enable_rls(chunk->table_id, InvalidOid, false) == RLS_ENABLED)
@@ -597,12 +596,6 @@ ts_chunk_insert_state_create(const Chunk *chunk, ChunkDispatch *dispatch)
 												 CHUNK_INSERT,
 												 true);
 
-	if (has_compressed_chunk && onconflict_action == ONCONFLICT_UPDATE)
-		ereport(ERROR,
-				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-				 errmsg(
-					 "INSERT with ON CONFLICT DO UPDATE is not supported on compressed chunks")));
-
 	rel = table_open(chunk->table_id, RowExclusiveLock);
 
 	MemoryContext old_mcxt = MemoryContextSwitchTo(cis_context);

src/nodes/chunk_dispatch/chunk_insert_state.h

@@ -67,4 +67,5 @@ typedef struct ChunkDispatch ChunkDispatch;
 extern ChunkInsertState *ts_chunk_insert_state_create(const Chunk *chunk, ChunkDispatch *dispatch);
 extern void ts_chunk_insert_state_destroy(ChunkInsertState *state);
 
+OnConflictAction chunk_dispatch_get_on_conflict_action(const ChunkDispatch *dispatch);
 #endif /* TIMESCALEDB_CHUNK_INSERT_STATE_H */

tsl/test/expected/compression.out

@@ -208,9 +208,26 @@ ERROR:  cannot update/delete rows from chunk "_hyper_1_1_chunk" as it is compres
 insert into foo values(10, 12, 12, 12)
 on conflict( a, b)
 do update set b = excluded.b;
-ERROR:  INSERT with ON CONFLICT DO UPDATE is not supported on compressed chunks
+SELECT * from foo ORDER BY a,b;
+ a  |  b  | c  | d  
+----+-----+----+----
+ 10 |  12 | 12 | 12
+ 20 |  11 | 20 |   
+ 30 | 222 | 40 |   
+(3 rows)
+
 --TEST2c Do DML directly on the chunk.
-insert into _timescaledb_internal._hyper_1_2_chunk values(10, 12, 12, 12);
+insert into _timescaledb_internal._hyper_1_2_chunk values(10, 12, 12, 12)
+on conflict( a, b)
+do update set b = excluded.b + 12;
+SELECT * from foo ORDER BY a,b;
+ a  |  b  | c  | d  
+----+-----+----+----
+ 10 |  24 | 12 | 12
+ 20 |  11 | 20 |   
+ 30 | 222 | 40 |   
+(3 rows)
+
 update _timescaledb_internal._hyper_1_2_chunk
 set b = 12;
 ERROR:  cannot update/delete rows from chunk "_hyper_1_2_chunk" as it is compressed
@@ -229,7 +246,7 @@ ERROR:  duplicate key value violates unique constraint "_hyper_1_2_chunk_foo_uni
 select * from _timescaledb_internal._hyper_1_2_chunk order by a;
  a  | b  | c  |  d  
 ----+----+----+-----
- 10 | 12 | 12 |  12
+ 10 | 24 | 12 |  12
  10 | 10 | 20 |    
  11 | 10 | 20 | 120
 (3 rows)

tsl/test/expected/compression_conflicts.out

@@ -274,3 +274,117 @@ SELECT count(*) FROM ONLY :CHUNK;
      1
 (1 row)
 
+CREATE OR REPLACE VIEW compressed_chunk_info_view AS
+SELECT
+   h.schema_name AS hypertable_schema,
+   h.table_name AS hypertable_name,
+   c.schema_name as chunk_schema,
+   c.table_name as chunk_name,
+   c.status as chunk_status,
+   comp.schema_name as compressed_chunk_schema,
+   comp.table_name as compressed_chunk_name
+FROM
+   _timescaledb_catalog.hypertable h JOIN
+  _timescaledb_catalog.chunk c ON h.id = c.hypertable_id
+   LEFT JOIN _timescaledb_catalog.chunk comp
+ON comp.id = c.compressed_chunk_id;
+CREATE TABLE compressed_ht (
+       time TIMESTAMP WITH TIME ZONE NOT NULL,
+       sensor_id INTEGER NOT NULL,
+       cpu double precision null,
+       temperature double precision null,
+       name varchar(100) default 'this is a default string value'
+);
+CREATE UNIQUE INDEX sensor_id_time_idx on compressed_ht(time, sensor_id);
+SELECT * FROM create_hypertable('compressed_ht', 'time',
+       chunk_time_interval => INTERVAL '2 months');
+WARNING:  column type "character varying" used for "name" does not follow best practices
+ hypertable_id | schema_name |  table_name   | created 
+---------------+-------------+---------------+---------
+             7 | public      | compressed_ht | t
+(1 row)
+
+-- create chunk 1
+INSERT INTO compressed_ht VALUES ('2017-12-28 01:10:28.192199+05:30', '1', 0.876, 4.123, 'chunk 1');
+INSERT INTO compressed_ht VALUES ('2017-12-24 01:10:28.192199+05:30', '1', 0.876, 4.123, 'chunk 1');
+-- create chunk 2
+INSERT INTO compressed_ht VALUES ('2017-03-28 01:10:28.192199+05:30', '2', 0.876, 4.123, 'chunk 2');
+INSERT INTO compressed_ht VALUES ('2017-03-12 01:10:28.192199+05:30', '3', 0.876, 4.123, 'chunk 2');
+-- create chunk 3
+INSERT INTO compressed_ht VALUES ('2022-01-18 01:10:28.192199+05:30', '4', 0.876, 4.123, 'chunk 3');
+INSERT INTO compressed_ht VALUES ('2022-01-08 01:10:28.192199+05:30', '4', 0.876, 4.123, 'chunk 3');
+INSERT INTO compressed_ht VALUES ('2022-01-11 01:10:28.192199+05:30', '5', 0.876, 4.123, 'chunk 3');
+INSERT INTO compressed_ht VALUES ('2022-01-24 01:10:28.192199+05:30', '6', 0.876, 4.123, 'chunk 3');
+ALTER TABLE compressed_ht SET (
+	timescaledb.compress,
+	timescaledb.compress_segmentby = 'sensor_id'
+);
+SELECT COMPRESS_CHUNK(SHOW_CHUNKS('compressed_ht'));
+             compress_chunk             
+----------------------------------------
+ _timescaledb_internal._hyper_7_7_chunk
+ _timescaledb_internal._hyper_7_8_chunk
+ _timescaledb_internal._hyper_7_9_chunk
+(3 rows)
+
+-- check compression status
+SELECT chunk_status,
+       chunk_name as "CHUNK_NAME"
+FROM compressed_chunk_info_view
+WHERE hypertable_name = 'compressed_ht' ORDER BY chunk_name;
+ chunk_status |    CHUNK_NAME    
+--------------+------------------
+            1 | _hyper_7_7_chunk
+            1 | _hyper_7_8_chunk
+            1 | _hyper_7_9_chunk
+(3 rows)
+
+-- should report 0 row
+SELECT COUNT(*) FROM compressed_ht WHERE name = 'ON CONFLICT DO UPDATE';
+ count 
+-------
+     0
+(1 row)
+
+INSERT INTO compressed_ht VALUES ('2017-12-28 01:10:28.192199+05:30', '1', 0.876, 4.123, 'new insert row')
+  ON conflict(sensor_id, time)
+DO UPDATE SET sensor_id = excluded.sensor_id , name = 'ON CONFLICT DO UPDATE';
+-- should report 1 row
+SELECT COUNT(*) FROM compressed_ht WHERE name = 'ON CONFLICT DO UPDATE';
+ count 
+-------
+     1
+(1 row)
+
+-- check that chunk 1 compression status is set to partial
+SELECT chunk_status,
+       chunk_name as "CHUNK_NAME"
+FROM compressed_chunk_info_view
+WHERE hypertable_name = 'compressed_ht' ORDER BY chunk_name;
+ chunk_status |    CHUNK_NAME    
+--------------+------------------
+            9 | _hyper_7_7_chunk
+            1 | _hyper_7_8_chunk
+            1 | _hyper_7_9_chunk
+(3 rows)
+
+INSERT INTO compressed_ht VALUES ('2022-01-24 01:10:28.192199+05:30', '6', 0.876, 4.123, 'new insert row')
+  ON conflict(sensor_id, time)
+DO UPDATE SET sensor_id = excluded.sensor_id , name = 'ON CONFLICT DO UPDATE' RETURNING *;
+                time                 | sensor_id |  cpu  | temperature |         name          
+-------------------------------------+-----------+-------+-------------+-----------------------
+ Sun Jan 23 11:40:28.192199 2022 PST |         6 | 0.876 |       4.123 | ON CONFLICT DO UPDATE
+(1 row)
+
+-- check that chunks 1 and 3 compression status is set to partial
+SELECT chunk_status,
+       chunk_name as "CHUNK_NAME"
+FROM compressed_chunk_info_view
+WHERE hypertable_name = 'compressed_ht' ORDER BY chunk_name;
+ chunk_status |    CHUNK_NAME    
+--------------+------------------
+            9 | _hyper_7_7_chunk
+            1 | _hyper_7_8_chunk
+            9 | _hyper_7_9_chunk
+(3 rows)
+

tsl/test/sql/compression.sql

@@ -85,9 +85,14 @@ set b = (select f1.newval from foo_join f1 left join lateral (select newval as n
 insert into foo values(10, 12, 12, 12)
 on conflict( a, b)
 do update set b = excluded.b;
+SELECT * from foo ORDER BY a,b;
 
 --TEST2c Do DML directly on the chunk.
-insert into _timescaledb_internal._hyper_1_2_chunk values(10, 12, 12, 12);
+insert into _timescaledb_internal._hyper_1_2_chunk values(10, 12, 12, 12)
+on conflict( a, b)
+do update set b = excluded.b + 12;
+SELECT * from foo ORDER BY a,b;
+
 update _timescaledb_internal._hyper_1_2_chunk
 set b = 12;
 delete from _timescaledb_internal._hyper_1_2_chunk;

tsl/test/sql/compression_conflicts.sql

@@ -210,3 +210,82 @@ INSERT INTO comp_conflicts_3 VALUES ('2020-01-01','d1',0.1) ON CONFLICT DO NOTHI
 -- data should have move into uncompressed chunk for conflict check
 SELECT count(*) FROM ONLY :CHUNK;
 
+CREATE OR REPLACE VIEW compressed_chunk_info_view AS
+SELECT
+   h.schema_name AS hypertable_schema,
+   h.table_name AS hypertable_name,
+   c.schema_name as chunk_schema,
+   c.table_name as chunk_name,
+   c.status as chunk_status,
+   comp.schema_name as compressed_chunk_schema,
+   comp.table_name as compressed_chunk_name
+FROM
+   _timescaledb_catalog.hypertable h JOIN
+  _timescaledb_catalog.chunk c ON h.id = c.hypertable_id
+   LEFT JOIN _timescaledb_catalog.chunk comp
+ON comp.id = c.compressed_chunk_id;
+
+CREATE TABLE compressed_ht (
+       time TIMESTAMP WITH TIME ZONE NOT NULL,
+       sensor_id INTEGER NOT NULL,
+       cpu double precision null,
+       temperature double precision null,
+       name varchar(100) default 'this is a default string value'
+);
+CREATE UNIQUE INDEX sensor_id_time_idx on compressed_ht(time, sensor_id);
+
+SELECT * FROM create_hypertable('compressed_ht', 'time',
+       chunk_time_interval => INTERVAL '2 months');
+
+-- create chunk 1
+INSERT INTO compressed_ht VALUES ('2017-12-28 01:10:28.192199+05:30', '1', 0.876, 4.123, 'chunk 1');
+INSERT INTO compressed_ht VALUES ('2017-12-24 01:10:28.192199+05:30', '1', 0.876, 4.123, 'chunk 1');
+
+-- create chunk 2
+INSERT INTO compressed_ht VALUES ('2017-03-28 01:10:28.192199+05:30', '2', 0.876, 4.123, 'chunk 2');
+INSERT INTO compressed_ht VALUES ('2017-03-12 01:10:28.192199+05:30', '3', 0.876, 4.123, 'chunk 2');
+
+-- create chunk 3
+INSERT INTO compressed_ht VALUES ('2022-01-18 01:10:28.192199+05:30', '4', 0.876, 4.123, 'chunk 3');
+INSERT INTO compressed_ht VALUES ('2022-01-08 01:10:28.192199+05:30', '4', 0.876, 4.123, 'chunk 3');
+INSERT INTO compressed_ht VALUES ('2022-01-11 01:10:28.192199+05:30', '5', 0.876, 4.123, 'chunk 3');
+INSERT INTO compressed_ht VALUES ('2022-01-24 01:10:28.192199+05:30', '6', 0.876, 4.123, 'chunk 3');
+
+ALTER TABLE compressed_ht SET (
+	timescaledb.compress,
+	timescaledb.compress_segmentby = 'sensor_id'
+);
+
+SELECT COMPRESS_CHUNK(SHOW_CHUNKS('compressed_ht'));
+
+-- check compression status
+SELECT chunk_status,
+       chunk_name as "CHUNK_NAME"
+FROM compressed_chunk_info_view
+WHERE hypertable_name = 'compressed_ht' ORDER BY chunk_name;
+
+-- should report 0 row
+SELECT COUNT(*) FROM compressed_ht WHERE name = 'ON CONFLICT DO UPDATE';
+
+INSERT INTO compressed_ht VALUES ('2017-12-28 01:10:28.192199+05:30', '1', 0.876, 4.123, 'new insert row')
+  ON conflict(sensor_id, time)
+DO UPDATE SET sensor_id = excluded.sensor_id , name = 'ON CONFLICT DO UPDATE';
+
+-- should report 1 row
+SELECT COUNT(*) FROM compressed_ht WHERE name = 'ON CONFLICT DO UPDATE';
+
+-- check that chunk 1 compression status is set to partial
+SELECT chunk_status,
+       chunk_name as "CHUNK_NAME"
+FROM compressed_chunk_info_view
+WHERE hypertable_name = 'compressed_ht' ORDER BY chunk_name;
+
+INSERT INTO compressed_ht VALUES ('2022-01-24 01:10:28.192199+05:30', '6', 0.876, 4.123, 'new insert row')
+  ON conflict(sensor_id, time)
+DO UPDATE SET sensor_id = excluded.sensor_id , name = 'ON CONFLICT DO UPDATE' RETURNING *;
+
+-- check that chunks 1 and 3 compression status is set to partial
+SELECT chunk_status,
+       chunk_name as "CHUNK_NAME"
+FROM compressed_chunk_info_view
+WHERE hypertable_name = 'compressed_ht' ORDER BY chunk_name;