On Sat, 17 Mar 2007, Steve Singer wrote:

The attached patch replaces the one I sent last week.  This version has the 
only_on execute script executed directly against the subscriber node by 
slonik as discussed on slony1-general.

Also per the discussion on general this patch also includes a fix so that 
subscribers that do not subscribe to the set a script is submitted against 
do not receive the script (As was the behaviour in the 1.1 series)

These patches are all against the 1.2 branch.


Steve

> On Sat, 17 Mar 2007, Steve Singer wrote:
>
> The attached patch against 1.2 should fix this bug by not
> executing your DDL on the event node.
>
> If your DDL has errors then you won't find out about it until slon tries to 
> execute it on the other node (slonik won't give you an error but making 
> changes just on your subscribers can have the potential for all sorts of 
> trouble if your not careful)
>
>
>
>> On Sat, 17 Mar 2007, Mikko Partio wrote:
>> 
>> Your getting this because your EVENT_NODE is 1 but you only want to execute 
>> the script on 2.
>> 
>> Slonik probably should have a check to see if you have specified an only 
>> excecute different than your event node and just submit the script into the 
>> queue at that stage.
>> 
>> Another option is to require the event node be equal to the only execute 
>> node.
>> 
>> 
>> 
>>> Hi,
>>> 
>>> slonik's EXECUTE SCRIPT -documentation says that:
>>> 
>>> EXECUTE ONLY ON = ival
>>> 
>>>   (Optional) The ID of the only node to actually execute the script.
>>>   This option causes the script to be propagated by all nodes but
>>>   executed only by one. The default is to execute the script on all
>>>   nodes that are subscribed to the set.
>>> 
>>> 
>>> In my experience this property is not working correctly, and here's the 
>>> proof ("tiuhti" is origin and "viuhti" subscriber):
>>> 
>>> slony1 at tiuhti:~$ psql -d cldb -c "CREATE TABLE testtable (id int)" -h 
>>> tiuhti
>>> CREATE TABLE
>>> slony1 at tiuhti:~$ psql -d cldb -c "CREATE TABLE testtable (id int)" -h 
>>> viuhti
>>> CREATE TABLE
>>> 
>>> slony1 at tiuhti:~$ cat drop_table_testtable.sql
>>> DROP TABLE testtable;
>>> 
>>> slony1 at tiuhti:~$ cat droptest.slonik
>>> #!/usr/bin/slonik
>>> 
>>> CLUSTER NAME=climate;
>>> 
>>> NODE 1 ADMIN CONNINFO = 'dbname=cldb host=tiuhti user=slony1';
>>> NODE 2 ADMIN CONNINFO = 'dbname=cldb host=viuhti user=slony1';
>>> 
>>> EXECUTE SCRIPT (
>>>       SET ID = 1,
>>>       FILENAME = '/home/slony1/drop_table_testtable.sql',
>>>       EVENT NODE = 1,
>>>       EXECUTE ONLY ON = 2
>>> );
>>> 
>>> slony1 at tiuhti:~$ slonik droptest.slonik
>>> DDL script consisting of 1 SQL statements
>>> DDL Statement 0: (0,21) [DROP TABLE testtable;]
>>> Submit DDL Event to subscribers...
>>> DDL on origin - PGRES_TUPLES_OK
>>> 
>>> slony1 at tiuhti:~$ psql -d cldb -c "\d testtable" -h viuhti
>>> Did not find any relation named "testtable".
>>> 
>>> This is what I expected, but
>>> 
>>> slony1 at tiuhti:~$ psql -d cldb -c "\d testtable" -h tiuhti
>>> Did not find any relation named "testtable".
>>> 
>>> Wooah - the script dropped table testtable from both nodes although I 
>>> specified the "execute only on" -option. Is there something I'm missing or 
>>> is there a bug?
>>> 
>>> Regards
>>> 
>>> MP
>>> 
>>> _______________________________________________
>>> Slony1-general mailing list
>>> Slony1-general at lists.slony.info
>>> http://lists.slony.info/mailman/listinfo/slony1-general
>>> 
>> 
>> _______________________________________________
>> Slony1-general mailing list
>> Slony1-general at lists.slony.info
>> http://lists.slony.info/mailman/listinfo/slony1-general
>> 
>
-------------- next part --------------
Index: src/backend/slony1_funcs.sql
===================================================================
RCS file: /slony1/slony1-engine/src/backend/slony1_funcs.sql,v
retrieving revision 1.98.2.13
diff -c -r1.98.2.13 slony1_funcs.sql
*** src/backend/slony1_funcs.sql	22 Mar 2007 20:41:27 -0000	1.98.2.13
--- src/backend/slony1_funcs.sql	24 Mar 2007 22:24:19 -0000
***************
*** 3683,3690 ****
--- 3683,3693 ----
  	-- ----
  	lock table @NAMESPACE at .sl_config_lock;
  
+ 	
  	-- ----
  	-- Check that the set exists and originates here
+ 	-- unless only_on_node was specified (then it can be applied to
+ 	-- that node because that is what the user wanted)
  	-- ----
  	select set_origin into v_set_origin
  			from @NAMESPACE at .sl_set
***************
*** 3693,3699 ****
  	if not found then
  		raise exception ''Slony-I: set % not found'', p_set_id;
  	end if;
! 	if v_set_origin <> @NAMESPACE at .getLocalNodeId(''_ at CLUSTERNAME@'') then
  		raise exception ''Slony-I: set % does not originate on local node'',
  				p_set_id;
  	end if;
--- 3696,3703 ----
  	if not found then
  		raise exception ''Slony-I: set % not found'', p_set_id;
  	end if;
! 	if v_set_origin <> @NAMESPACE at .getLocalNodeId(''_ at CLUSTERNAME@'') AND
! 	p_only_on_node=-1 then
  		raise exception ''Slony-I: set % does not originate on local node'',
  				p_set_id;
  	end if;
***************
*** 5904,5907 ****
  to specify fields for the passed-in tab_id.  
  
  In PG versions > 7.3, this looks like (field1,field2,...fieldn)';
- 
--- 5908,5910 ----
Index: src/slon/remote_worker.c
===================================================================
RCS file: /slony1/slony1-engine/src/slon/remote_worker.c,v
retrieving revision 1.124.2.12
diff -c -r1.124.2.12 remote_worker.c
*** src/slon/remote_worker.c	6 Mar 2007 18:47:45 -0000	1.124.2.12
--- src/slon/remote_worker.c	24 Mar 2007 22:24:22 -0000
***************
*** 268,273 ****
--- 268,276 ----
  
  static void compress_actionseq(const char *ssy_actionseq, SlonDString * action_subquery);
  
