busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix(translate): preserve ClickHouse datetime keys

Browse Source
This commit is contained in:
busya
2026-07-23 12:42:56 +03:00
parent cd4b91daa5
commit eeb3a05e42
8 changed files with 167 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
backend/src/plugins/translate/_batch_insert.py
View File

@@ -25,7 +25,7 @@ from ...core.db_executor import DbExecutor
from ...core.logger import belief_scope, logger
from ...models.translate import TranslationJob, TranslationRecord
from .orchestrator_sql_rows import dedup_rows_for_merge
from .sql_generator import SQLGenerator, _normalize_timestamp_value
from .sql_generator import SQLGenerator
from .superset_executor import SupersetSqlLabExecutor
@@ -85,6 +85,7 @@ async def insert_batch_to_target(
key_columns=job.target_key_cols,
upsert_strategy=job.upsert_strategy or "MERGE",
max_rows_per_statement=500,
column_types=getattr(job, "target_column_types", None),
)
total_inserted = 0
prepared = len(rows_for_sql)
@@ -199,14 +200,13 @@ def _build_insert_rows(
context_data[key] = str(val) if val is not None else ""
base_row: dict[str, object] = {}
if job.target_key_cols:
for k in job.target_key_cols:
raw = source_data.get(k)
if raw is not None:
normalized = _normalize_timestamp_value(raw)
base_row[k] = normalized if normalized else raw
else:
base_row[k] = None
target_key_cols = list(job.target_key_cols or [])
configured_source_keys = getattr(job, "source_key_cols", None)
source_key_cols = list(configured_source_keys) if isinstance(configured_source_keys, (list, tuple)) else target_key_cols
if len(source_key_cols) != len(target_key_cols):
raise ValueError("source_key_cols and target_key_cols must have equal lengths")
for source_key, target_key in zip(source_key_cols, target_key_cols, strict=True):
base_row[target_key] = source_data.get(source_key)
if job.target_source_column:
base_row[job.target_source_column] = rec.source_sql or ""
if job.target_source_language_column:

14
backend/src/plugins/translate/_run_service.py
View File

@@ -71,29 +71,33 @@ class RunExecutionService:
if not prev_records:
return source_rows
key_cols = job.target_key_cols or job.source_key_cols or []
if not key_cols:
target_key_cols = list(job.target_key_cols or [])
configured_source_keys = getattr(job, "source_key_cols", None)
source_key_cols = list(configured_source_keys) if isinstance(configured_source_keys, (list, tuple)) else target_key_cols
if len(source_key_cols) != len(target_key_cols):
raise ValueError("source_key_cols and target_key_cols must have equal lengths")
if not source_key_cols:
logger.explore("No key columns configured — skipping new-key-only filter", {"job_id": job.id})
return source_rows
existing_keys = set()
for rec in prev_records:
sd = rec.source_data or {}
key_tuple = tuple(str(sd.get(k, "")) for k in key_cols)
key_tuple = tuple(str(sd.get(k, "")) for k in source_key_cols)
existing_keys.add(key_tuple)
filtered = []
skipped = 0
for row in source_rows:
sd = row.get("source_data", {}) or {}
key_tuple = tuple(str(sd.get(k, "")) for k in key_cols)
key_tuple = tuple(str(sd.get(k, "")) for k in source_key_cols)
if key_tuple not in existing_keys:
filtered.append(row)
else:
skipped += 1
logger.reason(f"New-key-only filter: {len(source_rows)} total -> {len(filtered)} new, {skipped} skipped",
{"job_id": job.id, "prev_run_id": prev_run.id, "key_cols": key_cols})
{"job_id": job.id, "prev_run_id": prev_run.id, "key_cols": source_key_cols})
return filtered
def _load_preview_edits(self, job_id: str) -> None:

16
backend/src/plugins/translate/orchestrator_sql_rows.py
View File

