rserv/lib/rserv.c

/* rserv.c
 * Support functions for erServer replication.
 * (c) 2000 Vadim Mikheev, PostgreSQL Inc.
 */

#include "executor/spi.h"               /* this is what you need to work with SPI */
#include "commands/trigger.h"   /* -"- and triggers */
#include "utils/tqual.h"                /* -"- and SnapshotData */
#include <ctype.h>                              /* tolower () */

#ifdef PG_FUNCTION_INFO_V1
#define CurrentTriggerData ((TriggerData *) fcinfo->context)
#endif

#ifdef PG_FUNCTION_INFO_V1
PG_FUNCTION_INFO_V1(_rserv_log_);
PG_FUNCTION_INFO_V1(_rserv_sync_);
PG_FUNCTION_INFO_V1(_rserv_debug_);
Datum _rserv_log_(PG_FUNCTION_ARGS);
Datum _rserv_sync_(PG_FUNCTION_ARGS);
Datum _rserv_debug_(PG_FUNCTION_ARGS);
#else
HeapTuple       _rserv_log_(void);
int32           _rserv_sync_(int32);
int32           _rserv_debug_(int32);
#endif

static int      debug = 0;

static char* OutputValue(char *key, char *buf, int size);

#ifdef PG_FUNCTION_INFO_V1
Datum
_rserv_log_(PG_FUNCTION_ARGS)
#else
HeapTuple
_rserv_log_()
#endif
{
        Trigger    *trigger;            /* to get trigger name */
        int                     nargs;                  /* # of args specified in CREATE TRIGGER */
        char      **args;                       /* argument: argnum */
        Relation        rel;                    /* triggered relation */
        HeapTuple       tuple;                  /* tuple to return */
        HeapTuple       newtuple = NULL;/* tuple to return */
        TupleDesc       tupdesc;                /* tuple description */
        int                     keynum;
        char       *key;
        char       *okey;
        char       *newkey = NULL;
        int                     deleted;
        char            sql[8192];
        char            outbuf[8192];
        char            oidbuf[64];
        int                     ret;

        /* Called by trigger manager ? */
        if (!CurrentTriggerData)
                elog(ERROR, "_rserv_log_: triggers are not initialized");

        /* Should be called for ROW trigger */
        if (TRIGGER_FIRED_FOR_STATEMENT(CurrentTriggerData->tg_event))
                elog(ERROR, "_rserv_log_: can't process STATEMENT events");

        tuple = CurrentTriggerData->tg_trigtuple;

        trigger = CurrentTriggerData->tg_trigger;
        nargs = trigger->tgnargs;
        args = trigger->tgargs;

        if (nargs != 1)                 /* odd number of arguments! */
                elog(ERROR, "_rserv_log_: need in *one* argument");

        keynum = atoi(args[0]);

        if (keynum < 0 && keynum != ObjectIdAttributeNumber)
                elog(ERROR, "_rserv_log_: invalid keynum %d", keynum);

        rel = CurrentTriggerData->tg_relation;
        tupdesc = rel->rd_att;

        deleted = (TRIGGER_FIRED_BY_DELETE(CurrentTriggerData->tg_event)) ? 
                1 : 0;

        if (TRIGGER_FIRED_BY_UPDATE(CurrentTriggerData->tg_event))
                newtuple = CurrentTriggerData->tg_newtuple;

        /*
         * Setting CurrentTriggerData to NULL prevents direct calls to trigger
         * functions in queries. Normally, trigger functions have to be called
         * by trigger manager code only.
         */
        CurrentTriggerData = NULL;

        /* Connect to SPI manager */
        if ((ret = SPI_connect()) < 0)
                elog(ERROR, "_rserv_log_: SPI_connect returned %d", ret);

        if (keynum == ObjectIdAttributeNumber)
        {
                sprintf(oidbuf, "%u", tuple->t_data->t_oid);
                key = oidbuf;
        }
        else
                key = SPI_getvalue(tuple, tupdesc, keynum);

        if (key == NULL)
                elog(ERROR, "_rserv_log_: key must be not null");

        if (newtuple && keynum != ObjectIdAttributeNumber)
        {
                newkey = SPI_getvalue(newtuple, tupdesc, keynum);
                if (newkey == NULL)
                        elog(ERROR, "_rserv_log_: key must be not null");
                if (strcmp(newkey, key) == 0)
                        newkey = NULL;
                else
                        deleted = 1;    /* old key was deleted */
        }

        if (strpbrk(key, "\\    \n'"))
                okey = OutputValue(key, outbuf, sizeof(outbuf));
        else
                okey = key;

        sprintf(sql, "update _RSERV_LOG_ set logid = %d, logtime = now(), "
                        "deleted = %d where reloid = %u and key = '%s'", 
                        GetCurrentTransactionId(), deleted, rel->rd_id, okey);

        if (debug)
                elog(NOTICE, sql);

        ret = SPI_exec(sql, 0);

        if (ret < 0)
                elog(ERROR, "_rserv_log_: SPI_exec(update) returned %d", ret);

        /*
         * If no tuple was UPDATEd then do INSERT...
         */
        if (SPI_processed > 1)
                elog(ERROR, "_rserv_log_: duplicate tuples");
        else if (SPI_processed == 0)
        {
                sprintf(sql, "insert into _RSERV_LOG_ "
                                "(reloid, logid, logtime, deleted, key) "
                                "values (%u, %d, now(), %d, '%s')",
                                rel->rd_id, GetCurrentTransactionId(), 
                                deleted, okey);

                if (debug)
                        elog(NOTICE, sql);

                ret = SPI_exec(sql, 0);

                if (ret < 0)
                        elog(ERROR, "_rserv_log_: SPI_exec(insert) returned %d", ret);
        }

        if (okey != key && okey != outbuf)
                pfree(okey);

        if (newkey)
        {
                if (strpbrk(newkey, "\\ \n'"))
                        okey = OutputValue(newkey, outbuf, sizeof(outbuf));
                else
                        okey = newkey;

                sprintf(sql, "insert into _RSERV_LOG_ "
                                "(reloid, logid, logtime, deleted, key) "
                                "values (%u, %d, now(), 0, '%s')", 
                                rel->rd_id, GetCurrentTransactionId(), okey);

                if (debug)
                        elog(NOTICE, sql);

                ret = SPI_exec(sql, 0);

                if (ret < 0)
                        elog(ERROR, "_rserv_log_: SPI_exec returned %d", ret);

                if (okey != newkey && okey != outbuf)
                        pfree(okey);
        }

        SPI_finish();

#ifdef PG_FUNCTION_INFO_V1
        return (PointerGetDatum(tuple));
#else
        return (tuple);
#endif
}

