public inbox for [email protected]
help / color / mirror / Atom feed[PATCH] Fix a bug of physical replication slot.
5+ messages / 4 participants
[nested] [flat]
* [PATCH] Fix a bug of physical replication slot.
@ 2017-02-01 07:07 Kyotaro Horiguchi <[email protected]>
0 siblings, 0 replies; 5+ messages in thread
From: Kyotaro Horiguchi @ 2017-02-01 07:07 UTC (permalink / raw)
A physical-replication standby can stop just at the boundary of WAL
segments. restart_lsn of the slot on the master can be assumed to be
the same location. The last segment on the master will be removed
after some checkpoints for the case. If the first record of the next
segment is a continuation record, it is only on the master and its
beginning is only on the standby so the standby cannot restart because
the record to read is scattered to two sources.
This patch detains restart_lsn in the last sgement when the first page
of the next segment is a continuation record.
---
src/backend/replication/walsender.c | 107 +++++++++++++++++++++++++++++++++---
1 file changed, 100 insertions(+), 7 deletions(-)
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 5909b7d..cfbe70e 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -188,6 +188,12 @@ static volatile sig_atomic_t replication_active = false;
static LogicalDecodingContext *logical_decoding_ctx = NULL;
static XLogRecPtr logical_startptr = InvalidXLogRecPtr;
+/*
+ * Segment keep pointer for physical slots. Has a valid value only when it
+ * differs from the current flush pointer.
+ */
+static XLogRecPtr keepPtr = InvalidXLogRecPtr;
+
/* Signal handlers */
static void WalSndSigHupHandler(SIGNAL_ARGS);
static void WalSndXLogSendHandler(SIGNAL_ARGS);
@@ -220,7 +226,7 @@ static void WalSndPrepareWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, Tran
static void WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write);
static XLogRecPtr WalSndWaitForWal(XLogRecPtr loc);
-static void XLogRead(char *buf, XLogRecPtr startptr, Size count);
+static bool XLogRead(char *buf, XLogRecPtr startptr, Size count, bool noutfoundok);
/* Initialize walsender process before entering the main command loop */
@@ -541,6 +547,9 @@ StartReplication(StartReplicationCmd *cmd)
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
(errmsg("cannot use a logical replication slot for physical replication"))));
+
+ /* Restore keepPtr from replication slot */
+ keepPtr = MyReplicationSlot->data.restart_lsn;
}
/*
@@ -556,6 +565,10 @@ StartReplication(StartReplicationCmd *cmd)
else
FlushPtr = GetFlushRecPtr();
+ /* Set InvalidXLogRecPtr if catching up */
+ if (keepPtr == FlushPtr)
+ keepPtr = InvalidXLogRecPtr;
+
if (cmd->timeline != 0)
{
XLogRecPtr switchpoint;
@@ -777,7 +790,7 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req
count = flushptr - targetPagePtr;
/* now actually read the data, we know it's there */
- XLogRead(cur_page, targetPagePtr, XLOG_BLCKSZ);
+ XLogRead(cur_page, targetPagePtr, XLOG_BLCKSZ, false);
return count;
}
@@ -1563,7 +1576,7 @@ static void
ProcessStandbyReplyMessage(void)
{
XLogRecPtr writePtr,
- flushPtr,
+ flushPtr, oldFlushPtr,
applyPtr;
bool replyRequested;
@@ -1592,6 +1605,7 @@ ProcessStandbyReplyMessage(void)
WalSnd *walsnd = MyWalSnd;
SpinLockAcquire(&walsnd->mutex);
+ oldFlushPtr = walsnd->flush;
walsnd->write = writePtr;
walsnd->flush = flushPtr;
walsnd->apply = applyPtr;
@@ -1609,7 +1623,78 @@ ProcessStandbyReplyMessage(void)
if (SlotIsLogical(MyReplicationSlot))
LogicalConfirmReceivedLocation(flushPtr);
else
- PhysicalConfirmReceivedLocation(flushPtr);
+ {
+ /*
+ * On recovery, a continuation reocrd must be available from
+ * single WAL source. So physical replication slot should stay in
+ * the first segment for a continuation record spanning multiple
+ * segments. Since this doesn't look into individual record,
+ * keepPtr may stay a bit too behind.
+ *
+ * Since the objective is avoding to remove required segments,
+ * checking every segment is enough. But once keepPtr goes behind,
+ * check every page for quick restoration.
+ *
+ * keepPtr has a valid value only when it is behind flushPtr.
+ */
+ if (oldFlushPtr != InvalidXLogRecPtr &&
+ (keepPtr == InvalidXLogRecPtr ?
+ oldFlushPtr / XLOG_SEG_SIZE != flushPtr / XLOG_SEG_SIZE :
+ keepPtr / XLOG_BLCKSZ != flushPtr / XLOG_BLCKSZ))
+ {
+ XLogRecPtr rp;
+ XLogRecPtr oldKeepPtr = keepPtr; /* for debug */
+
+ if (keepPtr == InvalidXLogRecPtr)
+ keepPtr = oldFlushPtr;
+
+ rp = keepPtr - (keepPtr % XLOG_BLCKSZ);
+
+ /*
+ * We may have let the record at flushPtr sent, so it's worth
+ * looking
+ */
+ while (rp <= flushPtr)
+ {
+ XLogPageHeaderData header;
+
+ /*
+ * If the page header is not available for now, don't move
+ * keepPtr forward. We can read it by the next chance.
+ */
+ if(sentPtr - rp >= sizeof(XLogPageHeaderData))
+ {
+ bool found;
+ /*
+ * Fetch the page header of the next page. Move
+ * keepPtr forward only if when it is not a
+ * continuation page.
+ */
+ found = XLogRead((char *)&header, rp,
+ sizeof(XLogPageHeaderData), true);
+ if (found &&
+ (header.xlp_info & XLP_FIRST_IS_CONTRECORD) == 0)
+ keepPtr = rp;
+ }
+ rp += XLOG_BLCKSZ;
+ }
+
+ /*
+ * If keepPtr is on the same page with flushPtr, it means that
+ * we are catching up
+ */
+ if (keepPtr / XLOG_BLCKSZ == flushPtr / XLOG_BLCKSZ)
+ keepPtr = InvalidXLogRecPtr;
+
+ if (oldKeepPtr != keepPtr)
+ elog(LOG, "%lX => %lX / %lX",
+ oldKeepPtr, keepPtr, flushPtr);
+ }
+
+ /* keepPtr == InvalidXLogRecPtr means catching up */
+ PhysicalConfirmReceivedLocation(keepPtr != InvalidXLogRecPtr ?
+ keepPtr : flushPtr);
+ }
}
}
@@ -2031,6 +2116,7 @@ WalSndKill(int code, Datum arg)
/*
* Read 'count' bytes from WAL into 'buf', starting at location 'startptr'
+ * Returns false if the segment file is not found when notfoundok is true.
*
* XXX probably this should be improved to suck data directly from the
* WAL buffers when possible.
@@ -2040,8 +2126,8 @@ WalSndKill(int code, Datum arg)
* always be one descriptor left open until the process ends, but never
* more than one.
*/
-static void
-XLogRead(char *buf, XLogRecPtr startptr, Size count)
+static bool
+XLogRead(char *buf, XLogRecPtr startptr, Size count, bool notfoundok)
{
char *p;
XLogRecPtr recptr;
@@ -2118,10 +2204,15 @@ retry:
* removed or recycled.
*/
if (errno == ENOENT)
+ {
+ if (notfoundok)
+ return false;
+
ereport(ERROR,
(errcode_for_file_access(),
errmsg("requested WAL segment %s has already been removed",
XLogFileNameP(curFileTimeLine, sendSegNo))));
+ }
else
ereport(ERROR,
(errcode_for_file_access(),
@@ -2201,6 +2292,8 @@ retry:
goto retry;
}
}
+
+ return true;
}
/*
@@ -2405,7 +2498,7 @@ XLogSendPhysical(void)
* calls.
*/
enlargeStringInfo(&output_message, nbytes);
- XLogRead(&output_message.data[output_message.len], startptr, nbytes);
+ XLogRead(&output_message.data[output_message.len], startptr, nbytes, false);
output_message.len += nbytes;
output_message.data[output_message.len] = '\0';
--
2.9.2
----Next_Part(Wed_Feb_01_16_41_38_2017_419)--
Content-Type: text/plain
Content-Disposition: inline
Content-Transfer-Encoding: 8bit
MIME-Version: 1.0
--
Sent via pgsql-bugs mailing list ([email protected])
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-bugs
----Next_Part(Wed_Feb_01_16_41_38_2017_419)----
^ permalink raw reply [nested|flat] 5+ messages in thread
* [PATCH] Fix a bug of physical replication slot.
@ 2017-02-01 07:07 Kyotaro Horiguchi <[email protected]>
0 siblings, 0 replies; 5+ messages in thread
From: Kyotaro Horiguchi @ 2017-02-01 07:07 UTC (permalink / raw)
A physical-replication standby can stop just at the boundary of WAL
segments. restart_lsn of the slot on the master can be assumed to be
the same location. The last segment on the master will be removed
after some checkpoints for the case. If the first record of the next
segment is a continuation record, it is only on the master and its
beginning is only on the standby so the standby cannot restart because
the record to read is scattered to two sources.
This patch detains restart_lsn in the last sgement when the first page
of the next segment is a continuation record.
---
src/backend/replication/walsender.c | 104 +++++++++++++++++++++++++++++++++---
1 file changed, 97 insertions(+), 7 deletions(-)
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 76f09fb..0ec7ba9 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -188,6 +188,13 @@ static volatile sig_atomic_t replication_active = false;
static LogicalDecodingContext *logical_decoding_ctx = NULL;
static XLogRecPtr logical_startptr = InvalidXLogRecPtr;
+/*
+ * This variable corresponds to restart_lsn in pg_replication_slots for a
+ * physical slot. This has a valid value only when it differs from the current
+ * flush pointer.
+ */
+static XLogRecPtr restartLSN = InvalidXLogRecPtr;
+
/* Signal handlers */
static void WalSndSigHupHandler(SIGNAL_ARGS);
static void WalSndXLogSendHandler(SIGNAL_ARGS);
@@ -220,7 +227,7 @@ static void WalSndPrepareWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, Tran
static void WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write);
static XLogRecPtr WalSndWaitForWal(XLogRecPtr loc);
-static void XLogRead(char *buf, XLogRecPtr startptr, Size count);
+static bool XLogRead(char *buf, XLogRecPtr startptr, Size count, bool noutfoundok);
/* Initialize walsender process before entering the main command loop */
@@ -504,6 +511,9 @@ StartReplication(StartReplicationCmd *cmd)
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
(errmsg("cannot use a logical replication slot for physical replication"))));
+
+ /* Restore restartLSN from replication slot */
+ restartLSN = MyReplicationSlot->data.restart_lsn;
}
/*
@@ -519,6 +529,10 @@ StartReplication(StartReplicationCmd *cmd)
else
FlushPtr = GetFlushRecPtr();
+ /* Set InvalidXLogRecPtr if catching up */
+ if (restartLSN == FlushPtr)
+ restartLSN = InvalidXLogRecPtr;
+
if (cmd->timeline != 0)
{
XLogRecPtr switchpoint;
@@ -727,7 +741,7 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req
count = flushptr - targetPagePtr;
/* now actually read the data, we know it's there */
- XLogRead(cur_page, targetPagePtr, XLOG_BLCKSZ);
+ XLogRead(cur_page, targetPagePtr, XLOG_BLCKSZ, false);
return count;
}
@@ -1486,7 +1500,7 @@ static void
ProcessStandbyReplyMessage(void)
{
XLogRecPtr writePtr,
- flushPtr,
+ flushPtr, oldFlushPtr,
applyPtr;
bool replyRequested;
@@ -1515,6 +1529,7 @@ ProcessStandbyReplyMessage(void)
WalSnd *walsnd = MyWalSnd;
SpinLockAcquire(&walsnd->mutex);
+ oldFlushPtr = walsnd->flush;
walsnd->write = writePtr;
walsnd->flush = flushPtr;
walsnd->apply = applyPtr;
@@ -1532,7 +1547,74 @@ ProcessStandbyReplyMessage(void)
if (SlotIsLogical(MyReplicationSlot))
LogicalConfirmReceivedLocation(flushPtr);
else
- PhysicalConfirmReceivedLocation(flushPtr);
+ {
+ /*
+ * Recovery on standby requires that a continuation reocrd must be
+ * available from single WAL source. For the reason, physical
+ * replication slot should stay in the first segment for a
+ * continuation record spanning multiple segments. Since this
+ * doesn't look into individual record, restartLSN may stay a bit
+ * too behind.
+ *
+ * Since the objective is avoding to remove required segments,
+ * checking at the beginning of every segment is enough. But once
+ * restartLSN goes behind, check every page for quick restoration.
+ *
+ * restartLSN has a valid value only when it is behind flushPtr.
+ */
+ if (oldFlushPtr != InvalidXLogRecPtr &&
+ (restartLSN == InvalidXLogRecPtr ?
+ oldFlushPtr / XLOG_SEG_SIZE != flushPtr / XLOG_SEG_SIZE :
+ restartLSN / XLOG_BLCKSZ != flushPtr / XLOG_BLCKSZ))
+ {
+ XLogRecPtr rp;
+
+ if (restartLSN == InvalidXLogRecPtr)
+ restartLSN = oldFlushPtr;
+
+ rp = restartLSN - (restartLSN % XLOG_BLCKSZ);
+
+ /*
+ * We may have let the record at flushPtr sent, so it's worth
+ * looking
+ */
+ while (rp <= flushPtr)
+ {
+ XLogPageHeaderData header;
+
+ /*
+ * If the page header is not available for now, don't move
+ * restartLSN forward. We can read it by the next chance.
+ */
+ if(sentPtr - rp >= sizeof(XLogPageHeaderData))
+ {
+ bool found;
+ /*
+ * Fetch the page header of the next page. Move
+ * restartLSN forward only if it is not a continuation
+ * page.
+ */
+ found = XLogRead((char *)&header, rp,
+ sizeof(XLogPageHeaderData), true);
+ if (found &&
+ (header.xlp_info & XLP_FIRST_IS_CONTRECORD) == 0)
+ restartLSN = rp;
+ }
+ rp += XLOG_BLCKSZ;
+ }
+
+ /*
+ * If restartLSN is on the same page with flushPtr, it means
+ * that we are catching up.
+ */
+ if (restartLSN / XLOG_BLCKSZ == flushPtr / XLOG_BLCKSZ)
+ restartLSN = InvalidXLogRecPtr;
+ }
+
+ /* restartLSN == InvalidXLogRecPtr means catching up */
+ PhysicalConfirmReceivedLocation(restartLSN != InvalidXLogRecPtr ?
+ restartLSN : flushPtr);
+ }
}
}
@@ -1954,6 +2036,7 @@ WalSndKill(int code, Datum arg)
/*
* Read 'count' bytes from WAL into 'buf', starting at location 'startptr'
+ * Returns false if the segment file is not found when notfoundok is true.
*
* XXX probably this should be improved to suck data directly from the
* WAL buffers when possible.
@@ -1963,8 +2046,8 @@ WalSndKill(int code, Datum arg)
* always be one descriptor left open until the process ends, but never
* more than one.
*/
-static void
-XLogRead(char *buf, XLogRecPtr startptr, Size count)
+static bool
+XLogRead(char *buf, XLogRecPtr startptr, Size count, bool notfoundok)
{
char *p;
XLogRecPtr recptr;
@@ -2041,10 +2124,15 @@ retry:
* removed or recycled.
*/
if (errno == ENOENT)
+ {
+ if (notfoundok)
+ return false;
+
ereport(ERROR,
(errcode_for_file_access(),
errmsg("requested WAL segment %s has already been removed",
XLogFileNameP(curFileTimeLine, sendSegNo))));
+ }
else
ereport(ERROR,
(errcode_for_file_access(),
@@ -2124,6 +2212,8 @@ retry:
goto retry;
}
}
+
+ return true;
}
/*
@@ -2328,7 +2418,7 @@ XLogSendPhysical(void)
* calls.
*/
enlargeStringInfo(&output_message, nbytes);
- XLogRead(&output_message.data[output_message.len], startptr, nbytes);
+ XLogRead(&output_message.data[output_message.len], startptr, nbytes, false);
output_message.len += nbytes;
output_message.data[output_message.len] = '\0';
--
2.9.2
----Next_Part(Fri_Feb_03_12_16_49_2017_952)--
Content-Type: text/plain
Content-Disposition: inline
Content-Transfer-Encoding: 8bit
MIME-Version: 1.0
--
Sent via pgsql-hackers mailing list ([email protected])
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers
----Next_Part(Fri_Feb_03_12_16_49_2017_952)----
^ permalink raw reply [nested|flat] 5+ messages in thread
* [PATCH v24 09/10] new compression method extension for zlib
@ 2020-10-16 14:26 dilipkumar <[email protected]>
0 siblings, 0 replies; 5+ messages in thread
From: dilipkumar @ 2020-10-16 14:26 UTC (permalink / raw)
Dilip Kumar
---
contrib/Makefile | 1 +
contrib/cmzlib/.gitignore | 4 +
contrib/cmzlib/Makefile | 26 +++++
contrib/cmzlib/cmzlib--1.0.sql | 13 +++
contrib/cmzlib/cmzlib.c | 157 +++++++++++++++++++++++++++++
contrib/cmzlib/cmzlib.control | 5 +
contrib/cmzlib/expected/cmzlib.out | 53 ++++++++++
contrib/cmzlib/sql/cmzlib.sql | 22 ++++
8 files changed, 281 insertions(+)
create mode 100644 contrib/cmzlib/.gitignore
create mode 100644 contrib/cmzlib/Makefile
create mode 100644 contrib/cmzlib/cmzlib--1.0.sql
create mode 100644 contrib/cmzlib/cmzlib.c
create mode 100644 contrib/cmzlib/cmzlib.control
create mode 100644 contrib/cmzlib/expected/cmzlib.out
create mode 100644 contrib/cmzlib/sql/cmzlib.sql
diff --git a/contrib/Makefile b/contrib/Makefile
index f27e458482..9e452d8dd0 100644
--- a/contrib/Makefile
+++ b/contrib/Makefile
@@ -12,6 +12,7 @@ SUBDIRS = \
bloom \
btree_gin \
btree_gist \
+ cmzlib \
citext \
cube \
dblink \
diff --git a/contrib/cmzlib/.gitignore b/contrib/cmzlib/.gitignore
new file mode 100644
index 0000000000..5dcb3ff972
--- /dev/null
+++ b/contrib/cmzlib/.gitignore
@@ -0,0 +1,4 @@
+# Generated subdirectories
+/log/
+/results/
+/tmp_check/
diff --git a/contrib/cmzlib/Makefile b/contrib/cmzlib/Makefile
new file mode 100644
index 0000000000..956fbe7cc8
--- /dev/null
+++ b/contrib/cmzlib/Makefile
@@ -0,0 +1,26 @@
+# contrib/cmzlib/Makefile
+
+MODULE_big = cmzlib
+OBJS = \
+ $(WIN32RES) \
+ cmzlib.o
+
+EXTENSION = cmzlib
+DATA = cmzlib--1.0.sql
+PGFILEDESC = "zlib compression method "
+
+SHLIB_LINK += $(filter -lz, $(LIBS))
+
+REGRESS = cmzlib
+
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = contrib/cmzlib
+top_builddir = ../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/contrib/cmzlib/cmzlib--1.0.sql b/contrib/cmzlib/cmzlib--1.0.sql
new file mode 100644
index 0000000000..41f2f95870
--- /dev/null
+++ b/contrib/cmzlib/cmzlib--1.0.sql
@@ -0,0 +1,13 @@
+/* contrib/cm_lz4/cmzlib--1.0.sql */
+
+-- complain if script is sourced in psql, rather than via CREATE EXTENSION
+\echo Use "CREATE EXTENSION cmzlib" to load this file. \quit
+
+CREATE FUNCTION zlibhandler(internal)
+RETURNS compression_am_handler
+AS 'MODULE_PATHNAME'
+LANGUAGE C;
+
+-- Compression method
+CREATE ACCESS METHOD zlib TYPE COMPRESSION HANDLER zlibhandler;
+COMMENT ON ACCESS METHOD zlib IS 'zlib compression method';
diff --git a/contrib/cmzlib/cmzlib.c b/contrib/cmzlib/cmzlib.c
new file mode 100644
index 0000000000..686a7c7e0d
--- /dev/null
+++ b/contrib/cmzlib/cmzlib.c
@@ -0,0 +1,157 @@
+/*-------------------------------------------------------------------------
+ *
+ * cmzlib.c
+ * zlib compression method
+ *
+ * Copyright (c) 2015-2018, PostgreSQL Global Development Group
+ *
+ *
+ * IDENTIFICATION
+ * contrib/cmzlib/cmzlib.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+#include "access/compressamapi.h"
+#include "access/toast_internals.h"
+
+#include "fmgr.h"
+#include "utils/builtins.h"
+
+#include <zlib.h>
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(zlibhandler);
+
+void _PG_init(void);
+
+/*
+ * Module initialize function: initialize info about zlib
+ */
+void
+_PG_init(void)
+{
+
+}
+
+#define ZLIB_MAX_DICTIONARY_LENGTH 32768
+#define ZLIB_DICTIONARY_DELIM (" ,")
+
+typedef struct
+{
+ int level;
+ Bytef dict[ZLIB_MAX_DICTIONARY_LENGTH];
+ unsigned int dictlen;
+} zlib_state;
+
+/*
+ * zlib_cmcompress - compression routine for zlib compression method
+ *
+ * Compresses source into dest using the default compression level.
+ * Returns the compressed varlena, or NULL if compression fails.
+ */
+static struct varlena *
+zlib_cmcompress(const struct varlena *value, int32 header_size)
+{
+ int32 valsize,
+ len;
+ struct varlena *tmp = NULL;
+ z_streamp zp;
+ int res;
+ zlib_state state;
+
+ state.level = Z_DEFAULT_COMPRESSION;
+
+ zp = (z_streamp) palloc(sizeof(z_stream));
+ zp->zalloc = Z_NULL;
+ zp->zfree = Z_NULL;
+ zp->opaque = Z_NULL;
+
+ if (deflateInit(zp, state.level) != Z_OK)
+ elog(ERROR, "could not initialize compression library: %s", zp->msg);
+
+ valsize = VARSIZE_ANY_EXHDR(DatumGetPointer(value));
+ tmp = (struct varlena *) palloc(valsize + header_size);
+ zp->next_in = (void *) VARDATA_ANY(value);
+ zp->avail_in = valsize;
+ zp->avail_out = valsize;
+ zp->next_out = (void *) ((char *) tmp + header_size);
+
+ do
+ {
+ res = deflate(zp, Z_FINISH);
+ if (res == Z_STREAM_ERROR)
+ elog(ERROR, "could not compress data: %s", zp->msg);
+ } while (zp->avail_in != 0);
+
+ Assert(res == Z_STREAM_END);
+
+ len = valsize - zp->avail_out;
+ if (deflateEnd(zp) != Z_OK)
+ elog(ERROR, "could not close compression stream: %s", zp->msg);
+ pfree(zp);
+
+ if (len > 0)
+ {
+ SET_VARSIZE_COMPRESSED(tmp, len + header_size);
+ return tmp;
+ }
+
+ pfree(tmp);
+ return NULL;
+}
+
+/*
+ * zlib_cmdecompress - decompression routine for zlib compression method
+ *
+ * Returns the decompressed varlena.
+ */
+static struct varlena *
+zlib_cmdecompress(const struct varlena *value, int32 header_size)
+{
+ struct varlena *result;
+ z_streamp zp;
+ int res = Z_OK;
+
+ zp = (z_streamp) palloc(sizeof(z_stream));
+ zp->zalloc = Z_NULL;
+ zp->zfree = Z_NULL;
+ zp->opaque = Z_NULL;
+
+ if (inflateInit(zp) != Z_OK)
+ elog(ERROR, "could not initialize compression library: %s", zp->msg);
+
+ zp->next_in = (void *) ((char *) value + header_size);
+ zp->avail_in = VARSIZE(value) - header_size;
+ zp->avail_out = VARRAWSIZE_4B_C(value);
+
+ result = (struct varlena *) palloc(zp->avail_out + VARHDRSZ);
+ SET_VARSIZE(result, zp->avail_out + VARHDRSZ);
+ zp->next_out = (void *) VARDATA(result);
+
+ while (zp->avail_in > 0)
+ {
+ res = inflate(zp, 0);
+ if (!(res == Z_OK || res == Z_STREAM_END))
+ elog(ERROR, "could not uncompress data: %s", zp->msg);
+ }
+
+ if (inflateEnd(zp) != Z_OK)
+ elog(ERROR, "could not close compression library: %s", zp->msg);
+
+ pfree(zp);
+ return result;
+}
+
+const CompressionAmRoutine zlib_compress_methods = {
+ .type = T_CompressionAmRoutine,
+ .datum_compress = zlib_cmcompress,
+ .datum_decompress = zlib_cmdecompress,
+ .datum_decompress_slice = NULL};
+
+Datum
+zlibhandler(PG_FUNCTION_ARGS)
+{
+ PG_RETURN_POINTER(&zlib_compress_methods);
+}
diff --git a/contrib/cmzlib/cmzlib.control b/contrib/cmzlib/cmzlib.control
new file mode 100644
index 0000000000..2eb10f3a83
--- /dev/null
+++ b/contrib/cmzlib/cmzlib.control
@@ -0,0 +1,5 @@
+# cm_lz4 extension
+comment = 'cmzlib compression method'
+default_version = '1.0'
+module_pathname = '$libdir/cmzlib'
+relocatable = true
diff --git a/contrib/cmzlib/expected/cmzlib.out b/contrib/cmzlib/expected/cmzlib.out
new file mode 100644
index 0000000000..2b6fac7e0b
--- /dev/null
+++ b/contrib/cmzlib/expected/cmzlib.out
@@ -0,0 +1,53 @@
+CREATE EXTENSION cmzlib;
+-- zlib compression
+CREATE TABLE zlibtest(f1 TEXT COMPRESSION pglz);
+INSERT INTO zlibtest VALUES(repeat('1234567890',1004));
+INSERT INTO zlibtest VALUES(repeat('1234567890 one two three',1004));
+SELECT length(f1) FROM zlibtest;
+ length
+--------
+ 10040
+ 24096
+(2 rows)
+
+-- alter compression method with rewrite
+ALTER TABLE zlibtest ALTER COLUMN f1 SET COMPRESSION pglz;
+\d+ zlibtest
+ Table "public.zlibtest"
+ Column | Type | Collation | Nullable | Default | Storage | Compression | Stats target | Description
+--------+------+-----------+----------+---------+----------+-------------+--------------+-------------
+ f1 | text | | | | extended | pglz | |
+
+ALTER TABLE zlibtest ALTER COLUMN f1 SET COMPRESSION zlib;
+\d+ zlibtest
+ Table "public.zlibtest"
+ Column | Type | Collation | Nullable | Default | Storage | Compression | Stats target | Description
+--------+------+-----------+----------+---------+----------+-------------+--------------+-------------
+ f1 | text | | | | extended | zlib | |
+
+-- preserve old compression method
+ALTER TABLE zlibtest ALTER COLUMN f1 SET COMPRESSION pglz PRESERVE (zlib);
+INSERT INTO zlibtest VALUES (repeat('1234567890',1004));
+\d+ zlibtest
+ Table "public.zlibtest"
+ Column | Type | Collation | Nullable | Default | Storage | Compression | Stats target | Description
+--------+------+-----------+----------+---------+----------+-------------+--------------+-------------
+ f1 | text | | | | extended | pglz | |
+
+SELECT pg_column_compression(f1) FROM zlibtest;
+ pg_column_compression
+-----------------------
+ zlib
+ zlib
+ pglz
+(3 rows)
+
+SELECT length(f1) FROM zlibtest;
+ length
+--------
+ 10040
+ 24096
+ 10040
+(3 rows)
+
+DROP TABLE zlibtest;
diff --git a/contrib/cmzlib/sql/cmzlib.sql b/contrib/cmzlib/sql/cmzlib.sql
new file mode 100644
index 0000000000..ea8d206625
--- /dev/null
+++ b/contrib/cmzlib/sql/cmzlib.sql
@@ -0,0 +1,22 @@
+CREATE EXTENSION cmzlib;
+
+-- zlib compression
+CREATE TABLE zlibtest(f1 TEXT COMPRESSION pglz);
+INSERT INTO zlibtest VALUES(repeat('1234567890',1004));
+INSERT INTO zlibtest VALUES(repeat('1234567890 one two three',1004));
+SELECT length(f1) FROM zlibtest;
+
+-- alter compression method with rewrite
+ALTER TABLE zlibtest ALTER COLUMN f1 SET COMPRESSION pglz;
+\d+ zlibtest
+ALTER TABLE zlibtest ALTER COLUMN f1 SET COMPRESSION zlib;
+\d+ zlibtest
+
+-- preserve old compression method
+ALTER TABLE zlibtest ALTER COLUMN f1 SET COMPRESSION pglz PRESERVE (zlib);
+INSERT INTO zlibtest VALUES (repeat('1234567890',1004));
+\d+ zlibtest
+SELECT pg_column_compression(f1) FROM zlibtest;
+SELECT length(f1) FROM zlibtest;
+
+DROP TABLE zlibtest;
--
2.17.0
--m51xatjYGsM+13rf
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v24-0010-Support-compression-methods-options.patch"
^ permalink raw reply [nested|flat] 5+ messages in thread
* ERROR: type of parameter 1 (fruit2) does not match that when preparing the plan (fruit1)
@ 2022-04-30 23:23 d <[email protected]>
0 siblings, 1 reply; 5+ messages in thread
From: d @ 2022-04-30 23:23 UTC (permalink / raw)
To: [email protected]
Hi
This function should produce form_urlencoded data from a row.
It works on the first invoction.
but the second fails and references the table from the previous invocation.
In this case I am using pgbouncer but I have tested it without
and also without the urldecode on several platforms (pg13)
thanks,
dh
-------to reproduce --------------------------------------
CREATE OR REPLACE FUNCTION record_to_form_data(p_r record)
RETURNS text
LANGUAGE plpgsql
AS $function$
begin
return (
select string_agg(format('%s=%s',key,urlencode(value)),'&')
from
(select p_r.*) i,
hstore(i.*) as h,each(h) );
end;
$function$;
create table fruit1(id varchar not null,name varchar not null,color varchar);
create table fruit2(id varchar not null,name varchar not null);
insert into fruit1 values('1','apple','red');
insert into fruit2 values('1','apple');
select record_to_form_data(f.*) from fruit1 f;
select record_to_form_data(f.*) from fruit2 f;
--------------------------------
testit6=# select record_to_form_data(f.*) from fruit1 f;
record_to_form_data
---------------------------
id=1&name=apple&color=red
(1 row)
testit6=# select record_to_form_data(f.*) from fruit2 f;
ERROR: type of parameter 1 (fruit2) does not match that when preparing the plan (fruit1)
CONTEXT: SQL statement "SELECT (
select string_agg(format('%s=%s',key,urlencode(value)),'&')
from
(select p_r.*) i,
hstore(i.*) as h,each(h) )"
PL/pgSQL function record_to_form_data(record) line 6 at RETURN
testit6=# \c
psql (13.5 (Debian 13.5-0+deb11u1), server 13.6 (Debian 13.6-1.pgdg110+1))
SSL connection (protocol: TLSv1.3, cipher: TLS_AES_256_GCM_SHA384, bits: 256, compression: off)
You are now connected to database "testit6" as user "david".
testit6=# select record_to_form_data(f.*) from fruit2 f;
record_to_form_data
---------------------
id=1&name=apple
(1 row)
^ permalink raw reply [nested|flat] 5+ messages in thread
* Re: ERROR: type of parameter 1 (fruit2) does not match that when preparing the plan (fruit1)
@ 2022-05-01 15:58 David G. Johnston <[email protected]>
parent: d <[email protected]>
0 siblings, 0 replies; 5+ messages in thread
From: David G. Johnston @ 2022-05-01 15:58 UTC (permalink / raw)
To: d <[email protected]>; +Cc: PostgreSQL mailing lists <[email protected]>
On Sun, May 1, 2022 at 8:44 AM d <[email protected]> wrote:
> -------to reproduce --------------------------------------
> CREATE OR REPLACE FUNCTION record_to_form_data(p_r record)
> RETURNS text
> LANGUAGE plpgsql
> AS $function$
> begin
> return (
> select string_agg(format('%s=%s',key,urlencode(value)),'&')
> from
> (select p_r.*) i,
> hstore(i.*) as h,each(h) );
> end;
> $function$;
>
Not a bug, it is a documented limitation.
It is your use of "(select p_r.*)" that is problematic.
https://www.postgresql.org/docs/current/plpgsql-implementation.html#PLPGSQL-PLAN-CACHING
"""
The mutable nature of record variables presents another problem in this
connection. When fields of a record variable are used in expressions or
statements, the data types of the fields must not change from one call of
the function to the next, since each expression will be analyzed using the
data type that is present when the expression is first reached. EXECUTE can
be used to get around this problem when necessary.
"""
David J.
^ permalink raw reply [nested|flat] 5+ messages in thread
end of thread, other threads:[~2022-05-01 15:58 UTC | newest]
Thread overview: 5+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2017-02-01 07:07 [PATCH] Fix a bug of physical replication slot. Kyotaro Horiguchi <[email protected]>
2017-02-01 07:07 [PATCH] Fix a bug of physical replication slot. Kyotaro Horiguchi <[email protected]>
2020-10-16 14:26 [PATCH v24 09/10] new compression method extension for zlib dilipkumar <[email protected]>
2022-04-30 23:23 ERROR: type of parameter 1 (fruit2) does not match that when preparing the plan (fruit1) d <[email protected]>
2022-05-01 15:58 ` Re: ERROR: type of parameter 1 (fruit2) does not match that when preparing the plan (fruit1) David G. Johnston <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox