Update of /home/cvsd/slony1/slony1-engine/src/slon
In directory main.slony.info:/tmp/cvs-serv2496/src/slon

Modified Files:
	remote_worker.c 
Log Message:
This is the first step towards Slony-I version 2.0.

It drops all support for databases prior to Postgres version 8.3.
This is required because we now make use of new functionality in
Postgres, namely the trigger and rule support for session replication
role. As of now, every node (origin/subscriber/mixed) can be dumped
with pg_dump and result in a consistent snapshot of the database.

TODO:
  - Run DDL scripts as "local" replication role. Currently DDL
    scripts don't work at all.
  - Get rid of STORE TRIGGER and force users to make use of 8.3's
	advanced trigger and rule configuration.

Jan


Index: remote_worker.c
===================================================================
RCS file: /home/cvsd/slony1/slony1-engine/src/slon/remote_worker.c,v
retrieving revision 1.139
retrieving revision 1.140
diff -C2 -d -r1.139 -r1.140
*** remote_worker.c	31 May 2007 13:29:27 -0000	1.139
--- remote_worker.c	31 May 2007 16:46:18 -0000	1.140
***************
*** 336,340 ****
  	 */
  	(void) slon_mkquery(&query1,
! 				 "select %s.setSessionRole('_%s', 'slon'); ",
  				 rtcfg_namespace, rtcfg_cluster_name);
  	if (query_execute(node, local_dbconn, &query1) < 0)
--- 336,340 ----
  	 */
  	(void) slon_mkquery(&query1,
! 				 "set session_replication_role = replica; ",
  				 rtcfg_namespace, rtcfg_cluster_name);
  	if (query_execute(node, local_dbconn, &query1) < 0)
***************
*** 2355,2360 ****
  	PGresult   *res2;
  	PGresult   *res3;
- 	PGresult   *res4;
- 	int			nodeon73;
  	int			rc;
  	int			set_origin = 0;
--- 2355,2358 ----
***************
*** 2367,2377 ****
  	SlonDString ssy_action_list;
  	char		seqbuf[64];
- 
- #ifdef HAVE_PQPUTCOPYDATA
  	char	   *copydata = NULL;
- #else
- 	char		copybuf[8192];
- 	int			copydone;
- #endif
  	struct timeval tv_start;
  	struct timeval tv_start2;
--- 2365,2369 ----
***************
*** 2854,2886 ****
  		}
  
- 		(void) slon_mkquery(&query2, "select %s.pre74();",
- 					 rtcfg_namespace);
- 		res4 = PQexec(loc_dbconn, dstring_data(&query2));
- 
- 		if (PQresultStatus(res4) != PGRES_TUPLES_OK)
- 		{
- 			slon_log(SLON_ERROR, "remoteWorkerThread_%d: \"%s\" %s\n",
- 					 node->no_id, dstring_data(&query2),
- 					 PQresultErrorMessage(res4));
- 			PQclear(res4);
- 			PQclear(res3);
- 			PQclear(res1);
- 			slon_disconnectdb(pro_conn);
- 			dstring_free(&query1);
- 			dstring_free(&query2);
- 			dstring_free(&query3);
- 			dstring_free(&lsquery);
- 			dstring_free(&indexregenquery);
- 			archive_terminate(node);
- 			return -1;
- 		}
- 
- 		/* Are we running on < PG 7.4???  result =	*/
- 		nodeon73 = atoi(PQgetvalue(res4, 0, 0));
- 
- 		slon_log(SLON_DEBUG2, "remoteWorkerThread_%d: "
- 				 " nodeon73 is %d\n",
- 				 node->no_id, nodeon73);
- 
  		(void) slon_mkquery(&query1,
  					 "select %s.prepareTableForCopy(%d); "
--- 2846,2849 ----
***************
*** 2888,2892 ****
  					 rtcfg_namespace,
  					 tab_id, tab_fqname,
! 					 nodeon73 ? "" : PQgetvalue(res3, 0, 0)
  			);
  		res2 = PQexec(loc_dbconn, dstring_data(&query1));
--- 2851,2855 ----
  					 rtcfg_namespace,
  					 tab_id, tab_fqname,
! 					 PQgetvalue(res3, 0, 0)
  			);
  		res2 = PQexec(loc_dbconn, dstring_data(&query1));
***************
*** 2912,2916 ****
  			(void) slon_mkquery(&query1,
  			 "delete from %s;\ncopy %s %s from stdin;", tab_fqname, tab_fqname,
! 						 nodeon73 ? "" : PQgetvalue(res3, 0, 0));
  			rc = archive_append_ds(node, &query1);
  			if (rc < 0)
--- 2875,2879 ----
  			(void) slon_mkquery(&query1,
  			 "delete from %s;\ncopy %s %s from stdin;", tab_fqname, tab_fqname,
! 						 PQgetvalue(res3, 0, 0));
  			rc = archive_append_ds(node, &query1);
  			if (rc < 0)
***************
*** 2940,2949 ****
  					 PQresultErrorMessage(res2),
  					 PQerrorMessage(pro_dbconn));
