public inbox for [email protected]help / color / mirror / Atom feed
[PATCH v24 03/10] Add default_toast_compression GUC 3+ messages / 2 participants [nested] [flat]
* [PATCH v24 03/10] Add default_toast_compression GUC @ 2021-01-30 03:37 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 3+ messages in thread From: Justin Pryzby @ 2021-01-30 03:37 UTC (permalink / raw) --- src/backend/access/common/tupdesc.c | 2 +- src/backend/bootstrap/bootstrap.c | 3 +- src/backend/commands/amcmds.c | 143 +++++++++++++++++++++++++++- src/backend/commands/tablecmds.c | 4 +- src/backend/utils/init/postinit.c | 4 + src/backend/utils/misc/guc.c | 12 +++ src/include/access/amapi.h | 2 + src/include/access/compressamapi.h | 12 ++- 8 files changed, 176 insertions(+), 6 deletions(-) diff --git a/src/backend/access/common/tupdesc.c b/src/backend/access/common/tupdesc.c index ca26fab487..7afaea000b 100644 --- a/src/backend/access/common/tupdesc.c +++ b/src/backend/access/common/tupdesc.c @@ -668,7 +668,7 @@ TupleDescInitEntry(TupleDesc desc, att->attcollation = typeForm->typcollation; if (IsStorageCompressible(typeForm->typstorage)) - att->attcompression = DefaultCompressionOid; + att->attcompression = GetDefaultToastCompression(); else att->attcompression = InvalidOid; diff --git a/src/backend/bootstrap/bootstrap.c b/src/backend/bootstrap/bootstrap.c index 9b451eaa71..ec3376cf8a 100644 --- a/src/backend/bootstrap/bootstrap.c +++ b/src/backend/bootstrap/bootstrap.c @@ -732,8 +732,9 @@ DefineAttr(char *name, char *type, int attnum, int nullness) attrtypes[attnum]->attcacheoff = -1; attrtypes[attnum]->atttypmod = -1; attrtypes[attnum]->attislocal = true; + if (IsStorageCompressible(attrtypes[attnum]->attstorage)) - attrtypes[attnum]->attcompression = DefaultCompressionOid; + attrtypes[attnum]->attcompression = GetDefaultToastCompression(); else attrtypes[attnum]->attcompression = InvalidOid; diff --git a/src/backend/commands/amcmds.c b/src/backend/commands/amcmds.c index 3ad4a61739..1682afd2a4 100644 --- a/src/backend/commands/amcmds.c +++ b/src/backend/commands/amcmds.c @@ -13,8 +13,11 @@ */ #include "postgres.h" +#include "access/amapi.h" +#include "access/compressamapi.h" #include "access/htup_details.h" #include "access/table.h" +#include "access/xact.h" #include "catalog/catalog.h" #include "catalog/dependency.h" #include "catalog/indexing.h" @@ -27,13 +30,20 @@ #include "parser/parse_func.h" #include "utils/builtins.h" #include "utils/lsyscache.h" +#include "utils/guc.h" +#include "utils/inval.h" #include "utils/rel.h" #include "utils/syscache.h" - static Oid lookup_am_handler_func(List *handler_name, char amtype); static const char *get_am_type_string(char amtype); +/* Compile-time default */ +char *default_toast_compression = DEFAULT_TOAST_COMPRESSION; // maybe needs pg_dump support ? +/* Invalid means need to lookup the text value in the catalog */ +static Oid default_toast_compression_oid = InvalidOid; + +static void AccessMethodCallback(Datum arg, int cacheid, uint32 hashvalue); /* * CreateAccessMethod @@ -277,3 +287,134 @@ lookup_am_handler_func(List *handler_name, char amtype) return handlerOid; } + +/* check_hook: validate new default_toast_compression */ +bool +check_default_toast_compression(char **newval, void **extra, GucSource source) +{ + if (**newval == '\0') + { + GUC_check_errdetail("%s cannot be empty.", + "default_toast_compression"); + return false; + } + + if (strlen(*newval) >= NAMEDATALEN) + { + GUC_check_errdetail("%s is too long (maximum %d characters).", + "default_toast_compression", NAMEDATALEN - 1); + return false; + } + + /* + * If we aren't inside a transaction, or not connected to a database, we + * cannot do the catalog access necessary to verify the method. Must + * accept the value on faith. + */ + if (IsTransactionState() && MyDatabaseId != InvalidOid) + { + if (!OidIsValid(get_compression_am_oid(*newval, true))) + { + /* + * When source == PGC_S_TEST, don't throw a hard error for a + * nonexistent table access method, only a NOTICE. See comments in + * guc.h. + */ + if (source == PGC_S_TEST) + { + ereport(NOTICE, + (errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("compression access method \"%s\" does not exist", + *newval))); + } + else + { + GUC_check_errdetail("Compression access method \"%s\" does not exist.", + *newval); + return false; + } + } + } + + return true; +} + +/* + * assign_default_toast_compression: GUC assign_hook for default_toast_compression + */ +void +assign_default_toast_compression(const char *newval, void *extra) +{ + /* + * Invalidate setting, forcing it to be looked up as needed. + * This avoids trying to do database access during GUC initialization, + * or outside a transaction. + */ + + default_toast_compression = NULL; +fprintf(stderr, "set to null\n"); +} + + +/* + * GetDefaultToastCompression -- get the OID of the current toast compression + * + * This exists to hide and optimize the use of the default_toast_compression + * GUC variable. + */ +Oid +GetDefaultToastCompression(void) +{ + /* Avoid catalog access during bootstrap, and for default compression */ + if (strcmp(default_toast_compression, DEFAULT_TOAST_COMPRESSION) == 0) + return PGLZ_COMPRESSION_AM_OID; + + /* Cannot call get_compression_am_oid this early */ + // if (IsBootstrapProcessingMode()) + // return PGLZ_COMPRESSION_AM_OID; + Assert(!IsBootstrapProcessingMode()); + + /* + * If cached value isn't valid, look up the current default value, caching + * the result + */ + if (!OidIsValid(default_toast_compression_oid)) + default_toast_compression_oid = + get_am_type_oid(default_toast_compression, AMTYPE_COMPRESSION, + false); + + return default_toast_compression_oid; +} + +/* + * InitializeAccessMethods: initialize module during InitPostgres. + * + * This is called after we are up enough to be able to do catalog lookups. + */ +void +InitializeAccessMethods(void) +{ + if (IsBootstrapProcessingMode()) + return; + + /* + * In normal mode, arrange for a callback on any syscache invalidation + * of pg_am rows. + */ + CacheRegisterSyscacheCallback(AMOID, + AccessMethodCallback, + (Datum) 0); + /* Force cached default access method to be recomputed on next use */ + // default_toast_compression_oid = InvalidOid; +} + +/* + * AccessMethodCallback + * Syscache inval callback function + */ +static void +AccessMethodCallback(Datum arg, int cacheid, uint32 hashvalue) +{ + /* Force look up of compression oid on next use */ + default_toast_compression_oid = false; +} diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index dd81d5bf4e..72ba017814 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -11925,7 +11925,7 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel, if (!IsStorageCompressible(tform->typstorage)) attTup->attcompression = InvalidOid; else if (!OidIsValid(attTup->attcompression)) - attTup->attcompression = DefaultCompressionOid; + attTup->attcompression = GetDefaultToastCompression(); } else attTup->attcompression = InvalidOid; @@ -17762,7 +17762,7 @@ GetAttributeCompression(Form_pg_attribute att, char *compression) /* fallback to default compression if it's not specified */ if (compression == NULL) - return DefaultCompressionOid; + return GetDefaultToastCompression(); amoid = get_compression_am_oid(compression, false); diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c index e5965bc517..6a02bbd377 100644 --- a/src/backend/utils/init/postinit.c +++ b/src/backend/utils/init/postinit.c @@ -25,6 +25,7 @@ #include "access/session.h" #include "access/sysattr.h" #include "access/tableam.h" +#include "access/amapi.h" #include "access/xact.h" #include "access/xlog.h" #include "catalog/catalog.h" @@ -1060,6 +1061,9 @@ InitPostgres(const char *in_dbname, Oid dboid, const char *username, /* set default namespace search path */ InitializeSearchPath(); + /* set callback for changes to pg_am */ + InitializeAccessMethods(); + /* initialize client encoding */ InitializeClientEncoding(); diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index eafdb1118e..3a2b33fcd1 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -30,6 +30,7 @@ #include <unistd.h> #include "access/commit_ts.h" +#include "access/compressamapi.h" #include "access/gin.h" #include "access/rmgr.h" #include "access/tableam.h" @@ -3915,6 +3916,17 @@ static struct config_string ConfigureNamesString[] = check_default_table_access_method, NULL, NULL }, + { + {"default_toast_compression", PGC_USERSET, CLIENT_CONN_STATEMENT, + gettext_noop("Sets the default compression for new columns."), + NULL, + GUC_IS_NAME + }, + &default_toast_compression, + DEFAULT_TOAST_COMPRESSION, + check_default_toast_compression, assign_default_toast_compression, NULL, NULL + }, + { {"default_tablespace", PGC_USERSET, CLIENT_CONN_STATEMENT, gettext_noop("Sets the default tablespace to create tables and indexes in."), diff --git a/src/include/access/amapi.h b/src/include/access/amapi.h index d357ebb559..1513cafcf4 100644 --- a/src/include/access/amapi.h +++ b/src/include/access/amapi.h @@ -287,4 +287,6 @@ typedef struct IndexAmRoutine extern IndexAmRoutine *GetIndexAmRoutine(Oid amhandler); extern IndexAmRoutine *GetIndexAmRoutineByAmId(Oid amoid, bool noerror); +void InitializeAccessMethods(void); + #endif /* AMAPI_H */ diff --git a/src/include/access/compressamapi.h b/src/include/access/compressamapi.h index 5a8e23d926..d75a8e9df2 100644 --- a/src/include/access/compressamapi.h +++ b/src/include/access/compressamapi.h @@ -17,6 +17,7 @@ #include "catalog/pg_am_d.h" #include "nodes/nodes.h" +#include "utils/guc.h" /* * Built-in compression method-id. The toast compression header will store @@ -29,8 +30,17 @@ typedef enum CompressionId LZ4_COMPRESSION_ID = 1 } CompressionId; -/* Use default compression method if it is not specified. */ +/* Default compression method if not specified. */ +#define DEFAULT_TOAST_COMPRESSION "pglz" #define DefaultCompressionOid PGLZ_COMPRESSION_AM_OID + +/* GUC */ +extern char *default_toast_compression; +extern void assign_default_toast_compression(const char *newval, void *extra); +extern bool check_default_toast_compression(char **newval, void **extra, GucSource source); + +extern Oid GetDefaultToastCompression(void); + #define IsStorageCompressible(storage) ((storage) != TYPSTORAGE_PLAIN && \ (storage) != TYPSTORAGE_EXTERNAL) /* compression handler routines */ -- 2.17.0 --m51xatjYGsM+13rf Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v24-0004-default-to-with-lz4.patch" ^ permalink raw reply [nested|flat] 3+ messages in thread
* Re: psql's FETCH_COUNT (cursor) is not being respected for CTEs @ 2023-01-12 12:27 Daniel Verite <[email protected]> 0 siblings, 1 reply; 3+ messages in thread From: Daniel Verite @ 2023-01-12 12:27 UTC (permalink / raw) To: Tom Lane <[email protected]>; +Cc: Robert Haas <[email protected]>; Jakub Wartak <[email protected]>; pgsql-hackers Tom Lane wrote: > I agree that it seems like a good idea to try. > There will be more per-row overhead, but the increase in flexibility > is likely to justify that. Here's a POC patch implementing row-by-row fetching. If it wasn't for the per-row overhead, we could probably get rid of ExecQueryUsingCursor() and use row-by-row fetches whenever FETCH_COUNT is set, independently of the form of the query. However the difference in processing time seems to be substantial: on some quick tests with FETCH_COUNT=10000, I'm seeing almost a 1.5x increase on large datasets. I assume it's the cost of more allocations. I would have hoped that avoiding the FETCH queries and associated round-trips with the cursor method would compensate for that, but it doesn't appear to be the case, at least with a fast local connection. So in this patch, psql still uses the cursor method if the query starts with "select", and falls back to the row-by-row in the main code (ExecQueryAndProcessResults) otherwise. Anyway it solves the main issue of the over-consumption of memory for CTE and update/insert queries returning large resultsets. Best regards, -- Daniel Vérité https://postgresql.verite.pro/ Twitter: @DanielVerite Attachments: [text/x-patch] psql-fetchcount-single-row-mode.diff (11.8K, ../../[email protected]/2-psql-fetchcount-single-row-mode.diff) download | inline diff: diff --git a/src/bin/psql/common.c b/src/bin/psql/common.c index 00627830c4..d3de9d8336 100644 --- a/src/bin/psql/common.c +++ b/src/bin/psql/common.c @@ -372,6 +372,7 @@ AcceptResult(const PGresult *result, bool show_error) { case PGRES_COMMAND_OK: case PGRES_TUPLES_OK: + case PGRES_SINGLE_TUPLE: case PGRES_EMPTY_QUERY: case PGRES_COPY_IN: case PGRES_COPY_OUT: @@ -675,13 +676,13 @@ PrintNotifications(void) * Returns true if successful, false otherwise. */ static bool -PrintQueryTuples(const PGresult *result, const printQueryOpt *opt, +PrintQueryTuples(const PGresult **result, int nresults, const printQueryOpt *opt, FILE *printQueryFout) { bool ok = true; FILE *fout = printQueryFout ? printQueryFout : pset.queryFout; - printQuery(result, opt ? opt : &pset.popt, fout, false, pset.logfile); + printQueryChunks(result, nresults, opt ? opt : &pset.popt, fout, false, pset.logfile); fflush(fout); if (ferror(fout)) { @@ -958,7 +959,7 @@ PrintQueryResult(PGresult *result, bool last, else if (last && pset.crosstab_flag) success = PrintResultInCrosstab(result); else if (last || pset.show_all_results) - success = PrintQueryTuples(result, opt, printQueryFout); + success = PrintQueryTuples((const PGresult**)&result, 1, opt, printQueryFout); else success = true; @@ -1369,6 +1370,47 @@ DescribeQuery(const char *query, double *elapsed_msec) return OK; } +/* + * Check if an output stream for \g needs to be opened, and if + * yes, open it. + * Return false if an error occurred, true otherwise. + */ +static bool +SetupGOutput(PGresult *result, FILE **gfile_fout, bool *is_pipe) +{ + ExecStatusType status = PQresultStatus(result); + if (pset.gfname != NULL && /* there is a \g file or program */ + *gfile_fout == NULL && /* and it's not already opened */ + (status == PGRES_TUPLES_OK || + status == PGRES_SINGLE_TUPLE || + status == PGRES_COPY_OUT)) + { + if (openQueryOutputFile(pset.gfname, gfile_fout, is_pipe)) + { + if (is_pipe) + disable_sigpipe_trap(); + } + else + return false; + } + return true; +} + +static void +CloseGOutput(FILE *gfile_fout, bool is_pipe) +{ + /* close \g file if we opened it */ + if (gfile_fout) + { + if (is_pipe) + { + pclose(gfile_fout); + restore_sigpipe_trap(); + } + else + fclose(gfile_fout); + } +} /* * ExecQueryAndProcessResults: utility function for use by SendQuery() @@ -1400,10 +1442,16 @@ ExecQueryAndProcessResults(const char *query, bool success; instr_time before, after; + int fetch_count = pset.fetch_count; PGresult *result; + FILE *gfile_fout = NULL; bool gfile_is_pipe = false; + PGresult **result_array = NULL; /* to collect results in single row mode */ + int64 total_tuples = 0; + int ntuples; + if (timing) INSTR_TIME_SET_CURRENT(before); @@ -1424,6 +1472,33 @@ ExecQueryAndProcessResults(const char *query, return -1; } + /* + * If FETCH_COUNT is set and the context allows it, use the single row + * mode to fetch results and have no more than FETCH_COUNT rows in + * memory. + */ + if (fetch_count > 0 && !pset.crosstab_flag && !pset.gexec_flag && !is_watch + && !pset.gset_prefix && pset.show_all_results) + { + /* + * The row-by-row fetch is not enabled when SHOW_ALL_RESULTS is false, + * since we would need to accumulate all rows before knowing + * whether they need to be discarded or displayed, which contradicts + * FETCH_COUNT. + */ + if (!PQsetSingleRowMode(pset.db)) + { + pg_log_warning("fetching results in single row mode is unavailable"); + fetch_count = 0; + } + else + { + result_array = (PGresult**) pg_malloc(fetch_count * sizeof(PGresult*)); + } + } + else + fetch_count = 0; /* disable single-row mode */ + /* * If SIGINT is sent while the query is processing, the interrupt will be * consumed. The user's intention, though, is to cancel the entire watch @@ -1443,6 +1518,7 @@ ExecQueryAndProcessResults(const char *query, ExecStatusType result_status; PGresult *next_result; bool last; + bool partial_display = false; if (!AcceptResult(result, false)) { @@ -1569,6 +1645,85 @@ ExecQueryAndProcessResults(const char *query, success &= HandleCopyResult(&result, copy_stream); } + if (fetch_count > 0 && result_status == PGRES_SINGLE_TUPLE) + { + FILE *tuples_fout = printQueryFout; + printQueryOpt my_popt = pset.popt; + + ntuples = 0; + partial_display = true; + + success = SetupGOutput(result, &gfile_fout, &gfile_is_pipe); + if (gfile_fout) + tuples_fout = gfile_fout; + + /* initialize print options for partial table output */ + my_popt.topt.start_table = true; + my_popt.topt.stop_table = false; + my_popt.topt.prior_records = 0; + + while (success) + { + result_array[ntuples++] = result; + if (ntuples == fetch_count) + { + /* TODO: handle paging */ + /* display the current chunk of results */ + PrintQueryTuples(result_array, ntuples, &my_popt, tuples_fout); + /* clear and reuse result_array */ + for (int i=0; i < ntuples; i++) + PQclear(result_array[i]); + /* after the first result set, disallow header decoration */ + my_popt.topt.start_table = false; + my_popt.topt.prior_records += ntuples; + total_tuples += ntuples; + ntuples = 0; + } + + result = PQgetResult(pset.db); + if (result == NULL) + { + /* + * Error. We expect a PGRES_TUPLES_OK result with + * zero tuple in it to finish the row-by-row sequence. + */ + success = false; + break; + } + + if (PQresultStatus(result) == PGRES_TUPLES_OK) + { + /* TODO: merge this block with the code above? */ + /* + * The last row has been read. Display the last chunk of + * results and the footer. + */ + my_popt.topt.stop_table = true; + PrintQueryTuples(result_array, ntuples, &my_popt, tuples_fout); + for (int i=0; i < ntuples; i++) + PQclear(result_array[i]); + total_tuples += ntuples; + ntuples = 0; + + result = NULL; + { + /* + * fake SetResultVariables() as in ExecQueryUsingCursor(). + */ + char buf[32]; + + SetVariable(pset.vars, "ERROR", "false"); + SetVariable(pset.vars, "SQLSTATE", "00000"); + snprintf(buf, sizeof(buf), INT64_FORMAT, total_tuples); + SetVariable(pset.vars, "ROW_COUNT", buf); + } + break; + } + } + } + else + partial_display = false; + /* * Check PQgetResult() again. In the typical case of a single-command * string, it will return NULL. Otherwise, we'll have other results @@ -1597,7 +1752,7 @@ ExecQueryAndProcessResults(const char *query, } /* this may or may not print something depending on settings */ - if (result != NULL) + if (result != NULL && !partial_display) { /* * If results need to be printed into the file specified by \g, @@ -1606,25 +1761,10 @@ ExecQueryAndProcessResults(const char *query, * tuple output, but it's still used for status output. */ FILE *tuples_fout = printQueryFout; - bool do_print = true; - - if (PQresultStatus(result) == PGRES_TUPLES_OK && - pset.gfname) - { - if (gfile_fout == NULL) - { - if (openQueryOutputFile(pset.gfname, - &gfile_fout, &gfile_is_pipe)) - { - if (gfile_is_pipe) - disable_sigpipe_trap(); - } - else - success = do_print = false; - } + success = SetupGOutput(result, &gfile_fout, &gfile_is_pipe); + if (gfile_fout) tuples_fout = gfile_fout; - } - if (do_print) + if (success) success &= PrintQueryResult(result, last, opt, tuples_fout, printQueryFout); } @@ -1643,17 +1783,10 @@ ExecQueryAndProcessResults(const char *query, } } - /* close \g file if we opened it */ - if (gfile_fout) - { - if (gfile_is_pipe) - { - pclose(gfile_fout); - restore_sigpipe_trap(); - } - else - fclose(gfile_fout); - } + CloseGOutput(gfile_fout, gfile_is_pipe); + + if (result_array) + pg_free(result_array); /* may need this to recover from conn loss during COPY */ if (!CheckConnection()) diff --git a/src/fe_utils/print.c b/src/fe_utils/print.c index 3396f9b462..d8f0a29773 100644 --- a/src/fe_utils/print.c +++ b/src/fe_utils/print.c @@ -3533,17 +3533,42 @@ printTable(const printTableContent *cont, void printQuery(const PGresult *result, const printQueryOpt *opt, FILE *fout, bool is_pager, FILE *flog) +{ + printQueryChunks(&result, 1, opt, fout, is_pager, flog); +} + +/* + * Print the results of a query that may have been obtained by a + * succession of calls to PQgetResult in single-row mode. + * + * results: array of results of a successful query. They must have the same columns. + * nbresults: size of results + * opt: formatting options + * fout: where to print to + * is_pager: true if caller has already redirected fout to be a pager pipe + * flog: if not null, also print the data there (for --log-file option) + */ +void +printQueryChunks(const PGresult *results[], int nresults, const printQueryOpt *opt, + FILE *fout, bool is_pager, FILE *flog) { printTableContent cont; int i, r, c; + int nrows = 0; /* total number of rows */ + int ri; /* index into results[] */ if (cancel_pressed) return; + for (ri = 0; ri < nresults; ri++) + { + nrows += PQntuples(results[ri]); + } + printTableInit(&cont, &opt->topt, opt->title, - PQnfields(result), PQntuples(result)); + (nresults > 0) ? PQnfields(results[0]) : 0, nrows); /* Assert caller supplied enough translate_columns[] entries */ Assert(opt->translate_columns == NULL || @@ -3551,34 +3576,37 @@ printQuery(const PGresult *result, const printQueryOpt *opt, for (i = 0; i < cont.ncolumns; i++) { - printTableAddHeader(&cont, PQfname(result, i), + printTableAddHeader(&cont, PQfname(results[0], i), opt->translate_header, - column_type_alignment(PQftype(result, i))); + column_type_alignment(PQftype(results[0], i))); } /* set cells */ - for (r = 0; r < cont.nrows; r++) + for (ri = 0; ri < nresults; ri++) { - for (c = 0; c < cont.ncolumns; c++) + for (r = 0; r < PQntuples(results[ri]); r++) { - char *cell; - bool mustfree = false; - bool translate; - - if (PQgetisnull(result, r, c)) - cell = opt->nullPrint ? opt->nullPrint : ""; - else + for (c = 0; c < cont.ncolumns; c++) { - cell = PQgetvalue(result, r, c); - if (cont.aligns[c] == 'r' && opt->topt.numericLocale) + char *cell; + bool mustfree = false; + bool translate; + + if (PQgetisnull(results[ri], r, c)) + cell = opt->nullPrint ? opt->nullPrint : ""; + else { - cell = format_numeric_locale(cell); - mustfree = true; + cell = PQgetvalue(results[ri], r, c); + if (cont.aligns[c] == 'r' && opt->topt.numericLocale) + { + cell = format_numeric_locale(cell); + mustfree = true; + } } - } - translate = (opt->translate_columns && opt->translate_columns[c]); - printTableAddCell(&cont, cell, translate, mustfree); + translate = (opt->translate_columns && opt->translate_columns[c]); + printTableAddCell(&cont, cell, translate, mustfree); + } } } diff --git a/src/include/fe_utils/print.h b/src/include/fe_utils/print.h index 54f783c907..3befc41bdc 100644 --- a/src/include/fe_utils/print.h +++ b/src/include/fe_utils/print.h @@ -220,7 +220,10 @@ extern void printTableCleanup(printTableContent *const content); extern void printTable(const printTableContent *cont, FILE *fout, bool is_pager, FILE *flog); extern void printQuery(const PGresult *result, const printQueryOpt *opt, - FILE *fout, bool is_pager, FILE *flog); + FILE *fout, bool is_pager, FILE *flog); +extern void printQueryChunks(const PGresult *results[], int nresults, + const printQueryOpt *opt, + FILE *fout, bool is_pager, FILE *flog); extern char column_type_alignment(Oid); ^ permalink raw reply [nested|flat] 3+ messages in thread
* Re: psql's FETCH_COUNT (cursor) is not being respected for CTEs @ 2023-03-01 10:41 Daniel Verite <[email protected]> parent: Daniel Verite <[email protected]> 0 siblings, 0 replies; 3+ messages in thread From: Daniel Verite @ 2023-03-01 10:41 UTC (permalink / raw) To: pgsql-hackers; +Cc: Tom Lane <[email protected]>; Robert Haas <[email protected]>; Jakub Wartak <[email protected]> I wrote: > Here's a POC patch implementing row-by-row fetching. PFA an updated patch. Best regards, -- Daniel Vérité https://postgresql.verite.pro/ Twitter: @DanielVerite Attachments: [text/x-patch] psql-fetchcount-single-row-mode-v2.diff (12.5K, ../../[email protected]/2-psql-fetchcount-single-row-mode-v2.diff) download | inline diff: diff --git a/src/bin/psql/common.c b/src/bin/psql/common.c index f907f5d4e8..ad5e8a5de9 100644 --- a/src/bin/psql/common.c +++ b/src/bin/psql/common.c @@ -372,6 +372,7 @@ AcceptResult(const PGresult *result, bool show_error) { case PGRES_COMMAND_OK: case PGRES_TUPLES_OK: + case PGRES_SINGLE_TUPLE: case PGRES_EMPTY_QUERY: case PGRES_COPY_IN: case PGRES_COPY_OUT: @@ -675,13 +676,13 @@ PrintNotifications(void) * Returns true if successful, false otherwise. */ static bool -PrintQueryTuples(const PGresult *result, const printQueryOpt *opt, +PrintQueryTuples(const PGresult **result, int nresults, const printQueryOpt *opt, FILE *printQueryFout) { bool ok = true; FILE *fout = printQueryFout ? printQueryFout : pset.queryFout; - printQuery(result, opt ? opt : &pset.popt, fout, false, pset.logfile); + printQueryChunks(result, nresults, opt ? opt : &pset.popt, fout, false, pset.logfile); fflush(fout); if (ferror(fout)) { @@ -958,7 +959,7 @@ PrintQueryResult(PGresult *result, bool last, else if (last && pset.crosstab_flag) success = PrintResultInCrosstab(result); else if (last || pset.show_all_results) - success = PrintQueryTuples(result, opt, printQueryFout); + success = PrintQueryTuples((const PGresult**)&result, 1, opt, printQueryFout); else success = true; @@ -1371,6 +1372,47 @@ DescribeQuery(const char *query, double *elapsed_msec) return OK; } +/* + * Check if an output stream for \g needs to be opened, and if + * yes, open it. + * Return false if an error occurred, true otherwise. + */ +static bool +SetupGOutput(PGresult *result, FILE **gfile_fout, bool *is_pipe) +{ + ExecStatusType status = PQresultStatus(result); + if (pset.gfname != NULL && /* there is a \g file or program */ + *gfile_fout == NULL && /* and it's not already opened */ + (status == PGRES_TUPLES_OK || + status == PGRES_SINGLE_TUPLE || + status == PGRES_COPY_OUT)) + { + if (openQueryOutputFile(pset.gfname, gfile_fout, is_pipe)) + { + if (is_pipe) + disable_sigpipe_trap(); + } + else + return false; + } + return true; +} + +static void +CloseGOutput(FILE *gfile_fout, bool is_pipe) +{ + /* close \g file if we opened it */ + if (gfile_fout) + { + if (is_pipe) + { + pclose(gfile_fout); + restore_sigpipe_trap(); + } + else + fclose(gfile_fout); + } +} /* * ExecQueryAndProcessResults: utility function for use by SendQuery() @@ -1402,10 +1444,16 @@ ExecQueryAndProcessResults(const char *query, bool success; instr_time before, after; + int fetch_count = pset.fetch_count; PGresult *result; + FILE *gfile_fout = NULL; bool gfile_is_pipe = false; + PGresult **result_array = NULL; /* to collect results in single row mode */ + int64 total_tuples = 0; + int ntuples; + if (timing) INSTR_TIME_SET_CURRENT(before); else @@ -1428,6 +1476,33 @@ ExecQueryAndProcessResults(const char *query, return -1; } + /* + * If FETCH_COUNT is set and the context allows it, use the single row + * mode to fetch results and have no more than FETCH_COUNT rows in + * memory. + */ + if (fetch_count > 0 && !pset.crosstab_flag && !pset.gexec_flag && !is_watch + && !pset.gset_prefix && pset.show_all_results) + { + /* + * The row-by-row fetch is not enabled when SHOW_ALL_RESULTS is false, + * since we would need to accumulate all rows before knowing + * whether they need to be discarded or displayed, which contradicts + * FETCH_COUNT. + */ + if (!PQsetSingleRowMode(pset.db)) + { + pg_log_warning("fetching results in single row mode is unavailable"); + fetch_count = 0; + } + else + { + result_array = (PGresult**) pg_malloc(fetch_count * sizeof(PGresult*)); + } + } + else + fetch_count = 0; /* disable single-row mode */ + /* * If SIGINT is sent while the query is processing, the interrupt will be * consumed. The user's intention, though, is to cancel the entire watch @@ -1447,6 +1522,7 @@ ExecQueryAndProcessResults(const char *query, ExecStatusType result_status; PGresult *next_result; bool last; + bool partial_display = false; if (!AcceptResult(result, false)) { @@ -1573,6 +1649,96 @@ ExecQueryAndProcessResults(const char *query, success &= HandleCopyResult(&result, copy_stream); } + if (fetch_count > 0 && result_status == PGRES_SINGLE_TUPLE) + { + FILE *tuples_fout = printQueryFout; + printQueryOpt my_popt = pset.popt; + bool is_pager = false; + int flush_error = 0; + + ntuples = 0; + total_tuples = 0; + partial_display = true; + + success = SetupGOutput(result, &gfile_fout, &gfile_is_pipe); + if (gfile_fout) + tuples_fout = gfile_fout; + + /* initialize print options for partial table output */ + my_popt.topt.start_table = true; + my_popt.topt.stop_table = false; + my_popt.topt.prior_records = 0; + + while (success) + { + result_array[ntuples++] = result; + if (ntuples == fetch_count) + { + /* pager: open at most once per resultset */ + if (tuples_fout == stdout && !is_pager) + { + tuples_fout = PageOutput(INT_MAX, &(my_popt.topt)); + is_pager = true; + } + /* display the current chunk of results unless the output stream is not working */ + if (!flush_error) + { + PrintQueryTuples(result_array, ntuples, &my_popt, tuples_fout); + flush_error = fflush(tuples_fout); + } + /* clear and reuse result_array */ + for (int i=0; i < ntuples; i++) + PQclear(result_array[i]); + /* after the first result set, disallow header decoration */ + my_popt.topt.start_table = false; + my_popt.topt.prior_records += ntuples; + total_tuples += ntuples; + ntuples = 0; + } + + result = PQgetResult(pset.db); + if (result == NULL) + { + /* + * Error. We expect a PGRES_TUPLES_OK result with + * zero tuple in it to finish the row-by-row sequence. + */ + success = false; + break; + } + + if (PQresultStatus(result) == PGRES_TUPLES_OK) + { + /* TODO: merge this block with the code above? */ + /* + * The last row has been read. Display the last chunk of + * results and the footer. + */ + my_popt.topt.stop_table = true; + if (!flush_error) + { + PrintQueryTuples(result_array, ntuples, &my_popt, tuples_fout); + flush_error = fflush(tuples_fout); + } + for (int i=0; i < ntuples; i++) + PQclear(result_array[i]); + total_tuples += ntuples; + ntuples = 0; + + if (is_pager) + { + ClosePager(tuples_fout); + } + + result = NULL; + /*partial_display_rowcount = total_tuples;*/ + break; + } + } + } + else + partial_display = false; + /* * Check PQgetResult() again. In the typical case of a single-command * string, it will return NULL. Otherwise, we'll have other results @@ -1601,7 +1767,7 @@ ExecQueryAndProcessResults(const char *query, } /* this may or may not print something depending on settings */ - if (result != NULL) + if (result != NULL && !partial_display) { /* * If results need to be printed into the file specified by \g, @@ -1610,33 +1776,32 @@ ExecQueryAndProcessResults(const char *query, * tuple output, but it's still used for status output. */ FILE *tuples_fout = printQueryFout; - bool do_print = true; - - if (PQresultStatus(result) == PGRES_TUPLES_OK && - pset.gfname) - { - if (gfile_fout == NULL) - { - if (openQueryOutputFile(pset.gfname, - &gfile_fout, &gfile_is_pipe)) - { - if (gfile_is_pipe) - disable_sigpipe_trap(); - } - else - success = do_print = false; - } + success = SetupGOutput(result, &gfile_fout, &gfile_is_pipe); + if (gfile_fout) tuples_fout = gfile_fout; - } - if (do_print) + if (success) success &= PrintQueryResult(result, last, opt, tuples_fout, printQueryFout); } /* set variables on last result if all went well */ if (!is_watch && last && success) + { SetResultVariables(result, true); + if (partial_display) + { + /* + * fake SetResultVariables() as in ExecQueryUsingCursor(). + */ + char buf[32]; + SetVariable(pset.vars, "ERROR", "false"); + SetVariable(pset.vars, "SQLSTATE", "00000"); + snprintf(buf, sizeof(buf), INT64_FORMAT, total_tuples); + SetVariable(pset.vars, "ROW_COUNT", buf); + } + } + ClearOrSaveResult(result); result = next_result; @@ -1647,17 +1812,10 @@ ExecQueryAndProcessResults(const char *query, } } - /* close \g file if we opened it */ - if (gfile_fout) - { - if (gfile_is_pipe) - { - pclose(gfile_fout); - restore_sigpipe_trap(); - } - else - fclose(gfile_fout); - } + CloseGOutput(gfile_fout, gfile_is_pipe); + + if (result_array) + pg_free(result_array); /* may need this to recover from conn loss during COPY */ if (!CheckConnection()) diff --git a/src/fe_utils/print.c b/src/fe_utils/print.c index 3396f9b462..d8f0a29773 100644 --- a/src/fe_utils/print.c +++ b/src/fe_utils/print.c @@ -3533,17 +3533,42 @@ printTable(const printTableContent *cont, void printQuery(const PGresult *result, const printQueryOpt *opt, FILE *fout, bool is_pager, FILE *flog) +{ + printQueryChunks(&result, 1, opt, fout, is_pager, flog); +} + +/* + * Print the results of a query that may have been obtained by a + * succession of calls to PQgetResult in single-row mode. + * + * results: array of results of a successful query. They must have the same columns. + * nbresults: size of results + * opt: formatting options + * fout: where to print to + * is_pager: true if caller has already redirected fout to be a pager pipe + * flog: if not null, also print the data there (for --log-file option) + */ +void +printQueryChunks(const PGresult *results[], int nresults, const printQueryOpt *opt, + FILE *fout, bool is_pager, FILE *flog) { printTableContent cont; int i, r, c; + int nrows = 0; /* total number of rows */ + int ri; /* index into results[] */ if (cancel_pressed) return; + for (ri = 0; ri < nresults; ri++) + { + nrows += PQntuples(results[ri]); + } + printTableInit(&cont, &opt->topt, opt->title, - PQnfields(result), PQntuples(result)); + (nresults > 0) ? PQnfields(results[0]) : 0, nrows); /* Assert caller supplied enough translate_columns[] entries */ Assert(opt->translate_columns == NULL || @@ -3551,34 +3576,37 @@ printQuery(const PGresult *result, const printQueryOpt *opt, for (i = 0; i < cont.ncolumns; i++) { - printTableAddHeader(&cont, PQfname(result, i), + printTableAddHeader(&cont, PQfname(results[0], i), opt->translate_header, - column_type_alignment(PQftype(result, i))); + column_type_alignment(PQftype(results[0], i))); } /* set cells */ - for (r = 0; r < cont.nrows; r++) + for (ri = 0; ri < nresults; ri++) { - for (c = 0; c < cont.ncolumns; c++) + for (r = 0; r < PQntuples(results[ri]); r++) { - char *cell; - bool mustfree = false; - bool translate; - - if (PQgetisnull(result, r, c)) - cell = opt->nullPrint ? opt->nullPrint : ""; - else + for (c = 0; c < cont.ncolumns; c++) { - cell = PQgetvalue(result, r, c); - if (cont.aligns[c] == 'r' && opt->topt.numericLocale) + char *cell; + bool mustfree = false; + bool translate; + + if (PQgetisnull(results[ri], r, c)) + cell = opt->nullPrint ? opt->nullPrint : ""; + else { - cell = format_numeric_locale(cell); - mustfree = true; + cell = PQgetvalue(results[ri], r, c); + if (cont.aligns[c] == 'r' && opt->topt.numericLocale) + { + cell = format_numeric_locale(cell); + mustfree = true; + } } - } - translate = (opt->translate_columns && opt->translate_columns[c]); - printTableAddCell(&cont, cell, translate, mustfree); + translate = (opt->translate_columns && opt->translate_columns[c]); + printTableAddCell(&cont, cell, translate, mustfree); + } } } diff --git a/src/include/fe_utils/print.h b/src/include/fe_utils/print.h index 54f783c907..3befc41bdc 100644 --- a/src/include/fe_utils/print.h +++ b/src/include/fe_utils/print.h @@ -220,7 +220,10 @@ extern void printTableCleanup(printTableContent *const content); extern void printTable(const printTableContent *cont, FILE *fout, bool is_pager, FILE *flog); extern void printQuery(const PGresult *result, const printQueryOpt *opt, - FILE *fout, bool is_pager, FILE *flog); + FILE *fout, bool is_pager, FILE *flog); +extern void printQueryChunks(const PGresult *results[], int nresults, + const printQueryOpt *opt, + FILE *fout, bool is_pager, FILE *flog); extern char column_type_alignment(Oid); ^ permalink raw reply [nested|flat] 3+ messages in thread
end of thread, other threads:[~2023-03-01 10:41 UTC | newest] Thread overview: 3+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2021-01-30 03:37 [PATCH v24 03/10] Add default_toast_compression GUC Justin Pryzby <[email protected]> 2023-01-12 12:27 Re: psql's FETCH_COUNT (cursor) is not being respected for CTEs Daniel Verite <[email protected]> 2023-03-01 10:41 ` Re: psql's FETCH_COUNT (cursor) is not being respected for CTEs Daniel Verite <[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