Skip to content

Fixes 29147: add support for databricks contraints in non-incremental ingestion #29148

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mmigdiso wants to merge 4 commits into
base: main
Choose a base branch
from
+57 −1
Open

Fixes 29147: add support for databricks contraints in non-incremental ingestion #29148

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
107d4b5
add support for databricks contraints in non-incremental ingestion
Jun 17, 2026
4e9f380
fix exception handling, code cleanup
Jun 17, 2026
2e62c1c
failure to fetch the contraints is not an error but a warning
Jun 17, 2026
0d5a5f0
short circuit contraints collection if no schema or catalog is present
Jun 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 49 additions & 1 deletion ingestion/src/metadata/ingestion/source/database/unitycatalog/metadata.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,7 @@
UNITY_CATALOG_GET_ALL_TABLE_COLUMNS_TAGS,
UNITY_CATALOG_GET_ALL_TABLE_TAGS,
UNITY_CATALOG_GET_CATALOGS_TAGS,
UNITY_CATALOG_GET_TABLE_DDL,
UNITY_CATALOG_GET_TABLE_DDL, UNITY_CATALOG_TABLE_CONSTRAINTS,
)
from metadata.utils import fqn
from metadata.utils.filters import filter_by_database, filter_by_schema, filter_by_table
Expand Down Expand Up @@ -260,6 +260,7 @@ def _set_incremental_table_processor(self, catalog: str) -> None:
with self._state_lock:
self.incremental_table_processor = incremental_table_processor


def yield_database(self, database_name: str) -> Iterable[Either[CreateDatabaseRequest]]:
"""
From topology.
Expand Down Expand Up @@ -339,6 +340,40 @@ def yield_database_schema(self, schema_name: str) -> Iterable[Either[CreateDatab
yield Either(right=schema_request)
self.register_record_schema_request(schema_request=schema_request)

def _get_tables_with_constraints(self) -> set[tuple[str, str, str]]:
"""
Build and execute SQL query to fetch table constraints.
Handles cases where catalog_name and/or schema_name may be None.
"""
schema_name = self.context.get().database_schema
catalog_name = self.context.get().database
tables_with_constraints = set()
if catalog_name is None or schema_name is None:
return tables_with_constraints

sql = UNITY_CATALOG_TABLE_CONSTRAINTS
params = {}

if catalog_name is not None:
sql += " AND table_catalog = :catalog_name"
params["catalog_name"] = catalog_name
Comment on lines +343 to +359

@gitar-bot gitar-bot Bot Jun 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💡 Quality: No tests added for new constraint-fetch logic

The PR adds a new code path (_get_tables_with_constraints and the conditional full-fetch in get_tables_name_and_type) but includes no unit tests, and the PR checklist for tests is unchecked. Worth adding a test that mocks client.tables.list/get and the constraints query to verify only tables present in the constraints set trigger the extra get() call, and that constraints are propagated into the resulting table request.

Was this helpful? React with 👍 / 👎

if schema_name is not None:
sql += " AND table_schema = :schema_name"
params["schema_name"] = schema_name

try:
cursor = self.sql_connection.execute(text(sql), params)
Comment thread
gitar-bot[bot] marked this conversation as resolved.
for row in cursor:
table_identifier = (row.table_catalog, row.table_schema, row.table_name)
tables_with_constraints.add(table_identifier)
logger.debug(f"Table with constraints: {table_identifier}")
except Exception as exc:
logger.debug(traceback.format_exc())
logger.warning(
f"Error fetching table constraints for catalog [{catalog_name}], schema [{schema_name}]: {exc}"
)
return tables_with_constraints

def get_tables_name_and_type(self) -> Iterable[Tuple[str, TableType]]: # noqa: UP006
"""
Handle table and views.
Expand All @@ -356,10 +391,23 @@ def get_tables_name_and_type(self) -> Iterable[Tuple[str, TableType]]: # noqa:
if self.incremental.enabled and self.incremental_table_processor:
yield from self._get_incremental_tables(catalog_name, schema_name)
else:
table_with_constraints = self._get_tables_with_constraints()
for table in self.client.tables.list(
catalog_name=catalog_name,
schema_name=schema_name,
):
if (table.catalog_name, table.schema_name, table.name) in table_with_constraints:
# Only tables with constraints require full fetch; list() doesn't include constraint details
try:
table = self.client.tables.get(table.full_name)
except Exception as exc:
msg = (
f"Unexpected exception in fetching constraints "
f"Constraints will be ignored."
f"table [{table.full_name}]: {exc}. "
)
Comment on lines +404 to +408

@gitar-bot gitar-bot Bot Jun 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💡 Quality: Malformed warning message: unbalanced bracket and typo

The warning string built when client.tables.get() fails has a syntax/readability defect and a typo. The fragment f"(table [{table.full_name}]: {exc}. " opens ( and [ but never closes them, producing output like (table [cat.sch.tbl]: <err>. And "Contraints" should be "Constraints". Since this message is surfaced to users via self.status.warning, clean it up.

Fix the unbalanced brackets and the 'Contraints' typo.:

msg = (
    f"Unexpected exception fetching constraints "
    f"for table [{table.full_name}]: {exc}. "
    f"Constraints will be ignored."
)
  • Apply fix

Check the box to apply the fix or reply for a change | Was this helpful? React with 👍 / 👎

@mmigdiso mmigdiso Jun 18, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This comment is based on an outdated version of the code. The typo and the paranthesis mentioned in the comment doesn't exist.

logger.warning(msg)
self.status.warning(table.name, msg)
yield from self._process_table(table, catalog_name, schema_name)
Comment thread
gitar-bot[bot] marked this conversation as resolved.
Comment thread
greptile-apps[bot] marked this conversation as resolved.

def _get_incremental_tables(self, catalog_name: str, schema_name: str) -> Iterable[Tuple[str, TableType]]: # noqa: UP006
Expand Down
8 changes: 8 additions & 0 deletions ingestion/src/metadata/ingestion/source/database/unitycatalog/queries.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,3 +136,11 @@
WHERE 1=0
"""
)

UNITY_CATALOG_TABLE_CONSTRAINTS = textwrap.dedent(
"""
SELECT DISTINCT table_catalog, table_schema, table_name
FROM system.information_schema.table_constraints
WHERE 1=1
"""
)
Loading