>From e2cc7054bbab06c631d7c78491cb52143a4e47f9 Mon Sep 17 00:00:00 2001 From: Kyotaro Horiguchi Date: Mon, 29 Jun 2015 16:51:12 +0900 Subject: [PATCH] POC: Experimental fetch_by_size feature --- contrib/auto_explain/auto_explain.c | 8 +- contrib/pg_stat_statements/pg_stat_statements.c | 8 +- contrib/postgres_fdw/postgres_fdw.c | 92 +++++++++++++++----- src/backend/access/common/heaptuple.c | 42 +++++++++ src/backend/commands/copy.c | 2 +- src/backend/commands/createas.c | 2 +- src/backend/commands/explain.c | 2 +- src/backend/commands/extension.c | 2 +- src/backend/commands/matview.c | 2 +- src/backend/commands/portalcmds.c | 4 +- src/backend/commands/prepare.c | 2 +- src/backend/executor/execMain.c | 39 +++++++-- src/backend/executor/execUtils.c | 1 + src/backend/executor/functions.c | 2 +- src/backend/executor/spi.c | 4 +- src/backend/parser/gram.y | 65 ++++++++++++++ src/backend/tcop/postgres.c | 2 + src/backend/tcop/pquery.c | 109 +++++++++++++++++------- src/include/access/htup_details.h | 2 + src/include/executor/executor.h | 8 +- src/include/nodes/execnodes.h | 1 + src/include/nodes/parsenodes.h | 2 + src/include/tcop/pquery.h | 7 +- src/interfaces/ecpg/preproc/ecpg.addons | 83 ++++++++++++++++++ 24 files changed, 409 insertions(+), 82 deletions(-) diff --git a/contrib/auto_explain/auto_explain.c b/contrib/auto_explain/auto_explain.c index 2a184ed..f121a33 100644 --- a/contrib/auto_explain/auto_explain.c +++ b/contrib/auto_explain/auto_explain.c @@ -57,7 +57,7 @@ void _PG_fini(void); static void explain_ExecutorStart(QueryDesc *queryDesc, int eflags); static void explain_ExecutorRun(QueryDesc *queryDesc, ScanDirection direction, - long count); + long count, long size); static void explain_ExecutorFinish(QueryDesc *queryDesc); static void explain_ExecutorEnd(QueryDesc *queryDesc); @@ -232,15 +232,15 @@ explain_ExecutorStart(QueryDesc *queryDesc, int eflags) * ExecutorRun hook: all we need do is track nesting depth */ static void -explain_ExecutorRun(QueryDesc *queryDesc, ScanDirection direction, long count) +explain_ExecutorRun(QueryDesc *queryDesc, ScanDirection direction, long count, long size) { nesting_level++; PG_TRY(); { if (prev_ExecutorRun) - prev_ExecutorRun(queryDesc, direction, count); + prev_ExecutorRun(queryDesc, direction, count, size); else - standard_ExecutorRun(queryDesc, direction, count); + standard_ExecutorRun(queryDesc, direction, count, size); nesting_level--; } PG_CATCH(); diff --git a/contrib/pg_stat_statements/pg_stat_statements.c b/contrib/pg_stat_statements/pg_stat_statements.c index 0eb991c..593d406 100644 --- a/contrib/pg_stat_statements/pg_stat_statements.c +++ b/contrib/pg_stat_statements/pg_stat_statements.c @@ -289,7 +289,7 @@ static void pgss_post_parse_analyze(ParseState *pstate, Query *query); static void pgss_ExecutorStart(QueryDesc *queryDesc, int eflags); static void pgss_ExecutorRun(QueryDesc *queryDesc, ScanDirection direction, - long count); + long count, long size); static void pgss_ExecutorFinish(QueryDesc *queryDesc); static void pgss_ExecutorEnd(QueryDesc *queryDesc); static void pgss_ProcessUtility(Node *parsetree, const char *queryString, @@ -870,15 +870,15 @@ pgss_ExecutorStart(QueryDesc *queryDesc, int eflags) * ExecutorRun hook: all we need do is track nesting depth */ static void -pgss_ExecutorRun(QueryDesc *queryDesc, ScanDirection direction, long count) +pgss_ExecutorRun(QueryDesc *queryDesc, ScanDirection direction, long count, long size) { nested_level++; PG_TRY(); { if (prev_ExecutorRun) - prev_ExecutorRun(queryDesc, direction, count); + prev_ExecutorRun(queryDesc, direction, count, size); else - standard_ExecutorRun(queryDesc, direction, count); + standard_ExecutorRun(queryDesc, direction, count, size); nested_level--; } PG_CATCH(); diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index 40cac3b..0419cde 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -48,6 +48,11 @@ PG_MODULE_MAGIC; /* Default CPU cost to process 1 row (above and beyond cpu_tuple_cost). */ #define DEFAULT_FDW_TUPLE_COST 0.01 +/* Maximum tuples per fetch */ +#define MAX_FETCH_SIZE 10000 + +/* Maximum memory usable for retrieved data */ +#define MAX_FETCH_MEM (512 * 1024) /* * FDW-specific planner information kept in RelOptInfo.fdw_private for a * foreign table. This information is collected by postgresGetForeignRelSize. @@ -166,6 +171,8 @@ typedef struct PgFdwScanState MemoryContext batch_cxt; /* context holding current batch of tuples */ MemoryContext temp_cxt; /* context for per-tuple temporary data */ ExprContext *econtext; /* copy of ps_ExprContext of ForeignScanState */ + long max_palloced_mem; /* For test, remove me later */ + int max_numrows; } PgFdwScanState; /* @@ -331,6 +338,8 @@ static int postgresAcquireSampleRowsFunc(Relation relation, int elevel, double *totaldeadrows); static void analyze_row_processor(PGresult *res, int row, PgFdwAnalyzeState *astate); +static Size estimate_tuple_overhead(TupleDesc tupDesc, + List *retrieved_attrs); static HeapTuple make_tuple_from_result_row(PGresult *res, int row, Relation rel, @@ -1138,6 +1147,7 @@ postgresEndForeignScan(ForeignScanState *node) if (fsstate == NULL) return; + elog(LOG, "Max memory for tuple store = %ld, max numrows = %d", fsstate->max_palloced_mem, fsstate->max_numrows); /* Close the cursor if open, to prevent accumulation of cursors */ if (fsstate->cursor_exists) close_cursor(fsstate->conn, fsstate->cursor_number); @@ -2092,18 +2102,20 @@ fetch_more_data(PgFdwScanState *fsstate, fetch_mode cmd) { PgFdwConn *conn = fsstate->conn; char sql[64]; - int fetch_size; + int fetch_mem; + int tuple_overhead; int numrows, addrows, restrows; HeapTuple *tmptuples; int i; int fetch_buf_size; - /* The fetch size is arbitrary, but shouldn't be enormous. */ - fetch_size = 100; - - /* Make the query to fetch tuples */ - snprintf(sql, sizeof(sql), "FETCH %d FROM c%u", - fetch_size, fsstate->cursor_number); + tuple_overhead = estimate_tuple_overhead(fsstate->attinmeta->tupdesc, + fsstate->retrieved_attrs); + fetch_mem = MAX_FETCH_MEM - MAX_FETCH_SIZE * sizeof(HeapTuple); + snprintf(sql, sizeof(sql), "FETCH %d LIMIT %d (%d) FROM c%u", + MAX_FETCH_SIZE, + fetch_mem, tuple_overhead, + fsstate->cursor_number); if (PFCisAsyncRunning(conn)) { @@ -2123,17 +2135,15 @@ fetch_more_data(PgFdwScanState *fsstate, fetch_mode cmd) { /* Get result of running async fetch */ res = PFCgetResult(conn); - if (PQntuples(res) == fetch_size) - { - /* - * Connection state doesn't go to IDLE even if all data - * has been sent to client for asynchronous query. One - * more PQgetResult() is needed to reset the state to - * IDLE. See PQexecFinish() for details. - */ - if (PFCgetResult(conn) != NULL) - elog(ERROR, "Connection status error."); - } + + /* + * Connection state doesn't go to IDLE even if all data + * has been sent to client for asynchronous query. One + * more PQgetResult() is needed to reset the state to + * IDLE. See PQexecFinish() for details. + */ + if (PFCgetResult(conn) != NULL) + elog(ERROR, "Connection status error."); } PFCsetAsyncScan(conn, NULL); } @@ -2161,6 +2171,8 @@ fetch_more_data(PgFdwScanState *fsstate, fetch_mode cmd) /* allocate tuple storage */ tmptuples = fsstate->tuples; addrows = PQntuples(res); + if (fsstate->max_numrows < addrows) + fsstate->max_numrows = addrows; restrows = fsstate->num_tuples - fsstate->next_tuple; numrows = restrows + addrows; fetch_buf_size = numrows * sizeof(HeapTuple); @@ -2188,12 +2200,15 @@ fetch_more_data(PgFdwScanState *fsstate, fetch_mode cmd) fetch_buf_size += (HEAPTUPLESIZE + tup->t_len); } + if (fsstate->max_palloced_mem < fetch_buf_size) + fsstate->max_palloced_mem = fetch_buf_size; + /* Update fetch_ct_2 */ if (fsstate->fetch_ct_2 < 2) fsstate->fetch_ct_2++; - /* Must be EOF if we didn't get as many tuples as we asked for. */ - fsstate->eof_reached = (numrows < fetch_size); + /* Must be EOF if we have no new tuple here. */ + fsstate->eof_reached = (addrows == 0); PQclear(res); res = NULL; @@ -3007,6 +3022,43 @@ postgresImportForeignSchema(ImportForeignSchemaStmt *stmt, Oid serverOid) } /* + * Compute the estimated overhead of the result tuples + * See heap_form_tuple for the details of this calculation. + */ +static Size +estimate_tuple_overhead(TupleDesc tupDesc, + List *retrieved_attrs) +{ + Size size = 0; + int ncol = list_length(retrieved_attrs); + ListCell *lc; + + size += offsetof(HeapTupleHeaderData, t_bits); + size += BITMAPLEN(ncol); + + if (tupDesc->tdhasoid) + size += sizeof(Oid); + + size = MAXALIGN(size); + + size += sizeof(Datum) * ncol; + size += sizeof(bool) * ncol; + + foreach (lc, retrieved_attrs) + { + int i = lfirst_int(lc); + + if (i > 0) + { + if (tupDesc->attrs[i - 1]->attbyval) + size -= (sizeof(Datum) - tupDesc->attrs[i - 1]->attlen); + } + } + + return size; +} + +/* * Create a tuple from the specified row of the PGresult. * * rel is the local representation of the foreign table, attinmeta is diff --git a/src/backend/access/common/heaptuple.c b/src/backend/access/common/heaptuple.c index 09aea79..17525b5 100644 --- a/src/backend/access/common/heaptuple.c +++ b/src/backend/access/common/heaptuple.c @@ -133,6 +133,48 @@ heap_compute_data_size(TupleDesc tupleDesc, return data_length; } +Size +slot_compute_raw_data_size(TupleTableSlot *slot) +{ + TupleDesc tupleDesc = slot->tts_tupleDescriptor; + Datum *values = slot->tts_values; + bool *isnull = slot->tts_isnull; + Size data_length = 0; + int i; + int numberOfAttributes = tupleDesc->natts; + Form_pg_attribute *att = tupleDesc->attrs; + + if (slot->tts_nvalid < tupleDesc->natts) + heap_deform_tuple(slot->tts_tuple, tupleDesc, + slot->tts_values, slot->tts_isnull); + + for (i = 0; i < numberOfAttributes; i++) + { + Datum val; + Form_pg_attribute atti; + + if (isnull[i]) + continue; + + val = values[i]; + atti = att[i]; + + if (atti->attlen == -1) + { + data_length += toast_raw_datum_size(val); + } + else + { + data_length = att_align_datum(data_length, atti->attalign, + atti->attlen, val); + data_length = att_addlength_datum(data_length, atti->attlen, + val); + } + } + + return data_length; +} + /* * heap_fill_tuple * Load data portion of a tuple from values/isnull arrays diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c index 8904676..463fc67 100644 --- a/src/backend/commands/copy.c +++ b/src/backend/commands/copy.c @@ -1928,7 +1928,7 @@ CopyTo(CopyState cstate) else { /* run the plan --- the dest receiver will send tuples */ - ExecutorRun(cstate->queryDesc, ForwardScanDirection, 0L); + ExecutorRun(cstate->queryDesc, ForwardScanDirection, 0L, 0L, 0); processed = ((DR_copy *) cstate->queryDesc->dest)->processed; } diff --git a/src/backend/commands/createas.c b/src/backend/commands/createas.c index 41183f6..7612391 100644 --- a/src/backend/commands/createas.c +++ b/src/backend/commands/createas.c @@ -192,7 +192,7 @@ ExecCreateTableAs(CreateTableAsStmt *stmt, const char *queryString, dir = ForwardScanDirection; /* run the plan */ - ExecutorRun(queryDesc, dir, 0L); + ExecutorRun(queryDesc, dir, 0L, 0L, 0); /* save the rowcount if we're given a completionTag to fill */ if (completionTag) diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c index 0d1ecc2..4480343 100644 --- a/src/backend/commands/explain.c +++ b/src/backend/commands/explain.c @@ -498,7 +498,7 @@ ExplainOnePlan(PlannedStmt *plannedstmt, IntoClause *into, ExplainState *es, dir = ForwardScanDirection; /* run the plan */ - ExecutorRun(queryDesc, dir, 0L); + ExecutorRun(queryDesc, dir, 0L, 0L, 0); /* run cleanup too */ ExecutorFinish(queryDesc); diff --git a/src/backend/commands/extension.c b/src/backend/commands/extension.c index 2b1dcd0..bc116f9 100644 --- a/src/backend/commands/extension.c +++ b/src/backend/commands/extension.c @@ -733,7 +733,7 @@ execute_sql_string(const char *sql, const char *filename) dest, NULL, 0); ExecutorStart(qdesc, 0); - ExecutorRun(qdesc, ForwardScanDirection, 0); + ExecutorRun(qdesc, ForwardScanDirection, 0L, 0L, 0); ExecutorFinish(qdesc); ExecutorEnd(qdesc); diff --git a/src/backend/commands/matview.c b/src/backend/commands/matview.c index 5492e59..39e29ba 100644 --- a/src/backend/commands/matview.c +++ b/src/backend/commands/matview.c @@ -363,7 +363,7 @@ refresh_matview_datafill(DestReceiver *dest, Query *query, ExecutorStart(queryDesc, EXEC_FLAG_WITHOUT_OIDS); /* run the plan */ - ExecutorRun(queryDesc, ForwardScanDirection, 0L); + ExecutorRun(queryDesc, ForwardScanDirection, 0L, 0L, 0); /* and clean up */ ExecutorFinish(queryDesc); diff --git a/src/backend/commands/portalcmds.c b/src/backend/commands/portalcmds.c index 2794537..85fffc1 100644 --- a/src/backend/commands/portalcmds.c +++ b/src/backend/commands/portalcmds.c @@ -177,6 +177,8 @@ PerformPortalFetch(FetchStmt *stmt, nprocessed = PortalRunFetch(portal, stmt->direction, stmt->howMany, + stmt->howLarge, + stmt->tupoverhead, dest); /* Return command status if wanted */ @@ -375,7 +377,7 @@ PersistHoldablePortal(Portal portal) true); /* Fetch the result set into the tuplestore */ - ExecutorRun(queryDesc, ForwardScanDirection, 0L); + ExecutorRun(queryDesc, ForwardScanDirection, 0L, 0L, 0); (*queryDesc->dest->rDestroy) (queryDesc->dest); queryDesc->dest = NULL; diff --git a/src/backend/commands/prepare.c b/src/backend/commands/prepare.c index fb33d30..46fe4f8 100644 --- a/src/backend/commands/prepare.c +++ b/src/backend/commands/prepare.c @@ -291,7 +291,7 @@ ExecuteQuery(ExecuteStmt *stmt, IntoClause *intoClause, */ PortalStart(portal, paramLI, eflags, GetActiveSnapshot()); - (void) PortalRun(portal, count, false, dest, dest, completionTag); + (void) PortalRun(portal, count, 0L, 0, false, dest, dest, completionTag); PortalDrop(portal, false); diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c index 51a86b2..5f0de97 100644 --- a/src/backend/executor/execMain.c +++ b/src/backend/executor/execMain.c @@ -79,6 +79,8 @@ static void ExecutePlan(EState *estate, PlanState *planstate, CmdType operation, bool sendTuples, long numberTuples, + long sizeTuples, + int tupleOverhead, ScanDirection direction, DestReceiver *dest); static bool ExecCheckRTEPerms(RangeTblEntry *rte); @@ -277,17 +279,20 @@ standard_ExecutorStart(QueryDesc *queryDesc, int eflags) */ void ExecutorRun(QueryDesc *queryDesc, - ScanDirection direction, long count) + ScanDirection direction, long count, long size, int tupoverhead) { if (ExecutorRun_hook) - (*ExecutorRun_hook) (queryDesc, direction, count); + (*ExecutorRun_hook) (queryDesc, direction, + count, size, tupoverhead); else - standard_ExecutorRun(queryDesc, direction, count); + standard_ExecutorRun(queryDesc, direction, + count, size, tupoverhead); } void standard_ExecutorRun(QueryDesc *queryDesc, - ScanDirection direction, long count) + ScanDirection direction, + long count, long size, int tupoverhead) { EState *estate; CmdType operation; @@ -339,6 +344,8 @@ standard_ExecutorRun(QueryDesc *queryDesc, operation, sendTuples, count, + size, + tupoverhead, direction, dest); @@ -1551,22 +1558,27 @@ ExecutePlan(EState *estate, CmdType operation, bool sendTuples, long numberTuples, + long sizeTuples, + int tupleOverhead, ScanDirection direction, DestReceiver *dest) { TupleTableSlot *slot; long current_tuple_count; + long sent_size; /* * initialize local variables */ current_tuple_count = 0; - + sent_size = 0; /* * Set the direction. */ estate->es_direction = direction; + estate->es_stoppedbysize = false; + /* * Loop until we've processed the proper number of tuples from the plan. */ @@ -1621,6 +1633,23 @@ ExecutePlan(EState *estate, current_tuple_count++; if (numberTuples && numberTuples == current_tuple_count) break; + + if (sizeTuples > 0) + { + /* + * Count the size of tuples we've sent + * + * This needs all attributes deformed so a bit slow on some cases. + */ + sent_size += slot_compute_raw_data_size(slot) + tupleOverhead; + + /* Quit when the size limit will be exceeded by this tuple */ + if (sizeTuples < sent_size) + { + estate->es_stoppedbysize = true; + break; + } + } } } diff --git a/src/backend/executor/execUtils.c b/src/backend/executor/execUtils.c index e80bc22..6b59c05 100644 --- a/src/backend/executor/execUtils.c +++ b/src/backend/executor/execUtils.c @@ -126,6 +126,7 @@ CreateExecutorState(void) estate->es_preExecCallbacks = NULL; estate->es_processed = 0; + estate->es_stoppedbysize = false; estate->es_lastoid = InvalidOid; estate->es_top_eflags = 0; diff --git a/src/backend/executor/functions.c b/src/backend/executor/functions.c index ce49c47..7ab2e67 100644 --- a/src/backend/executor/functions.c +++ b/src/backend/executor/functions.c @@ -853,7 +853,7 @@ postquel_getnext(execution_state *es, SQLFunctionCachePtr fcache) /* Run regular commands to completion unless lazyEval */ long count = (es->lazyEval) ? 1L : 0L; - ExecutorRun(es->qd, ForwardScanDirection, count); + ExecutorRun(es->qd, ForwardScanDirection, count, 0L, 0); /* * If we requested run to completion OR there was no tuple returned, diff --git a/src/backend/executor/spi.c b/src/backend/executor/spi.c index d544ad9..f29c3a8 100644 --- a/src/backend/executor/spi.c +++ b/src/backend/executor/spi.c @@ -2399,7 +2399,7 @@ _SPI_pquery(QueryDesc *queryDesc, bool fire_triggers, long tcount) ExecutorStart(queryDesc, eflags); - ExecutorRun(queryDesc, ForwardScanDirection, tcount); + ExecutorRun(queryDesc, ForwardScanDirection, tcount, 0L, 0); _SPI_current->processed = queryDesc->estate->es_processed; _SPI_current->lastoid = queryDesc->estate->es_lastoid; @@ -2477,7 +2477,7 @@ _SPI_cursor_operation(Portal portal, FetchDirection direction, long count, /* Run the cursor */ nfetched = PortalRunFetch(portal, direction, - count, + count, 0L, 0, dest); /* diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index e0ff6f1..b7b061c 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -538,6 +538,8 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); %type opt_existing_window_name %type opt_if_not_exists +%type opt_overhead + /* * Non-keyword token types. These are hard-wired into the "flex" lexer. * They must be listed first so that their numeric codes do not depend on @@ -6066,6 +6068,16 @@ fetch_args: cursor_name n->howMany = $1; $$ = (Node *)n; } + | SignedIconst LIMIT Iconst opt_overhead opt_from_in cursor_name + { + FetchStmt *n = makeNode(FetchStmt); + n->portalname = $6; + n->direction = FETCH_FORWARD; + n->howMany = $1; + n->howLarge = $3; + n->tupoverhead = $4; + $$ = (Node *)n; + } | ALL opt_from_in cursor_name { FetchStmt *n = makeNode(FetchStmt); @@ -6074,6 +6086,16 @@ fetch_args: cursor_name n->howMany = FETCH_ALL; $$ = (Node *)n; } + | ALL LIMIT Iconst opt_overhead opt_from_in cursor_name + { + FetchStmt *n = makeNode(FetchStmt); + n->portalname = $6; + n->direction = FETCH_FORWARD; + n->howMany = FETCH_ALL; + n->howLarge = $3; + n->tupoverhead = $4; + $$ = (Node *)n; + } | FORWARD opt_from_in cursor_name { FetchStmt *n = makeNode(FetchStmt); @@ -6090,6 +6112,16 @@ fetch_args: cursor_name n->howMany = $2; $$ = (Node *)n; } + | FORWARD SignedIconst LIMIT Iconst opt_overhead opt_from_in cursor_name + { + FetchStmt *n = makeNode(FetchStmt); + n->portalname = $7; + n->direction = FETCH_FORWARD; + n->howMany = $2; + n->howLarge = $4; + n->tupoverhead = $5; + $$ = (Node *)n; + } | FORWARD ALL opt_from_in cursor_name { FetchStmt *n = makeNode(FetchStmt); @@ -6098,6 +6130,16 @@ fetch_args: cursor_name n->howMany = FETCH_ALL; $$ = (Node *)n; } + | FORWARD ALL LIMIT Iconst opt_overhead opt_from_in cursor_name + { + FetchStmt *n = makeNode(FetchStmt); + n->portalname = $7; + n->direction = FETCH_FORWARD; + n->howMany = FETCH_ALL; + n->howLarge = $4; + n->tupoverhead = $5; + $$ = (Node *)n; + } | BACKWARD opt_from_in cursor_name { FetchStmt *n = makeNode(FetchStmt); @@ -6114,6 +6156,16 @@ fetch_args: cursor_name n->howMany = $2; $$ = (Node *)n; } + | BACKWARD SignedIconst LIMIT Iconst opt_overhead opt_from_in cursor_name + { + FetchStmt *n = makeNode(FetchStmt); + n->portalname = $7; + n->direction = FETCH_BACKWARD; + n->howMany = $2; + n->howLarge = $4; + n->tupoverhead = $5; + $$ = (Node *)n; + } | BACKWARD ALL opt_from_in cursor_name { FetchStmt *n = makeNode(FetchStmt); @@ -6122,6 +6174,16 @@ fetch_args: cursor_name n->howMany = FETCH_ALL; $$ = (Node *)n; } + | BACKWARD ALL LIMIT Iconst opt_overhead opt_from_in cursor_name + { + FetchStmt *n = makeNode(FetchStmt); + n->portalname = $7; + n->direction = FETCH_BACKWARD; + n->howMany = FETCH_ALL; + n->howLarge = $4; + n->tupoverhead = $5; + $$ = (Node *)n; + } ; from_in: FROM {} @@ -6132,6 +6194,9 @@ opt_from_in: from_in {} | /* EMPTY */ {} ; +opt_overhead: '(' Iconst ')' { $$ = $2;} + | /* EMPTY */ { $$ = 0; } + ; /***************************************************************************** * diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c index ce4bdaf..70641eb 100644 --- a/src/backend/tcop/postgres.c +++ b/src/backend/tcop/postgres.c @@ -1103,6 +1103,7 @@ exec_simple_query(const char *query_string) */ (void) PortalRun(portal, FETCH_ALL, + 0L, 0, isTopLevel, receiver, receiver, @@ -1987,6 +1988,7 @@ exec_execute_message(const char *portal_name, long max_rows) completed = PortalRun(portal, max_rows, + 0L, 0, true, /* always top level */ receiver, receiver, diff --git a/src/backend/tcop/pquery.c b/src/backend/tcop/pquery.c index 9c14e8a..ce9541a 100644 --- a/src/backend/tcop/pquery.c +++ b/src/backend/tcop/pquery.c @@ -16,6 +16,7 @@ #include "postgres.h" #include "access/xact.h" +#include "access/htup_details.h" #include "commands/prepare.h" #include "executor/tstoreReceiver.h" #include "miscadmin.h" @@ -39,9 +40,11 @@ static void ProcessQuery(PlannedStmt *plan, DestReceiver *dest, char *completionTag); static void FillPortalStore(Portal portal, bool isTopLevel); -static uint32 RunFromStore(Portal portal, ScanDirection direction, long count, +static uint32 RunFromStore(Portal portal, ScanDirection direction, + long count, long size, int tupoverhead, bool *stoppedbysize, DestReceiver *dest); -static long PortalRunSelect(Portal portal, bool forward, long count, +static long PortalRunSelect(Portal portal, bool forward, + long count, long size, int tupoverhead, DestReceiver *dest); static void PortalRunUtility(Portal portal, Node *utilityStmt, bool isTopLevel, DestReceiver *dest, char *completionTag); @@ -51,6 +54,8 @@ static void PortalRunMulti(Portal portal, bool isTopLevel, static long DoPortalRunFetch(Portal portal, FetchDirection fdirection, long count, + long size, + int tupoverehad, DestReceiver *dest); static void DoPortalRewind(Portal portal); @@ -182,7 +187,7 @@ ProcessQuery(PlannedStmt *plan, /* * Run the plan to completion. */ - ExecutorRun(queryDesc, ForwardScanDirection, 0L); + ExecutorRun(queryDesc, ForwardScanDirection, 0L, 0L, 0); /* * Build command completion status string, if caller wants one. @@ -703,8 +708,8 @@ PortalSetResultFormat(Portal portal, int nFormats, int16 *formats) * suspended due to exhaustion of the count parameter. */ bool -PortalRun(Portal portal, long count, bool isTopLevel, - DestReceiver *dest, DestReceiver *altdest, +PortalRun(Portal portal, long count, long size, int tupoverhead, + bool isTopLevel, DestReceiver *dest, DestReceiver *altdest, char *completionTag) { bool result; @@ -787,7 +792,8 @@ PortalRun(Portal portal, long count, bool isTopLevel, /* * Now fetch desired portion of results. */ - nprocessed = PortalRunSelect(portal, true, count, dest); + nprocessed = PortalRunSelect(portal, true, + count, size, tupoverhead, dest); /* * If the portal result contains a command tag and the caller @@ -892,11 +898,14 @@ static long PortalRunSelect(Portal portal, bool forward, long count, + long size, + int tupoverhead, DestReceiver *dest) { QueryDesc *queryDesc; ScanDirection direction; uint32 nprocessed; + bool stoppedbysize; /* * NB: queryDesc will be NULL if we are fetching from a held cursor or a @@ -939,12 +948,15 @@ PortalRunSelect(Portal portal, count = 0; if (portal->holdStore) - nprocessed = RunFromStore(portal, direction, count, dest); + nprocessed = RunFromStore(portal, direction, + count, size, tupoverhead, + &stoppedbysize, dest); else { PushActiveSnapshot(queryDesc->snapshot); - ExecutorRun(queryDesc, direction, count); + ExecutorRun(queryDesc, direction, count, size, tupoverhead); nprocessed = queryDesc->estate->es_processed; + stoppedbysize = queryDesc->estate->es_stoppedbysize; PopActiveSnapshot(); } @@ -954,8 +966,9 @@ PortalRunSelect(Portal portal, if (nprocessed > 0) portal->atStart = false; /* OK to go backward now */ - if (count == 0 || - (unsigned long) nprocessed < (unsigned long) count) + if ((count == 0 || + (unsigned long) nprocessed < (unsigned long) count) && + !stoppedbysize) portal->atEnd = true; /* we retrieved 'em all */ oldPos = portal->portalPos; portal->portalPos += nprocessed; @@ -982,12 +995,15 @@ PortalRunSelect(Portal portal, count = 0; if (portal->holdStore) - nprocessed = RunFromStore(portal, direction, count, dest); + nprocessed = RunFromStore(portal, direction, + count, size, tupoverhead, + &stoppedbysize, dest); else { PushActiveSnapshot(queryDesc->snapshot); - ExecutorRun(queryDesc, direction, count); + ExecutorRun(queryDesc, direction, count, size, tupoverhead); nprocessed = queryDesc->estate->es_processed; + stoppedbysize = queryDesc->estate->es_stoppedbysize; PopActiveSnapshot(); } @@ -998,8 +1014,9 @@ PortalRunSelect(Portal portal, portal->atEnd = false; /* OK to go forward now */ portal->portalPos++; /* adjust for endpoint case */ } - if (count == 0 || - (unsigned long) nprocessed < (unsigned long) count) + if ((count == 0 || + (unsigned long) nprocessed < (unsigned long) count) && + !stoppedbysize) { portal->atStart = true; /* we retrieved 'em all */ portal->portalPos = 0; @@ -1088,11 +1105,15 @@ FillPortalStore(Portal portal, bool isTopLevel) * out for memory leaks. */ static uint32 -RunFromStore(Portal portal, ScanDirection direction, long count, - DestReceiver *dest) +RunFromStore(Portal portal, ScanDirection direction, + long count, long size_limit, int tupoverhead, + bool *stoppedbysize, DestReceiver *dest) { long current_tuple_count = 0; TupleTableSlot *slot; + long sent_size = 0; + + *stoppedbysize = false; slot = MakeSingleTupleTableSlot(portal->tupDesc); @@ -1122,6 +1143,9 @@ RunFromStore(Portal portal, ScanDirection direction, long count, break; (*dest->receiveSlot) (slot, dest); + /* Count the size of tuples we've sent */ + sent_size += slot_compute_raw_data_size(slot) + + tupoverhead; ExecClearTuple(slot); @@ -1133,10 +1157,19 @@ RunFromStore(Portal portal, ScanDirection direction, long count, current_tuple_count++; if (count && count == current_tuple_count) break; + + /* Quit when the size limit will be exceeded by this tuple */ + if (current_tuple_count > 0 && + size_limit > 0 && size_limit < sent_size) + { + *stoppedbysize = true; + break; + } } } (*dest->rShutdown) (dest); + elog(LOG, "Sent %ld bytes", sent_size); ExecDropSingleTupleTableSlot(slot); @@ -1385,6 +1418,8 @@ long PortalRunFetch(Portal portal, FetchDirection fdirection, long count, + long size, + int tupoverhead, DestReceiver *dest) { long result; @@ -1422,7 +1457,8 @@ PortalRunFetch(Portal portal, switch (portal->strategy) { case PORTAL_ONE_SELECT: - result = DoPortalRunFetch(portal, fdirection, count, dest); + result = DoPortalRunFetch(portal, fdirection, + count, size, tupoverhead, dest); break; case PORTAL_ONE_RETURNING: @@ -1439,7 +1475,8 @@ PortalRunFetch(Portal portal, /* * Now fetch desired portion of results. */ - result = DoPortalRunFetch(portal, fdirection, count, dest); + result = DoPortalRunFetch(portal, fdirection, + count, size, tupoverhead, dest); break; default: @@ -1484,6 +1521,8 @@ static long DoPortalRunFetch(Portal portal, FetchDirection fdirection, long count, + long size, + int tupoverhead, DestReceiver *dest) { bool forward; @@ -1526,7 +1565,7 @@ DoPortalRunFetch(Portal portal, { DoPortalRewind(portal); if (count > 1) - PortalRunSelect(portal, true, count - 1, + PortalRunSelect(portal, true, count - 1, 0L, 0, None_Receiver); } else @@ -1536,13 +1575,15 @@ DoPortalRunFetch(Portal portal, if (portal->atEnd) pos++; /* need one extra fetch if off end */ if (count <= pos) - PortalRunSelect(portal, false, pos - count + 1, + PortalRunSelect(portal, false, + pos - count + 1, 0L, 0, None_Receiver); else if (count > pos + 1) - PortalRunSelect(portal, true, count - pos - 1, + PortalRunSelect(portal, true, + count - pos - 1, 0L, 0, None_Receiver); } - return PortalRunSelect(portal, true, 1L, dest); + return PortalRunSelect(portal, true, 1L, 0L, 0, dest); } else if (count < 0) { @@ -1553,17 +1594,19 @@ DoPortalRunFetch(Portal portal, * (Is it worth considering case where count > half of size of * query? We could rewind once we know the size ...) */ - PortalRunSelect(portal, true, FETCH_ALL, None_Receiver); + PortalRunSelect(portal, true, + FETCH_ALL, 0L, 0, None_Receiver); if (count < -1) - PortalRunSelect(portal, false, -count - 1, None_Receiver); - return PortalRunSelect(portal, false, 1L, dest); + PortalRunSelect(portal, false, + -count - 1, 0, 0, None_Receiver); + return PortalRunSelect(portal, false, 1L, 0L, 0, dest); } else { /* count == 0 */ /* Rewind to start, return zero rows */ DoPortalRewind(portal); - return PortalRunSelect(portal, true, 0L, dest); + return PortalRunSelect(portal, true, 0L, 0L, 0, dest); } break; case FETCH_RELATIVE: @@ -1573,8 +1616,9 @@ DoPortalRunFetch(Portal portal, * Definition: advance count-1 rows, return next row (if any). */ if (count > 1) - PortalRunSelect(portal, true, count - 1, None_Receiver); - return PortalRunSelect(portal, true, 1L, dest); + PortalRunSelect(portal, true, + count - 1, 0L, 0, None_Receiver); + return PortalRunSelect(portal, true, 1L, 0L, 0, dest); } else if (count < 0) { @@ -1583,8 +1627,9 @@ DoPortalRunFetch(Portal portal, * any). */ if (count < -1) - PortalRunSelect(portal, false, -count - 1, None_Receiver); - return PortalRunSelect(portal, false, 1L, dest); + PortalRunSelect(portal, false, + -count - 1, 0L, 0, None_Receiver); + return PortalRunSelect(portal, false, 1L, 0L, 0, dest); } else { @@ -1630,7 +1675,7 @@ DoPortalRunFetch(Portal portal, */ if (on_row) { - PortalRunSelect(portal, false, 1L, None_Receiver); + PortalRunSelect(portal, false, 1L, 0L, 0, None_Receiver); /* Set up to fetch one row forward */ count = 1; forward = true; @@ -1652,7 +1697,7 @@ DoPortalRunFetch(Portal portal, return result; } - return PortalRunSelect(portal, forward, count, dest); + return PortalRunSelect(portal, forward, count, size, tupoverhead, dest); } /* diff --git a/src/include/access/htup_details.h b/src/include/access/htup_details.h index 55d483d..5f0c8f3 100644 --- a/src/include/access/htup_details.h +++ b/src/include/access/htup_details.h @@ -20,6 +20,7 @@ #include "access/transam.h" #include "storage/bufpage.h" +#include "executor/tuptable.h" /* * MaxTupleAttributeNumber limits the number of (user) columns in a tuple. * The key limit on this value is that the size of the fixed overhead for @@ -761,6 +762,7 @@ extern Datum fastgetattr(HeapTuple tup, int attnum, TupleDesc tupleDesc, /* prototypes for functions in common/heaptuple.c */ extern Size heap_compute_data_size(TupleDesc tupleDesc, Datum *values, bool *isnull); +extern Size slot_compute_raw_data_size(TupleTableSlot *slot); extern void heap_fill_tuple(TupleDesc tupleDesc, Datum *values, bool *isnull, char *data, Size data_size, diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h index 193a654..e2706a6 100644 --- a/src/include/executor/executor.h +++ b/src/include/executor/executor.h @@ -79,8 +79,8 @@ extern PGDLLIMPORT ExecutorStart_hook_type ExecutorStart_hook; /* Hook for plugins to get control in ExecutorRun() */ typedef void (*ExecutorRun_hook_type) (QueryDesc *queryDesc, - ScanDirection direction, - long count); + ScanDirection direction, + long count, long size, int tupoverhead); extern PGDLLIMPORT ExecutorRun_hook_type ExecutorRun_hook; /* Hook for plugins to get control in ExecutorFinish() */ @@ -175,9 +175,9 @@ extern TupleTableSlot *ExecFilterJunk(JunkFilter *junkfilter, extern void ExecutorStart(QueryDesc *queryDesc, int eflags); extern void standard_ExecutorStart(QueryDesc *queryDesc, int eflags); extern void ExecutorRun(QueryDesc *queryDesc, - ScanDirection direction, long count); + ScanDirection direction, long count, long size, int tupoverhead); extern void standard_ExecutorRun(QueryDesc *queryDesc, - ScanDirection direction, long count); + ScanDirection direction, long count, long size, int tupoverhead); extern void ExecutorFinish(QueryDesc *queryDesc); extern void standard_ExecutorFinish(QueryDesc *queryDesc); extern void ExecutorEnd(QueryDesc *queryDesc); diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index cb8d854..f8121ec 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -410,6 +410,7 @@ typedef struct EState PreExecCallbackItem *es_preExecCallbacks; /* pre-exec callbacks */ uint32 es_processed; /* # of tuples processed */ + bool es_stoppedbysize; /* true if processing stopped by size */ Oid es_lastoid; /* last oid processed (by INSERT) */ int es_top_eflags; /* eflags passed to ExecutorStart */ diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index 868905b..094c0ac 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -2393,6 +2393,8 @@ typedef struct FetchStmt NodeTag type; FetchDirection direction; /* see above */ long howMany; /* number of rows, or position argument */ + long howLarge; /* total bytes of rows */ + int tupoverhead; /* declared overhead per tuple in client */ char *portalname; /* name of portal (cursor) */ bool ismove; /* TRUE if MOVE */ } FetchStmt; diff --git a/src/include/tcop/pquery.h b/src/include/tcop/pquery.h index 8073a6e..021532c 100644 --- a/src/include/tcop/pquery.h +++ b/src/include/tcop/pquery.h @@ -17,7 +17,6 @@ #include "nodes/parsenodes.h" #include "utils/portal.h" - extern PGDLLIMPORT Portal ActivePortal; @@ -33,13 +32,15 @@ extern void PortalStart(Portal portal, ParamListInfo params, extern void PortalSetResultFormat(Portal portal, int nFormats, int16 *formats); -extern bool PortalRun(Portal portal, long count, bool isTopLevel, - DestReceiver *dest, DestReceiver *altdest, +extern bool PortalRun(Portal portal, long count, long size, int tupoverhead, + bool isTopLevel, DestReceiver *dest, DestReceiver *altdest, char *completionTag); extern long PortalRunFetch(Portal portal, FetchDirection fdirection, long count, + long size, + int tupoverhead, DestReceiver *dest); #endif /* PQUERY_H */ diff --git a/src/interfaces/ecpg/preproc/ecpg.addons b/src/interfaces/ecpg/preproc/ecpg.addons index b3b36cf..424f412 100644 --- a/src/interfaces/ecpg/preproc/ecpg.addons +++ b/src/interfaces/ecpg/preproc/ecpg.addons @@ -220,13 +220,56 @@ ECPG: fetch_argsNEXTopt_from_incursor_name addon ECPG: fetch_argsPRIORopt_from_incursor_name addon ECPG: fetch_argsFIRST_Popt_from_incursor_name addon ECPG: fetch_argsLAST_Popt_from_incursor_name addon + add_additional_variables($3, false); + if ($3[0] == ':') + { + free($3); + $3 = mm_strdup("$0"); + } ECPG: fetch_argsALLopt_from_incursor_name addon +ECPG: fetch_argsFORWARDopt_from_incursor_name addon +ECPG: fetch_argsBACKWARDopt_from_incursor_name addon add_additional_variables($3, false); if ($3[0] == ':') { free($3); $3 = mm_strdup("$0"); } +ECPG: fetch_argsALLLIMITIconstopt_overheadopt_from_incursor_name addon + add_additional_variables($6, false); + if ($6[0] == ':') + { + free($6); + $6 = mm_strdup("$0"); + } + if ($3[0] == '$') + { + free($3); + $3 = mm_strdup("$0"); + } + if ($4[0] == '$') + { + free($4); + $4 = mm_strdup("$0"); + } +ECPG: fetch_argsFORWARDALLLIMITIconstopt_overheadopt_from_incursor_name addon +ECPG: fetch_argsBACKWARDALLLIMITIconstopt_overheadopt_from_incursor_name addon + add_additional_variables($7, false); + if ($7[0] == ':') + { + free($7); + $7 = mm_strdup("$0"); + } + if ($4[0] == '$') + { + free($4); + $4 = mm_strdup("$0"); + } + if ($5[0] == '$') + { + free($5); + $5 = mm_strdup("$0"); + } ECPG: fetch_argsSignedIconstopt_from_incursor_name addon add_additional_variables($3, false); if ($3[0] == ':') @@ -234,11 +277,51 @@ ECPG: fetch_argsSignedIconstopt_from_incursor_name addon free($3); $3 = mm_strdup("$0"); } +ECPG: fetch_argsSignedIconstLIMITIconstopt_overheadopt_from_incursor_name addon + add_additional_variables($6, false); + if ($6[0] == ':') + { + free($6); + $6 = mm_strdup("$0"); + } if ($1[0] == '$') { free($1); $1 = mm_strdup("$0"); } + if ($3[0] == '$') + { + free($3); + $3 = mm_strdup("$0"); + } + if ($4[0] == '$') + { + free($4); + $4 = mm_strdup("$0"); + } +ECPG: fetch_argsFORWARDSignedIconstLIMITIconstopt_overheadopt_from_incursor_name addon +ECPG: fetch_argsBACKWARDSignedIconstLIMITIconstopt_overheadopt_from_incursor_name addon + add_additional_variables($7, false); + if ($7[0] == ':') + { + free($7); + $7 = mm_strdup("$0"); + } + if ($2[0] == '$') + { + free($2); + $2 = mm_strdup("$0"); + } + if ($4[0] == '$') + { + free($4); + $4 = mm_strdup("$0"); + } + if ($5[0] == '$') + { + free($5); + $5 = mm_strdup("$0"); + } ECPG: fetch_argsFORWARDALLopt_from_incursor_name addon ECPG: fetch_argsBACKWARDALLopt_from_incursor_name addon add_additional_variables($4, false); -- 1.8.3.1