public inbox for [email protected]  

help / color / mirror / Atom feed

[PATCH 2/7] Write WAL for empty nbtree index build
15+ messages / 3 participants
[nested] [flat]

* [PATCH 2/8] Write WAL for empty nbtree index build
@ 2018-10-11 01:03 Kyotaro Horiguchi <[email protected]>
  0 siblings, 0 replies; 15+ messages in thread

From: Kyotaro Horiguchi @ 2018-10-11 01:03 UTC (permalink / raw)

After relation truncation indexes are also rebuild. It doesn't emit
WAL in minimal mode and if truncation happened within its creation
transaction, crash recovery leaves an empty index heap, which is
considered broken. This patch forces to emit WAL when an index_build
turns into empty nbtree index.
---
 src/backend/access/nbtree/nbtsort.c | 11 +++++++++--
 1 file changed, 9 insertions(+), 2 deletions(-)

diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c
index 46e0831834..e65d4aab0f 100644
--- a/src/backend/access/nbtree/nbtsort.c
+++ b/src/backend/access/nbtree/nbtsort.c
@@ -622,8 +622,15 @@ _bt_blwritepage(BTWriteState *wstate, Page page, BlockNumber blkno)
 	/* Ensure rd_smgr is open (could have been closed by relcache flush!) */
 	RelationOpenSmgr(wstate->index);
 
