From: Andrey Lepikhov <a.lepikhov@postgrespro.ru>
To: Ashutosh Bapat <ashutosh.bapat@2ndquadrant.com>
Cc: Ashutosh Bapat <ashutosh.bapat.oss@gmail.com>
Cc: Etsuro Fujita <etsuro.fujita@gmail.com>
Cc: PostgreSQL-Dev <pgsql-hackers@postgresql.org>
Subject: Re: [POC] Fast COPY FROM command for the table with foreign partitions
Date: Sun, 12 Jul 2020 22:46:16 +0500
Message-ID: <ae25e79e-6122-720e-1349-fb832c6021df@postgrespro.ru> (raw)
In-Reply-To: <CAG-ACPUZ1FRwkRfOnBZN0PYvXSZrBRgaExyRT4QHaN20rf7A8Q@mail.gmail.com>
References: <3d0909dc-3691-a576-208a-90986e55489f@postgrespro.ru>
<CAPmGK15mAiaU85OcvEDE=7fHA3PvA7ChP82b7bEE57Yw20rF4A@mail.gmail.com>
<7a3c3bdd-dc43-68a7-7cc5-ae139f1de4a7@postgrespro.ru>
<CAExHW5uAtyAVL-iuu1Hsd0fycqS5UHoHCLfauYDLQwRucwC9Og@mail.gmail.com>
<bd29904a-4ad3-3619-eb0c-ff04f524f7a1@postgrespro.ru>
<CAG-ACPUZ1FRwkRfOnBZN0PYvXSZrBRgaExyRT4QHaN20rf7A8Q@mail.gmail.com>
22.06.2020 17:11, Ashutosh Bapat пишет:
>
>
> On Wed, 17 Jun 2020 at 11:54, Andrey V. Lepikhov
> <a.lepikhov@postgrespro.ru <mailto:a.lepikhov@postgrespro.ru>> wrote:
>
> On 6/15/20 10:26 AM, Ashutosh Bapat wrote:
> > Thanks Andrey for the patch. I am glad that the patch has taken care
> > of some corner cases already but there exist still more.
> >
> > COPY command constructed doesn't take care of dropped columns. There
> > is code in deparseAnalyzeSql which constructs list of columns for a
> > given foreign relation. 0002 patch attached here, moves that code
> to a
> > separate function and reuses it for COPY. If you find that code
> change
> > useful please include it in the main patch.
>
> Thanks, i included it.
>
> > 2. In the same case, if the foreign table declared locally didn't
> have
> > any non-dropped columns but the relation that it referred to on the
> > foreign server had some non-dropped columns, COPY command fails. I
> > added a test case for this in 0002 but haven't fixed it.
>
> I fixed it.
> This is very special corner case. The problem was that COPY FROM does
> not support semantics like the "INSERT INTO .. DEFAULT VALUES". To
> simplify the solution, i switched off bulk copying for this case.
>
> > I think this work is useful. Please add it to the next commitfest so
> > that it's tracked.
> Ok.
>
>
> It looks like we call BeginForeignInsert and EndForeignInsert even
> though actual copy is performed using BeginForeignCopy, ExecForeignCopy
> and EndForeignCopy. BeginForeignInsert constructs the INSERT query which
> looks unnecessary. Also some of the other PgFdwModifyState members are
> initialized unnecessarily. It also gives an impression that we are using
> INSERT underneath the copy. Instead a better way would be to
> call BeginForeignCopy instead of BeginForeignInsert and EndForeignCopy
> instead of EndForeignInsert, if we are going to use COPY protocol to
> copy data to the foreign server. Corresponding postgres_fdw
> implementations need to change in order to do that.
Fixed.
I replaced names of CopyIn FDW API. Also the partition routing
initializer calls BeginForeignInsert or BeginForeignCopyIn routines in
accordance with value of ResultRelInfo::UseBulkModifying.
I introduced this parameter because foreign partitions can be placed at
foreign servers with different types of foreign wrapper. Not all
wrappers can support CopyIn API.
Also I ran the Tomas Vondra benchmark. At my laptop we have results:
* regular: 5000 ms.
* Tomas buffering patch: 11000 ms.
* This CopyIn patch: 8000 ms.
--
regards,
Andrey Lepikhov
Postgres Professional
Attachments:v4-0001-Fast-COPY-FROM-into-the-foreign-or-sharded-table.patch (text/x-patch, 30.3K) download
From ac43384af911acd0a07b3fae0ab25a9131a4504c Mon Sep 17 00:00:00 2001
From: Andrey Lepikhov <a.lepikhov@postgrespro.ru>
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.cindex 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.outindex 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.cindex 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.hindex 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.sqlindex 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.cindex 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.cindex 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.cindex 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.hindex 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.hindex 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.hindex 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
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: pgsql-hackers@postgresql.org
Cc: a.lepikhov@postgrespro.ru, ashutosh.bapat@2ndquadrant.com, ashutosh.bapat.oss@gmail.com, etsuro.fujita@gmail.com
Subject: Re: [POC] Fast COPY FROM command for the table with foreign partitions
In-Reply-To: <ae25e79e-6122-720e-1349-fb832c6021df@postgrespro.ru>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by DDX for PostgreSQL; see mirroring instructions
for how to clone and mirror all data and code used for this inbox