From 487261102721d3e5120b5993ad85ca5fe019fbb2 Mon Sep 17 00:00:00 2001 From: lukasWuttke <54042461+LukasWodka@users.noreply.github.com> Date: Thu, 23 Jul 2026 08:56:44 +0200 Subject: [PATCH 1/5] fix(validators): stop doubling "Validator" in validation log lines (D1) (#392) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit validator.name already ends in "Validator" (e.g. "Data Validator"), so appending the word again produced "Data Validator Validator successfully passed" (and the same doubling in the failed/error paths). Drop the redundant literal in all three spots so the lines read "Data Validator successfully passed" / "Data Validator failed: …" / "Data Validator error: …". Log-string only, no logic change; updated the one docstring that quoted the old doubled message. Co-authored-by: Claude Opus 4.8 --- tests/test_data_validator.py | 41 ++++++++++++++++------------ tracebloc_ingestor/ingestors/base.py | 18 ++++++------ 2 files changed, 31 insertions(+), 28 deletions(-) diff --git a/tests/test_data_validator.py b/tests/test_data_validator.py index c39360a..4df3f0c 100644 --- a/tests/test_data_validator.py +++ b/tests/test_data_validator.py @@ -8,11 +8,11 @@ from tracebloc_ingestor.validators.data_validator import DataValidator - # --------------------------------------------------------------------------- # validate() top-level flow # --------------------------------------------------------------------------- + def test_no_schema_passes_with_warning(): result = DataValidator().validate(pd.DataFrame({"a": [1]})) assert result.is_valid @@ -73,7 +73,7 @@ def test_loads_from_json(tmp_path): Surfaced by an end-to-end cluster ingestion: a 20-record JSON file with explicit schema {id INT, age INT, score FLOAT, active BOOL, label - VARCHAR(20)} failed with "Data Validator Validator failed: No data found + VARCHAR(20)} failed with "Data Validator failed: No data found to validate" before any record was read. """ p = tmp_path / "d.json" @@ -186,9 +186,7 @@ def test_loads_from_json_scalar_array_fails_not_silently_passes(tmp_path): """ p = tmp_path / "scalars.json" p.write_text("[1, 2, 3]") - result = DataValidator(schema={"id": "INT", "label": "VARCHAR(8)"}).validate( - str(p) - ) + result = DataValidator(schema={"id": "INT", "label": "VARCHAR(8)"}).validate(str(p)) assert not result.is_valid assert "No data found" in result.errors[0] @@ -199,9 +197,7 @@ def test_loads_from_json_mixed_array_keeps_only_objects(tmp_path): the dict records (bugbot #233).""" p = tmp_path / "mixed.json" p.write_text('[{"id": 1, "label": "A"}, 42, {"id": 2, "label": "B"}]') - result = DataValidator(schema={"id": "INT", "label": "VARCHAR(8)"}).validate( - str(p) - ) + result = DataValidator(schema={"id": "INT", "label": "VARCHAR(8)"}).validate(str(p)) assert result.is_valid, f"expected valid; errors={result.errors}" assert result.metadata["rows_checked"] == 2 @@ -218,13 +214,9 @@ def test_loads_from_json_jsonl_surfaces_clear_error(tmp_path): """ p = tmp_path / "events.json" p.write_text( - '{"id":1,"label":"A"}\n' - '{"id":2,"label":"B"}\n' - '{"id":3,"label":"A"}\n' - ) - result = DataValidator(schema={"id": "INT", "label": "VARCHAR(8)"}).validate( - str(p) + '{"id":1,"label":"A"}\n' '{"id":2,"label":"B"}\n' '{"id":3,"label":"A"}\n' ) + result = DataValidator(schema={"id": "INT", "label": "VARCHAR(8)"}).validate(str(p)) assert not result.is_valid err = result.errors[0] assert "JSONL" in err or "newline-delimited" in err @@ -392,6 +384,7 @@ def test_dataframe_sparse_schema_field_still_passes(): # INT / BIGINT # --------------------------------------------------------------------------- + def test_int_valid(): df = pd.DataFrame({"n": [1, 2, 3]}) assert DataValidator(schema={"n": "INT"}).validate(df).is_valid @@ -420,7 +413,10 @@ def test_bigint_delegates_to_int(): # FLOAT / DOUBLE / DECIMAL # --------------------------------------------------------------------------- -@pytest.mark.parametrize("dtype", ["FLOAT", "DOUBLE", "DECIMAL(10,2)", "NUMERIC(8,3)", "NUMERIC"]) + +@pytest.mark.parametrize( + "dtype", ["FLOAT", "DOUBLE", "DECIMAL(10,2)", "NUMERIC(8,3)", "NUMERIC"] +) def test_float_family_valid(dtype): # NUMERIC is a MySQL alias for DECIMAL; #190 bugbot caught that the DDL # and ingestor type-cast layers accepted NUMERIC but the validator's @@ -440,6 +436,7 @@ def test_float_non_numeric_fails(): # Missing values are NULL, not "non-numeric" (regression) # --------------------------------------------------------------------------- + def test_int_with_missing_values_is_valid(): # NaN/empty in an INT column is a missing value (stored as NULL), not a # non-numeric one — and not a "non-integer" one either. @@ -480,6 +477,7 @@ def test_non_numeric_error_points_at_rows_without_leaking_values(): # Non-finite (inf) + realistic messy-data scenarios # --------------------------------------------------------------------------- + def test_float_infinity_is_rejected(): # inf survives pd.to_numeric (it is "numeric") but flows through the training # scaler unchanged and produces a NaN loss — so the gate must reject it. @@ -530,6 +528,7 @@ def test_scientific_notation_is_valid(): # VARCHAR / CHAR / TEXT # --------------------------------------------------------------------------- + def test_varchar_valid(): df = pd.DataFrame({"s": ["abc", "de"]}) assert DataValidator(schema={"s": "VARCHAR(255)"}).validate(df).is_valid @@ -614,8 +613,10 @@ def test_issue_188_full_repro_csv(make_csv): ) result = DataValidator( schema={ - "id": "INT", "feat": "FLOAT", - "code": "VARCHAR(10)", "label": "VARCHAR(8)", + "id": "INT", + "feat": "FLOAT", + "code": "VARCHAR(10)", + "label": "VARCHAR(8)", } ).validate(str(path)) assert result.is_valid, f"expected valid; errors={result.errors}" @@ -626,7 +627,7 @@ def test_issue_188_numeric_scalars_pass_string_family(dtype): # All three string-family validators share the same astype(str)!=value # over-rejection; the fix is applied uniformly. df = pd.DataFrame({"s": ["abc", 100, 200]}) # ints mixed in (CHAR case - # is special — see below) + # is special — see below) if "CHAR(" in dtype: # CHAR enforces fixed length, so use values that all stringify to 3. df = pd.DataFrame({"s": ["abc", 100, 200]}) @@ -650,6 +651,7 @@ def test_text_valid(): # BOOLEAN across dtypes # --------------------------------------------------------------------------- + def test_boolean_bool_dtype_valid(): df = pd.DataFrame({"b": [True, False]}) assert DataValidator(schema={"b": "BOOLEAN"}).validate(df).is_valid @@ -698,6 +700,7 @@ def test_boolean_float_invalid_fails(): # DATE / DATETIME / TIMESTAMP / TIME # --------------------------------------------------------------------------- + @pytest.mark.parametrize("dtype", ["DATE", "DATETIME", "TIMESTAMP", "TIME"]) def test_date_family_valid(dtype): df = pd.DataFrame({"d": ["2024-01-01", "2024-02-02"]}) @@ -737,6 +740,7 @@ def test_date_still_flags_real_bad_value_with_nulls_present(): # type parsing + auto-detect helper # --------------------------------------------------------------------------- + def test_constraints_are_stripped_from_type(): df = pd.DataFrame({"n": [1, 2]}) # "INT NOT NULL" should resolve to the INT validator. @@ -761,6 +765,7 @@ def test_detect_column_type(series, expected): # Streaming / large-file validation (bounded memory, full coverage) # --------------------------------------------------------------------------- + def test_validate_csv_streams_and_catches_late_chunk(tmp_path): # A bad value PAST the first chunk must still be caught — the validator # scans the whole file chunk by chunk (memory-bounded), not just a sample diff --git a/tracebloc_ingestor/ingestors/base.py b/tracebloc_ingestor/ingestors/base.py index 241b32c..73c03d6 100644 --- a/tracebloc_ingestor/ingestors/base.py +++ b/tracebloc_ingestor/ingestors/base.py @@ -552,7 +552,7 @@ def validate_data(self, source: Any) -> bool: if not result.is_valid: all_valid = False validation_errors.append( - f"{BOLD}{validator.name} Validator failed: {RESET} \n {RED}" + f"{BOLD}{validator.name} failed: {RESET} \n {RED}" ) validation_errors.extend(result.errors) validation_errors.append(f"{RESET}") @@ -563,12 +563,13 @@ def validate_data(self, source: Any) -> bool: f"{YELLOW}Validation warning - {validator.name}: {warning}{RESET}" ) if result.is_valid: - print( - f"{GREEN}{validator.name} Validator successfully passed{RESET}" - ) + # validator.name already ends in "Validator" (e.g. "Data + # Validator"), so don't append the word again — that produced + # "Data Validator Validator successfully passed". + print(f"{GREEN}{validator.name} successfully passed{RESET}") except Exception as e: all_valid = False - validation_errors.append(f"Validator {validator.name} error: {str(e)}") + validation_errors.append(f"{validator.name} error: {str(e)}") if not all_valid: error_summary = "\n".join(validation_errors) @@ -864,9 +865,7 @@ def _ingest_with_lock( and not self.unique_id_column and self._table_salt is None ): - self._table_salt = self.database.get_or_create_table_salt( - self.table_name - ) + self._table_salt = self.database.get_or_create_table_salt(self.table_name) if self.table is None: # Grouped categories get a composite (group, time) secondary @@ -1130,8 +1129,7 @@ def _ingest_with_lock( elif not label_counts: counts_helper = ( "get_label_sequence_counts" - if grouping is not None - and grouping.count_unit == "sequences" + if grouping is not None and grouping.count_unit == "sequences" else "get_label_counts" ) raise RuntimeError( From 10445cd5b69b26230ada390aae98d7089afa111c Mon Sep 17 00:00:00 2001 From: Divya Date: Thu, 23 Jul 2026 19:35:31 +0530 Subject: [PATCH 2/5] =?UTF-8?q?feat(backfill):=20metadata=20backfill=20run?= =?UTF-8?q?ner=20=E2=80=94=20GET=20dataset=20=E2=86=92=20recompute=20?= =?UTF-8?q?=E2=86=92=20send=20to=20backend=20(#378=20send=20side)=20(#393)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(backfill): metadata backfill runner — GET dataset, recompute, send to backend Wires data-ingestors #378 (build_dataset_metadata) to the backend metadata endpoints (#1166 POST, #1198 GET) into a one-time, per-client pre-cutover backfill sweep. For each registered dataset it: 1. GETs the backend record by ingestor_id (authoritative category + data_format, not recoverable from the table alone); 2. recomputes {schema, meta_data} from persisted rows via build_dataset_metadata, keyed on that category; 3. POSTs it back — the endpoint upserts GlobalMetaData and re-folds any competition built from the table. - APIClient.get_dataset_metadata / send_metadata_backfill — the two API calls, with local-mode mocks, 404→None (skip), and error surfacing. - Database.list_registered_runs — enumerate the client's REGISTERED ingest journal (the datasets to sweep). - metadata_backfill_runner — backfill_dataset (GET→build→POST, skips competitions/merged and already-new-shape datasets) + backfill_datasets (per-dataset guarded sweep) + a `tracebloc-backfill` console-script entrypoint reading DB/backend creds from env (no ingest.yaml). Idempotent and re-runnable; new ingests already emit new-shape metadata, so this is a rollout step, not always-on. Tests: runner orchestration/skip/guard paths, the two API methods (success/404/error/local/payload shape), and the DB enumeration. Co-Authored-By: Claude Opus 4.8 * chore: retrigger checks after retargeting PR base to develop The FR gate (staging/prod-promotion guard) ran against the original master base and its stale failure lingered on the head commit after retargeting to develop, where the gate does not apply. Empty commit to reset the head SHA. Co-Authored-By: Claude Opus 4.8 * fix(backfill): don't leak raw exception text / response bodies in the sweep (Bugbot #393) Per the org rule "error messages and logs must never embed customer cell values": - Runner: on a per-dataset failure, record and log the exception TYPE only (`type(exc).__name__`) instead of `str(exc)[:500]`, and use logger.error rather than logger.exception (no traceback). This runs in a Helm-hook Job whose output lands in install logs. - APIClient.get_dataset_metadata / send_metadata_backfill: raise HTTPError("HTTP ") without response.text. The backfill POST body carries this table's categorical vocab (customer cell values), so a backend error echoing it back must not reach the exception string / logs. The response object is still attached for deliberate programmatic inspection. Test asserts the recorded error is the exception type ("RuntimeError"), proving the raw message ("boom") is not surfaced. Co-Authored-By: Claude Opus 4.8 --------- Co-authored-by: Claude Opus 4.8 --- setup.py | 4 + tests/test_api_client_methods.py | 82 +++++++ tests/test_database.py | 25 ++ tests/test_metadata_backfill_runner.py | 181 ++++++++++++++ tracebloc_ingestor/api/client.py | 99 ++++++++ tracebloc_ingestor/database.py | 28 +++ .../metadata_backfill_runner.py | 232 ++++++++++++++++++ 7 files changed, 651 insertions(+) create mode 100644 tests/test_metadata_backfill_runner.py create mode 100644 tracebloc_ingestor/metadata_backfill_runner.py diff --git a/setup.py b/setup.py index c3893a4..b17d0ce 100644 --- a/setup.py +++ b/setup.py @@ -67,6 +67,10 @@ def _read_requirements(filename): # schema/ingest.v1.json, and dispatches to the right ingestor. "console_scripts": [ "tracebloc-ingest = tracebloc_ingestor.cli.run:main", + # One-time pre-cutover metadata backfill (backend#1166/#1198): reads + # DB + backend creds from the same env as the ingestor (no + # ingest.yaml) and sweeps this client's registered datasets. + "tracebloc-backfill = tracebloc_ingestor.metadata_backfill_runner:main", ], }, classifiers=[ diff --git a/tests/test_api_client_methods.py b/tests/test_api_client_methods.py index b7ca4b6..2830d2a 100644 --- a/tests/test_api_client_methods.py +++ b/tests/test_api_client_methods.py @@ -263,3 +263,85 @@ def test_refresh_token_false_when_reauth_raises(): client.token = "old_token" with patch.object(client, "authenticate", side_effect=RuntimeError("auth down")): assert client._refresh_token() is False + + +# --------------------------------------------------------------------------- +# get_dataset_metadata() — read-by-ingestor_id (backend#1198) +# --------------------------------------------------------------------------- + + +def test_get_dataset_metadata_success(): + client = _client() + body = {"table_name": "tb", "category": "tabular_classification", "meta_data": {}} + with patch.object(client.session, "get", return_value=_resp(200, body)): + result = client.get_dataset_metadata("ing-1") + assert result == body + + +def test_get_dataset_metadata_404_returns_none(): + """A 404 (no dataset for this ingestor_id owned by this edge) is a skip + signal for the sweep, not an error — return None rather than raise.""" + client = _client() + with patch.object(client.session, "get", return_value=_resp(404, text="nope")): + assert client.get_dataset_metadata("missing") is None + + +def test_get_dataset_metadata_http_error_raises(): + client = _client() + with patch.object(client.session, "get", return_value=_resp(500, text="boom")): + with pytest.raises(requests.exceptions.RequestException): + client.get_dataset_metadata("ing-1") + + +def test_get_dataset_metadata_local_mode(): + client = _client(EDGE_ENV="local") + with patch.object(client.session, "get") as get: + assert client.get_dataset_metadata("ing-1") is None + get.assert_not_called() + + +# --------------------------------------------------------------------------- +# send_metadata_backfill() — upsert recomputed metadata (backend#1166/#1198) +# --------------------------------------------------------------------------- + + +def test_send_metadata_backfill_success(): + client = _client() + body = {"table_name": "tb", "created": True, "competitions_refolded": 2} + with patch.object(client.session, "post", return_value=_resp(201, body)): + result = client.send_metadata_backfill("tb", {"x": {"dtype": "int"}}, {}) + assert result == body + + +def test_send_metadata_backfill_http_error_raises(): + client = _client() + with patch.object(client.session, "post", return_value=_resp(404, text="no table")): + with pytest.raises(requests.exceptions.RequestException): + client.send_metadata_backfill("tb", {"x": {"dtype": "int"}}, {}) + + +def test_send_metadata_backfill_local_mode(): + client = _client(EDGE_ENV="local") + with patch.object(client.session, "post") as post: + result = client.send_metadata_backfill("tb", {"x": {"dtype": "int"}}, {}) + post.assert_not_called() + assert result["table_name"] == "tb" + + +def test_send_metadata_backfill_payload_shape(): + import json as _json + + client = _client() + captured = {} + + def _capture(url, **kwargs): + captured["url"] = url + captured["data"] = kwargs.get("data") + return _resp(201, {"table_name": "tb", "created": False, "competitions_refolded": 0}) + + with patch.object(client.session, "post", side_effect=_capture): + client.send_metadata_backfill("tb", {"x": {"dtype": "int"}}, {"attributes": {}}) + + assert captured["url"].endswith("/global_meta/metadata_backfill/tb/") + sent = _json.loads(captured["data"]) + assert sent == {"schema": {"x": {"dtype": "int"}}, "meta_data": {"attributes": {}}} diff --git a/tests/test_database.py b/tests/test_database.py index ee502d9..93167df 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -1004,3 +1004,28 @@ def test_create_table_rejects_reserved_bookkeeping_tables(reserved): db = Database.__new__(Database) with pytest.raises(ValueError, match="reserved"): db.create_table(reserved, {"feature_0": "FLOAT"}) + + +# --------------------------------------------------------------------------- +# Run journal: enumeration for the metadata backfill sweep +# --------------------------------------------------------------------------- + + +def test_list_registered_runs_returns_registered_only(db, mock_engine_factory): + """The backfill sweep enumerates only REGISTERED runs, one row each, mapped + to {ingestor_id, table_name, task}. A NULL task (pre-task-column run) is + preserved as None.""" + _, _, conn = mock_engine_factory + conn.execute.return_value.fetchall.return_value = [ + ("run-a", "ta", "tabular_classification"), + ("run-b", "tb", None), + ] + result = db.list_registered_runs() + assert result == [ + {"ingestor_id": "run-a", "table_name": "ta", "task": "tabular_classification"}, + {"ingestor_id": "run-b", "table_name": "tb", "task": None}, + ] + select = next( + s for s in _executed_sql(conn) if "SELECT ingestor_id, table_name, task" in s + ) + assert "WHERE registered = 1" in select diff --git a/tests/test_metadata_backfill_runner.py b/tests/test_metadata_backfill_runner.py new file mode 100644 index 0000000..37e9c57 --- /dev/null +++ b/tests/test_metadata_backfill_runner.py @@ -0,0 +1,181 @@ +"""Tests for the metadata backfill runner — the GET → recompute → POST sweep +that wires build_dataset_metadata to the backend metadata-backfill endpoint. + +build_dataset_metadata and the APIClient are the units-under-test's collaborators +and are exercised elsewhere; here they are mocked so the runner's own +orchestration/skip/guard logic is what's covered. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +from tracebloc_ingestor import metadata_backfill_runner as runner +from tracebloc_ingestor.metadata_backfill_runner import ( + STATUS_ERROR, + STATUS_NOT_FOUND, + STATUS_OK, + STATUS_SKIPPED_COMPETITION, + STATUS_SKIPPED_CURRENT, + backfill_dataset, + backfill_datasets, +) + +# A pre-cutover plain dataset record as the GET returns it — no new-shape +# attributes yet, not a competition. +_PLAIN_RECORD = { + "id": 1, + "table_name": "tb", + "category": "tabular_classification", + "data_format": "tabular", + "intent": "train", + "ingestor_id": "ing-1", + "is_competition": False, + "source_dataset_ids": [], + "schema": {"x": "INT"}, + "meta_data": {}, +} + +_PAYLOAD = { + "schema": {"x": {"dtype": "int"}}, + "meta_data": {"attributes": {"feature_stats": {"x": {"count": 5}}}}, +} + + +def _api(record=None, send_result=None): + api = MagicMock() + api.get_dataset_metadata.return_value = record + api.send_metadata_backfill.return_value = send_result or { + "table_name": "tb", + "created": True, + "competitions_refolded": 2, + } + return api + + +# --------------------------------------------------------------------------- +# backfill_dataset — single dataset +# --------------------------------------------------------------------------- + + +def test_backfill_dataset_happy_path_gets_builds_and_posts(): + api = _api(record=dict(_PLAIN_RECORD)) + db = MagicMock() + with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD) as build: + result = backfill_dataset(db, api, "ing-1") + + # GET by ingestor_id, build keyed on the backend's category/data_format, + # then POST the recomputed payload back. + api.get_dataset_metadata.assert_called_once_with("ing-1") + build.assert_called_once() + assert build.call_args.kwargs["category"] == "tabular_classification" + assert build.call_args.kwargs["data_format"] == "tabular" + api.send_metadata_backfill.assert_called_once_with( + "tb", _PAYLOAD["schema"], _PAYLOAD["meta_data"] + ) + assert result.status == STATUS_OK + assert result.created is True + assert result.competitions_refolded == 2 + + +def test_backfill_dataset_not_found_returns_status_and_skips_build(): + api = _api(record=None) # GET 404 → None + db = MagicMock() + with patch.object(runner, "build_dataset_metadata") as build: + result = backfill_dataset(db, api, "missing") + assert result.status == STATUS_NOT_FOUND + build.assert_not_called() + api.send_metadata_backfill.assert_not_called() + + +def test_backfill_dataset_skips_competition(): + record = dict(_PLAIN_RECORD, is_competition=True) + api = _api(record=record) + db = MagicMock() + with patch.object(runner, "build_dataset_metadata") as build: + result = backfill_dataset(db, api, "ing-1") + assert result.status == STATUS_SKIPPED_COMPETITION + build.assert_not_called() + api.send_metadata_backfill.assert_not_called() + + +def test_backfill_dataset_skips_merged_with_source_ids(): + record = dict(_PLAIN_RECORD, source_dataset_ids=[7, 8]) + api = _api(record=record) + with patch.object(runner, "build_dataset_metadata") as build: + result = backfill_dataset(MagicMock(), api, "ing-1") + assert result.status == STATUS_SKIPPED_COMPETITION + build.assert_not_called() + + +def test_backfill_dataset_skips_when_already_new_shape(): + record = dict( + _PLAIN_RECORD, + meta_data={"attributes": {"feature_stats": {"x": {"count": 1}}}}, + ) + api = _api(record=record) + with patch.object(runner, "build_dataset_metadata") as build: + result = backfill_dataset(MagicMock(), api, "ing-1") + assert result.status == STATUS_SKIPPED_CURRENT + build.assert_not_called() + api.send_metadata_backfill.assert_not_called() + + +def test_backfill_dataset_reprocesses_current_when_skip_disabled(): + record = dict( + _PLAIN_RECORD, + meta_data={"attributes": {"feature_stats": {"x": {"count": 1}}}}, + ) + api = _api(record=record) + with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD): + result = backfill_dataset(MagicMock(), api, "ing-1", skip_if_current=False) + assert result.status == STATUS_OK + api.send_metadata_backfill.assert_called_once() + + +def test_backfill_dataset_error_when_record_missing_category(): + record = dict(_PLAIN_RECORD, category=None) + api = _api(record=record) + with patch.object(runner, "build_dataset_metadata") as build: + result = backfill_dataset(MagicMock(), api, "ing-1") + assert result.status == STATUS_ERROR + build.assert_not_called() + + +# --------------------------------------------------------------------------- +# backfill_datasets — the sweep +# --------------------------------------------------------------------------- + + +def test_backfill_datasets_enumerates_registered_runs_when_ids_omitted(): + db = MagicMock() + db.list_registered_runs.return_value = [ + {"ingestor_id": "a", "table_name": "ta", "task": "tabular_classification"}, + {"ingestor_id": "b", "table_name": "tb", "task": "tabular_classification"}, + ] + api = _api(record=dict(_PLAIN_RECORD)) + with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD): + results = backfill_datasets(db, api) + db.list_registered_runs.assert_called_once() + assert [r.ingestor_id for r in results] == ["a", "b"] + assert all(r.status == STATUS_OK for r in results) + + +def test_backfill_datasets_continues_past_a_failing_dataset(): + api = MagicMock() + # First GET raises (e.g. transient), second succeeds. + api.get_dataset_metadata.side_effect = [ + RuntimeError("boom"), + dict(_PLAIN_RECORD, ingestor_id="b"), + ] + api.send_metadata_backfill.return_value = { + "table_name": "tb", + "created": False, + "competitions_refolded": 0, + } + with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD): + results = backfill_datasets(MagicMock(), api, ["a", "b"]) + assert results[0].status == STATUS_ERROR + # Exception TYPE only — never the raw message (may embed customer cell values). + assert results[0].error == "RuntimeError" + assert results[1].status == STATUS_OK # sweep continued diff --git a/tracebloc_ingestor/api/client.py b/tracebloc_ingestor/api/client.py index d4ebbe7..8df074e 100644 --- a/tracebloc_ingestor/api/client.py +++ b/tracebloc_ingestor/api/client.py @@ -322,6 +322,105 @@ def send_ingest_summary( logger.error(f"{RED}Error sending ingest summary: {str(e)[:500]}{RESET}") raise + def get_dataset_metadata(self, ingestor_id: str) -> Optional[Dict[str, Any]]: + """Fetch a dataset's backend record by ``ingestor_id`` (backend#1198). + + Returns the core fields — ``table_name`` / ``category`` / ``data_format`` + / ``intent`` / ``is_competition`` / ``source_dataset_ids`` — plus the + currently-stored ``schema`` / ``meta_data``. Backs the pre-cutover + backfill runner: ``category`` and ``data_format`` drive the recompute and + are not recoverable from the table alone, so the runner reads them here + first. + + Returns ``None`` when the backend has no dataset for this ``ingestor_id`` + owned by this edge (404) so a sweep can skip it rather than crash. + + Raises: + requests.exceptions.HTTPError: on any non-404 error status. + """ + if self.config.EDGE_ENV == "local": + logger.info( + f"Mock: would fetch dataset metadata for ingestor_id={ingestor_id}" + ) + return None + + url = ( + f"{self.config.API_ENDPOINT}" + f"/global_meta/metadata_backfill/by-ingestor/{ingestor_id}/" + ) + response = self._authed_request("GET", url, timeout=API_TIMEOUT) + if response.status_code == 404: + logger.warning( + f"{YELLOW}No backend dataset for ingestor_id={ingestor_id} " + f"owned by this edge (404); skipping.{RESET}" + ) + return None + if response.status_code >= 400: + # Status only in the message — NOT response.text. A backend error + # body can echo the dataset's metadata (categorical vocab = customer + # cell values), and this exception surfaces in the backfill runner's + # install-log output. The response object is attached for a caller + # that needs to inspect it deliberately. (bugbot) + raise requests.exceptions.HTTPError( + f"HTTP {response.status_code}", response=response + ) + return self._parse_json(response, required=True) + + def send_metadata_backfill( + self, + table_name: str, + schema: Dict[str, Any], + meta_data: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + """Upsert recomputed ``{schema, meta_data}`` for a PRE-EXISTING table via + the metadata-backfill endpoint (backend#1166/#1198). + + Metadata-only: never creates a dataset (the table must already exist + backend-side — 404 otherwise). The POST also propagates the refresh into + any competition built from the table (re-fold). Idempotent, so a re-run + is a safe overwrite. + + Returns: + The backend body — ``{"table_name", "created", "competitions_refolded"}``. + + Raises: + requests.exceptions.HTTPError: If the API call fails after retries. + """ + if self.config.EDGE_ENV == "local": + logger.info( + f"Mock: would backfill metadata for {table_name} " + f"({len(schema)} schema column(s))" + ) + return { + "table_name": table_name, + "created": False, + "competitions_refolded": 0, + } + + payload = json.dumps({"schema": schema, "meta_data": meta_data or {}}) + response = self._authed_request( + "POST", + f"{self.config.API_ENDPOINT}/global_meta/metadata_backfill/{table_name}/", + extra_headers={"Content-Type": "application/json"}, + data=payload, + timeout=API_TIMEOUT, + ) + if response.status_code >= 400: + # Status only — NOT response.text (see get_dataset_metadata): the + # POST body carries this table's categorical vocab, so a backend + # error echoing it back would embed customer cell values in the + # runner's install-log output. (bugbot) + raise requests.exceptions.HTTPError( + f"HTTP {response.status_code}", response=response + ) + result = self._parse_json(response, required=True) + logger.info( + f"{GREEN}Backfilled metadata for {table_name}: " + f"created={result.get('created')}, " + f"competitions_refolded={result.get('competitions_refolded')}{RESET}" + ) + return result + def __del__(self): """Cleanup when the client is destroyed""" if hasattr(self, "session"): diff --git a/tracebloc_ingestor/database.py b/tracebloc_ingestor/database.py index b850b83..0481d7c 100644 --- a/tracebloc_ingestor/database.py +++ b/tracebloc_ingestor/database.py @@ -862,6 +862,34 @@ def reclaim_dead_run_rows( ) return reclaimed + def list_registered_runs(self) -> List[Dict[str, str]]: + """Return the REGISTERED ingest runs journalled on this client, one per + dataset: ``[{"ingestor_id", "table_name", "task"}, ...]``. + + The run journal (``record_ingest_started`` / ``mark_ingest_registered``) + is the authoritative local list of datasets this edge has ingested and + successfully registered with the backend — exactly the set the + pre-cutover metadata backfill sweeps. Only ``registered = 1`` rows are + returned, so a started-but-never-registered (dead) run is never + backfilled. ``task`` is the recorded category (may be NULL on runs + journalled before the column existed); the backfill runner re-reads the + authoritative category from the backend anyway. + """ + with self.engine.connect() as connection: + self._ensure_runs_table(connection) + rows = _execute_with_retry( + connection, + text( + f"SELECT ingestor_id, table_name, task " + f"FROM `{self.RUNS_TABLE}` WHERE registered = 1 " + f"ORDER BY started_at" + ), + ).fetchall() + return [ + {"ingestor_id": iid, "table_name": tname, "task": task} + for (iid, tname, task) in rows + ] + def get_label_counts(self, table_name: str, ingestor_id: str) -> Dict[str, int]: """ Return ``{label: row_count}`` for every label inserted by *ingestor_id*. diff --git a/tracebloc_ingestor/metadata_backfill_runner.py b/tracebloc_ingestor/metadata_backfill_runner.py new file mode 100644 index 0000000..5bbf105 --- /dev/null +++ b/tracebloc_ingestor/metadata_backfill_runner.py @@ -0,0 +1,232 @@ +"""Pre-cutover metadata backfill runner (backend#1166 / #1198 — the send side +of #378). + +Ties the backfill pieces together into a one-time, per-client sweep of datasets +ingested BEFORE the federated-alignment cutover. For each dataset it: + + 1. **GETs the backend record** by ``ingestor_id`` + (:meth:`APIClient.get_dataset_metadata`). The authoritative ``category`` and + ``data_format`` drive the recompute and are not recoverable from the table + alone, so they are read from the backend first. + 2. **Recomputes** the enriched ``{schema, meta_data}`` from the persisted rows + via :func:`build_dataset_metadata` — SQL aggregates, no row re-ingest — + keyed on that category. + 3. **POSTs** it back (:meth:`APIClient.send_metadata_backfill`), which upserts + ``GlobalMetaData`` and re-folds any competition built from the table. + +The set of datasets to sweep is the client's REGISTERED ingest journal +(:meth:`Database.list_registered_runs`); explicit ``ingestor_id``\\ s can be +passed instead. + +**Rollout, not always-on.** New ingests already emit the new-shape metadata +(#361), so this runs once per client during rollout. It is idempotent: a dataset +already carrying new-shape metadata is skipped, and the backend upsert is itself +a safe overwrite, so a re-run converges. + +**Scope.** Only plain/source datasets are recomputed. Competition/merged +datasets have no client-side raw table to read — their fold is rebuilt +backend-side by the re-fold this POST triggers — so they are skipped here +(detected via the backend record's ``is_competition`` / ``source_dataset_ids``). +``label_column`` and the uploader-declared scalar attributes are not persisted +backend-side, so they are not passed; the SQL-derivable enriched schema + +``feature_stats`` (the backfill's primary value) are recomputed regardless. See +:func:`build_dataset_metadata` for the per-fact limitations. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import Any, Dict, List, Optional + +from .api.client import APIClient +from .config import Config +from .database import Database +from .metadata_backfill import build_dataset_metadata +from .utils.logging import setup_logging + +logger = logging.getLogger(__name__) + + +# Outcome statuses for a single dataset's backfill attempt. +STATUS_OK = "ok" +STATUS_NOT_FOUND = "not_found" +STATUS_SKIPPED_COMPETITION = "skipped_competition" +STATUS_SKIPPED_CURRENT = "skipped_current" +STATUS_ERROR = "error" + + +@dataclass +class BackfillResult: + """The outcome of backfilling one dataset (see the ``STATUS_*`` constants).""" + + ingestor_id: str + table_name: Optional[str] = None + status: str = STATUS_OK + created: Optional[bool] = None + competitions_refolded: Optional[int] = None + error: Optional[str] = None + + +def _already_new_shape(record: Dict[str, Any]) -> bool: + """Whether the backend record already carries new-shape metadata, so the + recompute + POST can be skipped. + + Marker: a non-empty ``meta_data.attributes`` — the typed alignment contract + that only the post-cutover ingest (#361) / a prior backfill writes. A + pre-cutover row has no ``attributes``, so it is not skipped. Conservative: if + a category legitimately has empty attributes we simply re-POST (idempotent). + """ + meta_data = record.get("meta_data") or {} + return bool(meta_data.get("attributes")) + + +def backfill_dataset( + database: Database, + api_client: APIClient, + ingestor_id: str, + *, + skip_if_current: bool = True, +) -> BackfillResult: + """GET → recompute → POST for a single dataset. Never raises for an expected + skip/not-found; those are returned as a :class:`BackfillResult` status. + Unexpected failures (SQL / network) propagate to the caller. + """ + record = api_client.get_dataset_metadata(ingestor_id) + if record is None: + return BackfillResult(ingestor_id, status=STATUS_NOT_FOUND) + + table_name = record.get("table_name") + + # Competition/merged datasets have no client-side raw table for the SQL + # recompute; their reconciled fold is rebuilt backend-side (the re-fold this + # backfill triggers on source POSTs), so never recompute them here. + if record.get("is_competition") or record.get("source_dataset_ids"): + return BackfillResult( + ingestor_id, table_name, status=STATUS_SKIPPED_COMPETITION + ) + + if skip_if_current and _already_new_shape(record): + return BackfillResult(ingestor_id, table_name, status=STATUS_SKIPPED_CURRENT) + + category = record.get("category") + if not category or not table_name: + return BackfillResult( + ingestor_id, + table_name, + status=STATUS_ERROR, + error="backend record missing category/table_name", + ) + + payload = build_dataset_metadata( + database, + table_name, + category=category, + data_format=record.get("data_format"), + ) + result = api_client.send_metadata_backfill( + table_name, payload["schema"], payload.get("meta_data") + ) + return BackfillResult( + ingestor_id, + table_name, + status=STATUS_OK, + created=result.get("created"), + competitions_refolded=result.get("competitions_refolded"), + ) + + +def backfill_datasets( + database: Database, + api_client: APIClient, + ingestor_ids: Optional[List[str]] = None, + *, + skip_if_current: bool = True, +) -> List[BackfillResult]: + """Backfill every dataset in ``ingestor_ids``, or — when ``None`` — every + REGISTERED run in this client's ingest journal. + + Per-dataset guarded: an unexpected failure on one dataset is logged and + recorded as an ``error`` result, and the sweep continues to the next (a + single bad table can't abort the rollout). Returns one + :class:`BackfillResult` per dataset. + """ + if ingestor_ids is None: + ingestor_ids = [run["ingestor_id"] for run in database.list_registered_runs()] + + logger.info("Metadata backfill: %d dataset(s) to process.", len(ingestor_ids)) + + results: List[BackfillResult] = [] + for ingestor_id in ingestor_ids: + try: + result = backfill_dataset( + database, api_client, ingestor_id, skip_if_current=skip_if_current + ) + except Exception as exc: # noqa: BLE001 — one bad table must not abort the sweep + # Record/log the exception TYPE only — never str(exc) or a traceback. + # A SQL driver error or backend response body can embed customer cell + # values (e.g. categorical vocab), and this runs in a Helm-hook Job + # whose output lands in install logs. (bugbot) + error_type = type(exc).__name__ + logger.error( + "Metadata backfill failed for ingestor_id=%s (%s)", + ingestor_id, + error_type, + ) + result = BackfillResult( + ingestor_id, status=STATUS_ERROR, error=error_type + ) + results.append(result) + _log_result(result) + + _log_summary(results) + return results + + +def _log_result(result: BackfillResult) -> None: + if result.status == STATUS_OK: + logger.info( + " ✓ %s (%s): created=%s, competitions_refolded=%s", + result.table_name, + result.ingestor_id, + result.created, + result.competitions_refolded, + ) + elif result.status == STATUS_ERROR: + logger.error( + " ✗ %s (%s): %s", result.table_name, result.ingestor_id, result.error + ) + else: + logger.info( + " – %s (%s): %s", result.table_name, result.ingestor_id, result.status + ) + + +def _log_summary(results: List[BackfillResult]) -> None: + counts: Dict[str, int] = {} + for result in results: + counts[result.status] = counts.get(result.status, 0) + 1 + logger.info( + "Metadata backfill complete: %s", + ", ".join(f"{status}={count}" for status, count in sorted(counts.items())) + or "nothing to do", + ) + + +def main(argv: Optional[List[str]] = None) -> int: # pragma: no cover - thin shell + """Console-script entrypoint (``tracebloc-backfill``). + + Reads DB + backend credentials from the environment via :class:`Config` (the + same env the ingestor uses — ``MYSQL_HOST``, ``BACKEND_TOKEN``, ``EDGE_ENV``, + …); no ``ingest.yaml`` is needed. Returns a non-zero exit code iff any + dataset errored, so a rollout job surfaces failures. + """ + config = Config() + setup_logging(config) + + database = Database(config) + api_client = APIClient(config) # triggers config.validate() + + results = backfill_datasets(database, api_client) + errored = sum(1 for result in results if result.status == STATUS_ERROR) + return 1 if errored else 0 From 9b0f4395f82eb720843d37296d94f97d87b6da6e Mon Sep 17 00:00:00 2001 From: Divya Date: Fri, 24 Jul 2026 15:20:43 +0530 Subject: [PATCH 3/5] chore(release): bump version to 0.7.6 (#394) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * chore(release): bump version to 0.7.6 Co-Authored-By: Claude Opus 4.8 * fix(image-validator): auto-detect from first readable image, not files[0] test_too_small_result_still_reports_corrupt_files was order-dependent: auto-detection read only image_files[0], so when a corrupt file came first in glob order (deterministic on the Linux CI, not on macOS) it returned None, expected_resolution stayed unset, and _validate_image_resolutions bailed with "auto-detection failed" before the min-size floor check could run — masking the real verdict and making the outcome depend on filesystem ordering. Scan for the first *readable* image instead; if none are readable, expected_resolution stays unset and the honest "auto-detection failed" still surfaces. Adds an order-independence guard test that forces the corrupt file first (fails pre-fix, passes post-fix). Co-Authored-By: Claude Opus 4.8 --------- Co-authored-by: Claude Opus 4.8 Co-authored-by: shujaat --- tests/test_image_validator_min_size.py | 31 +++++++++++++++++++ tracebloc_ingestor/__init__.py | 2 +- .../validators/image_validator.py | 24 +++++++++----- 3 files changed, 49 insertions(+), 8 deletions(-) diff --git a/tests/test_image_validator_min_size.py b/tests/test_image_validator_min_size.py index 67d9df9..3427a74 100644 --- a/tests/test_image_validator_min_size.py +++ b/tests/test_image_validator_min_size.py @@ -196,6 +196,37 @@ def test_too_small_result_still_reports_corrupt_files(clean_env, images_dir): assert any("broken.jpg" in f for f in result.metadata["invalid_files"]) +def test_floor_wins_when_corrupt_file_is_first(clean_env, images_dir, monkeypatch): + """Order-independence guard: auto-detection must skip an unreadable first + file and detect from the first *readable* image. Before the fix, when the + corrupt file happened to come first in glob order, auto-detection gave up + and the result was 'auto-detection failed' instead of the floor verdict — + so the outcome depended on filesystem ordering (green on macOS, red on the + Linux CI). This forces the corrupt-first order explicitly.""" + src, add = images_dir + add("tiny.jpg", (8, 8)) + (src / "images" / "broken.jpg").write_bytes(b"not a real jpeg") + clean_env.setenv("SRC_PATH", str(src)) + + validator = ImageResolutionValidator() + real_get_files = validator._get_image_files + # Deterministically place the corrupt file first, whatever glob returns. + monkeypatch.setattr( + validator, + "_get_image_files", + lambda *a, **k: sorted( + real_get_files(*a, **k), + key=lambda p: 0 if p.name == "broken.jpg" else 1, + ), + ) + + result = validator.validate(None) + assert not result.is_valid + assert "below the minimum size" in result.errors[0] + assert "tiny.jpg" in result.errors[0] + assert any("broken.jpg" in f for f in result.metadata["invalid_files"]) + + # --- _meets_min_size unit ---------------------------------------------------- diff --git a/tracebloc_ingestor/__init__.py b/tracebloc_ingestor/__init__.py index 232a199..06f930a 100644 --- a/tracebloc_ingestor/__init__.py +++ b/tracebloc_ingestor/__init__.py @@ -28,7 +28,7 @@ # Single source of truth for the package version. setup.py parses this literal # (see _read_version in setup.py) so the two can't drift again (#175). Bump here # only — setup.py picks it up automatically. -__version__ = "0.7.5" +__version__ = "0.7.6" __all__ = [ "Config", diff --git a/tracebloc_ingestor/validators/image_validator.py b/tracebloc_ingestor/validators/image_validator.py index 2017a75..567e49d 100644 --- a/tracebloc_ingestor/validators/image_validator.py +++ b/tracebloc_ingestor/validators/image_validator.py @@ -172,14 +172,24 @@ def validate(self, path: Any, **kwargs) -> ValidationResult: metadata={"files_checked": 0}, ) - # Auto-detect resolution from first image if not specified + # Auto-detect resolution from the first *readable* image if not + # specified. Using image_files[0] blindly made auto-detection fail + # whenever the first file in (arbitrary) glob order happened to be + # corrupt/unreadable — which masked the real floor/uniformity + # verdict behind "auto-detection failed" and made the outcome + # depend on filesystem ordering. Skip unreadable files and detect + # from the first one that yields a resolution; if none do, leave + # expected_resolution unset and let _validate_image_resolutions + # report the honest "auto-detection failed". if auto_detect_resolution and not self.expected_resolution: - first_image_resolution = self._get_image_resolution(image_files[0]) - if first_image_resolution: - self.expected_resolution = first_image_resolution - logger.info( - f"Auto-detected expected resolution: {self.expected_resolution}" - ) + for candidate in image_files: + candidate_resolution = self._get_image_resolution(candidate) + if candidate_resolution: + self.expected_resolution = candidate_resolution + logger.info( + f"Auto-detected expected resolution: {self.expected_resolution}" + ) + break # Validate image resolutions return self._validate_image_resolutions(image_files) From 382b741c3e4242798afda4df459bbaf33f24e461 Mon Sep 17 00:00:00 2001 From: Divya Date: Fri, 24 Jul 2026 16:10:42 +0530 Subject: [PATCH 4/5] feat(backfill): discover dataset tables directly, not just the runs journal (#395) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(backfill): discover dataset tables directly, not just the runs journal The runner enumerated the runs journal (list_registered_runs), which only has entries for datasets ingested by the post-journal ingestor. The pre-cutover backlog the backfill exists for was ingested BEFORE the journal, so it had no entry and was silently swept 0 (verified on dev: empty journal, ~170 real dataset tables untouched). Enumerate dataset tables directly instead: - Database.list_dataset_ingestor_ids() — scan information_schema for tables carrying the framework `ingestor_id` column (excluding the run journal + salt store), then read each table's distinct non-null ingestor_ids. Covers journalled AND pre-journal datasets (the journal is a strict subset). - backfill_datasets() defaults to this discovery. A table appended by several ingest runs has multiple ingestor_ids; metadata is table-level, so a table is backfilled once — the first id the backend resolves wins, siblings are skipped (a not_found/error id doesn't mark the table done, so another id can resolve it). Explicit ingestor_ids still supported. Tests: table discovery is the default; framework tables excluded; multi-id table backfilled once; sibling id retried after a not_found. Co-Authored-By: Claude Opus 4.8 * chore(release): bump version to 0.7.7 Bundled with the table-discovery change so a single merge + v0.7.7 tag yields a deployable image carrying it. Co-Authored-By: Claude Opus 4.8 * fix(backfill): guard per-table discovery so one bad table can't abort the sweep (Bugbot #395) list_dataset_ingestor_ids scanned every dataset table with no per-table guard, so a single failing SELECT DISTINCT (scan timeout, table dropped mid-sweep, perms) aborted enumeration and skipped ALL tables. Wrap the per-table query in try/except: log the table name + exception TYPE only (never str(exc) — a driver message can embed cell values and feeds install logs) and continue. Co-Authored-By: Claude Opus 4.8 --------- Co-authored-by: Claude Opus 4.8 --- tests/test_database.py | 62 +++++++++++++++++++ tests/test_metadata_backfill_runner.py | 62 +++++++++++++++++-- tracebloc_ingestor/__init__.py | 2 +- tracebloc_ingestor/database.py | 62 +++++++++++++++++++ .../metadata_backfill_runner.py | 56 +++++++++++++---- 5 files changed, 226 insertions(+), 18 deletions(-) diff --git a/tests/test_database.py b/tests/test_database.py index 93167df..60a2cbe 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -1029,3 +1029,65 @@ def test_list_registered_runs_returns_registered_only(db, mock_engine_factory): s for s in _executed_sql(conn) if "SELECT ingestor_id, table_name, task" in s ) assert "WHERE registered = 1" in select + + +def test_list_dataset_ingestor_ids_scans_tables_excluding_framework( + db, mock_engine_factory +): + """Discovers dataset tables via information_schema (tables carrying an + ingestor_id column), skips the framework tables (run journal + salt store), + and returns each table's distinct non-null ingestor_id.""" + _, _, conn = mock_engine_factory + + def _result(rows): + r = MagicMock() + r.fetchall.return_value = rows + return r + + # 1st execute: information_schema table list (includes a framework table). + # then one execute per NON-framework table, in returned order. + conn.execute.side_effect = [ + _result([("ds_one",), ("tracebloc_ingest_runs",), ("ds_two",)]), + _result([("i1",), ("i2",)]), # ds_one distinct ids + _result([("i3",)]), # ds_two distinct ids + ] + + result = db.list_dataset_ingestor_ids() + assert result == [ + {"ingestor_id": "i1", "table_name": "ds_one"}, + {"ingestor_id": "i2", "table_name": "ds_one"}, + {"ingestor_id": "i3", "table_name": "ds_two"}, + ] + + executed = _executed_sql(conn) + # Discovery query hit information_schema for the ingestor_id column. + assert any("information_schema.columns" in s and "ingestor_id" in s for s in executed) + # The framework run-journal table was skipped — never SELECTed as a dataset. + assert not any("`tracebloc_ingest_runs`" in s for s in executed) + # Per-dataset-table id scans excluded null/empty ids. + assert any("FROM `ds_one`" in s and "IS NOT NULL" in s for s in executed) + + +def test_list_dataset_ingestor_ids_skips_a_table_that_errors(db, mock_engine_factory): + """A single unreadable table (scan timeout / dropped mid-sweep / perms) is + skipped and discovery continues — the sweep's 'one bad table can't abort the + rollout' guarantee must hold at enumeration time too. (bugbot)""" + _, _, conn = mock_engine_factory + + def _result(rows): + r = MagicMock() + r.fetchall.return_value = rows + return r + + conn.execute.side_effect = [ + _result([("ds_one",), ("ds_bad",), ("ds_two",)]), # information_schema + _result([("i1",)]), # ds_one + RuntimeError("scan timeout"), # ds_bad — per-table query fails + _result([("i2",)]), # ds_two still scanned + ] + + result = db.list_dataset_ingestor_ids() + assert result == [ + {"ingestor_id": "i1", "table_name": "ds_one"}, + {"ingestor_id": "i2", "table_name": "ds_two"}, + ] diff --git a/tests/test_metadata_backfill_runner.py b/tests/test_metadata_backfill_runner.py index 37e9c57..d797fef 100644 --- a/tests/test_metadata_backfill_runner.py +++ b/tests/test_metadata_backfill_runner.py @@ -147,20 +147,70 @@ def test_backfill_dataset_error_when_record_missing_category(): # --------------------------------------------------------------------------- -def test_backfill_datasets_enumerates_registered_runs_when_ids_omitted(): +def test_backfill_datasets_discovers_tables_when_ids_omitted(): + # Default enumeration scans dataset tables (covers the pre-cutover backlog), + # NOT the runs journal. db = MagicMock() - db.list_registered_runs.return_value = [ - {"ingestor_id": "a", "table_name": "ta", "task": "tabular_classification"}, - {"ingestor_id": "b", "table_name": "tb", "task": "tabular_classification"}, + db.list_dataset_ingestor_ids.return_value = [ + {"ingestor_id": "a", "table_name": "ta"}, + {"ingestor_id": "b", "table_name": "tb"}, ] - api = _api(record=dict(_PLAIN_RECORD)) + api = MagicMock() + api.get_dataset_metadata.side_effect = lambda iid: dict( + _PLAIN_RECORD, ingestor_id=iid, table_name={"a": "ta", "b": "tb"}[iid] + ) + api.send_metadata_backfill.side_effect = lambda table, schema, meta=None: { + "table_name": table, + "created": True, + "competitions_refolded": 0, + } with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD): results = backfill_datasets(db, api) - db.list_registered_runs.assert_called_once() + db.list_dataset_ingestor_ids.assert_called_once() + db.list_registered_runs.assert_not_called() assert [r.ingestor_id for r in results] == ["a", "b"] assert all(r.status == STATUS_OK for r in results) +def test_backfill_datasets_backfills_a_multi_id_table_once(): + # A table appended by several ingest runs has multiple ingestor_ids; its + # metadata is table-level, so it is backfilled ONCE and the sibling id is + # skipped (before any backend call). + db = MagicMock() + db.list_dataset_ingestor_ids.return_value = [ + {"ingestor_id": "run1", "table_name": "shared"}, + {"ingestor_id": "run2", "table_name": "shared"}, + ] + api = _api(record=dict(_PLAIN_RECORD, table_name="shared")) + with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD): + results = backfill_datasets(db, api) + assert len(results) == 1 # second id skipped by per-table dedup + assert results[0].table_name == "shared" + api.send_metadata_backfill.assert_called_once() + + +def test_backfill_datasets_retries_sibling_id_when_first_not_found(): + # If a table's first ingestor_id 404s, a sibling id is still tried (the table + # isn't marked done on not_found). + db = MagicMock() + db.list_dataset_ingestor_ids.return_value = [ + {"ingestor_id": "missing", "table_name": "shared"}, + {"ingestor_id": "good", "table_name": "shared"}, + ] + api = MagicMock() + api.get_dataset_metadata.side_effect = lambda iid: ( + None if iid == "missing" else dict(_PLAIN_RECORD, table_name="shared") + ) + api.send_metadata_backfill.return_value = { + "table_name": "shared", + "created": True, + "competitions_refolded": 0, + } + with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD): + results = backfill_datasets(db, api) + assert [r.status for r in results] == [STATUS_NOT_FOUND, STATUS_OK] + + def test_backfill_datasets_continues_past_a_failing_dataset(): api = MagicMock() # First GET raises (e.g. transient), second succeeds. diff --git a/tracebloc_ingestor/__init__.py b/tracebloc_ingestor/__init__.py index 06f930a..7308003 100644 --- a/tracebloc_ingestor/__init__.py +++ b/tracebloc_ingestor/__init__.py @@ -28,7 +28,7 @@ # Single source of truth for the package version. setup.py parses this literal # (see _read_version in setup.py) so the two can't drift again (#175). Bump here # only — setup.py picks it up automatically. -__version__ = "0.7.6" +__version__ = "0.7.7" __all__ = [ "Config", diff --git a/tracebloc_ingestor/database.py b/tracebloc_ingestor/database.py index 0481d7c..14f6462 100644 --- a/tracebloc_ingestor/database.py +++ b/tracebloc_ingestor/database.py @@ -890,6 +890,68 @@ def list_registered_runs(self) -> List[Dict[str, str]]: for (iid, tname, task) in rows ] + def list_dataset_ingestor_ids(self) -> List[Dict[str, str]]: + """Discover every dataset table directly and return its distinct + ``ingestor_id``\\ s: ``[{"ingestor_id", "table_name"}, ...]``. + + This is the enumeration the metadata backfill sweeps by default, in + preference to :meth:`list_registered_runs`. The pre-cutover backlog the + backfill exists for was ingested BEFORE the run journal existed, so those + datasets have no journal row — but their tables still carry the framework + ``ingestor_id`` column. Scanning for that column finds them; the journal + would miss them entirely. + + A dataset table is any table in this schema with an ``ingestor_id`` + column, minus the framework bookkeeping tables (the run journal — which + also has the column — and the salt store). Only non-null, non-empty + ``ingestor_id`` values are returned; a table appended by several runs + yields one pair per distinct id (the caller backfills the table once). + """ + framework_tables = {self.SALT_TABLE, self.RUNS_TABLE} + with self.engine.connect() as connection: + table_rows = _execute_with_retry( + connection, + text( + "SELECT DISTINCT table_name FROM information_schema.columns " + "WHERE table_schema = DATABASE() AND column_name = 'ingestor_id'" + ), + ).fetchall() + + pairs: List[Dict[str, str]] = [] + for (table_name,) in table_rows: + if table_name in framework_tables: + continue + # Identifier interpolated (bound params can't name a table); the name + # comes from information_schema on our own DB, and backticks are + # escaped, so it is not attacker-controlled free text. + safe_table = table_name.replace("`", "``") + # Per-table guard: a single unreadable table (timeout on a large + # unindexed scan, a table dropped mid-sweep, a permissions gap) must + # NOT abort discovery — otherwise every other table is skipped and the + # sweep's "one bad table can't stop the rollout" guarantee is lost. + # Log the table name + exception TYPE only (never str(exc): a driver + # message can embed cell values, and this feeds install logs). + try: + with self.engine.connect() as connection: + id_rows = _execute_with_retry( + connection, + text( + f"SELECT DISTINCT ingestor_id FROM `{safe_table}` " + f"WHERE ingestor_id IS NOT NULL AND ingestor_id <> ''" + ), + ).fetchall() + except Exception as exc: # noqa: BLE001 — skip the table, continue the sweep + logger.warning( + "list_dataset_ingestor_ids: skipping table %s — could not read " + "ingestor_ids (%s)", + table_name, + type(exc).__name__, + ) + continue + for (ingestor_id,) in id_rows: + pairs.append({"ingestor_id": ingestor_id, "table_name": table_name}) + return pairs + def get_label_counts(self, table_name: str, ingestor_id: str) -> Dict[str, int]: """ Return ``{label: row_count}`` for every label inserted by *ingestor_id*. diff --git a/tracebloc_ingestor/metadata_backfill_runner.py b/tracebloc_ingestor/metadata_backfill_runner.py index 5bbf105..a809558 100644 --- a/tracebloc_ingestor/metadata_backfill_runner.py +++ b/tracebloc_ingestor/metadata_backfill_runner.py @@ -143,21 +143,46 @@ def backfill_datasets( *, skip_if_current: bool = True, ) -> List[BackfillResult]: - """Backfill every dataset in ``ingestor_ids``, or — when ``None`` — every - REGISTERED run in this client's ingest journal. - - Per-dataset guarded: an unexpected failure on one dataset is logged and - recorded as an ``error`` result, and the sweep continues to the next (a - single bad table can't abort the rollout). Returns one - :class:`BackfillResult` per dataset. + """Backfill datasets. With an explicit ``ingestor_ids`` list, one attempt per + id; otherwise **discover every dataset table** in the client DB and backfill + each — see :meth:`Database.list_dataset_ingestor_ids`. + + Table discovery (not the runs journal) is the default because the whole point + of the backfill is the PRE-CUTOVER backlog — datasets ingested before the + runs journal existed, so they have no journal entry — yet their tables still + carry the framework ``ingestor_id`` column. Enumerating tables therefore + covers both journalled and pre-journal datasets; the journal is a strict + subset. + + A table can carry several ingestor_ids (multiple ingest runs into one table); + its metadata is table-level, so a table is backfilled once — the first of its + ids the backend resolves wins and the rest are skipped. + + Per-dataset guarded: an unexpected failure on one dataset is logged (type + only) and recorded as an ``error`` result; the sweep continues (a single bad + table can't abort the rollout). Returns one :class:`BackfillResult` per + attempt made. """ - if ingestor_ids is None: - ingestor_ids = [run["ingestor_id"] for run in database.list_registered_runs()] + if ingestor_ids is not None: + # Caller-supplied ids: one attempt each; the table is learned from the + # backend GET inside backfill_dataset. + items = [(ingestor_id, None) for ingestor_id in ingestor_ids] + else: + items = [ + (row["ingestor_id"], row["table_name"]) + for row in database.list_dataset_ingestor_ids() + ] - logger.info("Metadata backfill: %d dataset(s) to process.", len(ingestor_ids)) + logger.info("Metadata backfill: %d ingestor_id(s) to process.", len(items)) results: List[BackfillResult] = [] - for ingestor_id in ingestor_ids: + done_tables: set = set() + for ingestor_id, table_name in items: + # A table's metadata is backfilled once. Skip the remaining ids of a + # table already resolved — checked by the ENUMERATED name (no backend + # call) so multi-run tables don't re-POST. + if table_name is not None and table_name in done_tables: + continue try: result = backfill_dataset( database, api_client, ingestor_id, skip_if_current=skip_if_current @@ -178,6 +203,15 @@ def backfill_datasets( ) results.append(result) _log_result(result) + # Mark the table done only on a DEFINITIVE backend resolution (backfilled, + # already-current, or a competition we intentionally skip). not_found / + # error leave it open so a sibling ingestor_id can still resolve it. + if ( + result.status + in (STATUS_OK, STATUS_SKIPPED_CURRENT, STATUS_SKIPPED_COMPETITION) + and result.table_name + ): + done_tables.add(result.table_name) _log_summary(results) return results From 0a6bed20082b72e0f5eb1c102cfd78fe1df66e6d Mon Sep 17 00:00:00 2001 From: shujaat_tracebloc <153823837+shujaatTracebloc@users.noreply.github.com> Date: Fri, 24 Jul 2026 15:13:11 +0200 Subject: [PATCH 5/5] fix: normalize validator log labels (Bugbot, promotion PR #396) (#397) Append "Validator" only when validator.name doesn't already end in it, so names like "Ingestable Records" keep the suffix and names like "Data Validator" don't double it. - tracebloc_ingestor/ingestors/base.py Co-authored-by: Claude Opus 4.8 --- tests/test_ingestor_base.py | 64 ++++++++++++++++++++++++++++ tracebloc_ingestor/ingestors/base.py | 32 ++++++++++---- 2 files changed, 88 insertions(+), 8 deletions(-) diff --git a/tests/test_ingestor_base.py b/tests/test_ingestor_base.py index cac2254..6b79df5 100644 --- a/tests/test_ingestor_base.py +++ b/tests/test_ingestor_base.py @@ -610,6 +610,70 @@ def test_validate_data_validator_exception_raises(): ing.validate_data("src") +# --------------------------------------------------------------------------- +# validator label normalization (#396 Bugbot: log labels drop suffix) +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "name, expected", + [ + # Already ends in "Validator" — must NOT double it. + ("Data Validator", "Data Validator"), + # Case-insensitive on the existing suffix. + ("Pascal VOC XML validator", "Pascal VOC XML validator"), + # Missing the suffix — must gain it. + ("Ingestable Records", "Ingestable Records Validator"), + ("Text Content", "Text Content Validator"), + ("BIO Label", "BIO Label Validator"), + ("Keypoint Annotation", "Keypoint Annotation Validator"), + # Surrounding whitespace is stripped before deciding. + (" Data Validator ", "Data Validator"), + ], +) +def test_validator_label_normalizes(name, expected): + assert base_mod._validator_label(name) == expected + + +def test_validate_data_success_line_keeps_suffix_when_missing(capsys): + # A name that does NOT end in "Validator" (e.g. "Ingestable Records") must + # keep the suffix in its "successfully passed" line. + ing = make_ingestor(category=None) + v = MagicMock() + v.name = "Ingestable Records" + v.validate.return_value = ValidationResult(True, [], [], {}) + with patch.object(base_mod, "map_validators", return_value=[v]): + assert ing.validate_data("src") is True + out = capsys.readouterr().out + assert "Ingestable Records Validator successfully passed" in out + + +def test_validate_data_success_line_does_not_double_suffix(capsys): + # A name that already ends in "Validator" (e.g. "Data Validator") must not + # double the word. + ing = make_ingestor(category=None) + v = MagicMock() + v.name = "Data Validator" + v.validate.return_value = ValidationResult(True, [], [], {}) + with patch.object(base_mod, "map_validators", return_value=[v]): + assert ing.validate_data("src") is True + out = capsys.readouterr().out + assert "Data Validator successfully passed" in out + assert "Data Validator Validator" not in out + + +def test_validate_data_failure_message_keeps_suffix_when_missing(): + # The error-append line normalizes too, so a suffix-less name isn't stripped. + ing = make_ingestor(category=None) + v = MagicMock() + v.name = "Ingestable Records" + v.validate.return_value = ValidationResult(False, ["nope"], [], {}) + with patch.object(base_mod, "map_validators", return_value=[v]): + with pytest.raises(ValueError) as exc: + ing.validate_data("src") + assert "Ingestable Records Validator failed" in str(exc.value) + + # --------------------------------------------------------------------------- # ingest (full flow, Session patched) # --------------------------------------------------------------------------- diff --git a/tracebloc_ingestor/ingestors/base.py b/tracebloc_ingestor/ingestors/base.py index 73c03d6..8555de3 100644 --- a/tracebloc_ingestor/ingestors/base.py +++ b/tracebloc_ingestor/ingestors/base.py @@ -137,6 +137,20 @@ def _rows_state_clause(inserted_records: int) -> str: return "no rows were ingested, so nothing was left in the database" +def _validator_label(name: str) -> str: + """Human-readable label for a validator's preflight log/error lines. + + Validator ``.name`` values are inconsistent: some already end in + "Validator" (e.g. "Data Validator") while many do not (e.g. "Ingestable + Records", "Text Content", "BIO Label", "Keypoint Annotation"). Normalize at + the log site so every label reads " Validator" exactly once — + appending the word only when it's missing, so suffixed names don't double + ("Data Validator Validator") and unsuffixed names don't drop it. + """ + name = (name or "").strip() + return name if name.lower().endswith("validator") else f"{name} Validator" + + # NOTE: _TABULAR_FAMILY_CATEGORIES and _FILE_BEARING_CATEGORIES are imported # above from modalities.registry (derived from the per-category ModalitySpec # flags — backend#796 P3a). The ``self.category in `` checks throughout @@ -545,14 +559,15 @@ def validate_data(self, source: Any) -> bool: validation_errors = [] for validator in validators: + label = _validator_label(validator.name) try: - logger.info(f"{CYAN}Running validator: {validator.name}{RESET}") + logger.info(f"{CYAN}Running validator: {label}{RESET}") result = validator.validate(source) if not result.is_valid: all_valid = False validation_errors.append( - f"{BOLD}{validator.name} failed: {RESET} \n {RED}" + f"{BOLD}{label} failed: {RESET} \n {RED}" ) validation_errors.extend(result.errors) validation_errors.append(f"{RESET}") @@ -560,16 +575,17 @@ def validate_data(self, source: Any) -> bool: # Log warnings if any for warning in result.warnings: logger.warning( - f"{YELLOW}Validation warning - {validator.name}: {warning}{RESET}" + f"{YELLOW}Validation warning - {label}: {warning}{RESET}" ) if result.is_valid: - # validator.name already ends in "Validator" (e.g. "Data - # Validator"), so don't append the word again — that produced - # "Data Validator Validator successfully passed". - print(f"{GREEN}{validator.name} successfully passed{RESET}") + # Normalize the label so it always reads " Validator" + # exactly once: names that already end in "Validator" (e.g. + # "Data Validator") don't double, and names that don't (e.g. + # "Ingestable Records") keep the suffix. + print(f"{GREEN}{label} successfully passed{RESET}") except Exception as e: all_valid = False - validation_errors.append(f"{validator.name} error: {str(e)}") + validation_errors.append(f"{label} error: {str(e)}") if not all_valid: error_summary = "\n".join(validation_errors)