mirror of
https://github.com/hanzoai/datastore.git
synced 2026-09-26 13:55:07 +00:00
Groene AIandClaude Opus 4.8
49298e89bd
The test drives dict0, a config-based PostgreSQL dictionary with LIFETIME(1) and an invalidate_query, and asserts that changing a row other than the invalidate row (id != 0) does not reload the dictionary. SYSTEM RELOAD DICTIONARY performs a forced full reload but does not run the invalidate query, so the stored invalidate response is left empty. The first periodic check after the forced reload therefore always sees a "modified" source and reloads once more. If that follow-up reload lands after the test has changed the other rows (value * 2 WHERE id != 0), the dictionary picks up value 2 for id = 1 and the "no update should happen" assertion fails with '2' == '1'. On slow sanitizer / old-analyzer runs the periodic check (every 5s) lines up with the fixed sleep(5) window often enough to fail intermittently. It never fails on master. Wait for the invalidate baseline to settle (last_successful_update_time stops advancing across a full check period) after each reload, and verify the "no update" case directly by asserting last_successful_update_time does not advance. This drains the guaranteed post-forced-reload reload before the assertion and keeps the original intent: a change to the invalidate row triggers a reload, a change to other rows does not. Co-Authored-By: Claude Opus 4.8 <[email protected]>
777 lines
24 KiB
Python
777 lines
24 KiB
Python
import logging
|
|
import time
|
|
import socket
|
|
from contextlib import contextmanager
|
|
|
|
import pytest
|
|
|
|
from helpers.cluster import ClickHouseCluster
|
|
from helpers.config_cluster import pg_pass
|
|
from helpers.postgres_utility import get_postgres_conn
|
|
from helpers.port_forward import PortForward
|
|
from helpers.test_tools import assert_eq_with_retry
|
|
|
|
cluster = ClickHouseCluster(__file__)
|
|
node1 = cluster.add_instance(
|
|
"node1",
|
|
main_configs=[
|
|
"configs/config.xml",
|
|
"configs/dictionaries/postgres_dict.xml",
|
|
"configs/named_collections.xml",
|
|
"configs/bg_reconnect.xml",
|
|
],
|
|
with_postgres=True,
|
|
with_postgres_cluster=True,
|
|
)
|
|
|
|
|
|
def create_postgres_db(conn, name):
|
|
cursor = conn.cursor()
|
|
cursor.execute(f"CREATE DATABASE {name}")
|
|
|
|
|
|
def create_postgres_table(cursor, table_name):
|
|
cursor.execute(
|
|
f"""
|
|
CREATE TABLE IF NOT EXISTS {table_name} (
|
|
id Integer NOT NULL, key Integer NOT NULL, value Integer NOT NULL, PRIMARY KEY (id))
|
|
"""
|
|
)
|
|
|
|
|
|
def create_and_fill_postgres_table(cursor, table_name, port, host):
|
|
create_postgres_table(cursor, table_name)
|
|
# Fill postgres table using clickhouse postgres table function and check
|
|
table_func = f"""postgresql('{host}:{port}', 'postgres_database', '{table_name}', 'postgres', '{pg_pass}')"""
|
|
node1.query(
|
|
f"""INSERT INTO TABLE FUNCTION {table_func} SELECT number, number, number from numbers(10000)"""
|
|
)
|
|
result = node1.query(f"SELECT count() FROM {table_func}")
|
|
assert result.rstrip() == "10000"
|
|
|
|
|
|
def create_dict(table_name, index=0):
|
|
node1.query(
|
|
f"""
|
|
CREATE TABLE IF NOT EXISTS `test`.`dict_table_{table_name}` (
|
|
`key` UInt32, `value` UInt32
|
|
) ENGINE = Dictionary(dict{str(index)})
|
|
"""
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def started_cluster():
|
|
try:
|
|
cluster.start()
|
|
node1.query("DROP DATABASE IF EXISTS test")
|
|
node1.query("CREATE DATABASE test")
|
|
|
|
postgres_conn = get_postgres_conn(
|
|
ip=cluster.postgres_ip, port=cluster.postgres_port
|
|
)
|
|
logging.debug("postgres1 connected")
|
|
create_postgres_db(postgres_conn, "postgres_database")
|
|
|
|
postgres2_conn = get_postgres_conn(
|
|
ip=cluster.postgres2_ip, port=cluster.postgres_port
|
|
)
|
|
logging.debug("postgres2 connected")
|
|
create_postgres_db(postgres2_conn, "postgres_database")
|
|
|
|
yield cluster
|
|
|
|
finally:
|
|
cluster.shutdown()
|
|
|
|
|
|
def test_load_dictionaries(started_cluster):
|
|
conn = get_postgres_conn(
|
|
ip=started_cluster.postgres_ip,
|
|
database=True,
|
|
port=started_cluster.postgres_port,
|
|
)
|
|
cursor = conn.cursor()
|
|
table_name = "test0"
|
|
create_and_fill_postgres_table(
|
|
cursor,
|
|
table_name,
|
|
port=started_cluster.postgres_port,
|
|
host=started_cluster.postgres_ip,
|
|
)
|
|
create_dict(table_name)
|
|
dict_name = "dict0"
|
|
|
|
node1.query(f"SYSTEM RELOAD DICTIONARY {dict_name}")
|
|
assert (
|
|
node1.query(f"SELECT count() FROM `test`.`dict_table_{table_name}`").rstrip()
|
|
== "10000"
|
|
)
|
|
assert (
|
|
node1.query(f"SELECT dictGetUInt32('{dict_name}', 'key', toUInt64(0))") == "0\n"
|
|
)
|
|
assert (
|
|
node1.query(f"SELECT dictGetUInt32('{dict_name}', 'value', toUInt64(9999))")
|
|
== "9999\n"
|
|
)
|
|
|
|
cursor.execute(f"DROP TABLE IF EXISTS {table_name}")
|
|
node1.query(f"DROP TABLE IF EXISTS {table_name}")
|
|
node1.query(f"DROP DICTIONARY IF EXISTS {dict_name}")
|
|
|
|
|
|
def test_postgres_dictionaries_custom_query_full_load(started_cluster):
|
|
conn = get_postgres_conn(
|
|
ip=started_cluster.postgres_ip,
|
|
database=True,
|
|
port=started_cluster.postgres_port,
|
|
)
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute(
|
|
"CREATE TABLE IF NOT EXISTS test_table_1 (id Integer, value_1 Text);"
|
|
)
|
|
cursor.execute(
|
|
"CREATE TABLE IF NOT EXISTS test_table_2 (id Integer, value_2 Text);"
|
|
)
|
|
cursor.execute("INSERT INTO test_table_1 VALUES (1, 'Value_1');")
|
|
cursor.execute("INSERT INTO test_table_2 VALUES (1, 'Value_2');")
|
|
|
|
query = node1.query
|
|
query(
|
|
f"""
|
|
CREATE DICTIONARY test_dictionary_custom_query
|
|
(
|
|
id UInt64,
|
|
value_1 String,
|
|
value_2 String
|
|
)
|
|
PRIMARY KEY id
|
|
LAYOUT(FLAT())
|
|
SOURCE(PostgreSQL(
|
|
DB 'postgres_database'
|
|
HOST '{started_cluster.postgres_ip}'
|
|
PORT {started_cluster.postgres_port}
|
|
USER 'postgres'
|
|
PASSWORD '{pg_pass}'
|
|
QUERY $doc$SELECT id, value_1, value_2 FROM test_table_1 INNER JOIN test_table_2 USING (id);$doc$))
|
|
LIFETIME(0)
|
|
"""
|
|
)
|
|
|
|
result = query("SELECT id, value_1, value_2 FROM test_dictionary_custom_query")
|
|
|
|
assert result == "1\tValue_1\tValue_2\n"
|
|
|
|
query("DROP DICTIONARY test_dictionary_custom_query;")
|
|
|
|
cursor.execute("DROP TABLE test_table_2;")
|
|
cursor.execute("DROP TABLE test_table_1;")
|
|
|
|
|
|
def test_postgres_dictionaries_custom_query_partial_load_simple_key(started_cluster):
|
|
conn = get_postgres_conn(
|
|
ip=started_cluster.postgres_ip,
|
|
database=True,
|
|
port=started_cluster.postgres_port,
|
|
)
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute(
|
|
"CREATE TABLE IF NOT EXISTS test_table_1 (id Integer, value_1 Text);"
|
|
)
|
|
cursor.execute(
|
|
"CREATE TABLE IF NOT EXISTS test_table_2 (id Integer, value_2 Text);"
|
|
)
|
|
cursor.execute("INSERT INTO test_table_1 VALUES (1, 'Value_1');")
|
|
cursor.execute("INSERT INTO test_table_2 VALUES (1, 'Value_2');")
|
|
|
|
query = node1.query
|
|
query(
|
|
f"""
|
|
CREATE DICTIONARY test_dictionary_custom_query
|
|
(
|
|
id UInt64,
|
|
value_1 String,
|
|
value_2 String
|
|
)
|
|
PRIMARY KEY id
|
|
LAYOUT(DIRECT())
|
|
SOURCE(PostgreSQL(
|
|
DB 'postgres_database'
|
|
HOST '{started_cluster.postgres_ip}'
|
|
PORT {started_cluster.postgres_port}
|
|
USER 'postgres'
|
|
PASSWORD '{pg_pass}'
|
|
QUERY $doc$SELECT id, value_1, value_2 FROM test_table_1 INNER JOIN test_table_2 USING (id) WHERE {{condition}};$doc$))
|
|
"""
|
|
)
|
|
|
|
result = query(
|
|
"SELECT dictGet('test_dictionary_custom_query', ('value_1', 'value_2'), toUInt64(1))"
|
|
)
|
|
|
|
assert result == "('Value_1','Value_2')\n"
|
|
|
|
query("DROP DICTIONARY test_dictionary_custom_query;")
|
|
|
|
cursor.execute("DROP TABLE test_table_2;")
|
|
cursor.execute("DROP TABLE test_table_1;")
|
|
|
|
|
|
def test_postgres_dictionaries_custom_query_partial_load_complex_key(started_cluster):
|
|
conn = get_postgres_conn(
|
|
ip=started_cluster.postgres_ip,
|
|
database=True,
|
|
port=started_cluster.postgres_port,
|
|
)
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute(
|
|
"CREATE TABLE IF NOT EXISTS test_table_1 (id Integer, key Text, value_1 Text);"
|
|
)
|
|
cursor.execute(
|
|
"CREATE TABLE IF NOT EXISTS test_table_2 (id Integer, key Text, value_2 Text);"
|
|
)
|
|
cursor.execute("INSERT INTO test_table_1 VALUES (1, 'Key', 'Value_1');")
|
|
cursor.execute("INSERT INTO test_table_2 VALUES (1, 'Key', 'Value_2');")
|
|
|
|
query = node1.query
|
|
query(
|
|
f"""
|
|
CREATE DICTIONARY test_dictionary_custom_query
|
|
(
|
|
id UInt64,
|
|
key String,
|
|
value_1 String,
|
|
value_2 String
|
|
)
|
|
PRIMARY KEY id, key
|
|
LAYOUT(COMPLEX_KEY_DIRECT())
|
|
SOURCE(PostgreSQL(
|
|
DB 'postgres_database'
|
|
HOST '{started_cluster.postgres_ip}'
|
|
PORT {started_cluster.postgres_port}
|
|
USER 'postgres'
|
|
PASSWORD '{pg_pass}'
|
|
QUERY $doc$SELECT id, key, value_1, value_2 FROM test_table_1 INNER JOIN test_table_2 USING (id, key) WHERE {{condition}};$doc$))
|
|
"""
|
|
)
|
|
|
|
result = query(
|
|
"SELECT dictGet('test_dictionary_custom_query', ('value_1', 'value_2'), (toUInt64(1), 'Key'))"
|
|
)
|
|
|
|
assert result == "('Value_1','Value_2')\n"
|
|
|
|
query("DROP DICTIONARY test_dictionary_custom_query;")
|
|
|
|
cursor.execute("DROP TABLE test_table_2;")
|
|
cursor.execute("DROP TABLE test_table_1;")
|
|
|
|
|
|
def test_postgres_dict_complex_key_with_single_quote(started_cluster):
|
|
"""Regression test: string keys containing single quotes must not cause SQL injection.
|
|
|
|
ExternalQueryBuilder previously used backslash escaping (\\') which PostgreSQL
|
|
(standard_conforming_strings=on by default since 9.1) treats as a literal backslash,
|
|
breaking out of the string literal and allowing injection.
|
|
The fix switches to SQL-standard quote doubling ('').
|
|
"""
|
|
conn = get_postgres_conn(
|
|
ip=started_cluster.postgres_ip,
|
|
database=True,
|
|
port=started_cluster.postgres_port,
|
|
)
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute(
|
|
"CREATE TABLE IF NOT EXISTS test_single_quote (key Text PRIMARY KEY, value Text);"
|
|
)
|
|
cursor.execute("INSERT INTO test_single_quote VALUES ('it''s a key', 'found it');")
|
|
cursor.execute(
|
|
"INSERT INTO test_single_quote VALUES ('normal', 'normal value');"
|
|
)
|
|
|
|
query = node1.query
|
|
query(
|
|
f"""
|
|
CREATE DICTIONARY test_dict_single_quote
|
|
(
|
|
key String,
|
|
value String DEFAULT ''
|
|
)
|
|
PRIMARY KEY key
|
|
LAYOUT(COMPLEX_KEY_DIRECT())
|
|
SOURCE(PostgreSQL(
|
|
DB 'postgres_database'
|
|
HOST '{started_cluster.postgres_ip}'
|
|
PORT {started_cluster.postgres_port}
|
|
USER 'postgres'
|
|
PASSWORD '{pg_pass}'
|
|
TABLE 'test_single_quote'))
|
|
"""
|
|
)
|
|
|
|
# Key containing a single quote: must return the correct value, not an error.
|
|
result = query(
|
|
"SELECT dictGet('test_dict_single_quote', 'value', tuple('it''s a key'))"
|
|
)
|
|
assert result == "found it\n", f"Unexpected result: {result!r}"
|
|
|
|
# Key without a quote: must still work after the escaping change.
|
|
result = query(
|
|
"SELECT dictGet('test_dict_single_quote', 'value', tuple('normal'))"
|
|
)
|
|
assert result == "normal value\n", f"Unexpected result: {result!r}"
|
|
|
|
# Missing key must return the default, not raise an exception.
|
|
result = query(
|
|
"SELECT dictGet('test_dict_single_quote', 'value', tuple('no such key'))"
|
|
)
|
|
assert result == "\n", f"Unexpected result: {result!r}"
|
|
|
|
query("DROP DICTIONARY test_dict_single_quote;")
|
|
cursor.execute("DROP TABLE test_single_quote;")
|
|
|
|
|
|
def test_invalidate_query(started_cluster):
|
|
conn = get_postgres_conn(
|
|
ip=started_cluster.postgres_ip,
|
|
database=True,
|
|
port=started_cluster.postgres_port,
|
|
)
|
|
cursor = conn.cursor()
|
|
table_name = "test0"
|
|
create_and_fill_postgres_table(
|
|
cursor,
|
|
table_name,
|
|
port=started_cluster.postgres_port,
|
|
host=started_cluster.postgres_ip,
|
|
)
|
|
|
|
# invalidate query: SELECT value FROM test0 WHERE id = 0
|
|
dict_name = "dict0"
|
|
create_dict(table_name)
|
|
|
|
def last_update():
|
|
# 1970-01-01 00:00:00 means the dictionary was never updated yet.
|
|
return node1.query(
|
|
"SELECT toTimeZone(last_successful_update_time, 'UTC') "
|
|
f"FROM system.dictionaries WHERE name = '{dict_name}'"
|
|
).strip()
|
|
|
|
def wait_baseline_settled():
|
|
# SYSTEM RELOAD does a forced full reload but does not run the invalidate
|
|
# query, so the stored invalidate response is left empty. The first
|
|
# periodic check then always sees a "modified" source and reloads once
|
|
# more. Wait until the invalidate baseline settles, i.e.
|
|
# last_successful_update_time stops advancing across a full check period,
|
|
# so subsequent "no update" checks are reliable.
|
|
prev = last_update()
|
|
while True:
|
|
time.sleep(7) # dict lifetime is 1s and the periodic check runs every 5s
|
|
cur = last_update()
|
|
if cur == prev:
|
|
return cur
|
|
prev = cur
|
|
|
|
def dict_get(key):
|
|
return node1.query(
|
|
f"SELECT dictGetUInt32('{dict_name}', 'value', toUInt64({key}))"
|
|
).strip()
|
|
|
|
node1.query(f"SYSTEM RELOAD DICTIONARY {dict_name}")
|
|
wait_baseline_settled()
|
|
assert dict_get(0) == "0"
|
|
assert dict_get(1) == "1"
|
|
|
|
# update should happen: the invalidate query result (value at id = 0) changes
|
|
cursor.execute(f"UPDATE {table_name} SET value = value + 1 WHERE id = 0")
|
|
assert_eq_with_retry(
|
|
node1, f"SELECT dictGetUInt32('{dict_name}', 'value', toUInt64(0))", "1"
|
|
)
|
|
settled = wait_baseline_settled()
|
|
assert dict_get(0) == "1"
|
|
|
|
# no update should happen: the invalidate query result (value at id = 0) is
|
|
# unchanged, so the dictionary must not reload and pick up the change to
|
|
# other rows. Verify directly that no reload was triggered.
|
|
cursor.execute(f"UPDATE {table_name} SET value = value * 2 WHERE id != 0")
|
|
time.sleep(7) # more than one periodic check period
|
|
assert last_update() == settled
|
|
assert dict_get(0) == "1"
|
|
assert dict_get(1) == "1"
|
|
|
|
# update should happen: the invalidate query result (value at id = 0) changes
|
|
cursor.execute(f"UPDATE {table_name} SET value = value + 1 WHERE id = 0")
|
|
assert_eq_with_retry(
|
|
node1, f"SELECT dictGetUInt32('{dict_name}', 'value', toUInt64(0))", "2"
|
|
)
|
|
assert dict_get(0) == "2"
|
|
assert dict_get(1) == "2"
|
|
|
|
node1.query(f"DROP TABLE IF EXISTS {table_name}")
|
|
node1.query(f"DROP DICTIONARY IF EXISTS {dict_name}")
|
|
cursor.execute(f"DROP TABLE IF EXISTS {table_name}")
|
|
|
|
|
|
def test_dictionary_with_replicas(started_cluster):
|
|
conn1 = get_postgres_conn(
|
|
ip=started_cluster.postgres_ip,
|
|
port=started_cluster.postgres_port,
|
|
database=True,
|
|
)
|
|
cursor1 = conn1.cursor()
|
|
conn2 = get_postgres_conn(
|
|
ip=started_cluster.postgres2_ip,
|
|
port=started_cluster.postgres_port,
|
|
database=True,
|
|
)
|
|
cursor2 = conn2.cursor()
|
|
|
|
create_postgres_table(cursor1, "test1")
|
|
create_postgres_table(cursor2, "test1")
|
|
|
|
cursor1.execute(
|
|
"INSERT INTO test1 select i, i, i from generate_series(0, 99) as t(i);"
|
|
)
|
|
cursor2.execute(
|
|
"INSERT INTO test1 select i, i, i from generate_series(100, 199) as t(i);"
|
|
)
|
|
|
|
create_dict("test1", 1)
|
|
result = node1.query("SELECT * FROM `test`.`dict_table_test1` ORDER BY key")
|
|
|
|
# priority 0 - non running port
|
|
assert node1.contains_in_log("PostgreSQLConnectionPool: Connection error*")
|
|
|
|
# priority 1 - postgres2, table contains rows with values 100-200
|
|
# priority 2 - postgres1, table contains rows with values 0-100
|
|
expected = node1.query("SELECT number, number FROM numbers(100, 100)")
|
|
assert result == expected
|
|
|
|
cursor1.execute("DROP TABLE IF EXISTS test1")
|
|
cursor2.execute("DROP TABLE IF EXISTS test1")
|
|
|
|
node1.query("DROP TABLE IF EXISTS test1")
|
|
node1.query("DROP DICTIONARY IF EXISTS dict1")
|
|
|
|
|
|
def test_postgres_schema(started_cluster):
|
|
conn = get_postgres_conn(
|
|
ip=started_cluster.postgres_ip,
|
|
port=started_cluster.postgres_port,
|
|
database=True,
|
|
)
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("CREATE SCHEMA test_schema_1")
|
|
cursor.execute("CREATE TABLE test_schema_1.test_table (id integer, value integer)")
|
|
cursor.execute(
|
|
"INSERT INTO test_schema_1.test_table SELECT i, i FROM generate_series(0, 99) as t(i)"
|
|
)
|
|
|
|
node1.query(
|
|
f"""
|
|
DROP DICTIONARY IF EXISTS postgres_dict;
|
|
CREATE DICTIONARY postgres_dict (id UInt32, value UInt32)
|
|
PRIMARY KEY id
|
|
SOURCE(POSTGRESQL(
|
|
port 5432
|
|
host 'postgres1'
|
|
user 'postgres'
|
|
password '{pg_pass}'
|
|
db 'postgres_database'
|
|
table 'test_schema_1.test_table'))
|
|
LIFETIME(MIN 1 MAX 2)
|
|
LAYOUT(HASHED());
|
|
"""
|
|
)
|
|
|
|
result = node1.query("SELECT dictGetUInt32(postgres_dict, 'value', toUInt64(1))")
|
|
assert int(result.strip()) == 1
|
|
result = node1.query("SELECT dictGetUInt32(postgres_dict, 'value', toUInt64(99))")
|
|
assert int(result.strip()) == 99
|
|
node1.query("DROP DICTIONARY IF EXISTS postgres_dict")
|
|
cursor.execute("DROP TABLE test_schema_1.test_table")
|
|
cursor.execute("DROP SCHEMA test_schema_1")
|
|
|
|
|
|
def test_predefined_connection_configuration(started_cluster):
|
|
conn = get_postgres_conn(
|
|
ip=started_cluster.postgres_ip,
|
|
port=started_cluster.postgres_port,
|
|
database=True,
|
|
)
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("DROP TABLE IF EXISTS test_table")
|
|
cursor.execute("CREATE TABLE test_table (id integer, value integer)")
|
|
cursor.execute(
|
|
"INSERT INTO test_table SELECT i, i FROM generate_series(0, 99) as t(i)"
|
|
)
|
|
|
|
node1.query(
|
|
"""
|
|
DROP DICTIONARY IF EXISTS postgres_dict;
|
|
CREATE DICTIONARY postgres_dict (id UInt32, value UInt32)
|
|
PRIMARY KEY id
|
|
SOURCE(POSTGRESQL(NAME postgres1))
|
|
LIFETIME(MIN 1 MAX 2)
|
|
LAYOUT(HASHED());
|
|
"""
|
|
)
|
|
result = node1.query("SELECT dictGetUInt32(postgres_dict, 'value', toUInt64(99))")
|
|
assert int(result.strip()) == 99
|
|
|
|
cursor.execute("DROP SCHEMA IF EXISTS test_schema_2 CASCADE")
|
|
cursor.execute("CREATE SCHEMA test_schema_2")
|
|
cursor.execute("CREATE TABLE test_schema_2.test_table (id integer, value integer)")
|
|
cursor.execute(
|
|
"INSERT INTO test_schema_2.test_table SELECT i, 100 FROM generate_series(0, 99) as t(i)"
|
|
)
|
|
|
|
node1.query(
|
|
"""
|
|
DROP DICTIONARY postgres_dict;
|
|
CREATE DICTIONARY postgres_dict (id UInt32, value UInt32)
|
|
PRIMARY KEY id
|
|
SOURCE(POSTGRESQL(NAME postgres1 SCHEMA test_schema_2))
|
|
LIFETIME(MIN 1 MAX 2)
|
|
LAYOUT(HASHED());
|
|
"""
|
|
)
|
|
result = node1.query("SELECT dictGetUInt32(postgres_dict, 'value', toUInt64(99))")
|
|
assert int(result.strip()) == 100
|
|
|
|
node1.query(
|
|
"""
|
|
DROP DICTIONARY postgres_dict;
|
|
CREATE DICTIONARY postgres_dict (id UInt32, value UInt32)
|
|
PRIMARY KEY id
|
|
SOURCE(POSTGRESQL(NAME postgres2))
|
|
LIFETIME(MIN 1 MAX 2)
|
|
LAYOUT(HASHED());
|
|
"""
|
|
)
|
|
result = node1.query("SELECT dictGetUInt32(postgres_dict, 'value', toUInt64(99))")
|
|
assert int(result.strip()) == 100
|
|
|
|
node1.query("DROP DICTIONARY postgres_dict")
|
|
node1.query(
|
|
"""
|
|
CREATE DICTIONARY postgres_dict (id UInt32, value UInt32)
|
|
PRIMARY KEY id
|
|
SOURCE(POSTGRESQL(NAME postgres4))
|
|
LIFETIME(MIN 1 MAX 2)
|
|
LAYOUT(HASHED());
|
|
"""
|
|
)
|
|
result = node1.query_and_get_error(
|
|
"SELECT dictGetUInt32(postgres_dict, 'value', toUInt64(99))"
|
|
)
|
|
|
|
node1.query(
|
|
"""
|
|
DROP DICTIONARY postgres_dict;
|
|
CREATE DICTIONARY postgres_dict (id UInt32, value UInt32)
|
|
PRIMARY KEY id
|
|
SOURCE(POSTGRESQL(NAME postgres1 PORT 5432))
|
|
LIFETIME(MIN 1 MAX 2)
|
|
LAYOUT(HASHED());
|
|
"""
|
|
)
|
|
result = node1.query("SELECT dictGetUInt32(postgres_dict, 'value', toUInt64(99))")
|
|
assert int(result.strip()) == 99
|
|
|
|
|
|
def test_bad_configuration(started_cluster):
|
|
conn = get_postgres_conn(
|
|
ip=started_cluster.postgres_ip,
|
|
port=started_cluster.postgres_port,
|
|
database=True,
|
|
)
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("DROP SCHEMA IF EXISTS test_schema_3 CASCADE")
|
|
cursor.execute("CREATE SCHEMA test_schema_3")
|
|
cursor.execute("CREATE TABLE test_schema_3.test_table (id integer, value integer)")
|
|
|
|
node1.query(
|
|
f"""
|
|
DROP DICTIONARY IF EXISTS postgres_dict;
|
|
CREATE DICTIONARY postgres_dict (id UInt32, value UInt32)
|
|
PRIMARY KEY id
|
|
SOURCE(POSTGRESQL(
|
|
port 5432
|
|
host 'postgres1'
|
|
user 'postgres'
|
|
password '{pg_pass}'
|
|
dbbb 'postgres_database'
|
|
table 'test_schema_3.test_table'))
|
|
LIFETIME(MIN 1 MAX 2)
|
|
LAYOUT(HASHED());
|
|
"""
|
|
)
|
|
|
|
assert "Unexpected key `dbbb`" in node1.query_and_get_error(
|
|
"SELECT dictGetUInt32(postgres_dict, 'value', toUInt64(1))"
|
|
)
|
|
|
|
|
|
def test_named_collection_from_ddl(started_cluster):
|
|
cursor = started_cluster.postgres_conn.cursor()
|
|
cursor.execute("DROP TABLE IF EXISTS test_table")
|
|
cursor.execute("CREATE TABLE test_table (id integer, value integer)")
|
|
|
|
node1.query(
|
|
f"""
|
|
DROP NAMED COLLECTION IF EXISTS pg_conn;
|
|
CREATE NAMED COLLECTION pg_conn
|
|
AS user = 'postgres', password = '{pg_pass}', host = 'postgres1', port = 5432, database = 'postgres', table = 'test_table';
|
|
"""
|
|
)
|
|
|
|
cursor.execute(
|
|
"INSERT INTO test_table SELECT i, i FROM generate_series(0, 99) as t(i)"
|
|
)
|
|
|
|
node1.query(
|
|
"""
|
|
DROP DICTIONARY IF EXISTS postgres_dict;
|
|
CREATE DICTIONARY postgres_dict (id UInt32, value UInt32)
|
|
PRIMARY KEY id
|
|
SOURCE(POSTGRESQL(NAME pg_conn))
|
|
LIFETIME(MIN 1 MAX 2)
|
|
LAYOUT(HASHED());
|
|
"""
|
|
)
|
|
result = node1.query("SELECT dictGetUInt32(postgres_dict, 'value', toUInt64(99))")
|
|
assert int(result.strip()) == 99
|
|
|
|
node1.query(
|
|
f"""
|
|
DROP NAMED COLLECTION IF EXISTS pg_conn_2;
|
|
CREATE NAMED COLLECTION pg_conn_2
|
|
AS user = 'postgres', password = '{pg_pass}', host = 'postgres1', port = 5432, dbbb = 'postgres', table = 'test_table';
|
|
"""
|
|
)
|
|
node1.query(
|
|
"""
|
|
DROP DICTIONARY IF EXISTS postgres_dict;
|
|
CREATE DICTIONARY postgres_dict (id UInt32, value UInt32)
|
|
PRIMARY KEY id
|
|
SOURCE(POSTGRESQL(NAME pg_conn_2))
|
|
LIFETIME(MIN 1 MAX 2)
|
|
LAYOUT(HASHED());
|
|
"""
|
|
)
|
|
assert "Unexpected key `dbbb`" in node1.query_and_get_error(
|
|
"SELECT dictGetUInt32(postgres_dict, 'value', toUInt64(99))"
|
|
)
|
|
|
|
|
|
def test_background_dictionary_reconnect(started_cluster):
|
|
postgres_conn = get_postgres_conn(
|
|
ip=started_cluster.postgres_ip,
|
|
database=True,
|
|
port=started_cluster.postgres_port,
|
|
)
|
|
|
|
postgres_conn.cursor().execute("DROP TABLE IF EXISTS dict")
|
|
postgres_conn.cursor().execute(
|
|
"""
|
|
CREATE TABLE dict (
|
|
id integer NOT NULL, value text NOT NULL, PRIMARY KEY (id))
|
|
"""
|
|
)
|
|
|
|
postgres_conn.cursor().execute("INSERT INTO dict VALUES (1, 'Value_1')")
|
|
|
|
port_forward = PortForward()
|
|
port = port_forward.start((started_cluster.postgres_ip, started_cluster.postgres_port))
|
|
|
|
@contextmanager
|
|
def port_forward_manager():
|
|
try:
|
|
yield
|
|
finally:
|
|
port_forward.stop(True)
|
|
|
|
with port_forward_manager():
|
|
|
|
query = node1.query
|
|
query(
|
|
f"""
|
|
DROP DICTIONARY IF EXISTS dict;
|
|
CREATE DICTIONARY dict
|
|
(
|
|
id UInt64,
|
|
value String
|
|
)
|
|
PRIMARY KEY id
|
|
LAYOUT(DIRECT())
|
|
SOURCE(POSTGRESQL(
|
|
USER 'postgres'
|
|
PASSWORD '{pg_pass}'
|
|
DB 'postgres_database'
|
|
QUERY $doc$SELECT * FROM dict;$doc$
|
|
BACKGROUND_RECONNECT 'true'
|
|
REPLICA(HOST '{socket.gethostbyname(socket.gethostname())}' PORT {port} PRIORITY 1)))
|
|
"""
|
|
)
|
|
|
|
result = query("SELECT value FROM dict WHERE id = 1")
|
|
assert result == "Value_1\n"
|
|
|
|
class PostgreSQL_Instance:
|
|
pass
|
|
|
|
postgres_instance = PostgreSQL_Instance()
|
|
postgres_instance.ip_address = started_cluster.postgres_ip
|
|
|
|
query("TRUNCATE TABLE IF EXISTS system.text_log")
|
|
|
|
# Break connection to postgresql server
|
|
port_forward.stop(force=True)
|
|
|
|
# Exhaust possible connection pool and initiate reconnection attempts
|
|
for _ in range(5):
|
|
try:
|
|
result = query("SELECT value FROM dict WHERE id = 1")
|
|
except Exception:
|
|
pass
|
|
|
|
time.sleep(5)
|
|
|
|
# Restore connection to postgresql server
|
|
port_forward.start((started_cluster.postgres_ip, started_cluster.postgres_port), port)
|
|
|
|
time.sleep(5)
|
|
|
|
query("SYSTEM FLUSH LOGS")
|
|
|
|
assert (
|
|
int(
|
|
query(
|
|
"SELECT count() FROM system.text_log WHERE message like 'Reestablishing connection to % has failed: %'"
|
|
)
|
|
)
|
|
> 0
|
|
)
|
|
assert (
|
|
int(
|
|
query(
|
|
"SELECT count() FROM system.text_log WHERE message like 'Reestablishing connection to % has succeeded.'"
|
|
)
|
|
)
|
|
> 0
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cluster.start()
|
|
input("Cluster created, press any key to destroy...")
|
|
cluster.shutdown()
|