mirror of
https://github.com/hanzoai/datastore.git
synced 2026-09-26 13:55:07 +00:00
Alexey MilovidovandClaude Opus 4.8
295ced78d1
Follow-up to #105171, which added `ruff` with a deliberately narrow, bug-class-only rule set and left the unused-name and star-import rules off pending a dedicated cleanup pass. This enables that second batch in `pyproject.toml` and fixes every violation across the tree (the bulk in integration tests): - `F401` unused-import and `F841` unused-variable: removed, preserving any side-effecting right-hand side (e.g. `result = node.query(...)` becomes `node.query(...)`). Two `try: import ...` feature probes are kept with `# noqa: F401`. - `F541` f-string-missing-placeholders: dropped the redundant `f` prefix. - `F403`/`F405` star imports: every `from module import *` is expanded into explicit imports of the names actually used, taken from each name's real origin, so stdlib/third-party names import from their real source rather than through a test "convenience bundle". The now-empty `tests/integration/helpers/kafka/common_direct.py` bundle is removed and its three remaining consumers import `_VarintBytes`/`KafkaAdminClient` directly. Eliminating the star imports surfaced a real bug the wildcard had masked from `F821`: `generate_presets.py` in `test_prometheus_protocols` called an undefined `execute_instant_query_with_http_api`; fixed to the existing `execute_query_via_http_api`. No `black`/`isort` reformatting was applied (neither is enforced in the style check), so the diff is limited to the lint fixes themselves. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
249 lines
7.8 KiB
Python
249 lines
7.8 KiB
Python
import pytest
|
|
|
|
from helpers.cluster import ClickHouseCluster
|
|
from helpers.test_tools import assert_eq_with_retry
|
|
|
|
cluster = ClickHouseCluster(__file__)
|
|
|
|
replica1 = cluster.add_instance(
|
|
"replica1", with_zookeeper=True, main_configs=["configs/remote_servers.xml"]
|
|
)
|
|
replica2 = cluster.add_instance(
|
|
"replica2", with_zookeeper=True, main_configs=["configs/remote_servers.xml"]
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def start_cluster():
|
|
try:
|
|
cluster.start()
|
|
yield cluster
|
|
except Exception as ex:
|
|
print(ex)
|
|
finally:
|
|
cluster.shutdown()
|
|
|
|
|
|
# Retry parameters for queries that read from a `web` disk backed by
|
|
# `raw.githubusercontent.com`. The endpoint occasionally returns
|
|
# `Poco::TimeoutException: connect timed out` under CI network load,
|
|
# so all queries that touch the source table or its data are wrapped
|
|
# in `query_with_retry`.
|
|
WEB_RETRY_COUNT = 5
|
|
WEB_RETRY_SLEEP = 10
|
|
|
|
|
|
def cleanup(nodes):
|
|
for node in nodes:
|
|
node.query_with_retry(
|
|
"DROP TABLE IF EXISTS source SYNC",
|
|
retry_count=WEB_RETRY_COUNT,
|
|
sleep_time=WEB_RETRY_SLEEP,
|
|
)
|
|
node.query_with_retry(
|
|
"DROP TABLE IF EXISTS destination SYNC",
|
|
retry_count=WEB_RETRY_COUNT,
|
|
sleep_time=WEB_RETRY_SLEEP,
|
|
)
|
|
|
|
|
|
def create_source_table(node, table_name, replicated):
|
|
replica = node.name
|
|
engine = (
|
|
f"ReplicatedMergeTree('/clickhouse/tables/1/{table_name}', '{replica}')"
|
|
if replicated
|
|
else "MergeTree()"
|
|
)
|
|
node.query_with_retry(
|
|
"""
|
|
ATTACH TABLE {table_name} UUID 'cf712b4f-2ca8-435c-ac23-c4393efe52f7'
|
|
(
|
|
price UInt32,
|
|
date Date,
|
|
postcode1 LowCardinality(String),
|
|
postcode2 LowCardinality(String),
|
|
type Enum8('other' = 0, 'terraced' = 1, 'semi-detached' = 2, 'detached' = 3, 'flat' = 4),
|
|
is_new UInt8,
|
|
duration Enum8('unknown' = 0, 'freehold' = 1, 'leasehold' = 2),
|
|
addr1 String,
|
|
addr2 String,
|
|
street LowCardinality(String),
|
|
locality LowCardinality(String),
|
|
town LowCardinality(String),
|
|
district LowCardinality(String),
|
|
county LowCardinality(String)
|
|
)
|
|
ENGINE = {engine}
|
|
ORDER BY (postcode1, postcode2, addr1, addr2)
|
|
SETTINGS disk = disk(type = web, endpoint = 'https://raw.githubusercontent.com/ClickHouse/web-tables-demo/main/web/')
|
|
""".format(
|
|
table_name=table_name, engine=engine
|
|
)
|
|
)
|
|
|
|
|
|
def create_destination_table(node, table_name, replicated):
|
|
replica = node.name
|
|
engine = (
|
|
f"ReplicatedMergeTree('/clickhouse/tables/1/{table_name}', '{replica}')"
|
|
if replicated
|
|
else "MergeTree()"
|
|
)
|
|
node.query_with_retry(
|
|
"""
|
|
CREATE TABLE {table_name}
|
|
(
|
|
price UInt32,
|
|
date Date,
|
|
postcode1 LowCardinality(String),
|
|
postcode2 LowCardinality(String),
|
|
type Enum8('other' = 0, 'terraced' = 1, 'semi-detached' = 2, 'detached' = 3, 'flat' = 4),
|
|
is_new UInt8,
|
|
duration Enum8('unknown' = 0, 'freehold' = 1, 'leasehold' = 2),
|
|
addr1 String,
|
|
addr2 String,
|
|
street LowCardinality(String),
|
|
locality LowCardinality(String),
|
|
town LowCardinality(String),
|
|
district LowCardinality(String),
|
|
county LowCardinality(String)
|
|
)
|
|
ENGINE = {engine}
|
|
ORDER BY (postcode1, postcode2, addr1, addr2)
|
|
""".format(
|
|
table_name=table_name, engine=engine
|
|
)
|
|
)
|
|
|
|
|
|
def test_both_mergetree(start_cluster):
|
|
cleanup([replica1, replica2])
|
|
create_source_table(replica1, "source", False)
|
|
create_destination_table(replica1, "destination", False)
|
|
|
|
replica1.query_with_retry(
|
|
"ALTER TABLE destination ATTACH PARTITION tuple() FROM source",
|
|
retry_count=WEB_RETRY_COUNT,
|
|
sleep_time=WEB_RETRY_SLEEP,
|
|
)
|
|
|
|
expected = replica1.query_with_retry(
|
|
"SELECT toYear(date) AS year,round(avg(price)) AS price,"
|
|
"bar(price, 0, 1000000, 80) FROM source GROUP BY year ORDER BY year ASC",
|
|
retry_count=WEB_RETRY_COUNT,
|
|
sleep_time=WEB_RETRY_SLEEP,
|
|
)
|
|
assert_eq_with_retry(
|
|
replica1,
|
|
"SELECT toYear(date) AS year,round(avg(price)) AS price,"
|
|
"bar(price, 0, 1000000, 80) FROM destination GROUP BY year ORDER BY year ASC",
|
|
expected,
|
|
)
|
|
|
|
assert_eq_with_retry(
|
|
replica1, "SELECT town from destination LIMIT 1", "SCARBOROUGH"
|
|
)
|
|
|
|
cleanup([replica1])
|
|
|
|
|
|
def test_all_replicated(start_cluster):
|
|
cleanup([replica1, replica2])
|
|
create_source_table(replica1, "source", True)
|
|
create_destination_table(replica1, "destination", True)
|
|
create_destination_table(replica2, "destination", True)
|
|
|
|
replica1.query_with_retry(
|
|
"ALTER TABLE destination ATTACH PARTITION tuple() FROM source",
|
|
retry_count=WEB_RETRY_COUNT,
|
|
sleep_time=WEB_RETRY_SLEEP,
|
|
)
|
|
replica2.query("SYSTEM SYNC REPLICA destination")
|
|
|
|
expected = replica1.query_with_retry(
|
|
"SELECT toYear(date) AS year,round(avg(price)) AS price,"
|
|
"bar(price, 0, 1000000, 80) FROM source GROUP BY year ORDER BY year ASC",
|
|
retry_count=WEB_RETRY_COUNT,
|
|
sleep_time=WEB_RETRY_SLEEP,
|
|
)
|
|
assert_eq_with_retry(
|
|
replica1,
|
|
"SELECT toYear(date) AS year,round(avg(price)) AS price,"
|
|
"bar(price, 0, 1000000, 80) FROM destination GROUP BY year ORDER BY year ASC",
|
|
expected,
|
|
)
|
|
assert_eq_with_retry(
|
|
replica2,
|
|
"SELECT toYear(date) AS year,round(avg(price)) AS price,"
|
|
"bar(price, 0, 1000000, 80) FROM destination GROUP BY year ORDER BY year ASC",
|
|
expected,
|
|
)
|
|
|
|
assert_eq_with_retry(
|
|
replica1, "SELECT town from destination LIMIT 1", "SCARBOROUGH"
|
|
)
|
|
|
|
assert_eq_with_retry(
|
|
replica2, "SELECT town from destination LIMIT 1", "SCARBOROUGH"
|
|
)
|
|
|
|
cleanup([replica1, replica2])
|
|
|
|
|
|
def test_only_destination_replicated(start_cluster):
|
|
cleanup([replica1, replica2])
|
|
create_source_table(replica1, "source", False)
|
|
create_destination_table(replica1, "destination", True)
|
|
create_destination_table(replica2, "destination", True)
|
|
|
|
replica1.query_with_retry(
|
|
"ALTER TABLE destination ATTACH PARTITION tuple() FROM source",
|
|
retry_count=WEB_RETRY_COUNT,
|
|
sleep_time=WEB_RETRY_SLEEP,
|
|
)
|
|
replica2.query("SYSTEM SYNC REPLICA destination")
|
|
|
|
expected = replica1.query_with_retry(
|
|
"SELECT toYear(date) AS year,round(avg(price)) AS price,"
|
|
"bar(price, 0, 1000000, 80) FROM source GROUP BY year ORDER BY year ASC",
|
|
retry_count=WEB_RETRY_COUNT,
|
|
sleep_time=WEB_RETRY_SLEEP,
|
|
)
|
|
assert_eq_with_retry(
|
|
replica1,
|
|
"SELECT toYear(date) AS year,round(avg(price)) AS price,"
|
|
"bar(price, 0, 1000000, 80) FROM destination GROUP BY year ORDER BY year ASC",
|
|
expected,
|
|
)
|
|
assert_eq_with_retry(
|
|
replica2,
|
|
"SELECT toYear(date) AS year,round(avg(price)) AS price,"
|
|
"bar(price, 0, 1000000, 80) FROM destination GROUP BY year ORDER BY year ASC",
|
|
expected,
|
|
)
|
|
|
|
assert_eq_with_retry(
|
|
replica1, "SELECT town from destination LIMIT 1", "SCARBOROUGH"
|
|
)
|
|
|
|
assert_eq_with_retry(
|
|
replica2, "SELECT town from destination LIMIT 1", "SCARBOROUGH"
|
|
)
|
|
|
|
cleanup([replica1, replica2])
|
|
|
|
|
|
def test_not_work_on_different_disk(start_cluster):
|
|
cleanup([replica1, replica2])
|
|
# Replace and move should not work on replace
|
|
create_source_table(replica1, "source", False)
|
|
create_destination_table(replica2, "destination", False)
|
|
|
|
replica1.query_and_get_error(
|
|
"ALTER TABLE destination REPLACE PARTITION tuple() FROM source"
|
|
)
|
|
replica1.query_and_get_error(
|
|
"ALTER TABLE destination MOVE PARTITION tuple() FROM source"
|
|
)
|
|
cleanup([replica1, replica2])
|