From ac43384af911acd0a07b3fae0ab25a9131a4504c Mon Sep 17 00:00:00 2001 From: Andrey Lepikhov Date: Thu, 9 Jul 2020 11:16:56 +0500 Subject: [PATCH] Fast COPY FROM into the foreign or sharded table. 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. --- contrib/postgres_fdw/deparse.c | 60 ++++- .../postgres_fdw/expected/postgres_fdw.out | 33 ++- contrib/postgres_fdw/postgres_fdw.c | 130 ++++++++++ contrib/postgres_fdw/postgres_fdw.h | 1 + contrib/postgres_fdw/sql/postgres_fdw.sql | 28 ++ src/backend/commands/copy.c | 239 ++++++++++++------ src/backend/executor/execMain.c | 1 + src/backend/executor/execPartition.c | 34 ++- src/include/commands/copy.h | 5 + src/include/foreign/fdwapi.h | 15 ++ src/include/nodes/execnodes.h | 8 + 11 files changed, 456 insertions(+), 98 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 82fc1290ef..3a3cca5047 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 ----+----- @@ -8183,6 +8184,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..0db8d74320 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,10 @@ 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 TupleTableSlot *postgresExecForeignCopyIn(ResultRelInfo *resultRelInfo, TupleTableSlot **slots, int nslots); static int postgresIsForeignRelUpdatable(Relation rel); static bool postgresPlanDirectModify(PlannerInfo *root, ModifyTable *plan, @@ -533,6 +539,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; @@ -1847,6 +1856,9 @@ postgresExecForeignInsert(EState *estate, PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState; TupleTableSlot *rslot; + Assert(!resultRelInfo->UseBulkModifying || + resultRelInfo->ri_FdwRoutine->BeginForeignCopyIn == NULL); + /* * If the fmstate has aux_fmstate set, use the aux_fmstate (see * postgresBeginForeignInsert()) @@ -2051,6 +2063,124 @@ postgresEndForeignInsert(EState *estate, finish_foreign_modify(fmstate); } +/* + * + * postgresBeginForeignCopyIn + * Begin an COPY operation on a foreign table + */ +static void +postgresBeginForeignCopyIn(ModifyTableState *mtstate, + ResultRelInfo *resultRelInfo) +{ + PgFdwModifyState *fmstate; + Relation rel = resultRelInfo->ri_RelationDesc; + StringInfoData sql; + RangeTblEntry *rte; + + 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 = BeginForeignCopyTo(resultRelInfo->ri_RelationDesc); + 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); + + EndForeignCopyTo(fmstate->cstate); + pfree(fmstate->cstate); + fmstate->cstate = NULL; + finish_foreign_modify(fmstate); +} + +/* + * + * postgresExecForeignCopyIn + * Send a number of tuples to the foreign relation. + */ +static TupleTableSlot * +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); + + res = PQexec(conn, fmstate->query); + if (PQresultStatus(res) != PGRES_COPY_IN) + pgfdw_report_error(ERROR, res, conn, true, fmstate->query); + PQclear(res); + + PG_TRY(); + { + for (i = 0; i < nslots; i++) + { + char *buf = NextForeignCopyRow(fmstate->cstate, slots[i]); + + if (PQputCopyData(conn, buf, strlen(buf)) <= 0) + { + res = PQgetResult(conn); + pgfdw_report_error(ERROR, res, conn, true, fmstate->query); + } + } + + status = true; + } + PG_FINALLY(); + { + /* 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(); + return NULL; +} + /* * 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..73f98a3152 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -2293,6 +2293,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/src/backend/commands/copy.c b/src/backend/commands/copy.c index 3e199bdfd0..7338c63fe5 100644 --- a/src/backend/commands/copy.c +++ b/src/backend/commands/copy.c @@ -133,6 +133,7 @@ typedef struct CopyStateData 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; bool binary; /* binary format? */ bool freeze; /* freeze rows on loading? */ bool csv_mode; /* Comma Separated Value format? */ @@ -358,8 +359,11 @@ 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); + copy_data_dest_cb data_dest_cb, List *attnamelist, + List *options); static void EndCopyTo(CopyState cstate); +static void CopyToStart(CopyState cstate); +static void CopyToFinish(CopyState cstate); static uint64 DoCopyTo(CopyState cstate); static uint64 CopyTo(CopyState cstate); static void CopyOneRowTo(CopyState cstate, TupleTableSlot *slot); @@ -586,7 +590,9 @@ 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'); + CopySendChar(cstate, '\0'); + cstate->data_dest_cb(fe_msgbuf->data, fe_msgbuf->len); break; } @@ -1074,7 +1080,7 @@ DoCopy(ParseState *pstate, const CopyStmt *stmt, else { cstate = BeginCopyTo(pstate, rel, query, relid, - stmt->filename, stmt->is_program, + stmt->filename, stmt->is_program, NULL, stmt->attlist, stmt->options); *processed = DoCopyTo(cstate); /* copy from database to file */ EndCopyTo(cstate); @@ -1814,6 +1820,32 @@ EndCopy(CopyState cstate) pfree(cstate); } +static char *buf = NULL; +static void +data_dest_cb(void *outbuf, int len) +{ + buf = (char *) palloc(len); + memcpy(buf, (char *) outbuf, len); +} + +CopyState +BeginForeignCopyTo(Relation rel) +{ + CopyState cstate; + + cstate = BeginCopy(NULL, false, rel, NULL, InvalidOid, NIL, NIL); + cstate->copy_dest = COPY_CALLBACK; + cstate->data_dest_cb = data_dest_cb; + CopyToStart(cstate); + return cstate; +} + +void +EndForeignCopyTo(CopyState cstate) +{ + CopyToFinish(cstate); +} + /* * Setup CopyState to read tuples from a table or a query for COPY TO. */ @@ -1824,6 +1856,7 @@ BeginCopyTo(ParseState *pstate, Oid queryRelId, const char *filename, bool is_program, + copy_data_dest_cb data_dest_cb, List *attnamelist, List *options) { @@ -1879,6 +1912,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); @@ -1949,6 +1987,13 @@ BeginCopyTo(ParseState *pstate, return cstate; } +char * +NextForeignCopyRow(CopyState cstate, TupleTableSlot *slot) +{ + CopyOneRowTo(cstate, slot); + return buf; +} + /* * This intermediate routine exists mainly to localize the effects of setjmp * so we don't need to plaster a lot of variables with "volatile". @@ -1965,7 +2010,9 @@ DoCopyTo(CopyState cstate) if (fe_copy) SendCopyBegin(cstate); + CopyToStart(cstate); processed = CopyTo(cstate); + CopyToFinish(cstate); if (fe_copy) SendCopyEnd(cstate); @@ -2004,16 +2051,12 @@ EndCopyTo(CopyState cstate) EndCopy(cstate); } -/* - * Copy from relation or query TO file. - */ -static uint64 -CopyTo(CopyState cstate) +static void +CopyToStart(CopyState cstate) { TupleDesc tupDesc; int num_phys_attrs; ListCell *cur; - uint64 processed; if (cstate->rel) tupDesc = RelationGetDescr(cstate->rel); @@ -2103,6 +2146,29 @@ CopyTo(CopyState cstate) CopySendEndOfRow(cstate); } } +} + +static 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) { @@ -2134,17 +2200,6 @@ 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; } @@ -2444,53 +2499,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); + } + + /* + * 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]); + ExecClearTuple(slots[i]); + } } /* Mark that all slots are free */ @@ -2800,11 +2866,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(); @@ -2863,14 +2924,13 @@ CopyFrom(CopyState cstate) */ insertMethod = CIM_SINGLE; } - else if (resultRelInfo->ri_FdwRoutine != NULL || - cstate->volatile_defexprs) + else if (cstate->volatile_defexprs || list_length(cstate->attnumlist) == 0) { /* - * 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. + * 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 @@ -2910,6 +2970,24 @@ CopyFrom(CopyState cstate) estate, mycid, ti_options); } + if (insertMethod != CIM_SINGLE) + resultRelInfo->UseBulkModifying = true; + + /* + * Init COPY into foreign table. Initialization of copying into foreign + * partitions will be done later. + */ + if (target_resultRelInfo->ri_FdwRoutine != NULL) + { + if (target_resultRelInfo->UseBulkModifying && + 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); + } + /* * If not using batch mode (which allocates slots as needed) set up a * tuple slot too. When inserting into a partitioned table, we also need @@ -3033,7 +3111,7 @@ CopyFrom(CopyState cstate) leafpart_use_multi_insert = insertMethod == CIM_MULTI_CONDITIONAL && !has_before_insert_row_trig && !has_instead_insert_row_trig && - resultRelInfo->ri_FdwRoutine == NULL; + (resultRelInfo->ri_FdwRoutine == NULL || resultRelInfo->UseBulkModifying); /* Set the multi-insert buffer to use for this partition. */ if (leafpart_use_multi_insert) @@ -3292,10 +3370,17 @@ 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->UseBulkModifying && + 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); + target_resultRelInfo->UseBulkModifying = false; + } /* Tear down the multi-insert buffer data */ if (insertMethod != CIM_SINGLE) diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c index 4fdffad6f3..d3e8f1c720 100644 --- a/src/backend/executor/execMain.c +++ b/src/backend/executor/execMain.c @@ -1345,6 +1345,7 @@ InitResultRelInfo(ResultRelInfo *resultRelInfo, resultRelInfo->ri_PartitionRoot = partition_root; resultRelInfo->ri_PartitionInfo = NULL; /* may be set later */ resultRelInfo->ri_CopyMultiInsertBuffer = NULL; + resultRelInfo->UseBulkModifying = false; } /* diff --git a/src/backend/executor/execPartition.c b/src/backend/executor/execPartition.c index fb6ce49056..c216296811 100644 --- a/src/backend/executor/execPartition.c +++ b/src/backend/executor/execPartition.c @@ -526,6 +526,11 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate, rootrel, estate->es_instrument); + if (rootResultRelInfo->UseBulkModifying && + leaf_part_rri->ri_FdwRoutine != NULL && + leaf_part_rri->ri_FdwRoutine->BeginForeignCopyIn != NULL) + leaf_part_rri->UseBulkModifying = true; + /* * Verify result relation is a valid target for an INSERT. An UPDATE of a * partition-key becomes a DELETE+INSERT operation, so this check is still @@ -937,9 +942,16 @@ 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->UseBulkModifying) + { + Assert(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 +1133,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->UseBulkModifying) + { + 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/include/commands/copy.h b/src/include/commands/copy.h index c639833565..ef119a761a 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, @@ -41,4 +42,8 @@ extern uint64 CopyFrom(CopyState cstate); extern DestReceiver *CreateCopyDestReceiver(void); +extern CopyState BeginForeignCopyTo(Relation rel); +extern char *NextForeignCopyRow(CopyState cstate, TupleTableSlot *slot); +extern void EndForeignCopyTo(CopyState cstate); + #endif /* COPY_H */ diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h index 95556dfb15..073eeee2ca 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 TupleTableSlot *(*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 0187989fd1..8ac366a659 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -491,6 +491,14 @@ typedef struct ResultRelInfo /* For use by copy.c when performing multi-inserts */ struct CopyMultiInsertBuffer *ri_CopyMultiInsertBuffer; + + /* + * For use by copy.c: + * for partitioned relation "true" means that child relations are allowed for + * using bulk modify operations; for foreign relation (or foreign partition + * of) "true" value means that modify operations must use bulk FDW API. + */ + bool UseBulkModifying; } ResultRelInfo; /* ---------------- -- 2.17.1