+ static int process_ddl_script(SlonWorkMsg_event * event,SlonNode * node,
+ 							  PGconn * local_dbconn, char * seqbuf );
+ static int check_set_subscriber(int set_id, int node_id,PGconn * local_dbconn);
  
  /* ----------
   * slon_remoteWorkerThread
***************
*** 1339,1439 ****
  			}
  			else if (strcmp(event->ev_type, "DDL_SCRIPT") == 0)
  			{
! 				int			ddl_setid = (int)strtol(event->ev_data1, NULL, 10);
! 				char	   *ddl_script = event->ev_data2;
! 				int			ddl_only_on_node = (int)strtol(event->ev_data3, NULL, 10);
! 				int num_statements = -1, stmtno;
! 
! 				PGresult *res;
! 				ExecStatusType rstat;
! 
! 
! 				slon_appendquery(&query1,
! 						 "select %s.ddlScript_prepare_int(%d, %d); ",
! 						 rtcfg_namespace,
! 						 ddl_setid, ddl_only_on_node);
! 
! 				if (query_execute(node, local_dbconn, &query1) < 0) {
! 						slon_log(SLON_ERROR, "remoteWorkerThread_%d: DDL preparation failed - set %d - only on node %\n",
! 							 node->no_id, ddl_setid, ddl_only_on_node);
! 						slon_retry();
! 				}
! 
! 				num_statements = scan_for_statements (ddl_script);
! 				slon_log(SLON_CONFIG, "remoteWorkerThread_%d: DDL request with %d statements\n",
! 					 node->no_id, num_statements);
! 				if ((num_statements < 0) || (num_statements >= MAXSTATEMENTS)) {
! 					slon_log(SLON_ERROR, "remoteWorkerThread_%d: DDL had invalid number of statements - %d\n", 
! 						 node->no_id, num_statements);
! 					slon_retry();
! 				}
! 				
! 				for (stmtno=0; stmtno < num_statements;  stmtno++) {
! 					int startpos, endpos;
! 					char *dest;
! 					if (stmtno == 0)
! 						startpos = 0;
! 					else
! 						startpos = STMTS[stmtno-1];
! 
! 					endpos = STMTS[stmtno];
! 					dest = (char *) malloc (endpos - startpos + 1);
! 					if (dest == 0) {
! 						slon_log(SLON_ERROR, "remoteWorkerThread_%d: malloc() failure in DDL_SCRIPT - could not allocate %d bytes of memory\n", 
! 							 node->no_id, endpos - startpos + 1);
! 						slon_retry();
! 					}
! 					strncpy(dest, ddl_script + startpos, endpos-startpos);
! 					dest[STMTS[stmtno]-startpos] = 0;
! 					slon_mkquery(&query1, dest);
! 					slon_log(SLON_CONFIG, "remoteWorkerThread_%d: DDL Statement %d: [%s]\n", 
! 						 node->no_id, stmtno, dest);						 
! 					free(dest);
! 
! 					res = PQexec(local_dbconn, dstring_data(&query1));
! 
! 					if (PQresultStatus(res) != PGRES_COMMAND_OK && 
! 					    PQresultStatus(res) != PGRES_TUPLES_OK &&
! 					    PQresultStatus(res) != PGRES_EMPTY_QUERY)
! 					{
! 						rstat = PQresultStatus(res);
! 						slon_log(SLON_ERROR, "DDL Statement failed - %s\n", PQresStatus(rstat));
! 						dstring_free(&query1);
! 						slon_retry();
! 					}
! 					rstat = PQresultStatus(res);
! 					slon_log (SLON_CONFIG, "DDL success - %s\n", PQresStatus(rstat));
! 				}
! 	
! 				slon_mkquery(&query1, "select %s.ddlScript_complete_int(%d, %d); ", 
! 					     rtcfg_namespace,
! 					     ddl_setid,
! 					     ddl_only_on_node);
! 
! 				/* DDL_SCRIPT needs to be turned into a log shipping script */
! 				/* Note that the issue about parsing that mandates breaking 
! 				   up compound statements into
! 				   individually-processed statements does not apply to log
! 				   shipping as psql parses and processes each statement
! 				   individually */
! 
! 				if (archive_dir)
! 				{
! 					if ((ddl_only_on_node < 1) || (ddl_only_on_node == rtcfg_nodeid))
! 					{
! 
! 						if (archive_open(node, seqbuf) < 0)
! 							slon_retry();
! 						if (archive_tracking(node, rtcfg_namespace, 
! 								ddl_setid, seqbuf, seqbuf, 
! 								event->ev_timestamp_c) < 0)
! 							slon_retry();
! 						if (archive_append_str(node, ddl_script) < 0)
! 							slon_retry();
! 						if (archive_close(node) < 0)
! 							slon_retry();
! 					}
! 				}
  			}
  			else if (strcmp(event->ev_type, "RESET_CONFIG") == 0)
  			{
--- 1342,1348 ----
  			}
  			else if (strcmp(event->ev_type, "DDL_SCRIPT") == 0)
  			{
! 				process_ddl_script(event,node,local_dbconn,seqbuf);
  			}
  			else if (strcmp(event->ev_type, "RESET_CONFIG") == 0)
  			{
***************
*** 6097,6099 ****
--- 6006,6175 ----
  	}
  	slon_log(SLON_DEBUG4, " compressed actionseq subquery... %s\n", dstring_data(action_subquery));
  }
