fix(gooddata): preserve grain for aliased primary-key fields

converters/gooddata/src/ossie_gooddata/ossie_to_gooddata.py

@@ -91,6 +91,10 @@ def _build_target_info(sm: dict[str, Any]) -> dict[str, dict[str, Any]]:
         if not is_date:
             for f in ds.get("fields", []):
                 src = _get_source_column(f)
+                if src in col_to_attr:
+                    raise ValueError(
+                        f"Dataset '{ds_name}': source column '{src}' maps to multiple fields."
+                    )
                 col_to_attr[src] = f"attr.{ds_name}.{f['name']}"
         info[ds_name] = {"is_date": is_date, "col_to_attr": col_to_attr}
     return info
@@ -148,31 +152,28 @@ def _convert_ossie_dataset(
     # Regular dataset
     attributes: list[GdAttribute] = []
     facts: list[GdFact] = []
-    grain_ids: list[str] = []
-
-    pk_columns = set(ds.get("primary_key", []))
 
     for field_def in fields:
-        field_name = field_def["name"]
         is_dimension = field_def.get("dimension") is not None
 
         if is_dimension:
             attr = _convert_to_attribute(field_def, ds_name)
             attributes.append(attr)
-            if field_name in pk_columns:
-                grain_ids.append(attr.id)
         else:
             # Check MAQL expression to determine if fact or attribute
             maql_type = _detect_type_from_maql(field_def)
             if maql_type == "attribute":
                 attr = _convert_to_attribute(field_def, ds_name)
                 attributes.append(attr)
-                if field_name in pk_columns:
-                    grain_ids.append(attr.id)
             else:
                 facts.append(_convert_to_fact(field_def, ds_name))
 
-    grain = [GdGrain(id=gid, type="attribute") for gid in grain_ids]
+    attribute_ids_by_column = {attr.source_column: attr.id for attr in attributes}
+    grain = [
+        GdGrain(id=attribute_ids_by_column[column], type="attribute")
+        for column in ds.get("primary_key", [])
+        if column in attribute_ids_by_column
+    ]
 
     # Convert relationships from this dataset to GoodData references
     references = []

converters/gooddata/tests/test_ossie_to_gooddata.py

@@ -338,6 +338,98 @@ def test_grain_from_primary_key(ossie_tpcds_dict: dict):
     assert len(grain_ids) == 2
 
 
+@pytest.mark.parametrize(
+    "field",
+    [
+        {
+            "name": "customer_key",
+            "expression": {"dialects": [{"dialect": "ANSI_SQL", "expression": "customer_id"}]},
+            "dimension": {},
+        },
+        {
+            "name": "customer_key",
+            "expression": {
+                "dialects": [
+                    {"dialect": "ANSI_SQL", "expression": "customer_id"},
+                    {"dialect": "MAQL", "expression": "{label/customers.customer_key}"},
+                ]
+            },
+        },
+    ],
+    ids=["dimension", "maql-attribute"],
+)
+def test_grain_uses_source_column_for_aliased_attribute(field: dict):
+    """Verify physical primary keys select aliased GoodData grain attributes."""
+    model = {
+        "semantic_model": [
+            {
+                "name": "m",
+                "datasets": [
+                    {
+                        "name": "customers",
+                        "source": "db.s.customers",
+                        "primary_key": ["customer_id"],
+                        "fields": [field],
+                    }
+                ],
+            }
+        ]
+    }
+
+    customer = ossie_to_gooddata(model).ldm.datasets[0]
+
+    assert customer.attributes[0].source_column == "customer_id"
+    assert [grain.id for grain in customer.grain] == ["attr.customers.customer_key"]
+
+
+def test_duplicate_source_columns_are_rejected():
+    """Verify ambiguous grain and relationship targets fail instead of being misassigned."""
+    model = {
+        "semantic_model": [
+            {
+                "name": "m",
+                "datasets": [
+                    {
+                        "name": "customers",
+                        "primary_key": ["customer_id"],
+                        "fields": [
+                            _direct_field("customer_id", dimension={}),
+                            {
+                                "name": "customer_key",
+                                "expression": {
+                                    "dialects": [
+                                        {"dialect": "ANSI_SQL", "expression": "customer_id"}
+                                    ]
+                                },
+                                "dimension": {},
+                            },
+                        ],
+                    },
+                    {
+                        "name": "orders",
+                        "fields": [_direct_field("customer_id", dimension={})],
+                    },
+                ],
+                "relationships": [
+                    {
+                        "name": "orders_customer",
+                        "from": "orders",
+                        "to": "customers",
+                        "from_columns": ["customer_id"],
+                        "to_columns": ["customer_id"],
+                    }
+                ],
+            }
+        ]
+    }
+
+    with pytest.raises(
+        ValueError,
+        match="Dataset 'customers': source column 'customer_id' maps to multiple fields",
+    ):
+        ossie_to_gooddata(model)
+
+
 def test_relationships_become_references(ossie_tpcds_dict: dict):
     """Verify Ossie relationships become GoodData references."""
     result = ossie_to_gooddata(ossie_tpcds_dict)