public inbox for [email protected]help / color / mirror / Atom feed
[PATCH 2/2] Fast COPY FROM into the foreign or sharded table. 6+ messages / 5 participants [nested] [flat]
* [PATCH 2/2] Fast COPY FROM into the foreign or sharded table. @ 2020-07-09 06:16 Andrey Lepikhov <[email protected]> 0 siblings, 0 replies; 6+ messages in thread From: Andrey Lepikhov @ 2020-07-09 06:16 UTC (permalink / raw) This feature enables bulk COPY into foreign table in the case of multi inserts is possible and foreign table has non-zero number of columns. FDWAPI was extended by next routines: * BeginForeignCopy * EndForeignCopy * ExecForeignCopy BeginForeignCopy and EndForeignCopy initialize and free the CopyState of bulk COPY. The ExecForeignCopy routine send 'COPY ... FROM STDIN' command to the foreign server, in iterative manner send tuples by CopyTo() machinery, send EOF to this connection. Code that constructed list of columns for a given foreign relation in the deparseAnalyzeSql() routine is separated to the deparseRelColumnList(). It is reused in the deparseCopyFromSql(). Added TAP-tests on the specific corner cases of COPY FROM STDIN operation. By the analogy of CopyFrom() the CopyState structure was extended with data_dest_cb callback. It is used for send text representation of a tuple to a custom destination. The PgFdwModifyState structure is extended with the cstate field. It is needed for avoid repeated initialization of CopyState. ALso for this reason CopyTo() routine was split into the set of routines CopyToStart()/ CopyTo()/CopyToFinish(). Enum CopyInsertMethod removed. This logic implements by ri_usesMultiInsert field of the ResultRelInfo sructure. Discussion: https://www.postgresql.org/message-id/flat/3d0909dc-3691-a576-208a-90986e55489f%40postgrespro.ru Authors: Andrey Lepikhov, Ashutosh Bapat, Amit Langote --- contrib/postgres_fdw/deparse.c | 60 ++++- .../postgres_fdw/expected/postgres_fdw.out | 46 +++- contrib/postgres_fdw/postgres_fdw.c | 143 ++++++++++++ contrib/postgres_fdw/postgres_fdw.h | 1 + contrib/postgres_fdw/sql/postgres_fdw.sql | 45 ++++ doc/src/sgml/fdwhandler.sgml | 75 ++++++ src/backend/commands/copy.c | 220 +++++++++++------- src/backend/executor/execMain.c | 8 +- src/backend/executor/execPartition.c | 27 ++- src/include/commands/copy.h | 11 + src/include/foreign/fdwapi.h | 15 ++ 11 files changed, 548 insertions(+), 103 deletions(-) diff --git a/contrib/postgres_fdw/deparse.c b/contrib/postgres_fdw/deparse.c index ad37a74221..a37981ff66 100644 --- a/contrib/postgres_fdw/deparse.c +++ b/contrib/postgres_fdw/deparse.c @@ -184,6 +184,8 @@ static void appendAggOrderBy(List *orderList, List *targetList, static void appendFunctionName(Oid funcid, deparse_expr_cxt *context); static Node *deparseSortGroupClause(Index ref, List *tlist, bool force_colno, deparse_expr_cxt *context); +static List *deparseRelColumnList(StringInfo buf, Relation rel, + bool enclose_in_parens); /* * Helper functions @@ -1758,6 +1760,20 @@ deparseInsertSql(StringInfo buf, RangeTblEntry *rte, withCheckOptionList, returningList, retrieved_attrs); } +/* + * Deparse COPY FROM into given buf. + * We need to use list of parameters at each query. + */ +void +deparseCopyFromSql(StringInfo buf, Relation rel) +{ + appendStringInfoString(buf, "COPY "); + deparseRelation(buf, rel); + (void) deparseRelColumnList(buf, rel, true); + + appendStringInfoString(buf, " FROM STDIN "); +} + /* * deparse remote UPDATE statement * @@ -2061,6 +2077,30 @@ deparseAnalyzeSizeSql(StringInfo buf, Relation rel) */ void deparseAnalyzeSql(StringInfo buf, Relation rel, List **retrieved_attrs) +{ + appendStringInfoString(buf, "SELECT "); + *retrieved_attrs = deparseRelColumnList(buf, rel, false); + + /* Don't generate bad syntax for zero-column relation. */ + if (list_length(*retrieved_attrs) == 0) + appendStringInfoString(buf, "NULL"); + + /* + * Construct FROM clause + */ + appendStringInfoString(buf, " FROM "); + deparseRelation(buf, rel); +} + +/* + * Construct the list of columns of given foreign relation in the order they + * appear in the tuple descriptor of the relation. Ignore any dropped columns. + * Use column names on the foreign server instead of local names. + * + * Optionally enclose the list in parantheses. + */ +static List * +deparseRelColumnList(StringInfo buf, Relation rel, bool enclose_in_parens) { Oid relid = RelationGetRelid(rel); TupleDesc tupdesc = RelationGetDescr(rel); @@ -2069,10 +2109,8 @@ deparseAnalyzeSql(StringInfo buf, Relation rel, List **retrieved_attrs) List *options; ListCell *lc; bool first = true; + List *retrieved_attrs = NIL; - *retrieved_attrs = NIL; - - appendStringInfoString(buf, "SELECT "); for (i = 0; i < tupdesc->natts; i++) { /* Ignore dropped columns. */ @@ -2081,6 +2119,9 @@ deparseAnalyzeSql(StringInfo buf, Relation rel, List **retrieved_attrs) if (!first) appendStringInfoString(buf, ", "); + else if (enclose_in_parens) + appendStringInfoChar(buf, '('); + first = false; /* Use attribute name or column_name option. */ @@ -2100,18 +2141,13 @@ deparseAnalyzeSql(StringInfo buf, Relation rel, List **retrieved_attrs) appendStringInfoString(buf, quote_identifier(colname)); - *retrieved_attrs = lappend_int(*retrieved_attrs, i + 1); + retrieved_attrs = lappend_int(retrieved_attrs, i + 1); } - /* Don't generate bad syntax for zero-column relation. */ - if (first) - appendStringInfoString(buf, "NULL"); + if (enclose_in_parens && list_length(retrieved_attrs) > 0) + appendStringInfoChar(buf, ')'); - /* - * Construct FROM clause - */ - appendStringInfoString(buf, " FROM "); - deparseRelation(buf, rel); + return retrieved_attrs; } /* diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index 84bc0ee381..5206814f10 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -8084,8 +8084,9 @@ copy rem2 from stdin; copy rem2 from stdin; -- ERROR ERROR: new row for relation "loc2" violates check constraint "loc2_f1positive" DETAIL: Failing row contains (-1, xyzzy). -CONTEXT: remote SQL command: INSERT INTO public.loc2(f1, f2) VALUES ($1, $2) -COPY rem2, line 1: "-1 xyzzy" +CONTEXT: COPY loc2, line 1: "-1 xyzzy" +remote SQL command: COPY public.loc2(f1, f2) FROM STDIN +COPY rem2, line 2 select * from rem2; f1 | f2 ----+----- @@ -8096,6 +8097,19 @@ select * from rem2; alter foreign table rem2 drop constraint rem2_f1positive; alter table loc2 drop constraint loc2_f1positive; delete from rem2; +create table foo (a int) partition by list (a); +create table foo1 (like foo); +create foreign table ffoo1 partition of foo for values in (1) + server loopback options (table_name 'foo1'); +create table foo2 (like foo); +create foreign table ffoo2 partition of foo for values in (2) + server loopback options (table_name 'foo2'); +create function print_new_row() returns trigger language plpgsql as $$ + begin raise notice '%', new; return new; end; $$; +create trigger ffoo1_br_trig before insert on ffoo1 + for each row execute function print_new_row(); +copy foo from stdin; +NOTICE: (1) -- Test local triggers create trigger trig_stmt_before before insert on rem2 for each statement execute procedure trigger_func(); @@ -8204,6 +8218,34 @@ drop trigger rem2_trig_row_before on rem2; drop trigger rem2_trig_row_after on rem2; drop trigger loc2_trig_row_before_insert on loc2; delete from rem2; +alter table loc2 drop column f1; +alter table loc2 drop column f2; +copy rem2 from stdin; +ERROR: column "f1" of relation "loc2" does not exist +CONTEXT: remote SQL command: COPY public.loc2(f1, f2) FROM STDIN +COPY rem2, line 3 +alter table loc2 add column f1 int; +alter table loc2 add column f2 int; +select * from rem2; + f1 | f2 +----+---- +(0 rows) + +-- dropped columns locally and on the foreign server +alter table rem2 drop column f1; +alter table rem2 drop column f2; +copy rem2 from stdin; +select * from rem2; +-- +(2 rows) + +alter table loc2 drop column f1; +alter table loc2 drop column f2; +copy rem2 from stdin; +select * from rem2; +-- +(4 rows) + -- test COPY FROM with foreign table created in the same transaction create table loc3 (f1 int, f2 text); begin; diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index a31abce7c9..9685e731e0 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -18,6 +18,7 @@ #include "access/sysattr.h" #include "access/table.h" #include "catalog/pg_class.h" +#include "commands/copy.h" #include "commands/defrem.h" #include "commands/explain.h" #include "commands/vacuum.h" @@ -190,6 +191,7 @@ typedef struct PgFdwModifyState /* for update row movement if subplan result rel */ struct PgFdwModifyState *aux_fmstate; /* foreign-insert state, if * created */ + CopyState cstate; /* foreign COPY state, if used */ } PgFdwModifyState; /* @@ -356,6 +358,13 @@ static void postgresBeginForeignInsert(ModifyTableState *mtstate, ResultRelInfo *resultRelInfo); static void postgresEndForeignInsert(EState *estate, ResultRelInfo *resultRelInfo); +static void postgresBeginForeignCopy(ModifyTableState *mtstate, + ResultRelInfo *resultRelInfo); +static void postgresEndForeignCopy(EState *estate, + ResultRelInfo *resultRelInfo); +static void postgresExecForeignCopy(ResultRelInfo *resultRelInfo, + TupleTableSlot **slots, + int nslots); static int postgresIsForeignRelUpdatable(Relation rel); static bool postgresPlanDirectModify(PlannerInfo *root, ModifyTable *plan, @@ -533,6 +542,9 @@ postgres_fdw_handler(PG_FUNCTION_ARGS) routine->EndForeignModify = postgresEndForeignModify; routine->BeginForeignInsert = postgresBeginForeignInsert; routine->EndForeignInsert = postgresEndForeignInsert; + routine->BeginForeignCopy = postgresBeginForeignCopy; + routine->EndForeignCopy = postgresEndForeignCopy; + routine->ExecForeignCopy = postgresExecForeignCopy; routine->IsForeignRelUpdatable = postgresIsForeignRelUpdatable; routine->PlanDirectModify = postgresPlanDirectModify; routine->BeginDirectModify = postgresBeginDirectModify; @@ -2050,6 +2062,137 @@ postgresEndForeignInsert(EState *estate, finish_foreign_modify(fmstate); } +static PgFdwModifyState *copy_fmstate = NULL; + +static void +pgfdw_copy_dest_cb(void *buf, int len) +{ + PGconn *conn = copy_fmstate->conn; + + if (PQputCopyData(conn, (char *) buf, len) <= 0) + { + PGresult *res = PQgetResult(conn); + + pgfdw_report_error(ERROR, res, conn, true, copy_fmstate->query); + } +} + +/* + * + * postgresBeginForeignCopy + * Begin an COPY operation on a foreign table + */ +static void +postgresBeginForeignCopy(ModifyTableState *mtstate, + ResultRelInfo *resultRelInfo) +{ + PgFdwModifyState *fmstate; + StringInfoData sql; + RangeTblEntry *rte; + Relation rel = resultRelInfo->ri_RelationDesc; + + rte = exec_rt_fetch(resultRelInfo->ri_RangeTableIndex, mtstate->ps.state); + initStringInfo(&sql); + deparseCopyFromSql(&sql, rel); + + fmstate = create_foreign_modify(mtstate->ps.state, + rte, + resultRelInfo, + CMD_INSERT, + NULL, + sql.data, + NIL, + false, + NIL); + + fmstate->cstate = BeginCopyTo(NULL, NULL, RelationGetDescr(rel), NULL, + InvalidOid, NULL, false, pgfdw_copy_dest_cb, + NIL, NIL); + CopyToStart(fmstate->cstate); + resultRelInfo->ri_FdwState = fmstate; +} + +/* + * postgresEndForeignCopy + * Finish an COPY operation on a foreign table + */ +static void +postgresEndForeignCopy(EState *estate, ResultRelInfo *resultRelInfo) +{ + PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState; + + /* Check correct use of CopyIn FDW API. */ + Assert(fmstate->cstate != NULL); + CopyToFinish(fmstate->cstate); + pfree(fmstate->cstate); + fmstate->cstate = NULL; + finish_foreign_modify(fmstate); +} + +/* + * + * postgresExecForeignCopy + * Send a number of tuples to the foreign relation. + */ +static void +postgresExecForeignCopy(ResultRelInfo *resultRelInfo, + TupleTableSlot **slots, int nslots) +{ + PgFdwModifyState *fmstate = resultRelInfo->ri_FdwState; + PGresult *res; + PGconn *conn = fmstate->conn; + bool status = false; + int i; + + /* Check correct use of CopyIn FDW API. */ + Assert(fmstate->cstate != NULL); + Assert(copy_fmstate == NULL); + + res = PQexec(conn, fmstate->query); + if (PQresultStatus(res) != PGRES_COPY_IN) + pgfdw_report_error(ERROR, res, conn, true, fmstate->query); + PQclear(res); + + PG_TRY(); + { + copy_fmstate = fmstate; + for (i = 0; i < nslots; i++) + CopyOneRowTo(fmstate->cstate, slots[i]); + + status = true; + } + PG_FINALLY(); + { + copy_fmstate = NULL; /* Detect problems */ + + /* Finish COPY IN protocol. It is needed to do after successful copy or + * after an error. + */ + if (PQputCopyEnd(conn, status ? NULL : _("canceled by server")) <= 0 || + PQflush(conn)) + ereport(ERROR, + (errmsg("error returned by PQputCopyEnd: %s", + PQerrorMessage(conn)))); + + /* After successfully sending an EOF signal, check command status. */ + res = PQgetResult(conn); + if ((!status && PQresultStatus(res) != PGRES_FATAL_ERROR) || + (status && PQresultStatus(res) != PGRES_COMMAND_OK)) + pgfdw_report_error(ERROR, res, fmstate->conn, true, fmstate->query); + + PQclear(res); + /* Do this to ensure we've pumped libpq back to idle state */ + if (PQgetResult(conn) != NULL) + ereport(ERROR, + (errmsg("unexpected extra results during COPY of table: %s", + PQerrorMessage(conn)))); + + if (!status) + PG_RE_THROW(); + } + PG_END_TRY(); +} + /* * postgresIsForeignRelUpdatable * Determine whether a foreign table supports INSERT, UPDATE and/or diff --git a/contrib/postgres_fdw/postgres_fdw.h b/contrib/postgres_fdw/postgres_fdw.h index eef410db39..8fc5ff018f 100644 --- a/contrib/postgres_fdw/postgres_fdw.h +++ b/contrib/postgres_fdw/postgres_fdw.h @@ -162,6 +162,7 @@ extern void deparseInsertSql(StringInfo buf, RangeTblEntry *rte, List *targetAttrs, bool doNothing, List *withCheckOptionList, List *returningList, List **retrieved_attrs); +extern void deparseCopyFromSql(StringInfo buf, Relation rel); extern void deparseUpdateSql(StringInfo buf, RangeTblEntry *rte, Index rtindex, Relation rel, List *targetAttrs, diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql index d452d06343..1a56432f0f 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -2213,6 +2213,23 @@ alter table loc2 drop constraint loc2_f1positive; delete from rem2; +create table foo (a int) partition by list (a); +create table foo1 (like foo); +create foreign table ffoo1 partition of foo for values in (1) + server loopback options (table_name 'foo1'); +create table foo2 (like foo); +create foreign table ffoo2 partition of foo for values in (2) + server loopback options (table_name 'foo2'); +create function print_new_row() returns trigger language plpgsql as $$ + begin raise notice '%', new; return new; end; $$; +create trigger ffoo1_br_trig before insert on ffoo1 + for each row execute function print_new_row(); + +copy foo from stdin; +1 +2 +\. + -- Test local triggers create trigger trig_stmt_before before insert on rem2 for each statement execute procedure trigger_func(); @@ -2313,6 +2330,34 @@ drop trigger loc2_trig_row_before_insert on loc2; delete from rem2; +alter table loc2 drop column f1; +alter table loc2 drop column f2; +copy rem2 from stdin; +1 foo +2 bar +\. + +alter table loc2 add column f1 int; +alter table loc2 add column f2 int; +select * from rem2; + +-- dropped columns locally and on the foreign server +alter table rem2 drop column f1; +alter table rem2 drop column f2; +copy rem2 from stdin; + + +\. +select * from rem2; + +alter table loc2 drop column f1; +alter table loc2 drop column f2; +copy rem2 from stdin; + + +\. +select * from rem2; + -- test COPY FROM with foreign table created in the same transaction create table loc3 (f1 int, f2 text); begin; diff --git a/doc/src/sgml/fdwhandler.sgml b/doc/src/sgml/fdwhandler.sgml index 72fa127212..81728945ea 100644 --- a/doc/src/sgml/fdwhandler.sgml +++ b/doc/src/sgml/fdwhandler.sgml @@ -796,6 +796,81 @@ EndForeignInsert(EState *estate, <para> <programlisting> +void +BeginForeignCopy(ModifyTableState *mtstate, + ResultRelInfo *rinfo); +</programlisting> + + Begin executing an copy operation on a foreign table. This routine is + called right before the first call of <function>ExecForeignCopy</function> + routine for the foreign table. It should perform any initialization needed + prior to the actual COPY FROM operation. + Subsequently, <function>ExecForeignCopy</function> will be called for + a bulk of tuples to be copied into the foreign table. + </para> + + <para> + <literal>mtstate</literal> is the overall state of the + <structname>ModifyTable</structname> plan node being executed; global data about + the plan and execution state is available via this structure. + <literal>rinfo</literal> is the <structname>ResultRelInfo</structname> struct describing + the target foreign table. (The <structfield>ri_FdwState</structfield> field of + <structname>ResultRelInfo</structname> is available for the FDW to store any + private state it needs for this operation.) + </para> + + <para> + When this is called by a <command>COPY FROM</command> command, the + plan-related global data in <literal>mtstate</literal> is not provided. + </para> + + <para> + If the <function>BeginForeignCopy</function> pointer is set to + <literal>NULL</literal>, no action is taken for the initialization. + </para> + + <para> +<programlisting> +void +EndForeignCopy(EState *estate, + ResultRelInfo *rinfo); +</programlisting> + + End the copy operation and release resources. It is normally not important + to release palloc'd memory, but for example open files and connections + to remote servers should be cleaned up. + </para> + + <para> + If the <function>EndForeignCopy</function> pointer is set to + <literal>NULL</literal>, no action is taken for the termination. + </para> + + <para> +<programlisting> +TupleTableSlot * +ExecForeignCopy(ResultRelInfo *rinfo, + TupleTableSlot **slots, + int nslots); +</programlisting> + + Copy a bulk of tuples into the foreign table. + <literal>estate</literal> is global execution state for the query. + <literal>rinfo</literal> is the <structname>ResultRelInfo</structname> struct describing + the target foreign table. + <literal>slots</literal> contains the tuples to be inserted; it will match the + row-type definition of the foreign table. + <literal>nslots</literal> cis a number of tuples in the <literal>slots</literal> + </para> + + <para> + If the <function>ExecForeignCopy</function> pointer is set to + <literal>NULL</literal>, attempts to insert into the foreign table will fail + with an error message. + </para> + + <para> +<programlisting> int IsForeignRelUpdatable(Relation rel); </programlisting> diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c index 4e63926cb7..62dca2abff 100644 --- a/src/backend/commands/copy.c +++ b/src/backend/commands/copy.c @@ -118,11 +118,14 @@ typedef struct CopyStateData /* parameters from the COPY command */ Relation rel; /* relation to copy to or from */ + TupleDesc tupDesc; /* COPY TO will be used for manual tuple copying + * into the destination */ QueryDesc *queryDesc; /* executable query to copy from */ List *attnumlist; /* integer list of attnums to copy */ char *filename; /* filename, or NULL for STDIN/STDOUT */ bool is_program; /* is 'filename' a program to popen? */ copy_data_source_cb data_source_cb; /* function for reading data */ + copy_data_dest_cb data_dest_cb; /* function for writing data */ bool binary; /* binary format? */ bool freeze; /* freeze rows on loading? */ bool csv_mode; /* Comma Separated Value format? */ @@ -349,17 +352,12 @@ static const char BinarySignature[11] = "PGCOPY\n\377\r\n\0"; /* non-export function prototypes */ static CopyState BeginCopy(ParseState *pstate, bool is_from, Relation rel, - RawStmt *raw_query, Oid queryRelId, List *attnamelist, - List *options); + TupleDesc srcTupDesc, RawStmt *raw_query, + Oid queryRelId, List *attnamelist, List *options); static void EndCopy(CopyState cstate); static void ClosePipeToProgram(CopyState cstate); -static CopyState BeginCopyTo(ParseState *pstate, Relation rel, RawStmt *query, - Oid queryRelId, const char *filename, bool is_program, - List *attnamelist, List *options); -static void EndCopyTo(CopyState cstate); static uint64 DoCopyTo(CopyState cstate); static uint64 CopyTo(CopyState cstate); -static void CopyOneRowTo(CopyState cstate, TupleTableSlot *slot); static bool CopyReadLine(CopyState cstate); static bool CopyReadLineText(CopyState cstate); static int CopyReadAttributesText(CopyState cstate); @@ -585,7 +583,8 @@ CopySendEndOfRow(CopyState cstate) (void) pq_putmessage('d', fe_msgbuf->data, fe_msgbuf->len); break; case COPY_CALLBACK: - Assert(false); /* Not yet supported. */ + CopySendChar(cstate, '\n'); + cstate->data_dest_cb(fe_msgbuf->data, fe_msgbuf->len); break; } @@ -1114,8 +1113,8 @@ DoCopy(ParseState *pstate, const CopyStmt *stmt, } else { - cstate = BeginCopyTo(pstate, rel, query, relid, - stmt->filename, stmt->is_program, + cstate = BeginCopyTo(pstate, rel, NULL, query, relid, + stmt->filename, stmt->is_program, NULL, stmt->attlist, stmt->options); *processed = DoCopyTo(cstate); /* copy from database to file */ EndCopyTo(cstate); @@ -1497,6 +1496,7 @@ static CopyState BeginCopy(ParseState *pstate, bool is_from, Relation rel, + TupleDesc srcTupDesc, RawStmt *raw_query, Oid queryRelId, List *attnamelist, @@ -1532,6 +1532,11 @@ BeginCopy(ParseState *pstate, tupDesc = RelationGetDescr(cstate->rel); } + else if (srcTupDesc) + { + Assert(!raw_query && !is_from); + tupDesc = cstate->tupDesc = srcTupDesc; + } else { List *rewritten; @@ -1858,20 +1863,25 @@ EndCopy(CopyState cstate) /* * Setup CopyState to read tuples from a table or a query for COPY TO. */ -static CopyState +CopyState BeginCopyTo(ParseState *pstate, Relation rel, + TupleDesc tupDesc, RawStmt *query, Oid queryRelId, const char *filename, bool is_program, + copy_data_dest_cb data_dest_cb, List *attnamelist, List *options) { CopyState cstate; - bool pipe = (filename == NULL); + bool pipe = (filename == NULL) && (data_dest_cb == NULL); MemoryContext oldcontext; + /* Impossible to mix CopyTo modes */ + Assert(rel == NULL || tupDesc == NULL); + if (rel != NULL && rel->rd_rel->relkind != RELKIND_RELATION) { if (rel->rd_rel->relkind == RELKIND_VIEW) @@ -1910,8 +1920,9 @@ BeginCopyTo(ParseState *pstate, RelationGetRelationName(rel)))); } - cstate = BeginCopy(pstate, false, rel, query, queryRelId, attnamelist, - options); + cstate = BeginCopy(pstate, false, rel, tupDesc, query, queryRelId, + attnamelist, options); + oldcontext = MemoryContextSwitchTo(cstate->copycontext); if (pipe) @@ -1920,6 +1931,11 @@ BeginCopyTo(ParseState *pstate, if (whereToSendOutput != DestRemote) cstate->copy_file = stdout; } + else if (data_dest_cb) + { + cstate->copy_dest = COPY_CALLBACK; + cstate->data_dest_cb = data_dest_cb; + } else { cstate->filename = pstrdup(filename); @@ -2006,7 +2022,9 @@ DoCopyTo(CopyState cstate) if (fe_copy) SendCopyBegin(cstate); + CopyToStart(cstate); processed = CopyTo(cstate); + CopyToFinish(cstate); if (fe_copy) SendCopyEnd(cstate); @@ -2029,7 +2047,7 @@ DoCopyTo(CopyState cstate) /* * Clean up storage and release resources for COPY TO. */ -static void +void EndCopyTo(CopyState cstate) { if (cstate->queryDesc != NULL) @@ -2045,19 +2063,22 @@ EndCopyTo(CopyState cstate) EndCopy(cstate); } -/* - * Copy from relation or query TO file. +/* Start COPY TO operation. + * Separated to the routine to prevent duplicate operations in the case of + * manual mode, where tuples are copied to the destination one by one, by call of + * the CopyOneRowTo() routine. */ -static uint64 -CopyTo(CopyState cstate) +void +CopyToStart(CopyState cstate) { TupleDesc tupDesc; int num_phys_attrs; ListCell *cur; - uint64 processed; if (cstate->rel) tupDesc = RelationGetDescr(cstate->rel); + else if (cstate->tupDesc) + tupDesc = cstate->tupDesc; else tupDesc = cstate->queryDesc->tupDesc; num_phys_attrs = tupDesc->natts; @@ -2144,6 +2165,32 @@ CopyTo(CopyState cstate) CopySendEndOfRow(cstate); } } +} + +/* + * Finish COPY TO operation. + */ +void +CopyToFinish(CopyState cstate) +{ + if (cstate->binary) + { + /* Generate trailer for a binary copy */ + CopySendInt16(cstate, -1); + /* Need to flush out the trailer */ + CopySendEndOfRow(cstate); + } + + MemoryContextDelete(cstate->rowcontext); +} + +/* + * Copy from relation or query TO file. + */ +static uint64 +CopyTo(CopyState cstate) +{ + uint64 processed; if (cstate->rel) { @@ -2175,24 +2222,13 @@ CopyTo(CopyState cstate) ExecutorRun(cstate->queryDesc, ForwardScanDirection, 0L, true); processed = ((DR_copy *) cstate->queryDesc->dest)->processed; } - - if (cstate->binary) - { - /* Generate trailer for a binary copy */ - CopySendInt16(cstate, -1); - /* Need to flush out the trailer */ - CopySendEndOfRow(cstate); - } - - MemoryContextDelete(cstate->rowcontext); - return processed; } /* * Emit one row during CopyTo(). */ -static void +void CopyOneRowTo(CopyState cstate, TupleTableSlot *slot) { bool need_delim = false; @@ -2485,53 +2521,64 @@ CopyMultiInsertBufferFlush(CopyMultiInsertInfo *miinfo, cstate->line_buf_valid = false; save_cur_lineno = cstate->cur_lineno; - /* - * table_multi_insert may leak memory, so switch to short-lived memory - * context before calling it. - */ - oldcontext = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); - table_multi_insert(resultRelInfo->ri_RelationDesc, - slots, - nused, - mycid, - ti_options, - buffer->bistate); - MemoryContextSwitchTo(oldcontext); - - for (i = 0; i < nused; i++) + if (resultRelInfo->ri_RelationDesc->rd_rel->relkind == RELKIND_FOREIGN_TABLE) + { + /* Flush into foreign table or partition */ + resultRelInfo->ri_FdwRoutine->ExecForeignCopy(resultRelInfo, + slots, + nused); + } + else { /* - * If there are any indexes, update them for all the inserted tuples, - * and run AFTER ROW INSERT triggers. + * table_multi_insert may leak memory, so switch to short-lived memory + * context before calling it. */ - if (resultRelInfo->ri_NumIndices > 0) - { - List *recheckIndexes; - - cstate->cur_lineno = buffer->linenos[i]; - recheckIndexes = - ExecInsertIndexTuples(buffer->slots[i], estate, false, NULL, - NIL); - ExecARInsertTriggers(estate, resultRelInfo, - slots[i], recheckIndexes, - cstate->transition_capture); - list_free(recheckIndexes); - } + oldcontext = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); + + table_multi_insert(resultRelInfo->ri_RelationDesc, + slots, + nused, + mycid, + ti_options, + buffer->bistate); + MemoryContextSwitchTo(oldcontext); - /* - * There's no indexes, but see if we need to run AFTER ROW INSERT - * triggers anyway. - */ - else if (resultRelInfo->ri_TrigDesc != NULL && - (resultRelInfo->ri_TrigDesc->trig_insert_after_row || - resultRelInfo->ri_TrigDesc->trig_insert_new_table)) + for (i = 0; i < nused; i++) { - cstate->cur_lineno = buffer->linenos[i]; - ExecARInsertTriggers(estate, resultRelInfo, - slots[i], NIL, cstate->transition_capture); - } + /* + * If there are any indexes, update them for all the inserted tuples, + * and run AFTER ROW INSERT triggers. + */ + if (resultRelInfo->ri_NumIndices > 0) + { + List *recheckIndexes; + + cstate->cur_lineno = buffer->linenos[i]; + recheckIndexes = + ExecInsertIndexTuples(buffer->slots[i], estate, false, NULL, + NIL); + ExecARInsertTriggers(estate, resultRelInfo, + slots[i], recheckIndexes, + cstate->transition_capture); + list_free(recheckIndexes); + } - ExecClearTuple(slots[i]); + /* + * There's no indexes, but see if we need to run AFTER ROW INSERT + * triggers anyway. + */ + else if (resultRelInfo->ri_TrigDesc != NULL && + (resultRelInfo->ri_TrigDesc->trig_insert_after_row || + resultRelInfo->ri_TrigDesc->trig_insert_new_table)) + { + cstate->cur_lineno = buffer->linenos[i]; + ExecARInsertTriggers(estate, resultRelInfo, + slots[i], NIL, cstate->transition_capture); + } + + ExecClearTuple(slots[i]); + } } /* Mark that all slots are free */ @@ -2894,10 +2941,16 @@ CopyFrom(CopyState cstate) * Init COPY into foreign table. Initialization of copying into foreign * partitions will be done later. */ - if (target_resultRelInfo->ri_FdwRoutine != NULL && - target_resultRelInfo->ri_FdwRoutine->BeginForeignInsert != NULL) - target_resultRelInfo->ri_FdwRoutine->BeginForeignInsert(mtstate, + if (target_resultRelInfo->ri_FdwRoutine != NULL) + { + if (target_resultRelInfo->ri_usesMultiInsert && + target_resultRelInfo->ri_FdwRoutine->BeginForeignCopy != NULL) + target_resultRelInfo->ri_FdwRoutine->BeginForeignCopy(mtstate, resultRelInfo); + else if (target_resultRelInfo->ri_FdwRoutine->BeginForeignInsert != NULL) + target_resultRelInfo->ri_FdwRoutine->BeginForeignInsert(mtstate, + resultRelInfo); + } /* Prepare to catch AFTER triggers. */ AfterTriggerBeginQuery(); @@ -3295,10 +3348,16 @@ CopyFrom(CopyState cstate) ExecResetTupleTable(estate->es_tupleTable, false); /* Allow the FDW to shut down */ - if (target_resultRelInfo->ri_FdwRoutine != NULL && - target_resultRelInfo->ri_FdwRoutine->EndForeignInsert != NULL) - target_resultRelInfo->ri_FdwRoutine->EndForeignInsert(estate, + if (target_resultRelInfo->ri_FdwRoutine != NULL) + { + if (target_resultRelInfo->ri_usesMultiInsert && + target_resultRelInfo->ri_FdwRoutine->EndForeignCopy != NULL) + target_resultRelInfo->ri_FdwRoutine->EndForeignCopy(estate, target_resultRelInfo); + else if (target_resultRelInfo->ri_FdwRoutine->EndForeignInsert != NULL) + target_resultRelInfo->ri_FdwRoutine->EndForeignInsert(estate, + target_resultRelInfo); + } /* Tear down the multi-insert buffer data */ CopyMultiInsertInfoCleanup(&multiInsertInfo); @@ -3350,7 +3409,8 @@ BeginCopyFrom(ParseState *pstate, MemoryContext oldcontext; bool volatile_defexprs; - cstate = BeginCopy(pstate, true, rel, NULL, InvalidOid, attnamelist, options); + cstate = BeginCopy(pstate, true, rel, NULL, NULL, InvalidOid, attnamelist, + options); oldcontext = MemoryContextSwitchTo(cstate->copycontext); /* Initialize state variables */ diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c index 73f78f287a..8d31cd0f56 100644 --- a/src/backend/executor/execMain.c +++ b/src/backend/executor/execMain.c @@ -1383,9 +1383,13 @@ InitResultRelInfo(ResultRelInfo *resultRelInfo, */ resultRelInfo->ri_usesMultiInsert = false; } - else if (resultRelInfo->ri_FdwRoutine != NULL) + else if (resultRelInfo->ri_FdwRoutine != NULL && + resultRelInfo->ri_FdwRoutine->ExecForeignCopy == NULL) { - /* Foreign tables don't support multi-inserts. */ + /* + * Foreign tables don't support multi-inserts, unless their FDW + * provides the necessary COPY interface. + */ resultRelInfo->ri_usesMultiInsert = false; } else diff --git a/src/backend/executor/execPartition.c b/src/backend/executor/execPartition.c index 8d01f5098d..686d20362d 100644 --- a/src/backend/executor/execPartition.c +++ b/src/backend/executor/execPartition.c @@ -995,9 +995,14 @@ ExecInitRoutingInfo(ModifyTableState *mtstate, * If the partition is a foreign table, let the FDW init itself for * routing tuples to the partition. */ - if (partRelInfo->ri_FdwRoutine != NULL && - partRelInfo->ri_FdwRoutine->BeginForeignInsert != NULL) - partRelInfo->ri_FdwRoutine->BeginForeignInsert(mtstate, partRelInfo); + if (partRelInfo->ri_FdwRoutine != NULL) + { + if (partRelInfo->ri_usesMultiInsert && + partRelInfo->ri_FdwRoutine->BeginForeignCopy != NULL) + partRelInfo->ri_FdwRoutine->BeginForeignCopy(mtstate, partRelInfo); + else if (partRelInfo->ri_FdwRoutine->BeginForeignInsert != NULL) + partRelInfo->ri_FdwRoutine->BeginForeignInsert(mtstate, partRelInfo); + } partRelInfo->ri_PartitionInfo = partrouteinfo; partRelInfo->ri_CopyMultiInsertBuffer = NULL; @@ -1199,10 +1204,18 @@ ExecCleanupTupleRouting(ModifyTableState *mtstate, ResultRelInfo *resultRelInfo = proute->partitions[i]; /* Allow any FDWs to shut down */ - if (resultRelInfo->ri_FdwRoutine != NULL && - resultRelInfo->ri_FdwRoutine->EndForeignInsert != NULL) - resultRelInfo->ri_FdwRoutine->EndForeignInsert(mtstate->ps.state, - resultRelInfo); + if (resultRelInfo->ri_FdwRoutine != NULL) + { + if (resultRelInfo->ri_usesMultiInsert) + { + Assert(resultRelInfo->ri_FdwRoutine->EndForeignCopy != NULL); + resultRelInfo->ri_FdwRoutine->EndForeignCopy(mtstate->ps.state, + resultRelInfo); + } + else if (resultRelInfo->ri_FdwRoutine->EndForeignInsert != NULL) + resultRelInfo->ri_FdwRoutine->EndForeignInsert(mtstate->ps.state, + resultRelInfo); + } /* * Check if this result rel is one belonging to the node's subplans, diff --git a/src/include/commands/copy.h b/src/include/commands/copy.h index c639833565..08309149ea 100644 --- a/src/include/commands/copy.h +++ b/src/include/commands/copy.h @@ -22,6 +22,7 @@ /* CopyStateData is private in commands/copy.c */ typedef struct CopyStateData *CopyState; typedef int (*copy_data_source_cb) (void *outbuf, int minread, int maxread); +typedef void (*copy_data_dest_cb) (void *outbuf, int len); extern void DoCopy(ParseState *state, const CopyStmt *stmt, int stmt_location, int stmt_len, @@ -39,6 +40,16 @@ extern void CopyFromErrorCallback(void *arg); extern uint64 CopyFrom(CopyState cstate); +extern CopyState BeginCopyTo(ParseState *pstate, Relation rel, + TupleDesc tupDesc, RawStmt *query, + Oid queryRelId, const char *filename, bool is_program, + copy_data_dest_cb data_dest_cb, List *attnamelist, + List *options); +extern void EndCopyTo(CopyState cstate); +extern void CopyOneRowTo(CopyState cstate, TupleTableSlot *slot); +extern void CopyToStart(CopyState cstate); +extern void CopyToFinish(CopyState cstate); + extern DestReceiver *CreateCopyDestReceiver(void); #endif /* COPY_H */ diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h index 95556dfb15..e932bdf2f4 100644 --- a/src/include/foreign/fdwapi.h +++ b/src/include/foreign/fdwapi.h @@ -104,6 +104,16 @@ typedef void (*BeginForeignInsert_function) (ModifyTableState *mtstate, typedef void (*EndForeignInsert_function) (EState *estate, ResultRelInfo *rinfo); +typedef void (*BeginForeignCopy_function) (ModifyTableState *mtstate, + ResultRelInfo *rinfo); + +typedef void (*EndForeignCopy_function) (EState *estate, + ResultRelInfo *rinfo); + +typedef void (*ExecForeignCopy_function) (ResultRelInfo *rinfo, + TupleTableSlot **slots, + int nslots); + typedef int (*IsForeignRelUpdatable_function) (Relation rel); typedef bool (*PlanDirectModify_function) (PlannerInfo *root, @@ -220,6 +230,11 @@ typedef struct FdwRoutine IterateDirectModify_function IterateDirectModify; EndDirectModify_function EndDirectModify; + /* COPY a bulk of tuples into a foreign relation */ + BeginForeignCopy_function BeginForeignCopy; + EndForeignCopy_function EndForeignCopy; + ExecForeignCopy_function ExecForeignCopy; + /* Functions for SELECT FOR UPDATE/SHARE row locking */ GetForeignRowMarkType_function GetForeignRowMarkType; RefetchForeignRow_function RefetchForeignRow; -- 2.25.1 --------------0FC1AE37569A4E35540BBCE2-- ^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: Pluggable toaster @ 2023-02-05 21:10 Nikita Malakhov <[email protected]> 0 siblings, 1 reply; 6+ messages in thread From: Nikita Malakhov @ 2023-02-05 21:10 UTC (permalink / raw) To: Alvaro Herrera <[email protected]>; +Cc: vignesh C <[email protected]>; Aleksander Alekseev <[email protected]>; pgsql-hackers; Jacob Champion <[email protected]>; Andres Freund <[email protected]>; Matthias van de Meent <[email protected]>; Robert Haas <[email protected]>; Greg Stark <[email protected]>; Teodor Sigaev <[email protected]> Hi hackers! In response to opinion in thread on Compresson dictionaries for JSONb [1] >The approaches are completely different, >but they seem to be trying to fix the same problem: the fact that the >default TOAST stuff isn't good enough for JSONB. The problem, actually, is that the default TOAST is often not good for modern loads and amounts of data.Pluggable TOAST is based not only on pure enthusiasm, but on demands and tickets from production databases.The main demand is effective and transparent storage subsystem for large values for some problematic types of data, which we already have, with proven efficiency. So we're really quite surprised that it has got so little feedback. We've got some opinions on approach but there is no any general one on the approach itself except doubts about the TOAST mechanism needs revision at all. Currently we're busy revising the whole Pluggable TOAST API to make it available as an extension and based on hooks to minimize changes in the core. It will be available soon. > [1] https://postgr.es/m/[email protected] > -- Nikita Malakhov Postgres Professional https://postgrespro.ru/ ^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: Pluggable toaster @ 2023-02-06 10:49 Alvaro Herrera <[email protected]> parent: Nikita Malakhov <[email protected]> 0 siblings, 1 reply; 6+ messages in thread From: Alvaro Herrera @ 2023-02-06 10:49 UTC (permalink / raw) To: Nikita Malakhov <[email protected]>; +Cc: vignesh C <[email protected]>; Aleksander Alekseev <[email protected]>; pgsql-hackers; Jacob Champion <[email protected]>; Andres Freund <[email protected]>; Matthias van de Meent <[email protected]>; Robert Haas <[email protected]>; Greg Stark <[email protected]>; Teodor Sigaev <[email protected]> On 2023-Feb-06, Nikita Malakhov wrote: > Currently we're busy revising the whole Pluggable TOAST API to make it > available as an extension and based on hooks to minimize changes in > the core. It will be available soon. Hmm, I'm not sure why would PGDG want to accept such a thing. I read "minimize changes" as "open source Postgres can keep their crap implementation and companies will offer good ones for a fee". I'd rather have something that can give users direct benefit -- not hide it behind proprietary licenses forcing each company to implement their own performant toaster. -- Álvaro Herrera 48°01'N 7°57'E — https://www.EnterpriseDB.com/ ^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: Pluggable toaster @ 2023-02-06 13:38 Nikita Malakhov <[email protected]> parent: Alvaro Herrera <[email protected]> 0 siblings, 1 reply; 6+ messages in thread From: Nikita Malakhov @ 2023-02-06 13:38 UTC (permalink / raw) To: Alvaro Herrera <[email protected]>; +Cc: vignesh C <[email protected]>; Aleksander Alekseev <[email protected]>; pgsql-hackers; Jacob Champion <[email protected]>; Andres Freund <[email protected]>; Matthias van de Meent <[email protected]>; Robert Haas <[email protected]>; Greg Stark <[email protected]>; Teodor Sigaev <[email protected]> Hi! Existing TOAST has several very painful drawbacks - lack of UPDATE operation, bloating of TOAST tables, and limits that are implicitly implied on base tables by their TOAST tables, so it is seems not fair to say that Pluggable TOAST does not solve any problems but just introduces new ones. The main reason behind this decision is that keeping the first implementation on the side of the vanilla (I mean rebasing it) over time is very difficult due to the very invasive nature of this solution. So we decided to reduce changes in the core to the minimum necessary to make it available through the hooks, because the hooks part is very lightweight and simple to keep rebasing onto the vanilla core. We plan to keep this extension free with the PostgreSQL license, so any PostgreSQL user could benefit from the TOAST on steroids, and sometimes in the future it will be a much simpler task to integrate the Pluggable TOAST into the vanilla, along with our advanced TOAST implementations which we plan to keep under Open Source licenses too. On Mon, Feb 6, 2023 at 1:49 PM Alvaro Herrera <[email protected]> wrote: > On 2023-Feb-06, Nikita Malakhov wrote: > > > Currently we're busy revising the whole Pluggable TOAST API to make it > > available as an extension and based on hooks to minimize changes in > > the core. It will be available soon. > > Hmm, I'm not sure why would PGDG want to accept such a thing. I read > "minimize changes" as "open source Postgres can keep their crap > implementation and companies will offer good ones for a fee". I'd > rather have something that can give users direct benefit -- not hide it > behind proprietary licenses forcing each company to implement their own > performant toaster. > > -- > Álvaro Herrera 48°01'N 7°57'E — > https://www.EnterpriseDB.com/ > -- Regards, Nikita Malakhov Postgres Professional https://postgrespro.ru/ ^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: Pluggable toaster @ 2023-02-06 19:24 Andres Freund <[email protected]> parent: Nikita Malakhov <[email protected]> 0 siblings, 1 reply; 6+ messages in thread From: Andres Freund @ 2023-02-06 19:24 UTC (permalink / raw) To: Nikita Malakhov <[email protected]>; +Cc: Alvaro Herrera <[email protected]>; vignesh C <[email protected]>; Aleksander Alekseev <[email protected]>; pgsql-hackers; Jacob Champion <[email protected]>; Matthias van de Meent <[email protected]>; Robert Haas <[email protected]>; Greg Stark <[email protected]>; Teodor Sigaev <[email protected]> Hi, On 2023-02-06 16:38:01 +0300, Nikita Malakhov wrote: > So we decided to reduce changes in the core to the minimum necessary > to make it available through the hooks, because the hooks part is very > lightweight and simple to keep rebasing onto the vanilla core. At least I don't think we should accept such hooks. I don't think I am alone in that. Greetings, Andres Freund ^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: Pluggable toaster @ 2023-02-06 20:11 Matthias van de Meent <[email protected]> parent: Andres Freund <[email protected]> 0 siblings, 0 replies; 6+ messages in thread From: Matthias van de Meent @ 2023-02-06 20:11 UTC (permalink / raw) To: Andres Freund <[email protected]>; +Cc: Nikita Malakhov <[email protected]>; Alvaro Herrera <[email protected]>; vignesh C <[email protected]>; Aleksander Alekseev <[email protected]>; pgsql-hackers; Jacob Champion <[email protected]>; Robert Haas <[email protected]>; Greg Stark <[email protected]>; Teodor Sigaev <[email protected]> On Mon, 6 Feb 2023 at 20:24, Andres Freund <[email protected]> wrote: > > Hi, > > On 2023-02-06 16:38:01 +0300, Nikita Malakhov wrote: > > So we decided to reduce changes in the core to the minimum necessary > > to make it available through the hooks, because the hooks part is very > > lightweight and simple to keep rebasing onto the vanilla core. > > At least I don't think we should accept such hooks. I don't think I am alone > in that. +1 Assuming type-aware TOASTing is the goal, I don't think hooks are the right design to implement that. -Matthias van de Meent ^ permalink raw reply [nested|flat] 6+ messages in thread
end of thread, other threads:[~2023-02-06 20:11 UTC | newest] Thread overview: 6+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2020-07-09 06:16 [PATCH 2/2] Fast COPY FROM into the foreign or sharded table. Andrey Lepikhov <[email protected]> 2023-02-05 21:10 Re: Pluggable toaster Nikita Malakhov <[email protected]> 2023-02-06 10:49 ` Re: Pluggable toaster Alvaro Herrera <[email protected]> 2023-02-06 13:38 ` Re: Pluggable toaster Nikita Malakhov <[email protected]> 2023-02-06 19:24 ` Re: Pluggable toaster Andres Freund <[email protected]> 2023-02-06 20:11 ` Re: Pluggable toaster Matthias van de Meent <[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