Fix ON CONFLICT DO UPDATE on 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 Fix ON CONFLICT DO UPDATE on 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(false);
+			}
 			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/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;