#ifdef PG_FUNCTION_INFO_V1
Datum
_rserv_sync_(PG_FUNCTION_ARGS)
#else
int32
_rserv_sync_(int32 server)
#endif
{
#ifdef PG_FUNCTION_INFO_V1
        int32 server = PG_GETARG_INT32(0);
#endif
        char    sql[8192];
        char    buf[8192];
        char   *active = buf;
        uint32  xcnt;
        int             ret;

        if (SerializableSnapshot == NULL)
                elog(ERROR, "_rserv_sync_: SerializableSnapshot is NULL");

        buf[0] = 0;
        for (xcnt = 0; xcnt < SerializableSnapshot->xcnt; xcnt++)
        {
                sprintf(buf + strlen(buf), "%s%u", (xcnt) ? ", " : "",
                        SerializableSnapshot->xip[xcnt]);
        }

        if ((ret = SPI_connect()) < 0)
                elog(ERROR, "_rserv_sync_: SPI_connect returned %d", ret);

        sprintf(sql, "insert into _RSERV_SYNC_ "
                        "(server, syncid, synctime, status, minid, maxid, active) "
                        "values (%u, currval('_rserv_sync_seq_'), now(), 0, %d, %d, '%s')", 
                        server, SerializableSnapshot->xmin, SerializableSnapshot->xmax, active);

        ret = SPI_exec(sql, 0);

        if (ret < 0)
                elog(ERROR, "_rserv_sync_: SPI_exec returned %d", ret);

        SPI_finish();

        return (0);
}

