fix: special charcater decoding from datadiff username

Makefile

@@ -263,8 +263,8 @@ build-ingestion-base-local:  ## Builds the ingestion DEV docker operator with th
 	$(MAKE) install_dev generate
 	docker build -f ingestion/operators/docker/Dockerfile.ci . -t openmetadata/ingestion-base:local
 
-.PHONY: build-ingestion-base-slim-local
-build-ingestion-base-local:  ## Builds the ingestion DEV docker operator with the local ingestion files
+.PHONY: build-ingestion-slim-local
+build-ingestion-slim-local:  ## Builds the SLIM ingestion DEV docker operator with the local ingestion files
 	$(MAKE) install_dev generate
 	docker build -f ingestion/operators/docker/Dockerfile.ci . -t openmetadata/ingestion-base-slim:local --build-arg INGESTION_DEPENDENCY=slim
 

ingestion/src/metadata/data_quality/validations/models.py

@@ -3,7 +3,9 @@
 from typing import List, Optional, Union  # noqa: UP035
 
 from pydantic import BaseModel, Field
+from sqlalchemy.engine import make_url
 
+from metadata.data_quality.validations.utils import render_url_for_data_diff
 from metadata.generated.schema.entity.data.table import (
     Column,
     Table,
@@ -28,6 +30,17 @@ class TableParameter(BaseModel):
     key_columns: Optional[list[str]] = None  # noqa: UP045
     extra_columns: Optional[list[str]] = None  # noqa: UP045
 
+    @property
+    def data_diff_service_url(self) -> Union[str, dict]:  # noqa: UP007
+        """`serviceUrl` rendered for data-diff's own URI parser.
+
+        `serviceUrl` is a canonical SQLAlchemy URL, which encodes more than data-diff decodes.
+        Connection dicts are passed through: data-diff reads their values verbatim.
+        """
+        if isinstance(self.serviceUrl, dict):
+            return self.serviceUrl
+        return render_url_for_data_diff(make_url(self.serviceUrl))
+
 
 class TableDiffRuntimeParameters(BaseModel):
     table1: TableParameter

ingestion/src/metadata/data_quality/validations/table/sqlalchemy/tableDiff.py

@@ -13,10 +13,11 @@
 import random
 import string
 import traceback
+from contextlib import contextmanager
 from decimal import Decimal
 from functools import reduce
 from itertools import islice
-from typing import Dict, Iterable, List, Optional, Tuple, cast  # noqa: UP035
+from typing import Dict, Iterable, Iterator, List, Optional, Tuple, cast  # noqa: UP035
 from urllib.parse import urlparse
 
 import data_diff
@@ -78,6 +79,8 @@
     Dialects.UnityCatalog,
 ]
 
+DUPLICATE_KEY_MESSAGE = "Duplicate primary keys"
+
 
 class SchemaDiffResult(BaseModel):
     class Config:
@@ -164,6 +167,29 @@ def __init__(self, param: str, dialect: str):
         super().__init__(f"Unsupported dialect in param {param}: {dialect}")
 
 
