fix(model): galaxy joins preserved, DIM_/FACT_ naming banned, keys hidden, agg defaults

Four systemic model-quality fixes from the TRI_08190447_DXO review:

- _remove_diamond_joins(): directed-reachability pruning replaces the
undirected spanning tree that capped every model at N-1 joins and
silently dropped legitimate fact->dim joins in multi-fact schemas.
Only true diamonds (two paths between the same pair) and directed
cycles are removed now.
- Naming: blueprint prompt rule 2a + _table_ident() strip DIM_/FACT_
prefixes from table names (even when the user request includes them);
deployer strips them from display names, and collisions resolve to
"Broker Key (Broker Payment)" instead of "Dim Dim Broker Key" /
"fact_Fact ..." / "_2".
- Surrogate/FK key columns get is_hidden: true in model TML (present
for joins, invisible in search/Spotter).
- _determine_column_type(): token-based matching — INVOICES_PAID is a
measure again (endswith-'ID' trap), CORPORATE no longer matches RATE,
and RATIO/PCT/DAYS_TO_* default to AVERAGE instead of SUM.

Verified against the exact TRI structure: scratch/verify_fixes.py, 25/25.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

demoprep_app/scenario/blueprint.py

@@ -213,7 +213,7 @@ class DemoBlueprint:
                         performance=float(v.get("performance", 1.0) or 1.0),
                     ))
             dimensions.append(BlueprintDimension(
-                name=_ident(d.get("name", "")),
+                name=_table_ident(d.get("name", "")),
                 description=str(d.get("description") or ""),
                 attribute_columns=[_ident(c) for c in (d.get("attribute_columns") or [])],
                 values=[v for v in values if v.name],
@@ -248,9 +248,9 @@ class DemoBlueprint:
                     per=_ident(m.get("per") or m.get("per_entity") or ""),
                     description=str(m.get("description") or ""),
                 ))
