diff --git a/RELEASE b/RELEASE index 08ff73e5..646a60f9 100644 --- a/RELEASE +++ b/RELEASE @@ -1,208 +1,59 @@ #+OPTIONS: ^:{} * Slony-I Release Notes - -** 2.1.1 - -- Bug #246 :: Include path order changes - -** 2.1.0 -** 2.1.0 RC 2 - -- Bug #236 :: Fix default formatting of timestamp in logs - -- Add support for PostgreSQL 9.1. Simply marking it a supported version - -- Bug #237 :: Drop set did not remove truncate triggers. - -** 2.1.0 RC 1 - -- Bug #226 :: Cleanup thread needed change to expected PQResultStatus - -- Bug #229 :: Fix to set add table so it won't implicitly wait for an event - inside of a transaction block. - -- Bug #230 :: log_timestamps was always treated as false - -- Bug #231 :: Fix to regex in tools/start_slon.sh - -- Bug #227 :: Do not try to add views to a set with regular expressions in - set add table. Also fixes for escape characters in the regex - -- Clearer error message on 'create set' when no origin is specified. - -- Clearer error message on 'lock set' when no origin is specified. - -- Bug #218 :: Additional fixes, obtain sl_event_lock before sl_config_lock - to prevent deadlock. -- Include additional C headers required for Postgresql 9.2(master) - -** 2.1.0 (beta 3) - -- Bug #159 :: distclean deletes the slony_logshipper flex/bison generated files. - -- Bug #204 :: Additional fixes for failover to a non-direct subscriber - where circular paths in sl_subscribe could happen - -- Bug #214 :: Add slonikconfdump.sh back into tools - -- Bug #216 :: Use EPOCH for the timestamp in the monitor thread - to avoid differences in how Win32 handles timezones. - -- Bug #218 :: Acquire lock on sl_config_lock in much the same fashion as - with sl_event_lock (bug #196) - -- Bug #219 :: Default to PostgreSQL 8.4 support if the version of PG is newer - than what slony recognizes. - -- Bug #220 :: Fix issue where changing the provider of an existing - subscriber passed parameters in the wrong order to the - stored procedure. - -- Bug #222 :: Fixing issue where new connections that call the - logtrigger for the first time might block until an - in progress cleanupEvent finishes. - -- Bug #223 :: Clear out the last saved event number when uninstalling - a node. This prevents slonik from waiting for a node it just - uninstalled. - -- Bug #224 :: altperl code misspelled "PKEYDTABLES" - -*** Some items without bug numbers: - -- Fixed slonik memory corruption when it tries to get the last known - event_id from a node that doesn't yet have slony installed. - -- Remove rules for building PDFs for the original design docs, in - favor of having the built docs stored in a new "historical - documentation" directory. - -** 2.1.0 (beta 2) -- Bug 219 :: Default to PostgreSQL 8.4 support if the version of PG is newer than - what slony recognizes. - -- Fix threading issue in monitor_thread where slon_init can be called - without the mutex being locked. This function was written to assume - the mutex lock exists. - -- datarootdir warning from autoconf fixed - -- Bug #36 - (Reapplied) don't link with pgport on non-Win32 platforms - -- Misc compiler warnings including no longer calling get_share_dir from - Postgres if PGSHARE is not defined. Configure should define it. - -- Regenerated complexenv.png diagram - -- Obtain a lock on sl_event_lock during MOVE_SET - processing in remote_worker. - -** 2.1.0 (beta 2) - -- Bug #204 :: Upgrade from 2.0.x fails under PostgreSQL 8.3, fixing - issues with failover and non-direct subscribers - -- Bug #205 :: Upgrade fails from 2.0.0, return type from cloneNode changes - -- Bug #206 :: Do not perform an accidental rollback in a try block - -- Bug #208 :: Fixing issues with bulk adding of tables. - -- Bug #209 :: Default table ID problem - -- Bug #210 :: merge set should wait for both sets to be subscribed - before submitting the merge event. - -- Bug #211 :: Modify the altperl tools so they don't enclose statements - that wait for events inside of a 'try' block. - -** 2.1.0 (beta 1) - -*** Some items without bug numbers: - - Introduced tests using the clustertest framework, consisting of two - sets of tests: - - clustertest/regression - port of the old tests in the tests directory - - clustertest/disorder - new simulated order entry system with a fair bit of concurrency - - - Configure and Makefile cleanups and improvements, including - - Configure looks in common locations for some of the DocBook - tooling, so it is less likely to require manual intervention on - some common distributions - - Pick up Groff automagically - - Script to do virtually all the work documented in the [[http://slony.info/documentation/2.0/releasechecklist.html][Release Checklist]] - - Documentation updates - - Reorganized sections - - Trend towards having version-specific documentation, rather than - trying to have it document all versions at once - - Git changes - - Numerous .gitignore files set up to ignore generated files - - Add a resetSession function that will reset the session state - cached in the C stored functions. - -*** Changes from 2.0 branch that were also applied to HEAD: - - Bugs 40, 42, 75, 78, 102, 104, 115, 117, 121, 122, 124, 127, 132, - 136, 138, 139, 141, 144, 147, 148, 149, 154, 162 - -*** Items tracked from Bugzilla -- Bug #79 :: --prefix and --exec-prefix options don't function as expected - -- Bug #90 :: Builds On Solaris with Sun C Studio fail. This also addresses - issues building with Microsoft Visual Studio. - -- Bug #126 :: TCP KEEP alives can now be used from slon to postgresql(on by default) - -- Bug #134 :: TRUNCATE trigger usage - -- Bug #142 :: better document minimal security profile - -- Bug #146 :: Removed option to use DELETE to trim sl_log_* tables - only support TRUNCATE - -- Bug #152 :: eliminate debugging statements from DDL processing - -- Bug #156 :: add health check at slon startup time - -- Bug #163 :: change Slony-I tables to use TIMESTAMP WITH TIME ZONE rather than WITHOUT TIME ZONE - -- Bug #167 :: change queries against sl_log_* tables to avert slowdown when large backlog - -- Bug #170 :: remove SNMP support - -- Bug #172 :: add support for application_name GUC (in recent PG such as 9.0) - -- Bug #175 :: add monitoring thread which stows activity information in sl_components - -- Bug #179 :: Implicit/Automatic wait for. Slonik will now often perform an automatic - 'wait for event' when required to avoid race conditions. - -- Bug #180 :: Created a slonik DATE command - -- Bug #181 :: Support for bulk adding of tables and sequences and automatically finding - the next available table or sequence id. - -- Bug #185 :: change functions to use named parameters - -- Bug #187 :: add newline to slon PID files - -- Bug #188 :: test harness query had subquery returning multiple tuples - -- Bug #189 :: add slonik EXIT check for exit values 0..255, as that's all that Unix accepts - -- Bug #193 :: add back reset of vac_count so vacuums only done every few cleanup thread iterations - -- Bug #195 :: change slon_quote_* functions to IMMUTABLE - -- Bug #196 :: add guard lock on sl_event_lock to protect against events with XID going backwards - -- Bug #198 :: subscribeset() should verify that provider node exists - -- Bug #201 :: Allow command-line options to be configured from slon_tools.conf - -- Bug #204 :: Fix issue with FAILOVER to a non-direct subscriber introduced in 2.0.5 - -- Bug #206 :: Do not perform an accidental rollback in a try block - -** RELEASE 2.0.6 -- Bug #162 :: typenameTypeId() changes signatures in 9.1 - - +** 2.2.0 (beta 1) +*** Significant New Features + - Switch to use COPY protocol for handling sl_log_* data + + Since the schema of tables ~sl_log_1~ and ~sl_log_2~ are changing + substantially, capturing somewhat different information, there is + no way provided to convert old log data to use the new schema. + As a consequence, at the time of upgrade, it is essential that + these tables be empty. + + They need to /already/ be empty, as the ~upgradeSchema()~ + function runs locally on each node, and there is no readily + available mechanism to recover from the problem of discovering + that one node still has data that needs to be applied against + another node. We may make some /best efforts/ to prevent + problems, but there is only limited cross-node coordination that + may readily be done. + + We intend to offer several mechanisms to help this to happen, + including: + + - Preventing new log entries via ~LOCK SET~ :: It is recommended + that administrators /prevent/ data from being added to the + logs accidentally by locking all replication sets using [[http://slony.info/documentation/2.1/stmtlockset.html][LOCK + SET]], then unlock them via ~UNLOCK SET~ afterwards. For + systems with enormous numbers of replicated tables (/e.g./ - + on the order of thousands), this requires taking out + exclusive locks on all of them, as ~LOCK SET~ and ~UNLOCK + SET~ both need to submit an ~ALTER TABLE~ request against + each table in order to add/drop the "lock" trigger, and may + prove too expensive. + + - Preventing new log entries via ~pg_hba.conf~ :: If ~LOCK SET~ + is too expensive, an administrator may prefer to lock out + all application users during the upgrade process. + ~pg_hba.conf~ is useful for this. The process would + involve: + - Altering ~pg_hba.conf~ to contain entries denying access to + all other than the ~slony~ user + - Submitting ~SIGHUP~ signal to the Postgres backend so new + connections are rejected + - Verifying, via query to ~pg_stat_activity~, that there are no + users other than the ~slony~ user connected + + - Script to drain log tables ~sl_log_1~ and ~sl_log_2~ :: It runs + the following logic against each node: + - Calls the function ~cleanupEvent()~ to clean out any log + entries that are safe to delete + - Verify that log tables ~sl_log_1~ and ~sl_log_2~ are empty + + Once this script has run against all nodes, reporting that + the log tables are empty on each node, it is safe to upgrade + the schema for these tables. If some nodes are not ready, + it will be necessary to keep ~slon~ processes running to + apply the outstanding log entries, and so that confirmations + are passed back to all nodes in order that the cleanup can + be finalized. diff --git a/UPGRADING b/UPGRADING index 308c3a20..b6b1c5dd 100644 --- a/UPGRADING +++ b/UPGRADING @@ -1,6 +1,10 @@ -Upgrading from an earlier version of Slony-I to 1.2 ------------------------------------------------------ +Upgrading Slony +--------------- +Here are notes about updating Slony to a newer version. + +Upgrading from an earlier version of Slony-I to current: general +------------------------------------------------------------------ When upgrading Slony-I, the Slony installation on all nodes in a cluster must be upgraded at once, using the slonik command UPDATE FUNCTIONS. @@ -25,52 +29,6 @@ The proper upgrade procedure is thus: 4. Start all slons. -Upgrading is also discussed in more detail in the "Admin Guide." -The section is entitled "Slony-I Upgrade", distributed as +Upgrading is also discussed in more detail in the "Admin Guide." The +section is entitled "Slony-I Upgrade", distributed as doc/adminguide/slonyupgrade.html in the documentation package. - -Nullable Column issue in version 1.2 -===================================== - -In version 1.2, a check has been added to reject subscriptions to -tables where the candidate primary key has columns that are "Nullable" -(e.g. - that are not defined in the schema as NOT NULL). - -This was not checked in earlier versions, and people may have set up -constraints (e.g. - via triggers) to enforce NOT NULL as opposed to -using the NOT NULL column constraint. As a result, a replica running -on version 1.1 may have violations of this check. - -You may search for these violations via the following query: - -select c.relname as table_name, ic.relname as index_name, att.attname, att2.attnotnull -from _cluster.sl_table t, pg_catalog.pg_class c, pg_index i, pg_catalog.pg_class ic, pg_catalog.pg_attribute att, pg_catalog.pg_attribute att2 -where t.tab_reloid = c.oid - and t.tab_idxname = ic.relname - and ic.oid = i.indexrelid - and att.attrelid = i.indexrelid - and att2.attname = att.attname - and att2.attrelid = c.oid - and att2.attnotnull = 'f'; - -The problem may be rectified by submitting, for each entry found, a -query of the form: - alter table mytable alter column nullablecol set not null; - -It would minimally suffice to apply this to the schema of new -subscribers; such queries will run rapidly against empty tables. - -It will take longer to apply this change to a table that already -contains a great deal of data, as the alteration will scan the table -to verify that there are no tuples where the column is NULL. - -TABLE ADD KEY change in version 2.0 -===================================== - -In version 2.0, Slony-I drops out support for creating -system-generated "pseudo primary keys" that involve adding in a custom -Slony-I-generated column. - -There is documentation in the Admin Guide (slonyupgrade.sgml, -slonyupgrade.html) explaining approaches to "cleaning up" after having -used TABLE ADD KEY. diff --git a/clustertest/conf/slon.1.conf.sample b/clustertest/conf/slon.1.conf.sample index a4e834a8..43b94f2f 100644 --- a/clustertest/conf/slon.1.conf.sample +++ b/clustertest/conf/slon.1.conf.sample @@ -35,20 +35,6 @@ sync_interval_timeout=2000 # Range: [0,100], default: 6 sync_group_maxsize=5 -# Size above which an sl_log_? row's log_cmddata is considered large. -# Up to 500 rows of this size are allowed in memory at once. Rows larger -# than that count into the sync_max_largemem space allocated and free'd -# on demand. -# Range: [1024,32768], default: 8192 -sync_max_rowsize=16384 - -# Maximum amount of memory allowed for large rows. Note that the algorithm -# will stop fetching rows AFTER this amount is exceeded, not BEFORE. This -# is done to ensure that a single row exceeding this limit alone does not -# stall replication. -# Range: [1048576,1073741824], default: 5242880 -sync_max_largemem=3276800 - # If this parameter is 1, messages go both to syslog and the standard # output. A value of 2 sends output only to syslog (some messages will # still go to the standard output/error). The default is 0, which means diff --git a/clustertest/conf/slon.2.conf.sample b/clustertest/conf/slon.2.conf.sample index a4e834a8..43b94f2f 100644 --- a/clustertest/conf/slon.2.conf.sample +++ b/clustertest/conf/slon.2.conf.sample @@ -35,20 +35,6 @@ sync_interval_timeout=2000 # Range: [0,100], default: 6 sync_group_maxsize=5 -# Size above which an sl_log_? row's log_cmddata is considered large. -# Up to 500 rows of this size are allowed in memory at once. Rows larger -# than that count into the sync_max_largemem space allocated and free'd -# on demand. -# Range: [1024,32768], default: 8192 -sync_max_rowsize=16384 - -# Maximum amount of memory allowed for large rows. Note that the algorithm -# will stop fetching rows AFTER this amount is exceeded, not BEFORE. This -# is done to ensure that a single row exceeding this limit alone does not -# stall replication. -# Range: [1048576,1073741824], default: 5242880 -sync_max_largemem=3276800 - # If this parameter is 1, messages go both to syslog and the standard # output. A value of 2 sends output only to syslog (some messages will # still go to the standard output/error). The default is 0, which means diff --git a/clustertest/conf/slon.3.conf.sample b/clustertest/conf/slon.3.conf.sample index a4e834a8..43b94f2f 100644 --- a/clustertest/conf/slon.3.conf.sample +++ b/clustertest/conf/slon.3.conf.sample @@ -35,20 +35,6 @@ sync_interval_timeout=2000 # Range: [0,100], default: 6 sync_group_maxsize=5 -# Size above which an sl_log_? row's log_cmddata is considered large. -# Up to 500 rows of this size are allowed in memory at once. Rows larger -# than that count into the sync_max_largemem space allocated and free'd -# on demand. -# Range: [1024,32768], default: 8192 -sync_max_rowsize=16384 - -# Maximum amount of memory allowed for large rows. Note that the algorithm -# will stop fetching rows AFTER this amount is exceeded, not BEFORE. This -# is done to ensure that a single row exceeding this limit alone does not -# stall replication. -# Range: [1048576,1073741824], default: 5242880 -sync_max_largemem=3276800 - # If this parameter is 1, messages go both to syslog and the standard # output. A value of 2 sends output only to syslog (some messages will # still go to the standard output/error). The default is 0, which means diff --git a/clustertest/conf/slon.4.conf.sample b/clustertest/conf/slon.4.conf.sample index a4e834a8..43b94f2f 100644 --- a/clustertest/conf/slon.4.conf.sample +++ b/clustertest/conf/slon.4.conf.sample @@ -35,20 +35,6 @@ sync_interval_timeout=2000 # Range: [0,100], default: 6 sync_group_maxsize=5 -# Size above which an sl_log_? row's log_cmddata is considered large. -# Up to 500 rows of this size are allowed in memory at once. Rows larger -# than that count into the sync_max_largemem space allocated and free'd -# on demand. -# Range: [1024,32768], default: 8192 -sync_max_rowsize=16384 - -# Maximum amount of memory allowed for large rows. Note that the algorithm -# will stop fetching rows AFTER this amount is exceeded, not BEFORE. This -# is done to ensure that a single row exceeding this limit alone does not -# stall replication. -# Range: [1048576,1073741824], default: 5242880 -sync_max_largemem=3276800 - # If this parameter is 1, messages go both to syslog and the standard # output. A value of 2 sends output only to syslog (some messages will # still go to the standard output/error). The default is 0, which means diff --git a/clustertest/conf/slon.5.conf.sample b/clustertest/conf/slon.5.conf.sample index a4e834a8..43b94f2f 100644 --- a/clustertest/conf/slon.5.conf.sample +++ b/clustertest/conf/slon.5.conf.sample @@ -35,20 +35,6 @@ sync_interval_timeout=2000 # Range: [0,100], default: 6 sync_group_maxsize=5 -# Size above which an sl_log_? row's log_cmddata is considered large. -# Up to 500 rows of this size are allowed in memory at once. Rows larger -# than that count into the sync_max_largemem space allocated and free'd -# on demand. -# Range: [1024,32768], default: 8192 -sync_max_rowsize=16384 - -# Maximum amount of memory allowed for large rows. Note that the algorithm -# will stop fetching rows AFTER this amount is exceeded, not BEFORE. This -# is done to ensure that a single row exceeding this limit alone does not -# stall replication. -# Range: [1048576,1073741824], default: 5242880 -sync_max_largemem=3276800 - # If this parameter is 1, messages go both to syslog and the standard # output. A value of 2 sends output only to syslog (some messages will # still go to the standard output/error). The default is 0, which means diff --git a/doc/adminguide/loganalysis.sgml b/doc/adminguide/loganalysis.sgml index 8761d9fd..3d129bc8 100644 --- a/doc/adminguide/loganalysis.sgml +++ b/doc/adminguide/loganalysis.sgml @@ -736,9 +736,6 @@ application downtime may not be necessary. This indicates how long it took to get the first row from the LOG cursor that pulls in data from the sl_log tables. -ERROR: remoteWorkerThread_%d_d: large log_cmddata for actionseq %s not found - - &lslon; could not find the data for one of the very large sl_log table tuples that are pulled individually. This shouldn't happen. DEBUG2: remoteWorkerThread_%d_d:%.3f seconds until close cursor This indicates how long it took to complete reading data from the LOG cursor that pulls in data from the sl_log tables. diff --git a/doc/adminguide/slonconf.sgml b/doc/adminguide/slonconf.sgml index abb96dfe..7ad0cc26 100644 --- a/doc/adminguide/slonconf.sgml +++ b/doc/adminguide/slonconf.sgml @@ -580,45 +580,6 @@ - - sync_max_rowsize (integer) - - sync_max_rowsize configuration parameter - - - Size above which an sl_log_? row's - log_cmddata is considered large. Up to 500 - rows of this size are allowed in memory at once. Rows larger - than that count into the sync_max_largemem - space allocated and free()'ed on demand. - - - The default value is 8192, meaning that your expected - memory consumption (for the LOG cursor) should not exceed 8MB. - - - - - - sync_max_largemem (integer) - - sync_max_largemem configuration parameter - - - Maximum memory allocated for large rows, where - log_cmddata are larger than - sync_max_rowsize. - - Note that the algorithm reads rows until - after this value is exceeded. Otherwise, - a tuple larger than this value would stall replication. As a - result, don't assume that memory consumption will remain - smaller than this value. - - - The default value is 5242880. - - remote_listen_timeout (integer) diff --git a/share/slon.conf-sample b/share/slon.conf-sample index 7bb9d4c3..433c6955 100644 --- a/share/slon.conf-sample +++ b/share/slon.conf-sample @@ -37,20 +37,6 @@ # Range: [0,100], default: 6 #sync_group_maxsize=6 -# Size above which an sl_log_? row's log_cmddata is considered large. -# Up to 500 rows of this size are allowed in memory at once. Rows larger -# than that count into the sync_max_largemem space allocated and free'd -# on demand. -# Range: [1024,32768], default: 8192 -#sync_max_rowsize=8192 - -# Maximum amount of memory allowed for large rows. Note that the algorithm -# will stop fetching rows AFTER this amount is exceeded, not BEFORE. This -# is done to ensure that a single row exceeding this limit alone does not -# stall replication. -# Range: [1048576,1073741824], default: 5242880 -#sync_max_largemem=5242880 - # If this parameter is 1, messages go both to syslog and the standard # output. A value of 2 sends output only to syslog (some messages will # still go to the standard output/error). The default is 0, which means @@ -117,4 +103,4 @@ # archive_dir="/tmp/somewhere" # Should slon run the monitoring thread? -# monitor_threads=true \ No newline at end of file +# monitor_threads=true diff --git a/src/backend/slony1_base.sql b/src/backend/slony1_base.sql index d82e69db..decb1041 100644 --- a/src/backend/slony1_base.sql +++ b/src/backend/slony1_base.sql @@ -369,8 +369,11 @@ create table @NAMESPACE@.sl_log_1 ( log_txid bigint, log_tableid int4, log_actionseq int8, + log_tablenspname text, + log_tablerelname text, log_cmdtype char, - log_cmddata text + log_cmdupdncols int4, + log_cmdargs text[] ) WITHOUT OIDS; create index sl_log_1_idx1 on @NAMESPACE@.sl_log_1 (log_origin, log_txid, log_actionseq); @@ -383,8 +386,12 @@ comment on table @NAMESPACE@.sl_log_1 is 'Stores each change to be propagated to comment on column @NAMESPACE@.sl_log_1.log_origin is 'Origin node from which the change came'; comment on column @NAMESPACE@.sl_log_1.log_txid is 'Transaction ID on the origin node'; comment on column @NAMESPACE@.sl_log_1.log_tableid is 'The table ID (from sl_table.tab_id) that this log entry is to affect'; -comment on column @NAMESPACE@.sl_log_1.log_cmdtype is 'Replication action to take. U = Update, I = Insert, D = DELETE'; -comment on column @NAMESPACE@.sl_log_1.log_cmddata is 'The data needed to perform the log action'; +comment on column @NAMESPACE@.sl_log_1.log_actionseq is 'The sequence number in which actions will be applied on replicas'; +comment on column @NAMESPACE@.sl_log_1.log_tablenspname is 'The schema name of the table affected'; +comment on column @NAMESPACE@.sl_log_1.log_tablerelname is 'The table name of the table affected'; +comment on column @NAMESPACE@.sl_log_1.log_cmdtype is 'Replication action to take. U = Update, I = Insert, D = DELETE, T = TRUNCATE'; +comment on column @NAMESPACE@.sl_log_1.log_cmdupdncols is 'For cmdtype=U the number of updated columns in cmdargs'; +comment on column @NAMESPACE@.sl_log_1.log_cmdargs is 'The data needed to perform the log action on the replica'; -- ---------------------------------------------------------------------- -- TABLE sl_log_2 @@ -394,15 +401,12 @@ create table @NAMESPACE@.sl_log_2 ( log_txid bigint, log_tableid int4, log_actionseq int8, + log_tablenspname text, + log_tablerelname text, log_cmdtype char, - log_cmddata text + log_cmdupdncols int4, + log_cmdargs text[] ) WITHOUT OIDS; -comment on table @NAMESPACE@.sl_log_2 is 'Stores each change to be propagated to subscriber nodes'; -comment on column @NAMESPACE@.sl_log_2.log_origin is 'Origin node from which the change came'; -comment on column @NAMESPACE@.sl_log_2.log_txid is 'Transaction ID on the origin node'; -comment on column @NAMESPACE@.sl_log_2.log_tableid is 'The table ID (from sl_table.tab_id) that this log entry is to affect'; -comment on column @NAMESPACE@.sl_log_2.log_cmdtype is 'Replication action to take. U = Update, I = Insert, D = DELETE'; -comment on column @NAMESPACE@.sl_log_2.log_cmddata is 'The data needed to perform the log action'; create index sl_log_2_idx1 on @NAMESPACE@.sl_log_2 (log_origin, log_txid, log_actionseq); @@ -410,6 +414,17 @@ create index sl_log_2_idx1 on @NAMESPACE@.sl_log_2 -- create index sl_log_2_idx2 on @NAMESPACE@.sl_log_2 -- (log_txid); +comment on table @NAMESPACE@.sl_log_2 is 'Stores each change to be propagated to subscriber nodes'; +comment on column @NAMESPACE@.sl_log_2.log_origin is 'Origin node from which the change came'; +comment on column @NAMESPACE@.sl_log_2.log_txid is 'Transaction ID on the origin node'; +comment on column @NAMESPACE@.sl_log_2.log_tableid is 'The table ID (from sl_table.tab_id) that this log entry is to affect'; +comment on column @NAMESPACE@.sl_log_2.log_actionseq is 'The sequence number in which actions will be applied on replicas'; +comment on column @NAMESPACE@.sl_log_2.log_tablenspname is 'The schema name of the table affected'; +comment on column @NAMESPACE@.sl_log_2.log_tablerelname is 'The table name of the table affected'; +comment on column @NAMESPACE@.sl_log_2.log_cmdtype is 'Replication action to take. U = Update, I = Insert, D = DELETE, T = TRUNCATE'; +comment on column @NAMESPACE@.sl_log_2.log_cmdupdncols is 'For cmdtype=U the number of updated columns in cmdargs'; +comment on column @NAMESPACE@.sl_log_2.log_cmdargs is 'The data needed to perform the log action on the replica'; + -- ---------------------------------------------------------------------- -- TABLE sl_registry diff --git a/src/backend/slony1_funcs.c b/src/backend/slony1_funcs.c index 9f5c06c4..ac6c4c8b 100644 --- a/src/backend/slony1_funcs.c +++ b/src/backend/slony1_funcs.c @@ -36,6 +36,7 @@ #include "utils/guc.h" #include "utils/rel.h" #include "utils/relcache.h" +#include "utils/lsyscache.h" #ifdef HAVE_GETACTIVESNAPSHOT #include "utils/snapmgr.h" #endif @@ -119,9 +120,6 @@ typedef struct slony_I_cluster_status text *cmdtype_U; text *cmdtype_D; - text *cmddata_buf; - int cmddata_size; - struct slony_I_cluster_status *next; } Slony_I_ClusterStatus; @@ -131,7 +129,6 @@ static Slony_I_ClusterStatus * getClusterStatus(Name cluster_name, int need_plan_mask); static const char *slon_quote_identifier(const char *ident); -static char *slon_quote_literal(char *str); static int prepareLogPlan(Slony_I_ClusterStatus * cs, int log_status); @@ -287,14 +284,22 @@ _Slony_I_logTrigger(PG_FUNCTION_ARGS) TransactionId newXid = GetTopTransactionId(); Slony_I_ClusterStatus *cs; TriggerData *tg; - Datum argv[4]; + Datum argv[6]; text *cmdtype = NULL; + int32 cmdupdncols = 0; int rc; Name cluster_name; int32 tab_id; char *attkind; int attkind_idx; - int cmddata_need; + + char *olddatestyle = NULL; + Datum *cmdargs = NULL; + Datum *cmdargselem = NULL; + bool *cmdnulls = NULL; + bool *cmdnullselem = NULL; + int cmddims[1]; + int cmdlbs[1]; /* * Don't do any logging if the current session role isn't Origin. @@ -379,35 +384,41 @@ _Slony_I_logTrigger(PG_FUNCTION_ARGS) cs->currentXid = newXid; } + /* + * Save the current datestyle setting and switch to ISO (if not already) + */ + olddatestyle = GetConfigOptionByName("DateStyle", NULL); + if (!strstr(olddatestyle, "ISO")) + set_config_option("DateStyle", "ISO", PGC_USERSET, PGC_S_SESSION, + true, true); + /* - * Determine cmdtype and cmddata depending on the command type + * Determine cmdtype and cmdargs depending on the command type */ if (TRIGGER_FIRED_BY_INSERT(tg->tg_event)) { HeapTuple new_row = tg->tg_trigtuple; TupleDesc tupdesc = tg->tg_relation->rd_att; - char *col_ident; char *col_value; - int len_ident; - int len_value; int i; - int need_comma = false; - char *OldDateStyle; - char *cp = VARDATA(cs->cmddata_buf); /* * INSERT * - * cmdtype = 'I' cmddata = ("col" [, ...]) values ('value' [, ...]) + * cmdtype = 'I' cmdargs = colname, newval [, ...] */ cmdtype = cs->cmdtype_I; + cmdargselem = cmdargs = (Datum *)palloc(sizeof(Datum) * + ((tg->tg_relation->rd_att->natts * 2) + 2)); + cmdnullselem = cmdnulls = (bool *)palloc(sizeof(bool) * + ((tg->tg_relation->rd_att->natts * 2) + 2)); + /* * Specify all the columns */ - *cp++ = '('; for (i = 0; i < tg->tg_relation->rd_att->natts; i++) { /* @@ -416,98 +427,28 @@ _Slony_I_logTrigger(PG_FUNCTION_ARGS) if (tupdesc->attrs[i]->attisdropped) continue; - col_ident = (char *) slon_quote_identifier(SPI_fname(tupdesc, i + 1)); - cmddata_need = (cp - (char *) (cs->cmddata_buf)) + 16 + - (len_ident = strlen(col_ident)); - if (cs->cmddata_size < cmddata_need) - { - int have = (cp - (char *) (cs->cmddata_buf)); - - while (cs->cmddata_size < cmddata_need) - cs->cmddata_size *= 2; - cs->cmddata_buf = realloc(cs->cmddata_buf, cs->cmddata_size); - cp = (char *) (cs->cmddata_buf) + have; - } - - if (need_comma) - *cp++ = ','; - else - need_comma = true; - - memcpy(cp, col_ident, len_ident); - cp += len_ident; - } - - /* - * Append the string ") values (" - */ - *cp++ = ')'; - *cp++ = ' '; - *cp++ = 'v'; - *cp++ = 'a'; - *cp++ = 'l'; - *cp++ = 'u'; - *cp++ = 'e'; - *cp++ = 's'; - *cp++ = ' '; - *cp++ = '('; - - /* - * Append the values - */ - need_comma = false; - OldDateStyle = GetConfigOptionByName("DateStyle", NULL); - if (!strstr(OldDateStyle, "ISO")) - set_config_option("DateStyle", "ISO", PGC_USERSET, PGC_S_SESSION, true, true); - for (i = 0; i < tg->tg_relation->rd_att->natts; i++) - { /* - * Skip dropped columns + * Add the column name */ - if (tupdesc->attrs[i]->attisdropped) - continue; - + *cmdargselem++ = DirectFunctionCall1(textin, + CStringGetDatum(SPI_fname(tupdesc, i + 1))); + *cmdnullselem++ = false; + /* + * Add the column value + */ if ((col_value = SPI_getvalue(new_row, tupdesc, i + 1)) == NULL) { - col_value = "NULL"; + *cmdnullselem++ = true; } else { - col_value = slon_quote_literal(col_value); + *cmdargselem++ = DirectFunctionCall1(textin, + CStringGetDatum(col_value)); + *cmdnullselem++ = false; } - - cmddata_need = (cp - (char *) (cs->cmddata_buf)) + 16 + - (len_value = strlen(col_value)); - if (cs->cmddata_size < cmddata_need) - { - int have = (cp - (char *) (cs->cmddata_buf)); - - while (cs->cmddata_size < cmddata_need) - cs->cmddata_size *= 2; - cs->cmddata_buf = realloc(cs->cmddata_buf, cs->cmddata_size); - cp = (char *) (cs->cmddata_buf) + have; - } - - if (need_comma) - *cp++ = ','; - else - need_comma = true; - - memcpy(cp, col_value, len_value); - cp += len_value; } - if (!strstr(OldDateStyle, "ISO")) - set_config_option("DateStyle", OldDateStyle, PGC_USERSET, PGC_S_SESSION, true, true); - - /* - * Terminate and done - */ - *cp++ = ')'; - *cp = '\0'; - SET_VARSIZE(cs->cmddata_buf, - VARHDRSZ + (cp - VARDATA(cs->cmddata_buf))); } else if (TRIGGER_FIRED_BY_UPDATE(tg->tg_event)) { @@ -521,22 +462,24 @@ _Slony_I_logTrigger(PG_FUNCTION_ARGS) char *col_ident; char *col_value; - int len_ident; - int len_value; int i; - int need_comma = false; - int need_and = false; - char *OldDateStyle; - - char *cp = VARDATA(cs->cmddata_buf); /* * UPDATE * - * cmdtype = 'U' cmddata = "col_ident"='value' [, ...] where - * "pk_ident" = 'value' [ and ...] + * cmdtype = 'U' cmdargs = pkcolname, oldval [, ...] + * colname, newval [, ...] */ cmdtype = cs->cmdtype_U; + + cmdargselem = cmdargs = (Datum *)palloc(sizeof(Datum) * + ((tg->tg_relation->rd_att->natts * 4) + 3)); + cmdnullselem = cmdnulls = (bool *)palloc(sizeof(bool) * + ((tg->tg_relation->rd_att->natts * 4) + 3)); + + /* + * For all changed columns, add name+value pairs and count them. + */ for (i = 0; i < tg->tg_relation->rd_att->natts; i++) { /* @@ -612,95 +555,23 @@ _Slony_I_logTrigger(PG_FUNCTION_ARGS) } } - if (need_comma) - *cp++ = ','; - else - need_comma = true; - - col_ident = (char *) slon_quote_identifier(SPI_fname(tupdesc, i + 1)); + *cmdargselem++ = DirectFunctionCall1(textin, + CStringGetDatum(SPI_fname(tupdesc, i + 1))); + *cmdnullselem++ = false; if (new_isnull) - col_value = "NULL"; + *cmdnullselem++ = true; else { - OldDateStyle = GetConfigOptionByName("DateStyle", NULL); - if (!strstr(OldDateStyle, "ISO")) - set_config_option("DateStyle", "ISO", PGC_USERSET, PGC_S_SESSION, true, true); - col_value = slon_quote_literal(SPI_getvalue(new_row, tupdesc, i + 1)); - if (!strstr(OldDateStyle, "ISO")) - set_config_option("DateStyle", OldDateStyle, PGC_USERSET, PGC_S_SESSION, true, true); + *cmdargselem++ = DirectFunctionCall1(textin, + CStringGetDatum(SPI_getvalue(new_row, tupdesc, i + 1))); + *cmdnullselem++ = false; } - cmddata_need = (cp - (char *) (cs->cmddata_buf)) + 16 + - (len_ident = strlen(col_ident)) + - (len_value = strlen(col_value)); - if (cs->cmddata_size < cmddata_need) - { - int have = (cp - (char *) (cs->cmddata_buf)); - - while (cs->cmddata_size < cmddata_need) - cs->cmddata_size *= 2; - cs->cmddata_buf = realloc(cs->cmddata_buf, cs->cmddata_size); - cp = (char *) (cs->cmddata_buf) + have; - } - - memcpy(cp, col_ident, len_ident); - cp += len_ident; - *cp++ = '='; - memcpy(cp, col_value, len_value); - cp += len_value; + cmdupdncols++; } /* - * It can happen that the only UPDATE an application does is to set a - * column to the same value again. In that case, we'd end up here with - * no columns in the SET clause yet. We add the first key column here - * with it's old value to simulate the same for the replication - * engine. + * Add pairs of PK column names and values */ - if (!need_comma) - { - for (i = 0, attkind_idx = -1; i < tg->tg_relation->rd_att->natts; i++) - { - if (tupdesc->attrs[i]->attisdropped) - continue; - - attkind_idx++; - if (!attkind[attkind_idx]) - elog(ERROR, "Slony-I: no key columns found in logTrigger() attkind parameter"); - - if (attkind[attkind_idx] == 'k') - break; - } - col_ident = (char *) slon_quote_identifier(SPI_fname(tupdesc, i + 1)); - col_value = slon_quote_literal(SPI_getvalue(old_row, tupdesc, i + 1)); - - cmddata_need = (cp - (char *) (cs->cmddata_buf)) + 16 + - (len_ident = strlen(col_ident)) + - (len_value = strlen(col_value)); - if (cs->cmddata_size < cmddata_need) - { - int have = (cp - (char *) (cs->cmddata_buf)); - - while (cs->cmddata_size < cmddata_need) - cs->cmddata_size *= 2; - cs->cmddata_buf = realloc(cs->cmddata_buf, cs->cmddata_size); - cp = (char *) (cs->cmddata_buf) + have; - } - - memcpy(cp, col_ident, len_ident); - cp += len_ident; - *cp++ = '='; - memcpy(cp, col_value, len_value); - cp += len_value; - } - - *cp++ = ' '; - *cp++ = 'w'; - *cp++ = 'h'; - *cp++ = 'e'; - *cp++ = 'r'; - *cp++ = 'e'; - *cp++ = ' '; - for (i = 0, attkind_idx = -1; i < tg->tg_relation->rd_att->natts; i++) { /* @@ -714,45 +585,21 @@ _Slony_I_logTrigger(PG_FUNCTION_ARGS) break; if (attkind[attkind_idx] != 'k') continue; - col_ident = (char *) slon_quote_identifier(SPI_fname(tupdesc, i + 1)); - col_value = slon_quote_literal(SPI_getvalue(old_row, tupdesc, i + 1)); + col_ident = SPI_fname(tupdesc, i + 1); + col_value = SPI_getvalue(old_row, tupdesc, i + 1); if (col_value == NULL) elog(ERROR, "Slony-I: old key column %s.%s IS NULL on UPDATE", NameStr(tg->tg_relation->rd_rel->relname), col_ident); - cmddata_need = (cp - (char *) (cs->cmddata_buf)) + 16 + - (len_ident = strlen(col_ident)) + - (len_value = strlen(col_value)); - if (cs->cmddata_size < cmddata_need) - { - int have = (cp - (char *) (cs->cmddata_buf)); - - while (cs->cmddata_size < cmddata_need) - cs->cmddata_size *= 2; - cs->cmddata_buf = realloc(cs->cmddata_buf, cs->cmddata_size); - cp = (char *) (cs->cmddata_buf) + have; - } + *cmdargselem++ = DirectFunctionCall1(textin, + CStringGetDatum(col_ident)); + *cmdnullselem++ = false; - if (need_and) - { - *cp++ = ' '; - *cp++ = 'a'; - *cp++ = 'n'; - *cp++ = 'd'; - *cp++ = ' '; - } - else - need_and = true; - - memcpy(cp, col_ident, len_ident); - cp += len_ident; - *cp++ = '='; - memcpy(cp, col_value, len_value); - cp += len_value; + *cmdargselem++ = DirectFunctionCall1(textin, + CStringGetDatum(col_value)); + *cmdnullselem++ = false; } - *cp = '\0'; - SET_VARSIZE(cs->cmddata_buf, - VARHDRSZ + (cp - VARDATA(cs->cmddata_buf))); + } else if (TRIGGER_FIRED_BY_DELETE(tg->tg_event)) { @@ -760,19 +607,23 @@ _Slony_I_logTrigger(PG_FUNCTION_ARGS) TupleDesc tupdesc = tg->tg_relation->rd_att; char *col_ident; char *col_value; - int len_ident; - int len_value; int i; - int need_and = false; - char *cp = VARDATA(cs->cmddata_buf); /* * DELETE * - * cmdtype = 'D' cmddata = "pk_ident"='value' [and ...] + * cmdtype = 'D' cmdargs = pkcolname, oldval [, ...] */ cmdtype = cs->cmdtype_D; + cmdargselem = cmdargs = (Datum *)palloc(sizeof(Datum) * + ((tg->tg_relation->rd_att->natts * 2) + 2)); + cmdnullselem = cmdnulls = (bool *)palloc(sizeof(bool) * + ((tg->tg_relation->rd_att->natts * 2) + 2)); + + /* + * Add the PK columns + */ for (i = 0, attkind_idx = -1; i < tg->tg_relation->rd_att->natts; i++) { if (tupdesc->attrs[i]->attisdropped) @@ -783,55 +634,47 @@ _Slony_I_logTrigger(PG_FUNCTION_ARGS) break; if (attkind[attkind_idx] != 'k') continue; - col_ident = (char *) slon_quote_identifier(SPI_fname(tupdesc, i + 1)); - col_value = slon_quote_literal(SPI_getvalue(old_row, tupdesc, i + 1)); + + *cmdargselem++ = DirectFunctionCall1(textin, + CStringGetDatum(col_ident = SPI_fname(tupdesc, i + 1))); + *cmdnullselem++ = false; + + col_value = SPI_getvalue(old_row, tupdesc, i + 1); if (col_value == NULL) elog(ERROR, "Slony-I: old key column %s.%s IS NULL on DELETE", NameStr(tg->tg_relation->rd_rel->relname), col_ident); - - cmddata_need = (cp - (char *) (cs->cmddata_buf)) + 16 + - (len_ident = strlen(col_ident)) + - (len_value = strlen(col_value)); - if (cs->cmddata_size < cmddata_need) - { - int have = (cp - (char *) (cs->cmddata_buf)); - - while (cs->cmddata_size < cmddata_need) - cs->cmddata_size *= 2; - cs->cmddata_buf = realloc(cs->cmddata_buf, cs->cmddata_size); - cp = (char *) (cs->cmddata_buf) + have; - } - - if (need_and) - { - *cp++ = ' '; - *cp++ = 'a'; - *cp++ = 'n'; - *cp++ = 'd'; - *cp++ = ' '; - } - else - need_and = true; - - memcpy(cp, col_ident, len_ident); - cp += len_ident; - *cp++ = '='; - memcpy(cp, col_value, len_value); - cp += len_value; + *cmdargselem++ = DirectFunctionCall1(textin, + CStringGetDatum(col_value)); + *cmdnullselem++ = false; } - *cp = '\0'; - SET_VARSIZE(cs->cmddata_buf, - VARHDRSZ + (cp - VARDATA(cs->cmddata_buf))); } else elog(ERROR, "Slony-I: logTrigger() fired for unhandled event"); + /* + * Restore the datestyle + */ + if (!strstr(olddatestyle, "ISO")) + set_config_option("DateStyle", olddatestyle, + PGC_USERSET, PGC_S_SESSION, true, true); + /* * Construct the parameter array and insert the log row. */ + cmddims[0] = cmdargselem - cmdargs; + cmdlbs[0] = 1; + argv[0] = Int32GetDatum(tab_id); - argv[1] = PointerGetDatum(cmdtype); - argv[2] = PointerGetDatum(cs->cmddata_buf); + argv[1] = DirectFunctionCall1(textin, + CStringGetDatum(get_namespace_name( + RelationGetNamespace(tg->tg_relation)))); + argv[2] = DirectFunctionCall1(textin, + CStringGetDatum(RelationGetRelationName(tg->tg_relation))); + argv[3] = PointerGetDatum(cmdtype); + argv[4] = Int32GetDatum(cmdupdncols); + argv[5] = PointerGetDatum(construct_md_array(cmdargs, cmdnulls, 1, + cmddims, cmdlbs, TEXTOID, -1, false, 'i')); + SPI_execp(cs->plan_active_log, argv, NULL, 0); SPI_finish(); @@ -1018,50 +861,6 @@ _Slony_I_seqtrack(PG_FUNCTION_ARGS) } -static char * -slon_quote_literal(char *str) -{ - char *result; - char *cp1; - char *cp2; - int len; - int wl; - - if (str == NULL) - return NULL; - - len = strlen(str); - result = palloc(len * 2 + 3); - cp1 = str; - cp2 = result; - - *cp2++ = '\''; - while (len > 0) - { - if ((wl = pg_mblen((const char *) cp1)) != 1) - { - len -= wl; - - while (wl-- > 0) - *cp2++ = *cp1++; - continue; - } - - if (*cp1 == '\'') - *cp2++ = '\''; - if (*cp1 == '\\') - *cp2++ = '\\'; - *cp2++ = *cp1++; - len--; - } - - *cp2++ = '\''; - *cp2++ = '\0'; - - return result; -} - - /* * slon_quote_identifier - Quote an identifier only if needed * @@ -1342,7 +1141,7 @@ getClusterStatus(Name cluster_name, int need_plan_mask) /* * Also create the 3 rather static text values for the log_cmdtype - * parameter and initialize the cmddata_buf. + * parameter. */ cs->cmdtype_I = malloc(VARHDRSZ + 1); SET_VARSIZE(cs->cmdtype_I, VARHDRSZ + 1); @@ -1361,9 +1160,6 @@ getClusterStatus(Name cluster_name, int need_plan_mask) cs->clusterident); cs->plan_get_logstatus = SPI_saveplan(SPI_prepare(query, 0, NULL)); - cs->cmddata_size = 8192; - cs->cmddata_buf = (text *) malloc(8192); - cs->have_plan |= PLAN_INSERT_LOG_STATUS; } @@ -1375,6 +1171,11 @@ getClusterStatus(Name cluster_name, int need_plan_mask) * prepare the plan for the curren sl_log_x insert query. * */ + +#ifndef TEXTARRAYOID +#define TEXTARRAYOID 1009 +#endif + int prepareLogPlan(Slony_I_ClusterStatus * cs, int log_status) { @@ -1391,15 +1192,19 @@ int prepareLogPlan(Slony_I_ClusterStatus * cs, */ sprintf(query, "INSERT INTO %s.sl_log_1 " "(log_origin, log_txid, log_tableid, log_actionseq," - " log_cmdtype, log_cmddata) " + " log_tablenspname, log_tablerelname, " + " log_cmdtype, log_cmdupdncols, log_cmdargs) " "VALUES (%d, \"pg_catalog\".txid_current(), $1, " - "nextval('%s.sl_action_seq'), $2, $3); ", + "nextval('%s.sl_action_seq'), $2, $3, $4, $5, $6); ", cs->clusterident, cs->localNodeId, cs->clusterident); plan_types[0] = INT4OID; plan_types[1] = TEXTOID; plan_types[2] = TEXTOID; + plan_types[3] = TEXTOID; + plan_types[4] = INT4OID; + plan_types[5] = TEXTARRAYOID; - cs->plan_insert_log_1 = SPI_saveplan(SPI_prepare(query, 3, plan_types)); + cs->plan_insert_log_1 = SPI_saveplan(SPI_prepare(query, 6, plan_types)); if (cs->plan_insert_log_1 == NULL) elog(ERROR, "Slony-I: SPI_prepare() failed"); } @@ -1409,15 +1214,19 @@ int prepareLogPlan(Slony_I_ClusterStatus * cs, { sprintf(query, "INSERT INTO %s.sl_log_2 " "(log_origin, log_txid, log_tableid, log_actionseq," - " log_cmdtype, log_cmddata) " + " log_tablenspname, log_tablerelname, " + " log_cmdtype, log_cmdupdncols, log_cmdargs) " "VALUES (%d, \"pg_catalog\".txid_current(), $1, " - "nextval('%s.sl_action_seq'), $2, $3); ", + "nextval('%s.sl_action_seq'), $2, $3, $4, $5, $6); ", cs->clusterident, cs->localNodeId, cs->clusterident); plan_types[0] = INT4OID; plan_types[1] = TEXTOID; plan_types[2] = TEXTOID; + plan_types[3] = TEXTOID; + plan_types[4] = INT4OID; + plan_types[5] = TEXTARRAYOID; - cs->plan_insert_log_2 = SPI_saveplan(SPI_prepare(query, 3, plan_types)); + cs->plan_insert_log_2 = SPI_saveplan(SPI_prepare(query, 6, plan_types)); if (cs->plan_insert_log_2 == NULL) elog(ERROR, "Slony-I: SPI_prepare() failed"); } @@ -1446,8 +1255,6 @@ _Slony_I_resetSession(PG_FUNCTION_ARGS) free(cs->cmdtype_D); if(cs->cmdtype_U) free(cs->cmdtype_D); - if(cs->cmddata_buf) - free(cs->cmddata_buf); free(cs->clusterident); if(cs->plan_insert_event) SPI_freeplan(cs->plan_insert_event); diff --git a/src/backend/slony1_funcs.sql b/src/backend/slony1_funcs.sql index cc87e377..37ecad59 100644 --- a/src/backend/slony1_funcs.sql +++ b/src/backend/slony1_funcs.sql @@ -4996,19 +4996,27 @@ system switches between sl_log_1 and sl_log_2.'; -- -- Called by slonik during the function upgrade process. -- ---------------------------------------------------------------------- +create or replace function @NAMESPACE@.check_table_field_exists (p_namespace text, p_table text, p_field text) +returns bool as $$ +BEGIN + return exists ( + select 1 from "information_schema".columns + where table_schema = p_namespace + and table_name = p_table + and column_name = p_field + ); +END;$$ language plpgsql; + +comment on function @NAMESPACE@.check_table_field_exists (p_namespace text, p_table text, p_field text) +is 'Check if a table has a specific attribute'; + create or replace function @NAMESPACE@.add_missing_table_field (p_namespace text, p_table text, p_field text, p_type text) returns bool as $$ DECLARE v_row record; v_query text; BEGIN - select 1 into v_row from pg_namespace n, pg_class c, pg_attribute a - where @NAMESPACE@.slon_quote_brute(n.nspname) = p_namespace and - c.relnamespace = n.oid and - @NAMESPACE@.slon_quote_brute(c.relname) = p_table and - a.attrelid = c.oid and - @NAMESPACE@.slon_quote_brute(a.attname) = p_field; - if not found then + if not @NAMESPACE@.check_table_field_exists(p_namespace, p_table, p_field) then raise notice 'Upgrade table %.% - add field %', p_namespace, p_table, p_field; v_query := 'alter table ' || p_namespace || '.' || p_table || ' add column '; v_query := v_query || p_field || ' ' || p_type || ';'; @@ -5080,9 +5088,196 @@ create table @NAMESPACE@.sl_components ( v_query := 'create table @NAMESPACE@.sl_event_lock (dummy integer);'; execute v_query; end if; + + -- + -- On the upgrade to 2.2, we change the layout of sl_log_N by + -- adding columns log_tablenspname, log_tablerelname, and + -- log_cmdupdncols as well as changing log_cmddata into + -- log_cmdargs, which is a text array. + -- + if not @NAMESPACE@.check_table_field_exists('_@CLUSTERNAME@', 'sl_log_1', 'log_cmdargs') then + -- + -- Check that the cluster is completely caught up + -- + if @NAMESPACE@.check_unconfirmed_log() then + raise EXCEPTION 'cannot upgrade to new sl_log_N format due to existing unreplicated data'; + end if; + + -- + -- Drop tables sl_log_1 and sl_log_2 + -- + drop table @NAMESPACE@.sl_log_1; + drop table @NAMESPACE@.sl_log_2; + + -- + -- Create the new sl_log_1 + -- + create table @NAMESPACE@.sl_log_1 ( + log_origin int4, + log_txid bigint, + log_tableid int4, + log_actionseq int8, + log_tablenspname text, + log_tablerelname text, + log_cmdtype char, + log_cmdupdncols int4, + log_cmdargs text[] + ) without oids; + create index sl_log_1_idx1 on @NAMESPACE@.sl_log_1 + (log_origin, log_txid, log_actionseq); + + comment on table @NAMESPACE@.sl_log_1 is 'Stores each change to be propagated to subscriber nodes'; + comment on column @NAMESPACE@.sl_log_1.log_origin is 'Origin node from which the change came'; + comment on column @NAMESPACE@.sl_log_1.log_txid is 'Transaction ID on the origin node'; + comment on column @NAMESPACE@.sl_log_1.log_tableid is 'The table ID (from sl_table.tab_id) that this log entry is to affect'; + comment on column @NAMESPACE@.sl_log_1.log_actionseq is 'The sequence number in which actions will be applied on replicas'; + comment on column @NAMESPACE@.sl_log_1.log_tablenspname is 'The schema name of the table affected'; + comment on column @NAMESPACE@.sl_log_1.log_tablerelname is 'The table name of the table affected'; + comment on column @NAMESPACE@.sl_log_1.log_cmdtype is 'Replication action to take. U = Update, I = Insert, D = DELETE, T = TRUNCATE'; + comment on column @NAMESPACE@.sl_log_1.log_cmdupdncols is 'For cmdtype=U the number of updated columns in cmdargs'; + comment on column @NAMESPACE@.sl_log_1.log_cmdargs is 'The data needed to perform the log action on the replica'; + + -- + -- Create the new sl_log_2 + -- + create table @NAMESPACE@.sl_log_2 ( + log_origin int4, + log_txid bigint, + log_tableid int4, + log_actionseq int8, + log_tablenspname text, + log_tablerelname text, + log_cmdtype char, + log_cmdupdncols int4, + log_cmdargs text[] + ) without oids; + create index sl_log_2_idx1 on @NAMESPACE@.sl_log_2 + (log_origin, log_txid, log_actionseq); + + comment on table @NAMESPACE@.sl_log_2 is 'Stores each change to be propagated to subscriber nodes'; + comment on column @NAMESPACE@.sl_log_2.log_origin is 'Origin node from which the change came'; + comment on column @NAMESPACE@.sl_log_2.log_txid is 'Transaction ID on the origin node'; + comment on column @NAMESPACE@.sl_log_2.log_tableid is 'The table ID (from sl_table.tab_id) that this log entry is to affect'; + comment on column @NAMESPACE@.sl_log_2.log_actionseq is 'The sequence number in which actions will be applied on replicas'; + comment on column @NAMESPACE@.sl_log_2.log_tablenspname is 'The schema name of the table affected'; + comment on column @NAMESPACE@.sl_log_2.log_tablerelname is 'The table name of the table affected'; + comment on column @NAMESPACE@.sl_log_2.log_cmdtype is 'Replication action to take. U = Update, I = Insert, D = DELETE, T = TRUNCATE'; + comment on column @NAMESPACE@.sl_log_2.log_cmdupdncols is 'For cmdtype=U the number of updated columns in cmdargs'; + comment on column @NAMESPACE@.sl_log_2.log_cmdargs is 'The data needed to perform the log action on the replica'; + + -- + -- Put the log apply triggers back onto sl_log_1/2 + -- + create trigger apply_trigger + before INSERT on @NAMESPACE@.sl_log_1 + for each row execute procedure @NAMESPACE@.log_apply(); + alter table @NAMESPACE@.sl_log_1 + enable replica trigger apply_trigger; + create trigger apply_trigger + before INSERT on @NAMESPACE@.sl_log_2 + for each row execute procedure @NAMESPACE@.log_apply(); + alter table @NAMESPACE@.sl_log_2 + enable replica trigger apply_trigger; + end if; + return p_old; end; -$$ language plpgsql +$$ language plpgsql; + +create or replace function @NAMESPACE@.check_unconfirmed_log () +returns bool as $$ +declare + v_rc bool = false; + v_error bool = false; + v_origin integer; + v_allconf bigint; + v_allsnap txid_snapshot; + v_count bigint; +begin + -- + -- Loop over all nodes that are the origin of at least one set + -- + for v_origin in select distinct set_origin as no_id + from @NAMESPACE@.sl_set loop + -- + -- Per origin determine which is the highest event seqno + -- that is confirmed by all subscribers to any of the + -- origins sets. + -- + select into v_allconf min(max_seqno) from ( + select con_received, max(con_seqno) as max_seqno + from @NAMESPACE@.sl_confirm + where con_origin = v_origin + and con_received in ( + select distinct sub_receiver + from @NAMESPACE@.sl_set as SET, + @NAMESPACE@.sl_subscribe as SUB + where SET.set_id = SUB.sub_set + and SET.set_origin = v_origin + ) + group by con_received + ) as maxconfirmed; + if not found then + raise NOTICE 'check_unconfirmed_log(): cannot determine highest ev_seqno for node % confirmed by all subscribers', v_origin; + v_error = true; + continue; + end if; + + -- + -- Get the txid snapshot that corresponds with that event + -- + select into v_allsnap ev_snapshot + from @NAMESPACE@.sl_event + where ev_origin = v_origin + and ev_seqno = v_allconf; + if not found then + raise NOTICE 'check_unconfirmed_log(): cannot find event %,% in sl_event', v_origin, v_allconf; + v_error = true; + continue; + end if; + + -- + -- Count the number of log rows that appeard after that event. + -- + select into v_count count(*) from ( + select 1 from @NAMESPACE@.sl_log_1 + where log_origin = v_origin + and log_txid >= "pg_catalog".txid_snapshot_xmax(v_allsnap) + union all + select 1 from @NAMESPACE@.sl_log_1 + where log_origin = v_origin + and log_txid in ( + select * from "pg_catalog".txid_snapshot_xip(v_allsnap) + ) + union all + select 1 from @NAMESPACE@.sl_log_2 + where log_origin = v_origin + and log_txid >= "pg_catalog".txid_snapshot_xmax(v_allsnap) + union all + select 1 from @NAMESPACE@.sl_log_2 + where log_origin = v_origin + and log_txid in ( + select * from "pg_catalog".txid_snapshot_xip(v_allsnap) + ) + ) as cnt; + + if v_count > 0 then + raise NOTICE 'check_unconfirmed_log(): origin % has % log rows that have not propagated to all subscribers yet', v_origin, v_count; + v_rc = true; + end if; + end loop; + + if v_error then + raise EXCEPTION 'check_unconfirmed_log(): aborting due to previous inconsistency'; + end if; + + return v_rc; +end; +$$ language plpgsql; + +-- +-- XXX What is this doing here? +-- set search_path to @NAMESPACE@ ; @@ -5520,23 +5715,37 @@ comment on function @NAMESPACE@.slon_node_health_check() is 'called when slon st create or replace function @NAMESPACE@.log_truncate () returns trigger as $$ declare - c_command text; + c_nspname text; + c_relname text; c_log integer; c_node integer; c_tabid integer; begin c_tabid := tg_argv[0]; c_node := @NAMESPACE@.getLocalNodeId('_@CLUSTERNAME@'); - c_command := 'TRUNCATE TABLE ONLY "' || tab_nspname || '"."' || - tab_relname || '" CASCADE;' + select tab_nspname, tab_relname into c_nspname, c_relname from @NAMESPACE@.sl_table where tab_id = c_tabid; select last_value into c_log from @NAMESPACE@.sl_log_status; if c_log in (0, 2) then - insert into @NAMESPACE@.sl_log_1 (log_origin, log_txid, log_tableid, log_actionseq, log_cmdtype, log_cmddata) - values (c_node, pg_catalog.txid_current(), c_tabid, nextval('_@CLUSTERNAME@.sl_action_seq'), 'T', c_command); + insert into @NAMESPACE@.sl_log_1 ( + log_origin, log_txid, log_tableid, + log_actionseq, log_tablenspname, + log_tablerelname, log_cmdtype, + log_cmdupdncols, log_cmdargs + ) values ( + c_node, pg_catalog.txid_current(), c_tabid, + nextval('@NAMESPACE@.sl_action_seq'), c_nspname, + c_relname, 'T', 0, array[]::text[]); else -- (1, 3) - insert into @NAMESPACE@.sl_log_2 (log_origin, log_txid, log_tableid, log_actionseq, log_cmdtype, log_cmddata) - values (c_node, pg_catalog.txid_current, c_tabid, nextval('_@CLUSTERNAME@.sl_action_seq'), 'T', c_command); + insert into @NAMESPACE@.sl_log_2 ( + log_origin, log_txid, log_tableid, + log_actionseq, log_tablenspname, + log_tablerelname, log_cmdtype, + log_cmdupdncols, log_cmdargs + ) values ( + c_node, pg_catalog.txid_current(), c_tabid, + nextval('@NAMESPACE@.sl_action_seq'), c_nspname, + c_relname, 'T', 0, array[]::text[]); end if; return NULL; end @@ -5680,3 +5889,113 @@ tables that are already exclusivly locked by the current transaction are repaired. Otherwise all replicated tables with outdated trigger arguments are recreated.'; + +create or replace function @NAMESPACE@.log_apply() returns trigger +as $$ +declare + v_command text = 'not implemented yet'; + v_list1 text = ''; + v_list2 text = ''; + v_comma text = ''; + v_and text = ''; + v_idx integer = 1; + v_nargs integer; + v_i integer = 0; +begin + v_nargs = array_upper(NEW.log_cmdargs, 1); + + if NEW.log_cmdtype = 'I' then + while v_idx < v_nargs loop + v_list1 = v_list1 || v_comma || + @NAMESPACE@.slon_quote_brute(NEW.log_cmdargs[v_idx]); + v_idx = v_idx + 1; + v_list2 = v_list2 || v_comma || + pg_catalog.quote_literal(NEW.log_cmdargs[v_idx]); + v_idx = v_idx + 1; + + v_comma = ','; + end loop; + + v_command = 'INSERT INTO ' || + @NAMESPACE@.slon_quote_brute(NEW.log_tablenspname) || '.' || + @NAMESPACE@.slon_quote_brute(NEW.log_tablerelname) || ' (' || + v_list1 || ') VALUES (' || v_list2 || ')'; + + execute v_command; + end if; + + if NEW.log_cmdtype = 'U' then + v_command = 'UPDATE ONLY ' || + @NAMESPACE@.slon_quote_brute(NEW.log_tablenspname) || '.' || + @NAMESPACE@.slon_quote_brute(NEW.log_tablerelname) || ' SET '; + while v_i < NEW.log_cmdupdncols loop + v_command = v_command || v_comma || + @NAMESPACE@.slon_quote_brute(NEW.log_cmdargs[v_idx]) || '='; + v_idx = v_idx + 1; + v_command = v_command || + pg_catalog.quote_literal(NEW.log_cmdargs[v_idx]); + v_idx = v_idx + 1; + v_comma = ','; + v_i = v_i + 1; + end loop; + if NEW.log_cmdupdncols = 0 then + v_command = v_command || + @NAMESPACE@.slon_quote_brute(NEW.log_cmdargs[1]) || '=' || + @NAMESPACE@.slon_quote_brute(NEW.log_cmdargs[1]); + end if; + v_command = v_command || ' WHERE '; + while v_idx < v_nargs loop + v_command = v_command || v_and || + @NAMESPACE@.slon_quote_brute(NEW.log_cmdargs[v_idx]) || '='; + v_idx = v_idx + 1; + v_command = v_command || + pg_catalog.quote_literal(NEW.log_cmdargs[v_idx]); + v_idx = v_idx + 1; + + v_and = ' AND '; + end loop; + + execute v_command; + end if; + + if NEW.log_cmdtype = 'D' then + v_command = 'DELETE FROM ONLY ' || + @NAMESPACE@.slon_quote_brute(NEW.log_tablenspname) || '.' || + @NAMESPACE@.slon_quote_brute(NEW.log_tablerelname) || ' WHERE '; + while v_idx < v_nargs loop + v_command = v_command || v_and || + @NAMESPACE@.slon_quote_brute(NEW.log_cmdargs[v_idx]) || '='; + v_idx = v_idx + 1; + v_command = v_command || + pg_catalog.quote_literal(NEW.log_cmdargs[v_idx]); + v_idx = v_idx + 1; + + v_and = ' AND '; + end loop; + + execute v_command; + end if; + + if NEW.log_cmdtype = 'T' then + perform @NAMESPACE@.TruncateOnlyTable( + @NAMESPACE@.slon_quote_brute(NEW.log_tablenspname) || '.' || + @NAMESPACE@.slon_quote_brute(NEW.log_tablerelname)); + end if; + + return NEW; +end; +$$ language plpgsql; + +drop trigger if exists apply_trigger on @NAMESPACE@.sl_log_1; +create trigger apply_trigger + before INSERT on @NAMESPACE@.sl_log_1 + for each row execute procedure @NAMESPACE@.log_apply(); +alter table @NAMESPACE@.sl_log_1 + enable replica trigger apply_trigger; +drop trigger if exists apply_trigger on @NAMESPACE@.sl_log_2; +create trigger apply_trigger + before INSERT on @NAMESPACE@.sl_log_2 + for each row execute procedure @NAMESPACE@.log_apply(); +alter table @NAMESPACE@.sl_log_2 + enable replica trigger apply_trigger; + diff --git a/src/slon/confoptions.c b/src/slon/confoptions.c index 255d7ea2..81f55684 100644 --- a/src/slon/confoptions.c +++ b/src/slon/confoptions.c @@ -133,30 +133,6 @@ static struct config_int ConfigureNamesInt[] = 0, 2147483647 }, - { - { - (const char *) "sync_max_rowsize", /* conf name */ - gettext_noop("sl_log_? rows larger than that are read separately"), /* short desc */ - gettext_noop("sl_log_? rows with octet_length(log_cmddata) larger than this are read separately"), /* long desc */ - SLON_C_INT /* config type */ - }, - &sync_max_rowsize, /* var name */ - 8192, /* default val */ - 1024, /* min val */ - 32768 /* max val */ - }, - { - { - (const char *) "sync_max_largemem", /* conf name */ - gettext_noop("How much memory to allow for sl_log_? rows exceeding sync_max_rowsize"), /* short desc */ - gettext_noop("How much memory to allow for sl_log_? rows exceeding sync_max_rowsize"), /* long desc */ - SLON_C_INT /* config type */ - }, - &sync_max_largemem, /* var name */ - 5242880, /* default val */ - 1048576, /* min val */ - 1073741824 /* max val */ - }, { { diff --git a/src/slon/confoptions.h b/src/slon/confoptions.h index 8f62a678..6fd315e9 100644 --- a/src/slon/confoptions.h +++ b/src/slon/confoptions.h @@ -18,8 +18,6 @@ extern char *archive_dir; extern int slon_log_level; extern int sync_interval; extern int sync_interval_timeout; -extern int sync_max_rowsize; -extern int sync_max_largemem; extern int remote_listen_timeout; extern int sync_group_maxsize; diff --git a/src/slon/remote_worker.c b/src/slon/remote_worker.c index 1671f3de..f2d7d5b7 100644 --- a/src/slon/remote_worker.c +++ b/src/slon/remote_worker.c @@ -243,8 +243,6 @@ static struct node_confirm_status *node_confirm_tail = NULL; static pthread_mutex_t node_confirm_lock = PTHREAD_MUTEX_INITIALIZER; int sync_group_maxsize; -int sync_max_rowsize; -int sync_max_largemem; int explain_interval; time_t explain_lastsec; int explain_thistime; @@ -275,7 +273,6 @@ static void start_monitored_event(PerfMon *pm); static void monitor_provider_query(PerfMon *pm); static void monitor_subscriber_query(PerfMon *pm); static void monitor_subscriber_iud(PerfMon *pm); -static void monitor_largetuples(PerfMon *pm); static void adjust_provider_info(SlonNode *node, WorkerGroupData *wd, int cleanup); @@ -4067,7 +4064,7 @@ sync_event(SlonNode *node, SlonConn *local_conn, ntables_total += ntuples2; /* - * ... and build up a the provider query + * ... and build up the provider query */ for (sl_log_no = 1; sl_log_no <= 2; sl_log_no++) { @@ -4101,15 +4098,12 @@ sync_event(SlonNode *node, SlonConn *local_conn, */ slon_appendquery(provider_query, "select log_origin, log_txid, log_tableid, " - "log_actionseq, log_cmdtype, " - "octet_length(log_cmddata), " - "case when octet_length(log_cmddata) <= %d " - "then log_cmddata " - "else null end " + "log_actionseq, log_tablenspname, " + "log_tablerelname, log_cmdtype, " + "log_cmdupdncols, log_cmdargs " "from %s.sl_log_%d " "where log_origin = %d " "and log_tableid in (", - sync_max_rowsize, rtcfg_namespace, sl_log_no, node->no_id); for (tupno2 = 0; tupno2 < ntuples2; tupno2++) @@ -4169,15 +4163,12 @@ sync_event(SlonNode *node, SlonConn *local_conn, slon_appendquery(provider_query, "union all " "select log_origin, log_txid, log_tableid, " - "log_actionseq, log_cmdtype, " - "octet_length(log_cmddata), " - "case when octet_length(log_cmddata) <= %d " - "then log_cmddata " - "else null end " + "log_actionseq, log_tablenspname, " + "log_tablerelname, log_cmdtype, " + "log_cmdupdncols, log_cmdargs " "from %s.sl_log_%d " "where log_origin = %d " "and log_tableid in (", - sync_max_rowsize, rtcfg_namespace, sl_log_no, node->no_id); for (tupno2 = 0; tupno2 < ntuples2; tupno2++) @@ -4279,14 +4270,12 @@ sync_event(SlonNode *node, SlonConn *local_conn, slon_mkquery(provider_query, "declare LOG cursor for " "select log_origin, log_txid, log_tableid, " - "log_actionseq, log_cmdtype, " - "octet_length(log_cmddata), " - "case when octet_length(log_cmddata) <= %d " - "then log_cmddata " - "else null end " + "log_actionseq, log_tablenspname, " + "log_tablerelname, log_cmdtype, " + "log_cmdupdncols, log_cmdargs " "from %s.sl_log_1 " "where false", - sync_max_rowsize, rtcfg_namespace); + rtcfg_namespace); } } @@ -4794,7 +4783,6 @@ sync_helper(void *cdata) PGresult *res = NULL; PGresult *res2 = NULL; - PGresult *res3 = NULL; int ntuples; int tupno; @@ -4870,33 +4858,33 @@ sync_helper(void *cdata) rtcfg_namespace); start_monitored_event(&pm); - res3 = PQexec(dbconn, dstring_data(&query)); + res2 = PQexec(dbconn, dstring_data(&query)); monitor_provider_query(&pm); - rc = PQresultStatus(res3); + rc = PQresultStatus(res2); if (rc != PGRES_TUPLES_OK) { slon_log(SLON_ERROR, "remoteWorkerThread_%d: \"%s\" %s %s\n", node->no_id, dstring_data(&query), PQresStatus(rc), - PQresultErrorMessage(res3)); - PQclear(res3); + PQresultErrorMessage(res2)); + PQclear(res2); errors++; break; } - if (PQntuples(res3) != 1) + if (PQntuples(res2) != 1) { slon_log(SLON_ERROR, "remoteWorkerThread_%d: \"%s\" %s returned %d tuples\n", node->no_id, dstring_data(&query), - PQresStatus(rc), PQntuples(res3)); - PQclear(res3); + PQresStatus(rc), PQntuples(res2)); + PQclear(res2); errors++; break; } - log_status = strtol(PQgetvalue(res3, 0, 0), NULL, 10); - PQclear(res3); + log_status = strtol(PQgetvalue(res2, 0, 0), NULL, 10); + PQclear(res2); slon_log(SLON_DEBUG2, "remoteWorkerThread_%d_%d: current remote log_status = %d\n", node->no_id, provider->no_id, log_status); @@ -5018,34 +5006,8 @@ sync_helper(void *cdata) * available line buffers. */ pthread_mutex_lock(&(wd->workdata_lock)); - if (data_line_alloc == 0 || - wd->workdata_largemem > sync_max_largemem) + if (data_line_alloc == 0) { - /* - * First make sure that the overall memory usage is inside - * bounds. - */ - if (wd->workdata_largemem > sync_max_largemem) - { - slon_log(SLON_DEBUG4, - "remoteHelperThread_%d_%d: wait for oversize memory to free\n", - node->no_id, provider->no_id); - - while (wd->workdata_largemem > sync_max_largemem && - wd->workgroup_status == SLON_WG_BUSY) - { - pthread_cond_wait(&(wd->linepool_cond), &(wd->workdata_lock)); - } - if (wd->workgroup_status != SLON_WG_BUSY) - { - slon_log(SLON_DEBUG4, - "remoteHelperThread_%d_%d: abort operation\n", - node->no_id, provider->no_id); - errors++; - break; - } - } - /* * Second make sure that we have at least 1 line buffer. */ @@ -5157,60 +5119,14 @@ sync_helper(void *cdata) int log_tableid = strtol(PQgetvalue(res, tupno, 2), NULL, 10); char *log_actionseq = PQgetvalue(res, tupno, 3); - char *log_cmdtype = PQgetvalue(res, tupno, 4); - int log_cmdsize = strtol(PQgetvalue(res, tupno, 5), - NULL, 10); - char *log_cmddata = PQgetvalue(res, tupno, 6); - int largemem = 0; + char *log_tablenspname = PQgetvalue(res, tupno, 4); + char *log_tablerelname = PQgetvalue(res, tupno, 5); + char *log_cmdtype = PQgetvalue(res, tupno, 6); + char *log_cmdupdncols = PQgetvalue(res, tupno, 7); + char *log_cmdargs = PQgetvalue(res, tupno, 8); tupno++; - if (log_cmdsize >= sync_max_rowsize) - { - (void) slon_mkquery(&query2, - "select log_cmddata " - "from %s.sl_log_1 " - "where log_origin = '%s' " - " and log_txid = '%s' " - " and log_actionseq = '%s' " - "UNION ALL " - "select log_cmddata " - "from %s.sl_log_2 " - "where log_origin = '%s' " - " and log_txid = '%s' " - " and log_actionseq = '%s'", - rtcfg_namespace, - log_origin, log_txid, log_actionseq, - rtcfg_namespace, - log_origin, log_txid, log_actionseq); - start_monitored_event(&pm); - res2 = PQexec(dbconn, dstring_data(&query2)); - monitor_largetuples(&pm); - - if (PQresultStatus(res2) != PGRES_TUPLES_OK) - { - slon_log(SLON_ERROR, "remoteHelperThread_%d_%d: \"%s\" %s", - node->no_id, provider->no_id, - dstring_data(&query), - PQresultErrorMessage(res2)); - PQclear(res2); - errors++; - break; - } - if (PQntuples(res2) != 1) - { - slon_log(SLON_ERROR, "remoteHelperThread_%d_%d: large log_cmddata for actionseq %s not found\n", - node->no_id, provider->no_id, - log_actionseq); - PQclear(res2); - errors++; - break; - } - - log_cmddata = PQgetvalue(res2, 0, 0); - largemem = log_cmdsize; - } - /* * This can happen if the table belongs to a set that * already has a better sync status than the event we're @@ -5220,8 +5136,6 @@ sync_helper(void *cdata) if (log_tableid >= wd->tab_fqname_size || wd->tab_fqname[log_tableid] == NULL) { - if (largemem > 0) - PQclear(res2); continue; } @@ -5234,59 +5148,24 @@ sync_helper(void *cdata) slon_appendquery(&(line->log), "insert into %s.sl_log_%d " " (log_origin, log_txid, log_tableid, " - " log_actionseq, log_cmdtype, " - " log_cmddata) values " - " ('%s', '%s', '%d', '%s', '%q', '%q');\n", + " log_actionseq, log_tablenspname, " + " log_tablerelname, log_cmdtype, " + " log_cmdupdncols, " + " log_cmdargs) values " + " ('%s', '%s', '%d', '%s', '%q', '%q', '%q', '%q', '%q');\n", rtcfg_namespace, wd->active_log_table, log_origin, log_txid, log_tableid, - log_actionseq, log_cmdtype, log_cmddata); - largemem *= 2; + log_actionseq, log_tablenspname, + log_tablerelname, log_cmdtype, + log_cmdupdncols, log_cmdargs); } - /* - * Add the actual replicating command to the line buffer - */ - line->line_largemem += largemem; - switch (*log_cmdtype) - { - case 'I': - slon_appendquery(&(line->data), - "insert into %s %s;\n", - wd->tab_fqname[log_tableid], - log_cmddata); - pm.num_inserts++; - break; - - case 'U': - slon_appendquery(&(line->data), - "update only %s set %s;\n", - wd->tab_fqname[log_tableid], - log_cmddata); - pm.num_updates++; - break; - - case 'D': - slon_appendquery(&(line->data), - "delete from only %s where %s;\n", - wd->tab_fqname[log_tableid], - log_cmddata); - pm.num_deletes++; - break; - case 'T': - slon_appendquery(&(line->data), - "%s;\n", - log_cmddata); - pm.num_truncates++; - break; - } line_ncmds++; if (line_ncmds >= SLON_COMMANDS_PER_LINE) { if (data_line_last >= data_line_alloc) { - if (largemem > 0) - PQclear(res2); break; } @@ -5300,29 +5179,6 @@ sync_helper(void *cdata) line_ncmds = 0; } - - /* - * If this was a large log_cmddata entry (> - * sync_max_rowsize), add this to the memory usage of the - * workgroup and check if we are exceeding limits. - */ - if (largemem > 0) - { - PQclear(res2); - pthread_mutex_lock(&(wd->workdata_lock)); - wd->workdata_largemem += largemem; - if (wd->workdata_largemem >= sync_max_largemem) - { - /* - * This is it ... we exit the loop here and wait - * for the worker to apply enough of the large - * rows first. - */ - pthread_mutex_unlock(&(wd->workdata_lock)); - break; - } - pthread_mutex_unlock(&(wd->workdata_lock)); - } } /* @@ -6151,10 +6007,3 @@ static void monitor_subscriber_iud(PerfMon *perf_info) { (perf_info->subscr_iud__c) ++; } -static void monitor_largetuples(PerfMon *perf_info) { - double diff; - gettimeofday(&(perf_info->now_t), NULL); - diff = TIMEVAL_DIFF(&(perf_info->prev_t), &(perf_info->now_t)); - (perf_info->large_tuples_t) += diff; - (perf_info->large_tuples_c) ++; -} diff --git a/src/slon/slon.h b/src/slon/slon.h index 173d9fe7..e4342017 100644 --- a/src/slon/slon.h +++ b/src/slon/slon.h @@ -557,8 +557,6 @@ extern void *remoteListenThread_main(void *cdata); * ---------- */ extern int sync_group_maxsize; -extern int sync_max_rowsize; -extern int sync_max_largemem; extern int explain_interval; diff --git a/tests/run_test.sh b/tests/run_test.sh index 12c2f07a..d8eb8bba 100755 --- a/tests/run_test.sh +++ b/tests/run_test.sh @@ -493,8 +493,6 @@ build_slonconf() echo "sync_interval=2010" >> ${CONFFILE} echo "sync_interval_timeout=15000" >> ${CONFFILE} echo "sync_group_maxsize=8" >> ${CONFFILE} - echo "sync_max_rowsize=4096" >> ${CONFFILE} - echo "sync_max_largemem=1048576" >> ${CONFFILE} echo "syslog=1" >> ${CONFFILE} echo "log_timestamp=true" >> ${CONFFILE} echo "log_timestamp_format='%Y-%m-%d %H:%M:%S %Z'" >> ${CONFFILE} diff --git a/tools/check_unconfirmed_log.sh b/tools/check_unconfirmed_log.sh new file mode 100644 index 00000000..7cf6f767 --- /dev/null +++ b/tools/check_unconfirmed_log.sh @@ -0,0 +1,94 @@ +#!/bin/sh + +if [ $# -lt 2 ] ; then + echo "usage: `basename $0` clustername [psql-options] dbname" >&2 + exit 1 +fi + +cluster=$1; shift +nsp="\"_${cluster}\"" +psql="psql -qtA $*" +rc=0 + +# +# Get a list of all nodes that are origin to at least one set +# +origins=`${psql} <<_EOF_ + select distinct set_origin from ${nsp}.sl_set; +_EOF_ +` + +for origin in $origins ; do + # + # Per origin determine which is the highest event seqno + # that is confirmed by all subscribers to any of the origins + # sets. + # + all_confirmed=`${psql} <<_EOF_ + select min(max_seqno) from ( + select con_received, max(con_seqno) as max_seqno + from ${nsp}.sl_confirm + where con_origin = '${origin}' + and con_received in ( + select distinct sub_receiver from ${nsp}.sl_subscribe SUB, + ${nsp}.sl_set SET + where SET.set_id = SUB.sub_set + and SET.set_origin = '${origin}' + ) + group by con_received + ) as maxconfirmed; +_EOF_ +` + + # + # Get the txid snapshot that corresponds with that event + # + all_snapshot=`${psql} <<_EOF_ + select ev_snapshot from ${nsp}.sl_event + where ev_origin = '${origin}' + and ev_seqno = '${all_confirmed}'; +_EOF_ +` + + # + # Count the number of unconfirmed sl_log rows that appeared after + # that event was created. + # + unconfirmed=`${psql} <<_EOF_ + select count(*) from ( + select log_origin from ${nsp}.sl_log_1 + where log_origin = '${origin}' + and log_txid >= "pg_catalog".txid_snapshot_xmax('${all_snapshot}') + union all + select log_origin from ${nsp}.sl_log_1 + where log_origin = '${origin}' + and log_txid in ( + select * from "pg_catalog".txid_snapshot_xip('${all_snapshot}') + ) + union all + select log_origin from ${nsp}.sl_log_2 + where log_origin = '${origin}' + and log_txid >= "pg_catalog".txid_snapshot_xmax('${all_snapshot}') + union all + select log_origin from ${nsp}.sl_log_2 + where log_origin = '${origin}' + and log_txid in ( + select * from "pg_catalog".txid_snapshot_xip('${all_snapshot}') + ) + ) AS cnt; +_EOF_ +` + # + # Report the result + # + if [ $unconfirmed -gt 0 ] ; then + echo "origin $origin: ERROR - node has $unconfirmed unconfirmed log rows" + echo "origin $origin: the last event confirmed by all subscribers" + echo "origin $origin: is $all_confirmed" + rc=2 + else + echo "origin $origin: OK" + fi +done + +exit ${rc} diff --git a/tools/mkservice/slon-mkservice.sh b/tools/mkservice/slon-mkservice.sh index ac27e6a7..1e203230 100644 --- a/tools/mkservice/slon-mkservice.sh +++ b/tools/mkservice/slon-mkservice.sh @@ -304,20 +304,6 @@ log_timestamp 0 # multilog will insert a tai64n timestamp # If the value is left empty, this logic will be ignored. #lag_interval -# Size above which an sl_log_? row's log_cmddata is considered large. Up to -# 500 rows of this size are allowed in memory at once. Rows larger than that -# count into the sync_max_largemem space allocated and free()'ed on demand. -# The default value is 8192, meaning that your expected memory consumption -# (for the LOG cursor) should not exceed 8MB. -#sync_max_rowsize 8192 - -# Maximum memory allocated for large rows, where log_cmddata are larger than -# sync_max_rowsize. Note that the algorithm reads rows until after this value -# is exceeded. Otherwise, a tuple larger than this value would stall -# replication. As a result, don't assume that memory consumption will remain -# smaller than this value. The default value is 5242880. -#sync_max_largemem 5242880 - # How long should the remote listener wait before treating the event selection # criteria as having timed out? # Range: [30-30000], default 300 diff --git a/tools/mkslonconf.sh b/tools/mkslonconf.sh index 5995dcef..c9f5359c 100755 --- a/tools/mkslonconf.sh +++ b/tools/mkslonconf.sh @@ -65,20 +65,6 @@ cat <<_EOF_ > $conffile # Range: [0,100], default: 6 #sync_group_maxsize=6 -# Size above which an sl_log_? row's log_cmddata is considered large. -# Up to 500 rows of this size are allowed in memory at once. Rows larger -# than that count into the sync_max_largemem space allocated and free'd -# on demand. -# Range: [1024,32768], default: 8192 -#sync_max_rowsize=8192 - -# Maximum amount of memory allowed for large rows. Note that the algorithm -# will stop fetching rows AFTER this amount is exceeded, not BEFORE. This -# is done to ensure that a single row exceeding this limit alone does not -# stall replication. -# Range: [1048576,1073741824], default: 5242880 -#sync_max_largemem=5242880 - # If this parameter is 1, messages go both to syslog and the standard # output. A value of 2 sends output only to syslog (some messages will # still go to the standard output/error). The default is 0, which means