mirror of
https://github.com/postgres/postgres.git
synced 2025-07-07 00:36:50 +03:00
Add support for COPY TO callback functions
This is useful as a way for extensions to process COPY TO rows in the way they see fit (say auditing, analytics, backend, etc.) without the need to invoke an external process running as the OS user running the backend through PROGRAM that requires superuser rights. COPY FROM already provides a similar callback for logical replication. For COPY TO, the callback is triggered when we are ready to send a row in CopySendEndOfRow(), which is the same code path as when sending a row to a frontend or a pipe/file. A small test module, test_copy_callbacks, is added to provide some coverage for this facility. Author: Bilva Sanaba, Nathan Bossart Discussion: https://postgr.es/m/253C21D1-FCEB-41D9-A2AF-E6517015B7D7@amazon.com
This commit is contained in:
@ -51,6 +51,7 @@ typedef enum CopyDest
|
||||
{
|
||||
COPY_FILE, /* to file (or a piped program) */
|
||||
COPY_FRONTEND, /* to frontend */
|
||||
COPY_CALLBACK /* to callback function */
|
||||
} CopyDest;
|
||||
|
||||
/*
|
||||
@ -85,6 +86,7 @@ typedef struct CopyToStateData
|
||||
List *attnumlist; /* integer list of attnums to copy */
|
||||
char *filename; /* filename, or NULL for STDOUT */
|
||||
bool is_program; /* is 'filename' a program to popen? */
|
||||
copy_data_dest_cb data_dest_cb; /* function for writing data */
|
||||
|
||||
CopyFormatOptions opts;
|
||||
Node *whereClause; /* WHERE condition (or NULL) */
|
||||
@ -247,6 +249,9 @@ CopySendEndOfRow(CopyToState cstate)
|
||||
/* Dump the accumulated row as one CopyData message */
|
||||
(void) pq_putmessage('d', fe_msgbuf->data, fe_msgbuf->len);
|
||||
break;
|
||||
case COPY_CALLBACK:
|
||||
cstate->data_dest_cb(fe_msgbuf->data, fe_msgbuf->len);
|
||||
break;
|
||||
}
|
||||
|
||||
/* Update the progress */
|
||||
@ -336,6 +341,17 @@ EndCopy(CopyToState cstate)
|
||||
|
||||
/*
|
||||
* Setup CopyToState to read tuples from a table or a query for COPY TO.
|
||||
*
|
||||
* 'rel': Relation to be copied
|
||||
* 'raw_query': Query whose results are to be copied
|
||||
* 'queryRelId': OID of base relation to convert to a query (for RLS)
|
||||
* 'filename': Name of server-local file to write, NULL for STDOUT
|
||||
* 'is_program': true if 'filename' is program to execute
|
||||
* 'data_dest_cb': Callback that processes the output data
|
||||
* 'attnamelist': List of char *, columns to include. NIL selects all cols.
|
||||
* 'options': List of DefElem. See copy_opt_item in gram.y for selections.
|
||||
*
|
||||
* Returns a CopyToState, to be passed to DoCopyTo() and related functions.
|
||||
*/
|
||||
CopyToState
|
||||
BeginCopyTo(ParseState *pstate,
|
||||
@ -344,11 +360,12 @@ BeginCopyTo(ParseState *pstate,
|
||||
Oid queryRelId,
|
||||
const char *filename,
|
||||
bool is_program,
|
||||
copy_data_dest_cb data_dest_cb,
|
||||
List *attnamelist,
|
||||
List *options)
|
||||
{
|
||||
CopyToState cstate;
|
||||
bool pipe = (filename == NULL);
|
||||
bool pipe = (filename == NULL && data_dest_cb == NULL);
|
||||
TupleDesc tupDesc;
|
||||
int num_phys_attrs;
|
||||
MemoryContext oldcontext;
|
||||
@ -656,7 +673,13 @@ BeginCopyTo(ParseState *pstate,
|
||||
|
||||
cstate->copy_dest = COPY_FILE; /* default */
|
||||
|
||||
if (pipe)
|
||||
if (data_dest_cb)
|
||||
{
|
||||
progress_vals[1] = PROGRESS_COPY_TYPE_CALLBACK;
|
||||
cstate->copy_dest = COPY_CALLBACK;
|
||||
cstate->data_dest_cb = data_dest_cb;
|
||||
}
|
||||
else if (pipe)
|
||||
{
|
||||
progress_vals[1] = PROGRESS_COPY_TYPE_PIPE;
|
||||
|
||||
@ -765,11 +788,13 @@ EndCopyTo(CopyToState cstate)
|
||||
|
||||
/*
|
||||
* Copy from relation or query TO file.
|
||||
*
|
||||
* Returns the number of rows processed.
|
||||
*/
|
||||
uint64
|
||||
DoCopyTo(CopyToState cstate)
|
||||
{
|
||||
bool pipe = (cstate->filename == NULL);
|
||||
bool pipe = (cstate->filename == NULL && cstate->data_dest_cb == NULL);
|
||||
bool fe_copy = (pipe && whereToSendOutput == DestRemote);
|
||||
TupleDesc tupDesc;
|
||||
int num_phys_attrs;
|
||||
|
Reference in New Issue
Block a user