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;