diff --git a/backend/src/plugins/translate/_batch_insert.py b/backend/src/plugins/translate/_batch_insert.py index cd06b7280..ff6d65b5d 100644 --- a/backend/src/plugins/translate/_batch_insert.py +++ b/backend/src/plugins/translate/_batch_insert.py @@ -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: diff --git a/backend/src/plugins/translate/_run_service.py b/backend/src/plugins/translate/_run_service.py index be0d5a4c8..e145ca8af 100644 --- a/backend/src/plugins/translate/_run_service.py +++ b/backend/src/plugins/translate/_run_service.py @@ -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: diff --git a/backend/src/plugins/translate/orchestrator_sql_rows.py b/backend/src/plugins/translate/orchestrator_sql_rows.py index 3903e53d6..92281a5ce 100644 --- a/backend/src/plugins/translate/orchestrator_sql_rows.py +++ b/backend/src/plugins/translate/orchestrator_sql_rows.py @@ -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: diff --git a/backend/src/plugins/translate/preview.py b/backend/src/plugins/translate/preview.py index 0d2375730..17ac8c3ba 100644 --- a/backend/src/plugins/translate/preview.py +++ b/backend/src/plugins/translate/preview.py @@ -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) diff --git a/backend/src/plugins/translate/sql_generator.py b/backend/src/plugins/translate/sql_generator.py index 0c9658741..57b6f02b8 100644 --- a/backend/src/plugins/translate/sql_generator.py +++ b/backend/src/plugins/translate/sql_generator.py @@ -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)) diff --git a/backend/src/schemas/translate.py b/backend/src/schemas/translate.py index 4a9c4f8e5..d61fd1513 100644 --- a/backend/src/schemas/translate.py +++ b/backend/src/schemas/translate.py @@ -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"): diff --git a/backend/tests/plugins/translate/test_orchestrator_sql_rows.py b/backend/tests/plugins/translate/test_orchestrator_sql_rows.py index 371d40911..bdeea2dcd 100644 --- a/backend/tests/plugins/translate/test_orchestrator_sql_rows.py +++ b/backend/tests/plugins/translate/test_orchestrator_sql_rows.py @@ -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): diff --git a/backend/tests/plugins/translate/test_sql_generator.py b/backend/tests/plugins/translate/test_sql_generator.py index cf4b76aa1..89054d890 100644 --- a/backend/tests/plugins/translate/test_sql_generator.py +++ b/backend/tests/plugins/translate/test_sql_generator.py @@ -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')" # ---------------------------------------------------------------------------