-            requested_dims = [_ident(n) for n in (f.get("dimension_names") or dim_names)]
+            requested_dims = [_table_ident(n) for n in (f.get("dimension_names") or dim_names)]
             facts.append(FactSpec(
-                name=_ident(f.get("name", "FACT")),
+                name=_table_ident(f.get("name", "FACT")),
                 grain=str(f.get("grain") or ""),
                 dimension_names=requested_dims,
                 measures=measures,
@@ -586,6 +586,22 @@ def _ident(text: Any) -> str:
     return cleaned[:60]
 
 
+def _table_ident(text: Any) -> str:
+    """Normalize a TABLE name: SQL-safe UPPER_SNAKE with no DIM_/FACT_ prefix.
+
+    We never use warehouse-style prefixes — BROKER vs LOAD_TRANSACTIONS reads
+    fine to data people, and the prefixes cascade into ugly derived column
+    names (DIM_BROKER_KEY -> "Dim Broker Key"). If the user's request named
+    tables WITH prefixes, directives matching already strips them too, so the
+    stripped blueprint still satisfies the directive.
+    """
+    name = _ident(text)
+    for prefix in ("DIM_", "FACT_"):
+        if name.startswith(prefix) and len(name) > len(prefix):
+            return name[len(prefix):]
+    return name
+
+
 def _looks_generic(name: str) -> bool:
     """Detect placeholder values like 'Product 3' or 'Category A'."""
     import re

demoprep_app/scenario/blueprint_author.py

@@ -99,11 +99,11 @@ EXAMPLE_RETAIL = {
 # about the company's actual business (selling data products), not the
 # generic operations of its industry.
 EXAMPLE_DATA_PRODUCTS = {
-    "notes": "User explicitly requested DIM_CUSTOMER_ACCOUNT, DIM_DATA_PRODUCT, FACT_PRODUCT_USAGE — all present verbatim.",
+    "notes": "User explicitly requested DIM_CUSTOMER_ACCOUNT, DIM_DATA_PRODUCT, FACT_PRODUCT_USAGE — all honored, with the DIM_/FACT_ prefixes dropped per our naming convention.",
     "date": {"grain": "month", "months_of_history": 24, "include_date_dimension": False},
     "dimensions": [
         {
-            "name": "DIM_CUSTOMER_ACCOUNT",
+            "name": "CUSTOMER_ACCOUNT",
             "attribute_columns": ["CUSTOMER_TYPE", "CONTRACT_TIER"],
             "values": [
                 {"name": "Meridian Pharma Insights", "attributes": {"CUSTOMER_TYPE": "Pharmaceutical", "CONTRACT_TIER": "Enterprise"}, "performance": 1.5},
@@ -113,7 +113,7 @@ EXAMPLE_DATA_PRODUCTS = {
             ],
         },
         {
-            "name": "DIM_DATA_PRODUCT",
+            "name": "DATA_PRODUCT",
             "attribute_columns": ["PRODUCT_FAMILY"],
             "values": [
                 {"name": "Open Claims Feed", "attributes": {"PRODUCT_FAMILY": "Patient Data"}, "mix_weight": 1.6},
@@ -125,9 +125,9 @@ EXAMPLE_DATA_PRODUCTS = {
     ],
     "facts": [
         {
-            "name": "FACT_PRODUCT_USAGE",
+            "name": "PRODUCT_USAGE",
             "grain": "one row per customer per data product per month",
-            "dimension_names": ["DIM_CUSTOMER_ACCOUNT", "DIM_DATA_PRODUCT"],
+            "dimension_names": ["CUSTOMER_ACCOUNT", "DATA_PRODUCT"],
             "date_column": "MONTH_DATE",
             "measures": [
                 {"name": "QUERIES_RUN", "kind": "base", "distribution": "lognormal", "params": {"mean": 4200, "sigma": 0.7}, "fmt": "int"},
@@ -147,7 +147,7 @@ EXAMPLE_DATA_PRODUCTS = {
             "id": "nlq_adoption",
             "headline": "Natural-language querying took off after the embedded relaunch",
             "insight_type": "trend_break",
-            "dimension": "DIM_CUSTOMER_ACCOUNT", "dimension_value": "Meridian Pharma Insights",
+            "dimension": "CUSTOMER_ACCOUNT", "dimension_value": "Meridian Pharma Insights",
             "measure": "NL_QUESTIONS", "magnitude": 1.6, "window_start": "2026-01-01",
             "spotter_question": "Show natural language questions by customer account by month",
             "expected_finding": "Meridian's NLQ volume breaks upward from January 2026",
@@ -198,6 +198,10 @@ and grounded in this company's actual world.
 2. Star schema: 2-6 dimensions and 1-3 fact tables. Multiple facts may share dimensions
    (e.g. a usage fact and a journey fact both keyed to customers and products). Every
    measure name must be unique across all facts.
+2a. Table names are plain UPPER_SNAKE business nouns — NEVER prefixed with DIM_ or FACT_.
+   BROKER vs LOAD_TRANSACTIONS reads fine to data people; the prefixes cascade into ugly
+   derived column names. If the user's request names tables with DIM_/FACT_ prefixes,
+   honor the tables but drop the prefixes (DIM_CUSTOMER_ACCOUNT -> CUSTOMER_ACCOUNT).
 3. Base measures get a distribution (lognormal/normal/uniform) with realistic params for a
    company of this size. Derived measures get arithmetic formulas over measures in the
    same fact.

sprint_2026_04.md

@@ -256,6 +256,13 @@ should tell. KPI targets, growth trends, and anomaly patterns live in the matrix
 
 ### Sprint 4 (this sprint)
 
+- [x] **Model-quality fixes from TRI_08190447_DXO review (Aug 24)** — investigated mani's Triumph model (sebe); found four systemic pipeline defects and fixed all in code (NOT applied to the existing TRI objects — that demo stays as-is per boone):
+  - **Join spanning-tree bug**: `_remove_diamond_joins()` reduced every model to N−1 joins via undirected union-find, silently dropping legitimate fact→dim joins in galaxy schemas (TRI lost FACT_BROKER_PAYMENT→DIM_BROKER and FACT_CARRIER_CAPACITY→DIM_LANE — payment metrics unsliceable by broker). Rewritten to directed-reachability: only prunes true diamonds (two paths between same pair) and directed cycles; shared conformed dims are kept ✅
+  - **DIM_/FACT_ naming banned end-to-end**: blueprint prompt rule 2a + `_table_ident()` normalization in `blueprint.py` (strips prefixes even when the user request includes them; directives matching already prefix-tolerant); Example B updated. Deployer `_strip_dim_fact_prefix()` strips prefixes from display names; conflict resolution now gives the home table the clean name ("Broker Key") and others a readable suffix ("Broker Key (Broker Payment)") — no more "Dim Dim Broker Key" / "fact_Fact …" / "_2" ✅
+  - **Key columns hidden**: surrogate PKs + FKs get `is_hidden: true` in model TML (still present for joins) ✅
+  - **Column classification**: token-based matching in `_determine_column_type()` — INVOICES_PAID no longer an ATTRIBUTE (endswith-'ID' trap), CORPORATE/GENERATED no longer match 'RATE'; RATIO/PCT/DAYS_TO_* now AVERAGE instead of SUM ✅
+  - Verified offline against the exact TRI structure: `scratch/verify_fixes.py` — 25/25 pass. Repaired TRI TMLs staged in scratch for reference only (`model_tri_fixed.tml.yml`, `lb_*_fixed.tml.yml`), not imported.
+
 - [x] **Table-create latency root cause + fix (Aug 10)** — ~250s/import was the unscoped TS connection scanning all ~507 Snowflake DBs visible to SE_ROLE (measured identical on sebe AND secloud; NOT instance load, July SRE docs corrected). Fix: (1) connection TML now scoped via required `database` property (`create_connection_tml`), (2) demos write to monthly-rotating DB `<SNOWFLAKE_DATABASE>_<YYYY_MM>` (`get_demo_database()`/`ensure_demo_database()` in `snowflake_auth.py`) since bloated DEMOBUILD (1,003 schemas) still cost 33s vs 0.7s fresh. Probe: `tests/ts_table_create_deepdive.py` ✅
 - [x] **Perf fix verified + deployed (Aug 10)** — app e2e: 655s Grade B, 0 timeouts (pre-fix record: 1,084–2,711s, 12/24 timeouts). Deployed to `hf-test` AND MCP Space `thoughtspot-demoprep/mcp` (`df44611`, rollback `3d735c7`); MCP Tixr rehearsal on new code: PASS, 9.2 min vs 16–20 min. Full research notes: `docs/build_time_deep_dive_2026_08_10.md` ✅
 - [x] **CHECKOUT_* column-eating bug found + fixed (Aug 10)** — the CREATE TABLE CHECK-constraint sanitizer (`cdw_connector._sanitize_create_table_statement`) used prefix matching, silently deleting any column named CHECK* (CHECKOUT_INITIATED, CHECKOUT_ATTEMPTS...) while the dataset still carried values → the "random" schema/data-mismatch failures (Cloudscape, Atlas Freight, Tixr 02ec3e7). Fixed with keyword+paren matching, 4 regression tests, proven end-to-end into Snowflake. Deployed to MCP Space (`bd28045`); Tixr rehearsal PASS 10.0 min ✅

thoughtspot_deployer.py

@@ -133,6 +133,19 @@ def _to_snake_case(name: str) -> str:
     return _apply_naming_style(name, "snake_case")
 
 
+def _strip_dim_fact_prefix(name: str) -> str:
+    """Drop a leading DIM_/FACT_ token from a physical name for display purposes.
+
+    We never surface warehouse-style prefixes in ThoughtSpot column names:
+    DIM_BROKER_KEY -> BROKER_KEY, FACT_LOAD_TRANSACTION_KEY -> LOAD_TRANSACTION_KEY.
+    """
+    upper = (name or "").upper()
+    for prefix in ("DIM_", "FACT_"):
+        if upper.startswith(prefix) and len(name) > len(prefix):
+            return name[len(prefix):]
+    return name
+
+
 def _infer_liveboard_context_from_custom_request(text: str) -> Tuple[str, str]:
     """Infer a coarse vertical/function label for custom liveboard question generation."""
     normalized = (text or "").lower()
@@ -1236,23 +1249,19 @@ class ThoughtSpotDeployer:
                     original_name=original_col_name
                 )
                 
-                # If the display name is still used, make it unique with table prefixes
+                # If the display name is still used, disambiguate with a readable
+                # "(Table)" suffix — never a table-name prefix
                 original_display_name = display_name
-                counter = 1
+                counter = 2
                 while display_name.lower() in used_display_names:
-                    # Use table prefix for all tables consistently
-                    if counter == 1:
-                        # Generate consistent short prefix from table name
-                        if len(table_name_upper) <= 4:
-                            prefix = table_name_upper.lower()
-                        else:
-                            prefix = table_name_upper[:4].lower()
-                        # Use snake_case: prefix_name
-                        display_name = f"{prefix}_{original_display_name}"
+                    if display_name == original_display_name:
+                        table_label = _apply_naming_style(
+                            _strip_dim_fact_prefix(table_name_upper), self.column_naming_style
+                        ) or table_name_upper
+                        display_name = f"{original_display_name} ({table_label})"
                     else:
-                        # Fallback: add number
-                        display_name = f"{original_display_name}_{counter}"
-                    counter += 1
+                        display_name = f"{original_display_name} {counter}"
+                        counter += 1
                 
                 used_display_names.add(display_name.lower())
                 
@@ -1393,56 +1402,39 @@ class ThoughtSpotDeployer:
         
         # Add columns with proper global conflict resolution (same as working version)
         used_display_names = set()
-        
+
+        # Key columns (surrogate PKs and FKs) must be IN the model for joins to
+        # work, but they're join plumbing, not demo content — mark them hidden.
+        fk_columns = set()
+        for fk in foreign_keys or []:
+            fk_columns.add((fk.get('from_table', '').upper(), fk.get('from_column', '').upper()))
+            fk_columns.add((fk.get('to_table', '').upper(), fk.get('to_column', '').upper()))
+
         for table_name, columns in tables.items():
             table_name_upper = table_name.upper()
             for col in columns:
                 col_name = col['name'].upper()
                 original_col_name = col.get('original_name', col['name'])  # Use original casing for display
 
-                # NOTE: We used to skip FK/PK columns, but ThoughtSpot requires them for joins
-                # Even though users don't search "customer 23455", the join columns must be present
-                # in the model's columns section for the joins to work properly.
-                #
-                # SKIP foreign key columns - they're join keys, not analytics columns
-                # if self._is_foreign_key_column(col_name, table_name_upper, foreign_keys):
-                #     print(f"   ⏭️  Skipping FK column: {table_name_upper}.{col_name}")
-                #     continue
-                #
-                # SKIP surrogate primary keys (numeric IDs) - nobody searches "customer 23455"
-                # if self._is_surrogate_primary_key(col, col_name):
-                #     print(f"   ⏭️  Skipping surrogate PK: {table_name_upper}.{col_name}")
-                #     continue
-
                 # Start with basic conflict resolution
                 display_name = self._resolve_column_name_conflict(
                     col_name, table_name_upper, column_name_counts,
                     original_name=original_col_name
                 )
                 
-                # If the display name is still used, make it unique with table prefixes
+                # If the display name is still used, disambiguate with a readable
+                # "(Table)" suffix — never a table-name prefix
                 original_display_name = display_name
-                counter = 1
+                counter = 2
                 while display_name.lower() in used_display_names:
-                    # Use snake_case: prefix_name
-                    if counter == 1:
-                        if table_name_upper == 'CUSTOMERS':
-                            display_name = f"cust_{original_display_name}"
-                        elif table_name_upper == 'PRODUCTS':
-                            display_name = f"prod_{original_display_name}"
-                        elif table_name_upper == 'ORDERS':
-                            display_name = f"order_{original_display_name}"
-                        elif table_name_upper == 'ORDERITEMS':
-                            display_name = f"item_{original_display_name}"
-                        elif table_name_upper == 'SALES':
-                            display_name = f"sale_{original_display_name}"
-                        elif table_name_upper == 'SALESREPS':
-                            display_name = f"rep_{original_display_name}"
-                        else:
-                            display_name = f"{table_name_upper[:4].lower()}_{original_display_name}"
+                    if display_name == original_display_name:
+                        table_label = _apply_naming_style(
+                            _strip_dim_fact_prefix(table_name_upper), self.column_naming_style
+                        ) or table_name_upper
+                        display_name = f"{original_display_name} ({table_label})"
                     else:
-                        display_name = f"{original_display_name}_{counter}"
-                    counter += 1
+                        display_name = f"{original_display_name} {counter}"
+                        counter += 1
                 
                 used_display_names.add(display_name.lower())
                 
@@ -1461,6 +1453,11 @@ class ThoughtSpotDeployer:
                 if aggregation:
                     column_def['properties']['aggregation'] = aggregation
 
+                # Hide surrogate/foreign key columns — required for joins,
+                # but nobody demos "broker key 23455"
+                if col_name.endswith('_KEY') or (table_name_upper, col_name) in fk_columns:
+                    column_def['properties']['is_hidden'] = True
+
                 # Add calendar property for DATE columns so ThoughtSpot enables
                 # time bucketing (.weekly, .monthly, etc.) on them
                 if self._map_data_type(col['type']) == 'DATE':
@@ -1487,12 +1484,19 @@ class ThoughtSpotDeployer:
         return yaml_output
     
     def _remove_diamond_joins(self, model_tables: list):
-        """Remove join cycles that ThoughtSpot rejects using a spanning tree.
-        
-        ThoughtSpot requires exactly ONE path between any two tables in a model.
-        Uses Kruskal's algorithm: add edges from highest-priority (fact) tables first,
-        skip any edge that would create a cycle. Priority order ensures fact table
-        joins are preserved and dimension-to-dimension "snowflake" joins are pruned.
+        """Remove ONLY joins that create a second directed path between two tables.
+
+        The old implementation reduced the join graph to an undirected spanning
+        tree (max N-1 joins). That silently broke galaxy schemas: two facts
+        sharing conformed dimensions is standard and ThoughtSpot supports it,
+        but the union-find pass saw an undirected cycle and dropped legitimate
+        fact->dim joins (e.g. FACT_BROKER_PAYMENT lost its DIM_BROKER join in
+        the TRI build, making every payment metric unsliceable by broker).
+
+        What ThoughtSpot actually rejects is ambiguity: two DIRECTED join paths
+        from one table to another (a diamond, e.g. F->D1->X and F->D2->X) or a
+        directed cycle. Multiple facts pointing at one shared dim is fine —
+        no table ends up with two routes to any other table.
         """
         def edge_key(src_name: str, join_def: dict):
             return (
@@ -1508,58 +1512,66 @@ class ThoughtSpotDeployer:
             src_name = t['name']
             for j in t.get('joins', []):
                 all_edges.append((src_name, j.get('with'), j, edge_key(src_name, j)))
-        
+
         if not all_edges:
             print(f"   ✅ No joins to check for cycles")
             return
-        
+
         out_degree = {}
         for t in model_tables:
             out_degree[t['name']] = len(t.get('joins', []))
-        
+
         in_degree = {t['name']: 0 for t in model_tables}
         for src, dst, _, _ in all_edges:
             in_degree[dst] = in_degree.get(dst, 0) + 1
-        
+
+        # Priority order: when a genuine diamond has to be broken, the
+        # fact-side join (high out-degree source) survives and the
+        # dim-to-dim snowflake edge is the one pruned.
         all_edges.sort(key=lambda e: (-out_degree.get(e[0], 0), -in_degree.get(e[1], 0), e[0], e[1]))
-        
-        parent = {t['name']: t['name'] for t in model_tables}
-        def find(x):
-            while parent[x] != x:
-                parent[x] = parent[parent[x]]
-                x = parent[x]
-            return x
-        def union(a, b):
-            ra, rb = find(a), find(b)
-            if ra == rb:
-                return False
-            parent[ra] = rb
-            return True
-        
+
+        nodes = {t['name'] for t in model_tables}
+        reach = {n: set() for n in nodes}    # nodes reachable from n via kept joins
+        parents = {n: set() for n in nodes}  # nodes that can reach n via kept joins
+
         kept_edge_keys = set()
         removed = []
         for src, dst, join_def, e_key in all_edges:
-            if union(src, dst):
-                kept_edge_keys.add(e_key)
-            else:
+            if src not in nodes or dst not in nodes or src == dst:
                 removed.append(f"{src}->{dst} ({join_def.get('on', '')})")
-        
+                continue
+            sources = {src} | parents[src]
+            targets = {dst} | reach[dst]
+            # Directed cycle: something downstream of dst already reaches src.
+            # Diamond: some ancestor of src already reaches some target — the
+            # new edge would give it a second path there.
+            if (sources & targets) or any(
+                t_ in reach[s_] for s_ in sources for t_ in targets
+            ):
+                removed.append(f"{src}->{dst} ({join_def.get('on', '')})")
+                continue
+            kept_edge_keys.add(e_key)
+            for s_ in sources:
+                reach[s_].update(targets)
+            for t_ in targets:
+                parents[t_].update(sources)
+
         for t in model_tables:
             if 'joins' not in t:
                 continue
             src_name = t['name']
             t['joins'] = [j for j in t['joins'] if edge_key(src_name, j) in kept_edge_keys]
-        
+
         for t in model_tables:
             if 'joins' in t and not t['joins']:
                 del t['joins']
-        
+
         if removed:
-            print(f"   🔶 Removed {len(removed)} redundant joins (cycle prevention):")
+            print(f"   🔶 Removed {len(removed)} joins that created ambiguous paths or cycles:")
             for r in removed:
                 print(f"      - {r}")
         else:
-            print(f"   ✅ No join cycles detected")
+            print(f"   ✅ No ambiguous join paths detected")
     
     def _generate_constraint_id(self) -> str:
         """Generate a constraint ID similar to ThoughtSpot's system constraints"""
@@ -1619,33 +1631,26 @@ class ThoughtSpotDeployer:
             column_name_counts: Dict tracking column name occurrences
             original_name: Original casing of column name (for proper camelCase detection)
         """
-        # Use original name if provided (preserves camelCase boundaries)
-        name_for_styling = original_name if original_name else col_name
-        
+        # Use original name if provided (preserves camelCase boundaries),
+        # and never surface DIM_/FACT_ warehouse prefixes in display names
+        name_for_styling = _strip_dim_fact_prefix(original_name if original_name else col_name)
+
         # Apply configured naming style
         styled_name = _apply_naming_style(name_for_styling, self.column_naming_style)
-        
+
         if len(column_name_counts.get(col_name, [])) <= 1:
             # No conflict - use styled name directly
             return styled_name
-        
-        # For conflicts, generate prefix dynamically from table name
-        if len(table_name) <= 4:
-            prefix = table_name.lower()
-        elif 'sales' in table_name.lower() and 'rep' in table_name.lower():
-            prefix = 'rep'  # Special case for readability
-        elif 'customer' in table_name.lower():
-            prefix = 'cust'  # Common abbreviation
-        elif 'product' in table_name.lower():
-            prefix = 'prod'  # Common abbreviation
-        elif 'order' in table_name.lower():
-            prefix = 'order'  # Common abbreviation
-        else:
-            prefix = table_name[:4].lower()  # First 4 characters
-        
-        # Apply naming style to prefix + name combination
-        prefixed_name = f"{prefix}_{styled_name}" if styled_name else prefix
-        return _apply_naming_style(prefixed_name, self.column_naming_style)
+
+        # Cross-table collision (e.g. BROKER_KEY exists on the BROKER dim and as
+        # an FK on fact tables): the column's home table keeps the clean name,
+        # every other table gets a readable "(Table)" suffix.
+        business_table = _strip_dim_fact_prefix(table_name)
+        if _strip_dim_fact_prefix(col_name).upper().startswith(business_table.upper()):
+            return styled_name
+
+        table_label = _apply_naming_style(business_table, self.column_naming_style) or business_table
+        return f"{styled_name} ({table_label})"
     
     def _get_table_prefix(self, table_name: str) -> str:
         """Get appropriate prefix for table to avoid column conflicts"""
@@ -1680,18 +1685,28 @@ class ThoughtSpotDeployer:
         
         # Numeric types should be measures (unless they're IDs or keys)
         if base_type in ['NUMBER', 'DECIMAL', 'FLOAT', 'DOUBLE', 'INT', 'INTEGER', 'BIGINT']:
-            # Skip ID/KEY columns - they're join keys, not analytics columns
-            if col_upper.endswith('ID') or col_upper.endswith('KEY') or col_upper.endswith('_CODE'):
+            # Skip ID/KEY columns - they're join keys, not analytics columns.
+            # Match whole name tokens, not substrings: a bare endswith('ID')
+            # misclassified INVOICES_PAID (and anything ending PAID/VALID/GRID)
+            # as an attribute.
+            tokens = col_upper.split('_')
+            if tokens[-1] in ('ID', 'KEY', 'CODE') or col_upper in ('ID', 'KEY'):
                 return 'ATTRIBUTE', None
-            
-            # All other numeric columns are measures
-            # Determine aggregation based on column name patterns
-            if any(word in col_upper for word in ['QUANTITY', 'QTY', 'COUNT', 'SOLD']):
+
+            # All other numeric columns are measures.
+            # Aggregation from whole-word tokens (substring matching hit
+            # CORPORATE/GENERATED for 'RATE'). Ratios, rates, percentages and
+            # per-row durations must AVERAGE — summing a rate is meaningless.
+            token_set = set(tokens)
+            if token_set & {'RATING', 'SCORE', 'MARGIN', 'PERCENT', 'PCT', 'RATE', 'RATIO', 'AVG', 'AVERAGE'}:
+                return 'MEASURE', 'AVERAGE'
+            elif 'DAYS' in token_set and ('TO' in token_set or 'SINCE' in token_set):
+                # DAYS_TO_PAY / DAYS_SINCE_X are per-row durations, not additive
+                return 'MEASURE', 'AVERAGE'
+            elif token_set & {'QUANTITY', 'QTY', 'COUNT', 'SOLD'}:
                 return 'MEASURE', 'SUM'
-            elif any(word in col_upper for word in ['PRICE', 'COST', 'REVENUE', 'AMOUNT', 'TOTAL', 'PROFIT', 'DISCOUNT', 'SHIPPING', 'TAX']):
+            elif token_set & {'PRICE', 'COST', 'REVENUE', 'AMOUNT', 'TOTAL', 'PROFIT', 'DISCOUNT', 'SHIPPING', 'TAX'}:
                 return 'MEASURE', 'SUM'
-            elif any(word in col_upper for word in ['RATING', 'SCORE', 'MARGIN', 'PERCENT', 'RATE']):
-                return 'MEASURE', 'AVERAGE'
             else:
                 # Default: numeric = measure with SUM
                 return 'MEASURE', 'SUM'