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]>
90 lines
3.5 KiB
Python
90 lines
3.5 KiB
Python
import time
|
|
|
|
import pytest
|
|
|
|
from helpers.cluster import ClickHouseCluster
|
|
|
|
cluster = ClickHouseCluster(__file__)
|
|
|
|
instance_test_mutations = cluster.add_instance(
|
|
"test_mutations_with_projection",
|
|
main_configs=["configs/config.xml"],
|
|
user_configs=["configs/users.xml"],
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def started_cluster():
|
|
try:
|
|
cluster.start()
|
|
instance_test_mutations.query(
|
|
"""
|
|
CREATE TABLE video_log
|
|
(
|
|
`datetime` DateTime, -- 20,000 records per second
|
|
`user_id` UInt64, -- Cardinality == 100,000,000
|
|
`device_id` UInt64, -- Cardinality == 200,000,000
|
|
`video_id` UInt64, -- Cardinality == 100,00000
|
|
`domain` LowCardinality(String), -- Cardinality == 100
|
|
`bytes` UInt64, -- Ranging from 128 to 1152
|
|
`duration` UInt64, -- Ranging from 100 to 400
|
|
PROJECTION p_norm (SELECT datetime, device_id, bytes, duration ORDER BY device_id),
|
|
PROJECTION p_agg (SELECT toStartOfHour(datetime) AS hour, domain, sum(bytes), avg(duration) GROUP BY hour, domain)
|
|
)
|
|
ENGINE = MergeTree
|
|
PARTITION BY toDate(datetime) -- Daily partitioning
|
|
ORDER BY (user_id, device_id, video_id) -- Can only favor one column here
|
|
SETTINGS index_granularity = 1000;
|
|
"""
|
|
)
|
|
|
|
instance_test_mutations.query(
|
|
"""CREATE TABLE rng (`user_id_raw` UInt64, `device_id_raw` UInt64, `video_id_raw` UInt64, `domain_raw` UInt64, `bytes_raw` UInt64, `duration_raw` UInt64) ENGINE = GenerateRandom(1024);"""
|
|
)
|
|
|
|
instance_test_mutations.query(
|
|
"""INSERT INTO video_log SELECT toUnixTimestamp(toDateTime(today())) + (rowNumberInAllBlocks() / 20000), user_id_raw % 100000000 AS user_id, device_id_raw % 200000000 AS device_id, video_id_raw % 100000000 AS video_id, domain_raw % 100, (bytes_raw % 1024) + 128, (duration_raw % 300) + 100 FROM rng LIMIT 500000;"""
|
|
)
|
|
|
|
instance_test_mutations.query("""OPTIMIZE TABLE video_log FINAL;""")
|
|
|
|
yield cluster
|
|
finally:
|
|
cluster.shutdown()
|
|
|
|
|
|
def test_mutations_with_multi_level_merge_of_projections(started_cluster):
|
|
try:
|
|
instance_test_mutations.query(
|
|
"""ALTER TABLE video_log UPDATE bytes = bytes + 10086 WHERE 1;"""
|
|
)
|
|
|
|
def count_and_changed():
|
|
return instance_test_mutations.query(
|
|
"SELECT count(), countIf(bytes > 10000) FROM video_log SETTINGS force_index_by_date = 0, force_primary_key = 0 FORMAT CSV"
|
|
).splitlines()
|
|
|
|
all_done = False
|
|
for wait_times_for_mutation in range(
|
|
100
|
|
): # wait for replication 80 seconds max
|
|
time.sleep(0.8)
|
|
|
|
if count_and_changed() == ["500000,500000"]:
|
|
all_done = True
|
|
break
|
|
|
|
print(
|
|
instance_test_mutations.query(
|
|
"SELECT mutation_id, command, parts_to_do, is_done, latest_failed_part, latest_fail_reason, parts_to_do_names FROM system.mutations WHERE table = 'video_log' SETTINGS force_index_by_date = 0, force_primary_key = 0 FORMAT TSVWithNames"
|
|
)
|
|
)
|
|
|
|
assert (count_and_changed(), all_done) == (["500000,500000"], True)
|
|
assert instance_test_mutations.query(
|
|
"SELECT DISTINCT arraySort(projections) FROM system.parts WHERE table = 'video_log' SETTINGS force_index_by_date = 0, force_primary_key = 0 FORMAT TSVRaw"
|
|
).splitlines() == ["['p_agg','p_norm']"]
|
|
|
|
finally:
|
|
instance_test_mutations.query("""DROP TABLE video_log""")
|