diff --git a/src/keboola_agent_cli/commands/lineage.py b/src/keboola_agent_cli/commands/lineage.py index 7ffb7ec3..9350f1c8 100644 --- a/src/keboola_agent_cli/commands/lineage.py +++ b/src/keboola_agent_cli/commands/lineage.py @@ -20,6 +20,7 @@ import typer from ..errors import ErrorCode +from ..output import OutputFormatter from ..services.deep_lineage_service import DeepLineageService, LineageGraph from ._helpers import ( check_cli_permission, @@ -340,8 +341,13 @@ def lineage_show( display_opts = {"show_columns": columns, "filter_column": column} - if upstream: - query_result = service.query_upstream(graph, upstream, project or "", depth) + # Both directions render identically; only the service call and the label + # differ. Passing both --upstream and --downstream emits both, in order. + for direction, identifier in (("upstream", upstream), ("downstream", downstream)): + if not identifier: + continue + query = service.query_upstream if direction == "upstream" else service.query_downstream + query_result = query(graph, identifier, project or "", depth) if "error" in query_result: suggestions = query_result.get("suggestions", []) msg = query_result["error"] @@ -350,38 +356,32 @@ def lineage_show( formatter.error(message=msg, error_code=ErrorCode.NODE_NOT_FOUND) raise typer.Exit(code=1) - if formatter.json_mode: - if column: - query_result = _filter_column_json(query_result, column) - formatter.output(query_result) - elif format in ("mermaid", "html", "er"): - _output_mermaid_or_html(formatter, service, graph, query_result, "upstream", format) - else: - _format_lineage_tree(formatter, graph, query_result, "upstream", **display_opts) - - if downstream: - query_result = service.query_downstream(graph, downstream, project or "", depth) - if "error" in query_result: - suggestions = query_result.get("suggestions", []) - msg = query_result["error"] - if suggestions: - msg += "\nDid you mean: " + ", ".join(suggestions[:5]) - formatter.error(message=msg, error_code=ErrorCode.NODE_NOT_FOUND) - raise typer.Exit(code=1) + _emit_query_warnings(formatter, query_result) if formatter.json_mode: if column: query_result = _filter_column_json(query_result, column) formatter.output(query_result) elif format in ("mermaid", "html", "er"): - _output_mermaid_or_html(formatter, service, graph, query_result, "downstream", format) + _output_mermaid_or_html(formatter, service, graph, query_result, direction, format) else: - _format_lineage_tree(formatter, graph, query_result, "downstream", **display_opts) + _format_lineage_tree(formatter, graph, query_result, direction, **display_opts) # -- Output formatting helpers ---------------------------------------------- +def _emit_query_warnings(formatter: OutputFormatter, query_result: dict) -> None: + """Print non-fatal warnings carried by a lineage query result. + + Currently only the ambiguous-identifier warning (the same table id exists + in several projects). ``formatter.warning`` is a no-op in JSON mode -- + there the warnings ride along in the emitted payload instead. + """ + for warning in query_result.get("warnings", []): + formatter.warning(warning) + + def _output_mermaid_or_html( formatter, service, diff --git a/src/keboola_agent_cli/services/deep_lineage_service.py b/src/keboola_agent_cli/services/deep_lineage_service.py index d310d6f7..85e579b8 100644 --- a/src/keboola_agent_cli/services/deep_lineage_service.py +++ b/src/keboola_agent_cli/services/deep_lineage_service.py @@ -371,6 +371,47 @@ def to_dict(self) -> dict: # --------------------------------------------------------------------------- +# Keep a warning readable when a bare table name matches across many buckets. +_MAX_LISTED_CANDIDATES = 5 + + +def _format_candidates(items: list[str]) -> str: + """Join ids for display, trimming a long tail rather than printing all of it.""" + head = items[:_MAX_LISTED_CANDIDATES] + tail = f", +{len(items) - len(head)} more" if len(items) > len(head) else "" + return ", ".join(head) + tail + + +def _ambiguity_warning(identifier: str, candidates: list[str]) -> str: + """Build the warning shown when an identifier matches more than one node. + + Two different shapes reach this. The one #568 reports is the same + ``bucket_id.table_name`` living in several project namespaces (a bucket + shared from one project and linked into another) -- there every candidate + differs only by project, and ``:`` is a retry that + resolves. The other comes from the name-only fallback in + ``_find_node_candidates``, which matches a bare table name across buckets: + those candidates can share a single project, so counting them as projects + would tell the user a table "exists in 2 projects (alpha, alpha)", and + ``:`` would not resolve because the real node ids + carry a bucket. That case gets the full ids instead. + """ + shown = candidates[0] + if all(fqn.partition(":")[2] == identifier for fqn in candidates): + projects = sorted({fqn.partition(":")[0] for fqn in candidates}) + return ( + f"'{identifier}' exists in {len(projects)} projects " + f"({_format_candidates(projects)}); showing '{shown}' only. " + f"Query a specific one with '--upstream/--downstream " + f":{identifier}' or scope with --project." + ) + return ( + f"'{identifier}' matches {len(candidates)} nodes " + f"({_format_candidates(candidates)}); showing '{shown}' only. " + f"Query a specific one by its full id, e.g. '--upstream/--downstream {shown}'." + ) + + class DeepLineageService: """Business logic for column-level lineage from sync'd data on disk. @@ -473,18 +514,7 @@ def query_upstream( depth: int = 10, ) -> dict[str, Any]: """Query upstream dependencies of a node.""" - fqn = self._find_node(graph, identifier, project) - if not fqn: - return { - "error": f"Node not found: {identifier}", - "suggestions": self._suggest(graph, identifier), - } - return { - "node": fqn, - "direction": "upstream", - "node_info": self._node_info(graph, fqn), - "edges": graph.get_upstream(fqn, depth), - } + return self._query(graph, identifier, project, depth, direction="upstream") def query_downstream( self, @@ -494,18 +524,48 @@ def query_downstream( depth: int = 10, ) -> dict[str, Any]: """Query downstream dependents of a node.""" - fqn = self._find_node(graph, identifier, project) - if not fqn: + return self._query(graph, identifier, project, depth, direction="downstream") + + def _query( + self, + graph: LineageGraph, + identifier: str, + project: str, + depth: int, + *, + direction: str, + ) -> dict[str, Any]: + """Resolve ``identifier`` and walk the graph in ``direction``. + + When an unqualified identifier resolves to more than one project (the + same ``bucket_id.table_name`` exists in several project namespaces -- + typically a shared/linked bucket), the first candidate is still used, + but the result carries ``ambiguous_matches`` and a human-readable + ``warnings`` entry so callers can surface the ambiguity instead of + presenting one project's answer as the whole picture. + """ + candidates = self._find_node_candidates(graph, identifier, project) + if not candidates: return { "error": f"Node not found: {identifier}", "suggestions": self._suggest(graph, identifier), } - return { + + fqn = candidates[0] + result: dict[str, Any] = { "node": fqn, - "direction": "downstream", + "direction": direction, "node_info": self._node_info(graph, fqn), - "edges": graph.get_downstream(fqn, depth), + "edges": ( + graph.get_upstream(fqn, depth) + if direction == "upstream" + else graph.get_downstream(fqn, depth) + ), } + if len(candidates) > 1: + result["ambiguous_matches"] = candidates + result["warnings"] = [_ambiguity_warning(identifier, candidates)] + return result # --- Internal methods --- @@ -1004,30 +1064,39 @@ def _graph_from_dict(data: dict) -> LineageGraph: return graph def _find_node(self, graph: LineageGraph, identifier: str, project: str = "") -> str | None: - if ":" in identifier: - all_fqns = set(graph.tables) | set(graph.configurations) - for e in graph.edges: - all_fqns.add(e.source_fqn) - all_fqns.add(e.target_fqn) - return identifier if identifier in all_fqns else None - + candidates = self._find_node_candidates(graph, identifier, project) + return candidates[0] if candidates else None + + def _find_node_candidates( + self, graph: LineageGraph, identifier: str, project: str = "" + ) -> list[str]: + """Return every FQN ``identifier`` could refer to, best match first. + + A fully-qualified ``project:bucket_id.table_name`` (or an identifier + combined with an explicit ``project``) is unambiguous by construction, + so at most one candidate comes back. A bare ``bucket_id.table_name`` + can exist in several project namespaces at once -- a bucket shared + from one project and linked into another yields a node per project -- + in which case every match is returned so the caller can report the + ambiguity rather than silently answering for one of them. + """ all_fqns = set(graph.tables) | set(graph.configurations) for e in graph.edges: all_fqns.add(e.source_fqn) all_fqns.add(e.target_fqn) + if ":" in identifier: + return [identifier] if identifier in all_fqns else [] + if project: fqn = f"{project}:{identifier}" - return fqn if fqn in all_fqns else None + return [fqn] if fqn in all_fqns else [] - matches = [f for f in all_fqns if f.endswith(f":{identifier}")] - if len(matches) == 1: - return matches[0] + matches = sorted(f for f in all_fqns if f.endswith(f":{identifier}")) if matches: - return sorted(matches)[0] + return matches - partial = [f for f in all_fqns if f.split(":")[-1].endswith(f".{identifier}")] - return sorted(partial)[0] if partial else None + return sorted(f for f in all_fqns if f.split(":")[-1].endswith(f".{identifier}")) # --- Mermaid / HTML rendering --- diff --git a/tests/test_deep_lineage_service.py b/tests/test_deep_lineage_service.py index bb4bd1f3..ac529a42 100644 --- a/tests/test_deep_lineage_service.py +++ b/tests/test_deep_lineage_service.py @@ -528,6 +528,169 @@ def test_cache_roundtrip(self, tmp_path: Path) -> None: assert len(graph.edges) >= 2 +# --------------------------------------------------------------------------- +# Ambiguous identifier resolution (issue #568) +# --------------------------------------------------------------------------- + + +SHARED_TABLE = "in.c-shared.orders" + + +def _shared_table_graph() -> LineageGraph: + """Graph where the same table id is a node in two projects. + + Mirrors a bucket shared from ``alpha`` and linked into ``beta``: each + project has its own node for ``in.c-shared.orders``, and each has a + dependent config the other one does not have. + """ + graph = LineageGraph() + for project, config_id in (("alpha", "100"), ("beta", "200")): + graph.add_edge( + Edge( + source_fqn=f"{project}:{SHARED_TABLE}", + target_fqn=f"{project}:config/{config_id}", + source_type="table", + target_type="config", + edge_type="reads", + detection="input_mapping", + ) + ) + graph.add_edge( + Edge( + source_fqn=f"{project}:in.c-raw.source", + target_fqn=f"{project}:{SHARED_TABLE}", + source_type="table", + target_type="table", + edge_type="writes", + detection="input_mapping", + ) + ) + return graph + + +class TestAmbiguousNodeResolution: + """An unqualified table id present in N projects must warn, not go silent.""" + + def _service(self, tmp_path: Path) -> DeepLineageService: + (tmp_path / "cfg").mkdir(exist_ok=True) + return DeepLineageService(config_store=ConfigStore(config_dir=tmp_path / "cfg")) + + def test_downstream_warns_and_lists_every_candidate(self, tmp_path: Path) -> None: + service = self._service(tmp_path) + result = service.query_downstream(_shared_table_graph(), SHARED_TABLE) + + assert "error" not in result + # Still deterministic: the alphabetically first project is used. + assert result["node"] == f"alpha:{SHARED_TABLE}" + assert result["ambiguous_matches"] == [ + f"alpha:{SHARED_TABLE}", + f"beta:{SHARED_TABLE}", + ] + assert len(result["warnings"]) == 1 + warning = result["warnings"][0] + assert "2 projects" in warning + assert "alpha" in warning and "beta" in warning + + def test_upstream_warns_too(self, tmp_path: Path) -> None: + service = self._service(tmp_path) + result = service.query_upstream(_shared_table_graph(), SHARED_TABLE) + + assert result["direction"] == "upstream" + assert result["ambiguous_matches"] == [ + f"alpha:{SHARED_TABLE}", + f"beta:{SHARED_TABLE}", + ] + + def test_explicit_fqn_is_not_ambiguous(self, tmp_path: Path) -> None: + service = self._service(tmp_path) + result = service.query_downstream(_shared_table_graph(), f"beta:{SHARED_TABLE}") + + assert result["node"] == f"beta:{SHARED_TABLE}" + assert "ambiguous_matches" not in result + assert "warnings" not in result + assert result["edges"][0]["target"] == "beta:config/200" + + def test_project_scope_disambiguates(self, tmp_path: Path) -> None: + service = self._service(tmp_path) + result = service.query_downstream(_shared_table_graph(), SHARED_TABLE, project="beta") + + assert result["node"] == f"beta:{SHARED_TABLE}" + assert "warnings" not in result + + def test_unique_identifier_stays_quiet(self, tmp_path: Path) -> None: + service = self._service(tmp_path) + graph = LineageGraph() + graph.add_edge( + Edge( + source_fqn="alpha:in.c-solo.events", + target_fqn="alpha:config/1", + source_type="table", + target_type="config", + edge_type="reads", + detection="input_mapping", + ) + ) + + result = service.query_downstream(graph, "in.c-solo.events") + + assert result["node"] == "alpha:in.c-solo.events" + assert "ambiguous_matches" not in result + assert "warnings" not in result + + def _one_project_two_buckets(self) -> LineageGraph: + """A bare table name that the name-only fallback matches twice in ONE project.""" + graph = LineageGraph() + for bucket in ("in.c-a", "in.c-b"): + graph.add_edge( + Edge( + source_fqn=f"alpha:{bucket}.orders", + target_fqn=f"alpha:config/{bucket}", + source_type="table", + target_type="config", + edge_type="reads", + detection="input_mapping", + ) + ) + return graph + + def test_name_only_match_does_not_claim_several_projects(self, tmp_path: Path) -> None: + """Candidates sharing one project must not be counted as that many projects.""" + service = self._service(tmp_path) + + result = service.query_downstream(self._one_project_two_buckets(), "orders") + + assert result["ambiguous_matches"] == ["alpha:in.c-a.orders", "alpha:in.c-b.orders"] + warning = result["warnings"][0] + assert "2 projects" not in warning + assert "alpha, alpha" not in warning + assert "2 nodes" in warning + + def test_name_only_match_suggests_a_retry_that_resolves(self, tmp_path: Path) -> None: + """The remedy the warning offers must name a node that actually exists. + + ``:orders`` does not -- the real ids carry a bucket -- so the + warning has to point at the full candidate ids instead. + """ + service = self._service(tmp_path) + graph = self._one_project_two_buckets() + + warning = service.query_downstream(graph, "orders")["warnings"][0] + + assert ":orders" not in warning + suggested = "alpha:in.c-a.orders" + assert suggested in warning + retry = service.query_downstream(graph, suggested) + assert "error" not in retry + assert retry["node"] == suggested + + def test_unknown_identifier_still_errors(self, tmp_path: Path) -> None: + service = self._service(tmp_path) + result = service.query_downstream(_shared_table_graph(), "in.c-nope.missing") + + assert "error" in result + assert "warnings" not in result + + # --------------------------------------------------------------------------- # CLI tests via CliRunner # --------------------------------------------------------------------------- @@ -608,6 +771,61 @@ def test_missing_cache_file(self) -> None: ) assert result.exit_code == 1 + def _ambiguous_cache(self, tmp_path: Path) -> Path: + cache_path = tmp_path / "shared-lineage.json" + cache_path.write_text(json.dumps(_shared_table_graph().to_dict())) + return cache_path + + def test_ambiguous_query_warns_on_stderr(self, tmp_path: Path) -> None: + """Human mode: the ambiguity is a stderr warning, stdout stays the tree.""" + from keboola_agent_cli.cli import app + + runner_local = __import__("typer.testing", fromlist=["CliRunner"]).CliRunner() + result = runner_local.invoke( + app, + [ + "lineage", + "show", + "--load", + str(self._ambiguous_cache(tmp_path)), + "--downstream", + SHARED_TABLE, + ], + ) + + assert result.exit_code == 0, result.output + stderr = " ".join(result.stderr.split()) + assert "Warning:" in stderr + assert "exists in 2 projects" in stderr + assert "alpha" in stderr and "beta" in stderr + + def test_ambiguous_query_json_carries_candidates(self, tmp_path: Path) -> None: + """JSON mode: warnings ride along in the payload, stdout stays parseable.""" + from keboola_agent_cli.cli import app + + runner_local = __import__("typer.testing", fromlist=["CliRunner"]).CliRunner() + result = runner_local.invoke( + app, + [ + "--json", + "lineage", + "show", + "--load", + str(self._ambiguous_cache(tmp_path)), + "--downstream", + SHARED_TABLE, + ], + ) + + assert result.exit_code == 0, result.output + data = json.loads(result.stdout)["data"] + assert data["node"] == f"alpha:{SHARED_TABLE}" + assert data["ambiguous_matches"] == [ + f"alpha:{SHARED_TABLE}", + f"beta:{SHARED_TABLE}", + ] + assert data["warnings"] + # --------------------------------------------------------------------------- # Sync layout handling: flat (single project in CWD) vs. nested (multi-project)