- Notifications
You must be signed in to change notification settings - Fork35
Logical replication support#42
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to ourterms of service andprivacy statement. We’ll occasionally send you account related emails.
Already on GitHub?Sign in to your account
Merged
Uh oh!
There was an error while loading.Please reload this page.
Merged
Changes from1 commit
Commits
Show all changes
16 commits Select commitHold shift + click to select a range
ca5b546 Logical replication
zilderbb01c7d Skip logical replication test on PostgreSQL versions below 10
zilder782484b Fix logical replication for python3
zilderf8b95c6 Merge branch 'master' into logical
zilderb1cba73 Some minor refactoring of logical replication
zilder954879a Added options_string() func
zilder0741c70 Merge branch 'master' into logical
zilderf4e0bd0 Minor refactoring
zilderf48623b Add failing logical replication test for 9.6
zilder138c6cc Added test for Publication.add_tables()
zilderf652bf4 Merge branch 'master' into logical. Also testgres.pubsub was added to…
zilderbc1002f Additional subscription catchup test
zilder08ed6ef Some refactoring of logical replication API and formatting
zilder4b279ef minor refactoring
zilderd60cdcb Merge branch 'master' into logical
zilder50e02ff change safe_psql() call to execute() in pubsub.py
zilderFile filter
Filter by extension
Conversations
Failed to load comments.
Loading
Uh oh!
There was an error while loading.Please reload this page.
Jump to
Jump to file
Failed to load files.
Loading
Uh oh!
There was an error while loading.Please reload this page.
Diff view
Diff view
NextNext commit
Logical replication
- Loading branch information
Uh oh!
There was an error while loading.Please reload this page.
commitca5b5465a07e45f91f6258bfcc73a4dd06ae9e8c
There are no files selected for viewing
58 changes: 55 additions & 3 deletionstestgres/node.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -48,10 +48,13 @@ | ||
| ExecUtilException, \ | ||
| QueryException, \ | ||
| StartNodeException, \ | ||
| TimeoutException, \ | ||
| InitNodeException | ||
| from .logger import TestgresLogger | ||
| from .pubsub import Publication, Subscription | ||
| from .utils import \ | ||
| eprint, \ | ||
| get_bin_path, \ | ||
| @@ -278,6 +281,7 @@ def default_conf(self, | ||
| fsync=False, | ||
| unix_sockets=True, | ||
| allow_streaming=True, | ||
| allow_logical=False, | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others.Learn more. Why don't we enable this by default? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others.Learn more. Because it is not supported on postgres versions below 10 and there is specific message when someone's trying to enable this feature on those versions. Besides it produces extra WAL data and hence could work slightly slower. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others.Learn more. Ah, i see. | ||
| log_statement='all'): | ||
| """ | ||
| Apply default settings to this node. | ||
| @@ -286,6 +290,7 @@ def default_conf(self, | ||
| fsync: should this node use fsync to keep data safe? | ||
| unix_sockets: should we enable UNIX sockets? | ||
| allow_streaming: should this node add a hba entry for replication? | ||
| allow_logical: can this node be used as a logical replication publisher? | ||
| log_statement: one of ('all', 'off', 'mod', 'ddl'). | ||
| Returns: | ||
| @@ -365,6 +370,12 @@ def get_auth_method(t): | ||
| wal_keep_segments, | ||
| wal_level)) | ||
| if allow_logical: | ||
| if not pg_version_ge('10'): | ||
| raise InitNodeException("Logical replication is only " | ||
| "available for Postgres 10 and newer") | ||
| conf.write(u"wal_level = logical\n") | ||
| # disable UNIX sockets if asked to | ||
| if not unix_sockets: | ||
| conf.write(u"unix_socket_directories = ''\n") | ||
| @@ -751,7 +762,8 @@ def poll_query_until(self, | ||
| expected=True, | ||
| commit=True, | ||
| raise_programming_error=True, | ||
| raise_internal_error=True, | ||
| zero_rows_is_ok=False): | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others.Learn more. Isn't There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others.Learn more. In | ||
| """ | ||
| Run a query once per second until it returns 'expected'. | ||
| Query should return a single value (1 row, 1 column). | ||
| @@ -788,7 +800,12 @@ def poll_query_until(self, | ||
| raise QueryException('Query returned None', query) | ||
| if len(res) == 0: | ||
| if zero_rows_is_ok: | ||
| time.sleep(sleep_time) | ||
| attempts += 1 | ||
| continue | ||
| else: | ||
| raise QueryException('Query returned 0 rows', query) | ||
| if len(res[0]) == 0: | ||
| raise QueryException('Query returned 0 columns', query) | ||
| @@ -902,6 +919,41 @@ def catchup(self, dbname=None, username=None): | ||
| except Exception as e: | ||
| raise_from(CatchUpException("Failed to catch up", poll_lsn), e) | ||
| def publish(self, | ||
| pubname, | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others.Learn more. Maybe change | ||
| tables=None, | ||
| dbname=None, | ||
| username=None): | ||
| """ | ||
| Create publication for logical replication | ||
| Args: | ||
| pubname: publication name | ||
| tables: tables names list | ||
| dbname: database name where objects or interest are located | ||
| username: replication username | ||
| """ | ||
| return Publication(pubname, self, tables, dbname, username) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others.Learn more. Could you please use keyword args instead of positional args? | ||
| def subscribe(self, | ||
| publication, | ||
| subname, | ||
| dbname=None, | ||
| username=None, | ||
| **kwargs): | ||
| """ | ||
| Create subscription for logical replication | ||
| Args: | ||
| subname: subscription name | ||
| publication: publication object obtained from publish() | ||
| """ | ||
| return Subscription(subname, self, publication, | ||
| dbname=dbname, | ||
| username=username, | ||
| **kwargs) | ||
| def pgbench(self, | ||
| dbname=None, | ||
| username=None, | ||
161 changes: 161 additions & 0 deletionstestgres/pubsub.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,161 @@ | ||
| # coding: utf-8 | ||
| from six import raise_from | ||
| from .defaults import default_dbname, default_username | ||
| from .exceptions import CatchUpException | ||
| from .utils import pg_version_ge | ||
| class Publication(object): | ||
| def __init__(self, pubname, node, tables=None, dbname=None, username=None): | ||
| """ | ||
| Constructor | ||
| Args: | ||
| pubname: publication name | ||
| node: publisher's node | ||
| tables: tables list or None for all tables | ||
| dbname: database name used to connect and perform subscription | ||
| username: username used to connect to the database | ||
| """ | ||
| self.name = pubname | ||
| self.node = node | ||
| self.dbname = dbname or default_dbname() | ||
| self.username = username or default_username() | ||
| # create publication in database | ||
| t = 'table ' + ', '.join(tables) if tables else 'all tables' | ||
| query = "create publication {} for {}" | ||
| node.safe_psql(query.format(pubname, t), | ||
| dbname=dbname, | ||
| username=username) | ||
| def close(self, dbname=None, username=None): | ||
| """ | ||
| Drop publication | ||
| """ | ||
| self.node.safe_psql("drop publication {}".format(self.name), | ||
| dbname=dbname, username=username) | ||
| def add_tables(self, tables, dbname=None, username=None): | ||
| """ | ||
| Add tables | ||
| Args: | ||
| tables: a list of tables to add to the publication | ||
| """ | ||
| if not tables: | ||
| raise ValueError("Tables list is empty") | ||
| query = "alter publication {} add table {}" | ||
| self.node.safe_psql(query.format(self.name, ', '.join(tables)), | ||
| dbname=dbname or self.dbname, | ||
| username=username or self.username) | ||
| class Subscription(object): | ||
| def __init__(self, | ||
| subname, | ||
| node, | ||
| publication, | ||
| dbname=None, | ||
| username=None, | ||
| **kwargs): | ||
| """ | ||
| Constructor | ||
| Args: | ||
| subname: subscription name | ||
| node: subscriber's node | ||
| publication: Publication object we are subscribing to | ||
| dbname: database name used to connect and perform subscription | ||
| username: username used to connect to the database | ||
| **kwargs: subscription parameters (see CREATE SUBSCRIPTION | ||
| in PostgreSQL documentation for more information) | ||
| """ | ||
| self.name = subname | ||
| self.node = node | ||
| self.pub = publication | ||
| # connection info | ||
| conninfo = ( | ||
| u"dbname={} user={} host={} port={}" | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others.Learn more. I think we could extract common | ||
| ).format(self.pub.dbname, | ||
| self.pub.username, | ||
| self.pub.node.host, | ||
| self.pub.node.port) | ||
| query = ( | ||
| "create subscription {} connection '{}' publication {}" | ||
| ).format(subname, conninfo, self.pub.name) | ||
| # additional parameters | ||
| if kwargs: | ||
| params = ','.join('{}={}'.format(k, v) for k, v in kwargs.iteritems()) | ||
| query += " with ({})".format(params) | ||
| node.safe_psql(query, dbname=dbname, username=username) | ||
| def disable(self, dbname=None, username=None): | ||
| """ | ||
| Disables the running subscription. | ||
| """ | ||
| query = "alter subscription {} disable" | ||
| self.node.safe_psql(query.format(self.name), | ||
| dbname=None, | ||
| username=None) | ||
| def enable(self, dbname=None, username=None): | ||
| """ | ||
| Enables the previously disabled subscription. | ||
| """ | ||
| query = "alter subscription {} enable" | ||
| self.node.safe_psql(query.format(self.name), | ||
| dbname=None, | ||
| username=None) | ||
| def refresh(self, copy_data=True, dbname=None, username=None): | ||
| """ | ||
| Disables the running subscription. | ||
| """ | ||
| query = "alter subscription {} refresh publication with (copy_data={})" | ||
| self.node.safe_psql(query.format(self.name, copy_data), | ||
| dbname=dbname, | ||
| username=username) | ||
| def close(self, dbname=None, username=None): | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others.Learn more. Shouldn't it be called | ||
| """ | ||
| Drops subscription | ||
| """ | ||
| self.node.safe_psql("drop subscription {}".format(self.name), | ||
| dbname=dbname, username=username) | ||
| def catchup(self, username=None): | ||
| """ | ||
| Wait until subscription catches up with publication. | ||
| Args: | ||
| username: remote node's user name | ||
| """ | ||
| if pg_version_ge('10'): | ||
| query = ( | ||
| "select pg_current_wal_lsn() - replay_lsn = 0 " | ||
| "from pg_stat_replication where application_name = '{}'" | ||
| ).format(self.name) | ||
| else: | ||
| query = ( | ||
| "select pg_current_xlog_location() - replay_location = 0 " | ||
| "from pg_stat_replication where application_name = '{}'" | ||
| ).format(self.name) | ||
| try: | ||
| # wait until this LSN reaches subscriber | ||
| self.pub.node.poll_query_until( | ||
| query=query, | ||
| dbname=self.pub.dbname, | ||
| username=username or self.pub.username, | ||
| max_attempts=60, | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others.Learn more. IMHO it's better to replace | ||
| zero_rows_is_ok=True) # statistics may have not updated yet | ||
| except Exception as e: | ||
| raise_from(CatchUpException("Failed to catch up", query), e) | ||
66 changes: 66 additions & 0 deletionstests/test_simple.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.