public inbox for [email protected]help / color / mirror / Atom feed
[PATCH] Fast COPY FROM into the foreign or sharded table. 6+ messages / 3 participants [nested] [flat]
* [PATCH] 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: * BeginForeignCopyIn * EndForeignCopyIn * ExecForeignCopyIn BeginForeignCopyIn and EndForeignCopyIn initialize and free the CopyState of bulk COPY. The ExecForeignCopyIn 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 | 74 ++++ src/backend/commands/copy.c | 398 +++++++++--------- src/backend/commands/tablecmds.c | 1 + src/backend/executor/execMain.c | 53 +++ src/backend/executor/execPartition.c | 28 +- src/backend/replication/logical/worker.c | 2 +- src/include/commands/copy.h | 11 + src/include/executor/executor.h | 1 + src/include/foreign/fdwapi.h | 15 + src/include/nodes/execnodes.h | 9 +- 15 files changed, 669 insertions(+), 218 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 90db550b92..de2638109b 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -8063,8 +8063,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 ----+----- @@ -8075,6 +8076,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(); @@ -8183,6 +8197,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 9fc53cad68..19cf119d08 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 postgresBeginForeignCopyIn(ModifyTableState *mtstate, + ResultRelInfo *resultRelInfo); +static void postgresEndForeignCopyIn(EState *estate, + ResultRelInfo *resultRelInfo); +static void postgresExecForeignCopyIn(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->BeginForeignCopyIn = postgresBeginForeignCopyIn; + routine->EndForeignCopyIn = postgresEndForeignCopyIn; + routine->ExecForeignCopyIn = postgresExecForeignCopyIn; routine->IsForeignRelUpdatable = postgresIsForeignRelUpdatable; routine->PlanDirectModify = postgresPlanDirectModify; routine->BeginDirectModify = postgresBeginDirectModify; @@ -2051,6 +2063,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); + } +} + +/* + * + * postgresBeginForeignCopyIn + * Begin an COPY operation on a foreign table + */ +static void +postgresBeginForeignCopyIn(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; +} + +/* + * postgresEndForeignCopyIn + * Finish an COPY operation on a foreign table + */ +static void +postgresEndForeignCopyIn(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); +} + +/* + * + * postgresExecForeignCopyIn + * Send a number of tuples to the foreign relation. + */ +static void +postgresExecForeignCopyIn(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 83971665e3..aa0b26de77 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -2193,6 +2193,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(); @@ -2293,6 +2310,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 74793035d7..e8fd91a7bc 100644 --- a/doc/src/sgml/fdwhandler.sgml +++ b/doc/src/sgml/fdwhandler.sgml @@ -795,6 +795,80 @@ EndForeignInsert(EState *estate, <para> <programlisting> +void +BeginForeignCopyIn(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>ExecForeignCopyIn</function> + routine for the foreign table. It should perform any initialization needed + prior to the actual COPY FROM operation. + Subsequently, <function>ExecForeignCopyIn</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>BeginForeignCopyIn</function> pointer is set to + <literal>NULL</literal>, no action is taken for the initialization. + </para> + + <para> +void +EndForeignCopyIn(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>EndForeignCopyIn</function> pointer is set to + <literal>NULL</literal>, no action is taken for the termination. + </para> + + <para> +<programlisting> +TupleTableSlot * +ExecForeignCopyIn(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>ExecForeignCopyIn</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 db7d24a511..43cd8c011e 100644 --- a/src/backend/commands/copy.c +++ b/src/backend/commands/copy.c @@ -85,16 +85,6 @@ typedef enum EolType EOL_CRNL } EolType; -/* - * Represents the heap insert method to be used during COPY FROM. - */ -typedef enum CopyInsertMethod -{ - CIM_SINGLE, /* use table_tuple_insert or fdw routine */ - CIM_MULTI, /* always use table_multi_insert */ - CIM_MULTI_CONDITIONAL /* use table_multi_insert only if valid */ -} CopyInsertMethod; - /* * This struct contains all the state variables used throughout a COPY * operation. For simplicity, we use the same struct for all variants of COPY, @@ -128,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? */ @@ -359,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); @@ -595,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; } @@ -1124,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); @@ -1507,6 +1496,7 @@ static CopyState BeginCopy(ParseState *pstate, bool is_from, Relation rel, + TupleDesc srcTupDesc, RawStmt *raw_query, Oid queryRelId, List *attnamelist, @@ -1542,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; @@ -1868,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) @@ -1920,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) @@ -1930,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); @@ -2016,7 +2022,9 @@ DoCopyTo(CopyState cstate) if (fe_copy) SendCopyBegin(cstate); + CopyToStart(cstate); processed = CopyTo(cstate); + CopyToFinish(cstate); if (fe_copy) SendCopyEnd(cstate); @@ -2039,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) @@ -2055,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; @@ -2154,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) { @@ -2185,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; @@ -2495,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->ExecForeignCopyIn(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 */ @@ -2715,12 +2752,11 @@ CopyFrom(CopyState cstate) CommandId mycid = GetCurrentCommandId(true); int ti_options = 0; /* start with default options for insert */ BulkInsertState bistate = NULL; - CopyInsertMethod insertMethod; + bool use_multi_insert; CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */ uint64 processed = 0; bool has_before_insert_row_trig; bool has_instead_insert_row_trig; - bool leafpart_use_multi_insert = false; Assert(cstate->rel); @@ -2820,6 +2856,52 @@ CopyFrom(CopyState cstate) ti_options |= TABLE_INSERT_FROZEN; } + /* + * It's generally more efficient to prepare a bunch of tuples for + * insertion, and insert them in bulk, for example, with one + * table_multi_insert() call than call table_tuple_insert() separately + * for every tuple. However, there are a number of reasons why we might + * not be able to do this. We check some conditions below while some + * other target relation properties are left for InitResultRelInfo() to + * check, because they must also be checked for partitions which are + * initialized later. + */ + if (cstate->volatile_defexprs || list_length(cstate->attnumlist) == 0) + { + /* + * Can't support bufferization of copy into foreign tables without any + * defined columns or if there are any volatile default expressions in the + * table. Similarly to the trigger case above, such expressions may query + * the table we're inserting into. + * + * Note: It does not matter if any partitions have any volatile + * default expressions as we use the defaults from the target of the + * COPY command. + */ + use_multi_insert = false; + } + else if (contain_volatile_functions(cstate->whereClause)) + { + /* + * Can't support multi-inserts if there are any volatile function + * expressions in WHERE clause. Similarly to the trigger case above, + * such expressions may query the table we're inserting into. + */ + use_multi_insert = false; + } + else + { + /* + * Looks okay to try multi-insert, but that may change once we + * check few more properties in InitResultRelInfo(). + * + * For partitioned tables, whether or not to use multi-insert depends + * on the individual parition's properties which are also checked in + * InitResultRelInfo(). + */ + use_multi_insert = true; + } + /* * We need a ResultRelInfo so we can use the regular executor's * index-entry-making machinery. (There used to be a huge amount of code @@ -2830,6 +2912,7 @@ CopyFrom(CopyState cstate) cstate->rel, 1, /* must match rel's position in range_table */ NULL, + use_multi_insert, 0); target_resultRelInfo = resultRelInfo; @@ -2854,11 +2937,6 @@ CopyFrom(CopyState cstate) mtstate->operation = CMD_INSERT; mtstate->resultRelInfo = estate->es_result_relations; - if (resultRelInfo->ri_FdwRoutine != NULL && - resultRelInfo->ri_FdwRoutine->BeginForeignInsert != NULL) - resultRelInfo->ri_FdwRoutine->BeginForeignInsert(mtstate, - resultRelInfo); - /* Prepare to catch AFTER triggers. */ AfterTriggerBeginQuery(); @@ -2886,82 +2964,23 @@ CopyFrom(CopyState cstate) cstate->qualexpr = ExecInitQual(castNode(List, cstate->whereClause), &mtstate->ps); + if (resultRelInfo->ri_usesMultiInsert) + CopyMultiInsertInfoInit(&multiInsertInfo, resultRelInfo, cstate, + estate, mycid, ti_options); + /* - * It's generally more efficient to prepare a bunch of tuples for - * insertion, and insert them in one table_multi_insert() call, than call - * table_tuple_insert() separately for every tuple. However, there are a - * number of reasons why we might not be able to do this. These are - * explained below. + * Init COPY into foreign table. Initialization of copying into foreign + * partitions will be done later. */ - if (resultRelInfo->ri_TrigDesc != NULL && - (resultRelInfo->ri_TrigDesc->trig_insert_before_row || - resultRelInfo->ri_TrigDesc->trig_insert_instead_row)) - { - /* - * Can't support multi-inserts when there are any BEFORE/INSTEAD OF - * triggers on the table. Such triggers might query the table we're - * inserting into and act differently if the tuples that have already - * been processed and prepared for insertion are not there. - */ - insertMethod = CIM_SINGLE; - } - else if (proute != NULL && resultRelInfo->ri_TrigDesc != NULL && - resultRelInfo->ri_TrigDesc->trig_insert_new_table) - { - /* - * For partitioned tables we can't support multi-inserts when there - * are any statement level insert triggers. It might be possible to - * allow partitioned tables with such triggers in the future, but for - * now, CopyMultiInsertInfoFlush expects that any before row insert - * and statement level insert triggers are on the same relation. - */ - insertMethod = CIM_SINGLE; - } - else if (resultRelInfo->ri_FdwRoutine != NULL || - cstate->volatile_defexprs) - { - /* - * Can't support multi-inserts to foreign tables or if there are any - * volatile default expressions in the table. Similarly to the - * trigger case above, such expressions may query the table we're - * inserting into. - * - * Note: It does not matter if any partitions have any volatile - * default expressions as we use the defaults from the target of the - * COPY command. - */ - insertMethod = CIM_SINGLE; - } - else if (contain_volatile_functions(cstate->whereClause)) - { - /* - * Can't support multi-inserts if there are any volatile function - * expressions in WHERE clause. Similarly to the trigger case above, - * such expressions may query the table we're inserting into. - */ - insertMethod = CIM_SINGLE; - } - else + if (target_resultRelInfo->ri_FdwRoutine != NULL) { - /* - * For partitioned tables, we may still be able to perform bulk - * inserts. However, the possibility of this depends on which types - * of triggers exist on the partition. We must disable bulk inserts - * if the partition is a foreign table or it has any before row insert - * or insert instead triggers (same as we checked above for the parent - * table). Since the partition's resultRelInfos are initialized only - * when we actually need to insert the first tuple into them, we must - * have the intermediate insert method of CIM_MULTI_CONDITIONAL to - * flag that we must later determine if we can use bulk-inserts for - * the partition being inserted into. - */ - if (proute) - insertMethod = CIM_MULTI_CONDITIONAL; - else - insertMethod = CIM_MULTI; - - CopyMultiInsertInfoInit(&multiInsertInfo, resultRelInfo, cstate, - estate, mycid, ti_options); + if (target_resultRelInfo->ri_usesMultiInsert && + target_resultRelInfo->ri_FdwRoutine->BeginForeignCopyIn != NULL) + target_resultRelInfo->ri_FdwRoutine->BeginForeignCopyIn(mtstate, + resultRelInfo); + else if (target_resultRelInfo->ri_FdwRoutine->BeginForeignInsert != NULL) + target_resultRelInfo->ri_FdwRoutine->BeginForeignInsert(mtstate, + resultRelInfo); } /* @@ -2970,7 +2989,7 @@ CopyFrom(CopyState cstate) * one, even if we might batch insert, to read the tuple in the root * partition's form. */ - if (insertMethod == CIM_SINGLE || insertMethod == CIM_MULTI_CONDITIONAL) + if (!resultRelInfo->ri_usesMultiInsert || proute) { singleslot = table_slot_create(resultRelInfo->ri_RelationDesc, &estate->es_tupleTable); @@ -3013,7 +3032,7 @@ CopyFrom(CopyState cstate) ResetPerTupleExprContext(estate); /* select slot to (initially) load row into */ - if (insertMethod == CIM_SINGLE || proute) + if (!target_resultRelInfo->ri_usesMultiInsert || proute) { myslot = singleslot; Assert(myslot != NULL); @@ -3021,7 +3040,6 @@ CopyFrom(CopyState cstate) else { Assert(resultRelInfo == target_resultRelInfo); - Assert(insertMethod == CIM_MULTI); myslot = CopyMultiInsertInfoNextFreeSlot(&multiInsertInfo, resultRelInfo); @@ -3080,24 +3098,14 @@ CopyFrom(CopyState cstate) has_instead_insert_row_trig = (resultRelInfo->ri_TrigDesc && resultRelInfo->ri_TrigDesc->trig_insert_instead_row); - /* - * Disable multi-inserts when the partition has BEFORE/INSTEAD - * OF triggers, or if the partition is a foreign partition. - */ - leafpart_use_multi_insert = insertMethod == CIM_MULTI_CONDITIONAL && - !has_before_insert_row_trig && - !has_instead_insert_row_trig && - resultRelInfo->ri_FdwRoutine == NULL; - /* Set the multi-insert buffer to use for this partition. */ - if (leafpart_use_multi_insert) + if (resultRelInfo->ri_usesMultiInsert) { if (resultRelInfo->ri_CopyMultiInsertBuffer == NULL) CopyMultiInsertInfoSetupBuffer(&multiInsertInfo, resultRelInfo); } - else if (insertMethod == CIM_MULTI_CONDITIONAL && - !CopyMultiInsertInfoIsEmpty(&multiInsertInfo)) + else if (!CopyMultiInsertInfoIsEmpty(&multiInsertInfo)) { /* * Flush pending inserts if this partition can't use @@ -3149,7 +3157,7 @@ CopyFrom(CopyState cstate) * rowtype. */ map = resultRelInfo->ri_PartitionInfo->pi_RootToPartitionMap; - if (insertMethod == CIM_SINGLE || !leafpart_use_multi_insert) + if (!resultRelInfo->ri_usesMultiInsert) { /* non batch insert */ if (map != NULL) @@ -3168,9 +3176,6 @@ CopyFrom(CopyState cstate) */ TupleTableSlot *batchslot; - /* no other path available for partitioned table */ - Assert(insertMethod == CIM_MULTI_CONDITIONAL); - batchslot = CopyMultiInsertInfoNextFreeSlot(&multiInsertInfo, resultRelInfo); @@ -3241,7 +3246,7 @@ CopyFrom(CopyState cstate) ExecPartitionCheck(resultRelInfo, myslot, estate, true); /* Store the slot in the multi-insert buffer, when enabled. */ - if (insertMethod == CIM_MULTI || leafpart_use_multi_insert) + if (resultRelInfo->ri_usesMultiInsert) { /* * The slot previously might point into the per-tuple @@ -3316,11 +3321,8 @@ CopyFrom(CopyState cstate) } /* Flush any remaining buffered tuples */ - if (insertMethod != CIM_SINGLE) - { - if (!CopyMultiInsertInfoIsEmpty(&multiInsertInfo)) - CopyMultiInsertInfoFlush(&multiInsertInfo, NULL); - } + if (!CopyMultiInsertInfoIsEmpty(&multiInsertInfo)) + CopyMultiInsertInfoFlush(&multiInsertInfo, NULL); /* Done, clean up */ error_context_stack = errcallback.previous; @@ -3346,14 +3348,19 @@ 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, - target_resultRelInfo); + if (target_resultRelInfo->ri_FdwRoutine != NULL) + { + if (target_resultRelInfo->ri_usesMultiInsert && + target_resultRelInfo->ri_FdwRoutine->EndForeignCopyIn != NULL) + target_resultRelInfo->ri_FdwRoutine->EndForeignCopyIn(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 */ - if (insertMethod != CIM_SINGLE) - CopyMultiInsertInfoCleanup(&multiInsertInfo); + CopyMultiInsertInfoCleanup(&multiInsertInfo); ExecCloseIndices(target_resultRelInfo); @@ -3402,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/commands/tablecmds.c b/src/backend/commands/tablecmds.c index cd989c95e5..2629ceb432 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -1786,6 +1786,7 @@ ExecuteTruncateGuts(List *explicit_rels, List *relids, List *relids_logged, rel, 0, /* dummy rangetable index */ NULL, + false, 0); resultRelInfo++; } diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c index 4fdffad6f3..e08c8b29df 100644 --- a/src/backend/executor/execMain.c +++ b/src/backend/executor/execMain.c @@ -851,6 +851,7 @@ InitPlan(QueryDesc *queryDesc, int eflags) resultRelation, resultRelationIndex, NULL, + false, estate->es_instrument); resultRelInfo++; } @@ -883,6 +884,7 @@ InitPlan(QueryDesc *queryDesc, int eflags) resultRelDesc, resultRelIndex, NULL, + false, estate->es_instrument); resultRelInfo++; } @@ -1278,6 +1280,7 @@ InitResultRelInfo(ResultRelInfo *resultRelInfo, Relation resultRelationDesc, Index resultRelationIndex, Relation partition_root, + bool use_multi_insert, int instrument_options) { List *partition_check = NIL; @@ -1345,6 +1348,55 @@ InitResultRelInfo(ResultRelInfo *resultRelInfo, resultRelInfo->ri_PartitionRoot = partition_root; resultRelInfo->ri_PartitionInfo = NULL; /* may be set later */ resultRelInfo->ri_CopyMultiInsertBuffer = NULL; + + /* + * If the caller has asked to use "multi-insert" mode, check if the + * relation allows it and if it does set ri_usesMultiInsert to true. + */ + if (!use_multi_insert) + { + /* Caller didn't ask for it. */ + resultRelInfo->ri_usesMultiInsert = false; + } + else if (resultRelInfo->ri_TrigDesc != NULL && + (resultRelInfo->ri_TrigDesc->trig_insert_before_row || + resultRelInfo->ri_TrigDesc->trig_insert_instead_row)) + { + /* + * Can't support multi-inserts when there are any BEFORE/INSTEAD OF + * triggers on the table. Such triggers might query the table we're + * inserting into and act differently if the tuples that have already + * been processed and prepared for insertion are not there. + */ + resultRelInfo->ri_usesMultiInsert = false; + } + else if (resultRelationDesc->rd_rel->relkind == RELKIND_PARTITIONED_TABLE && + resultRelInfo->ri_TrigDesc != NULL && + resultRelInfo->ri_TrigDesc->trig_insert_new_table) + { + /* + * For partitioned tables we can't support multi-inserts when there + * are any statement level insert triggers. It might be possible to + * allow partitioned tables with such triggers in the future, but for + * now, CopyMultiInsertInfoFlush expects that any before row insert + * and statement level insert triggers are on the same relation. + */ + resultRelInfo->ri_usesMultiInsert = false; + } + else if (resultRelInfo->ri_FdwRoutine != NULL && + resultRelInfo->ri_FdwRoutine->ExecForeignCopyIn == NULL) + { + /* + * For a foreign table, we can't support multi-inserts unless its FDW + * provides the necessary COPY interface. + */ + resultRelInfo->ri_usesMultiInsert = false; + } + else + { + /* OK, caller can use multi-insert on this relation. */ + resultRelInfo->ri_usesMultiInsert = true; + } } /* @@ -1434,6 +1486,7 @@ ExecGetTriggerResultRel(EState *estate, Oid relid) rel, 0, /* dummy rangetable index */ NULL, + false, estate->es_instrument); estate->es_trig_target_relations = lappend(estate->es_trig_target_relations, rInfo); diff --git a/src/backend/executor/execPartition.c b/src/backend/executor/execPartition.c index 79fcbd6b06..7b72a09fb7 100644 --- a/src/backend/executor/execPartition.c +++ b/src/backend/executor/execPartition.c @@ -524,6 +524,7 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate, partrel, node ? node->rootRelation : 1, rootrel, + rootResultRelInfo->ri_usesMultiInsert, estate->es_instrument); /* @@ -937,9 +938,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->BeginForeignCopyIn != NULL) + partRelInfo->ri_FdwRoutine->BeginForeignCopyIn(mtstate, partRelInfo); + else if (partRelInfo->ri_FdwRoutine->BeginForeignInsert != NULL) + partRelInfo->ri_FdwRoutine->BeginForeignInsert(mtstate, partRelInfo); + } partRelInfo->ri_PartitionInfo = partrouteinfo; partRelInfo->ri_CopyMultiInsertBuffer = NULL; @@ -1121,10 +1127,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->EndForeignCopyIn != NULL); + resultRelInfo->ri_FdwRoutine->EndForeignCopyIn(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/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index b576e342cb..9f9cf2dbdb 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -211,7 +211,7 @@ create_estate_for_relation(LogicalRepRelMapEntry *rel) ExecInitRangeTable(estate, list_make1(rte)); resultRelInfo = makeNode(ResultRelInfo); - InitResultRelInfo(resultRelInfo, rel->localrel, 1, NULL, 0); + InitResultRelInfo(resultRelInfo, rel->localrel, 1, NULL, false, 0); estate->es_result_relations = resultRelInfo; estate->es_num_result_relations = 1; 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/executor/executor.h b/src/include/executor/executor.h index 415e117407..72612bd5a6 100644 --- a/src/include/executor/executor.h +++ b/src/include/executor/executor.h @@ -189,6 +189,7 @@ extern void InitResultRelInfo(ResultRelInfo *resultRelInfo, Relation resultRelationDesc, Index resultRelationIndex, Relation partition_root, + bool use_multi_insert, int instrument_options); extern ResultRelInfo *ExecGetTriggerResultRel(EState *estate, Oid relid); extern void ExecCleanUpTriggerState(EState *estate); diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h index 95556dfb15..11ea451fe4 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 (*BeginForeignCopyIn_function) (ModifyTableState *mtstate, + ResultRelInfo *rinfo); + +typedef void (*EndForeignCopyIn_function) (EState *estate, + ResultRelInfo *rinfo); + +typedef void (*ExecForeignCopyIn_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 */ + BeginForeignCopyIn_function BeginForeignCopyIn; + EndForeignCopyIn_function EndForeignCopyIn; + ExecForeignCopyIn_function ExecForeignCopyIn; + /* Functions for SELECT FOR UPDATE/SHARE row locking */ GetForeignRowMarkType_function GetForeignRowMarkType; RefetchForeignRow_function RefetchForeignRow; diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index 0b42dd6f94..89ae9afaa4 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -489,7 +489,14 @@ typedef struct ResultRelInfo /* Additional information specific to partition tuple routing */ struct PartitionRoutingInfo *ri_PartitionInfo; - /* For use by copy.c when performing multi-inserts */ + /* + * The following fields are currently only relevant to copy.c. + * + * True if okay to use multi-insert on this relation + */ + bool ri_usesMultiInsert; + + /* Buffer allocated to this relation when using multi-insert mode */ struct CopyMultiInsertBuffer *ri_CopyMultiInsertBuffer; } ResultRelInfo; -- 2.17.1 --------------2563DFF8F134DCE2289A1427-- ^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: refactoring relation extension and BufferAlloc(), faster COPY @ 2023-02-10 16:38 Heikki Linnakangas <[email protected]> 0 siblings, 1 reply; 6+ messages in thread From: Heikki Linnakangas @ 2023-02-10 16:38 UTC (permalink / raw) To: Andres Freund <[email protected]>; +Cc: vignesh C <[email protected]>; pgsql-hackers; Thomas Munro <[email protected]>; Melanie Plageman <[email protected]>; Yura Sokolov <[email protected]>; Robert Haas <[email protected]>; David Rowley <[email protected]> I'll continue reviewing this, but here's some feedback on the first two patches: v2-0001-aio-Add-some-error-checking-around-pinning.patch: I wonder if the extra assertion in LockBufHdr() is worth the overhead. It won't add anything without assertions, of course, but still. No objections if you think it's worth it. v2-0002-hio-Release-extension-lock-before-initializing-pa.patch: Looks as far as it goes. It's a bit silly that we use RBM_ZERO_AND_LOCK, which zeroes the page, and then we call PageInit to zero the page again. RBM_ZERO_AND_LOCK only zeroes the page if it wasn't in the buffer cache previously, but with P_NEW, that is always true. - Heikki ^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: refactoring relation extension and BufferAlloc(), faster COPY @ 2023-02-11 21:03 Heikki Linnakangas <[email protected]> parent: Heikki Linnakangas <[email protected]> 0 siblings, 1 reply; 6+ messages in thread From: Heikki Linnakangas @ 2023-02-11 21:03 UTC (permalink / raw) To: Andres Freund <[email protected]>; +Cc: vignesh C <[email protected]>; pgsql-hackers; Thomas Munro <[email protected]>; Melanie Plageman <[email protected]>; Yura Sokolov <[email protected]>; Robert Haas <[email protected]>; David Rowley <[email protected]> > v2-0005-bufmgr-Acquire-and-clean-victim-buffer-separately.patch This can be applied separately from the rest of the patches, which is nice. Some small comments on it: * Needs a rebase, it conflicted slightly with commit f30d62c2fc. * GetVictimBuffer needs a comment to explain what it does. In particular, mention that it returns a buffer that's pinned and known !BM_TAG_VALID. * I suggest renaming 'cur_buf' and other such local variables in GetVictimBufffer to just 'buf'. 'cur' prefix suggests that there is some other buffer involved too, but there is no 'prev' or 'next' or 'other' buffer. The old code called it just 'buf' too, and before this patch it actually was a bit confusing because there were two buffers involved. But with this patch, GetVictimBuffer only deals with one buffer at a time. * This FIXME: > /* OK, do the I/O */ > /* FIXME: These used the wrong smgr before afaict? */ > { > SMgrRelation smgr = smgropen(BufTagGetRelFileLocator(&buf_hdr->tag), > InvalidBackendId); > > TRACE_POSTGRESQL_BUFFER_WRITE_DIRTY_START(buf_hdr->tag.forkNum, > buf_hdr->tag.blockNum, > smgr->smgr_rlocator.locator.spcOid, > smgr->smgr_rlocator.locator.dbOid, > smgr->smgr_rlocator.locator.relNumber); > > FlushBuffer(buf_hdr, smgr, IOOBJECT_RELATION, io_context); > LWLockRelease(content_lock); > > ScheduleBufferTagForWriteback(&BackendWritebackContext, > &buf_hdr->tag); > > TRACE_POSTGRESQL_BUFFER_WRITE_DIRTY_DONE(buf_hdr->tag.forkNum, > buf_hdr->tag.blockNum, > smgr->smgr_rlocator.locator.spcOid, > smgr->smgr_rlocator.locator.dbOid, > smgr->smgr_rlocator.locator.relNumber); > } I believe that was intentional. The probes previously reported the block and relation whose read *caused* the eviction. It was not just the smgr but also the blockNum and forkNum that referred to the block that was being read. There's another pair of probe points, TRACE_POSTGRESQL_BUFFER_FLUSH_START/DONE, inside FlushBuffer that indicate the page that is being flushed. I see that reporting the evicted page is more convenient with this patch, otherwise you'd need to pass the smgr and blocknum of the page that's being read to InvalidateVictimBuffer(). IMHO you can just remove these probe points. We don't need to bend over backwards to maintain specific probe points. * InvalidateVictimBuffer reads the buffer header with an atomic read op, just to check if BM_TAG_VALID is set. If it's not, it does nothing (except for a few Asserts). But the caller has already read the buffer header. Consider refactoring it so that the caller checks VM_TAG_VALID, and only calls InvalidateVictimBuffer if it's set, saving one atomic read in InvalidateVictimBuffer. I think it would be just as readable, so no loss there. I doubt the atomic read makes any measurable performance difference, but it looks redundant. * I don't understand this comment: > /* > * Clear out the buffer's tag and flags and usagecount. We must do > * this to ensure that linear scans of the buffer array don't think > * the buffer is valid. > * > * XXX: This is a pre-existing comment I just moved, but isn't it > * entirely bogus with regard to the tag? We can't do anything with > * the buffer without taking BM_VALID / BM_TAG_VALID into > * account. Likely doesn't matter because we're already dirtying the > * cacheline, but still. > * > */ > ClearBufferTag(&buf_hdr->tag); > buf_state &= ~(BUF_FLAG_MASK | BUF_USAGECOUNT_MASK); > UnlockBufHdr(buf_hdr, buf_state); What exactly is wrong with clearing the tag? What does dirtying the cacheline have to do with the correctness here? * pgindent - Heikki ^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: refactoring relation extension and BufferAlloc(), faster COPY @ 2023-02-11 21:36 Andres Freund <[email protected]> parent: Heikki Linnakangas <[email protected]> 0 siblings, 2 replies; 6+ messages in thread From: Andres Freund @ 2023-02-11 21:36 UTC (permalink / raw) To: Heikki Linnakangas <[email protected]>; +Cc: vignesh C <[email protected]>; pgsql-hackers; Thomas Munro <[email protected]>; Melanie Plageman <[email protected]>; Yura Sokolov <[email protected]>; Robert Haas <[email protected]>; David Rowley <[email protected]> Hi, On 2023-02-11 23:03:56 +0200, Heikki Linnakangas wrote: > > v2-0005-bufmgr-Acquire-and-clean-victim-buffer-separately.patch > This can be applied separately from the rest of the patches, which is nice. > Some small comments on it: Thanks for looking at these! > * Needs a rebase, it conflicted slightly with commit f30d62c2fc. Will work on that. > * GetVictimBuffer needs a comment to explain what it does. In particular, > mention that it returns a buffer that's pinned and known !BM_TAG_VALID. Will add. > * I suggest renaming 'cur_buf' and other such local variables in > GetVictimBufffer to just 'buf'. 'cur' prefix suggests that there is some > other buffer involved too, but there is no 'prev' or 'next' or 'other' > buffer. The old code called it just 'buf' too, and before this patch it > actually was a bit confusing because there were two buffers involved. But > with this patch, GetVictimBuffer only deals with one buffer at a time. Hm. Yea. I probably ended up with these names because initially GetVictimBuffer() wasn't a separate function, and I indeed constantly got confused by which buffer was referenced. > * This FIXME: > > > /* OK, do the I/O */ > > /* FIXME: These used the wrong smgr before afaict? */ > > { > > SMgrRelation smgr = smgropen(BufTagGetRelFileLocator(&buf_hdr->tag), > > InvalidBackendId); > > > > TRACE_POSTGRESQL_BUFFER_WRITE_DIRTY_START(buf_hdr->tag.forkNum, > > buf_hdr->tag.blockNum, > > smgr->smgr_rlocator.locator.spcOid, > > smgr->smgr_rlocator.locator.dbOid, > > smgr->smgr_rlocator.locator.relNumber); > > > > FlushBuffer(buf_hdr, smgr, IOOBJECT_RELATION, io_context); > > LWLockRelease(content_lock); > > > > ScheduleBufferTagForWriteback(&BackendWritebackContext, > > &buf_hdr->tag); > > > > TRACE_POSTGRESQL_BUFFER_WRITE_DIRTY_DONE(buf_hdr->tag.forkNum, > > buf_hdr->tag.blockNum, > > smgr->smgr_rlocator.locator.spcOid, > > smgr->smgr_rlocator.locator.dbOid, > > smgr->smgr_rlocator.locator.relNumber); > > } > > I believe that was intentional. The probes previously reported the block and > relation whose read *caused* the eviction. It was not just the smgr but also > the blockNum and forkNum that referred to the block that was being read. You're probably right. It's certainly not understandable from our docs though: <row> <entry><literal>buffer-write-dirty-start</literal></entry> <entry><literal>(ForkNumber, BlockNumber, Oid, Oid, Oid)</literal></entry> <entry>Probe that fires when a server process begins to write a dirty buffer. (If this happens often, it implies that <xref linkend="guc-shared-buffers"/> is too small or the background writer control parameters need adjustment.) arg0 and arg1 contain the fork and block numbers of the page. arg2, arg3, and arg4 contain the tablespace, database, and relation OIDs identifying the relation.</entry> </row> > I see that reporting the evicted page is more convenient with this patch, > otherwise you'd need to pass the smgr and blocknum of the page that's being > read to InvalidateVictimBuffer(). IMHO you can just remove these probe > points. We don't need to bend over backwards to maintain specific probe > points. Agreed. > * InvalidateVictimBuffer reads the buffer header with an atomic read op, > just to check if BM_TAG_VALID is set. It's not a real atomic op, in the sense of being special instruction. It does force the compiler to actually read from memory, but that's it. But you're right, even that is unnecessary. I think it ended up that way because I also wanted the full buf_hdr, and it seemed somewhat error prone to pass in both. > * I don't understand this comment: > > > /* > > * Clear out the buffer's tag and flags and usagecount. We must do > > * this to ensure that linear scans of the buffer array don't think > > * the buffer is valid. > > * > > * XXX: This is a pre-existing comment I just moved, but isn't it > > * entirely bogus with regard to the tag? We can't do anything with > > * the buffer without taking BM_VALID / BM_TAG_VALID into > > * account. Likely doesn't matter because we're already dirtying the > > * cacheline, but still. > > * > > */ > > ClearBufferTag(&buf_hdr->tag); > > buf_state &= ~(BUF_FLAG_MASK | BUF_USAGECOUNT_MASK); > > UnlockBufHdr(buf_hdr, buf_state); > > What exactly is wrong with clearing the tag? What does dirtying the > cacheline have to do with the correctness here? There's nothing wrong with clearing out the tag, but I don't think it's a hard requirement today, and certainly not for the reason stated above. Validity of the buffer isn't determined by the tag, it's determined by BM_VALID (or, if you interpret valid more widely, BM_TAG_VALID). Without either having pinned the buffer, or holding the buffer header spinlock, the tag can change at any time. And code like DropDatabaseBuffers() knows that, and re-checks the the tag after locking the buffer header spinlock. Afaict, there'd be no correctness issue with removing the ClearBufferTag(). There would be an efficiency issue though, because when encountering an invalid buffer, we'd unnecessarily enter InvalidateBuffer(), which'd find that BM_[TAG_]VALID isn't set, and not to anything. Even though it's not a correctness issue, it seems to me that DropRelationsAllBuffers() etc ought to check if the buffer is BM_TAG_VALID, before doing anything further. Particularly in DropRelationsAllBuffers(), the check we do for each buffer isn't cheap. Doing it for buffers that don't even have a tag seems .. not smart. Greetings, Andres Freund ^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: refactoring relation extension and BufferAlloc(), faster COPY @ 2023-02-11 22:04 Andres Freund <[email protected]> parent: Andres Freund <[email protected]> 1 sibling, 0 replies; 6+ messages in thread From: Andres Freund @ 2023-02-11 22:04 UTC (permalink / raw) To: Heikki Linnakangas <[email protected]>; +Cc: vignesh C <[email protected]>; pgsql-hackers; Thomas Munro <[email protected]>; Melanie Plageman <[email protected]>; Yura Sokolov <[email protected]>; Robert Haas <[email protected]>; David Rowley <[email protected]> On 2023-02-11 13:36:51 -0800, Andres Freund wrote: > Even though it's not a correctness issue, it seems to me that > DropRelationsAllBuffers() etc ought to check if the buffer is BM_TAG_VALID, > before doing anything further. Particularly in DropRelationsAllBuffers(), the > check we do for each buffer isn't cheap. Doing it for buffers that don't even > have a tag seems .. not smart. There's a small regression for a single relation, but after that it's a clear benefit. 32GB shared buffers, empty. The test creates N new relations and then rolls back. tps tps num relations HEAD precheck 1 46.11 45.22 2 43.24 44.87 4 35.14 44.20 8 28.72 42.79 I don't understand the regression at 1, TBH. I think it must be a random code layout issue, because the same pre-check in DropRelationBuffers() (exercised via TRUNCATE of a newly created relation), shows a tiny speedup. ^ permalink raw reply [nested|flat] 6+ messages in thread
* Re: refactoring relation extension and BufferAlloc(), faster COPY @ 2023-02-21 15:16 Heikki Linnakangas <[email protected]> parent: Andres Freund <[email protected]> 1 sibling, 0 replies; 6+ messages in thread From: Heikki Linnakangas @ 2023-02-21 15:16 UTC (permalink / raw) To: Andres Freund <[email protected]>; +Cc: vignesh C <[email protected]>; pgsql-hackers; Thomas Munro <[email protected]>; Melanie Plageman <[email protected]>; Yura Sokolov <[email protected]>; Robert Haas <[email protected]>; David Rowley <[email protected]> On 11/02/2023 23:36, Andres Freund wrote: > On 2023-02-11 23:03:56 +0200, Heikki Linnakangas wrote: >> * I don't understand this comment: >> >>> /* >>> * Clear out the buffer's tag and flags and usagecount. We must do >>> * this to ensure that linear scans of the buffer array don't think >>> * the buffer is valid. >>> * >>> * XXX: This is a pre-existing comment I just moved, but isn't it >>> * entirely bogus with regard to the tag? We can't do anything with >>> * the buffer without taking BM_VALID / BM_TAG_VALID into >>> * account. Likely doesn't matter because we're already dirtying the >>> * cacheline, but still. >>> * >>> */ >>> ClearBufferTag(&buf_hdr->tag); >>> buf_state &= ~(BUF_FLAG_MASK | BUF_USAGECOUNT_MASK); >>> UnlockBufHdr(buf_hdr, buf_state); >> >> What exactly is wrong with clearing the tag? What does dirtying the >> cacheline have to do with the correctness here? > > There's nothing wrong with clearing out the tag, but I don't think it's a hard > requirement today, and certainly not for the reason stated above. > > Validity of the buffer isn't determined by the tag, it's determined by > BM_VALID (or, if you interpret valid more widely, BM_TAG_VALID). > > Without either having pinned the buffer, or holding the buffer header > spinlock, the tag can change at any time. And code like DropDatabaseBuffers() > knows that, and re-checks the the tag after locking the buffer header > spinlock. > > Afaict, there'd be no correctness issue with removing the > ClearBufferTag(). There would be an efficiency issue though, because when > encountering an invalid buffer, we'd unnecessarily enter InvalidateBuffer(), > which'd find that BM_[TAG_]VALID isn't set, and not to anything. Okay, gotcha. > Even though it's not a correctness issue, it seems to me that > DropRelationsAllBuffers() etc ought to check if the buffer is BM_TAG_VALID, > before doing anything further. Particularly in DropRelationsAllBuffers(), the > check we do for each buffer isn't cheap. Doing it for buffers that don't even > have a tag seems .. not smart. Depends on what percentage of buffers are valid, I guess. If all buffers are valid, checking BM_TAG_VALID first would lose. In practice, I doubt it makes any measurable difference either way. Since we're micro-optimizing, I noticed that BufTagMatchesRelFileLocator() compares the fields in order "spcOid, dbOid, relNumber". Before commit 82ac34db20, we used RelFileLocatorEqual(), which has this comment: /* * Note: RelFileLocatorEquals and RelFileLocatorBackendEquals compare relNumber * first since that is most likely to be different in two unequal * RelFileLocators. It is probably redundant to compare spcOid if the other * fields are found equal, but do it anyway to be sure. Likewise for checking * the backend ID in RelFileLocatorBackendEquals. */ So we lost that micro-optimization. Should we reorder the checks in BufTagMatchesRelFileLocator()? - Heikki ^ permalink raw reply [nested|flat] 6+ messages in thread
end of thread, other threads:[~2023-02-21 15:16 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] Fast COPY FROM into the foreign or sharded table. Andrey Lepikhov <[email protected]> 2023-02-10 16:38 Re: refactoring relation extension and BufferAlloc(), faster COPY Heikki Linnakangas <[email protected]> 2023-02-11 21:03 ` Re: refactoring relation extension and BufferAlloc(), faster COPY Heikki Linnakangas <[email protected]> 2023-02-11 21:36 ` Re: refactoring relation extension and BufferAlloc(), faster COPY Andres Freund <[email protected]> 2023-02-11 22:04 ` Re: refactoring relation extension and BufferAlloc(), faster COPY Andres Freund <[email protected]> 2023-02-21 15:16 ` Re: refactoring relation extension and BufferAlloc(), faster COPY Heikki Linnakangas <[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