fix(unitycatalog): dedup cross-window lineage edges, enforce chunk-size minimum

- Add bounded LRU dedup so an edge spanning multiple day-windows is emitted
  once instead of once per window; surface the deduplicated count in the
  per-run summary.
- Add `minimum: 1` to the lineageQueryChunkSize connection schema so invalid
  values are rejected by UI/API, matching the runtime clamp.
- Correct the lineage summary log to credit database/schema/table filter
  patterns rather than only databaseFilterPattern.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: ulixius9

ingestion/src/metadata/ingestion/source/database/unitycatalog/lineage.py

@@ -61,6 +61,8 @@
 
 TABLE_RESOLUTION_CACHE_SIZE = 1000
 
+EDGE_DEDUP_CACHE_SIZE = 1000
+
 DEFAULT_LINEAGE_CHUNK_DAYS = 7
 
 
@@ -90,6 +92,7 @@ def __init__(
         self.connection_obj = get_connection(self.service_connection)
         self.engine = get_sqlalchemy_connection(self.service_connection)
         self._table_cache: LRUCache = LRUCache(maxsize=TABLE_RESOLUTION_CACHE_SIZE)
+        self._seen_edges: LRUCache = LRUCache(maxsize=EDGE_DEDUP_CACHE_SIZE)
         self._chunk_days = self._resolve_chunk_days()
         self.test_connection()
 
@@ -273,17 +276,18 @@ def _yield_table_lineage(self) -> Iterable[Either[AddLineageRequest]]:
         """
         Stream table/column lineage one day-window at a time, emitting one
         request per resolved edge. Per-row and per-window failures both surface
-        as Either(left) instead of being swallowed. Edges dropped by
-        `databaseFilterPattern` and edges whose tables are absent from
-        OpenMetadata are counted separately so the summary distinguishes
+        as Either(left) instead of being swallowed. Edges dropped by the
+        database/schema/table filter patterns and edges whose tables are absent
+        from OpenMetadata are counted separately so the summary distinguishes
         intentional filtering from missing metadata.
         """
-        stats = {"emitted": 0, "filtered": 0, "unresolved": 0, "failed": 0}
+        stats = {"emitted": 0, "duplicate": 0, "filtered": 0, "unresolved": 0, "failed": 0}
         for window_start, window_end in self._iter_date_windows():
             yield from self._yield_window_lineage(window_start, window_end, stats)
         logger.info(
-            f"Table lineage: emitted {stats['emitted']} edges, filtered {stats['filtered']} "
-            f"(databaseFilterPattern), unresolved {stats['unresolved']} (tables not in OpenMetadata), "
+            f"Table lineage: emitted {stats['emitted']} edges, deduplicated {stats['duplicate']} "
+            f"cross-window duplicates, filtered {stats['filtered']} (database/schema/table filter "
+            f"patterns), unresolved {stats['unresolved']} (tables not in OpenMetadata), "
             f"failed {stats['failed']} (row/window errors)"
         )
 
@@ -310,6 +314,23 @@ def _yield_window_lineage(
             )
 
     def _yield_row_lineage(self, row: Any, stats: dict) -> Iterable[Either[AddLineageRequest]]:
+        """
+        Skip edges already streamed in an earlier window so an edge whose events
+        span multiple windows is emitted once instead of once per window. The
+        dedup set is a bounded LRU, so only recently-seen edges are suppressed;
+        edges past the cache window may re-emit (lineage adds are idempotent).
+        """
+        edge_key = (
+            row.source_table_full_name.lower(),
+            row.target_table_full_name.lower(),
+        )
+        if edge_key in self._seen_edges:
+            stats["duplicate"] += 1
+        else:
+            self._seen_edges[edge_key] = True
+            yield from self._process_unseen_row(row, stats)
+
+    def _process_unseen_row(self, row: Any, stats: dict) -> Iterable[Either[AddLineageRequest]]:
         if self._is_filtered_table(row.source_table_full_name) or self._is_filtered_table(row.target_table_full_name):
             stats["filtered"] += 1
         else:

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

@@ -276,6 +276,20 @@ def test_emits_resolved_edges(self, lineage_source):
         assert len(results) == 1
         assert isinstance(results[0].right, AddLineageRequest)
 
+    def test_dedupes_edges_across_windows(self, lineage_source):
+        source_table = _make_table("source", "local_unitycatalog.cat.schema.source")
+        target_table = _make_table("target", "local_unitycatalog.cat.schema.target")
+        lineage_source.metadata.get_by_name.side_effect = [source_table, target_table]
+
+        rows = [LineageRow("cat.schema.source", "cat.schema.target", None)]
+        with patch.object(lineage_source, "_iter_date_windows", return_value=[("s1", "e1"), ("s2", "e2")]):
+            _mock_query_rows(lineage_source, rows)
+            results = list(lineage_source._yield_table_lineage())
+
+        assert len(results) == 1
+        assert isinstance(results[0].right, AddLineageRequest)
+        assert lineage_source.metadata.get_by_name.call_count == 2
+
     def test_skips_unresolved_edges(self, lineage_source):
         lineage_source.metadata.get_by_name.return_value = None
 

openmetadata-spec/src/main/resources/json/schema/entity/services/connections/database/unityCatalogConnection.json

@@ -84,7 +84,8 @@
       "title": "Lineage Query Chunk Size",
       "description": "Number of days of lineage scanned per query against the Unity Catalog system tables. The configured lineage lookback window is split into chunks of this size and streamed one chunk at a time so each scan stays bounded. Larger values issue fewer queries (faster); smaller values keep each scan smaller.",
       "type": "integer",
-      "default": 7
+      "default": 7,
+      "minimum": 1
     },
     "connectionOptions": {
       "title": "Connection Options",