From 193ea59fadec708e753b5daa33ce887f66ad9115 Mon Sep 17 00:00:00 2001 From: JuliaEdom Date: Sun, 5 Jul 2026 23:31:09 +0300 Subject: [PATCH 1/4] =?UTF-8?q?feat(g2):=20S2b=20=E2=80=94=20remove=20X5?= =?UTF-8?q?=20Retail=20Hero=20loader=20package,=20de-brand=20spec.yaml?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Delete warehouse/agentflow/dv2/loaders/x5_retail_hero/ entirely (no successor generator, per the S2a benchmark-fate decision) and its dedicated test tests/unit/test_x5_retail_hero_loader.py. Relocate the vault-generic row models (VaultRow, Hub*, Link*, SatOrderHeader, SatOrderPricing, SatLinkOrderProduct) from the deleted loader's schemas.py into a neutral warehouse/agentflow/dv2/loaders/vault_rows.py, since PostgresVaultWriter's own tests (test_dv2_postgres_ingestion.py) are the only coverage for that class and must keep running. Surgical edits: - test_dv2_postgres_ingestion.py: re-point fixtures at vault_rows, drop the loader-sink tests (_open_sink/_DryRunSink/_PostgresSink/_ClickHouseSink, which die with the loader). - test_dv2_supplier_reference.py: replace the X5-loader hash-equality pin with self-contained known-vector digests for vault_mapping's own MD5 canonicalisation. - test_dv2_business_vault_ddl.py / test_dv2_postgres_ddl.py: drop the negative "x5__ not in raw" regression guard (the string itself violates the no-X5-anywhere mandate); keep the positive "mp__ in raw" check. De-brand spec.yaml: "1c (X5 Retail Hero)" -> "1c (branch order feed)" across all 5 sat_order_header__1c__* entries, then regenerate the 10 dependent satellite DDL files (5 ClickHouse + 5 PostgreSQL) via generate_satellites.py so the generated header comments come out de-X5'd too. Part of the G2 audit follow-up (see x5-benchmark-decision.md for the full disposition list; deleted in this step's final commit). --- tests/unit/test_dv2_business_vault_ddl.py | 9 +- tests/unit/test_dv2_postgres_ddl.py | 9 +- tests/unit/test_dv2_postgres_ingestion.py | 110 ++--- tests/unit/test_dv2_supplier_reference.py | 40 +- tests/unit/test_x5_retail_hero_loader.py | 99 ---- .../schemas.py => vault_rows.py} | 42 +- .../dv2/loaders/x5_retail_hero/README.md | 67 --- .../dv2/loaders/x5_retail_hero/__init__.py | 3 - .../x5_retail_hero/branch_distributor.py | 61 --- .../dv2/loaders/x5_retail_hero/loader.py | 453 ------------------ .../dv2/loaders/x5_retail_hero/mappers.py | 410 ---------------- .../loaders/x5_retail_hero/requirements.txt | 6 - .../satellites/sat_order_header__1c__ala.sql | 4 +- .../satellites/sat_order_header__1c__dxb.sql | 4 +- .../satellites/sat_order_header__1c__ekb.sql | 4 +- .../satellites/sat_order_header__1c__msk.sql | 4 +- .../satellites/sat_order_header__1c__spb.sql | 4 +- .../satellites/sat_order_header__1c__ala.sql | 4 +- .../satellites/sat_order_header__1c__dxb.sql | 4 +- .../satellites/sat_order_header__1c__ekb.sql | 4 +- .../satellites/sat_order_header__1c__msk.sql | 4 +- .../satellites/sat_order_header__1c__spb.sql | 4 +- warehouse/agentflow/dv2/spec.yaml | 20 +- 23 files changed, 110 insertions(+), 1259 deletions(-) delete mode 100644 tests/unit/test_x5_retail_hero_loader.py rename warehouse/agentflow/dv2/loaders/{x5_retail_hero/schemas.py => vault_rows.py} (69%) delete mode 100644 warehouse/agentflow/dv2/loaders/x5_retail_hero/README.md delete mode 100644 warehouse/agentflow/dv2/loaders/x5_retail_hero/__init__.py delete mode 100644 warehouse/agentflow/dv2/loaders/x5_retail_hero/branch_distributor.py delete mode 100644 warehouse/agentflow/dv2/loaders/x5_retail_hero/loader.py delete mode 100644 warehouse/agentflow/dv2/loaders/x5_retail_hero/mappers.py delete mode 100644 warehouse/agentflow/dv2/loaders/x5_retail_hero/requirements.txt diff --git a/tests/unit/test_dv2_business_vault_ddl.py b/tests/unit/test_dv2_business_vault_ddl.py index de0ed367..0b1ab306 100644 --- a/tests/unit/test_dv2_business_vault_ddl.py +++ b/tests/unit/test_dv2_business_vault_ddl.py @@ -82,13 +82,10 @@ def test_customer_mdm_views_admit_all_source_conventions(): f"hard-coded record_source = '1c__' filter in " f"bv_customer_mdm__{branch} reintroduces audit #12" ) - # B2 (domain.md §5.4): the legend's marketplace-feed vocabulary replaces - # the Kaggle dataset name in the third-source-convention example (checked - # on the RAW text — the example lives in the header comment, which body + # domain.md §5.3/§5.4: the header comment's source-convention example + # must document the consolidated marketplace-feed prefix (checked on + # the RAW text — the example lives in the header comment, which body # strips). - assert "x5__" not in raw, ( - f"stale Kaggle-dataset record_source prefix x5__ leaked back into bv_customer_mdm__{branch}.sql" - ) assert "mp__" in raw, ( f"bv_customer_mdm__{branch}.sql should document the mp__ marketplace-feed convention (domain.md §5.3)" ) diff --git a/tests/unit/test_dv2_postgres_ddl.py b/tests/unit/test_dv2_postgres_ddl.py index 877056a5..9b00d6c7 100644 --- a/tests/unit/test_dv2_postgres_ddl.py +++ b/tests/unit/test_dv2_postgres_ddl.py @@ -163,12 +163,9 @@ def test_customer_mdm_views_admit_all_source_conventions(): assert "record_source = '1c__" not in body, ( "hard-coded record_source = '1c__' filter reintroduces audit #12" ) - # B2 (domain.md §5.4): the legend's marketplace-feed vocabulary replaces the - # Kaggle dataset name in the third-source-convention example (checked on the - # RAW text — the example lives in the header comment, which body strips). - assert "x5__" not in raw, ( - "stale Kaggle-dataset record_source prefix x5__ leaked back into 03_business_vault.sql" - ) + # domain.md §5.3/§5.4: the header comment's source-convention example must + # document the consolidated marketplace-feed prefix (checked on the RAW + # text — the example lives in the header comment, which body strips). assert "mp__" in raw, ( "03_business_vault.sql should document the mp__ marketplace-feed convention (domain.md §5.3)" ) diff --git a/tests/unit/test_dv2_postgres_ingestion.py b/tests/unit/test_dv2_postgres_ingestion.py index dfe313ce..742e4371 100644 --- a/tests/unit/test_dv2_postgres_ingestion.py +++ b/tests/unit/test_dv2_postgres_ingestion.py @@ -1,15 +1,14 @@ """Unit tests for the DV2 PostgreSQL ingestion repoint (no Docker). -The raw vault moved off ClickHouse onto PostgreSQL; these tests pin that the X5 -loader and the supplier reference can actually *feed* that PostgreSQL vault -without a live database: +The raw vault moved off ClickHouse onto PostgreSQL; these tests pin that the +per-branch order feed and the supplier reference can actually *feed* that +PostgreSQL vault without a live database: * ``build_insert_sql`` renders an idempotent, sqlglot-valid INSERT; -* every column the loaders insert exists in the committed PostgreSQL DDL (the +* every column the feeds insert exists in the committed PostgreSQL DDL (the guard that catches a model/DDL or generic-vs-entity-name drift); * ``PostgresVaultWriter`` streams rows through a fake DB-API connection in the - right column order and batches, with hash keys preserved as ``bytes``; -* the X5 loader selects the right sink per ``--target``. + right column order and batches, with hash keys preserved as ``bytes``. A live apply + ``bv_order_canonical`` query against real data is a separate single-node Mac smoke (see dv2/postgres/README.md), mirroring the Flink and @@ -22,21 +21,18 @@ from datetime import datetime from pathlib import Path -import click import pytest import sqlglot from click.testing import CliRunner from sqlglot import exp import warehouse.agentflow.dv2.loaders.pg_vault_writer as pgw -from warehouse.agentflow.dv2.loaders import pg_vault_writer +from warehouse.agentflow.dv2.loaders import pg_vault_writer, vault_rows from warehouse.agentflow.dv2.loaders.pg_vault_writer import ( PostgresVaultWriter, build_insert_sql, satellite_hash_key, ) -from warehouse.agentflow.dv2.loaders.x5_retail_hero import loader -from warehouse.agentflow.dv2.loaders.x5_retail_hero import schemas as x5 from warehouse.agentflow.dv2.reference import load_postgres from warehouse.agentflow.dv2.reference.generator import build_reference from warehouse.agentflow.dv2.reference.vault_mapping import VAULT_DB_COLUMNS, map_reference @@ -171,23 +167,31 @@ def test_build_insert_sql_satellite_requires_single_hash_key(): # --- inserted columns exist in the committed PostgreSQL DDL ------------------- -# Tables the X5 loader actually emits (mappers.py), with the row model whose +# Tables the per-branch order feed (1c) inserts into, with the row model whose # fields become the INSERT column list. -X5_TABLES = [ - ("hub_customer", "01_hubs.sql", x5.HubCustomer), - ("hub_product", "01_hubs.sql", x5.HubProduct), - ("hub_store", "01_hubs.sql", x5.HubStore), - ("hub_order", "01_hubs.sql", x5.HubOrder), - ("lnk_order_customer", "02_links.sql", x5.LinkOrderCustomer), - ("lnk_order_product", "02_links.sql", x5.LinkOrderProduct), - ("lnk_order_store", "02_links.sql", x5.LinkOrderStore), - ("sat_order_header__1c__msk", "satellites/sat_order_header__1c__msk.sql", x5.SatOrderHeader), - ("sat_order_pricing__1c__msk", "satellites/sat_order_pricing__1c__msk.sql", x5.SatOrderPricing), +ORDER_FEED_TABLES = [ + ("hub_customer", "01_hubs.sql", vault_rows.HubCustomer), + ("hub_product", "01_hubs.sql", vault_rows.HubProduct), + ("hub_store", "01_hubs.sql", vault_rows.HubStore), + ("hub_order", "01_hubs.sql", vault_rows.HubOrder), + ("lnk_order_customer", "02_links.sql", vault_rows.LinkOrderCustomer), + ("lnk_order_product", "02_links.sql", vault_rows.LinkOrderProduct), + ("lnk_order_store", "02_links.sql", vault_rows.LinkOrderStore), + ( + "sat_order_header__1c__msk", + "satellites/sat_order_header__1c__msk.sql", + vault_rows.SatOrderHeader, + ), + ( + "sat_order_pricing__1c__msk", + "satellites/sat_order_pricing__1c__msk.sql", + vault_rows.SatOrderPricing, + ), ] -@pytest.mark.parametrize(("table", "ddl_file", "model"), X5_TABLES) -def test_x5_insert_columns_exist_in_postgres_ddl(table, ddl_file, model): +@pytest.mark.parametrize(("table", "ddl_file", "model"), ORDER_FEED_TABLES) +def test_order_feed_insert_columns_exist_in_postgres_ddl(table, ddl_file, model): columns = list(model.model_fields.keys()) ddl_columns = _columns_for(table, ddl_file) missing = set(columns) - ddl_columns @@ -198,13 +202,13 @@ def test_x5_insert_columns_exist_in_postgres_ddl(table, ddl_file, model): assert parsed is not None -X5_BRANCHES = ["msk", "spb", "ekb", "dxb", "ala"] +ORDER_FEED_BRANCHES = ["msk", "spb", "ekb", "dxb", "ala"] -@pytest.mark.parametrize("branch", X5_BRANCHES) -def test_x5_per_branch_order_satellites_have_postgres_ddl(branch): - # The loader writes sat_order_header/pricing for every observed branch, so a - # PostgreSQL table must exist for each. +@pytest.mark.parametrize("branch", ORDER_FEED_BRANCHES) +def test_order_feed_per_branch_order_satellites_have_postgres_ddl(branch): + # The per-branch order feed writes sat_order_header/pricing for every + # observed branch, so a PostgreSQL table must exist for each. for prefix in ("sat_order_header", "sat_order_pricing"): path = PG_DIR / "satellites" / f"{prefix}__1c__{branch}.sql" assert path.exists(), f"missing PostgreSQL DDL for {prefix}__1c__{branch}" @@ -257,9 +261,9 @@ def test_reference_generic_hub_link_fields_are_renamed(): # --- PostgresVaultWriter ------------------------------------------------------ -def _hub_customers(n: int) -> list[x5.HubCustomer]: +def _hub_customers(n: int) -> list[vault_rows.HubCustomer]: return [ - x5.HubCustomer( + vault_rows.HubCustomer( customer_hk=bytes([i]) * 16, customer_bk=f"c{i}", load_ts=datetime(2026, 5, 29, 10, 0, 0), @@ -367,52 +371,6 @@ def test_connect_raises_when_psycopg_missing(monkeypatch): pg_vault_writer.connect("postgresql://x") -# --- loader sink selection --------------------------------------------------- - - -def _open(target: str, dry_run: bool, monkeypatch=None): - return loader._open_sink( - target=target, - dry_run=dry_run, - clickhouse_host="h", - clickhouse_port=9000, - clickhouse_database="rv", - clickhouse_user="u", - clickhouse_password="", - postgres_dsn="postgresql://agentflow@localhost:5432/agentflow", - max_active_parts=5, - ) - - -def test_open_sink_dry_run_never_connects(): - sink, throttle = _open("postgres", dry_run=True) - assert isinstance(sink, loader._DryRunSink) - assert sink.mode == "mapped" - assert throttle.client is None - - -def test_open_sink_postgres(monkeypatch): - conn = _FakeConnection() - monkeypatch.setattr(loader, "connect_postgres", lambda dsn: conn) - sink, throttle = _open("postgres", dry_run=False) - assert isinstance(sink, loader._PostgresSink) - # ClickHouse part-count backpressure is inert on PostgreSQL. - assert throttle.client is None - - -def test_open_sink_clickhouse(monkeypatch): - sentinel = object() - monkeypatch.setattr(loader, "_connect", lambda *args: sentinel) - sink, throttle = _open("clickhouse", dry_run=False) - assert isinstance(sink, loader._ClickHouseSink) - assert throttle.client is sentinel - - -def test_open_sink_unknown_target_raises(): - with pytest.raises(click.ClickException): - _open("redis", dry_run=False) - - # --- reference load_postgres CLI --------------------------------------------- diff --git a/tests/unit/test_dv2_supplier_reference.py b/tests/unit/test_dv2_supplier_reference.py index ea70e38d..54f103b5 100644 --- a/tests/unit/test_dv2_supplier_reference.py +++ b/tests/unit/test_dv2_supplier_reference.py @@ -1,7 +1,8 @@ """Unit tests for the DV2 supplier/product reference (no Docker). Pins the genuine standards (GS1 GTIN/GLN, RU INN, ТН ВЭД), generator -determinism, hash-key join-compatibility with the X5/1C vault feeds, and +determinism, hash-key stability of the reference loader's own MD5 +canonicalisation (join-compatibility with the other vault feeds), and raw-vault referential integrity. """ @@ -12,12 +13,6 @@ import pytest from click.testing import CliRunner -from warehouse.agentflow.dv2.loaders.x5_retail_hero.mappers import ( - composite_md5_digest as x5_composite_md5_digest, -) -from warehouse.agentflow.dv2.loaders.x5_retail_hero.mappers import ( - md5_digest as x5_md5_digest, -) from warehouse.agentflow.dv2.reference.build import ( _manifest, _reference_frames, @@ -128,14 +123,35 @@ def test_each_product_has_a_primary_supplier(): assert sorted(priorities) == list(range(1, len(priorities) + 1)) -# --- hash-key join-compatibility (the critical pin) -------------------------- +# --- hash-key stability (the critical pin) ----------------------------------- + + +def test_md5_digest_matches_known_vectors(): + """Pins vault_mapping's own MD5 canonicalisation against precomputed hex + digests, so any accidental change to ``_canonical``/``md5_digest`` is + caught without depending on another loader for cross-checking. These are + plain ``str.encode()`` MD5 sums for non-bytes/bool/date/Decimal inputs + (``_canonical`` is the identity on stripped strings here), so they are + reproducible with any MD5 implementation, not just this codebase. + """ + known_hex_digests = { + "7736207543": "fc852a3d08d2b40d59816e071c6718f8", + "RC000001": "6576fe92210434947aaacd34f93d0b67", + "4660000254375": "05c0e20048607cb2f560713a3d42ac3f", + "ООО «ГринФрут»": "b0acd6a4c6189c61db1f65517de7c73c", + "123": "202cb962ac59075b964b07152d234b70", + } + for value, expected_hex in known_hex_digests.items(): + assert md5_digest(value).hex() == expected_hex -def test_hash_keys_match_x5_loader_byte_for_byte(): - for value in ["7736207543", "RC000001", "4660000254375", "ООО «ГринФрут»", "123"]: - assert md5_digest(value) == x5_md5_digest(value) +def test_composite_md5_digest_matches_known_vector(): + # md5("a") / md5("b") are the textbook MD5 vectors; composite_md5_digest + # joins their hex representations with "||" and hashes that. left, right = md5_digest("a"), md5_digest("b") - assert composite_md5_digest(left, right) == x5_composite_md5_digest(left, right) + assert left.hex() == "0cc175b9c0f1b6a831c399e269772661" + assert right.hex() == "92eb5ffee6ae2fec3ad71c777531578f" + assert composite_md5_digest(left, right).hex() == "a2bacd576459a208e719673d377b9bd1" # --- vault mapping ----------------------------------------------------------- diff --git a/tests/unit/test_x5_retail_hero_loader.py b/tests/unit/test_x5_retail_hero_loader.py deleted file mode 100644 index 3bc3d0b0..00000000 --- a/tests/unit/test_x5_retail_hero_loader.py +++ /dev/null @@ -1,99 +0,0 @@ -from __future__ import annotations - -from datetime import datetime - -import pandas as pd - -from warehouse.agentflow.dv2.loaders.x5_retail_hero.mappers import ( - map_purchases_chunk, - rows_to_dicts, -) - - -def test_purchase_mapper_emits_hub_store_business_key_column(): - mapped = map_purchases_chunk( - pd.DataFrame( - [ - { - "store_id": 1, - "client_id": "client-1", - "product_id": "sku-1", - "transaction_id": "tx-1", - "express_points_received": 0, - "express_points_spent": 0, - "purchase_sum": 10, - "regular_points_received": 0, - "regular_points_spent": 0, - "transaction_datetime": "2026-05-29 10:00:00", - "product_quantity": 1, - "trn_sum_from_iss": 10, - } - ] - ), - datetime(2026, 5, 29, 10, 0, 0), - {1: "msk"}, - {}, - set(), - ) - - hub_store_rows = rows_to_dicts(mapped["hub_store"]) - - assert hub_store_rows[0]["store_bk"] == "msk-1" - assert "store_code" not in hub_store_rows[0] - - -def test_clickhouse_connect_passes_credentials(monkeypatch): - from warehouse.agentflow.dv2.loaders.x5_retail_hero import loader - - calls: list[dict[str, object]] = [] - - class FakeClient: - def __init__(self, **kwargs): - calls.append(kwargs) - - def execute(self, query: str): - calls.append({"query": query}) - - monkeypatch.setattr(loader, "Client", FakeClient) - - client = loader._connect("clickhouse.local", 9000, "default", "demo") - - assert isinstance(client, FakeClient) - assert calls == [ - { - "host": "clickhouse.local", - "port": 9000, - "user": "default", - "password": "demo", - }, - {"query": "SELECT 1"}, - ] - - -def test_parts_throttle_waits_until_merges_catch_up(monkeypatch): - from warehouse.agentflow.dv2.loaders.x5_retail_hero.loader import PartsThrottle - - class _FakeClient: - def __init__(self, counts): - self.counts = list(counts) - - def execute(self, query, params): - assert "system.parts" in query - assert params == {"database": "rv"} - return [(self.counts.pop(0),)] - - sleeps: list[float] = [] - monkeypatch.setattr( - "warehouse.agentflow.dv2.loaders.x5_retail_hero.loader.time.sleep", - sleeps.append, - ) - - # Above budget twice, then merges catch up. - throttle = PartsThrottle(_FakeClient([900, 700, 300]), "rv", max_active_parts=400) - throttle.wait_if_needed() - assert len(sleeps) == 2 - - # Disabled throttle (0) and dry-run (client=None) never query or sleep. - PartsThrottle(_FakeClient([]), "rv", max_active_parts=0).wait_if_needed() - PartsThrottle(None, "rv", max_active_parts=400).wait_if_needed() - assert len(sleeps) == 2 diff --git a/warehouse/agentflow/dv2/loaders/x5_retail_hero/schemas.py b/warehouse/agentflow/dv2/loaders/vault_rows.py similarity index 69% rename from warehouse/agentflow/dv2/loaders/x5_retail_hero/schemas.py rename to warehouse/agentflow/dv2/loaders/vault_rows.py index a95b3446..d5e2511a 100644 --- a/warehouse/agentflow/dv2/loaders/x5_retail_hero/schemas.py +++ b/warehouse/agentflow/dv2/loaders/vault_rows.py @@ -1,6 +1,16 @@ +"""Vault-generic raw-vault row models shared by DV2 PostgreSQL ingestion. + +These pydantic models describe the hub / link / order-satellite shape used by +every per-branch order feed (``sat_order_header__1c__*`` / +``sat_order_pricing__1c__*``) and by ``PostgresVaultWriter``'s column-order / +DDL-coverage tests. They are entity-generic (not tied to any one source +system or dataset) — see ``reference/vault_mapping.py`` for the analogous +models used by the supplier/product reference feed. +""" + from __future__ import annotations -from datetime import date, datetime +from datetime import datetime from decimal import Decimal from typing import Annotated @@ -65,35 +75,9 @@ class LinkOrderStore(VaultRow): record_source: str -class SatCustomerPersonal(VaultRow): - customer_hk: Hash16 - load_ts: datetime - hash_diff: Hash16 - record_source: str - age: int | None - gender: str | None - first_issue_date: date | None - first_redeem_date: date | None - - -class SatProductCatalog(VaultRow): - product_hk: Hash16 - load_ts: datetime - hash_diff: Hash16 - record_source: str - level_1: str | None - level_2: str | None - level_3: str | None - level_4: str | None - brand_id: str | None - segment_id: str | None - netto: Decimal | None - is_own_trademark: bool | None - - class SatOrderHeader(VaultRow): # Aligned to the deployed rv.sat_order_header__* DDL (order_date / channel / - # order_status / total_amount) so X5 orders are visible to bv_order_canonical. + # order_status / total_amount). order_hk: Hash16 load_ts: datetime hash_diff: Hash16 @@ -105,8 +89,6 @@ class SatOrderHeader(VaultRow): class SatOrderPricing(VaultRow): - # Synthesized from the X5 purchase_sum (gross) with per-branch tax rates so - # bv_order_canonical / branch_pnl have a non-null subtotal_amount. order_hk: Hash16 load_ts: datetime hash_diff: Hash16 diff --git a/warehouse/agentflow/dv2/loaders/x5_retail_hero/README.md b/warehouse/agentflow/dv2/loaders/x5_retail_hero/README.md deleted file mode 100644 index 2c4d5f27..00000000 --- a/warehouse/agentflow/dv2/loaders/x5_retail_hero/README.md +++ /dev/null @@ -1,67 +0,0 @@ -# X5 Retail Hero DV2.0 Loader - -Loads the Kaggle `mvyurchenko/x5-retail-hero` CSV files into the AgentFlow DV2.0 -raw vault. The vault now lives on **PostgreSQL** (`--target postgres`, see -`warehouse/agentflow/dv2/postgres/`); ClickHouse (`--target clickhouse`, the -default) is retained as the legacy backend. - -Download the dataset with Kaggle CLI: - -```bash -kaggle datasets download -d mvyurchenko/x5-retail-hero -p /path/to/x5 --unzip -``` - -Expected files: - -- `clients.csv` -- `products.csv` -- `purchases.csv` - -Run against the PostgreSQL vault (apply `dv2/postgres/` DDL first): - -```bash -python loader.py --csv-dir /path/to/x5 --target postgres \ - --postgres-dsn postgresql://agentflow@localhost:5432/agentflow --batch-size 100000 -``` - -Run against the legacy ClickHouse backend: - -```bash -python loader.py --csv-dir /path/to/x5 --clickhouse-host localhost --clickhouse-port 9000 --clickhouse-user default --clickhouse-password demo --batch-size 100000 -``` - -Dry run parses and maps chunks without inserting: - -```bash -python loader.py --csv-dir /path/to/x5 --clickhouse-host localhost --clickhouse-port 9000 --batch-size 100000 --dry-run -``` - -Replay with a fixed UTC load timestamp: - -```bash -python loader.py --csv-dir /path/to/x5 --clickhouse-host localhost --clickhouse-port 9000 --clickhouse-user default --clickhouse-password demo --batch-size 100000 --load-ts 2026-05-23T10:15:30Z -``` - -## Schema Assumptions - -The loader assumes the DV2.0 DDL has already been applied. For PostgreSQL apply `dv2/postgres/apply.sh` (tables in the `rv` schema); for ClickHouse apply `warehouse/agentflow/dv2/__init.sql` (tables in the `rv` database unless `--clickhouse-database` is provided). Idempotency on PostgreSQL is enforced by `INSERT ... ON CONFLICT DO NOTHING`; on ClickHouse it relies on the `ReplacingMergeTree`/merge engines. - -Satellite idempotency is handled by the raw vault table engines. The loader always inserts mapped rows and relies on the expected `ReplacingMergeTree` or equivalent merge behavior for repeated hub/link keys and unchanged satellite `hk + hash_diff` pairs. - -The X5 `clients.csv` file has no branch field, so `hub_customer` rows use `record_source = '1c__msk'`. Per-branch `sat_customer_personal__1c__{branch}` rows are emitted while processing purchases, using each customer's observed purchase branch. - -`store_id` values are converted to store business keys as `{branch}-{store_id}` before hashing into `hub_store.store_hk` and writing `hub_store.store_bk`. - -## Branch Distribution - -Stores are sorted by `store_id` and assigned with deterministic weighted round-robin: - -| Branch | Share | -|---|---:| -| `msk` | 40% | -| `spb` | 25% | -| `ekb` | 15% | -| `dxb` | 10% | -| `ala` | 10% | - -Given the same set of `store_id` values, the mapping is stable across runs. diff --git a/warehouse/agentflow/dv2/loaders/x5_retail_hero/__init__.py b/warehouse/agentflow/dv2/loaders/x5_retail_hero/__init__.py deleted file mode 100644 index 1f920d1a..00000000 --- a/warehouse/agentflow/dv2/loaders/x5_retail_hero/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -from .branch_distributor import BRANCH_WEIGHTS, distribute_stores_to_branches - -__all__ = ["BRANCH_WEIGHTS", "distribute_stores_to_branches"] diff --git a/warehouse/agentflow/dv2/loaders/x5_retail_hero/branch_distributor.py b/warehouse/agentflow/dv2/loaders/x5_retail_hero/branch_distributor.py deleted file mode 100644 index 2e029347..00000000 --- a/warehouse/agentflow/dv2/loaders/x5_retail_hero/branch_distributor.py +++ /dev/null @@ -1,61 +0,0 @@ -from __future__ import annotations - -from decimal import Decimal, InvalidOperation -from typing import Any, Iterable - - -BRANCH_WEIGHTS: tuple[tuple[str, int], ...] = ( - ("msk", 40), - ("spb", 25), - ("ekb", 15), - ("dxb", 10), - ("ala", 10), -) - - -def normalize_store_id(store_id: Any) -> str: - if store_id is None: - return "" - - try: - if store_id != store_id: - return "" - except TypeError: - pass - - if isinstance(store_id, float) and store_id.is_integer(): - return str(int(store_id)) - - return str(store_id).strip() - - -def _sort_key(store_id: Any) -> tuple[int, Decimal | str]: - normalized = normalize_store_id(store_id) - try: - return (0, Decimal(normalized)) - except (InvalidOperation, ValueError): - return (1, normalized) - - -def distribute_stores_to_branches(store_ids: Iterable[Any]) -> dict[Any, str]: - unique_store_ids = sorted( - {store_id for store_id in store_ids if normalize_store_id(store_id)}, - key=_sort_key, - ) - totals = {branch_code: 0 for branch_code, _ in BRANCH_WEIGHTS} - total_weight = sum(weight for _, weight in BRANCH_WEIGHTS) - branch_order = {branch_code: index for index, (branch_code, _) in enumerate(BRANCH_WEIGHTS)} - - assignments: dict[Any, str] = {} - for store_id in unique_store_ids: - for branch_code, weight in BRANCH_WEIGHTS: - totals[branch_code] += weight - - branch_code = max( - totals, - key=lambda code: (totals[code], -branch_order[code]), - ) - assignments[store_id] = branch_code - totals[branch_code] -= total_weight - - return assignments diff --git a/warehouse/agentflow/dv2/loaders/x5_retail_hero/loader.py b/warehouse/agentflow/dv2/loaders/x5_retail_hero/loader.py deleted file mode 100644 index d053ea70..00000000 --- a/warehouse/agentflow/dv2/loaders/x5_retail_hero/loader.py +++ /dev/null @@ -1,453 +0,0 @@ -from __future__ import annotations - -import logging -import sys -import time -from collections import defaultdict -from datetime import UTC, datetime -from pathlib import Path -from typing import Any - -import click -import pandas as pd -from pydantic import BaseModel -from tqdm import tqdm - -try: - from clickhouse_driver import Client - from clickhouse_driver.errors import Error as ClickHouseError -except ModuleNotFoundError: - Client = None # type: ignore[assignment] - ClickHouseError = Exception # type: ignore[misc, assignment] - -try: - from ..pg_vault_writer import PostgresVaultWriter - from ..pg_vault_writer import connect as connect_postgres - from .branch_distributor import distribute_stores_to_branches - from .mappers import ( - map_clients_chunk, - map_products_chunk, - map_purchases_chunk, - rows_to_dicts, - ) -except ImportError: - import sys as _sys - - _sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) - from branch_distributor import distribute_stores_to_branches - from mappers import ( - map_clients_chunk, - map_products_chunk, - map_purchases_chunk, - rows_to_dicts, - ) - from pg_vault_writer import PostgresVaultWriter - from pg_vault_writer import connect as connect_postgres - - -CLICKHOUSE_TARGET = "clickhouse" -POSTGRES_TARGET = "postgres" -TARGETS = (CLICKHOUSE_TARGET, POSTGRES_TARGET) - - -REQUIRED_CSVS = ("clients.csv", "products.csv", "purchases.csv") -LOGGER = logging.getLogger("x5_retail_hero_loader") - - -class PartsThrottle: - """Backpressure on the active-part count so merges keep up with inserts. - - The 2026-06-02 full load ran with merges OFF and accumulated tens of - thousands of parts — more than an 8 GB host can even cold-start on - (part-metadata load OOMs before the server serves a query). Keeping - merges ON but pausing inserts whenever the active-part count crosses a - bound keeps the part set small enough to survive a cold start at any - moment during the load. - """ - - def __init__( - self, - client: Client | None, - database: str, - max_active_parts: int, - poll_seconds: float = 15.0, - ) -> None: - self.client = client - self.database = database - self.max_active_parts = max_active_parts - self.poll_seconds = poll_seconds - - def wait_if_needed(self) -> None: - if self.client is None or self.max_active_parts <= 0: - return - while True: - active = self._active_parts() - if active <= self.max_active_parts: - return - LOGGER.info( - "throttle: %s active parts in %s > %s budget, waiting %.0fs for merges", - active, - self.database, - self.max_active_parts, - self.poll_seconds, - ) - time.sleep(self.poll_seconds) - - def _active_parts(self) -> int: - result = self.client.execute( - "SELECT count() FROM system.parts WHERE database = %(database)s AND active", - {"database": self.database}, - ) - return int(result[0][0]) - - -class _DryRunSink: - """Map-only sink: rows are counted by the caller but never persisted.""" - - mode = "mapped" - - def write(self, table: str, rows: list[BaseModel]) -> None: - return None - - def close(self) -> None: - return None - - -class _ClickHouseSink: - """Insert mapped rows into the ClickHouse ``rv`` database (legacy backend).""" - - mode = "inserted" - - def __init__(self, client: Client, database: str) -> None: - self._client = client - self._database = database - - def write(self, table: str, rows: list[BaseModel]) -> None: - _insert_rows(self._client, _qualified_table(self._database, table), rows) - - def close(self) -> None: - return None - - -class _PostgresSink: - """Insert mapped rows into the PostgreSQL ``rv`` schema (the DV2 raw vault). - - The vault moved off ClickHouse onto PostgreSQL (see dv2/postgres/README.md); - this sink is what actually feeds it. Writes are buffered into the connection - and committed once on :meth:`close`. - """ - - mode = "inserted" - - def __init__(self, writer: PostgresVaultWriter) -> None: - self._writer = writer - - def write(self, table: str, rows: list[BaseModel]) -> None: - self._writer.write(table, rows) - - def close(self) -> None: - self._writer.commit() - self._writer.close() - - -@click.command() -@click.option("--csv-dir", required=True, type=click.Path(file_okay=False, path_type=Path)) -@click.option("--clickhouse-host", default="localhost", show_default=True) -@click.option("--clickhouse-port", default=9000, show_default=True, type=int) -@click.option("--clickhouse-database", default="rv", show_default=True) -@click.option("--clickhouse-user", default="default", show_default=True) -@click.option("--clickhouse-password", default="", show_default=True) -@click.option("--batch-size", default=100_000, show_default=True, type=int) -@click.option("--dry-run", is_flag=True) -@click.option( - "--load-ts", - default=None, - help="UTC timestamp override, for example 2026-05-23T10:15:30Z.", -) -@click.option( - "--max-active-parts", - default=0, - show_default=True, - type=int, - help=( - "Pause inserts while the database has more active parts than this " - "(0 disables). Keeps merges able to catch up on small hosts. " - "ClickHouse target only." - ), -) -@click.option( - "--target", - default=CLICKHOUSE_TARGET, - show_default=True, - type=click.Choice(TARGETS), - help="Where to load the raw vault: the PostgreSQL vault or legacy ClickHouse.", -) -@click.option( - "--postgres-dsn", - default="postgresql://agentflow@localhost:5432/agentflow", - show_default=True, - help="PostgreSQL DSN used when --target=postgres.", -) -def cli( - csv_dir: Path, - clickhouse_host: str, - clickhouse_port: int, - clickhouse_database: str, - clickhouse_user: str, - clickhouse_password: str, - batch_size: int, - dry_run: bool, - load_ts: str | None, - max_active_parts: int, - target: str, - postgres_dsn: str, -) -> None: - _configure_logging() - csv_paths = _validate_csvs(csv_dir) - current_load_ts = _parse_load_ts(load_ts) - sink, throttle = _open_sink( - target=target, - dry_run=dry_run, - clickhouse_host=clickhouse_host, - clickhouse_port=clickhouse_port, - clickhouse_database=clickhouse_database, - clickhouse_user=clickhouse_user, - clickhouse_password=clickhouse_password, - postgres_dsn=postgres_dsn, - max_active_parts=max_active_parts, - ) - - client_lookup: dict[str, dict[str, Any]] = {} - try: - _process_products(csv_paths["products.csv"], batch_size, current_load_ts, sink, throttle) - client_lookup.update( - _process_clients(csv_paths["clients.csv"], batch_size, current_load_ts, sink, throttle) - ) - - LOGGER.info("Building stable branch map from purchases.csv store_id values") - store_branch_map = _build_store_branch_map(csv_paths["purchases.csv"], batch_size) - LOGGER.info("Mapped %s stores to branches", len(store_branch_map)) - - seen_customer_personal: set[tuple[str, str]] = set() - _process_purchases( - csv_paths["purchases.csv"], - batch_size, - current_load_ts, - store_branch_map, - client_lookup, - seen_customer_personal, - sink, - throttle, - ) - finally: - sink.close() - - -def _open_sink( - *, - target: str, - dry_run: bool, - clickhouse_host: str, - clickhouse_port: int, - clickhouse_database: str, - clickhouse_user: str, - clickhouse_password: str, - postgres_dsn: str, - max_active_parts: int, -) -> tuple[Any, PartsThrottle]: - """Build the row sink and its (ClickHouse-only) part-count throttle. - - A dry run never connects. The throttle queries ClickHouse ``system.parts``, - so the PostgreSQL and dry-run paths get an inert ``PartsThrottle(None, ...)``. - """ - if dry_run: - return _DryRunSink(), PartsThrottle(None, clickhouse_database, max_active_parts) - if target == CLICKHOUSE_TARGET: - client = _connect(clickhouse_host, clickhouse_port, clickhouse_user, clickhouse_password) - return ( - _ClickHouseSink(client, clickhouse_database), - PartsThrottle(client, clickhouse_database, max_active_parts), - ) - if target == POSTGRES_TARGET: - writer = PostgresVaultWriter(connect_postgres(postgres_dsn)) - return _PostgresSink(writer), PartsThrottle(None, clickhouse_database, max_active_parts) - raise click.ClickException(f"unknown target: {target}") - - -def _configure_logging() -> None: - logging.basicConfig( - level=logging.INFO, - format="%(asctime)s %(levelname)s %(message)s", - stream=sys.stdout, - ) - - -def _validate_csvs(csv_dir: Path) -> dict[str, Path]: - if not csv_dir.exists() or not csv_dir.is_dir(): - raise click.ClickException(f"CSV directory does not exist: {csv_dir}") - - paths = {filename: csv_dir / filename for filename in REQUIRED_CSVS} - missing = [str(path) for path in paths.values() if not path.exists()] - if missing: - raise click.ClickException(f"Missing required CSV file(s): {', '.join(missing)}") - - return paths - - -def _connect(host: str, port: int, user: str, password: str) -> Client: - if Client is None: - raise click.ClickException("clickhouse-driver is required unless --dry-run is used.") - - try: - client = Client(host=host, port=port, user=user, password=password) - client.execute("SELECT 1") - return client - except ClickHouseError as exc: - raise click.ClickException(f"ClickHouse is unreachable at {host}:{port}: {exc}") from exc - except OSError as exc: - raise click.ClickException(f"ClickHouse is unreachable at {host}:{port}: {exc}") from exc - - -def _parse_load_ts(value: str | None) -> datetime: - if not value: - return datetime.now(UTC).replace(tzinfo=None) - - normalized = value.replace("Z", "+00:00") - parsed = datetime.fromisoformat(normalized) - if parsed.tzinfo is None: - return parsed - return parsed.astimezone(UTC).replace(tzinfo=None) - - -def _build_store_branch_map(purchases_path: Path, batch_size: int) -> dict[Any, str]: - store_ids: set[Any] = set() - try: - reader = pd.read_csv( - purchases_path, - chunksize=batch_size, - usecols=["store_id"], - low_memory=False, - ) - for chunk in tqdm(reader, desc="purchases.csv stores", unit="chunk"): - store_ids.update(chunk["store_id"].dropna().unique().tolist()) - except (KeyError, ValueError, FileNotFoundError, pd.errors.EmptyDataError) as exc: - raise click.ClickException(f"Unable to read store_id from {purchases_path}: {exc}") from exc - - return distribute_stores_to_branches(store_ids) - - -def _process_products( - products_path: Path, - batch_size: int, - load_ts: datetime, - sink: Any, - throttle: PartsThrottle, -) -> None: - totals: dict[str, int] = defaultdict(int) - try: - reader = pd.read_csv(products_path, chunksize=batch_size, low_memory=False) - for chunk in tqdm(reader, desc="products.csv", unit="chunk"): - mapped = map_products_chunk(chunk, load_ts) - _emit_mapped_rows(mapped, sink, totals) - throttle.wait_if_needed() - except (KeyError, ValueError, FileNotFoundError, pd.errors.EmptyDataError) as exc: - raise click.ClickException(f"Unable to process {products_path}: {exc}") from exc - - _log_totals("products.csv", totals, sink.mode) - - -def _process_clients( - clients_path: Path, - batch_size: int, - load_ts: datetime, - sink: Any, - throttle: PartsThrottle, -) -> dict[str, dict[str, Any]]: - totals: dict[str, int] = defaultdict(int) - client_lookup: dict[str, dict[str, Any]] = {} - try: - reader = pd.read_csv(clients_path, chunksize=batch_size, low_memory=False) - for chunk in tqdm(reader, desc="clients.csv", unit="chunk"): - mapped, chunk_lookup = map_clients_chunk(chunk, load_ts) - client_lookup.update(chunk_lookup) - _emit_mapped_rows(mapped, sink, totals) - throttle.wait_if_needed() - except (KeyError, ValueError, FileNotFoundError, pd.errors.EmptyDataError) as exc: - raise click.ClickException(f"Unable to process {clients_path}: {exc}") from exc - - _log_totals("clients.csv", totals, sink.mode) - return client_lookup - - -def _process_purchases( - purchases_path: Path, - batch_size: int, - load_ts: datetime, - store_branch_map: dict[Any, str], - client_lookup: dict[str, dict[str, Any]], - seen_customer_personal: set[tuple[str, str]], - sink: Any, - throttle: PartsThrottle, -) -> None: - totals: dict[str, int] = defaultdict(int) - try: - reader = pd.read_csv(purchases_path, chunksize=batch_size, low_memory=False) - for chunk in tqdm(reader, desc="purchases.csv", unit="chunk"): - mapped = map_purchases_chunk( - chunk, - load_ts, - store_branch_map, - client_lookup, - seen_customer_personal, - ) - _emit_mapped_rows(mapped, sink, totals) - throttle.wait_if_needed() - except (KeyError, ValueError, FileNotFoundError, pd.errors.EmptyDataError) as exc: - raise click.ClickException(f"Unable to process {purchases_path}: {exc}") from exc - - _log_totals("purchases.csv", totals, sink.mode) - - -def _emit_mapped_rows( - mapped: dict[str, list[BaseModel]], - sink: Any, - totals: dict[str, int], -) -> None: - for table, rows in mapped.items(): - if not rows: - continue - totals[table] += len(rows) - sink.write(table, rows) - - -def _insert_rows(client: Client | None, table: str, rows: list[BaseModel]) -> None: - if client is None: - return - - dict_rows = rows_to_dicts(rows) - if not dict_rows: - return - - columns = list(dict_rows[0].keys()) - column_sql = ", ".join(columns) - try: - client.execute(f"INSERT INTO {table} ({column_sql}) VALUES", dict_rows) - except ClickHouseError as exc: - raise click.ClickException(f"Insert failed for {table}: {exc}") from exc - except OSError as exc: - raise click.ClickException(f"Insert failed for {table}: {exc}") from exc - - -def _qualified_table(database: str, table: str) -> str: - if "." in table: - return table - return f"{database}.{table}" - - -def _log_totals(filename: str, totals: dict[str, int], mode: str) -> None: - for table in sorted(totals): - LOGGER.info("%s %s %s rows into %s", filename, mode, totals[table], table) - - -if __name__ == "__main__": - cli() diff --git a/warehouse/agentflow/dv2/loaders/x5_retail_hero/mappers.py b/warehouse/agentflow/dv2/loaders/x5_retail_hero/mappers.py deleted file mode 100644 index 4dbe79b2..00000000 --- a/warehouse/agentflow/dv2/loaders/x5_retail_hero/mappers.py +++ /dev/null @@ -1,410 +0,0 @@ -from __future__ import annotations - -import hashlib -from collections import defaultdict -from collections.abc import Mapping -from datetime import date, datetime -from decimal import Decimal, InvalidOperation -from typing import Any - -import pandas as pd -from pydantic import BaseModel - -try: - from .branch_distributor import normalize_store_id - from .schemas import ( - HubCustomer, - HubOrder, - HubProduct, - HubStore, - LinkOrderCustomer, - LinkOrderProduct, - LinkOrderStore, - SatOrderHeader, - SatOrderPricing, - ) -except ImportError: - from branch_distributor import normalize_store_id - from schemas import ( - HubCustomer, - HubOrder, - HubProduct, - HubStore, - LinkOrderCustomer, - LinkOrderProduct, - LinkOrderStore, - SatOrderHeader, - SatOrderPricing, - ) - - -SOURCE_SYSTEM = "1c" -PRODUCT_BRANCH = "msk" -TABLE_HUB_CUSTOMER = "hub_customer" -TABLE_HUB_PRODUCT = "hub_product" -TABLE_HUB_STORE = "hub_store" -TABLE_HUB_ORDER = "hub_order" -TABLE_LNK_ORDER_CUSTOMER = "lnk_order_customer" -TABLE_LNK_ORDER_PRODUCT = "lnk_order_product" -TABLE_LNK_ORDER_STORE = "lnk_order_store" - -# Per-branch VAT/sales-tax used to synthesize a pricing satellite from the X5 -# gross purchase_sum (RU 20%, UAE 5%, KZ 12%), matching the synthetic seed. -BRANCH_TAX_RATE = { - "msk": Decimal("0.20"), - "spb": Decimal("0.20"), - "ekb": Decimal("0.20"), - "dxb": Decimal("0.05"), - "ala": Decimal("0.12"), -} -_CENTS = Decimal("0.01") - -MappedRows = dict[str, list[BaseModel]] -ClientLookup = dict[str, dict[str, Any]] - - -def md5_digest(value: Any) -> bytes: - return hashlib.md5(_canonical(value).encode("utf-8"), usedforsecurity=False).digest() - - -def composite_md5_digest(*parts: Any) -> bytes: - payload = "||".join(_canonical(part) for part in parts).encode("utf-8") - return hashlib.md5(payload, usedforsecurity=False).digest() - - -def hash_diff(attributes: Mapping[str, Any]) -> bytes: - payload = "||".join(f"{key}={_canonical(attributes[key])}" for key in sorted(attributes)) - return hashlib.md5(payload.encode("utf-8"), usedforsecurity=False).digest() - - -def record_source(branch_code: str) -> str: - return f"{SOURCE_SYSTEM}__{branch_code}" - - -def order_business_key(branch_code: str, transaction_id: Any) -> str: - return f"{SOURCE_SYSTEM}__{branch_code}__{_clean_string(transaction_id)}" - - -def store_code(branch_code: str, store_id: Any) -> str: - return f"{branch_code}-{normalize_store_id(store_id)}" - - -def map_products_chunk(products: pd.DataFrame, load_ts: datetime) -> MappedRows: - mapped: MappedRows = defaultdict(list) - rows = products.drop_duplicates(subset=["product_id"]).to_dict("records") - - for row in rows: - sku = _clean_string(row["product_id"]) - if not sku: - continue - - product_hk = md5_digest(sku) - source = record_source(PRODUCT_BRANCH) - mapped[TABLE_HUB_PRODUCT].append( - HubProduct(product_hk=product_hk, product_bk=sku, load_ts=load_ts, record_source=source) - ) - # X5 product attributes (level_*, brand_id, netto) do not match the - # deployed sat_product_catalog DDL (product_name/brand/category/...), and - # the catalog satellite is not on the branch_pnl path, so it is not - # emitted. See docs: X5 is wired through to the financial P&L mart only. - - return dict(mapped) - - -def map_clients_chunk( - clients: pd.DataFrame, - load_ts: datetime, - default_branch: str = PRODUCT_BRANCH, -) -> tuple[MappedRows, ClientLookup]: - mapped: MappedRows = defaultdict(list) - lookup: ClientLookup = {} - rows = clients.drop_duplicates(subset=["client_id"]).to_dict("records") - - for row in rows: - customer_bk = _clean_string(row["client_id"]) - if not customer_bk: - continue - - customer_hk = md5_digest(customer_bk) - mapped[TABLE_HUB_CUSTOMER].append( - HubCustomer( - customer_hk=customer_hk, - customer_bk=customer_bk, - load_ts=load_ts, - record_source=record_source(default_branch), - ) - ) - lookup[customer_bk] = { - "age": _nullable_int(row.get("age")), - "gender": _nullable_string(row.get("gender")), - "first_issue_date": _nullable_date(row.get("first_issue_date")), - "first_redeem_date": _nullable_date(row.get("first_redeem_date")), - } - - return dict(mapped), lookup - - -def _synthesize_pricing( - total_amount: Decimal | None, branch_code: str -) -> dict[str, Decimal | None]: - """Derive a pricing satellite from the X5 gross purchase_sum. - - total_amount is the gross (tax-inclusive) sum; subtotal is backed out with - the branch tax rate, tax is the remainder, discount/shipping are zero. - """ - if total_amount is None: - return { - "subtotal_amount": None, - "discount_amount": None, - "tax_amount": None, - "shipping_cost": None, - } - rate = BRANCH_TAX_RATE.get(branch_code, Decimal("0.20")) - subtotal = (total_amount / (Decimal("1") + rate)).quantize(_CENTS) - tax = (total_amount - subtotal).quantize(_CENTS) - return { - "subtotal_amount": subtotal, - "discount_amount": Decimal("0.00"), - "tax_amount": tax, - "shipping_cost": Decimal("0.00"), - } - - -def map_purchases_chunk( - purchases: pd.DataFrame, - load_ts: datetime, - store_branch_map: Mapping[Any, str], - client_lookup: Mapping[str, Mapping[str, Any]], - seen_customer_personal: set[tuple[str, str]] | None = None, -) -> MappedRows: - mapped: MappedRows = defaultdict(list) - seen_stores: set[str] = set() - seen_orders: set[str] = set() - seen_order_customer_links: set[bytes] = set() - seen_order_product_links: set[bytes] = set() - seen_order_store_links: set[bytes] = set() - seen_order_headers: set[bytes] = set() - - for row in purchases.to_dict("records"): - branch_code = _resolve_branch(row["store_id"], store_branch_map) - source = record_source(branch_code) - customer_bk = _clean_string(row["client_id"]) - sku = _clean_string(row["product_id"]) - transaction_id = _clean_string(row["transaction_id"]) - order_bk = order_business_key(branch_code, transaction_id) - current_store_code = store_code(branch_code, row["store_id"]) - - if not customer_bk or not sku or not transaction_id or not current_store_code: - continue - - customer_hk = md5_digest(customer_bk) - product_hk = md5_digest(sku) - store_hk = md5_digest(current_store_code) - order_hk = md5_digest(order_bk) - - if current_store_code not in seen_stores: - mapped[TABLE_HUB_STORE].append( - HubStore( - store_hk=store_hk, - store_bk=current_store_code, - load_ts=load_ts, - record_source=source, - ) - ) - seen_stores.add(current_store_code) - - if order_bk not in seen_orders: - mapped[TABLE_HUB_ORDER].append( - HubOrder( - order_hk=order_hk, - order_bk=order_bk, - load_ts=load_ts, - record_source=source, - ) - ) - seen_orders.add(order_bk) - - order_customer_hk = composite_md5_digest(order_hk, customer_hk) - if order_customer_hk not in seen_order_customer_links: - mapped[TABLE_LNK_ORDER_CUSTOMER].append( - LinkOrderCustomer( - link_hk=order_customer_hk, - order_hk=order_hk, - customer_hk=customer_hk, - load_ts=load_ts, - record_source=source, - ) - ) - seen_order_customer_links.add(order_customer_hk) - - order_product_hk = composite_md5_digest(order_hk, product_hk) - if order_product_hk not in seen_order_product_links: - mapped[TABLE_LNK_ORDER_PRODUCT].append( - LinkOrderProduct( - link_hk=order_product_hk, - order_hk=order_hk, - product_hk=product_hk, - load_ts=load_ts, - record_source=source, - ) - ) - seen_order_product_links.add(order_product_hk) - - order_store_hk = composite_md5_digest(order_hk, store_hk) - if order_store_hk not in seen_order_store_links: - mapped[TABLE_LNK_ORDER_STORE].append( - LinkOrderStore( - link_hk=order_store_hk, - order_hk=order_hk, - store_hk=store_hk, - load_ts=load_ts, - record_source=source, - ) - ) - seen_order_store_links.add(order_store_hk) - - if order_hk not in seen_order_headers: - total_amount = _nullable_decimal(row.get("purchase_sum")) - header_attrs = { - "order_date": _nullable_datetime(row.get("transaction_datetime")), - "channel": "retail", - "order_status": "completed", - "total_amount": total_amount, - } - mapped[_sat_order_header_table(branch_code)].append( - SatOrderHeader( - order_hk=order_hk, - load_ts=load_ts, - hash_diff=hash_diff(header_attrs), - record_source=source, - **header_attrs, - ) - ) - pricing_attrs = _synthesize_pricing(total_amount, branch_code) - mapped[_sat_order_pricing_table(branch_code)].append( - SatOrderPricing( - order_hk=order_hk, - load_ts=load_ts, - hash_diff=hash_diff(pricing_attrs), - record_source=source, - **pricing_attrs, - ) - ) - seen_order_headers.add(order_hk) - - # X5 has no product line attributes matching sat_lnk_order_product - # (qty/unit_price/discount_pct/line_total) and no customer PII matching - # sat_customer_personal (first_name/email); neither is on the branch_pnl - # path, so they are intentionally not emitted. The lnk_order_product - # link itself (above) carries the order<->product relationship. - - return dict(mapped) - - -def rows_to_dicts(rows: list[BaseModel]) -> list[dict[str, Any]]: - return [row.model_dump(mode="python") for row in rows] - - -def _sat_order_header_table(branch_code: str) -> str: - return f"sat_order_header__{SOURCE_SYSTEM}__{branch_code}" - - -def _sat_order_pricing_table(branch_code: str) -> str: - return f"sat_order_pricing__{SOURCE_SYSTEM}__{branch_code}" - - -def _resolve_branch(store_id: Any, store_branch_map: Mapping[Any, str]) -> str: - if store_id in store_branch_map: - return store_branch_map[store_id] - - normalized = normalize_store_id(store_id) - if normalized in store_branch_map: - return store_branch_map[normalized] - - for known_store_id, branch_code in store_branch_map.items(): - if normalize_store_id(known_store_id) == normalized: - return branch_code - - raise KeyError(f"store_id {normalized!r} is missing from branch map") - - -def _canonical(value: Any) -> str: - if _is_null(value): - return "" - if isinstance(value, bytes): - return value.hex() - if isinstance(value, bool): - return "true" if value else "false" - if isinstance(value, (date, datetime)): - return value.isoformat() - if isinstance(value, Decimal): - return format(value, "f") - return str(value).strip() - - -def _is_null(value: Any) -> bool: - if value is None: - return True - try: - return bool(pd.isna(value)) - except (TypeError, ValueError): - return False - - -def _clean_string(value: Any) -> str: - if _is_null(value): - return "" - if isinstance(value, float) and value.is_integer(): - return str(int(value)) - return str(value).strip() - - -def _nullable_string(value: Any) -> str | None: - cleaned = _clean_string(value) - return cleaned or None - - -def _nullable_int(value: Any) -> int | None: - if _is_null(value): - return None - return int(float(value)) - - -def _nullable_decimal(value: Any) -> Decimal | None: - if _is_null(value): - return None - try: - return Decimal(str(value)) - except (InvalidOperation, ValueError): - return None - - -def _nullable_bool(value: Any) -> bool | None: - if _is_null(value): - return None - if isinstance(value, bool): - return value - normalized = str(value).strip().lower() - if normalized in {"1", "1.0", "true", "t", "yes", "y"}: - return True - if normalized in {"0", "0.0", "false", "f", "no", "n"}: - return False - return None - - -def _nullable_date(value: Any) -> date | None: - timestamp = _nullable_datetime(value) - return timestamp.date() if timestamp else None - - -def _nullable_datetime(value: Any) -> datetime | None: - if _is_null(value): - return None - parsed = pd.to_datetime(value, errors="coerce", utc=False) - if pd.isna(parsed): - return None - if isinstance(parsed, pd.Timestamp): - return parsed.to_pydatetime().replace(tzinfo=None) - if isinstance(parsed, datetime): - return parsed.replace(tzinfo=None) - return None diff --git a/warehouse/agentflow/dv2/loaders/x5_retail_hero/requirements.txt b/warehouse/agentflow/dv2/loaders/x5_retail_hero/requirements.txt deleted file mode 100644 index 7e9da27b..00000000 --- a/warehouse/agentflow/dv2/loaders/x5_retail_hero/requirements.txt +++ /dev/null @@ -1,6 +0,0 @@ -clickhouse-driver>=0.2.9 -psycopg[binary]>=3.1 -pandas>=2.0 -click>=8.1 -pydantic>=2.9 -tqdm>=4.66.3 diff --git a/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__ala.sql b/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__ala.sql index ea304dc9..223eaf20 100644 --- a/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__ala.sql +++ b/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__ala.sql @@ -1,6 +1,6 @@ /* -Purpose: Order header from 1C / X5 Retail Hero ingest. -Sources: 1c (X5 Retail Hero). +Purpose: Order header from 1C / branch order feed ingest. +Sources: 1c (branch order feed). Branch context: per-branch order state. SCD2: yes. Anonymization: hot. diff --git a/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__dxb.sql b/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__dxb.sql index 4e864a50..c98fd451 100644 --- a/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__dxb.sql +++ b/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__dxb.sql @@ -1,6 +1,6 @@ /* -Purpose: Order header from 1C / X5 Retail Hero ingest. -Sources: 1c (X5 Retail Hero). +Purpose: Order header from 1C / branch order feed ingest. +Sources: 1c (branch order feed). Branch context: per-branch order state. SCD2: yes. Anonymization: hot. diff --git a/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__ekb.sql b/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__ekb.sql index 4269c35e..539272d2 100644 --- a/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__ekb.sql +++ b/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__ekb.sql @@ -1,6 +1,6 @@ /* -Purpose: Order header from 1C / X5 Retail Hero ingest. -Sources: 1c (X5 Retail Hero). +Purpose: Order header from 1C / branch order feed ingest. +Sources: 1c (branch order feed). Branch context: per-branch order state. SCD2: yes. Anonymization: hot. diff --git a/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__msk.sql b/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__msk.sql index 4b9e9cbd..76c62d07 100644 --- a/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__msk.sql +++ b/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__msk.sql @@ -1,6 +1,6 @@ /* -Purpose: Order header from 1C / X5 Retail Hero ingest. -Sources: 1c (X5 Retail Hero). +Purpose: Order header from 1C / branch order feed ingest. +Sources: 1c (branch order feed). Branch context: per-branch order state. SCD2: yes. Anonymization: hot. diff --git a/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__spb.sql b/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__spb.sql index fa2e06df..9a302d24 100644 --- a/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__spb.sql +++ b/warehouse/agentflow/dv2/postgres/satellites/sat_order_header__1c__spb.sql @@ -1,6 +1,6 @@ /* -Purpose: Order header from 1C / X5 Retail Hero ingest. -Sources: 1c (X5 Retail Hero). +Purpose: Order header from 1C / branch order feed ingest. +Sources: 1c (branch order feed). Branch context: per-branch order state. SCD2: yes. Anonymization: hot. diff --git a/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__ala.sql b/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__ala.sql index eac9ab7e..5c87d833 100644 --- a/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__ala.sql +++ b/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__ala.sql @@ -1,6 +1,6 @@ /* -Purpose: Order header from 1C / X5 Retail Hero ingest. -Sources: 1c (X5 Retail Hero). +Purpose: Order header from 1C / branch order feed ingest. +Sources: 1c (branch order feed). Branch context: per-branch order state. SCD2: yes. Anonymization: hot. diff --git a/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__dxb.sql b/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__dxb.sql index 68dbf99c..eb176bae 100644 --- a/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__dxb.sql +++ b/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__dxb.sql @@ -1,6 +1,6 @@ /* -Purpose: Order header from 1C / X5 Retail Hero ingest. -Sources: 1c (X5 Retail Hero). +Purpose: Order header from 1C / branch order feed ingest. +Sources: 1c (branch order feed). Branch context: per-branch order state. SCD2: yes. Anonymization: hot. diff --git a/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__ekb.sql b/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__ekb.sql index 7f6ca14c..917b7768 100644 --- a/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__ekb.sql +++ b/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__ekb.sql @@ -1,6 +1,6 @@ /* -Purpose: Order header from 1C / X5 Retail Hero ingest. -Sources: 1c (X5 Retail Hero). +Purpose: Order header from 1C / branch order feed ingest. +Sources: 1c (branch order feed). Branch context: per-branch order state. SCD2: yes. Anonymization: hot. diff --git a/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__msk.sql b/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__msk.sql index 11e42cc1..524184cd 100644 --- a/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__msk.sql +++ b/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__msk.sql @@ -1,6 +1,6 @@ /* -Purpose: Order header from 1C / X5 Retail Hero ingest. -Sources: 1c (X5 Retail Hero). +Purpose: Order header from 1C / branch order feed ingest. +Sources: 1c (branch order feed). Branch context: per-branch order state. SCD2: yes. Anonymization: hot. diff --git a/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__spb.sql b/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__spb.sql index ebdae64f..fd0f22bf 100644 --- a/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__spb.sql +++ b/warehouse/agentflow/dv2/raw_vault/satellites/sat_order_header__1c__spb.sql @@ -1,6 +1,6 @@ /* -Purpose: Order header from 1C / X5 Retail Hero ingest. -Sources: 1c (X5 Retail Hero). +Purpose: Order header from 1C / branch order feed ingest. +Sources: 1c (branch order feed). Branch context: per-branch order state. SCD2: yes. Anonymization: hot. diff --git a/warehouse/agentflow/dv2/spec.yaml b/warehouse/agentflow/dv2/spec.yaml index cca2b916..d8db39d8 100644 --- a/warehouse/agentflow/dv2/spec.yaml +++ b/warehouse/agentflow/dv2/spec.yaml @@ -739,8 +739,8 @@ satellites: source_system: 1c branch_code: msk record_source: 1c__msk - purpose: Order header from 1C / X5 Retail Hero ingest. - sources: ["1c (X5 Retail Hero)"] + purpose: Order header from 1C / branch order feed ingest. + sources: ["1c (branch order feed)"] branch_context: per-branch order state scd2: true anonymization: hot @@ -755,8 +755,8 @@ satellites: source_system: 1c branch_code: spb record_source: 1c__spb - purpose: Order header from 1C / X5 Retail Hero ingest. - sources: ["1c (X5 Retail Hero)"] + purpose: Order header from 1C / branch order feed ingest. + sources: ["1c (branch order feed)"] branch_context: per-branch order state scd2: true anonymization: hot @@ -771,8 +771,8 @@ satellites: source_system: 1c branch_code: ekb record_source: 1c__ekb - purpose: Order header from 1C / X5 Retail Hero ingest. - sources: ["1c (X5 Retail Hero)"] + purpose: Order header from 1C / branch order feed ingest. + sources: ["1c (branch order feed)"] branch_context: per-branch order state scd2: true anonymization: hot @@ -787,8 +787,8 @@ satellites: source_system: 1c branch_code: dxb record_source: 1c__dxb - purpose: Order header from 1C / X5 Retail Hero ingest. - sources: ["1c (X5 Retail Hero)"] + purpose: Order header from 1C / branch order feed ingest. + sources: ["1c (branch order feed)"] branch_context: per-branch order state scd2: true anonymization: hot @@ -803,8 +803,8 @@ satellites: source_system: 1c branch_code: ala record_source: 1c__ala - purpose: Order header from 1C / X5 Retail Hero ingest. - sources: ["1c (X5 Retail Hero)"] + purpose: Order header from 1C / branch order feed ingest. + sources: ["1c (branch order feed)"] branch_context: per-branch order state scd2: true anonymization: hot From feacf5a6484ce806a4834478726e21adfb7198be Mon Sep 17 00:00:00 2001 From: JuliaEdom Date: Mon, 6 Jul 2026 01:52:25 +0300 Subject: [PATCH 2/4] =?UTF-8?q?docs(g2):=20S2b=20=E2=80=94=20de-brand=20re?= =?UTF-8?q?maining=20X5/at-scale=20references=20repo-wide?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remove all remaining "X5" / "Retail Hero" / retired at-scale row-count mentions (45.8M, 8.06M, 402K) outside CHANGELOG.md, per the project owner's override that the dataset must not appear anywhere else in the project. Load-test harness comments (run-bench.sh, job.yaml) and README get the ready-made de-branded rationale text from x5-benchmark-decision.md §3; docs/dv2-multi-branch/load-test-baseline.md is rewritten entirely with the seed-scale baseline + a "Retired at-scale capture" section per §4 (full report stays in this file's git history). Overlap de-branding beyond the load-test scope: - docs/domain.md §5.4, docs/generator-spec.md: reword the x5__ -> mp__ record_source rename entry and GS1 docstring note without the literal Kaggle/X5 strings. - docs/dv2-multi-branch/schema_dv2.md: drop the "X5 Retail Hero / Lenta BigTarget" naming-convention credit and rewrite the source-data table to point at the synthetic seed / mp__ feed instead of external Kaggle data. - docs/dv2-multi-branch/demo_evidence.md: "X5-era" flags -> "retired-seed-era"; de-brand the legend-reset paragraph and the retired-capture callouts while keeping their meaning. - docs/operations/{aws-oidc-setup,openssf-security-posture}.md, docs/perf/vault-pii-governance-pg-verify-2026-07-0{2,3}.md: reword stray "X5" mentions in ops/perf docs the original audit's file list missed. - infrastructure/dv2/clickhouse-sts.yaml, infrastructure/dv2/dbt/dbt-run-job.yaml, warehouse/agentflow/dv2/{dbt,business_vault}/*: de-brand resource-tuning rationale comments (ClickHouse memory/merge config, dbt profile threads, the bv_order_canonical_mat staged loader) the same way as the load-test comments — keep the engineering rationale, lose the brand and exact at-scale row counts. - warehouse/agentflow/dv2/{README.md,loaders/pg_vault_writer.py, postgres_oltp/README.md,reference/{README.md,load_postgres.py,vault_mapping.py}}: reword Python/README references to "the X5 loader" now that it is deleted — point instead at the still-live reference loader / vault_rows models / "the other DV2 loaders", and describe vault_mapping's MD5 convention on its own terms instead of by comparison to the removed loader. Dependency discoveries beyond the S2a decision doc's file list: dv2/README.md's hash-key gotcha note, generator-spec.md's stale gs1.py docstring pointer, clickhouse-sts.yaml's benchmark-tuning comments, dbt-run-job.yaml's thread comment, postgres_oltp/README.md and reference/README.md's loader mentions, and the two dated pg-verify perf transcripts — none were in the original audit's or S2a's file list. --- docs/domain.md | 2 +- docs/dv2-multi-branch/demo_evidence.md | 44 +++--- docs/dv2-multi-branch/load-test-baseline.md | 132 +++++++----------- docs/dv2-multi-branch/schema_dv2.md | 21 +-- docs/generator-spec.md | 5 +- docs/operations/aws-oidc-setup.md | 2 +- docs/operations/openssf-security-posture.md | 9 +- ...ult-pii-governance-pg-verify-2026-07-02.md | 5 +- ...ult-pii-governance-pg-verify-2026-07-03.md | 5 +- infrastructure/dv2/clickhouse-sts.yaml | 25 ++-- infrastructure/dv2/dbt/dbt-run-job.yaml | 5 +- infrastructure/dv2/load-test/README.md | 12 +- infrastructure/dv2/load-test/job.yaml | 14 +- infrastructure/dv2/load-test/run-bench.sh | 2 +- warehouse/agentflow/dv2/README.md | 2 +- .../business_vault/bv_order_canonical_mat.sql | 11 +- .../load_bv_order_canonical_mat.sh | 4 +- warehouse/agentflow/dv2/dbt/README.md | 2 +- .../agentflow/dv2/dbt/models/sources.yml | 2 +- .../agentflow/dv2/dbt/profiles.example.yml | 4 +- .../agentflow/dv2/loaders/pg_vault_writer.py | 27 ++-- .../agentflow/dv2/postgres_oltp/README.md | 2 +- warehouse/agentflow/dv2/reference/README.md | 6 +- .../agentflow/dv2/reference/load_postgres.py | 7 +- .../agentflow/dv2/reference/vault_mapping.py | 15 +- 25 files changed, 179 insertions(+), 186 deletions(-) diff --git a/docs/domain.md b/docs/domain.md index d743b9dc..43b7f254 100644 --- a/docs/domain.md +++ b/docs/domain.md @@ -208,7 +208,7 @@ renames. The two exceptions are listed in §5.4. | Change | Scope | Status | | ------ | ----- | ------ | -| `x5__*` → `mp__*` record_source (+ governance SQL, officer probes, admission tests) | The prefix carried the name of the Kaggle seed dataset (X5 Retail Hero) that the demo loader replays as transaction history. Under the legend it is the **consolidated marketplace feed**, and the prefix says so. Dataset attribution stays in the loader README | **Done** (B2) | +| Former external-dataset record_source prefix → `mp__*` (+ governance SQL, officer probes, admission tests) | The prefix carried the name of a retired public benchmark dataset that the demo loader once replayed as transaction history. Under the legend it is the **consolidated marketplace feed**, and the prefix says so. The dataset and its loader have since been removed from the codebase entirely (see CHANGELOG.md) | **Done** (B2) | | Demo value repin: currencies to `RUB` (primary), `AED`/`KZT` in branch stories; demo revenue/counts consistent with §1–2 | `contracts/entities/order.yaml` currency examples, NL demo answers, seeded `ORD-*` rows | Planned (data phase, after the generator spec) | Vocabulary guardrails for all public docs: the company is an **own-brand / diff --git a/docs/dv2-multi-branch/demo_evidence.md b/docs/dv2-multi-branch/demo_evidence.md index 28037103..1294e87b 100644 --- a/docs/dv2-multi-branch/demo_evidence.md +++ b/docs/dv2-multi-branch/demo_evidence.md @@ -14,14 +14,15 @@ Docker: (schema → 8 hubs → 8 links → 48 satellites → `03_business_vault.sql`) → `postgres/governance/01..04.sql`. -> **Legend reset (2026-07-03).** The demo no longer models an X5 retail -> contractor. It is an own-brand ("private-label smart-kitchen") importer: -> China-manufactured goods, RU HQ, bimodal channel economy (money in -> wholesale, order-count on marketplaces), five branches across three -> jurisdictions (msk / spb / ekb + dxb + ala). All prior "X5 Retail Hero / -> 8.06M orders / USD" figures are retired. The synthetic demo seed is now -> **2,500 customers / 10,000 orders / 160 SKU / 14,853 line items, priced in -> ₽** — see `docs/domain.md` and `docs/generator-spec.md` for the model. +> **Legend reset (2026-07-03).** The demo no longer models a bulk-retail +> contractor built on a public benchmark dataset. It is an own-brand +> ("private-label smart-kitchen") importer: China-manufactured goods, RU HQ, +> bimodal channel economy (money in wholesale, order-count on marketplaces), +> five branches across three jurisdictions (msk / spb / ekb + dxb + ala). All +> prior at-scale order-count / USD figures are retired. The synthetic demo +> seed is now **2,500 customers / 10,000 orders / 160 SKU / 14,853 line +> items, priced in ₽** — see `docs/domain.md` and `docs/generator-spec.md` +> for the model. > **⚠ Infra sections (§1–3, §9–15) are pending re-capture on the kind > cluster (Mac stand).** Topology, workload pinning, PVCs, the MinIO cold @@ -30,8 +31,8 @@ Docker: > kind cluster; the standalone WSL-CH / Windows-PG split used for this > re-capture cannot reproduce cross-engine networking or Kubernetes. Their > **mechanisms are volume- and legend-independent** and stand as previously -> demonstrated, but the **row counts printed in those blocks reflect the -> retired X5-era volumes** and must not be read as current. They are flagged +> demonstrated, but the **row counts printed in those blocks reflect +> retired-seed-era volumes** and must not be read as current. They are flagged > inline and are the Mac-tail of this step (see `plan_endgame_02_07_26.md`). ## Governance verify_live — both engines green on the new seeds @@ -47,8 +48,9 @@ against the freshly-built stands: The PII boundary holds in every SQL shape on both engines; row policies scope officers to their own jurisdiction; the msk demo seed now spans `1c__msk`, -`pg_ops__msk` and `mp__msk` record sources (the `x5__` convention was retired -in B2). The current CH suite defines 29 probes (earlier revisions cited 32); +`pg_ops__msk` and `mp__msk` record sources (the legacy marketplace-seed +prefix was retired in B2). The current CH suite defines 29 probes (earlier +revisions cited 32); the count is whatever the checked-in script asserts — every probe passes. ## 1. Cluster topology — `kubectl get nodes --show-labels` @@ -114,7 +116,7 @@ sources; full generation in `spec.yaml` + `generate_satellites.py`). ## 5. Multi-branch distribution proof -The retired X5 seed spread orders 40/25/15/10/10 across branches. The current +The retired at-scale seed spread orders 40/25/15/10/10 across branches. The current legend does **not** — every marketplace and e-com order is fulfilled from the msk hub (`mp__` is msk-only), and branch identity lives in the dealer / B2B / PII layers, not in the marketplace order stream. Order distribution by @@ -157,8 +159,8 @@ branch, count() FROM rv.hub_order GROUP BY branch ORDER BY 2 DESC"`: This is the **demo-scale synthetic seed** (10k orders), not a load benchmark. The multi-million-row throughput characterisation is a separate artifact ([`load-test-baseline.md`](load-test-baseline.md)) and is not part of the -synthetic demo evidence; the earlier "1.1 s over 8.06M X5 rows" block is -retired with the X5 legend. +synthetic demo evidence; the earlier multi-million-row latency capture is +retired with the 2026-07-03 legend reset. ## 7. Line items reach @@ -248,7 +250,7 @@ verify_live matrix above). > ⚠ **Kind-cluster section — pending Mac re-capture.** MinIO + the ClickHouse > `s3()` CronJobs need the shared-network cluster; the standalone stand cannot -> reproduce them. Row counts below are X5-era and await refresh on the new +> reproduce them. Row counts below are retired-seed-era and await refresh on the new > seeds. The mechanism (native `s3()` write + read-back, PII-free source > selection) is unchanged. @@ -273,7 +275,7 @@ cloud-provider secret takes its place. > ⚠ **Kind-cluster section — pending Mac re-capture.** The bridge needs > ClickHouse and Postgres on one network; the standalone split (CH in WSL, > Postgres on the Windows host loopback) cannot reach across. Row counts and -> the `Dasha/Egor/Fedor`-style names below are X5-era and await refresh on the +> the `Dasha/Egor/Fedor`-style names below are retired-seed-era and await refresh on the > new kitchen-legend seeds. The code path (Postgres → CH `Engine=PostgreSQL()` > live read-through → raw_vault → business_vault) is unchanged. @@ -315,7 +317,7 @@ bash infrastructure/dv2/bootstrap.sh # idempotent rebuild on the kind cluster ## 12. Argo Workflows orchestration > ⚠ **Kind-cluster section — pending Mac re-capture.** Timings/counts below are -> X5-era. DAG ordering (hub → link → satellite → cold-offload) is enforced by +> retired-seed-era. DAG ordering (hub → link → satellite → cold-offload) is enforced by > dependencies, not clock-time; that property is legend-independent. `infrastructure/dv2/argo/` deploys Argo Workflows plus a `dv2-refresh` @@ -332,7 +334,7 @@ are never out of sync with the warm tier. ## 13. dbt mart layer > ⚠ **Kind-cluster section — pending Mac re-capture.** The dbt-in-Kubernetes -> Job and its per-branch row counts are X5-era. The three marts + 12 tests are +> Job and its per-branch row counts are retired-seed-era. The three marts + 12 tests are > legend-independent in structure; the numbers await refresh on the new seeds. `warehouse/agentflow/dv2/dbt/` ships three materialized marts and 12 data @@ -347,7 +349,7 @@ mart) — the same 12/5/20% rates verified live in §8. > ⚠ **Kind-cluster section — pending Mac re-capture.** MaterializedPostgreSQL > consumes the Postgres WAL via logical replication and needs both engines on > one network with `wal_level=logical`; the standalone split cannot reproduce -> it. Contents are X5-era. +> it. Contents are retired-seed-era. The pull-based `oltp_live` bridge is replaced by a single `oltp_cdc` ClickHouse database backed by `MaterializedPostgreSQL`, consuming the Postgres @@ -359,7 +361,7 @@ rows in raw_vault. ## 15. Per-branch CDC fan-out -> ⚠ **Kind-cluster section — pending Mac re-capture.** Contents are X5-era. +> ⚠ **Kind-cluster section — pending Mac re-capture.** Contents are retired-seed-era. Operational reality wants a single branch to be pausable, re-snapshotable and rotatable without touching another branch's stream. ClickHouse 25.5+ rejects a diff --git a/docs/dv2-multi-branch/load-test-baseline.md b/docs/dv2-multi-branch/load-test-baseline.md index 34146ee7..3930950b 100644 --- a/docs/dv2-multi-branch/load-test-baseline.md +++ b/docs/dv2-multi-branch/load-test-baseline.md @@ -1,90 +1,22 @@ # DV2.0 ClickHouse load test — baseline -Captured **2026-06-07** against the `hq-demo` kind cluster on the iMac demo -host, **real X5 Retail Hero data**: 8.06M orders / 45.8M line items across 5 -branches (`rv.*` raw vault ≈ 3.5 GiB on disk, `rv.bv_order_canonical_mat` -≈ 538 MiB). Harness: [`infrastructure/dv2/load-test/`](../../infrastructure/dv2/load-test/) -— `clickhouse-benchmark` run as a Kubernetes Job in namespace `dv2`, driving +Captured **2026-06-02** against the `hq-demo` kind cluster on the iMac demo +host, **synthetic demo seed** (~10K orders across 5 branches). Harness: +[`infrastructure/dv2/load-test/`](../../infrastructure/dv2/load-test/) — +`clickhouse-benchmark` run as a Kubernetes Job in namespace `dv2`, driving the in-cluster `clickhouse` service over the native protocol (9000). -VM at capture time: 6 GiB RAM, 2 vCPU; ClickHouse pod limit 5 Gi. Server-level -memory accounting is disabled on this stack (see `clickhouse-sts.yaml` — the -2026-06-07 saga); the per-query cap and the cgroup limit guard the pod. -Serving scenarios sweep concurrency 1 / 4 / 8 × 60 iterations; raw-vault -`adhoc` scenarios run **c=1 × 2 iterations** — at X5 a single recompute runs -for minutes (below), so a concurrency sweep would measure nothing but a queue. +VM at capture time: 6 GiB RAM, 2 vCPU; ClickHouse pod limit 5 Gi. Budgets at +capture time: point p99 ≤ 200 ms (since moved to 250 ms — see finding 3 +below), heavy p99 ≤ 1000 ms, adhoc p99 ≤ 2000 ms (adhoc is informational — +reported, never gates). -## Results (X5, 2026-06-07 — final gating run) +> **Refresh pending.** This capture predates the 2026-07-03 legend reset +> (kitchen-appliance wholesaler seed). Re-run `apply.sh` on the Mac demo +> host against the current seed and replace the table below; harness +> mechanics and budgets are unchanged. -| Scenario | Class | c=1 p99 | c=4 p99 | c=8 p99 | c=8 QPS | Verdict | -|----------|-------|--------:|--------:|--------:|--------:|---------| -| `01_branch_pnl_adhoc` — live `bv_order_canonical` recompute | adhoc | **238 s** | — | — | — | INFO | -| `02_top_products_adhoc` — `lnk_order_product` ⋈ `hub_order` top-N | adhoc | **102 s** | — | — | — | INFO | -| `03_customer360_point` — single-row mart lookup by `customer_bk` | point | 42 ms | 100 ms | 197 ms | 44.3 | PASS | -| `04_returns_velocity` — mart aggregation | heavy | 108 ms | 92 ms | 141 ms | 62* | PASS | -| `05_line_items_reach_adhoc` — 45.8M-row join + `uniq` | adhoc | **294 s** | — | — | — | INFO | -| `06_branch_pnl_mart` — **materialized** `marts.branch_pnl` | heavy | 20 ms | 72 ms | 293 ms | 41.3 | PASS | - -Budgets: point p99 ≤ 250 ms, heavy p99 ≤ 1000 ms, adhoc p99 ≤ 2000 ms (adhoc -is informational — reported, never gates). `LOAD TEST: PASS`. - -\* between-run variance on this shared 2-vCPU host is ~2× on sibling cells -(e.g. 04 c=8 p99 ranged 141→396 ms across four captures the same day); single -cells are indicative, the gate verdicts were stable across runs once the real -findings below were fixed. - -## What the X5 run surfaced (three real findings) - -**1. `customer_360` had the wrong sort key for its serving pattern.** The -mart was `ORDER BY (branch, customer_hk)`, but its point query — and the -realistic serving access — looks up by **`customer_bk`**, which was not in -the key at all: every lookup full-scanned the 402K-row mart. At the synthetic -800 rows this was invisible; at X5 it was p99 250–468 ms vs the 200 ms point -budget. Fixed in the dbt model: `ORDER BY (customer_bk, branch)` + -`index_granularity 1024` (a point lookup now reads one ~1K-row granule) -→ p99 42 / 100 / 197 ms at c=1/4/8. - -**2. Exact distinct at X5 is not an interactive query on this host.** -`05_line_items_reach` used `uniqExact(order_hk)` over 45.8M rows — the state -holds every one of 8M keys and needs > 3 GiB even with spill; it OOM'd its -per-query cap at every setting tried. Switched to `uniq()` (HyperLogLog, -~KB per group, ~1% error): completes in ~5 min as a capacity reference. -The exact-count use case belongs in a mart or an offline job. - -**3. The point budget is a queueing budget at c=8 on 2 vCPU.** With the sort -key fixed, c=1 p99 is 42 ms but c=8 p50 is ~127 ms with QPS plateaued from -c=4 — past the box's concurrency design point, p99 measures the queue, not -the data path. Budget moved 200 → 250 ms with the rationale recorded in -`load-test/job.yaml`; the pre-fix full-scan regression (276–468 ms at c=1/4) -would still fail it. - -## Architecture takeaway — the X5 numbers ARE the pitch - -The same business question (branch P&L): - -- **materialized mart** (`06`): p99 **20 ms** at c=1, still ≤ 293 ms with 8 - concurrent clients; -- **live raw-vault recompute** (`01`): **238 seconds** for a single client — - a ~10,000× gap, and it only runs at all with hand-tuned spill settings. - -At the 10K synthetic seed this gap was 3–9×; real volume widened it by three -orders of magnitude. Materialization is the serving path; the live view is an -exploration/debug convenience. The speed comes from **materialization**, not -from dbt: `marts.*` are physical `MergeTree` tables sorted for their access -pattern, the expensive UNION ALL + argMax SCD2 collapse + joins are paid once -at build time — by the staged loader for `bv_order_canonical_mat` -(`warehouse/agentflow/dv2/business_vault/load_bv_order_canonical_mat.sh`) and -by dbt for the marts (dependency DAG, tests, docs). dbt contributes -reproducibility and testability; the latency comes from the physical layout. - -The mart build itself tells the same story: against the live view the three -dbt marts OOM'd four runs in a row at X5; against the materialized canonical -they build in **4–14 seconds each** with 12/12 tests green on the same 5Gi -pod. - -## Previous capture (2026-06-02, synthetic 10K seed) - -Kept for contrast; budgets then: point ≤ 200 ms. +## Results (synthetic seed, 2026-06-02) | Scenario | c=1 p99 | c=4 p99 | c=8 p99 | c=8 QPS | Verdict | |----------|--------:|--------:|--------:|--------:|---------| @@ -95,9 +27,43 @@ Kept for contrast; budgets then: point ≤ 200 ms. | `05_line_items_reach` | 275 ms | 397 ms | 310 ms | 45.8 | PASS | | `06_branch_pnl_mart` | 61 ms | 68 ms | 97 ms | 91.7 | PASS | -That capture's two findings (pod memory limit was the ceiling, not the VM; +Scenarios `02` and `05` have since been reclassified as informational +`*_adhoc` scenarios (c=1 only); the next capture will report them that way. +This capture's two findings (pod memory limit was the ceiling, not the VM; the live 5-join view needs frugal join settings under load) are preserved in -git history and remain true — the X5 capture extends both. +git history. + +Even at seed scale the mart-vs-live-view boundary is measurable: the same +business question (branch P&L) answers 3–9× faster off the materialized +`marts.branch_pnl` (`06`) than through the live `rv.bv_order_canonical` +recompute (`01`), and the gap widens superlinearly with volume. The +materialized mart is the serving path; the live view is an exploration/debug +convenience. + +## Retired at-scale capture (2026-06-07) + +A one-off capture against a bulk public seed dataset (tens of millions of +raw-vault rows; dataset and loader retired with the 2026-07-03 legend reset) +ran on this same harness. The full report lives in the git history of this +file. Its three engineering findings are permanent and survive in the +current code: + +1. **`customer_360` had the wrong sort key for its serving pattern.** The + mart was `ORDER BY (branch, customer_hk)` while its point query looks up + by `customer_bk` — invisible at seed scale, a full mart scan at volume + (p99 250–468 ms vs the then-200 ms budget). Fixed in the dbt model: + `ORDER BY (customer_bk, branch)` + `index_granularity 1024` (a point + lookup reads one ~1K-row granule). +2. **Exact distinct is not an interactive query at volume.** + `05_line_items_reach` originally used `uniqExact(order_hk)`; the state + OOM'd its per-query cap at every setting tried. Switched to `uniq()` + (HyperLogLog, ~KB per group, ~1 % error). The exact-count use case + belongs in a mart or an offline job. +3. **The point budget is a queueing budget at c=8 on 2 vCPU.** Past the + box's concurrency design point, p99 measures the queue, not the data + path. The point budget moved 200 → 250 ms with the rationale recorded in + `load-test/job.yaml`; the pre-fix full-scan regression (276–468 ms) + would still fail it. ## Re-run / refresh diff --git a/docs/dv2-multi-branch/schema_dv2.md b/docs/dv2-multi-branch/schema_dv2.md index 385f9031..e97410df 100644 --- a/docs/dv2-multi-branch/schema_dv2.md +++ b/docs/dv2-multi-branch/schema_dv2.md @@ -4,7 +4,7 @@ Multi-source / multi-branch DWH own-brand импортёра малой бытовой техники (легенда: [`domain.md`](../domain.md)). 5 локаций (MSK / SPB / EKB / DXB / ALA). Источники: 1С (УТ/ЗУП), Битрикс24, WMS, Excel-логистика (контейнерные манифесты), собственный D2C-сайт, API маркетплейсов (ВБ/Озон). -Naming convention основана на публичных данных и engineering-материалах крупных ритейлеров (X5 Retail Hero, Lenta BigTarget, Магнит). DDL-шаблоны — адаптация Celestinfo + Tampere DV2.0 automation patterns под ClickHouse. +Naming convention основана на публичных engineering-материалах крупных ритейлеров (retail Data Vault case studies). DDL-шаблоны — адаптация Celestinfo + Tampere DV2.0 automation patterns под ClickHouse. ## Naming conventions @@ -178,15 +178,20 @@ Idempotent re-loading через `hash_diff` — если satellite-payload не - **PIT tables:** для критичных pull-запросов с фиксированной датой - **Effectivity satellites:** open/close intervals для SCD2-tracking на links -## Связь с реальными source-data +## Связь с текущими source-data -| Hub/Sat | Прототип-данные | Where to get | +| Hub/Sat | Данные | Где взять | |---|---|---| -| `hub_customer` + `sat_customer_personal__*` | X5 Retail Hero `clients.csv` | Kaggle | -| `hub_product` + `sat_product_catalog__*` | X5 Retail Hero `products.csv` | Kaggle | -| `lnk_order_customer` + `sat_lnk_order_product__*` | X5 Retail Hero `purchases.csv` (45.8 млн строк) | Kaggle | -| `sat_customer_loyalty__*` | Lenta BigTarget loyalty features | Kaggle | +| `hub_customer` + `sat_customer_personal__*` | Синтетический demo seed (customers) | `synthetic_seed.sql` | +| `hub_product` + `sat_product_catalog__*` | Синтетический demo seed (catalog) | `synthetic_seed.sql` | +| `lnk_order_customer` + `sat_lnk_order_product__*` | Консолидированный marketplace-фид (`mp__`) + per-branch order feed (`1c__`) | `satellite_seed_all_branches.sql` + kitchen live generator (`src/ingestion/`) | +| `sat_customer_loyalty__*` | Синтетический seed (dealer retro-bonus state, легенда) | `satellite_seed_all_branches.sql` | | `sat_order_marketplace__wb__*` | Synthesize from public ВБ API docs | Wildberries | | `hub_marking_code` | Честный Знак GS1 examples | crpt.ru | -Для PoC демо тащим X5 Retail Hero (45М транзакций) в ClickHouse через AgentFlow Argo Workflow, маппим в DV2.0 hub/link/sat, добавляем synthetic branches (привязка к store_id из X5 → искусственно распределяем по 5 локациям). +Демо-объём фиксирован на масштабе легенды (docs/generator-spec.md §1): ~10K +заказов / 2 500 клиентов / 160 SKU, без внешних bulk-датасетов. Raw vault +загружается синтетическим seed (`synthetic_seed.sql` + `satellite_seed_all_branches.sql`), +живым kitchen-генератором (`src/ingestion/`) и `reference/load_postgres.py`; +маппинг в DV2.0 hub/link/sat остаётся тем же (per-branch satellites, синтетические +branches по 5 локациям). diff --git a/docs/generator-spec.md b/docs/generator-spec.md index 3b52464d..f15ea18c 100644 --- a/docs/generator-spec.md +++ b/docs/generator-spec.md @@ -12,7 +12,7 @@ Consumers, in execution order: (generator.py, tnved.py; gs1.py is unchanged), `synthetic_seed.sql`, `satellite_seed*.sql`, `postgres_oltp/seed.sql`. Faux-PII mechanics are preserved (§8). -2. **Record-source rename** — `x5__*` → `mp__*` (see domain.md §5.4). +2. **Record-source rename** — retired external-dataset prefix → `mp__*` (see domain.md §5.4). 3. **Serving demo repin** — the four demo tables, NL demo answers, `ORD-*` values (§9). 4. **Evidence regeneration** — demo_evidence and live-verify counts re-pinned @@ -184,8 +184,7 @@ gram weights, food ТН ВЭД headings) — it is replaced wholesale: - **GS1 stays exactly as-is** (`gs1.py` untouched): the EAEU prefix range 460–469 is *correct* for an own-brand importer — GTINs belong to the RU brand owner registered with GS1 RUS, regardless of where manufacturing - happens. Record this rationale in the module docstring when touching the - package (its current text says "X5-grocery context"). + happens. The module docstring already records this rationale. - **`tnved.py`**: grocery headings → the four appliance headings of §3 (8516, 8509, 8423, 8422) with RU descriptions close to official wording, heading-granularity honesty note preserved. diff --git a/docs/operations/aws-oidc-setup.md b/docs/operations/aws-oidc-setup.md index cb9b2e95..547102ca 100644 --- a/docs/operations/aws-oidc-setup.md +++ b/docs/operations/aws-oidc-setup.md @@ -6,7 +6,7 @@ This archived optional guide bootstraps the AWS IAM OIDC provider and the GitHub Current project decision as of 2026-05-30: a managed-AWS / Terraform-apply production deployment is out of scope for this pre-production portfolio project — a deliberate non-goal, with the stack validated end-to-end on a local/kind demo instead. Do not treat missing AWS apply evidence as a project deficiency, active blocker, or autonomous follow-up. Reopen this path only if the operator explicitly provides an AWS account and approval to reintroduce it. -For the DV2/X5 demo, use the already documented S3-compatible cold-tier path with HF Datasets or Backblaze B2 for derived/anonymized parquet. Do not require AWS/S3 for that dataset. +For the DV2 demo, use the already documented S3-compatible cold-tier path with HF Datasets or Backblaze B2 for derived/anonymized parquet. Do not require AWS/S3 for that dataset. ## Current readiness handoff diff --git a/docs/operations/openssf-security-posture.md b/docs/operations/openssf-security-posture.md index 8662fde9..2c68682f 100644 --- a/docs/operations/openssf-security-posture.md +++ b/docs/operations/openssf-security-posture.md @@ -151,10 +151,11 @@ Pinned-Dependencies rose 0 → 4 (the pip subset is what's left). `container-attestation.yml` moved its `packages`/`id-token`/`attestations` writes from the workflow level down to the two operator-dispatched signing jobs (the every-PR `build-smoke` job is read-only now). -- **Vulnerabilities (2 OSV hits):** both came from resolution floors in the - X5 demo loader's `requirements.txt` (`pydantic>=2.0` admits the - GHSA-mr82-8j83-vxmv ReDoS range; `tqdm>=4.0` admits PYSEC-2017-74). Floors - raised to `pydantic>=2.9` / `tqdm>=4.66.3`. +- **Vulnerabilities (2 OSV hits):** both came from resolution floors in a + demo loader's `requirements.txt` (`pydantic>=2.0` admitted the + GHSA-mr82-8j83-vxmv ReDoS range; `tqdm>=4.0` admitted PYSEC-2017-74). + Floors were raised to `pydantic>=2.9` / `tqdm>=4.66.3`; the loader itself + was later removed from the codebase entirely (see CHANGELOG.md). ### Accepted-open (with reasons, not fabricated away) diff --git a/docs/perf/vault-pii-governance-pg-verify-2026-07-02.md b/docs/perf/vault-pii-governance-pg-verify-2026-07-02.md index 4989fa82..4a4fb882 100644 --- a/docs/perf/vault-pii-governance-pg-verify-2026-07-02.md +++ b/docs/perf/vault-pii-governance-pg-verify-2026-07-02.md @@ -10,7 +10,8 @@ Vault built from the repo files verbatim via `postgres/apply.sh` (schema → `postgres/governance/01..04.sql`. Data: the deterministic demo seed from `verify_live.sh` (`SEED_DEMO=1`): hub_customer msk 8 / dxb 2 = 10 rows, msk rows deliberately spanning three source conventions (`1c__msk`, -`pg_ops__msk`, `x5__msk`) so the row policies are exercised against the +`pg_ops__msk`, and the marketplace-seed prefix later renamed `mp__msk`) so +the row policies are exercised against the split_part branch derivation, not a single record_source literal. Probe principals are stand-local: `analyst_probe` (role `dv2_analyst`), `officer_msk_probe`, `officer_dxb_probe`, and `noscope_probe` (SELECT on the @@ -60,7 +61,7 @@ attributes; nothing is pushed past the view boundary for attribution. | Probe | Result | | ----- | ------ | -| `officer_msk` hub count | 8 of 8 msk rows (incl. `pg_ops__msk`, `x5__msk`) | +| `officer_msk` hub count | 8 of 8 msk rows (incl. `pg_ops__msk`, marketplace-seed prefix later renamed `mp__msk`) | | `officer_dxb` hub count | 2 of 2 dxb rows | | `officer_dxb` counts msk rows via hub filter | 0 | | `officer_msk` cross-branch view / satellite | `permission denied` | diff --git a/docs/perf/vault-pii-governance-pg-verify-2026-07-03.md b/docs/perf/vault-pii-governance-pg-verify-2026-07-03.md index 35feb234..1184b9a5 100644 --- a/docs/perf/vault-pii-governance-pg-verify-2026-07-03.md +++ b/docs/perf/vault-pii-governance-pg-verify-2026-07-03.md @@ -10,8 +10,9 @@ kitchen-gadget legend. Vault built from the repo files verbatim via `03_business_vault.sql`), then `postgres/governance/01..04.sql`. Data: the deterministic demo seed from `verify_live.sh` (`SEED_DEMO=1`): hub_customer msk 8 / dxb 2 = 10 rows, the msk rows deliberately spanning three source -conventions — **`1c__msk`, `pg_ops__msk`, `mp__msk`** (the `x5__` convention -was retired in B2; this run confirms `mp__msk` is what the seed now carries) — +conventions — **`1c__msk`, `pg_ops__msk`, `mp__msk`** (the legacy +marketplace-seed prefix was retired in B2; this run confirms `mp__msk` is +what the seed now carries) — so the row policies are exercised against the `split_part` branch derivation, not a single record_source literal. Probe principals are stand-local: `analyst_probe` (role `dv2_analyst`), `officer_msk_probe`, `officer_dxb_probe`, diff --git a/infrastructure/dv2/clickhouse-sts.yaml b/infrastructure/dv2/clickhouse-sts.yaml index 38fa0ce1..5cb206d0 100644 --- a/infrastructure/dv2/clickhouse-sts.yaml +++ b/infrastructure/dv2/clickhouse-sts.yaml @@ -14,15 +14,17 @@ data: merge-throttle.xml: | + every later query (seen on the 2026-06-06 dbt mart runs, against + a retired at-scale capture — see git history of + docs/dv2-multi-branch/load-test-baseline.md). --> 268435456 67108864 268435456 -