Added support for COPY FROM, for loading tables.

Some additional work was done related to the combiner and
error handling to make this code a little cleaner.

This was written by Andrei Martsinchyk.

4 files changed, 20 insertions, 3 deletions

diff --git a/src/include/pgxc/combiner.h b/src/include/pgxc/combiner.h
index 2596e50677..7926cbd9ff 100644
--- a/src/include/pgxc/combiner.h
+++ b/src/include/pgxc/combiner.h
@@ -46,6 +46,9 @@ typedef struct
 	int			row_count;
 	RequestType request_type;
 	int			description_count;
+	uint64		copy_in_count;
+	uint64		copy_out_count;
+	bool		inErrorState;
 	List	   *simple_aggregates;
 }	ResponseCombinerData;
 
@@ -59,5 +62,6 @@ extern int CombineResponse(ResponseCombiner combiner, char msg_type,
 extern bool ValidateAndCloseCombiner(ResponseCombiner combiner);
 extern bool ValidateAndResetCombiner(ResponseCombiner combiner);
 extern void AssignCombinerAggregates(ResponseCombiner combiner, List *simple_aggregates);
+extern void CloseCombiner(ResponseCombiner combiner);
 
 #endif   /* COMBINER_H */
diff --git a/src/include/pgxc/datanode.h b/src/include/pgxc/datanode.h
index 165be530d9..4f75ba24af 100644
--- a/src/include/pgxc/datanode.h
+++ b/src/include/pgxc/datanode.h
@@ -33,10 +33,16 @@ typedef enum
 	DN_CONNECTION_STATE_COMPLETED,
 	DN_CONNECTION_STATE_ERROR_NOT_READY,	/* error, but need ReadyForQuery message */
 	DN_CONNECTION_STATE_ERROR_READY,		/* error and received ReadyForQuery */
-	DN_CONNECTION_STATE_ERROR_FATAL			/* fatal error */
-
+	DN_CONNECTION_STATE_ERROR_FATAL,		/* fatal error */
+	DN_CONNECTION_STATE_COPY_IN,
+	DN_CONNECTION_STATE_COPY_OUT
 }	DNConnectionState;
 
+#define DN_CONNECTION_STATE_ERROR(dnconn) \
+	(dnconn)->state == DN_CONNECTION_STATE_ERROR_FATAL \
+	|| (dnconn)->state == DN_CONNECTION_STATE_ERROR_NOT_READY \
+	|| (dnconn)->state == DN_CONNECTION_STATE_ERROR_READY
+
 struct data_node_handle
 {
 	/* fd of the connection */
@@ -74,7 +80,11 @@ extern void DataNodeBegin(void);
 extern int	DataNodeCommit(CommandDest dest);
 extern int	DataNodeRollback(CommandDest dest);
 
-extern int	DataNodeExec(const char *query, Exec_Nodes *exec_nodes, CommandDest dest, Snapshot snapshot, bool force_autocommit, List *simple_aggregates, bool is_read_only);
+extern int	DataNodeExec(const char *query, Exec_Nodes *exec_nodes, CombineType combine_type, CommandDest dest, Snapshot snapshot, bool force_autocommit, List *simple_aggregates, bool is_read_only);
+
+extern DataNodeHandle** DataNodeCopyBegin(const char *query, List *nodelist, Snapshot snapshot);
+extern int DataNodeCopyIn(char *data_row, int len, Exec_Nodes *exec_nodes, DataNodeHandle** copy_connections);
+extern uint64 DataNodeCopyFinish(DataNodeHandle** copy_connections, int primary_data_node, CombineType combine_type, CommandDest dest);
 
 extern int primary_data_node;
 #endif
diff --git a/src/include/pgxc/locator.h b/src/include/pgxc/locator.h
index 6fa32fb73b..1d5280c765 100644
--- a/src/include/pgxc/locator.h
+++ b/src/include/pgxc/locator.h
@@ -52,6 +52,7 @@ typedef struct
 {
 	List	   *primarynodelist; 
 	List	   *nodelist;
+	char	    baselocatortype;
 }	Exec_Nodes;
 
 
diff --git a/src/include/pgxc/planner.h b/src/include/pgxc/planner.h
index c738312ce1..04ec6fcc0e 100644
--- a/src/include/pgxc/planner.h
+++ b/src/include/pgxc/planner.h
@@ -16,6 +16,7 @@
 #define PGXCPLANNER_H
 
 #include "pgxc/locator.h"
+#include "pgxc/combiner.h"
 
 
 /* for Query_Plan.exec_loc_type can have these OR'ed*/
@@ -30,6 +31,7 @@ typedef struct
 {
 	char	   *sql_statement;
 	Exec_Nodes *exec_nodes;
+	CombineType combine_type;
 	List	   *simple_aggregates;		/* simple aggregate to combine on this
 										 * step */
 }	Query_Step;