text: -- upgrade catalogue script 2.0.5 to 2.0.6
  CREATE OR REPLACE FUNCTION sch_chameleon.fn_replay_mysql(integer,integer,boolean)
  RETURNS sch_chameleon.ty_replay_status AS
  $BODY$
  	DECLARE
  		p_i_max_events		ALIAS FOR $1;
  		p_i_id_source		ALIAS FOR $2;
  		p_b_exit_on_error	ALIAS FOR $3;
  		v_ty_status		sch_chameleon.ty_replay_status;
  		v_r_statements		record;
  		v_i_id_batch		bigint;
  		v_v_log_table		text;
  		v_t_ddl			text;
  		v_t_main_sql		text;
  		v_t_delete_sql		text;
  		v_i_replayed		integer;
  		v_i_skipped		integer;
  		v_i_ddl			integer;
  		v_i_evt_replay		bigint[];
  		v_i_evt_queue		bigint[];
  		v_ts_evt_source		timestamp without time zone;
  		v_tab_enabled		boolean;
  
  	BEGIN
  		v_i_replayed:=0;
  		v_i_ddl:=0;
  		v_i_skipped:=0;
  		v_ty_status.b_continue:=FALSE;
  		v_ty_status.b_error:=FALSE;		
  		RAISE DEBUG 'Searching batches to replay for source id: %', p_i_id_source;
  		v_i_id_batch:= (
  			SELECT 
  				bat.i_id_batch 
  			FROM 
  				sch_chameleon.t_replica_batch bat
  				INNER JOIN  sch_chameleon.t_batch_events evt
  				ON
  					evt.i_id_batch=bat.i_id_batch
  			WHERE 
  					bat.b_started 
  				AND	bat.b_processed 
  				AND	NOT bat.b_replayed
  				AND	bat.i_id_source=p_i_id_source
  			ORDER BY 
  				bat.ts_created 
  			LIMIT 1
  			)
  		;
  
  		v_v_log_table:=(
  			SELECT 
  				v_log_table
  			FROM 
  				sch_chameleon.t_replica_batch 
  			WHERE 
  				i_id_batch=v_i_id_batch
  			)
  		;
  		IF v_i_id_batch IS NULL 
  		THEN
  			RAISE DEBUG 'There are no batches available for replay';
  			RETURN v_ty_status;
  		END IF;
  		
  		RAISE DEBUG 'Found id_batch %, data in log table %', v_i_id_batch,v_v_log_table;
  		RAISE DEBUG 'Building a list of event id with max length %...', p_i_max_events;
  		v_i_evt_replay:=(
  			SELECT 
  				i_id_event[1:p_i_max_events] 
  			FROM 
  				sch_chameleon.t_batch_events 
  			WHERE 
  				i_id_batch=v_i_id_batch
  		);
  		
  		
  		v_i_evt_queue:=(
  			SELECT 
  				i_id_event[p_i_max_events+1:array_length(i_id_event,1)] 
  			FROM 
  				sch_chameleon.t_batch_events 
  			WHERE 
  				i_id_batch=v_i_id_batch
  		);
  		
  		RAISE DEBUG 'Finding the last executed event''s timestamp...';
  		v_ts_evt_source:=(
  			SELECT 
  				to_timestamp(i_my_event_time)
  			FROM	
  				sch_chameleon.t_log_replica
  			WHERE
  					i_id_event=v_i_evt_replay[array_length(v_i_evt_replay,1)]
  				AND	i_id_batch=v_i_id_batch
  		);
  		
  		RAISE DEBUG 'Generating the main loop sql';
  
  		v_t_main_sql:=format('
  			SELECT 
  				i_id_event AS i_id_event,
  				enm_binlog_event,
  				(enm_binlog_event=''ddl'')::integer as i_ddl,
  				(enm_binlog_event<>''ddl'')::integer as i_replay,
  				t_binlog_name,
  				i_binlog_position,
  				v_table_name,
  				v_schema_name,
  				t_pk_data,
  				CASE
  					WHEN enm_binlog_event = ''ddl''
  					THEN 
  						t_query
  					WHEN enm_binlog_event = ''insert''
  					THEN
  						format(
  							''INSERT INTO %%I.%%I %%s;'',
  							v_schema_name,
  							v_table_name,
  							t_dec_data
  							
  						)
  					WHEN enm_binlog_event = ''update''
  					THEN
  						format(
  							''UPDATE %%I.%%I SET %%s WHERE %%s;'',
  							v_schema_name,
  							v_table_name,
  							t_dec_data,
  							t_pk_data
  						)
  					WHEN enm_binlog_event = ''delete''
  					THEN
  						format(
  							''DELETE FROM %%I.%%I WHERE %%s;'',
  							v_schema_name,
  							v_table_name,
  							t_pk_data
  						)
  					
  				END AS t_sql
  			FROM 
  			(
  				SELECT 
  					pk.i_id_event,
  					pk.v_table_name,
  					pk.v_schema_name,
  					pk.enm_binlog_event,
  					pk.t_binlog_name,
  					pk.i_binlog_position,
  					pk.t_query as t_query,
  					pk.ts_event_datetime,
  					pk.t_dec_data,
  					string_agg(DISTINCT 
  							CASE
  								WHEN pk.v_table_pkey IS NOT NULL
  								THEN
  									format(
  										''%%I=%%L'',
  										pk.v_table_pkey,
  										CASE 
  											WHEN pk.enm_binlog_event = ''update''
  											THEN
  												pk.jsb_event_before->>v_table_pkey
  											ELSE
  												pk.jsb_event_after->>v_table_pkey
  										END 	
  								
  									)
  							END
  					,'' AND '') as  t_pk_data
  					
  				FROM
  				(
  					SELECT 
  						dec.i_id_event,
  						dec.v_table_name,
  						dec.v_schema_name,
  						dec.enm_binlog_event,
  						dec.t_binlog_name,
  						dec.i_binlog_position,
  						dec.t_query as t_query,
  						dec.ts_event_datetime,
  						CASE
  							WHEN dec.enm_binlog_event = ''insert''
  							THEN
  							format(''(%%s) VALUES (%%s)'',string_agg(format(''%%I'',dec.t_column),'',''),string_agg(format(''%%L'',dec.jsb_event_after->>t_column),'',''))
  							WHEN dec.enm_binlog_event = ''update''
  							THEN
  								string_agg(format(''%%I=%%L'',dec.t_column,dec.jsb_event_after->>t_column),'','')
  							
  						END AS t_dec_data,
  						unnest(v_table_pkey) as v_table_pkey,
  						dec.jsb_event_after,
  						dec.jsb_event_before
  						
  					FROM 
  					(
  						SELECT 
  							log.i_id_event,
  							log.v_table_name,
  							log.v_schema_name,
  							log.enm_binlog_event,
  							log.t_binlog_name,
  							log.i_binlog_position,
  							coalesce(log.jsb_event_after,''{"foo":"bar"}''::jsonb) as jsb_event_after,
  							(jsonb_each_text(coalesce(log.jsb_event_after,''{"foo":"bar"}''::jsonb))).key AS t_column,
  							log.jsb_event_before,
  							log.t_query as t_query,
  							log.ts_event_datetime,
  							v_table_pkey
  						FROM 
  							sch_chameleon.%I log
  							INNER JOIN sch_chameleon.t_replica_tables tab
  								ON 
  										tab.v_table_name=log.v_table_name
  									AND	tab.v_schema_name=log.v_schema_name
  						WHERE
  								tab.b_replica_enabled
  							AND	i_id_event = ANY(%L)
  						
  					) dec
  					GROUP BY 
  						dec.i_id_event,
  						dec.v_table_name,
  						dec.v_schema_name,
  						dec.enm_binlog_event,
  						dec.t_query,
  						dec.ts_event_datetime,
  						dec.t_binlog_name,
  						dec.i_binlog_position,
  						dec.v_table_pkey,
  						dec.jsb_event_after,
  						dec.jsb_event_before
  					) pk
  				GROUP BY 
  					pk.i_id_event,
  					pk.v_table_name,
  					pk.v_schema_name,
  					pk.enm_binlog_event,
  					pk.t_binlog_name,
  					pk.i_binlog_position,
  					pk.t_query,
  					pk.ts_event_datetime,
  					pk.t_dec_data
  				
  					
  			) par
  			ORDER BY 
  				i_id_event ASC
  			;		
  		',v_v_log_table,v_i_evt_replay);
  		RAISE DEBUG '%',v_t_main_sql;
  		FOR v_r_statements IN EXECUTE v_t_main_sql
  		LOOP
  			
  			BEGIN
  				EXECUTE v_r_statements.t_sql;
  				v_i_ddl:=v_i_ddl+v_r_statements.i_ddl;
  				v_i_replayed:=v_i_replayed+v_r_statements.i_replay;
  				
  				
  			EXCEPTION
  				WHEN OTHERS
  				THEN
  				RAISE NOTICE 'An error occurred when replaying data for the table %.%',v_r_statements.v_schema_name,v_r_statements.v_table_name;
  				RAISE NOTICE 'SQLSTATE: % - ERROR MESSAGE %',SQLSTATE, SQLERRM;
  				RAISE DEBUG 'SQL EXECUTED: % ',v_r_statements.t_sql;
  				RAISE NOTICE 'The table %.% has been removed from the replica',v_r_statements.v_schema_name,v_r_statements.v_table_name;
  				v_ty_status.v_table_error:=array_append(v_ty_status.v_table_error, format('%I.%I SQLSTATE: %s - ERROR MESSAGE: %s',v_r_statements.v_schema_name,v_r_statements.v_table_name,SQLSTATE, SQLERRM)::character varying) ;
  				RAISE NOTICE 'Adding error log entry for table %.% ',v_r_statements.v_schema_name,v_r_statements.v_table_name;
  				INSERT INTO sch_chameleon.t_error_log
  							(
  								i_id_batch, 
  								i_id_source,
  								v_schema_name, 
  								v_table_name, 
  								t_table_pkey, 
  								t_binlog_name, 
  								i_binlog_position, 
  								ts_error, 
  								t_sql,
  								t_error_message
  							)
  							SELECT 
  								i_id_batch, 
  								p_i_id_source,
  								v_schema_name, 
  								v_table_name, 
  								v_r_statements.t_pk_data as t_table_pkey, 
  								t_binlog_name, 
  								i_binlog_position, 
  								clock_timestamp(), 
  								quote_literal(v_r_statements.t_sql) as t_sql,
  								format('%s - %s',SQLSTATE, SQLERRM) as t_error_message
  							FROM
  								sch_chameleon.t_log_replica  log
  							WHERE 
  								log.i_id_event=v_r_statements.i_id_event
  						;
  				IF p_b_exit_on_error
  				THEN
  					v_ty_status.b_continue:=FALSE;
  					v_ty_status.b_error:=TRUE;
  					RETURN v_ty_status;
  				ELSE
  				
  					RAISE NOTICE 'Statement %', v_r_statements.t_sql;
  					UPDATE sch_chameleon.t_replica_tables 
  						SET 
  							b_replica_enabled=FALSE
  					WHERE
  							v_schema_name=v_r_statements.v_schema_name
  						AND	v_table_name=v_r_statements.v_table_name
  					;
  
  					RAISE NOTICE 'Deleting the log entries for the table %.% ',v_r_statements.v_schema_name,v_r_statements.v_table_name;
  					DELETE FROM sch_chameleon.t_log_replica  log
  					WHERE
  							v_table_name=v_r_statements.v_table_name
  						AND	v_schema_name=v_r_statements.v_schema_name
  						AND 	i_id_batch=v_i_id_batch
  					;
  				END IF;
  			END;
  		END LOOP;
  		IF v_ts_evt_source IS NOT NULL
  		THEN
  			UPDATE sch_chameleon.t_last_replayed
  				SET
  					ts_last_replayed=v_ts_evt_source
  			WHERE 	
  				i_id_source=p_i_id_source
  			;
  		END IF;
  		IF v_i_replayed=0 AND v_i_ddl=0
  		THEN
  			DELETE FROM sch_chameleon.t_log_replica
  			WHERE
      			    i_id_batch=v_i_id_batch
  			;
  				
  			GET DIAGNOSTICS v_i_skipped = ROW_COUNT;
  			RAISE DEBUG 'SKIPPED ROWS: % ',v_i_skipped;
  
  			UPDATE ONLY sch_chameleon.t_replica_batch  
  			SET 
  				b_replayed=True,
  				i_skipped=v_i_skipped,
  				ts_replayed=clock_timestamp()
  				
  			WHERE
  				i_id_batch=v_i_id_batch
  			;
  
  			DELETE FROM sch_chameleon.t_batch_events
  			WHERE
  				i_id_batch=v_i_id_batch
  			;
  
  			v_ty_status.b_continue:=FALSE;
  		ELSE
  			UPDATE ONLY sch_chameleon.t_replica_batch  
  			SET 
  				i_ddl=coalesce(i_ddl,0)+v_i_ddl,
  				i_replayed=coalesce(i_replayed,0)+v_i_replayed,
  				i_skipped=v_i_skipped,
  				ts_replayed=clock_timestamp()
  				
  			WHERE
  				i_id_batch=v_i_id_batch
  			;
  
  			UPDATE sch_chameleon.t_batch_events
  				SET
  					i_id_event = v_i_evt_queue
  			WHERE
  				i_id_batch=v_i_id_batch
  			;
  
  			DELETE FROM sch_chameleon.t_log_replica
  			WHERE
  					i_id_batch=v_i_id_batch
  				AND 	i_id_event=ANY(v_i_evt_replay) 
  			;
  			v_ty_status.b_continue:=TRUE;
  			RETURN v_ty_status;
  		END IF;
  		v_i_id_batch:= (
  			SELECT 
  				bat.i_id_batch 
  			FROM 
  				sch_chameleon.t_replica_batch bat
  				INNER JOIN  sch_chameleon.t_batch_events evt
  				ON
  					evt.i_id_batch=bat.i_id_batch
  			WHERE 
  					bat.b_started 
  				AND	bat.b_processed 
  				AND	NOT bat.b_replayed
  				AND	bat.i_id_source=p_i_id_source
  			ORDER BY 
  				bat.ts_created 
  			LIMIT 1
  			)
  		;
  		
  		IF v_i_id_batch IS NOT NULL
  		THEN
  			v_ty_status.b_continue:=TRUE;
  		END IF;
  		
  		
  		RETURN v_ty_status;
  
  		
  		
  	END;
  	
  $BODY$
  LANGUAGE plpgsql;
  