diff --git a/CHANGELOG.md b/CHANGELOG.md index 672afdd14..8b182bf77 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -409,6 +409,31 @@ true until the next version shipped. only for a role the scan would admit, and gives everyone else the flat estimate, as the Parquet wrapper has since #330. `test/fdw_plan_privilege.sh` and `test_fdw_plan_privilege.py` pin both wrappers' gates. +- **A plpgsql EXCEPTION block that rolls back no longer loses the enclosing statement's + rows (#1341).** A statement ending inside the block flushed every buffered write, + including the enclosing INSERT's or DELETE's, under the block's subtransaction; the + rollback then took them away. `INSERT INTO t SELECT f(g) FROM generate_series(1, + 5000) g`, where `f`'s block wrote a columnar table and raised, kept 1 row. Now a + statement's end flushes only its own subtransaction's buffers. The same loss through a + READ of that table inside the block is #1360, not yet fixed. +- **A unique index is no longer violated by an insert racing an uncommitted delete of + its key (#1341).** The index fetch answered a row whose delete was still in progress + as gone, so the uniqueness check did not wait; when the delete rolled back, the key had + two rows. It now reports the row present, as heap does, and the check waits. +- **An index built after a DELETE serves a transaction whose snapshot began before it + (#1341).** The build left deleted rows out, so that transaction's index scan missed a + row its table scan returned. Deleted rows are now indexed as not alive, as heap indexes + a recently dead tuple; a UNIQUE index still builds over a key whose other holder was + deleted. +- **`INSERT ... ON CONFLICT DO NOTHING` works on a columnar table with a unique index + (#1341).** It raised `0A000` ("speculative insert is not supported yet") on any table + with a primary key or unique index, which is every table it is used on. + `docs/limitations.md` said it worked. Speculative insertion now inserts the row as an + ordinary row, and withdraws it again (a delete mark) when a concurrent inserter won the + key. `ON CONFLICT DO UPDATE` still refuses when a row conflicts, because updating it + takes a row lock; with no conflict it inserts. Tested in `am_contracts`, in both + twins, including eight sessions racing on the same keys: 0 duplicate keys, against + 407 to 507 with the withdrawal disabled. - Changing a text column's collation without a rewrite no longer gives wrong answers (#1327). `ALTER COLUMN ... TYPE text COLLATE ` keeps the table's files, so the zone maps kept minima and maxima ordered under the old diff --git a/docs/limitations.md b/docs/limitations.md index c5fcbd4cb..9bd7666fb 100644 --- a/docs/limitations.md +++ b/docs/limitations.md @@ -652,8 +652,10 @@ other than where the feature is used: partitions. Drop the constraint first if the table must be columnar. A conversion of the referencing side is permitted. A conversion of a table that no foreign key refers to is also permitted. -- `INSERT ... ON CONFLICT DO UPDATE` takes a row lock and raises the same error. - `ON CONFLICT DO NOTHING` does not, and works. +- `INSERT ... ON CONFLICT DO UPDATE` raises the same error when a conflicting row + exists, because updating it takes a row lock. When no row conflicts, it inserts. + `ON CONFLICT DO NOTHING` takes no row lock and works, with or without a conflict + target. `CREATE TABLE` refuses an unlogged columnar table for the same reason. Both cases follow one rule. This access method refuses a configuration that it cannot diff --git a/src/columnar_delete_vector.c b/src/columnar_delete_vector.c index 110fd633a..174889d1d 100644 --- a/src/columnar_delete_vector.c +++ b/src/columnar_delete_vector.c @@ -439,6 +439,25 @@ PgColumnarFlushAllDeleteVectors(void) delete_vector_flush_buffer((DeleteVectorBuffer *) lfirst(lc)); } +/* + * PgColumnarFlushDeleteVectorsOfSubXact + * Flush only the delete buffers the given subtransaction owns, for the reason + * PgColumnarFlushPendingWritesOfSubXact gives (#1341). + */ +void +PgColumnarFlushDeleteVectorsOfSubXact(SubTransactionId subid) +{ + ListCell *lc; + + foreach(lc, PgColumnarDeleteVectorBuffers) + { + DeleteVectorBuffer *buf = (DeleteVectorBuffer *) lfirst(lc); + + if (buf->subid == subid) + delete_vector_flush_buffer(buf); + } +} + /* * PgColumnarDiscardAllDeleteVectors * Forget all pending delete buffers (transaction end). diff --git a/src/columnar_delete_vector.h b/src/columnar_delete_vector.h index 5168bff77..aa7465967 100644 --- a/src/columnar_delete_vector.h +++ b/src/columnar_delete_vector.h @@ -24,6 +24,7 @@ extern void PgColumnarMarkRowDeleted(Relation rel, uint64 rowNumber); extern bool PgColumnarDeleteVectorBufferedDeleted(Relation rel, uint64 rowNumber); extern void PgColumnarFlushAllDeleteVectors(void); +extern void PgColumnarFlushDeleteVectorsOfSubXact(SubTransactionId subid); extern void PgColumnarDiscardAllDeleteVectors(void); diff --git a/src/columnar_reader.c b/src/columnar_reader.c index fac0e306f..2c9b0d003 100644 --- a/src/columnar_reader.c +++ b/src/columnar_reader.c @@ -292,6 +292,9 @@ struct PgColumnarReadState */ char *nativeDeleteMask; uint32 nativeDeleteMaskLen; + /* return deleted rows too, flagged; the index build asks for this (#1341) */ + bool emitDeleted; + bool lastRowDeleted; MemoryContext readContext; /* whole scan */ MemoryContext stripeContext; /* reset per stripe */ @@ -3453,13 +3456,35 @@ pgcolumnar_native_next_row(PgColumnarReadState *rs, Datum *values, bool *nulls, *rowNumber = rs->nativeGroup->firstRowNumber + rs->rowInGroup; rs->rowInGroup++; - if (deleted) + if (deleted && !rs->emitDeleted) continue; /* row deleted: cursors advanced, do not emit */ + rs->lastRowDeleted = deleted; return true; } } +/* + * PgColumnarReadSetEmitDeleted + * Return rows the snapshot sees as deleted as well, and let the caller ask + * PgColumnarReadLastRowDeleted which ones they were (#1341). The index build + * needs them: a row deleted after an older snapshot began is still visible to + * that snapshot, so a new index must hold it, as heap's build holds a + * recently-dead tuple. Only the unfiltered path honours this; a filter never + * sees a deleted row. + */ +void +PgColumnarReadSetEmitDeleted(PgColumnarReadState *readState, bool emit) +{ + readState->emitDeleted = emit; +} + +bool +PgColumnarReadLastRowDeleted(PgColumnarReadState *readState) +{ + return readState->lastRowDeleted; +} + bool PgColumnarReadNextRow(PgColumnarReadState *readState, Datum *values, bool *nulls, uint64 *rowNumber) diff --git a/src/columnar_reader.h b/src/columnar_reader.h index dab74e881..c211ddd1d 100644 --- a/src/columnar_reader.h +++ b/src/columnar_reader.h @@ -44,6 +44,8 @@ extern bool PgColumnarReadFoldColumn(PgColumnarReadState *readState, int attidx, const char **validity, const char **packed, int16 *attlen, const uint32 **vecRawLen); +extern void PgColumnarReadSetEmitDeleted(PgColumnarReadState *readState, bool emit); +extern bool PgColumnarReadLastRowDeleted(PgColumnarReadState *readState); extern void PgColumnarReadSetProjection(PgColumnarReadState *readState, Bitmapset *projectedColumns); diff --git a/src/columnar_tableam.c b/src/columnar_tableam.c index 6f4f211d1..c5a0744ed 100644 --- a/src/columnar_tableam.c +++ b/src/columnar_tableam.c @@ -1986,6 +1986,32 @@ pgcolumnar_index_fetch_tuple(struct IndexFetchTableData *scan, ItemPointer tid, else if (PgColumnarBufferedRowByNumber(rel, rowNumber, slot->tts_values, slot->tts_isnull)) ExecStoreVirtualTuple(slot); + else if (snapshot != NULL && snapshot->snapshot_type == SNAPSHOT_DIRTY && + (TransactionIdIsValid(snapshot->xmin) || + TransactionIdIsValid(snapshot->xmax)) && + PgColumnarRowIsLive(rel, SnapshotAny, rowNumber)) + { + /* + * A DELETE STILL IN PROGRESS IS NOT A DELETE YET (#1341). The row reads + * as gone only because another transaction's uncommitted delete-vector + * row says so, and the catalog read above recorded that transaction in + * the dirty snapshot. Answering "no such row" let _bt_check_unique insert + * a duplicate without waiting: when the delete then rolled back, the key + * had two live rows. Measured before this change: an INSERT of k racing + * an uncommitted DELETE of k, then a ROLLBACK, committed and left two + * rows with k. + * + * Heap answers such a row as present, with xmax naming the deleter, and + * the caller waits on it and asks again with the outcome known. Report + * the row the same way. The snapshot's xmin/xmax are left as the catalog + * read set them, so the caller waits on that transaction. If the in-progress + * change was not the one that deleted this row, the cost is a wait and a + * re-check, never a wrong answer. The row's values are not needed by the + * uniqueness check, so they are fetched under SnapshotAny, as for a row + * the caller already knows the fate of. + */ + PgColumnarSlotStoreDeferred(slot, rel, SnapshotAny, rowNumber); + } else return false; @@ -2137,20 +2163,54 @@ pgcolumnar_index_delete_tuples(Relation rel, TM_IndexDeleteOp *delstate) } #endif +/* + * Speculative insertion, which INSERT ... ON CONFLICT uses whenever an arbiter + * index exists (#1341). docs/limitations.md promised DO NOTHING "works", and it + * raised 0A000 on any table with a unique index, which is the only table ON + * CONFLICT arbitrates on. + * + * The executor's pre-check already skips a conflict it can see, without + * inserting. What reaches here is a row that looked free: insert it as an + * ordinary row, and if the arbiter index then reports a conflict (a concurrent + * inserter won), withdraw it again by marking it deleted. + * + * NO SPECULATIVE TOKEN. Heap stamps the token so a waiter can wait on the + * insertion rather than on the whole transaction. Here a waiter waits on the + * transaction, which is correct and only coarser. The same unique-key lock the + * ordinary insert takes (issue #5) serialises two inserters of one key. + * + * WITHDRAWN, NOT "SUPER-DELETED". Heap makes an aborted speculative tuple + * invisible to everyone at once. A delete mark is transactional, but the row was + * never visible outside this transaction, and it commits or aborts together + * with the mark. So no other session can ever see it, and this one never does + * either, since the executor does not return a withdrawn row. + * + * DO UPDATE is still refused, and not here: a conflict sends the executor to + * lock the existing row, and row locking is unsupported (0A000). + */ static void pgcolumnar_tuple_insert_speculative(Relation rel, TupleTableSlot *slot, CommandId cid, COLUMNAR_TABLE_OPTIONS options, struct BulkInsertStateData *bistate, uint32 specToken) { - COLUMNAR_UNSUPPORTED("speculative insert"); + pgcolumnar_tuple_insert(rel, slot, cid, options, bistate); } static void pgcolumnar_tuple_complete_speculative(Relation rel, TupleTableSlot *slot, uint32 specToken, bool succeeded) { - COLUMNAR_UNSUPPORTED("speculative insert"); + if (succeeded) + return; + + /* + * The row may still be buffered, and a delete mark needs the row group + * that covers it. Flush first, as the projection back-fill does before it + * reads this transaction's rows. + */ + PgColumnarFlushWriteStateForRelation(RelationGetRelid(rel)); + PgColumnarMarkRowDeleted(rel, PgColumnarItemPointerToRowNumber(&slot->tts_tid)); } /* @@ -2363,6 +2423,7 @@ pgcolumnar_index_build_range_scan(Relation table_rel, Relation index_rel, Datum indexValues[INDEX_MAX_KEYS]; bool indexNulls[INDEX_MAX_KEYS]; double reltuples = 0; + bool alive; uint64 rowNumber; if (start_blockno != 0 || numblocks != InvalidBlockNumber) @@ -2476,6 +2537,9 @@ pgcolumnar_index_build_range_scan(Relation table_rel, Relation index_rel, ownReadState = true; } + /* deleted rows too, flagged, so the build can index them as not alive (#1341) */ + PgColumnarReadSetEmitDeleted(readState, true); + /* * Say which branch ran and how wide the reader it produced will actually * read, so a test can assert the projection NARROWED rather than infer it @@ -2506,8 +2570,10 @@ pgcolumnar_index_build_range_scan(Relation table_rel, Relation index_rel, &rowNumber)) break; ExecStoreVirtualTuple(slot); + alive = !PgColumnarReadLastRowDeleted(readState); - reltuples += 1; + if (alive) + reltuples += 1; MemoryContextReset(econtext->ecxt_per_tuple_memory); @@ -2518,7 +2584,19 @@ pgcolumnar_index_build_range_scan(Relation table_rel, Relation index_rel, PgColumnarRowNumberToItemPointer(rowNumber, &slot->tts_tid); - callback(index_rel, &slot->tts_tid, indexValues, indexNulls, true, + /* + * A DELETED ROW IS INDEXED TOO, AS NOT ALIVE (#1341). The build reads + * under a current snapshot, but a transaction whose snapshot began before + * the delete still sees the row, and may use this index. Leaving the row + * out made that transaction's index scan miss a row it can see. Measured + * before this change: REPEATABLE READ, then a committed DELETE of row + * 4242, then CREATE INDEX; the old snapshot's index lookup of 4242 found + * 0 rows, where its scan of the table found 1. Heap indexes a + * recently-dead tuple the same way, as not alive, so a unique index does + * not count it as a duplicate. The fetch still filters by each caller's + * snapshot, so a new snapshot does not see the row by this entry. + */ + callback(index_rel, &slot->tts_tid, indexValues, indexNulls, alive, callback_state); } @@ -2781,8 +2859,15 @@ pgcolumnar_executor_end(QueryDesc *queryDesc) else standard_ExecutorEnd(queryDesc); - PgColumnarFlushAllPendingWrites(); - PgColumnarFlushAllDeleteVectors(); + /* + * ONLY THIS SUBTRANSACTION'S BUFFERS (#1341). A statement ending inside a + * plpgsql EXCEPTION block runs here while the block's subtransaction is + * current; flushing the enclosing statement's buffers now would attribute + * them to the block, and its rollback would lose them. They are flushed when + * their own statement ends, back in their own subtransaction. + */ + PgColumnarFlushPendingWritesOfSubXact(GetCurrentSubTransactionId()); + PgColumnarFlushDeleteVectorsOfSubXact(GetCurrentSubTransactionId()); /* * The fetch cache is scoped to a statement, so release it here rather than diff --git a/src/columnar_write_state.c b/src/columnar_write_state.c index e08fcb050..7ba410358 100644 --- a/src/columnar_write_state.c +++ b/src/columnar_write_state.c @@ -3864,6 +3864,34 @@ PgColumnarFlushAllPendingWrites(void) } } +/* + * PgColumnarFlushPendingWritesOfSubXact + * Flush only the write states the given subtransaction owns (#1341). + * + * A flush writes catalog rows under the subtransaction that is CURRENT, not + * the one that buffered the rows. So flushing an enclosing statement's buffer + * while a child subtransaction is current -- a statement inside a plpgsql + * EXCEPTION block ends -- attributes the parent's rows to the child, and the + * block's rollback takes them away. Measured before this: an INSERT ... SELECT + * f(g) of 5000 rows, where f's EXCEPTION block wrote a columnar table and rolled + * back, kept one row. A parent's buffer waits for its own statement's end. + */ +void +PgColumnarFlushPendingWritesOfSubXact(SubTransactionId subid) +{ + ListCell *lc; + + foreach(lc, PgColumnarWriteStates) + { + PgColumnarWriteState *writeState = (PgColumnarWriteState *) lfirst(lc); + + if (writeState->subid != subid) + continue; + pgcolumnar_flush_row_group(writeState); + flush_ws_projections(writeState); + } +} + /* * PgColumnarDiscardAllPendingWrites * Forget all pending write states. The backing memory is freed with the diff --git a/src/columnar_write_state.h b/src/columnar_write_state.h index 91d126512..2fef7bc89 100644 --- a/src/columnar_write_state.h +++ b/src/columnar_write_state.h @@ -36,6 +36,7 @@ extern bool PgColumnarBufferedRowByNumber(Relation rel, uint64 rowNumber, extern void PgColumnarForgetWriteStateForRelation(Oid relid); extern void PgColumnarFlushAllPendingWrites(void); +extern void PgColumnarFlushPendingWritesOfSubXact(SubTransactionId subid); extern void PgColumnarDiscardAllPendingWrites(void); diff --git a/test/am_contracts.sh b/test/am_contracts.sh new file mode 100755 index 000000000..900847a48 --- /dev/null +++ b/test/am_contracts.sh @@ -0,0 +1,211 @@ +#!/usr/bin/env bash +# +# Contracts the access method owes core DML and transactions, each one stated in +# docs/ or owed by core's table-AM interface, and none of them tested before #1341. +# One section per contract, each a promise someone relies on without reading the C. +# +# Usage: test/am_contracts.sh [PG_CONFIG] +# Written fresh for pgColumnar. + +set -uo pipefail +. "$(dirname "${BASH_SOURCE[0]}")/lib.sh" +pgc_setup "${1:-/usr/local/pg17/bin/pg_config}" + +# state SQL -> "ok" when it succeeds, or the SQLSTATE it raised. A refusal is decided by +# its code, not its wording. +state() { + local out + out="$(env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" \ + -At -v VERBOSITY=sqlstate -v ON_ERROR_STOP=1 -c "$1" 2>&1)" && { echo ok; return; } + sed -n 's/^.*ERROR:[[:space:]]*\([0-9A-Z]\{5\}\).*$/\1/p' <<<"$out" | head -1 +} + +# ---- INSERT ... ON CONFLICT (docs/limitations.md:655-656) ---------------------- +# +# The documented contract: DO NOTHING "works", and DO UPDATE "takes a row lock and +# raises the same error" as other row locking. Both on a table whose unique index is +# what ON CONFLICT arbitrates on -- the case anybody uses it for. +psql_run "CREATE TABLE oc (id int PRIMARY KEY, v text) USING pgcolumnar;" +psql_run "INSERT INTO oc SELECT g, 'v' || g FROM generate_series(1, 10) g;" +check "premise: the conflict table holds ten rows" "$(q 'SELECT count(*) FROM oc')" "10" + +check "ON CONFLICT DO NOTHING skips the duplicate and inserts the new row" \ + "$(state "INSERT INTO oc VALUES (5, 'dup'), (11, 'new') ON CONFLICT DO NOTHING;")|$(q "SELECT count(*) || '|' || (SELECT v FROM oc WHERE id = 5) FROM oc")" \ + "ok|11|v5" +check "and so does ON CONFLICT (id) DO NOTHING, naming its arbiter" \ + "$(state "INSERT INTO oc VALUES (6, 'dup'), (12, 'new') ON CONFLICT (id) DO NOTHING;")|$(q "SELECT count(*) || '|' || (SELECT v FROM oc WHERE id = 6) FROM oc")" \ + "ok|12|v6" +check "ON CONFLICT DO UPDATE is refused as unsupported (0A000), and changes nothing" \ + "$(state "INSERT INTO oc VALUES (7, 'upd') ON CONFLICT (id) DO UPDATE SET v = excluded.v;")|$(q "SELECT count(*) || '|' || (SELECT v FROM oc WHERE id = 7) FROM oc")" \ + "0A000|12|v7" + +check "ON CONFLICT DO UPDATE with no conflicting row inserts it" \ + "$(state "INSERT INTO oc VALUES (13, 'upd') ON CONFLICT (id) DO UPDATE SET v = excluded.v;")|$(q "SELECT count(*) || '|' || (SELECT v FROM oc WHERE id = 13) FROM oc")" \ + "ok|13|upd" +check "a key twice in one DO NOTHING statement is inserted once, the first time" \ + "$(state "INSERT INTO oc VALUES (30, 'first'), (30, 'second') ON CONFLICT DO NOTHING;")|$(q "SELECT count(*) || '|' || min(v) FROM oc WHERE id = 30")" \ + "ok|1|first" + +# THE WITHDRAWAL PATH. A row that looked free at the pre-check but conflicts once +# inserted, because another session inserted the key and had not committed. The +# first session holds key 20 uncommitted; the second runs DO NOTHING on 20 and 21, +# waits, and must end with exactly one row 20 -- the first session's -- and its 21. +env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" -Atq \ + -c "BEGIN; INSERT INTO oc VALUES (20, 'holder'); SELECT pg_sleep(3); COMMIT;" >/dev/null 2>&1 & +_oc_holder=$! +sleep 1 +_oc_racer="$(state "INSERT INTO oc VALUES (20, 'racer'), (21, 'racer') ON CONFLICT DO NOTHING;")" +wait "$_oc_holder" +check "a DO NOTHING racing an uncommitted insert of the same key waits, then skips it" \ + "$_oc_racer|$(q "SELECT count(*) FILTER (WHERE id = 20) || '|' || min(v) FILTER (WHERE id = 20) || '|' || count(*) FILTER (WHERE id = 21) FROM oc")" \ + "ok|1|holder|1" +check "and the unique index answers the key with that one row" \ + "$(q "SET enable_seqscan = off; SELECT count(*) || '|' || min(v) FROM oc WHERE id = 20;" | tail -1)" "1|holder" + +# THE WITHDRAWAL ITSELF, which the arm above does not reach: there the pre-check sees +# the holder's row and waits. A row is withdrawn only when another session inserts +# the key after this one's pre-check and before its own insert -- a window no +# ordering of two sessions can force. So eight sessions race on the same keys, each +# taken from the clock, for a few seconds. Measured: 0 duplicate keys on three runs; +# with the withdrawal disabled, 407 to 507. +psql_run "CREATE TABLE ocs (id bigint PRIMARY KEY, w int) USING pgcolumnar;" +_oc_sql="$PGC_WORKDIR/oc_race.sql" +for _i in $(seq 1 1500); do + echo "INSERT INTO ocs VALUES ((extract(epoch FROM clock_timestamp()) * 200)::bigint, 1) ON CONFLICT DO NOTHING;" +done > "$_oc_sql" +_oc_pids="" +for _w in 1 2 3 4 5 6 7 8; do + env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" -q \ + -f "$_oc_sql" > "$PGC_WORKDIR/oc_race.$_w.out" 2>&1 & + _oc_pids="$_oc_pids $!" +done +wait $_oc_pids +_oc_errors="$(cat "$PGC_WORKDIR"/oc_race.*.out | grep -c 'ERROR' || true)" +check "premise: the racing sessions contended for keys" \ + "$(q "SELECT (count(*) < 8 * 1500)::text FROM ocs")" "true" +check "sessions racing DO NOTHING on the same keys leave each key once, with no error" \ + "$(q "SELECT count(*) - count(DISTINCT id) FROM ocs")|$_oc_errors" "0|0" + +# ---- a TRUNCATE that rolls back restores every row (#1341) ------------------------- +# +# TRUNCATE swaps the table onto new storage (set_new_filelocator), so the old +# storage's rows and catalog entries must survive until the swap commits, and come +# back whole if it does not -- at transaction level and to a savepoint. +psql_run "CREATE TABLE tr (id int PRIMARY KEY, v int) USING pgcolumnar;" +psql_run "SELECT pgcolumnar.set_options('tr', stripe_row_limit => 1000);" +psql_run "INSERT INTO tr SELECT g, g * 3 FROM generate_series(1, 5000) g;" +_tr_before="$(q "SELECT count(*) || '|' || sum(v) FROM tr")" +check "premise: the truncate table holds 5000 rows" "$_tr_before" "5000|37507500" +psql_run "BEGIN; TRUNCATE tr; ROLLBACK;" +check "a TRUNCATE rolled back at transaction level restores every row" \ + "$(q "SELECT count(*) || '|' || sum(v) FROM tr")" "$_tr_before" +check "and the primary key still finds them by index" \ + "$(q "SET enable_seqscan = off; SELECT v FROM tr WHERE id = 4321;" | tail -1)" "12963" +psql_run "BEGIN; SAVEPOINT s; TRUNCATE tr; ROLLBACK TO SAVEPOINT s; INSERT INTO tr VALUES (5001, 1); COMMIT;" +check "a TRUNCATE rolled back to a savepoint restores every row, and the transaction goes on" \ + "$(q "SELECT count(*) || '|' || sum(v) FROM tr")" "5001|37507501" +check "and the primary key still refuses a duplicate of a restored row" \ + "$(state "INSERT INTO tr VALUES (17, 0);")" "23505" + +# AND THE ROWS THE PARENT BUFFERED BEFORE THE SAVEPOINT. TRUNCATE discards the cached +# write state; rows inserted earlier in the same transaction may still be in it, +# unflushed, and ROLLBACK TO SAVEPOINT must bring them back with the table. 2500 rows, +# not a multiple of the 1000-row stripe, so the last 500 are still buffered. +psql_run "BEGIN; INSERT INTO tr SELECT g, 1 FROM generate_series(6001, 8500) g; SAVEPOINT s; TRUNCATE tr; INSERT INTO tr SELECT g, 2 FROM generate_series(1, 5) g; ROLLBACK TO SAVEPOINT s; COMMIT;" +check "rows the parent buffered before a savepoint survive a TRUNCATE rolled back to it" \ + "$(q "SELECT count(*) || '|' || sum(v) FROM tr")" "7501|37510001" + +# ---- an aborted inner block keeps the outer statement's rows (#1341) --------------- +# +# A plpgsql EXCEPTION block is a subtransaction. A statement inside it ends, and its +# ExecutorEnd flushes every pending write -- the OUTER statement's buffered rows and +# delete marks too. If what that flush writes belongs to the inner subtransaction, +# the block's rollback takes the outer rows with it. Each arm is run on heap too, and +# the columnar answer must be heap's. +psql_run "CREATE TABLE sx_side (id int) USING pgcolumnar;" +psql_run "CREATE FUNCTION sx_f(i int) RETURNS int LANGUAGE plpgsql AS \$\$ +BEGIN + BEGIN + INSERT INTO sx_side VALUES (i); + PERFORM 1 / 0; + EXCEPTION WHEN division_by_zero THEN NULL; + END; + RETURN i; +END \$\$;" +for am in pgcolumnar heap; do + psql_run "CREATE TABLE sx_$am (id int) USING $am;" + psql_run "INSERT INTO sx_$am SELECT sx_f(g) FROM generate_series(1, 5000) g;" +done +check "an outer insert survives an aborted inner block, as on heap" \ + "$(q "SELECT count(*) || '|' || sum(id) FROM sx_pgcolumnar")" "$(q "SELECT count(*) || '|' || sum(id) FROM sx_heap")" +check "and the inner block's own rows are gone" "$(q "SELECT count(*) FROM sx_side")" "0" +for am in pgcolumnar heap; do + psql_run "DELETE FROM sx_$am WHERE id % 2 = 0 AND sx_f(id) IS NOT NULL;" +done +check "an outer delete survives an aborted inner block, as on heap" \ + "$(q "SELECT count(*) || '|' || sum(id) FROM sx_pgcolumnar")" "$(q "SELECT count(*) || '|' || sum(id) FROM sx_heap")" + +# ---- an insert racing an uncommitted delete of its key (#1341) ---------------------- +# +# Session 1 deletes key k and has not committed; session 2 inserts k. On heap session +# 2 waits for session 1: a rollback leaves k taken (23505), a commit frees it. One row +# with k either way -- never two. +psql_run "CREATE TABLE ud (k int UNIQUE, v text) USING pgcolumnar;" +psql_run "INSERT INTO ud VALUES (5, 'a'), (6, 'b'), (7, 'c');" +# race HOLD_SQL INSERT_SQL -> the inserter's state, after the holder has ended +race() { + env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" -Atq \ + -c "BEGIN; $1; SELECT pg_sleep(3); $2;" >/dev/null 2>&1 & + local holder=$! + sleep 1 + local got; got="$(state "$3")" + wait "$holder" + echo "$got" +} +check "an insert of a key whose delete is uncommitted waits, and fails when the delete rolls back" \ + "$(race "DELETE FROM ud WHERE k = 5" "ROLLBACK" "INSERT INTO ud VALUES (5, 'racer');")|$(q "SELECT count(*) || '|' || min(v) FROM ud WHERE k = 5")" \ + "23505|1|a" +check "and succeeds when the delete commits, leaving one row with the key" \ + "$(race "DELETE FROM ud WHERE k = 6" "COMMIT" "INSERT INTO ud VALUES (6, 'racer');")|$(q "SELECT count(*) || '|' || min(v) FROM ud WHERE k = 6")" \ + "ok|1|racer" +check "an insert of a key an uncommitted UPDATE moved away fails when the update rolls back" \ + "$(race "UPDATE ud SET k = 70 WHERE k = 7" "ROLLBACK" "INSERT INTO ud VALUES (7, 'racer');")|$(q "SELECT count(*) || '|' || min(v) FROM ud WHERE k = 7")" \ + "23505|1|c" + +# ---- an index built after a delete serves an older snapshot that sees the row (#1341) - +# +# Session 1 holds a REPEATABLE READ snapshot from before a committed DELETE of row 4242; +# then an index is built. Session 1 still sees row 4242, so an index scan must find it, +# as on heap, which indexes rows not yet dead to every snapshot. +psql_run "CREATE TABLE ib (id int, v int) USING pgcolumnar;" +psql_run "INSERT INTO ib SELECT g, g FROM generate_series(1, 10000) g;" +_ib_out="$PGC_WORKDIR/ib_rr.out" +env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" -Atq -v ON_ERROR_STOP=1 \ + -c "BEGIN ISOLATION LEVEL REPEATABLE READ; SELECT count(*) FROM ib; SELECT pg_sleep(4); + SET LOCAL enable_seqscan = off; SET LOCAL enable_bitmapscan = off; + EXPLAIN (COSTS OFF) SELECT count(*) FROM ib WHERE id = 4242; + SELECT 'rows:' || count(*) FROM ib WHERE id = 4242; COMMIT;" > "$_ib_out" 2>&1 & +_ib_s1=$! +sleep 1 +psql_run "DELETE FROM ib WHERE id = 4242;" +psql_run "CREATE INDEX ib_id ON ib (id);" +wait "$_ib_s1" +check "premise: the index exists and is valid" \ + "$(q "SELECT indisvalid FROM pg_index WHERE indexrelid = 'ib_id'::regclass")" "t" +check "premise: the older snapshot's lookup is an index scan" \ + "$(grep -cE 'Index (Only )?Scan using ib_id' "$_ib_out")" "1" +check "an index built after a delete finds the row for an older snapshot that sees it" \ + "$(sed -n 's/^rows://p' "$_ib_out")" "1" +check "and a new snapshot does not see the deleted row by that index" \ + "$(q "SET enable_seqscan = off; SET enable_bitmapscan = off; SELECT count(*) FROM ib WHERE id = 4242;" | tail -1)" "0" + +# The cost of indexing deleted rows, pinned: they go in as not alive, so a UNIQUE +# index still builds over a key whose only other holder is a deleted row. +psql_run "CREATE TABLE ibu (k int) USING pgcolumnar;" +psql_run "INSERT INTO ibu VALUES (5), (6);" +psql_run "DELETE FROM ibu WHERE k = 5;" +psql_run "INSERT INTO ibu VALUES (5);" +check "a unique index builds over a key whose other holder is a deleted row" \ + "$(state "CREATE UNIQUE INDEX ibu_k ON ibu (k);")|$(q "SET enable_seqscan = off; SELECT count(*) FROM ibu WHERE k = 5;" | tail -1)" "ok|1" + +pgc_summary diff --git a/test/check_ledger.tsv b/test/check_ledger.tsv index 5452efc59..63c475123 100644 --- a/test/check_ledger.tsv +++ b/test/check_ledger.tsv @@ -1,3 +1,30 @@ +am_contracts am_contracts ON CONFLICT DO NOTHING skips the duplicate and inserts the new row 15;16;17;18;19 2026-10-14 speculative insert refused as unsupported, as before #1341 +am_contracts am_contracts ON CONFLICT DO UPDATE is refused as unsupported (0A000), and changes nothing 15;16;17;18;19 2026-10-14 speculative insert refused as unsupported, as before #1341 +am_contracts am_contracts ON CONFLICT DO UPDATE with no conflicting row inserts it 15;16;17;18;19 2026-10-14 speculative insert refused as unsupported, as before #1341 +am_contracts am_contracts a DO NOTHING racing an uncommitted insert of the same key waits, then skips it 15;16;17;18;19 2026-10-14 speculative insert refused as unsupported, as before #1341 +am_contracts am_contracts a TRUNCATE rolled back at transaction level restores every row 15;16;17;18;19 2026-10-14 TRUNCATE empties the old storage in place (RelationTruncate before the swap) +am_contracts am_contracts a TRUNCATE rolled back to a savepoint restores every row, and the transaction goes on 15;16;17;18;19 2026-10-14 TRUNCATE empties the old storage in place (RelationTruncate before the swap) +am_contracts am_contracts a key twice in one DO NOTHING statement is inserted once, the first time 15;16;17;18;19 2026-10-14 speculative insert refused as unsupported, as before #1341 +am_contracts am_contracts a unique index builds over a key whose other holder is a deleted row 15;16;17;18;19 never - +am_contracts am_contracts an index built after a delete finds the row for an older snapshot that sees it 15;16;17;18;19 2026-10-14 the index build skips deleted rows, as before #1341 +am_contracts am_contracts an insert of a key an uncommitted UPDATE moved away fails when the update rolls back 15;16;17;18;19 2026-10-14 the dirty-snapshot index fetch answers an in-progress delete as final, as before #1341 +am_contracts am_contracts an insert of a key whose delete is uncommitted waits, and fails when the delete rolls back 15;16;17;18;19 2026-10-14 the dirty-snapshot index fetch answers an in-progress delete as final, as before #1341 +am_contracts am_contracts an outer delete survives an aborted inner block, as on heap 15;16;17;18;19 2026-10-14 ExecutorEnd flushes every subtransaction's buffers, as before #1341 +am_contracts am_contracts an outer insert survives an aborted inner block, as on heap 15;16;17;18;19 2026-10-14 ExecutorEnd flushes every subtransaction's buffers, as before #1341 +am_contracts am_contracts and a new snapshot does not see the deleted row by that index 15;16;17;18;19 never - +am_contracts am_contracts and so does ON CONFLICT (id) DO NOTHING, naming its arbiter 15;16;17;18;19 2026-10-14 speculative insert refused as unsupported, as before #1341 +am_contracts am_contracts and succeeds when the delete commits, leaving one row with the key 15;16;17;18;19 2026-10-14 ExecutorEnd flushes no pending writes +am_contracts am_contracts and the inner block's own rows are gone 15;16;17;18;19 never - +am_contracts am_contracts and the primary key still finds them by index 15;16;17;18;19 2026-10-14 TRUNCATE empties the old storage in place (RelationTruncate before the swap) +am_contracts am_contracts and the primary key still refuses a duplicate of a restored row 15;16;17;18;19 2026-10-14 TRUNCATE empties the old storage in place (RelationTruncate before the swap) +am_contracts am_contracts and the unique index answers the key with that one row 15;16;17;18;19 never - +am_contracts am_contracts premise: the conflict table holds ten rows 15;16;17;18;19 never - +am_contracts am_contracts premise: the index exists and is valid 15;16;17;18;19 never - +am_contracts am_contracts premise: the older snapshot's lookup is an index scan 15;16;17;18;19 never - +am_contracts am_contracts premise: the racing sessions contended for keys 15;16;17;18;19 never - +am_contracts am_contracts premise: the truncate table holds 5000 rows 15;16;17;18;19 never - +am_contracts am_contracts rows the parent buffered before a savepoint survive a TRUNCATE rolled back to it 15;16;17;18;19 2026-10-14 ExecutorEnd flushes no pending writes;TRUNCATE empties the old storage in place (RelationTruncate before the swap) +am_contracts am_contracts sessions racing DO NOTHING on the same keys leave each key once, with no error 15;16;17;18;19 2026-10-14 a speculative row that lost the race is never withdrawn;speculative insert refused as unsupported, as before #1341 analyze_stats analyze_stats ANALYZE now collects statistics at all 15;16;17;18;19 never - analyze_stats analyze_stats ANALYZE on a wide table completes well inside its derived cap 15;16;17;18;19 never - analyze_stats analyze_stats a fast machine still gets the floor, not a bound that shrinks with it 15;16;17;18;19 2026-09-24 the cap loses its absolute floor, so the bound shrinks with the machine it runs on;the floor moved by one, so the boundary where floor and derived bound meet is off by one diff --git a/test/check_ledger_budget.txt b/test/check_ledger_budget.txt index 1f2808e78..94318cae3 100644 --- a/test/check_ledger_budget.txt +++ b/test/check_ledger_budget.txt @@ -2137,4 +2137,17 @@ suites_not_covered 237 # never 2020 + 21 = 2041 # dated 608 + 7 = 615 # 2041 + 615 == 2656 -checks_never_observed_red 2041 +# +# #1341, am_contracts (jdatcmd): 2041 -> 2050. A new suite, seeded from five-major runs +# (27/27 on each of 15-19), so suites_not_covered does not rise. 27 rows: 18 arrive +# dated by seven mutations on PG 18 -- the speculative-insert stub, the withdrawal +# disabled, no ExecutorEnd flush, TRUNCATE emptying the old storage in place, the +# flush-every-subtransaction ExecutorEnd, the dirty fetch answering an in-progress +# delete as final, and the index build skipping deleted rows -- each merged with +# --target naming every check it reddened. 9 arrive never: 5 premises and 4 controls. +# +# ledger rows 2656 + 27 = 2683 +# never 2041 + 27 - 18 = 2050 +# dated 615 + 18 = 633 +# 2050 + 633 == 2683 +checks_never_observed_red 2050 diff --git a/test/pytest/TESTS.md b/test/pytest/TESTS.md index eb7bedce3..603fb4b75 100644 --- a/test/pytest/TESTS.md +++ b/test/pytest/TESTS.md @@ -155,6 +155,7 @@ behaviour, the source of that number is named. - [107. test_fdw_plan_privilege.py: planning a foreign table reads no server file for a role the scan would refuse](#107-test_fdw_plan_privilegepy-planning-a-foreign-table-reads-no-server-file-for-a-role-the-scan-would-refuse) - [108. test_arrow_import.py: import_arrow round trips, reads a foreign file, and refuses what it cannot decode](#108-test_arrow_importpy-import_arrow-round-trips-reads-a-foreign-file-and-refuses-what-it-cannot-decode) - [109. test_entry_point_privilege.py: every relation-taking entry point has a decided privilege bar](#109-test_entry_point_privilegepy-every-relation-taking-entry-point-has-a-decided-privilege-bar) +- [110. test_am_contracts.py: contracts the access method owes core DML and transactions](#110-test_am_contractspy-contracts-the-access-method-owes-core-dml-and-transactions) ## 1. How to read a test in here @@ -6887,3 +6888,21 @@ The pytest twin of `test/entry_point_privilege.sh` (#569, ported for #1341), eve | test | what it holds | | --- | --- | | `test_every_relation_taking_entry_point_has_a_decided_bar` | the enumeration premises, the symbol-set agreement, the bucket and class counts, the five 42501 refusals with their owner premises, and the four file siblings refused at the table | + +## 110. test_am_contracts.py: contracts the access method owes core DML and transactions + +The pytest twin of `test/am_contracts.sh` (#1341). Each section is a promise from `docs/` or from core's table-AM interface that no test exercised. The first is `INSERT ... ON CONFLICT`: `docs/limitations.md` said `DO NOTHING` works, and on any table with a unique index it raised `0A000`, because the access method refused speculative insertion. Speculative insertion now inserts the row as an ordinary row, and withdraws it with a delete mark when a concurrent inserter won the key. + +THE WITHDRAWAL IS REACHED ONLY BY A RACE. When one session holds a key uncommitted, the executor's pre-check sees it and waits, so it never inserts. A row is withdrawn only when another session inserts the key between this one's pre-check and its insert. So the third test races eight sessions on clock-derived keys. Measured: 0 duplicate keys clean; 407 to 507 with the withdrawal disabled. + +### Every test + +| test | what it holds | +| --- | --- | +| `test_insert_on_conflict` | `DO NOTHING` skips a duplicate and inserts the rest, with and without a conflict target; a key twice in one statement goes in once; `DO UPDATE` is refused (`0A000`) on a conflicting row and changes nothing, and inserts when nothing conflicts | +| `test_insert_on_conflict_withdraws_a_row_a_concurrent_insert_beat` | a `DO NOTHING` racing another session's uncommitted insert of the same key waits, then skips it: one row, the holder's, by scan and by the unique index | +| `test_insert_on_conflict_racing_sessions_leave_each_key_once` | eight sessions racing `DO NOTHING` on the same keys leave each key once, with no error; a premise requires that they contended | +| `test_a_truncate_that_rolls_back_restores_every_row` | a `TRUNCATE` rolled back at transaction level, and one rolled back to a savepoint, restore every row, with the primary key still answering and refusing; rows the parent buffered before the savepoint survive too | +| `test_an_aborted_inner_block_keeps_the_outer_statements_rows` | an outer `INSERT` and an outer `DELETE` whose expression calls a function with an aborted EXCEPTION block keep heap's answer. Before #1341's fix the INSERT kept 1 row of 5000. The same loss through a read inside the block is #1360 | +| `test_an_insert_racing_an_uncommitted_delete_of_its_key` | an insert of a key whose delete (or key-changing update) is uncommitted waits, then fails with 23505 if the delete rolls back and succeeds if it commits: never two rows with the key. Before #1341's fix the rollback left two | +| `test_an_index_built_after_a_delete_serves_an_older_snapshot` | a REPEATABLE READ snapshot from before a committed DELETE still finds the row through an index built after it (an Index Scan, asserted), and a new snapshot does not; a UNIQUE index still builds over a key whose other holder is deleted. Before #1341's fix the old snapshot's lookup found 0 | diff --git a/test/pytest/expected_tests.txt b/test/pytest/expected_tests.txt index 2cb5a0c97..02b56ef40 100644 --- a/test/pytest/expected_tests.txt +++ b/test/pytest/expected_tests.txt @@ -653,4 +653,7 @@ guard_tests 433 # test_ttl_expire.py, by collection over the cluster files on this tree. # cluster_tests 550 -> 551 for #1341's test_entry_point_privilege.py, the twin of # entry_point_privilege.sh: one test, by collection over the cluster files on this tree. -cluster_tests 551 +# +# cluster_tests 551 -> 558 for #1341's test_am_contracts.py, the twin of +# test/am_contracts.sh: seven tests, by collection over the cluster files on this tree. +cluster_tests 558 diff --git a/test/pytest/test_am_contracts.py b/test/pytest/test_am_contracts.py new file mode 100644 index 000000000..8ab47c2d4 --- /dev/null +++ b/test/pytest/test_am_contracts.py @@ -0,0 +1,307 @@ +"""Contracts the access method owes core DML and transactions (#1341). + +Each is stated in docs/ or owed by core's table-AM interface, and none was tested. +Independent of test/am_contracts.sh: same public seam, own fixtures. Check names +match the shell twin so the two compare by name. +""" + +import re +import threading +import time + +import psycopg + + +def _state(conn, sql): + """-> "ok", or the SQLSTATE the statement raised. A refusal is decided by its code.""" + try: + with conn.cursor() as cur: + cur.execute(sql) + return "ok" + except psycopg.Error as e: + return e.sqlstate + + +def _one(conn, sql): + with conn.cursor() as cur: + cur.execute(sql) + return cur.fetchone()[0] + + +def test_insert_on_conflict(pgc_conn, expect): + """docs/limitations.md:655-656: DO NOTHING works, DO UPDATE raises the row-lock error. + + On a table with a primary key, which is the index ON CONFLICT arbitrates on. + """ + with pgc_conn.cursor() as cur: + cur.execute("CREATE TABLE oc (id int PRIMARY KEY, v text) USING pgcolumnar") + cur.execute("INSERT INTO oc SELECT g, 'v' || g FROM generate_series(1, 10) g") + expect.num(_one(pgc_conn, "SELECT count(*) FROM oc"), 10, + "premise: the conflict table holds ten rows") + after = "SELECT count(*) || '|' || (SELECT v FROM oc WHERE id = {k}) FROM oc" + got = _state(pgc_conn, "INSERT INTO oc VALUES (5, 'dup'), (11, 'new') ON CONFLICT DO NOTHING") + expect.text(f"{got}|{_one(pgc_conn, after.format(k=5))}", "ok|11|v5", + "ON CONFLICT DO NOTHING skips the duplicate and inserts the new row") + got = _state(pgc_conn, "INSERT INTO oc VALUES (6, 'dup'), (12, 'new') ON CONFLICT (id) DO NOTHING") + expect.text(f"{got}|{_one(pgc_conn, after.format(k=6))}", "ok|12|v6", + "and so does ON CONFLICT (id) DO NOTHING, naming its arbiter") + got = _state(pgc_conn, "INSERT INTO oc VALUES (7, 'upd') ON CONFLICT (id) DO UPDATE SET v = excluded.v") + expect.text(f"{got}|{_one(pgc_conn, after.format(k=7))}", "0A000|12|v7", + "ON CONFLICT DO UPDATE is refused as unsupported (0A000), and changes nothing") + got = _state(pgc_conn, "INSERT INTO oc VALUES (13, 'upd') ON CONFLICT (id) DO UPDATE SET v = excluded.v") + expect.text(f"{got}|{_one(pgc_conn, after.format(k=13))}", "ok|13|upd", + "ON CONFLICT DO UPDATE with no conflicting row inserts it") + got = _state(pgc_conn, "INSERT INTO oc VALUES (30, 'first'), (30, 'second') ON CONFLICT DO NOTHING") + twice = _one(pgc_conn, "SELECT count(*) || '|' || min(v) FROM oc WHERE id = 30") + expect.text(f"{got}|{twice}", "ok|1|first", + "a key twice in one DO NOTHING statement is inserted once, the first time") + + +def test_insert_on_conflict_withdraws_a_row_a_concurrent_insert_beat(pgc_cluster, pgc_conn, expect): + """The withdrawal path: a row free at the pre-check that conflicts once inserted. + + A second connection holds key 20 uncommitted. DO NOTHING on 20 and 21 waits, then + must leave exactly one row 20 -- the holder's -- and its own 21. + """ + with pgc_conn.cursor() as cur: + cur.execute("CREATE TABLE ocr (id int PRIMARY KEY, v text) USING pgcolumnar") + cur.execute("INSERT INTO ocr SELECT g, 'v' || g FROM generate_series(1, 10) g") + cur.execute("SELECT current_schema()") + schema = cur.fetchone()[0] + holder = psycopg.connect(pgc_cluster.dsn(), autocommit=False) + try: + with holder.cursor() as hc: + hc.execute(f'SET search_path TO "{schema}"') + hc.execute("INSERT INTO ocr VALUES (20, 'holder')") + timer = threading.Timer(2.0, holder.commit) + timer.start() + started = time.monotonic() + got = _state(pgc_conn, "INSERT INTO ocr VALUES (20, 'racer'), (21, 'racer') ON CONFLICT DO NOTHING") + waited = time.monotonic() - started + timer.join() + finally: + holder.close() + expect.at_least(int(waited * 10), 10, "premise: the racer waited for the holder to commit") + expect.text(f"{got}|" + _one(pgc_conn, "SELECT count(*) FILTER (WHERE id = 20) || '|' || " + "min(v) FILTER (WHERE id = 20) || '|' || " + "count(*) FILTER (WHERE id = 21) FROM ocr"), + "ok|1|holder|1", + "a DO NOTHING racing an uncommitted insert of the same key waits, then skips it") + with pgc_conn.cursor() as cur: + cur.execute("SET enable_seqscan = off") + cur.execute("SELECT count(*) || '|' || min(v) FROM ocr WHERE id = 20") + idx = cur.fetchone()[0] + cur.execute("RESET enable_seqscan") + expect.text(idx, "1|holder", "and the unique index answers the key with that one row") + + +def test_insert_on_conflict_racing_sessions_leave_each_key_once(pgc_cluster, pgc_conn, expect): + """The withdrawal itself, which a two-session arm does not reach. + + A row is withdrawn only when another session inserts the key after this one's + pre-check and before its own insert, and no ordering of two sessions can force + that. So eight sessions race on clock-derived keys for a few seconds. Measured: + 0 duplicate keys on three runs; with the withdrawal disabled, 407 to 507. + """ + with pgc_conn.cursor() as cur: + cur.execute("CREATE TABLE ocs (id bigint PRIMARY KEY, w int) USING pgcolumnar") + cur.execute("SELECT current_schema()") + schema = cur.fetchone()[0] + errors, attempts, stop = [], [], time.monotonic() + 5 + + def worker(): + n = 0 + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as c: + c.execute(f'SET search_path TO "{schema}"') + while time.monotonic() < stop: + try: + c.execute("INSERT INTO ocs VALUES ((extract(epoch FROM clock_timestamp()) * 200)::bigint, 1)" + " ON CONFLICT DO NOTHING") + except psycopg.Error as e: + errors.append(e.sqlstate) + n += 1 + attempts.append(n) + + threads = [threading.Thread(target=worker) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + rows = _one(pgc_conn, "SELECT count(*) FROM ocs") + expect.text(str(rows < sum(attempts)).lower(), "true", + "premise: the racing sessions contended for keys") + dup = _one(pgc_conn, "SELECT count(*) - count(DISTINCT id) FROM ocs") + expect.text(f"{dup}|{len(errors)}", "0|0", + "sessions racing DO NOTHING on the same keys leave each key once, with no error") + + +def test_a_truncate_that_rolls_back_restores_every_row(pgc_conn, expect): + """TRUNCATE swaps the table onto new storage, so the old storage's rows and catalog + entries must come back whole if the swap does not commit (#1341).""" + with pgc_conn.cursor() as cur: + cur.execute("CREATE TABLE tr (id int PRIMARY KEY, v int) USING pgcolumnar") + cur.execute("SELECT pgcolumnar.set_options('tr', stripe_row_limit => 1000)") + cur.execute("INSERT INTO tr SELECT g, g * 3 FROM generate_series(1, 5000) g") + shape = "SELECT count(*) || '|' || sum(v) FROM tr" + before = _one(pgc_conn, shape) + expect.text(before, "5000|37507500", "premise: the truncate table holds 5000 rows") + # psycopg rolls the block back when psycopg.Rollback is raised inside it. + with pgc_conn.transaction(): + with pgc_conn.cursor() as cur: + cur.execute("TRUNCATE tr") + raise psycopg.Rollback() + expect.text(_one(pgc_conn, shape), before, + "a TRUNCATE rolled back at transaction level restores every row") + with pgc_conn.cursor() as cur: + cur.execute("SET enable_seqscan = off") + cur.execute("SELECT v FROM tr WHERE id = 4321") + by_index = cur.fetchone()[0] + cur.execute("RESET enable_seqscan") + expect.num(by_index, 12963, "and the primary key still finds them by index") + with pgc_conn.transaction(): + with pgc_conn.cursor() as cur: + cur.execute("SAVEPOINT s") + cur.execute("TRUNCATE tr") + cur.execute("ROLLBACK TO SAVEPOINT s") + cur.execute("INSERT INTO tr VALUES (5001, 1)") + expect.text(_one(pgc_conn, shape), "5001|37507501", + "a TRUNCATE rolled back to a savepoint restores every row, and the transaction goes on") + expect.text(_state(pgc_conn, "INSERT INTO tr VALUES (17, 0)"), "23505", + "and the primary key still refuses a duplicate of a restored row") + # And the rows the parent buffered before the savepoint: TRUNCATE discards the + # cached write state, and they may still be in it, unflushed. + with pgc_conn.transaction(): + with pgc_conn.cursor() as cur: + cur.execute("INSERT INTO tr SELECT g, 1 FROM generate_series(6001, 8500) g") + cur.execute("SAVEPOINT s") + cur.execute("TRUNCATE tr") + cur.execute("INSERT INTO tr SELECT g, 2 FROM generate_series(1, 5) g") + cur.execute("ROLLBACK TO SAVEPOINT s") + expect.text(_one(pgc_conn, shape), "7501|37510001", + "rows the parent buffered before a savepoint survive a TRUNCATE rolled back to it") + + +def test_an_aborted_inner_block_keeps_the_outer_statements_rows(pgc_conn, expect): + """A plpgsql EXCEPTION block is a subtransaction, and a statement inside it flushes + every pending write at its ExecutorEnd -- the outer statement's too. If that flush + belongs to the inner subtransaction, the block's rollback takes the outer rows with + it (#1341). Each arm runs on heap as well; the columnar answer must be heap's.""" + with pgc_conn.cursor() as cur: + cur.execute("CREATE TABLE sx_side (id int) USING pgcolumnar") + cur.execute("""CREATE FUNCTION sx_f(i int) RETURNS int LANGUAGE plpgsql AS $$ +BEGIN + BEGIN + INSERT INTO sx_side VALUES (i); + PERFORM 1 / 0; + EXCEPTION WHEN division_by_zero THEN NULL; + END; + RETURN i; +END $$""") + for am in ("pgcolumnar", "heap"): + cur.execute(f"CREATE TABLE sx_{am} (id int) USING {am}") + cur.execute(f"INSERT INTO sx_{am} SELECT sx_f(g) FROM generate_series(1, 5000) g") + shape = "SELECT count(*) || '|' || sum(id) FROM sx_{}" + expect.text(_one(pgc_conn, shape.format("pgcolumnar")), _one(pgc_conn, shape.format("heap")), + "an outer insert survives an aborted inner block, as on heap") + expect.num(_one(pgc_conn, "SELECT count(*) FROM sx_side"), 0, + "and the inner block's own rows are gone") + with pgc_conn.cursor() as cur: + for am in ("pgcolumnar", "heap"): + cur.execute(f"DELETE FROM sx_{am} WHERE id % 2 = 0 AND sx_f(id) IS NOT NULL") + expect.text(_one(pgc_conn, shape.format("pgcolumnar")), _one(pgc_conn, shape.format("heap")), + "an outer delete survives an aborted inner block, as on heap") + + +def test_an_insert_racing_an_uncommitted_delete_of_its_key(pgc_cluster, pgc_conn, expect): + """Session 1 removes key k and has not committed; session 2 inserts k (#1341). + + On heap session 2 waits for session 1: a rollback leaves k taken (23505), a commit + frees it. One row with k either way, never two. + """ + with pgc_conn.cursor() as cur: + cur.execute("CREATE TABLE ud (k int UNIQUE, v text) USING pgcolumnar") + cur.execute("INSERT INTO ud VALUES (5, 'a'), (6, 'b'), (7, 'c')") + cur.execute("SELECT current_schema()") + schema = cur.fetchone()[0] + + def race(hold, end, insert): + holder = psycopg.connect(pgc_cluster.dsn(), autocommit=False) + try: + with holder.cursor() as hc: + hc.execute(f'SET search_path TO "{schema}"') + hc.execute(hold) + timer = threading.Timer(2.0, getattr(holder, end)) + timer.start() + got = _state(pgc_conn, insert) + timer.join() + finally: + holder.close() + return got + + row = "SELECT count(*) || '|' || min(v) FROM ud WHERE k = {}" + got = race("DELETE FROM ud WHERE k = 5", "rollback", "INSERT INTO ud VALUES (5, 'racer')") + expect.text(f"{got}|{_one(pgc_conn, row.format(5))}", "23505|1|a", + "an insert of a key whose delete is uncommitted waits, and fails when the delete rolls back") + got = race("DELETE FROM ud WHERE k = 6", "commit", "INSERT INTO ud VALUES (6, 'racer')") + expect.text(f"{got}|{_one(pgc_conn, row.format(6))}", "ok|1|racer", + "and succeeds when the delete commits, leaving one row with the key") + got = race("UPDATE ud SET k = 70 WHERE k = 7", "rollback", "INSERT INTO ud VALUES (7, 'racer')") + expect.text(f"{got}|{_one(pgc_conn, row.format(7))}", "23505|1|c", + "an insert of a key an uncommitted UPDATE moved away fails when the update rolls back") + + +def test_an_index_built_after_a_delete_serves_an_older_snapshot(pgc_cluster, pgc_conn, expect): + """Session 1 holds a REPEATABLE READ snapshot from before a committed DELETE of row + 4242; then an index is built. Session 1 still sees the row, so an index scan must + find it, as on heap, which indexes rows not yet dead to every snapshot (#1341).""" + with pgc_conn.cursor() as cur: + cur.execute("CREATE TABLE ib (id int, v int) USING pgcolumnar") + cur.execute("INSERT INTO ib SELECT g, g FROM generate_series(1, 10000) g") + cur.execute("SELECT current_schema()") + schema = cur.fetchone()[0] + old = psycopg.connect(pgc_cluster.dsn(), autocommit=False) + try: + with old.cursor() as oc: + oc.execute(f'SET search_path TO "{schema}"') + oc.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ") + oc.execute("SELECT count(*) FROM ib") + with pgc_conn.cursor() as cur: + cur.execute("DELETE FROM ib WHERE id = 4242") + cur.execute("CREATE INDEX ib_id ON ib (id)") + oc.execute("SET LOCAL enable_seqscan = off") + oc.execute("SET LOCAL enable_bitmapscan = off") + oc.execute("EXPLAIN (COSTS OFF) SELECT count(*) FROM ib WHERE id = 4242") + plan = "\n".join(r[0] for r in oc.fetchall()) + oc.execute("SELECT count(*) FROM ib WHERE id = 4242") + seen = oc.fetchone()[0] + old.rollback() + finally: + old.close() + expect.text(str(_one(pgc_conn, "SELECT indisvalid FROM pg_index WHERE indexrelid = 'ib_id'::regclass")), + "True", "premise: the index exists and is valid") + expect.num(len(re.findall(r"Index (Only )?Scan using ib_id", plan)), 1, + "premise: the older snapshot's lookup is an index scan") + expect.num(seen, 1, "an index built after a delete finds the row for an older snapshot that sees it") + with pgc_conn.cursor() as cur: + cur.execute("SET enable_seqscan = off") + cur.execute("SET enable_bitmapscan = off") + cur.execute("SELECT count(*) FROM ib WHERE id = 4242") + now = cur.fetchone()[0] + cur.execute("RESET enable_seqscan") + cur.execute("RESET enable_bitmapscan") + expect.num(now, 0, "and a new snapshot does not see the deleted row by that index") + + # The cost of indexing deleted rows, pinned: they go in as not alive, so a UNIQUE + # index still builds over a key whose only other holder is a deleted row. + with pgc_conn.cursor() as cur: + cur.execute("CREATE TABLE ibu (k int) USING pgcolumnar") + cur.execute("INSERT INTO ibu VALUES (5), (6)") + cur.execute("DELETE FROM ibu WHERE k = 5") + cur.execute("INSERT INTO ibu VALUES (5)") + built = _state(pgc_conn, "CREATE UNIQUE INDEX ibu_k ON ibu (k)") + with pgc_conn.cursor() as cur: + cur.execute("SET enable_seqscan = off") + cur.execute("SELECT count(*) FROM ibu WHERE k = 5") + five = cur.fetchone()[0] + cur.execute("RESET enable_seqscan") + expect.text(f"{built}|{five}", "ok|1", "a unique index builds over a key whose other holder is a deleted row") diff --git a/test/pytest/test_compare_to_bash.py b/test/pytest/test_compare_to_bash.py index d705ec243..7860fdaca 100644 --- a/test/pytest/test_compare_to_bash.py +++ b/test/pytest/test_compare_to_bash.py @@ -87,6 +87,7 @@ # directions for the same reason: a one-way list rots into a permanent exemption. COMPLETE = ["advisory_lock_class", "fdw_plan_privilege", + "am_contracts", "analyze_differential", "analyze_function", "analyze_reltuples", diff --git a/test/run_all_versions.sh b/test/run_all_versions.sh index 9171883be..55e8b31b1 100755 --- a/test/run_all_versions.sh +++ b/test/run_all_versions.sh @@ -42,6 +42,7 @@ SUITES=( advisory_lock_class alter_am_cleanup alter_column_type + am_contracts analyze_differential analyze_function analyze_reltuples