agora inbox for [email protected]help / color / mirror / Atom feed
[PATCH v9 1/4] INSERT SELECT to use BulkInsertState and multi_insert 9+ messages / 5 participants [nested] [flat]
* [PATCH v9 1/4] INSERT SELECT to use BulkInsertState and multi_insert @ 2020-05-08 07:17 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Justin Pryzby @ 2020-05-08 07:17 UTC (permalink / raw) Renames structures; Move MultipleInsert functions from copyfrom.c to (tentatively) nodeModifyTable.h; Move into MultiInsertInfo: transition_capture and cur_lineno (via cstate->miinfo); Dynamically switch to multi-insert mode based on the number of insertions. This is intended to accomodate 1) the original use case of INSERT using a small ring buffer to avoid leaving behind dirty buffers; and, 2) Automatically using multi-inserts for batch operations; 3) allow the old behavior of leaving behind dirty buffers, which might allow INSERT to run more quickly, at the cost of leaving behind many dirty buffers which other backends may have to write out. XXX: for (1), the bulk-insert state is used even if not multi-insert, including for a VALUES. TODO: use cstate->miinfo.cur_lineno++ instead of mtstate->miinfo->ntuples --- src/backend/commands/copyfrom.c | 394 +---------------------- src/backend/commands/copyfromparse.c | 10 +- src/backend/executor/execMain.c | 2 +- src/backend/executor/execPartition.c | 2 +- src/backend/executor/nodeModifyTable.c | 196 +++++++++-- src/backend/utils/misc/guc.c | 10 + src/include/commands/copyfrom_internal.h | 5 +- src/include/executor/nodeModifyTable.h | 371 +++++++++++++++++++++ src/include/nodes/execnodes.h | 16 +- src/test/regress/expected/insert.out | 43 +++ src/test/regress/sql/insert.sql | 20 ++ src/tools/pgindent/typedefs.list | 4 +- 12 files changed, 658 insertions(+), 415 deletions(-) diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c index c39cc736ed..8221a2c5d3 100644 --- a/src/backend/commands/copyfrom.c +++ b/src/backend/commands/copyfrom.c @@ -46,54 +46,6 @@ #include "utils/rel.h" #include "utils/snapmgr.h" -/* - * No more than this many tuples per CopyMultiInsertBuffer - * - * Caution: Don't make this too big, as we could end up with this many - * CopyMultiInsertBuffer items stored in CopyMultiInsertInfo's - * multiInsertBuffers list. Increasing this can cause quadratic growth in - * memory requirements during copies into partitioned tables with a large - * number of partitions. - */ -#define MAX_BUFFERED_TUPLES 1000 - -/* - * Flush buffers if there are >= this many bytes, as counted by the input - * size, of tuples stored. - */ -#define MAX_BUFFERED_BYTES 65535 - -/* Trim the list of buffers back down to this number after flushing */ -#define MAX_PARTITION_BUFFERS 32 - -/* Stores multi-insert data related to a single relation in CopyFrom. */ -typedef struct CopyMultiInsertBuffer -{ - TupleTableSlot *slots[MAX_BUFFERED_TUPLES]; /* Array to store tuples */ - ResultRelInfo *resultRelInfo; /* ResultRelInfo for 'relid' */ - BulkInsertState bistate; /* BulkInsertState for this rel */ - int nused; /* number of 'slots' containing tuples */ - uint64 linenos[MAX_BUFFERED_TUPLES]; /* Line # of tuple in copy - * stream */ -} CopyMultiInsertBuffer; - -/* - * Stores one or many CopyMultiInsertBuffers and details about the size and - * number of tuples which are stored in them. This allows multiple buffers to - * exist at once when COPYing into a partitioned table. - */ -typedef struct CopyMultiInsertInfo -{ - List *multiInsertBuffers; /* List of tracked CopyMultiInsertBuffers */ - int bufferedTuples; /* number of tuples buffered over all buffers */ - int bufferedBytes; /* number of bytes from all buffered tuples */ - CopyFromState cstate; /* Copy state for this CopyMultiInsertInfo */ - EState *estate; /* Executor state used for COPY */ - CommandId mycid; /* Command Id used for COPY */ - int ti_options; /* table insert options */ -} CopyMultiInsertInfo; - - /* non-export function prototypes */ static char *limit_printout_length(const char *str); @@ -111,7 +63,7 @@ CopyFromErrorCallback(void *arg) char curlineno_str[32]; snprintf(curlineno_str, sizeof(curlineno_str), UINT64_FORMAT, - cstate->cur_lineno); + cstate->miinfo.cur_lineno); if (cstate->opts.binary) { @@ -206,317 +158,6 @@ limit_printout_length(const char *str) return res; } -/* - * Allocate memory and initialize a new CopyMultiInsertBuffer for this - * ResultRelInfo. - */ -static CopyMultiInsertBuffer * -CopyMultiInsertBufferInit(ResultRelInfo *rri) -{ - CopyMultiInsertBuffer *buffer; - - buffer = (CopyMultiInsertBuffer *) palloc(sizeof(CopyMultiInsertBuffer)); - memset(buffer->slots, 0, sizeof(TupleTableSlot *) * MAX_BUFFERED_TUPLES); - buffer->resultRelInfo = rri; - buffer->bistate = GetBulkInsertState(); - buffer->nused = 0; - - return buffer; -} - -/* - * Make a new buffer for this ResultRelInfo. - */ -static inline void -CopyMultiInsertInfoSetupBuffer(CopyMultiInsertInfo *miinfo, - ResultRelInfo *rri) -{ - CopyMultiInsertBuffer *buffer; - - buffer = CopyMultiInsertBufferInit(rri); - - /* Setup back-link so we can easily find this buffer again */ - rri->ri_CopyMultiInsertBuffer = buffer; - /* Record that we're tracking this buffer */ - miinfo->multiInsertBuffers = lappend(miinfo->multiInsertBuffers, buffer); -} - -/* - * Initialize an already allocated CopyMultiInsertInfo. - * - * If rri is a non-partitioned table then a CopyMultiInsertBuffer is set up - * for that table. - */ -static void -CopyMultiInsertInfoInit(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri, - CopyFromState cstate, EState *estate, CommandId mycid, - int ti_options) -{ - miinfo->multiInsertBuffers = NIL; - miinfo->bufferedTuples = 0; - miinfo->bufferedBytes = 0; - miinfo->cstate = cstate; - miinfo->estate = estate; - miinfo->mycid = mycid; - miinfo->ti_options = ti_options; - - /* - * Only setup the buffer when not dealing with a partitioned table. - * Buffers for partitioned tables will just be setup when we need to send - * tuples their way for the first time. - */ - if (rri->ri_RelationDesc->rd_rel->relkind != RELKIND_PARTITIONED_TABLE) - CopyMultiInsertInfoSetupBuffer(miinfo, rri); -} - -/* - * Returns true if the buffers are full - */ -static inline bool -CopyMultiInsertInfoIsFull(CopyMultiInsertInfo *miinfo) -{ - if (miinfo->bufferedTuples >= MAX_BUFFERED_TUPLES || - miinfo->bufferedBytes >= MAX_BUFFERED_BYTES) - return true; - return false; -} - -/* - * Returns true if we have no buffered tuples - */ -static inline bool -CopyMultiInsertInfoIsEmpty(CopyMultiInsertInfo *miinfo) -{ - return miinfo->bufferedTuples == 0; -} - -/* - * Write the tuples stored in 'buffer' out to the table. - */ -static inline void -CopyMultiInsertBufferFlush(CopyMultiInsertInfo *miinfo, - CopyMultiInsertBuffer *buffer) -{ - MemoryContext oldcontext; - int i; - uint64 save_cur_lineno; - CopyFromState cstate = miinfo->cstate; - EState *estate = miinfo->estate; - CommandId mycid = miinfo->mycid; - int ti_options = miinfo->ti_options; - bool line_buf_valid = cstate->line_buf_valid; - int nused = buffer->nused; - ResultRelInfo *resultRelInfo = buffer->resultRelInfo; - TupleTableSlot **slots = buffer->slots; - - /* - * Print error context information correctly, if one of the operations - * below fail. - */ - cstate->line_buf_valid = false; - save_cur_lineno = cstate->cur_lineno; - - /* - * table_multi_insert may leak memory, so switch to short-lived memory - * context before calling it. - */ - oldcontext = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); - table_multi_insert(resultRelInfo->ri_RelationDesc, - slots, - nused, - mycid, - ti_options, - buffer->bistate); - MemoryContextSwitchTo(oldcontext); - - for (i = 0; i < nused; i++) - { - /* - * If there are any indexes, update them for all the inserted tuples, - * and run AFTER ROW INSERT triggers. - */ - if (resultRelInfo->ri_NumIndices > 0) - { - List *recheckIndexes; - - cstate->cur_lineno = buffer->linenos[i]; - recheckIndexes = - ExecInsertIndexTuples(resultRelInfo, - buffer->slots[i], estate, false, false, - NULL, NIL); - ExecARInsertTriggers(estate, resultRelInfo, - slots[i], recheckIndexes, - cstate->transition_capture); - list_free(recheckIndexes); - } - - /* - * There's no indexes, but see if we need to run AFTER ROW INSERT - * triggers anyway. - */ - else if (resultRelInfo->ri_TrigDesc != NULL && - (resultRelInfo->ri_TrigDesc->trig_insert_after_row || - resultRelInfo->ri_TrigDesc->trig_insert_new_table)) - { - cstate->cur_lineno = buffer->linenos[i]; - ExecARInsertTriggers(estate, resultRelInfo, - slots[i], NIL, cstate->transition_capture); - } - - ExecClearTuple(slots[i]); - } - - /* Mark that all slots are free */ - buffer->nused = 0; - - /* reset cur_lineno and line_buf_valid to what they were */ - cstate->line_buf_valid = line_buf_valid; - cstate->cur_lineno = save_cur_lineno; -} - -/* - * Drop used slots and free member for this buffer. - * - * The buffer must be flushed before cleanup. - */ -static inline void -CopyMultiInsertBufferCleanup(CopyMultiInsertInfo *miinfo, - CopyMultiInsertBuffer *buffer) -{ - int i; - - /* Ensure buffer was flushed */ - Assert(buffer->nused == 0); - - /* Remove back-link to ourself */ - buffer->resultRelInfo->ri_CopyMultiInsertBuffer = NULL; - - FreeBulkInsertState(buffer->bistate); - - /* Since we only create slots on demand, just drop the non-null ones. */ - for (i = 0; i < MAX_BUFFERED_TUPLES && buffer->slots[i] != NULL; i++) - ExecDropSingleTupleTableSlot(buffer->slots[i]); - - table_finish_bulk_insert(buffer->resultRelInfo->ri_RelationDesc, - miinfo->ti_options); - - pfree(buffer); -} - -/* - * Write out all stored tuples in all buffers out to the tables. - * - * Once flushed we also trim the tracked buffers list down to size by removing - * the buffers created earliest first. - * - * Callers should pass 'curr_rri' is the ResultRelInfo that's currently being - * used. When cleaning up old buffers we'll never remove the one for - * 'curr_rri'. - */ -static inline void -CopyMultiInsertInfoFlush(CopyMultiInsertInfo *miinfo, ResultRelInfo *curr_rri) -{ - ListCell *lc; - - foreach(lc, miinfo->multiInsertBuffers) - { - CopyMultiInsertBuffer *buffer = (CopyMultiInsertBuffer *) lfirst(lc); - - CopyMultiInsertBufferFlush(miinfo, buffer); - } - - miinfo->bufferedTuples = 0; - miinfo->bufferedBytes = 0; - - /* - * Trim the list of tracked buffers down if it exceeds the limit. Here we - * remove buffers starting with the ones we created first. It seems less - * likely that these older ones will be needed than the ones that were - * just created. - */ - while (list_length(miinfo->multiInsertBuffers) > MAX_PARTITION_BUFFERS) - { - CopyMultiInsertBuffer *buffer; - - buffer = (CopyMultiInsertBuffer *) linitial(miinfo->multiInsertBuffers); - - /* - * We never want to remove the buffer that's currently being used, so - * if we happen to find that then move it to the end of the list. - */ - if (buffer->resultRelInfo == curr_rri) - { - miinfo->multiInsertBuffers = list_delete_first(miinfo->multiInsertBuffers); - miinfo->multiInsertBuffers = lappend(miinfo->multiInsertBuffers, buffer); - buffer = (CopyMultiInsertBuffer *) linitial(miinfo->multiInsertBuffers); - } - - CopyMultiInsertBufferCleanup(miinfo, buffer); - miinfo->multiInsertBuffers = list_delete_first(miinfo->multiInsertBuffers); - } -} - -/* - * Cleanup allocated buffers and free memory - */ -static inline void -CopyMultiInsertInfoCleanup(CopyMultiInsertInfo *miinfo) -{ - ListCell *lc; - - foreach(lc, miinfo->multiInsertBuffers) - CopyMultiInsertBufferCleanup(miinfo, lfirst(lc)); - - list_free(miinfo->multiInsertBuffers); -} - -/* - * Get the next TupleTableSlot that the next tuple should be stored in. - * - * Callers must ensure that the buffer is not full. - * - * Note: 'miinfo' is unused but has been included for consistency with the - * other functions in this area. - */ -static inline TupleTableSlot * -CopyMultiInsertInfoNextFreeSlot(CopyMultiInsertInfo *miinfo, - ResultRelInfo *rri) -{ - CopyMultiInsertBuffer *buffer = rri->ri_CopyMultiInsertBuffer; - int nused = buffer->nused; - - Assert(buffer != NULL); - Assert(nused < MAX_BUFFERED_TUPLES); - - if (buffer->slots[nused] == NULL) - buffer->slots[nused] = table_slot_create(rri->ri_RelationDesc, NULL); - return buffer->slots[nused]; -} - -/* - * Record the previously reserved TupleTableSlot that was reserved by - * CopyMultiInsertInfoNextFreeSlot as being consumed. - */ -static inline void -CopyMultiInsertInfoStore(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri, - TupleTableSlot *slot, int tuplen, uint64 lineno) -{ - CopyMultiInsertBuffer *buffer = rri->ri_CopyMultiInsertBuffer; - - Assert(buffer != NULL); - Assert(slot == buffer->slots[buffer->nused]); - - /* Store the line number so we can properly report any errors later */ - buffer->linenos[buffer->nused] = lineno; - - /* Record this slot as being used */ - buffer->nused++; - - /* Update how many tuples are stored and their size */ - miinfo->bufferedTuples++; - miinfo->bufferedBytes += tuplen; -} - /* * Copy FROM file to relation. */ @@ -538,7 +179,6 @@ CopyFrom(CopyFromState cstate) int ti_options = 0; /* start with default options for insert */ BulkInsertState bistate = NULL; CopyInsertMethod insertMethod; - CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */ uint64 processed = 0; bool has_before_insert_row_trig; bool has_instead_insert_row_trig; @@ -725,7 +365,7 @@ CopyFrom(CopyFromState cstate) * For partitioned tables we can't support multi-inserts when there * are any statement level insert triggers. It might be possible to * allow partitioned tables with such triggers in the future, but for - * now, CopyMultiInsertInfoFlush expects that any before row insert + * now, MultiInsertInfoFlush expects that any before row insert * and statement level insert triggers are on the same relation. */ insertMethod = CIM_SINGLE; @@ -773,7 +413,8 @@ CopyFrom(CopyFromState cstate) else insertMethod = CIM_MULTI; - CopyMultiInsertInfoInit(&multiInsertInfo, resultRelInfo, cstate, + MultiInsertInfoInit(&cstate->miinfo, resultRelInfo, + cstate->transition_capture, estate, mycid, ti_options); } @@ -836,7 +477,7 @@ CopyFrom(CopyFromState cstate) Assert(resultRelInfo == target_resultRelInfo); Assert(insertMethod == CIM_MULTI); - myslot = CopyMultiInsertInfoNextFreeSlot(&multiInsertInfo, + myslot = MultiInsertInfoNextFreeSlot(&cstate->miinfo, resultRelInfo); } @@ -905,18 +546,18 @@ CopyFrom(CopyFromState cstate) /* Set the multi-insert buffer to use for this partition. */ if (leafpart_use_multi_insert) { - if (resultRelInfo->ri_CopyMultiInsertBuffer == NULL) - CopyMultiInsertInfoSetupBuffer(&multiInsertInfo, + if (resultRelInfo->ri_MultiInsertBuffer == NULL) + MultiInsertInfoSetupBuffer(&cstate->miinfo, resultRelInfo); } else if (insertMethod == CIM_MULTI_CONDITIONAL && - !CopyMultiInsertInfoIsEmpty(&multiInsertInfo)) + !MultiInsertInfoIsEmpty(&cstate->miinfo)) { /* * Flush pending inserts if this partition can't use * batching, so rows are visible to triggers etc. */ - CopyMultiInsertInfoFlush(&multiInsertInfo, resultRelInfo); + MultiInsertInfoFlush(&cstate->miinfo, resultRelInfo); } if (bistate != NULL) @@ -962,7 +603,7 @@ CopyFrom(CopyFromState cstate) /* no other path available for partitioned table */ Assert(insertMethod == CIM_MULTI_CONDITIONAL); - batchslot = CopyMultiInsertInfoNextFreeSlot(&multiInsertInfo, + batchslot = MultiInsertInfoNextFreeSlot(&cstate->miinfo, resultRelInfo); if (map != NULL) @@ -1042,17 +683,17 @@ CopyFrom(CopyFromState cstate) ExecMaterializeSlot(myslot); /* Add this tuple to the tuple buffer */ - CopyMultiInsertInfoStore(&multiInsertInfo, + MultiInsertInfoStore(&cstate->miinfo, resultRelInfo, myslot, cstate->line_buf.len, - cstate->cur_lineno); + cstate->miinfo.cur_lineno); /* * If enough inserts have queued up, then flush all * buffers out to their tables. */ - if (CopyMultiInsertInfoIsFull(&multiInsertInfo)) - CopyMultiInsertInfoFlush(&multiInsertInfo, resultRelInfo); + if (MultiInsertInfoIsFull(&cstate->miinfo)) + MultiInsertInfoFlush(&cstate->miinfo, resultRelInfo); } else { @@ -1113,8 +754,8 @@ CopyFrom(CopyFromState cstate) /* Flush any remaining buffered tuples */ if (insertMethod != CIM_SINGLE) { - if (!CopyMultiInsertInfoIsEmpty(&multiInsertInfo)) - CopyMultiInsertInfoFlush(&multiInsertInfo, NULL); + if (!MultiInsertInfoIsEmpty(&cstate->miinfo)) + MultiInsertInfoFlush(&cstate->miinfo, NULL); } /* Done, clean up */ @@ -1148,7 +789,7 @@ CopyFrom(CopyFromState cstate) /* Tear down the multi-insert buffer data */ if (insertMethod != CIM_SINGLE) - CopyMultiInsertInfoCleanup(&multiInsertInfo); + MultiInsertInfoCleanup(&cstate->miinfo); /* Close all the partitioned tables, leaf partitions, and their indices */ if (proute) @@ -1327,7 +968,6 @@ BeginCopyFrom(ParseState *pstate, cstate->reached_eof = false; cstate->eol_type = EOL_UNKNOWN; cstate->cur_relname = RelationGetRelationName(cstate->rel); - cstate->cur_lineno = 0; cstate->cur_attname = NULL; cstate->cur_attval = NULL; diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c index 4c74067f84..02efad8b31 100644 --- a/src/backend/commands/copyfromparse.c +++ b/src/backend/commands/copyfromparse.c @@ -460,14 +460,14 @@ NextCopyFromRawFields(CopyFromState cstate, char ***fields, int *nfields) Assert(!cstate->opts.binary); /* on input just throw the header line away */ - if (cstate->cur_lineno == 0 && cstate->opts.header_line) + if (cstate->miinfo.cur_lineno == 0 && cstate->opts.header_line) { - cstate->cur_lineno++; + cstate->miinfo.cur_lineno++; if (CopyReadLine(cstate)) return false; /* done */ } - cstate->cur_lineno++; + cstate->miinfo.cur_lineno++; /* Actually read the line into memory here */ done = CopyReadLine(cstate); @@ -608,7 +608,7 @@ NextCopyFrom(CopyFromState cstate, ExprContext *econtext, int16 fld_count; ListCell *cur; - cstate->cur_lineno++; + cstate->miinfo.cur_lineno++; if (!CopyGetInt16(cstate, &fld_count)) { @@ -916,7 +916,7 @@ CopyReadLineText(CopyFromState cstate) * at all --- is cur_lineno a physical or logical count?) */ if (in_quote && c == (cstate->eol_type == EOL_NL ? '\n' : '\r')) - cstate->cur_lineno++; + cstate->miinfo.cur_lineno++; } /* Process \r */ diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c index f4dd47acc7..79dbf0b3e8 100644 --- a/src/backend/executor/execMain.c +++ b/src/backend/executor/execMain.c @@ -1247,7 +1247,7 @@ InitResultRelInfo(ResultRelInfo *resultRelInfo, * ExecInitRoutingInfo */ resultRelInfo->ri_PartitionTupleSlot = NULL; /* ditto */ resultRelInfo->ri_ChildToRootMap = NULL; - resultRelInfo->ri_CopyMultiInsertBuffer = NULL; + resultRelInfo->ri_MultiInsertBuffer = NULL; } /* diff --git a/src/backend/executor/execPartition.c b/src/backend/executor/execPartition.c index 941731a0a9..067bfd11ba 100644 --- a/src/backend/executor/execPartition.c +++ b/src/backend/executor/execPartition.c @@ -993,7 +993,7 @@ ExecInitRoutingInfo(ModifyTableState *mtstate, partRelInfo->ri_FdwRoutine->BeginForeignInsert != NULL) partRelInfo->ri_FdwRoutine->BeginForeignInsert(mtstate, partRelInfo); - partRelInfo->ri_CopyMultiInsertBuffer = NULL; + partRelInfo->ri_MultiInsertBuffer = NULL; /* * Keep track of it in the PartitionTupleRouting->partitions array. diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c index 921e695419..a53cbeeb2c 100644 --- a/src/backend/executor/nodeModifyTable.c +++ b/src/backend/executor/nodeModifyTable.c @@ -43,6 +43,7 @@ #include "access/xact.h" #include "catalog/catalog.h" #include "commands/trigger.h" +#include "commands/copy.h" #include "executor/execPartition.h" #include "executor/executor.h" #include "executor/nodeModifyTable.h" @@ -72,6 +73,8 @@ static TupleTableSlot *ExecPrepareTupleRouting(ModifyTableState *mtstate, ResultRelInfo *targetRelInfo, TupleTableSlot *slot, ResultRelInfo **partRelInfo); +/* guc */ +int bulk_insert_ntuples = 1000; /* * Verify that the tuples to be produced by INSERT or UPDATE match the @@ -389,6 +392,8 @@ ExecInsert(ModifyTableState *mtstate, ModifyTable *node = (ModifyTable *) mtstate->ps.plan; OnConflictAction onconflict = node->onConflictAction; PartitionTupleRouting *proute = mtstate->mt_partition_tuple_routing; + TupleTableSlot *batchslot = NULL; + bool use_multi_insert = false; /* * If the input result relation is a partitioned table, find the leaf @@ -408,6 +413,66 @@ ExecInsert(ModifyTableState *mtstate, resultRelationDesc = resultRelInfo->ri_RelationDesc; + /* Use bulk insert after a threshold number of tuples */ + // XXX: maybe this should only be done if it's not a partitioned table or + // if the partitions don't support miinfo, which uses its own bistates + mtstate->ntuples++; + if (mtstate->bistate == NULL && + mtstate->operation == CMD_INSERT && + mtstate->ntuples > bulk_insert_ntuples && + bulk_insert_ntuples >= 0) + { + elog(DEBUG1, "enabling bulk insert"); + mtstate->bistate = GetBulkInsertState(); + } + + if (!mtstate->miinfo) + { + /* + * If multi-inserts aren't possible for this statement at all, so don't + * check further + */ + } else if (proute == NULL) + { + if (mtstate->miinfo->ntuples++ >= bulk_insert_ntuples && + bulk_insert_ntuples >= 0) + use_multi_insert = true; + } + else + { + /* + * If a partitioned table itself allows multi-insert, and bistate + * indicates we've inserted the threshold number of tuples, check if + * the partition also supports it. + */ + + /* Determine which triggers exist on this partition */ + // XXX copyfrom.c only checks triggers when the partition changes, + // so maybe use_multi_insert should be in mtstate ? + bool has_before_insert_row_trig = (resultRelInfo->ri_TrigDesc && + resultRelInfo->ri_TrigDesc->trig_insert_before_row); + + bool has_instead_insert_row_trig = (resultRelInfo->ri_TrigDesc && + resultRelInfo->ri_TrigDesc->trig_insert_instead_row); + + /* + * Disable multi-inserts when the partition has BEFORE/INSTEAD + * OF triggers, or if the partition is a foreign partition. + * The number of tuples eligible for multi-insert is tracked separately + * from the total number of tuples in case it's not supported for some + * partitions. + */ + if (!has_before_insert_row_trig && + !has_instead_insert_row_trig && + resultRelInfo->ri_FdwRoutine == NULL && + mtstate->miinfo->ntuples++ >= bulk_insert_ntuples && + bulk_insert_ntuples >= 0) + use_multi_insert = true; + } + + if (use_multi_insert && mtstate->miinfo->ntuples - 1 == bulk_insert_ntuples) + elog(DEBUG1, "enabling multi insert"); + /* * BEFORE ROW INSERT Triggers. * @@ -594,7 +659,7 @@ ExecInsert(ModifyTableState *mtstate, table_tuple_insert_speculative(resultRelationDesc, slot, estate->es_output_cid, 0, - NULL, + mtstate->bistate, specToken); /* insert index entries for tuple */ @@ -629,12 +694,39 @@ ExecInsert(ModifyTableState *mtstate, /* Since there was no insertion conflict, we're done */ } + else if (use_multi_insert) + { + if (resultRelInfo->ri_MultiInsertBuffer == NULL) + MultiInsertInfoSetupBuffer(mtstate->miinfo, resultRelInfo); + + batchslot = MultiInsertInfoNextFreeSlot(mtstate->miinfo, resultRelInfo); + ExecCopySlot(batchslot, slot); + + MultiInsertInfoStore(mtstate->miinfo, resultRelInfo, batchslot, 0, 0); // XXX: tuplen/lineno + + if (MultiInsertInfoIsFull(mtstate->miinfo)) + MultiInsertInfoFlush(mtstate->miinfo, resultRelInfo); + } else { + if (proute && mtstate->prevResultRelInfo != resultRelInfo) + { + if (mtstate->bistate) + ReleaseBulkInsertStatePin(mtstate->bistate); + mtstate->prevResultRelInfo = resultRelInfo; + } + + /* + * Flush pending inserts if this partition can't use + * batching, so rows are visible to triggers etc. + */ + if (mtstate->miinfo) + MultiInsertInfoFlush(mtstate->miinfo, resultRelInfo); + /* insert the tuple normally */ table_tuple_insert(resultRelationDesc, slot, estate->es_output_cid, - 0, NULL); + 0, mtstate->bistate); /* insert index entries for tuple */ if (resultRelInfo->ri_NumIndices > 0) @@ -647,32 +739,36 @@ ExecInsert(ModifyTableState *mtstate, if (canSetTag) (estate->es_processed)++; - /* - * If this insert is the result of a partition key update that moved the - * tuple to a new partition, put this row into the transition NEW TABLE, - * if there is one. We need to do this separately for DELETE and INSERT - * because they happen on different tables. - */ - ar_insert_trig_tcs = mtstate->mt_transition_capture; - if (mtstate->operation == CMD_UPDATE && mtstate->mt_transition_capture - && mtstate->mt_transition_capture->tcs_update_new_table) + /* Triggers were already run in the batch insert case */ + if (batchslot == NULL) { - ExecARUpdateTriggers(estate, resultRelInfo, NULL, - NULL, - slot, - NULL, - mtstate->mt_transition_capture); - /* - * We've already captured the NEW TABLE row, so make sure any AR - * INSERT trigger fired below doesn't capture it again. + * If this insert is the result of a partition key update that moved the + * tuple to a new partition, put this row into the transition NEW TABLE, + * if there is one. We need to do this separately for DELETE and INSERT + * because they happen on different tables. */ - ar_insert_trig_tcs = NULL; - } + ar_insert_trig_tcs = mtstate->mt_transition_capture; + if (mtstate->operation == CMD_UPDATE && mtstate->mt_transition_capture + && mtstate->mt_transition_capture->tcs_update_new_table) + { + ExecARUpdateTriggers(estate, resultRelInfo, NULL, + NULL, + slot, + NULL, + mtstate->mt_transition_capture); + + /* + * We've already captured the NEW TABLE row, so make sure any AR + * INSERT trigger fired below doesn't capture it again. + */ + ar_insert_trig_tcs = NULL; + } - /* AFTER ROW INSERT Triggers */ - ExecARInsertTriggers(estate, resultRelInfo, slot, recheckIndexes, - ar_insert_trig_tcs); + /* AFTER ROW INSERT Triggers */ + ExecARInsertTriggers(estate, resultRelInfo, slot, recheckIndexes, + ar_insert_trig_tcs); + } list_free(recheckIndexes); @@ -2229,6 +2325,45 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags) mtstate->mt_arowmarks = (List **) palloc0(sizeof(List *) * nplans); mtstate->mt_nplans = nplans; + mtstate->bistate = NULL; + + /* + * Set miinfo if it can support multi-insert. This is the equivalent of + * CIM_MULTI_* et al in copyfrom.c + */ + + if (operation != CMD_INSERT || + node->onConflictAction != ONCONFLICT_NONE) + mtstate->miinfo = NULL; + else if (mtstate->rootResultRelInfo->ri_TrigDesc != NULL && + (mtstate->rootResultRelInfo->ri_TrigDesc->trig_insert_before_row || + // mtstate->rootResultRelInfo->ri_TrigDesc->trig_insert_after_row || // XXX or any row level triggers at all? + mtstate->rootResultRelInfo->ri_TrigDesc->trig_insert_instead_row)) + /* + * Can't support multi-inserts when there are any BEFORE/INSTEAD OF + * triggers on the table. + */ + mtstate->miinfo = NULL; + else if (node->rootRelation > 0 && + mtstate->rootResultRelInfo->ri_TrigDesc != NULL && + mtstate->rootResultRelInfo->ri_TrigDesc->trig_insert_new_table) + /* + * For partitioned tables we can't support multi-inserts when there + * are any statement level insert triggers. + */ + mtstate->miinfo = NULL; + else if (mtstate->rootResultRelInfo->ri_FdwRoutine != NULL + /* || cstate->volatile_defexprs */ ) + // XXX contain_volatile_functions_not_nextval((Node *) defexpr); + /* Can't support multi-inserts to foreign tables or if there are any */ + mtstate->miinfo = NULL; + else + { + mtstate->miinfo = calloc(sizeof(*mtstate->miinfo), 1); + MultiInsertInfoInit(mtstate->miinfo, mtstate->rootResultRelInfo, + mtstate->mt_transition_capture, + estate, GetCurrentCommandId(true), 0); + } /* set up epqstate with dummy subplan data for the moment */ EvalPlanQualInit(&mtstate->mt_epqstate, estate, NULL, NIL, node->epqParam); @@ -2693,6 +2828,19 @@ ExecEndModifyTable(ModifyTableState *node) resultRelInfo); } + if (node->bistate) + { + FreeBulkInsertState(node->bistate); + table_finish_bulk_insert(node->rootResultRelInfo->ri_RelationDesc, 0); + } + + if (node->miinfo) + { + if (!MultiInsertInfoIsEmpty(node->miinfo)) + MultiInsertInfoFlush(node->miinfo, node->resultRelInfo); // root ? + MultiInsertInfoCleanup(node->miinfo); + } + /* * Close all the partitioned tables, leaf partitions, and their indices * and release the slot used for tuple routing, if set. diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index 17579eeaca..b6f493e0c8 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -54,6 +54,7 @@ #include "libpq/pqformat.h" #include "miscadmin.h" #include "optimizer/cost.h" +#include "executor/nodeModifyTable.h" #include "optimizer/geqo.h" #include "optimizer/optimizer.h" #include "optimizer/paths.h" @@ -3445,6 +3446,15 @@ static struct config_int ConfigureNamesInt[] = NULL, NULL, NULL }, + { + {"bulk_insert_ntuples", PGC_USERSET, CLIENT_CONN_STATEMENT, + gettext_noop("Enable bulk insertions after this number of tuples."), + gettext_noop("A ring buffer of limited size will be used and updates done in batch"), + }, + &bulk_insert_ntuples, + 1000, -1, INT_MAX, + }, + /* End-of-list marker */ { {NULL, 0, 0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h index e37942df39..372414b1c0 100644 --- a/src/include/commands/copyfrom_internal.h +++ b/src/include/commands/copyfrom_internal.h @@ -15,6 +15,7 @@ #define COPYFROM_INTERNAL_H #include "commands/copy.h" +#include "executor/nodeModifyTable.h" #include "commands/trigger.h" /* @@ -92,10 +93,12 @@ typedef struct CopyFromStateData /* these are just for error messages, see CopyFromErrorCallback */ const char *cur_relname; /* table name for error messages */ - uint64 cur_lineno; /* line number for error messages */ const char *cur_attname; /* current att for error messages */ const char *cur_attval; /* current att value for error messages */ + /* For bulk inserts and for error callback */ + MultiInsertInfo miinfo; + /* * Working state */ diff --git a/src/include/executor/nodeModifyTable.h b/src/include/executor/nodeModifyTable.h index 83e2965531..30542a542a 100644 --- a/src/include/executor/nodeModifyTable.h +++ b/src/include/executor/nodeModifyTable.h @@ -13,8 +13,12 @@ #ifndef NODEMODIFYTABLE_H #define NODEMODIFYTABLE_H +#include "commands/trigger.h" +#include "executor/executor.h" // XXX #include "nodes/execnodes.h" +extern PGDLLIMPORT int bulk_insert_ntuples; + extern void ExecComputeStoredGenerated(ResultRelInfo *resultRelInfo, EState *estate, TupleTableSlot *slot, CmdType cmdtype); @@ -23,4 +27,371 @@ extern ModifyTableState *ExecInitModifyTable(ModifyTable *node, EState *estate, extern void ExecEndModifyTable(ModifyTableState *node); extern void ExecReScanModifyTable(ModifyTableState *node); +/* Bulk insert stuff which used to live in copy.c */ + +/* + * No more than this many tuples per MultiInsertBuffer + * + * Caution: Don't make this too big, as we could end up with this many + * MultiInsertBuffer items stored in MultiInsertInfo's + * multiInsertBuffers list. Increasing this can cause quadratic growth in + * memory requirements during copies into partitioned tables with a large + * number of partitions. + */ +#define MAX_BUFFERED_TUPLES 1000 + +/* + * Flush buffers if there are >= this many bytes, as counted by the input + * size, of tuples stored. + */ +#define MAX_BUFFERED_BYTES 65535 + +/* Trim the list of buffers back down to this number after flushing */ +#define MAX_PARTITION_BUFFERS 32 + +/* Stores multi-insert data related to a single relation in CopyFrom. */ +typedef struct MultiInsertBuffer +{ + TupleTableSlot *slots[MAX_BUFFERED_TUPLES]; /* Array to store tuples */ + ResultRelInfo *resultRelInfo; /* ResultRelInfo for 'relid' */ + BulkInsertState bistate; /* BulkInsertState for this rel */ + int nused; /* number of 'slots' containing tuples */ + uint64 linenos[MAX_BUFFERED_TUPLES]; /* Line # of tuple in copy + * stream */ +} MultiInsertBuffer; + +/* + * Stores one or many MultiInsertBuffers and details about the size and + * number of tuples which are stored in them. This allows multiple buffers to + * exist at once when COPY/INSERTing into a partitioned table. + */ +typedef struct MultiInsertInfo +{ + List *multiInsertBuffers; /* List of tracked MultiInsertBuffers */ + int bufferedTuples; /* number of tuples buffered over all buffers */ + int bufferedBytes; /* number of bytes from all buffered tuples */ + TransitionCaptureState *transition_capture; + EState *estate; /* Executor state */ + CommandId mycid; /* Command Id */ + int ti_options; /* table insert options */ + size_t ntuples; /* Number of rows *eligible* for multi-insert */ + + /* Line number for errors in copyfrom.c */ + uint64 cur_lineno; + bool line_buf_valid; +} MultiInsertInfo; + + +/* + * Allocate memory and initialize a new MultiInsertBuffer for this + * ResultRelInfo. + */ +static MultiInsertBuffer * +MultiInsertBufferInit(ResultRelInfo *rri) +{ + MultiInsertBuffer *buffer; + + buffer = (MultiInsertBuffer *) palloc(sizeof(MultiInsertBuffer)); + memset(buffer->slots, 0, sizeof(TupleTableSlot *) * MAX_BUFFERED_TUPLES); + buffer->resultRelInfo = rri; + buffer->bistate = GetBulkInsertState(); + buffer->nused = 0; + + return buffer; +} + +/* + * Make a new buffer for this ResultRelInfo. + */ +static inline void +MultiInsertInfoSetupBuffer(MultiInsertInfo *miinfo, + ResultRelInfo *rri) +{ + MultiInsertBuffer *buffer; + + buffer = MultiInsertBufferInit(rri); + + /* Setup back-link so we can easily find this buffer again */ + rri->ri_MultiInsertBuffer = buffer; + /* Record that we're tracking this buffer */ + miinfo->multiInsertBuffers = lappend(miinfo->multiInsertBuffers, buffer); +} + +/* + * Initialize an already allocated MultiInsertInfo. + * + * If rri is a non-partitioned table then a MultiInsertBuffer is set up + * for that table. + */ +static inline void +MultiInsertInfoInit(MultiInsertInfo *miinfo, ResultRelInfo *rri, + TransitionCaptureState *transition_capture, + EState *estate, CommandId mycid, int ti_options) +{ + miinfo->multiInsertBuffers = NIL; + miinfo->bufferedTuples = 0; + miinfo->bufferedBytes = 0; + miinfo->transition_capture = transition_capture; + miinfo->estate = estate; + miinfo->mycid = mycid; + miinfo->ti_options = ti_options; + miinfo->cur_lineno = 0; + + /* + * Only setup the buffer when not dealing with a partitioned table. + * Buffers for partitioned tables will just be setup when we need to send + * tuples their way for the first time. + */ + if (rri->ri_RelationDesc->rd_rel->relkind != RELKIND_PARTITIONED_TABLE) + MultiInsertInfoSetupBuffer(miinfo, rri); +} + +/* + * Returns true if the buffers are full + */ +static inline bool +MultiInsertInfoIsFull(MultiInsertInfo *miinfo) +{ + if (miinfo->bufferedTuples >= MAX_BUFFERED_TUPLES || + miinfo->bufferedBytes >= MAX_BUFFERED_BYTES) + return true; + return false; +} + +/* + * Returns true if we have no buffered tuples + */ +static inline bool +MultiInsertInfoIsEmpty(MultiInsertInfo *miinfo) +{ + return miinfo->bufferedTuples == 0; +} + +/* + * Write the tuples stored in 'buffer' out to the table. + */ +static inline void +MultiInsertBufferFlush(MultiInsertInfo *miinfo, + MultiInsertBuffer *buffer) +{ + MemoryContext oldcontext; + int i; + uint64 save_cur_lineno; + EState *estate = miinfo->estate; + CommandId mycid = miinfo->mycid; + int ti_options = miinfo->ti_options; + bool line_buf_valid = miinfo->line_buf_valid; + int nused = buffer->nused; + ResultRelInfo *resultRelInfo = buffer->resultRelInfo; + TupleTableSlot **slots = buffer->slots; + + /* + * Print error context information correctly, if one of the operations + * below fail. + */ + miinfo->line_buf_valid = false; + save_cur_lineno = miinfo->cur_lineno; + + /* + * table_multi_insert may leak memory, so switch to short-lived memory + * context before calling it. + */ + oldcontext = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); // XXX requires executor.h + table_multi_insert(resultRelInfo->ri_RelationDesc, + slots, + nused, + mycid, + ti_options, + buffer->bistate); + MemoryContextSwitchTo(oldcontext); + + for (i = 0; i < nused; i++) + { + /* + * If there are any indexes, update them for all the inserted tuples, + * and run AFTER ROW INSERT triggers. + */ + if (resultRelInfo->ri_NumIndices > 0) + { + List *recheckIndexes; + + miinfo->cur_lineno = buffer->linenos[i]; + recheckIndexes = + ExecInsertIndexTuples(resultRelInfo, + buffer->slots[i], estate, false, false, + NULL, NIL); + + ExecARInsertTriggers(estate, resultRelInfo, + slots[i], recheckIndexes, + miinfo->transition_capture); + list_free(recheckIndexes); + } + + /* + * There's no indexes, but see if we need to run AFTER ROW INSERT + * triggers anyway. + */ + else if (resultRelInfo->ri_TrigDesc != NULL && + (resultRelInfo->ri_TrigDesc->trig_insert_after_row || + resultRelInfo->ri_TrigDesc->trig_insert_new_table)) + { + miinfo->cur_lineno = buffer->linenos[i]; + ExecARInsertTriggers(estate, resultRelInfo, + slots[i], NIL, miinfo->transition_capture); + } + + ExecClearTuple(slots[i]); + } + + /* Mark that all slots are free */ + buffer->nused = 0; + + /* reset cur_lineno and line_buf_valid to what they were */ + miinfo->line_buf_valid = line_buf_valid; + miinfo->cur_lineno = save_cur_lineno; +} + +/* + * Drop used slots and free member for this buffer. + * + * The buffer must be flushed before cleanup. + */ +static inline void +MultiInsertBufferCleanup(MultiInsertInfo *miinfo, + MultiInsertBuffer *buffer) +{ + int i; + + /* Ensure buffer was flushed */ + Assert(buffer->nused == 0); + + /* Remove back-link to ourself */ + buffer->resultRelInfo->ri_MultiInsertBuffer = NULL; + + FreeBulkInsertState(buffer->bistate); + + /* Since we only create slots on demand, just drop the non-null ones. */ + for (i = 0; i < MAX_BUFFERED_TUPLES && buffer->slots[i] != NULL; i++) + ExecDropSingleTupleTableSlot(buffer->slots[i]); + + table_finish_bulk_insert(buffer->resultRelInfo->ri_RelationDesc, + miinfo->ti_options); + + pfree(buffer); +} + +/* + * Write out all stored tuples in all buffers out to the tables. + * + * Once flushed we also trim the tracked buffers list down to size by removing + * the buffers created earliest first. + * + * Callers should pass 'curr_rri' is the ResultRelInfo that's currently being + * used. When cleaning up old buffers we'll never remove the one for + * 'curr_rri'. + */ +static inline void +MultiInsertInfoFlush(MultiInsertInfo *miinfo, ResultRelInfo *curr_rri) +{ + ListCell *lc; + + foreach(lc, miinfo->multiInsertBuffers) + { + MultiInsertBuffer *buffer = (MultiInsertBuffer *) lfirst(lc); + + MultiInsertBufferFlush(miinfo, buffer); + } + + miinfo->bufferedTuples = 0; + miinfo->bufferedBytes = 0; + + /* + * Trim the list of tracked buffers down if it exceeds the limit. Here we + * remove buffers starting with the ones we created first. It seems less + * likely that these older ones will be needed than the ones that were + * just created. + */ + while (list_length(miinfo->multiInsertBuffers) > MAX_PARTITION_BUFFERS) + { + MultiInsertBuffer *buffer; + + buffer = (MultiInsertBuffer *) linitial(miinfo->multiInsertBuffers); + + /* + * We never want to remove the buffer that's currently being used, so + * if we happen to find that then move it to the end of the list. + */ + if (buffer->resultRelInfo == curr_rri) + { + miinfo->multiInsertBuffers = list_delete_first(miinfo->multiInsertBuffers); + miinfo->multiInsertBuffers = lappend(miinfo->multiInsertBuffers, buffer); + buffer = (MultiInsertBuffer *) linitial(miinfo->multiInsertBuffers); + } + + MultiInsertBufferCleanup(miinfo, buffer); + miinfo->multiInsertBuffers = list_delete_first(miinfo->multiInsertBuffers); + } +} + +/* + * Cleanup allocated buffers and free memory + */ +static inline void +MultiInsertInfoCleanup(MultiInsertInfo *miinfo) +{ + ListCell *lc; + + foreach(lc, miinfo->multiInsertBuffers) + MultiInsertBufferCleanup(miinfo, lfirst(lc)); + + list_free(miinfo->multiInsertBuffers); +} + +/* + * Get the next TupleTableSlot that the next tuple should be stored in. + * + * Callers must ensure that the buffer is not full. + * + * Note: 'miinfo' is unused but has been included for consistency with the + * other functions in this area. + */ +static inline TupleTableSlot * +MultiInsertInfoNextFreeSlot(MultiInsertInfo *miinfo, + ResultRelInfo *rri) +{ + MultiInsertBuffer *buffer = rri->ri_MultiInsertBuffer; + int nused = buffer->nused; + + Assert(buffer != NULL); + Assert(nused < MAX_BUFFERED_TUPLES); + + if (buffer->slots[nused] == NULL) + buffer->slots[nused] = table_slot_create(rri->ri_RelationDesc, NULL); + return buffer->slots[nused]; +} + +/* + * Record the previously reserved TupleTableSlot that was reserved by + * MultiInsertInfoNextFreeSlot as being consumed. + */ +static inline void +MultiInsertInfoStore(MultiInsertInfo *miinfo, ResultRelInfo *rri, + TupleTableSlot *slot, int tuplen, uint64 lineno) +{ + MultiInsertBuffer *buffer = rri->ri_MultiInsertBuffer; + + Assert(buffer != NULL); + Assert(slot == buffer->slots[buffer->nused]); + + /* Store the line number so we can properly report any errors later */ + buffer->linenos[buffer->nused] = lineno; + + /* Record this slot as being used */ + buffer->nused++; + + /* Update how many tuples are stored and their size */ + miinfo->bufferedTuples++; + miinfo->bufferedBytes += tuplen; +} + #endif /* NODEMODIFYTABLE_H */ diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index 48c3f570fa..8c7ca6627d 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -14,6 +14,7 @@ #ifndef EXECNODES_H #define EXECNODES_H +#include "access/heapam.h" #include "access/tupconvert.h" #include "executor/instrument.h" #include "fmgr.h" @@ -32,6 +33,9 @@ #include "utils/tuplesort.h" #include "utils/tuplestore.h" +/* This would be a circular inclusion */ +// #include "executor/nodeModifyTable.h" + struct PlanState; /* forward references in this file */ struct ParallelHashJoinState; struct ExecRowMark; @@ -39,8 +43,8 @@ struct ExprState; struct ExprContext; struct RangeTblEntry; /* avoid including parsenodes.h here */ struct ExprEvalStep; /* avoid including execExpr.h everywhere */ -struct CopyMultiInsertBuffer; - +// struct MultiInsertBuffer; +// struct MultiInsertInfo; /* ---------------- * ExprState node @@ -498,8 +502,8 @@ typedef struct ResultRelInfo */ TupleConversionMap *ri_ChildToRootMap; - /* for use by copyfrom.c when performing multi-inserts */ - struct CopyMultiInsertBuffer *ri_CopyMultiInsertBuffer; + /* for use by copyfrom.c/modifyTable when performing multi-inserts */ + struct MultiInsertBuffer *ri_MultiInsertBuffer; } ResultRelInfo; /* ---------------- @@ -1165,6 +1169,10 @@ typedef struct ModifyTableState List **mt_arowmarks; /* per-subplan ExecAuxRowMark lists */ EPQState mt_epqstate; /* for evaluating EvalPlanQual rechecks */ bool fireBSTriggers; /* do we need to fire stmt triggers? */ + BulkInsertState bistate; /* state for bulk insert like INSERT SELECT, when miinfo cannot be used */ + ResultRelInfo *prevResultRelInfo; /* last child inserted with bistate */ + struct MultiInsertInfo *miinfo; + size_t ntuples; /* Number of tuples inserted; */ /* * Slot for storing tuples in the root partitioned table's rowtype during diff --git a/src/test/regress/expected/insert.out b/src/test/regress/expected/insert.out index da50ee3b67..bc4c1a4fc2 100644 --- a/src/test/regress/expected/insert.out +++ b/src/test/regress/expected/insert.out @@ -462,6 +462,49 @@ Partitions: part_aa_bb FOR VALUES IN ('aa', 'bb'), part_xx_yy FOR VALUES IN ('xx', 'yy'), PARTITIONED, part_default DEFAULT, PARTITIONED +-- bulk inserts +truncate hash_parted; +begin; +create index on hash_parted(a); +-- make sure small inserts are flushed +insert into hash_parted values(11); +insert into hpart0 values(12); +select * from hash_parted; + a +---- + 12 + 11 +(2 rows) + +-- exercise bulk insert to partitions +SET client_min_messages=debug1; +insert into hash_parted select generate_series(1,9999); +DEBUG: enabling bulk insert +DEBUG: enabling multi insert +RESET client_min_messages; +select count(1) from hash_parted; + count +------- + 10001 +(1 row) + +commit; +-- test that index was updated +vacuum analyze hash_parted; +explain(costs off) +select * from hash_parted where a=13; + QUERY PLAN +---------------------------------------------------------- + Index Only Scan using hpart1_a_idx on hpart1 hash_parted + Index Cond: (a = 13) +(2 rows) + +select * from hash_parted where a=13; + a +---- + 13 +(1 row) + -- cleanup drop table range_parted, list_parted; drop table hash_parted; diff --git a/src/test/regress/sql/insert.sql b/src/test/regress/sql/insert.sql index 963faa1614..a74eb3826a 100644 --- a/src/test/regress/sql/insert.sql +++ b/src/test/regress/sql/insert.sql @@ -280,6 +280,26 @@ from hash_parted order by part; -- partitions \d+ list_parted +-- bulk inserts +truncate hash_parted; +begin; +create index on hash_parted(a); +-- make sure small inserts are flushed +insert into hash_parted values(11); +insert into hpart0 values(12); +select * from hash_parted; +-- exercise bulk insert to partitions +SET client_min_messages=debug1; +insert into hash_parted select generate_series(1,9999); +RESET client_min_messages; +select count(1) from hash_parted; +commit; +-- test that index was updated +vacuum analyze hash_parted; +explain(costs off) +select * from hash_parted where a=13; +select * from hash_parted where a=13; + -- cleanup drop table range_parted, list_parted; drop table hash_parted; diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 943142ced8..2274f78843 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -423,8 +423,6 @@ ConvertRowtypeExpr CookedConstraint CopyDest CopyInsertMethod -CopyMultiInsertBuffer -CopyMultiInsertInfo CopyState CopyStateData CopyStmt @@ -1401,6 +1399,8 @@ ModifyTableState MorphOpaque MsgType MultiAssignRef +MultiInsertBuffer +MultiInsertInfo MultiSortSupport MultiSortSupportData MultiXactId -- 2.17.0 --3yNHWXBV/QO9xKNm Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v9-0002-WIP-Check-for-volatile-defaults.patch" ^ permalink raw reply [nested|flat] 9+ messages in thread
* pg_recvlogical prints bogus error when interrupted @ 2022-10-19 21:39 Andres Freund <[email protected]> 2022-10-20 07:58 ` Re: pg_recvlogical prints bogus error when interrupted Bharath Rupireddy <[email protected]> 0 siblings, 1 reply; 9+ messages in thread From: Andres Freund @ 2022-10-19 21:39 UTC (permalink / raw) To: pgsql-hackers Hi, While reviewing https://postgr.es/m/CAD21AoBe2o2D%3Dxyycsxw2bQOD%3DzPj7ETuJ5VYGN%3DdpoTiCMRJQ%40mail.gmail.com I noticed that pg_recvlogical prints "pg_recvlogical: error: unexpected termination of replication stream: " when signalled with SIGINT/TERM. Oddly enough, that looks to have "always" been the case, even though clearly the code tried to make provisions for a different outcome. It looks to me like all that's needed is to gate the block printing the message with an !time_to_abort. I also then noticed that we don't fsync the output file in cases of errors - that seems wrong to me? Looks to me like that block should be moved till after the error:? Greetings, Andres Freund ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: pg_recvlogical prints bogus error when interrupted 2022-10-19 21:39 pg_recvlogical prints bogus error when interrupted Andres Freund <[email protected]> @ 2022-10-20 07:58 ` Bharath Rupireddy <[email protected]> 2022-10-21 02:21 ` Re: pg_recvlogical prints bogus error when interrupted Kyotaro Horiguchi <[email protected]> 0 siblings, 1 reply; 9+ messages in thread From: Bharath Rupireddy @ 2022-10-20 07:58 UTC (permalink / raw) To: Andres Freund <[email protected]>; +Cc: pgsql-hackers On Thu, Oct 20, 2022 at 3:10 AM Andres Freund <[email protected]> wrote: > > Hi, > > While reviewing > https://postgr.es/m/CAD21AoBe2o2D%3Dxyycsxw2bQOD%3DzPj7ETuJ5VYGN%3DdpoTiCMRJQ%40mail.gmail.com > I noticed that pg_recvlogical prints > "pg_recvlogical: error: unexpected termination of replication stream: " > > when signalled with SIGINT/TERM. > > Oddly enough, that looks to have "always" been the case, even though clearly > the code tried to make provisions for a different outcome. > > > It looks to me like all that's needed is to gate the block printing the > message with an !time_to_abort. +1. How about emitting a message like its friend pg_receivewal, like the attached patch? > I also then noticed that we don't fsync the output file in cases of errors - > that seems wrong to me? Looks to me like that block should be moved till after > the error:? How about something like the attached patch? -- Bharath Rupireddy PostgreSQL Contributors Team RDS Open Source Databases Amazon Web Services: https://aws.amazon.com Attachments: [application/octet-stream] v1-0001-pg_recvlogical-fixes.patch (1.2K, ../../CALj2ACUAd==wVcQxbkhxAmoegs2g_b+OX8moL68NzjtnFeoM4A@mail.gmail.com/2-v1-0001-pg_recvlogical-fixes.patch) download | inline diff: From bfbc76e6a85b40e3b9000e22d1c53d06e8799a2a Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Thu, 20 Oct 2022 07:57:04 +0000 Subject: [PATCH v1] pg_recvlogical fixes --- src/bin/pg_basebackup/pg_recvlogical.c | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/src/bin/pg_basebackup/pg_recvlogical.c b/src/bin/pg_basebackup/pg_recvlogical.c index 5f2e6af445..4ce4ee648c 100644 --- a/src/bin/pg_basebackup/pg_recvlogical.c +++ b/src/bin/pg_basebackup/pg_recvlogical.c @@ -614,7 +614,14 @@ StreamLogicalLog(void) res = PQgetResult(conn); } - if (PQresultStatus(res) != PGRES_COMMAND_OK) + + if (time_to_abort) + { + if (verbose) + pg_log_info("received interrupt signal, exiting"); + goto error; + } + else if (PQresultStatus(res) != PGRES_COMMAND_OK) { pg_log_error("unexpected termination of replication stream: %s", PQresultErrorMessage(res)); @@ -634,6 +641,14 @@ StreamLogicalLog(void) } outfd = -1; error: + if (outfd != -1) + { + TimestampTz t; + + t = feGetCurrentTimestamp(); + /* no need to jump to error on failure here, we're finishing anyway */ + OutputFsync(t); + } if (copybuf != NULL) { PQfreemem(copybuf); -- 2.34.1 ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: pg_recvlogical prints bogus error when interrupted 2022-10-19 21:39 pg_recvlogical prints bogus error when interrupted Andres Freund <[email protected]> 2022-10-20 07:58 ` Re: pg_recvlogical prints bogus error when interrupted Bharath Rupireddy <[email protected]> @ 2022-10-21 02:21 ` Kyotaro Horiguchi <[email protected]> 2022-10-24 02:45 ` Re: pg_recvlogical prints bogus error when interrupted Bharath Rupireddy <[email protected]> 0 siblings, 1 reply; 9+ messages in thread From: Kyotaro Horiguchi @ 2022-10-21 02:21 UTC (permalink / raw) To: [email protected]; +Cc: [email protected]; pgsql-hackers At Thu, 20 Oct 2022 13:28:45 +0530, Bharath Rupireddy <[email protected]> wrote in > On Thu, Oct 20, 2022 at 3:10 AM Andres Freund <[email protected]> wrote: > > > > Hi, > > > > While reviewing > > https://postgr.es/m/CAD21AoBe2o2D%3Dxyycsxw2bQOD%3DzPj7ETuJ5VYGN%3DdpoTiCMRJQ%40mail.gmail.com > > I noticed that pg_recvlogical prints > > "pg_recvlogical: error: unexpected termination of replication stream: " > > > > when signalled with SIGINT/TERM. > > > > Oddly enough, that looks to have "always" been the case, even though clearly > > the code tried to make provisions for a different outcome. > > > > > > It looks to me like all that's needed is to gate the block printing the > > message with an !time_to_abort. +1 > +1. How about emitting a message like its friend pg_receivewal, like > the attached patch? I'm not a fan of treating SIGINT as an error in this case. It calls prepareToTerminate() when time_to_abort and everything goes fine after then. So I think we should do the same thing after receiving an interrupt. This also does file-sync naturally as a part of normal shutdown. I'm also not a fan of doing fsync at error. > > I also then noticed that we don't fsync the output file in cases of errors - > > that seems wrong to me? Looks to me like that block should be moved till after > > the error:? > > How about something like the attached patch? regards. -- Kyotaro Horiguchi NTT Open Source Software Center diff --git a/src/bin/pg_basebackup/pg_recvlogical.c b/src/bin/pg_basebackup/pg_recvlogical.c index 5f2e6af445..e33c204df0 100644 --- a/src/bin/pg_basebackup/pg_recvlogical.c +++ b/src/bin/pg_basebackup/pg_recvlogical.c @@ -55,6 +55,7 @@ static const char *plugin = "test_decoding"; /* Global State */ static int outfd = -1; static volatile sig_atomic_t time_to_abort = false; +static volatile sig_atomic_t interrupted = false; static volatile sig_atomic_t output_reopen = false; static bool output_isfile; static TimestampTz output_last_fsync = -1; @@ -206,6 +207,7 @@ StreamLogicalLog(void) char *copybuf = NULL; TimestampTz last_status = -1; int i; + XLogRecPtr cur_record_lsn = InvalidXLogRecPtr; PQExpBuffer query; output_written_lsn = InvalidXLogRecPtr; @@ -275,7 +277,6 @@ StreamLogicalLog(void) int bytes_written; TimestampTz now; int hdr_len; - XLogRecPtr cur_record_lsn = InvalidXLogRecPtr; if (copybuf != NULL) { @@ -487,7 +488,7 @@ StreamLogicalLog(void) if (endposReached) { - prepareToTerminate(conn, endpos, true, InvalidXLogRecPtr); + cur_record_lsn = InvalidXLogRecPtr; time_to_abort = true; break; } @@ -527,7 +528,6 @@ StreamLogicalLog(void) */ if (!flushAndSendFeedback(conn, &now)) goto error; - prepareToTerminate(conn, endpos, false, cur_record_lsn); time_to_abort = true; break; } @@ -572,12 +572,14 @@ StreamLogicalLog(void) /* endpos was exactly the record we just processed, we're done */ if (!flushAndSendFeedback(conn, &now)) goto error; - prepareToTerminate(conn, endpos, false, cur_record_lsn); time_to_abort = true; break; } } + if (time_to_abort) + prepareToTerminate(conn, endpos, false, cur_record_lsn); + res = PQgetResult(conn); if (PQresultStatus(res) == PGRES_COPY_OUT) { @@ -657,6 +659,7 @@ static void sigexit_handler(SIGNAL_ARGS) { time_to_abort = true; + interrupted = true; } /* @@ -1031,6 +1034,8 @@ prepareToTerminate(PGconn *conn, XLogRecPtr endpos, bool keepalive, XLogRecPtr l if (keepalive) pg_log_info("end position %X/%X reached by keepalive", LSN_FORMAT_ARGS(endpos)); + else if (interrupted) + pg_log_info("interrupted after %X/%X", LSN_FORMAT_ARGS(lsn)); else pg_log_info("end position %X/%X reached by WAL record at %X/%X", LSN_FORMAT_ARGS(endpos), LSN_FORMAT_ARGS(lsn)); Attachments: [text/plain] pg_recvlogical_graceful_interrupt.txt (2.3K, ../../[email protected]/2-pg_recvlogical_graceful_interrupt.txt) download | inline diff: diff --git a/src/bin/pg_basebackup/pg_recvlogical.c b/src/bin/pg_basebackup/pg_recvlogical.c index 5f2e6af445..e33c204df0 100644 --- a/src/bin/pg_basebackup/pg_recvlogical.c +++ b/src/bin/pg_basebackup/pg_recvlogical.c @@ -55,6 +55,7 @@ static const char *plugin = "test_decoding"; /* Global State */ static int outfd = -1; static volatile sig_atomic_t time_to_abort = false; +static volatile sig_atomic_t interrupted = false; static volatile sig_atomic_t output_reopen = false; static bool output_isfile; static TimestampTz output_last_fsync = -1; @@ -206,6 +207,7 @@ StreamLogicalLog(void) char *copybuf = NULL; TimestampTz last_status = -1; int i; + XLogRecPtr cur_record_lsn = InvalidXLogRecPtr; PQExpBuffer query; output_written_lsn = InvalidXLogRecPtr; @@ -275,7 +277,6 @@ StreamLogicalLog(void) int bytes_written; TimestampTz now; int hdr_len; - XLogRecPtr cur_record_lsn = InvalidXLogRecPtr; if (copybuf != NULL) { @@ -487,7 +488,7 @@ StreamLogicalLog(void) if (endposReached) { - prepareToTerminate(conn, endpos, true, InvalidXLogRecPtr); + cur_record_lsn = InvalidXLogRecPtr; time_to_abort = true; break; } @@ -527,7 +528,6 @@ StreamLogicalLog(void) */ if (!flushAndSendFeedback(conn, &now)) goto error; - prepareToTerminate(conn, endpos, false, cur_record_lsn); time_to_abort = true; break; } @@ -572,12 +572,14 @@ StreamLogicalLog(void) /* endpos was exactly the record we just processed, we're done */ if (!flushAndSendFeedback(conn, &now)) goto error; - prepareToTerminate(conn, endpos, false, cur_record_lsn); time_to_abort = true; break; } } + if (time_to_abort) + prepareToTerminate(conn, endpos, false, cur_record_lsn); + res = PQgetResult(conn); if (PQresultStatus(res) == PGRES_COPY_OUT) { @@ -657,6 +659,7 @@ static void sigexit_handler(SIGNAL_ARGS) { time_to_abort = true; + interrupted = true; } /* @@ -1031,6 +1034,8 @@ prepareToTerminate(PGconn *conn, XLogRecPtr endpos, bool keepalive, XLogRecPtr l if (keepalive) pg_log_info("end position %X/%X reached by keepalive", LSN_FORMAT_ARGS(endpos)); + else if (interrupted) + pg_log_info("interrupted after %X/%X", LSN_FORMAT_ARGS(lsn)); else pg_log_info("end position %X/%X reached by WAL record at %X/%X", LSN_FORMAT_ARGS(endpos), LSN_FORMAT_ARGS(lsn)); ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: pg_recvlogical prints bogus error when interrupted 2022-10-19 21:39 pg_recvlogical prints bogus error when interrupted Andres Freund <[email protected]> 2022-10-20 07:58 ` Re: pg_recvlogical prints bogus error when interrupted Bharath Rupireddy <[email protected]> 2022-10-21 02:21 ` Re: pg_recvlogical prints bogus error when interrupted Kyotaro Horiguchi <[email protected]> @ 2022-10-24 02:45 ` Bharath Rupireddy <[email protected]> 2022-10-27 11:26 ` Re: pg_recvlogical prints bogus error when interrupted Bharath Rupireddy <[email protected]> 2022-10-27 23:11 ` Re: pg_recvlogical prints bogus error when interrupted Andres Freund <[email protected]> 0 siblings, 2 replies; 9+ messages in thread From: Bharath Rupireddy @ 2022-10-24 02:45 UTC (permalink / raw) To: Kyotaro Horiguchi <[email protected]>; +Cc: [email protected]; pgsql-hackers On Fri, Oct 21, 2022 at 7:52 AM Kyotaro Horiguchi <[email protected]> wrote: > > > +1. How about emitting a message like its friend pg_receivewal, like > > the attached patch? > > I'm not a fan of treating SIGINT as an error in this case. It calls > prepareToTerminate() when time_to_abort and everything goes fine after > then. So I think we should do the same thing after receiving an > interrupt. This also does file-sync naturally as a part of normal > shutdown. I'm also not a fan of doing fsync at error. I think the pg_recvlogical can gracefully exit on both SIGINT and SIGTERM to keep things simple. > > > I also then noticed that we don't fsync the output file in cases of errors - > > > that seems wrong to me? Looks to me like that block should be moved till after > > > the error:? > > > > How about something like the attached patch? The attached patch (pg_recvlogical_graceful_interrupt.text) has a couple of problems, I believe. We're losing prepareToTerminate() with keepalive true and we're not skipping pg_log_error("unexpected termination of replication stream: %s" upon interrupt, after all we're here discussing how to avoid it. I came up with the attached v2 patch, please have a look. -- Bharath Rupireddy PostgreSQL Contributors Team RDS Open Source Databases Amazon Web Services: https://aws.amazon.com Attachments: [application/x-patch] v2-0001-pg_recvlogical-fixes.patch (1.9K, ../../CALj2ACWrLuSpXT87HA91Y=1_SefBM2uL9BCffm2VPretBCfgWQ@mail.gmail.com/2-v2-0001-pg_recvlogical-fixes.patch) download | inline diff: From 56e25373796b114254f5995701b07b05381f28ef Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Sat, 22 Oct 2022 08:35:16 +0000 Subject: [PATCH v2] pg_recvlogical fixes --- src/bin/pg_basebackup/pg_recvlogical.c | 25 ++++++++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) diff --git a/src/bin/pg_basebackup/pg_recvlogical.c b/src/bin/pg_basebackup/pg_recvlogical.c index 5f2e6af445..849e9d9071 100644 --- a/src/bin/pg_basebackup/pg_recvlogical.c +++ b/src/bin/pg_basebackup/pg_recvlogical.c @@ -54,7 +54,8 @@ static const char *plugin = "test_decoding"; /* Global State */ static int outfd = -1; -static volatile sig_atomic_t time_to_abort = false; +static bool time_to_abort = false; +static volatile sig_atomic_t ready_to_exit = false; static volatile sig_atomic_t output_reopen = false; static bool output_isfile; static TimestampTz output_last_fsync = -1; @@ -283,6 +284,23 @@ StreamLogicalLog(void) copybuf = NULL; } + /* When we get SIGINT/SIGTERM, we exit */ + if (ready_to_exit) + { + /* + * Try informing the server about our exit, but don't wait around + * or retry on failure. + */ + (void) PQputCopyEnd(conn, NULL); + (void) PQflush(conn); + time_to_abort = ready_to_exit; + + if (verbose) + pg_log_info("received interrupt signal, exiting"); + + break; + } + /* * Potentially send a status message to the primary. */ @@ -614,7 +632,8 @@ StreamLogicalLog(void) res = PQgetResult(conn); } - if (PQresultStatus(res) != PGRES_COMMAND_OK) + if (!ready_to_exit && + PQresultStatus(res) != PGRES_COMMAND_OK) { pg_log_error("unexpected termination of replication stream: %s", PQresultErrorMessage(res)); @@ -656,7 +675,7 @@ error: static void sigexit_handler(SIGNAL_ARGS) { - time_to_abort = true; + ready_to_exit = true; } /* -- 2.34.1 ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: pg_recvlogical prints bogus error when interrupted 2022-10-19 21:39 pg_recvlogical prints bogus error when interrupted Andres Freund <[email protected]> 2022-10-20 07:58 ` Re: pg_recvlogical prints bogus error when interrupted Bharath Rupireddy <[email protected]> 2022-10-21 02:21 ` Re: pg_recvlogical prints bogus error when interrupted Kyotaro Horiguchi <[email protected]> 2022-10-24 02:45 ` Re: pg_recvlogical prints bogus error when interrupted Bharath Rupireddy <[email protected]> @ 2022-10-27 11:26 ` Bharath Rupireddy <[email protected]> 1 sibling, 0 replies; 9+ messages in thread From: Bharath Rupireddy @ 2022-10-27 11:26 UTC (permalink / raw) To: Kyotaro Horiguchi <[email protected]>; +Cc: [email protected]; pgsql-hackers On Mon, Oct 24, 2022 at 8:15 AM Bharath Rupireddy <[email protected]> wrote: > > On Fri, Oct 21, 2022 at 7:52 AM Kyotaro Horiguchi > <[email protected]> wrote: > > > > > +1. How about emitting a message like its friend pg_receivewal, like > > > the attached patch? > > > > I'm not a fan of treating SIGINT as an error in this case. It calls > > prepareToTerminate() when time_to_abort and everything goes fine after > > then. So I think we should do the same thing after receiving an > > interrupt. This also does file-sync naturally as a part of normal > > shutdown. I'm also not a fan of doing fsync at error. > > I think the pg_recvlogical can gracefully exit on both SIGINT and > SIGTERM to keep things simple. > > > > > I also then noticed that we don't fsync the output file in cases of errors - > > > > that seems wrong to me? Looks to me like that block should be moved till after > > > > the error:? > > > > > > How about something like the attached patch? > > The attached patch (pg_recvlogical_graceful_interrupt.text) has a > couple of problems, I believe. We're losing prepareToTerminate() with > keepalive true and we're not skipping pg_log_error("unexpected > termination of replication stream: %s" upon interrupt, after all we're > here discussing how to avoid it. > > I came up with the attached v2 patch, please have a look. FWIW, I added it to CF - https://commitfest.postgresql.org/40/3966/. -- Bharath Rupireddy PostgreSQL Contributors Team RDS Open Source Databases Amazon Web Services: https://aws.amazon.com ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: pg_recvlogical prints bogus error when interrupted 2022-10-19 21:39 pg_recvlogical prints bogus error when interrupted Andres Freund <[email protected]> 2022-10-20 07:58 ` Re: pg_recvlogical prints bogus error when interrupted Bharath Rupireddy <[email protected]> 2022-10-21 02:21 ` Re: pg_recvlogical prints bogus error when interrupted Kyotaro Horiguchi <[email protected]> 2022-10-24 02:45 ` Re: pg_recvlogical prints bogus error when interrupted Bharath Rupireddy <[email protected]> @ 2022-10-27 23:11 ` Andres Freund <[email protected]> 2022-10-28 03:11 ` Re: pg_recvlogical prints bogus error when interrupted Bharath Rupireddy <[email protected]> 1 sibling, 1 reply; 9+ messages in thread From: Andres Freund @ 2022-10-27 23:11 UTC (permalink / raw) To: Bharath Rupireddy <[email protected]>; +Cc: Kyotaro Horiguchi <[email protected]>; pgsql-hackers Hi, On 2022-10-24 08:15:11 +0530, Bharath Rupireddy wrote: > I came up with the attached v2 patch, please have a look. Thanks for working on this! > + /* When we get SIGINT/SIGTERM, we exit */ > + if (ready_to_exit) > + { > + /* > + * Try informing the server about our exit, but don't wait around > + * or retry on failure. > + */ > + (void) PQputCopyEnd(conn, NULL); > + (void) PQflush(conn); > + time_to_abort = ready_to_exit; This doesn't strike me as great - because the ready_to_exit isn't checked in the loop around StreamLogicalLog(), we'll reconnect if something else causes StreamLogicalLog() to return. Why do we need both time_to_abort and ready_to_exit? Perhaps worth noting that time_to_abort is still an sig_atomic_t, but isn't modified in a signal handler, which seems a bit unnecessarily confusing. Greetings, Andres Freund ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: pg_recvlogical prints bogus error when interrupted 2022-10-19 21:39 pg_recvlogical prints bogus error when interrupted Andres Freund <[email protected]> 2022-10-20 07:58 ` Re: pg_recvlogical prints bogus error when interrupted Bharath Rupireddy <[email protected]> 2022-10-21 02:21 ` Re: pg_recvlogical prints bogus error when interrupted Kyotaro Horiguchi <[email protected]> 2022-10-24 02:45 ` Re: pg_recvlogical prints bogus error when interrupted Bharath Rupireddy <[email protected]> 2022-10-27 23:11 ` Re: pg_recvlogical prints bogus error when interrupted Andres Freund <[email protected]> @ 2022-10-28 03:11 ` Bharath Rupireddy <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Bharath Rupireddy @ 2022-10-28 03:11 UTC (permalink / raw) To: Andres Freund <[email protected]>; +Cc: Kyotaro Horiguchi <[email protected]>; pgsql-hackers On Fri, Oct 28, 2022 at 4:41 AM Andres Freund <[email protected]> wrote: > > On 2022-10-24 08:15:11 +0530, Bharath Rupireddy wrote: > > > > + /* When we get SIGINT/SIGTERM, we exit */ > > + if (ready_to_exit) > > + { > > + /* > > + * Try informing the server about our exit, but don't wait around > > + * or retry on failure. > > + */ > > + (void) PQputCopyEnd(conn, NULL); > > + (void) PQflush(conn); > > + time_to_abort = ready_to_exit; > > This doesn't strike me as great - because the ready_to_exit isn't checked in > the loop around StreamLogicalLog(), we'll reconnect if something else causes > StreamLogicalLog() to return. Fixed. > Why do we need both time_to_abort and ready_to_exit? Intention to have ready_to_exit is to be able to distinguish between SIGINT/SIGTERM and aborting when endpos is reached so that necessary code is skipped/executed and proper logs are printed. > Perhaps worth noting that > time_to_abort is still an sig_atomic_t, but isn't modified in a signal > handler, which seems a bit unnecessarily confusing. time_to_abort is just a static variable, no? +static bool time_to_abort = false; +static volatile sig_atomic_t ready_to_exit = false; Please see the attached v3 patch. -- Bharath Rupireddy PostgreSQL Contributors Team RDS Open Source Databases Amazon Web Services: https://aws.amazon.com Attachments: [application/octet-stream] v3-0001-Fix-pg_recvlogical-error-message-upon-SIGINT-SIGT.patch (2.2K, ../../CALj2ACWwpz0tCMnUD28-XXLrAF7h9rzxJq2M6Nb_0GYA2cxANA@mail.gmail.com/2-v3-0001-Fix-pg_recvlogical-error-message-upon-SIGINT-SIGT.patch) download | inline diff: From b5f940a6af46d487e164f3de17055be1289ae41b Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Fri, 28 Oct 2022 03:09:46 +0000 Subject: [PATCH v3] Fix pg_recvlogical error message upon SIGINT/SIGTERM --- src/bin/pg_basebackup/pg_recvlogical.c | 29 ++++++++++++++++++++++---- 1 file changed, 25 insertions(+), 4 deletions(-) diff --git a/src/bin/pg_basebackup/pg_recvlogical.c b/src/bin/pg_basebackup/pg_recvlogical.c index 5f2e6af445..8f1d4d581f 100644 --- a/src/bin/pg_basebackup/pg_recvlogical.c +++ b/src/bin/pg_basebackup/pg_recvlogical.c @@ -54,7 +54,8 @@ static const char *plugin = "test_decoding"; /* Global State */ static int outfd = -1; -static volatile sig_atomic_t time_to_abort = false; +static bool time_to_abort = false; +static volatile sig_atomic_t ready_to_exit = false; static volatile sig_atomic_t output_reopen = false; static bool output_isfile; static TimestampTz output_last_fsync = -1; @@ -283,6 +284,23 @@ StreamLogicalLog(void) copybuf = NULL; } + /* When we get SIGINT/SIGTERM, we exit */ + if (ready_to_exit) + { + /* + * Try informing the server about our exit, but don't wait around + * or retry on failure. + */ + (void) PQputCopyEnd(conn, NULL); + (void) PQflush(conn); + time_to_abort = true; + + if (verbose) + pg_log_info("received interrupt signal, exiting"); + + break; + } + /* * Potentially send a status message to the primary. */ @@ -614,7 +632,10 @@ StreamLogicalLog(void) res = PQgetResult(conn); } - if (PQresultStatus(res) != PGRES_COMMAND_OK) + + /* It is not unexepected termination error when Ctrl-C'ed. */ + if (!ready_to_exit && + PQresultStatus(res) != PGRES_COMMAND_OK) { pg_log_error("unexpected termination of replication stream: %s", PQresultErrorMessage(res)); @@ -656,7 +677,7 @@ error: static void sigexit_handler(SIGNAL_ARGS) { - time_to_abort = true; + ready_to_exit = true; } /* @@ -976,7 +997,7 @@ main(int argc, char **argv) while (true) { StreamLogicalLog(); - if (time_to_abort) + if (ready_to_exit || time_to_abort) { /* * We've been Ctrl-C'ed or reached an exit limit condition. That's -- 2.34.1 ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH v15 7/8] Row pattern recognition patch (tests). @ 2024-03-28 10:30 Tatsuo Ishii <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Tatsuo Ishii @ 2024-03-28 10:30 UTC (permalink / raw) --- src/test/regress/expected/rpr.out | 821 +++++++++++++++++++++++++++++ src/test/regress/parallel_schedule | 2 +- src/test/regress/sql/rpr.sql | 392 ++++++++++++++ 3 files changed, 1214 insertions(+), 1 deletion(-) create mode 100644 src/test/regress/expected/rpr.out create mode 100644 src/test/regress/sql/rpr.sql diff --git a/src/test/regress/expected/rpr.out b/src/test/regress/expected/rpr.out new file mode 100644 index 0000000000..e8998ebf45 --- /dev/null +++ b/src/test/regress/expected/rpr.out @@ -0,0 +1,821 @@ +-- +-- Test for row pattern definition clause +-- +CREATE TEMP TABLE stock ( + company TEXT, + tdate DATE, + price INTEGER +); +INSERT INTO stock VALUES ('company1', '2023-07-01', 100); +INSERT INTO stock VALUES ('company1', '2023-07-02', 200); +INSERT INTO stock VALUES ('company1', '2023-07-03', 150); +INSERT INTO stock VALUES ('company1', '2023-07-04', 140); +INSERT INTO stock VALUES ('company1', '2023-07-05', 150); +INSERT INTO stock VALUES ('company1', '2023-07-06', 90); +INSERT INTO stock VALUES ('company1', '2023-07-07', 110); +INSERT INTO stock VALUES ('company1', '2023-07-08', 130); +INSERT INTO stock VALUES ('company1', '2023-07-09', 120); +INSERT INTO stock VALUES ('company1', '2023-07-10', 130); +INSERT INTO stock VALUES ('company2', '2023-07-01', 50); +INSERT INTO stock VALUES ('company2', '2023-07-02', 2000); +INSERT INTO stock VALUES ('company2', '2023-07-03', 1500); +INSERT INTO stock VALUES ('company2', '2023-07-04', 1400); +INSERT INTO stock VALUES ('company2', '2023-07-05', 1500); +INSERT INTO stock VALUES ('company2', '2023-07-06', 60); +INSERT INTO stock VALUES ('company2', '2023-07-07', 1100); +INSERT INTO stock VALUES ('company2', '2023-07-08', 1300); +INSERT INTO stock VALUES ('company2', '2023-07-09', 1200); +INSERT INTO stock VALUES ('company2', '2023-07-10', 1300); +SELECT * FROM stock; + company | tdate | price +----------+------------+------- + company1 | 07-01-2023 | 100 + company1 | 07-02-2023 | 200 + company1 | 07-03-2023 | 150 + company1 | 07-04-2023 | 140 + company1 | 07-05-2023 | 150 + company1 | 07-06-2023 | 90 + company1 | 07-07-2023 | 110 + company1 | 07-08-2023 | 130 + company1 | 07-09-2023 | 120 + company1 | 07-10-2023 | 130 + company2 | 07-01-2023 | 50 + company2 | 07-02-2023 | 2000 + company2 | 07-03-2023 | 1500 + company2 | 07-04-2023 | 1400 + company2 | 07-05-2023 | 1500 + company2 | 07-06-2023 | 60 + company2 | 07-07-2023 | 1100 + company2 | 07-08-2023 | 1300 + company2 | 07-09-2023 | 1200 + company2 | 07-10-2023 | 1300 +(20 rows) + +-- basic test using PREV +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w, + nth_value(tdate, 2) OVER w AS nth_second + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + company | tdate | price | first_value | last_value | nth_second +----------+------------+-------+-------------+------------+------------ + company1 | 07-01-2023 | 100 | 100 | 140 | 07-02-2023 + company1 | 07-02-2023 | 200 | | | + company1 | 07-03-2023 | 150 | | | + company1 | 07-04-2023 | 140 | | | + company1 | 07-05-2023 | 150 | | | + company1 | 07-06-2023 | 90 | 90 | 120 | 07-07-2023 + company1 | 07-07-2023 | 110 | | | + company1 | 07-08-2023 | 130 | | | + company1 | 07-09-2023 | 120 | | | + company1 | 07-10-2023 | 130 | | | + company2 | 07-01-2023 | 50 | 50 | 1400 | 07-02-2023 + company2 | 07-02-2023 | 2000 | | | + company2 | 07-03-2023 | 1500 | | | + company2 | 07-04-2023 | 1400 | | | + company2 | 07-05-2023 | 1500 | | | + company2 | 07-06-2023 | 60 | 60 | 1200 | 07-07-2023 + company2 | 07-07-2023 | 1100 | | | + company2 | 07-08-2023 | 1300 | | | + company2 | 07-09-2023 | 1200 | | | + company2 | 07-10-2023 | 1300 | | | +(20 rows) + +-- basic test using PREV. UP appears twice +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w, + nth_value(tdate, 2) OVER w AS nth_second + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+ UP+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + company | tdate | price | first_value | last_value | nth_second +----------+------------+-------+-------------+------------+------------ + company1 | 07-01-2023 | 100 | 100 | 150 | 07-02-2023 + company1 | 07-02-2023 | 200 | | | + company1 | 07-03-2023 | 150 | | | + company1 | 07-04-2023 | 140 | | | + company1 | 07-05-2023 | 150 | | | + company1 | 07-06-2023 | 90 | 90 | 130 | 07-07-2023 + company1 | 07-07-2023 | 110 | | | + company1 | 07-08-2023 | 130 | | | + company1 | 07-09-2023 | 120 | | | + company1 | 07-10-2023 | 130 | | | + company2 | 07-01-2023 | 50 | 50 | 1500 | 07-02-2023 + company2 | 07-02-2023 | 2000 | | | + company2 | 07-03-2023 | 1500 | | | + company2 | 07-04-2023 | 1400 | | | + company2 | 07-05-2023 | 1500 | | | + company2 | 07-06-2023 | 60 | 60 | 1300 | 07-07-2023 + company2 | 07-07-2023 | 1100 | | | + company2 | 07-08-2023 | 1300 | | | + company2 | 07-09-2023 | 1200 | | | + company2 | 07-10-2023 | 1300 | | | +(20 rows) + +-- basic test using PREV. Use '*' +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w, + nth_value(tdate, 2) OVER w AS nth_second + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP* DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + company | tdate | price | first_value | last_value | nth_second +----------+------------+-------+-------------+------------+------------ + company1 | 07-01-2023 | 100 | 100 | 140 | 07-02-2023 + company1 | 07-02-2023 | 200 | | | + company1 | 07-03-2023 | 150 | | | + company1 | 07-04-2023 | 140 | | | + company1 | 07-05-2023 | 150 | 150 | 90 | 07-06-2023 + company1 | 07-06-2023 | 90 | | | + company1 | 07-07-2023 | 110 | 110 | 120 | 07-08-2023 + company1 | 07-08-2023 | 130 | | | + company1 | 07-09-2023 | 120 | | | + company1 | 07-10-2023 | 130 | | | + company2 | 07-01-2023 | 50 | 50 | 1400 | 07-02-2023 + company2 | 07-02-2023 | 2000 | | | + company2 | 07-03-2023 | 1500 | | | + company2 | 07-04-2023 | 1400 | | | + company2 | 07-05-2023 | 1500 | 1500 | 60 | 07-06-2023 + company2 | 07-06-2023 | 60 | | | + company2 | 07-07-2023 | 1100 | 1100 | 1200 | 07-08-2023 + company2 | 07-08-2023 | 1300 | | | + company2 | 07-09-2023 | 1200 | | | + company2 | 07-10-2023 | 1300 | | | +(20 rows) + +-- basic test with none greedy pattern +SELECT company, tdate, price, count(*) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (A A A) + DEFINE + A AS price >= 140 AND price <= 150 +); + company | tdate | price | count +----------+------------+-------+------- + company1 | 07-01-2023 | 100 | 0 + company1 | 07-02-2023 | 200 | 0 + company1 | 07-03-2023 | 150 | 3 + company1 | 07-04-2023 | 140 | + company1 | 07-05-2023 | 150 | + company1 | 07-06-2023 | 90 | 0 + company1 | 07-07-2023 | 110 | 0 + company1 | 07-08-2023 | 130 | 0 + company1 | 07-09-2023 | 120 | 0 + company1 | 07-10-2023 | 130 | 0 + company2 | 07-01-2023 | 50 | 0 + company2 | 07-02-2023 | 2000 | 0 + company2 | 07-03-2023 | 1500 | 0 + company2 | 07-04-2023 | 1400 | 0 + company2 | 07-05-2023 | 1500 | 0 + company2 | 07-06-2023 | 60 | 0 + company2 | 07-07-2023 | 1100 | 0 + company2 | 07-08-2023 | 1300 | 0 + company2 | 07-09-2023 | 1200 | 0 + company2 | 07-10-2023 | 1300 | 0 +(20 rows) + +-- last_value() should remain consistent +SELECT company, tdate, price, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ORDER BY tdate + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + company | tdate | price | last_value +----------+------------+-------+------------ + company1 | 07-01-2023 | 100 | 140 + company1 | 07-02-2023 | 200 | + company1 | 07-03-2023 | 150 | + company1 | 07-04-2023 | 140 | + company1 | 07-05-2023 | 150 | + company1 | 07-06-2023 | 90 | 120 + company1 | 07-07-2023 | 110 | + company1 | 07-08-2023 | 130 | + company1 | 07-09-2023 | 120 | + company1 | 07-10-2023 | 130 | + company2 | 07-01-2023 | 50 | 1400 + company2 | 07-02-2023 | 2000 | + company2 | 07-03-2023 | 1500 | + company2 | 07-04-2023 | 1400 | + company2 | 07-05-2023 | 1500 | + company2 | 07-06-2023 | 60 | 1200 + company2 | 07-07-2023 | 1100 | + company2 | 07-08-2023 | 1300 | + company2 | 07-09-2023 | 1200 | + company2 | 07-10-2023 | 1300 | +(20 rows) + +-- omit "START" in DEFINE but it is ok because "START AS TRUE" is +-- implicitly defined. per spec. +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w, + nth_value(tdate, 2) OVER w AS nth_second + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + company | tdate | price | first_value | last_value | nth_second +----------+------------+-------+-------------+------------+------------ + company1 | 07-01-2023 | 100 | 100 | 140 | 07-02-2023 + company1 | 07-02-2023 | 200 | | | + company1 | 07-03-2023 | 150 | | | + company1 | 07-04-2023 | 140 | | | + company1 | 07-05-2023 | 150 | | | + company1 | 07-06-2023 | 90 | 90 | 120 | 07-07-2023 + company1 | 07-07-2023 | 110 | | | + company1 | 07-08-2023 | 130 | | | + company1 | 07-09-2023 | 120 | | | + company1 | 07-10-2023 | 130 | | | + company2 | 07-01-2023 | 50 | 50 | 1400 | 07-02-2023 + company2 | 07-02-2023 | 2000 | | | + company2 | 07-03-2023 | 1500 | | | + company2 | 07-04-2023 | 1400 | | | + company2 | 07-05-2023 | 1500 | | | + company2 | 07-06-2023 | 60 | 60 | 1200 | 07-07-2023 + company2 | 07-07-2023 | 1100 | | | + company2 | 07-08-2023 | 1300 | | | + company2 | 07-09-2023 | 1200 | | | + company2 | 07-10-2023 | 1300 | | | +(20 rows) + +-- the first row start with less than or equal to 100 +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (LOWPRICE UP+ DOWN+) + DEFINE + LOWPRICE AS price <= 100, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + company | tdate | price | first_value | last_value +----------+------------+-------+-------------+------------ + company1 | 07-01-2023 | 100 | 100 | 140 + company1 | 07-02-2023 | 200 | | + company1 | 07-03-2023 | 150 | | + company1 | 07-04-2023 | 140 | | + company1 | 07-05-2023 | 150 | | + company1 | 07-06-2023 | 90 | 90 | 120 + company1 | 07-07-2023 | 110 | | + company1 | 07-08-2023 | 130 | | + company1 | 07-09-2023 | 120 | | + company1 | 07-10-2023 | 130 | | + company2 | 07-01-2023 | 50 | 50 | 1400 + company2 | 07-02-2023 | 2000 | | + company2 | 07-03-2023 | 1500 | | + company2 | 07-04-2023 | 1400 | | + company2 | 07-05-2023 | 1500 | | + company2 | 07-06-2023 | 60 | 60 | 1200 + company2 | 07-07-2023 | 1100 | | + company2 | 07-08-2023 | 1300 | | + company2 | 07-09-2023 | 1200 | | + company2 | 07-10-2023 | 1300 | | +(20 rows) + +-- second row raises 120% +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (LOWPRICE UP+ DOWN+) + DEFINE + LOWPRICE AS price <= 100, + UP AS price > PREV(price) * 1.2, + DOWN AS price < PREV(price) +); + company | tdate | price | first_value | last_value +----------+------------+-------+-------------+------------ + company1 | 07-01-2023 | 100 | 100 | 140 + company1 | 07-02-2023 | 200 | | + company1 | 07-03-2023 | 150 | | + company1 | 07-04-2023 | 140 | | + company1 | 07-05-2023 | 150 | | + company1 | 07-06-2023 | 90 | | + company1 | 07-07-2023 | 110 | | + company1 | 07-08-2023 | 130 | | + company1 | 07-09-2023 | 120 | | + company1 | 07-10-2023 | 130 | | + company2 | 07-01-2023 | 50 | 50 | 1400 + company2 | 07-02-2023 | 2000 | | + company2 | 07-03-2023 | 1500 | | + company2 | 07-04-2023 | 1400 | | + company2 | 07-05-2023 | 1500 | | + company2 | 07-06-2023 | 60 | | + company2 | 07-07-2023 | 1100 | | + company2 | 07-08-2023 | 1300 | | + company2 | 07-09-2023 | 1200 | | + company2 | 07-10-2023 | 1300 | | +(20 rows) + +-- using NEXT +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UPDOWN) + DEFINE + START AS TRUE, + UPDOWN AS price > PREV(price) AND price > NEXT(price) +); + company | tdate | price | first_value | last_value +----------+------------+-------+-------------+------------ + company1 | 07-01-2023 | 100 | 100 | 200 + company1 | 07-02-2023 | 200 | | + company1 | 07-03-2023 | 150 | | + company1 | 07-04-2023 | 140 | 140 | 150 + company1 | 07-05-2023 | 150 | | + company1 | 07-06-2023 | 90 | | + company1 | 07-07-2023 | 110 | 110 | 130 + company1 | 07-08-2023 | 130 | | + company1 | 07-09-2023 | 120 | | + company1 | 07-10-2023 | 130 | | + company2 | 07-01-2023 | 50 | 50 | 2000 + company2 | 07-02-2023 | 2000 | | + company2 | 07-03-2023 | 1500 | | + company2 | 07-04-2023 | 1400 | 1400 | 1500 + company2 | 07-05-2023 | 1500 | | + company2 | 07-06-2023 | 60 | | + company2 | 07-07-2023 | 1100 | 1100 | 1300 + company2 | 07-08-2023 | 1300 | | + company2 | 07-09-2023 | 1200 | | + company2 | 07-10-2023 | 1300 | | +(20 rows) + +-- using AFTER MATCH SKIP TO NEXT ROW +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + AFTER MATCH SKIP TO NEXT ROW + INITIAL + PATTERN (START UPDOWN) + DEFINE + START AS TRUE, + UPDOWN AS price > PREV(price) AND price > NEXT(price) +); + company | tdate | price | first_value | last_value +----------+------------+-------+-------------+------------ + company1 | 07-01-2023 | 100 | 100 | 200 + company1 | 07-02-2023 | 200 | | + company1 | 07-03-2023 | 150 | | + company1 | 07-04-2023 | 140 | 140 | 150 + company1 | 07-05-2023 | 150 | | + company1 | 07-06-2023 | 90 | | + company1 | 07-07-2023 | 110 | 110 | 130 + company1 | 07-08-2023 | 130 | | + company1 | 07-09-2023 | 120 | | + company1 | 07-10-2023 | 130 | | + company2 | 07-01-2023 | 50 | 50 | 2000 + company2 | 07-02-2023 | 2000 | | + company2 | 07-03-2023 | 1500 | | + company2 | 07-04-2023 | 1400 | 1400 | 1500 + company2 | 07-05-2023 | 1500 | | + company2 | 07-06-2023 | 60 | | + company2 | 07-07-2023 | 1100 | 1100 | 1300 + company2 | 07-08-2023 | 1300 | | + company2 | 07-09-2023 | 1200 | | + company2 | 07-10-2023 | 1300 | | +(20 rows) + +-- match everything +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ORDER BY tdate + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + AFTER MATCH SKIP PAST LAST ROW + INITIAL + PATTERN (A+) + DEFINE + A AS TRUE +); + company | tdate | price | first_value | last_value +----------+------------+-------+-------------+------------ + company1 | 07-01-2023 | 100 | 100 | 130 + company1 | 07-02-2023 | 200 | | + company1 | 07-03-2023 | 150 | | + company1 | 07-04-2023 | 140 | | + company1 | 07-05-2023 | 150 | | + company1 | 07-06-2023 | 90 | | + company1 | 07-07-2023 | 110 | | + company1 | 07-08-2023 | 130 | | + company1 | 07-09-2023 | 120 | | + company1 | 07-10-2023 | 130 | | + company2 | 07-01-2023 | 50 | 50 | 1300 + company2 | 07-02-2023 | 2000 | | + company2 | 07-03-2023 | 1500 | | + company2 | 07-04-2023 | 1400 | | + company2 | 07-05-2023 | 1500 | | + company2 | 07-06-2023 | 60 | | + company2 | 07-07-2023 | 1100 | | + company2 | 07-08-2023 | 1300 | | + company2 | 07-09-2023 | 1200 | | + company2 | 07-10-2023 | 1300 | | +(20 rows) + +-- backtracking with reclassification of rows +-- using AFTER MATCH SKIP PAST LAST ROW +SELECT company, tdate, price, first_value(tdate) OVER w, last_value(tdate) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ORDER BY tdate + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + AFTER MATCH SKIP PAST LAST ROW + INITIAL + PATTERN (A+ B+) + DEFINE + A AS price > 100, + B AS price > 100 +); + company | tdate | price | first_value | last_value +----------+------------+-------+-------------+------------ + company1 | 07-01-2023 | 100 | | + company1 | 07-02-2023 | 200 | 07-02-2023 | 07-05-2023 + company1 | 07-03-2023 | 150 | | + company1 | 07-04-2023 | 140 | | + company1 | 07-05-2023 | 150 | | + company1 | 07-06-2023 | 90 | | + company1 | 07-07-2023 | 110 | 07-07-2023 | 07-10-2023 + company1 | 07-08-2023 | 130 | | + company1 | 07-09-2023 | 120 | | + company1 | 07-10-2023 | 130 | | + company2 | 07-01-2023 | 50 | | + company2 | 07-02-2023 | 2000 | 07-02-2023 | 07-05-2023 + company2 | 07-03-2023 | 1500 | | + company2 | 07-04-2023 | 1400 | | + company2 | 07-05-2023 | 1500 | | + company2 | 07-06-2023 | 60 | | + company2 | 07-07-2023 | 1100 | 07-07-2023 | 07-10-2023 + company2 | 07-08-2023 | 1300 | | + company2 | 07-09-2023 | 1200 | | + company2 | 07-10-2023 | 1300 | | +(20 rows) + +-- backtracking with reclassification of rows +-- using AFTER MATCH SKIP TO NEXT ROW +SELECT company, tdate, price, first_value(tdate) OVER w, last_value(tdate) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ORDER BY tdate + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + AFTER MATCH SKIP TO NEXT ROW + INITIAL + PATTERN (A+ B+) + DEFINE + A AS price > 100, + B AS price > 100 +); + company | tdate | price | first_value | last_value +----------+------------+-------+-------------+------------ + company1 | 07-01-2023 | 100 | | + company1 | 07-02-2023 | 200 | 07-02-2023 | 07-05-2023 + company1 | 07-03-2023 | 150 | 07-03-2023 | 07-05-2023 + company1 | 07-04-2023 | 140 | 07-04-2023 | 07-05-2023 + company1 | 07-05-2023 | 150 | | + company1 | 07-06-2023 | 90 | | + company1 | 07-07-2023 | 110 | 07-07-2023 | 07-10-2023 + company1 | 07-08-2023 | 130 | 07-08-2023 | 07-10-2023 + company1 | 07-09-2023 | 120 | 07-09-2023 | 07-10-2023 + company1 | 07-10-2023 | 130 | | + company2 | 07-01-2023 | 50 | | + company2 | 07-02-2023 | 2000 | 07-02-2023 | 07-05-2023 + company2 | 07-03-2023 | 1500 | 07-03-2023 | 07-05-2023 + company2 | 07-04-2023 | 1400 | 07-04-2023 | 07-05-2023 + company2 | 07-05-2023 | 1500 | | + company2 | 07-06-2023 | 60 | | + company2 | 07-07-2023 | 1100 | 07-07-2023 | 07-10-2023 + company2 | 07-08-2023 | 1300 | 07-08-2023 | 07-10-2023 + company2 | 07-09-2023 | 1200 | 07-09-2023 | 07-10-2023 + company2 | 07-10-2023 | 1300 | | +(20 rows) + +-- ROWS BETWEEN CURRENT ROW AND offset FOLLOWING +SELECT company, tdate, price, first_value(tdate) OVER w, last_value(tdate) OVER w, + count(*) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ORDER BY tdate + ROWS BETWEEN CURRENT ROW AND 2 FOLLOWING + AFTER MATCH SKIP PAST LAST ROW + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + company | tdate | price | first_value | last_value | count +----------+------------+-------+-------------+------------+------- + company1 | 07-01-2023 | 100 | 07-01-2023 | 07-03-2023 | 3 + company1 | 07-02-2023 | 200 | | | + company1 | 07-03-2023 | 150 | | | + company1 | 07-04-2023 | 140 | 07-04-2023 | 07-06-2023 | 3 + company1 | 07-05-2023 | 150 | | | + company1 | 07-06-2023 | 90 | | | + company1 | 07-07-2023 | 110 | 07-07-2023 | 07-09-2023 | 3 + company1 | 07-08-2023 | 130 | | | + company1 | 07-09-2023 | 120 | | | + company1 | 07-10-2023 | 130 | | | 0 + company2 | 07-01-2023 | 50 | 07-01-2023 | 07-03-2023 | 3 + company2 | 07-02-2023 | 2000 | | | + company2 | 07-03-2023 | 1500 | | | + company2 | 07-04-2023 | 1400 | 07-04-2023 | 07-06-2023 | 3 + company2 | 07-05-2023 | 1500 | | | + company2 | 07-06-2023 | 60 | | | + company2 | 07-07-2023 | 1100 | 07-07-2023 | 07-09-2023 | 3 + company2 | 07-08-2023 | 1300 | | | + company2 | 07-09-2023 | 1200 | | | + company2 | 07-10-2023 | 1300 | | | 0 +(20 rows) + +-- +-- Aggregates +-- +-- using AFTER MATCH SKIP PAST LAST ROW +SELECT company, tdate, price, + first_value(price) OVER w, + last_value(price) OVER w, + max(price) OVER w, + min(price) OVER w, + sum(price) OVER w, + avg(price) OVER w, + count(price) OVER w +FROM stock +WINDOW w AS ( +PARTITION BY company +ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING +AFTER MATCH SKIP PAST LAST ROW +INITIAL +PATTERN (START UP+ DOWN+) +DEFINE +START AS TRUE, +UP AS price > PREV(price), +DOWN AS price < PREV(price) +); + company | tdate | price | first_value | last_value | max | min | sum | avg | count +----------+------------+-------+-------------+------------+------+-----+------+-----------------------+------- + company1 | 07-01-2023 | 100 | 100 | 140 | 200 | 100 | 590 | 147.5000000000000000 | 4 + company1 | 07-02-2023 | 200 | | | | | | | + company1 | 07-03-2023 | 150 | | | | | | | + company1 | 07-04-2023 | 140 | | | | | | | + company1 | 07-05-2023 | 150 | | | | | | | 0 + company1 | 07-06-2023 | 90 | 90 | 120 | 130 | 90 | 450 | 112.5000000000000000 | 4 + company1 | 07-07-2023 | 110 | | | | | | | + company1 | 07-08-2023 | 130 | | | | | | | + company1 | 07-09-2023 | 120 | | | | | | | + company1 | 07-10-2023 | 130 | | | | | | | 0 + company2 | 07-01-2023 | 50 | 50 | 1400 | 2000 | 50 | 4950 | 1237.5000000000000000 | 4 + company2 | 07-02-2023 | 2000 | | | | | | | + company2 | 07-03-2023 | 1500 | | | | | | | + company2 | 07-04-2023 | 1400 | | | | | | | + company2 | 07-05-2023 | 1500 | | | | | | | 0 + company2 | 07-06-2023 | 60 | 60 | 1200 | 1300 | 60 | 3660 | 915.0000000000000000 | 4 + company2 | 07-07-2023 | 1100 | | | | | | | + company2 | 07-08-2023 | 1300 | | | | | | | + company2 | 07-09-2023 | 1200 | | | | | | | + company2 | 07-10-2023 | 1300 | | | | | | | 0 +(20 rows) + +-- using AFTER MATCH SKIP TO NEXT ROW +SELECT company, tdate, price, + first_value(price) OVER w, + last_value(price) OVER w, + max(price) OVER w, + min(price) OVER w, + sum(price) OVER w, + avg(price) OVER w, + count(price) OVER w +FROM stock +WINDOW w AS ( +PARTITION BY company +ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING +AFTER MATCH SKIP TO NEXT ROW +INITIAL +PATTERN (START UP+ DOWN+) +DEFINE +START AS TRUE, +UP AS price > PREV(price), +DOWN AS price < PREV(price) +); + company | tdate | price | first_value | last_value | max | min | sum | avg | count +----------+------------+-------+-------------+------------+------+------+------+-----------------------+------- + company1 | 07-01-2023 | 100 | 100 | 140 | 200 | 100 | 590 | 147.5000000000000000 | 4 + company1 | 07-02-2023 | 200 | | | | | | | 0 + company1 | 07-03-2023 | 150 | | | | | | | 0 + company1 | 07-04-2023 | 140 | 140 | 90 | 150 | 90 | 380 | 126.6666666666666667 | 3 + company1 | 07-05-2023 | 150 | | | | | | | 0 + company1 | 07-06-2023 | 90 | 90 | 120 | 130 | 90 | 450 | 112.5000000000000000 | 4 + company1 | 07-07-2023 | 110 | 110 | 120 | 130 | 110 | 360 | 120.0000000000000000 | 3 + company1 | 07-08-2023 | 130 | | | | | | | 0 + company1 | 07-09-2023 | 120 | | | | | | | 0 + company1 | 07-10-2023 | 130 | | | | | | | 0 + company2 | 07-01-2023 | 50 | 50 | 1400 | 2000 | 50 | 4950 | 1237.5000000000000000 | 4 + company2 | 07-02-2023 | 2000 | | | | | | | 0 + company2 | 07-03-2023 | 1500 | | | | | | | 0 + company2 | 07-04-2023 | 1400 | 1400 | 60 | 1500 | 60 | 2960 | 986.6666666666666667 | 3 + company2 | 07-05-2023 | 1500 | | | | | | | 0 + company2 | 07-06-2023 | 60 | 60 | 1200 | 1300 | 60 | 3660 | 915.0000000000000000 | 4 + company2 | 07-07-2023 | 1100 | 1100 | 1200 | 1300 | 1100 | 3600 | 1200.0000000000000000 | 3 + company2 | 07-08-2023 | 1300 | | | | | | | 0 + company2 | 07-09-2023 | 1200 | | | | | | | 0 + company2 | 07-10-2023 | 1300 | | | | | | | 0 +(20 rows) + +-- JOIN case +CREATE TEMP TABLE t1 (i int, v1 int); +CREATE TEMP TABLE t2 (j int, v2 int); +INSERT INTO t1 VALUES(1,10); +INSERT INTO t1 VALUES(1,11); +INSERT INTO t1 VALUES(1,12); +INSERT INTO t2 VALUES(2,10); +INSERT INTO t2 VALUES(2,11); +INSERT INTO t2 VALUES(2,12); +SELECT * FROM t1, t2 WHERE t1.v1 <= 11 AND t2.v2 <= 11; + i | v1 | j | v2 +---+----+---+---- + 1 | 10 | 2 | 10 + 1 | 10 | 2 | 11 + 1 | 11 | 2 | 10 + 1 | 11 | 2 | 11 +(4 rows) + +SELECT *, count(*) OVER w FROM t1, t2 +WINDOW w AS ( + PARTITION BY t1.i + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (A) + DEFINE + A AS v1 <= 11 AND v2 <= 11 +); + i | v1 | j | v2 | count +---+----+---+----+------- + 1 | 10 | 2 | 10 | 1 + 1 | 10 | 2 | 11 | 1 + 1 | 10 | 2 | 12 | 0 + 1 | 11 | 2 | 10 | 1 + 1 | 11 | 2 | 11 | 1 + 1 | 11 | 2 | 12 | 0 + 1 | 12 | 2 | 10 | 0 + 1 | 12 | 2 | 11 | 0 + 1 | 12 | 2 | 12 | 0 +(9 rows) + +-- WITH case +WITH wstock AS ( + SELECT * FROM stock WHERE tdate < '2023-07-08' +) +SELECT tdate, price, +first_value(tdate) OVER w, +count(*) OVER w + FROM wstock + WINDOW w AS ( + PARTITION BY company + ORDER BY tdate + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + tdate | price | first_value | count +------------+-------+-------------+------- + 07-01-2023 | 100 | 07-01-2023 | 4 + 07-02-2023 | 200 | | + 07-03-2023 | 150 | | + 07-04-2023 | 140 | | + 07-05-2023 | 150 | | 0 + 07-06-2023 | 90 | | 0 + 07-07-2023 | 110 | | 0 + 07-01-2023 | 50 | 07-01-2023 | 4 + 07-02-2023 | 2000 | | + 07-03-2023 | 1500 | | + 07-04-2023 | 1400 | | + 07-05-2023 | 1500 | | 0 + 07-06-2023 | 60 | | 0 + 07-07-2023 | 1100 | | 0 +(14 rows) + +-- +-- Error cases +-- +-- row pattern definition variable name must not appear more than once +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + ORDER BY tdate + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price), + UP AS price > PREV(price) +); +ERROR: syntax error at or near "ORDER" +LINE 6: ORDER BY tdate + ^ +-- pattern variable name must appear in DEFINE +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ORDER BY tdate + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+ END) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); +ERROR: syntax error at or near "END" +LINE 8: PATTERN (START UP+ DOWN+ END) + ^ +-- FRAME must start at current row when row patttern recognition is used +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ORDER BY tdate + ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); +ERROR: FRAME must start at current row when row patttern recognition is used +-- SEEK is not supported +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ORDER BY tdate + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + AFTER MATCH SKIP TO NEXT ROW + SEEK + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); +ERROR: SEEK is not supported +LINE 8: SEEK + ^ +HINT: Use INITIAL. diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule index 5ac6e871f5..db6978fb5c 100644 --- a/src/test/regress/parallel_schedule +++ b/src/test/regress/parallel_schedule @@ -98,7 +98,7 @@ test: publication subscription # Another group of parallel tests # select_views depends on create_view # ---------- -test: select_views portals_p2 foreign_key cluster dependency guc bitmapops combocid tsearch tsdicts foreign_data window xmlmap functional_deps advisory_lock indirect_toast equivclass +test: select_views portals_p2 foreign_key cluster dependency guc bitmapops combocid tsearch tsdicts foreign_data window xmlmap functional_deps advisory_lock indirect_toast equivclass rpr # ---------- # Another group of parallel tests (JSON related) diff --git a/src/test/regress/sql/rpr.sql b/src/test/regress/sql/rpr.sql new file mode 100644 index 0000000000..0a69cc0e11 --- /dev/null +++ b/src/test/regress/sql/rpr.sql @@ -0,0 +1,392 @@ +-- +-- Test for row pattern definition clause +-- + +CREATE TEMP TABLE stock ( + company TEXT, + tdate DATE, + price INTEGER +); +INSERT INTO stock VALUES ('company1', '2023-07-01', 100); +INSERT INTO stock VALUES ('company1', '2023-07-02', 200); +INSERT INTO stock VALUES ('company1', '2023-07-03', 150); +INSERT INTO stock VALUES ('company1', '2023-07-04', 140); +INSERT INTO stock VALUES ('company1', '2023-07-05', 150); +INSERT INTO stock VALUES ('company1', '2023-07-06', 90); +INSERT INTO stock VALUES ('company1', '2023-07-07', 110); +INSERT INTO stock VALUES ('company1', '2023-07-08', 130); +INSERT INTO stock VALUES ('company1', '2023-07-09', 120); +INSERT INTO stock VALUES ('company1', '2023-07-10', 130); +INSERT INTO stock VALUES ('company2', '2023-07-01', 50); +INSERT INTO stock VALUES ('company2', '2023-07-02', 2000); +INSERT INTO stock VALUES ('company2', '2023-07-03', 1500); +INSERT INTO stock VALUES ('company2', '2023-07-04', 1400); +INSERT INTO stock VALUES ('company2', '2023-07-05', 1500); +INSERT INTO stock VALUES ('company2', '2023-07-06', 60); +INSERT INTO stock VALUES ('company2', '2023-07-07', 1100); +INSERT INTO stock VALUES ('company2', '2023-07-08', 1300); +INSERT INTO stock VALUES ('company2', '2023-07-09', 1200); +INSERT INTO stock VALUES ('company2', '2023-07-10', 1300); + +SELECT * FROM stock; + +-- basic test using PREV +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w, + nth_value(tdate, 2) OVER w AS nth_second + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + +-- basic test using PREV. UP appears twice +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w, + nth_value(tdate, 2) OVER w AS nth_second + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+ UP+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + +-- basic test using PREV. Use '*' +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w, + nth_value(tdate, 2) OVER w AS nth_second + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP* DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + +-- basic test with none greedy pattern +SELECT company, tdate, price, count(*) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (A A A) + DEFINE + A AS price >= 140 AND price <= 150 +); + +-- last_value() should remain consistent +SELECT company, tdate, price, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ORDER BY tdate + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + +-- omit "START" in DEFINE but it is ok because "START AS TRUE" is +-- implicitly defined. per spec. +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w, + nth_value(tdate, 2) OVER w AS nth_second + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + +-- the first row start with less than or equal to 100 +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (LOWPRICE UP+ DOWN+) + DEFINE + LOWPRICE AS price <= 100, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + +-- second row raises 120% +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (LOWPRICE UP+ DOWN+) + DEFINE + LOWPRICE AS price <= 100, + UP AS price > PREV(price) * 1.2, + DOWN AS price < PREV(price) +); + +-- using NEXT +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UPDOWN) + DEFINE + START AS TRUE, + UPDOWN AS price > PREV(price) AND price > NEXT(price) +); + +-- using AFTER MATCH SKIP TO NEXT ROW +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + AFTER MATCH SKIP TO NEXT ROW + INITIAL + PATTERN (START UPDOWN) + DEFINE + START AS TRUE, + UPDOWN AS price > PREV(price) AND price > NEXT(price) +); + +-- match everything + +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ORDER BY tdate + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + AFTER MATCH SKIP PAST LAST ROW + INITIAL + PATTERN (A+) + DEFINE + A AS TRUE +); + +-- backtracking with reclassification of rows +-- using AFTER MATCH SKIP PAST LAST ROW +SELECT company, tdate, price, first_value(tdate) OVER w, last_value(tdate) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ORDER BY tdate + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + AFTER MATCH SKIP PAST LAST ROW + INITIAL + PATTERN (A+ B+) + DEFINE + A AS price > 100, + B AS price > 100 +); + +-- backtracking with reclassification of rows +-- using AFTER MATCH SKIP TO NEXT ROW +SELECT company, tdate, price, first_value(tdate) OVER w, last_value(tdate) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ORDER BY tdate + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + AFTER MATCH SKIP TO NEXT ROW + INITIAL + PATTERN (A+ B+) + DEFINE + A AS price > 100, + B AS price > 100 +); + +-- ROWS BETWEEN CURRENT ROW AND offset FOLLOWING +SELECT company, tdate, price, first_value(tdate) OVER w, last_value(tdate) OVER w, + count(*) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ORDER BY tdate + ROWS BETWEEN CURRENT ROW AND 2 FOLLOWING + AFTER MATCH SKIP PAST LAST ROW + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + +-- +-- Aggregates +-- + +-- using AFTER MATCH SKIP PAST LAST ROW +SELECT company, tdate, price, + first_value(price) OVER w, + last_value(price) OVER w, + max(price) OVER w, + min(price) OVER w, + sum(price) OVER w, + avg(price) OVER w, + count(price) OVER w +FROM stock +WINDOW w AS ( +PARTITION BY company +ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING +AFTER MATCH SKIP PAST LAST ROW +INITIAL +PATTERN (START UP+ DOWN+) +DEFINE +START AS TRUE, +UP AS price > PREV(price), +DOWN AS price < PREV(price) +); + +-- using AFTER MATCH SKIP TO NEXT ROW +SELECT company, tdate, price, + first_value(price) OVER w, + last_value(price) OVER w, + max(price) OVER w, + min(price) OVER w, + sum(price) OVER w, + avg(price) OVER w, + count(price) OVER w +FROM stock +WINDOW w AS ( +PARTITION BY company +ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING +AFTER MATCH SKIP TO NEXT ROW +INITIAL +PATTERN (START UP+ DOWN+) +DEFINE +START AS TRUE, +UP AS price > PREV(price), +DOWN AS price < PREV(price) +); + +-- JOIN case +CREATE TEMP TABLE t1 (i int, v1 int); +CREATE TEMP TABLE t2 (j int, v2 int); +INSERT INTO t1 VALUES(1,10); +INSERT INTO t1 VALUES(1,11); +INSERT INTO t1 VALUES(1,12); +INSERT INTO t2 VALUES(2,10); +INSERT INTO t2 VALUES(2,11); +INSERT INTO t2 VALUES(2,12); + +SELECT * FROM t1, t2 WHERE t1.v1 <= 11 AND t2.v2 <= 11; + +SELECT *, count(*) OVER w FROM t1, t2 +WINDOW w AS ( + PARTITION BY t1.i + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (A) + DEFINE + A AS v1 <= 11 AND v2 <= 11 +); + +-- WITH case +WITH wstock AS ( + SELECT * FROM stock WHERE tdate < '2023-07-08' +) +SELECT tdate, price, +first_value(tdate) OVER w, +count(*) OVER w + FROM wstock + WINDOW w AS ( + PARTITION BY company + ORDER BY tdate + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + +-- +-- Error cases +-- + +-- row pattern definition variable name must not appear more than once +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + ORDER BY tdate + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price), + UP AS price > PREV(price) +); + +-- pattern variable name must appear in DEFINE +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ORDER BY tdate + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+ END) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + +-- FRAME must start at current row when row patttern recognition is used +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ORDER BY tdate + ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING + INITIAL + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); + +-- SEEK is not supported +SELECT company, tdate, price, first_value(price) OVER w, last_value(price) OVER w + FROM stock + WINDOW w AS ( + PARTITION BY company + ORDER BY tdate + ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING + AFTER MATCH SKIP TO NEXT ROW + SEEK + PATTERN (START UP+ DOWN+) + DEFINE + START AS TRUE, + UP AS price > PREV(price), + DOWN AS price < PREV(price) +); -- 2.25.1 ----Next_Part(Thu_Mar_28_19_59_25_2024_076)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="v15-0008-Allow-to-print-raw-parse-tree.patch" ^ permalink raw reply [nested|flat] 9+ messages in thread
end of thread, other threads:[~2024-03-28 10:30 UTC | newest] Thread overview: 9+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2020-05-08 07:17 [PATCH v9 1/4] INSERT SELECT to use BulkInsertState and multi_insert Justin Pryzby <[email protected]> 2022-10-19 21:39 pg_recvlogical prints bogus error when interrupted Andres Freund <[email protected]> 2022-10-20 07:58 ` Re: pg_recvlogical prints bogus error when interrupted Bharath Rupireddy <[email protected]> 2022-10-21 02:21 ` Re: pg_recvlogical prints bogus error when interrupted Kyotaro Horiguchi <[email protected]> 2022-10-24 02:45 ` Re: pg_recvlogical prints bogus error when interrupted Bharath Rupireddy <[email protected]> 2022-10-27 11:26 ` Re: pg_recvlogical prints bogus error when interrupted Bharath Rupireddy <[email protected]> 2022-10-27 23:11 ` Re: pg_recvlogical prints bogus error when interrupted Andres Freund <[email protected]> 2022-10-28 03:11 ` Re: pg_recvlogical prints bogus error when interrupted Bharath Rupireddy <[email protected]> 2024-03-28 10:30 [PATCH v15 7/8] Row pattern recognition patch (tests). 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