+ 
+ 
+ /**
+  *
+  * Process a ddl_script command.
+  */
+ static int process_ddl_script(SlonWorkMsg_event * event,SlonNode * node,
+ 							  PGconn * local_dbconn,
+ 							  char * seqbuf) 
+ {
+ 	int			ddl_setid = (int)strtol(event->ev_data1, NULL, 10);
+ 	char	   *ddl_script = event->ev_data2;
+ 	int			ddl_only_on_node = (int)strtol(event->ev_data3, NULL, 10);
+ 	int num_statements = -1, stmtno;
+ 	int node_in_set;
+ 	int localNodeId;
+ 	PGresult *res;
+ 	ExecStatusType rstat;
+ 	SlonDString query1;
+ 
+ 	
+ 
+ 	dstring_init(&query1);
+ 	/**
+ 	 * Check to make sure this node is part of the set
+ 	 */
+ 	slon_log(SLON_INFO, "Checking local node id");
+ 	localNodeId = db_getLocalNodeId(local_dbconn);
+ 	slon_log(SLON_INFO,"Found local node id");
+ 	node_in_set = check_set_subscriber(ddl_setid,localNodeId,local_dbconn);
+ 	
+ 	if(!node_in_set) {
+ 		/**
+ 		 *
+ 		 * Node is not part of the set.  
+ 		 * Do not forward teh DDL to the node,
+ 		 * nor should it be included in the log for log-shipping.
+ 		 */
+ 		slon_log(SLON_INFO,"Not forwarding DDL to node %d for set %d\n",
+ 				 node->no_id,ddl_setid);
+ 		
+ 	}
+ 	else 
+ 	{
+ 		slon_appendquery(&query1,
+ 						 "select %s.ddlScript_prepare_int(%d, %d); ",
+ 						 rtcfg_namespace,
+ 						 ddl_setid, ddl_only_on_node);
+ 		
+ 		if (query_execute(node, local_dbconn, &query1) < 0) {
+ 			slon_log(SLON_ERROR, "remoteWorkerThread_%d: DDL preparation failed - set %d - only on node %\n",
+ 					 node->no_id, ddl_setid, ddl_only_on_node);			
+ 			slon_retry();
+ 		}
+ 		
+ 		num_statements = scan_for_statements (ddl_script);
+ 		slon_log(SLON_CONFIG, "remoteWorkerThread_%d: DDL request with %d statements\n",
+ 				 node->no_id, num_statements);
+ 		if ((num_statements < 0) || (num_statements >= MAXSTATEMENTS)) {
+ 			slon_log(SLON_ERROR, "remoteWorkerThread_%d: DDL had invalid number of statements - %d\n", 
+ 					 node->no_id, num_statements);
+ 			slon_retry();
+ 		}
+ 		
+ 		for (stmtno=0; stmtno < num_statements;  stmtno++) {
+ 			int startpos, endpos;
+ 			char *dest;
+ 			if (stmtno == 0)
+ 				startpos = 0;
+ 			else
+ 				startpos = STMTS[stmtno-1];
+ 			
+ 			endpos = STMTS[stmtno];
+ 			dest = (char *) malloc (endpos - startpos + 1);
+ 			if (dest == 0) {
+ 				slon_log(SLON_ERROR, "remoteWorkerThread_%d: malloc() failure in DDL_SCRIPT - could not allocate %d bytes of memory\n", 
+ 						 node->no_id, endpos - startpos + 1);
+ 				slon_retry();
+ 			}
+ 			strncpy(dest, ddl_script + startpos, endpos-startpos);
+ 			dest[STMTS[stmtno]-startpos] = 0;
+ 			slon_mkquery(&query1, dest);
+ 			slon_log(SLON_CONFIG, "remoteWorkerThread_%d: DDL Statement %d: [%s]\n", 
+ 					 node->no_id, stmtno, dest);						 
+ 			free(dest);
+ 			
+ 			res = PQexec(local_dbconn, dstring_data(&query1));
+ 			
+ 			if (PQresultStatus(res) != PGRES_COMMAND_OK && 
+ 				PQresultStatus(res) != PGRES_TUPLES_OK &&
+ 				PQresultStatus(res) != PGRES_EMPTY_QUERY)
+ 				{
+ 					rstat = PQresultStatus(res);
+ 					slon_log(SLON_ERROR, "DDL Statement failed - %s\n", PQresStatus(rstat));
+ 					dstring_free(&query1);
+ 					slon_retry();
+ 				}
+ 			rstat = PQresultStatus(res);
+ 			slon_log (SLON_CONFIG, "DDL success - %s\n", PQresStatus(rstat));
+ 		}
+ 		
+ 		slon_mkquery(&query1, "select %s.ddlScript_complete_int(%d, %d); ", 
+ 					 rtcfg_namespace,
+ 					 ddl_setid,
+ 					 ddl_only_on_node);
+ 		
+ 		/* DDL_SCRIPT needs to be turned into a log shipping script */
+ 		/* Note that the issue about parsing that mandates breaking 
+ 		   up compound statements into
+ 		   individually-processed statements does not apply to log
+ 		   shipping as psql parses and processes each statement
+ 		   individually */
+ 		
+ 		if (archive_dir)
+ 			{
+ 				if ((ddl_only_on_node < 1) || (ddl_only_on_node == rtcfg_nodeid))
+ 					{
+ 						
+ 						if (archive_open(node, seqbuf) < 0)
+ 							slon_retry();
+ 						if (archive_tracking(node, rtcfg_namespace, 
+ 											 ddl_setid, seqbuf, seqbuf, 
+ 											 event->ev_timestamp_c) < 0)
+ 							slon_retry();
+ 						if (archive_append_str(node, ddl_script) < 0)
+ 							slon_retry();
+ 						if (archive_close(node) < 0)
+ 							slon_retry();
+ 					}
+ 			}
+ 	}/*else node a subscriber */
+ 	
+ 	dstring_free(&query1);
+ 
+ }
+ 
+ /**
+  * Checks to see if the node specified is a member of the set.
+  *
+  */
+ static int check_set_subscriber(int set_id, int node_id,PGconn * local_dbconn) 
+ {
+   
+   
+   SlonDString query1;
+   PGresult* res;
+   dstring_init(&query1);
+ 
+   slon_appendquery(&query1,"select 1 from %s.sl_subscribe WHERE sub_set=%d AND sub_receiver=%d for update"
+ 	       ,rtcfg_namespace,set_id,node_id);
+   res = PQexec(local_dbconn,dstring_data(&query1));
+   if(PQresultStatus(res)!=PGRES_TUPLES_OK) {
+     slon_log(SLON_ERROR,"remoteWorkerThread_%d: DDL preperation can not check set membership"
+ 	     ,node_id);
+ 	dstring_free(&query1);
+     slon_retry();
+   }
+   dstring_free(&query1);
+   if(PQntuples(res)==0) {
+     PQclear(res);
+     return 0;
+   }
+   PQclear(res);
+   return 1;
+ 
+ 
+ }
Index: src/slonik/slonik.c
===================================================================
RCS file: /slony1/slony1-engine/src/slonik/slonik.c,v
retrieving revision 1.67.2.6
diff -c -r1.67.2.6 slonik.c
*** src/slonik/slonik.c	15 Mar 2007 18:52:02 -0000	1.67.2.6
--- src/slonik/slonik.c	24 Mar 2007 22:24:24 -0000
***************
*** 3849,3861 ****
  	PGresult *res;
  	ExecStatusType rstat;
  
  #define PARMCOUNT 1  
  
          const char *params[PARMCOUNT];
          int paramlens[PARMCOUNT];
          int paramfmts[PARMCOUNT];
  