-	/* XLOG stuff */
-	if (wstate->btws_use_wal)
+	/* XLOG stuff
+	 *
+	 * Even if minimal mode, WAL is required here if truncation happened after
+	 * being created in the same transaction. It is not needed otherwise but
+	 * we don't bother identifying the case precisely.
+	 */
+	if (wstate->btws_use_wal ||
+		(RelationNeedsWAL(wstate->index) &&
+		 (blkno == BTREE_METAPAGE && BTPageGetMeta(page)->btm_root == 0)))
 	{
 		/* We use the heap NEWPAGE record type for this */
 		log_newpage(&wstate->index->rd_node, MAIN_FORKNUM, blkno, page, true);
-- 
2.16.3


----Next_Part(Tue_Mar_26_16_35_07_2019_128)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="v9-0003-Move-XLOG-stuff-from-heap_insert-and-heap_delete.patch"



^ permalink  raw  reply  [nested|flat] 15+ messages in thread

* [PATCH 2/4] Write WAL for empty nbtree index build
@ 2018-10-11 01:03 Kyotaro Horiguchi <[email protected]>
  0 siblings, 0 replies; 15+ messages in thread

From: Kyotaro Horiguchi @ 2018-10-11 01:03 UTC (permalink / raw)

After relation truncation indexes are also rebuild. It doesn't emit
WAL in minimal mode and if truncation happened within its creation
transaction, crash recovery leaves an empty index heap, which is
considered broken. This patch forces to emit WAL when an index_build
turns into empty nbtree index.
---
 src/backend/access/nbtree/nbtsort.c | 15 +++++++++++++--
 1 file changed, 13 insertions(+), 2 deletions(-)

diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c
index dc398e1186..70d4380533 100644
--- a/src/backend/access/nbtree/nbtsort.c
+++ b/src/backend/access/nbtree/nbtsort.c
@@ -611,8 +611,14 @@ _bt_blwritepage(BTWriteState *wstate, Page page, BlockNumber blkno)
 	/* Ensure rd_smgr is open (could have been closed by relcache flush!) */
 	RelationOpenSmgr(wstate->index);
 
-	/* XLOG stuff */
-	if (wstate->btws_use_wal)
+	/* XLOG stuff
+	 *
+	 * Even if minimal mode, WAL is required here if truncation happened after
+	 * being created in the same transaction. It is not needed otherwise but
+	 * we don't bother identifying the case precisely.
+	 */
+	if (wstate->btws_use_wal ||
+		(blkno == BTREE_METAPAGE && BTPageGetMeta(page)->btm_root == 0))
 	{
 		/* We use the heap NEWPAGE record type for this */
 		log_newpage(&wstate->index->rd_node, MAIN_FORKNUM, blkno, page, true);
@@ -1056,6 +1062,11 @@ _bt_uppershutdown(BTWriteState *wstate, BTPageState *state)
 	 * set to point to "P_NONE").  This changes the index to the "valid" state
 	 * by filling in a valid magic number in the metapage.
 	 */
+	/*
+	 * If no tuple was inserted, it's possible that we are truncating a
+	 * relation. We need to emit WAL for the metapage in the case. However it
+	 * is not required elsewise,
+	 */
 	metapage = (Page) palloc(BLCKSZ);
 	_bt_initmetapage(metapage, rootblkno, rootlevel);
 	_bt_blwritepage(wstate, metapage, BTREE_METAPAGE);
-- 
2.16.3


----Next_Part(Mon_Mar_04_12_24_48_2019_788)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="v7-0003-Add-infrastructure-to-WAL-logging-skip-feature.patch"



^ permalink  raw  reply  [nested|flat] 15+ messages in thread

* [PATCH 2/7] Write WAL for empty nbtree index build
@ 2018-10-11 01:03 Kyotaro Horiguchi <[email protected]>
  0 siblings, 0 replies; 15+ messages in thread

From: Kyotaro Horiguchi @ 2018-10-11 01:03 UTC (permalink / raw)

After relation truncation indexes are also rebuild. It doesn't emit
WAL in minimal mode and if truncation happened within its creation
transaction, crash recovery leaves an empty index heap, which is
considered broken. This patch forces to emit WAL when an index_build
turns into empty nbtree index.
---
 src/backend/access/nbtree/nbtsort.c | 12 ++++++++++--
 1 file changed, 10 insertions(+), 2 deletions(-)

diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c
index 14d9545768..5551a9c227 100644
--- a/src/backend/access/nbtree/nbtsort.c
+++ b/src/backend/access/nbtree/nbtsort.c
@@ -622,8 +622,16 @@ _bt_blwritepage(BTWriteState *wstate, Page page, BlockNumber blkno)
 	/* Ensure rd_smgr is open (could have been closed by relcache flush!) */
 	RelationOpenSmgr(wstate->index);
 
-	/* XLOG stuff */
-	if (wstate->btws_use_wal)
+	/* XLOG stuff
+	 *
+	 * Even when wal_level is minimal, WAL is required here if truncation
+	 * happened after being created in the same transaction. This is hacky but
+	 * we cannot use BufferNeedsWAL() stuff for nbtree since it can emit
+	 * atomic WAL records on multiple buffers.
+	 */
+	if (wstate->btws_use_wal ||
+		(RelationNeedsWAL(wstate->index) &&
+		 (blkno == BTREE_METAPAGE && BTPageGetMeta(page)->btm_root == 0)))
 	{
 		/* We use the heap NEWPAGE record type for this */
 		log_newpage(&wstate->index->rd_node, MAIN_FORKNUM, blkno, page, true);
-- 
2.16.3


----Next_Part(Tue_Apr_02_19_54_06_2019_801)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="v10-0003-Move-XLOG-stuff-from-heap_insert-and-heap_delete.patch"



^ permalink  raw  reply  [nested|flat] 15+ messages in thread

* [PATCH 2/7] Write WAL for empty nbtree index build
@ 2018-10-11 01:03 Kyotaro Horiguchi <[email protected]>
  0 siblings, 0 replies; 15+ messages in thread

From: Kyotaro Horiguchi @ 2018-10-11 01:03 UTC (permalink / raw)

After relation truncation indexes are also rebuild. It doesn't emit
WAL in minimal mode and if truncation happened within its creation
transaction, crash recovery leaves an empty index heap, which is
considered broken. This patch forces to emit WAL when an index_build
turns into empty nbtree index.
---
 src/backend/access/nbtree/nbtsort.c | 12 ++++++++++--
 1 file changed, 10 insertions(+), 2 deletions(-)

diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c
index 9ac4c1e1c0..a31d58025f 100644
--- a/src/backend/access/nbtree/nbtsort.c
+++ b/src/backend/access/nbtree/nbtsort.c
@@ -654,8 +654,16 @@ _bt_blwritepage(BTWriteState *wstate, Page page, BlockNumber blkno)
 	/* Ensure rd_smgr is open (could have been closed by relcache flush!) */
 	RelationOpenSmgr(wstate->index);
 
-	/* XLOG stuff */
-	if (wstate->btws_use_wal)
+	/* XLOG stuff
+	 *
+	 * Even when wal_level is minimal, WAL is required here if truncation
+	 * happened after being created in the same transaction. This is hacky but
+	 * we cannot use BufferNeedsWAL() stuff for nbtree since it can emit
+	 * atomic WAL records on multiple buffers.
+	 */
+	if (wstate->btws_use_wal ||
+		(RelationNeedsWAL(wstate->index) &&
+		 (blkno == BTREE_METAPAGE && BTPageGetMeta(page)->btm_root == 0)))
 	{
 		/* We use the heap NEWPAGE record type for this */
 		log_newpage(&wstate->index->rd_node, MAIN_FORKNUM, blkno, page, true);
-- 
2.16.3


----Next_Part(Fri_Apr_05_12_55_20_2019_986)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="v11-0003-Move-XLOG-stuff-from-heap_insert-and-heap_delete.patch"



^ permalink  raw  reply  [nested|flat] 15+ messages in thread

* [PATCH 2/7] Write WAL for empty nbtree index build
@ 2018-10-11 01:03 Kyotaro Horiguchi <[email protected]>
  0 siblings, 0 replies; 15+ messages in thread

From: Kyotaro Horiguchi @ 2018-10-11 01:03 UTC (permalink / raw)

After relation truncation indexes are also rebuild. It doesn't emit
WAL in minimal mode and if truncation happened within its creation
transaction, crash recovery leaves an empty index heap, which is
considered broken. This patch forces to emit WAL when an index_build
turns into empty nbtree index.
---
 src/backend/access/nbtree/nbtsort.c | 11 +++++++++--
 1 file changed, 9 insertions(+), 2 deletions(-)

diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c
index 2762a2d548..70fe3bec32 100644
--- a/src/backend/access/nbtree/nbtsort.c
+++ b/src/backend/access/nbtree/nbtsort.c
@@ -622,8 +622,15 @@ _bt_blwritepage(BTWriteState *wstate, Page page, BlockNumber blkno)
 	/* Ensure rd_smgr is open (could have been closed by relcache flush!) */
 	RelationOpenSmgr(wstate->index);
 
-	/* XLOG stuff */
-	if (wstate->btws_use_wal)
+	/* XLOG stuff
+	 *
+	 * Even if minimal mode, WAL is required here if truncation happened after
+	 * being created in the same transaction. It is not needed otherwise but
+	 * we don't bother identifying the case precisely.
+	 */
+	if (wstate->btws_use_wal ||
+		(RelationNeedsWAL(wstate->index) &&
+		 (blkno == BTREE_METAPAGE && BTPageGetMeta(page)->btm_root == 0)))
 	{
 		/* We use the heap NEWPAGE record type for this */
 		log_newpage(&wstate->index->rd_node, MAIN_FORKNUM, blkno, page, true);
-- 
2.16.3


----Next_Part(Mon_Mar_25_21_32_04_2019_838)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="v8-0003-Move-XLOG-stuff-from-heap_insert-and-heap_delete.patch"



^ permalink  raw  reply  [nested|flat] 15+ messages in thread

* Re: Move global variables of pgoutput to plugin private scope.
@ 2023-09-20 01:42 Peter Smith <[email protected]>
  0 siblings, 0 replies; 15+ messages in thread

From: Peter Smith @ 2023-09-20 01:42 UTC (permalink / raw)
  To: Zhijie Hou (Fujitsu) <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]>

Hi Hou-san.

Given there are some issues raised about the 0001 patch [1] I am
skipping that one until I see the replies.

Meanwhile, here are some review comments for the patches v1-0002 and v1-0003

////////////////////
v1-0002

======
Commit message

1.
The pgoutput module uses a global variable(publish_no_origin) to cache the
action for the origin filter. But we only initialize publish_no_origin when
user specifies the "origin" in the output paramters which means we could refer
to an uninitialized variable if user didn't specify the paramter.

~

1a.

typos
/variable(publish_no_origin)/variable (publish_no_origin)/
/paramters/parameters/
/paramter./paramter./

~

1b.
"...we could refer to an uninitialized variable"

I'm not sure what this means. Previously it was static, so it wouldn't
be "uninitialised"; it would be false. Perhaps there might be a stale
value from a previous pgoutput, but IIUC that's the point made by your
next paragraph ("Besides, we don't...")

~~~

2.
To improve it, the patch stores the map within the private data of the output
plugin so that it will get initialized and reset along with the output plugin
context.

2a.
/To improve it,/To fix this/

~

2b.
"stores the map"

What map? This might be a cut/paste error from the v1-0001 patch comment.

////////////////////
v1-0003

======
Commit message

1.
Missing patch comment.

======
src/backend/replication/pgoutput/pgoutput.c

2. maybe_send_schema

- if (in_streaming)
+ if (data->in_streaming)
  set_schema_sent_in_streamed_txn((PGOutputData *) ctx->output_plugin_private,
  relentry, topxid);
~

Since you added a new 'data' variable, you might as well make use of
it here instead of doing "(PGOutputData *) ctx->output_plugin_private"
again.

======
src/include/replication/pgoutput.h

3.
  MemoryContext cachectx; /* private memory context for cache data */

+ bool in_streaming;
+

Even though there was no comment previously when this was static, IMO
it is better to comment on all the structure fields where possible.

------
[1] https://www.postgresql.org/message-id/ZQk1Ca_eFDTmBiZy%40paquier.xyz

Kind Regards,
Peter Smith.
Fujitsu Australia






^ permalink  raw  reply  [nested|flat] 15+ messages in thread

* [PATCH v17] Add CopyFromRoutine/CopyToRountine
@ 2024-03-04 04:52 Sutou Kouhei <[email protected]>
  0 siblings, 0 replies; 15+ messages in thread

From: Sutou Kouhei @ 2024-03-04 04:52 UTC (permalink / raw)

They are for implementing custom COPY FROM/TO format. But this is not
enough to implement custom COPY FROM/TO format yet. We'll export some
APIs to receive/send data and add "format" option to COPY FROM/TO
later.

Existing text/csv/binary format implementations don't use
CopyFromRoutine/CopyToRoutine for now. We have a patch for it but we
defer it. Because there are some mysterious profile results in spite
of we get faster runtimes. See [1] for details.

[1] https://www.postgresql.org/message-id/ZdbtQJ-p5H1_EDwE%40paquier.xyz

Note that this doesn't change existing text/csv/binary format
implementations.
---
 src/backend/commands/copyfrom.c          |  24 +++++-
 src/backend/commands/copyfromparse.c     |   5 ++
 src/backend/commands/copyto.c            |  25 +++++-
 src/include/commands/copyapi.h           | 100 +++++++++++++++++++++++
 src/include/commands/copyfrom_internal.h |   4 +
 src/tools/pgindent/typedefs.list         |   2 +
 6 files changed, 155 insertions(+), 5 deletions(-)
 create mode 100644 src/include/commands/copyapi.h

diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index c3bc897028..9bf2f6497e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -1623,12 +1623,22 @@ BeginCopyFrom(ParseState *pstate,
 
 		/* Fetch the input function and typioparam info */
 		if (cstate->opts.binary)
+		{
 			getTypeBinaryInputInfo(att->atttypid,
 								   &in_func_oid, &typioparams[attnum - 1]);
+			fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+		}
+		else if (cstate->routine)
+			cstate->routine->CopyFromInFunc(cstate, att->atttypid,
+											&in_functions[attnum - 1],
+											&typioparams[attnum - 1]);
+
 		else
+		{
 			getTypeInputInfo(att->atttypid,
 							 &in_func_oid, &typioparams[attnum - 1]);
-		fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+			fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+		}
 
 		/* Get default info if available */
 		defexprs[attnum - 1] = NULL;
@@ -1768,10 +1778,13 @@ BeginCopyFrom(ParseState *pstate,
 		/* Read and verify binary header */
 		ReceiveCopyBinaryHeader(cstate);
 	}
-
-	/* create workspace for CopyReadAttributes results */
-	if (!cstate->opts.binary)
+	else if (cstate->routine)
 	{
+		cstate->routine->CopyFromStart(cstate, tupDesc);
+	}
+	else
+	{
+		/* create workspace for CopyReadAttributes results */
 		AttrNumber	attr_count = list_length(cstate->attnumlist);
 
 		cstate->max_fields = attr_count;
@@ -1789,6 +1802,9 @@ BeginCopyFrom(ParseState *pstate,
 void
 EndCopyFrom(CopyFromState cstate)
 {
+	if (cstate->routine)
+		cstate->routine->CopyFromEnd(cstate);
+
 	/* No COPY FROM related resources except memory. */
 	if (cstate->is_program)
 	{
diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c
index 7cacd0b752..8b15080585 100644
--- a/src/backend/commands/copyfromparse.c
+++ b/src/backend/commands/copyfromparse.c
@@ -978,6 +978,11 @@ NextCopyFrom(CopyFromState cstate, ExprContext *econtext,
 
 		Assert(fieldno == attr_count);
 	}
+	else if (cstate->routine)
+	{
+		if (!cstate->routine->CopyFromOneRow(cstate, econtext, values, nulls))
+			return false;
+	}
 	else
 	{
 		/* binary */
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index 20ffc90363..b4a7c9c8b9 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -24,6 +24,7 @@
 #include "access/xact.h"
 #include "access/xlog.h"
 #include "commands/copy.h"
+#include "commands/copyapi.h"
 #include "commands/progress.h"
 #include "executor/execdesc.h"
 #include "executor/executor.h"
@@ -71,6 +72,9 @@ typedef enum CopyDest
  */
 typedef struct CopyToStateData
 {
+	/* format routine */
+	const CopyToRoutine *routine;
+
 	/* low-level state data */
 	CopyDest	copy_dest;		/* type of copy source/destination */
 	FILE	   *copy_file;		/* used if copy_dest == COPY_FILE */
@@ -777,14 +781,22 @@ DoCopyTo(CopyToState cstate)
 		Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
 
 		if (cstate->opts.binary)
+		{
 			getTypeBinaryOutputInfo(attr->atttypid,
 									&out_func_oid,
 									&isvarlena);
+			fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+		}
+		else if (cstate->routine)
+			cstate->routine->CopyToOutFunc(cstate, attr->atttypid,
+										   &cstate->out_functions[attnum - 1]);
 		else
+		{
 			getTypeOutputInfo(attr->atttypid,
 							  &out_func_oid,
 							  &isvarlena);
-		fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+			fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+		}
 	}
 
 	/*
@@ -811,6 +823,8 @@ DoCopyTo(CopyToState cstate)
 		tmp = 0;
 		CopySendInt32(cstate, tmp);
 	}
+	else if (cstate->routine)
+		cstate->routine->CopyToStart(cstate, tupDesc);
 	else
 	{
 		/*
@@ -892,6 +906,8 @@ DoCopyTo(CopyToState cstate)
 		/* Need to flush out the trailer */
 		CopySendEndOfRow(cstate);
 	}
+	else if (cstate->routine)
+		cstate->routine->CopyToEnd(cstate);
 
 	MemoryContextDelete(cstate->rowcontext);
 
@@ -916,6 +932,13 @@ CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot)
 	MemoryContextReset(cstate->rowcontext);
 	oldcontext = MemoryContextSwitchTo(cstate->rowcontext);
 
+	if (cstate->routine)
+	{
+		cstate->routine->CopyToOneRow(cstate, slot);
+		MemoryContextSwitchTo(oldcontext);
+		return;
+	}
+
 	if (cstate->opts.binary)
 	{
 		/* Binary per-tuple header */
diff --git a/src/include/commands/copyapi.h b/src/include/commands/copyapi.h
new file mode 100644
index 0000000000..635c4cbff2
--- /dev/null
+++ b/src/include/commands/copyapi.h
@@ -0,0 +1,100 @@
+/*-------------------------------------------------------------------------
+ *
+ * copyapi.h
+ *	  API for COPY TO/FROM handlers
+ *
+ *
+ * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/commands/copyapi.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef COPYAPI_H
+#define COPYAPI_H
+
+#include "executor/tuptable.h"
+#include "nodes/execnodes.h"
+
+/* These are private in commands/copy[from|to].c */
+typedef struct CopyFromStateData *CopyFromState;
+typedef struct CopyToStateData *CopyToState;
+
+/*
+ * API structure for a COPY FROM format implementation.  Note this must be
+ * allocated in a server-lifetime manner, typically as a static const struct.
+ */
+typedef struct CopyFromRoutine
+{
+	/*
+	 * Called when COPY FROM is started to set up the input functions
+	 * associated to the relation's attributes writing to.  `finfo` can be
+	 * optionally filled to provide the catalog information of the input
+	 * function.  `typioparam` can be optionally filled to define the OID of
+	 * the type to pass to the input function.  `atttypid` is the OID of data
+	 * type used by the relation's attribute.
+	 */
+	void		(*CopyFromInFunc) (CopyFromState cstate, Oid atttypid,
+								   FmgrInfo *finfo, Oid *typioparam);
+
+	/*
+	 * Called when COPY FROM is started.
+	 *
+	 * `tupDesc` is the tuple descriptor of the relation where the data needs
+	 * to be copied.  This can be used for any initialization steps required
+	 * by a format.
+	 */
+	void		(*CopyFromStart) (CopyFromState cstate, TupleDesc tupDesc);
+
+	/*
+	 * Copy one row to a set of `values` and `nulls` of size tupDesc->natts.
+	 *
+	 * 'econtext' is used to evaluate default expression for each column that
+	 * is either not read from the file or is using the DEFAULT option of COPY
+	 * FROM.  It is NULL if no default values are used.
+	 *
+	 * Returns false if there are no more tuples to copy.
+	 */
+	bool		(*CopyFromOneRow) (CopyFromState cstate, ExprContext *econtext,
+								   Datum *values, bool *nulls);
+
+	/* Called when COPY FROM has ended. */
+	void		(*CopyFromEnd) (CopyFromState cstate);
+} CopyFromRoutine;
+
+/*
+ * API structure for a COPY TO format implementation.   Note this must be
+ * allocated in a server-lifetime manner, typically as a static const struct.
+ */
+typedef struct CopyToRoutine
+{
+	/*
+	 * Called when COPY TO is started to set up the output functions
+	 * associated to the relation's attributes reading from.  `finfo` can be
+	 * optionally filled.  `atttypid` is the OID of data type used by the
+	 * relation's attribute.
+	 */
+	void		(*CopyToOutFunc) (CopyToState cstate, Oid atttypid,
+								  FmgrInfo *finfo);
+
+	/*
+	 * Called when COPY TO is started.
+	 *
+	 * `tupDesc` is the tuple descriptor of the relation from where the data
+	 * is read.
+	 */
+	void		(*CopyToStart) (CopyToState cstate, TupleDesc tupDesc);
+
+	/*
+	 * Copy one row for COPY TO.
+	 *
+	 * `slot` is the tuple slot where the data is emitted.
+	 */
+	void		(*CopyToOneRow) (CopyToState cstate, TupleTableSlot *slot);
+
+	/* Called when COPY TO has ended */
+	void		(*CopyToEnd) (CopyToState cstate);
+} CopyToRoutine;
+
+#endif							/* COPYAPI_H */
diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h
index cad52fcc78..509b9e92a1 100644
--- a/src/include/commands/copyfrom_internal.h
+++ b/src/include/commands/copyfrom_internal.h
@@ -15,6 +15,7 @@
 #define COPYFROM_INTERNAL_H
 
 #include "commands/copy.h"
+#include "commands/copyapi.h"
 #include "commands/trigger.h"
 #include "nodes/miscnodes.h"
 
@@ -58,6 +59,9 @@ typedef enum CopyInsertMethod
  */
 typedef struct CopyFromStateData
 {
+	/* format routine */
+	const CopyFromRoutine *routine;
+
 	/* low-level state data */
 	CopySource	copy_src;		/* type of copy source */
 	FILE	   *copy_file;		/* used if copy_src == COPY_FILE */
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index ee40a341d3..a5ae161ca5 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -475,6 +475,7 @@ ConvertRowtypeExpr
 CookedConstraint
 CopyDest
 CopyFormatOptions
+CopyFromRoutine
 CopyFromState
 CopyFromStateData
 CopyHeaderChoice
@@ -484,6 +485,7 @@ CopyMultiInsertInfo
 CopyOnErrorChoice
 CopySource
 CopyStmt
+CopyToRoutine
 CopyToState
 CopyToStateData
 Cost
-- 
2.43.0


----Next_Part(Tue_Mar__5_17_18_08_2024_468)----





^ permalink  raw  reply  [nested|flat] 15+ messages in thread

* [PATCH v17] Add CopyFromRoutine/CopyToRountine
@ 2024-03-04 04:52 Sutou Kouhei <[email protected]>
  0 siblings, 0 replies; 15+ messages in thread

From: Sutou Kouhei @ 2024-03-04 04:52 UTC (permalink / raw)

They are for implementing custom COPY FROM/TO format. But this is not
enough to implement custom COPY FROM/TO format yet. We'll export some
APIs to receive/send data and add "format" option to COPY FROM/TO
later.

Existing text/csv/binary format implementations don't use
CopyFromRoutine/CopyToRoutine for now. We have a patch for it but we
defer it. Because there are some mysterious profile results in spite
of we get faster runtimes. See [1] for details.

[1] https://www.postgresql.org/message-id/ZdbtQJ-p5H1_EDwE%40paquier.xyz

Note that this doesn't change existing text/csv/binary format
implementations.
---
 src/backend/commands/copyfrom.c          |  24 +++++-
 src/backend/commands/copyfromparse.c     |   5 ++
 src/backend/commands/copyto.c            |  25 +++++-
 src/include/commands/copyapi.h           | 100 +++++++++++++++++++++++
 src/include/commands/copyfrom_internal.h |   4 +
 src/tools/pgindent/typedefs.list         |   2 +
 6 files changed, 155 insertions(+), 5 deletions(-)
 create mode 100644 src/include/commands/copyapi.h

diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index c3bc897028..9bf2f6497e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -1623,12 +1623,22 @@ BeginCopyFrom(ParseState *pstate,
 
 		/* Fetch the input function and typioparam info */
 		if (cstate->opts.binary)
+		{
 			getTypeBinaryInputInfo(att->atttypid,
 								   &in_func_oid, &typioparams[attnum - 1]);
+			fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+		}
+		else if (cstate->routine)
+			cstate->routine->CopyFromInFunc(cstate, att->atttypid,
+											&in_functions[attnum - 1],
+											&typioparams[attnum - 1]);
+
 		else
+		{
 			getTypeInputInfo(att->atttypid,
 							 &in_func_oid, &typioparams[attnum - 1]);
-		fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+			fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+		}
 
 		/* Get default info if available */
 		defexprs[attnum - 1] = NULL;
@@ -1768,10 +1778,13 @@ BeginCopyFrom(ParseState *pstate,
 		/* Read and verify binary header */
 		ReceiveCopyBinaryHeader(cstate);
 	}
-
-	/* create workspace for CopyReadAttributes results */
-	if (!cstate->opts.binary)
+	else if (cstate->routine)
 	{
+		cstate->routine->CopyFromStart(cstate, tupDesc);
+	}
+	else
+	{
+		/* create workspace for CopyReadAttributes results */
 		AttrNumber	attr_count = list_length(cstate->attnumlist);
 
 		cstate->max_fields = attr_count;
@@ -1789,6 +1802,9 @@ BeginCopyFrom(ParseState *pstate,
 void
 EndCopyFrom(CopyFromState cstate)
 {
+	if (cstate->routine)
+		cstate->routine->CopyFromEnd(cstate);
+
 	/* No COPY FROM related resources except memory. */
 	if (cstate->is_program)
 	{
diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c
index 7cacd0b752..8b15080585 100644
--- a/src/backend/commands/copyfromparse.c
+++ b/src/backend/commands/copyfromparse.c
@@ -978,6 +978,11 @@ NextCopyFrom(CopyFromState cstate, ExprContext *econtext,
 
 		Assert(fieldno == attr_count);
 	}
+	else if (cstate->routine)
+	{
+		if (!cstate->routine->CopyFromOneRow(cstate, econtext, values, nulls))
+			return false;
+	}
 	else
 	{
 		/* binary */
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index 20ffc90363..b4a7c9c8b9 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -24,6 +24,7 @@
 #include "access/xact.h"
 #include "access/xlog.h"
 #include "commands/copy.h"
+#include "commands/copyapi.h"
 #include "commands/progress.h"
 #include "executor/execdesc.h"
 #include "executor/executor.h"
@@ -71,6 +72,9 @@ typedef enum CopyDest
  */
 typedef struct CopyToStateData
 {
+	/* format routine */
+	const CopyToRoutine *routine;
+
 	/* low-level state data */
 	CopyDest	copy_dest;		/* type of copy source/destination */
 	FILE	   *copy_file;		/* used if copy_dest == COPY_FILE */
@@ -777,14 +781,22 @@ DoCopyTo(CopyToState cstate)
 		Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
 
 		if (cstate->opts.binary)
+		{
 			getTypeBinaryOutputInfo(attr->atttypid,
 									&out_func_oid,
 									&isvarlena);
+			fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+		}
+		else if (cstate->routine)
+			cstate->routine->CopyToOutFunc(cstate, attr->atttypid,
+										   &cstate->out_functions[attnum - 1]);
 		else
+		{
 			getTypeOutputInfo(attr->atttypid,
 							  &out_func_oid,
 							  &isvarlena);
-		fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+			fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+		}
 	}
 
 	/*
@@ -811,6 +823,8 @@ DoCopyTo(CopyToState cstate)
 		tmp = 0;
 		CopySendInt32(cstate, tmp);
 	}
+	else if (cstate->routine)
+		cstate->routine->CopyToStart(cstate, tupDesc);
 	else
 	{
 		/*
@@ -892,6 +906,8 @@ DoCopyTo(CopyToState cstate)
 		/* Need to flush out the trailer */
 		CopySendEndOfRow(cstate);
 	}
+	else if (cstate->routine)
+		cstate->routine->CopyToEnd(cstate);
 
 	MemoryContextDelete(cstate->rowcontext);
 
@@ -916,6 +932,13 @@ CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot)
 	MemoryContextReset(cstate->rowcontext);
 	oldcontext = MemoryContextSwitchTo(cstate->rowcontext);
 
+	if (cstate->routine)
+	{
+		cstate->routine->CopyToOneRow(cstate, slot);
+		MemoryContextSwitchTo(oldcontext);
+		return;
+	}
+
 	if (cstate->opts.binary)
 	{
 		/* Binary per-tuple header */
diff --git a/src/include/commands/copyapi.h b/src/include/commands/copyapi.h
new file mode 100644
index 0000000000..635c4cbff2
--- /dev/null
+++ b/src/include/commands/copyapi.h
@@ -0,0 +1,100 @@
+/*-------------------------------------------------------------------------
+ *
+ * copyapi.h
+ *	  API for COPY TO/FROM handlers
+ *
+ *
+ * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/commands/copyapi.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef COPYAPI_H
+#define COPYAPI_H
+
+#include "executor/tuptable.h"
+#include "nodes/execnodes.h"
+
+/* These are private in commands/copy[from|to].c */
+typedef struct CopyFromStateData *CopyFromState;
+typedef struct CopyToStateData *CopyToState;
+
+/*
+ * API structure for a COPY FROM format implementation.  Note this must be
+ * allocated in a server-lifetime manner, typically as a static const struct.
+ */
+typedef struct CopyFromRoutine
+{
+	/*
+	 * Called when COPY FROM is started to set up the input functions
+	 * associated to the relation's attributes writing to.  `finfo` can be
+	 * optionally filled to provide the catalog information of the input
+	 * function.  `typioparam` can be optionally filled to define the OID of
+	 * the type to pass to the input function.  `atttypid` is the OID of data
+	 * type used by the relation's attribute.
+	 */
+	void		(*CopyFromInFunc) (CopyFromState cstate, Oid atttypid,
+								   FmgrInfo *finfo, Oid *typioparam);
+
+	/*
+	 * Called when COPY FROM is started.
+	 *
+	 * `tupDesc` is the tuple descriptor of the relation where the data needs
+	 * to be copied.  This can be used for any initialization steps required
+	 * by a format.
+	 */
+	void		(*CopyFromStart) (CopyFromState cstate, TupleDesc tupDesc);
+
+	/*
+	 * Copy one row to a set of `values` and `nulls` of size tupDesc->natts.
+	 *
+	 * 'econtext' is used to evaluate default expression for each column that
+	 * is either not read from the file or is using the DEFAULT option of COPY
+	 * FROM.  It is NULL if no default values are used.
+	 *
+	 * Returns false if there are no more tuples to copy.
+	 */
+	bool		(*CopyFromOneRow) (CopyFromState cstate, ExprContext *econtext,
+								   Datum *values, bool *nulls);
+
+	/* Called when COPY FROM has ended. */
+	void		(*CopyFromEnd) (CopyFromState cstate);
+} CopyFromRoutine;
+
+/*
+ * API structure for a COPY TO format implementation.   Note this must be
+ * allocated in a server-lifetime manner, typically as a static const struct.
+ */
+typedef struct CopyToRoutine
+{
+	/*
+	 * Called when COPY TO is started to set up the output functions
+	 * associated to the relation's attributes reading from.  `finfo` can be
+	 * optionally filled.  `atttypid` is the OID of data type used by the
+	 * relation's attribute.
+	 */
+	void		(*CopyToOutFunc) (CopyToState cstate, Oid atttypid,
+								  FmgrInfo *finfo);
+
+	/*
+	 * Called when COPY TO is started.
+	 *
+	 * `tupDesc` is the tuple descriptor of the relation from where the data
+	 * is read.
+	 */
+	void		(*CopyToStart) (CopyToState cstate, TupleDesc tupDesc);
+
+	/*
+	 * Copy one row for COPY TO.
+	 *
+	 * `slot` is the tuple slot where the data is emitted.
+	 */
+	void		(*CopyToOneRow) (CopyToState cstate, TupleTableSlot *slot);
+
+	/* Called when COPY TO has ended */
+	void		(*CopyToEnd) (CopyToState cstate);
+} CopyToRoutine;
+
+#endif							/* COPYAPI_H */
diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h
index cad52fcc78..509b9e92a1 100644
--- a/src/include/commands/copyfrom_internal.h
+++ b/src/include/commands/copyfrom_internal.h
@@ -15,6 +15,7 @@
 #define COPYFROM_INTERNAL_H
 
 #include "commands/copy.h"
+#include "commands/copyapi.h"
 #include "commands/trigger.h"
 #include "nodes/miscnodes.h"
 
@@ -58,6 +59,9 @@ typedef enum CopyInsertMethod
  */
 typedef struct CopyFromStateData
 {
+	/* format routine */
+	const CopyFromRoutine *routine;
+
 	/* low-level state data */
 	CopySource	copy_src;		/* type of copy source */
 	FILE	   *copy_file;		/* used if copy_src == COPY_FILE */
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index ee40a341d3..a5ae161ca5 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -475,6 +475,7 @@ ConvertRowtypeExpr
 CookedConstraint
 CopyDest
 CopyFormatOptions
+CopyFromRoutine
 CopyFromState
 CopyFromStateData
 CopyHeaderChoice
@@ -484,6 +485,7 @@ CopyMultiInsertInfo
 CopyOnErrorChoice
 CopySource
 CopyStmt
+CopyToRoutine
 CopyToState
 CopyToStateData
 Cost
-- 
2.43.0


----Next_Part(Tue_Mar__5_17_18_08_2024_468)----





^ permalink  raw  reply  [nested|flat] 15+ messages in thread

* [PATCH v16] Add CopyFromRoutine/CopyToRountine
@ 2024-03-04 04:52 Sutou Kouhei <[email protected]>
  0 siblings, 0 replies; 15+ messages in thread

From: Sutou Kouhei @ 2024-03-04 04:52 UTC (permalink / raw)

They are for implementing custom COPY FROM/TO format. But this is not
enough to implement custom COPY FROM/TO format yet. We'll export some
APIs to receive/send data and add "format" option to COPY FROM/TO
later.

Existing text/csv/binary format implementations don't use
CopyFromRoutine/CopyToRoutine for now. We have a patch for it but we
defer it. Because there are some mysterious profile results in spite
of we get faster runtimes. See [1] for details.

[1] https://www.postgresql.org/message-id/ZdbtQJ-p5H1_EDwE%40paquier.xyz

Note that this doesn't change existing text/csv/binary format
implementations. There are many diffs for CopyOneRowTo() but they're
caused by indentation. They don't change implementations.
---
 src/backend/commands/copyfrom.c          |  24 +++++-
 src/backend/commands/copyfromparse.c     |   5 ++
 src/backend/commands/copyto.c            | 103 ++++++++++++++---------
 src/include/commands/copyapi.h           | 100 ++++++++++++++++++++++
 src/include/commands/copyfrom_internal.h |   4 +
 src/tools/pgindent/typedefs.list         |   2 +
 6 files changed, 193 insertions(+), 45 deletions(-)
 create mode 100644 src/include/commands/copyapi.h

diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index c3bc897028..9bf2f6497e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -1623,12 +1623,22 @@ BeginCopyFrom(ParseState *pstate,
 
 		/* Fetch the input function and typioparam info */
 		if (cstate->opts.binary)
+		{
 			getTypeBinaryInputInfo(att->atttypid,
 								   &in_func_oid, &typioparams[attnum - 1]);
+			fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+		}
+		else if (cstate->routine)
+			cstate->routine->CopyFromInFunc(cstate, att->atttypid,
+											&in_functions[attnum - 1],
+											&typioparams[attnum - 1]);
+
 		else
+		{
 			getTypeInputInfo(att->atttypid,
 							 &in_func_oid, &typioparams[attnum - 1]);
-		fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+			fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+		}
 
 		/* Get default info if available */
 		defexprs[attnum - 1] = NULL;
@@ -1768,10 +1778,13 @@ BeginCopyFrom(ParseState *pstate,
 		/* Read and verify binary header */
 		ReceiveCopyBinaryHeader(cstate);
 	}
-
-	/* create workspace for CopyReadAttributes results */
-	if (!cstate->opts.binary)
+	else if (cstate->routine)
 	{
+		cstate->routine->CopyFromStart(cstate, tupDesc);
+	}
+	else
+	{
+		/* create workspace for CopyReadAttributes results */
 		AttrNumber	attr_count = list_length(cstate->attnumlist);
 
 		cstate->max_fields = attr_count;
@@ -1789,6 +1802,9 @@ BeginCopyFrom(ParseState *pstate,
 void
 EndCopyFrom(CopyFromState cstate)
 {
+	if (cstate->routine)
+		cstate->routine->CopyFromEnd(cstate);
+
 	/* No COPY FROM related resources except memory. */
 	if (cstate->is_program)
 	{
diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c
index 7cacd0b752..8b15080585 100644
--- a/src/backend/commands/copyfromparse.c
+++ b/src/backend/commands/copyfromparse.c
@@ -978,6 +978,11 @@ NextCopyFrom(CopyFromState cstate, ExprContext *econtext,
 
 		Assert(fieldno == attr_count);
 	}
+	else if (cstate->routine)
+	{
+		if (!cstate->routine->CopyFromOneRow(cstate, econtext, values, nulls))
+			return false;
+	}
 	else
 	{
 		/* binary */
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index 20ffc90363..6080627c83 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -24,6 +24,7 @@
 #include "access/xact.h"
 #include "access/xlog.h"
 #include "commands/copy.h"
+#include "commands/copyapi.h"
 #include "commands/progress.h"
 #include "executor/execdesc.h"
 #include "executor/executor.h"
@@ -71,6 +72,9 @@ typedef enum CopyDest
  */
 typedef struct CopyToStateData
 {
+	/* format routine */
+	const CopyToRoutine *routine;
+
 	/* low-level state data */
 	CopyDest	copy_dest;		/* type of copy source/destination */
 	FILE	   *copy_file;		/* used if copy_dest == COPY_FILE */
@@ -777,14 +781,22 @@ DoCopyTo(CopyToState cstate)
 		Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
 
 		if (cstate->opts.binary)
+		{
 			getTypeBinaryOutputInfo(attr->atttypid,
 									&out_func_oid,
 									&isvarlena);
+			fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+		}
+		else if (cstate->routine)
+			cstate->routine->CopyToOutFunc(cstate, attr->atttypid,
+										   &cstate->out_functions[attnum - 1]);
 		else
+		{
 			getTypeOutputInfo(attr->atttypid,
 							  &out_func_oid,
 							  &isvarlena);
-		fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+			fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+		}
 	}
 
 	/*
@@ -811,6 +823,8 @@ DoCopyTo(CopyToState cstate)
 		tmp = 0;
 		CopySendInt32(cstate, tmp);
 	}
+	else if (cstate->routine)
+		cstate->routine->CopyToStart(cstate, tupDesc);
 	else
 	{
 		/*
@@ -892,6 +906,8 @@ DoCopyTo(CopyToState cstate)
 		/* Need to flush out the trailer */
 		CopySendEndOfRow(cstate);
 	}
+	else if (cstate->routine)
+		cstate->routine->CopyToEnd(cstate);
 
 	MemoryContextDelete(cstate->rowcontext);
 
@@ -916,61 +932,66 @@ CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot)
 	MemoryContextReset(cstate->rowcontext);
 	oldcontext = MemoryContextSwitchTo(cstate->rowcontext);
 
-	if (cstate->opts.binary)
+	if (cstate->routine)
+		cstate->routine->CopyToOneRow(cstate, slot);
+	else
 	{
-		/* Binary per-tuple header */
-		CopySendInt16(cstate, list_length(cstate->attnumlist));
-	}
-
-	/* Make sure the tuple is fully deconstructed */
-	slot_getallattrs(slot);
-
-	foreach(cur, cstate->attnumlist)
-	{
-		int			attnum = lfirst_int(cur);
-		Datum		value = slot->tts_values[attnum - 1];
-		bool		isnull = slot->tts_isnull[attnum - 1];
-
-		if (!cstate->opts.binary)
+		if (cstate->opts.binary)
 		{
-			if (need_delim)
-				CopySendChar(cstate, cstate->opts.delim[0]);
-			need_delim = true;
+			/* Binary per-tuple header */
+			CopySendInt16(cstate, list_length(cstate->attnumlist));
 		}
 
-		if (isnull)
-		{
-			if (!cstate->opts.binary)
-				CopySendString(cstate, cstate->opts.null_print_client);
-			else
-				CopySendInt32(cstate, -1);
-		}
-		else
+		/* Make sure the tuple is fully deconstructed */
+		slot_getallattrs(slot);
+
+		foreach(cur, cstate->attnumlist)
 		{
+			int			attnum = lfirst_int(cur);
+			Datum		value = slot->tts_values[attnum - 1];
+			bool		isnull = slot->tts_isnull[attnum - 1];
+
 			if (!cstate->opts.binary)
 			{
-				string = OutputFunctionCall(&out_functions[attnum - 1],
-											value);
-				if (cstate->opts.csv_mode)
-					CopyAttributeOutCSV(cstate, string,
-										cstate->opts.force_quote_flags[attnum - 1]);
+				if (need_delim)
+					CopySendChar(cstate, cstate->opts.delim[0]);
+				need_delim = true;
+			}
+
+			if (isnull)
+			{
+				if (!cstate->opts.binary)
+					CopySendString(cstate, cstate->opts.null_print_client);
 				else
-					CopyAttributeOutText(cstate, string);
+					CopySendInt32(cstate, -1);
 			}
 			else
 			{
-				bytea	   *outputbytes;
+				if (!cstate->opts.binary)
+				{
+					string = OutputFunctionCall(&out_functions[attnum - 1],
+												value);
+					if (cstate->opts.csv_mode)
+						CopyAttributeOutCSV(cstate, string,
+											cstate->opts.force_quote_flags[attnum - 1]);
+					else
+						CopyAttributeOutText(cstate, string);
+				}
+				else
+				{
+					bytea	   *outputbytes;
 
-				outputbytes = SendFunctionCall(&out_functions[attnum - 1],
-											   value);
-				CopySendInt32(cstate, VARSIZE(outputbytes) - VARHDRSZ);
-				CopySendData(cstate, VARDATA(outputbytes),
-							 VARSIZE(outputbytes) - VARHDRSZ);
+					outputbytes = SendFunctionCall(&out_functions[attnum - 1],
+												   value);
+					CopySendInt32(cstate, VARSIZE(outputbytes) - VARHDRSZ);
+					CopySendData(cstate, VARDATA(outputbytes),
+								 VARSIZE(outputbytes) - VARHDRSZ);
+				}
 			}
 		}
-	}
 
-	CopySendEndOfRow(cstate);
+		CopySendEndOfRow(cstate);
+	}
 
 	MemoryContextSwitchTo(oldcontext);
 }
diff --git a/src/include/commands/copyapi.h b/src/include/commands/copyapi.h
new file mode 100644
index 0000000000..635c4cbff2
--- /dev/null
+++ b/src/include/commands/copyapi.h
@@ -0,0 +1,100 @@
+/*-------------------------------------------------------------------------
+ *
+ * copyapi.h
+ *	  API for COPY TO/FROM handlers
+ *
+ *
+ * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/commands/copyapi.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef COPYAPI_H
+#define COPYAPI_H
+
+#include "executor/tuptable.h"
+#include "nodes/execnodes.h"
+
+/* These are private in commands/copy[from|to].c */
+typedef struct CopyFromStateData *CopyFromState;
+typedef struct CopyToStateData *CopyToState;
+
+/*
+ * API structure for a COPY FROM format implementation.  Note this must be
+ * allocated in a server-lifetime manner, typically as a static const struct.
+ */
+typedef struct CopyFromRoutine
+{
+	/*
+	 * Called when COPY FROM is started to set up the input functions
+	 * associated to the relation's attributes writing to.  `finfo` can be
+	 * optionally filled to provide the catalog information of the input
+	 * function.  `typioparam` can be optionally filled to define the OID of
+	 * the type to pass to the input function.  `atttypid` is the OID of data
+	 * type used by the relation's attribute.
+	 */
+	void		(*CopyFromInFunc) (CopyFromState cstate, Oid atttypid,
+								   FmgrInfo *finfo, Oid *typioparam);
+
+	/*
+	 * Called when COPY FROM is started.
+	 *
+	 * `tupDesc` is the tuple descriptor of the relation where the data needs
+	 * to be copied.  This can be used for any initialization steps required
+	 * by a format.
+	 */
+	void		(*CopyFromStart) (CopyFromState cstate, TupleDesc tupDesc);
+
+	/*
+	 * Copy one row to a set of `values` and `nulls` of size tupDesc->natts.
+	 *
+	 * 'econtext' is used to evaluate default expression for each column that
+	 * is either not read from the file or is using the DEFAULT option of COPY
+	 * FROM.  It is NULL if no default values are used.
+	 *
+	 * Returns false if there are no more tuples to copy.
+	 */
+	bool		(*CopyFromOneRow) (CopyFromState cstate, ExprContext *econtext,
+								   Datum *values, bool *nulls);
+
+	/* Called when COPY FROM has ended. */
+	void		(*CopyFromEnd) (CopyFromState cstate);
+} CopyFromRoutine;
+
+/*
+ * API structure for a COPY TO format implementation.   Note this must be
+ * allocated in a server-lifetime manner, typically as a static const struct.
+ */
+typedef struct CopyToRoutine
+{
+	/*
+	 * Called when COPY TO is started to set up the output functions
+	 * associated to the relation's attributes reading from.  `finfo` can be
+	 * optionally filled.  `atttypid` is the OID of data type used by the
+	 * relation's attribute.
+	 */
+	void		(*CopyToOutFunc) (CopyToState cstate, Oid atttypid,
+								  FmgrInfo *finfo);
+
+	/*
+	 * Called when COPY TO is started.
+	 *
+	 * `tupDesc` is the tuple descriptor of the relation from where the data
+	 * is read.
+	 */
+	void		(*CopyToStart) (CopyToState cstate, TupleDesc tupDesc);
+
+	/*
+	 * Copy one row for COPY TO.
+	 *
+	 * `slot` is the tuple slot where the data is emitted.
+	 */
+	void		(*CopyToOneRow) (CopyToState cstate, TupleTableSlot *slot);
+
+	/* Called when COPY TO has ended */
+	void		(*CopyToEnd) (CopyToState cstate);
+} CopyToRoutine;
+
+#endif							/* COPYAPI_H */
diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h
index cad52fcc78..509b9e92a1 100644
--- a/src/include/commands/copyfrom_internal.h
+++ b/src/include/commands/copyfrom_internal.h
@@ -15,6 +15,7 @@
 #define COPYFROM_INTERNAL_H
 
 #include "commands/copy.h"
+#include "commands/copyapi.h"
 #include "commands/trigger.h"
 #include "nodes/miscnodes.h"
 
@@ -58,6 +59,9 @@ typedef enum CopyInsertMethod
  */
 typedef struct CopyFromStateData
 {
+	/* format routine */
+	const CopyFromRoutine *routine;
+
 	/* low-level state data */
 	CopySource	copy_src;		/* type of copy source */
 	FILE	   *copy_file;		/* used if copy_src == COPY_FILE */
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index ee40a341d3..a5ae161ca5 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -475,6 +475,7 @@ ConvertRowtypeExpr
 CookedConstraint
 CopyDest
 CopyFormatOptions
+CopyFromRoutine
 CopyFromState
 CopyFromStateData
 CopyHeaderChoice
@@ -484,6 +485,7 @@ CopyMultiInsertInfo
 CopyOnErrorChoice
 CopySource
 CopyStmt
+CopyToRoutine
 CopyToState
 CopyToStateData
 Cost
-- 
2.43.0


----Next_Part(Mon_Mar__4_14_11_08_2024_631)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="v16-0001-Add-CopyFromRoutine-CopyToRountine-w.patch"



^ permalink  raw  reply  [nested|flat] 15+ messages in thread

* [PATCH v16] Add CopyFromRoutine/CopyToRountine
@ 2024-03-04 04:52 Sutou Kouhei <[email protected]>
  0 siblings, 0 replies; 15+ messages in thread

From: Sutou Kouhei @ 2024-03-04 04:52 UTC (permalink / raw)

They are for implementing custom COPY FROM/TO format. But this is not
enough to implement custom COPY FROM/TO format yet. We'll export some
APIs to receive/send data and add "format" option to COPY FROM/TO
later.

Existing text/csv/binary format implementations don't use
CopyFromRoutine/CopyToRoutine for now. We have a patch for it but we
defer it. Because there are some mysterious profile results in spite
of we get faster runtimes. See [1] for details.

[1] https://www.postgresql.org/message-id/ZdbtQJ-p5H1_EDwE%40paquier.xyz

Note that this doesn't change existing text/csv/binary format
implementations. There are many diffs for CopyOneRowTo() but they're
caused by indentation. They don't change implementations.
---
 src/backend/commands/copyfrom.c          |  22 ++++-
 src/backend/commands/copyfromparse.c     |   5 ++
 src/backend/commands/copyto.c            |  21 +++++
 src/include/commands/copyapi.h           | 100 +++++++++++++++++++++++
 src/include/commands/copyfrom_internal.h |   4 +
 src/tools/pgindent/typedefs.list         |   2 +
 6 files changed, 151 insertions(+), 3 deletions(-)
 create mode 100644 src/include/commands/copyapi.h

diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index c3bc897028..9bf2f6497e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -1623,12 +1623,22 @@ BeginCopyFrom(ParseState *pstate,
 
 		/* Fetch the input function and typioparam info */
 		if (cstate->opts.binary)
+		{
 			getTypeBinaryInputInfo(att->atttypid,
 								   &in_func_oid, &typioparams[attnum - 1]);
+			fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+		}
+		else if (cstate->routine)
+			cstate->routine->CopyFromInFunc(cstate, att->atttypid,
+											&in_functions[attnum - 1],
+											&typioparams[attnum - 1]);
+
 		else
+		{
 			getTypeInputInfo(att->atttypid,
 							 &in_func_oid, &typioparams[attnum - 1]);
 			fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+		}
 
 		/* Get default info if available */
 		defexprs[attnum - 1] = NULL;
@@ -1768,10 +1778,13 @@ BeginCopyFrom(ParseState *pstate,
 		/* Read and verify binary header */
 		ReceiveCopyBinaryHeader(cstate);
 	}
-
-	/* create workspace for CopyReadAttributes results */
-	if (!cstate->opts.binary)
+	else if (cstate->routine)
 	{
+		cstate->routine->CopyFromStart(cstate, tupDesc);
+	}
+	else
+	{
+		/* create workspace for CopyReadAttributes results */
 		AttrNumber	attr_count = list_length(cstate->attnumlist);
 
 		cstate->max_fields = attr_count;
@@ -1789,6 +1802,9 @@ BeginCopyFrom(ParseState *pstate,
 void
 EndCopyFrom(CopyFromState cstate)
 {
+	if (cstate->routine)
+		cstate->routine->CopyFromEnd(cstate);
+
 	/* No COPY FROM related resources except memory. */
 	if (cstate->is_program)
 	{
diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c
index 7cacd0b752..8b15080585 100644
--- a/src/backend/commands/copyfromparse.c
+++ b/src/backend/commands/copyfromparse.c
@@ -978,6 +978,11 @@ NextCopyFrom(CopyFromState cstate, ExprContext *econtext,
 
 		Assert(fieldno == attr_count);
 	}
+	else if (cstate->routine)
+	{
+		if (!cstate->routine->CopyFromOneRow(cstate, econtext, values, nulls))
+			return false;
+	}
 	else
 	{
 		/* binary */
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index 20ffc90363..6080627c83 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -24,6 +24,7 @@
 #include "access/xact.h"
 #include "access/xlog.h"
 #include "commands/copy.h"
+#include "commands/copyapi.h"
 #include "commands/progress.h"
 #include "executor/execdesc.h"
 #include "executor/executor.h"
@@ -71,6 +72,9 @@ typedef enum CopyDest
  */
 typedef struct CopyToStateData
 {
+	/* format routine */
+	const CopyToRoutine *routine;
+
 	/* low-level state data */
 	CopyDest	copy_dest;		/* type of copy source/destination */
 	FILE	   *copy_file;		/* used if copy_dest == COPY_FILE */
@@ -777,15 +781,23 @@ DoCopyTo(CopyToState cstate)
 		Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
 
 		if (cstate->opts.binary)
+		{
 			getTypeBinaryOutputInfo(attr->atttypid,
 									&out_func_oid,
 									&isvarlena);
+			fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+		}
+		else if (cstate->routine)
+			cstate->routine->CopyToOutFunc(cstate, attr->atttypid,
+										   &cstate->out_functions[attnum - 1]);
 		else
+		{
 			getTypeOutputInfo(attr->atttypid,
 							  &out_func_oid,
 							  &isvarlena);
 			fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
 		}
+	}
 
 	/*
 	 * Create a temporary memory context that we can reset once per row to
@@ -811,6 +823,8 @@ DoCopyTo(CopyToState cstate)
 		tmp = 0;
 		CopySendInt32(cstate, tmp);
 	}
+	else if (cstate->routine)
+		cstate->routine->CopyToStart(cstate, tupDesc);
 	else
 	{
 		/*
@@ -892,6 +906,8 @@ DoCopyTo(CopyToState cstate)
 		/* Need to flush out the trailer */
 		CopySendEndOfRow(cstate);
 	}
+	else if (cstate->routine)
+		cstate->routine->CopyToEnd(cstate);
 
 	MemoryContextDelete(cstate->rowcontext);
 
@@ -916,6 +932,10 @@ CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot)
 	MemoryContextReset(cstate->rowcontext);
 	oldcontext = MemoryContextSwitchTo(cstate->rowcontext);
 
+	if (cstate->routine)
+		cstate->routine->CopyToOneRow(cstate, slot);
+	else
+	{
 		if (cstate->opts.binary)
 		{
 			/* Binary per-tuple header */
@@ -971,6 +991,7 @@ CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot)
 		}
 
 		CopySendEndOfRow(cstate);
+	}
 
 	MemoryContextSwitchTo(oldcontext);
 }
diff --git a/src/include/commands/copyapi.h b/src/include/commands/copyapi.h
new file mode 100644
index 0000000000..635c4cbff2
--- /dev/null
+++ b/src/include/commands/copyapi.h
@@ -0,0 +1,100 @@
+/*-------------------------------------------------------------------------
+ *
+ * copyapi.h
+ *	  API for COPY TO/FROM handlers
+ *
+ *
+ * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/commands/copyapi.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef COPYAPI_H
+#define COPYAPI_H
+
+#include "executor/tuptable.h"
+#include "nodes/execnodes.h"
+
+/* These are private in commands/copy[from|to].c */
+typedef struct CopyFromStateData *CopyFromState;
+typedef struct CopyToStateData *CopyToState;
+
+/*
+ * API structure for a COPY FROM format implementation.  Note this must be
+ * allocated in a server-lifetime manner, typically as a static const struct.
+ */
+typedef struct CopyFromRoutine
+{
+	/*
+	 * Called when COPY FROM is started to set up the input functions
+	 * associated to the relation's attributes writing to.  `finfo` can be
+	 * optionally filled to provide the catalog information of the input
+	 * function.  `typioparam` can be optionally filled to define the OID of
+	 * the type to pass to the input function.  `atttypid` is the OID of data
+	 * type used by the relation's attribute.
+	 */
+	void		(*CopyFromInFunc) (CopyFromState cstate, Oid atttypid,
+								   FmgrInfo *finfo, Oid *typioparam);
+
+	/*
+	 * Called when COPY FROM is started.
+	 *
+	 * `tupDesc` is the tuple descriptor of the relation where the data needs
+	 * to be copied.  This can be used for any initialization steps required
+	 * by a format.
+	 */
+	void		(*CopyFromStart) (CopyFromState cstate, TupleDesc tupDesc);
+
+	/*
+	 * Copy one row to a set of `values` and `nulls` of size tupDesc->natts.
+	 *
+	 * 'econtext' is used to evaluate default expression for each column that
+	 * is either not read from the file or is using the DEFAULT option of COPY
+	 * FROM.  It is NULL if no default values are used.
+	 *
+	 * Returns false if there are no more tuples to copy.
+	 */
+	bool		(*CopyFromOneRow) (CopyFromState cstate, ExprContext *econtext,
+								   Datum *values, bool *nulls);
+
+	/* Called when COPY FROM has ended. */
+	void		(*CopyFromEnd) (CopyFromState cstate);
+} CopyFromRoutine;
+
+/*
+ * API structure for a COPY TO format implementation.   Note this must be
+ * allocated in a server-lifetime manner, typically as a static const struct.
+ */
+typedef struct CopyToRoutine
+{
+	/*
+	 * Called when COPY TO is started to set up the output functions
+	 * associated to the relation's attributes reading from.  `finfo` can be
+	 * optionally filled.  `atttypid` is the OID of data type used by the
+	 * relation's attribute.
+	 */
+	void		(*CopyToOutFunc) (CopyToState cstate, Oid atttypid,
+								  FmgrInfo *finfo);
+
+	/*
+	 * Called when COPY TO is started.
+	 *
+	 * `tupDesc` is the tuple descriptor of the relation from where the data
+	 * is read.
+	 */
+	void		(*CopyToStart) (CopyToState cstate, TupleDesc tupDesc);
+
+	/*
+	 * Copy one row for COPY TO.
+	 *
+	 * `slot` is the tuple slot where the data is emitted.
+	 */
+	void		(*CopyToOneRow) (CopyToState cstate, TupleTableSlot *slot);
+
+	/* Called when COPY TO has ended */
+	void		(*CopyToEnd) (CopyToState cstate);
+} CopyToRoutine;
+
+#endif							/* COPYAPI_H */
diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h
index cad52fcc78..509b9e92a1 100644
--- a/src/include/commands/copyfrom_internal.h
+++ b/src/include/commands/copyfrom_internal.h
@@ -15,6 +15,7 @@
 #define COPYFROM_INTERNAL_H
 
 #include "commands/copy.h"
+#include "commands/copyapi.h"
 #include "commands/trigger.h"
 #include "nodes/miscnodes.h"
 
@@ -58,6 +59,9 @@ typedef enum CopyInsertMethod
  */
 typedef struct CopyFromStateData
 {
+	/* format routine */
+	const CopyFromRoutine *routine;
+
 	/* low-level state data */
 	CopySource	copy_src;		/* type of copy source */
 	FILE	   *copy_file;		/* used if copy_src == COPY_FILE */
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index ee40a341d3..a5ae161ca5 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -475,6 +475,7 @@ ConvertRowtypeExpr
 CookedConstraint
 CopyDest
 CopyFormatOptions
+CopyFromRoutine
 CopyFromState
 CopyFromStateData
 CopyHeaderChoice
@@ -484,6 +485,7 @@ CopyMultiInsertInfo
 CopyOnErrorChoice
 CopySource
 CopyStmt
+CopyToRoutine
 CopyToState
 CopyToStateData
 Cost
-- 
2.43.0


----Next_Part(Mon_Mar__4_14_11_08_2024_631)----





^ permalink  raw  reply  [nested|flat] 15+ messages in thread

* [PATCH v16] Add CopyFromRoutine/CopyToRountine
@ 2024-03-04 04:52 Sutou Kouhei <[email protected]>
  0 siblings, 0 replies; 15+ messages in thread

From: Sutou Kouhei @ 2024-03-04 04:52 UTC (permalink / raw)

They are for implementing custom COPY FROM/TO format. But this is not
enough to implement custom COPY FROM/TO format yet. We'll export some
APIs to receive/send data and add "format" option to COPY FROM/TO
later.

Existing text/csv/binary format implementations don't use
CopyFromRoutine/CopyToRoutine for now. We have a patch for it but we
defer it. Because there are some mysterious profile results in spite
of we get faster runtimes. See [1] for details.

[1] https://www.postgresql.org/message-id/ZdbtQJ-p5H1_EDwE%40paquier.xyz

Note that this doesn't change existing text/csv/binary format
implementations. There are many diffs for CopyOneRowTo() but they're
caused by indentation. They don't change implementations.
---
 src/backend/commands/copyfrom.c          |  22 ++++-
 src/backend/commands/copyfromparse.c     |   5 ++
 src/backend/commands/copyto.c            |  21 +++++
 src/include/commands/copyapi.h           | 100 +++++++++++++++++++++++
 src/include/commands/copyfrom_internal.h |   4 +
 src/tools/pgindent/typedefs.list         |   2 +
 6 files changed, 151 insertions(+), 3 deletions(-)
 create mode 100644 src/include/commands/copyapi.h

diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index c3bc897028..9bf2f6497e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -1623,12 +1623,22 @@ BeginCopyFrom(ParseState *pstate,
 
 		/* Fetch the input function and typioparam info */
 		if (cstate->opts.binary)
+		{
 			getTypeBinaryInputInfo(att->atttypid,
 								   &in_func_oid, &typioparams[attnum - 1]);
+			fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+		}
+		else if (cstate->routine)
+			cstate->routine->CopyFromInFunc(cstate, att->atttypid,
+											&in_functions[attnum - 1],
+											&typioparams[attnum - 1]);
+
 		else
+		{
 			getTypeInputInfo(att->atttypid,
 							 &in_func_oid, &typioparams[attnum - 1]);
 			fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+		}
 
 		/* Get default info if available */
 		defexprs[attnum - 1] = NULL;
@@ -1768,10 +1778,13 @@ BeginCopyFrom(ParseState *pstate,
 		/* Read and verify binary header */
 		ReceiveCopyBinaryHeader(cstate);
 	}
-
-	/* create workspace for CopyReadAttributes results */
-	if (!cstate->opts.binary)
+	else if (cstate->routine)
 	{
+		cstate->routine->CopyFromStart(cstate, tupDesc);
+	}
+	else
+	{
+		/* create workspace for CopyReadAttributes results */
 		AttrNumber	attr_count = list_length(cstate->attnumlist);
 
 		cstate->max_fields = attr_count;
@@ -1789,6 +1802,9 @@ BeginCopyFrom(ParseState *pstate,
 void
 EndCopyFrom(CopyFromState cstate)
 {
+	if (cstate->routine)
+		cstate->routine->CopyFromEnd(cstate);
+
 	/* No COPY FROM related resources except memory. */
 	if (cstate->is_program)
 	{
diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c
index 7cacd0b752..8b15080585 100644
--- a/src/backend/commands/copyfromparse.c
+++ b/src/backend/commands/copyfromparse.c
@@ -978,6 +978,11 @@ NextCopyFrom(CopyFromState cstate, ExprContext *econtext,
 
 		Assert(fieldno == attr_count);
 	}
+	else if (cstate->routine)
+	{
+		if (!cstate->routine->CopyFromOneRow(cstate, econtext, values, nulls))
+			return false;
+	}
 	else
 	{
 		/* binary */
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index 20ffc90363..6080627c83 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -24,6 +24,7 @@
 #include "access/xact.h"
 #include "access/xlog.h"
 #include "commands/copy.h"
+#include "commands/copyapi.h"
 #include "commands/progress.h"
 #include "executor/execdesc.h"
 #include "executor/executor.h"
@@ -71,6 +72,9 @@ typedef enum CopyDest
  */
 typedef struct CopyToStateData
 {
+	/* format routine */
+	const CopyToRoutine *routine;
+
 	/* low-level state data */
 	CopyDest	copy_dest;		/* type of copy source/destination */
 	FILE	   *copy_file;		/* used if copy_dest == COPY_FILE */
@@ -777,15 +781,23 @@ DoCopyTo(CopyToState cstate)
 		Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
 
 		if (cstate->opts.binary)
+		{
 			getTypeBinaryOutputInfo(attr->atttypid,
 									&out_func_oid,
 									&isvarlena);
+			fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+		}
+		else if (cstate->routine)
+			cstate->routine->CopyToOutFunc(cstate, attr->atttypid,
+										   &cstate->out_functions[attnum - 1]);
 		else
+		{
 			getTypeOutputInfo(attr->atttypid,
 							  &out_func_oid,
 							  &isvarlena);
 			fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
 		}
+	}
 
 	/*
 	 * Create a temporary memory context that we can reset once per row to
@@ -811,6 +823,8 @@ DoCopyTo(CopyToState cstate)
 		tmp = 0;
 		CopySendInt32(cstate, tmp);
 	}
+	else if (cstate->routine)
+		cstate->routine->CopyToStart(cstate, tupDesc);
 	else
 	{
 		/*
@@ -892,6 +906,8 @@ DoCopyTo(CopyToState cstate)
 		/* Need to flush out the trailer */
 		CopySendEndOfRow(cstate);
 	}
+	else if (cstate->routine)
+		cstate->routine->CopyToEnd(cstate);
 
 	MemoryContextDelete(cstate->rowcontext);
 
@@ -916,6 +932,10 @@ CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot)
 	MemoryContextReset(cstate->rowcontext);
 	oldcontext = MemoryContextSwitchTo(cstate->rowcontext);
 
+	if (cstate->routine)
+		cstate->routine->CopyToOneRow(cstate, slot);
+	else
+	{
 		if (cstate->opts.binary)
 		{
 			/* Binary per-tuple header */
@@ -971,6 +991,7 @@ CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot)
 		}
 
 		CopySendEndOfRow(cstate);
+	}
 
 	MemoryContextSwitchTo(oldcontext);
 }
diff --git a/src/include/commands/copyapi.h b/src/include/commands/copyapi.h
new file mode 100644
index 0000000000..635c4cbff2
--- /dev/null
+++ b/src/include/commands/copyapi.h
@@ -0,0 +1,100 @@
+/*-------------------------------------------------------------------------
+ *
+ * copyapi.h
+ *	  API for COPY TO/FROM handlers
+ *
+ *
+ * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/commands/copyapi.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef COPYAPI_H
+#define COPYAPI_H
+
+#include "executor/tuptable.h"
+#include "nodes/execnodes.h"
+
+/* These are private in commands/copy[from|to].c */
+typedef struct CopyFromStateData *CopyFromState;
+typedef struct CopyToStateData *CopyToState;
+
+/*
+ * API structure for a COPY FROM format implementation.  Note this must be
+ * allocated in a server-lifetime manner, typically as a static const struct.
+ */
+typedef struct CopyFromRoutine
+{
+	/*
+	 * Called when COPY FROM is started to set up the input functions
+	 * associated to the relation's attributes writing to.  `finfo` can be
+	 * optionally filled to provide the catalog information of the input
+	 * function.  `typioparam` can be optionally filled to define the OID of
+	 * the type to pass to the input function.  `atttypid` is the OID of data
+	 * type used by the relation's attribute.
+	 */
+	void		(*CopyFromInFunc) (CopyFromState cstate, Oid atttypid,
+								   FmgrInfo *finfo, Oid *typioparam);
+
+	/*
+	 * Called when COPY FROM is started.
+	 *
+	 * `tupDesc` is the tuple descriptor of the relation where the data needs
+	 * to be copied.  This can be used for any initialization steps required
+	 * by a format.
+	 */
+	void		(*CopyFromStart) (CopyFromState cstate, TupleDesc tupDesc);
+
+	/*
+	 * Copy one row to a set of `values` and `nulls` of size tupDesc->natts.
+	 *
+	 * 'econtext' is used to evaluate default expression for each column that
+	 * is either not read from the file or is using the DEFAULT option of COPY
+	 * FROM.  It is NULL if no default values are used.
+	 *
+	 * Returns false if there are no more tuples to copy.
+	 */
+	bool		(*CopyFromOneRow) (CopyFromState cstate, ExprContext *econtext,
+								   Datum *values, bool *nulls);
+
+	/* Called when COPY FROM has ended. */
+	void		(*CopyFromEnd) (CopyFromState cstate);
+} CopyFromRoutine;
+
+/*
+ * API structure for a COPY TO format implementation.   Note this must be
+ * allocated in a server-lifetime manner, typically as a static const struct.
+ */
+typedef struct CopyToRoutine
+{
+	/*
+	 * Called when COPY TO is started to set up the output functions
+	 * associated to the relation's attributes reading from.  `finfo` can be
+	 * optionally filled.  `atttypid` is the OID of data type used by the
+	 * relation's attribute.
+	 */
+	void		(*CopyToOutFunc) (CopyToState cstate, Oid atttypid,
+								  FmgrInfo *finfo);
+
+	/*
+	 * Called when COPY TO is started.
+	 *
+	 * `tupDesc` is the tuple descriptor of the relation from where the data
+	 * is read.
+	 */
+	void		(*CopyToStart) (CopyToState cstate, TupleDesc tupDesc);
+
+	/*
+	 * Copy one row for COPY TO.
+	 *
+	 * `slot` is the tuple slot where the data is emitted.
+	 */
+	void		(*CopyToOneRow) (CopyToState cstate, TupleTableSlot *slot);
+
+	/* Called when COPY TO has ended */
+	void		(*CopyToEnd) (CopyToState cstate);
+} CopyToRoutine;
+
+#endif							/* COPYAPI_H */
diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h
index cad52fcc78..509b9e92a1 100644
--- a/src/include/commands/copyfrom_internal.h
+++ b/src/include/commands/copyfrom_internal.h
@@ -15,6 +15,7 @@
 #define COPYFROM_INTERNAL_H
 
 #include "commands/copy.h"
+#include "commands/copyapi.h"
 #include "commands/trigger.h"
 #include "nodes/miscnodes.h"
 
@@ -58,6 +59,9 @@ typedef enum CopyInsertMethod
  */
 typedef struct CopyFromStateData
 {
+	/* format routine */
+	const CopyFromRoutine *routine;
+
 	/* low-level state data */
 	CopySource	copy_src;		/* type of copy source */
 	FILE	   *copy_file;		/* used if copy_src == COPY_FILE */
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index ee40a341d3..a5ae161ca5 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -475,6 +475,7 @@ ConvertRowtypeExpr
 CookedConstraint
 CopyDest
 CopyFormatOptions
+CopyFromRoutine
 CopyFromState
 CopyFromStateData
 CopyHeaderChoice
@@ -484,6 +485,7 @@ CopyMultiInsertInfo
 CopyOnErrorChoice
 CopySource
 CopyStmt
+CopyToRoutine
 CopyToState
 CopyToStateData
 Cost
-- 
2.43.0


----Next_Part(Mon_Mar__4_14_11_08_2024_631)----





^ permalink  raw  reply  [nested|flat] 15+ messages in thread

* [PATCH v16] Add CopyFromRoutine/CopyToRountine
@ 2024-03-04 04:52 Sutou Kouhei <[email protected]>
  0 siblings, 0 replies; 15+ messages in thread

From: Sutou Kouhei @ 2024-03-04 04:52 UTC (permalink / raw)

They are for implementing custom COPY FROM/TO format. But this is not
enough to implement custom COPY FROM/TO format yet. We'll export some
APIs to receive/send data and add "format" option to COPY FROM/TO
later.

Existing text/csv/binary format implementations don't use
CopyFromRoutine/CopyToRoutine for now. We have a patch for it but we
defer it. Because there are some mysterious profile results in spite
of we get faster runtimes. See [1] for details.

[1] https://www.postgresql.org/message-id/ZdbtQJ-p5H1_EDwE%40paquier.xyz

Note that this doesn't change existing text/csv/binary format
implementations. There are many diffs for CopyOneRowTo() but they're
caused by indentation. They don't change implementations.
---
 src/backend/commands/copyfrom.c          |  24 +++++-
 src/backend/commands/copyfromparse.c     |   5 ++
 src/backend/commands/copyto.c            | 103 ++++++++++++++---------
 src/include/commands/copyapi.h           | 100 ++++++++++++++++++++++
 src/include/commands/copyfrom_internal.h |   4 +
 src/tools/pgindent/typedefs.list         |   2 +
 6 files changed, 193 insertions(+), 45 deletions(-)
 create mode 100644 src/include/commands/copyapi.h

diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index c3bc897028..9bf2f6497e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -1623,12 +1623,22 @@ BeginCopyFrom(ParseState *pstate,
 
 		/* Fetch the input function and typioparam info */
 		if (cstate->opts.binary)
+		{
 			getTypeBinaryInputInfo(att->atttypid,
 								   &in_func_oid, &typioparams[attnum - 1]);
+			fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+		}
+		else if (cstate->routine)
+			cstate->routine->CopyFromInFunc(cstate, att->atttypid,
+											&in_functions[attnum - 1],
+											&typioparams[attnum - 1]);
+
 		else
+		{
 			getTypeInputInfo(att->atttypid,
 							 &in_func_oid, &typioparams[attnum - 1]);
-		fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+			fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+		}
 
 		/* Get default info if available */
 		defexprs[attnum - 1] = NULL;
@@ -1768,10 +1778,13 @@ BeginCopyFrom(ParseState *pstate,
 		/* Read and verify binary header */
 		ReceiveCopyBinaryHeader(cstate);
 	}
-
-	/* create workspace for CopyReadAttributes results */
-	if (!cstate->opts.binary)
+	else if (cstate->routine)
 	{
+		cstate->routine->CopyFromStart(cstate, tupDesc);
+	}
+	else
+	{
+		/* create workspace for CopyReadAttributes results */
 		AttrNumber	attr_count = list_length(cstate->attnumlist);
 
 		cstate->max_fields = attr_count;
@@ -1789,6 +1802,9 @@ BeginCopyFrom(ParseState *pstate,
 void
 EndCopyFrom(CopyFromState cstate)
 {
+	if (cstate->routine)
+		cstate->routine->CopyFromEnd(cstate);
+
 	/* No COPY FROM related resources except memory. */
 	if (cstate->is_program)
 	{
diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c
index 7cacd0b752..8b15080585 100644
--- a/src/backend/commands/copyfromparse.c
+++ b/src/backend/commands/copyfromparse.c
@@ -978,6 +978,11 @@ NextCopyFrom(CopyFromState cstate, ExprContext *econtext,
 
 		Assert(fieldno == attr_count);
 	}
+	else if (cstate->routine)
+	{
+		if (!cstate->routine->CopyFromOneRow(cstate, econtext, values, nulls))
+			return false;
+	}
 	else
 	{
 		/* binary */
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index 20ffc90363..6080627c83 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -24,6 +24,7 @@
 #include "access/xact.h"
 #include "access/xlog.h"
 #include "commands/copy.h"
+#include "commands/copyapi.h"
 #include "commands/progress.h"
 #include "executor/execdesc.h"
 #include "executor/executor.h"
@@ -71,6 +72,9 @@ typedef enum CopyDest
  */
 typedef struct CopyToStateData
 {
+	/* format routine */
+	const CopyToRoutine *routine;
+
 	/* low-level state data */
 	CopyDest	copy_dest;		/* type of copy source/destination */
 	FILE	   *copy_file;		/* used if copy_dest == COPY_FILE */
@@ -777,14 +781,22 @@ DoCopyTo(CopyToState cstate)
 		Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
 
 		if (cstate->opts.binary)
+		{
 			getTypeBinaryOutputInfo(attr->atttypid,
 									&out_func_oid,
 									&isvarlena);
+			fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+		}
+		else if (cstate->routine)
+			cstate->routine->CopyToOutFunc(cstate, attr->atttypid,
+										   &cstate->out_functions[attnum - 1]);
 		else
+		{
 			getTypeOutputInfo(attr->atttypid,
 							  &out_func_oid,
 							  &isvarlena);
-		fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+			fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+		}
 	}
 
 	/*
@@ -811,6 +823,8 @@ DoCopyTo(CopyToState cstate)
 		tmp = 0;
 		CopySendInt32(cstate, tmp);
 	}
+	else if (cstate->routine)
+		cstate->routine->CopyToStart(cstate, tupDesc);
 	else
 	{
 		/*
@@ -892,6 +906,8 @@ DoCopyTo(CopyToState cstate)
 		/* Need to flush out the trailer */
 		CopySendEndOfRow(cstate);
 	}
+	else if (cstate->routine)
+		cstate->routine->CopyToEnd(cstate);
 
 	MemoryContextDelete(cstate->rowcontext);
 
@@ -916,61 +932,66 @@ CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot)
 	MemoryContextReset(cstate->rowcontext);
 	oldcontext = MemoryContextSwitchTo(cstate->rowcontext);
 
-	if (cstate->opts.binary)
+	if (cstate->routine)
+		cstate->routine->CopyToOneRow(cstate, slot);
+	else
 	{
-		/* Binary per-tuple header */
-		CopySendInt16(cstate, list_length(cstate->attnumlist));
-	}
-
-	/* Make sure the tuple is fully deconstructed */
-	slot_getallattrs(slot);
-
-	foreach(cur, cstate->attnumlist)
-	{
-		int			attnum = lfirst_int(cur);
-		Datum		value = slot->tts_values[attnum - 1];
-		bool		isnull = slot->tts_isnull[attnum - 1];
-
-		if (!cstate->opts.binary)
+		if (cstate->opts.binary)
 		{
-			if (need_delim)
-				CopySendChar(cstate, cstate->opts.delim[0]);
-			need_delim = true;
+			/* Binary per-tuple header */
+			CopySendInt16(cstate, list_length(cstate->attnumlist));
 		}
 
-		if (isnull)
-		{
-			if (!cstate->opts.binary)
-				CopySendString(cstate, cstate->opts.null_print_client);
-			else
-				CopySendInt32(cstate, -1);
-		}
-		else
+		/* Make sure the tuple is fully deconstructed */
+		slot_getallattrs(slot);
+
+		foreach(cur, cstate->attnumlist)
 		{
+			int			attnum = lfirst_int(cur);
+			Datum		value = slot->tts_values[attnum - 1];
+			bool		isnull = slot->tts_isnull[attnum - 1];
+
 			if (!cstate->opts.binary)
 			{
-				string = OutputFunctionCall(&out_functions[attnum - 1],
-											value);
-				if (cstate->opts.csv_mode)
-					CopyAttributeOutCSV(cstate, string,
-										cstate->opts.force_quote_flags[attnum - 1]);
+				if (need_delim)
+					CopySendChar(cstate, cstate->opts.delim[0]);
+				need_delim = true;
+			}
+
+			if (isnull)
+			{
+				if (!cstate->opts.binary)
+					CopySendString(cstate, cstate->opts.null_print_client);
 				else
-					CopyAttributeOutText(cstate, string);
+					CopySendInt32(cstate, -1);
 			}
 			else
 			{
-				bytea	   *outputbytes;
+				if (!cstate->opts.binary)
+				{
+					string = OutputFunctionCall(&out_functions[attnum - 1],
+												value);
+					if (cstate->opts.csv_mode)
+						CopyAttributeOutCSV(cstate, string,
+											cstate->opts.force_quote_flags[attnum - 1]);
+					else
+						CopyAttributeOutText(cstate, string);
+				}
+				else
+				{
+					bytea	   *outputbytes;
 
-				outputbytes = SendFunctionCall(&out_functions[attnum - 1],
-											   value);
-				CopySendInt32(cstate, VARSIZE(outputbytes) - VARHDRSZ);
-				CopySendData(cstate, VARDATA(outputbytes),
-							 VARSIZE(outputbytes) - VARHDRSZ);
+					outputbytes = SendFunctionCall(&out_functions[attnum - 1],
+												   value);
+					CopySendInt32(cstate, VARSIZE(outputbytes) - VARHDRSZ);
+					CopySendData(cstate, VARDATA(outputbytes),
+								 VARSIZE(outputbytes) - VARHDRSZ);
+				}
 			}
 		}
-	}
 
-	CopySendEndOfRow(cstate);
+		CopySendEndOfRow(cstate);
+	}
 
 	MemoryContextSwitchTo(oldcontext);
 }
diff --git a/src/include/commands/copyapi.h b/src/include/commands/copyapi.h
new file mode 100644
index 0000000000..635c4cbff2
--- /dev/null
+++ b/src/include/commands/copyapi.h
@@ -0,0 +1,100 @@
+/*-------------------------------------------------------------------------
+ *
+ * copyapi.h
+ *	  API for COPY TO/FROM handlers
+ *
+ *
+ * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/commands/copyapi.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef COPYAPI_H
+#define COPYAPI_H
+
+#include "executor/tuptable.h"
+#include "nodes/execnodes.h"
+
+/* These are private in commands/copy[from|to].c */
+typedef struct CopyFromStateData *CopyFromState;
+typedef struct CopyToStateData *CopyToState;
+
+/*
+ * API structure for a COPY FROM format implementation.  Note this must be
+ * allocated in a server-lifetime manner, typically as a static const struct.
+ */
+typedef struct CopyFromRoutine
+{
+	/*
+	 * Called when COPY FROM is started to set up the input functions
+	 * associated to the relation's attributes writing to.  `finfo` can be
+	 * optionally filled to provide the catalog information of the input
+	 * function.  `typioparam` can be optionally filled to define the OID of
+	 * the type to pass to the input function.  `atttypid` is the OID of data
+	 * type used by the relation's attribute.
+	 */
+	void		(*CopyFromInFunc) (CopyFromState cstate, Oid atttypid,
+								   FmgrInfo *finfo, Oid *typioparam);
+
+	/*
+	 * Called when COPY FROM is started.
+	 *
+	 * `tupDesc` is the tuple descriptor of the relation where the data needs
+	 * to be copied.  This can be used for any initialization steps required
+	 * by a format.
+	 */
+	void		(*CopyFromStart) (CopyFromState cstate, TupleDesc tupDesc);
+
+	/*
+	 * Copy one row to a set of `values` and `nulls` of size tupDesc->natts.
+	 *
+	 * 'econtext' is used to evaluate default expression for each column that
+	 * is either not read from the file or is using the DEFAULT option of COPY
+	 * FROM.  It is NULL if no default values are used.
+	 *
+	 * Returns false if there are no more tuples to copy.
+	 */
+	bool		(*CopyFromOneRow) (CopyFromState cstate, ExprContext *econtext,
+								   Datum *values, bool *nulls);
+
+	/* Called when COPY FROM has ended. */
+	void		(*CopyFromEnd) (CopyFromState cstate);
+} CopyFromRoutine;
+
+/*
+ * API structure for a COPY TO format implementation.   Note this must be
+ * allocated in a server-lifetime manner, typically as a static const struct.
+ */
+typedef struct CopyToRoutine
+{
+	/*
+	 * Called when COPY TO is started to set up the output functions
+	 * associated to the relation's attributes reading from.  `finfo` can be
+	 * optionally filled.  `atttypid` is the OID of data type used by the
+	 * relation's attribute.
+	 */
+	void		(*CopyToOutFunc) (CopyToState cstate, Oid atttypid,
+								  FmgrInfo *finfo);
+
+	/*
+	 * Called when COPY TO is started.
+	 *
+	 * `tupDesc` is the tuple descriptor of the relation from where the data
+	 * is read.
+	 */
+	void		(*CopyToStart) (CopyToState cstate, TupleDesc tupDesc);
+
+	/*
+	 * Copy one row for COPY TO.
+	 *
+	 * `slot` is the tuple slot where the data is emitted.
+	 */
+	void		(*CopyToOneRow) (CopyToState cstate, TupleTableSlot *slot);
+
+	/* Called when COPY TO has ended */
+	void		(*CopyToEnd) (CopyToState cstate);
+} CopyToRoutine;
+
+#endif							/* COPYAPI_H */
diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h
index cad52fcc78..509b9e92a1 100644
--- a/src/include/commands/copyfrom_internal.h
+++ b/src/include/commands/copyfrom_internal.h
@@ -15,6 +15,7 @@
 #define COPYFROM_INTERNAL_H
 
 #include "commands/copy.h"
+#include "commands/copyapi.h"
 #include "commands/trigger.h"
 #include "nodes/miscnodes.h"
 
@@ -58,6 +59,9 @@ typedef enum CopyInsertMethod
  */
 typedef struct CopyFromStateData
 {
+	/* format routine */
+	const CopyFromRoutine *routine;
+
 	/* low-level state data */
 	CopySource	copy_src;		/* type of copy source */
 	FILE	   *copy_file;		/* used if copy_src == COPY_FILE */
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index ee40a341d3..a5ae161ca5 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -475,6 +475,7 @@ ConvertRowtypeExpr
 CookedConstraint
 CopyDest
 CopyFormatOptions
+CopyFromRoutine
 CopyFromState
 CopyFromStateData
 CopyHeaderChoice
@@ -484,6 +485,7 @@ CopyMultiInsertInfo
 CopyOnErrorChoice
 CopySource
 CopyStmt
+CopyToRoutine
 CopyToState
 CopyToStateData
 Cost
-- 
2.43.0


----Next_Part(Mon_Mar__4_14_11_08_2024_631)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="v16-0001-Add-CopyFromRoutine-CopyToRountine-w.patch"



^ permalink  raw  reply  [nested|flat] 15+ messages in thread

* [PATCH v16] Add CopyFromRoutine/CopyToRountine
@ 2024-03-04 04:52 Sutou Kouhei <[email protected]>
  0 siblings, 0 replies; 15+ messages in thread

From: Sutou Kouhei @ 2024-03-04 04:52 UTC (permalink / raw)

They are for implementing custom COPY FROM/TO format. But this is not
enough to implement custom COPY FROM/TO format yet. We'll export some
APIs to receive/send data and add "format" option to COPY FROM/TO
later.

Existing text/csv/binary format implementations don't use
CopyFromRoutine/CopyToRoutine for now. We have a patch for it but we
defer it. Because there are some mysterious profile results in spite
of we get faster runtimes. See [1] for details.

[1] https://www.postgresql.org/message-id/ZdbtQJ-p5H1_EDwE%40paquier.xyz

Note that this doesn't change existing text/csv/binary format
implementations. There are many diffs for CopyOneRowTo() but they're
caused by indentation. They don't change implementations.
---
 src/backend/commands/copyfrom.c          |  24 +++++-
 src/backend/commands/copyfromparse.c     |   5 ++
 src/backend/commands/copyto.c            | 103 ++++++++++++++---------
 src/include/commands/copyapi.h           | 100 ++++++++++++++++++++++
 src/include/commands/copyfrom_internal.h |   4 +
 src/tools/pgindent/typedefs.list         |   2 +
 6 files changed, 193 insertions(+), 45 deletions(-)
 create mode 100644 src/include/commands/copyapi.h

diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index c3bc897028..9bf2f6497e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -1623,12 +1623,22 @@ BeginCopyFrom(ParseState *pstate,
 
 		/* Fetch the input function and typioparam info */
 		if (cstate->opts.binary)
+		{
 			getTypeBinaryInputInfo(att->atttypid,
 								   &in_func_oid, &typioparams[attnum - 1]);
+			fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+		}
+		else if (cstate->routine)
+			cstate->routine->CopyFromInFunc(cstate, att->atttypid,
+											&in_functions[attnum - 1],
+											&typioparams[attnum - 1]);
+
 		else
+		{
 			getTypeInputInfo(att->atttypid,
 							 &in_func_oid, &typioparams[attnum - 1]);
-		fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+			fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+		}
 
 		/* Get default info if available */
 		defexprs[attnum - 1] = NULL;
@@ -1768,10 +1778,13 @@ BeginCopyFrom(ParseState *pstate,
 		/* Read and verify binary header */
 		ReceiveCopyBinaryHeader(cstate);
 	}
-
-	/* create workspace for CopyReadAttributes results */
-	if (!cstate->opts.binary)
+	else if (cstate->routine)
 	{
+		cstate->routine->CopyFromStart(cstate, tupDesc);
+	}
+	else
+	{
+		/* create workspace for CopyReadAttributes results */
 		AttrNumber	attr_count = list_length(cstate->attnumlist);
 
 		cstate->max_fields = attr_count;
@@ -1789,6 +1802,9 @@ BeginCopyFrom(ParseState *pstate,
 void
 EndCopyFrom(CopyFromState cstate)
 {
+	if (cstate->routine)
+		cstate->routine->CopyFromEnd(cstate);
+
 	/* No COPY FROM related resources except memory. */
 	if (cstate->is_program)
 	{
diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c
index 7cacd0b752..8b15080585 100644
--- a/src/backend/commands/copyfromparse.c
+++ b/src/backend/commands/copyfromparse.c
@@ -978,6 +978,11 @@ NextCopyFrom(CopyFromState cstate, ExprContext *econtext,
 
 		Assert(fieldno == attr_count);
 	}
+	else if (cstate->routine)
+	{
+		if (!cstate->routine->CopyFromOneRow(cstate, econtext, values, nulls))
+			return false;
+	}
 	else
 	{
 		/* binary */
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index 20ffc90363..6080627c83 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -24,6 +24,7 @@
 #include "access/xact.h"
 #include "access/xlog.h"
 #include "commands/copy.h"
+#include "commands/copyapi.h"
 #include "commands/progress.h"
 #include "executor/execdesc.h"
 #include "executor/executor.h"
@@ -71,6 +72,9 @@ typedef enum CopyDest
  */
 typedef struct CopyToStateData
 {
+	/* format routine */
+	const CopyToRoutine *routine;
+
 	/* low-level state data */
 	CopyDest	copy_dest;		/* type of copy source/destination */
 	FILE	   *copy_file;		/* used if copy_dest == COPY_FILE */
@@ -777,14 +781,22 @@ DoCopyTo(CopyToState cstate)
 		Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
 
 		if (cstate->opts.binary)
+		{
 			getTypeBinaryOutputInfo(attr->atttypid,
 									&out_func_oid,
 									&isvarlena);
+			fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+		}
+		else if (cstate->routine)
+			cstate->routine->CopyToOutFunc(cstate, attr->atttypid,
+										   &cstate->out_functions[attnum - 1]);
 		else
+		{
 			getTypeOutputInfo(attr->atttypid,
 							  &out_func_oid,
 							  &isvarlena);
-		fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+			fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+		}
 	}
 
 	/*
@@ -811,6 +823,8 @@ DoCopyTo(CopyToState cstate)
 		tmp = 0;
 		CopySendInt32(cstate, tmp);
 	}
+	else if (cstate->routine)
+		cstate->routine->CopyToStart(cstate, tupDesc);
 	else
 	{
 		/*
@@ -892,6 +906,8 @@ DoCopyTo(CopyToState cstate)
 		/* Need to flush out the trailer */
 		CopySendEndOfRow(cstate);
 	}
+	else if (cstate->routine)
+		cstate->routine->CopyToEnd(cstate);
 
 	MemoryContextDelete(cstate->rowcontext);
 
@@ -916,61 +932,66 @@ CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot)
 	MemoryContextReset(cstate->rowcontext);
 	oldcontext = MemoryContextSwitchTo(cstate->rowcontext);
 
-	if (cstate->opts.binary)
+	if (cstate->routine)
+		cstate->routine->CopyToOneRow(cstate, slot);
+	else
 	{
-		/* Binary per-tuple header */
-		CopySendInt16(cstate, list_length(cstate->attnumlist));
-	}
-
-	/* Make sure the tuple is fully deconstructed */
-	slot_getallattrs(slot);
-
-	foreach(cur, cstate->attnumlist)
-	{
-		int			attnum = lfirst_int(cur);
-		Datum		value = slot->tts_values[attnum - 1];
-		bool		isnull = slot->tts_isnull[attnum - 1];
-
-		if (!cstate->opts.binary)
+		if (cstate->opts.binary)
 		{
-			if (need_delim)
-				CopySendChar(cstate, cstate->opts.delim[0]);
-			need_delim = true;
+			/* Binary per-tuple header */
+			CopySendInt16(cstate, list_length(cstate->attnumlist));
 		}
 
-		if (isnull)
-		{
-			if (!cstate->opts.binary)
-				CopySendString(cstate, cstate->opts.null_print_client);
-			else
-				CopySendInt32(cstate, -1);
-		}
-		else
+		/* Make sure the tuple is fully deconstructed */
+		slot_getallattrs(slot);
+
+		foreach(cur, cstate->attnumlist)
 		{
+			int			attnum = lfirst_int(cur);
+			Datum		value = slot->tts_values[attnum - 1];
+			bool		isnull = slot->tts_isnull[attnum - 1];
+
 			if (!cstate->opts.binary)
 			{
-				string = OutputFunctionCall(&out_functions[attnum - 1],
-											value);
-				if (cstate->opts.csv_mode)
-					CopyAttributeOutCSV(cstate, string,
-										cstate->opts.force_quote_flags[attnum - 1]);
+				if (need_delim)
+					CopySendChar(cstate, cstate->opts.delim[0]);
+				need_delim = true;
+			}
+
+			if (isnull)
+			{
+				if (!cstate->opts.binary)
+					CopySendString(cstate, cstate->opts.null_print_client);
 				else
-					CopyAttributeOutText(cstate, string);
+					CopySendInt32(cstate, -1);
 			}
 			else
 			{
-				bytea	   *outputbytes;
+				if (!cstate->opts.binary)
+				{
+					string = OutputFunctionCall(&out_functions[attnum - 1],
+												value);
+					if (cstate->opts.csv_mode)
+						CopyAttributeOutCSV(cstate, string,
+											cstate->opts.force_quote_flags[attnum - 1]);
+					else
+						CopyAttributeOutText(cstate, string);
+				}
+				else
+				{
+					bytea	   *outputbytes;
 
-				outputbytes = SendFunctionCall(&out_functions[attnum - 1],
-											   value);
-				CopySendInt32(cstate, VARSIZE(outputbytes) - VARHDRSZ);
-				CopySendData(cstate, VARDATA(outputbytes),
-							 VARSIZE(outputbytes) - VARHDRSZ);
+					outputbytes = SendFunctionCall(&out_functions[attnum - 1],
+												   value);
+					CopySendInt32(cstate, VARSIZE(outputbytes) - VARHDRSZ);
+					CopySendData(cstate, VARDATA(outputbytes),
+								 VARSIZE(outputbytes) - VARHDRSZ);
+				}
 			}
 		}
-	}
 
-	CopySendEndOfRow(cstate);
+		CopySendEndOfRow(cstate);
+	}
 
 	MemoryContextSwitchTo(oldcontext);
 }
diff --git a/src/include/commands/copyapi.h b/src/include/commands/copyapi.h
new file mode 100644
index 0000000000..635c4cbff2
--- /dev/null
+++ b/src/include/commands/copyapi.h
@@ -0,0 +1,100 @@
+/*-------------------------------------------------------------------------
+ *
+ * copyapi.h
+ *	  API for COPY TO/FROM handlers
+ *
+ *
+ * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/commands/copyapi.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef COPYAPI_H
+#define COPYAPI_H
+
+#include "executor/tuptable.h"
+#include "nodes/execnodes.h"
+
+/* These are private in commands/copy[from|to].c */
+typedef struct CopyFromStateData *CopyFromState;
+typedef struct CopyToStateData *CopyToState;
+
+/*
+ * API structure for a COPY FROM format implementation.  Note this must be
+ * allocated in a server-lifetime manner, typically as a static const struct.
+ */
+typedef struct CopyFromRoutine
+{
+	/*
+	 * Called when COPY FROM is started to set up the input functions
+	 * associated to the relation's attributes writing to.  `finfo` can be
+	 * optionally filled to provide the catalog information of the input
+	 * function.  `typioparam` can be optionally filled to define the OID of
+	 * the type to pass to the input function.  `atttypid` is the OID of data
+	 * type used by the relation's attribute.
+	 */
+	void		(*CopyFromInFunc) (CopyFromState cstate, Oid atttypid,
+								   FmgrInfo *finfo, Oid *typioparam);
+
+	/*
+	 * Called when COPY FROM is started.
+	 *
+	 * `tupDesc` is the tuple descriptor of the relation where the data needs
+	 * to be copied.  This can be used for any initialization steps required
+	 * by a format.
+	 */
+	void		(*CopyFromStart) (CopyFromState cstate, TupleDesc tupDesc);
+
+	/*
+	 * Copy one row to a set of `values` and `nulls` of size tupDesc->natts.
+	 *
+	 * 'econtext' is used to evaluate default expression for each column that
+	 * is either not read from the file or is using the DEFAULT option of COPY
+	 * FROM.  It is NULL if no default values are used.
+	 *
+	 * Returns false if there are no more tuples to copy.
+	 */
+	bool		(*CopyFromOneRow) (CopyFromState cstate, ExprContext *econtext,
+								   Datum *values, bool *nulls);
+
+	/* Called when COPY FROM has ended. */
+	void		(*CopyFromEnd) (CopyFromState cstate);
+} CopyFromRoutine;
+
+/*
+ * API structure for a COPY TO format implementation.   Note this must be
+ * allocated in a server-lifetime manner, typically as a static const struct.
+ */
+typedef struct CopyToRoutine
+{
+	/*
+	 * Called when COPY TO is started to set up the output functions
+	 * associated to the relation's attributes reading from.  `finfo` can be
+	 * optionally filled.  `atttypid` is the OID of data type used by the
+	 * relation's attribute.
+	 */
+	void		(*CopyToOutFunc) (CopyToState cstate, Oid atttypid,
+								  FmgrInfo *finfo);
+
+	/*
+	 * Called when COPY TO is started.
+	 *
+	 * `tupDesc` is the tuple descriptor of the relation from where the data
+	 * is read.
+	 */
+	void		(*CopyToStart) (CopyToState cstate, TupleDesc tupDesc);
+
+	/*
+	 * Copy one row for COPY TO.
+	 *
+	 * `slot` is the tuple slot where the data is emitted.
+	 */
+	void		(*CopyToOneRow) (CopyToState cstate, TupleTableSlot *slot);
+
+	/* Called when COPY TO has ended */
+	void		(*CopyToEnd) (CopyToState cstate);
+} CopyToRoutine;
+
+#endif							/* COPYAPI_H */
diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h
index cad52fcc78..509b9e92a1 100644
--- a/src/include/commands/copyfrom_internal.h
+++ b/src/include/commands/copyfrom_internal.h
@@ -15,6 +15,7 @@
 #define COPYFROM_INTERNAL_H
 
 #include "commands/copy.h"
+#include "commands/copyapi.h"
 #include "commands/trigger.h"
 #include "nodes/miscnodes.h"
 
@@ -58,6 +59,9 @@ typedef enum CopyInsertMethod
  */
 typedef struct CopyFromStateData
 {
+	/* format routine */
+	const CopyFromRoutine *routine;
+
 	/* low-level state data */
 	CopySource	copy_src;		/* type of copy source */
 	FILE	   *copy_file;		/* used if copy_src == COPY_FILE */
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index ee40a341d3..a5ae161ca5 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -475,6 +475,7 @@ ConvertRowtypeExpr
 CookedConstraint
 CopyDest
 CopyFormatOptions
+CopyFromRoutine
 CopyFromState
 CopyFromStateData
 CopyHeaderChoice
@@ -484,6 +485,7 @@ CopyMultiInsertInfo
 CopyOnErrorChoice
 CopySource
 CopyStmt
+CopyToRoutine
 CopyToState
 CopyToStateData
 Cost
-- 
2.43.0


----Next_Part(Mon_Mar__4_14_11_08_2024_631)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="v16-0001-Add-CopyFromRoutine-CopyToRountine-w.patch"



^ permalink  raw  reply  [nested|flat] 15+ messages in thread

* [PATCH v16] Add CopyFromRoutine/CopyToRountine
@ 2024-03-04 04:52 Sutou Kouhei <[email protected]>
  0 siblings, 0 replies; 15+ messages in thread

From: Sutou Kouhei @ 2024-03-04 04:52 UTC (permalink / raw)

They are for implementing custom COPY FROM/TO format. But this is not
enough to implement custom COPY FROM/TO format yet. We'll export some
APIs to receive/send data and add "format" option to COPY FROM/TO
later.

Existing text/csv/binary format implementations don't use
CopyFromRoutine/CopyToRoutine for now. We have a patch for it but we
defer it. Because there are some mysterious profile results in spite
of we get faster runtimes. See [1] for details.

[1] https://www.postgresql.org/message-id/ZdbtQJ-p5H1_EDwE%40paquier.xyz

Note that this doesn't change existing text/csv/binary format
implementations. There are many diffs for CopyOneRowTo() but they're
caused by indentation. They don't change implementations.
---
 src/backend/commands/copyfrom.c          |  22 ++++-
 src/backend/commands/copyfromparse.c     |   5 ++
 src/backend/commands/copyto.c            |  21 +++++
 src/include/commands/copyapi.h           | 100 +++++++++++++++++++++++
 src/include/commands/copyfrom_internal.h |   4 +
 src/tools/pgindent/typedefs.list         |   2 +
 6 files changed, 151 insertions(+), 3 deletions(-)
 create mode 100644 src/include/commands/copyapi.h

diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index c3bc897028..9bf2f6497e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -1623,12 +1623,22 @@ BeginCopyFrom(ParseState *pstate,
 
 		/* Fetch the input function and typioparam info */
 		if (cstate->opts.binary)
+		{
 			getTypeBinaryInputInfo(att->atttypid,
 								   &in_func_oid, &typioparams[attnum - 1]);
+			fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+		}
+		else if (cstate->routine)
+			cstate->routine->CopyFromInFunc(cstate, att->atttypid,
+											&in_functions[attnum - 1],
+											&typioparams[attnum - 1]);
+
 		else
+		{
 			getTypeInputInfo(att->atttypid,
 							 &in_func_oid, &typioparams[attnum - 1]);
 			fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+		}
 
 		/* Get default info if available */
 		defexprs[attnum - 1] = NULL;
@@ -1768,10 +1778,13 @@ BeginCopyFrom(ParseState *pstate,
 		/* Read and verify binary header */
 		ReceiveCopyBinaryHeader(cstate);
 	}
-
-	/* create workspace for CopyReadAttributes results */
-	if (!cstate->opts.binary)
+	else if (cstate->routine)
 	{
+		cstate->routine->CopyFromStart(cstate, tupDesc);
+	}
+	else
+	{
+		/* create workspace for CopyReadAttributes results */
 		AttrNumber	attr_count = list_length(cstate->attnumlist);
 
 		cstate->max_fields = attr_count;
@@ -1789,6 +1802,9 @@ BeginCopyFrom(ParseState *pstate,
 void
 EndCopyFrom(CopyFromState cstate)
 {
+	if (cstate->routine)
+		cstate->routine->CopyFromEnd(cstate);
+
 	/* No COPY FROM related resources except memory. */
 	if (cstate->is_program)
 	{
diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c
index 7cacd0b752..8b15080585 100644
--- a/src/backend/commands/copyfromparse.c
+++ b/src/backend/commands/copyfromparse.c
@@ -978,6 +978,11 @@ NextCopyFrom(CopyFromState cstate, ExprContext *econtext,
 
 		Assert(fieldno == attr_count);
 	}
+	else if (cstate->routine)
+	{
+		if (!cstate->routine->CopyFromOneRow(cstate, econtext, values, nulls))
+			return false;
+	}
 	else
 	{
 		/* binary */
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index 20ffc90363..6080627c83 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -24,6 +24,7 @@
 #include "access/xact.h"
 #include "access/xlog.h"
 #include "commands/copy.h"
+#include "commands/copyapi.h"
 #include "commands/progress.h"
 #include "executor/execdesc.h"
 #include "executor/executor.h"
@@ -71,6 +72,9 @@ typedef enum CopyDest
  */
 typedef struct CopyToStateData
 {
+	/* format routine */
+	const CopyToRoutine *routine;
+
 	/* low-level state data */
 	CopyDest	copy_dest;		/* type of copy source/destination */
 	FILE	   *copy_file;		/* used if copy_dest == COPY_FILE */
@@ -777,15 +781,23 @@ DoCopyTo(CopyToState cstate)
 		Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
 
 		if (cstate->opts.binary)
+		{
 			getTypeBinaryOutputInfo(attr->atttypid,
 									&out_func_oid,
 									&isvarlena);
+			fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+		}
+		else if (cstate->routine)
+			cstate->routine->CopyToOutFunc(cstate, attr->atttypid,
+										   &cstate->out_functions[attnum - 1]);
 		else
+		{
 			getTypeOutputInfo(attr->atttypid,
 							  &out_func_oid,
 							  &isvarlena);
 			fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
 		}
+	}
 
 	/*
 	 * Create a temporary memory context that we can reset once per row to
@@ -811,6 +823,8 @@ DoCopyTo(CopyToState cstate)
 		tmp = 0;
 		CopySendInt32(cstate, tmp);
 	}
+	else if (cstate->routine)
+		cstate->routine->CopyToStart(cstate, tupDesc);
 	else
 	{
 		/*
@@ -892,6 +906,8 @@ DoCopyTo(CopyToState cstate)
 		/* Need to flush out the trailer */
 		CopySendEndOfRow(cstate);
 	}
+	else if (cstate->routine)
+		cstate->routine->CopyToEnd(cstate);
 
 	MemoryContextDelete(cstate->rowcontext);
 
@@ -916,6 +932,10 @@ CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot)
 	MemoryContextReset(cstate->rowcontext);
 	oldcontext = MemoryContextSwitchTo(cstate->rowcontext);
 
+	if (cstate->routine)
+		cstate->routine->CopyToOneRow(cstate, slot);
+	else
+	{
 		if (cstate->opts.binary)
 		{
 			/* Binary per-tuple header */
@@ -971,6 +991,7 @@ CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot)
 		}
 
 		CopySendEndOfRow(cstate);
+	}
 
 	MemoryContextSwitchTo(oldcontext);
 }
diff --git a/src/include/commands/copyapi.h b/src/include/commands/copyapi.h
new file mode 100644
index 0000000000..635c4cbff2
--- /dev/null
+++ b/src/include/commands/copyapi.h
@@ -0,0 +1,100 @@
+/*-------------------------------------------------------------------------
+ *
+ * copyapi.h
+ *	  API for COPY TO/FROM handlers
+ *
+ *
+ * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/commands/copyapi.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef COPYAPI_H
+#define COPYAPI_H
+
+#include "executor/tuptable.h"
+#include "nodes/execnodes.h"
+
+/* These are private in commands/copy[from|to].c */
+typedef struct CopyFromStateData *CopyFromState;
+typedef struct CopyToStateData *CopyToState;
+
+/*
+ * API structure for a COPY FROM format implementation.  Note this must be
+ * allocated in a server-lifetime manner, typically as a static const struct.
+ */
+typedef struct CopyFromRoutine
+{
+	/*
+	 * Called when COPY FROM is started to set up the input functions
+	 * associated to the relation's attributes writing to.  `finfo` can be
+	 * optionally filled to provide the catalog information of the input
+	 * function.  `typioparam` can be optionally filled to define the OID of
+	 * the type to pass to the input function.  `atttypid` is the OID of data
+	 * type used by the relation's attribute.
+	 */
+	void		(*CopyFromInFunc) (CopyFromState cstate, Oid atttypid,
+								   FmgrInfo *finfo, Oid *typioparam);
+
+	/*
+	 * Called when COPY FROM is started.
+	 *
+	 * `tupDesc` is the tuple descriptor of the relation where the data needs
+	 * to be copied.  This can be used for any initialization steps required
+	 * by a format.
+	 */
+	void		(*CopyFromStart) (CopyFromState cstate, TupleDesc tupDesc);
+
+	/*
+	 * Copy one row to a set of `values` and `nulls` of size tupDesc->natts.
+	 *
+	 * 'econtext' is used to evaluate default expression for each column that
+	 * is either not read from the file or is using the DEFAULT option of COPY
+	 * FROM.  It is NULL if no default values are used.
+	 *
+	 * Returns false if there are no more tuples to copy.
+	 */
+	bool		(*CopyFromOneRow) (CopyFromState cstate, ExprContext *econtext,
+								   Datum *values, bool *nulls);
+
+	/* Called when COPY FROM has ended. */
+	void		(*CopyFromEnd) (CopyFromState cstate);
+} CopyFromRoutine;
+
+/*
+ * API structure for a COPY TO format implementation.   Note this must be
+ * allocated in a server-lifetime manner, typically as a static const struct.
+ */
+typedef struct CopyToRoutine
+{
+	/*
+	 * Called when COPY TO is started to set up the output functions
+	 * associated to the relation's attributes reading from.  `finfo` can be
+	 * optionally filled.  `atttypid` is the OID of data type used by the
+	 * relation's attribute.
+	 */
+	void		(*CopyToOutFunc) (CopyToState cstate, Oid atttypid,
+								  FmgrInfo *finfo);
+
+	/*
+	 * Called when COPY TO is started.
+	 *
+	 * `tupDesc` is the tuple descriptor of the relation from where the data
+	 * is read.
+	 */
+	void		(*CopyToStart) (CopyToState cstate, TupleDesc tupDesc);
+
+	/*
+	 * Copy one row for COPY TO.
+	 *
+	 * `slot` is the tuple slot where the data is emitted.
+	 */
+	void		(*CopyToOneRow) (CopyToState cstate, TupleTableSlot *slot);
+
+	/* Called when COPY TO has ended */
+	void		(*CopyToEnd) (CopyToState cstate);
+} CopyToRoutine;
+
+#endif							/* COPYAPI_H */
diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h
index cad52fcc78..509b9e92a1 100644
--- a/src/include/commands/copyfrom_internal.h
+++ b/src/include/commands/copyfrom_internal.h
@@ -15,6 +15,7 @@
 #define COPYFROM_INTERNAL_H
 
 #include "commands/copy.h"
+#include "commands/copyapi.h"
 #include "commands/trigger.h"
 #include "nodes/miscnodes.h"
 
@@ -58,6 +59,9 @@ typedef enum CopyInsertMethod
  */
 typedef struct CopyFromStateData
 {
+	/* format routine */
+	const CopyFromRoutine *routine;
+
 	/* low-level state data */
 	CopySource	copy_src;		/* type of copy source */
 	FILE	   *copy_file;		/* used if copy_src == COPY_FILE */
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index ee40a341d3..a5ae161ca5 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -475,6 +475,7 @@ ConvertRowtypeExpr
 CookedConstraint
 CopyDest
 CopyFormatOptions
+CopyFromRoutine
 CopyFromState
 CopyFromStateData
 CopyHeaderChoice
@@ -484,6 +485,7 @@ CopyMultiInsertInfo
 CopyOnErrorChoice
 CopySource
 CopyStmt
+CopyToRoutine
 CopyToState
 CopyToStateData
 Cost
-- 
2.43.0


----Next_Part(Mon_Mar__4_14_11_08_2024_631)----





^ permalink  raw  reply  [nested|flat] 15+ messages in thread

* [PATCH v17] Add CopyFromRoutine/CopyToRountine
@ 2024-03-04 04:52 Sutou Kouhei <[email protected]>
  0 siblings, 0 replies; 15+ messages in thread

From: Sutou Kouhei @ 2024-03-04 04:52 UTC (permalink / raw)

They are for implementing custom COPY FROM/TO format. But this is not
enough to implement custom COPY FROM/TO format yet. We'll export some
APIs to receive/send data and add "format" option to COPY FROM/TO
later.

Existing text/csv/binary format implementations don't use
CopyFromRoutine/CopyToRoutine for now. We have a patch for it but we
defer it. Because there are some mysterious profile results in spite
of we get faster runtimes. See [1] for details.

[1] https://www.postgresql.org/message-id/ZdbtQJ-p5H1_EDwE%40paquier.xyz

Note that this doesn't change existing text/csv/binary format
implementations.
---
 src/backend/commands/copyfrom.c          |  24 +++++-
 src/backend/commands/copyfromparse.c     |   5 ++
 src/backend/commands/copyto.c            |  25 +++++-
 src/include/commands/copyapi.h           | 100 +++++++++++++++++++++++
 src/include/commands/copyfrom_internal.h |   4 +
 src/tools/pgindent/typedefs.list         |   2 +
 6 files changed, 155 insertions(+), 5 deletions(-)
 create mode 100644 src/include/commands/copyapi.h

diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index c3bc897028..9bf2f6497e 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -1623,12 +1623,22 @@ BeginCopyFrom(ParseState *pstate,
 
 		/* Fetch the input function and typioparam info */
 		if (cstate->opts.binary)
+		{
 			getTypeBinaryInputInfo(att->atttypid,
 								   &in_func_oid, &typioparams[attnum - 1]);
+			fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+		}
+		else if (cstate->routine)
+			cstate->routine->CopyFromInFunc(cstate, att->atttypid,
+											&in_functions[attnum - 1],
+											&typioparams[attnum - 1]);
+
 		else
+		{
 			getTypeInputInfo(att->atttypid,
 							 &in_func_oid, &typioparams[attnum - 1]);
-		fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+			fmgr_info(in_func_oid, &in_functions[attnum - 1]);
+		}
 
 		/* Get default info if available */
 		defexprs[attnum - 1] = NULL;
@@ -1768,10 +1778,13 @@ BeginCopyFrom(ParseState *pstate,
 		/* Read and verify binary header */
 		ReceiveCopyBinaryHeader(cstate);
 	}
-
-	/* create workspace for CopyReadAttributes results */
-	if (!cstate->opts.binary)
+	else if (cstate->routine)
 	{
+		cstate->routine->CopyFromStart(cstate, tupDesc);
+	}
+	else
+	{
+		/* create workspace for CopyReadAttributes results */
 		AttrNumber	attr_count = list_length(cstate->attnumlist);
 
 		cstate->max_fields = attr_count;
@@ -1789,6 +1802,9 @@ BeginCopyFrom(ParseState *pstate,
 void
 EndCopyFrom(CopyFromState cstate)
 {
+	if (cstate->routine)
+		cstate->routine->CopyFromEnd(cstate);
+
 	/* No COPY FROM related resources except memory. */
 	if (cstate->is_program)
 	{
diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c
index 7cacd0b752..8b15080585 100644
--- a/src/backend/commands/copyfromparse.c
+++ b/src/backend/commands/copyfromparse.c
@@ -978,6 +978,11 @@ NextCopyFrom(CopyFromState cstate, ExprContext *econtext,
 
 		Assert(fieldno == attr_count);
 	}
+	else if (cstate->routine)
+	{
+		if (!cstate->routine->CopyFromOneRow(cstate, econtext, values, nulls))
+			return false;
+	}
 	else
 	{
 		/* binary */
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index 20ffc90363..b4a7c9c8b9 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -24,6 +24,7 @@
 #include "access/xact.h"
 #include "access/xlog.h"
 #include "commands/copy.h"
+#include "commands/copyapi.h"
 #include "commands/progress.h"
 #include "executor/execdesc.h"
 #include "executor/executor.h"
@@ -71,6 +72,9 @@ typedef enum CopyDest
  */
 typedef struct CopyToStateData
 {
+	/* format routine */
+	const CopyToRoutine *routine;
+
 	/* low-level state data */
 	CopyDest	copy_dest;		/* type of copy source/destination */
 	FILE	   *copy_file;		/* used if copy_dest == COPY_FILE */
@@ -777,14 +781,22 @@ DoCopyTo(CopyToState cstate)
 		Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
 
 		if (cstate->opts.binary)
+		{
 			getTypeBinaryOutputInfo(attr->atttypid,
 									&out_func_oid,
 									&isvarlena);
+			fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+		}
+		else if (cstate->routine)
+			cstate->routine->CopyToOutFunc(cstate, attr->atttypid,
+										   &cstate->out_functions[attnum - 1]);
 		else
+		{
 			getTypeOutputInfo(attr->atttypid,
 							  &out_func_oid,
 							  &isvarlena);
-		fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+			fmgr_info(out_func_oid, &cstate->out_functions[attnum - 1]);
+		}
 	}
 
 	/*
@@ -811,6 +823,8 @@ DoCopyTo(CopyToState cstate)
 		tmp = 0;
 		CopySendInt32(cstate, tmp);
 	}
+	else if (cstate->routine)
+		cstate->routine->CopyToStart(cstate, tupDesc);
 	else
 	{
 		/*
@@ -892,6 +906,8 @@ DoCopyTo(CopyToState cstate)
 		/* Need to flush out the trailer */
 		CopySendEndOfRow(cstate);
 	}
+	else if (cstate->routine)
+		cstate->routine->CopyToEnd(cstate);
 
 	MemoryContextDelete(cstate->rowcontext);
 
@@ -916,6 +932,13 @@ CopyOneRowTo(CopyToState cstate, TupleTableSlot *slot)
 	MemoryContextReset(cstate->rowcontext);
 	oldcontext = MemoryContextSwitchTo(cstate->rowcontext);
 
+	if (cstate->routine)
+	{
+		cstate->routine->CopyToOneRow(cstate, slot);
+		MemoryContextSwitchTo(oldcontext);
+		return;
+	}
+
 	if (cstate->opts.binary)
 	{
 		/* Binary per-tuple header */
diff --git a/src/include/commands/copyapi.h b/src/include/commands/copyapi.h
new file mode 100644
index 0000000000..635c4cbff2
--- /dev/null
+++ b/src/include/commands/copyapi.h
@@ -0,0 +1,100 @@
+/*-------------------------------------------------------------------------
+ *
+ * copyapi.h
+ *	  API for COPY TO/FROM handlers
+ *
+ *
+ * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/commands/copyapi.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef COPYAPI_H
+#define COPYAPI_H
+
+#include "executor/tuptable.h"
+#include "nodes/execnodes.h"
+
+/* These are private in commands/copy[from|to].c */
+typedef struct CopyFromStateData *CopyFromState;
+typedef struct CopyToStateData *CopyToState;
+
+/*
+ * API structure for a COPY FROM format implementation.  Note this must be
+ * allocated in a server-lifetime manner, typically as a static const struct.
+ */
+typedef struct CopyFromRoutine
+{
+	/*
+	 * Called when COPY FROM is started to set up the input functions
+	 * associated to the relation's attributes writing to.  `finfo` can be
+	 * optionally filled to provide the catalog information of the input
+	 * function.  `typioparam` can be optionally filled to define the OID of
+	 * the type to pass to the input function.  `atttypid` is the OID of data
+	 * type used by the relation's attribute.
+	 */
+	void		(*CopyFromInFunc) (CopyFromState cstate, Oid atttypid,
+								   FmgrInfo *finfo, Oid *typioparam);
+
+	/*
+	 * Called when COPY FROM is started.
+	 *
+	 * `tupDesc` is the tuple descriptor of the relation where the data needs
+	 * to be copied.  This can be used for any initialization steps required
+	 * by a format.
+	 */
+	void		(*CopyFromStart) (CopyFromState cstate, TupleDesc tupDesc);
+
+	/*
+	 * Copy one row to a set of `values` and `nulls` of size tupDesc->natts.
+	 *
+	 * 'econtext' is used to evaluate default expression for each column that
+	 * is either not read from the file or is using the DEFAULT option of COPY
+	 * FROM.  It is NULL if no default values are used.
+	 *
+	 * Returns false if there are no more tuples to copy.
+	 */
+	bool		(*CopyFromOneRow) (CopyFromState cstate, ExprContext *econtext,
+								   Datum *values, bool *nulls);
+
+	/* Called when COPY FROM has ended. */
+	void		(*CopyFromEnd) (CopyFromState cstate);
+} CopyFromRoutine;
+
+/*
+ * API structure for a COPY TO format implementation.   Note this must be
+ * allocated in a server-lifetime manner, typically as a static const struct.
+ */
+typedef struct CopyToRoutine
+{
+	/*
+	 * Called when COPY TO is started to set up the output functions
+	 * associated to the relation's attributes reading from.  `finfo` can be
+	 * optionally filled.  `atttypid` is the OID of data type used by the
+	 * relation's attribute.
+	 */
+	void		(*CopyToOutFunc) (CopyToState cstate, Oid atttypid,
+								  FmgrInfo *finfo);
+
+	/*
+	 * Called when COPY TO is started.
+	 *
+	 * `tupDesc` is the tuple descriptor of the relation from where the data
+	 * is read.
+	 */
+	void		(*CopyToStart) (CopyToState cstate, TupleDesc tupDesc);
+
+	/*
+	 * Copy one row for COPY TO.
+	 *
+	 * `slot` is the tuple slot where the data is emitted.
+	 */
+	void		(*CopyToOneRow) (CopyToState cstate, TupleTableSlot *slot);
+
+	/* Called when COPY TO has ended */
+	void		(*CopyToEnd) (CopyToState cstate);
+} CopyToRoutine;
+
+#endif							/* COPYAPI_H */
diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h
index cad52fcc78..509b9e92a1 100644
--- a/src/include/commands/copyfrom_internal.h
+++ b/src/include/commands/copyfrom_internal.h
@@ -15,6 +15,7 @@
 #define COPYFROM_INTERNAL_H
 
 #include "commands/copy.h"
+#include "commands/copyapi.h"
 #include "commands/trigger.h"
 #include "nodes/miscnodes.h"
 
@@ -58,6 +59,9 @@ typedef enum CopyInsertMethod
  */
 typedef struct CopyFromStateData
 {
+	/* format routine */
+	const CopyFromRoutine *routine;
+
 	/* low-level state data */
 	CopySource	copy_src;		/* type of copy source */
 	FILE	   *copy_file;		/* used if copy_src == COPY_FILE */
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index ee40a341d3..a5ae161ca5 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -475,6 +475,7 @@ ConvertRowtypeExpr
 CookedConstraint
 CopyDest
 CopyFormatOptions
+CopyFromRoutine
 CopyFromState
 CopyFromStateData
 CopyHeaderChoice
@@ -484,6 +485,7 @@ CopyMultiInsertInfo
 CopyOnErrorChoice
 CopySource
 CopyStmt
+CopyToRoutine
 CopyToState
 CopyToStateData
 Cost
-- 
2.43.0


----Next_Part(Tue_Mar__5_17_18_08_2024_468)----





^ permalink  raw  reply  [nested|flat] 15+ messages in thread

end of thread, other threads:[~2024-03-04 04:52 UTC | newest]

Thread overview: 15+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2018-10-11 01:03 [PATCH 2/8] Write WAL for empty nbtree index build Kyotaro Horiguchi <[email protected]>
2018-10-11 01:03 [PATCH 2/4] Write WAL for empty nbtree index build Kyotaro Horiguchi <[email protected]>
2018-10-11 01:03 [PATCH 2/7] Write WAL for empty nbtree index build Kyotaro Horiguchi <[email protected]>
2018-10-11 01:03 [PATCH 2/7] Write WAL for empty nbtree index build Kyotaro Horiguchi <[email protected]>
2018-10-11 01:03 [PATCH 2/7] Write WAL for empty nbtree index build Kyotaro Horiguchi <[email protected]>
2023-09-20 01:42 Re: Move global variables of pgoutput to plugin private scope. Peter Smith <[email protected]>
2024-03-04 04:52 [PATCH v17] Add CopyFromRoutine/CopyToRountine Sutou Kouhei <[email protected]>
2024-03-04 04:52 [PATCH v17] Add CopyFromRoutine/CopyToRountine Sutou Kouhei <[email protected]>
2024-03-04 04:52 [PATCH v16] Add CopyFromRoutine/CopyToRountine Sutou Kouhei <[email protected]>
2024-03-04 04:52 [PATCH v16] Add CopyFromRoutine/CopyToRountine Sutou Kouhei <[email protected]>
2024-03-04 04:52 [PATCH v16] Add CopyFromRoutine/CopyToRountine Sutou Kouhei <[email protected]>
2024-03-04 04:52 [PATCH v16] Add CopyFromRoutine/CopyToRountine Sutou Kouhei <[email protected]>
2024-03-04 04:52 [PATCH v16] Add CopyFromRoutine/CopyToRountine Sutou Kouhei <[email protected]>
2024-03-04 04:52 [PATCH v16] Add CopyFromRoutine/CopyToRountine Sutou Kouhei <[email protected]>
2024-03-04 04:52 [PATCH v17] Add CopyFromRoutine/CopyToRountine Sutou Kouhei <[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