+class DuplicateKeyError(Exception):
+    """A diff key column is not unique, which makes a row-level diff undefined.
+
+    The message names the key columns only. The duplicated values themselves are row data, which
+    this test result is not the place to publish, and finding them costs a scan of the table.
+    """
+
+    def __init__(
+        self,
+        key_columns: List[str],  # noqa: UP006
+        table: Optional[str] = None,  # noqa: UP045
+    ):
+        if len(key_columns) == 1:
+            subject = f"Key column '{key_columns[0]}' is"
+        else:
+            subject = f"Key columns ({', '.join(repr(c) for c in key_columns)}) are"
+        location = f"in {table}" if table else "in one of the compared tables"
+        super().__init__(
+            f"{subject} not unique {location}. A row-level diff needs a unique key: "
+            "pick a unique column, or add more columns to the key."
+        )
+
+
 def masked(s: str, mask: bool = True) -> str:
     """Mask a string if masked is True otherwise return the string.
     Only for development purposes, do not use in production.
@@ -204,6 +230,14 @@ def _run_validation(self):
         try:
             self._validate_dialects()
             return self._run()
+        except DuplicateKeyError as e:
+            logger.error(f"[Data Diff]: {e}")
+            result = TestCaseResult(
+                timestamp=self.execution_date,  # type: ignore
+                testCaseStatus=TestCaseStatus.Aborted,
+                result=str(e),
+            )
+            return result  # noqa: RET504
         except DataDiffMismatchingKeyTypesError as e:
             result = TestCaseResult(
                 timestamp=self.execution_date,  # type: ignore
@@ -260,36 +294,37 @@ def _run(self) -> TestCaseResult:
             self.runtime_params.table2.extra_columns = common_columns
         table_diff_iter = self.get_table_diff()
 
-        if not threshold or self.test_case.computePassedFailedRowCount:
-            stats = table_diff_iter.get_stats_dict()
-            if stats["total"] > 0:
-                logger.debug("Sample of failed rows:")
-                # depending on the data, this require scanning a lot of data
-                # so we only log the sample in debug mode. data can be sensitive
-                # so it is masked by default
-                for s in islice(
-                    self.safe_table_diff_iterator(),
-                    10 if logger.level <= logging.DEBUG else 0,
-                ):
-                    logger.debug("%s", str([s[0]] + [masked(st) for st in s[1]]))
-            test_case_result = self.get_row_diff_test_case_result(
+        with self._duplicate_keys_named(table_diff_iter):
+            if not threshold or self.test_case.computePassedFailedRowCount:
+                stats = table_diff_iter.get_stats_dict()
+                if stats["total"] > 0:
+                    logger.debug("Sample of failed rows:")
+                    # depending on the data, this require scanning a lot of data
+                    # so we only log the sample in debug mode. data can be sensitive
+                    # so it is masked by default
+                    for s in islice(
+                        self.safe_table_diff_iterator(),
+                        10 if logger.level <= logging.DEBUG else 0,
+                    ):
+                        logger.debug("%s", str([s[0]] + [masked(st) for st in s[1]]))
+                test_case_result = self.get_row_diff_test_case_result(
+                    threshold,
+                    stats["total"],
+                    stats["updated"],
+                    stats["exclusive_A"],
+                    stats["exclusive_B"],
+                    column_diff,
+                )
+                count = self._compute_row_count(self.runner, None)  # type: ignore
+                test_case_result.passedRows = stats["unchanged"]
+                test_case_result.passedRowsPercentage = test_case_result.passedRows / count * 100
+                test_case_result.failedRowsPercentage = test_case_result.failedRows / count * 100
+                return test_case_result
+            return self.get_row_diff_test_case_result(
                 threshold,
-                stats["total"],
-                stats["updated"],
-                stats["exclusive_A"],
-                stats["exclusive_B"],
+                self.calculate_diffs_with_limit(table_diff_iter, threshold),
                 column_diff,
             )
-            count = self._compute_row_count(self.runner, None)  # type: ignore
-            test_case_result.passedRows = stats["unchanged"]
-            test_case_result.passedRowsPercentage = test_case_result.passedRows / count * 100
-            test_case_result.failedRowsPercentage = test_case_result.failedRows / count * 100
-            return test_case_result
-        return self.get_row_diff_test_case_result(
-            threshold,
-            self.calculate_diffs_with_limit(table_diff_iter, threshold),
-            column_diff,
-        )
 
     def get_incomparable_columns(self) -> List[str]:  # noqa: UP006
         """Get the columns that have types that are not comparable between the two tables. For example
@@ -300,7 +335,7 @@ def get_incomparable_columns(self) -> List[str]:  # noqa: UP006
         """
 
         table1 = data_diff.connect_to_table(
-            self.runtime_params.table1.serviceUrl,
+            self.runtime_params.table1.data_diff_service_url,
             self.runtime_params.table1.path,
             self.runtime_params.table1.key_columns,
             extra_columns=self.runtime_params.extraColumns,
@@ -313,7 +348,7 @@ def get_incomparable_columns(self) -> List[str]:  # noqa: UP006
             else None,
         ).with_schema()
         table2 = data_diff.connect_to_table(
-            self.runtime_params.table2.serviceUrl,
+            self.runtime_params.table2.data_diff_service_url,
             self.runtime_params.table2.path,
             self.runtime_params.table2.key_columns,
             extra_columns=self.runtime_params.extraColumns,
@@ -377,7 +412,7 @@ def get_table_diff(self) -> DiffResultWrapper:
         """Calls data_diff.diff_tables with the parameters from the test case."""
         left_where, right_where = self.sample_where_clause()
         table1 = data_diff.connect_to_table(
-            self.runtime_params.table1.serviceUrl,
+            self.runtime_params.table1.data_diff_service_url,
             self.runtime_params.table1.path,
             self.runtime_params.table1.key_columns,  # type: ignore
             extra_columns=self.runtime_params.table1.extra_columns,
@@ -391,7 +426,7 @@ def get_table_diff(self) -> DiffResultWrapper:
             else None,
         )
         table2 = data_diff.connect_to_table(
-            self.runtime_params.table2.serviceUrl,
+            self.runtime_params.table2.data_diff_service_url,
             self.runtime_params.table2.path,
             self.runtime_params.table2.key_columns,  # type: ignore
             extra_columns=self.runtime_params.table2.extra_columns,
@@ -586,6 +621,60 @@ def _validate_dialects(self):
             if dialect not in SUPPORTED_DIALECTS:
                 raise UnsupportedDialectError(name, dialect)
 
+    @contextmanager
+    def _duplicate_keys_named(self, table_diff_iter: DiffResultWrapper) -> Iterator[None]:
+        """Re-raise data-diff's duplicate-key failures with the key columns named.
+
+        A key that is not unique makes a row-level diff undefined, and data-diff reports it in two
+        equally opaque ways: joindiff validates the key itself and raises
+        `ValueError("Duplicate primary keys")`, while hashdiff only trips over it in `_get_stats`,
+        which folds rows into a `{key: sign}` map and asserts a key never repeats with the same
+        sign - a bare `AssertionError`. Neither names the key, so we do.
+
+        We deliberately do not check uniqueness up front: that is a COUNT/COUNT DISTINCT over both
+        tables on every run, far too expensive on a large table to pay for an error that only
+        happens when the key is misconfigured. Both paths here are reached only once the diff has
+        already failed, and neither queries anything.
+        """
+        try:
+            yield
+        except ValueError as exc:
+            if DUPLICATE_KEY_MESSAGE not in str(exc):
+                raise
+            # joindiff does not say which of the two tables it found the duplicates in.
+            raise DuplicateKeyError(self._diff_key_columns()) from exc
+        except AssertionError as exc:
+            table_param = self._table_with_duplicate_keys(table_diff_iter)
+            if table_param is None:
+                # Some other invariant inside data-diff; nothing useful to add.
+                raise
+            raise DuplicateKeyError(
+                list(table_param.key_columns or []),
+                table_param.fullyQualifiedName or table_param.path,
+            ) from exc
+
+    def _diff_key_columns(self) -> List[str]:  # noqa: UP006
+        """The key columns the diff runs on. Both sides diff on the same key."""
+        return list(self.runtime_params.table1.key_columns or self.runtime_params.table2.key_columns or [])
+
+    def _table_with_duplicate_keys(self, table_diff_iter: DiffResultWrapper) -> Optional[TableParameter]:  # noqa: UP045
+        """The table whose rows repeat a key, or None if the diffed rows show no duplicate.
+
+        Read off the rows `_get_stats` had already materialised into `result_list` when it failed,
+        so this costs no query. A row is `(sign, values)` with the key first; "-" rows come from
+        table1 and "+" rows from table2.
+        """
+        key_length = len(self._diff_key_columns())
+        if not key_length:
+            return None
+        seen = set()
+        for sign, values in table_diff_iter.result_list:
+            marker = (sign, tuple(values[:key_length]))
+            if marker in seen:
+                return self.runtime_params.table1 if sign == "-" else self.runtime_params.table2
+            seen.add(marker)
+        return None
+
     def get_column_diff(self) -> Optional[ColumnDiffResult]:  # noqa: UP045
         """Get the column diff between the two tables. If there are no differences, return None."""
         removed, added = self.get_changed_added_columns(

ingestion/src/metadata/data_quality/validations/utils.py

@@ -3,12 +3,22 @@
 """
 
 from typing import Any, Callable, List, Optional, TypeVar, Union  # noqa: UP035
