Release develop → master

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=[

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": {}}}

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

tests/test_database.py

@@ -1004,3 +1004,90 @@ 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
+
+
+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"},
+    ]

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 ----------------------------------------------------