agora inbox for pgsql-hackers@postgresql.org
help / color / mirror / Atom feed[PATCH 08/10] Make pg_waldump not use callback but call the function directly
9+ messages / 6 participants
[nested] [flat]
* [PATCH 08/10] Make pg_waldump not use callback but call the function directly
@ 2019-04-18 06:50 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
0 siblings, 0 replies; 9+ messages in thread
From: Kyotaro Horiguchi @ 2019-04-18 06:50 UTC (permalink / raw)
This patch does the similar thing to the change in logical rep. Moves
callback from XLogReaderState from the parameter of
XLogFindNextRecord. Then invalidate the parameters callback and
private for XLogReaderAllocate.
---
src/backend/access/transam/xlogreader.c | 17 +++++++----------
src/bin/pg_waldump/pg_waldump.c | 21 +++++++++------------
src/include/access/xlogreader.h | 14 +++++---------
3 files changed, 21 insertions(+), 31 deletions(-)
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index f953924a72..9b38fc7829 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -996,7 +996,8 @@ XLogReaderValidatePageHeader(XLogReaderState *state, XLogRecPtr recptr,
* debugging purposes.
*/
XLogRecPtr
-XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
+XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr,
+ XLogFindNextRecordCB read_page, void *private)
{
XLogReaderState saved_state = *state;
XLogRecPtr tmpRecPtr;
@@ -1008,6 +1009,8 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
Assert(!XLogRecPtrIsInvalid(RecPtr));
+ XLREAD_RESET(state);
+
/*
* skip over potential continuation data, keeping in mind that it may span
* multiple pages
@@ -1035,9 +1038,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
/* Read the page containing the record */
while(XLogNeedData(state, targetPagePtr, targetRecOff))
- state->read_page(state, state->loadPagePtr, state->loadLen,
- state->currRecPtr, state->readBuf,
- &state->readPageTLI);
+ read_page(state, private);
if (state->readLen < 0)
goto err;
@@ -1048,9 +1049,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
/* make sure we have enough data for the page header */
while (XLogNeedData(state, targetPagePtr, pageHeaderSize))
- state->read_page(state, state->loadPagePtr, state->loadLen,
- state->currRecPtr, state->readBuf,
- &state->readPageTLI);
+ read_page(state, private);
if (state->readLen < 0)
goto err;
@@ -1097,9 +1096,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
{
if (result == XLREAD_NEED_DATA)
{
- state->read_page(state, state->loadPagePtr, state->loadLen,
- state->currRecPtr, state->readBuf,
- &state->readPageTLI);
+ read_page(state, private);
continue;
}
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index acee7ae199..1966da493f 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -422,10 +422,12 @@ XLogDumpXLogRead(const char *directory, TimeLineID timeline_id,
* XLogReader read_page callback
*/
static void
-XLogDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
- XLogRecPtr targetPtr, char *readBuff, TimeLineID *curFileTLI)
+XLogDumpReadPage(XLogReaderState *state, void *priv)
{
- XLogDumpPrivate *private = state->private_data;
+ XLogRecPtr targetPagePtr = state->loadPagePtr;
+ int reqLen = state->loadLen;
+ char *readBuff = state->readBuf;
+ XLogDumpPrivate *private = (XLogDumpPrivate *) priv;
int count = XLOG_BLCKSZ;
if (private->endptr != InvalidXLogRecPtr)
@@ -1095,13 +1097,13 @@ main(int argc, char **argv)
/* done with argument parsing, do the actual work */
/* we have everything we need, start reading */
- xlogreader_state = XLogReaderAllocate(WalSegSz, XLogDumpReadPage,
- &private);
+ xlogreader_state = XLogReaderAllocate(WalSegSz, NULL, NULL);
if (!xlogreader_state)
fatal_error("out of memory");
/* first find a valid recptr to start from */
- first_record = XLogFindNextRecord(xlogreader_state, private.startptr);
+ first_record = XLogFindNextRecord(xlogreader_state, private.startptr,
+ &XLogDumpReadPage, (void*) &private);
if (first_record == InvalidXLogRecPtr)
fatal_error("could not find a valid record after %X/%X",
@@ -1128,12 +1130,7 @@ main(int argc, char **argv)
while (XLogReadRecord(xlogreader_state,
first_record, &record, &errormsg) ==
XLREAD_NEED_DATA)
- xlogreader_state->read_page(xlogreader_state,
- xlogreader_state->loadPagePtr,
- xlogreader_state->loadLen,
- xlogreader_state->currRecPtr,
- xlogreader_state->readBuf,
- &xlogreader_state->readPageTLI);
+ XLogDumpReadPage(xlogreader_state, (void *) &private);
if (!record)
{
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h
index 9bfa9e8d54..b231cb330c 100644
--- a/src/include/access/xlogreader.h
+++ b/src/include/access/xlogreader.h
@@ -29,14 +29,6 @@
typedef struct XLogReaderState XLogReaderState;
-/* Function type definition for the read_page callback */
-typedef void (*XLogPageReadCB) (XLogReaderState *xlogreader,
- XLogRecPtr targetPagePtr,
- int reqLen,
- XLogRecPtr targetRecPtr,
- char *readBuf,
- TimeLineID *pageTLI);
-
typedef struct
{
/* Is this block ref in use? */
@@ -263,7 +255,11 @@ extern bool XLogReaderValidatePageHeader(XLogReaderState *state,
extern void XLogReaderInvalReadState(XLogReaderState *state);
#ifdef FRONTEND
-extern XLogRecPtr XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr);
+/* Function type definition for the read_page callback */
+typedef void (*XLogFindNextRecordCB) (XLogReaderState *xlogreader,
+ void *private);
+extern XLogRecPtr XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr,
+ XLogFindNextRecordCB read_page, void *private);
#endif /* FRONTEND */
/* Functions for decoding an XLogRecord */
--
2.16.3
----Next_Part(Fri_Apr_26_17_40_34_2019_888)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
filename="v2-0009-Make-pg_rewind-not-use-callback-but-call-the-functio.patch"
^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH 08/10] Make pg_waldump not use callback but call the function directly
@ 2019-04-18 06:50 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
0 siblings, 0 replies; 9+ messages in thread
From: Kyotaro Horiguchi @ 2019-04-18 06:50 UTC (permalink / raw)
This patch does the similar thing to the change in logical rep. Moves
callback from XLogReaderState from the parameter of
XLogFindNextRecord. Then invalidate the parameters callback and
private for XLogReaderAllocate.
---
src/backend/access/transam/xlogreader.c | 15 +++++----------
src/bin/pg_waldump/pg_waldump.c | 21 +++++++++------------
src/include/access/xlogreader.h | 14 +++++---------
3 files changed, 19 insertions(+), 31 deletions(-)
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index 4ab0655af5..004eaac021 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -1016,7 +1016,8 @@ XLogReaderValidatePageHeader(XLogReaderState *state, XLogRecPtr recptr,
* debugging purposes.
*/
XLogRecPtr
-XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
+XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr,
+ XLogFindNextRecordCB read_page, void *private)
{
XLogReaderState saved_state = *state;
XLogRecPtr tmpRecPtr;
@@ -1055,9 +1056,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
/* Read the page containing the record */
while(XLogNeedData(state, targetPagePtr, targetRecOff))
- state->read_page(state, state->loadPagePtr, state->loadLen,
- state->currRecPtr, state->readBuf,
- &state->readPageTLI);
+ read_page(state, private);
if (state->readLen < 0)
goto err;
@@ -1068,9 +1067,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
/* make sure we have enough data for the page header */
while (XLogNeedData(state, targetPagePtr, pageHeaderSize))
- state->read_page(state, state->loadPagePtr, state->loadLen,
- state->currRecPtr, state->readBuf,
- &state->readPageTLI);
+ read_page(state, private);
if (state->readLen < 0)
goto err;
@@ -1117,9 +1114,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
{
if (result == XLREAD_NEED_DATA)
{
- state->read_page(state, state->loadPagePtr, state->loadLen,
- state->currRecPtr, state->readBuf,
- &state->readPageTLI);
+ read_page(state, private);
continue;
}
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index e2e93f144a..8fe6823b32 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -422,10 +422,12 @@ XLogDumpXLogRead(const char *directory, TimeLineID timeline_id,
* XLogReader read_page callback
*/
static void
-XLogDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
- XLogRecPtr targetPtr, char *readBuff, TimeLineID *curFileTLI)
+XLogDumpReadPage(XLogReaderState *state, void *priv)
{
- XLogDumpPrivate *private = state->private_data;
+ XLogRecPtr targetPagePtr = state->loadPagePtr;
+ int reqLen = state->loadLen;
+ char *readBuff = state->readBuf;
+ XLogDumpPrivate *private = (XLogDumpPrivate *) priv;
int count = XLOG_BLCKSZ;
if (private->endptr != InvalidXLogRecPtr)
@@ -1102,13 +1104,13 @@ main(int argc, char **argv)
/* done with argument parsing, do the actual work */
/* we have everything we need, start reading */
- xlogreader_state = XLogReaderAllocate(WalSegSz, XLogDumpReadPage,
- &private);
+ xlogreader_state = XLogReaderAllocate(WalSegSz, NULL, NULL);
if (!xlogreader_state)
fatal_error("out of memory");
/* first find a valid recptr to start from */
- first_record = XLogFindNextRecord(xlogreader_state, private.startptr);
+ first_record = XLogFindNextRecord(xlogreader_state, private.startptr,
+ &XLogDumpReadPage, (void*) &private);
if (first_record == InvalidXLogRecPtr)
fatal_error("could not find a valid record after %X/%X",
@@ -1135,12 +1137,7 @@ main(int argc, char **argv)
while (XLogReadRecord(xlogreader_state,
first_record, &record, &errormsg) ==
XLREAD_NEED_DATA)
- xlogreader_state->read_page(xlogreader_state,
- xlogreader_state->loadPagePtr,
- xlogreader_state->loadLen,
- xlogreader_state->currRecPtr,
- xlogreader_state->readBuf,
- &xlogreader_state->readPageTLI);
+ XLogDumpReadPage(xlogreader_state, (void *) &private);
if (!record)
{
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h
index bc0c642906..b4ace71a75 100644
--- a/src/include/access/xlogreader.h
+++ b/src/include/access/xlogreader.h
@@ -29,14 +29,6 @@
typedef struct XLogReaderState XLogReaderState;
-/* Function type definition for the read_page callback */
-typedef void (*XLogPageReadCB) (XLogReaderState *xlogreader,
- XLogRecPtr targetPagePtr,
- int reqLen,
- XLogRecPtr targetRecPtr,
- char *readBuf,
- TimeLineID *pageTLI);
-
typedef struct
{
/* Is this block ref in use? */
@@ -252,7 +244,11 @@ extern bool XLogReaderValidatePageHeader(XLogReaderState *state,
extern void XLogReaderInvalReadState(XLogReaderState *state);
#ifdef FRONTEND
-extern XLogRecPtr XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr);
+/* Function type definition for the read_page callback */
+typedef void (*XLogFindNextRecordCB) (XLogReaderState *xlogreader,
+ void *private);
+extern XLogRecPtr XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr,
+ XLogFindNextRecordCB read_page, void *private);
#endif /* FRONTEND */
/* Functions for decoding an XLogRecord */
--
2.16.3
----Next_Part(Wed_Jul_10_13_18_10_2019_842)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
filename="v4-0009-Make-pg_rewind-not-use-callback-but-call-the-functio.patch"
^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH 08/10] Make pg_waldump not use callback but call the function directly
@ 2019-04-18 06:50 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
0 siblings, 0 replies; 9+ messages in thread
From: Kyotaro Horiguchi @ 2019-04-18 06:50 UTC (permalink / raw)
This patch does the similar thing to the change in logical rep. Moves
callback from XLogReaderState from the parameter of
XLogFindNextRecord. Then invalidate the parameters callback and
private for XLogReaderAllocate.
---
src/backend/access/transam/xlogreader.c | 15 +++++----------
src/bin/pg_waldump/pg_waldump.c | 21 +++++++++------------
src/include/access/xlogreader.h | 14 +++++---------
3 files changed, 19 insertions(+), 31 deletions(-)
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index ce901c358a..0bf8dac408 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -1026,7 +1026,8 @@ XLogReaderValidatePageHeader(XLogReaderState *state, XLogRecPtr recptr,
* debugging purposes.
*/
XLogRecPtr
-XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
+XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr,
+ XLogFindNextRecordCB read_page, void *private)
{
XLogReaderState saved_state = *state;
XLogRecPtr tmpRecPtr;
@@ -1065,9 +1066,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
/* Read the page containing the record */
while(XLogNeedData(state, targetPagePtr, targetRecOff))
- state->read_page(state, state->loadPagePtr, state->loadLen,
- state->currRecPtr, state->readBuf,
- &state->readPageTLI);
+ read_page(state, private);
if (state->readLen < 0)
goto err;
@@ -1078,9 +1077,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
/* make sure we have enough data for the page header */
while (XLogNeedData(state, targetPagePtr, pageHeaderSize))
- state->read_page(state, state->loadPagePtr, state->loadLen,
- state->currRecPtr, state->readBuf,
- &state->readPageTLI);
+ read_page(state, private);
if (state->readLen < 0)
goto err;
@@ -1127,9 +1124,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
{
if (result == XLREAD_NEED_DATA)
{
- state->read_page(state, state->loadPagePtr, state->loadLen,
- state->currRecPtr, state->readBuf,
- &state->readPageTLI);
+ read_page(state, private);
continue;
}
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index 54717c9320..3125633327 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -422,10 +422,12 @@ XLogDumpXLogRead(const char *directory, TimeLineID timeline_id,
* XLogReader read_page callback
*/
static void
-XLogDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
- XLogRecPtr targetPtr, char *readBuff, TimeLineID *curFileTLI)
+XLogDumpReadPage(XLogReaderState *state, void *priv)
{
- XLogDumpPrivate *private = state->private_data;
+ XLogRecPtr targetPagePtr = state->loadPagePtr;
+ int reqLen = state->loadLen;
+ char *readBuff = state->readBuf;
+ XLogDumpPrivate *private = (XLogDumpPrivate *) priv;
int count = XLOG_BLCKSZ;
if (private->endptr != InvalidXLogRecPtr)
@@ -1095,13 +1097,13 @@ main(int argc, char **argv)
/* done with argument parsing, do the actual work */
/* we have everything we need, start reading */
- xlogreader_state = XLogReaderAllocate(WalSegSz, XLogDumpReadPage,
- &private);
+ xlogreader_state = XLogReaderAllocate(WalSegSz, NULL, NULL);
if (!xlogreader_state)
fatal_error("out of memory");
/* first find a valid recptr to start from */
- first_record = XLogFindNextRecord(xlogreader_state, private.startptr);
+ first_record = XLogFindNextRecord(xlogreader_state, private.startptr,
+ &XLogDumpReadPage, (void*) &private);
if (first_record == InvalidXLogRecPtr)
fatal_error("could not find a valid record after %X/%X",
@@ -1128,12 +1130,7 @@ main(int argc, char **argv)
while (XLogReadRecord(xlogreader_state,
first_record, &record, &errormsg) ==
XLREAD_NEED_DATA)
- xlogreader_state->read_page(xlogreader_state,
- xlogreader_state->loadPagePtr,
- xlogreader_state->loadLen,
- xlogreader_state->currRecPtr,
- xlogreader_state->readBuf,
- &xlogreader_state->readPageTLI);
+ XLogDumpReadPage(xlogreader_state, (void *) &private);
if (!record)
{
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h
index bc0c642906..b4ace71a75 100644
--- a/src/include/access/xlogreader.h
+++ b/src/include/access/xlogreader.h
@@ -29,14 +29,6 @@
typedef struct XLogReaderState XLogReaderState;
-/* Function type definition for the read_page callback */
-typedef void (*XLogPageReadCB) (XLogReaderState *xlogreader,
- XLogRecPtr targetPagePtr,
- int reqLen,
- XLogRecPtr targetRecPtr,
- char *readBuf,
- TimeLineID *pageTLI);
-
typedef struct
{
/* Is this block ref in use? */
@@ -252,7 +244,11 @@ extern bool XLogReaderValidatePageHeader(XLogReaderState *state,
extern void XLogReaderInvalReadState(XLogReaderState *state);
#ifdef FRONTEND
-extern XLogRecPtr XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr);
+/* Function type definition for the read_page callback */
+typedef void (*XLogFindNextRecordCB) (XLogReaderState *xlogreader,
+ void *private);
+extern XLogRecPtr XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr,
+ XLogFindNextRecordCB read_page, void *private);
#endif /* FRONTEND */
/* Functions for decoding an XLogRecord */
--
2.16.3
----Next_Part(Fri_May_24_11_56_24_2019_374)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
filename="v3-0009-Make-pg_rewind-not-use-callback-but-call-the-functio.patch"
^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH 08/10] Make pg_waldump not use callback but call the function directly
@ 2019-04-18 06:50 Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
0 siblings, 0 replies; 9+ messages in thread
From: Kyotaro Horiguchi @ 2019-04-18 06:50 UTC (permalink / raw)
This patch does the similar thing to the change in logical rep. Moves
callback from XLogReaderState from the parameter of
XLogFindNextRecord. Then invalidate the parameters callback and
private for XLogReaderAllocate.
---
src/backend/access/transam/xlogreader.c | 17 +++++++----------
src/bin/pg_waldump/pg_waldump.c | 21 +++++++++------------
src/include/access/xlogreader.h | 14 +++++---------
3 files changed, 21 insertions(+), 31 deletions(-)
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index 05a57a1ebd..72b82a17d6 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -1006,7 +1006,8 @@ XLogReaderValidatePageHeader(XLogReaderState *state, XLogRecPtr recptr,
* debugging purposes.
*/
XLogRecPtr
-XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
+XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr,
+ XLogFindNextRecordCB read_page, void *private)
{
XLogReaderState saved_state = *state;
XLogRecPtr tmpRecPtr;
@@ -1018,6 +1019,8 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
Assert(!XLogRecPtrIsInvalid(RecPtr));
+ XLREAD_RESET(state);
+
/*
* skip over potential continuation data, keeping in mind that it may span
* multiple pages
@@ -1045,9 +1048,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
/* Read the page containing the record */
while(XLogNeedData(state, targetPagePtr, targetRecOff))
- state->read_page(state, state->loadPagePtr, state->loadLen,
- state->currRecPtr, state->readBuf,
- &state->readPageTLI);
+ read_page(state, private);
if (state->readLen < 0)
goto err;
@@ -1058,9 +1059,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
/* make sure we have enough data for the page header */
while (XLogNeedData(state, targetPagePtr, pageHeaderSize))
- state->read_page(state, state->loadPagePtr, state->loadLen,
- state->currRecPtr, state->readBuf,
- &state->readPageTLI);
+ read_page(state, private);
if (state->readLen < 0)
goto err;
@@ -1107,9 +1106,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr)
{
if (result == XLREAD_NEED_DATA)
{
- state->read_page(state, state->loadPagePtr, state->loadLen,
- state->currRecPtr, state->readBuf,
- &state->readPageTLI);
+ read_page(state, private);
continue;
}
diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c
index acee7ae199..1966da493f 100644
--- a/src/bin/pg_waldump/pg_waldump.c
+++ b/src/bin/pg_waldump/pg_waldump.c
@@ -422,10 +422,12 @@ XLogDumpXLogRead(const char *directory, TimeLineID timeline_id,
* XLogReader read_page callback
*/
static void
-XLogDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
- XLogRecPtr targetPtr, char *readBuff, TimeLineID *curFileTLI)
+XLogDumpReadPage(XLogReaderState *state, void *priv)
{
- XLogDumpPrivate *private = state->private_data;
+ XLogRecPtr targetPagePtr = state->loadPagePtr;
+ int reqLen = state->loadLen;
+ char *readBuff = state->readBuf;
+ XLogDumpPrivate *private = (XLogDumpPrivate *) priv;
int count = XLOG_BLCKSZ;
if (private->endptr != InvalidXLogRecPtr)
@@ -1095,13 +1097,13 @@ main(int argc, char **argv)
/* done with argument parsing, do the actual work */
/* we have everything we need, start reading */
- xlogreader_state = XLogReaderAllocate(WalSegSz, XLogDumpReadPage,
- &private);
+ xlogreader_state = XLogReaderAllocate(WalSegSz, NULL, NULL);
if (!xlogreader_state)
fatal_error("out of memory");
/* first find a valid recptr to start from */
- first_record = XLogFindNextRecord(xlogreader_state, private.startptr);
+ first_record = XLogFindNextRecord(xlogreader_state, private.startptr,
+ &XLogDumpReadPage, (void*) &private);
if (first_record == InvalidXLogRecPtr)
fatal_error("could not find a valid record after %X/%X",
@@ -1128,12 +1130,7 @@ main(int argc, char **argv)
while (XLogReadRecord(xlogreader_state,
first_record, &record, &errormsg) ==
XLREAD_NEED_DATA)
- xlogreader_state->read_page(xlogreader_state,
- xlogreader_state->loadPagePtr,
- xlogreader_state->loadLen,
- xlogreader_state->currRecPtr,
- xlogreader_state->readBuf,
- &xlogreader_state->readPageTLI);
+ XLogDumpReadPage(xlogreader_state, (void *) &private);
if (!record)
{
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h
index 338dc2c14d..5f85c79424 100644
--- a/src/include/access/xlogreader.h
+++ b/src/include/access/xlogreader.h
@@ -29,14 +29,6 @@
typedef struct XLogReaderState XLogReaderState;
-/* Function type definition for the read_page callback */
-typedef void (*XLogPageReadCB) (XLogReaderState *xlogreader,
- XLogRecPtr targetPagePtr,
- int reqLen,
- XLogRecPtr targetRecPtr,
- char *readBuf,
- TimeLineID *pageTLI);
-
typedef struct
{
/* Is this block ref in use? */
@@ -263,7 +255,11 @@ extern bool XLogReaderValidatePageHeader(XLogReaderState *state,
extern void XLogReaderInvalReadState(XLogReaderState *state);
#ifdef FRONTEND
-extern XLogRecPtr XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr);
+/* Function type definition for the read_page callback */
+typedef void (*XLogFindNextRecordCB) (XLogReaderState *xlogreader,
+ void *private);
+extern XLogRecPtr XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr,
+ XLogFindNextRecordCB read_page, void *private);
#endif /* FRONTEND */
/* Functions for decoding an XLogRecord */
--
2.16.3
----Next_Part(Thu_Apr_18_21_02_57_2019_406)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
filename="0009-Make-pg_rewind-not-use-callback-but-call-the-functio.patch"
^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH v3 1/2] Use "template" initdb in tests
@ 2023-02-03 05:51 Andres Freund <andres@anarazel.de>
0 siblings, 0 replies; 9+ messages in thread
From: Andres Freund @ 2023-02-03 05:51 UTC (permalink / raw)
Discussion: https://postgr.es/m/20220120021859.3zpsfqn4z7ob7afz@alap3.anarazel.de
---
meson.build | 30 ++++++++++
.cirrus.yml | 3 +-
src/test/perl/PostgreSQL/Test/Cluster.pm | 46 ++++++++++++++-
src/test/regress/pg_regress.c | 74 ++++++++++++++++++------
src/Makefile.global.in | 52 +++++++++--------
5 files changed, 161 insertions(+), 44 deletions(-)
diff --git a/meson.build b/meson.build
index 04ea3488522..47429a18c3f 100644
--- a/meson.build
+++ b/meson.build
@@ -3056,8 +3056,10 @@ testport = 40000
test_env = environment()
temp_install_bindir = test_install_location / get_option('bindir')
+test_initdb_template = meson.build_root() / 'tmp_install' / 'initdb-template'
test_env.set('PG_REGRESS', pg_regress.full_path())
test_env.set('REGRESS_SHLIB', regress_module.full_path())
+test_env.set('INITDB_TEMPLATE', test_initdb_template)
# Test suites that are not safe by default but can be run if selected
# by the user via the whitespace-separated list in variable PG_TEST_EXTRA.
@@ -3072,6 +3074,34 @@ if library_path_var != ''
endif
+# Create (and remove old) initdb template directory. Tests use that, where
+# possible, to make it cheaper to run tests.
+#
+# Use python to remove the old cached initdb, as we cannot rely on a working
+# 'rm' binary on windows.
+test('initdb_cache',
+ python,
+ args: [
+ '-c', '''
+import shutil
+import sys
+import subprocess
+
+shutil.rmtree(sys.argv[1], ignore_errors=True)
+sp = subprocess.run(sys.argv[2:] + [sys.argv[1]])
+sys.exit(sp.returncode)
+''',
+ test_initdb_template,
+ temp_install_bindir / 'initdb',
+ '-A', 'trust', '-N', '--no-instructions'
+ ],
+ priority: setup_tests_priority - 1,
+ timeout: 300,
+ is_parallel: false,
+ env: test_env,
+ suite: ['setup'])
+
+
###############################################################
# Test Generation
diff --git a/.cirrus.yml b/.cirrus.yml
index d260f15c4e2..8fce17dff08 100644
--- a/.cirrus.yml
+++ b/.cirrus.yml
@@ -115,8 +115,9 @@ task:
test_minimal_script: |
su postgres <<-EOF
ulimit -c unlimited
+ meson test $MTEST_ARGS --suite setup
meson test $MTEST_ARGS --num-processes ${TEST_JOBS} \
- tmp_install cube/regress pg_ctl/001_start_stop
+ cube/regress pg_ctl/001_start_stop
EOF
on_failure:
diff --git a/src/test/perl/PostgreSQL/Test/Cluster.pm b/src/test/perl/PostgreSQL/Test/Cluster.pm
index 5e161dbee60..4d449c35de9 100644
--- a/src/test/perl/PostgreSQL/Test/Cluster.pm
+++ b/src/test/perl/PostgreSQL/Test/Cluster.pm
@@ -522,8 +522,50 @@ sub init
mkdir $self->backup_dir;
mkdir $self->archive_dir;
- PostgreSQL::Test::Utils::system_or_bail('initdb', '-D', $pgdata, '-A',
- 'trust', '-N', @{ $params{extra} });
+ # If available and if there aren't any parameters, use a previously
+ # initdb'd cluster as a template by copying it. For a lot of tests, that's
+ # substantially cheaper. Do so only if there aren't parameters, it doesn't
+ # seem worth figuring out whether they affect compatibility.
+ #
+ # There's very similar code in pg_regress.c, but we can't easily
+ # deduplicate it until we require perl at build time.
+ if (defined $params{extra} or !defined $ENV{INITDB_TEMPLATE})
+ {
+ note("initializing database system by running initdb");
+ PostgreSQL::Test::Utils::system_or_bail('initdb', '-D', $pgdata, '-A',
+ 'trust', '-N', @{ $params{extra} });
+ }
+ else
+ {
+ my @copycmd;
+ my $expected_exitcode;
+
+ note("initializing database system by copying initdb template");
+
+ if ($PostgreSQL::Test::Utils::windows_os)
+ {
+ @copycmd = qw(robocopy /E /NJS /NJH /NFL /NDL /NP);
+ $expected_exitcode = 1; # 1 denotes files were copied
+ }
+ else
+ {
+ @copycmd = qw(cp -a);
+ $expected_exitcode = 0;
+ }
+
+ @copycmd = (@copycmd, $ENV{INITDB_TEMPLATE}, $pgdata);
+
+ my $ret = PostgreSQL::Test::Utils::system_log(@copycmd);
+
+ # See http://perldoc.perl.org/perlvar.html#%24CHILD_ERROR
+ if ($ret & 127 or $ret >> 8 != $expected_exitcode)
+ {
+ BAIL_OUT(
+ sprintf("failed to execute command \"%s\": $ret",
+ join(" ", @copycmd)));
+ }
+ }
+
PostgreSQL::Test::Utils::system_or_bail($ENV{PG_REGRESS},
'--config-auth', $pgdata, @{ $params{auth_extra} });
diff --git a/src/test/regress/pg_regress.c b/src/test/regress/pg_regress.c
index b68632320a7..407e3915cec 100644
--- a/src/test/regress/pg_regress.c
+++ b/src/test/regress/pg_regress.c
@@ -2295,6 +2295,7 @@ regression_main(int argc, char *argv[],
FILE *pg_conf;
const char *env_wait;
int wait_seconds;
+ const char *initdb_template_dir;
/*
* Prepare the temp instance
@@ -2316,25 +2317,64 @@ regression_main(int argc, char *argv[],
if (!directory_exists(buf))
make_directory(buf);
- /* initdb */
initStringInfo(&cmd);
- appendStringInfo(&cmd,
- "\"%s%sinitdb\" -D \"%s/data\" --no-clean --no-sync",
- bindir ? bindir : "",
- bindir ? "/" : "",
- temp_instance);
- if (debug)
- appendStringInfo(&cmd, " --debug");
- if (nolocale)
- appendStringInfo(&cmd, " --no-locale");
- appendStringInfo(&cmd, " > \"%s/log/initdb.log\" 2>&1", outputdir);
- fflush(NULL);
- if (system(cmd.data))
+
+ /*
+ * Create data directory.
+ *
+ * If available, use a previously initdb'd cluster as a template by
+ * copying it. For a lot of tests, that's substantially cheaper.
+ *
+ * There's very similar code in Cluster.pm, but we can't easily de
+ * duplicate it until we require perl at build time.
+ */
+ initdb_template_dir = getenv("INITDB_TEMPLATE");
+ if (initdb_template_dir == NULL || nolocale || debug)
{
- bail("initdb failed\n"
- "# Examine \"%s/log/initdb.log\" for the reason.\n"
- "# Command was: %s",
- outputdir, cmd.data);
+ note("initializing database system by running initdb");
+
+ appendStringInfo(&cmd,
+ "\"%s%sinitdb\" -D \"%s/data\" --no-clean --no-sync",
+ bindir ? bindir : "",
+ bindir ? "/" : "",
+ temp_instance);
+ if (debug)
+ appendStringInfo(&cmd, " --debug");
+ if (nolocale)
+ appendStringInfo(&cmd, " --no-locale");
+ appendStringInfo(&cmd, " > \"%s/log/initdb.log\" 2>&1", outputdir);
+ fflush(NULL);
+ if (system(cmd.data))
+ {
+ bail("initdb failed\n"
+ "# Examine \"%s/log/initdb.log\" for the reason.\n"
+ "# Command was: %s",
+ outputdir, cmd.data);
+ }
+ }
+ else
+ {
+#ifndef WIN32
+ const char *copycmd = "cp -a \"%s\" \"%s/data\"";
+ int expected_exitcode = 0;
+#else
+ const char *copycmd = "robocopy /E /NJS /NJH /NFL /NDL /NP \"%s\" \"%s/data\"";
+ int expected_exitcode = 1; /* 1 denotes files were copied */
+#endif
+
+ note("initializing database system by copying initdb template");
+
+ appendStringInfo(&cmd,
+ copycmd,
+ initdb_template_dir,
+ temp_instance);
+ if (system(cmd.data) != expected_exitcode)
+ {
+ bail("copying of initdb template failed\n"
+ "# Examine \"%s/log/initdb.log\" for the reason.\n"
+ "# Command was: %s",
+ outputdir, cmd.data);
+ }
}
pfree(cmd.data);
diff --git a/src/Makefile.global.in b/src/Makefile.global.in
index df9f721a41a..0b4ca0eb6ae 100644
--- a/src/Makefile.global.in
+++ b/src/Makefile.global.in
@@ -397,30 +397,6 @@ check: temp-install
.PHONY: temp-install
-temp-install: | submake-generated-headers
-ifndef NO_TEMP_INSTALL
-ifneq ($(abs_top_builddir),)
-ifeq ($(MAKELEVEL),0)
- rm -rf '$(abs_top_builddir)'/tmp_install
- $(MKDIR_P) '$(abs_top_builddir)'/tmp_install/log
- $(MAKE) -C '$(top_builddir)' DESTDIR='$(abs_top_builddir)'/tmp_install install >'$(abs_top_builddir)'/tmp_install/log/install.log 2>&1
- $(MAKE) -j1 $(if $(CHECKPREP_TOP),-C $(CHECKPREP_TOP),) checkprep >>'$(abs_top_builddir)'/tmp_install/log/install.log 2>&1
-endif
-endif
-endif
-
-# Tasks to run serially at the end of temp-install. Some EXTRA_INSTALL
-# entries appear more than once in the tree, and parallel installs of the same
-# file can fail with EEXIST.
-checkprep:
- $(if $(EXTRA_INSTALL),for extra in $(EXTRA_INSTALL); do $(MAKE) -C '$(top_builddir)'/$$extra DESTDIR='$(abs_top_builddir)'/tmp_install install || exit; done)
-
-PROVE = @PROVE@
-# There are common routines in src/test/perl, and some test suites have
-# extra perl modules in their own directory.
-PG_PROVE_FLAGS = -I $(top_srcdir)/src/test/perl/ -I $(srcdir)
-# User-supplied prove flags such as --verbose can be provided in PROVE_FLAGS.
-PROVE_FLAGS =
# prepend to path if already set, else just set it
define add_to_path
@@ -437,8 +413,36 @@ ld_library_path_var = LD_LIBRARY_PATH
with_temp_install = \
PATH="$(abs_top_builddir)/tmp_install$(bindir):$(CURDIR):$$PATH" \
$(call add_to_path,$(strip $(ld_library_path_var)),$(abs_top_builddir)/tmp_install$(libdir)) \
+ INITDB_TEMPLATE='$(abs_top_builddir)'/tmp_install/initdb-template \
$(with_temp_install_extra)
+temp-install: | submake-generated-headers
+ifndef NO_TEMP_INSTALL
+ifneq ($(abs_top_builddir),)
+ifeq ($(MAKELEVEL),0)
+ rm -rf '$(abs_top_builddir)'/tmp_install
+ $(MKDIR_P) '$(abs_top_builddir)'/tmp_install/log
+ $(MAKE) -C '$(top_builddir)' DESTDIR='$(abs_top_builddir)'/tmp_install install >'$(abs_top_builddir)'/tmp_install/log/install.log 2>&1
+ $(MAKE) -j1 $(if $(CHECKPREP_TOP),-C $(CHECKPREP_TOP),) checkprep >>'$(abs_top_builddir)'/tmp_install/log/install.log 2>&1
+
+ $(with_temp_install) initdb -A trust -N --no-instructions '$(abs_top_builddir)'/tmp_install/initdb-template >>'$(abs_top_builddir)'/tmp_install/log/initdb-template.log 2>&1
+endif
+endif
+endif
+
+# Tasks to run serially at the end of temp-install. Some EXTRA_INSTALL
+# entries appear more than once in the tree, and parallel installs of the same
+# file can fail with EEXIST.
+checkprep:
+ $(if $(EXTRA_INSTALL),for extra in $(EXTRA_INSTALL); do $(MAKE) -C '$(top_builddir)'/$$extra DESTDIR='$(abs_top_builddir)'/tmp_install install || exit; done)
+
+PROVE = @PROVE@
+# There are common routines in src/test/perl, and some test suites have
+# extra perl modules in their own directory.
+PG_PROVE_FLAGS = -I $(top_srcdir)/src/test/perl/ -I $(srcdir)
+# User-supplied prove flags such as --verbose can be provided in PROVE_FLAGS.
+PROVE_FLAGS =
+
ifeq ($(enable_tap_tests),yes)
ifndef PGXS
--
2.38.0
--m6x7pdqtqaxzqass--
^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH 6/6] Use multiple snapshots to copy the data.
@ 2026-01-08 16:47 Antonin Houska <ah@cybertec.at>
0 siblings, 0 replies; 9+ messages in thread
From: Antonin Houska @ 2026-01-08 16:47 UTC (permalink / raw)
REPACK (CONCURRENTLY) does not prevent applications from using the table that
is being processed, however it can prevent the xmin horizon from advancing and
thus restrict VACUUM for the whole database. This patch adds the ability to
use particular snapshot only for certain range of pages. Each time that number
of pages is processed, a new snapshot is built, which supposedly has its xmin
higher than the previous snapshot.
The data copying works as follows:
1. Have the logical decoding system build a snapshot S0 for range R0 at
LSN0. This snapshot sees all the data changes whose commit records have
LSN < LSN0.
2. Copy the pages in that range to the new relation. The changes not visible
to the snapshot (because their transactions are still running) will
appear in the output of the logical decoding system as soon as their
commit records appear in WAL.
3. Perform logical decoding of all changes we find in WAL for the table
we're repacking, put them aside and remember that out of these we can
only apply those that affect the range R0 in the old
relation. (Naturally, we cannot apply ones that belong to other pages
because it's impossible to UPDATE / DELETE a row in the new relation if
it hasn't been copied yet.) Once the decoding is done, consider LSN1 to
be the position of the end of the last WAL record decoded.
4. Build a new snapshot S1 at position LSN1, i.e. one that sees all the data
whose commit records are at WAL positions < LSN1. Use this snapshot to
copy the range of pages R1.
5. Perform logical decoding like in step 3, but remember, that out of this
next set, only changes belonging to ranges R0 *and* R1 in the old table
can be applied.
6. etc
Note that the changes decoded above should not be applied to the new relation
until the whole relation has been copied. The point is that we need "identity
index" to apply UPDATE and DELETE statements, and bulk creation of indexes on
the already copied heap is probably better than retail insertions during the
copying.
Special attention needs to be paid to UPDATES that span page ranges. For
example, if the old tuple is in range R0, but the new tuple is in R1, and R1
hasn't been copied yet, we only DELETE the old version from the new
relation. The new version will be handled during processing of range R1. The
snapshot S1 will be based on WAL position following that UPDATE, so it'll see
the new tuple if its transaction's commit record is at WAL position lower than
the position where we built the snapshot. On the other hand, if the commit
record appears at higher position than the that of the snapshot, the
corresponding INSERT will be decoded and replayed sometime later: once the
scan of R1 started, changes of tuples belonging to it are no longer filtered
out.
Likewise, if the old tuple is in range R1 (not yet copied) but the new tuple
is in R0, we only perform INSERT on the new relation. The deletion of the old
version will either be visible to the snapshot S1 (i.e. the snapshot won't see
the old version), or replayed later.
This approach introduces one limitation though: if the USING INDEX clause is
specified, an explicit sort is always used. Index scan wouldn't work because
it does not return the tuples sorted by CTID. That way we wouldn't be able to
split the copying into ranges of pages. I'm not sure it's serious. If REPACK
runs concurrently and does not restrict VACUUM, the execution time should not
be critical.
A new GUC repack_snapshot_after can be used to set the number of pages per
snapshot. It's currently classified as DEVELOPER_OPTIONS and may be replaced
by a constant after enough evaluation is done.
---
src/backend/access/heap/heapam_handler.c | 144 ++++-
src/backend/commands/cluster.c | 589 +++++++++++++-----
src/backend/replication/logical/decode.c | 47 +-
src/backend/replication/logical/logical.c | 30 +-
.../replication/logical/reorderbuffer.c | 50 ++
src/backend/replication/logical/snapbuild.c | 27 +-
.../pgoutput_repack/pgoutput_repack.c | 2 +
src/backend/utils/misc/guc_parameters.dat | 10 +
src/backend/utils/misc/guc_tables.c | 1 +
src/include/access/tableam.h | 14 +-
src/include/commands/cluster.h | 72 +++
src/include/replication/logical.h | 2 +-
src/include/replication/reorderbuffer.h | 1 +
src/include/replication/snapbuild.h | 2 +-
src/tools/pgindent/typedefs.list | 3 +-
15 files changed, 803 insertions(+), 191 deletions(-)
diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c
index 475c536ce43..9c02d91d327 100644
--- a/src/backend/access/heap/heapam_handler.c
+++ b/src/backend/access/heap/heapam_handler.c
@@ -33,6 +33,7 @@
#include "catalog/index.h"
#include "catalog/storage.h"
#include "catalog/storage_xlog.h"
+#include "commands/cluster.h"
#include "commands/progress.h"
#include "executor/executor.h"
#include "miscadmin.h"
@@ -686,12 +687,12 @@ static void
heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
Relation OldIndex, bool use_sort,
TransactionId OldestXmin,
- Snapshot snapshot,
TransactionId *xid_cutoff,
MultiXactId *multi_cutoff,
double *num_tuples,
double *tups_vacuumed,
- double *tups_recently_dead)
+ double *tups_recently_dead,
+ void *tableam_data)
{
RewriteState rwstate = NULL;
IndexScanDesc indexScan;
@@ -707,7 +708,10 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
bool *isnull;
BufferHeapTupleTableSlot *hslot;
BlockNumber prev_cblock = InvalidBlockNumber;
- bool concurrent = snapshot != NULL;
+ ConcurrentChangeContext *ctx = (ConcurrentChangeContext *) tableam_data;
+ bool concurrent = ctx != NULL;
+ Snapshot snapshot = NULL;
+ BlockNumber range_end = InvalidBlockNumber;
/* Remember if it's a system catalog */
is_system_catalog = IsSystemRelation(OldHeap);
@@ -744,8 +748,9 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
* that still need to be copied, we scan with SnapshotAny and use
* HeapTupleSatisfiesVacuum for the visibility test.
*
- * In the CONCURRENTLY case, we do regular MVCC visibility tests, using
- * the snapshot passed by the caller.
+ * In the CONCURRENTLY case, we do regular MVCC visibility tests. The
+ * snapshot changes several times during the scan so that we do not block
+ * the progress of the xmin horizon for VACUUM too much.
*/
if (OldIndex != NULL && !use_sort)
{
@@ -773,10 +778,15 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
pgstat_progress_update_param(PROGRESS_REPACK_PHASE,
PROGRESS_REPACK_PHASE_SEQ_SCAN_HEAP);
- tableScan = table_beginscan(OldHeap,
- snapshot ? snapshot : SnapshotAny,
- 0, (ScanKey) NULL);
+ tableScan = table_beginscan(OldHeap, SnapshotAny, 0, (ScanKey) NULL);
heapScan = (HeapScanDesc) tableScan;
+
+ /*
+ * In CONCURRENTLY mode we scan the table by ranges of blocks and the
+ * algorithm below expects forward direction. (No other direction
+ * should be set here regardless concurrently anyway.)
+ */
+ Assert(heapScan->rs_dir == ForwardScanDirection || !concurrent);
indexScan = NULL;
/* Set total heap blocks */
@@ -787,6 +797,24 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
slot = table_slot_create(OldHeap, NULL);
hslot = (BufferHeapTupleTableSlot *) slot;
+ if (concurrent)
+ {
+ /*
+ * Do not block the progress of xmin horizons.
+ *
+ * TODO Analyze thoroughly if this might have bad consequences.
+ */
+ PopActiveSnapshot();
+ InvalidateCatalogSnapshot();
+
+ /*
+ * Wait until the worker has the initial snapshot and retrieve it.
+ */
+ snapshot = repack_get_snapshot(ctx);
+
+ PushActiveSnapshot(snapshot);
+ }
+
/*
* Scan through the OldHeap, either in OldIndex order or sequentially;
* copy each tuple into the NewHeap, or transiently to the tuplesort
@@ -803,6 +831,13 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
if (indexScan != NULL)
{
+ /*
+ * Index scan should not be used in the CONCURRENTLY case because
+ * it returns tuples in random order, so we could not split the
+ * scan into a series of page ranges.
+ */
+ Assert(!concurrent);
+
if (!index_getnext_slot(indexScan, ForwardScanDirection, slot))
break;
@@ -824,6 +859,18 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
*/
pgstat_progress_update_param(PROGRESS_REPACK_HEAP_BLKS_SCANNED,
heapScan->rs_nblocks);
+
+ if (concurrent)
+ {
+ PopActiveSnapshot();
+
+ /*
+ * For the last range, there are no restriction on block
+ * numbers, so the concurrent data changes pertaining to
+ * this range can decoded (and applied) anytime after this
+ * loop.
+ */
+ }
break;
}
@@ -922,6 +969,75 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
continue;
}
}
+ else
+ {
+ BlockNumber blkno;
+ bool visible;
+
+ /*
+ * With CONCURRENTLY, we use each snapshot only for certain range
+ * of pages, so that VACUUM does not get block for too long. So
+ * first check if the tuple falls into the current range.
+ */
+ blkno = BufferGetBlockNumber(buf);
+
+ /* The first block of the scan? */
+ if (!BlockNumberIsValid(ctx->first_block))
+ {
+ Assert(!BlockNumberIsValid(range_end));
+
+ ctx->first_block = blkno;
+ range_end = repack_blocks_per_snapshot;
+ }
+ else
+ {
+ Assert(BlockNumberIsValid(range_end));
+
+ /* End of the current range? */
+ if (blkno >= range_end)
+ {
+ XLogRecPtr end_of_wal;
+
+ PopActiveSnapshot();
+
+ /*
+ * XXX It might be worth Assert(CatalogSnapshot == NULL)
+ * here, however that symbol is not external.
+ */
+
+ /*
+ * Decode all the concurrent data changes committed so far
+ * - these will be applicable to the current range.
+ */
+ end_of_wal = GetFlushRecPtr(NULL);
+ repack_get_concurrent_changes(ctx, end_of_wal, range_end,
+ true, false);
+
+ /*
+ * Define the next range.
+ */
+ range_end = blkno + repack_blocks_per_snapshot;
+
+ /*
+ * Get the snapshot for the next range - it should have
+ * been built at the position right after the last change
+ * decoded. Data present in the next range of blocks will
+ * either be visible to the snapshot or appear in the next
+ * batch of decoded changes.
+ */
+ snapshot = repack_get_snapshot(ctx);
+ PushActiveSnapshot(snapshot);
+ }
+ }
+
+ /* Finally check the tuple visibility. */
+ LockBuffer(buf, BUFFER_LOCK_SHARE);
+ visible = HeapTupleSatisfiesVisibility(tuple, snapshot, buf);
+ LockBuffer(buf, BUFFER_LOCK_UNLOCK);
+
+ if (!visible)
+ continue;
+ }
*num_tuples += 1;
if (tuplesort != NULL)
@@ -956,6 +1072,18 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
}
}
+ if (concurrent)
+ {
+ XLogRecPtr end_of_wal;
+
+ /* Decode the changes belonging to the last range. */
+ end_of_wal = GetFlushRecPtr(NULL);
+ repack_get_concurrent_changes(ctx, end_of_wal, InvalidBlockNumber,
+ false, false);
+
+ PushActiveSnapshot(GetTransactionSnapshot());
+ }
+
if (indexScan != NULL)
index_endscan(indexScan);
if (tableScan != NULL)
diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c
index 5232fbfb57d..8affa859abc 100644
--- a/src/backend/commands/cluster.c
+++ b/src/backend/commands/cluster.c
@@ -111,52 +111,27 @@ typedef struct
static RelFileLocator repacked_rel_locator = {.relNumber = InvalidOid};
static RelFileLocator repacked_rel_toast_locator = {.relNumber = InvalidOid};
-/*
- * Everything we need to call ExecInsertIndexTuples().
- */
-typedef struct IndexInsertState
-{
- ResultRelInfo *rri;
- EState *estate;
-} IndexInsertState;
-
/* The WAL segment being decoded. */
static XLogSegNo repack_current_segment = 0;
/*
- * The first file exported by the decoding worker must contain a snapshot, the
- * following ones contain the data changes.
+ * When REPACK (CONCURRENTLY) copies data to the new heap, a new snapshot is
+ * built after processing this many pages.
*/
-#define WORKER_FILE_SNAPSHOT 0
+int repack_blocks_per_snapshot = 1024;
/*
- * Information needed to apply concurrent data changes.
+ * Remember here to which pages should applied to changes recorded in given
+ * file.
*/
-typedef struct ChangeDest
+typedef struct RepackApplyRange
{
- /* The relation the changes are applied to. */
- Relation rel;
+ /* The first block of the next range. */
+ BlockNumber end;
- /*
- * The following is needed to find the existing tuple if the change is
- * UPDATE or DELETE. 'ident_key' should have all the fields except for
- * 'sk_argument' initialized.
- */
- Relation ident_index;
- ScanKey ident_key;
- int ident_key_nentries;
-
- /* Needed to update indexes of rel_dst. */
- IndexInsertState *iistate;
-
- /*
- * Sequential number of the file containing the changes.
- *
- * TODO This field makes the structure name less descriptive. Should we
- * rename it, e.g. to ChangeApplyInfo?
- */
- int file_seq;
-} ChangeDest;
+ /* File containing the changes to be applied to blocks in this range. */
+ char *fname;
+} RepackApplyRange;
/*
* Layout of shared memory used for communication between backend and the
@@ -167,6 +142,9 @@ typedef struct DecodingWorkerShared
/* Is the decoding initialized? */
bool initialized;
+ /* Set to request a snapshot. */
+ bool snapshot_requested;
+
/*
* Once the worker has reached this LSN, it should close the current
* output file and either create a new one or exit, according to the field
@@ -174,6 +152,8 @@ typedef struct DecodingWorkerShared
* the WAL available and keep checking this field. It is ok if the worker
* had already decoded records whose LSN is >= lsn_upto before this field
* has been set.
+ *
+ * Set a valid LSN to request data changes.
*/
XLogRecPtr lsn_upto;
@@ -184,7 +164,8 @@ typedef struct DecodingWorkerShared
SharedFileSet sfs;
/* Number of the last file exported by the worker. */
- int last_exported;
+ int last_exported_snapshot;
+ int last_exported_changes;
/* Synchronize access to the fields above. */
slock_t mutex;
@@ -226,26 +207,14 @@ typedef struct DecodingWorkerShared
* the fileset name.)
*/
static inline void
-DecodingWorkerFileName(char *fname, Oid relid, uint32 seq)
+DecodingWorkerFileName(char *fname, Oid relid, uint32 seq, bool snapshot)
{
- snprintf(fname, MAXPGPATH, "%u-%u", relid, seq);
+ if (!snapshot)
+ snprintf(fname, MAXPGPATH, "%u-%u", relid, seq);
+ else
+ snprintf(fname, MAXPGPATH, "%u-%u-snapshot", relid, seq);
}
-/*
- * Backend-local information to control the decoding worker.
- */
-typedef struct DecodingWorker
-{
- /* The worker. */
- BackgroundWorkerHandle *handle;
-
- /* DecodingWorkerShared is in this segment. */
- dsm_segment *seg;
-
- /* Handle of the error queue. */
- shm_mq_handle *error_mqh;
-} DecodingWorker;
-
/* Pointer to currently running decoding worker. */
static DecodingWorker *decoding_worker = NULL;
@@ -262,11 +231,11 @@ static void check_repack_concurrently_requirements(Relation rel);
static void rebuild_relation(Relation OldHeap, Relation index, bool verbose,
bool concurrent);
static void copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex,
- Snapshot snapshot,
bool verbose,
bool *pSwapToastByContent,
TransactionId *pFreezeXid,
- MultiXactId *pCutoffMulti);
+ MultiXactId *pCutoffMulti,
+ ConcurrentChangeContext *ctx);
static List *get_tables_to_repack(RepackCommand cmd, bool usingindex,
MemoryContext permcxt);
static List *get_tables_to_repack_partitioned(RepackCommand cmd,
@@ -276,9 +245,12 @@ static bool cluster_is_permitted_for_relation(RepackCommand cmd,
Oid relid, Oid userid);
static LogicalDecodingContext *setup_logical_decoding(Oid relid);
-static bool decode_concurrent_changes(LogicalDecodingContext *ctx,
+static bool decode_concurrent_changes(LogicalDecodingContext *decoding_ctx,
DecodingWorkerShared *shared);
-static void apply_concurrent_changes(BufFile *file, ChangeDest *dest);
+static void apply_concurrent_changes(ConcurrentChangeContext *ctx);
+static void apply_concurrent_changes_file(ConcurrentChangeContext *ctx,
+ BufFile *file,
+ BlockNumber range_end);
static void apply_concurrent_insert(Relation rel, HeapTuple tup,
IndexInsertState *iistate,
TupleTableSlot *index_slot);
@@ -287,12 +259,14 @@ static void apply_concurrent_update(Relation rel, HeapTuple tup,
IndexInsertState *iistate,
TupleTableSlot *index_slot);
static void apply_concurrent_delete(Relation rel, HeapTuple tup_target);
-static HeapTuple find_target_tuple(Relation rel, ChangeDest *dest,
+static bool is_tuple_in_block_range(HeapTuple tup, BlockNumber start,
+ BlockNumber end);
+static HeapTuple find_target_tuple(Relation rel,
+ ConcurrentChangeContext *ctx,
HeapTuple tup_key,
TupleTableSlot *ident_slot);
-static void process_concurrent_changes(XLogRecPtr end_of_wal,
- ChangeDest *dest,
- bool done);
+static void repack_add_block_range(ConcurrentChangeContext *ctx,
+ BlockNumber end, char *fname);
static IndexInsertState *get_index_insert_state(Relation relation,
Oid ident_index_id,
Relation *ident_index_p);
@@ -303,7 +277,8 @@ static void cleanup_logical_decoding(LogicalDecodingContext *ctx);
static void rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap,
Relation cl_index,
TransactionId frozenXid,
- MultiXactId cutoffMulti);
+ MultiXactId cutoffMulti,
+ ConcurrentChangeContext *ctx);
static List *build_new_indexes(Relation NewHeap, Relation OldHeap, List *OldIndexes);
static Relation process_single_relation(RepackStmt *stmt,
LOCKMODE lockmode,
@@ -314,9 +289,8 @@ static Oid determine_clustered_index(Relation rel, bool usingindex,
static void start_decoding_worker(Oid relid);
static void stop_decoding_worker(void);
static void repack_worker_internal(dsm_segment *seg);
-static void export_initial_snapshot(Snapshot snapshot,
- DecodingWorkerShared *shared);
-static Snapshot get_initial_snapshot(DecodingWorker *worker);
+static void export_snapshot(Snapshot snapshot,
+ DecodingWorkerShared *shared);
static void ProcessRepackMessage(StringInfo msg);
static const char *RepackCommandAsString(RepackCommand cmd);
@@ -402,7 +376,15 @@ ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel)
{
rel = process_single_relation(stmt, lockmode, isTopLevel, ¶ms);
if (rel == NULL)
+ {
+ /*
+ * The original transaction was committed, so the current
+ * portal will not pop the active snapshot.
+ */
+ PopActiveSnapshot();
+
return; /* all done */
+ }
}
/*
@@ -1020,6 +1002,15 @@ check_repack_concurrently_requirements(Relation rel)
RelationGetRelationName(rel)),
(errhint("Relation \"%s\" has no identity index.",
RelationGetRelationName(rel)))));
+
+ /*
+ * In the CONCURRENTLY mode we don't want to use the same snapshot
+ * throughout the whole processing, as it could block the progress of xmin
+ * horizon.
+ */
+ if (IsolationUsesXactSnapshot())
+ ereport(ERROR,
+ (errmsg("REPACK (CONCURRENTLY) does not support transaction isolation higher than READ COMMITTED")));
}
@@ -1050,7 +1041,7 @@ rebuild_relation(Relation OldHeap, Relation index, bool verbose, bool concurrent
bool swap_toast_by_content;
TransactionId frozenXid;
MultiXactId cutoffMulti;
- Snapshot snapshot = NULL;
+ ConcurrentChangeContext *ctx = NULL;
#if USE_ASSERT_CHECKING
LOCKMODE lmode;
@@ -1062,6 +1053,13 @@ rebuild_relation(Relation OldHeap, Relation index, bool verbose, bool concurrent
if (concurrent)
{
+ /*
+ * This is only needed here to gather the data changes and range
+ * information during the copying. The fields needed to apply the
+ * changes be filled later.
+ */
+ ctx = palloc0_object(ConcurrentChangeContext);
+
/*
* The worker needs to be member of the locking group we're the leader
* of. We ought to become the leader before the worker starts. The
@@ -1087,13 +1085,7 @@ rebuild_relation(Relation OldHeap, Relation index, bool verbose, bool concurrent
* REPACK CONCURRENTLY.
*/
start_decoding_worker(tableOid);
-
- /*
- * Wait until the worker has the initial snapshot and retrieve it.
- */
- snapshot = get_initial_snapshot(decoding_worker);
-
- PushActiveSnapshot(snapshot);
+ ctx->worker = decoding_worker;
}
/* for CLUSTER or REPACK USING INDEX, mark the index as the one to use */
@@ -1117,21 +1109,25 @@ rebuild_relation(Relation OldHeap, Relation index, bool verbose, bool concurrent
NewHeap = table_open(OIDNewHeap, NoLock);
/* Copy the heap data into the new table in the desired order */
- copy_table_data(NewHeap, OldHeap, index, snapshot, verbose,
- &swap_toast_by_content, &frozenXid, &cutoffMulti);
-
- /* The historic snapshot won't be needed anymore. */
- if (snapshot)
+ if (concurrent)
{
- PopActiveSnapshot();
- UpdateActiveSnapshotCommandId();
+ ctx->first_block = InvalidBlockNumber;
+ ctx->block_ranges = NIL;
}
+ copy_table_data(NewHeap, OldHeap, index, verbose, &swap_toast_by_content,
+ &frozenXid, &cutoffMulti, ctx);
if (concurrent)
{
+ /*
+ * Make sure the active snapshot can see the data copied, so the rows
+ * can be updated / deleted.
+ */
+ UpdateActiveSnapshotCommandId();
+
Assert(!swap_toast_by_content);
rebuild_relation_finish_concurrent(NewHeap, OldHeap, index,
- frozenXid, cutoffMulti);
+ frozenXid, cutoffMulti, ctx);
pgstat_progress_update_param(PROGRESS_REPACK_PHASE,
PROGRESS_REPACK_PHASE_FINAL_CLEANUP);
@@ -1295,9 +1291,6 @@ make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod,
/*
* Do the physical copying of table data.
*
- * 'snapshot' and 'decoding_ctx': see table_relation_copy_for_cluster(). Pass
- * iff concurrent processing is required.
- *
* There are three output parameters:
* *pSwapToastByContent is set true if toast tables must be swapped by content.
* *pFreezeXid receives the TransactionId used as freeze cutoff point.
@@ -1305,8 +1298,9 @@ make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod,
*/
static void
copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex,
- Snapshot snapshot, bool verbose, bool *pSwapToastByContent,
- TransactionId *pFreezeXid, MultiXactId *pCutoffMulti)
+ bool verbose, bool *pSwapToastByContent,
+ TransactionId *pFreezeXid, MultiXactId *pCutoffMulti,
+ ConcurrentChangeContext *ctx)
{
Relation relRelation;
HeapTuple reltup;
@@ -1323,7 +1317,7 @@ copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex,
int elevel = verbose ? INFO : DEBUG2;
PGRUsage ru0;
char *nspname;
- bool concurrent = snapshot != NULL;
+ bool concurrent = ctx != NULL;
LOCKMODE lmode;
lmode = concurrent ? ShareUpdateExclusiveLock : AccessExclusiveLock;
@@ -1435,8 +1429,18 @@ copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex,
* provided, else plain seqscan.
*/
if (OldIndex != NULL && OldIndex->rd_rel->relam == BTREE_AM_OID)
- use_sort = plan_cluster_use_sort(RelationGetRelid(OldHeap),
- RelationGetRelid(OldIndex));
+ {
+ if (!concurrent)
+ use_sort = plan_cluster_use_sort(RelationGetRelid(OldHeap),
+ RelationGetRelid(OldIndex));
+ else
+
+ /*
+ * To use multiple snapshots, we need to process the table
+ * sequentially.
+ */
+ use_sort = true;
+ }
else
use_sort = false;
@@ -1465,11 +1469,11 @@ copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex,
* values (e.g. because the AM doesn't use freezing).
*/
table_relation_copy_for_cluster(OldHeap, NewHeap, OldIndex, use_sort,
- cutoffs.OldestXmin, snapshot,
+ cutoffs.OldestXmin,
&cutoffs.FreezeLimit,
&cutoffs.MultiXactCutoff,
&num_tuples, &tups_vacuumed,
- &tups_recently_dead);
+ &tups_recently_dead, ctx);
/* return selected values to caller, get set as relfrozenxid/minmxid */
*pFreezeXid = cutoffs.FreezeLimit;
@@ -2361,6 +2365,8 @@ cluster_is_permitted_for_relation(RepackCommand cmd, Oid relid, Oid userid)
* instead return the opened and locked relcache entry, so that caller can
* process the partitions using the multiple-table handling code. In this
* case, if an index name is given, it's up to the caller to resolve it.
+ *
+ * A new transaction is started in either case.
*/
static Relation
process_single_relation(RepackStmt *stmt, LOCKMODE lockmode, bool isTopLevel,
@@ -2373,6 +2379,25 @@ process_single_relation(RepackStmt *stmt, LOCKMODE lockmode, bool isTopLevel,
Assert(stmt->command == REPACK_COMMAND_CLUSTER ||
stmt->command == REPACK_COMMAND_REPACK);
+ /*
+ * Since REPACK (CONCURRENTLY) pops the active snapshot during the
+ * processing (it creates and pushes snapshots on its own), and since that
+ * snapshot can be referenced by the current portal, we need to make sure
+ * that the portal has no dangling pointer to the snapshot. Starting a new
+ * transaction seems to be the simplest way.
+ */
+ PopActiveSnapshot();
+ CommitTransactionCommand();
+
+ /* Start a new transaction. */
+ StartTransactionCommand();
+
+ /*
+ * Functions in indexes may want a snapshot set. Note that the portal is
+ * not aware of this one, so the caller needs to pop it explicitly.
+ */
+ PushActiveSnapshot(GetTransactionSnapshot());
+
/* Find, lock, and check permissions on the table. */
tableOid = RangeVarGetRelidExtended(stmt->relation->relation,
lockmode,
@@ -2675,6 +2700,7 @@ decode_concurrent_changes(LogicalDecodingContext *ctx,
DecodingWorkerShared *shared)
{
RepackDecodingState *dstate;
+ bool snapshot_requested;
XLogRecPtr lsn_upto;
bool done;
char fname[MAXPGPATH];
@@ -2682,11 +2708,14 @@ decode_concurrent_changes(LogicalDecodingContext *ctx,
dstate = (RepackDecodingState *) ctx->output_writer_private;
/* Open the output file. */
- DecodingWorkerFileName(fname, shared->relid, shared->last_exported + 1);
+ DecodingWorkerFileName(fname, shared->relid,
+ shared->last_exported_changes + 1,
+ false);
dstate->file = BufFileCreateFileSet(&shared->sfs.fs, fname);
SpinLockAcquire(&shared->mutex);
lsn_upto = shared->lsn_upto;
+ snapshot_requested = shared->snapshot_requested;
done = shared->done;
SpinLockRelease(&shared->mutex);
@@ -2752,6 +2781,7 @@ decode_concurrent_changes(LogicalDecodingContext *ctx,
{
SpinLockAcquire(&shared->mutex);
lsn_upto = shared->lsn_upto;
+ snapshot_requested = shared->snapshot_requested;
/* 'done' should be set at the same time as 'lsn_upto' */
done = shared->done;
SpinLockRelease(&shared->mutex);
@@ -2796,27 +2826,105 @@ decode_concurrent_changes(LogicalDecodingContext *ctx,
*/
BufFileClose(dstate->file);
dstate->file = NULL;
+
+ /*
+ * Before publishing the data changes, export the snapshot too if
+ * requested. Publishing both at once makes sense because both are needed
+ * at the same time, and it's simpler.
+ */
+ if (snapshot_requested)
+ {
+ Snapshot snapshot;
+
+ snapshot = SnapBuildSnapshotForRepack(ctx->snapshot_builder);
+ export_snapshot(snapshot, shared);
+
+ /*
+ * Adjust the replication slot's xmin so that VACUUM can do more work.
+ */
+ LogicalIncreaseXminForSlot(InvalidXLogRecPtr, snapshot->xmin, false);
+ FreeSnapshot(snapshot);
+ }
+ else
+ {
+ /*
+ * If data changes were requested but no following snapshot, we don't
+ * care about xmin horizon because the heap copying should be done by
+ * now.
+ */
+ LogicalIncreaseXminForSlot(InvalidXLogRecPtr, InvalidTransactionId,
+ false);
+
+ }
+
+ /*
+ * Now increase the counter(s) to announce that the output is
+ * available.
+ */
SpinLockAcquire(&shared->mutex);
+ shared->last_exported_changes++;
shared->lsn_upto = InvalidXLogRecPtr;
- shared->last_exported++;
+ if (snapshot_requested)
+ {
+ shared->last_exported_snapshot++;
+ shared->snapshot_requested = false;
+ }
SpinLockRelease(&shared->mutex);
+
ConditionVariableSignal(&shared->cv);
return done;
}
/*
- * Apply changes stored in 'file'.
+ * Apply all concurrent changes.
*/
static void
-apply_concurrent_changes(BufFile *file, ChangeDest *dest)
+apply_concurrent_changes(ConcurrentChangeContext *ctx)
+{
+ DecodingWorkerShared *shared;
+
+ shared = (DecodingWorkerShared *) dsm_segment_address(decoding_worker->seg);
+
+ foreach_ptr(RepackApplyRange, range, ctx->block_ranges)
+ {
+ BufFile *file;
+
+ file = BufFileOpenFileSet(&shared->sfs.fs, range->fname, O_RDONLY,
+ false);
+
+ /*
+ * If range end is valid, the start should be as well.
+ */
+ Assert(!BlockNumberIsValid(range->end) ||
+ BlockNumberIsValid(ctx->first_block));
+
+ apply_concurrent_changes_file(ctx, file, range->end);
+ BufFileClose(file);
+
+ pfree(range->fname);
+ pfree(range);
+ }
+
+ /* Get ready for the next decoding. */
+ ctx->block_ranges = NIL;
+ ctx->first_block = InvalidBlockNumber;
+}
+
+/*
+ * Apply concurrent changes stored in 'file'.
+ */
+static void
+apply_concurrent_changes_file(ConcurrentChangeContext *ctx, BufFile *file,
+ BlockNumber range_end)
{
char kind;
uint32 t_len;
- Relation rel = dest->rel;
+ Relation rel = ctx->rel;
TupleTableSlot *index_slot,
*ident_slot;
HeapTuple tup_old = NULL;
+ bool check_range = BlockNumberIsValid(range_end);
/* TupleTableSlot is needed to pass the tuple to ExecInsertIndexTuples(). */
index_slot = MakeSingleTupleTableSlot(RelationGetDescr(rel),
@@ -2844,8 +2952,8 @@ apply_concurrent_changes(BufFile *file, ChangeDest *dest)
tup->t_data = (HeapTupleHeader) ((char *) tup + HEAPTUPLESIZE);
BufFileReadExact(file, tup->t_data, t_len);
tup->t_len = t_len;
- ItemPointerSetInvalid(&tup->t_self);
- tup->t_tableOid = RelationGetRelid(dest->rel);
+ tup->t_tableOid = RelationGetRelid(ctx->rel);
+ BufFileReadExact(file, &tup->t_self, sizeof(tup->t_self));
if (kind == CHANGE_UPDATE_OLD)
{
@@ -2856,7 +2964,10 @@ apply_concurrent_changes(BufFile *file, ChangeDest *dest)
{
Assert(tup_old == NULL);
- apply_concurrent_insert(rel, tup, dest->iistate, index_slot);
+ if (!check_range ||
+ is_tuple_in_block_range(tup, ctx->first_block, range_end))
+ apply_concurrent_insert(rel, tup, ctx->iistate,
+ index_slot);
pfree(tup);
}
@@ -2877,16 +2988,52 @@ apply_concurrent_changes(BufFile *file, ChangeDest *dest)
/*
* Find the tuple to be updated or deleted.
*/
- tup_exist = find_target_tuple(rel, dest, tup_key, ident_slot);
- if (tup_exist == NULL)
- elog(ERROR, "failed to find target tuple");
+ if (!check_range ||
+ (is_tuple_in_block_range(tup_key, ctx->first_block,
+ range_end)))
+ {
+ /* The change needs to be applied to this tuple. */
+ tup_exist = find_target_tuple(rel, ctx, tup_key, ident_slot);
+ if (tup_exist == NULL)
+ elog(ERROR, "failed to find target tuple");
- if (kind == CHANGE_UPDATE_NEW)
- apply_concurrent_update(rel, tup, tup_exist, dest->iistate,
- index_slot);
+ if (kind == CHANGE_DELETE)
+ apply_concurrent_delete(rel, tup_exist);
+ else
+ {
+ /* UPDATE */
+ if (!check_range || tup == tup_key ||
+ is_tuple_in_block_range(tup, ctx->first_block,
+ range_end))
+ /* The new tuple is in the same range. */
+ apply_concurrent_update(rel, tup, tup_exist,
+ ctx->iistate, index_slot);
+ else
+
+ /*
+ * The new key is in the other range, so only delete
+ * it from the current one. The new version should be
+ * visible to the snapshot that we'll use to copy the
+ * other block.
+ */
+ apply_concurrent_delete(rel, tup_exist);
+ }
+ }
else
- apply_concurrent_delete(rel, tup_exist);
-
+ {
+ /*
+ * The change belongs to another range, so we don't need to
+ * bother with the old tuple: the snapshot used for the other
+ * range won't see it, so it won't be copied. However, the new
+ * tuple still may need to go to the range we are checking. In
+ * that case, simply insert it there.
+ */
+ if (kind == CHANGE_UPDATE_NEW && tup != tup_key &&
+ is_tuple_in_block_range(tup, ctx->first_block,
+ range_end))
+ apply_concurrent_insert(rel, tup, ctx->iistate,
+ index_slot);
+ }
if (tup_old != NULL)
{
pfree(tup_old);
@@ -3025,6 +3172,33 @@ apply_concurrent_delete(Relation rel, HeapTuple tup_target)
pgstat_progress_incr_param(PROGRESS_REPACK_HEAP_TUPLES_DELETED, 1);
}
+/*
+ * Check if tuple originates from given range of blocks that have already been
+ * copied.
+ */
+static bool
+is_tuple_in_block_range(HeapTuple tup, BlockNumber start, BlockNumber end)
+{
+ BlockNumber blknum;
+
+ Assert(BlockNumberIsValid(start) && BlockNumberIsValid(end));
+
+ blknum = ItemPointerGetBlockNumber(&tup->t_self);
+ Assert(BlockNumberIsValid(blknum));
+
+ if (start < end)
+ {
+ return blknum >= start && blknum < end;
+ }
+ else
+ {
+ /* Has the scan position wrapped around? */
+ Assert(start > end);
+
+ return blknum >= start || blknum < end;
+ }
+}
+
/*
* Find the tuple to be updated or deleted.
*
@@ -3034,10 +3208,10 @@ apply_concurrent_delete(Relation rel, HeapTuple tup_target)
* it when he no longer needs the tuple returned.
*/
static HeapTuple
-find_target_tuple(Relation rel, ChangeDest *dest, HeapTuple tup_key,
- TupleTableSlot *ident_slot)
+find_target_tuple(Relation rel, ConcurrentChangeContext *ctx,
+ HeapTuple tup_key, TupleTableSlot *ident_slot)
{
- Relation ident_index = dest->ident_index;
+ Relation ident_index = ctx->ident_index;
IndexScanDesc scan;
Form_pg_index ident_form;
int2vector *ident_indkey;
@@ -3045,14 +3219,14 @@ find_target_tuple(Relation rel, ChangeDest *dest, HeapTuple tup_key,
/* XXX no instrumentation for now */
scan = index_beginscan(rel, ident_index, GetActiveSnapshot(),
- NULL, dest->ident_key_nentries, 0);
+ NULL, ctx->ident_key_nentries, 0);
/*
* Scan key is passed by caller, so it does not have to be constructed
* multiple times. Key entries have all fields initialized, except for
* sk_argument.
*/
- index_rescan(scan, dest->ident_key, dest->ident_key_nentries, NULL, 0);
+ index_rescan(scan, ctx->ident_key, ctx->ident_key_nentries, NULL, 0);
/* Info needed to retrieve key values from heap tuple. */
ident_form = ident_index->rd_index;
@@ -3087,15 +3261,22 @@ find_target_tuple(Relation rel, ChangeDest *dest, HeapTuple tup_key,
}
/*
- * Decode and apply concurrent changes, up to (and including) the record whose
- * LSN is 'end_of_wal'.
+ * Get concurrent changes, up to (and including) the record whose LSN is
+ * 'end_of_wal', from the decoding worker. If 'range_end' is a valid block
+ * number, the changes should only be applied to blocks greater than or equal
+ * to ctx->first_block and lower than range_end.
+ *
+ * If 'request_snapshot' is true, the snapshot built at LSN following the last
+ * data change needs to be exported too.
*/
-static void
-process_concurrent_changes(XLogRecPtr end_of_wal, ChangeDest *dest, bool done)
+extern void
+repack_get_concurrent_changes(ConcurrentChangeContext *ctx,
+ XLogRecPtr end_of_wal,
+ BlockNumber range_end,
+ bool request_snapshot, bool done)
{
DecodingWorkerShared *shared;
char fname[MAXPGPATH];
- BufFile *file;
pgstat_progress_update_param(PROGRESS_REPACK_PHASE,
PROGRESS_REPACK_PHASE_CATCH_UP);
@@ -3104,6 +3285,8 @@ process_concurrent_changes(XLogRecPtr end_of_wal, ChangeDest *dest, bool done)
shared = (DecodingWorkerShared *) dsm_segment_address(decoding_worker->seg);
SpinLockAcquire(&shared->mutex);
shared->lsn_upto = end_of_wal;
+ Assert(!shared->snapshot_requested);
+ shared->snapshot_requested = request_snapshot;
shared->done = done;
SpinLockRelease(&shared->mutex);
@@ -3118,30 +3301,52 @@ process_concurrent_changes(XLogRecPtr end_of_wal, ChangeDest *dest, bool done)
int last_exported;
SpinLockAcquire(&shared->mutex);
- last_exported = shared->last_exported;
+ last_exported = shared->last_exported_changes;
SpinLockRelease(&shared->mutex);
/*
* Has the worker exported the file we are waiting for?
*/
- if (last_exported == dest->file_seq)
+ if (last_exported == ctx->file_seq_changes)
break;
ConditionVariableSleep(&shared->cv, WAIT_EVENT_REPACK_WORKER_EXPORT);
}
ConditionVariableCancelSleep();
- /* Open the file. */
- DecodingWorkerFileName(fname, shared->relid, dest->file_seq);
- file = BufFileOpenFileSet(&shared->sfs.fs, fname, O_RDONLY, false);
- apply_concurrent_changes(file, dest);
+ /*
+ * Remember the file name so we can apply the changes when appropriate.
+ * One particular reason to postpone the replay is that indexes haven't
+ * been built yet on the new heap.
+ */
+ DecodingWorkerFileName(fname, shared->relid, ctx->file_seq_changes,
+ false);
+ repack_add_block_range(ctx, range_end, fname);
- BufFileClose(file);
+#ifdef USE_ASSERT_CHECKING
+ /* No file is exported until the worker exports the next one. */
+ SpinLockAcquire(&shared->mutex);
+ Assert(XLogRecPtrIsInvalid(shared->lsn_upto));
+ SpinLockRelease(&shared->mutex);
+#endif
+
+ /* Get ready for the next set of changes. */
+ ctx->file_seq_changes++;
+}
+
+static void
+repack_add_block_range(ConcurrentChangeContext *ctx, BlockNumber end,
+ char *fname)
+{
+ RepackApplyRange *range;
- /* Get ready for the next file. */
- dest->file_seq++;
+ range = palloc_object(RepackApplyRange);
+ range->end = end;
+ range->fname = pstrdup(fname);
+ ctx->block_ranges = lappend(ctx->block_ranges, range);
}
+
/*
* Initialize IndexInsertState for index specified by ident_index_id.
*
@@ -3284,7 +3489,8 @@ static void
rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap,
Relation cl_index,
TransactionId frozenXid,
- MultiXactId cutoffMulti)
+ MultiXactId cutoffMulti,
+ ConcurrentChangeContext *ctx)
{
LOCKMODE lockmode_old PG_USED_FOR_ASSERTS_ONLY;
List *ind_oids_new;
@@ -3303,7 +3509,6 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap,
Relation *ind_refs,
*ind_refs_p;
int nind;
- ChangeDest chgdst;
/* Like in cluster_rel(). */
lockmode_old = ShareUpdateExclusiveLock;
@@ -3360,12 +3565,18 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap,
(errmsg("identity index missing on the new relation")));
/* Gather information to apply concurrent changes. */
- chgdst.rel = NewHeap;
- chgdst.iistate = get_index_insert_state(NewHeap, ident_idx_new,
- &chgdst.ident_index);
- chgdst.ident_key = build_identity_key(ident_idx_new, OldHeap,
- &chgdst.ident_key_nentries);
- chgdst.file_seq = WORKER_FILE_SNAPSHOT + 1;
+ ctx->rel = NewHeap;
+ ctx->iistate = get_index_insert_state(NewHeap, ident_idx_new,
+ &ctx->ident_index);
+ ctx->ident_key = build_identity_key(ident_idx_new, OldHeap,
+ &ctx->ident_key_nentries);
+
+ /*
+ * Replay the concurrent data changes gathered during heap copying. This
+ * had to wait until after the index build because the identity index is
+ * needed to apply UPDATE and DELETE changes.
+ */
+ apply_concurrent_changes(ctx);
/*
* During testing, wait for another backend to perform concurrent data
@@ -3383,11 +3594,13 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap,
end_of_wal = GetFlushRecPtr(NULL);
/*
- * Apply concurrent changes first time, to minimize the time we need to
- * hold AccessExclusiveLock. (Quite some amount of WAL could have been
+ * Decode and apply concurrent changes again, to minimize the time we need
+ * to hold AccessExclusiveLock. (Quite some amount of WAL could have been
* written during the data copying and index creation.)
*/
- process_concurrent_changes(end_of_wal, &chgdst, false);
+ repack_get_concurrent_changes(ctx, end_of_wal, InvalidBlockNumber, false,
+ false);
+ apply_concurrent_changes(ctx);
/*
* Acquire AccessExclusiveLock on the table, its TOAST relation (if there
@@ -3482,10 +3695,13 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap,
end_of_wal = GetFlushRecPtr(NULL);
/*
- * Apply the concurrent changes again. Indicate that the decoding worker
- * won't be needed anymore.
+ * Decode and apply the concurrent changes again. Indicate that the
+ * decoding worker won't be needed anymore.
*/
- process_concurrent_changes(end_of_wal, &chgdst, true);
+ repack_get_concurrent_changes(ctx, end_of_wal, InvalidBlockNumber, false,
+ true);
+ apply_concurrent_changes(ctx);
+
/* Remember info about rel before closing OldHeap */
relpersistence = OldHeap->rd_rel->relpersistence;
@@ -3536,8 +3752,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap,
table_close(NewHeap, NoLock);
/* Cleanup what we don't need anymore. (And close the identity index.) */
- pfree(chgdst.ident_key);
- free_index_insert_state(chgdst.iistate);
+ pfree(ctx->ident_key);
+ free_index_insert_state(ctx->iistate);
/*
* Swap the relations and their TOAST relations and TOAST indexes. This
@@ -3578,6 +3794,23 @@ build_new_indexes(Relation NewHeap, Relation OldHeap, List *OldIndexes)
char *newName;
Relation ind;
+ /*
+ * Try to reduce the impact on VACUUM.
+ *
+ * The individual builds might still be a problem, but that's a
+ * separate issue.
+ *
+ * TODO Can we somehow use the fact that the new heap is not yet
+ * visible to other transaction, and thus cannot be vacuumed? Perhaps
+ * by preventing snapshots from setting MyProc->xmin temporarily. (All
+ * the snapshots that might have participated in the build, including
+ * the catalog snapshots, must not be used for other tables of
+ * course.)
+ */
+ PopActiveSnapshot();
+ InvalidateCatalogSnapshot();
+ PushActiveSnapshot(GetTransactionSnapshot());
+
ind = index_open(ind_oid, ShareUpdateExclusiveLock);
newName = ChooseRelationName(get_rel_name(ind_oid),
@@ -3616,10 +3849,14 @@ start_decoding_worker(Oid relid)
BUFFERALIGN(REPACK_ERROR_QUEUE_SIZE);
seg = dsm_create(size, 0);
shared = (DecodingWorkerShared *) dsm_segment_address(seg);
+ shared->initialized = false;
shared->lsn_upto = InvalidXLogRecPtr;
shared->done = false;
+ /* Snapshot is the first thing we need from the worker. */
+ shared->snapshot_requested = true;
SharedFileSetInit(&shared->sfs, seg);
- shared->last_exported = -1;
+ shared->last_exported_changes = -1;
+ shared->last_exported_snapshot = -1;
SpinLockInit(&shared->mutex);
shared->dbid = MyDatabaseId;
@@ -3828,6 +4065,9 @@ repack_worker_internal(dsm_segment *seg)
*/
SpinLockAcquire(&shared->mutex);
Assert(XLogRecPtrIsInvalid(shared->lsn_upto));
+ /* Initially we're expected to provide a snapshot and only that. */
+ Assert(shared->snapshot_requested &&
+ XLogRecPtrIsInvalid(shared->lsn_upto));
sfs = &shared->sfs;
SpinLockRelease(&shared->mutex);
@@ -3845,8 +4085,22 @@ repack_worker_internal(dsm_segment *seg)
ConditionVariableSignal(&shared->cv);
/* Build the initial snapshot and export it. */
- snapshot = SnapBuildInitialSnapshotForRepack(decoding_ctx->snapshot_builder);
- export_initial_snapshot(snapshot, shared);
+ snapshot = SnapBuildSnapshotForRepack(decoding_ctx->snapshot_builder);
+ export_snapshot(snapshot, shared);
+
+ /*
+ * Adjust the replication slot's xmin so that VACUUM can do more work.
+ */
+ LogicalIncreaseXminForSlot(InvalidXLogRecPtr, snapshot->xmin, false);
+ FreeSnapshot(snapshot);
+
+ /* Increase the counter to tell the backend that the file is available. */
+ SpinLockAcquire(&shared->mutex);
+ Assert(shared->snapshot_requested);
+ shared->last_exported_snapshot++;
+ shared->snapshot_requested = false;
+ SpinLockRelease(&shared->mutex);
+ ConditionVariableSignal(&shared->cv);
/*
* Only historic snapshots should be used now. Do not let us restrict the
@@ -3866,7 +4120,7 @@ repack_worker_internal(dsm_segment *seg)
* Make snapshot available to the backend that launched the decoding worker.
*/
static void
-export_initial_snapshot(Snapshot snapshot, DecodingWorkerShared *shared)
+export_snapshot(Snapshot snapshot, DecodingWorkerShared *shared)
{
char fname[MAXPGPATH];
BufFile *file;
@@ -3876,28 +4130,23 @@ export_initial_snapshot(Snapshot snapshot, DecodingWorkerShared *shared)
snap_size = EstimateSnapshotSpace(snapshot);
snap_space = (char *) palloc(snap_size);
SerializeSnapshot(snapshot, snap_space);
- FreeSnapshot(snapshot);
- DecodingWorkerFileName(fname, shared->relid, shared->last_exported + 1);
+ DecodingWorkerFileName(fname, shared->relid,
+ shared->last_exported_snapshot + 1,
+ true);
file = BufFileCreateFileSet(&shared->sfs.fs, fname);
/* To make restoration easier, write the snapshot size first. */
BufFileWrite(file, &snap_size, sizeof(snap_size));
BufFileWrite(file, snap_space, snap_size);
pfree(snap_space);
BufFileClose(file);
-
- /* Increase the counter to tell the backend that the file is available. */
- SpinLockAcquire(&shared->mutex);
- shared->last_exported++;
- SpinLockRelease(&shared->mutex);
- ConditionVariableSignal(&shared->cv);
}
/*
- * Get the initial snapshot from the decoding worker.
+ * Get snapshot from the decoding worker.
*/
-static Snapshot
-get_initial_snapshot(DecodingWorker *worker)
+extern Snapshot
+repack_get_snapshot(ConcurrentChangeContext *ctx)
{
DecodingWorkerShared *shared;
char fname[MAXPGPATH];
@@ -3905,13 +4154,15 @@ get_initial_snapshot(DecodingWorker *worker)
Size snap_size;
char *snap_space;
Snapshot snapshot;
+ DecodingWorker *worker = ctx->worker;
shared = (DecodingWorkerShared *) dsm_segment_address(worker->seg);
/*
- * The worker needs to initialize the logical decoding, which usually
- * takes some time. Therefore it makes sense to prepare for the sleep
- * first.
+ * For the first snapshot request, the worker needs to initialize the
+ * logical decoding, which usually takes some time. Therefore it makes
+ * sense to prepare for the sleep first. Does it make sense to skip the
+ * preparation on the next requests?
*/
ConditionVariablePrepareToSleep(&shared->cv);
for (;;)
@@ -3919,13 +4170,13 @@ get_initial_snapshot(DecodingWorker *worker)
int last_exported;
SpinLockAcquire(&shared->mutex);
- last_exported = shared->last_exported;
+ last_exported = shared->last_exported_snapshot;
SpinLockRelease(&shared->mutex);
/*
* Has the worker exported the file we are waiting for?
*/
- if (last_exported == WORKER_FILE_SNAPSHOT)
+ if (last_exported == ctx->file_seq_snapshot)
break;
ConditionVariableSleep(&shared->cv, WAIT_EVENT_REPACK_WORKER_EXPORT);
@@ -3933,17 +4184,27 @@ get_initial_snapshot(DecodingWorker *worker)
ConditionVariableCancelSleep();
/* Read the snapshot from a file. */
- DecodingWorkerFileName(fname, shared->relid, WORKER_FILE_SNAPSHOT);
+ DecodingWorkerFileName(fname, shared->relid, ctx->file_seq_snapshot,
+ true);
file = BufFileOpenFileSet(&shared->sfs.fs, fname, O_RDONLY, false);
BufFileReadExact(file, &snap_size, sizeof(snap_size));
snap_space = (char *) palloc(snap_size);
BufFileReadExact(file, snap_space, snap_size);
BufFileClose(file);
+#ifdef USE_ASSERT_CHECKING
+ SpinLockAcquire(&shared->mutex);
+ Assert(!shared->snapshot_requested);
+ SpinLockRelease(&shared->mutex);
+#endif
+
/* Restore it. */
snapshot = RestoreSnapshot(snap_space);
pfree(snap_space);
+ /* Get ready for the next snapshot. */
+ ctx->file_seq_snapshot++;
+
return snapshot;
}
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index dc8c7be2aca..8f42238ab21 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -920,6 +920,7 @@ DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
xl_heap_insert *xlrec;
ReorderBufferChange *change;
RelFileLocator target_locator;
+ BlockNumber blknum;
xlrec = (xl_heap_insert *) XLogRecGetData(r);
@@ -931,7 +932,7 @@ DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
return;
/* only interested in our database */
- XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
+ XLogRecGetBlockTag(r, 0, &target_locator, NULL, &blknum);
if (target_locator.dbOid != ctx->slot->data.database)
return;
@@ -956,6 +957,15 @@ DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple);
+ /*
+ * REPACK (CONCURRENTLY) needs block number to check if the corresponding
+ * part of the table was already copied.
+ */
+ if (am_decoding_for_repack())
+ /* offnum is not really needed, but let's set valid pointer. */
+ ItemPointerSet(&change->data.tp.newtuple->t_self, blknum,
+ xlrec->offnum);
+
change->data.tp.clear_toast_afterwards = true;
ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
@@ -977,11 +987,12 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
ReorderBufferChange *change;
char *data;
RelFileLocator target_locator;
+ BlockNumber new_blknum;
xlrec = (xl_heap_update *) XLogRecGetData(r);
/* only interested in our database */
- XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
+ XLogRecGetBlockTag(r, 0, &target_locator, NULL, &new_blknum);
if (target_locator.dbOid != ctx->slot->data.database)
return;
@@ -1007,12 +1018,27 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
ReorderBufferAllocTupleBuf(ctx->reorder, tuplelen);
DecodeXLogTuple(data, datalen, change->data.tp.newtuple);
+
+ /*
+ * REPACK (CONCURRENTLY) needs block number to check if the
+ * corresponding part of the table was already copied.
+ */
+ if (am_decoding_for_repack())
+ /* offnum is not really needed, but let's set valid pointer. */
+ ItemPointerSet(&change->data.tp.newtuple->t_self,
+ new_blknum, xlrec->new_offnum);
}
if (xlrec->flags & XLH_UPDATE_CONTAINS_OLD)
{
Size datalen;
Size tuplelen;
+ BlockNumber old_blknum;
+
+ if (XLogRecHasBlockRef(r, 1))
+ XLogRecGetBlockTag(r, 1, NULL, NULL, &old_blknum);
+ else
+ XLogRecGetBlockTag(r, 0, NULL, NULL, &old_blknum);
/* caution, remaining data in record is not aligned */
data = XLogRecGetData(r) + SizeOfHeapUpdate;
@@ -1023,6 +1049,11 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
ReorderBufferAllocTupleBuf(ctx->reorder, tuplelen);
DecodeXLogTuple(data, datalen, change->data.tp.oldtuple);
+ /* See above. */
+ if (am_decoding_for_repack())
+ ItemPointerSet(&change->data.tp.oldtuple->t_self,
+ old_blknum, xlrec->old_offnum);
+
}
change->data.tp.clear_toast_afterwards = true;
@@ -1043,6 +1074,7 @@ DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
xl_heap_delete *xlrec;
ReorderBufferChange *change;
RelFileLocator target_locator;
+ BlockNumber blknum;
xlrec = (xl_heap_delete *) XLogRecGetData(r);
@@ -1056,7 +1088,7 @@ DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
return;
/* only interested in our database */
- XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
+ XLogRecGetBlockTag(r, 0, &target_locator, NULL, &blknum);
if (target_locator.dbOid != ctx->slot->data.database)
return;
@@ -1088,6 +1120,15 @@ DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete,
datalen, change->data.tp.oldtuple);
+
+ /*
+ * REPACK (CONCURRENTLY) needs block number to check if the
+ * corresponding part of the table was already copied.
+ */
+ if (am_decoding_for_repack())
+ /* offnum is not really needed, but let's set valid pointer. */
+ ItemPointerSet(&change->data.tp.oldtuple->t_self, blknum,
+ xlrec->offnum);
}
change->data.tp.clear_toast_afterwards = true;
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 35a46988285..76119c5ecaa 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -1661,14 +1661,17 @@ update_progress_txn_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
/*
* Set the required catalog xmin horizon for historic snapshots in the current
- * replication slot.
+ * replication slot if catalog is TRUE, or xmin if catalog is FALSE.
*
* Note that in the most cases, we won't be able to immediately use the xmin
* to increase the xmin horizon: we need to wait till the client has confirmed
- * receiving current_lsn with LogicalConfirmReceivedLocation().
+ * receiving current_lsn with LogicalConfirmReceivedLocation(). However,
+ * catalog=FALSE is only allowed for temporary replication slots, so the
+ * horizon is applied immediately.
*/
void
-LogicalIncreaseXminForSlot(XLogRecPtr current_lsn, TransactionId xmin)
+LogicalIncreaseXminForSlot(XLogRecPtr current_lsn, TransactionId xmin,
+ bool catalog)
{
bool updated_xmin = false;
ReplicationSlot *slot;
@@ -1679,6 +1682,27 @@ LogicalIncreaseXminForSlot(XLogRecPtr current_lsn, TransactionId xmin)
Assert(slot != NULL);
SpinLockAcquire(&slot->mutex);
+ if (!catalog)
+ {
+ /*
+ * The non-catalog horizon can only advance in temporary slots, so
+ * update it in the shared memory immediately (w/o requiring prior
+ * saving to disk).
+ */
+ Assert(slot->data.persistency == RS_TEMPORARY);
+
+ /*
+ * The horizon must not go backwards, however it's o.k. to become
+ * invalid.
+ */
+ Assert(!TransactionIdIsValid(slot->effective_xmin) ||
+ !TransactionIdIsValid(xmin) ||
+ TransactionIdFollowsOrEquals(xmin, slot->effective_xmin));
+
+ slot->effective_xmin = xmin;
+ SpinLockRelease(&slot->mutex);
+ return;
+ }
/*
* don't overwrite if we already have a newer xmin. This can happen if we
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index a0293f6ec7c..3003cadd76e 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -3734,6 +3734,56 @@ ReorderBufferXidHasCatalogChanges(ReorderBuffer *rb, TransactionId xid)
return rbtxn_has_catalog_changes(txn);
}
+/*
+ * Check if a transaction (or its subtransaction) contains a heap change.
+ */
+bool
+ReorderBufferXidHasHeapChanges(ReorderBuffer *rb, TransactionId xid)
+{
+ ReorderBufferTXN *txn;
+ dlist_iter iter;
+
+ txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
+ false);
+ if (txn == NULL)
+ return false;
+
+ dlist_foreach(iter, &txn->changes)
+ {
+ ReorderBufferChange *change;
+
+ change = dlist_container(ReorderBufferChange, node, iter.cur);
+
+ switch (change->action)
+ {
+ case REORDER_BUFFER_CHANGE_INSERT:
+ case REORDER_BUFFER_CHANGE_UPDATE:
+ case REORDER_BUFFER_CHANGE_DELETE:
+ return true;
+ default:
+ break;
+ }
+ }
+
+ /* Check subtransactions. */
+
+ /*
+ * TODO Verify that subtransactions must be assigned to the top-level
+ * transactions by now.
+ */
+ dlist_foreach(iter, &txn->subtxns)
+ {
+ ReorderBufferTXN *subtxn;
+
+ subtxn = dlist_container(ReorderBufferTXN, node, iter.cur);
+
+ if (ReorderBufferXidHasHeapChanges(rb, subtxn->xid))
+ return true;
+ }
+
+ return false;
+}
+
/*
* ReorderBufferXidHasBaseSnapshot
* Have we already set the base snapshot for the given txn/subtxn?
diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
index e238bcd73cd..fbc24de6e24 100644
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -128,6 +128,7 @@
#include "access/heapam_xlog.h"
#include "access/transam.h"
#include "access/xact.h"
+#include "commands/cluster.h"
#include "common/file_utils.h"
#include "miscadmin.h"
#include "pgstat.h"
@@ -496,7 +497,7 @@ SnapBuildInitialSnapshot(SnapBuild *builder)
* we do not set MyProc->xmin). XXX Do we yet need to add some restrictions?
*/
Snapshot
-SnapBuildInitialSnapshotForRepack(SnapBuild *builder)
+SnapBuildSnapshotForRepack(SnapBuild *builder)
{
Snapshot snap;
@@ -1035,6 +1036,28 @@ SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
}
}
+ /*
+ * Is REPACKED (CONCURRENTLY) is being run by this backend?
+ */
+ else if (am_decoding_for_repack())
+ {
+ Assert(builder->building_full_snapshot);
+
+ /*
+ * In this special mode, heap changes of other relations should not be
+ * decoded at all - see heap_decode(). Thus if we find a single heap
+ * change in this transaction (or its subtransaction), we know that
+ * this transaction changes the relation being repacked.
+ */
+ if (ReorderBufferXidHasHeapChanges(builder->reorder, xid))
+
+ /*
+ * Record the commit so we can build snapshots for the relation
+ * being repacked.
+ */
+ needs_timetravel = true;
+ }
+
for (nxact = 0; nxact < nsubxacts; nxact++)
{
TransactionId subxid = subxacts[nxact];
@@ -1240,7 +1263,7 @@ SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xact
xmin = running->oldestRunningXid;
elog(DEBUG3, "xmin: %u, xmax: %u, oldest running: %u, oldest xmin: %u",
builder->xmin, builder->xmax, running->oldestRunningXid, xmin);
- LogicalIncreaseXminForSlot(lsn, xmin);
+ LogicalIncreaseXminForSlot(lsn, xmin, true);
/*
* Also tell the slot where we can restart decoding from. We don't want to
diff --git a/src/backend/replication/pgoutput_repack/pgoutput_repack.c b/src/backend/replication/pgoutput_repack/pgoutput_repack.c
index fb9956d392d..be1c3ec9626 100644
--- a/src/backend/replication/pgoutput_repack/pgoutput_repack.c
+++ b/src/backend/replication/pgoutput_repack/pgoutput_repack.c
@@ -195,6 +195,8 @@ store_change(LogicalDecodingContext *ctx, ConcurrentChangeKind kind,
BufFileWrite(dstate->file, &tuple->t_len, sizeof(tuple->t_len));
/* ... and the tuple itself. */
BufFileWrite(dstate->file, tuple->t_data, tuple->t_len);
+ /* CTID is needed as well, to check block ranges. */
+ BufFileWrite(dstate->file, &tuple->t_self, sizeof(tuple->t_self));
/* Free the flat copy if created above. */
if (flattened)
diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat
index 7c60b125564..24f29f0016e 100644
--- a/src/backend/utils/misc/guc_parameters.dat
+++ b/src/backend/utils/misc/guc_parameters.dat
@@ -2424,6 +2424,16 @@
boot_val => 'true',
},
+# TODO Tune boot_val, 1024 is probably too low.
+{ name => 'repack_snapshot_after', type => 'int', context => 'PGC_USERSET', group => 'DEVELOPER_OPTIONS',
+ short_desc => 'Number of pages after which REPACK (CONCURRENTLY) builds a new snapshot.',
+ flags => 'GUC_UNIT_BLOCKS | GUC_NOT_IN_SAMPLE',
+ variable => 'repack_blocks_per_snapshot',
+ boot_val => '1024',
+ min => '1',
+ max => 'INT_MAX',
+}
+
{ name => 'reserved_connections', type => 'int', context => 'PGC_POSTMASTER', group => 'CONN_AUTH_SETTINGS',
short_desc => 'Sets the number of connection slots reserved for roles with privileges of pg_use_reserved_connections.',
variable => 'ReservedConnections',
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index 73ff6ad0a32..55c761de759 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -42,6 +42,7 @@
#include "catalog/namespace.h"
#include "catalog/storage.h"
#include "commands/async.h"
+#include "commands/cluster.h"
#include "commands/extension.h"
#include "commands/event_trigger.h"
#include "commands/tablespace.h"
diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h
index 15760363a1a..03ba89e6989 100644
--- a/src/include/access/tableam.h
+++ b/src/include/access/tableam.h
@@ -629,12 +629,12 @@ typedef struct TableAmRoutine
Relation OldIndex,
bool use_sort,
TransactionId OldestXmin,
- Snapshot snapshot,
TransactionId *xid_cutoff,
MultiXactId *multi_cutoff,
double *num_tuples,
double *tups_vacuumed,
- double *tups_recently_dead);
+ double *tups_recently_dead,
+ void *tableam_data);
/*
* React to VACUUM command on the relation. The VACUUM can be triggered by
@@ -1647,8 +1647,6 @@ table_relation_copy_data(Relation rel, const RelFileLocator *newrlocator)
* not needed for the relation's AM
* - *xid_cutoff - ditto
* - *multi_cutoff - ditto
- * - snapshot - if != NULL, ignore data changes done by transactions that this
- * (MVCC) snapshot considers still in-progress or in the future.
*
* Output parameters:
* - *xid_cutoff - rel's new relfrozenxid value, may be invalid
@@ -1661,19 +1659,19 @@ table_relation_copy_for_cluster(Relation OldTable, Relation NewTable,
Relation OldIndex,
bool use_sort,
TransactionId OldestXmin,
- Snapshot snapshot,
TransactionId *xid_cutoff,
MultiXactId *multi_cutoff,
double *num_tuples,
double *tups_vacuumed,
- double *tups_recently_dead)
+ double *tups_recently_dead,
+ void *tableam_data)
{
OldTable->rd_tableam->relation_copy_for_cluster(OldTable, NewTable, OldIndex,
use_sort, OldestXmin,
- snapshot,
xid_cutoff, multi_cutoff,
num_tuples, tups_vacuumed,
- tups_recently_dead);
+ tups_recently_dead,
+ tableam_data);
}
/*
diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h
index 1b05d5d418b..438ee0d751e 100644
--- a/src/include/commands/cluster.h
+++ b/src/include/commands/cluster.h
@@ -46,6 +46,72 @@ typedef struct ClusterParams
* The following definitions are used by REPACK CONCURRENTLY.
*/
+extern PGDLLIMPORT int repack_blocks_per_snapshot;
+
+/*
+ * Everything we need to call ExecInsertIndexTuples().
+ */
+typedef struct IndexInsertState
+{
+ ResultRelInfo *rri;
+ EState *estate;
+} IndexInsertState;
+
+/*
+ * Backend-local information to control the decoding worker.
+ */
+typedef struct DecodingWorker
+{
+ /* The worker. */
+ BackgroundWorkerHandle *handle;
+
+ /* DecodingWorkerShared is in this segment. */
+ dsm_segment *seg;
+
+ /* Handle of the error queue. */
+ shm_mq_handle *error_mqh;
+} DecodingWorker;
+
+/*
+ * Information needed to handle concurrent data changes.
+ */
+typedef struct ConcurrentChangeContext
+{
+ /* The relation the changes are applied to. */
+ Relation rel;
+
+ /*
+ * Background worker performing logical decoding of concurrent data
+ * changes.
+ */
+ DecodingWorker *worker;
+
+ /*
+ * Sequential numbers of the most recent files containing snapshots and
+ * data changes respectively. These files are created by the decoding
+ * worker.
+ */
+ int file_seq_snapshot;
+ int file_seq_changes;
+
+ /*
+ * The following is needed to find the existing tuple if the change is
+ * UPDATE or DELETE. 'ident_key' should have all the fields except for
+ * 'sk_argument' initialized.
+ */
+ Relation ident_index;
+ ScanKey ident_key;
+ int ident_key_nentries;
+
+ /* Needed to update indexes of rel_dst. */
+ IndexInsertState *iistate;
+
+ /* The first block of the scan used to copy the heap. */
+ BlockNumber first_block;
+ /* List of RepackApplyRange objects. */
+ List *block_ranges;
+} ConcurrentChangeContext;
+
/*
* Stored as a single byte in the output file.
*/
@@ -103,6 +169,12 @@ extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap,
extern bool am_decoding_for_repack(void);
extern bool change_useless_for_repack(XLogRecordBuffer *buf);
+extern void repack_get_concurrent_changes(struct ConcurrentChangeContext *ctx,
+ XLogRecPtr end_of_wal,
+ BlockNumber range_end,
+ bool request_snapshot,
+ bool done);
+extern Snapshot repack_get_snapshot(struct ConcurrentChangeContext *ctx);
extern void RepackWorkerMain(Datum main_arg);
#endif /* CLUSTER_H */
diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h
index 5b43e181135..432dca928e3 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -137,7 +137,7 @@ extern bool DecodingContextReady(LogicalDecodingContext *ctx);
extern void FreeDecodingContext(LogicalDecodingContext *ctx);
extern void LogicalIncreaseXminForSlot(XLogRecPtr current_lsn,
- TransactionId xmin);
+ TransactionId xmin, bool catalog);
extern void LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn,
XLogRecPtr restart_lsn);
extern void LogicalConfirmReceivedLocation(XLogRecPtr lsn);
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 314e35592c0..19df5f4a9ee 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -763,6 +763,7 @@ extern void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRe
extern void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn);
extern bool ReorderBufferXidHasCatalogChanges(ReorderBuffer *rb, TransactionId xid);
+extern bool ReorderBufferXidHasHeapChanges(ReorderBuffer *rb, TransactionId xid);
extern bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *rb, TransactionId xid);
extern bool ReorderBufferRememberPrepareInfo(ReorderBuffer *rb, TransactionId xid,
diff --git a/src/include/replication/snapbuild.h b/src/include/replication/snapbuild.h
index 5ee267d1c90..b20a4d1a93d 100644
--- a/src/include/replication/snapbuild.h
+++ b/src/include/replication/snapbuild.h
@@ -73,7 +73,7 @@ extern void FreeSnapshotBuilder(SnapBuild *builder);
extern void SnapBuildSnapDecRefcount(Snapshot snap);
extern Snapshot SnapBuildInitialSnapshot(SnapBuild *builder);
-extern Snapshot SnapBuildInitialSnapshotForRepack(SnapBuild *builder);
+extern Snapshot SnapBuildSnapshotForRepack(SnapBuild *builder);
extern Snapshot SnapBuildMVCCFromHistoric(Snapshot snapshot, bool in_place);
extern const char *SnapBuildExportSnapshot(SnapBuild *builder);
extern void SnapBuildClearExportedSnapshot(void);
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index d1a694f9008..220a2b43aa1 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -419,7 +419,6 @@ CatCacheHeader
CatalogId
CatalogIdMapEntry
CatalogIndexState
-ChangeDest
ChangeVarNodes_callback
ChangeVarNodes_context
CheckPoint
@@ -496,6 +495,7 @@ CompressFileHandle
CompressionLocation
CompressorState
ComputeXidHorizonsResult
+ConcurrentChangeContext
ConcurrentChangeKind
ConditionVariable
ConditionVariableMinimallyPadded
@@ -2575,6 +2575,7 @@ ReorderBufferTupleCidKey
ReorderBufferUpdateProgressTxnCB
ReorderTuple
RepOriginId
+RepackApplyRange
RepackCommand
RepackDecodingState
RepackStmt
--
2.47.3
--=-=-=--
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: Wrong results with equality search using trigram index and non-deterministic collation
@ 2026-05-04 11:53 David Geier <geidav.pg@gmail.com>
0 siblings, 1 reply; 9+ messages in thread
From: David Geier @ 2026-05-04 11:53 UTC (permalink / raw)
To: Laurenz Albe <laurenz.albe@cybertec.at>; pgsql-hackers@lists.postgresql.org
>> I think we should change that. It's very counter intuitive that a query
>> can change behavior when the planner flips from using e.g. a Seq Scan to
>> a Bitmap Index Scan or the other way around. There's already a patch for
>> that, see [1].
>>
>> [1]
>> https://www.postgresql.org/message-id/flat/db087c3e-230e-4119-8a03-8b5d74956bc2%40gmail.com
>
>
> That's not only unintuitive, it is a clear bug.
> An index is not allowed to change the semantics of a query.
>
> Does your patch fix the bug, that is, will the query with "WHERE t = 'right'"
> return both results? That's the case that is mostly in need of fixing.
> I am not sure if the behavior for the % operator should also be considered
> a bug.
Not yet :).
>>> I don't know what the correct fix would be. Perhaps just refusing to use
>>> the index for equality comparisons with non-deterministic collations.
>>
>> If we merge [1], then not only = but also LIKE would be incorrect. How
>> about disabling CREATE INDEX USING gin on columns with non-deterministic
>> collations?
>
> Oh, I see. So your patch won't fix the bug.
Indeed.
> I am not sure if refusing to *create* the index is the best solution.
> Perhaps a warning will be better:
>
> WARNING: GIN indexes won't be used columns with non-deterministic collations
At this point we could more generally say
WARNING: pg_trgm does not use inferred column collations but always
uses the default database collation.
But I'm hoping the patch in [1] gets merged and we can also use the
inferred collation to fix the bug you found.
>> Or is there maybe a way to make these cases work correctly for
>> non-deterministic collations by applying the collation when extracting
>> the search trigrams? I take a look into that.
Attached patch makes your case work, including the % case. It builds on
top of the other patches from [1] that makes pg_trgm use the inferred
collation trigram extraction.
Instead of using btint4cmp() to compare trigrams, the patch uses a
collation-aware string comparison function.
This is just a PoC. I haven't given much thought to the details but e.g.
when three consecutive characters exceed 3 bytes then compact_trigram()
uses a truncated 32-bit hash value as trigram instead. Such trigrams
won't work in all cases. We could omit them from the query string but
for languages where the majority of trigrams are hashed or where the
query string consists of only a few trigrams, the look-up performance
would suffer.
I guess better would be using a collation-aware hash function that maps
different values that compare equal to the same hash value. hashtext()
does that already. The new comparison function would then have to
distinguish between plain text trigrams and hash trigrams.
Alternatively, we could store all trigrams as hashes but that would
break functions such as show_trgm().
--
David Geier
Attachments:
[text/x-patch] v1-0003-Use-correct-collation-for-comparison.patch (5.5K, ../../9b850976-8f0d-4957-9308-e1c053a35559@gmail.com/2-v1-0003-Use-correct-collation-for-comparison.patch)
download | inline diff:
From 5c86bfd3e83b2bde7706746c59ac148b1553e717 Mon Sep 17 00:00:00 2001
From: David Geier <geidav.pg@gmail.com>
Date: Thu, 23 Apr 2026 11:08:33 +0200
Subject: [PATCH v1 3/3] Use correct collation for comparison
---
contrib/pg_trgm/Makefile | 6 +--
contrib/pg_trgm/meson.build | 1 +
contrib/pg_trgm/pg_trgm--1.6--1.7.sql | 20 +++++++++
contrib/pg_trgm/pg_trgm.control | 2 +-
contrib/pg_trgm/trgm_gin.c | 65 +++++++++++++++++++++++++++
5 files changed, 90 insertions(+), 4 deletions(-)
create mode 100644 contrib/pg_trgm/pg_trgm--1.6--1.7.sql
diff --git a/contrib/pg_trgm/Makefile b/contrib/pg_trgm/Makefile
index 26b3028b75e..556b76f49f2 100644
--- a/contrib/pg_trgm/Makefile
+++ b/contrib/pg_trgm/Makefile
@@ -9,9 +9,9 @@ OBJS = \
trgm_regexp.o
EXTENSION = pg_trgm
-DATA = pg_trgm--1.5--1.6.sql pg_trgm--1.4--1.5.sql pg_trgm--1.3--1.4.sql \
- pg_trgm--1.3.sql pg_trgm--1.2--1.3.sql pg_trgm--1.1--1.2.sql \
- pg_trgm--1.0--1.1.sql
+DATA = pg_trgm--1.6--1.7.sql pg_trgm--1.5--1.6.sql pg_trgm--1.4--1.5.sql \
+ pg_trgm--1.3--1.4.sql pg_trgm--1.3.sql pg_trgm--1.2--1.3.sql \
+ pg_trgm--1.1--1.2.sql pg_trgm--1.0--1.1.sql
PGFILEDESC = "pg_trgm - trigram matching"
REGRESS = pg_trgm pg_utf8_trgm pg_word_trgm pg_strict_word_trgm pg_trgm_collation
diff --git a/contrib/pg_trgm/meson.build b/contrib/pg_trgm/meson.build
index 5eafa774435..9fdf8c2d07e 100644
--- a/contrib/pg_trgm/meson.build
+++ b/contrib/pg_trgm/meson.build
@@ -28,6 +28,7 @@ install_data(
'pg_trgm--1.3.sql',
'pg_trgm--1.4--1.5.sql',
'pg_trgm--1.5--1.6.sql',
+ 'pg_trgm--1.6--1.7.sql',
'pg_trgm.control',
kwargs: contrib_data_args,
)
diff --git a/contrib/pg_trgm/pg_trgm--1.6--1.7.sql b/contrib/pg_trgm/pg_trgm--1.6--1.7.sql
new file mode 100644
index 00000000000..dc5552e375e
--- /dev/null
+++ b/contrib/pg_trgm/pg_trgm--1.6--1.7.sql
@@ -0,0 +1,20 @@
+/* contrib/pg_trgm/pg_trgm--1.6--1.7.sql */
+
+-- complain if script is sourced in psql, rather than via ALTER EXTENSION
+\echo Use "ALTER EXTENSION pg_trgm UPDATE TO '1.7'" to load this file. \quit
+
+-- Create collation-aware comparison function for trigrams
+CREATE FUNCTION gin_compare_value_trgm(int4, int4)
+RETURNS int4
+AS 'MODULE_PATHNAME'
+LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
+
+-- Replace btint4cmp with gin_compare_value_trgm in the operator family
+-- This ensures trigram comparisons respect collation settings
+-- First drop the old function, then add the new one
+ALTER OPERATOR FAMILY gin_trgm_ops USING gin DROP
+ FUNCTION 1 (text, text);
+
+ALTER OPERATOR FAMILY gin_trgm_ops USING gin ADD
+ FUNCTION 1 (text, text) gin_compare_value_trgm (int4, int4);
+
diff --git a/contrib/pg_trgm/pg_trgm.control b/contrib/pg_trgm/pg_trgm.control
index 1d6a9ddf259..6e3ee43c510 100644
--- a/contrib/pg_trgm/pg_trgm.control
+++ b/contrib/pg_trgm/pg_trgm.control
@@ -1,6 +1,6 @@
# pg_trgm extension
comment = 'text similarity measurement and index searching based on trigrams'
-default_version = '1.6'
+default_version = '1.7'
module_pathname = '$libdir/pg_trgm'
relocatable = true
trusted = true
diff --git a/contrib/pg_trgm/trgm_gin.c b/contrib/pg_trgm/trgm_gin.c
index 14a892c657d..dadbc789349 100644
--- a/contrib/pg_trgm/trgm_gin.c
+++ b/contrib/pg_trgm/trgm_gin.c
@@ -5,8 +5,11 @@
#include "access/gin.h"
#include "access/stratnum.h"
+#include "catalog/pg_collation.h"
+#include "catalog/pg_type.h"
#include "fmgr.h"
#include "trgm.h"
+#include "utils/pg_locale.h"
#include "varatt.h"
PG_FUNCTION_INFO_V1(gin_extract_trgm);
@@ -14,6 +17,7 @@ PG_FUNCTION_INFO_V1(gin_extract_value_trgm);
PG_FUNCTION_INFO_V1(gin_extract_query_trgm);
PG_FUNCTION_INFO_V1(gin_trgm_consistent);
PG_FUNCTION_INFO_V1(gin_trgm_triconsistent);
+PG_FUNCTION_INFO_V1(gin_compare_value_trgm);
/*
* This function can only be called if a pre-9.1 version of the GIN operator
@@ -169,6 +173,67 @@ gin_extract_query_trgm(PG_FUNCTION_ARGS)
PG_RETURN_POINTER(entries);
}
+/*
+ * Compare two trigram values for GIN index.
+ * This function considers the active collation when comparing trigrams,
+ * unlike btint4cmp which treats them as plain integers.
+ */
+Datum
+gin_compare_value_trgm(PG_FUNCTION_ARGS)
+{
+ int32 a = PG_GETARG_INT32(0);
+ int32 b = PG_GETARG_INT32(1);
+ Oid collid = PG_GET_COLLATION();
+ pg_locale_t locale = 0;
+
+ /*
+ * If a non-default collation is specified, we need to compare the
+ * trigrams character-by-character using the collation's rules.
+ */
+ if (collid != DEFAULT_COLLATION_OID)
+ locale = pg_newlocale_from_collation(collid);
+
+ if (locale && locale->collate_is_c)
+ locale = 0; /* C collation can use simple comparison */
+
+ if (locale && locale->collate)
+ {
+ /*
+ * For non-C collations, extract the three bytes from each trigram
+ * and compare them using the collation's comparison function.
+ */
+ char str_a[3];
+ char str_b[3];
+ int result;
+
+ /* Extract bytes from the packed integer representation */
+ str_a[0] = (a >> 16) & 0xFF;
+ str_a[1] = (a >> 8) & 0xFF;
+ str_a[2] = a & 0xFF;
+
+ str_b[0] = (b >> 16) & 0xFF;
+ str_b[1] = (b >> 8) & 0xFF;
+ str_b[2] = b & 0xFF;
+
+ /* Use collation-aware comparison */
+ result = pg_strncoll(str_a, 3, str_b, 3, locale);
+ PG_RETURN_INT32(result);
+ }
+ else
+ {
+ /*
+ * For C collation or default collation, simple integer comparison
+ * is sufficient and faster.
+ */
+ if (a < b)
+ PG_RETURN_INT32(-1);
+ else if (a > b)
+ PG_RETURN_INT32(1);
+ else
+ PG_RETURN_INT32(0);
+ }
+}
+
Datum
gin_trgm_consistent(PG_FUNCTION_ARGS)
{
--
2.51.0
[text/x-patch] v1-0002-Use-correct-collation-for-finding-word-boundaries.patch (9.7K, ../../9b850976-8f0d-4957-9308-e1c053a35559@gmail.com/3-v1-0002-Use-correct-collation-for-finding-word-boundaries.patch)
download | inline diff:
From a62959930214cee04e9f190f908746067d8d1a62 Mon Sep 17 00:00:00 2001
From: David Geier <geidav.pg@gmail.com>
Date: Fri, 23 Jan 2026 15:39:06 +0100
Subject: [PATCH v1 2/3] Use correct collation for finding word boundaries
pg_trgm finds all words in the input string and creates trigrams for
them. Word characters are alpha-numeric characters. What qualifies as
alpha-numeric character depends on the collation. Previously, pg_trgm
always used the default collation. Now the specified collation is used
instead.
---
.../pg_trgm/expected/pg_trgm_collation.out | 13 ++++++++++++
contrib/pg_trgm/sql/pg_trgm_collation.sql | 5 +++++
contrib/pg_trgm/trgm.h | 6 +++---
contrib/pg_trgm/trgm_op.c | 20 +++++++++----------
contrib/pg_trgm/trgm_regexp.c | 2 +-
src/backend/tsearch/ts_locale.c | 8 ++++----
src/include/tsearch/ts_locale.h | 3 ++-
7 files changed, 38 insertions(+), 19 deletions(-)
diff --git a/contrib/pg_trgm/expected/pg_trgm_collation.out b/contrib/pg_trgm/expected/pg_trgm_collation.out
index 472ce867665..0cc53edd821 100644
--- a/contrib/pg_trgm/expected/pg_trgm_collation.out
+++ b/contrib/pg_trgm/expected/pg_trgm_collation.out
@@ -41,3 +41,16 @@ SELECT similarity('ıstanbul' COLLATE "C", 'ISTANBUL' COLLATE "C");
0.54545456
(1 row)
+-- Test that word boundary identification uses specified collation
+SELECT show_trgm('helloıtestIworldİcode' COLLATE "tr-x-icu");
+ show_trgm
+-------------------------------------------------------------------------------------------------------------------------------
+ {0x8fc0a2,0x93dfbf,0x1bf43c," h"," he",0x22d44f,0x4398ff,cod,"de ",dic,ell,est,hel,ico,ldi,llo,ode,orl,0x71b8f5,rld,tes,wor}
+(1 row)
+
+SELECT show_trgm('helloıtestIworldİcode' COLLATE "C");
+ show_trgm
+-------------------------------------------------------------------------------------------------------------
+ {" c"," h"," t"," co"," he"," te",cod,"de ",ell,est,hel,iwo,"ld ",llo,"lo ",ode,orl,rld,sti,tes,tiw,wor}
+(1 row)
+
diff --git a/contrib/pg_trgm/sql/pg_trgm_collation.sql b/contrib/pg_trgm/sql/pg_trgm_collation.sql
index afb3973a8b3..e1a5c7c5fa8 100644
--- a/contrib/pg_trgm/sql/pg_trgm_collation.sql
+++ b/contrib/pg_trgm/sql/pg_trgm_collation.sql
@@ -22,3 +22,8 @@ SELECT show_trgm('ISTANBUL' COLLATE "C");
SELECT similarity('ıstanbul' COLLATE "tr-x-icu", 'ISTANBUL' COLLATE "tr-x-icu");
SELECT similarity('ıstanbul' COLLATE "C", 'ISTANBUL' COLLATE "C");
+-- Test that word boundary identification uses specified collation
+
+SELECT show_trgm('helloıtestIworldİcode' COLLATE "tr-x-icu");
+SELECT show_trgm('helloıtestIworldİcode' COLLATE "C");
+
diff --git a/contrib/pg_trgm/trgm.h b/contrib/pg_trgm/trgm.h
index b6911e91458..3c4db129e20 100644
--- a/contrib/pg_trgm/trgm.h
+++ b/contrib/pg_trgm/trgm.h
@@ -47,9 +47,9 @@ typedef char trgm[3];
} while(0)
extern int (*CMPTRGM) (const void *a, const void *b);
-#define ISWORDCHR(c, len) (t_isalnum_with_len(c, len))
-#define ISPRINTABLECHAR(a) ( isascii( *(unsigned char*)(a) ) && (isalnum( *(unsigned char*)(a) ) || *(unsigned char*)(a)==' ') )
-#define ISPRINTABLETRGM(t) ( ISPRINTABLECHAR( ((char*)(t)) ) && ISPRINTABLECHAR( ((char*)(t))+1 ) && ISPRINTABLECHAR( ((char*)(t))+2 ) )
+#define ISWORDCHR(c, len, collation) (t_isalnum_with_len_collation(c, len, collation))
+#define ISPRINTABLECHAR(a) ( isascii( *(unsigned char*)(a) ) && (isalnum( *(unsigned char*)(a) ) || *(unsigned char*)(a)==' ') )
+#define ISPRINTABLETRGM(t) ( ISPRINTABLECHAR( ((char*)(t)) ) && ISPRINTABLECHAR( ((char*)(t))+1 ) && ISPRINTABLECHAR( ((char*)(t))+2 ) )
#define ISESCAPECHAR(x) (*(x) == '\\') /* Wildcard escape character */
#define ISWILDCARDCHAR(x) (*(x) == '_' || *(x) == '%') /* Wildcard
diff --git a/contrib/pg_trgm/trgm_op.c b/contrib/pg_trgm/trgm_op.c
index c04d7265153..dd88b1fddf9 100644
--- a/contrib/pg_trgm/trgm_op.c
+++ b/contrib/pg_trgm/trgm_op.c
@@ -335,7 +335,7 @@ show_limit(PG_FUNCTION_ARGS)
* endword points to the character after word
*/
static char *
-find_word(char *str, int lenstr, char **endword)
+find_word(char *str, int lenstr, char **endword, Oid collation)
{
char *beginword = str;
const char *endstr = str + lenstr;
@@ -344,7 +344,7 @@ find_word(char *str, int lenstr, char **endword)
{
int clen = pg_mblen_range(beginword, endstr);
- if (ISWORDCHR(beginword, clen))
+ if (ISWORDCHR(beginword, clen, collation))
break;
beginword += clen;
}
@@ -357,7 +357,7 @@ find_word(char *str, int lenstr, char **endword)
{
int clen = pg_mblen_range(*endword, endstr);
- if (!ISWORDCHR(*endword, clen))
+ if (!ISWORDCHR(*endword, clen, collation))
break;
*endword += clen;
}
@@ -533,7 +533,7 @@ generate_trgm_only(growable_trgm_array *dst, char *str, int slen, Oid collation,
}
eword = str;
- while ((bword = find_word(eword, slen - (eword - str), &eword)) != NULL)
+ while ((bword = find_word(eword, slen - (eword - str), &eword, collation)) != NULL)
{
int oldlen;
@@ -950,7 +950,7 @@ calc_word_similarity(char *str1, int slen1, char *str2, int slen2,
*/
static const char *
get_wildcard_part(const char *str, int lenstr,
- char *buf, int *bytelen)
+ char *buf, int *bytelen, Oid collation)
{
const char *beginword = str;
const char *endword;
@@ -973,7 +973,7 @@ get_wildcard_part(const char *str, int lenstr,
if (in_escape)
{
- if (ISWORDCHR(beginword, clen))
+ if (ISWORDCHR(beginword, clen, collation))
break;
in_escape = false;
in_leading_wildcard_meta = false;
@@ -984,7 +984,7 @@ get_wildcard_part(const char *str, int lenstr,
in_escape = true;
else if (ISWILDCARDCHAR(beginword))
in_leading_wildcard_meta = true;
- else if (ISWORDCHR(beginword, clen))
+ else if (ISWORDCHR(beginword, clen, collation))
break;
else
in_leading_wildcard_meta = false;
@@ -1022,7 +1022,7 @@ get_wildcard_part(const char *str, int lenstr,
clen = pg_mblen_range(endword, endstr);
if (in_escape)
{
- if (ISWORDCHR(endword, clen))
+ if (ISWORDCHR(endword, clen, collation))
{
memcpy(s, endword, clen);
s += clen;
@@ -1049,7 +1049,7 @@ get_wildcard_part(const char *str, int lenstr,
in_trailing_wildcard_meta = true;
break;
}
- else if (ISWORDCHR(endword, clen))
+ else if (ISWORDCHR(endword, clen, collation))
{
memcpy(s, endword, clen);
s += clen;
@@ -1113,7 +1113,7 @@ generate_wildcard_trgm(const char *str, int slen, Oid collation)
*/
eword = str;
while ((eword = get_wildcard_part(eword, slen - (eword - str),
- buf, &bytelen)) != NULL)
+ buf, &bytelen, collation)) != NULL)
{
char *word;
diff --git a/contrib/pg_trgm/trgm_regexp.c b/contrib/pg_trgm/trgm_regexp.c
index b4086a75e17..36a10f24b09 100644
--- a/contrib/pg_trgm/trgm_regexp.c
+++ b/contrib/pg_trgm/trgm_regexp.c
@@ -811,7 +811,7 @@ getColorInfo(regex_t *regex, TrgmNFA *trgmNFA, Oid collation)
if (!clen)
continue; /* ok to ignore it altogether */
- if (ISWORDCHR(c.bytes, clen))
+ if (ISWORDCHR(c.bytes, clen, collation))
colorInfo->wordChars[colorInfo->wordCharsCount++] = c;
else
colorInfo->containsNonWord = true;
diff --git a/src/backend/tsearch/ts_locale.c b/src/backend/tsearch/ts_locale.c
index df02ffb12fd..6f331e054a2 100644
--- a/src/backend/tsearch/ts_locale.c
+++ b/src/backend/tsearch/ts_locale.c
@@ -26,27 +26,27 @@ static void tsearch_readline_callback(void *arg);
#define GENERATE_T_ISCLASS_DEF(character_class) \
/* mblen shall be that of the first character */ \
int \
-t_is##character_class##_with_len(const char *ptr, int mblen) \
+t_is##character_class##_with_len_collation(const char *ptr, int mblen, Oid collation) \
{ \
pg_wchar wstr[WC_BUF_LEN]; \
int wlen pg_attribute_unused(); \
wlen = pg_mb2wchar_with_len(ptr, wstr, mblen); \
Assert(wlen <= 1); \
/* pass single character, or NUL if empty */ \
- return pg_isw##character_class(wstr[0], pg_database_locale()); \
+ return pg_isw##character_class(wstr[0], pg_newlocale_from_collation(collation)); \
} \
\
/* ptr shall point to a NUL-terminated string */ \
int \
t_is##character_class##_cstr(const char *ptr) \
{ \
- return t_is##character_class##_with_len(ptr, pg_mblen_cstr(ptr)); \
+ return t_is##character_class##_with_len_collation(ptr, pg_mblen_cstr(ptr), DEFAULT_COLLATION_OID); \
} \
/* ptr shall point to a string with pre-validated encoding */ \
int \
t_is##character_class##_unbounded(const char *ptr) \
{ \
- return t_is##character_class##_with_len(ptr, pg_mblen_unbounded(ptr)); \
+ return t_is##character_class##_with_len_collation(ptr, pg_mblen_unbounded(ptr), DEFAULT_COLLATION_OID); \
} \
/* historical name for _unbounded */ \
int \
diff --git a/src/include/tsearch/ts_locale.h b/src/include/tsearch/ts_locale.h
index f4edf300c2b..43a606505a8 100644
--- a/src/include/tsearch/ts_locale.h
+++ b/src/include/tsearch/ts_locale.h
@@ -18,6 +18,7 @@
#include "lib/stringinfo.h"
#include "mb/pg_wchar.h"
+#include "catalog/pg_collation.h"
#include "utils/pg_locale.h"
/* working state for tsearch_readline (should be a local var in caller) */
@@ -56,7 +57,7 @@ ts_copychar_cstr(void *dest, const void *src)
#define COPYCHAR ts_copychar_cstr
#define GENERATE_T_ISCLASS_DECL(character_class) \
-extern int t_is##character_class##_with_len(const char *ptr, int mblen); \
+extern int t_is##character_class##_with_len_collation(const char *ptr, int mblen, Oid collation); \
extern int t_is##character_class##_cstr(const char *ptr); \
extern int t_is##character_class##_unbounded(const char *ptr); \
\
--
2.51.0
[text/x-patch] v1-0001-Use-correct-collation-for-lowercasing.patch (19.4K, ../../9b850976-8f0d-4957-9308-e1c053a35559@gmail.com/4-v1-0001-Use-correct-collation-for-lowercasing.patch)
download | inline diff:
From dee7d76a76608b51bad4d98399e4d6a7484a2714 Mon Sep 17 00:00:00 2001
From: David Geier <geidav.pg@gmail.com>
Date: Wed, 21 Jan 2026 14:54:28 +0100
Subject: [PATCH v1 1/3] Use correct collation for lowercasing
pg_trgm converts the input words to lowercase before extracting the
trigrams. The lowercase conversion depends on the collation. Previously,
pg_trgm always used the default collation. Now, the specified collation
is used instead.
---
contrib/pg_trgm/Makefile | 2 +-
.../pg_trgm/expected/pg_trgm_collation.out | 43 +++++++++++++
.../pg_trgm/expected/pg_trgm_collation_1.out | 9 +++
contrib/pg_trgm/meson.build | 1 +
contrib/pg_trgm/sql/pg_trgm_collation.sql | 24 ++++++++
contrib/pg_trgm/trgm.h | 4 +-
contrib/pg_trgm/trgm_gin.c | 7 ++-
contrib/pg_trgm/trgm_gist.c | 10 ++--
contrib/pg_trgm/trgm_op.c | 60 ++++++++++---------
contrib/pg_trgm/trgm_regexp.c | 20 +++----
10 files changed, 132 insertions(+), 48 deletions(-)
create mode 100644 contrib/pg_trgm/expected/pg_trgm_collation.out
create mode 100644 contrib/pg_trgm/expected/pg_trgm_collation_1.out
create mode 100644 contrib/pg_trgm/sql/pg_trgm_collation.sql
diff --git a/contrib/pg_trgm/Makefile b/contrib/pg_trgm/Makefile
index c1756993ec7..26b3028b75e 100644
--- a/contrib/pg_trgm/Makefile
+++ b/contrib/pg_trgm/Makefile
@@ -14,7 +14,7 @@ DATA = pg_trgm--1.5--1.6.sql pg_trgm--1.4--1.5.sql pg_trgm--1.3--1.4.sql \
pg_trgm--1.0--1.1.sql
PGFILEDESC = "pg_trgm - trigram matching"
-REGRESS = pg_trgm pg_utf8_trgm pg_word_trgm pg_strict_word_trgm
+REGRESS = pg_trgm pg_utf8_trgm pg_word_trgm pg_strict_word_trgm pg_trgm_collation
ifdef USE_PGXS
PG_CONFIG = pg_config
diff --git a/contrib/pg_trgm/expected/pg_trgm_collation.out b/contrib/pg_trgm/expected/pg_trgm_collation.out
new file mode 100644
index 00000000000..472ce867665
--- /dev/null
+++ b/contrib/pg_trgm/expected/pg_trgm_collation.out
@@ -0,0 +1,43 @@
+/*
+ * This test is for ICU collations.
+ */
+/* skip test if not UTF8 server encoding or no ICU collations installed */
+SELECT getdatabaseencoding() <> 'UTF8' OR
+ (SELECT count(*) FROM pg_collation WHERE collprovider = 'i' AND collname <> 'unicode') = 0
+ AS skip_test \gset
+\if :skip_test
+\quit
+\endif
+-- Test that lowercase conversion of trigrams uses specified collation
+CREATE TABLE test(col TEXT COLLATE "tr-x-icu");
+INSERT INTO test VALUES ('ISTANBUL');
+SELECT show_trgm(col) FROM test;
+ show_trgm
+--------------------------------------------------------
+ {0xf31e1a,0xfe581d,0x3efd30,anb,bul,nbu,sta,tan,"ul "}
+(1 row)
+
+SELECT show_trgm('ISTANBUL' COLLATE "tr-x-icu");
+ show_trgm
+--------------------------------------------------------
+ {0xf31e1a,0xfe581d,0x3efd30,anb,bul,nbu,sta,tan,"ul "}
+(1 row)
+
+SELECT show_trgm('ISTANBUL' COLLATE "C");
+ show_trgm
+---------------------------------------------
+ {" i"," is",anb,bul,ist,nbu,sta,tan,"ul "}
+(1 row)
+
+SELECT similarity('ıstanbul' COLLATE "tr-x-icu", 'ISTANBUL' COLLATE "tr-x-icu");
+ similarity
+------------
+ 1
+(1 row)
+
+SELECT similarity('ıstanbul' COLLATE "C", 'ISTANBUL' COLLATE "C");
+ similarity
+------------
+ 0.54545456
+(1 row)
+
diff --git a/contrib/pg_trgm/expected/pg_trgm_collation_1.out b/contrib/pg_trgm/expected/pg_trgm_collation_1.out
new file mode 100644
index 00000000000..25c99c4abf0
--- /dev/null
+++ b/contrib/pg_trgm/expected/pg_trgm_collation_1.out
@@ -0,0 +1,9 @@
+/*
+ * This test is for ICU collations.
+ */
+/* skip test if not UTF8 server encoding or no ICU collations installed */
+SELECT getdatabaseencoding() <> 'UTF8' OR
+ (SELECT count(*) FROM pg_collation WHERE collprovider = 'i' AND collname <> 'unicode') = 0
+ AS skip_test \gset
+\if :skip_test
+\quit
diff --git a/contrib/pg_trgm/meson.build b/contrib/pg_trgm/meson.build
index 3ecf95ba862..5eafa774435 100644
--- a/contrib/pg_trgm/meson.build
+++ b/contrib/pg_trgm/meson.build
@@ -42,6 +42,7 @@ tests += {
'pg_utf8_trgm',
'pg_word_trgm',
'pg_strict_word_trgm',
+ 'pg_trgm_collation',
],
},
}
diff --git a/contrib/pg_trgm/sql/pg_trgm_collation.sql b/contrib/pg_trgm/sql/pg_trgm_collation.sql
new file mode 100644
index 00000000000..afb3973a8b3
--- /dev/null
+++ b/contrib/pg_trgm/sql/pg_trgm_collation.sql
@@ -0,0 +1,24 @@
+/*
+ * This test is for ICU collations.
+ */
+
+/* skip test if not UTF8 server encoding or no ICU collations installed */
+SELECT getdatabaseencoding() <> 'UTF8' OR
+ (SELECT count(*) FROM pg_collation WHERE collprovider = 'i' AND collname <> 'unicode') = 0
+ AS skip_test \gset
+\if :skip_test
+\quit
+\endif
+
+-- Test that lowercase conversion of trigrams uses specified collation
+
+CREATE TABLE test(col TEXT COLLATE "tr-x-icu");
+INSERT INTO test VALUES ('ISTANBUL');
+SELECT show_trgm(col) FROM test;
+SELECT show_trgm('ISTANBUL' COLLATE "tr-x-icu");
+
+SELECT show_trgm('ISTANBUL' COLLATE "C");
+
+SELECT similarity('ıstanbul' COLLATE "tr-x-icu", 'ISTANBUL' COLLATE "tr-x-icu");
+SELECT similarity('ıstanbul' COLLATE "C", 'ISTANBUL' COLLATE "C");
+
diff --git a/contrib/pg_trgm/trgm.h b/contrib/pg_trgm/trgm.h
index ca23aad4dd9..b6911e91458 100644
--- a/contrib/pg_trgm/trgm.h
+++ b/contrib/pg_trgm/trgm.h
@@ -119,8 +119,8 @@ extern double strict_word_similarity_threshold;
extern double index_strategy_get_limit(StrategyNumber strategy);
extern uint32 trgm2int(trgm *ptr);
extern void compact_trigram(trgm *tptr, char *str, int bytelen);
-extern TRGM *generate_trgm(char *str, int slen);
-extern TRGM *generate_wildcard_trgm(const char *str, int slen);
+extern TRGM *generate_trgm(char *str, int slen, Oid collation);
+extern TRGM *generate_wildcard_trgm(const char *str, int slen, Oid collation);
extern float4 cnt_sml(TRGM *trg1, TRGM *trg2, bool inexact);
extern bool trgm_contained_by(TRGM *trg1, TRGM *trg2);
extern bool *trgm_presence_map(TRGM *query, TRGM *key);
diff --git a/contrib/pg_trgm/trgm_gin.c b/contrib/pg_trgm/trgm_gin.c
index 5766b3e9955..14a892c657d 100644
--- a/contrib/pg_trgm/trgm_gin.c
+++ b/contrib/pg_trgm/trgm_gin.c
@@ -42,7 +42,7 @@ gin_extract_value_trgm(PG_FUNCTION_ARGS)
*nentries = 0;
- trg = generate_trgm(VARDATA_ANY(val), VARSIZE_ANY_EXHDR(val));
+ trg = generate_trgm(VARDATA_ANY(val), VARSIZE_ANY_EXHDR(val), PG_GET_COLLATION());
trglen = ARRNELEM(trg);
if (trglen > 0)
@@ -93,7 +93,7 @@ gin_extract_query_trgm(PG_FUNCTION_ARGS)
case WordSimilarityStrategyNumber:
case StrictWordSimilarityStrategyNumber:
case EqualStrategyNumber:
- trg = generate_trgm(VARDATA_ANY(val), VARSIZE_ANY_EXHDR(val));
+ trg = generate_trgm(VARDATA_ANY(val), VARSIZE_ANY_EXHDR(val), PG_GET_COLLATION());
break;
case ILikeStrategyNumber:
#ifndef IGNORECASE
@@ -107,7 +107,8 @@ gin_extract_query_trgm(PG_FUNCTION_ARGS)
* potentially-matching string must include.
*/
trg = generate_wildcard_trgm(VARDATA_ANY(val),
- VARSIZE_ANY_EXHDR(val));
+ VARSIZE_ANY_EXHDR(val),
+ PG_GET_COLLATION());
break;
case RegExpICaseStrategyNumber:
#ifndef IGNORECASE
diff --git a/contrib/pg_trgm/trgm_gist.c b/contrib/pg_trgm/trgm_gist.c
index 11812b2984e..d9102400442 100644
--- a/contrib/pg_trgm/trgm_gist.c
+++ b/contrib/pg_trgm/trgm_gist.c
@@ -123,7 +123,7 @@ gtrgm_compress(PG_FUNCTION_ARGS)
TRGM *res;
text *val = DatumGetTextPP(entry->key);
- res = generate_trgm(VARDATA_ANY(val), VARSIZE_ANY_EXHDR(val));
+ res = generate_trgm(VARDATA_ANY(val), VARSIZE_ANY_EXHDR(val), PG_GET_COLLATION());
retval = palloc_object(GISTENTRY);
gistentryinit(*retval, PointerGetDatum(res),
entry->rel, entry->page,
@@ -242,7 +242,8 @@ gtrgm_consistent(PG_FUNCTION_ARGS)
case StrictWordSimilarityStrategyNumber:
case EqualStrategyNumber:
qtrg = generate_trgm(VARDATA(query),
- querysize - VARHDRSZ);
+ querysize - VARHDRSZ,
+ PG_GET_COLLATION());
break;
case ILikeStrategyNumber:
#ifndef IGNORECASE
@@ -251,7 +252,8 @@ gtrgm_consistent(PG_FUNCTION_ARGS)
pg_fallthrough;
case LikeStrategyNumber:
qtrg = generate_wildcard_trgm(VARDATA(query),
- querysize - VARHDRSZ);
+ querysize - VARHDRSZ,
+ PG_GET_COLLATION());
break;
case RegExpICaseStrategyNumber:
#ifndef IGNORECASE
@@ -475,7 +477,7 @@ gtrgm_distance(PG_FUNCTION_ARGS)
{
char *newcache;
- qtrg = generate_trgm(VARDATA(query), querysize - VARHDRSZ);
+ qtrg = generate_trgm(VARDATA(query), querysize - VARHDRSZ, PG_GET_COLLATION());
newcache = MemoryContextAlloc(fcinfo->flinfo->fn_mcxt,
MAXALIGN(querysize) +
diff --git a/contrib/pg_trgm/trgm_op.c b/contrib/pg_trgm/trgm_op.c
index 22bcc3c3361..c04d7265153 100644
--- a/contrib/pg_trgm/trgm_op.c
+++ b/contrib/pg_trgm/trgm_op.c
@@ -490,7 +490,7 @@ done:
* bounds_p: where to return bounds of trigrams (if needed).
*/
static void
-generate_trgm_only(growable_trgm_array *dst, char *str, int slen, TrgmBound **bounds_p)
+generate_trgm_only(growable_trgm_array *dst, char *str, int slen, Oid collation, TrgmBound **bounds_p)
{
size_t buflen;
char *buf;
@@ -542,7 +542,7 @@ generate_trgm_only(growable_trgm_array *dst, char *str, int slen, TrgmBound **bo
{
char *lowered;
- lowered = str_tolower(bword, eword - bword, DEFAULT_COLLATION_OID);
+ lowered = str_tolower(bword, eword - bword, collation);
bytelen = strlen(lowered);
/* grow the buffer if necessary */
@@ -596,13 +596,13 @@ generate_trgm_only(growable_trgm_array *dst, char *str, int slen, TrgmBound **bo
* Returns the sorted array of unique trigrams.
*/
TRGM *
-generate_trgm(char *str, int slen)
+generate_trgm(char *str, int slen, Oid collation)
{
TRGM *trg;
growable_trgm_array arr;
int len;
- generate_trgm_only(&arr, str, slen, NULL);
+ generate_trgm_only(&arr, str, slen, collation, NULL);
len = arr.length;
trg = arr.datum;
trg->flag = ARRKEY;
@@ -857,7 +857,7 @@ iterate_word_similarity(int *trg2indexes,
*/
static float4
calc_word_similarity(char *str1, int slen1, char *str2, int slen2,
- uint8 flags)
+ uint8 flags, Oid collation)
{
bool *found;
pos_trgm *ptrg;
@@ -875,9 +875,9 @@ calc_word_similarity(char *str1, int slen1, char *str2, int slen2,
/* Make positional trigrams */
- generate_trgm_only(&trg1, str1, slen1, NULL);
+ generate_trgm_only(&trg1, str1, slen1, collation, NULL);
len1 = trg1.length;
- generate_trgm_only(&trg2, str2, slen2, (flags & WORD_SIMILARITY_STRICT) ? &bounds : NULL);
+ generate_trgm_only(&trg2, str2, slen2, collation, (flags & WORD_SIMILARITY_STRICT) ? &bounds : NULL);
len2 = trg2.length;
ptrg = make_positional_trgm(GETARR(trg1.datum), len1, GETARR(trg2.datum), len2);
@@ -1086,7 +1086,7 @@ get_wildcard_part(const char *str, int lenstr,
* " a", "bcd" would be extracted.
*/
TRGM *
-generate_wildcard_trgm(const char *str, int slen)
+generate_wildcard_trgm(const char *str, int slen, Oid collation)
{
TRGM *trg;
growable_trgm_array arr;
@@ -1118,7 +1118,7 @@ generate_wildcard_trgm(const char *str, int slen)
char *word;
#ifdef IGNORECASE
- word = str_tolower(buf, bytelen, DEFAULT_COLLATION_OID);
+ word = str_tolower(buf, bytelen, collation);
bytelen = strlen(word);
#else
word = buf;
@@ -1177,7 +1177,7 @@ show_trgm(PG_FUNCTION_ARGS)
trgm *ptr;
int i;
- trg = generate_trgm(VARDATA_ANY(in), VARSIZE_ANY_EXHDR(in));
+ trg = generate_trgm(VARDATA_ANY(in), VARSIZE_ANY_EXHDR(in), PG_GET_COLLATION());
d = palloc_array(Datum, 1 + ARRNELEM(trg));
for (i = 0, ptr = GETARR(trg); i < ARRNELEM(trg); i++, ptr++)
@@ -1344,8 +1344,8 @@ similarity(PG_FUNCTION_ARGS)
*trg2;
float4 res;
- trg1 = generate_trgm(VARDATA_ANY(in1), VARSIZE_ANY_EXHDR(in1));
- trg2 = generate_trgm(VARDATA_ANY(in2), VARSIZE_ANY_EXHDR(in2));
+ trg1 = generate_trgm(VARDATA_ANY(in1), VARSIZE_ANY_EXHDR(in1), PG_GET_COLLATION());
+ trg2 = generate_trgm(VARDATA_ANY(in2), VARSIZE_ANY_EXHDR(in2), PG_GET_COLLATION());
res = cnt_sml(trg1, trg2, false);
@@ -1366,7 +1366,7 @@ word_similarity(PG_FUNCTION_ARGS)
res = calc_word_similarity(VARDATA_ANY(in1), VARSIZE_ANY_EXHDR(in1),
VARDATA_ANY(in2), VARSIZE_ANY_EXHDR(in2),
- 0);
+ 0, PG_GET_COLLATION());
PG_FREE_IF_COPY(in1, 0);
PG_FREE_IF_COPY(in2, 1);
@@ -1382,7 +1382,7 @@ strict_word_similarity(PG_FUNCTION_ARGS)
res = calc_word_similarity(VARDATA_ANY(in1), VARSIZE_ANY_EXHDR(in1),
VARDATA_ANY(in2), VARSIZE_ANY_EXHDR(in2),
- WORD_SIMILARITY_STRICT);
+ WORD_SIMILARITY_STRICT, PG_GET_COLLATION());
PG_FREE_IF_COPY(in1, 0);
PG_FREE_IF_COPY(in2, 1);
@@ -1392,9 +1392,10 @@ strict_word_similarity(PG_FUNCTION_ARGS)
Datum
similarity_dist(PG_FUNCTION_ARGS)
{
- float4 res = DatumGetFloat4(DirectFunctionCall2(similarity,
- PG_GETARG_DATUM(0),
- PG_GETARG_DATUM(1)));
+ float4 res = DatumGetFloat4(DirectFunctionCall2Coll(similarity,
+ PG_GET_COLLATION(),
+ PG_GETARG_DATUM(0),
+ PG_GETARG_DATUM(1)));
PG_RETURN_FLOAT4(1.0 - res);
}
@@ -1402,9 +1403,10 @@ similarity_dist(PG_FUNCTION_ARGS)
Datum
similarity_op(PG_FUNCTION_ARGS)
{
- float4 res = DatumGetFloat4(DirectFunctionCall2(similarity,
- PG_GETARG_DATUM(0),
- PG_GETARG_DATUM(1)));
+ float4 res = DatumGetFloat4(DirectFunctionCall2Coll(similarity,
+ PG_GET_COLLATION(),
+ PG_GETARG_DATUM(0),
+ PG_GETARG_DATUM(1)));
PG_RETURN_BOOL(res >= similarity_threshold);
}
@@ -1418,7 +1420,7 @@ word_similarity_op(PG_FUNCTION_ARGS)
res = calc_word_similarity(VARDATA_ANY(in1), VARSIZE_ANY_EXHDR(in1),
VARDATA_ANY(in2), VARSIZE_ANY_EXHDR(in2),
- WORD_SIMILARITY_CHECK_ONLY);
+ WORD_SIMILARITY_CHECK_ONLY, PG_GET_COLLATION());
PG_FREE_IF_COPY(in1, 0);
PG_FREE_IF_COPY(in2, 1);
@@ -1434,7 +1436,7 @@ word_similarity_commutator_op(PG_FUNCTION_ARGS)
res = calc_word_similarity(VARDATA_ANY(in2), VARSIZE_ANY_EXHDR(in2),
VARDATA_ANY(in1), VARSIZE_ANY_EXHDR(in1),
- WORD_SIMILARITY_CHECK_ONLY);
+ WORD_SIMILARITY_CHECK_ONLY, PG_GET_COLLATION());
PG_FREE_IF_COPY(in1, 0);
PG_FREE_IF_COPY(in2, 1);
@@ -1450,7 +1452,7 @@ word_similarity_dist_op(PG_FUNCTION_ARGS)
res = calc_word_similarity(VARDATA_ANY(in1), VARSIZE_ANY_EXHDR(in1),
VARDATA_ANY(in2), VARSIZE_ANY_EXHDR(in2),
- 0);
+ 0, PG_GET_COLLATION());
PG_FREE_IF_COPY(in1, 0);
PG_FREE_IF_COPY(in2, 1);
@@ -1466,7 +1468,7 @@ word_similarity_dist_commutator_op(PG_FUNCTION_ARGS)
res = calc_word_similarity(VARDATA_ANY(in2), VARSIZE_ANY_EXHDR(in2),
VARDATA_ANY(in1), VARSIZE_ANY_EXHDR(in1),
- 0);
+ 0, PG_GET_COLLATION());
PG_FREE_IF_COPY(in1, 0);
PG_FREE_IF_COPY(in2, 1);
@@ -1482,7 +1484,8 @@ strict_word_similarity_op(PG_FUNCTION_ARGS)
res = calc_word_similarity(VARDATA_ANY(in1), VARSIZE_ANY_EXHDR(in1),
VARDATA_ANY(in2), VARSIZE_ANY_EXHDR(in2),
- WORD_SIMILARITY_CHECK_ONLY | WORD_SIMILARITY_STRICT);
+ WORD_SIMILARITY_CHECK_ONLY | WORD_SIMILARITY_STRICT,
+ PG_GET_COLLATION());
PG_FREE_IF_COPY(in1, 0);
PG_FREE_IF_COPY(in2, 1);
@@ -1498,7 +1501,8 @@ strict_word_similarity_commutator_op(PG_FUNCTION_ARGS)
res = calc_word_similarity(VARDATA_ANY(in2), VARSIZE_ANY_EXHDR(in2),
VARDATA_ANY(in1), VARSIZE_ANY_EXHDR(in1),
- WORD_SIMILARITY_CHECK_ONLY | WORD_SIMILARITY_STRICT);
+ WORD_SIMILARITY_CHECK_ONLY | WORD_SIMILARITY_STRICT,
+ PG_GET_COLLATION());
PG_FREE_IF_COPY(in1, 0);
PG_FREE_IF_COPY(in2, 1);
@@ -1514,7 +1518,7 @@ strict_word_similarity_dist_op(PG_FUNCTION_ARGS)
res = calc_word_similarity(VARDATA_ANY(in1), VARSIZE_ANY_EXHDR(in1),
VARDATA_ANY(in2), VARSIZE_ANY_EXHDR(in2),
- WORD_SIMILARITY_STRICT);
+ WORD_SIMILARITY_STRICT, PG_GET_COLLATION());
PG_FREE_IF_COPY(in1, 0);
PG_FREE_IF_COPY(in2, 1);
@@ -1530,7 +1534,7 @@ strict_word_similarity_dist_commutator_op(PG_FUNCTION_ARGS)
res = calc_word_similarity(VARDATA_ANY(in2), VARSIZE_ANY_EXHDR(in2),
VARDATA_ANY(in1), VARSIZE_ANY_EXHDR(in1),
- WORD_SIMILARITY_STRICT);
+ WORD_SIMILARITY_STRICT, PG_GET_COLLATION());
PG_FREE_IF_COPY(in1, 0);
PG_FREE_IF_COPY(in2, 1);
diff --git a/contrib/pg_trgm/trgm_regexp.c b/contrib/pg_trgm/trgm_regexp.c
index b4eaeec6090..b4086a75e17 100644
--- a/contrib/pg_trgm/trgm_regexp.c
+++ b/contrib/pg_trgm/trgm_regexp.c
@@ -479,11 +479,11 @@ typedef struct
/* prototypes for private functions */
static TRGM *createTrgmNFAInternal(regex_t *regex, TrgmPackedGraph **graph,
- MemoryContext rcontext);
+ MemoryContext rcontext, Oid collation);
static void RE_compile(regex_t *regex, text *text_re,
int cflags, Oid collation);
-static void getColorInfo(regex_t *regex, TrgmNFA *trgmNFA);
-static int convertPgWchar(pg_wchar c, trgm_mb_char *result);
+static void getColorInfo(regex_t *regex, TrgmNFA *trgmNFA, Oid collation);
+static int convertPgWchar(pg_wchar c, trgm_mb_char *result, Oid collation);
static void transformGraph(TrgmNFA *trgmNFA);
static void processState(TrgmNFA *trgmNFA, TrgmState *state);
static void addKey(TrgmNFA *trgmNFA, TrgmState *state, TrgmStateKey *key);
@@ -551,7 +551,7 @@ createTrgmNFA(text *text_re, Oid collation,
REG_ADVANCED | REG_NOSUB, collation);
#endif
- trg = createTrgmNFAInternal(®ex, graph, rcontext);
+ trg = createTrgmNFAInternal(®ex, graph, rcontext, collation);
/* Clean up all the cruft we created (including regex) */
MemoryContextSwitchTo(oldcontext);
@@ -565,7 +565,7 @@ createTrgmNFA(text *text_re, Oid collation,
*/
static TRGM *
createTrgmNFAInternal(regex_t *regex, TrgmPackedGraph **graph,
- MemoryContext rcontext)
+ MemoryContext rcontext, Oid collation)
{
TRGM *trg;
TrgmNFA trgmNFA;
@@ -573,7 +573,7 @@ createTrgmNFAInternal(regex_t *regex, TrgmPackedGraph **graph,
trgmNFA.regex = regex;
/* Collect color information from the regex */
- getColorInfo(regex, &trgmNFA);
+ getColorInfo(regex, &trgmNFA, collation);
#ifdef TRGM_REGEXP_DEBUG
printSourceNFA(regex, trgmNFA.colorInfo, trgmNFA.ncolors);
@@ -762,7 +762,7 @@ RE_compile(regex_t *regex, text *text_re, int cflags, Oid collation)
* Fill TrgmColorInfo structure for each color using regex export functions.
*/
static void
-getColorInfo(regex_t *regex, TrgmNFA *trgmNFA)
+getColorInfo(regex_t *regex, TrgmNFA *trgmNFA, Oid collation)
{
int colorsCount = pg_reg_getnumcolors(regex);
int i;
@@ -807,7 +807,7 @@ getColorInfo(regex_t *regex, TrgmNFA *trgmNFA)
for (j = 0; j < charsCount; j++)
{
trgm_mb_char c;
- int clen = convertPgWchar(chars[j], &c);
+ int clen = convertPgWchar(chars[j], &c, collation);
if (!clen)
continue; /* ok to ignore it altogether */
@@ -827,7 +827,7 @@ getColorInfo(regex_t *regex, TrgmNFA *trgmNFA)
* byte length.
*/
static int
-convertPgWchar(pg_wchar c, trgm_mb_char *result)
+convertPgWchar(pg_wchar c, trgm_mb_char *result, Oid collation)
{
/* "s" has enough space for a multibyte character and a trailing NUL */
char s[MAX_MULTIBYTE_CHAR_LEN + 1];
@@ -860,7 +860,7 @@ convertPgWchar(pg_wchar c, trgm_mb_char *result)
*/
#ifdef IGNORECASE
{
- char *lowerCased = str_tolower(s, clen, DEFAULT_COLLATION_OID);
+ char *lowerCased = str_tolower(s, clen, collation);
if (strcmp(lowerCased, s) != 0)
{
--
2.51.0
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: Wrong results with equality search using trigram index and non-deterministic collation
@ 2026-05-07 13:05 Laurenz Albe <laurenz.albe@cybertec.at>
parent: David Geier <geidav.pg@gmail.com>
0 siblings, 1 reply; 9+ messages in thread
From: Laurenz Albe @ 2026-05-07 13:05 UTC (permalink / raw)
To: David Geier <geidav.pg@gmail.com>; pgsql-hackers@lists.postgresql.org
On Mon, 2026-05-04 at 13:53 +0200, David Geier wrote:
> >
> Attached patch makes your case work, including the % case. It builds on
> top of the other patches from [1] that makes pg_trgm use the inferred
> collation trigram extraction.
>
> Instead of using btint4cmp() to compare trigrams, the patch uses a
> collation-aware string comparison function.
Thanks! I tried your patch, and it does indeed fix the bug I reported.
I looked at your patch, and it is pretty straightforward.
("git am" complained about an empty line at the end of
"pg_trgm--1.6--1.7.sql", but that's merely cosmetic.)
> This is just a PoC. I haven't given much thought to the details but e.g.
> when three consecutive characters exceed 3 bytes then compact_trigram()
> uses a truncated 32-bit hash value as trigram instead. Such trigrams
> won't work in all cases. We could omit them from the query string but
> for languages where the majority of trigrams are hashed or where the
> query string consists of only a few trigrams, the look-up performance
> would suffer.
Does that mean that you could end up with wrong results (which would not
be acceptable), or that you could end up with false positives that
later get eliminated by the recheck (which would be fine)?
I am worried about collations that have digraphs - the letters would be
split when trigrams are formed, and that might cause trouble.
And indeed, I am able to break it with a "quadrigraph":
CREATE COLLATION crazy (
PROVIDER = icu,
LOCALE = 'da-DK',
DETERMINISTIC = FALSE,
RULES = '& a = zzzz'
);
CREATE TABLE boom2 (id integer PRIMARY KEY, t text COLLATE crazy);
INSERT INTO boom2 VALUES (1, 'myad'), (2, 'myzzzzd');
SELECT * FROM boom2 WHERE t = 'myad';
id │ t
════╪═════════
1 │ myad
2 │ myzzzzd
(2 rows)
CREATE INDEX trgm_idx2 ON boom2 USING gin (t gin_trgm_ops);
SET enable_seqscan = off;
SELECT * FROM boom2 WHERE t = 'myad';
id │ t
════╪══════
1 │ myad
(1 row)
> I guess better would be using a collation-aware hash function that maps
> different values that compare equal to the same hash value. hashtext()
> does that already. The new comparison function would then have to
> distinguish between plain text trigrams and hash trigrams.
> Alternatively, we could store all trigrams as hashes but that would
> break functions such as show_trgm().
But that would probably not fix the above problem, right?
My initial thought about this bug was to just not consider a trigram
index if a non-deterministic collation is involved, but I can't see
how that could be done in the planner.
Still, I think that the first two patches of your set do the right thing.
Yours,
Laurenz Albe
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: Wrong results with equality search using trigram index and non-deterministic collation
@ 2026-05-07 19:00 Zsolt Parragi <zsolt.parragi@percona.com>
parent: Laurenz Albe <laurenz.albe@cybertec.at>
0 siblings, 0 replies; 9+ messages in thread
From: Zsolt Parragi @ 2026-05-07 19:00 UTC (permalink / raw)
To: Laurenz Albe <laurenz.albe@cybertec.at>; +Cc: David Geier <geidav.pg@gmail.com>; pgsql-hackers@lists.postgresql.org
Hello
> Does that mean that you could end up with wrong results (which would not
> be acceptable), or that you could end up with false positives that
> later get eliminated by the recheck (which would be fine)?
+ /*
+ * For non-C collations, extract the three bytes from each trigram
+ * and compare them using the collation's comparison function.
+ */
...
+ /* Use collation-aware comparison */
+ result = pg_strncoll(str_a, 3, str_b, 3, locale);
+ PG_RETURN_INT32(result);
For non-C collations, isn't the trigram likely a hash rather than a
proper string, where pg_strncoll won't work properly?
^ permalink raw reply [nested|flat] 9+ messages in thread
end of thread, other threads:[~2026-05-07 19:00 UTC | newest]
Thread overview: 9+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2019-04-18 06:50 [PATCH 08/10] Make pg_waldump not use callback but call the function directly Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2019-04-18 06:50 [PATCH 08/10] Make pg_waldump not use callback but call the function directly Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2019-04-18 06:50 [PATCH 08/10] Make pg_waldump not use callback but call the function directly Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2019-04-18 06:50 [PATCH 08/10] Make pg_waldump not use callback but call the function directly Kyotaro Horiguchi <horiguchi.kyotaro@lab.ntt.co.jp>
2023-02-03 05:51 [PATCH v3 1/2] Use "template" initdb in tests Andres Freund <andres@anarazel.de>
2026-01-08 16:47 [PATCH 6/6] Use multiple snapshots to copy the data. Antonin Houska <ah@cybertec.at>
2026-05-04 11:53 Re: Wrong results with equality search using trigram index and non-deterministic collation David Geier <geidav.pg@gmail.com>
2026-05-07 13:05 ` Re: Wrong results with equality search using trigram index and non-deterministic collation Laurenz Albe <laurenz.albe@cybertec.at>
2026-05-07 19:00 ` Re: Wrong results with equality search using trigram index and non-deterministic collation Zsolt Parragi <zsolt.parragi@percona.com>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox