diff --git a/ingestion/src/metadata/ingestion/source/database/common_db_source.py b/ingestion/src/metadata/ingestion/source/database/common_db_source.py index e96380c20639..744ad4b13636 100644 --- a/ingestion/src/metadata/ingestion/source/database/common_db_source.py +++ b/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,7 +659,6 @@ 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 @@ -667,14 +666,12 @@ def _prepare_foreign_constraints( # pylint: disable=too-many-arguments, too-man 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"), diff --git a/ingestion/src/metadata/utils/constraints.py b/ingestion/src/metadata/utils/constraints.py index e4883c6027f3..c86a2aa893d4 100644 --- a/ingestion/src/metadata/utils/constraints.py +++ b/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.) diff --git a/ingestion/tests/integration/mysql/test_foreign_key_case.py b/ingestion/tests/integration/mysql/test_foreign_key_case.py new file mode 100644 index 000000000000..b03ec99e7a9d --- /dev/null +++ b/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 diff --git a/ingestion/tests/unit/topology/database/test_common_db_source.py b/ingestion/tests/unit/topology/database/test_common_db_source.py index 1c9262eafd2b..0a21efd15227 100644 --- a/ingestion/tests/unit/topology/database/test_common_db_source.py +++ b/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."""