+from urllib.parse import quote
+
+from sqlalchemy.engine import URL
 
 from metadata.generated.schema.tests.testCase import TestCaseParameterValue
+from metadata.utils.logger import test_suite_logger
+
+logger = test_suite_logger()
 
 T = TypeVar("T", bound=Callable)
 R = TypeVar("R")
 
+# Characters that terminate the userinfo (or the whole authority) while data-diff parses the URI.
+# They must stay percent-encoded even though data-diff will not decode them back.
+USERNAME_RESERVED_CHARACTERS = ":/?#"
+
 
 def get_test_case_param_value(
     test_case_param_vals: List[TestCaseParameterValue],  # noqa: UP006
@@ -54,6 +64,44 @@ def get_bool_test_case_param(
     return str_val.lower() == "true"
 
 
+def _encode_username_for_data_diff(username: str) -> str:
+    """Percent-encode only what data-diff's URI parser needs to locate the userinfo boundaries."""
+    return "".join(f"%{ord(char):02X}" if char in USERNAME_RESERVED_CHARACTERS else char for char in username)
+
+
+def render_url_for_data_diff(url: URL) -> str:
+    """Render `url` so that data-diff reads back the values it was built from.
+
+    `URL.render_as_string` percent-encodes the username, but data-diff only decodes the password
+    (`CustomParseResult`), the host and the query string when it parses a URI. An encoded username
+    therefore reaches the driver still encoded, and `user@corp.com` tries to authenticate as
+    `user%40corp.com`. We hand data-diff the decoded username and keep the password encoded, so that
+    every component survives exactly one encode/decode round trip.
+    """
+    if url.username is None:
+        return url.render_as_string(hide_password=False)
+
+    userinfo = _encode_username_for_data_diff(url.username)
+    if userinfo != url.username:
+        logger.warning(
+            "[Data Diff]: The username contains characters reserved by the connection URI (%s). "
+            "data-diff does not decode them, so authentication may fail.",
+            ", ".join(sorted(set(url.username) & set(USERNAME_RESERVED_CHARACTERS))),
+        )
+    if url.password is not None:
+        userinfo += f":{quote(str(url.password), safe=' +')}"
+
+    authority = URL.create(
+        drivername=url.drivername,
+        host=url.host,
+        port=url.port,
+        database=url.database,
+        query=url.query,
+    ).render_as_string(hide_password=False)
+    scheme, _, rest = authority.partition("://")
+    return f"{scheme}://{userinfo}@{rest}"
+
+
 def casefold_if_string(value: Any) -> Any:
     """Case fold the value if it is a string.