From 7aa2110485b4218276dfa9eff6e7129ae8d69110 Mon Sep 17 00:00:00 2001 From: Hannu Krosing Date: Sun, 23 Aug 2026 20:13:04 +0000 Subject: [PATCH v4 4/9] Support Direct TOAST in logical decoding, replication, and online REPACK Add logical replication and decoding support for Direct TOAST tuples: - reorderbuffer.c: Support direct TOAST chunk reconstruction by keying the toast hash table on tuple TID (via a union key) when chunk_id is InvalidOid. Reconstruct hierarchical direct TOAST DAGs bottom-up from the WAL change stream. - decode.c: Record the tuple's physical TID in decode_heap() when processing toast inserts. - proto.c & pgoutput.c: Correctly handle unchanged direct TOAST attributes in the logical replication output plugin. - Add isolation tests using injection points (repack_direct_toast.spec). --- src/backend/commands/repack.c | 82 ++++++- src/backend/replication/logical/decode.c | 7 +- src/backend/replication/logical/proto.c | 2 +- .../replication/logical/reorderbuffer.c | 208 ++++++++++++++---- src/backend/replication/pgoutput/pgoutput.c | 4 +- src/backend/replication/pgrepack/pgrepack.c | 2 +- src/test/modules/injection_points/Makefile | 1 + .../expected/repack_direct_toast.out | 126 +++++++++++ src/test/modules/injection_points/meson.build | 1 + .../specs/repack_direct_toast.spec | 180 +++++++++++++++ 10 files changed, 570 insertions(+), 43 deletions(-) create mode 100644 src/test/modules/injection_points/expected/repack_direct_toast.out create mode 100644 src/test/modules/injection_points/specs/repack_direct_toast.spec diff --git a/src/backend/commands/repack.c b/src/backend/commands/repack.c index 57c156e6805..d881b2d5db0 100644 --- a/src/backend/commands/repack.c +++ b/src/backend/commands/repack.c @@ -37,6 +37,7 @@ #include "access/multixact.h" #include "access/relscan.h" #include "access/tableam.h" +#include "access/genam.h" #include "access/toast_internals.h" #include "access/transam.h" #include "access/xact.h" @@ -45,6 +46,7 @@ #include "catalog/dependency.h" #include "catalog/heap.h" #include "catalog/index.h" +#include "catalog/pg_depend.h" #include "catalog/namespace.h" #include "catalog/objectaccess.h" #include "catalog/pg_am.h" @@ -498,6 +500,54 @@ RepackLockLevel(bool concurrent) * 'cmd' indicates which command is being executed, to be used for error * messages. */ +/* + * Find the parent relation OID of a TOAST table from pg_depend. + */ +static Oid +toast_get_parent_relid(Oid toastrelid) +{ + Relation depRel; + ScanKeyData key[3]; + SysScanDesc scan; + HeapTuple depTuple; + Oid parentrelid = InvalidOid; + + depRel = table_open(DependRelationId, AccessShareLock); + + ScanKeyInit(&key[0], + Anum_pg_depend_classid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(RelationRelationId)); + ScanKeyInit(&key[1], + Anum_pg_depend_objid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(toastrelid)); + ScanKeyInit(&key[2], + Anum_pg_depend_objsubid, + BTEqualStrategyNumber, F_INT4EQ, + Int32GetDatum(0)); + + scan = systable_beginscan(depRel, DependDependerIndexId, true, + NULL, 3, key); + + while (HeapTupleIsValid(depTuple = systable_getnext(scan))) + { + Form_pg_depend dep = (Form_pg_depend) GETSTRUCT(depTuple); + + if (dep->refclassid == RelationRelationId && + dep->deptype == DEPENDENCY_INTERNAL) + { + parentrelid = dep->refobjid; + break; + } + } + + systable_endscan(scan); + table_close(depRel, AccessShareLock); + + return parentrelid; +} + void cluster_rel(RepackCommand cmd, Relation OldHeap, Oid indexOid, ClusterParams *params, bool isTopLevel) @@ -647,6 +697,36 @@ cluster_rel(RepackCommand cmd, Relation OldHeap, Oid indexOid, goto out; } + + if (OldHeap->rd_rel->relkind == RELKIND_TOASTVALUE) + { + /* + * If this TOAST table's parent uses Direct TOAST, disallow VACUUM FULL, + * CLUSTER, or REPACK directly on the TOAST table. Rebuilding the TOAST + * table independently would invalidate the physical TIDs stored in the + * parent relation's tuples. + */ + Oid parentrelid = toast_get_parent_relid(RelationGetRelid(OldHeap)); + + if (OidIsValid(parentrelid)) + { + Relation parentrel = relation_open(parentrelid, AccessShareLock); + bool is_direct = (RelationGetToastFlavour(parentrel) == TOAST_FLAVOUR_DIRECT); + + relation_close(parentrel, AccessShareLock); + + if (is_direct) + { + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("cannot %s direct TOAST table directly", + RepackCommandAsString(cmd)), + errhint("Execute %s on the parent table instead.", + RepackCommandAsString(cmd)))); + } + } + } + Assert(OldHeap->rd_rel->relkind == RELKIND_RELATION || OldHeap->rd_rel->relkind == RELKIND_MATVIEW || OldHeap->rd_rel->relkind == RELKIND_TOASTVALUE); @@ -2992,7 +3072,7 @@ adjust_toast_pointers(Relation relation, TupleTableSlot *dest, TupleTableSlot *s slot_getsomeattrs(dest, i + 1); varlena_dst = (varlena *) DatumGetPointer(dest->tts_values[i]); - if (!VARATT_IS_EXTERNAL_ONDISK(varlena_dst)) + if (!VARATT_IS_EXTERNAL_ONDISK(varlena_dst) && !VARATT_IS_EXTERNAL_DIRECT(varlena_dst)) continue; slot_getsomeattrs(src, i + 1); diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index 4a739230264..482937af04b 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -921,6 +921,7 @@ DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) xl_heap_insert *xlrec; ReorderBufferChange *change; RelFileLocator target_locator; + BlockNumber blknum; xlrec = (xl_heap_insert *) XLogRecGetData(r); @@ -932,7 +933,7 @@ DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) return; /* only interested in our database */ - XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL); + XLogRecGetBlockTag(r, 0, &target_locator, NULL, &blknum); if (target_locator.dbOid != ctx->slot->data.database) return; @@ -957,6 +958,10 @@ DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple); + /* Set the TID for toast relation inserts, needed for direct toast */ + if (xlrec->flags & XLH_INSERT_ON_TOAST_RELATION) + ItemPointerSet(&change->data.tp.newtuple->t_self, blknum, xlrec->offnum); + change->data.tp.clear_toast_afterwards = true; ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c index 86ad97cd937..50f29523499 100644 --- a/src/backend/replication/logical/proto.c +++ b/src/backend/replication/logical/proto.c @@ -812,7 +812,7 @@ logicalrep_write_tuple(StringInfo out, Relation rel, TupleTableSlot *slot, continue; } - if (att->attlen == -1 && VARATT_IS_EXTERNAL_ONDISK(DatumGetPointer(values[i]))) + if (att->attlen == -1 && VARATT_IS_EXTERNAL_ONDISK_OR_DIRECT(DatumGetPointer(values[i]))) { /* * Unchanged toasted datum. (Note that we don't promise to detect diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index ede117d339e..025d7562cee 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -110,6 +110,9 @@ #include "storage/sinval.h" #include "utils/builtins.h" #include "utils/inval.h" +#include "utils/array.h" +#include "catalog/pg_type.h" +#include "utils/lsyscache.h" #include "utils/memutils.h" #include "utils/rel.h" #include "utils/relfilenumbermap.h" @@ -174,10 +177,21 @@ typedef struct ReorderBufferIterTXNState ReorderBufferIterTXNEntry entries[FLEXIBLE_ARRAY_MEMBER]; } ReorderBufferIterTXNState; +typedef struct ReorderBufferToastKey +{ + bool is_direct; + union + { + Oid8 chunk_id; + ItemPointerData tid; + } u; +} ReorderBufferToastKey; + /* toast datastructures */ typedef struct ReorderBufferToastEnt { - Oid8 chunk_id; /* toast_table.chunk_id */ + ReorderBufferToastKey key; /* toast_table.chunk_id or chunk TID */ + Oid8 chunk_id; /* toast_table.chunk_id (kept for compatibility) */ int32 last_chunk_seq; /* toast_table.chunk_seq of the last chunk we * have seen */ Size num_chunks; /* number of chunks we've already seen */ @@ -5010,7 +5024,7 @@ ReorderBufferToastInitHash(ReorderBuffer *rb, ReorderBufferTXN *txn) Assert(txn->toast_hash == NULL); - hash_ctl.keysize = sizeof(Oid8); + hash_ctl.keysize = sizeof(ReorderBufferToastKey); hash_ctl.entrysize = sizeof(ReorderBufferToastEnt); hash_ctl.hcxt = rb->context; txn->toast_hash = hash_create("ReorderBufferToastHash", 5, &hash_ctl, @@ -5034,9 +5048,12 @@ ReorderBufferToastAppendChunk(ReorderBuffer *rb, ReorderBufferTXN *txn, bool isnull; Pointer chunk; TupleDesc desc = RelationGetDescr(relation); - Oid8 chunk_id; + Oid8 chunk_id = 0; int32 chunk_seq; - Oid valueid_type; + ReorderBufferToastKey key; + bool is_direct = false; + Datum chunk_tids_datum; + bool chunk_tids_isnull = true; Datum valueid_datum; if (txn->toast_hash == NULL) @@ -5045,52 +5062,146 @@ ReorderBufferToastAppendChunk(ReorderBuffer *rb, ReorderBufferTXN *txn, Assert(IsToastRelation(relation)); newtup = change->data.tp.newtuple; - valueid_type = TupleDescAttr(desc, 0)->atttypid; valueid_datum = fastgetattr(newtup, 1, desc, &isnull); - if (valueid_type == OID8OID) - chunk_id = DatumGetObjectId8(valueid_datum); + if (isnull) + { + is_direct = true; + memset(&key, 0, sizeof(key)); + key.is_direct = true; + key.u.tid = newtup->t_self; + if (!ItemPointerIsValid(&key.u.tid)) + elog(ERROR, "invalid TID for direct toast chunk"); + } else - chunk_id = DatumGetObjectId(valueid_datum); - Assert(!isnull); + { + Oid valueid_type = TupleDescAttr(desc, 0)->atttypid; + + if (valueid_type == OID8OID) + chunk_id = DatumGetObjectId8(valueid_datum); + else + chunk_id = DatumGetObjectId(valueid_datum); + + memset(&key, 0, sizeof(key)); + key.is_direct = false; + key.u.chunk_id = chunk_id; + } chunk_seq = DatumGetInt32(fastgetattr(newtup, 2, desc, &isnull)); Assert(!isnull); + if (desc->natts >= 4) + { + chunk_tids_datum = fastgetattr(newtup, 4, desc, &chunk_tids_isnull); + } + ent = (ReorderBufferToastEnt *) - hash_search(txn->toast_hash, &chunk_id, HASH_ENTER, &found); + hash_search(txn->toast_hash, &key, HASH_ENTER, &found); + + chunk = DatumGetPointer(fastgetattr(newtup, 3, desc, &isnull)); + if (!isnull) + { + /* calculate size so we can allocate the right size at once later */ + if (!VARATT_IS_EXTENDED(chunk)) + chunksize = VARSIZE(chunk) - VARHDRSZ; + else if (VARATT_IS_SHORT(chunk)) + /* could happen due to heap_form_tuple doing its thing */ + chunksize = VARSIZE_SHORT(chunk) - VARHDRSZ_SHORT; + else + elog(ERROR, "unexpected type of toast chunk"); + } + else + { + chunksize = 0; + } if (!found) { - Assert(ent->chunk_id == chunk_id); + ent->key = key; + ent->chunk_id = is_direct ? 0 : chunk_id; ent->num_chunks = 0; ent->last_chunk_seq = 0; ent->size = 0; ent->reconstructed = NULL; dlist_init(&ent->chunks); - if (chunk_seq != 0) + if (!is_direct && chunk_seq != 0) elog(ERROR, "got sequence entry %d for toast chunk " OID8_FORMAT " instead of seq 0", chunk_seq, chunk_id); } - else if (found && chunk_seq != ent->last_chunk_seq + 1) - elog(ERROR, "got sequence entry %d for toast chunk " OID8_FORMAT " instead of seq %d", - chunk_seq, chunk_id, ent->last_chunk_seq + 1); + else if (found && !is_direct) + { + if (chunk_seq != ent->last_chunk_seq + 1) + elog(ERROR, "got sequence entry %d for toast chunk " OID8_FORMAT " instead of seq %d", + chunk_seq, chunk_id, ent->last_chunk_seq + 1); + } + else if (found && is_direct) + { + elog(ERROR, "duplicate TID in direct toast hash"); + } - chunk = DatumGetPointer(fastgetattr(newtup, 3, desc, &isnull)); - Assert(!isnull); + /* Group previous chunks if this is a node with chunk_tids */ + if (is_direct && !chunk_tids_isnull) + { + ArrayType *arr = DatumGetArrayTypeP(chunk_tids_datum); + Oid eltype = ARR_ELEMTYPE(arr); + int16 typlen; + bool typbyval; + char typalign; + Datum *elems; + bool *nulls; + int nelems; + int i; + + Assert(eltype == TIDOID); + get_typlenbyvalalign(eltype, &typlen, &typbyval, &typalign); + deconstruct_array(arr, eltype, typlen, typbyval, typalign, + &elems, &nulls, &nelems); + + for (i = 0; i < nelems; i++) + { + ItemPointer tid = DatumGetItemPointer(elems[i]); + ReorderBufferToastKey prev_key; + ReorderBufferToastEnt *ent_prev; + dlist_mutable_iter miter; - /* calculate size so we can allocate the right size at once later */ - if (!VARATT_IS_EXTENDED(chunk)) - chunksize = VARSIZE(chunk) - VARHDRSZ; - else if (VARATT_IS_SHORT(chunk)) - /* could happen due to heap_form_tuple doing its thing */ - chunksize = VARSIZE_SHORT(chunk) - VARHDRSZ_SHORT; - else - elog(ERROR, "unexpected type of toast chunk"); + if (ItemPointerEquals(tid, &key.u.tid)) + continue; + + memset(&prev_key, 0, sizeof(prev_key)); + prev_key.is_direct = true; + prev_key.u.tid = *tid; + + ent_prev = (ReorderBufferToastEnt *) + hash_search(txn->toast_hash, &prev_key, HASH_FIND, NULL); + + if (ent_prev == NULL) + elog(ERROR, "could not find previous direct toast chunk"); + + /* Move chunks to the current entry */ + dlist_foreach_modify(miter, &ent_prev->chunks) + { + ReorderBufferChange *c = dlist_container(ReorderBufferChange, node, miter.cur); + dlist_delete(miter.cur); + dlist_push_tail(&ent->chunks, &c->node); + } + + ent->size += ent_prev->size; + ent->num_chunks += ent_prev->num_chunks; + + /* Remove from hash */ + hash_search(txn->toast_hash, &prev_key, HASH_REMOVE, NULL); + } + + pfree(elems); + pfree(nulls); + } - ent->size += chunksize; ent->last_chunk_seq = chunk_seq; - ent->num_chunks++; - dlist_push_tail(&ent->chunks, &change->node); + if (chunksize > 0) + { + ent->size += chunksize; + ent->num_chunks++; + dlist_push_tail(&ent->chunks, &change->node); + } } /* @@ -5181,7 +5292,10 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn, varlena *reconstructed; dlist_iter it; Size data_done = 0; - Oid8 toast_valueid; + ReorderBufferToastKey key; + int32 rawsize; + uint32 extsize; + bool is_compressed; if (attr->attisdropped) continue; @@ -5197,19 +5311,39 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn, /* ok, we know we have a toast datum */ varlena_pointer = (varlena *) DatumGetPointer(attrs[natt]); - /* no need to do anything if the tuple isn't external */ - if (!VARATT_IS_EXTERNAL_ONDISK(varlena_pointer)) + /* no need to do anything if the tuple isn't external ondisk or direct */ + if (!VARATT_IS_EXTERNAL_ONDISK(varlena_pointer) && + !VARATT_IS_EXTERNAL_DIRECT(varlena_pointer)) continue; - toast_external_info_get(varlena_pointer, &toast_ext_data); - toast_valueid = toast_ext_data.valueid; + memset(&key, 0, sizeof(key)); + if (VARATT_IS_EXTERNAL_DIRECT(varlena_pointer)) + { + varatt_direct toast_pointer_direct; + + VARATT_EXTERNAL_GET_POINTER_DIRECT(toast_pointer_direct, varlena_pointer); + key.is_direct = true; + key.u.tid = toast_pointer_direct.va_tid; + rawsize = toast_pointer_direct.va_rawsize; + extsize = VARATT_DIRECT_GET_EXTSIZE(toast_pointer_direct); + is_compressed = VARATT_DIRECT_IS_COMPRESSED(toast_pointer_direct); + } + else + { + toast_external_info_get(varlena_pointer, &toast_ext_data); + key.is_direct = false; + key.u.chunk_id = toast_ext_data.valueid; + rawsize = toast_ext_data.rawsize; + extsize = VARATT_EXTINFO_GET_EXTSIZE(toast_ext_data.extinfo); + is_compressed = VARATT_EXTINFO_IS_COMPRESSED(toast_ext_data.extinfo, toast_ext_data.rawsize); + } /* * Check whether the toast tuple changed, replace if so. */ ent = (ReorderBufferToastEnt *) hash_search(txn->toast_hash, - &toast_valueid, + &key, HASH_FIND, NULL); if (ent == NULL) @@ -5220,7 +5354,7 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn, free[natt] = true; - reconstructed = palloc0(toast_ext_data.rawsize); + reconstructed = palloc0(rawsize); ent->reconstructed = reconstructed; @@ -5245,10 +5379,10 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn, VARSIZE(chunk) - VARHDRSZ); data_done += VARSIZE(chunk) - VARHDRSZ; } - Assert(data_done == VARATT_EXTINFO_GET_EXTSIZE(toast_ext_data.extinfo)); + Assert(data_done == extsize); /* make sure its marked as compressed or not */ - if (VARATT_EXTINFO_IS_COMPRESSED(toast_ext_data.extinfo, toast_ext_data.rawsize)) + if (is_compressed) SET_VARSIZE_COMPRESSED(reconstructed, data_done + VARHDRSZ); else SET_VARSIZE(reconstructed, data_done + VARHDRSZ); diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 484ffbe2cee..3703c47148c 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -1402,8 +1402,8 @@ pgoutput_row_filter(Relation relation, TupleTableSlot *old_slot, * VARTAG_INDIRECT. See ReorderBufferToastReplace. */ if (att->attlen == -1 && - VARATT_IS_EXTERNAL_ONDISK(DatumGetPointer(new_slot->tts_values[i])) && - !VARATT_IS_EXTERNAL_ONDISK(DatumGetPointer(old_slot->tts_values[i]))) + VARATT_IS_EXTERNAL_ONDISK_OR_DIRECT(DatumGetPointer(new_slot->tts_values[i])) && + !VARATT_IS_EXTERNAL_ONDISK_OR_DIRECT(DatumGetPointer(old_slot->tts_values[i]))) { if (!tmp_new_slot) { diff --git a/src/backend/replication/pgrepack/pgrepack.c b/src/backend/replication/pgrepack/pgrepack.c index a1cbc8db68a..38fb231ad98 100644 --- a/src/backend/replication/pgrepack/pgrepack.c +++ b/src/backend/replication/pgrepack/pgrepack.c @@ -272,7 +272,7 @@ repack_store_change(LogicalDecodingContext *ctx, Relation relation, * We get here if the table has external values but only * in-line values are being updated now. */ - Assert(VARATT_IS_EXTERNAL_ONDISK(varlen)); + Assert(VARATT_IS_EXTERNAL_ONDISK(varlen) || VARATT_IS_EXTERNAL_DIRECT(varlen)); } } diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile index 9d8b4b3540c..46292337475 100644 --- a/src/test/modules/injection_points/Makefile +++ b/src/test/modules/injection_points/Makefile @@ -22,6 +22,7 @@ ISOLATION = basic \ repack_temporal \ repack_temporal_multirange \ repack_toast \ + repack_direct_toast \ ri_fastpath_reindex \ ri_fastpath_snapshot \ syscache-update-pruned \ diff --git a/src/test/modules/injection_points/expected/repack_direct_toast.out b/src/test/modules/injection_points/expected/repack_direct_toast.out new file mode 100644 index 00000000000..aa73c3a24cc --- /dev/null +++ b/src/test/modules/injection_points/expected/repack_direct_toast.out @@ -0,0 +1,126 @@ +Parsed test spec with 2 sessions + +starting permutation: s1_wait_before_lock s2_updates s2_check s2_wakeup_before_lock s1_check +injection_points_attach +----------------------- + +(1 row) + +step s1_wait_before_lock: + REPACK (CONCURRENTLY) repack_direct_toast; + +step s2_updates: + DELETE FROM repack_direct_toast WHERE i=1; + INSERT INTO repack_direct_toast(i, j, k) VALUES (1, gen_external(), gen_compressible(1)); + + -- existing toast data unchanged. (This covers the case where we + -- adjust the toast pointer.) + UPDATE repack_direct_toast SET i=i+300 where i % 10 = 2 RETURNING OLD.i, NEW.i; + + -- "j" is here an external indirect, written to the file separately. + UPDATE repack_direct_toast SET j=gen_external() where i % 10 = 3 RETURNING OLD.i, NEW.i; + + -- the updated value of "j" is compressed. + UPDATE repack_direct_toast SET j=gen_compressible(1), k=k||'' where i % 10 = 4 RETURNING i; + + -- the updated value of "j" is compressed externally. + UPDATE repack_direct_toast SET j=gen_compressible_external(2) where i % 10 = 5 RETURNING i; + + -- the updated value of "j" stays inline. + UPDATE repack_direct_toast SET j=gen_inline(), k=repeat(k,5) where i % 10 = 6 RETURNING i; + + -- updated value of "j" is a short varlena; "k" is written separately. + UPDATE repack_direct_toast SET j=gen_short(), k=gen_external() where i % 10 = 7 RETURNING i; + + i| i +--+--- + 2|302 +12|312 +(2 rows) + + i| i +--+-- + 3| 3 +13|13 +(2 rows) + + i +-- + 4 +14 +(2 rows) + + i +-- + 5 +15 +(2 rows) + + i +-- + 6 +16 +(2 rows) + + i +-- + 7 +17 +(2 rows) + +step s2_check: + INSERT INTO relfilenodes(node) + SELECT c2.relfilenode + FROM pg_class c1 JOIN pg_class c2 ON c2.oid = c1.oid OR c2.oid = c1.reltoastrelid + WHERE c1.relname='repack_direct_toast'; + + INSERT INTO data_s2(i, j, j_toast, k, k_toast) + SELECT i, j, COALESCE(pg_column_toast_chunk_id(j), 0) AS j_toast, + k, COALESCE(pg_column_toast_chunk_id(k), 0) AS k_toast + FROM repack_direct_toast; + +step s2_wakeup_before_lock: + SELECT injection_points_wakeup('repack-concurrently-before-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step s1_wait_before_lock: <... completed> +step s1_check: + INSERT INTO relfilenodes(node) + SELECT c2.relfilenode + FROM pg_class c1 JOIN pg_class c2 ON c2.oid = c1.oid OR c2.oid = c1.reltoastrelid + WHERE c1.relname='repack_direct_toast'; + + SELECT count(DISTINCT node) FROM relfilenodes; + + INSERT INTO data_s1(i, j, j_toast, k, k_toast) + SELECT i, + j, COALESCE(pg_column_toast_chunk_id(j), 0) AS j_toast, + k, COALESCE(pg_column_toast_chunk_id(k), 0) AS k_toast + FROM repack_direct_toast; + + -- this should be empty + SELECT d1.i, substring(d1.j FOR 12) AS d1_j, substring(d1.k FOR 12) AS d1_k, + d2.i, substring(d2.j FOR 12) AS d2_j, substring(d2.k FOR 12) AS d2_k, + d1.j_toast as d1_j_tst, d2.j_toast as d2_j_tst, + d1.k_toast as d1_k_tst, d2.k_toast AS d2_k_tst + FROM data_s1 d1 FULL JOIN data_s2 d2 USING (i, j, k) + WHERE d1.i ISNULL OR d2.i ISNULL; + +count +----- + 4 +(1 row) + +i|d1_j|d1_k|i|d2_j|d2_k|d1_j_tst|d2_j_tst|d1_k_tst|d2_k_tst +-+----+----+-+----+----+--------+--------+--------+-------- +(0 rows) + +injection_points_detach +----------------------- + +(1 row) + diff --git a/src/test/modules/injection_points/meson.build b/src/test/modules/injection_points/meson.build index 80a09f34d78..409d4131135 100644 --- a/src/test/modules/injection_points/meson.build +++ b/src/test/modules/injection_points/meson.build @@ -51,6 +51,7 @@ tests += { 'repack_temporal', 'repack_temporal_multirange', 'repack_toast', + 'repack_direct_toast', 'ri_fastpath_reindex', 'ri_fastpath_snapshot', 'syscache-update-pruned', diff --git a/src/test/modules/injection_points/specs/repack_direct_toast.spec b/src/test/modules/injection_points/specs/repack_direct_toast.spec new file mode 100644 index 00000000000..b67cbd43278 --- /dev/null +++ b/src/test/modules/injection_points/specs/repack_direct_toast.spec @@ -0,0 +1,180 @@ +# REPACK (CONCURRENTLY) with Direct TOAST; +# +# Test handling of Direct TOAST. At the same time, no tuplesort. +setup +{ + + CREATE EXTENSION IF NOT EXISTS injection_points; + + -- Generate text consisting of repeated strings so that it can be + -- compressed easily. + CREATE FUNCTION gen_compressible(seed int) + RETURNS text + LANGUAGE sql IMMUTABLE as $$ + SELECT repeat(md5((seed * 1000)::text), 50); + $$; + + -- Like above, but too big even after compression. + CREATE FUNCTION gen_compressible_external(seed int) + RETURNS text + LANGUAGE sql IMMUTABLE as $$ + SELECT repeat(md5((seed * 1000)::text), 10000); + $$; + + -- Generate a string of random characters that is not likely to be + -- compressed, but is big enough to be stored externally. + CREATE FUNCTION gen_external() + RETURNS text + LANGUAGE sql as $$ + SELECT string_agg(chr(65 + trunc(25 * random())::int), '') + FROM generate_series(1, 2048) s(x); + $$; + + -- Not compressible like above, but small enough to stay in-line. + CREATE FUNCTION gen_inline() + RETURNS text + LANGUAGE sql as $$ + SELECT string_agg(chr(65 + trunc(25 * random())::int), '') + FROM generate_series(1, 1024) s(x); + $$; + + -- A varlena short enough to have a one-byte header. + CREATE FUNCTION gen_short() + RETURNS text + LANGUAGE sql as $$ + SELECT string_agg(chr(65 + trunc(25 * random())::int), '') + FROM generate_series(1, 120) s(x); + $$; + + CREATE TABLE repack_direct_toast(drop1 int, i int PRIMARY KEY, drop2 int, + j text COMPRESSION pglz, k text COMPRESSION pglz) WITH (toast_flavour = 'direct'); + INSERT INTO repack_direct_toast(drop1, i, drop2, j, k) + SELECT 42, gs, 42, gen_external(), gen_compressible(gs) FROM generate_series(1, 10) gs; + ALTER TABLE repack_direct_toast DROP COLUMN drop1, DROP COLUMN drop2; + ALTER TABLE repack_direct_toast ALTER COLUMN k SET COMPRESSION default; + INSERT INTO repack_direct_toast(i, j, k) + SELECT gs, gen_external(), gen_compressible(142857) FROM generate_series(11, 20) gs; + + ALTER TABLE repack_direct_toast SET (toast_tuple_target = 128); + + CREATE TABLE relfilenodes(node oid); + + CREATE TABLE data_s1 (i int, j text, j_toast oid8, k text, k_toast oid8); + CREATE TABLE data_s2 (LIKE data_s1); +} + + +teardown +{ + DROP TABLE repack_direct_toast; + DROP EXTENSION injection_points; + DROP FUNCTION gen_compressible(int); + DROP FUNCTION gen_compressible_external(int); + DROP FUNCTION gen_external(); + DROP FUNCTION gen_inline(); + DROP FUNCTION gen_short(); + + DROP TABLE relfilenodes; + DROP TABLE data_s1; + DROP TABLE data_s2; +} + +session s1 +setup +{ + SELECT injection_points_set_local(); + SELECT injection_points_attach('repack-concurrently-before-lock', 'wait'); +} + +# Perform the initial load and wait for s2 to do some data changes. +step s1_wait_before_lock +{ + REPACK (CONCURRENTLY) repack_direct_toast; +} + +# Check the table, after REPACK has completed. s2 must have saved the data +# as it was visible to it. We check that the relfilenode changed in addition +# to verifying that the actual data matches. +step s1_check +{ + INSERT INTO relfilenodes(node) + SELECT c2.relfilenode + FROM pg_class c1 JOIN pg_class c2 ON c2.oid = c1.oid OR c2.oid = c1.reltoastrelid + WHERE c1.relname='repack_direct_toast'; + + SELECT count(DISTINCT node) FROM relfilenodes; + + INSERT INTO data_s1(i, j, j_toast, k, k_toast) + SELECT i, + j, COALESCE(pg_column_toast_chunk_id(j), 0) AS j_toast, + k, COALESCE(pg_column_toast_chunk_id(k), 0) AS k_toast + FROM repack_direct_toast; + + -- this should be empty + SELECT d1.i, substring(d1.j FOR 12) AS d1_j, substring(d1.k FOR 12) AS d1_k, + d2.i, substring(d2.j FOR 12) AS d2_j, substring(d2.k FOR 12) AS d2_k, + d1.j_toast as d1_j_tst, d2.j_toast as d2_j_tst, + d1.k_toast as d1_k_tst, d2.k_toast AS d2_k_tst + FROM data_s1 d1 FULL JOIN data_s2 d2 USING (i, j, k) + WHERE d1.i ISNULL OR d2.i ISNULL; +} +teardown +{ + SELECT injection_points_detach('repack-concurrently-before-lock'); +} + +session s2 + +# Test different kinds of toast data changes. +step s2_updates +{ + DELETE FROM repack_direct_toast WHERE i=1; + INSERT INTO repack_direct_toast(i, j, k) VALUES (1, gen_external(), gen_compressible(1)); + + -- existing toast data unchanged. (This covers the case where we + -- adjust the toast pointer.) + UPDATE repack_direct_toast SET i=i+300 where i % 10 = 2 RETURNING OLD.i, NEW.i; + + -- "j" is here an external indirect, written to the file separately. + UPDATE repack_direct_toast SET j=gen_external() where i % 10 = 3 RETURNING OLD.i, NEW.i; + + -- the updated value of "j" is compressed. + UPDATE repack_direct_toast SET j=gen_compressible(1), k=k||'' where i % 10 = 4 RETURNING i; + + -- the updated value of "j" is compressed externally. + UPDATE repack_direct_toast SET j=gen_compressible_external(2) where i % 10 = 5 RETURNING i; + + -- the updated value of "j" stays inline. + UPDATE repack_direct_toast SET j=gen_inline(), k=repeat(k,5) where i % 10 = 6 RETURNING i; + + -- updated value of "j" is a short varlena; "k" is written separately. + UPDATE repack_direct_toast SET j=gen_short(), k=gen_external() where i % 10 = 7 RETURNING i; +} + +# Check the table from the perspective of s2. This saves data so that it can +# be verified later. +step s2_check +{ + INSERT INTO relfilenodes(node) + SELECT c2.relfilenode + FROM pg_class c1 JOIN pg_class c2 ON c2.oid = c1.oid OR c2.oid = c1.reltoastrelid + WHERE c1.relname='repack_direct_toast'; + + INSERT INTO data_s2(i, j, j_toast, k, k_toast) + SELECT i, j, COALESCE(pg_column_toast_chunk_id(j), 0) AS j_toast, + k, COALESCE(pg_column_toast_chunk_id(k), 0) AS k_toast + FROM repack_direct_toast; +} +step s2_wakeup_before_lock +{ + SELECT injection_points_wakeup('repack-concurrently-before-lock'); +} + +# Test if data changes introduced while one session is performing REPACK +# CONCURRENTLY find their way into the table. +permutation + s1_wait_before_lock + s2_updates + s2_check + s2_wakeup_before_lock + s1_check -- 2.55.0.1082.g2b9226bbc0-goog