Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

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
funbringer merged 16 commits intopostgrespro:masterfromzilder:logical
Jun 1, 2018
Merged
Show file tree
Hide file tree
Changes fromall commits
Commits
Show all changes
16 commits
Select commitHold shift + click to select a range
ca5b546
Logical replication
zilderMar 16, 2018
bb01c7d
Skip logical replication test on PostgreSQL versions below 10
zilderMar 16, 2018
782484b
Fix logical replication for python3
zilderMar 16, 2018
f8b95c6
Merge branch 'master' into logical
zilderMar 19, 2018
b1cba73
Some minor refactoring of logical replication
zilderMar 22, 2018
954879a
Added options_string() func
zilderMar 22, 2018
0741c70
Merge branch 'master' into logical
zilderMar 22, 2018
f4e0bd0
Minor refactoring
zilderMar 22, 2018
f48623b
Add failing logical replication test for 9.6
zilderMar 22, 2018
138c6cc
Added test for Publication.add_tables()
zilderMar 22, 2018
f652bf4
Merge branch 'master' into logical. Also testgres.pubsub was added to…
zilderMar 26, 2018
bc1002f
Additional subscription catchup test
zilderMar 27, 2018
08ed6ef
Some refactoring of logical replication API and formatting
zilderMar 27, 2018
4b279ef
minor refactoring
zilderMay 31, 2018
d60cdcb
Merge branch 'master' into logical
zilderMay 31, 2018
50e02ff
change safe_psql() call to execute() in pubsub.py
zilderJun 1, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 16 additions & 1 deletiondocs/source/testgres.rst
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -59,4 +59,19 @@ testgres.node
.. automethod:: __init__

.. autoclass:: testgres.node.ProcessProxy
:members:
:members:

testgres.pubsub
---------------

.. automodule:: testgres.pubsub

.. autoclass:: testgres.node.Publication
:members:

.. automethod:: __init__

.. autoclass:: testgres.node.Subscription
:members:

.. automethod:: __init__
8 changes: 7 additions & 1 deletiontestgres/connection.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -27,7 +27,12 @@ class NodeConnection(object):
Transaction wrapper returned by Node
"""

def __init__(self, node, dbname=None, username=None, password=None):
def __init__(self,
node,
dbname=None,
username=None,
password=None,
autocommit=False):

# Set default arguments
dbname = dbname or default_dbname()
Expand All@@ -42,6 +47,7 @@ def __init__(self, node, dbname=None, username=None, password=None):
host=node.host,
port=node.port)

self._connection.autocommit = autocommit
self._cursor = self.connection.cursor()

@property
Expand Down
3 changes: 3 additions & 0 deletionstestgres/consts.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -29,3 +29,6 @@
MAX_REPLICATION_SLOTS = 10
MAX_WAL_SENDERS = 10
WAL_KEEP_SEGMENTS = 20

# logical replication settings
LOGICAL_REPL_MAX_CATCHUP_ATTEMPTS = 60
93 changes: 72 additions & 21 deletionstestgres/node.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -57,11 +57,14 @@
QueryException, \
StartNodeException, \
TimeoutException, \
InitNodeException, \
TestgresException, \
BackupException

from .logger import TestgresLogger

from .pubsub import Publication, Subscription

from .utils import \
eprint, \
get_bin_path, \
Expand All@@ -70,6 +73,7 @@
reserve_port, \
release_port, \
execute_utility, \
options_string, \
clean_on_error

from .backup import NodeBackup
Expand DownExpand Up@@ -300,24 +304,24 @@ def _create_recovery_conf(self, username, slot=None):
master = self.master
assert master is not None

conninfo =(
u"application_name={} "
u"port={} "
u"user={} "
).format(self.name, master.port, username) # yapf: disable
conninfo ={
"application_name": self.name,
"port": master.port,
"user": username
} # yapf: disable

# host is tricky
try:
import ipaddress
ipaddress.ip_address(master.host)
conninfo += u"hostaddr={}".format(master.host)
conninfo["hostaddr"] =master.host
except ValueError:
conninfo += u"host={}".format(master.host)
conninfo["host"] =master.host

line = (
"primary_conninfo='{}'\n"
"standby_mode=on\n"
).format(conninfo) # yapf: disable
).format(options_string(**conninfo)) # yapf: disable

if slot:
# Connect to master for some additional actions
Expand DownExpand Up@@ -413,6 +417,7 @@ def default_conf(self,
fsync=False,
unix_sockets=True,
allow_streaming=True,
allow_logical=False,
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others.Learn more.

Why don't we enable this by default?

Copy link
CollaboratorAuthor

Choose a reason for hiding this comment

The 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.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others.Learn more.

Ah, i see.

log_statement='all'):
"""
Apply default settings to this node.
Expand All@@ -421,6 +426,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:
Expand DownExpand Up@@ -497,6 +503,13 @@ def get_auth_method(t):
WAL_KEEP_SEGMENTS,
wal_level)) # yapf: disable

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")
Expand DownExpand Up@@ -937,13 +950,14 @@ def poll_query_until(self,
if res is None:
raise QueryException('Query returned None', query)

if len(res) == 0:
raise QueryException('Query returned 0 rows', query)

if len(res[0]) == 0:
raise QueryException('Query returned 0 columns', query)

if res[0][0] == expected:
# result set is not empty
if len(res):
if len(res[0]) == 0:
raise QueryException('Query returned 0 columns', query)
if res[0][0] == expected:
return # done
# empty result set is considered as None
elif expected is None:
return # done

except ProgrammingError as e:
Expand DownExpand Up@@ -982,13 +996,11 @@ def execute(self,

with self.connect(dbname=dbname,
username=username,
password=password) as node_con: # yapf: disable
password=password,
autocommit=commit) as node_con: # yapf: disable

res = node_con.execute(query)

if commit:
node_con.commit()

return res

def backup(self, **kwargs):
Expand DownExpand Up@@ -1052,6 +1064,37 @@ def catchup(self, dbname=None, username=None):
except Exception as e:
raise_from(CatchUpException("Failed to catch up", poll_lsn), e)

def publish(self, name, **kwargs):
"""
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(name=name, node=self, **kwargs)

def subscribe(self, publication, name, dbname=None, username=None,
**params):
"""
Create subscription for logical replication

Args:
name: subscription name
publication: publication object obtained from publish()
dbname: database name
username: replication username
params: subscription parameters (see documentation on `CREATE SUBSCRIPTION
<https://www.postgresql.org/docs/current/static/sql-createsubscription.html>`_
for details)
"""
# yapf: disable
return Subscription(name=name, node=self, publication=publication,
dbname=dbname, username=username, **params)
# yapf: enable

def pgbench(self,
dbname=None,
username=None,
Expand DownExpand Up@@ -1150,14 +1193,21 @@ def pgbench_run(self, dbname=None, username=None, options=[], **kwargs):

return execute_utility(_params, self.utils_log_file)

def connect(self, dbname=None, username=None, password=None):
def connect(self,
dbname=None,
username=None,
password=None,
autocommit=False):
"""
Connect to a database.

Args:
dbname: database name to connect to.
username: database user name.
password: user's password.
autocommit: commit each statement automatically. Also it should be
set to `True` for statements requiring to be run outside
a transaction? such as `VACUUM` or `CREATE DATABASE`.

Returns:
An instance of :class:`.NodeConnection`.
Expand All@@ -1166,4 +1216,5 @@ def connect(self, dbname=None, username=None, password=None):
return NodeConnection(node=self,
dbname=dbname,
username=username,
password=password) # yapf: disable
password=password,
autocommit=autocommit) # yapf: disable
Loading
[8]ページ先頭
©2009-2025 Movatter.jp