! 	adminfo1 = get_active_adminfo((SlonikStmt *) stmt, stmt->ev_origin);
  	if (adminfo1 == NULL)
  		return -1;
  
--- 3849,3865 ----
  	PGresult *res;
  	ExecStatusType rstat;
  
+ 
  #define PARMCOUNT 1  
  
          const char *params[PARMCOUNT];
          int paramlens[PARMCOUNT];
          int paramfmts[PARMCOUNT];
  
! 	if(stmt->only_on_node>-1) {
! 		adminfo1 = get_active_adminfo((SlonikStmt*) stmt,stmt->only_on_node);
! 	}
! 	  adminfo1 = get_active_adminfo((SlonikStmt *) stmt, stmt->ev_origin);
  	if (adminfo1 == NULL)
  		return -1;
  
***************
*** 3935,3967 ****
  		/* printf ("Success - %s\n", PQresStatus(rstat)); */
  	}
  	
! 	printf("Submit DDL Event to subscribers...\n");
! 
! 	slon_mkquery(&query, "select \"_%s\".ddlScript_complete(%d, $1::text, %d); ", 
! 		     stmt->hdr.script->clustername,
! 		     stmt->ddl_setid,  
! 		     stmt->only_on_node);
! 
! 	paramlens[PARMCOUNT-1] = 0;
! 	paramfmts[PARMCOUNT-1] = 0;
! 	params[PARMCOUNT-1] = dstring_data(&script);
! 
! 	res = PQexecParams(adminfo1->dbconn, dstring_data(&query), PARMCOUNT,
! 			   NULL, params, paramlens, paramfmts, 0);
! 	
! 	if (PQresultStatus(res) != PGRES_COMMAND_OK && 
! 	    PQresultStatus(res) != PGRES_TUPLES_OK &&
! 	    PQresultStatus(res) != PGRES_EMPTY_QUERY)
! 	{
! 		rstat = PQresultStatus(res);
! 		printf("Event submission for DDL failed - %s\n", PQresStatus(rstat));
! 		dstring_free(&query);
! 		return -1;
! 	} else {
! 		rstat = PQresultStatus(res);
! 		printf ("DDL on origin - %s\n", PQresStatus(rstat));
! 	}
! 	
  	dstring_free(&script);
  	dstring_free(&query);
  	return 0;
