Fixes #34933: Match FK referred columns to the parent table's column case

ingestion/src/metadata/ingestion/source/database/common_db_source.py

@@ -73,7 +73,7 @@
 from metadata.ingestion.source.database.sqlalchemy_source import SqlAlchemySource
 from metadata.ingestion.source.database.stored_procedures_mixin import QueryByProcedure
 from metadata.utils import fqn
-from metadata.utils.constraints import get_relationship_type
+from metadata.utils.constraints import get_relationship_type, resolve_column_names
 from metadata.utils.filters import filter_by_table
 from metadata.utils.helpers import retry_with_docker_host
 from metadata.utils.logger import ingestion_logger
@@ -659,22 +659,19 @@ def _prepare_foreign_constraints(  # pylint: disable=too-many-arguments, too-man
         """
         Method to prepare the foreign constraints
         """
-        referred_column_fqns = []
         database_name = (column.get("referred_database") if supports_database else None) or self.context.get().database  # pyright: ignore[reportAttributeAccessIssue]
 
         referred_schema = column.get("referred_schema") or schema_name
         referred_table_fqn = (
             f"{self.context.get().database_service}.{database_name}.{referred_schema}.{column.get('referred_table')}"
         )
         referred_table = self.metadata.get_by_name(entity=Table, fqn=referred_table_fqn)
-        if referred_table:
-            for referred_column in column.get("referred_columns"):
-                col_fqn = fqn._build(  # pylint: disable=protected-access
-                    referred_table_fqn, referred_column, quote=False
-                )
-                if col_fqn:
-                    referred_column_fqns.append(FullyQualifiedEntityName(col_fqn))
-        else:
+        referred_columns = (
+            resolve_column_names(column["referred_columns"], referred_table.columns) if referred_table else None
+        )
+        if referred_table is None or referred_columns is None:
+            # The referred table, or a column added to it in this run, may only reach the
+            # server later in the run, so retry once every table has been ingested.
             if add_to_global:
                 column_and_referred_columns = ColumnAndReferredColumn(
                     table_name=table_name,
@@ -683,14 +680,27 @@ def _prepare_foreign_constraints(  # pylint: disable=too-many-arguments, too-man
                     column=column,
                 )
                 self.context.get_global().foreign_tables.append(column_and_referred_columns)
+            elif referred_table:
+                logger.warning(
+                    "Skipping foreign key of [%s.%s]: referred columns %s don't each match exactly one column of [%s]",
+                    schema_name,
+                    table_name,
+                    column["referred_columns"],
+                    referred_table_fqn,
+                )
             return None
-        relationship_type = None
-        if referred_table:
-            relationship_type = get_relationship_type(
-                column,  # sqlalchemy foreign column
-                referred_table.columns,  # referred table columns
-                columns,  # current table om columns
+        referred_column_fqns = []
+        for referred_column in referred_columns:
+            col_fqn = fqn._build(  # pylint: disable=protected-access
+                referred_table_fqn, referred_column, quote=False
             )
+            if col_fqn:
+                referred_column_fqns.append(FullyQualifiedEntityName(col_fqn))
+        relationship_type = get_relationship_type(
+            {**column, "referred_columns": referred_columns},
+            referred_table.columns,  # referred table columns
+            columns,  # current table om columns
+        )
         return TableConstraint(
             constraintType=ConstraintType.FOREIGN_KEY,
             columns=column.get("constrained_columns"),

ingestion/src/metadata/utils/constraints.py

@@ -38,6 +38,29 @@ def _is_column_unique(column: dict, columns: list[Column]) -> bool:
     return False
 
 
+def resolve_column_names(column_names: list[str], columns: list[Column]) -> list[str] | None:
+    """
+    Map column names reported by the source to the names stored for the table.
+
+    Sources with case-insensitive identifiers can report a name in a different case than the
+    column was created with (MySQL keeps a foreign key's referenced column as written in the
+    FK definition), while the server matches column names case-sensitively. An exact match
+    wins, then a single case-insensitive match. Returns None if any name matches no column
+    or several.
+    """
+    stored_names = [model_str(col.name) for col in columns or []]
+    resolved = []
+    for name in column_names:
+        if name in stored_names:
+            resolved.append(name)
+            continue
+        case_insensitive_matches = [stored for stored in stored_names if stored.lower() == name.lower()]
+        if len(case_insensitive_matches) != 1:
+            return None
+        resolved.append(case_insensitive_matches[0])
+    return resolved
+
+
 def get_relationship_type(column: dict, referred_table_columns: list[Column], columns: list[Column]) -> str:
     """
     Determine the type of relationship (one-to-one, one-to-many, etc.)

ingestion/tests/integration/mysql/test_foreign_key_case.py

@@ -0,0 +1,81 @@
+#  Copyright 2025 Collate
+#  Licensed under the Collate Community License, Version 1.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  https://github.com/open-metadata/OpenMetadata/blob/main/ingestion/LICENSE
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+"""A MySQL foreign key whose referenced column differs in case from the parent column
+(`REFERENCES parent (id)` while the column is `Id`) must not get the table rejected."""
+
+import pytest
+from sqlalchemy import create_engine, text
+
+from metadata.generated.schema.entity.data.table import (
+    ConstraintType,
+    RelationshipType,
+    Table,
+)
+from metadata.ingestion.ometa.utils import model_str
+from metadata.workflow.metadata import MetadataWorkflow
+
+
+@pytest.fixture(scope="module")
+def mysql_engine(mysql_container):
+    engine = create_engine(mysql_container.get_connection_url())
+    yield engine
+    engine.dispose()
+
+
+@pytest.fixture()
+def child_with_lowercase_referred_column(mysql_engine):
+    # MySQL rewrites the referenced column to the parent's casing when the parent exists.
+    # Created before the parent with FK checks off, the FK keeps `id` as written.
+    with mysql_engine.connect() as conn:
+        conn.execute(text("SET foreign_key_checks = 0"))
+        conn.execute(
+            text(
+                "CREATE TABLE employees.fk_case_child ("
+                "  Id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,"
+                "  Reason TINYINT NOT NULL,"
+                "  KEY fk_reason (Reason),"
+                "  CONSTRAINT fk_case_child_ibfk_1 FOREIGN KEY (Reason) REFERENCES fk_case_parent (id)"
+                ")"
+            )
+        )
+        conn.execute(text("CREATE TABLE employees.fk_case_parent (Id TINYINT NOT NULL PRIMARY KEY, Name VARCHAR(100))"))
+        conn.execute(text("SET foreign_key_checks = 1"))
+        child_ddl = conn.execute(text("SHOW CREATE TABLE employees.fk_case_child")).one()[1]
+        conn.commit()
+    assert "REFERENCES `fk_case_parent` (`id`)" in child_ddl
+    yield
+    with mysql_engine.connect() as conn:
+        conn.execute(text("DROP TABLE IF EXISTS employees.fk_case_child, employees.fk_case_parent"))
+        conn.commit()
+
+
+def test_foreign_key_uses_stored_casing_of_referred_column(
+    patch_passwords_for_db_services,
+    run_workflow,
+    ingestion_config,
+    metadata,
+    db_service,
+    child_with_lowercase_referred_column,
+):
+    # On a new service the parent may not be in OpenMetadata yet when the child is processed,
+    # which defers the FK to the end of the run. The second run sends it inline with the child
+    # table: the request the server used to reject with `Invalid column name id`.
+    run_workflow(MetadataWorkflow, ingestion_config)
+    run_workflow(MetadataWorkflow, ingestion_config)
+
+    schema_fqn = f"{db_service.fullyQualifiedName.root}.default.employees"
+    child = metadata.get_by_name(entity=Table, fqn=f"{schema_fqn}.fk_case_child", fields=["tableConstraints"])
+    foreign_keys = [c for c in child.tableConstraints or [] if c.constraintType == ConstraintType.FOREIGN_KEY]
+    assert [[model_str(column) for column in fk.referredColumns] for fk in foreign_keys] == [
+        [f"{schema_fqn}.fk_case_parent.Id"]
+    ]
+    assert foreign_keys[0].relationshipType == RelationshipType.MANY_TO_ONE

ingestion/tests/unit/topology/database/test_common_db_source.py

@@ -26,8 +26,10 @@
 
 from metadata.generated.schema.entity.data.table import (
     Column,
+    Constraint,
     ConstraintType,
     DataType,
+    RelationshipType,
     Table,
     TableConstraint,
 )
@@ -278,6 +280,113 @@ def test_referred_table_not_found_appends_to_global_foreign_tables(self, source)
         assert len(source.context.get_global.return_value.foreign_tables) == 1
 
 
+LIMITS_COLUMNS = [
+    Column(name="Id", dataType=DataType.BIGINT, constraint=Constraint.PRIMARY_KEY),
+    Column(name="Reason", dataType=DataType.TINYINT, constraint=Constraint.NOT_NULL),
+    Column(name="RegionRef", dataType=DataType.VARCHAR, dataLength=10),
+]
+REFERRED_TABLE_FQN = "test_service.test_db.risk.limitadjustmentreasons"
+
+
+def _referred_table(*columns: Column) -> Table:
+    return Table(
+        id=uuid.uuid4(),
+        name="limitadjustmentreasons",
+        fullyQualifiedName=REFERRED_TABLE_FQN,
+        columns=list(columns),
+    )
+
+
+class TestPrepareForeignConstraintsReferredColumnCase:
+    """MySQL can keep a foreign key's referenced column as written (`id`) while the parent column
+    is `Id`. The server matches referred columns case-sensitively and rejects the whole table."""
+
+    @staticmethod
+    def _prepare(source, referred_columns, constrained_columns=("Reason",), add_to_global=True):
+        return source._prepare_foreign_constraints(
+            supports_database=False,
+            column={
+                "referred_schema": "risk",
+                "referred_table": "limitadjustmentreasons",
+                "referred_columns": referred_columns,
+                "constrained_columns": list(constrained_columns),
+            },
+            table_name="limits",
+            schema_name="risk",
+            db_name="test_db",
+            columns=LIMITS_COLUMNS,
+            add_to_global=add_to_global,
+        )
+
+    @pytest.mark.parametrize("add_to_global", [True, False], ids=["inline", "deferred-pass"])
+    def test_referred_column_differing_in_case_uses_stored_name(self, source, add_to_global):
+        source.metadata.get_by_name.return_value = _referred_table(
+            Column(name="Id", dataType=DataType.TINYINT, constraint=Constraint.PRIMARY_KEY),
+            Column(name="Name", dataType=DataType.VARCHAR, dataLength=100),
+        )
+
+        result = self._prepare(source, ["id"], add_to_global=add_to_global)
+
+        assert [model_str(column) for column in result.referredColumns] == [f"{REFERRED_TABLE_FQN}.Id"]
+
+    def test_relationship_type_sees_referred_primary_key_despite_case(self, source):
+        source.metadata.get_by_name.return_value = _referred_table(
+            Column(name="Id", dataType=DataType.TINYINT, constraint=Constraint.PRIMARY_KEY),
+        )
+
+        result = self._prepare(source, ["id"])
+
+        assert result.relationshipType == RelationshipType.MANY_TO_ONE
+
+    def test_composite_foreign_key_resolves_each_referred_column(self, source):
+        source.metadata.get_by_name.return_value = _referred_table(
+            Column(name="Region", dataType=DataType.VARCHAR, dataLength=10),
+            Column(name="Code", dataType=DataType.TINYINT),
+        )
+
+        result = self._prepare(source, ["region", "Code"], constrained_columns=("RegionRef", "Reason"))
+
+        assert [model_str(column) for column in result.referredColumns] == [
+            f"{REFERRED_TABLE_FQN}.Region",
+            f"{REFERRED_TABLE_FQN}.Code",
+        ]
+
+    def test_exact_match_wins_over_case_insensitive_match(self, source):
+        """Case-sensitive sources (e.g. quoted identifiers in Postgres) can hold both `id` and `ID`."""
+        source.metadata.get_by_name.return_value = _referred_table(
+            Column(name="id", dataType=DataType.INT),
+            Column(name="ID", dataType=DataType.INT),
+        )
+
+        result = self._prepare(source, ["ID"])
+
+        assert [model_str(column) for column in result.referredColumns] == [f"{REFERRED_TABLE_FQN}.ID"]
+
+    @pytest.mark.parametrize("referred_column", ["missing", "Id"], ids=["no-match", "ambiguous-match"])
+    def test_unresolvable_referred_column_is_deferred(self, source, referred_column):
+        """The parent may still gain the column later in the run, so retry once all tables are in."""
+        source.metadata.get_by_name.return_value = _referred_table(
+            Column(name="id", dataType=DataType.INT),
+            Column(name="ID", dataType=DataType.INT),
+        )
+
+        result = self._prepare(source, [referred_column])
+
+        assert result is None
+        assert len(source.context.get_global.return_value.foreign_tables) == 1
+
+    def test_unresolvable_referred_column_is_dropped_on_deferred_pass(self, source):
+        """Sending it would get the whole table rejected by the server."""
+        source.metadata.get_by_name.return_value = _referred_table(
+            Column(name="Id", dataType=DataType.TINYINT, constraint=Constraint.PRIMARY_KEY),
+        )
+
+        result = self._prepare(source, ["missing"], add_to_global=False)
+
+        assert result is None
+        assert source.context.get_global.return_value.foreign_tables == []
+
+
 class TestNormalizeTableConstraints:
     """Test DatabaseServiceSource.normalize_table_constraints."""