#ifdef PG_FUNCTION_INFO_V1
Datum
_rserv_debug_(PG_FUNCTION_ARGS)
#else
int32
_rserv_debug_(int32 newval)
#endif
{
#ifdef PG_FUNCTION_INFO_V1
        int32   newval = PG_GETARG_INT32(0);
#endif
        int32   oldval = debug;

        debug = newval;

        return (oldval);
}

#define ExtendBy        1024

static char*
OutputValue(char *key, char *buf, int size)
{
        int                     i = 0;
        char       *out = buf;
        char       *subst = NULL;
        int                     slen = 0;

        size--;
        for ( ; ; )
        {
                switch (*key)
                {
                        case '\\':      subst ="\\\\";
                                                slen = 2;
                                                break;
                        case '  ':      subst = "\\011";
                                                slen = 4;
                                                break;
                        case '\n':      subst = "\\012";
                                                slen = 4;
                                                break;
                        case '\'':      subst = "\\047";
                                                slen = 4;
                                                break;
                        case '\0':      out[i] = 0;
                                                return(out);
                        default:        slen = 1;
                                                break;
                }

                if (i + slen >= size)
                {
                        if (out == buf)
                        {
                                out = (char*) palloc(size + ExtendBy);
                                strncpy(out, buf, i);
                                size += ExtendBy;
                        }
                        else
                        {
                                out = (char*) repalloc(out, size + ExtendBy);
                                size += ExtendBy;
                        }
                }

                if (slen == 1)
                        out[i++] = *key;
                else
                {
                        memcpy(out + i, subst, slen);
                        i += slen;
                }
                key++;
        }

        return(out);

}
1	dpavlin	1.1	/* rserv.c
2			* Support functions for erServer replication.
3			* (c) 2000 Vadim Mikheev, PostgreSQL Inc.
4			*/
5
6			#include "executor/spi.h" /* this is what you need to work with SPI */
7			#include "commands/trigger.h" /* -"- and triggers */
8			#include "utils/tqual.h" /* -"- and SnapshotData */
9			#include <ctype.h> /* tolower () */
10
11			#ifdef PG_FUNCTION_INFO_V1
12			#define CurrentTriggerData ((TriggerData *) fcinfo->context)
13			#endif
14
15			#ifdef PG_FUNCTION_INFO_V1
16			PG_FUNCTION_INFO_V1(_rserv_log_);
17			PG_FUNCTION_INFO_V1(_rserv_sync_);
18			PG_FUNCTION_INFO_V1(_rserv_debug_);
19			Datum _rserv_log_(PG_FUNCTION_ARGS);
20			Datum _rserv_sync_(PG_FUNCTION_ARGS);
21			Datum _rserv_debug_(PG_FUNCTION_ARGS);
22			#else
23			HeapTuple _rserv_log_(void);
24			int32 _rserv_sync_(int32);
25			int32 _rserv_debug_(int32);
26			#endif
27
28			static int debug = 0;
29
30			static char* OutputValue(char key, char buf, int size);
31
32			#ifdef PG_FUNCTION_INFO_V1
33			Datum
34			_rserv_log_(PG_FUNCTION_ARGS)
35			#else
36			HeapTuple
37			_rserv_log_()
38			#endif
39			{
40			Trigger trigger; / to get trigger name */
41			int nargs; /* # of args specified in CREATE TRIGGER */
42			char *args; / argument: argnum */
43			Relation rel; /* triggered relation */
44			HeapTuple tuple; /* tuple to return */
45			HeapTuple newtuple = NULL;/* tuple to return */
46			TupleDesc tupdesc; /* tuple description */
47			int keynum;
48			char *key;
49			char *okey;
50			char *newkey = NULL;
51			int deleted;
52			char sql[8192];
53			char outbuf[8192];
54			char oidbuf[64];
55			int ret;
56
57			/* Called by trigger manager ? */
58			if (!CurrentTriggerData)
59			elog(ERROR, "_rserv_log_: triggers are not initialized");
60
61			/* Should be called for ROW trigger */
62			if (TRIGGER_FIRED_FOR_STATEMENT(CurrentTriggerData->tg_event))
63			elog(ERROR, "_rserv_log_: can't process STATEMENT events");
64
65			tuple = CurrentTriggerData->tg_trigtuple;
66
67			trigger = CurrentTriggerData->tg_trigger;
68			nargs = trigger->tgnargs;
69			args = trigger->tgargs;
70
71			if (nargs != 1) /* odd number of arguments! */
72			elog(ERROR, "_rserv_log_: need in one argument");
73
74			keynum = atoi(args[0]);
75
76			if (keynum < 0 && keynum != ObjectIdAttributeNumber)
77			elog(ERROR, "_rserv_log_: invalid keynum %d", keynum);
78
79			rel = CurrentTriggerData->tg_relation;
80			tupdesc = rel->rd_att;
81
82			deleted = (TRIGGER_FIRED_BY_DELETE(CurrentTriggerData->tg_event)) ?
83			1 : 0;
84
85			if (TRIGGER_FIRED_BY_UPDATE(CurrentTriggerData->tg_event))
86			newtuple = CurrentTriggerData->tg_newtuple;
87
88			/*
89			* Setting CurrentTriggerData to NULL prevents direct calls to trigger
90			* functions in queries. Normally, trigger functions have to be called
91			* by trigger manager code only.
92			*/
93			CurrentTriggerData = NULL;
94
95			/* Connect to SPI manager */
96			if ((ret = SPI_connect()) < 0)
97			elog(ERROR, "_rserv_log_: SPI_connect returned %d", ret);
98
99			if (keynum == ObjectIdAttributeNumber)
100			{
101			sprintf(oidbuf, "%u", tuple->t_data->t_oid);
102			key = oidbuf;
103			}
104			else
105			key = SPI_getvalue(tuple, tupdesc, keynum);
106
107			if (key == NULL)
108			elog(ERROR, "_rserv_log_: key must be not null");
109
110			if (newtuple && keynum != ObjectIdAttributeNumber)
111			{
112			newkey = SPI_getvalue(newtuple, tupdesc, keynum);
113			if (newkey == NULL)
114			elog(ERROR, "_rserv_log_: key must be not null");
115			if (strcmp(newkey, key) == 0)
116			newkey = NULL;
117			else
118			deleted = 1; /* old key was deleted */
119			}
120
121			if (strpbrk(key, "\\ \n'"))
122			okey = OutputValue(key, outbuf, sizeof(outbuf));
123			else
124			okey = key;
125
126			sprintf(sql, "update _RSERV_LOG_ set logid = %d, logtime = now(), "
127			"deleted = %d where reloid = %u and key = '%s'",
128			GetCurrentTransactionId(), deleted, rel->rd_id, okey);
129
130			if (debug)
131			elog(NOTICE, sql);
132
133			ret = SPI_exec(sql, 0);
134
135			if (ret < 0)
136			elog(ERROR, "_rserv_log_: SPI_exec(update) returned %d", ret);
137
138			/*
139			* If no tuple was UPDATEd then do INSERT...
140			*/
141			if (SPI_processed > 1)
142			elog(ERROR, "_rserv_log_: duplicate tuples");
143			else if (SPI_processed == 0)
144			{
145			sprintf(sql, "insert into _RSERV_LOG_ "
146			"(reloid, logid, logtime, deleted, key) "
147			"values (%u, %d, now(), %d, '%s')",
148			rel->rd_id, GetCurrentTransactionId(),
149			deleted, okey);
150
151			if (debug)
152			elog(NOTICE, sql);
153
154			ret = SPI_exec(sql, 0);
155
156			if (ret < 0)
157			elog(ERROR, "_rserv_log_: SPI_exec(insert) returned %d", ret);
158			}
159
160			if (okey != key && okey != outbuf)
161			pfree(okey);
162
163			if (newkey)
164			{
165			if (strpbrk(newkey, "\\ \n'"))
166			okey = OutputValue(newkey, outbuf, sizeof(outbuf));
167			else
168			okey = newkey;
169
170			sprintf(sql, "insert into _RSERV_LOG_ "
171			"(reloid, logid, logtime, deleted, key) "
172			"values (%u, %d, now(), 0, '%s')",
173			rel->rd_id, GetCurrentTransactionId(), okey);
174
175			if (debug)
176			elog(NOTICE, sql);
177
178			ret = SPI_exec(sql, 0);
179
180			if (ret < 0)
181			elog(ERROR, "_rserv_log_: SPI_exec returned %d", ret);
182
183			if (okey != newkey && okey != outbuf)
184			pfree(okey);
185			}
186
187			SPI_finish();
188
189			#ifdef PG_FUNCTION_INFO_V1
190			return (PointerGetDatum(tuple));
191			#else
192			return (tuple);
193			#endif
194			}
195
196			#ifdef PG_FUNCTION_INFO_V1
197			Datum
198			_rserv_sync_(PG_FUNCTION_ARGS)
199			#else
200			int32
201			_rserv_sync_(int32 server)
202			#endif
203			{
204			#ifdef PG_FUNCTION_INFO_V1
205			int32 server = PG_GETARG_INT32(0);
206			#endif
207			char sql[8192];
208			char buf[8192];
209			char *active = buf;
210			uint32 xcnt;
211			int ret;
212
213			if (SerializableSnapshot == NULL)
214			elog(ERROR, "_rserv_sync_: SerializableSnapshot is NULL");
215
216			buf[0] = 0;
217			for (xcnt = 0; xcnt < SerializableSnapshot->xcnt; xcnt++)
218			{
219			sprintf(buf + strlen(buf), "%s%u", (xcnt) ? ", " : "",
220			SerializableSnapshot->xip[xcnt]);
221			}
222
223			if ((ret = SPI_connect()) < 0)
224			elog(ERROR, "_rserv_sync_: SPI_connect returned %d", ret);
225
226			sprintf(sql, "insert into _RSERV_SYNC_ "
227			"(server, syncid, synctime, status, minid, maxid, active) "
228			"values (%u, currval('_rserv_sync_seq_'), now(), 0, %d, %d, '%s')",
229			server, SerializableSnapshot->xmin, SerializableSnapshot->xmax, active);
230
231			ret = SPI_exec(sql, 0);
232
233			if (ret < 0)
234			elog(ERROR, "_rserv_sync_: SPI_exec returned %d", ret);
235
236			SPI_finish();
237
238			return (0);
239			}
240
241			#ifdef PG_FUNCTION_INFO_V1
242			Datum
243			_rserv_debug_(PG_FUNCTION_ARGS)
244			#else
245			int32
246			_rserv_debug_(int32 newval)
247			#endif
248			{
249			#ifdef PG_FUNCTION_INFO_V1
250			int32 newval = PG_GETARG_INT32(0);
251			#endif
252			int32 oldval = debug;
253
254			debug = newval;
255
256			return (oldval);
257			}
258
259			#define ExtendBy 1024
260
261			static char*
262			OutputValue(char key, char buf, int size)
263			{
264			int i = 0;
265			char *out = buf;
266			char *subst = NULL;
267			int slen = 0;
268
269			size--;
270			for ( ; ; )
271			{
272			switch (*key)
273			{
274			case '\\': subst ="\\\\";
275			slen = 2;
276			break;
277			case ' ': subst = "\\011";
278			slen = 4;
279			break;
280			case '\n': subst = "\\012";
281			slen = 4;
282			break;
283			case '\'': subst = "\\047";
284			slen = 4;
285			break;
286			case '\0': out[i] = 0;
287			return(out);
288			default: slen = 1;
289			break;
290			}
291
292			if (i + slen >= size)
293			{
294			if (out == buf)
295			{
296			out = (char*) palloc(size + ExtendBy);
297			strncpy(out, buf, i);
298			size += ExtendBy;
299			}
300			else
301			{
302			out = (char*) repalloc(out, size + ExtendBy);
303			size += ExtendBy;
304			}
305			}
306
307			if (slen == 1)
308			out[i++] = *key;
309			else
310			{
311			memcpy(out + i, subst, slen);
312			i += slen;
313			}
314			key++;
315			}
316
317			return(out);
318
319			}