--- 3939,3972 ----
  		/* printf ("Success - %s\n", PQresStatus(rstat)); */
  	}
  	
! 	if(stmt->only_on_node==-1) {
! 		printf("Submit DDL Event to subscribers...\n");
! 		
! 		slon_mkquery(&query, "select \"_%s\".ddlScript_complete(%d, $1::text, %d); ", 
! 					 stmt->hdr.script->clustername,
! 					 stmt->ddl_setid,  
! 					 stmt->only_on_node);
! 		
! 		paramlens[PARMCOUNT-1] = 0;
! 		paramfmts[PARMCOUNT-1] = 0;
! 		params[PARMCOUNT-1] = dstring_data(&script);
! 		
! 		res = PQexecParams(adminfo1->dbconn, dstring_data(&query), PARMCOUNT,
! 						   NULL, params, paramlens, paramfmts, 0);
! 		
! 		if (PQresultStatus(res) != PGRES_COMMAND_OK && 
! 			PQresultStatus(res) != PGRES_TUPLES_OK &&
! 			PQresultStatus(res) != PGRES_EMPTY_QUERY)
! 			{
! 				rstat = PQresultStatus(res);
! 				printf("Event submission for DDL failed - %s\n", PQresStatus(rstat));
! 				dstring_free(&query);
! 				return -1;
! 			} else {
! 				rstat = PQresultStatus(res);
! 				printf ("DDL on origin - %s\n", PQresStatus(rstat));
! 			}
! 	}/*only on node*/
  	dstring_free(&script);
  	dstring_free(&query);
  	return 0;