pg.ddx.io pgsql-hackers@postgresql.org mailing list archive
help / color / mirror / Atom feed[PATCH 07/16] Log enough data into the wal to reconstruct logical changes from it if wal_level=logical
5+ messages / 3 participants
[nested] [flat]
* [PATCH 07/16] Log enough data into the wal to reconstruct logical changes from it if wal_level=logical
@ 2012-06-13 11:28 Andres Freund <andres@2ndquadrant.com>
2012-06-13 15:27 ` Re: [PATCH 07/16] Log enough data into the wal to reconstruct logical changes from it if wal_level=logical Kevin Grittner <Kevin.Grittner@wicourts.gov>
2012-06-27 13:55 ` Re: [PATCH 07/16] Log enough data into the wal to reconstruct logical changes from it if wal_level=logical Heikki Linnakangas <heikki.linnakangas@enterprisedb.com>
0 siblings, 2 replies; 5+ messages in thread
From: Andres Freund @ 2012-06-13 11:28 UTC (permalink / raw)
To: pgsql-hackers
From: Andres Freund <andres@anarazel.de>
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(-)
inline patch 1: src/backend/access/heap/heapam.c, src/backend/access/transam/xlog.c, src/backend/catalog/index.c and 3 more (+207/-12 lines) download inline-1.patch
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 9519e73..9149d53 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -52,6 +52,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"
@@ -1937,10 +1938,19 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
xl_heap_insert xlrec;
xl_heap_header xlhdr;
XLogRecPtr recptr;
- XLogRecData rdata[3];
+ XLogRecData rdata[4];
Page page = BufferGetPage(buffer);
uint8 info = XLOG_HEAP_INSERT;
+ /*
+ * For the logical replication case we need the tuple even if were
+ * doing a full page write. We could alternatively store a pointer into
+ * the fpw though.
+ * For that to work we add another rdata entry for the buffer in that
+ * case.
+ */
+ bool need_tuple = wal_level == WAL_LEVEL_LOGICAL;
+
xlrec.all_visible_cleared = all_visible_cleared;
xlrec.target.node = relation->rd_node;
xlrec.target.tid = heaptup->t_self;
@@ -1960,18 +1970,32 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
*/
rdata[1].data = (char *) &xlhdr;
rdata[1].len = SizeOfHeapHeader;
- rdata[1].buffer = buffer;
+ rdata[1].buffer = need_tuple ? InvalidBuffer : buffer;
rdata[1].buffer_std = true;
rdata[1].next = &(rdata[2]);
/* PG73FORMAT: write bitmap [+ padding] [+ oid] + data */
rdata[2].data = (char *) heaptup->t_data + offsetof(HeapTupleHeaderData, t_bits);
rdata[2].len = heaptup->t_len - offsetof(HeapTupleHeaderData, t_bits);
- rdata[2].buffer = buffer;
+ rdata[2].buffer = need_tuple ? InvalidBuffer : buffer;
rdata[2].buffer_std = true;
rdata[2].next = NULL;
/*
+ * add record for the buffer without actual content thats removed if
+ * fpw is done for that buffer
+ */
+ if(need_tuple){
+ rdata[2].next = &(rdata[3]);
+
+ rdata[3].data = NULL;
+ rdata[3].len = 0;
+ rdata[3].buffer = buffer;
+ rdata[3].buffer_std = true;
+ rdata[3].next = 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.
@@ -1980,7 +2004,7 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
PageGetMaxOffsetNumber(page) == FirstOffsetNumber)
{
info |= XLOG_HEAP_INIT_PAGE;
- rdata[1].buffer = rdata[2].buffer = InvalidBuffer;
+ rdata[1].buffer = rdata[2].buffer = rdata[3].buffer = InvalidBuffer;
}
recptr = XLogInsert(RM_HEAP_ID, info, rdata);
@@ -2568,7 +2592,9 @@ l1:
{
xl_heap_delete xlrec;
XLogRecPtr recptr;
- XLogRecData rdata[2];
+ XLogRecData rdata[4];
+
+ bool need_tuple = wal_level == WAL_LEVEL_LOGICAL && relation->rd_id >= FirstNormalObjectId;
xlrec.all_visible_cleared = all_visible_cleared;
xlrec.target.node = relation->rd_node;
@@ -2584,6 +2610,73 @@ l1:
rdata[1].buffer_std = true;
rdata[1].next = NULL;
+ /*
+ * 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 = InvalidOid;
+ int16 pknratts;
+ int16 pkattnum[INDEX_MAX_KEYS];
+ Oid pktypoid[INDEX_MAX_KEYS];
+ Oid pkopclass[INDEX_MAX_KEYS];
+ TupleDesc desc = 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, pktypoid, pkopclass);
+
+ if(!indexoid){
+ elog(WARNING, "Could not find primary key for table with oid %u",
+ relation->rd_id);
+ goto no_index_found;
+ }
+
+ index_rel = index_open(indexoid, AccessShareLock);
+
+ indexdesc = RelationGetDescr(index_rel);
+
+ for(natt = 0; natt < indexdesc->natts; natt++){
+ idxvals[natt] =
+ fastgetattr(&tp, pkattnum[natt], desc, &idxisnull[natt]);
+ Assert(!idxisnull[natt]);
+ }
+
+ idxtuple = heap_form_tuple(indexdesc, idxvals, idxisnull);
+
+ xlhdr.t_infomask2 = idxtuple->t_data->t_infomask2;
+ xlhdr.t_infomask = idxtuple->t_data->t_infomask;
+ xlhdr.t_hoff = idxtuple->t_data->t_hoff;
+
+ rdata[1].next = &(rdata[2]);
+ rdata[2].data = (char*)&xlhdr;
+ rdata[2].len = SizeOfHeapHeader;
+ rdata[2].buffer = InvalidBuffer;
+ rdata[2].next = NULL;
+
+ rdata[2].next = &(rdata[3]);
+ rdata[3].data = (char *) idxtuple->t_data + offsetof(HeapTupleHeaderData, t_bits);
+ rdata[3].len = idxtuple->t_len - offsetof(HeapTupleHeaderData, t_bits);
+ rdata[3].buffer = InvalidBuffer;
+ rdata[3].next = NULL;
+
+ heap_close(index_rel, NoLock);
+ no_index_found:
+ ;
+ }
+
recptr = XLogInsert(RM_HEAP_ID, XLOG_HEAP_DELETE, rdata);
PageSetLSN(page, recptr);
@@ -4413,9 +4506,14 @@ log_heap_update(Relation reln, Buffer oldbuf, ItemPointerData from,
xl_heap_header xlhdr;
uint8 info;
XLogRecPtr recptr;
- XLogRecData rdata[4];
+ XLogRecData rdata[5];
Page page = BufferGetPage(newbuf);
+ /*
+ * Just as for XLOG_HEAP_INSERT we need to make sure the tuple
+ */
+ bool need_tuple = wal_level == WAL_LEVEL_LOGICAL;
+
/* Caller should not call me on a non-WAL-logged relation */
Assert(RelationNeedsWAL(reln));
@@ -4446,28 +4544,43 @@ log_heap_update(Relation reln, Buffer oldbuf, ItemPointerData from,
xlhdr.t_hoff = newtup->t_data->t_hoff;
/*
- * 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 containing
+ * tuples separately from the buffer if we do logical replication that
+ * is...
*/
rdata[2].data = (char *) &xlhdr;
rdata[2].len = SizeOfHeapHeader;
- rdata[2].buffer = newbuf;
+ rdata[2].buffer = need_tuple ? InvalidBuffer : newbuf;
rdata[2].buffer_std = true;
rdata[2].next = &(rdata[3]);
/* PG73FORMAT: write bitmap [+ padding] [+ oid] + data */
rdata[3].data = (char *) newtup->t_data + offsetof(HeapTupleHeaderData, t_bits);
rdata[3].len = newtup->t_len - offsetof(HeapTupleHeaderData, t_bits);
- rdata[3].buffer = newbuf;
+ rdata[3].buffer = need_tuple ? InvalidBuffer : newbuf;
rdata[3].buffer_std = true;
rdata[3].next = NULL;
+ /*
+ * separate storage for the buffer reference of the new page in the
+ * wal_level=logical case
+ */
+ if(need_tuple){
+ rdata[3].next = &(rdata[4]);
+
+ rdata[4].data = NULL,
+ rdata[4].len = 0;
+ rdata[4].buffer = newbuf;
+ rdata[4].buffer_std = true;
+ rdata[4].next = NULL;
+ }
+
/* If new tuple is the single and first tuple on page... */
if (ItemPointerGetOffsetNumber(&(newtup->t_self)) == FirstOffsetNumber &&
PageGetMaxOffsetNumber(page) == FirstOffsetNumber)
{
info |= XLOG_HEAP_INIT_PAGE;
- rdata[2].buffer = rdata[3].buffer = InvalidBuffer;
+ rdata[2].buffer = rdata[3].buffer = rdata[4].buffer = InvalidBuffer;
}
recptr = XLogInsert(RM_HEAP_ID, info, rdata);
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 166efb0..c6feed0 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -105,6 +105,7 @@ const struct config_enum_entry wal_level_options[] = {
{"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}
};
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index 9e8b1cc..4cddcac 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"
@@ -3311,3 +3312,76 @@ ResetReindexPending(void)
{
pendingReindexedIndexes = 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 bool 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 = NULL;
+ Datum indclassDatum;
+ bool isnull;
+ oidvector *indclass;
+ int i;
+ Form_pg_index indexStruct = NULL;
+
+ *indexOid = InvalidOid;
+
+ indexoidlist = RelationGetIndexList(pkrel);
+
+ foreach(indexoidscan, indexoidlist)
+ {
+ Oid indexoid = lfirst_oid(indexoidscan);
+
+ indexTuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(indexoid));
+ if(!HeapTupleIsValid(indexTuple))
+ elog(ERROR, "cache lookup failed for index %u", indexoid);
+
+ indexStruct = (Form_pg_index) GETSTRUCT(indexTuple);
+ if(indexStruct->indisprimary && indexStruct->indimmediate)
+ {
+ *indexOid = indexoid;
+ break;
+ }
+ ReleaseSysCache(indexTuple);
+
+ }
+ list_free(indexoidlist);
+
+ if (!OidIsValid(*indexOid))
+ return;
+
+ /* Must get indclass the hard way */
+ indclassDatum = SysCacheGetAttr(INDEXRELID, indexTuple,
+ Anum_pg_index_indclass, &isnull);
+ Assert(!isnull);
+ indclass = (oidvector *) DatumGetPointer(indclassDatum);
+
+ *nratts = indexStruct->indnatts;
+ /*
+ * Now build the list of PK attributes from the indkey definition (we
+ * assume a primary key cannot have expressional elements)
+ */
+ for (i = 0; i < indexStruct->indnatts; i++)
+ {
+ int pkattno = indexStruct->indkey.values[i];
+
+ attnums[i] = pkattno;
+ atttypids[i] = attnumTypeId(pkrel, pkattno);
+ opclasses[i] = indclass->values[i];
+ }
+
+ ReleaseSysCache(indexTuple);
+}
diff --git a/src/bin/pg_controldata/pg_controldata.c b/src/bin/pg_controldata/pg_controldata.c
index c00183a..47715c9 100644
--- a/src/bin/pg_controldata/pg_controldata.c
+++ b/src/bin/pg_controldata/pg_controldata.c
@@ -82,6 +82,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 df5f232..2843aca 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -199,7 +199,8 @@ typedef enum WalLevel
{
WAL_LEVEL_MINIMAL = 0,
WAL_LEVEL_ARCHIVE,
- WAL_LEVEL_HOT_STANDBY
+ WAL_LEVEL_HOT_STANDBY,
+ WAL_LEVEL_LOGICAL
} WalLevel;
extern int wal_level;
diff --git a/src/include/catalog/index.h b/src/include/catalog/index.h
index 7c8198f..2ba0ac3 100644
--- a/src/include/catalog/index.h
+++ b/src/include/catalog/index.h
@@ -101,4 +101,8 @@ extern bool ReindexIsProcessingHeap(Oid heapOid);
extern bool ReindexIsProcessingIndex(Oid indexOid);
extern Oid IndexGetRelation(Oid indexId, bool missing_ok);
+extern void relationFindPrimaryKey(Relation pkrel, Oid *indexOid,
+ int16 *nratts, int16 *attnums, Oid *atttypids,
+ Oid *opclasses);
+
#endif /* INDEX_H */
--
1.7.10.rc3.3.g19a6c.dirty
* Re: [PATCH 07/16] Log enough data into the wal to reconstruct logical changes from it if wal_level=logical
2012-06-13 11:28 [PATCH 07/16] Log enough data into the wal to reconstruct logical changes from it if wal_level=logical Andres Freund <andres@2ndquadrant.com>
@ 2012-06-13 15:27 ` Kevin Grittner <Kevin.Grittner@wicourts.gov>
2012-06-13 15:57 ` Re: [PATCH 07/16] Log enough data into the wal to reconstruct logical changes from it if wal_level=logical Andres Freund <andres@2ndquadrant.com>
1 sibling, 1 reply; 5+ messages in thread
From: Kevin Grittner @ 2012-06-13 15:27 UTC (permalink / raw)
To: Andres Freund <andres@2ndquadrant.com>; pgsql-hackers
Andres Freund <andres@2ndquadrant.com> wrote:
> This adds a new wal_level value 'logical'
>
> Missing cases:
> - heap_multi_insert
> - primary key changes for updates
> - no primary key
> - LOG_NEWPAGE
First, Wow!
I look forward to the point where we can replace our trigger-based
replication with this! Your "missing cases" for primary key issues
would not cause us any pain for our current system, since we require
a primary key and don't support updates to PKs for replicated
tables. While I don't expect that the first cut of this will be able
to replace our replication-related functionality, I'm interested in
making sure it can be extended in that direction, so I have a couple
things to consider:
(1) For our usage, with dozens of source databases feeding into
multiple aggregate databases and interfaces, DDL replication is not
of much if any interest. It should be easy enough to ignore as long
as it is low volume, so that doesn't worry me too much; but if I'm
missing something any you run across any logical WAL logging for DDL
which does generate a lot of WAL traffic, it would be nice to have a
way to turn that off at generation time rather than filtering it or
ignoring it later. (Probably won't be an issue, just a head-up.)
(2) To match the functionality we now have, we would need the
logical stream to include the *before* image of the whole tuple for
each row updated or deleted. I understand that this is not needed
for the use cases you are initially targeting; I just hope the
design leaves this option open without needing to disturb other use
cases. Perhaps this would require yet another wal_level value.
Perhaps rather than testing the current value directly for
determining whether to log something, the GUC processing could set
some booleans for faster testing and less code churn when the
initial implementation is expanded to support other use cases (like
ours).
(3) Similar to point 2, it would be extremely desirable to be able
to determine table name and columns names for the tuples in a stream
from that stream, without needing to query a hot standby or similar
digging into other sources of information. Not only will the
various source databases all have different OID values for the same
objects, and the aggregate targets have different values from each
other and the sources, but some targets don't have the tables at
all. I'm talking about our database transaction repository and the
interfaces to business partners which we currently drive off of the
same transaction stream which drives replication.
Would it be helpful or just a distraction if I were to provide a
more detailed description of our whole replication / transaction
store / interface area?
If it would be useful, I could also describe some other replication
patterns I have seen over the years. In particular, one which might
be interesting is where subsets of the data are distributed to
multiple standalone machines which have intermittent or unreliable
connections to a central site, which periodically collects data from
all the remote sites, recalculates distribution, and sends
transactions back out to those remote sites to add, remove, and
update rows based on the distribution rules and the new data.
-Kevin
^ permalink raw reply [nested|flat] 5+ messages in thread
* Re: [PATCH 07/16] Log enough data into the wal to reconstruct logical changes from it if wal_level=logical
2012-06-13 11:28 [PATCH 07/16] Log enough data into the wal to reconstruct logical changes from it if wal_level=logical Andres Freund <andres@2ndquadrant.com>
2012-06-13 15:27 ` Re: [PATCH 07/16] Log enough data into the wal to reconstruct logical changes from it if wal_level=logical Kevin Grittner <Kevin.Grittner@wicourts.gov>
@ 2012-06-13 15:57 ` Andres Freund <andres@2ndquadrant.com>
0 siblings, 0 replies; 5+ messages in thread
From: Andres Freund @ 2012-06-13 15:57 UTC (permalink / raw)
To: pgsql-hackers; +Cc: Kevin Grittner <Kevin.Grittner@wicourts.gov>
On Wednesday, June 13, 2012 05:27:06 PM Kevin Grittner wrote:
> Andres Freund <andres@2ndquadrant.com> wrote:
> > This adds a new wal_level value 'logical'
> >
> > Missing cases:
> > - heap_multi_insert
> > - primary key changes for updates
> > - no primary key
> > - LOG_NEWPAGE
>
> First, Wow!
Thanks ;) I hope you will still be convinced after reading some of the code :P
> I look forward to the point where we can replace our trigger-based
> replication with this! Your "missing cases" for primary key issues
> would not cause us any pain for our current system, since we require
> a primary key and don't support updates to PKs for replicated
> tables. While I don't expect that the first cut of this will be able
> to replace our replication-related functionality, I'm interested in
> making sure it can be extended in that direction, so I have a couple
> things to consider:
Ok.
> (1) For our usage, with dozens of source databases feeding into
> multiple aggregate databases and interfaces, DDL replication is not
> of much if any interest. It should be easy enough to ignore as long
> as it is low volume, so that doesn't worry me too much; but if I'm
> missing something any you run across any logical WAL logging for DDL
> which does generate a lot of WAL traffic, it would be nice to have a
> way to turn that off at generation time rather than filtering it or
> ignoring it later. (Probably won't be an issue, just a head-up.)
I don't really see a problem there. I don't yet have a mental image of the
API/Parameters to START_LOGICAL_REPLICATION to specify filters on the source
side, but this should be possible.
> (2) To match the functionality we now have, we would need the
> logical stream to include the *before* image of the whole tuple for
> each row updated or deleted. I understand that this is not needed
> for the use cases you are initially targeting; I just hope the
> design leaves this option open without needing to disturb other use
> cases. Perhaps this would require yet another wal_level value.
> Perhaps rather than testing the current value directly for
> determining whether to log something, the GUC processing could set
> some booleans for faster testing and less code churn when the
> initial implementation is expanded to support other use cases (like
> ours).
Hm. I don't see a big problem implementing this although I have to say that I
am a bit hesitant to do this without in-core users of it for fear of silent
breakage. WAL is kind of a central thing... ;). But then, the implementation
should be relatively easy.
I don't see a need to break the wal level down into some booleans: changing
the test from wal_level >= WAL_LEVEL_LOGICAL into something else shouldn't
result in any measurable difference in that codepath.
I definitely have the use-case of replicating into databases where you need to
do some transformation in mind.
> (3) Similar to point 2, it would be extremely desirable to be able
> to determine table name and columns names for the tuples in a stream
> from that stream, without needing to query a hot standby or similar
> digging into other sources of information. Not only will the
> various source databases all have different OID values for the same
> objects, and the aggregate targets have different values from each
> other and the sources, but some targets don't have the tables at
> all. I'm talking about our database transaction repository and the
> interfaces to business partners which we currently drive off of the
> same transaction stream which drives replication.
I don't forsee this as a realistic thing. I think the required changes would
be way to intrusive for too little gain. The performance and space
requirements would probably also be rather noticeable. I think you will have
to live with the mentioned 'proxy' pg instances which only contain the catalog
(which normally isn't very big anyway) doing the decoding into your target
database (which then doesn't need the same oids).
For how I imagine those proxy instances check my mail to merlin earlier today,
I described it in some more detail there.
If I can find the time I should possibly develop (another) prototype of such a
proxy instance. I don't forsee it needing much more infrastructure/code.
Is that a problem for your use-case? If yes, why?
> Would it be helpful or just a distraction if I were to provide a
> more detailed description of our whole replication / transaction
> store / interface area?
If you have it ready, yes. Otherwise I think I have a good enough image
already. I tried to listen to you at pgcon and I have developed similar things
before.
To be honest, I would prefer you spending the time on checking some of the
code if not ;)
> If it would be useful, I could also describe some other replication
> patterns I have seen over the years. In particular, one which might
> be interesting is where subsets of the data are distributed to
> multiple standalone machines which have intermittent or unreliable
> connections to a central site, which periodically collects data from
> all the remote sites, recalculates distribution, and sends
> transactions back out to those remote sites to add, remove, and
> update rows based on the distribution rules and the new data.
I don't think its that relevant for now. I think that time is spent more
wisely when we get to conflict resolution and user interface...
Thanks!
Andres
--
Andres Freund http://www.2ndQuadrant.com/
PostgreSQL Development, 24x7 Support, Training & Services
^ permalink raw reply [nested|flat] 5+ messages in thread
* Re: [PATCH 07/16] Log enough data into the wal to reconstruct logical changes from it if wal_level=logical
2012-06-13 11:28 [PATCH 07/16] Log enough data into the wal to reconstruct logical changes from it if wal_level=logical Andres Freund <andres@2ndquadrant.com>
@ 2012-06-27 13:55 ` Heikki Linnakangas <heikki.linnakangas@enterprisedb.com>
1 sibling, 0 replies; 5+ messages in thread
From: Heikki Linnakangas @ 2012-06-27 13:55 UTC (permalink / raw)
To: Andres Freund <andres@2ndquadrant.com>; +Cc: pgsql-hackers
On 13.06.2012 14:28, Andres Freund wrote:
> @@ -2584,6 +2610,73 @@ l1:
> rdata[1].buffer_std = true;
> rdata[1].next = NULL;
>
> + /*
> + * 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;
> ...
> + relationFindPrimaryKey(relation,&indexoid,&pknratts, pkattnum, pktypoid, pkopclass);
> +
> + if(!indexoid){
> + elog(WARNING, "Could not find primary key for table with oid %u",
> + relation->rd_id);
> + goto no_index_found;
> + }
> +
> + index_rel = index_open(indexoid, AccessShareLock);
> +
> + indexdesc = RelationGetDescr(index_rel);
> +
> + for(natt = 0; natt< indexdesc->natts; natt++){
> + idxvals[natt] =
> + fastgetattr(&tp, pkattnum[natt], desc,&idxisnull[natt]);
> + Assert(!idxisnull[natt]);
> + }
> +
> + idxtuple = heap_form_tuple(indexdesc, idxvals, idxisnull);
> +
> + xlhdr.t_infomask2 = idxtuple->t_data->t_infomask2;
> + xlhdr.t_infomask = idxtuple->t_data->t_infomask;
> + xlhdr.t_hoff = idxtuple->t_data->t_hoff;
> +
> + rdata[1].next =&(rdata[2]);
> + rdata[2].data = (char*)&xlhdr;
> + rdata[2].len = SizeOfHeapHeader;
> + rdata[2].buffer = InvalidBuffer;
> + rdata[2].next = NULL;
> +
> + rdata[2].next =&(rdata[3]);
> + rdata[3].data = (char *) idxtuple->t_data + offsetof(HeapTupleHeaderData, t_bits);
> + rdata[3].len = idxtuple->t_len - offsetof(HeapTupleHeaderData, t_bits);
> + rdata[3].buffer = InvalidBuffer;
> + rdata[3].next = NULL;
> +
> + heap_close(index_rel, NoLock);
> + no_index_found:
> + ;
> + }
> +
> recptr = XLogInsert(RM_HEAP_ID, XLOG_HEAP_DELETE, rdata);
>
> PageSetLSN(page, recptr);
It's not cool to do all that primary key lookup stuff within the
critical section, while holding a lock on the page.
--
Heikki Linnakangas
EnterpriseDB http://www.enterprisedb.com
^ permalink raw reply [nested|flat] 5+ messages in thread
* [PATCH 6/8] Log enough data into the wal to reconstruct logical changes from it if wal_level=logical
@ 2012-09-15 00:39 Andres Freund <andres@2ndquadrant.com>
0 siblings, 0 replies; 5+ messages in thread
From: Andres Freund @ 2012-09-15 00:39 UTC (permalink / raw)
To: Andres Freund <andres@2ndquadrant.com>; pgsql-hackers@postgresql.org, Robert Haas <robertmhaas@gmail.com>; Kevin Grittner <Kevin.Grittner@wicourts.gov>; Alvaro Herrera <alvherre@2ndquadrant.com>; Simon Riggs <simon@2ndquadrant.com>
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(-)
Attachments:
0006-Log-enough-data-into-the-wal-to-reconstruct-logical-.patch (text/x-patch, 11.4K) download
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.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, CommandId cid,
xl_heap_insert xlrec;
xl_heap_header xlhdr;
XLogRecPtr recptr;
- XLogRecData rdata[3];
+ XLogRecData rdata[4];
Page page = BufferGetPage(buffer);
uint8 info = XLOG_HEAP_INSERT;
+ /*
+ * For the logical replication case we need the tuple even if were
+ * doing a full page write. We could alternatively store a pointer into
+ * the fpw though.
+ * For that to work we add another rdata entry for the buffer in that
+ * case.
+ */
+ bool need_tuple = wal_level == WAL_LEVEL_LOGICAL;
+
xlrec.all_visible_cleared = all_visible_cleared;
xlrec.target.node = relation->rd_node;
xlrec.target.tid = heaptup->t_self;
@@ -1961,18 +1971,32 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
*/
rdata[1].data = (char *) &xlhdr;
rdata[1].len = SizeOfHeapHeader;
- rdata[1].buffer = buffer;
+ rdata[1].buffer = need_tuple ? InvalidBuffer : buffer;
rdata[1].buffer_std = true;
rdata[1].next = &(rdata[2]);
/* PG73FORMAT: write bitmap [+ padding] [+ oid] + data */
rdata[2].data = (char *) heaptup->t_data + offsetof(HeapTupleHeaderData, t_bits);
rdata[2].len = heaptup->t_len - offsetof(HeapTupleHeaderData, t_bits);
- rdata[2].buffer = buffer;
+ rdata[2].buffer = need_tuple ? InvalidBuffer : buffer;
rdata[2].buffer_std = true;
rdata[2].next = NULL;
/*
+ * add record for the buffer without actual content thats removed if
+ * fpw is done for that buffer
+ */
+ if(need_tuple){
+ rdata[2].next = &(rdata[3]);
+
+ rdata[3].data = NULL;
+ rdata[3].len = 0;
+ rdata[3].buffer = buffer;
+ rdata[3].buffer_std = true;
+ rdata[3].next = 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, CommandId cid,
PageGetMaxOffsetNumber(page) == FirstOffsetNumber)
{
info |= XLOG_HEAP_INIT_PAGE;
- rdata[1].buffer = rdata[2].buffer = InvalidBuffer;
+ rdata[1].buffer = rdata[2].buffer = rdata[3].buffer = InvalidBuffer;
}
recptr = 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 = wal_level == WAL_LEVEL_LOGICAL && relation->rd_id >= FirstNormalObjectId;
xlrec.all_visible_cleared = all_visible_cleared;
xlrec.target.node = relation->rd_node;
@@ -2585,6 +2611,73 @@ l1:
rdata[1].buffer_std = true;
rdata[1].next = NULL;
+ /*
+ * 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 = InvalidOid;
+ int16 pknratts;
+ int16 pkattnum[INDEX_MAX_KEYS];
+ Oid pktypoid[INDEX_MAX_KEYS];
+ Oid pkopclass[INDEX_MAX_KEYS];
+ TupleDesc desc = 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, pktypoid, pkopclass);
+
+ if(!indexoid){
+ elog(WARNING, "Could not find primary key for table with oid %u",
+ relation->rd_id);
+ goto no_index_found;
+ }
+
+ index_rel = index_open(indexoid, AccessShareLock);
+
+ indexdesc = RelationGetDescr(index_rel);
+
+ for(natt = 0; natt < indexdesc->natts; natt++){
+ idxvals[natt] =
+ fastgetattr(&tp, pkattnum[natt], desc, &idxisnull[natt]);
+ Assert(!idxisnull[natt]);
+ }
+
+ idxtuple = heap_form_tuple(indexdesc, idxvals, idxisnull);
+
+ xlhdr.t_infomask2 = idxtuple->t_data->t_infomask2;
+ xlhdr.t_infomask = idxtuple->t_data->t_infomask;
+ xlhdr.t_hoff = idxtuple->t_data->t_hoff;
+
+ rdata[1].next = &(rdata[2]);
+ rdata[2].data = (char*)&xlhdr;
+ rdata[2].len = SizeOfHeapHeader;
+ rdata[2].buffer = InvalidBuffer;
+ rdata[2].next = NULL;
+
+ rdata[2].next = &(rdata[3]);
+ rdata[3].data = (char *) idxtuple->t_data + offsetof(HeapTupleHeaderData, t_bits);
+ rdata[3].len = idxtuple->t_len - offsetof(HeapTupleHeaderData, t_bits);
+ rdata[3].buffer = InvalidBuffer;
+ rdata[3].next = NULL;
+
+ heap_close(index_rel, NoLock);
+ no_index_found:
+ ;
+ }
+
recptr = XLogInsert(RM_HEAP_ID, XLOG_HEAP_DELETE, rdata);
PageSetLSN(page, recptr);
@@ -4414,9 +4507,14 @@ log_heap_update(Relation reln, Buffer oldbuf, ItemPointerData from,
xl_heap_header xlhdr;
uint8 info;
XLogRecPtr recptr;
- XLogRecData rdata[4];
+ XLogRecData rdata[5];
Page page = BufferGetPage(newbuf);
+ /*
+ * Just as for XLOG_HEAP_INSERT we need to make sure the tuple
+ */
+ bool need_tuple = wal_level == WAL_LEVEL_LOGICAL;
+
/* Caller should not call me on a non-WAL-logged relation */
Assert(RelationNeedsWAL(reln));
@@ -4447,28 +4545,43 @@ log_heap_update(Relation reln, Buffer oldbuf, ItemPointerData from,
xlhdr.t_hoff = newtup->t_data->t_hoff;
/*
- * 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 containing
+ * tuples separately from the buffer if we do logical replication that
+ * is...
*/
rdata[2].data = (char *) &xlhdr;
rdata[2].len = SizeOfHeapHeader;
- rdata[2].buffer = newbuf;
+ rdata[2].buffer = need_tuple ? InvalidBuffer : newbuf;
rdata[2].buffer_std = true;
rdata[2].next = &(rdata[3]);
/* PG73FORMAT: write bitmap [+ padding] [+ oid] + data */
rdata[3].data = (char *) newtup->t_data + offsetof(HeapTupleHeaderData, t_bits);
rdata[3].len = newtup->t_len - offsetof(HeapTupleHeaderData, t_bits);
- rdata[3].buffer = newbuf;
+ rdata[3].buffer = need_tuple ? InvalidBuffer : newbuf;
rdata[3].buffer_std = true;
rdata[3].next = NULL;
+ /*
+ * separate storage for the buffer reference of the new page in the
+ * wal_level=logical case
+ */
+ if(need_tuple){
+ rdata[3].next = &(rdata[4]);
+
+ rdata[4].data = NULL,
+ rdata[4].len = 0;
+ rdata[4].buffer = newbuf;
+ rdata[4].buffer_std = true;
+ rdata[4].next = NULL;
+ }
+
/* If new tuple is the single and first tuple on page... */
if (ItemPointerGetOffsetNumber(&(newtup->t_self)) == FirstOffsetNumber &&
PageGetMaxOffsetNumber(page) == FirstOffsetNumber)
{
info |= XLOG_HEAP_INIT_PAGE;
- rdata[2].buffer = rdata[3].buffer = InvalidBuffer;
+ rdata[2].buffer = rdata[3].buffer = rdata[4].buffer = InvalidBuffer;
}
recptr = XLogInsert(RM_HEAP_ID, info, rdata);
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/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[] = {
{"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}
};
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 = 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 bool 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 = NULL;
+ Datum indclassDatum;
+ bool isnull;
+ oidvector *indclass;
+ int i;
+ Form_pg_index indexStruct = NULL;
+
+ *indexOid = InvalidOid;
+
+ indexoidlist = RelationGetIndexList(pkrel);
+
+ foreach(indexoidscan, indexoidlist)
+ {
+ Oid indexoid = lfirst_oid(indexoidscan);
+
+ indexTuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(indexoid));
+ if(!HeapTupleIsValid(indexTuple))
+ elog(ERROR, "cache lookup failed for index %u", indexoid);
+
+ indexStruct = (Form_pg_index) GETSTRUCT(indexTuple);
+ if(indexStruct->indisprimary && indexStruct->indimmediate)
+ {
+ *indexOid = indexoid;
+ break;
+ }
+ ReleaseSysCache(indexTuple);
+
+ }
+ list_free(indexoidlist);
+
+ if (!OidIsValid(*indexOid))
+ return;
+
+ /* Must get indclass the hard way */
+ indclassDatum = SysCacheGetAttr(INDEXRELID, indexTuple,
+ Anum_pg_index_indclass, &isnull);
+ Assert(!isnull);
+ indclass = (oidvector *) DatumGetPointer(indclassDatum);
+
+ *nratts = indexStruct->indnatts;
+ /*
+ * Now build the list of PK attributes from the indkey definition (we
+ * assume a primary key cannot have expressional elements)
+ */
+ for (i = 0; i < indexStruct->indnatts; i++)
+ {
+ int pkattno = indexStruct->indkey.values[i];
+
+ attnums[i] = pkattno;
+ atttypids[i] = attnumTypeId(pkrel, pkattno);
+ opclasses[i] = indclass->values[i];
+ }
+
+ ReleaseSysCache(indexTuple);
+}
diff --git a/src/bin/pg_controldata/pg_controldata.c b/src/bin/pg_controldata/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 = 0,
WAL_LEVEL_ARCHIVE,
- WAL_LEVEL_HOT_STANDBY
+ WAL_LEVEL_HOT_STANDBY,
+ WAL_LEVEL_LOGICAL
} WalLevel;
extern int wal_level;
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);
+extern void relationFindPrimaryKey(Relation pkrel, Oid *indexOid,
+ int16 *nratts, int16 *attnums, Oid *atttypids,
+ Oid *opclasses);
+
#endif /* INDEX_H */
end of thread, other threads:[~2012-09-15 00:39 UTC | newest]
Thread overview: 5+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2012-06-13 11:28 [PATCH 07/16] Log enough data into the wal to reconstruct logical changes from it if wal_level=logical Andres Freund <andres@2ndquadrant.com>
2012-06-13 15:27 ` Kevin Grittner <Kevin.Grittner@wicourts.gov>
2012-06-13 15:57 ` Andres Freund <andres@2ndquadrant.com>
2012-06-27 13:55 ` Heikki Linnakangas <heikki.linnakangas@enterprisedb.com>
2012-09-15 00:39 [PATCH 6/8] Log enough data into the wal to reconstruct logical changes from it if wal_level=logical Andres Freund <andres@2ndquadrant.com>
This inbox is served by DDX for PostgreSQL; see mirroring instructions
for how to clone and mirror all data and code used for this inbox