Received: from magus.postgresql.org ([87.238.57.229]) by malur.postgresql.org with esmtp (Exim 4.72) (envelope-from ) id 1TCgQn-0005kr-CD for pgsql-hackers@postgresql.org; Sat, 15 Sep 2012 00:40:21 +0000 Received: from mail.anarazel.de ([217.115.131.40]) by magus.postgresql.org with esmtp (Exim 4.72) (envelope-from ) id 1TCgQk-00005m-M8 for pgsql-hackers@postgresql.org; Sat, 15 Sep 2012 00:40:20 +0000 Received: by mail.anarazel.de (Postfix, from userid 108) id 3006C4C0012; Sat, 15 Sep 2012 02:40:17 +0200 (CEST) X-Spam-Checker-Version: SpamAssassin 3.2.5 (2008-06-10) on mail X-Spam-Level: X-Spam-Status: No, score=-3.7 required=5.0 tests=ALL_TRUSTED,AWL,BAYES_00 autolearn=unavailable version=3.2.5 Received: from anarazel.de (p5DDC5E43.dip.t-dialin.net [93.220.94.67]) (Authenticated sender: andres@anarazel.de) by mail.anarazel.de (Postfix) with ESMTPSA id 7B0DB4C0017; Sat, 15 Sep 2012 02:39:48 +0200 (CEST) Received: by anarazel.de (Postfix, from userid 1000) id 6D4B61C017A; Sat, 15 Sep 2012 02:39:44 +0200 (CEST) From: Andres Freund To: Andres Freund , pgsql-hackers@postgresql.org, Robert Haas , Kevin Grittner , Alvaro Herrera , Simon Riggs Subject: [PATCH 6/8] Log enough data into the wal to reconstruct logical changes from it if wal_level=logical Date: Sat, 15 Sep 2012 02:39:33 +0200 Message-Id: <1347669575-14371-6-git-send-email-andres@2ndquadrant.com> X-Mailer: git-send-email 1.7.12.289.g0ce9864.dirty In-Reply-To: <201209150233.25616.andres@2ndquadrant.com> References: <201209150233.25616.andres@2ndquadrant.com> MIME-Version: 1.0 Content-Type: multipart/mixed; boundary="------------1.7.12.289.g0ce9864.dirty" X-Pg-Spam-Score: -3.1 (---) X-Archive-Number: 201209/578 X-Sequence-Number: 213606 From: Andres Freund This is a multi-part message in MIME format. --------------1.7.12.289.g0ce9864.dirty Content-Type: text/plain; charset=UTF-8; format=fixed Content-Transfer-Encoding: quoted-printable This adds a new wal_level value 'logical' Missing cases: - heap_multi_insert - primary key changes for updates - no primary key - LOG_NEWPAGE --- src/backend/access/heap/heapam.c | 135 ++++++++++++++++++++++++++= +++--- src/backend/access/transam/xlog.c | 1 + src/backend/catalog/index.c | 74 +++++++++++++++++ src/bin/pg_controldata/pg_controldata.c | 2 + src/include/access/xlog.h | 3 +- src/include/catalog/index.h | 4 + 6 files changed, 207 insertions(+), 12 deletions(-) --------------1.7.12.289.g0ce9864.dirty Content-Type: text/x-patch; name="0006-Log-enough-data-into-the-wal-to-reconstruct-logical-.patch" Content-Disposition: attachment; filename="0006-Log-enough-data-into-the-wal-to-reconstruct-logical-.patch" Content-Transfer-Encoding: quoted-printable diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/h= eapam.c index f56b577..190ae03 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -53,6 +53,7 @@ #include "access/xact.h" #include "access/xlogutils.h" #include "catalog/catalog.h" +#include "catalog/index.h" #include "catalog/namespace.h" #include "miscadmin.h" #include "pgstat.h" @@ -1938,10 +1939,19 @@ heap_insert(Relation relation, HeapTuple tup, Com= mandId cid, xl_heap_insert xlrec; xl_heap_header xlhdr; XLogRecPtr recptr; - XLogRecData rdata[3]; + XLogRecData rdata[4]; Page page =3D BufferGetPage(buffer); uint8 info =3D XLOG_HEAP_INSERT; =20 + /* + * For the logical replication case we need the tuple even if were + * doing a full page write. We could alternatively store a pointer int= o + * the fpw though. + * For that to work we add another rdata entry for the buffer in that + * case. + */ + bool need_tuple =3D wal_level =3D=3D WAL_LEVEL_LOGICAL; + xlrec.all_visible_cleared =3D all_visible_cleared; xlrec.target.node =3D relation->rd_node; xlrec.target.tid =3D heaptup->t_self; @@ -1961,18 +1971,32 @@ heap_insert(Relation relation, HeapTuple tup, Com= mandId cid, */ rdata[1].data =3D (char *) &xlhdr; rdata[1].len =3D SizeOfHeapHeader; - rdata[1].buffer =3D buffer; + rdata[1].buffer =3D need_tuple ? InvalidBuffer : buffer; rdata[1].buffer_std =3D true; rdata[1].next =3D &(rdata[2]); =20 /* PG73FORMAT: write bitmap [+ padding] [+ oid] + data */ rdata[2].data =3D (char *) heaptup->t_data + offsetof(HeapTupleHeaderD= ata, t_bits); rdata[2].len =3D heaptup->t_len - offsetof(HeapTupleHeaderData, t_bits= ); - rdata[2].buffer =3D buffer; + rdata[2].buffer =3D need_tuple ? InvalidBuffer : buffer; rdata[2].buffer_std =3D true; rdata[2].next =3D NULL; =20 /* + * add record for the buffer without actual content thats removed if + * fpw is done for that buffer + */ + if(need_tuple){ + rdata[2].next =3D &(rdata[3]); + + rdata[3].data =3D NULL; + rdata[3].len =3D 0; + rdata[3].buffer =3D buffer; + rdata[3].buffer_std =3D true; + rdata[3].next =3D NULL; + } + + /* * If this is the single and first tuple on page, we can reinit the * page instead of restoring the whole thing. Set flag, and hide * buffer references from XLogInsert. @@ -1981,7 +2005,7 @@ heap_insert(Relation relation, HeapTuple tup, Comma= ndId cid, PageGetMaxOffsetNumber(page) =3D=3D FirstOffsetNumber) { info |=3D XLOG_HEAP_INIT_PAGE; - rdata[1].buffer =3D rdata[2].buffer =3D InvalidBuffer; + rdata[1].buffer =3D rdata[2].buffer =3D rdata[3].buffer =3D InvalidBu= ffer; } =20 recptr =3D XLogInsert(RM_HEAP_ID, info, rdata); @@ -2569,7 +2593,9 @@ l1: { xl_heap_delete xlrec; XLogRecPtr recptr; - XLogRecData rdata[2]; + XLogRecData rdata[4]; + + bool need_tuple =3D wal_level =3D=3D WAL_LEVEL_LOGICAL && relation->rd= _id >=3D FirstNormalObjectId; =20 xlrec.all_visible_cleared =3D all_visible_cleared; xlrec.target.node =3D relation->rd_node; @@ -2585,6 +2611,73 @@ l1: rdata[1].buffer_std =3D true; rdata[1].next =3D NULL; =20 + /* + * XXX: We could decide not to log changes when the origin is not the + * local node, that should reduce redundant logging. + */ + if(need_tuple){ + xl_heap_header xlhdr; + + Oid indexoid =3D InvalidOid; + int16 pknratts; + int16 pkattnum[INDEX_MAX_KEYS]; + Oid pktypoid[INDEX_MAX_KEYS]; + Oid pkopclass[INDEX_MAX_KEYS]; + TupleDesc desc =3D RelationGetDescr(relation); + Relation index_rel; + TupleDesc indexdesc; + int natt; + + Datum idxvals[INDEX_MAX_KEYS]; + bool idxisnull[INDEX_MAX_KEYS]; + HeapTuple idxtuple; + + MemSet(pkattnum, 0, sizeof(pkattnum)); + MemSet(pktypoid, 0, sizeof(pktypoid)); + MemSet(pkopclass, 0, sizeof(pkopclass)); + MemSet(idxvals, 0, sizeof(idxvals)); + MemSet(idxisnull, 0, sizeof(idxisnull)); + relationFindPrimaryKey(relation, &indexoid, &pknratts, pkattnum, pkty= poid, pkopclass); + + if(!indexoid){ + elog(WARNING, "Could not find primary key for table with oid %u", + relation->rd_id); + goto no_index_found; + } + + index_rel =3D index_open(indexoid, AccessShareLock); + + indexdesc =3D RelationGetDescr(index_rel); + + for(natt =3D 0; natt < indexdesc->natts; natt++){ + idxvals[natt] =3D + fastgetattr(&tp, pkattnum[natt], desc, &idxisnull[natt]); + Assert(!idxisnull[natt]); + } + + idxtuple =3D heap_form_tuple(indexdesc, idxvals, idxisnull); + + xlhdr.t_infomask2 =3D idxtuple->t_data->t_infomask2; + xlhdr.t_infomask =3D idxtuple->t_data->t_infomask; + xlhdr.t_hoff =3D idxtuple->t_data->t_hoff; + + rdata[1].next =3D &(rdata[2]); + rdata[2].data =3D (char*)&xlhdr; + rdata[2].len =3D SizeOfHeapHeader; + rdata[2].buffer =3D InvalidBuffer; + rdata[2].next =3D NULL; + + rdata[2].next =3D &(rdata[3]); + rdata[3].data =3D (char *) idxtuple->t_data + offsetof(HeapTupleHeade= rData, t_bits); + rdata[3].len =3D idxtuple->t_len - offsetof(HeapTupleHeaderData, t_bi= ts); + rdata[3].buffer =3D InvalidBuffer; + rdata[3].next =3D NULL; + + heap_close(index_rel, NoLock); + no_index_found: + ; + } + recptr =3D XLogInsert(RM_HEAP_ID, XLOG_HEAP_DELETE, rdata); =20 PageSetLSN(page, recptr); @@ -4414,9 +4507,14 @@ log_heap_update(Relation reln, Buffer oldbuf, Item= PointerData from, xl_heap_header xlhdr; uint8 info; XLogRecPtr recptr; - XLogRecData rdata[4]; + XLogRecData rdata[5]; Page page =3D BufferGetPage(newbuf); =20 + /* + * Just as for XLOG_HEAP_INSERT we need to make sure the tuple + */ + bool need_tuple =3D wal_level =3D=3D WAL_LEVEL_LOGICAL; + /* Caller should not call me on a non-WAL-logged relation */ Assert(RelationNeedsWAL(reln)); =20 @@ -4447,28 +4545,43 @@ log_heap_update(Relation reln, Buffer oldbuf, Ite= mPointerData from, xlhdr.t_hoff =3D newtup->t_data->t_hoff; =20 /* - * As with insert records, we need not store the rdata[2] segment if we - * decide to store the whole buffer instead. + * As with insert's logging , we need not store the the Datum containin= g + * tuples separately from the buffer if we do logical replication that + * is... */ rdata[2].data =3D (char *) &xlhdr; rdata[2].len =3D SizeOfHeapHeader; - rdata[2].buffer =3D newbuf; + rdata[2].buffer =3D need_tuple ? InvalidBuffer : newbuf; rdata[2].buffer_std =3D true; rdata[2].next =3D &(rdata[3]); =20 /* PG73FORMAT: write bitmap [+ padding] [+ oid] + data */ rdata[3].data =3D (char *) newtup->t_data + offsetof(HeapTupleHeaderDat= a, t_bits); rdata[3].len =3D newtup->t_len - offsetof(HeapTupleHeaderData, t_bits); - rdata[3].buffer =3D newbuf; + rdata[3].buffer =3D need_tuple ? InvalidBuffer : newbuf; rdata[3].buffer_std =3D true; rdata[3].next =3D NULL; =20 + /* + * separate storage for the buffer reference of the new page in the + * wal_level=3Dlogical case + */ + if(need_tuple){ + rdata[3].next =3D &(rdata[4]); + + rdata[4].data =3D NULL, + rdata[4].len =3D 0; + rdata[4].buffer =3D newbuf; + rdata[4].buffer_std =3D true; + rdata[4].next =3D NULL; + } + /* If new tuple is the single and first tuple on page... */ if (ItemPointerGetOffsetNumber(&(newtup->t_self)) =3D=3D FirstOffsetNum= ber && PageGetMaxOffsetNumber(page) =3D=3D FirstOffsetNumber) { info |=3D XLOG_HEAP_INIT_PAGE; - rdata[2].buffer =3D rdata[3].buffer =3D InvalidBuffer; + rdata[2].buffer =3D rdata[3].buffer =3D rdata[4].buffer =3D InvalidBuf= fer; } =20 recptr =3D XLogInsert(RM_HEAP_ID, info, rdata); diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/trans= am/xlog.c index ff56c26..53a0bc8 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -107,6 +107,7 @@ const struct config_enum_entry wal_level_options[] =3D= { {"minimal", WAL_LEVEL_MINIMAL, false}, {"archive", WAL_LEVEL_ARCHIVE, false}, {"hot_standby", WAL_LEVEL_HOT_STANDBY, false}, + {"logical", WAL_LEVEL_LOGICAL, false}, {NULL, 0, false} }; =20 diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c index 464950b..8145997 100644 --- a/src/backend/catalog/index.c +++ b/src/backend/catalog/index.c @@ -49,6 +49,7 @@ #include "nodes/nodeFuncs.h" #include "optimizer/clauses.h" #include "parser/parser.h" +#include "parser/parse_relation.h" #include "storage/bufmgr.h" #include "storage/lmgr.h" #include "storage/predicate.h" @@ -3322,3 +3323,76 @@ ResetReindexPending(void) { pendingReindexedIndexes =3D NIL; } + +/* + * relationFindPrimaryKey + * Find primary key for a relation if it exists. + * + * If no primary key is found *indexOid is set to InvalidOid + * + * This is quite similar to tablecmd.c's transformFkeyGetPrimaryKey. + * + * XXX: It might be a good idea to change pg_class.relhaspkey into an bo= ol to + * make this more efficient. + */ +void +relationFindPrimaryKey(Relation pkrel, Oid *indexOid, + int16 *nratts, int16 *attnums, Oid *atttypids, + Oid *opclasses){ + List *indexoidlist; + ListCell *indexoidscan; + HeapTuple indexTuple =3D NULL; + Datum indclassDatum; + bool isnull; + oidvector *indclass; + int i; + Form_pg_index indexStruct =3D NULL; + + *indexOid =3D InvalidOid; + + indexoidlist =3D RelationGetIndexList(pkrel); + + foreach(indexoidscan, indexoidlist) + { + Oid indexoid =3D lfirst_oid(indexoidscan); + + indexTuple =3D SearchSysCache1(INDEXRELID, ObjectIdGetDatum(indexoid))= ; + if(!HeapTupleIsValid(indexTuple)) + elog(ERROR, "cache lookup failed for index %u", indexoid); + + indexStruct =3D (Form_pg_index) GETSTRUCT(indexTuple); + if(indexStruct->indisprimary && indexStruct->indimmediate) + { + *indexOid =3D indexoid; + break; + } + ReleaseSysCache(indexTuple); + + } + list_free(indexoidlist); + + if (!OidIsValid(*indexOid)) + return; + + /* Must get indclass the hard way */ + indclassDatum =3D SysCacheGetAttr(INDEXRELID, indexTuple, + Anum_pg_index_indclass, &isnull); + Assert(!isnull); + indclass =3D (oidvector *) DatumGetPointer(indclassDatum); + + *nratts =3D indexStruct->indnatts; + /* + * Now build the list of PK attributes from the indkey definition (we + * assume a primary key cannot have expressional elements) + */ + for (i =3D 0; i < indexStruct->indnatts; i++) + { + int pkattno =3D indexStruct->indkey.values[i]; + + attnums[i] =3D pkattno; + atttypids[i] =3D attnumTypeId(pkrel, pkattno); + opclasses[i] =3D indclass->values[i]; + } + + ReleaseSysCache(indexTuple); +} diff --git a/src/bin/pg_controldata/pg_controldata.c b/src/bin/pg_control= data/pg_controldata.c index 129c4d0..10080d0 100644 --- a/src/bin/pg_controldata/pg_controldata.c +++ b/src/bin/pg_controldata/pg_controldata.c @@ -77,6 +77,8 @@ wal_level_str(WalLevel wal_level) return "archive"; case WAL_LEVEL_HOT_STANDBY: return "hot_standby"; + case WAL_LEVEL_LOGICAL: + return "logical"; } return _("unrecognized wal_level"); } diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index 2893f3b..7d90416 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -200,7 +200,8 @@ typedef enum WalLevel { WAL_LEVEL_MINIMAL =3D 0, WAL_LEVEL_ARCHIVE, - WAL_LEVEL_HOT_STANDBY + WAL_LEVEL_HOT_STANDBY, + WAL_LEVEL_LOGICAL } WalLevel; extern int wal_level; =20 diff --git a/src/include/catalog/index.h b/src/include/catalog/index.h index eb417ce..3de0a29 100644 --- a/src/include/catalog/index.h +++ b/src/include/catalog/index.h @@ -102,4 +102,8 @@ extern bool ReindexIsProcessingHeap(Oid heapOid); extern bool ReindexIsProcessingIndex(Oid indexOid); extern Oid IndexGetRelation(Oid indexId, bool missing_ok); =20 +extern void relationFindPrimaryKey(Relation pkrel, Oid *indexOid, + int16 *nratts, int16 *attnums, Oid *a= tttypids, + Oid *opclasses); + #endif /* INDEX_H */ --------------1.7.12.289.g0ce9864.dirty--