public inbox for [email protected]help / color / mirror / Atom feed
[PATCH 4/4] Fix WAL skipping feature. 9+ messages / 5 participants [nested] [flat]
* [PATCH 4/4] Fix WAL skipping feature. @ 2018-10-11 07:18 Kyotaro Horiguchi <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Kyotaro Horiguchi @ 2018-10-11 07:18 UTC (permalink / raw) This patch repalces WAL-skipping means from HEAP_INSERT_SKIP_WAL to pending-sync tracking infrastructure. --- src/backend/access/heap/heapam.c | 70 ++++++++++++++++++++++----------- src/backend/access/heap/pruneheap.c | 3 +- src/backend/access/heap/rewriteheap.c | 3 -- src/backend/access/heap/vacuumlazy.c | 6 +-- src/backend/access/heap/visibilitymap.c | 3 +- src/backend/commands/copy.c | 13 +++--- src/backend/commands/createas.c | 9 ++--- src/backend/commands/matview.c | 6 +-- src/backend/commands/tablecmds.c | 5 +-- src/include/access/heapam.h | 9 ++--- 10 files changed, 72 insertions(+), 55 deletions(-) diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index 5ea5ff5848..c66a468335 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -28,6 +28,28 @@ * the POSTGRES heap access method used for all POSTGRES * relations. * + * WAL CONSIDERATIONS + * All heap operations are normally WAL-logged. but there are a few + * exceptions. Temporary and unlogged relations never need to be + * WAL-logged, but we can also skip WAL-logging for a table that was + * created in the same transaction, if we don't need WAL for PITR or + * WAL archival purposes (i.e. if wal_level=minimal), and we fsync() + * the file to disk at COMMIT instead. + * + * The same-relation optimization is not employed automatically on all + * updates to a table that was created in the same transacton, because + * for a small number of changes, it's cheaper to just create the WAL + * records than fsyncing() the whole relation at COMMIT. It is only + * worthwhile for (presumably) large operations like COPY, CLUSTER, + * or VACUUM FULL. Use heap_register_sync() to initiate such an + * operation; it will cause any subsequent updates to the table to skip + * WAL-logging, if possible, and cause the heap to be synced to disk at + * COMMIT. + * + * To make that work, all modifications to heap must use + * BufferNeedsWAL() to check if WAL-logging is needed in this transaction + * for the given block. + * *------------------------------------------------------------------------- */ #include "postgres.h" @@ -2127,12 +2149,6 @@ ReleaseBulkInsertStatePin(BulkInsertState bistate) * The new tuple is stamped with current transaction ID and the specified * command ID. * - * If the HEAP_INSERT_SKIP_WAL option is specified, the new tuple is not - * logged in WAL, even for a non-temp relation. Safe usage of this behavior - * requires that we arrange that all new tuples go into new pages not - * containing any tuples from other transactions, and that the relation gets - * fsync'd before commit. (See also heap_sync() comments) - * * The HEAP_INSERT_SKIP_FSM option is passed directly to * RelationGetBufferForTuple, which see for more info. * @@ -2239,7 +2255,7 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid, MarkBufferDirty(buffer); /* XLOG stuff */ - if (!(options & HEAP_INSERT_SKIP_WAL) && RelationNeedsWAL(relation)) + if (BufferNeedsWAL(relation, buffer)) { xl_heap_insert xlrec; xl_heap_header xlhdr; @@ -2414,7 +2430,6 @@ heap_multi_insert(Relation relation, HeapTuple *tuples, int ntuples, int ndone; PGAlignedBlock scratch; Page page; - bool needwal; Size saveFreeSpace; bool need_tuple_data = RelationIsLogicallyLogged(relation); bool need_cids = RelationIsAccessibleInLogicalDecoding(relation); @@ -2422,7 +2437,6 @@ heap_multi_insert(Relation relation, HeapTuple *tuples, int ntuples, /* currently not needed (thus unsupported) for heap_multi_insert() */ AssertArg(!(options & HEAP_INSERT_NO_LOGICAL)); - needwal = !(options & HEAP_INSERT_SKIP_WAL) && RelationNeedsWAL(relation); saveFreeSpace = RelationGetTargetPageFreeSpace(relation, HEAP_DEFAULT_FILLFACTOR); @@ -2464,6 +2478,7 @@ heap_multi_insert(Relation relation, HeapTuple *tuples, int ntuples, Buffer vmbuffer = InvalidBuffer; bool all_visible_cleared = false; int nthispage; + bool needwal; CHECK_FOR_INTERRUPTS(); @@ -2475,6 +2490,7 @@ heap_multi_insert(Relation relation, HeapTuple *tuples, int ntuples, InvalidBuffer, options, bistate, &vmbuffer, NULL); page = BufferGetPage(buffer); + needwal = BufferNeedsWAL(relation, buffer); /* NO EREPORT(ERROR) from here till changes are logged */ START_CRIT_SECTION(); @@ -3037,7 +3053,7 @@ l1: * NB: heap_abort_speculative() uses the same xlog record and replay * routines. */ - if (RelationNeedsWAL(relation)) + if (BufferNeedsWAL(relation, buffer)) { xl_heap_delete xlrec; xl_heap_header xlhdr; @@ -3776,7 +3792,7 @@ l2: MarkBufferDirty(buffer); - if (RelationNeedsWAL(relation)) + if (BufferNeedsWAL(relation, buffer)) { xl_heap_lock xlrec; XLogRecPtr recptr; @@ -3991,7 +4007,8 @@ l2: MarkBufferDirty(buffer); /* XLOG stuff */ - if (RelationNeedsWAL(relation)) + if (BufferNeedsWAL(relation, buffer) || + BufferNeedsWAL(relation, newbuf)) { XLogRecPtr recptr; @@ -4881,7 +4898,7 @@ failed: * (Also, in a PITR log-shipping or 2PC environment, we have to have XLOG * entries for everything anyway.) */ - if (RelationNeedsWAL(relation)) + if (BufferNeedsWAL(relation, *buffer)) { xl_heap_lock xlrec; XLogRecPtr recptr; @@ -5625,7 +5642,7 @@ l4: MarkBufferDirty(buf); /* XLOG stuff */ - if (RelationNeedsWAL(rel)) + if (BufferNeedsWAL(rel, buf)) { xl_heap_lock_updated xlrec; XLogRecPtr recptr; @@ -5785,7 +5802,7 @@ heap_finish_speculative(Relation relation, HeapTuple tuple) htup->t_ctid = tuple->t_self; /* XLOG stuff */ - if (RelationNeedsWAL(relation)) + if (BufferNeedsWAL(relation, buffer)) { xl_heap_confirm xlrec; XLogRecPtr recptr; @@ -5918,7 +5935,7 @@ heap_abort_speculative(Relation relation, HeapTuple tuple) * The WAL records generated here match heap_delete(). The same recovery * routines are used. */ - if (RelationNeedsWAL(relation)) + if (BufferNeedsWAL(relation, buffer)) { xl_heap_delete xlrec; XLogRecPtr recptr; @@ -6027,7 +6044,7 @@ heap_inplace_update(Relation relation, HeapTuple tuple) MarkBufferDirty(buffer); /* XLOG stuff */ - if (RelationNeedsWAL(relation)) + if (BufferNeedsWAL(relation, buffer)) { xl_heap_inplace xlrec; XLogRecPtr recptr; @@ -7224,7 +7241,7 @@ log_heap_clean(Relation reln, Buffer buffer, XLogRecPtr recptr; /* Caller should not call me on a non-WAL-logged relation */ - Assert(RelationNeedsWAL(reln)); + Assert(BufferNeedsWAL(reln, buffer)); xlrec.latestRemovedXid = latestRemovedXid; xlrec.nredirected = nredirected; @@ -7272,7 +7289,7 @@ log_heap_freeze(Relation reln, Buffer buffer, TransactionId cutoff_xid, XLogRecPtr recptr; /* Caller should not call me on a non-WAL-logged relation */ - Assert(RelationNeedsWAL(reln)); + Assert(BufferNeedsWAL(reln, buffer)); /* nor when there are no tuples to freeze */ Assert(ntuples > 0); @@ -7357,7 +7374,7 @@ log_heap_update(Relation reln, Buffer oldbuf, int bufflags; /* Caller should not call me on a non-WAL-logged relation */ - Assert(RelationNeedsWAL(reln)); + Assert(BufferNeedsWAL(reln, newbuf) || BufferNeedsWAL(reln, oldbuf)); XLogBeginInsert(); @@ -8961,9 +8978,16 @@ heap2_redo(XLogReaderState *record) * heap_sync - sync a heap, for use when no WAL has been written * * This forces the heap contents (including TOAST heap if any) down to disk. - * If we skipped using WAL, and WAL is otherwise needed, we must force the - * relation down to disk before it's safe to commit the transaction. This - * requires writing out any dirty buffers and then doing a forced fsync. + * If we did any changes to the heap bypassing the buffer manager, we must + * force the relation down to disk before it's safe to commit the + * transaction, because the direct modifications will not be flushed by + * the next checkpoint. + * + * We used to also use this after batch operations like COPY and CLUSTER, + * if we skipped using WAL and WAL is otherwise needed, but there were + * corner-cases involving other WAL-logged operations to the same + * relation, where that was not enough. heap_register_sync() should be + * used for that purpose instead. * * Indexes are not touched. (Currently, index operations associated with * the commands that use this are WAL-logged and so do not need fsync. diff --git a/src/backend/access/heap/pruneheap.c b/src/backend/access/heap/pruneheap.c index a3e51922d8..a05659b168 100644 --- a/src/backend/access/heap/pruneheap.c +++ b/src/backend/access/heap/pruneheap.c @@ -20,6 +20,7 @@ #include "access/htup_details.h" #include "access/xlog.h" #include "catalog/catalog.h" +#include "catalog/storage.h" #include "miscadmin.h" #include "pgstat.h" #include "storage/bufmgr.h" @@ -258,7 +259,7 @@ heap_page_prune(Relation relation, Buffer buffer, TransactionId OldestXmin, /* * Emit a WAL HEAP_CLEAN record showing what we did */ - if (RelationNeedsWAL(relation)) + if (BufferNeedsWAL(relation, buffer)) { XLogRecPtr recptr; diff --git a/src/backend/access/heap/rewriteheap.c b/src/backend/access/heap/rewriteheap.c index f5cf9ffc9c..1e9c07c9b2 100644 --- a/src/backend/access/heap/rewriteheap.c +++ b/src/backend/access/heap/rewriteheap.c @@ -654,9 +654,6 @@ raw_heap_insert(RewriteState state, HeapTuple tup) { int options = HEAP_INSERT_SKIP_FSM; - if (!state->rs_use_wal) - options |= HEAP_INSERT_SKIP_WAL; - /* * While rewriting the heap for VACUUM FULL / CLUSTER, make sure data * for the TOAST table are not logically decoded. The main heap is diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c index 9416c31889..1f66685c88 100644 --- a/src/backend/access/heap/vacuumlazy.c +++ b/src/backend/access/heap/vacuumlazy.c @@ -929,7 +929,7 @@ lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats, * page has been previously WAL-logged, and if not, do that * now. */ - if (RelationNeedsWAL(onerel) && + if (BufferNeedsWAL(onerel, buf) && PageGetLSN(page) == InvalidXLogRecPtr) log_newpage_buffer(buf, true); @@ -1193,7 +1193,7 @@ lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats, } /* Now WAL-log freezing if necessary */ - if (RelationNeedsWAL(onerel)) + if (BufferNeedsWAL(onerel, buf)) { XLogRecPtr recptr; @@ -1575,7 +1575,7 @@ lazy_vacuum_page(Relation onerel, BlockNumber blkno, Buffer buffer, MarkBufferDirty(buffer); /* XLOG stuff */ - if (RelationNeedsWAL(onerel)) + if (BufferNeedsWAL(onerel, buffer)) { XLogRecPtr recptr; diff --git a/src/backend/access/heap/visibilitymap.c b/src/backend/access/heap/visibilitymap.c index 64dfe06b26..1f5f7d92dd 100644 --- a/src/backend/access/heap/visibilitymap.c +++ b/src/backend/access/heap/visibilitymap.c @@ -88,6 +88,7 @@ #include "access/heapam_xlog.h" #include "access/visibilitymap.h" #include "access/xlog.h" +#include "catalog/storage.h" #include "miscadmin.h" #include "port/pg_bitutils.h" #include "storage/bufmgr.h" @@ -276,7 +277,7 @@ visibilitymap_set(Relation rel, BlockNumber heapBlk, Buffer heapBuf, map[mapByte] |= (flags << mapOffset); MarkBufferDirty(vmBuf); - if (RelationNeedsWAL(rel)) + if (BufferNeedsWAL(rel, heapBuf)) { if (XLogRecPtrIsInvalid(recptr)) { diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c index 5dd6fe02c6..db7a94ff6e 100644 --- a/src/backend/commands/copy.c +++ b/src/backend/commands/copy.c @@ -2390,8 +2390,7 @@ CopyFrom(CopyState cstate) * - data is being written to relfilenode created in this transaction * then we can skip writing WAL. It's safe because if the transaction * doesn't commit, we'll discard the table (or the new relfilenode file). - * If it does commit, we'll have done the heap_sync at the bottom of this - * routine first. + * If it does commit, commit will do heap_sync(). * * As mentioned in comments in utils/rel.h, the in-same-transaction test * is not always set correctly, since in rare cases rd_newRelfilenodeSubid @@ -2437,7 +2436,7 @@ CopyFrom(CopyState cstate) { hi_options |= HEAP_INSERT_SKIP_FSM; if (!XLogIsNeeded()) - hi_options |= HEAP_INSERT_SKIP_WAL; + heap_register_sync(cstate->rel); } /* @@ -3087,11 +3086,11 @@ CopyFrom(CopyState cstate) FreeExecutorState(estate); /* - * If we skipped writing WAL, then we need to sync the heap (but not - * indexes since those use WAL anyway) + * If we skipped writing WAL, then we will sync the heap at the end of + * the transaction. (We used to do it here, but it was later found out + * that to be safe, we must also avoid WAL-logging any subsequent + * actions on the pages we skipped WAL for). Indexes always use WAL. */ - if (hi_options & HEAP_INSERT_SKIP_WAL) - heap_sync(cstate->rel); return processed; } diff --git a/src/backend/commands/createas.c b/src/backend/commands/createas.c index 6517ecb738..17fb78ba78 100644 --- a/src/backend/commands/createas.c +++ b/src/backend/commands/createas.c @@ -556,8 +556,9 @@ intorel_startup(DestReceiver *self, int operation, TupleDesc typeinfo) * We can skip WAL-logging the insertions, unless PITR or streaming * replication is in use. We can skip the FSM in any case. */ - myState->hi_options = HEAP_INSERT_SKIP_FSM | - (XLogIsNeeded() ? 0 : HEAP_INSERT_SKIP_WAL); + if (!XLogIsNeeded()) + heap_register_sync(intoRelationDesc); + myState->hi_options = HEAP_INSERT_SKIP_FSM; myState->bistate = GetBulkInsertState(); /* Not using WAL requires smgr_targblock be initially invalid */ @@ -603,9 +604,7 @@ intorel_shutdown(DestReceiver *self) FreeBulkInsertState(myState->bistate); - /* If we skipped using WAL, must heap_sync before commit */ - if (myState->hi_options & HEAP_INSERT_SKIP_WAL) - heap_sync(myState->rel); + /* If we skipped using WAL, we will sync the relation at commit */ /* close rel, but keep lock until commit */ table_close(myState->rel, NoLock); diff --git a/src/backend/commands/matview.c b/src/backend/commands/matview.c index 5a47be4b33..5f447c6d94 100644 --- a/src/backend/commands/matview.c +++ b/src/backend/commands/matview.c @@ -463,7 +463,7 @@ transientrel_startup(DestReceiver *self, int operation, TupleDesc typeinfo) */ myState->hi_options = HEAP_INSERT_SKIP_FSM | HEAP_INSERT_FROZEN; if (!XLogIsNeeded()) - myState->hi_options |= HEAP_INSERT_SKIP_WAL; + heap_register_sync(transientrel); myState->bistate = GetBulkInsertState(); /* Not using WAL requires smgr_targblock be initially invalid */ @@ -509,9 +509,7 @@ transientrel_shutdown(DestReceiver *self) FreeBulkInsertState(myState->bistate); - /* If we skipped using WAL, must heap_sync before commit */ - if (myState->hi_options & HEAP_INSERT_SKIP_WAL) - heap_sync(myState->transientrel); + /* If we skipped using WAL, we will sync the relation at commit */ /* close transientrel, but keep lock until commit */ table_close(myState->transientrel, NoLock); diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index 6190b3f605..94d7876b8c 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -4617,8 +4617,9 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode) bistate = GetBulkInsertState(); hi_options = HEAP_INSERT_SKIP_FSM; + if (!XLogIsNeeded()) - hi_options |= HEAP_INSERT_SKIP_WAL; + heap_register_sync(newrel); } else { @@ -4886,8 +4887,6 @@ ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode) FreeBulkInsertState(bistate); /* If we skipped writing WAL, then we need to sync the heap. */ - if (hi_options & HEAP_INSERT_SKIP_WAL) - heap_sync(newrel); table_close(newrel, NoLock); } diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h index fab5052868..32a365021a 100644 --- a/src/include/access/heapam.h +++ b/src/include/access/heapam.h @@ -27,11 +27,10 @@ /* "options" flag bits for heap_insert */ -#define HEAP_INSERT_SKIP_WAL 0x0001 -#define HEAP_INSERT_SKIP_FSM 0x0002 -#define HEAP_INSERT_FROZEN 0x0004 -#define HEAP_INSERT_SPECULATIVE 0x0008 -#define HEAP_INSERT_NO_LOGICAL 0x0010 +#define HEAP_INSERT_SKIP_FSM 0x0001 +#define HEAP_INSERT_FROZEN 0x0002 +#define HEAP_INSERT_SPECULATIVE 0x0004 +#define HEAP_INSERT_NO_LOGICAL 0x0008 typedef struct BulkInsertStateData *BulkInsertState; -- 2.16.3 ----Next_Part(Mon_Mar_04_12_24_48_2019_788)---- ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: PL/Python: Fix return in the middle of PG_TRY() block. @ 2023-05-03 20:33 Tom Lane <[email protected]> 0 siblings, 1 reply; 9+ messages in thread From: Tom Lane @ 2023-05-03 20:33 UTC (permalink / raw) To: Nathan Bossart <[email protected]>; +Cc: Andres Freund <[email protected]>; Xing Guo <[email protected]>; pgsql-hackers Nathan Bossart <[email protected]> writes: > Here's a new version of the patch. Besides adding comments and a commit > message, I made sure to decrement the reference count for pltargs in the > PG_CATCH block (which means that pltargs likely needs to be volatile). Hmm, actually I think these changes should allow you to *remove* some volatile markers. IIUC, you need volatile for variables that are declared outside PG_TRY but modified within it. That is the case for these pointers as the code stands, but your patch is changing them to the non-risky case where they are assigned once before entering PG_TRY. (My mental model of this is that without "volatile", the compiler may keep the variable in a register, creating the hazard that longjmp will revert the variable's value to what it was at setjmp time thanks to the register save/restore that those functions do. But if it hasn't changed value since entering PG_TRY, then that doesn't matter.) > I'm > not too wild about moving the chunk of code for pltargs like this, but I > haven't thought of a better option. We could error instead of returning > NULL, but IIUC that would go against d0aa965's stated purpose. Agreed, throwing an error in these situations doesn't improve matters. regards, tom lane ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: PL/Python: Fix return in the middle of PG_TRY() block. @ 2023-05-03 20:58 Nathan Bossart <[email protected]> parent: Tom Lane <[email protected]> 0 siblings, 1 reply; 9+ messages in thread From: Nathan Bossart @ 2023-05-03 20:58 UTC (permalink / raw) To: Tom Lane <[email protected]>; +Cc: Andres Freund <[email protected]>; Xing Guo <[email protected]>; pgsql-hackers On Wed, May 03, 2023 at 04:33:32PM -0400, Tom Lane wrote: > Nathan Bossart <[email protected]> writes: >> Here's a new version of the patch. Besides adding comments and a commit >> message, I made sure to decrement the reference count for pltargs in the >> PG_CATCH block (which means that pltargs likely needs to be volatile). > > Hmm, actually I think these changes should allow you to *remove* some > volatile markers. IIUC, you need volatile for variables that are declared > outside PG_TRY but modified within it. That is the case for these > pointers as the code stands, but your patch is changing them to the > non-risky case where they are assigned once before entering PG_TRY. > > (My mental model of this is that without "volatile", the compiler > may keep the variable in a register, creating the hazard that longjmp > will revert the variable's value to what it was at setjmp time thanks > to the register save/restore that those functions do. But if it hasn't > changed value since entering PG_TRY, then that doesn't matter.) Ah, I think you are right. elog.h states as follows: * Note: if a local variable of the function containing PG_TRY is modified * in the PG_TRY section and used in the PG_CATCH section, that variable * must be declared "volatile" for POSIX compliance. This is not mere * pedantry; we have seen bugs from compilers improperly optimizing code * away when such a variable was not marked. Beware that gcc's -Wclobbered * warnings are just about entirely useless for catching such oversights. With this change, pltdata isn't modified in the PG_TRY section, and the only modification of pltargs happens after all elogs. It might be worth keeping pltargs volatile in case someone decides to add an elog() in the future, but I see no need to keep it for pltdata. -- Nathan Bossart Amazon Web Services: https://aws.amazon.com ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: PL/Python: Fix return in the middle of PG_TRY() block. @ 2023-05-04 04:54 Nathan Bossart <[email protected]> parent: Nathan Bossart <[email protected]> 0 siblings, 1 reply; 9+ messages in thread From: Nathan Bossart @ 2023-05-04 04:54 UTC (permalink / raw) To: Tom Lane <[email protected]>; +Cc: Andres Freund <[email protected]>; Xing Guo <[email protected]>; pgsql-hackers On Wed, May 03, 2023 at 01:58:38PM -0700, Nathan Bossart wrote: > With this change, pltdata isn't modified in the PG_TRY section, and the > only modification of pltargs happens after all elogs. It might be worth > keeping pltargs volatile in case someone decides to add an elog() in the > future, but I see no need to keep it for pltdata. Here's a new patch that removes the volatile marker from pltdata. -- Nathan Bossart Amazon Web Services: https://aws.amazon.com Attachments: [text/x-diff] v4-0001-Fix-improper-returns-in-PG_TRY-blocks.patch (3.5K, ../../20230504045413.GA2202617@nathanxps13/2-v4-0001-Fix-improper-returns-in-PG_TRY-blocks.patch) download | inline diff: From 53d2d942b8ee553439da6f324186a62ebb7fce1f Mon Sep 17 00:00:00 2001 From: Nathan Bossart <[email protected]> Date: Wed, 3 May 2023 11:32:43 -0700 Subject: [PATCH v4 1/1] Fix improper returns in PG_TRY blocks. If we exit a PG_TRY block early via "continue", "break", "goto", or "return", we'll skip unwinding its exception stack. This change moves a couple of such "return" statements in PL/Python out of PG_TRY blocks. This was introduced in d0aa965c0a and affects all supported versions. We might also be able to add compile-time checks to avoid recurrence, but that is left as a future exercise. Reported-by: Mikhail Gribkov, Xing Guo Author: Xing Guo Reviewed-by: Michael Paquier, Andres Freund, Tom Lane Discussion: https://postgr.es/m/CAMEv5_v5Y%2B-D%3DCO1%2Bqoe16sAmgC4sbbQjz%2BUtcHmB6zcgS%2B5Ew%40mail.gmail.com Discussion: https://postgr.es/m/CACpMh%2BCMsGMRKFzFMm3bYTzQmMU5nfEEoEDU2apJcc4hid36AQ%40mail.gmail.com Backpatch-through: 11 (all supported versions) --- src/pl/plpython/plpy_exec.c | 52 +++++++++++++++++++++++++------------ 1 file changed, 35 insertions(+), 17 deletions(-) diff --git a/src/pl/plpython/plpy_exec.c b/src/pl/plpython/plpy_exec.c index 923703535a..1e3efd4d86 100644 --- a/src/pl/plpython/plpy_exec.c +++ b/src/pl/plpython/plpy_exec.c @@ -414,12 +414,17 @@ PLy_function_build_args(FunctionCallInfo fcinfo, PLyProcedure *proc) PyObject *volatile args = NULL; int i; + /* + * Make any Py*_New() calls before the PG_TRY block so that we can quickly + * return NULL on failure. We can't return within the PG_TRY block, else + * we'd miss unwinding the exception stack. + */ + args = PyList_New(proc->nargs); + if (!args) + return NULL; + PG_TRY(); { - args = PyList_New(proc->nargs); - if (!args) - return NULL; - for (i = 0; i < proc->nargs; i++) { PLyDatumToOb *arginfo = &proc->args[i]; @@ -683,19 +688,34 @@ PLy_trigger_build_args(FunctionCallInfo fcinfo, PLyProcedure *proc, HeapTuple *r *pltlevel, *pltrelid, *plttablename, - *plttableschema; - PyObject *pltargs, + *plttableschema, *pytnew, - *pytold; - PyObject *volatile pltdata = NULL; + *pytold, + *pltdata; + PyObject *volatile pltargs = NULL; char *stroid; - PG_TRY(); + /* + * Make any Py*_New() calls before the PG_TRY block so that we can quickly + * return NULL on failure. We can't return within the PG_TRY block, else + * we'd miss unwinding the exception stack. + */ + pltdata = PyDict_New(); + if (!pltdata) + return NULL; + + if (tdata->tg_trigger->tgnargs) { - pltdata = PyDict_New(); - if (!pltdata) + pltargs = PyList_New(tdata->tg_trigger->tgnargs); + if (!pltargs) + { + Py_DECREF(pltdata); return NULL; + } + } + PG_TRY(); + { pltname = PLyUnicode_FromString(tdata->tg_trigger->tgname); PyDict_SetItemString(pltdata, "name", pltname); Py_DECREF(pltname); @@ -835,12 +855,9 @@ PLy_trigger_build_args(FunctionCallInfo fcinfo, PLyProcedure *proc, HeapTuple *r int i; PyObject *pltarg; - pltargs = PyList_New(tdata->tg_trigger->tgnargs); - if (!pltargs) - { - Py_DECREF(pltdata); - return NULL; - } + /* pltargs should have been allocated before the PG_TRY block. */ + Assert(pltargs); + for (i = 0; i < tdata->tg_trigger->tgnargs; i++) { pltarg = PLyUnicode_FromString(tdata->tg_trigger->tgargs[i]); @@ -861,6 +878,7 @@ PLy_trigger_build_args(FunctionCallInfo fcinfo, PLyProcedure *proc, HeapTuple *r } PG_CATCH(); { + Py_XDECREF(pltargs); Py_XDECREF(pltdata); PG_RE_THROW(); } -- 2.25.1 ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: PL/Python: Fix return in the middle of PG_TRY() block. @ 2023-05-04 05:01 Nathan Bossart <[email protected]> parent: Nathan Bossart <[email protected]> 0 siblings, 1 reply; 9+ messages in thread From: Nathan Bossart @ 2023-05-04 05:01 UTC (permalink / raw) To: Tom Lane <[email protected]>; +Cc: Andres Freund <[email protected]>; Xing Guo <[email protected]>; pgsql-hackers On Wed, May 03, 2023 at 09:54:13PM -0700, Nathan Bossart wrote: > Here's a new patch that removes the volatile marker from pltdata. Gah, right after I sent that, I realized we can remove one more volatile marker. Sorry for the noise. -- Nathan Bossart Amazon Web Services: https://aws.amazon.com Attachments: [text/x-diff] v5-0001-Fix-improper-returns-in-PG_TRY-blocks.patch (3.6K, ../../20230504050108.GA2203557@nathanxps13/2-v5-0001-Fix-improper-returns-in-PG_TRY-blocks.patch) download | inline diff: From 24a4e91b3cdd66b985fab70f3aed8f6a202b31fe Mon Sep 17 00:00:00 2001 From: Nathan Bossart <[email protected]> Date: Wed, 3 May 2023 11:32:43 -0700 Subject: [PATCH v5 1/1] Fix improper returns in PG_TRY blocks. If we exit a PG_TRY block early via "continue", "break", "goto", or "return", we'll skip unwinding its exception stack. This change moves a couple of such "return" statements in PL/Python out of PG_TRY blocks. This was introduced in d0aa965c0a and affects all supported versions. We might also be able to add compile-time checks to avoid recurrence, but that is left as a future exercise. Reported-by: Mikhail Gribkov, Xing Guo Author: Xing Guo Reviewed-by: Michael Paquier, Andres Freund, Tom Lane Discussion: https://postgr.es/m/CAMEv5_v5Y%2B-D%3DCO1%2Bqoe16sAmgC4sbbQjz%2BUtcHmB6zcgS%2B5Ew%40mail.gmail.com Discussion: https://postgr.es/m/CACpMh%2BCMsGMRKFzFMm3bYTzQmMU5nfEEoEDU2apJcc4hid36AQ%40mail.gmail.com Backpatch-through: 11 (all supported versions) --- src/pl/plpython/plpy_exec.c | 54 ++++++++++++++++++++++++------------- 1 file changed, 36 insertions(+), 18 deletions(-) diff --git a/src/pl/plpython/plpy_exec.c b/src/pl/plpython/plpy_exec.c index 923703535a..fcb363891a 100644 --- a/src/pl/plpython/plpy_exec.c +++ b/src/pl/plpython/plpy_exec.c @@ -411,15 +411,20 @@ static PyObject * PLy_function_build_args(FunctionCallInfo fcinfo, PLyProcedure *proc) { PyObject *volatile arg = NULL; - PyObject *volatile args = NULL; + PyObject *args = NULL; int i; + /* + * Make any Py*_New() calls before the PG_TRY block so that we can quickly + * return NULL on failure. We can't return within the PG_TRY block, else + * we'd miss unwinding the exception stack. + */ + args = PyList_New(proc->nargs); + if (!args) + return NULL; + PG_TRY(); { - args = PyList_New(proc->nargs); - if (!args) - return NULL; - for (i = 0; i < proc->nargs; i++) { PLyDatumToOb *arginfo = &proc->args[i]; @@ -683,19 +688,34 @@ PLy_trigger_build_args(FunctionCallInfo fcinfo, PLyProcedure *proc, HeapTuple *r *pltlevel, *pltrelid, *plttablename, - *plttableschema; - PyObject *pltargs, + *plttableschema, *pytnew, - *pytold; - PyObject *volatile pltdata = NULL; + *pytold, + *pltdata; + PyObject *volatile pltargs = NULL; char *stroid; - PG_TRY(); + /* + * Make any Py*_New() calls before the PG_TRY block so that we can quickly + * return NULL on failure. We can't return within the PG_TRY block, else + * we'd miss unwinding the exception stack. + */ + pltdata = PyDict_New(); + if (!pltdata) + return NULL; + + if (tdata->tg_trigger->tgnargs) { - pltdata = PyDict_New(); - if (!pltdata) + pltargs = PyList_New(tdata->tg_trigger->tgnargs); + if (!pltargs) + { + Py_DECREF(pltdata); return NULL; + } + } + PG_TRY(); + { pltname = PLyUnicode_FromString(tdata->tg_trigger->tgname); PyDict_SetItemString(pltdata, "name", pltname); Py_DECREF(pltname); @@ -835,12 +855,9 @@ PLy_trigger_build_args(FunctionCallInfo fcinfo, PLyProcedure *proc, HeapTuple *r int i; PyObject *pltarg; - pltargs = PyList_New(tdata->tg_trigger->tgnargs); - if (!pltargs) - { - Py_DECREF(pltdata); - return NULL; - } + /* pltargs should have been allocated before the PG_TRY block. */ + Assert(pltargs); + for (i = 0; i < tdata->tg_trigger->tgnargs; i++) { pltarg = PLyUnicode_FromString(tdata->tg_trigger->tgargs[i]); @@ -861,6 +878,7 @@ PLy_trigger_build_args(FunctionCallInfo fcinfo, PLyProcedure *proc, HeapTuple *r } PG_CATCH(); { + Py_XDECREF(pltargs); Py_XDECREF(pltdata); PG_RE_THROW(); } -- 2.25.1 ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: PL/Python: Fix return in the middle of PG_TRY() block. @ 2023-05-04 12:39 Tom Lane <[email protected]> parent: Nathan Bossart <[email protected]> 0 siblings, 1 reply; 9+ messages in thread From: Tom Lane @ 2023-05-04 12:39 UTC (permalink / raw) To: Nathan Bossart <[email protected]>; +Cc: Andres Freund <[email protected]>; Xing Guo <[email protected]>; pgsql-hackers Nathan Bossart <[email protected]> writes: > Gah, right after I sent that, I realized we can remove one more volatile > marker. Sorry for the noise. Hmm, I'm not sure why PLy_trigger_build_args's pltargs needs to gain a "volatile" here? LGTM otherwise. regards, tom lane ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: PL/Python: Fix return in the middle of PG_TRY() block. @ 2023-05-04 23:42 Nathan Bossart <[email protected]> parent: Tom Lane <[email protected]> 0 siblings, 1 reply; 9+ messages in thread From: Nathan Bossart @ 2023-05-04 23:42 UTC (permalink / raw) To: Tom Lane <[email protected]>; +Cc: Andres Freund <[email protected]>; Xing Guo <[email protected]>; pgsql-hackers On Thu, May 04, 2023 at 08:39:03AM -0400, Tom Lane wrote: > Hmm, I'm not sure why PLy_trigger_build_args's pltargs needs to > gain a "volatile" here? LGTM otherwise. I removed that new "volatile" marker before committing. I was trying to future-proof a bit and follow elog.h's recommendation to the letter, but following your mental model upthread, it doesn't seem to be strictly necessary, and we'd need to set pltargs to NULL after decrementing its reference count in the PG_TRY section for such future-proofing to be effective, anyway. Thank you for reviewing! -- Nathan Bossart Amazon Web Services: https://aws.amazon.com ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: PL/Python: Fix return in the middle of PG_TRY() block. @ 2023-05-05 01:40 Xing Guo <[email protected]> parent: Nathan Bossart <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Xing Guo @ 2023-05-05 01:40 UTC (permalink / raw) To: Nathan Bossart <[email protected]>; +Cc: pgsql-hackers; Andres Freund <[email protected]>; Tom Lane <[email protected]> Sorry for not responding to this thread for a long time and a huge thank you for pushing it forward! Best Regards, Xing On Fri, May 5, 2023 at 7:42 AM Nathan Bossart <[email protected]> wrote: > On Thu, May 04, 2023 at 08:39:03AM -0400, Tom Lane wrote: > > Hmm, I'm not sure why PLy_trigger_build_args's pltargs needs to > > gain a "volatile" here? LGTM otherwise. > > I removed that new "volatile" marker before committing. I was trying to > future-proof a bit and follow elog.h's recommendation to the letter, but > following your mental model upthread, it doesn't seem to be strictly > necessary, and we'd need to set pltargs to NULL after decrementing its > reference count in the PG_TRY section for such future-proofing to be > effective, anyway. > > Thank you for reviewing! > > -- > Nathan Bossart > Amazon Web Services: https://aws.amazon.com > ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH v11 4/7] Row pattern recognition patch (executor). @ 2023-11-08 06:57 Tatsuo Ishii <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Tatsuo Ishii @ 2023-11-08 06:57 UTC (permalink / raw) --- src/backend/executor/nodeWindowAgg.c | 1420 +++++++++++++++++++++++++- src/backend/utils/adt/windowfuncs.c | 37 +- src/include/catalog/pg_proc.dat | 6 + src/include/nodes/execnodes.h | 26 + 4 files changed, 1476 insertions(+), 13 deletions(-) diff --git a/src/backend/executor/nodeWindowAgg.c b/src/backend/executor/nodeWindowAgg.c index 77724a6daa..ea5e73c969 100644 --- a/src/backend/executor/nodeWindowAgg.c +++ b/src/backend/executor/nodeWindowAgg.c @@ -36,6 +36,7 @@ #include "access/htup_details.h" #include "catalog/objectaccess.h" #include "catalog/pg_aggregate.h" +#include "catalog/pg_collation_d.h" #include "catalog/pg_proc.h" #include "executor/executor.h" #include "executor/nodeWindowAgg.h" @@ -48,6 +49,7 @@ #include "utils/acl.h" #include "utils/builtins.h" #include "utils/datum.h" +#include "utils/fmgroids.h" #include "utils/expandeddatum.h" #include "utils/lsyscache.h" #include "utils/memutils.h" @@ -159,6 +161,40 @@ typedef struct WindowStatePerAggData bool restart; /* need to restart this agg in this cycle? */ } WindowStatePerAggData; +/* + * Set of StringInfo. Used in RPR. + */ +typedef struct StringSet { + StringInfo *str_set; + Size set_size; /* current array allocation size in number of items */ + int set_index; /* current used size */ +} StringSet; + +/* + * Allowed subsequent PATTERN variables positions. + * Used in RPR. + * + * pos represents the pattern variable defined order in DEFINE caluase. For + * example. "DEFINE START..., UP..., DOWN ..." and "PATTERN START UP DOWN UP" + * will create: + * VariablePos[0].pos[0] = 0; START + * VariablePos[1].pos[0] = 1; UP + * VariablePos[1].pos[1] = 3; UP + * VariablePos[2].pos[0] = 2; DOWN + * + * Note that UP has two pos because UP appears in PATTERN twice. + * + * By using this strucrture, we can know which pattern variable can be followed + * by which pattern variable(s). For example, START can be followed by UP and + * DOWN since START's pos is 0, and UP's pos is 1 or 3, DOWN's pos is 2. + * DOWN can be followed by UP since UP's pos is either 1 or 3. + * + */ +#define NUM_ALPHABETS 26 /* we allow [a-z] variable initials */ +typedef struct VariablePos { + int pos[NUM_ALPHABETS]; /* postion(s) in PATTERN */ +} VariablePos; + static void initialize_windowaggregate(WindowAggState *winstate, WindowStatePerFunc perfuncstate, WindowStatePerAgg peraggstate); @@ -182,8 +218,9 @@ static void begin_partition(WindowAggState *winstate); static void spool_tuples(WindowAggState *winstate, int64 pos); static void release_partition(WindowAggState *winstate); -static int row_is_in_frame(WindowAggState *winstate, int64 pos, +static int row_is_in_frame(WindowAggState *winstate, int64 pos, TupleTableSlot *slot); + static void update_frameheadpos(WindowAggState *winstate); static void update_frametailpos(WindowAggState *winstate); static void update_grouptailpos(WindowAggState *winstate); @@ -195,9 +232,42 @@ static Datum GetAggInitVal(Datum textInitVal, Oid transtype); static bool are_peers(WindowAggState *winstate, TupleTableSlot *slot1, TupleTableSlot *slot2); -static bool window_gettupleslot(WindowObject winobj, int64 pos, - TupleTableSlot *slot); +static int WinGetSlotInFrame(WindowObject winobj, TupleTableSlot *slot, + int relpos, int seektype, bool set_mark, + bool *isnull, bool *isout); +static bool window_gettupleslot(WindowObject winobj, int64 pos, TupleTableSlot *slot); + +static void attno_map(Node *node); +static bool attno_map_walker(Node *node, void *context); +static int row_is_in_reduced_frame(WindowObject winobj, int64 pos); +static bool rpr_is_defined(WindowAggState *winstate); + +static void create_reduced_frame_map(WindowAggState *winstate); +static int get_reduced_frame_map(WindowAggState *winstate, int64 pos); +static void register_reduced_frame_map(WindowAggState *winstate, int64 pos, int val); +static void clear_reduced_frame_map(WindowAggState *winstate); +static void update_reduced_frame(WindowObject winobj, int64 pos); + +static int64 evaluate_pattern(WindowObject winobj, int64 current_pos, + char *vname, StringInfo encoded_str, bool *result); + +static bool get_slots(WindowObject winobj, int64 current_pos); + +static int search_str_set(char *pattern, StringSet *str_set, VariablePos *variable_pos); +static char pattern_initial(WindowAggState *winstate, char *vname); +static int do_pattern_match(char *pattern, char *encoded_str); + +static StringSet *string_set_init(void); +static void string_set_add(StringSet *string_set, StringInfo str); +static StringInfo string_set_get(StringSet *string_set, int index); +static int string_set_get_size(StringSet *string_set); +static void string_set_discard(StringSet *string_set); +static VariablePos *variable_pos_init(void); +static void variable_pos_register(VariablePos *variable_pos, char initial, int pos); +static bool variable_pos_compare(VariablePos *variable_pos, char initial1, char initial2); +static int variable_pos_fetch(VariablePos *variable_pos, char initial, int index); +static void variable_pos_discard(VariablePos *variable_pos); /* * initialize_windowaggregate @@ -673,6 +743,7 @@ eval_windowaggregates(WindowAggState *winstate) WindowObject agg_winobj; TupleTableSlot *agg_row_slot; TupleTableSlot *temp_slot; + bool agg_result_isnull; numaggs = winstate->numaggs; if (numaggs == 0) @@ -778,6 +849,9 @@ eval_windowaggregates(WindowAggState *winstate) * Note that we don't strictly need to restart in the last case, but if * we're going to remove all rows from the aggregation anyway, a restart * surely is faster. + * + * - if RPR is enabled and skip mode is SKIP TO NEXT ROW, + * we restart aggregation too. *---------- */ numaggs_restart = 0; @@ -788,8 +862,11 @@ eval_windowaggregates(WindowAggState *winstate) (winstate->aggregatedbase != winstate->frameheadpos && !OidIsValid(peraggstate->invtransfn_oid)) || (winstate->frameOptions & FRAMEOPTION_EXCLUSION) || - winstate->aggregatedupto <= winstate->frameheadpos) + winstate->aggregatedupto <= winstate->frameheadpos || + (rpr_is_defined(winstate) && + winstate->rpSkipTo == ST_NEXT_ROW)) { + elog(DEBUG1, "peraggstate->restart is set"); peraggstate->restart = true; numaggs_restart++; } @@ -861,8 +938,10 @@ eval_windowaggregates(WindowAggState *winstate) * If we created a mark pointer for aggregates, keep it pushed up to frame * head, so that tuplestore can discard unnecessary rows. */ +#ifdef NOT_USED if (agg_winobj->markptr >= 0) WinSetMarkPosition(agg_winobj, winstate->frameheadpos); +#endif /* * Now restart the aggregates that require it. @@ -917,6 +996,29 @@ eval_windowaggregates(WindowAggState *winstate) { winstate->aggregatedupto = winstate->frameheadpos; ExecClearTuple(agg_row_slot); + + /* + * If RPR is defined, we do not use aggregatedupto_nonrestarted. To + * avoid assertion failure below, we reset aggregatedupto_nonrestarted + * to frameheadpos. + */ + if (rpr_is_defined(winstate)) + aggregatedupto_nonrestarted = winstate->frameheadpos; + } + + agg_result_isnull = false; + /* RPR is defined? */ + if (rpr_is_defined(winstate)) + { + /* + * If the skip mode is SKIP TO PAST LAST ROW and we already know that + * current row is a skipped row, we don't need to accumulate rows, + * just return NULL. Note that for unamtched row, we need to do + * aggregation so that count(*) shows 0, rather than NULL. + */ + if (winstate->rpSkipTo == ST_PAST_LAST_ROW && + get_reduced_frame_map(winstate, winstate->currentpos) == RF_SKIPPED) + agg_result_isnull = true; } /* @@ -930,6 +1032,11 @@ eval_windowaggregates(WindowAggState *winstate) { int ret; + elog(DEBUG1, "===== loop in frame starts: " INT64_FORMAT, winstate->aggregatedupto); + + if (agg_result_isnull) + break; + /* Fetch next row if we didn't already */ if (TupIsNull(agg_row_slot)) { @@ -945,9 +1052,28 @@ eval_windowaggregates(WindowAggState *winstate) ret = row_is_in_frame(winstate, winstate->aggregatedupto, agg_row_slot); if (ret < 0) break; + if (ret == 0) goto next_tuple; + if (rpr_is_defined(winstate)) + { + /* + * If the row status at currentpos is already decided and current + * row status is not decided yet, it means we passed the last + * reduced frame. Time to break the loop. + */ + if (get_reduced_frame_map(winstate, winstate->currentpos) != RF_NOT_DETERMINED && + get_reduced_frame_map(winstate, winstate->aggregatedupto) == RF_NOT_DETERMINED) + break; + /* + * Otherwise we need to calculate the reduced frame. + */ + ret = row_is_in_reduced_frame(winstate->agg_winobj, winstate->aggregatedupto); + if (ret == -1) /* unmatched row */ + break; + } + /* Set tuple context for evaluation of aggregate arguments */ winstate->tmpcontext->ecxt_outertuple = agg_row_slot; @@ -976,6 +1102,7 @@ next_tuple: ExecClearTuple(agg_row_slot); } + /* The frame's end is not supposed to move backwards, ever */ Assert(aggregatedupto_nonrestarted <= winstate->aggregatedupto); @@ -996,6 +1123,16 @@ next_tuple: peraggstate, result, isnull); + /* + * RPR is defined and we just return NULL because skip mode is SKIP + * TO PAST LAST ROW and current row is skipped row. + */ + if (agg_result_isnull) + { + *isnull = true; + *result = (Datum) 0; + } + /* * save the result in case next row shares the same frame. * @@ -1090,6 +1227,7 @@ begin_partition(WindowAggState *winstate) winstate->framehead_valid = false; winstate->frametail_valid = false; winstate->grouptail_valid = false; + create_reduced_frame_map(winstate); winstate->spooled_rows = 0; winstate->currentpos = 0; winstate->frameheadpos = 0; @@ -2053,6 +2191,8 @@ ExecWindowAgg(PlanState *pstate) CHECK_FOR_INTERRUPTS(); + elog(DEBUG1, "ExecWindowAgg called. pos: " INT64_FORMAT , winstate->currentpos); + if (winstate->status == WINDOWAGG_DONE) return NULL; @@ -2221,6 +2361,17 @@ ExecWindowAgg(PlanState *pstate) /* don't evaluate the window functions when we're in pass-through mode */ if (winstate->status == WINDOWAGG_RUN) { + /* + * If RPR is defined and skip mode is next row, we need to clear existing + * reduced frame info so that we newly calculate the info starting from + * current row. + */ + if (rpr_is_defined(winstate)) + { + if (winstate->rpSkipTo == ST_NEXT_ROW) + clear_reduced_frame_map(winstate); + } + /* * Evaluate true window functions */ @@ -2388,6 +2539,9 @@ ExecInitWindowAgg(WindowAgg *node, EState *estate, int eflags) TupleDesc scanDesc; ListCell *l; + TargetEntry *te; + Expr *expr; + /* check for unsupported flags */ Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK))); @@ -2483,6 +2637,16 @@ ExecInitWindowAgg(WindowAgg *node, EState *estate, int eflags) winstate->temp_slot_2 = ExecInitExtraTupleSlot(estate, scanDesc, &TTSOpsMinimalTuple); + winstate->prev_slot = ExecInitExtraTupleSlot(estate, scanDesc, + &TTSOpsMinimalTuple); + + winstate->next_slot = ExecInitExtraTupleSlot(estate, scanDesc, + &TTSOpsMinimalTuple); + + winstate->null_slot = ExecInitExtraTupleSlot(estate, scanDesc, + &TTSOpsMinimalTuple); + winstate->null_slot = ExecStoreAllNullTuple(winstate->null_slot); + /* * create frame head and tail slots only if needed (must create slots in * exactly the same cases that update_frameheadpos and update_frametailpos @@ -2667,6 +2831,39 @@ ExecInitWindowAgg(WindowAgg *node, EState *estate, int eflags) winstate->inRangeAsc = node->inRangeAsc; winstate->inRangeNullsFirst = node->inRangeNullsFirst; + /* Set up SKIP TO type */ + winstate->rpSkipTo = node->rpSkipTo; + /* Set up row pattern recognition PATTERN clause */ + winstate->patternVariableList = node->patternVariable; + winstate->patternRegexpList = node->patternRegexp; + + /* Set up row pattern recognition DEFINE clause */ + winstate->defineInitial = node->defineInitial; + winstate->defineVariableList = NIL; + winstate->defineClauseList = NIL; + if (node->defineClause != NIL) + { + /* + * Tweak arg var of PREV/NEXT so that it refers to scan/inner slot. + */ + foreach(l, node->defineClause) + { + char *name; + ExprState *exps; + + te = lfirst(l); + name = te->resname; + expr = te->expr; + + elog(DEBUG1, "defineVariable name: %s", name); + winstate->defineVariableList = lappend(winstate->defineVariableList, + makeString(pstrdup(name))); + attno_map((Node *)expr); + exps = ExecInitExpr(expr, (PlanState *) winstate); + winstate->defineClauseList = lappend(winstate->defineClauseList, exps); + } + } + winstate->all_first = true; winstate->partition_spooled = false; winstate->more_partitions = false; @@ -2674,6 +2871,57 @@ ExecInitWindowAgg(WindowAgg *node, EState *estate, int eflags) return winstate; } +/* + * Rewrite varno of Var node that is the argument of PREV/NET so that it sees + * scan tuple (PREV) or inner tuple (NEXT). + */ +static void +attno_map(Node *node) +{ + (void) expression_tree_walker(node, attno_map_walker, NULL); +} + +static bool +attno_map_walker(Node *node, void *context) +{ + FuncExpr *func; + int nargs; + Expr *expr; + Var *var; + + if (node == NULL) + return false; + + if (IsA(node, FuncExpr)) + { + func = (FuncExpr *)node; + + if (func->funcid == F_PREV || func->funcid == F_NEXT) + { + /* sanity check */ + nargs = list_length(func->args); + if (list_length(func->args) != 1) + elog(ERROR, "PREV/NEXT must have 1 argument but function %d has %d args", func->funcid, nargs); + + expr = (Expr *) lfirst(list_head(func->args)); + if (!IsA(expr, Var)) + elog(ERROR, "PREV/NEXT's arg is not Var"); /* XXX: is it possible that arg type is Const? */ + var = (Var *)expr; + + if (func->funcid == F_PREV) + /* + * Rewrite varno from OUTER_VAR to regular var no so that the + * var references scan tuple. + */ + var->varno = var->varnosyn; + else + var->varno = INNER_VAR; + elog(DEBUG1, "PREV/NEXT's varno is rewritten to: %d", var->varno); + } + } + return expression_tree_walker(node, attno_map_walker, NULL); +} + /* ----------------- * ExecEndWindowAgg * ----------------- @@ -2723,6 +2971,8 @@ ExecReScanWindowAgg(WindowAggState *node) ExecClearTuple(node->agg_row_slot); ExecClearTuple(node->temp_slot_1); ExecClearTuple(node->temp_slot_2); + ExecClearTuple(node->prev_slot); + ExecClearTuple(node->next_slot); if (node->framehead_slot) ExecClearTuple(node->framehead_slot); if (node->frametail_slot) @@ -3083,7 +3333,7 @@ window_gettupleslot(WindowObject winobj, int64 pos, TupleTableSlot *slot) return false; if (pos < winobj->markpos) - elog(ERROR, "cannot fetch row before WindowObject's mark position"); + elog(ERROR, "cannot fetch row: " INT64_FORMAT " before WindowObject's mark position: " INT64_FORMAT, pos, winobj->markpos ); oldcontext = MemoryContextSwitchTo(winstate->ss.ps.ps_ExprContext->ecxt_per_query_memory); @@ -3403,14 +3653,54 @@ WinGetFuncArgInFrame(WindowObject winobj, int argno, WindowAggState *winstate; ExprContext *econtext; TupleTableSlot *slot; - int64 abs_pos; - int64 mark_pos; Assert(WindowObjectIsValid(winobj)); winstate = winobj->winstate; econtext = winstate->ss.ps.ps_ExprContext; slot = winstate->temp_slot_1; + if (WinGetSlotInFrame(winobj, slot, + relpos, seektype, set_mark, + isnull, isout) == 0) + { + econtext->ecxt_outertuple = slot; + return ExecEvalExpr((ExprState *) list_nth(winobj->argstates, argno), + econtext, isnull); + } + + if (isout) + *isout = true; + *isnull = true; + return (Datum) 0; +} + +/* + * WinGetSlotInFrame + * slot: TupleTableSlot to store the result + * relpos: signed rowcount offset from the seek position + * seektype: WINDOW_SEEK_HEAD or WINDOW_SEEK_TAIL + * set_mark: If the row is found/in frame and set_mark is true, the mark is + * moved to the row as a side-effect. + * isnull: output argument, receives isnull status of result + * isout: output argument, set to indicate whether target row position + * is out of frame (can pass NULL if caller doesn't care about this) + * + * Returns 0 if we successfullt got the slot. false if out of frame. + * (also isout is set) + */ +static int +WinGetSlotInFrame(WindowObject winobj, TupleTableSlot *slot, + int relpos, int seektype, bool set_mark, + bool *isnull, bool *isout) +{ + WindowAggState *winstate; + int64 abs_pos; + int64 mark_pos; + int num_reduced_frame; + + Assert(WindowObjectIsValid(winobj)); + winstate = winobj->winstate; + switch (seektype) { case WINDOW_SEEK_CURRENT: @@ -3477,11 +3767,21 @@ WinGetFuncArgInFrame(WindowObject winobj, int argno, winstate->frameOptions); break; } + num_reduced_frame = row_is_in_reduced_frame(winobj, winstate->frameheadpos); + if (num_reduced_frame < 0) + goto out_of_frame; + else if (num_reduced_frame > 0) + if (relpos >= num_reduced_frame) + goto out_of_frame; break; case WINDOW_SEEK_TAIL: /* rejecting relpos > 0 is easy and simplifies code below */ if (relpos > 0) goto out_of_frame; + + /* RPR cares about frame head pos. Need to call update_frameheadpos */ + update_frameheadpos(winstate); + update_frametailpos(winstate); abs_pos = winstate->frametailpos - 1 + relpos; @@ -3548,6 +3848,12 @@ WinGetFuncArgInFrame(WindowObject winobj, int argno, mark_pos = 0; /* keep compiler quiet */ break; } + + num_reduced_frame = row_is_in_reduced_frame(winobj, winstate->frameheadpos + relpos); + if (num_reduced_frame < 0) + goto out_of_frame; + else if (num_reduced_frame > 0) + abs_pos = winstate->frameheadpos + relpos + num_reduced_frame - 1; break; default: elog(ERROR, "unrecognized window seek type: %d", seektype); @@ -3566,15 +3872,13 @@ WinGetFuncArgInFrame(WindowObject winobj, int argno, *isout = false; if (set_mark) WinSetMarkPosition(winobj, mark_pos); - econtext->ecxt_outertuple = slot; - return ExecEvalExpr((ExprState *) list_nth(winobj->argstates, argno), - econtext, isnull); + return 0; out_of_frame: if (isout) *isout = true; *isnull = true; - return (Datum) 0; + return -1; } /* @@ -3605,3 +3909,1097 @@ WinGetFuncArgCurrent(WindowObject winobj, int argno, bool *isnull) return ExecEvalExpr((ExprState *) list_nth(winobj->argstates, argno), econtext, isnull); } + +/* + * rpr_is_defined + * return true if Row pattern recognition is defined. + */ +static +bool rpr_is_defined(WindowAggState *winstate) +{ + return winstate->patternVariableList != NIL; +} + +/* + * row_is_in_reduced_frame + * Determine whether a row is in the current row's reduced window frame according + * to row pattern matching + * + * The row must has been already determined that it is in a full window frame + * and fetched it into slot. + * + * Returns: + * = 0, RPR is not defined. + * >0, if the row is the first in the reduced frame. Return the number of rows in the reduced frame. + * -1, if the row is unmatched row + * -2, if the row is in the reduced frame but needed to be skipped because of + * AFTER MATCH SKIP PAST LAST ROW + */ +static +int row_is_in_reduced_frame(WindowObject winobj, int64 pos) +{ + WindowAggState *winstate = winobj->winstate; + int state; + int rtn; + + if (!rpr_is_defined(winstate)) + { + /* + * RPR is not defined. Assume that we are always in the the reduced + * window frame. + */ + rtn = 0; + elog(DEBUG1, "row_is_in_reduced_frame returns %d: pos: " INT64_FORMAT, rtn, pos); + return rtn; + } + + state = get_reduced_frame_map(winstate, pos); + + if (state == RF_NOT_DETERMINED) + { + update_frameheadpos(winstate); + update_reduced_frame(winobj, pos); + } + + state = get_reduced_frame_map(winstate, pos); + + switch (state) + { + int64 i; + int num_reduced_rows; + + case RF_FRAME_HEAD: + num_reduced_rows = 1; + for (i = pos + 1; get_reduced_frame_map(winstate,i) == RF_SKIPPED; i++) + num_reduced_rows++; + rtn = num_reduced_rows; + break; + + case RF_SKIPPED: + rtn = -2; + break; + + case RF_UNMATCHED: + rtn = -1; + break; + + default: + elog(ERROR, "Unrecognized state: %d at: " INT64_FORMAT, state, pos); + break; + } + + elog(DEBUG1, "row_is_in_reduced_frame returns %d: pos: " INT64_FORMAT, rtn, pos); + return rtn; +} + +#define REDUCED_FRAME_MAP_INIT_SIZE 1024L + +/* + * Create reduced frame map + */ +static +void create_reduced_frame_map(WindowAggState *winstate) +{ + winstate->reduced_frame_map = + MemoryContextAlloc(winstate->partcontext, REDUCED_FRAME_MAP_INIT_SIZE); + winstate->alloc_sz = REDUCED_FRAME_MAP_INIT_SIZE; + clear_reduced_frame_map(winstate); +} + +/* + * Clear reduced frame map + */ +static +void clear_reduced_frame_map(WindowAggState *winstate) +{ + Assert(winstate->reduced_frame_map != NULL); + MemSet(winstate->reduced_frame_map, RF_NOT_DETERMINED, + winstate->alloc_sz); +} + +/* + * Get reduced frame map specified by pos + */ +static +int get_reduced_frame_map(WindowAggState *winstate, int64 pos) +{ + Assert(winstate->reduced_frame_map != NULL); + + if (pos < 0 || pos >= winstate->alloc_sz) + elog(ERROR, "wrong pos: " INT64_FORMAT, pos); + + return winstate->reduced_frame_map[pos]; +} + +/* + * Add/replace reduced frame map member at pos. + * If there's no enough space, expand the map. + */ +static +void register_reduced_frame_map(WindowAggState *winstate, int64 pos, int val) +{ + int64 realloc_sz; + + Assert(winstate->reduced_frame_map != NULL); + + if (pos < 0) + elog(ERROR, "wrong pos: " INT64_FORMAT, pos); + + if (pos > winstate->alloc_sz - 1) + { + realloc_sz = winstate->alloc_sz * 2; + + winstate->reduced_frame_map = + repalloc(winstate->reduced_frame_map, realloc_sz); + + MemSet(winstate->reduced_frame_map + winstate->alloc_sz, + RF_NOT_DETERMINED, realloc_sz - winstate->alloc_sz); + + winstate->alloc_sz = realloc_sz; + } + + winstate->reduced_frame_map[pos] = val; +} + +/* + * update_reduced_frame + * Update reduced frame info. + */ +static +void update_reduced_frame(WindowObject winobj, int64 pos) +{ + WindowAggState *winstate = winobj->winstate; + ListCell *lc1, *lc2; + bool expression_result; + int num_matched_rows; + int64 original_pos; + bool anymatch; + StringInfo encoded_str; + StringInfo pattern_str = makeStringInfo(); + StringSet *str_set; + int str_set_index = 0; + int initial_index; + VariablePos *variable_pos; + bool greedy = false; + int64 result_pos, i; + + /* + * Set of pattern variables evaluated to true. + * Each character corresponds to pattern variable. + * Example: + * str_set[0] = "AB"; + * str_set[1] = "AC"; + * In this case at row 0 A and B are true, and A and C are true in row 1. + */ + + /* initialize pattern variables set */ + str_set = string_set_init(); + + /* save original pos */ + original_pos = pos; + + /* + * Check if the pattern does not include any greedy quantifier. + * If it does not, we can just apply the pattern to each row. + * If it succeeds, we are done. + */ + foreach(lc1, winstate->patternRegexpList) + { + char *quantifier = strVal(lfirst(lc1)); + if (*quantifier == '+' || *quantifier == '*') + { + greedy = true; + break; + } + } + if (!greedy) + { + num_matched_rows = 0; + + foreach(lc1, winstate->patternVariableList) + { + char *vname = strVal(lfirst(lc1)); + + encoded_str = makeStringInfo(); + + elog(DEBUG1, "pos: " INT64_FORMAT " pattern vname: %s", pos, vname); + + expression_result = false; + + /* evaluate row pattern against current row */ + result_pos = evaluate_pattern(winobj, pos, vname, encoded_str, &expression_result); + if (!expression_result || result_pos < 0) + { + elog(DEBUG1, "expression result is false or out of frame"); + register_reduced_frame_map(winstate, original_pos, RF_UNMATCHED); + return; + } + /* move to next row */ + pos++; + num_matched_rows++; + } + elog(DEBUG1, "pattern matched"); + + register_reduced_frame_map(winstate, original_pos, RF_FRAME_HEAD); + + for (i = original_pos + 1; i < original_pos + num_matched_rows; i++) + { + register_reduced_frame_map(winstate, i, RF_SKIPPED); + } + return; + } + + /* + * Greedy quantifiers included. + * Loop over until none of pattern matches or encounters end of frame. + */ + for (;;) + { + result_pos = -1; + + /* + * Loop over each PATTERN variable. + */ + anymatch = false; + encoded_str = makeStringInfo(); + + forboth(lc1, winstate->patternVariableList, lc2, winstate->patternRegexpList) + { + char *vname = strVal(lfirst(lc1)); + char *quantifier = strVal(lfirst(lc2)); + + elog(DEBUG1, "pos: " INT64_FORMAT " pattern vname: %s quantifier: %s", pos, vname, quantifier); + + expression_result = false; + + /* evaluate row pattern against current row */ + result_pos = evaluate_pattern(winobj, pos, vname, encoded_str, &expression_result); + if (expression_result) + { + elog(DEBUG1, "expression result is true"); + anymatch = true; + } + + /* + * If out of frame, we are done. + */ + if (result_pos < 0) + break; + } + + if (!anymatch) + { + /* none of patterns matched. */ + break; + } + + string_set_add(str_set, encoded_str); + + elog(DEBUG1, "pos: " INT64_FORMAT " str_set_index: %d encoded_str: %s", pos, str_set_index, encoded_str->data); + + /* move to next row */ + pos++; + + if (result_pos < 0) + { + /* out of frame */ + break; + } + } + + if (string_set_get_size(str_set) == 0) + { + /* no match found in the first row */ + register_reduced_frame_map(winstate, original_pos, RF_UNMATCHED); + return; + } + + elog(DEBUG2, "pos: " INT64_FORMAT " encoded_str: %s", pos, encoded_str->data); + + /* build regular expression */ + pattern_str = makeStringInfo(); + appendStringInfoChar(pattern_str, '^'); + initial_index = 0; + + variable_pos = variable_pos_init(); + + forboth (lc1, winstate->patternVariableList, lc2, winstate->patternRegexpList) + { + char *vname = strVal(lfirst(lc1)); + char *quantifier = strVal(lfirst(lc2)); + char initial; + + initial = pattern_initial(winstate, vname); + Assert(initial != 0); + appendStringInfoChar(pattern_str, initial); + if (quantifier[0]) + appendStringInfoChar(pattern_str, quantifier[0]); + + /* + * Register the initial at initial_index. If the initial appears more + * than once, all of it's initial_index will be recorded. This could + * happen if a pattern variable appears in the PATTERN clause more + * than once like "UP DOWN UP" "UP UP UP". + */ + variable_pos_register(variable_pos, initial, initial_index); + + initial_index++; + } + + elog(DEBUG2, "pos: " INT64_FORMAT " pattern: %s", pos, pattern_str->data); + + /* look for matching pattern variable sequence */ + elog(DEBUG1, "search_str_set started"); + num_matched_rows = search_str_set(pattern_str->data, str_set, variable_pos); + elog(DEBUG1, "search_str_set returns: %d", num_matched_rows); + + variable_pos_discard(variable_pos); + string_set_discard(str_set); + + /* + * We are at the first row in the reduced frame. Save the number of + * matched rows as the number of rows in the reduced frame. + */ + if (num_matched_rows <= 0) + { + /* no match */ + register_reduced_frame_map(winstate, original_pos, RF_UNMATCHED); + } + else + { + register_reduced_frame_map(winstate, original_pos, RF_FRAME_HEAD); + + for (i = original_pos + 1; i < original_pos + num_matched_rows; i++) + { + register_reduced_frame_map(winstate, i, RF_SKIPPED); + } + } + + return; +} + +/* + * Perform pattern matching using pattern against str_set. pattern is a + * regular expression derived from PATTERN clause. Note that the regular + * expression string is prefixed by '^' and followed by initials represented + * in a same way as str_set. str_set is a set of StringInfo. Each StringInfo + * has a string comprising initials of pattern variable strings being true in + * a row. The initials are one of [a-y], parallel to the order of variable + * names in DEFINE clause. For example, if DEFINE has variables START, UP and + * DOWN, PATTERN HAS START, UP and DOWN, then the initials in PATTERN will be + * 'a', 'b' and 'c'. + * + * variable_pos is an array representing the order of pattern variable string + * initials in PATTERN clause. For example initial 'a' potion is in + * variable_pos[0].pos[0] = 0. Note that if the pattern is "START UP DOWN UP" + * (UP appears twice), then "UP" (initial is 'b') has two position 1 and + * 3. Thus variable_pos for b is variable_pos[1].pos[0] = 1 and + * variable_pos[1].pos[1] = 3. + * + * Returns the longest number of the matching rows. + */ +static +int search_str_set(char *pattern, StringSet *str_set, VariablePos *variable_pos) +{ +#define MAX_CANDIDATE_NUM 10000 /* max pattern match candidate size */ +#define FREEZED_CHAR 'Z' /* a pattern is freezed if it ends with the char */ +#define DISCARD_CHAR 'z' /* a pattern is not need to keep */ + + int set_size; /* number of rows in the set */ + int resultlen; + int index; + StringSet *old_str_set, *new_str_set; + int new_str_size; + int len; + + set_size = string_set_get_size(str_set); + new_str_set = string_set_init(); + len = 0; + resultlen = 0; + + /* + * Generate all possible pattern variable name initials as a set of + * StringInfo named "new_str_set". For example, if we have two rows + * having "ab" (row 0) and "ac" (row 1) in the input str_set, new_str_set + * will have set of StringInfo "aa", "ac", "ba" and "bc" in the end. + */ + elog(DEBUG1, "pattern: %s set_size: %d", pattern, set_size); + for (index = 0; index < set_size; index++) + { + StringInfo str; /* search target row */ + char *p; + int old_set_size; + int i; + + elog(DEBUG1, "index: %d", index); + + if (index == 0) + { + /* copy variables in row 0 */ + str = string_set_get(str_set, index); + p = str->data; + + /* + * Loop over each new pattern variable char. + */ + while (*p) + { + StringInfo new = makeStringInfo(); + + /* add pattern variable char */ + appendStringInfoChar(new, *p); + /* add new one to string set */ + string_set_add(new_str_set, new); + elog(DEBUG1, "old_str: NULL new_str: %s", new->data); + p++; /* next pattern variable */ + } + } + else /* index != 0 */ + { + old_str_set = new_str_set; + new_str_set = string_set_init(); + str = string_set_get(str_set, index); + old_set_size = string_set_get_size(old_str_set); + + /* + * Loop over each rows in the previous result set. + */ + for (i = 0; i < old_set_size ; i++) + { + StringInfo new; + char last_old_char; + int old_str_len; + StringInfo old = string_set_get(old_str_set, i); + + p = old->data; + old_str_len = strlen(p); + if (old_str_len > 0) + last_old_char = p[old_str_len - 1]; + else + last_old_char = '\0'; + + /* Is this old set freezed? */ + if (last_old_char == FREEZED_CHAR) + { + /* if shorter match. we can discard it */ + if ((old_str_len - 1) < resultlen) + { + elog(DEBUG1, "discard this old set because shorter match: %s", old->data); + continue; + } + + elog(DEBUG1, "keep this old set: %s", old->data); + + /* move the old set to new_str_set */ + string_set_add(new_str_set, old); + old_str_set->str_set[i] = NULL; + continue; + } + /* Can this old set be discarded? */ + else if (last_old_char == DISCARD_CHAR) + { + elog(DEBUG1, "discard this old set: %s", old->data); + continue; + } + + elog(DEBUG1, "str->data: %s", str->data); + + /* + * loop over each pattern variable initial char in the input + * set. + */ + for (p = str->data; *p; p++) + { + /* + * Optimization. Check if the row's pattern variable + * initial character position is greater than or equal to + * the old set's last pattern variable initial character + * position. For example, if the old set's last pattern + * variable initials are "ab", then the new pattern + * variable initial can be "b" or "c" but can not be "a", + * if the initials in PATTERN is something like "a b c" or + * "a b+ c+" etc. This optimization is possible when we + * only allow "+" quantifier. + */ + if (variable_pos_compare(variable_pos, last_old_char, *p)) + { + /* copy source string */ + new = makeStringInfo(); + enlargeStringInfo(new, old->len + 1); + appendStringInfoString(new, old->data); + /* add pattern variable char */ + appendStringInfoChar(new, *p); + elog(DEBUG1, "old_str: %s new_str: %s", old->data, new->data); + + /* + * Adhoc optimization. If the first letter in the + * input string is the first and second position one + * and there's no associated quatifier '+', then we + * can dicard the input because there's no chace to + * expand the string further. + * + * For example, pattern "abc" cannot match "aa". + */ + elog(DEBUG1, "pattern[1]:%c pattern[2]:%c new[0]:%c new[1]:%c", + pattern[1], pattern[2], new->data[0], new->data[1]); + if (pattern[1] == new->data[0] && + pattern[1] == new->data[1] && + pattern[2] != '+' && + pattern[1] != pattern[2]) + { + elog(DEBUG1, "discard this new data: %s", + new->data); + pfree(new->data); + pfree(new); + continue; + } + + /* add new one to string set */ + string_set_add(new_str_set, new); + } + else + { + /* + * We are freezing this pattern string. Since there's + * no chance to expand the string further, we perform + * pattern matching against the string. If it does not + * match, we can discard it. + */ + len = do_pattern_match(pattern, old->data); + + if (len <= 0) + { + /* no match. we can discard it */ + continue; + } + + else if (len <= resultlen) + { + /* shorter match. we can discard it */ + continue; + } + else + { + /* match length is the longest so far */ + + int new_index; + + /* remember the longest match */ + resultlen = len; + + /* freeze the pattern string */ + new = makeStringInfo(); + enlargeStringInfo(new, old->len + 1); + appendStringInfoString(new, old->data); + /* add freezed mark */ + appendStringInfoChar(new, FREEZED_CHAR); + elog(DEBUG1, "old_str: %s new_str: %s", old->data, new->data); + string_set_add(new_str_set, new); + + /* + * Search new_str_set to find out freezed + * entries that have shorter match length. + * Mark them as "discard" so that they are + * discarded in the next round. + */ + + /* new_index_size should be the one before */ + new_str_size = string_set_get_size(new_str_set) - 1; + + /* loop over new_str_set */ + for (new_index = 0; new_index < new_str_size; new_index++) + { + char new_last_char; + int new_str_len; + + new = string_set_get(new_str_set, new_index); + new_str_len = strlen(new->data); + if (new_str_len > 0) + { + new_last_char = new->data[new_str_len - 1]; + if (new_last_char == FREEZED_CHAR && + (new_str_len - 1) <= len) + { + /* mark this set to discard in the next round */ + appendStringInfoChar(new, DISCARD_CHAR); + elog(DEBUG1, "add discard char: %s", new->data); + } + } + } + } + } + } + } + /* we no longer need old string set */ + string_set_discard(old_str_set); + } + } + + /* + * Perform pattern matching to find out the longest match. + */ + new_str_size = string_set_get_size(new_str_set); + elog(DEBUG1, "new_str_size: %d", new_str_size); + len = 0; + resultlen = 0; + + for (index = 0; index < new_str_size; index++) + { + StringInfo s; + + s = string_set_get(new_str_set, index); + if (s == NULL) + continue; /* no data */ + + elog(DEBUG1, "target string: %s", s->data); + + len = do_pattern_match(pattern, s->data); + if (len > resultlen) + { + /* remember the longest match */ + resultlen = len; + + /* + * If the size of result set is equal to the number of rows in the + * set, we are done because it's not possible that the number of + * matching rows exceeds the number of rows in the set. + */ + if (resultlen >= set_size) + break; + } + } + + /* we no longer need new string set */ + string_set_discard(new_str_set); + + return resultlen; +} + +/* + * do_pattern_match + * perform pattern match using pattern against encoded_str. + * returns matching number of rows if matching is succeeded. + * Otherwise returns 0. + */ +static +int do_pattern_match(char *pattern, char *encoded_str) +{ + Datum d; + text *res; + char *substr; + int len = 0; + text *pattern_text, *encoded_str_text; + + pattern_text = cstring_to_text(pattern); + encoded_str_text = cstring_to_text(encoded_str); + + /* + * We first perform pattern matching using regexp_instr, then call + * textregexsubstr to get matched substring to know how long the + * matched string is. That is the number of rows in the reduced window + * frame. The reason why we can't call textregexsubstr in the first + * place is, it errors out if pattern does not match. + */ + if (DatumGetInt32(DirectFunctionCall2Coll(regexp_instr, DEFAULT_COLLATION_OID, + PointerGetDatum(encoded_str_text), + PointerGetDatum(pattern_text)))) + { + d = DirectFunctionCall2Coll(textregexsubstr, + DEFAULT_COLLATION_OID, + PointerGetDatum(encoded_str_text), + PointerGetDatum(pattern_text)); + if (d != 0) + { + res = DatumGetTextPP(d); + substr = text_to_cstring(res); + len = strlen(substr); + pfree(substr); + } + } + pfree(encoded_str_text); + pfree(pattern_text); + + return len; +} + + +/* + * Evaluate expression associated with PATTERN variable vname. + * relpos is relative row position in a frame (starting from 0). + * "quantifier" is the quatifier part of the PATTERN regular expression. + * Currently only '+' is allowed. + * result is out paramater representing the expression evaluation result + * is true of false. + * Return values are: + * >=0: the last match absolute row position + * other wise out of frame. + */ +static +int64 evaluate_pattern(WindowObject winobj, int64 current_pos, + char *vname, StringInfo encoded_str, bool *result) +{ + WindowAggState *winstate = winobj->winstate; + ExprContext *econtext = winstate->ss.ps.ps_ExprContext; + ListCell *lc1, *lc2, *lc3; + ExprState *pat; + Datum eval_result; + bool out_of_frame = false; + bool isnull; + TupleTableSlot *slot; + + forthree (lc1, winstate->defineVariableList, lc2, winstate->defineClauseList, lc3, winstate->defineInitial) + { + char initial; + char *name = strVal(lfirst(lc1)); + + if (strcmp(vname, name)) + continue; + + initial = *(strVal(lfirst(lc3))); + + /* set expression to evaluate */ + pat = lfirst(lc2); + + /* get current, previous and next tuples */ + if (!get_slots(winobj, current_pos)) + { + out_of_frame = true; + } + else + { + /* evaluate the expression */ + eval_result = ExecEvalExpr(pat, econtext, &isnull); + if (isnull) + { + /* expression is NULL */ + elog(DEBUG1, "expression for %s is NULL at row: " INT64_FORMAT, vname, current_pos); + *result = false; + } + else + { + if (!DatumGetBool(eval_result)) + { + /* expression is false */ + elog(DEBUG1, "expression for %s is false at row: " INT64_FORMAT, vname, current_pos); + *result = false; + } + else + { + /* expression is true */ + elog(DEBUG1, "expression for %s is true at row: " INT64_FORMAT, vname, current_pos); + appendStringInfoChar(encoded_str, initial); + *result = true; + } + } + + slot = winstate->temp_slot_1; + if (slot != winstate->null_slot) + ExecClearTuple(slot); + slot = winstate->prev_slot; + if (slot != winstate->null_slot) + ExecClearTuple(slot); + slot = winstate->next_slot; + if (slot != winstate->null_slot) + ExecClearTuple(slot); + + break; + } + + if (out_of_frame) + { + *result = false; + return -1; + } + } + return current_pos; +} + +/* + * Get current, previous and next tuples. + * Returns false if current row is out of partition/full frame. + */ +static +bool get_slots(WindowObject winobj, int64 current_pos) +{ + WindowAggState *winstate = winobj->winstate; + TupleTableSlot *slot; + int ret; + ExprContext *econtext; + + econtext = winstate->ss.ps.ps_ExprContext; + + /* set up current row tuple slot */ + slot = winstate->temp_slot_1; + if (!window_gettupleslot(winobj, current_pos, slot)) + { + elog(DEBUG1, "current row is out of partition at:" INT64_FORMAT, current_pos); + return false; + } + ret = row_is_in_frame(winstate, current_pos, slot); + if (ret <= 0) + { + elog(DEBUG1, "current row is out of frame at: " INT64_FORMAT, current_pos); + ExecClearTuple(slot); + return false; + } + econtext->ecxt_outertuple = slot; + + /* for PREV */ + if (current_pos > 0) + { + slot = winstate->prev_slot; + if (!window_gettupleslot(winobj, current_pos - 1, slot)) + { + elog(DEBUG1, "previous row is out of partition at: " INT64_FORMAT, current_pos - 1); + econtext->ecxt_scantuple = winstate->null_slot; + } + else + { + ret = row_is_in_frame(winstate, current_pos - 1, slot); + if (ret <= 0) + { + elog(DEBUG1, "previous row is out of frame at: " INT64_FORMAT, current_pos - 1); + ExecClearTuple(slot); + econtext->ecxt_scantuple = winstate->null_slot; + } + else + { + econtext->ecxt_scantuple = slot; + } + } + } + else + econtext->ecxt_scantuple = winstate->null_slot; + + /* for NEXT */ + slot = winstate->next_slot; + if (!window_gettupleslot(winobj, current_pos + 1, slot)) + { + elog(DEBUG1, "next row is out of partiton at: " INT64_FORMAT, current_pos + 1); + econtext->ecxt_innertuple = winstate->null_slot; + } + else + { + ret = row_is_in_frame(winstate, current_pos + 1, slot); + if (ret <= 0) + { + elog(DEBUG1, "next row is out of frame at: " INT64_FORMAT, current_pos + 1); + ExecClearTuple(slot); + econtext->ecxt_innertuple = winstate->null_slot; + } + else + econtext->ecxt_innertuple = slot; + } + return true; +} + +/* + * Return pattern variable initial character + * matching with pattern variable name vname. + * If not found, return 0. + */ +static +char pattern_initial(WindowAggState *winstate, char *vname) +{ + char initial; + char *name; + ListCell *lc1, *lc2; + + forboth (lc1, winstate->defineVariableList, lc2, winstate->defineInitial) + { + name = strVal(lfirst(lc1)); /* DEFINE variable name */ + initial = *(strVal(lfirst(lc2))); /* DEFINE variable initial */ + + + if (!strcmp(name, vname)) + return initial; /* found */ + } + return 0; +} + +/* + * string_set_init + * Create dynamic set of StringInfo. + */ +static +StringSet *string_set_init(void) +{ +/* Initial allocation size of str_set */ +#define STRING_SET_ALLOC_SIZE 1024 + + StringSet *string_set; + Size set_size; + + string_set = palloc0(sizeof(StringSet)); + string_set->set_index = 0; + set_size = STRING_SET_ALLOC_SIZE; + string_set->str_set = palloc(set_size * sizeof(StringInfo)); + string_set->set_size = set_size; + + return string_set; +} + +/* + * Add StringInfo str to StringSet string_set. + */ +static +void string_set_add(StringSet *string_set, StringInfo str) +{ + Size set_size; + + set_size = string_set->set_size; + if (string_set->set_index >= set_size) + { + set_size *= 2; + string_set->str_set = repalloc(string_set->str_set, + set_size * sizeof(StringInfo)); + string_set->set_size = set_size; + } + + string_set->str_set[string_set->set_index++] = str; + + return; +} + +/* + * Returns StringInfo specified by index. + * If there's no data yet, returns NULL. + */ +static +StringInfo string_set_get(StringSet *string_set, int index) +{ + /* no data? */ + if (index == 0 && string_set->set_index == 0) + return NULL; + + if (index < 0 ||index >= string_set->set_index) + elog(ERROR, "invalid index: %d", index); + + return string_set->str_set[index]; +} + +/* + * Returns the size of StringSet. + */ +static +int string_set_get_size(StringSet *string_set) +{ + return string_set->set_index; +} + +/* + * Discard StringSet. + * All memory including StringSet itself is freed. + */ +static +void string_set_discard(StringSet *string_set) +{ + int i; + + for (i = 0; i < string_set->set_index; i++) + { + StringInfo str = string_set->str_set[i]; + if (str) + { + pfree(str->data); + pfree(str); + } + } + pfree(string_set->str_set); + pfree(string_set); +} + +/* + * Create and initialize variable postion structure + */ +static +VariablePos *variable_pos_init(void) +{ + VariablePos *variable_pos; + + variable_pos = palloc(sizeof(VariablePos) * NUM_ALPHABETS); + MemSet(variable_pos, -1, sizeof(VariablePos) * NUM_ALPHABETS); + return variable_pos; +} + +/* + * Register pattern variable whose initial is initial into postion index. + * pos is position of initial. + * If pos is already registered, register it at next empty slot. + */ +static +void variable_pos_register(VariablePos *variable_pos, char initial, int pos) +{ + int index = initial - 'a'; + int slot; + int i; + + if (pos < 0 || pos > NUM_ALPHABETS) + elog(ERROR, "initial is not valid char: %c", initial); + + for (i = 0; i < NUM_ALPHABETS; i++) + { + slot = variable_pos[index].pos[i]; + if (slot < 0) + { + /* empty slot found */ + variable_pos[index].pos[i] = pos; + return; + } + } + elog(ERROR, "no empty slot for initial: %c", initial); +} + +/* + * Returns true if initial1 can be followed by initial2 + */ +static +bool variable_pos_compare(VariablePos *variable_pos, char initial1, char initial2) +{ + int index1, index2; + int pos1, pos2; + + for (index1 = 0; ; index1++) + { + pos1 = variable_pos_fetch(variable_pos, initial1, index1); + if (pos1 < 0) + break; + + for (index2 = 0; ; index2++) + { + pos2 = variable_pos_fetch(variable_pos, initial2, index2); + if (pos2 < 0) + break; + if (pos1 <= pos2) + return true; + } + } + return false; +} + +/* + * Fetch position of pattern variable whose initial is initial, and whose index + * is index. If no postion was registered by initial, index, returns -1. + */ +static +int variable_pos_fetch(VariablePos *variable_pos, char initial, int index) +{ + int pos = initial - 'a'; + + if (pos < 0 || pos > NUM_ALPHABETS) + elog(ERROR, "initial is not valid char: %c", initial); + + if (index < 0 || index > NUM_ALPHABETS) + elog(ERROR, "index is not valid: %d", index); + + return variable_pos[pos].pos[index]; +} + +/* + * Discard VariablePos + */ +static +void variable_pos_discard(VariablePos *variable_pos) +{ + pfree(variable_pos); +} diff --git a/src/backend/utils/adt/windowfuncs.c b/src/backend/utils/adt/windowfuncs.c index 0bfbac00d7..a4a11c7f8d 100644 --- a/src/backend/utils/adt/windowfuncs.c +++ b/src/backend/utils/adt/windowfuncs.c @@ -13,6 +13,9 @@ */ #include "postgres.h" +#include "catalog/pg_collation_d.h" +#include "executor/executor.h" +#include "nodes/execnodes.h" #include "nodes/parsenodes.h" #include "nodes/supportnodes.h" #include "utils/builtins.h" @@ -37,11 +40,19 @@ typedef struct int64 remainder; /* (total rows) % (bucket num) */ } ntile_context; +/* + * rpr process information. + * Used for AFTER MATCH SKIP PAST LAST ROW + */ +typedef struct SkipContext +{ + int64 pos; /* last row absolute position */ +} SkipContext; + static bool rank_up(WindowObject winobj); static Datum leadlag_common(FunctionCallInfo fcinfo, bool forward, bool withoffset, bool withdefault); - /* * utility routine for *_rank functions. */ @@ -674,7 +685,7 @@ window_last_value(PG_FUNCTION_ARGS) bool isnull; result = WinGetFuncArgInFrame(winobj, 0, - 0, WINDOW_SEEK_TAIL, true, + 0, WINDOW_SEEK_TAIL, false, &isnull, NULL); if (isnull) PG_RETURN_NULL(); @@ -714,3 +725,25 @@ window_nth_value(PG_FUNCTION_ARGS) PG_RETURN_DATUM(result); } + +/* + * prev + * Dummy function to invoke RPR's navigation operator "PREV". + * This is *not* a window function. + */ +Datum +window_prev(PG_FUNCTION_ARGS) +{ + PG_RETURN_DATUM(PG_GETARG_DATUM(0)); +} + +/* + * next + * Dummy function to invoke RPR's navigation operation "NEXT". + * This is *not* a window function. + */ +Datum +window_next(PG_FUNCTION_ARGS) +{ + PG_RETURN_DATUM(PG_GETARG_DATUM(0)); +} diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index f14aed422a..6df54a3cab 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -10423,6 +10423,12 @@ { oid => '3114', descr => 'fetch the Nth row value', proname => 'nth_value', prokind => 'w', prorettype => 'anyelement', proargtypes => 'anyelement int4', prosrc => 'window_nth_value' }, +{ oid => '6122', descr => 'previous value', + proname => 'prev', provolatile => 's', prorettype => 'anyelement', + proargtypes => 'anyelement', prosrc => 'window_prev' }, +{ oid => '6123', descr => 'next value', + proname => 'next', provolatile => 's', prorettype => 'anyelement', + proargtypes => 'anyelement', prosrc => 'window_next' }, # functions for range types { oid => '3832', descr => 'I/O', diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index 5d7f17dee0..d8f92720bc 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -2470,6 +2470,11 @@ typedef enum WindowAggStatus * tuples during spool */ } WindowAggStatus; +#define RF_NOT_DETERMINED 0 +#define RF_FRAME_HEAD 1 +#define RF_SKIPPED 2 +#define RF_UNMATCHED 3 + typedef struct WindowAggState { ScanState ss; /* its first field is NodeTag */ @@ -2518,6 +2523,15 @@ typedef struct WindowAggState int64 groupheadpos; /* current row's peer group head position */ int64 grouptailpos; /* " " " " tail position (group end+1) */ + /* these fields are used in Row pattern recognition: */ + RPSkipTo rpSkipTo; /* Row Pattern Skip To type */ + List *patternVariableList; /* list of row pattern variables names (list of String) */ + List *patternRegexpList; /* list of row pattern regular expressions ('+' or ''. list of String) */ + List *defineVariableList; /* list of row pattern definition variables (list of String) */ + List *defineClauseList; /* expression for row pattern definition + * search conditions ExprState list */ + List *defineInitial; /* list of row pattern definition variable initials (list of String) */ + MemoryContext partcontext; /* context for partition-lifespan data */ MemoryContext aggcontext; /* shared context for aggregate working data */ MemoryContext curaggcontext; /* current aggregate's working data */ @@ -2554,6 +2568,18 @@ typedef struct WindowAggState TupleTableSlot *agg_row_slot; TupleTableSlot *temp_slot_1; TupleTableSlot *temp_slot_2; + + /* temporary slots for RPR */ + TupleTableSlot *prev_slot; /* PREV row navigation operator */ + TupleTableSlot *next_slot; /* NEXT row navigation operator */ + TupleTableSlot *null_slot; /* all NULL slot */ + + /* + * Each byte corresponds to a row positioned at absolute its pos in + * partition. See above definition for RF_* + */ + char *reduced_frame_map; + int64 alloc_sz; /* size of the map */ } WindowAggState; /* ---------------- -- 2.25.1 ----Next_Part(Wed_Nov__8_16_37_05_2023_872)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="v11-0005-Row-pattern-recognition-patch-docs.patch" ^ permalink raw reply [nested|flat] 9+ messages in thread
end of thread, other threads:[~2023-11-08 06:57 UTC | newest] Thread overview: 9+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2018-10-11 07:18 [PATCH 4/4] Fix WAL skipping feature. Kyotaro Horiguchi <[email protected]> 2023-05-03 20:33 Re: PL/Python: Fix return in the middle of PG_TRY() block. Tom Lane <[email protected]> 2023-05-03 20:58 ` Re: PL/Python: Fix return in the middle of PG_TRY() block. Nathan Bossart <[email protected]> 2023-05-04 04:54 ` Re: PL/Python: Fix return in the middle of PG_TRY() block. Nathan Bossart <[email protected]> 2023-05-04 05:01 ` Re: PL/Python: Fix return in the middle of PG_TRY() block. Nathan Bossart <[email protected]> 2023-05-04 12:39 ` Re: PL/Python: Fix return in the middle of PG_TRY() block. Tom Lane <[email protected]> 2023-05-04 23:42 ` Re: PL/Python: Fix return in the middle of PG_TRY() block. Nathan Bossart <[email protected]> 2023-05-05 01:40 ` Re: PL/Python: Fix return in the middle of PG_TRY() block. Xing Guo <[email protected]> 2023-11-08 06:57 [PATCH v11 4/7] Row pattern recognition patch (executor). Tatsuo Ishii <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox