From c136833f4a385574474b246a381014abeb631377 Mon Sep 17 00:00:00 2001 From: Sutou Kouhei Date: Fri, 26 Jan 2024 16:46:51 +0900 Subject: [PATCH v9 1/2] Extract COPY TO format implementations This is a part of making COPY format extendable. See also these past discussions: * New Copy Formats - avro/orc/parquet: https://www.postgresql.org/message-id/flat/20180210151304.fonjztsynewldfba%40gmail.com * Make COPY extendable in order to support Parquet and other formats: https://www.postgresql.org/message-id/flat/CAJ7c6TM6Bz1c3F04Cy6%2BSzuWfKmr0kU8c_3Stnvh_8BR0D6k8Q%40mail.gmail.com This doesn't change the current behavior. This just introduces CopyToRoutine, which just has function pointers of format implementation like TupleTableSlotOps, and use it for existing "text", "csv" and "binary" format implementations. Note that CopyToRoutine can't be used from extensions yet because CopySend*() aren't exported yet. Extensions can't send formatted data to a destination without CopySend*(). They will be exported by subsequent patches. Here is a benchmark result with/without this change because there was a discussion that we should care about performance regression: https://www.postgresql.org/message-id/3741749.1655952719%40sss.pgh.pa.us > I think that step 1 ought to be to convert the existing formats into > plug-ins, and demonstrate that there's no significant loss of > performance. You can see that there is no significant loss of performance: Data: Random 32 bit integers: CREATE TABLE data (int32 integer); SELECT setseed(0.29); INSERT INTO data SELECT random() * 10000 FROM generate_series(1, ${n_records}); The number of records: 100K, 1M and 10M 100K without this change: format,elapsed time (ms) text,10.561 csv,10.868 binary,10.287 100K with this change: format,elapsed time (ms) text,9.962 csv,10.453 binary,9.473 1M without this change: format,elapsed time (ms) text,103.265 csv,109.789 binary,104.078 1M with this change: format,elapsed time (ms) text,98.612 csv,101.908 binary,94.456 10M without this change: format,elapsed time (ms) text,1060.614 csv,1065.272 binary,1025.875 10M with this change: format,elapsed time (ms) text,1020.050 csv,1031.279 binary,954.792 --- contrib/file_fdw/file_fdw.c | 2 +- src/backend/commands/copy.c | 82 ++++- src/backend/commands/copyfrom.c | 2 +- src/backend/commands/copyto.c | 587 +++++++++++++++++++++++--------- src/include/commands/copy.h | 8 +- src/include/commands/copyapi.h | 62 ++++ 6 files changed, 560 insertions(+), 183 deletions(-) create mode 100644 src/include/commands/copyapi.h diff --git a/contrib/file_fdw/file_fdw.c b/contrib/file_fdw/file_fdw.c index 249d82d3a0..9e4e819858 100644 --- a/contrib/file_fdw/file_fdw.c +++ b/contrib/file_fdw/file_fdw.c @@ -329,7 +329,7 @@ file_fdw_validator(PG_FUNCTION_ARGS) /* * Now apply the core COPY code's validation logic for more checks. */ - ProcessCopyOptions(NULL, NULL, true, other_options); + ProcessCopyOptions(NULL, NULL, true, NULL, other_options); /* * Either filename or program option is required for file_fdw foreign diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c index cc0786c6f4..dd0fe7f0bb 100644 --- a/src/backend/commands/copy.c +++ b/src/backend/commands/copy.c @@ -442,6 +442,9 @@ defGetCopyOnErrorChoice(DefElem *def, ParseState *pstate, bool is_from) * a list of options. In that usage, 'opts_out' can be passed as NULL and * the collected data is just leaked until CurrentMemoryContext is reset. * + * 'cstate' is CopyToState* for !is_from, CopyFromState* for is_from. 'cstate' + * may be NULL. For example, file_fdw uses NULL. + * * Note that additional checking, such as whether column names listed in FORCE * QUOTE actually exist, has to be applied later. This just checks for * self-consistency of the options list. @@ -450,6 +453,7 @@ void ProcessCopyOptions(ParseState *pstate, CopyFormatOptions *opts_out, bool is_from, + void *cstate, List *options) { bool format_specified = false; @@ -457,6 +461,7 @@ ProcessCopyOptions(ParseState *pstate, bool header_specified = false; bool on_error_specified = false; ListCell *option; + List *unknown_options = NIL; /* Support external use for option sanity checking */ if (opts_out == NULL) @@ -464,30 +469,58 @@ ProcessCopyOptions(ParseState *pstate, opts_out->file_encoding = -1; - /* Extract options from the statement node tree */ + /* + * Extract only the "format" option to detect target routine as the first + * step + */ foreach(option, options) { DefElem *defel = lfirst_node(DefElem, option); if (strcmp(defel->defname, "format") == 0) { - char *fmt = defGetString(defel); - if (format_specified) errorConflictingDefElem(defel, pstate); format_specified = true; - if (strcmp(fmt, "text") == 0) - /* default format */ ; - else if (strcmp(fmt, "csv") == 0) - opts_out->csv_mode = true; - else if (strcmp(fmt, "binary") == 0) - opts_out->binary = true; + + if (is_from) + { + char *fmt = defGetString(defel); + + if (strcmp(fmt, "text") == 0) + /* default format */ ; + else if (strcmp(fmt, "csv") == 0) + { + opts_out->csv_mode = true; + } + else if (strcmp(fmt, "binary") == 0) + { + opts_out->binary = true; + } + else + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("COPY format \"%s\" not recognized", fmt), + parser_errposition(pstate, defel->location))); + } else - ereport(ERROR, - (errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("COPY format \"%s\" not recognized", fmt), - parser_errposition(pstate, defel->location))); + ProcessCopyOptionFormatTo(pstate, opts_out, cstate, defel); } + } + if (!format_specified) + /* Set the default format. */ + ProcessCopyOptionFormatTo(pstate, opts_out, cstate, NULL); + + /* + * Extract options except "format" from the statement node tree. Unknown + * options are processed later. + */ + foreach(option, options) + { + DefElem *defel = lfirst_node(DefElem, option); + + if (strcmp(defel->defname, "format") == 0) + continue; else if (strcmp(defel->defname, "freeze") == 0) { if (freeze_specified) @@ -616,11 +649,7 @@ ProcessCopyOptions(ParseState *pstate, opts_out->on_error = defGetCopyOnErrorChoice(defel, pstate, is_from); } else - ereport(ERROR, - (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("option \"%s\" not recognized", - defel->defname), - parser_errposition(pstate, defel->location))); + unknown_options = lappend(unknown_options, defel); } /* @@ -821,6 +850,23 @@ ProcessCopyOptions(ParseState *pstate, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("NULL specification and DEFAULT specification cannot be the same"))); } + + /* Process not built-in options. */ + foreach(option, unknown_options) + { + DefElem *defel = lfirst_node(DefElem, option); + bool processed = false; + + if (!is_from) + processed = opts_out->to_routine->CopyToProcessOption(cstate, defel); + if (!processed) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("option \"%s\" not recognized", + defel->defname), + parser_errposition(pstate, defel->location))); + } + list_free(unknown_options); } /* diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c index 1fe70b9133..fb3d4d9296 100644 --- a/src/backend/commands/copyfrom.c +++ b/src/backend/commands/copyfrom.c @@ -1416,7 +1416,7 @@ BeginCopyFrom(ParseState *pstate, oldcontext = MemoryContextSwitchTo(cstate->copycontext); /* Extract options from the statement node tree */ - ProcessCopyOptions(pstate, &cstate->opts, true /* is_from */ , options); + ProcessCopyOptions(pstate, &cstate->opts, true /* is_from */ , cstate, options); /* Process the target relation */ cstate->rel = rel; diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c index d3dc3fc854..4fb41f04fc 100644 --- a/src/backend/commands/copyto.c +++ b/src/backend/commands/copyto.c @@ -24,6 +24,7 @@ #include "access/xact.h" #include "access/xlog.h" #include "commands/copy.h" +#include "commands/defrem.h" #include "commands/progress.h" #include "executor/execdesc.h" #include "executor/executor.h" @@ -131,6 +132,427 @@ static void CopySendEndOfRow(CopyToState cstate); static void CopySendInt32(CopyToState cstate, int32 val); static void CopySendInt16(CopyToState cstate, int16 val); +/* + * CopyToRoutine implementations. + */ + +/* + * CopyToRoutine implementation for "text" and "csv". CopyToTextBased*() are + * shared by both of "text" and "csv". CopyToText*() are only for "text" and + * CopyToCSV*() are only for "csv". + * + * We can use the same functions for all callbacks by referring + * cstate->opts.csv_mode but splitting callbacks to eliminate "if + * (cstate->opts.csv_mode)" branches from all callbacks has performance + * merit when many tuples are copied. So we use separated callbacks for "text" + * and "csv". + */ + +static void +CopyToTextBasedInit(CopyToState cstate) +{ +} + +/* + * All "text" and "csv" options are parsed in ProcessCopyOptions(). We may + * move the code to here later. + */ +static bool +CopyToTextBasedProcessOption(CopyToState cstate, DefElem *defel) +{ + return false; +} + +static void +CopyToTextBasedFillCopyOutResponse(CopyToState cstate, StringInfoData *buf) +{ + int16 format = 0; + int natts = list_length(cstate->attnumlist); + int i; + + pq_sendbyte(buf, format); /* overall format */ + pq_sendint16(buf, natts); + for (i = 0; i < natts; i++) + pq_sendint16(buf, format); /* per-column formats */ +} + +static void +CopyToTextBasedSendEndOfRow(CopyToState cstate) +{ + switch (cstate->copy_dest) + { + case COPY_FILE: + /* Default line termination depends on platform */ +#ifndef WIN32 + CopySendChar(cstate, '\n'); +#else + CopySendString(cstate, "\r\n"); +#endif + break; + case COPY_FRONTEND: + /* The FE/BE protocol uses \n as newline for all platforms */ + CopySendChar(cstate, '\n'); + break; + default: + break; + } + CopySendEndOfRow(cstate); +} + +typedef void (*CopyAttributeOutHeaderFunction) (CopyToState cstate, char *string); + +/* + * We can use CopyAttributeOutText() directly but define this for consistency + * with CopyAttributeOutCSVHeader(). "static inline" will prevent performance + * penalty by this wrapping. + */ +static inline void +CopyAttributeOutTextHeader(CopyToState cstate, char *string) +{ + CopyAttributeOutText(cstate, string); +} + +static inline void +CopyAttributeOutCSVHeader(CopyToState cstate, char *string) +{ + CopyAttributeOutCSV(cstate, string, false, + list_length(cstate->attnumlist) == 1); +} + +/* + * We don't use this function as a callback directly. We define + * CopyToTextStart() and CopyToCSVStart() and use them instead. It's for + * eliminating a "if (cstate->opts.csv_mode)" branch. This callback is called + * only once per COPY TO. So this optimization may be meaningless but done for + * consistency with CopyToTextBasedOneRow(). + * + * This must initialize cstate->out_functions for CopyToTextBasedOneRow(). + */ +static inline void +CopyToTextBasedStart(CopyToState cstate, TupleDesc tupDesc, CopyAttributeOutHeaderFunction out) +{ + int num_phys_attrs; + ListCell *cur; + + num_phys_attrs = tupDesc->natts; + /* Get info about the columns we need to process. */ + cstate->out_functions = (FmgrInfo *) palloc(num_phys_attrs * sizeof(FmgrInfo)); + foreach(cur, cstate->attnumlist) + { + int attnum = lfirst_int(cur); + Oid out_func_oid; + bool isvarlena; + Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1); + + getTypeOutputInfo(attr->atttypid, &out_func_oid, &isvarlena); + fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]); + } + + /* + * For non-binary copy, we need to convert null_print to file encoding, + * because it will be sent directly with CopySendString. + */ + if (cstate->need_transcoding) + cstate->opts.null_print_client = pg_server_to_any(cstate->opts.null_print, + cstate->opts.null_print_len, + cstate->file_encoding); + + /* if a header has been requested send the line */ + if (cstate->opts.header_line) + { + bool hdr_delim = false; + + foreach(cur, cstate->attnumlist) + { + int attnum = lfirst_int(cur); + char *colname; + + if (hdr_delim) + CopySendChar(cstate, cstate->opts.delim[0]); + hdr_delim = true; + + colname = NameStr(TupleDescAttr(tupDesc, attnum - 1)->attname); + + out(cstate, colname); + } + + CopyToTextBasedSendEndOfRow(cstate); + } +} + +static void +CopyToTextStart(CopyToState cstate, TupleDesc tupDesc) +{ + CopyToTextBasedStart(cstate, tupDesc, CopyAttributeOutTextHeader); +} + +static void +CopyToCSVStart(CopyToState cstate, TupleDesc tupDesc) +{ + CopyToTextBasedStart(cstate, tupDesc, CopyAttributeOutCSVHeader); +} + +typedef void (*CopyAttributeOutValueFunction) (CopyToState cstate, char *string, int attnum); + +static inline void +CopyAttributeOutTextValue(CopyToState cstate, char *string, int attnum) +{ + CopyAttributeOutText(cstate, string); +} + +static inline void +CopyAttributeOutCSVValue(CopyToState cstate, char *string, int attnum) +{ + CopyAttributeOutCSV(cstate, string, + cstate->opts.force_quote_flags[attnum - 1], + list_length(cstate->attnumlist) == 1); +} + +/* + * We don't use this function as a callback directly. We define + * CopyToTextOneRow() and CopyToCSVOneRow() and use them instead. It's for + * eliminating a "if (cstate->opts.csv_mode)" branch. This callback is called + * per tuple. So this optimization will be valuable when many tuples are + * copied. + * + * cstate->out_functions must be initialized in CopyToTextBasedStart(). + */ +static void +CopyToTextBasedOneRow(CopyToState cstate, TupleTableSlot *slot, CopyAttributeOutValueFunction out) +{ + bool need_delim = false; + FmgrInfo *out_functions = cstate->out_functions; + ListCell *cur; + + foreach(cur, cstate->attnumlist) + { + int attnum = lfirst_int(cur); + Datum value = slot->tts_values[attnum - 1]; + bool isnull = slot->tts_isnull[attnum - 1]; + + if (need_delim) + CopySendChar(cstate, cstate->opts.delim[0]); + need_delim = true; + + if (isnull) + { + CopySendString(cstate, cstate->opts.null_print_client); + } + else + { + char *string; + + string = OutputFunctionCall(&out_functions[attnum - 1], value); + out(cstate, string, attnum); + } + } + + CopyToTextBasedSendEndOfRow(cstate); +} + +static void +CopyToTextOneRow(CopyToState cstate, TupleTableSlot *slot) +{ + CopyToTextBasedOneRow(cstate, slot, CopyAttributeOutTextValue); +} + +static void +CopyToCSVOneRow(CopyToState cstate, TupleTableSlot *slot) +{ + CopyToTextBasedOneRow(cstate, slot, CopyAttributeOutCSVValue); +} + +static void +CopyToTextBasedEnd(CopyToState cstate) +{ +} + +/* + * CopyToRoutine implementation for "binary". + */ + +static void +CopyToBinaryInit(CopyToState cstate) +{ +} + +/* + * All "binary" options are parsed in ProcessCopyOptions(). We may move the + * code to here later. + */ +static bool +CopyToBinaryProcessOption(CopyToState cstate, DefElem *defel) +{ + return false; +} + +static void +CopyToBinaryFillCopyOutResponse(CopyToState cstate, StringInfoData *buf) +{ + int16 format = 1; + int natts = list_length(cstate->attnumlist); + int i; + + pq_sendbyte(buf, format); /* overall format */ + pq_sendint16(buf, natts); + for (i = 0; i < natts; i++) + pq_sendint16(buf, format); /* per-column formats */ +} + +/* + * This must initialize cstate->out_functions for CopyToBinaryOneRow(). + */ +static void +CopyToBinaryStart(CopyToState cstate, TupleDesc tupDesc) +{ + int num_phys_attrs; + ListCell *cur; + + num_phys_attrs = tupDesc->natts; + /* Get info about the columns we need to process. */ + cstate->out_functions = (FmgrInfo *) palloc(num_phys_attrs * sizeof(FmgrInfo)); + foreach(cur, cstate->attnumlist) + { + int attnum = lfirst_int(cur); + Oid out_func_oid; + bool isvarlena; + Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1); + + getTypeBinaryOutputInfo(attr->atttypid, &out_func_oid, &isvarlena); + fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]); + } + + { + /* Generate header for a binary copy */ + int32 tmp; + + /* Signature */ + CopySendData(cstate, BinarySignature, 11); + /* Flags field */ + tmp = 0; + CopySendInt32(cstate, tmp); + /* No header extension */ + tmp = 0; + CopySendInt32(cstate, tmp); + } +} + +/* + * cstate->out_functions must be initialized in CopyToBinaryStart(). + */ +static void +CopyToBinaryOneRow(CopyToState cstate, TupleTableSlot *slot) +{ + FmgrInfo *out_functions = cstate->out_functions; + ListCell *cur; + + /* Binary per-tuple header */ + CopySendInt16(cstate, list_length(cstate->attnumlist)); + + foreach(cur, cstate->attnumlist) + { + int attnum = lfirst_int(cur); + Datum value = slot->tts_values[attnum - 1]; + bool isnull = slot->tts_isnull[attnum - 1]; + + if (isnull) + { + CopySendInt32(cstate, -1); + } + else + { + bytea *outputbytes; + + outputbytes = SendFunctionCall(&out_functions[attnum - 1], value); + CopySendInt32(cstate, VARSIZE(outputbytes) - VARHDRSZ); + CopySendData(cstate, VARDATA(outputbytes), + VARSIZE(outputbytes) - VARHDRSZ); + } + } + + CopySendEndOfRow(cstate); +} + +static void +CopyToBinaryEnd(CopyToState cstate) +{ + /* Generate trailer for a binary copy */ + CopySendInt16(cstate, -1); + /* Need to flush out the trailer */ + CopySendEndOfRow(cstate); +} + +/* + * CopyToTextBased*() are shared with "csv". CopyToText*() are only for "text". + */ +static const CopyToRoutine CopyToRoutineText = { + .CopyToInit = CopyToTextBasedInit, + .CopyToProcessOption = CopyToTextBasedProcessOption, + .CopyToFillCopyOutResponse = CopyToTextBasedFillCopyOutResponse, + .CopyToStart = CopyToTextStart, + .CopyToOneRow = CopyToTextOneRow, + .CopyToEnd = CopyToTextBasedEnd, +}; + +/* + * CopyToTextBased*() are shared with "text". CopyToCSV*() are only for "csv". + */ +static const CopyToRoutine CopyToRoutineCSV = { + .CopyToInit = CopyToTextBasedInit, + .CopyToProcessOption = CopyToTextBasedProcessOption, + .CopyToFillCopyOutResponse = CopyToTextBasedFillCopyOutResponse, + .CopyToStart = CopyToCSVStart, + .CopyToOneRow = CopyToCSVOneRow, + .CopyToEnd = CopyToTextBasedEnd, +}; + +static const CopyToRoutine CopyToRoutineBinary = { + .CopyToInit = CopyToBinaryInit, + .CopyToProcessOption = CopyToBinaryProcessOption, + .CopyToFillCopyOutResponse = CopyToBinaryFillCopyOutResponse, + .CopyToStart = CopyToBinaryStart, + .CopyToOneRow = CopyToBinaryOneRow, + .CopyToEnd = CopyToBinaryEnd, +}; + +/* + * Process the "format" option for COPY TO. + * + * If defel is NULL, the default format "text" is used. + */ +void +ProcessCopyOptionFormatTo(ParseState *pstate, + CopyFormatOptions *opts_out, + CopyToState cstate, + DefElem *defel) +{ + char *format; + + if (defel) + format = defGetString(defel); + else + format = "text"; + + if (strcmp(format, "text") == 0) + opts_out->to_routine = &CopyToRoutineText; + else if (strcmp(format, "csv") == 0) + { + opts_out->csv_mode = true; + opts_out->to_routine = &CopyToRoutineCSV; + } + else if (strcmp(format, "binary") == 0) + { + opts_out->binary = true; + opts_out->to_routine = &CopyToRoutineBinary; + } + else + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("COPY format \"%s\" not recognized", format), + parser_errposition(pstate, defel->location))); + + opts_out->to_routine->CopyToInit(cstate); +} /* * Send copy start/stop messages for frontend copies. These have changed @@ -140,15 +562,9 @@ static void SendCopyBegin(CopyToState cstate) { StringInfoData buf; - int natts = list_length(cstate->attnumlist); - int16 format = (cstate->opts.binary ? 1 : 0); - int i; pq_beginmessage(&buf, PqMsg_CopyOutResponse); - pq_sendbyte(&buf, format); /* overall format */ - pq_sendint16(&buf, natts); - for (i = 0; i < natts; i++) - pq_sendint16(&buf, format); /* per-column formats */ + cstate->opts.to_routine->CopyToFillCopyOutResponse(cstate, &buf); pq_endmessage(&buf); cstate->copy_dest = COPY_FRONTEND; } @@ -198,16 +614,6 @@ CopySendEndOfRow(CopyToState cstate) switch (cstate->copy_dest) { case COPY_FILE: - if (!cstate->opts.binary) - { - /* Default line termination depends on platform */ -#ifndef WIN32 - CopySendChar(cstate, '\n'); -#else - CopySendString(cstate, "\r\n"); -#endif - } - if (fwrite(fe_msgbuf->data, fe_msgbuf->len, 1, cstate->copy_file) != 1 || ferror(cstate->copy_file)) @@ -242,10 +648,6 @@ CopySendEndOfRow(CopyToState cstate) } break; case COPY_FRONTEND: - /* The FE/BE protocol uses \n as newline for all platforms */ - if (!cstate->opts.binary) - CopySendChar(cstate, '\n'); - /* Dump the accumulated row as one CopyData message */ (void) pq_putmessage(PqMsg_CopyData, fe_msgbuf->data, fe_msgbuf->len); break; @@ -431,7 +833,7 @@ BeginCopyTo(ParseState *pstate, oldcontext = MemoryContextSwitchTo(cstate->copycontext); /* Extract options from the statement node tree */ - ProcessCopyOptions(pstate, &cstate->opts, false /* is_from */ , options); + ProcessCopyOptions(pstate, &cstate->opts, false /* is_from */ , cstate, options); /* Process the source/target relation or query */ if (rel) @@ -748,8 +1150,6 @@ DoCopyTo(CopyToState cstate) bool pipe = (cstate->filename == NULL && cstate->data_dest_cb == NULL); bool fe_copy = (pipe && whereToSendOutput == DestRemote); TupleDesc tupDesc; - int num_phys_attrs; - ListCell *cur; uint64 processed; if (fe_copy) @@ -759,32 +1159,11 @@ DoCopyTo(CopyToState cstate) tupDesc = RelationGetDescr(cstate->rel); else tupDesc = cstate->queryDesc->tupDesc; - num_phys_attrs = tupDesc->natts; cstate->opts.null_print_client = cstate->opts.null_print; /* default */ /* We use fe_msgbuf as a per-row buffer regardless of copy_dest */ cstate->fe_msgbuf = makeStringInfo(); - /* Get info about the columns we need to process. */ - cstate->out_functions = (FmgrInfo *) palloc(num_phys_attrs * sizeof(FmgrInfo)); - foreach(cur, cstate->attnumlist) - { - int attnum = lfirst_int(cur); - Oid out_func_oid; - bool isvarlena; - Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1); - - if (cstate->opts.binary) - getTypeBinaryOutputInfo(attr->atttypid, - &out_func_oid, - &isvarlena); - else - getTypeOutputInfo(attr->atttypid, - &out_func_oid, - &isvarlena); - fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]); - } - /* * Create a temporary memory context that we can reset once per row to * recover palloc'd memory. This avoids any problems with leaks inside @@ -795,57 +1174,7 @@ DoCopyTo(CopyToState cstate) "COPY TO", ALLOCSET_DEFAULT_SIZES); - if (cstate->opts.binary) - { - /* Generate header for a binary copy */ - int32 tmp; - - /* Signature */ - CopySendData(cstate, BinarySignature, 11); - /* Flags field */ - tmp = 0; - CopySendInt32(cstate, tmp); - /* No header extension */ - tmp = 0; - CopySendInt32(cstate, tmp); - } - else - { - /* - * For non-binary copy, we need to convert null_print to file - * encoding, because it will be sent directly with CopySendString. - */ - if (cstate->need_transcoding) - cstate->opts.null_print_client = pg_server_to_any(cstate->opts.null_print, - cstate->opts.null_print_len, - cstate->file_encoding); - - /* if a header has been requested send the line */ - if (cstate->opts.header_line) - { - bool hdr_delim = false; - - foreach(cur, cstate->attnumlist) - { - int attnum = lfirst_int(cur); - char *colname; - - if (hdr_delim) - CopySendChar(cstate, cstate->opts.delim[0]); - hdr_delim = true; - - colname = NameStr(TupleDescAttr(tupDesc, attnum - 1)->attname); - - if (cstate->opts.csv_mode) - CopyAttributeOutCSV(cstate, colname, false, - list_length(cstate->attnumlist) == 1); - else - CopyAttributeOutText(cstate, colname); - } - - CopySendEndOfRow(cstate); - } - } + cstate->opts.to_routine->CopyToStart(cstate, tupDesc); if (cstate->rel) { @@ -884,13 +1213,7 @@ DoCopyTo(CopyToState cstate) processed = ((DR_copy *) cstate->queryDesc->dest)->processed; } - if (cstate->opts.binary) - { - /* Generate trailer for a binary copy */ - CopySendInt16(cstate, -1); - /* Need to flush out the trailer */ - CopySendEndOfRow(cstate); - } + cstate->opts.to_routine->CopyToEnd(cstate); MemoryContextDelete(cstate->rowcontext); @@ -906,71 +1229,15 @@ DoCopyTo(CopyToState cstate) static void CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot) { - bool need_delim = false; - FmgrInfo *out_functions = cstate->out_functions; MemoryContext oldcontext; - ListCell *cur; - char *string; MemoryContextReset(cstate->rowcontext); oldcontext = MemoryContextSwitchTo(cstate->rowcontext); - if (cstate->opts.binary) - { - /* Binary per-tuple header */ - CopySendInt16(cstate, list_length(cstate->attnumlist)); - } - /* Make sure the tuple is fully deconstructed */ slot_getallattrs(slot); - foreach(cur, cstate->attnumlist) - { - int attnum = lfirst_int(cur); - Datum value = slot->tts_values[attnum - 1]; - bool isnull = slot->tts_isnull[attnum - 1]; - - if (!cstate->opts.binary) - { - if (need_delim) - CopySendChar(cstate, cstate->opts.delim[0]); - need_delim = true; - } - - if (isnull) - { - if (!cstate->opts.binary) - CopySendString(cstate, cstate->opts.null_print_client); - else - CopySendInt32(cstate, -1); - } - else - { - if (!cstate->opts.binary) - { - string = OutputFunctionCall(&out_functions[attnum - 1], - value); - if (cstate->opts.csv_mode) - CopyAttributeOutCSV(cstate, string, - cstate->opts.force_quote_flags[attnum - 1], - list_length(cstate->attnumlist) == 1); - else - CopyAttributeOutText(cstate, string); - } - else - { - bytea *outputbytes; - - outputbytes = SendFunctionCall(&out_functions[attnum - 1], - value); - CopySendInt32(cstate, VARSIZE(outputbytes) - VARHDRSZ); - CopySendData(cstate, VARDATA(outputbytes), - VARSIZE(outputbytes) - VARHDRSZ); - } - } - } - - CopySendEndOfRow(cstate); + cstate->opts.to_routine->CopyToOneRow(cstate, slot); MemoryContextSwitchTo(oldcontext); } diff --git a/src/include/commands/copy.h b/src/include/commands/copy.h index b3da3cb0be..de316cfd81 100644 --- a/src/include/commands/copy.h +++ b/src/include/commands/copy.h @@ -14,6 +14,7 @@ #ifndef COPY_H #define COPY_H +#include "commands/copyapi.h" #include "nodes/execnodes.h" #include "nodes/parsenodes.h" #include "parser/parse_node.h" @@ -74,11 +75,11 @@ typedef struct CopyFormatOptions bool convert_selectively; /* do selective binary conversion? */ CopyOnErrorChoice on_error; /* what to do when error happened */ List *convert_select; /* list of column names (can be NIL) */ + const CopyToRoutine *to_routine; /* callback routines for COPY TO */ } CopyFormatOptions; -/* These are private in commands/copy[from|to].c */ +/* This is private in commands/copyfrom.c */ typedef struct CopyFromStateData *CopyFromState; -typedef struct CopyToStateData *CopyToState; typedef int (*copy_data_source_cb) (void *outbuf, int minread, int maxread); typedef void (*copy_data_dest_cb) (void *data, int len); @@ -87,7 +88,8 @@ extern void DoCopy(ParseState *pstate, const CopyStmt *stmt, int stmt_location, int stmt_len, uint64 *processed); -extern void ProcessCopyOptions(ParseState *pstate, CopyFormatOptions *opts_out, bool is_from, List *options); +extern void ProcessCopyOptions(ParseState *pstate, CopyFormatOptions *opts_out, bool is_from, void *cstate, List *options); +extern void ProcessCopyOptionFormatTo(ParseState *pstate, CopyFormatOptions *opts_out, CopyToState cstate, DefElem *defel); extern CopyFromState BeginCopyFrom(ParseState *pstate, Relation rel, Node *whereClause, const char *filename, bool is_program, copy_data_source_cb data_source_cb, List *attnamelist, List *options); diff --git a/src/include/commands/copyapi.h b/src/include/commands/copyapi.h new file mode 100644 index 0000000000..f8901cac51 --- /dev/null +++ b/src/include/commands/copyapi.h @@ -0,0 +1,62 @@ +/*------------------------------------------------------------------------- + * + * copyapi.h + * API for COPY TO/FROM handlers + * + * + * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * src/include/commands/copyapi.h + * + *------------------------------------------------------------------------- + */ +#ifndef COPYAPI_H +#define COPYAPI_H + +#include "executor/tuptable.h" +#include "nodes/parsenodes.h" + +/* This is private in commands/copyto.c */ +typedef struct CopyToStateData *CopyToState; + +/* Routines for a COPY TO format implementation. */ +typedef struct CopyToRoutine +{ + /* + * Called when this CopyToRoutine is chosen. This can be used for + * initialization. + */ + void (*CopyToInit) (CopyToState cstate); + + /* + * Called for processing one COPY TO option. This will return false when + * the given option is invalid. + */ + bool (*CopyToProcessOption) (CopyToState cstate, DefElem *defel); + + /* + * Called when COPY TO via the PostgreSQL protocol is started. This must + * fill buf as a valid CopyOutResponse message: + * + */ + /*-- + * +--------+--------+--------+--------+--------+ +--------+--------+ + * | Format | N attributes | Attr1's format |...| AttrN's format | + * +--------+--------+--------+--------+--------+ +--------+--------+ + * 0: text 0: text 0: text + * 1: binary 1: binary 1: binary + */ + void (*CopyToFillCopyOutResponse) (CopyToState cstate, StringInfoData *buf); + + /* Called when COPY TO is started. This will send a header. */ + void (*CopyToStart) (CopyToState cstate, TupleDesc tupDesc); + + /* Copy one row for COPY TO. */ + void (*CopyToOneRow) (CopyToState cstate, TupleTableSlot *slot); + + /* Called when COPY TO is ended. This will send a trailer. */ + void (*CopyToEnd) (CopyToState cstate); +} CopyToRoutine; + +#endif /* COPYAPI_H */ -- 2.43.0