From ca76727164698209256028f7328414949c6cf908 Mon Sep 17 00:00:00 2001 From: Sven Klemm Date: Thu, 16 Jul 2026 22:00:26 +0200 Subject: [PATCH] Support ON CONFLICT DO SELECT on hypertables for PostgreSQL 19 PostgreSQL 19 added INSERT ... ON CONFLICT DO SELECT, which returns the conflicting row instead of updating it and can optionally lock it. Upstream changes: Add support for INSERT ... ON CONFLICT DO SELECT. https://github.com/postgres/postgres/commit/88327092ff0 --- .unreleased/pr_10279 | 1 + src/chunk_insert_state.c | 86 ++++--- src/nodes/modify_hypertable_exec.c | 301 +++++++++++++++++++++++++ tsl/test/expected/upsert_do_select.out | 207 +++++++++++++++++ tsl/test/sql/CMakeLists.txt | 2 +- tsl/test/sql/upsert_do_select.sql | 149 ++++++++++++ 6 files changed, 709 insertions(+), 37 deletions(-) create mode 100644 .unreleased/pr_10279 create mode 100644 tsl/test/expected/upsert_do_select.out create mode 100644 tsl/test/sql/upsert_do_select.sql diff --git a/.unreleased/pr_10279 b/.unreleased/pr_10279 new file mode 100644 index 00000000000..9d749f6a2f7 --- /dev/null +++ b/.unreleased/pr_10279 @@ -0,0 +1 @@ +Implements: #10279 Support ON CONFLICT DO SELECT with hypertables diff --git a/src/chunk_insert_state.c b/src/chunk_insert_state.c index 032d8e67d11..fc326b9de1f 100644 --- a/src/chunk_insert_state.c +++ b/src/chunk_insert_state.c @@ -230,6 +230,9 @@ setup_on_conflict_state(ResultRelInfo *ht_rri, ModifyTableState *mtstate, ChunkI Relation hyper_rel = ht_rri->ri_RelationDesc; ModifyTable *mt = castNode(ModifyTable, mtstate->ps.plan); + /* DO UPDATE has a SET clause; DO SELECT only fetches the conflicting row. */ + bool do_update = mt->onConflictAction == ONCONFLICT_UPDATE; + OnConflictActionState *onconfl = makeNode(OnConflictActionState); memcpy(onconfl, ht_rri->ri_onConflict, sizeof(OnConflictActionState)); chunk_rri->ri_onConflict = onconfl; @@ -237,7 +240,7 @@ setup_on_conflict_state(ResultRelInfo *ht_rri, ModifyTableState *mtstate, ChunkI chunk_rri->ri_RootToChildMap = map; chunk_rri->ri_RootToChildMapValid = true; - Assert(mt->onConflictSet); + Assert(!do_update || mt->onConflictSet); Assert(ht_rri->ri_onConflict != NULL); /* @@ -271,18 +274,6 @@ setup_on_conflict_state(ResultRelInfo *ht_rri, ModifyTableState *mtstate, ChunkI } else { - List *onconflset; - List *onconflcols; - - /* - * Translate expressions in onConflictSet to account for - * different attribute numbers. For that, map partition - * varattnos twice: first to catch the EXCLUDED - * pseudo-relation (INNER_VAR), and second to handle the main - * target relation (firstVarno). - */ - onconflset = copyObject(mt->onConflictSet); - Assert(map->outdesc == RelationGetDescr(chunk_rel)); if (!chunk_map) @@ -291,33 +282,52 @@ setup_on_conflict_state(ResultRelInfo *ht_rri, ModifyTableState *mtstate, ChunkI convert_tuples_by_name(RelationGetDescr(chunk_rel), RelationGetDescr(hyper_rel)); } - onconflset = translate_clause(onconflset, chunk_map, ht_rri->ri_RangeTableIndex, chunk_rel); - chunk_rri->ri_ChildToRootMap = chunk_map; chunk_rri->ri_ChildToRootMapValid = true; - /* Finally, adjust the target colnos to match the chunk. */ - if (chunk_map) + if (do_update) { - onconflcols = adjust_chunk_colnos(mt->onConflictCols, chunk_rri); + List *onconflset; + List *onconflcols; + + /* + * Translate expressions in onConflictSet to account for + * different attribute numbers. For that, map partition + * varattnos twice: first to catch the EXCLUDED + * pseudo-relation (INNER_VAR), and second to handle the main + * target relation (firstVarno). + */ + onconflset = copyObject(mt->onConflictSet); + onconflset = + translate_clause(onconflset, chunk_map, ht_rri->ri_RangeTableIndex, chunk_rel); + + /* + * Finally, adjust the target colnos to match the chunk. When the + * descriptors match (e.g. a direct chunk insert) there is no + * child-to-root map and the parent colnos apply unchanged. + */ + if (chunk_map) + { + onconflcols = adjust_chunk_colnos(mt->onConflictCols, chunk_rri); + } + else + { + onconflcols = mt->onConflictCols; + } + + /* create the tuple slot for the UPDATE SET projection */ + onconfl->oc_ProjSlot = table_slot_create(chunk_rel, NULL); + state->conflproj_slot = onconfl->oc_ProjSlot; + + /* build UPDATE SET projection state */ + onconfl->oc_ProjInfo = ExecBuildUpdateProjection(onconflset, + true, + onconflcols, + RelationGetDescr(chunk_rel), + mtstate->ps.ps_ExprContext, + onconfl->oc_ProjSlot, + &mtstate->ps); } - else - { - onconflcols = mt->onConflictCols; - } - - /* create the tuple slot for the UPDATE SET projection */ - onconfl->oc_ProjSlot = table_slot_create(chunk_rel, NULL); - state->conflproj_slot = onconfl->oc_ProjSlot; - - /* build UPDATE SET projection state */ - onconfl->oc_ProjInfo = ExecBuildUpdateProjection(onconflset, - true, - onconflcols, - RelationGetDescr(chunk_rel), - mtstate->ps.ps_ExprContext, - onconfl->oc_ProjSlot, - &mtstate->ps); Node *onconflict_where = mt->onConflictWhere; @@ -406,7 +416,11 @@ adjust_projections(ResultRelInfo *ht_rri, ModifyTableState *mtstate, ChunkInsert { set_arbiter_indexes(cis, ht_rri->ri_onConflictArbiterIndexes); - if (onConflictAction == ONCONFLICT_UPDATE) + if (onConflictAction == ONCONFLICT_UPDATE +#if PG19_GE + || onConflictAction == ONCONFLICT_SELECT +#endif + ) { setup_on_conflict_state(ht_rri, mtstate, cis, chunk_map); } diff --git a/src/nodes/modify_hypertable_exec.c b/src/nodes/modify_hypertable_exec.c index 9453bee8b2f..1d441002caf 100644 --- a/src/nodes/modify_hypertable_exec.c +++ b/src/nodes/modify_hypertable_exec.c @@ -165,6 +165,20 @@ static bool ExecOnConflictUpdate(ModifyTableContext *context, TupleTableSlot *excludedSlot, bool canSetTag, TupleTableSlot **returning); +#if PG19_GE +static bool ExecOnConflictLockRow(ModifyTableContext *context, + TupleTableSlot *existing, + ItemPointer conflictTid, + Relation relation, + LockTupleMode lockmode, + bool isUpdate); +static bool ExecOnConflictSelect(ModifyTableContext *context, + ResultRelInfo *resultRelInfo, + ItemPointer conflictTid, + TupleTableSlot *excludedSlot, + bool canSetTag, + TupleTableSlot **returning); +#endif static TupleTableSlot *ExecPrepareTupleRouting(ModifyTableState *mtstate, EState *estate, @@ -850,6 +864,31 @@ ExecInsert(ModifyTableContext *context, else goto vlock; } +#if PG19_GE + else if (onconflict == ONCONFLICT_SELECT) + { + /* + * In case of ON CONFLICT DO SELECT, optionally lock the + * conflicting tuple, fetch it and project RETURNING on it. + * Be prepared to retry if locking fails because of a + * concurrent UPDATE/DELETE to the conflict tuple. + */ + TupleTableSlot *returning = NULL; + + if (ExecOnConflictSelect(context, + resultRelInfo, + &conflictTid, + slot, + canSetTag, + &returning)) + { + InstrCountTuples2(&mtstate->ps, 1); + return returning; + } + else + goto vlock; + } +#endif else { /* @@ -2235,6 +2274,268 @@ ExecOnConflictUpdate(ModifyTableContext *context, return true; } +#if PG19_GE +/* + * ExecOnConflictLockRow --- lock the row for ON CONFLICT DO SELECT/UPDATE + * + * Try to lock tuple for update as part of speculative insertion for ON + * CONFLICT DO UPDATE or ON CONFLICT DO SELECT FOR UPDATE/SHARE. + * + * Returns true if the row is successfully locked, or false if the caller must + * retry the INSERT from scratch. + * + * copied verbatim from ExecOnConflictLockRow in executor/nodeModifyTable.c + */ +static bool +ExecOnConflictLockRow(ModifyTableContext *context, + TupleTableSlot *existing, + ItemPointer conflictTid, + Relation relation, + LockTupleMode lockmode, + bool isUpdate) +{ + TM_FailureData tmfd; + TM_Result test; + Datum xminDatum; + TransactionId xmin; + bool isnull; + + /* + * Lock tuple with lockmode. Don't follow updates when tuple cannot be + * locked without doing so. A row locking conflict here means our + * previous conclusion that the tuple is conclusively committed is not + * true anymore. + */ + test = table_tuple_lock(relation, conflictTid, + context->estate->es_snapshot, + existing, context->estate->es_output_cid, + lockmode, LockWaitBlock, 0, + &tmfd); + switch (test) + { + case TM_Ok: + /* success! */ + break; + + case TM_Invisible: + + /* + * This can occur when a just inserted tuple is updated again in + * the same command. E.g. because multiple rows with the same + * conflicting key values are inserted. + * + * This is somewhat similar to the ExecUpdate() TM_SelfModified + * case. We do not want to proceed because it would lead to the + * same row being updated a second time in some unspecified order, + * and in contrast to plain UPDATEs there's no historical behavior + * to break. + * + * It is the user's responsibility to prevent this situation from + * occurring. These problems are why the SQL standard similarly + * specifies that for SQL MERGE, an exception must be raised in + * the event of an attempt to update the same row twice. + */ + xminDatum = slot_getsysattr(existing, + MinTransactionIdAttributeNumber, + &isnull); + Assert(!isnull); + xmin = DatumGetTransactionId(xminDatum); + + if (TransactionIdIsCurrentTransactionId(xmin)) + ereport(ERROR, + (errcode(ERRCODE_CARDINALITY_VIOLATION), + /* translator: %s is a SQL command name */ + errmsg("%s command cannot affect row a second time", + isUpdate ? "ON CONFLICT DO UPDATE" : "ON CONFLICT DO SELECT"), + errhint("Ensure that no rows proposed for insertion within the same command have duplicate constrained values."))); + + /* This shouldn't happen */ + elog(ERROR, "attempted to lock invisible tuple"); + break; + + case TM_SelfModified: + + /* + * This state should never be reached. As a dirty snapshot is used + * to find conflicting tuples, speculative insertion wouldn't have + * seen this row to conflict with. + */ + elog(ERROR, "unexpected self-updated tuple"); + break; + + case TM_Updated: + if (IsolationUsesXactSnapshot()) + ereport(ERROR, + (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), + errmsg("could not serialize access due to concurrent update"))); + + /* + * Tell caller to try again from the very start. + * + * It does not make sense to use the usual EvalPlanQual() style + * loop here, as the new version of the row might not conflict + * anymore, or the conflicting tuple has actually been deleted. + */ + ExecClearTuple(existing); + return false; + + case TM_Deleted: + if (IsolationUsesXactSnapshot()) + ereport(ERROR, + (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), + errmsg("could not serialize access due to concurrent delete"))); + + /* see TM_Updated case */ + ExecClearTuple(existing); + return false; + + default: + elog(ERROR, "unrecognized table_tuple_lock status: %u", test); + } + + /* Success, the tuple is locked. */ + return true; +} + +/* + * ExecOnConflictSelect --- execute SELECT of INSERT ON CONFLICT DO SELECT + * + * If SELECT FOR UPDATE/SHARE is specified, try to lock tuple as part of + * speculative insertion. If a qual originating from ON CONFLICT DO SELECT is + * satisfied, select (but still lock row, even though it may not satisfy + * estate's snapshot). + * + * Returns true if we're done (with or without a select), or false if the + * caller must retry the INSERT from scratch. + * + * copied and modified version of ExecOnConflictSelect from + * executor/nodeModifyTable.c + */ +static bool +ExecOnConflictSelect(ModifyTableContext *context, + ResultRelInfo *resultRelInfo, + ItemPointer conflictTid, + TupleTableSlot *excludedSlot, + bool canSetTag, + TupleTableSlot **returning) +{ + ModifyTableState *mtstate = context->mtstate; + ExprContext *econtext = mtstate->ps.ps_ExprContext; + Relation relation = resultRelInfo->ri_RelationDesc; + ExprState *onConflictSelectWhere = resultRelInfo->ri_onConflict->oc_WhereClause; + TupleTableSlot *existing = resultRelInfo->ri_onConflict->oc_Existing; + LockClauseStrength lockStrength = resultRelInfo->ri_onConflict->oc_LockStrength; + + /* + * Parse analysis should have blocked ON CONFLICT for all system + * relations, which includes these. There's no fundamental obstacle to + * supporting this; we'd just need to handle LOCKTAG_TUPLE appropriately. + */ + Assert(!resultRelInfo->ri_needLockTagTuple); + + /* Fetch/lock existing tuple, according to the requested lock strength */ + if (lockStrength == LCS_NONE) + { + if (!table_tuple_fetch_row_version(relation, + conflictTid, + SnapshotAny, + existing)) + elog(ERROR, "failed to fetch conflicting tuple for ON CONFLICT"); + } + else + { + LockTupleMode lockmode; + + switch (lockStrength) + { + case LCS_FORKEYSHARE: + lockmode = LockTupleKeyShare; + break; + case LCS_FORSHARE: + lockmode = LockTupleShare; + break; + case LCS_FORNOKEYUPDATE: + lockmode = LockTupleNoKeyExclusive; + break; + case LCS_FORUPDATE: + lockmode = LockTupleExclusive; + break; + default: + elog(ERROR, "Unexpected lock strength %d", (int) lockStrength); + } + + if (!ExecOnConflictLockRow(context, existing, conflictTid, + resultRelInfo->ri_RelationDesc, lockmode, false)) + return false; + } + + /* + * Verify that the tuple is visible to our MVCC snapshot if the current + * isolation level mandates that. See comments in ExecOnConflictUpdate(). + */ + ExecCheckTupleVisible(context->estate, relation, existing); + + /* + * Make tuple and any needed join variables available to ExecQual. The + * EXCLUDED tuple is installed in ecxt_innertuple, while the target's + * existing tuple is installed in the scantuple. EXCLUDED has been made + * to reference INNER_VAR in setrefs.c, but there is no other redirection. + */ + econtext->ecxt_scantuple = existing; + econtext->ecxt_innertuple = excludedSlot; + econtext->ecxt_outertuple = NULL; + + if (!ExecQual(onConflictSelectWhere, econtext)) + { + ExecClearTuple(existing); /* see return below */ + InstrCountFiltered1(&mtstate->ps, 1); + return true; /* done with the tuple */ + } + + if (resultRelInfo->ri_WithCheckOptions != NIL) + { + /* + * Check target's existing tuple against SELECT-applicable USING + * security barrier quals (if any), enforced here as RLS checks/WCOs. + * + * The rewriter creates WCOs from the USING quals of SELECT policies, + * and stores them as WCOs of "kind" WCO_RLS_CONFLICT_CHECK. If FOR + * UPDATE/SHARE was specified, UPDATE permissions are required on the + * target table, and the rewriter also adds WCOs built from the USING + * quals of UPDATE policies, using WCOs of the same kind, and this + * check enforces them too. + */ + ExecWithCheckOptions(WCO_RLS_CONFLICT_CHECK, resultRelInfo, + existing, + mtstate->ps.state); + } + + /* RETURNING is required for DO SELECT */ + Assert(resultRelInfo->ri_projectReturning); + + /* uses TimescaleDB's ExecProcessReturning signature */ + *returning = + ExecProcessReturning(resultRelInfo, CMD_INSERT, existing, existing, context->planSlot); + + if (canSetTag) + context->estate->es_processed++; + + /* + * Before releasing the existing tuple, make sure that the returning slot + * has a local copy of any pass-by-reference values. + */ + ExecMaterializeSlot(*returning); + + /* + * Clear out existing tuple, as there might not be another conflict among + * the next input rows. Don't want to hold resources till the end of the + * query. + */ + ExecClearTuple(existing); + + return true; +} +#endif static void fireASTriggers(ModifyTableState *node); static void fireBSTriggers(ModifyTableState *node); diff --git a/tsl/test/expected/upsert_do_select.out b/tsl/test/expected/upsert_do_select.out new file mode 100644 index 00000000000..1a48131c7ba --- /dev/null +++ b/tsl/test/expected/upsert_do_select.out @@ -0,0 +1,207 @@ +-- This file and its contents are licensed under the Timescale License. +-- Please see the included NOTICE for copyright information and +-- LICENSE-TIMESCALE for a copy of the license. +-- Test INSERT ... ON CONFLICT DO SELECT on hypertables (PG19+ feature) +CREATE TABLE ht_ocs( + time timestamptz NOT NULL, + device int NOT NULL, + value float, + label text, + PRIMARY KEY (time) +) WITH (tsdb.hypertable, tsdb.partition_column = 'time', tsdb.chunk_interval = '1 day'); +INSERT INTO ht_ocs VALUES ('2024-01-01 01:00', 1, 10.0, 'a'); +INSERT INTO ht_ocs VALUES ('2024-01-02 01:00', 2, 20.0, 'b'); +-- Get-or-create: conflicting row is returned unchanged, no new tuple written +INSERT INTO ht_ocs VALUES ('2024-01-01 01:00', 99, 99.0, 'x') +ON CONFLICT (time) DO SELECT RETURNING *; + time | device | value | label +------------------------------+--------+-------+------- + Mon Jan 01 01:00:00 2024 PST | 1 | 10 | a + +-- No conflict: the freshly inserted row is returned +INSERT INTO ht_ocs VALUES ('2024-01-03 01:00', 3, 30.0, 'c') +ON CONFLICT (time) DO SELECT RETURNING *; + time | device | value | label +------------------------------+--------+-------+------- + Wed Jan 03 01:00:00 2024 PST | 3 | 30 | c + +-- Multi-value insert mixing new rows and conflicts, spanning several chunks +INSERT INTO ht_ocs VALUES + ('2024-01-01 01:00', 1, 11.0, 'a2'), + ('2024-01-04 01:00', 4, 40.0, 'd'), + ('2024-01-02 01:00', 2, 22.0, 'b2') +ON CONFLICT (time) DO SELECT RETURNING time, device, value; + time | device | value +------------------------------+--------+------- + Mon Jan 01 01:00:00 2024 PST | 1 | 10 + Thu Jan 04 01:00:00 2024 PST | 4 | 40 + Tue Jan 02 01:00:00 2024 PST | 2 | 20 + +SELECT * FROM ht_ocs ORDER BY time; + time | device | value | label +------------------------------+--------+-------+------- + Mon Jan 01 01:00:00 2024 PST | 1 | 10 | a + Tue Jan 02 01:00:00 2024 PST | 2 | 20 | b + Wed Jan 03 01:00:00 2024 PST | 3 | 30 | c + Thu Jan 04 01:00:00 2024 PST | 4 | 40 | d + +-- Projected RETURNING list +INSERT INTO ht_ocs VALUES ('2024-01-01 01:00', 1, 0, 'z') +ON CONFLICT (time) DO SELECT RETURNING device, value * 2 AS double_value; + device | double_value +--------+-------------- + 1 | 20 + +-- WHERE clause: returns the row only when the qual matches +INSERT INTO ht_ocs VALUES ('2024-01-01 01:00', 1, 0, 'z') +ON CONFLICT (time) DO SELECT WHERE ht_ocs.device = 1 RETURNING device; + device +-------- + 1 + +INSERT INTO ht_ocs VALUES ('2024-01-01 01:00', 1, 0, 'z') +ON CONFLICT (time) DO SELECT WHERE ht_ocs.device = 12345 RETURNING device; + device +-------- + +-- Lock the conflicting row FOR UPDATE / FOR SHARE +INSERT INTO ht_ocs VALUES ('2024-01-02 01:00', 2, 0, 'z') +ON CONFLICT (time) DO SELECT FOR UPDATE RETURNING time, device; + time | device +------------------------------+-------- + Tue Jan 02 01:00:00 2024 PST | 2 + +INSERT INTO ht_ocs VALUES ('2024-01-02 01:00', 2, 0, 'z') +ON CONFLICT (time) DO SELECT FOR SHARE RETURNING time, device; + time | device +------------------------------+-------- + Tue Jan 02 01:00:00 2024 PST | 2 + +-- RETURNING is mandatory for DO SELECT +\set ON_ERROR_STOP 0 +INSERT INTO ht_ocs VALUES ('2024-01-01 01:00', 1, 0, 'z') ON CONFLICT (time) DO SELECT; +ERROR: ON CONFLICT DO SELECT requires a RETURNING clause at character 59 +\set ON_ERROR_STOP 1 +-- Chunk tuple descriptor differs from the hypertable root (dropped column). +-- The column is dropped after the first chunk exists, so the new chunk below +-- exercises the attribute-remapping path in the per-chunk ON CONFLICT setup. +ALTER TABLE ht_ocs DROP COLUMN label; +INSERT INTO ht_ocs VALUES ('2024-02-01 01:00', 5, 50.0); +INSERT INTO ht_ocs VALUES ('2024-02-01 01:00', 6, 60.0) +ON CONFLICT (time) DO SELECT RETURNING *; + time | device | value +------------------------------+--------+------- + Thu Feb 01 01:00:00 2024 PST | 5 | 50 + +-- DO SELECT against a compressed chunk decompresses the conflicting batch so +-- the existing row can be fetched and returned. +ALTER TABLE ht_ocs SET (timescaledb.compress, timescaledb.compress_segmentby = 'device'); +NOTICE: updated compression settings will only apply to future compressions +SELECT count(compress_chunk(c)) FROM show_chunks('ht_ocs') c; + count +------- + 5 + +SELECT ch AS "CHUNK" FROM show_chunks('ht_ocs', newer_than => '2024-01-01 00:00+00'::timestamptz, older_than => '2024-01-02 00:00+00'::timestamptz) ch \gset +-- fully compressed: the uncompressed chunk is empty +SELECT count(*) FROM ONLY :CHUNK; + count +------- + 0 + +INSERT INTO ht_ocs VALUES ('2024-01-01 01:00', 1, 0) +ON CONFLICT (time) DO SELECT RETURNING time, device, value; + time | device | value +------------------------------+--------+------- + Mon Jan 01 01:00:00 2024 PST | 1 | 10 + +-- the conflicting batch was decompressed into the uncompressed chunk +SELECT count(*) FROM ONLY :CHUNK; + count +------- + 1 + +-- The conflicting compressed row must be returned unchanged, not duplicated +SELECT count(*), min(value) FROM ht_ocs WHERE time = '2024-01-01 01:00'; + count | min +-------+----- + 1 | 10 + +-- With enable_dml_decompression off the required decompression is refused +SET timescaledb.enable_dml_decompression = off; +\set ON_ERROR_STOP 0 +INSERT INTO ht_ocs VALUES ('2024-02-01 01:00', 6, 0) +ON CONFLICT (time) DO SELECT RETURNING time, device; +ERROR: inserting into compressed chunk with unique constraints disabled +\set ON_ERROR_STOP 1 +RESET timescaledb.enable_dml_decompression; +DROP TABLE ht_ocs; +-- DO SELECT honors max_tuples_decompressed_per_dml_transaction: a conflict that +-- needs to decompress a batch larger than the limit is refused. +CREATE TABLE ocs_limit( + time timestamptz NOT NULL, + device int NOT NULL, + value float, + PRIMARY KEY (time, device) +) WITH (tsdb.hypertable, tsdb.partition_column = 'time', tsdb.chunk_interval = '1 day'); +INSERT INTO ocs_limit +SELECT '2024-01-01 01:00'::timestamptz + (g || ' second')::interval, 1, g +FROM generate_series(1, 10) g; +ALTER TABLE ocs_limit SET (timescaledb.compress, timescaledb.compress_segmentby = 'device'); +NOTICE: updated compression settings will only apply to future compressions +SELECT count(compress_chunk(c)) FROM show_chunks('ocs_limit') c; + count +------- + 1 + +SET timescaledb.max_tuples_decompressed_per_dml_transaction = 5; +\set VERBOSITY default +\set ON_ERROR_STOP 0 +INSERT INTO ocs_limit VALUES ('2024-01-01 01:00:01', 1, 0) +ON CONFLICT (time, device) DO SELECT RETURNING device; +ERROR: tuple decompression limit exceeded by operation +DETAIL: current limit: 5, tuples decompressed: 10 +HINT: Consider increasing timescaledb.max_tuples_decompressed_per_dml_transaction or set to 0 (unlimited). +\set ON_ERROR_STOP 1 +\set VERBOSITY terse +RESET timescaledb.max_tuples_decompressed_per_dml_transaction; +DROP TABLE ocs_limit; +-- Only the batch matching the insert's segmentby value is decompressed, not +-- every batch in the chunk. +CREATE TABLE ocs_batch( + time timestamptz NOT NULL, + device int NOT NULL, + value float, + PRIMARY KEY (time, device) +) WITH (tsdb.hypertable, tsdb.partition_column = 'time', tsdb.chunk_interval = '1 day'); +-- three device segments, five rows each, all in one chunk -> three batches +INSERT INTO ocs_batch +SELECT '2024-01-01 01:00'::timestamptz + (g || ' second')::interval, d, g +FROM generate_series(1, 5) g, generate_series(1, 3) d; +ALTER TABLE ocs_batch SET (timescaledb.compress, timescaledb.compress_segmentby = 'device'); +NOTICE: updated compression settings will only apply to future compressions +SELECT count(compress_chunk(c)) FROM show_chunks('ocs_batch') c; + count +------- + 1 + +SELECT ch AS "CHUNK" FROM show_chunks('ocs_batch') ch \gset +-- fully compressed: the uncompressed chunk is empty +SELECT count(*) FROM ONLY :CHUNK; + count +------- + 0 + +-- conflict on device 1 only decompresses that segment's batch (5 of 15 rows) +INSERT INTO ocs_batch VALUES ('2024-01-01 01:00:01', 1, 0) +ON CONFLICT (time, device) DO SELECT RETURNING device; + device +-------- + 1 + +SELECT count(*) FROM ONLY :CHUNK; + count +------- + 5 + +DROP TABLE ocs_batch; diff --git a/tsl/test/sql/CMakeLists.txt b/tsl/test/sql/CMakeLists.txt index 289e86d52f9..055846029cb 100644 --- a/tsl/test/sql/CMakeLists.txt +++ b/tsl/test/sql/CMakeLists.txt @@ -237,7 +237,7 @@ if((${PG_VERSION_MAJOR} GREATER_EQUAL "18")) endif() if((${PG_VERSION_MAJOR} GREATER_EQUAL "19")) - list(APPEND TEST_FILES repack.sql) + list(APPEND TEST_FILES upsert_do_select.sql repack.sql) endif() set(SOLO_TESTS diff --git a/tsl/test/sql/upsert_do_select.sql b/tsl/test/sql/upsert_do_select.sql new file mode 100644 index 00000000000..8292f8fea83 --- /dev/null +++ b/tsl/test/sql/upsert_do_select.sql @@ -0,0 +1,149 @@ +-- This file and its contents are licensed under the Timescale License. +-- Please see the included NOTICE for copyright information and +-- LICENSE-TIMESCALE for a copy of the license. + +-- Test INSERT ... ON CONFLICT DO SELECT on hypertables (PG19+ feature) + +CREATE TABLE ht_ocs( + time timestamptz NOT NULL, + device int NOT NULL, + value float, + label text, + PRIMARY KEY (time) +) WITH (tsdb.hypertable, tsdb.partition_column = 'time', tsdb.chunk_interval = '1 day'); + +INSERT INTO ht_ocs VALUES ('2024-01-01 01:00', 1, 10.0, 'a'); +INSERT INTO ht_ocs VALUES ('2024-01-02 01:00', 2, 20.0, 'b'); + +-- Get-or-create: conflicting row is returned unchanged, no new tuple written +INSERT INTO ht_ocs VALUES ('2024-01-01 01:00', 99, 99.0, 'x') +ON CONFLICT (time) DO SELECT RETURNING *; + +-- No conflict: the freshly inserted row is returned +INSERT INTO ht_ocs VALUES ('2024-01-03 01:00', 3, 30.0, 'c') +ON CONFLICT (time) DO SELECT RETURNING *; + +-- Multi-value insert mixing new rows and conflicts, spanning several chunks +INSERT INTO ht_ocs VALUES + ('2024-01-01 01:00', 1, 11.0, 'a2'), + ('2024-01-04 01:00', 4, 40.0, 'd'), + ('2024-01-02 01:00', 2, 22.0, 'b2') +ON CONFLICT (time) DO SELECT RETURNING time, device, value; + +SELECT * FROM ht_ocs ORDER BY time; + +-- Projected RETURNING list +INSERT INTO ht_ocs VALUES ('2024-01-01 01:00', 1, 0, 'z') +ON CONFLICT (time) DO SELECT RETURNING device, value * 2 AS double_value; + +-- WHERE clause: returns the row only when the qual matches +INSERT INTO ht_ocs VALUES ('2024-01-01 01:00', 1, 0, 'z') +ON CONFLICT (time) DO SELECT WHERE ht_ocs.device = 1 RETURNING device; + +INSERT INTO ht_ocs VALUES ('2024-01-01 01:00', 1, 0, 'z') +ON CONFLICT (time) DO SELECT WHERE ht_ocs.device = 12345 RETURNING device; + +-- Lock the conflicting row FOR UPDATE / FOR SHARE +INSERT INTO ht_ocs VALUES ('2024-01-02 01:00', 2, 0, 'z') +ON CONFLICT (time) DO SELECT FOR UPDATE RETURNING time, device; + +INSERT INTO ht_ocs VALUES ('2024-01-02 01:00', 2, 0, 'z') +ON CONFLICT (time) DO SELECT FOR SHARE RETURNING time, device; + +-- RETURNING is mandatory for DO SELECT +\set ON_ERROR_STOP 0 +INSERT INTO ht_ocs VALUES ('2024-01-01 01:00', 1, 0, 'z') ON CONFLICT (time) DO SELECT; +\set ON_ERROR_STOP 1 + +-- Chunk tuple descriptor differs from the hypertable root (dropped column). +-- The column is dropped after the first chunk exists, so the new chunk below +-- exercises the attribute-remapping path in the per-chunk ON CONFLICT setup. +ALTER TABLE ht_ocs DROP COLUMN label; +INSERT INTO ht_ocs VALUES ('2024-02-01 01:00', 5, 50.0); +INSERT INTO ht_ocs VALUES ('2024-02-01 01:00', 6, 60.0) +ON CONFLICT (time) DO SELECT RETURNING *; + +-- DO SELECT against a compressed chunk decompresses the conflicting batch so +-- the existing row can be fetched and returned. +ALTER TABLE ht_ocs SET (timescaledb.compress, timescaledb.compress_segmentby = 'device'); +SELECT count(compress_chunk(c)) FROM show_chunks('ht_ocs') c; + +SELECT ch AS "CHUNK" FROM show_chunks('ht_ocs', newer_than => '2024-01-01 00:00+00'::timestamptz, older_than => '2024-01-02 00:00+00'::timestamptz) ch \gset + +-- fully compressed: the uncompressed chunk is empty +SELECT count(*) FROM ONLY :CHUNK; + +INSERT INTO ht_ocs VALUES ('2024-01-01 01:00', 1, 0) +ON CONFLICT (time) DO SELECT RETURNING time, device, value; + +-- the conflicting batch was decompressed into the uncompressed chunk +SELECT count(*) FROM ONLY :CHUNK; + +-- The conflicting compressed row must be returned unchanged, not duplicated +SELECT count(*), min(value) FROM ht_ocs WHERE time = '2024-01-01 01:00'; + +-- With enable_dml_decompression off the required decompression is refused +SET timescaledb.enable_dml_decompression = off; +\set ON_ERROR_STOP 0 +INSERT INTO ht_ocs VALUES ('2024-02-01 01:00', 6, 0) +ON CONFLICT (time) DO SELECT RETURNING time, device; +\set ON_ERROR_STOP 1 +RESET timescaledb.enable_dml_decompression; + +DROP TABLE ht_ocs; + +-- DO SELECT honors max_tuples_decompressed_per_dml_transaction: a conflict that +-- needs to decompress a batch larger than the limit is refused. +CREATE TABLE ocs_limit( + time timestamptz NOT NULL, + device int NOT NULL, + value float, + PRIMARY KEY (time, device) +) WITH (tsdb.hypertable, tsdb.partition_column = 'time', tsdb.chunk_interval = '1 day'); + +INSERT INTO ocs_limit +SELECT '2024-01-01 01:00'::timestamptz + (g || ' second')::interval, 1, g +FROM generate_series(1, 10) g; + +ALTER TABLE ocs_limit SET (timescaledb.compress, timescaledb.compress_segmentby = 'device'); +SELECT count(compress_chunk(c)) FROM show_chunks('ocs_limit') c; + +SET timescaledb.max_tuples_decompressed_per_dml_transaction = 5; +\set VERBOSITY default +\set ON_ERROR_STOP 0 +INSERT INTO ocs_limit VALUES ('2024-01-01 01:00:01', 1, 0) +ON CONFLICT (time, device) DO SELECT RETURNING device; +\set ON_ERROR_STOP 1 +\set VERBOSITY terse +RESET timescaledb.max_tuples_decompressed_per_dml_transaction; + +DROP TABLE ocs_limit; + +-- Only the batch matching the insert's segmentby value is decompressed, not +-- every batch in the chunk. +CREATE TABLE ocs_batch( + time timestamptz NOT NULL, + device int NOT NULL, + value float, + PRIMARY KEY (time, device) +) WITH (tsdb.hypertable, tsdb.partition_column = 'time', tsdb.chunk_interval = '1 day'); + +-- three device segments, five rows each, all in one chunk -> three batches +INSERT INTO ocs_batch +SELECT '2024-01-01 01:00'::timestamptz + (g || ' second')::interval, d, g +FROM generate_series(1, 5) g, generate_series(1, 3) d; + +ALTER TABLE ocs_batch SET (timescaledb.compress, timescaledb.compress_segmentby = 'device'); +SELECT count(compress_chunk(c)) FROM show_chunks('ocs_batch') c; +SELECT ch AS "CHUNK" FROM show_chunks('ocs_batch') ch \gset + +-- fully compressed: the uncompressed chunk is empty +SELECT count(*) FROM ONLY :CHUNK; + +-- conflict on device 1 only decompresses that segment's batch (5 of 15 rows) +INSERT INTO ocs_batch VALUES ('2024-01-01 01:00:01', 1, 0) +ON CONFLICT (time, device) DO SELECT RETURNING device; + +SELECT count(*) FROM ONLY :CHUNK; + +DROP TABLE ocs_batch;