1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
|
"""Shared window antijoin cleanup for analytics marts and upload scripts."""
from __future__ import annotations
import re
from typing import Any, Dict, List, Optional, Sequence, Tuple
import ydb
from ydb_wrapper import YDBWrapper
def validate_identifier(value: str, arg_name: str) -> None:
if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", value or ""):
raise ValueError(
f"Invalid {arg_name}: {value!r}. Allowed pattern: [A-Za-z_][A-Za-z0-9_]*"
)
def validate_interval_expression(interval_expr: str) -> None:
# Example: 365 * Interval("P1D")
if not re.fullmatch(r'\d+\s*\*\s*Interval\("P[0-9A-Z]+"\)', interval_expr or ""):
raise ValueError(
f"Invalid cleanup window interval: {interval_expr!r}. "
'Expected format like: 365 * Interval("P1D")'
)
def normalize_pk_value(value: Any) -> Any:
"""Normalize YDB scan values for PK comparison (Utf8 often arrives as bytes)."""
if isinstance(value, bytes):
return value.decode('utf-8', errors='replace')
return value
def pk_tuple(row: Dict[str, Any], primary_keys: Sequence[str]) -> Tuple[Any, ...]:
return tuple(normalize_pk_value(row[key]) for key in primary_keys)
def to_typed_param(value: Any, type_name: str) -> Any:
primitive = getattr(ydb.PrimitiveType, type_name, None)
if primitive is None:
return value
return ydb.TypedValue(value=value, value_type=primitive)
def delete_stale_pk_rows(
ydb_wrapper: YDBWrapper,
table_path: str,
primary_keys: Sequence[str],
pk_type_map: Dict[str, str],
stale_rows: Sequence[Dict[str, Any]],
query_name_prefix: Optional[str] = None,
) -> None:
if not stale_rows:
print("Cleanup mode=window_antijoin: no stale rows to delete")
return
prefix = query_name_prefix or table_path.split('/')[-1]
print(f"Cleanup mode=window_antijoin: deleting stale rows: {len(stale_rows)}")
preview_limit = 10
chunk_size = 100
total = len(stale_rows)
for idx, row in enumerate(stale_rows, 1):
if idx <= preview_limit:
preview = ", ".join([f"{key}={row[key]!r}" for key in primary_keys])
print(f"Stale row {idx}/{total}: {preview}")
if idx == preview_limit + 1:
print("Stale row preview limit reached; continuing without per-row logging")
for start_idx in range(0, total, chunk_size):
chunk = stale_rows[start_idx:start_idx + chunk_size]
chunk_no = start_idx // chunk_size + 1
declare_lines: List[str] = []
predicates: List[str] = []
params: Dict[str, Any] = {}
for row_idx, row in enumerate(chunk):
row_param_names: List[str] = []
for key in primary_keys:
param_name = f"${key}_{row_idx}"
declare_lines.append(f"DECLARE {param_name} AS {pk_type_map[key]};")
params[param_name] = to_typed_param(normalize_pk_value(row[key]), pk_type_map[key])
row_param_names.append(param_name)
predicates.append(
"(" + " AND ".join(
[f"`{key}` = {row_param_names[key_idx]}" for key_idx, key in enumerate(primary_keys)]
) + ")"
)
declare_block = "\n".join(declare_lines)
delete_query = f"""
{declare_block}
DELETE FROM `{table_path}`
WHERE {' OR '.join(predicates)};
"""
ydb_wrapper.execute_dml(
delete_query,
parameters=params,
query_name=f"{prefix}_cleanup_stale_pk_chunk_{chunk_no}",
)
deleted = min(start_idx + chunk_size, total)
print(f"Deleted stale rows: {deleted}/{total} (chunks: {chunk_no})")
def cleanup_window_antijoin(
ydb_wrapper: YDBWrapper,
table_path: str,
source_rows: Sequence[Dict[str, Any]],
primary_keys: Sequence[str],
pk_type_map: Dict[str, str],
window_predicate: str,
query_name: str,
) -> None:
"""Delete table rows in window that are absent from source_rows (by primary key).
``window_predicate`` is interpolated into SQL as-is; callers must build it from
validated identifiers / escaped literals (see ``validate_identifier``, ``_sql_escape``).
"""
for key in primary_keys:
validate_identifier(key, "primary_keys")
target_pk_projection = ", ".join([f"t.`{key}` AS `{key}`" for key in primary_keys])
target_pk_query = f"""
SELECT {target_pk_projection}
FROM `{table_path}` AS t
WHERE {window_predicate};
"""
target_pk_rows = ydb_wrapper.execute_scan_query(
target_pk_query, f"{query_name}_cleanup_target_pks"
)
source_pk_set = {pk_tuple(row, primary_keys) for row in source_rows}
stale_rows = [
row for row in target_pk_rows
if pk_tuple(row, primary_keys) not in source_pk_set
]
delete_stale_pk_rows(
ydb_wrapper, table_path, primary_keys, pk_type_map, stale_rows, query_name_prefix=query_name
)
def cleanup_window_antijoin_by_date_key(
ydb_wrapper: YDBWrapper,
table_path: str,
source_rows: Sequence[Dict[str, Any]],
primary_keys: Sequence[str],
pk_type_map: Dict[str, str],
window_key: str,
window_interval: str,
query_name: str,
) -> None:
"""Date-key variant used by data_mart_executor (CurrentUtcDate() window)."""
validate_identifier(window_key, "window_key")
validate_interval_expression(window_interval)
window_predicate = f"t.`{window_key}` >= CurrentUtcDate() - {window_interval}"
cleanup_window_antijoin(
ydb_wrapper,
table_path,
source_rows,
primary_keys,
pk_type_map,
window_predicate,
query_name,
)
|