@@ -7,7 +7,6 @@
import json
from ...models.translate import TranslationJob, TranslationRecord
from .sql_generator import _normalize_timestamp_value
# #region Plugin.OrchestratorSqlRows.BuildColumns [C:2] [TYPE Function] [SEMANTICS sql, columns, build]
@@ -166,14 +165,13 @@ def build_rows(
context_data = {key: str(source_data.get(key, "")) for key in context_keys}
base_row: dict[str, object] = {}
if job.target_key_cols:
for k in job.target_key_cols:
raw = source_data.get(k)
if raw is not None:
normalized = _normalize_timestamp_value(raw)
base_row[k] = normalized if normalized else raw
else:
base_row[k] = None
target_key_cols = list(job.target_key_cols or [])
configured_source_keys = getattr(job, "source_key_cols", None)
source_key_cols = list(configured_source_keys) if isinstance(configured_source_keys, (list, tuple)) else target_key_cols
if len(source_key_cols) != len(target_key_cols):
raise ValueError("source_key_cols and target_key_cols must have equal lengths")
for source_key, target_key in zip(source_key_cols, target_key_cols, strict=True):
base_row[target_key] = source_data.get(source_key)
if job.target_source_column:
base_row[job.target_source_column] = rec.source_sql or ""
if job.target_source_language_column:

17
backend/src/plugins/translate/preview.py
View File

@@ -220,8 +220,21 @@ class TranslationPreview:
source_row = meta.get("source_row", {})
source_data = None
if job.target_key_cols:
source_data = {k: source_row.get(k) for k in job.target_key_cols if k in source_row}
configured_source_keys = getattr(job, "source_key_cols", None)
source_key_cols = (
list(configured_source_keys)
if isinstance(configured_source_keys, (list, tuple))
else list(job.target_key_cols or [])
)
target_key_cols = list(job.target_key_cols or [])
if len(source_key_cols) != len(target_key_cols):
raise ValueError("source_key_cols and target_key_cols must have equal lengths")
if source_key_cols:
source_data = {
source_key: source_row.get(source_key)
for source_key in source_key_cols
if source_key in source_row
}
elif source_row:
source_data = dict(source_row)

99
backend/src/plugins/translate/sql_generator.py
View File

@@ -12,6 +12,8 @@
# @REJECTED ORM-based insert bypasses Superset's SQL Lab audit trail.
from datetime import UTC, datetime
from decimal import Decimal
import math
from typing import Any
from ...core.logger import belief_scope, logger
@@ -27,39 +29,35 @@ BACKTICK_DIALECTS = {"clickhouse", "clickhousedb", "mysql"}
# #region Plugin.SqlGenerator.NormalizeTimestampValue [C:2] [TYPE Function] [SEMANTICS translate,sql,timestamp]
# @BRIEF Detect Unix timestamp strings (seconds or millis) and convert to 'YYYY-MM-DD' for Date columns.
def _normalize_timestamp_value(value: Any) -> str | None:
"""Detect Unix timestamp values and convert to date string.
Handles:
- Integer/float Unix timestamps (seconds or milliseconds)
- String representations of Unix timestamps (e.g. '1726358400000.0')
Returns 'YYYY-MM-DD' if conversion succeeds, None if value is not a timestamp.
"""
# Try numeric conversion first
# @BRIEF Convert Unix timestamp values to a ClickHouse temporal literal only when the target type is explicit.
# @RATIONALE Value magnitude alone cannot distinguish a timestamp from a numeric or string primary key.
# @REJECTED Dialect-wide timestamp heuristics — they corrupt legitimate IDs and discard DateTime precision.
def _normalize_timestamp_value(value: Any, target_type: str | None = None) -> str | None:
"""Convert a Unix timestamp to a target ClickHouse temporal literal."""
if not target_type:
return None
normalized_type = target_type.strip().lower()
if not normalized_type.startswith(("date", "datetime")):
return None
try:
ts = float(value)
except (ValueError, TypeError):
return None
# Heuristic: Unix timestamps in seconds are ~10 digits (1e9 range for 2001-2033)
# Unix timestamps in milliseconds are ~13 digits (1e12 range for 2001-2033)
if 1e9 <= ts < 1e12:
# Already in seconds
if 0 <= ts < 1e12:
pass
elif 1e12 <= ts < 1e15:
# Milliseconds — convert to seconds
ts = ts / 1000.0
ts /= 1000.0
else:
# Not a plausible Unix timestamp
return None
try:
dt = datetime.fromtimestamp(ts, tz=UTC)
return dt.strftime("%Y-%m-%d")
except (OSError, OverflowError, ValueError):
return None
if normalized_type == "date":
return dt.strftime("%Y-%m-%d")
if normalized_type.startswith("datetime64"):
return dt.strftime("%Y-%m-%d %H:%M:%S.%f").rstrip("0").rstrip(".")
return dt.strftime("%Y-%m-%d %H:%M:%S")
# #endregion Plugin.SqlGenerator.NormalizeTimestampValue
@@ -94,28 +92,29 @@ def _quote_identifier(identifier: str, dialect: str) -> str:
# #region Plugin.SqlGenerator.EncodeSqlValue [C:2] [TYPE Function] [SEMANTICS translate,sql,encoding]
# @BRIEF Encode a Python value into a SQL-safe literal for INSERT VALUES, with ClickHouse timestamp normalization.
def _encode_sql_value(value: Any, dialect: str | None = None) -> str:
"""Encode a Python value into a SQL-safe literal.
For ClickHouse dialect, attempts to detect Unix timestamp strings
and convert them to 'YYYY-MM-DD' format for Date column compatibility.
"""
# @BRIEF Encode a Python value into a SQL-safe literal with optional target-column type metadata.
def _encode_sql_value(
value: Any,
dialect: str | None = None,
target_type: str | None = None,
) -> str:
"""Encode a Python value into a SQL-safe literal."""
if value is None:
return "NULL"
if isinstance(value, bool):
return "TRUE" if value else "FALSE"
if isinstance(value, float) and not math.isfinite(value):
raise ValueError("NaN and infinity are not supported in SQL INSERT values")
if isinstance(value, Decimal):
return str(value)
# For ClickHouse: try to normalize timestamp-like values (both string and numeric)
if dialect in CLICKHOUSE_DIALECTS:
if (isinstance(value, str) and value) or isinstance(value, (int, float)):
normalized = _normalize_timestamp_value(value)
if normalized:
return f"'{normalized}'"
if dialect in CLICKHOUSE_DIALECTS and target_type:
normalized = _normalize_timestamp_value(value, target_type)
if normalized:
return f"'{normalized}'"
if isinstance(value, (int, float)):
return str(value)
# String — escape single quotes by doubling them
escaped = str(value).replace("'", "''")
return f"'{escaped}'"
# #endregion Plugin.SqlGenerator.EncodeSqlValue
@@ -123,20 +122,21 @@ def _encode_sql_value(value: Any, dialect: str | None = None) -> str:
# #region Plugin.SqlGenerator.BuildValuesClause [C:2] [TYPE Function] [SEMANTICS translate,sql,values]
# @BRIEF Build a VALUES clause for multiple rows with per-column value encoding and dialect-aware quoting.
def _build_values_clause(columns: list[str], rows: list[dict[str, Any]], dialect: str | None = None) -> str:
"""Build VALUES (...) clause for multiple rows.
NOTE: columns may be quoted (e.g. '"col"' or '`col`') for the SQL column list,
but row dicts have UNQUOTED keys. Strip quotes before value lookup.
"""
def _build_values_clause(
columns: list[str],
rows: list[dict[str, Any]],
dialect: str | None = None,
column_types: dict[str, str] | None = None,
) -> str:
"""Build VALUES (...) clause for multiple rows."""
value_groups = []
for row in rows:
values = []
for col in columns:
# Strip quoting for value lookup — row keys are unquoted
lookup_key = col.strip('"').strip('`').strip('[]')
val = row.get(lookup_key)
values.append(_encode_sql_value(val, dialect=dialect))
target_type = (column_types or {}).get(lookup_key)
values.append(_encode_sql_value(val, dialect=dialect, target_type=target_type))
value_groups.append(f"({', '.join(values)})")
return ",\n".join(value_groups)
# #endregion Plugin.SqlGenerator.BuildValuesClause
@@ -153,6 +153,7 @@ def generate_insert_sql(
columns: list[str],
rows: list[dict[str, Any]],
dialect: str | None = None,
column_types: dict[str, str] | None = None,
) -> str:
"""Generate a plain INSERT SQL."""
with belief_scope("generate_insert_sql"):
@@ -164,7 +165,7 @@ def generate_insert_sql(
raise ValueError("At least one row is required for INSERT SQL generation")
col_list = ", ".join(columns)
values = _build_values_clause(columns, rows, dialect=dialect)
values = _build_values_clause(columns, rows, dialect=dialect, column_types=column_types)
table_ref = target_table
if target_schema:
@@ -192,6 +193,7 @@ def generate_upsert_sql(
columns: list[str],
key_columns: list[str],
rows: list[dict[str, Any]],
column_types: dict[str, str] | None = None,
) -> str:
"""Generate INSERT ... ON CONFLICT (key_cols) DO UPDATE SET ... SQL."""
with belief_scope("generate_upsert_sql"):
@@ -206,7 +208,7 @@ def generate_upsert_sql(
col_list = ", ".join(columns)
key_list = ", ".join(key_columns)
values = _build_values_clause(columns, rows)
values = _build_values_clause(columns, rows, column_types=column_types)
table_ref = target_table
if target_schema:
@@ -252,6 +254,7 @@ class SQLGenerator:
rows: list[dict[str, Any]],
key_columns: list[str] | None = None,
upsert_strategy: str = "MERGE",
column_types: dict[str, str] | None = None,
) -> tuple[str, int]:
"""Generate dialect-appropriate INSERT/UPSERT SQL.
@@ -314,6 +317,7 @@ class SQLGenerator:
columns=quoted_columns,
key_columns=quoted_key_columns,
rows=rows,
column_types=column_types,
)
else:
sql = generate_insert_sql(
@@ -322,6 +326,7 @@ class SQLGenerator:
columns=quoted_columns,
rows=rows,
dialect=dialect,
column_types=column_types,
)
elif dialect in CLICKHOUSE_DIALECTS:
# ClickHouse: plain INSERT, no ON CONFLICT support
@@ -331,6 +336,7 @@ class SQLGenerator:
columns=quoted_columns,
rows=rows,
dialect=dialect,
column_types=column_types,
)
if use_upsert:
logger.reason("ClickHouse UPSERT not supported; using plain INSERT", {
@@ -345,6 +351,7 @@ class SQLGenerator:
columns=quoted_columns,
rows=rows,
dialect=dialect,
column_types=column_types,
)
if use_upsert:
logger.reason(
@@ -374,6 +381,7 @@ class SQLGenerator:
key_columns: list[str] | None = None,
upsert_strategy: str = "MERGE",
max_rows_per_statement: int = 500,
column_types: dict[str, str] | None = None,
) -> list[tuple[str, int]]:
"""Generate SQL in batches, splitting large row sets into multiple statements.
@@ -395,6 +403,7 @@ class SQLGenerator:
rows=chunk,
key_columns=key_columns,
upsert_strategy=upsert_strategy,
column_types=column_types,
)
statements.append((sql, count))

4
backend/src/schemas/translate.py
View File

@@ -74,6 +74,8 @@ class TranslateJobCreate(BaseModel):
def validate_insert_method(self):
if self.insert_method == "direct_db" and not self.connection_id:
raise ValueError("connection_id is required when insert_method='direct_db'")
if self.source_key_cols and self.target_key_cols and len(self.source_key_cols) != len(self.target_key_cols):
raise ValueError("source_key_cols and target_key_cols must have equal lengths")
if self.multi_lang_mode is not None and self.multi_lang_mode not in ("single_call", "per_language"):
raise ValueError("multi_lang_mode must be 'single_call' or 'per_language'")
if self.batch_aggressiveness is not None and self.batch_aggressiveness not in ("safe", "balanced", "fast"):
@@ -127,6 +129,8 @@ class TranslateJobUpdate(BaseModel):
@model_validator(mode="after")
def validate_performance_enums(self):
if self.source_key_cols and self.target_key_cols and len(self.source_key_cols) != len(self.target_key_cols):
raise ValueError("source_key_cols and target_key_cols must have equal lengths")
if self.multi_lang_mode is not None and self.multi_lang_mode not in ("single_call", "per_language"):
raise ValueError("multi_lang_mode must be 'single_call' or 'per_language'")
if self.batch_aggressiveness is not None and self.batch_aggressiveness not in ("safe", "balanced", "fast"):

16
backend/tests/plugins/translate/test_orchestrator_sql_rows.py
View File

@@ -26,6 +26,7 @@ from src.plugins.translate.orchestrator_sql_rows import (
@pytest.fixture
def mock_job():
job = MagicMock()
job.source_key_cols = ["product_id", "store_id"]
job.target_key_cols = ["product_id", "store_id"]
job.target_language_column = "lang"
job.target_source_column = "source_text"
@@ -269,6 +270,21 @@ class TestBuildRows:
assert original["store_id"] is None
assert original["product_id"] == 1
def test_source_keys_map_to_target_keys(self, mock_job):
mock_job.source_key_cols = ["source_id", "source_event"]
mock_job.target_key_cols = ["id", "event_time"]
rec = self._make_record(
source_data={"source_id": 42, "source_event": "2024-01-15 12:00:00"},
)
rows = build_rows(
records=[rec], job=mock_job, effective_target="translated_name",
primary_language="en", context_keys=[],
)
original = [row for row in rows if row["is_original"] == 1][0]
assert original["id"] == 42
assert original["event_time"] == "2024-01-15 12:00:00"
assert "source_id" not in original
# ── Lines 96-106: translation language iteration ──
def test_translation_loop_creates_translated_rows(self, mock_job):

64
backend/tests/plugins/translate/test_sql_generator.py
View File

@@ -11,6 +11,7 @@
# @TEST_EDGE: unknown_dialect -> ANSI fallback
# @TEST_EDGE: timestamp_normalization -> Converts Unix timestamps
from datetime import UTC, datetime
from decimal import Decimal
from typing import Any
import pytest
@@ -44,20 +45,20 @@ class TestNormalizeTimestampValue:
def test_unix_seconds(self):
# 2024-01-15 12:00:00 UTC = 1705320000
result = _normalize_timestamp_value(1705320000)
result = _normalize_timestamp_value(1705320000, "Date")
assert result == "2024-01-15"
def test_unix_milliseconds(self):
# 2024-01-15 12:00:00 UTC = 1705320000000 ms
result = _normalize_timestamp_value(1705320000000)
result = _normalize_timestamp_value(1705320000000, "Date")
assert result == "2024-01-15"
def test_string_unix_timestamp(self):
result = _normalize_timestamp_value("1705320000")
result = _normalize_timestamp_value("1705320000", "Date")
assert result == "2024-01-15"
def test_string_unix_millis(self):
result = _normalize_timestamp_value("1705320000000.0")
result = _normalize_timestamp_value("1705320000000.0", "Date")
assert result == "2024-01-15"
def test_out_of_range_small(self):
@@ -74,11 +75,11 @@ class TestNormalizeTimestampValue:
def test_overflow_error(self):
# datetime.fromtimestamp can raise OverflowError for extreme values
# 1e15 is still within range for most systems
result = _normalize_timestamp_value(9999999999) # ~2286
result = _normalize_timestamp_value(9999999999, "Date") # ~2286
assert result is not None
def test_float_milliseconds(self):
result = _normalize_timestamp_value(1705320000000.0)
result = _normalize_timestamp_value(1705320000000.0, "Date")
assert result == "2024-01-15"
def test_empty_string(self):
@@ -160,13 +161,33 @@ class TestEncodeSqlValue:
def test_string_with_quote(self):
assert _encode_sql_value("it's") == "'it''s'"
def test_clickhouse_with_timestamp_string(self):
result = _encode_sql_value("1705320000", dialect="clickhouse")
def test_epoch_zero_requires_explicit_date_type(self):
assert _normalize_timestamp_value(0, "Date") == "1970-01-01"
assert _normalize_timestamp_value(0) is None
def test_clickhouse_timestamp_requires_date_type(self):
result = _encode_sql_value("1705320000", dialect="clickhouse", target_type="Date")
assert result == "'2024-01-15'"
def test_clickhouse_with_timestamp_number(self):
def test_clickhouse_timestamp_like_id_is_preserved(self):
result = _encode_sql_value(1705320000, dialect="clickhouse")
assert result == "'2024-01-15'"
assert result == "1705320000"
def test_clickhouse_datetime_preserves_time(self):
result = _encode_sql_value(1705320000123, dialect="clickhouse", target_type="DateTime")
assert result == "'2024-01-15 12:00:00'"
def test_clickhouse_datetime64_preserves_fraction(self):
result = _encode_sql_value(1705320000123, dialect="clickhouse", target_type="DateTime64(3)")
assert result == "'2024-01-15 12:00:00.123'"
@pytest.mark.parametrize("value", [float("nan"), float("inf"), float("-inf")])
def test_non_finite_float_is_rejected(self, value):
with pytest.raises(ValueError, match="NaN and infinity"):
_encode_sql_value(value, dialect="clickhouse")
def test_decimal_is_numeric(self):
assert _encode_sql_value(Decimal("1.2300"), dialect="clickhouse") == "1.2300"
def test_clickhouse_regular_string(self):
result = _encode_sql_value("hello", dialect="clickhouse")
@@ -181,7 +202,28 @@ class TestEncodeSqlValue:
def test_non_clickhouse_passes_through(self):
result = _encode_sql_value("1705320000", dialect="postgresql")
assert result == "'1705320000'" # Not normalized
assert result == "'1705320000'"
def test_generate_batch_forwards_column_types_to_clickhouse(self):
statements = SQLGenerator.generate_batch(
dialect="clickhouse",
target_schema=None,
target_table="events",
columns=["created_at"],
rows=[{"created_at": 1705320000123}],
upsert_strategy="INSERT",
column_types={"created_at": "DateTime64(3)"},
)
assert "'2024-01-15 12:00:00.123'" in statements[0][0]
def test_values_clause_uses_column_types(self):
result = _build_values_clause(
["event_id", "created_at"],
[{"event_id": 1705320000, "created_at": 1705320000123}],
dialect="clickhouse",
column_types={"created_at": "DateTime64(3)"},
)
assert result == "(1705320000, '2024-01-15 12:00:00.123')"
# ---------------------------------------------------------------------------