- #ifdef HAVE_PQPUTCOPYDATA
  			PQputCopyEnd(loc_dbconn, "Slony-I: copy set operation failed");
- #else
- 			PQputline(loc_dbconn, "\\.\n");
- 			PQendcopy(loc_dbconn);
- #endif
  			PQclear(res3);
  			PQclear(res2);
--- 2903,2907 ----
***************
*** 2962,2966 ****
  		 * Copy the data over
  		 */
- #ifdef HAVE_PQPUTCOPYDATA
  		while ((rc = PQgetCopyData(pro_dbconn, &copydata, 0)) > 0)
  		{
--- 2920,2923 ----
***************
*** 3117,3247 ****
  			}
  		}
- #else							/* ! HAVE_PQPUTCOPYDATA */
- 		copydone = false;
- 		while (!copydone)
- 		{
- 			rc = PQgetline(pro_dbconn, copybuf, sizeof(copybuf));
- 
- 			if (copybuf[0] == '\\' &&
- 				copybuf[1] == '.' &&
- 				copybuf[2] == '\0')
- 			{
- 				copydone = true;
- 			}
- 			else
- 			{
- 				switch (rc)
- 				{
- 				case EOF:
- 				  copydone = true;
- 				  break;
- 				case 0:
- 				  PQputline(loc_dbconn, copybuf);
- 				  PQputline(loc_dbconn, "\n");
- 				  if (archive_dir) {
- 				    rc = archive_append_str(node, copybuf);
- 				    if (rc < 0) {
- 				      PQclear(res2);
- 				      PQclear(res1);
- 				      slon_disconnectdb(pro_conn);
- 				      dstring_free(&query1);
- 				      dstring_free(&query2);
- 				      dstring_free(&query3);
- 				      dstring_free(&lsquery);
- 				      dstring_free(&indexregenquery);
- 				      archive_terminate(node);
- 				      return -1;
- 				    }
- 				  }
- 				  break;
- 				case 1:
- 				  PQputline(loc_dbconn, copybuf);
- 				  if (archive_dir) {
- 				    rc = archive_append_data(node, copybuf, strlen(copybuf));
- 				    if (rc < 0) {
- 				      PQclear(res2);
- 				      PQclear(res1);
- 				      slon_disconnectdb(pro_conn);
- 				      dstring_free(&query1);
- 				      dstring_free(&query2);
- 				      dstring_free(&query3);
- 				      dstring_free(&lsquery);
- 				      dstring_free(&indexregenquery);
- 				      archive_terminate(node);
- 				      return -1;
- 				    }
- 
- 				  }
- 				  break;
- 
- 				}
- 			}
- 		}
- 		PQputline(loc_dbconn, "\\.\n");
- 		if (archive_dir)
- 		{
- 			rc = archive_append_str(node, "\\.");
- 			if (rc < 0) {
- 			  PQclear(res2);
- 			  PQclear(res1);
- 			  slon_disconnectdb(pro_conn);
- 			  dstring_free(&query1);
- 			  dstring_free(&query2);
- 			  dstring_free(&query3);
- 			  dstring_free(&lsquery);
- 			  dstring_free(&indexregenquery);
- 			  archive_terminate(node);
- 			  return -1;
- 			}
- 		}
- 
- 		/*
- 		 * End the COPY to stdout on the provider
- 		 */
- 		if (PQendcopy(pro_dbconn) != 0)
- 		{
- 			slon_log(SLON_ERROR, "remoteWorkerThread_%d: "
- 					 "PGendcopy() on provider%s",
- 					 node->no_id, PQerrorMessage(pro_dbconn));
- 			PQclear(res3);
- 			PQendcopy(loc_dbconn);
- 			PQclear(res2);
- 			PQclear(res1);
- 			slon_disconnectdb(pro_conn);
- 			dstring_free(&query1);
- 			dstring_free(&query2);
- 			dstring_free(&query3);
- 			dstring_free(&lsquery);
- 			dstring_free(&indexregenquery);
- 			archive_terminate(node);
- 			return -1;
- 		}
- 		PQclear(res3);
- 
- 		/*
- 		 * Check that the COPY to stdout on the provider node finished
- 		 * successful.
- 		 */
- 
- 		/*
- 		 * End the COPY from stdin on the local node with success
- 		 */
- 		if (PQendcopy(loc_dbconn) != 0)
- 		{
- 			slon_log(SLON_ERROR, "remoteWorkerThread_%d: "
- 					 "PGendcopy() on local DB%s",
- 					 node->no_id, PQerrorMessage(loc_dbconn));
- 			PQclear(res2);
- 			PQclear(res1);
- 			slon_disconnectdb(pro_conn);
- 			dstring_free(&query1);
- 			dstring_free(&query2);
- 			dstring_free(&query3);
- 			dstring_free(&lsquery);
- 			dstring_free(&indexregenquery);
- 			archive_terminate(node);
- 			return -1;
- 		}
- #endif   /* HAVE_PQPUTCOPYDATA */
  
  		PQclear(res2);
--- 3074,3077 ----