reata__sqllineage-224

{
  "status": "ok",
  "stdout": "Source Tables: [Table: <default>.foo, Column: <default>.foo.col]\nTarget Tables: [Table: <default>.foo, Column: <default>.foo.col]\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "sqllineage/core/holders.py:178:    def source_tables(self) -> Set[Table]:\nsqllineage/core/holders.py:182:        source_tables = {\nsqllineage/core/holders.py:187:        source_tables |= self._selfloop_tables\nsqllineage/core/holders.py:188:        source_tables |= self._sourceonly_tables\nsqllineage/core/holders.py:189:        return source_tables\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "sqllineage/runner.py",
  "content": "import logging\nfrom typing import Dict, List, Tuple\n\nimport sqlparse\nfrom sqlparse.sql import Statement\n\nfrom sqllineage.core import LineageAnalyzer\nfrom sqllineage.core.holders import SQLLineageHolder\nfrom sqllineage.core.models import Column, Table\nfrom sqllineage.drawing import draw_lineage_graph\nfrom sqllineage.io import to_cytoscape\nfrom sqllineage.utils.constant import LineageLevel\n\n\nlogger = logging.getLogger(__name__)\n\n\ndef lazy_method(func):\n    def wrapper(*args, **kwargs):\n        self = args[0]\n        if not self._evaluated:\n            self._eval()\n        return func(*args, **kwargs)\n\n    return wrapper\n\n\ndef lazy_property(func):\n    return property(lazy_method(func))\n\n\nclass LineageRunner(object):\n    def __init__(\n        self,\n        sql: str,\n        encoding: str = None,\n        verbose: bool = False,\n        draw_options: Dict[str, str] = None,\n    ):\n        \"\"\"\n        The entry point of SQLLineage after command line options are parsed.\n\n        :param sql: a string representation of SQL statements.\n        :param encoding: the encoding for sql string\n        :param verbose: verbose flag indicate whether statement-wise lineage result will be shown\n        \"\"\"\n        self._encoding = encoding\n        self._sql = sql\n        self._verbose = verbose\n        self._draw_options = draw_options if draw_options else {}\n        self._evaluated = False\n        self._stmt: List[Statement] = []\n\n    @lazy_method\n    def __str__(self):\n        \"\"\"\n        print out the Lineage Summary.\n        \"\"\"\n        statements = self.statements(strip_comments=True)\n        source_tables = \"\\n    \".join(str(t) for t in self.source_tables)\n        target_tables = \"\\n    \".join(str(t) for t in self.target_tables)\n        combined = f\"\"\"Statements(#): {len(statements)}\nSource Tables:\n    {source_tables}\nTarget Tables:\n    {target_tables}\n\"\"\"\n        if self.intermediate_tables:\n            intermediate_tables = \"\\n    \".join(\n                str(t) for t in self.intermediate_tables\n            )\n            combined += f\"\"\"Intermediate Tables:\n    {intermediate_tables}\"\"\"\n        if self._verbose:\n            result = \"\"\n            for i, holder in enumerate(self._stmt_holders):\n                stmt_short = statements[i].replace(\"\\n\", \"\")\n                if len(stmt_short) > 50:\n                    stmt_short = stmt_short[:50] + \"...\"\n                content = str(holder).replace(\"\\n\", \"\\n    \")\n                result += f\"\"\"Statement #{i + 1}: {stmt_short}\n    {content}\n\"\"\"\n            combined = result + \"==========\\nSummary:\\n\" + combined\n        return combined\n\n    @lazy_method\n    def to_cytoscape(self, level=LineageLevel.TABLE) -> List[Dict[str, Dict[str, str]]]:\n        \"\"\"\n        to turn the DAG into cytoscape format.\n        \"\"\"\n        if level == LineageLevel.COLUMN:\n            return to_cytoscape(self._sql_holder.column_lineage_graph, compound=True)\n        else:\n            return to_cytoscape(self._sql_holder.table_lineage_graph)\n\n    def draw(self) -> None:\n        \"\"\"\n        to draw the lineage directed graph\n        \"\"\"\n        draw_options = self._draw_options\n        if draw_options.get(\"f\") is None:\n            draw_options.pop(\"f\", None)\n            draw_options[\"e\"] = self._sql\n        return draw_lineage_graph(**draw_options)\n\n    @lazy_method\n    def statements(self, **kwargs) -> List[str]:\n        \"\"\"\n        a list of statements.\n\n        :param kwargs: the key arguments that will be passed to `sqlparse.format`\n        \"\"\"\n        return [sqlparse.format(s.value, **kwargs) for s in self.statements_parsed]\n\n    @lazy_property\n    def statements_parsed(self) -> List[Statement]:\n        \"\"\"\n        a list of :class:`sqlparse.sql.Statement`\n        \"\"\"\n        return self._stmt\n\n    @lazy_property\n    def source_tables(self) -> List[Table]:\n        \"\"\"\n        a list of source :class:`sqllineage.models.Table`\n        \"\"\"\n        return sorted(self._sql_holder.source_tables, key=lambda x: str(x))\n\n    @lazy_property\n    def target_tables(self) -> List[Table]:\n        \"\"\"\n        a list of target :class:`sqllineage.models.Table`\n        \"\"\"\n        return sorted(self._sql_holder.target_tables, key=lambda x: str(x))\n\n    @lazy_property\n    def intermediate_tables(self) -> List[Table]:\n        \"\"\"\n        a list of intermediate :class:`sqllineage.models.Table`\n        \"\"\"\n        return sorted(self._sql_holder.intermediate_tables, key=lambda x: str(x))\n\n    @lazy_method\n    def get_column_lineage(self, exclude_subquery=True) -> List[Tuple[Column, Column]]:\n        \"\"\"\n        a list of column tuple :class:`sqllineage.models.Column`\n        \"\"\"\n        # sort by target column, and then source column\n        return sorted(",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 181,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "sqllineage/core/holders.py",
  "content": "\n    @property\n    def rename(self) -> Set[Tuple[Table, Table]]:\n        return {\n            (src, tgt)\n            for src, tgt, attr in self.graph.edges(data=True)\n            if attr.get(\"type\") == EdgeType.RENAME\n        }\n\n    def add_rename(self, src: Table, tgt: Table) -> None:\n        self.graph.add_edge(src, tgt, type=EdgeType.RENAME)\n\n    @staticmethod\n    def of(holder: SubQueryLineageHolder):\n        stmt_holder = StatementLineageHolder()\n        stmt_holder.graph = holder.graph\n        return stmt_holder\n\n\nclass SQLLineageHolder(ColumnLineageMixin):\n    def __init__(self, graph: DiGraph):\n        \"\"\"\n        The combined lineage result in representation of Directed Acyclic Graph.\n\n        :param graph: the Directed Acyclic Graph holding all the combined lineage result.\n        \"\"\"\n        self.graph = graph\n        self._selfloop_tables = self.__retrieve_tag_tables(NodeTag.SELFLOOP)\n        self._sourceonly_tables = self.__retrieve_tag_tables(NodeTag.SOURCE_ONLY)\n        self._targetonly_tables = self.__retrieve_tag_tables(NodeTag.TARGET_ONLY)\n\n    @property\n    def table_lineage_graph(self) -> DiGraph:\n        \"\"\"\n        The table level DiGraph held by SQLLineageHolder\n        \"\"\"\n        table_nodes = [n for n in self.graph.nodes if isinstance(n, DATASET_CLASSES)]\n        return self.graph.subgraph(table_nodes)\n\n    @property\n    def column_lineage_graph(self) -> DiGraph:\n        \"\"\"\n        The column level DiGraph held by SQLLineageHolder\n        \"\"\"\n        column_nodes = [n for n in self.graph.nodes if isinstance(n, Column)]\n        return self.graph.subgraph(column_nodes)\n\n    @property\n    def source_tables(self) -> Set[Table]:\n        \"\"\"\n        a list of source :class:`sqllineage.models.Table`\n        \"\"\"\n        source_tables = {\n            table for table, deg in self.table_lineage_graph.in_degree if deg == 0\n        }.intersection(\n            {table for table, deg in self.table_lineage_graph.out_degree if deg > 0}\n        )\n        source_tables |= self._selfloop_tables\n        source_tables |= self._sourceonly_tables\n        return source_tables\n\n    @property\n    def target_tables(self) -> Set[Table]:\n        \"\"\"\n        a list of target :class:`sqllineage.models.Table`\n        \"\"\"\n        target_tables = {\n            table for table, deg in self.table_lineage_graph.out_degree if deg == 0\n        }.intersection(\n            {table for table, deg in self.table_lineage_graph.in_degree if deg > 0}\n        )\n        target_tables |= self._selfloop_tables\n        target_tables |= self._targetonly_tables\n        return target_tables\n\n    @property\n    def intermediate_tables(self) -> Set[Table]:\n        \"\"\"\n        a list of intermediate :class:`sqllineage.models.Table`\n        \"\"\"\n        intermediate_tables = {\n            table for table, deg in self.table_lineage_graph.in_degree if deg > 0\n        }.intersection(\n            {table for table, deg in self.table_lineage_graph.out_degree if deg > 0}\n        )\n        intermediate_tables -= self.__retrieve_tag_tables(NodeTag.SELFLOOP)\n        return intermediate_tables\n\n    def __retrieve_tag_tables(self, tag) -> Set[Table]:\n        return {\n            table",
  "start_line": 130,
  "end_line": 220,
  "total_lines": 280,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "sqllineage/core/holders.py",
  "content": "import itertools\nfrom typing import Set, Tuple, Union\n\nimport networkx as nx\nfrom networkx import DiGraph\n\nfrom sqllineage.core.models import Column, Path, SubQuery, Table\nfrom sqllineage.utils.constant import EdgeType, NodeTag\n\nDATASET_CLASSES = (Path, Table)\n\n\nclass ColumnLineageMixin:\n    def get_column_lineage(self, exclude_subquery=True) -> Set[Tuple[Column, ...]]:\n        self.graph: DiGraph  # For mypy attribute checking\n        # filter all the column node in the graph\n        column_nodes = [n for n in self.graph.nodes if isinstance(n, Column)]\n        column_graph = self.graph.subgraph(column_nodes)\n        source_columns = {column for column, deg in column_graph.in_degree if deg == 0}\n        # if a column lineage path ends at SubQuery, then it should be pruned\n        target_columns = {\n            node\n            for node, deg in column_graph.out_degree\n            if isinstance(node, Column) and deg == 0\n        }\n        if exclude_subquery:\n            target_columns = {\n                node for node in target_columns if isinstance(node.parent, Table)\n            }\n        columns = set()\n        for (source, target) in itertools.product(source_columns, target_columns):\n            simple_paths = list(nx.all_simple_paths(self.graph, source, target))\n            if len(simple_paths) == 1:\n                columns.add(tuple(simple_paths[0]))\n            # we can ignore when simple path doesn't exist, but could there be more than one simple path?\n        return columns\n\n\nclass SubQueryLineageHolder(ColumnLineageMixin):\n    \"\"\"\n    SubQuery/Query Level Lineage Result.\n\n    SubQueryLineageHolder will hold attributes like read, write, cte\n\n    Each of them is a Set[:class:`sqllineage.models.Table`].\n\n    This is the most atomic representation of lineage result.\n    \"\"\"\n\n    def __init__(self) -> None:\n        self.graph = nx.DiGraph()\n\n    def __or__(self, other):\n        self.graph = nx.compose(self.graph, other.graph)\n        return self\n\n    def _property_getter(self, prop) -> Set[Union[SubQuery, Table]]:\n        return {t for t, attr in self.graph.nodes(data=True) if attr.get(prop) is True}\n\n    def _property_setter(self, value, prop) -> None:\n        self.graph.add_node(value, **{prop: True})\n\n    @property\n    def read(self) -> Set[Union[SubQuery, Table]]:\n        return self._property_getter(NodeTag.READ)\n\n    def add_read(self, value) -> None:\n        self._property_setter(value, NodeTag.READ)\n        # the same table can be added (in SQL: joined) multiple times with different alias\n        if hasattr(value, \"alias\"):\n            self.graph.add_edge(value, value.alias, type=EdgeType.HAS_ALIAS)\n\n    @property\n    def write(self) -> Set[Union[SubQuery, Table]]:\n        return self._property_getter(NodeTag.WRITE)\n\n    def add_write(self, value) -> None:\n        self._property_setter(value, NodeTag.WRITE)\n\n    @property\n    def cte(self) -> Set[SubQuery]:\n        return self._property_getter(NodeTag.CTE)  # type: ignore\n\n    def add_cte(self, value) -> None:\n        self._property_setter(value, NodeTag.CTE)\n\n    def add_column_lineage(self, src: Column, tgt: Column) -> None:\n        self.graph.add_edge(src, tgt, type=EdgeType.LINEAGE)\n        self.graph.add_edge(tgt.parent, tgt, type=EdgeType.HAS_COLUMN)\n        if src.parent is not None:\n            # starting NetworkX v2.6, None is not allowed as node, see https://github.com/networkx/networkx/pull/4892\n            self.graph.add_edge(src.parent, src, type=EdgeType.HAS_COLUMN)\n\n\nclass StatementLineageHolder(SubQueryLineageHolder, ColumnLineageMixin):\n    \"\"\"\n    Statement Level Lineage Result.\n\n    Based on SubQueryLineageHolder, StatementLineageHolder holds extra attributes like drop and rename\n",
  "start_line": 1,
  "end_line": 100,
  "total_lines": 280,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "sqllineage/core/models.py",
  "content": "import warnings\nfrom typing import Dict, List, Optional, Set, Union\n\nfrom sqlparse import tokens as T\nfrom sqlparse.engine import grouping\nfrom sqlparse.sql import (\n    Case,\n    Comparison,\n    Function,\n    Identifier,\n    IdentifierList,\n    Operation,\n    Parenthesis,\n    Token,\n    TokenList,\n)\nfrom sqlparse.utils import imt\n\nfrom sqllineage.exceptions import SQLLineageException\nfrom sqllineage.utils.entities import ColumnQualifierTuple\nfrom sqllineage.utils.helpers import escape_identifier_name\nfrom sqllineage.utils.sqlparse import get_parameters, is_subquery\n\n\nclass Schema:\n    unknown = \"<default>\"\n\n    def __init__(self, name: str = unknown):\n        \"\"\"\n        Data Class for Schema\n\n        :param name: schema name\n        \"\"\"\n        self.raw_name = escape_identifier_name(name)\n\n    def __str__(self):\n        return self.raw_name.lower()\n\n    def __repr__(self):\n        return \"Schema: \" + str(self)\n\n    def __eq__(self, other):\n        return type(self) is type(other) and str(self) == str(other)\n\n    def __hash__(self):\n        return hash(str(self))\n\n    def __bool__(self):\n        return str(self) != self.unknown\n\n\nclass Table:\n    def __init__(self, name: str, schema: Schema = Schema(), **kwargs):\n        \"\"\"\n        Data Class for Table\n\n        :param name: table name\n        :param schema: schema as defined by :class:`Schema`\n        \"\"\"\n        if \".\" not in name:\n            self.schema = schema\n            self.raw_name = escape_identifier_name(name)\n        else:\n            schema_name, table_name = name.rsplit(\".\", 1)\n            if len(schema_name.split(\".\")) > 2:\n                # allow db.schema as schema_name, but a.b.c as schema_name is forbidden\n                raise SQLLineageException(\"Invalid format for table name: %s.\", name)\n            self.schema = Schema(schema_name)\n            self.raw_name = escape_identifier_name(table_name)\n            if schema:\n                warnings.warn(\"Name is in schema.table format, schema param is ignored\")\n        self.alias = kwargs.pop(\"alias\", self.raw_name)\n\n    def __str__(self):\n        return f\"{self.schema}.{self.raw_name.lower()}\"\n\n    def __repr__(self):\n        return \"Table: \" + str(self)\n\n    def __eq__(self, other):\n        return type(self) is type(other) and str(self) == str(other)\n\n    def __hash__(self):\n        return hash(str(self))\n\n    @staticmethod\n    def of(identifier: Identifier) -> \"Table\":\n        # rewrite identifier's get_real_name method, by matching the last dot instead of the first dot, so that the\n        # real name for a.b.c will be c instead of b\n        dot_idx, _ = identifier._token_matching(\n            lambda token: imt(token, m=(T.Punctuation, \".\")),\n            start=len(identifier.tokens),\n            reverse=True,\n        )\n        real_name = identifier._get_first_name(dot_idx, real_name=True)\n        # rewrite identifier's get_parent_name accordingly\n        parent_name = (\n            \"\".join(\n                [\n                    escape_identifier_name(token.value)\n                    for token in identifier.tokens[:dot_idx]\n                ]\n            )\n            if dot_idx\n            else None\n        )\n        schema = Schema(parent_name) if parent_name is not None else Schema()\n        alias = identifier.get_alias()\n        kwargs = {\"alias\": alias} if alias else {}\n        return Table(real_name, schema, **kwargs)\n\n\nclass Path:\n    def __init__(self, uri: str):\n        self.uri = escape_identifier_name(uri)\n\n    def __str__(self):\n        return self.uri\n\n    def __repr__(self):\n        return \"Path: \" + str(self)\n\n    def __eq__(self, other):\n        return type(self) is type(other) and self.uri == other.uri\n\n    def __hash__(self):\n        return hash(self.uri)\n\n\nclass Partition:\n    pass\n\n\nclass SubQuery:\n    def __init__(self, token: Parenthesis, alias: Optional[str]):\n        \"\"\"\n        Data Class for SubQuery\n\n        :param token: subquery token\n        :param alias: subquery name\n        \"\"\"\n        self.token = token\n        self._query = token.value\n        self.alias = alias if alias is not None else f\"subquery_{hash(self)}\"\n\n    def __str__(self):\n        return self.alias\n\n    def __repr__(self):\n        return \"SubQuery: \" + str(self)",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 348,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "sqllineage/core/models.py",
  "content": "\n    def __eq__(self, other):\n        return type(self) is type(other) and self._query == other._query\n\n    def __hash__(self):\n        return hash(self._query)\n\n    @staticmethod\n    def of(parenthesis: Parenthesis, alias: Optional[str]) -> \"SubQuery\":\n        return SubQuery(parenthesis, alias)\n\n\nclass Column:\n    def __init__(self, name: str, **kwargs):\n        \"\"\"\n        Data Class for Column\n\n        :param name: column name\n        :param parent: :class:`Table` or :class:`SubQuery`\n        :param kwargs:\n        \"\"\"\n        self._parent: Set[Union[Table, SubQuery]] = set()\n        self.raw_name = escape_identifier_name(name)\n        self.source_columns = kwargs.pop(\"source_columns\", ((self.raw_name, None),))\n\n    def __str__(self):\n        return (\n            f\"{self.parent}.{self.raw_name.lower()}\"\n            if self.parent is not None and not isinstance(self.parent, Path)\n            else f\"{self.raw_name.lower()}\"\n        )\n\n    def __repr__(self):\n        return \"Column: \" + str(self)\n\n    def __eq__(self, other):\n        return type(self) is type(other) and str(self) == str(other)\n\n    def __hash__(self):\n        return hash(str(self))\n\n    @property\n    def parent(self) -> Optional[Union[Table, SubQuery]]:\n        return list(self._parent)[0] if len(self._parent) == 1 else None\n\n    @parent.setter\n    def parent(self, value: Union[Table, SubQuery]):\n        self._parent.add(value)\n\n    @property\n    def parent_candidates(self) -> List[Union[Table, SubQuery]]:\n        return sorted(self._parent, key=lambda p: str(p))\n\n    @staticmethod\n    def of(token: Token):\n        if isinstance(token, Identifier):\n            alias = token.get_alias()\n            if alias:\n                # handle column alias, including alias for column name or Case, Function\n                kw_idx, kw = token.token_next_by(m=(T.Keyword, \"AS\"))\n                if kw_idx is None:\n                    # alias without AS\n                    kw_idx, _ = token.token_next_by(i=Identifier)\n                if kw_idx is None:\n                    # invalid syntax: col AS, without alias\n                    return Column(alias)\n                else:\n                    idx, _ = token.token_prev(kw_idx, skip_cm=True)\n                    expr = grouping.group(TokenList(token.tokens[: idx + 1]))[0]\n                    source_columns = Column._extract_source_columns(expr)\n                    return Column(alias, source_columns=source_columns)\n            else:\n                # select column name directly without alias\n                return Column(\n                    token.get_real_name(),\n                    source_columns=((token.get_real_name(), token.get_parent_name()),),\n                )\n        else:\n            # Wildcard, Case, Function without alias (thus not recognized as an Identifier)\n            source_columns = Column._extract_source_columns(token)\n            return Column(token.value, source_columns=source_columns)\n\n    @staticmethod\n    def _extract_source_columns(token: Token) -> List[ColumnQualifierTuple]:\n        if isinstance(token, Function):\n            # max(col1) AS col2\n            source_columns = [\n                cqt\n                for tk in get_parameters(token)\n                for cqt in Column._extract_source_columns(tk)\n            ]\n        elif isinstance(token, Parenthesis):\n            if is_subquery(token):\n                # This is to avoid circular import\n                from sqllineage.runner import LineageRunner\n\n                # (SELECT avg(col1) AS col1 FROM tab3), used after WHEN or THEN in CASE clause\n                src_cols = [\n                    lineage[0]\n                    for lineage in LineageRunner(token.value).get_column_lineage(\n                        exclude_subquery=False\n                    )\n                ]\n                source_columns = [\n                    ColumnQualifierTuple(src_col.raw_name, src_col.parent.raw_name)\n                    for src_col in src_cols\n                ]\n            else:\n                # (col1 + col2) AS col3\n                source_columns = [\n                    cqt\n                    for tk in token.tokens[1:-1]\n                    for cqt in Column._extract_source_columns(tk)\n                ]\n        elif isinstance(token, Operation):\n            # col1 + col2 AS col3\n            source_columns = [\n                cqt\n                for tk in token.get_sublists()\n                for cqt in Column._extract_source_columns(tk)\n            ]\n        elif isinstance(token, Case):\n            # CASE WHEN col1 = 2 THEN \"V1\" WHEN col1 = \"2\" THEN \"V2\" END AS col2\n            source_columns = [\n                cqt\n                for tk in token.get_sublists()\n                for cqt in Column._extract_source_columns(tk)\n            ]\n        elif isinstance(token, Comparison):\n            source_columns = Column._extract_source_columns(\n                token.left\n            ) + Column._extract_source_columns(token.right)\n        elif isinstance(token, IdentifierList):\n            source_columns = [\n                cqt\n                for tk in token.get_sublists()\n                for cqt in Column._extract_source_columns(tk)\n            ]\n        elif isinstance(token, Identifier):\n            if token.get_real_name():\n                # col1 AS col2\n                source_columns = [\n                    ColumnQualifierTuple(token.get_real_name(), token.get_parent_name())\n                ]\n            else:\n                # col1=1 AS int\n                source_columns = [\n                    cqt\n                    for tk in token.get_sublists()\n                    for cqt in Column._extract_source_columns(tk)",
  "start_line": 151,
  "end_line": 300,
  "total_lines": 348,
  "is_truncated": true
}

{
  "status": "ok",
  "stdout": "sqllineage/core/holders.py:8:from sqllineage.utils.constant import EdgeType, NodeTag\nsqllineage/core/holders.py:65:        return self._property_getter(NodeTag.READ)\nsqllineage/core/holders.py:68:        self._property_setter(value, NodeTag.READ)\nsqllineage/core/holders.py:75:        return self._property_getter(NodeTag.WRITE)\nsqllineage/core/holders.py:78:        self._property_setter(value, NodeTag.WRITE)\nsqllineage/core/holders.py:82:        return self._property_getter(NodeTag.CTE)  # type: ignore\nsqllineage/core/holders.py:85:        self._property_setter(value, NodeTag.CTE)\nsqllineage/core/holders.py:126:        return self._property_getter(NodeTag.DROP)  # type: ignore\nsqllineage/core/holders.py:129:        self._property_setter(value, NodeTag.DROP)\nsqllineage/core/holders.py:157:        self._selfloop_tables = self.__retrieve_tag_tables(NodeTag.SELFLOOP)\nsqllineage/core/holders.py:158:        self._sourceonly_tables = self.__retrieve_tag_tables(NodeTag.SOURCE_ONLY)\nsqllineage/core/holders.py:159:        self._targetonly_tables = self.__retrieve_tag_tables(NodeTag.TARGET_ONLY)\nsqllineage/core/holders.py:215:        intermediate_tables -= self.__retrieve_tag_tables(NodeTag.SELFLOOP)\nsqllineage/core/holders.py:248:                    g.add_nodes_from(read, **{NodeTag.SOURCE_ONLY: True})\nsqllineage/core/holders.py:251:                    g.add_nodes_from(write, **{NodeTag.TARGET_ONLY: True})\nsqllineage/core/holders.py:258:            g.nodes[table][NodeTag.SELFLOOP] = True\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "sqllineage/core/holders.py",
  "content": "        )\n        target_tables |= self._selfloop_tables\n        target_tables |= self._targetonly_tables\n        return target_tables\n\n    @property\n    def intermediate_tables(self) -> Set[Table]:\n        \"\"\"\n        a list of intermediate :class:`sqllineage.models.Table`\n        \"\"\"\n        intermediate_tables = {\n            table for table, deg in self.table_lineage_graph.in_degree if deg > 0\n        }.intersection(\n            {table for table, deg in self.table_lineage_graph.out_degree if deg > 0}\n        )\n        intermediate_tables -= self.__retrieve_tag_tables(NodeTag.SELFLOOP)\n        return intermediate_tables\n\n    def __retrieve_tag_tables(self, tag) -> Set[Table]:\n        return {\n            table\n            for table, attr in self.graph.nodes(data=True)\n            if attr.get(tag) is True\n        }\n\n    @staticmethod\n    def of(*args: StatementLineageHolder):\n        \"\"\"\n        To assemble multiple :class:`sqllineage.holders.StatementLineageHolder` into\n        :class:`sqllineage.holders.SQLLineageHolder`\n        \"\"\"\n        g = DiGraph()\n        for holder in args:\n            g = nx.compose(g, holder.graph)\n            if holder.drop:\n                for table in holder.drop:\n                    if g.has_node(table) and g.degree[table] == 0:\n                        g.remove_node(table)\n            elif holder.rename:\n                for (table_old, table_new) in holder.rename:\n                    g = nx.relabel_nodes(g, {table_old: table_new})\n                    g.remove_edge(table_new, table_new)\n                    if g.degree[table_new] == 0:\n                        g.remove_node(table_new)\n            else:\n                read, write = holder.read, holder.write\n                if len(read) > 0 and len(write) == 0:\n                    # source only table comes from SELECT statement\n                    g.add_nodes_from(read, **{NodeTag.SOURCE_ONLY: True})\n                elif len(read) == 0 and len(write) > 0:\n                    # target only table comes from case like: 1) INSERT/UPDATE constant values; 2) CREATE TABLE\n                    g.add_nodes_from(write, **{NodeTag.TARGET_ONLY: True})\n                else:\n                    g.add_nodes_from(read)\n                    g.add_nodes_from(write)\n                    for source, target in itertools.product(read, write):\n                        g.add_edge(source, target, type=EdgeType.LINEAGE)\n        for table in {e[0] for e in nx.selfloop_edges(g)}:\n            g.nodes[table][NodeTag.SELFLOOP] = True\n        # find all the columns that we can't assign accurately to a parent table (with multiple parent candidates)\n        unresolved_cols = [\n            (s, t)\n            for s, t in g.edges\n            if isinstance(s, Column) and len(s.parent_candidates) > 1\n        ]\n        for unresolved_col, tgt_col in unresolved_cols:\n            # check if there's only one parent candidate contains the column with same name\n            src_cols = []\n            for parent in unresolved_col.parent_candidates:\n                src_col = Column(unresolved_col.raw_name)\n                src_col.parent = parent\n                if g.has_edge(parent, src_col):\n                    src_cols.append(src_col)\n            if len(src_cols) == 1:\n                g.add_edge(src_cols[0], tgt_col, type=EdgeType.LINEAGE)\n                g.remove_edge(unresolved_col, tgt_col)\n        # when unresolved column got resolved, it will be orphan node, and we can remove it\n        for node in [n for n, deg in g.degree if deg == 0]:\n            if isinstance(node, Column) and len(node.parent_candidates) > 1:\n                g.remove_node(node)\n        return SQLLineageHolder(g)",
  "start_line": 200,
  "end_line": 280,
  "total_lines": 280,
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "Source Tables: []\nTarget Tables: []\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "Source Tables: [Table: <default>.foo]\nTarget Tables: []\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "Source Tables: [Table: <default>.foo, Column: <default>.foo.col]\nTarget Tables: [Table: <default>.foo, Column: <default>.foo.col]\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "sqllineage/core/holders.py",
  "occurrences": 1,
  "strategy": "exact",
  "diff": "--- a/sqllineage/core/holders.py\n+++ b/sqllineage/core/holders.py\n@@ -219,7 +219,7 @@\n         return {\n             table\n             for table, attr in self.graph.nodes(data=True)\n-            if attr.get(tag) is True\n+            if isinstance(table, DATASET_CLASSES) and attr.get(tag) is True\n         }\n \n     @staticmethod\n",
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "Source Tables: [Table: <default>.foo]\nTarget Tables: [Table: <default>.foo]\nCase 2 Source: [Table: <default>.foo]\nCase 3 Source: [Table: <default>.bar]\nCase 3 Target: [Table: <default>.foo]\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "sqllineage/core/holders.py:8:from sqllineage.utils.constant import EdgeType, NodeTag\nsqllineage/core/holders.py:65:        return self._property_getter(NodeTag.READ)\nsqllineage/core/holders.py:68:        self._property_setter(value, NodeTag.READ)\nsqllineage/core/holders.py:75:        return self._property_getter(NodeTag.WRITE)\nsqllineage/core/holders.py:78:        self._property_setter(value, NodeTag.WRITE)\nsqllineage/core/holders.py:82:        return self._property_getter(NodeTag.CTE)  # type: ignore\nsqllineage/core/holders.py:85:        self._property_setter(value, NodeTag.CTE)\nsqllineage/core/holders.py:126:        return self._property_getter(NodeTag.DROP)  # type: ignore\nsqllineage/core/holders.py:129:        self._property_setter(value, NodeTag.DROP)\nsqllineage/core/holders.py:157:        self._selfloop_tables = self.__retrieve_tag_tables(NodeTag.SELFLOOP)\nsqllineage/core/holders.py:158:        self._sourceonly_tables = self.__retrieve_tag_tables(NodeTag.SOURCE_ONLY)\nsqllineage/core/holders.py:159:        self._targetonly_tables = self.__retrieve_tag_tables(NodeTag.TARGET_ONLY)\nsqllineage/core/holders.py:215:        intermediate_tables -= self.__retrieve_tag_tables(NodeTag.SELFLOOP)\nsqllineage/core/holders.py:248:                    g.add_nodes_from(read, **{NodeTag.SOURCE_ONLY: True})\nsqllineage/core/holders.py:251:                    g.add_nodes_from(write, **{NodeTag.TARGET_ONLY: True})\nsqllineage/core/holders.py:258:            g.nodes[table][NodeTag.SELFLOOP] = True\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "sqllineage/core/holders.py",
  "content": "import itertools\nfrom typing import Set, Tuple, Union\n\nimport networkx as nx\nfrom networkx import DiGraph\n\nfrom sqllineage.core.models import Column, Path, SubQuery, Table\nfrom sqllineage.utils.constant import EdgeType, NodeTag\n\nDATASET_CLASSES = (Path, Table)\n\n\nclass ColumnLineageMixin:\n    def get_column_lineage(self, exclude_subquery=True) -> Set[Tuple[Column, ...]]:\n        self.graph: DiGraph  # For mypy attribute checking\n        # filter all the column node in the graph\n        column_nodes = [n for n in self.graph.nodes if isinstance(n, Column)]\n        column_graph = self.graph.subgraph(column_nodes)\n        source_columns = {column for column, deg in column_graph.in_degree if deg == 0}\n        # if a column lineage path ends at SubQuery, then it should be pruned\n        target_columns = {\n            node\n            for node, deg in column_graph.out_degree\n            if isinstance(node, Column) and deg == 0\n        }\n        if exclude_subquery:\n            target_columns = {\n                node for node in target_columns if isinstance(node.parent, Table)\n            }\n        columns = set()\n        for (source, target) in itertools.product(source_columns, target_columns):\n            simple_paths = list(nx.all_simple_paths(self.graph, source, target))\n            if len(simple_paths) == 1:\n                columns.add(tuple(simple_paths[0]))\n            # we can ignore when simple path doesn't exist, but could there be more than one simple path?\n        return columns\n\n\nclass SubQueryLineageHolder(ColumnLineageMixin):\n    \"\"\"\n    SubQuery/Query Level Lineage Result.\n\n    SubQueryLineageHolder will hold attributes like read, write, cte\n\n    Each of them is a Set[:class:`sqllineage.models.Table`].\n\n    This is the most atomic representation of lineage result.\n    \"\"\"\n\n    def __init__(self) -> None:\n        self.graph = nx.DiGraph()\n\n    def __or__(self, other):\n        self.graph = nx.compose(self.graph, other.graph)\n        return self\n\n    def _property_getter(self, prop) -> Set[Union[SubQuery, Table]]:\n        return {t for t, attr in self.graph.nodes(data=True) if attr.get(prop) is True}\n\n    def _property_setter(self, value, prop) -> None:\n        self.graph.add_node(value, **{prop: True})\n\n    @property\n    def read(self) -> Set[Union[SubQuery, Table]]:\n        return self._property_getter(NodeTag.READ)\n\n    def add_read(self, value) -> None:\n        self._property_setter(value, NodeTag.READ)\n        # the same table can be added (in SQL: joined) multiple times with different alias\n        if hasattr(value, \"alias\"):\n            self.graph.add_edge(value, value.alias, type=EdgeType.HAS_ALIAS)\n\n    @property\n    def write(self) -> Set[Union[SubQuery, Table]]:\n        return self._property_getter(NodeTag.WRITE)\n\n    def add_write(self, value) -> None:\n        self._property_setter(value, NodeTag.WRITE)\n\n    @property\n    def cte(self) -> Set[SubQuery]:\n        return self._property_getter(NodeTag.CTE)  # type: ignore\n\n    def add_cte(self, value) -> None:\n        self._property_setter(value, NodeTag.CTE)\n\n    def add_column_lineage(self, src: Column, tgt: Column) -> None:\n        self.graph.add_edge(src, tgt, type=EdgeType.LINEAGE)\n        self.graph.add_edge(tgt.parent, tgt, type=EdgeType.HAS_COLUMN)\n        if src.parent is not None:\n            # starting NetworkX v2.6, None is not allowed as node, see https://github.com/networkx/networkx/pull/4892\n            self.graph.add_edge(src.parent, src, type=EdgeType.HAS_COLUMN)\n\n\nclass StatementLineageHolder(SubQueryLineageHolder, ColumnLineageMixin):\n    \"\"\"\n    Statement Level Lineage Result.\n\n    Based on SubQueryLineageHolder, StatementLineageHolder holds extra attributes like drop and rename\n\n    For drop, it is a Set[:class:`sqllineage.models.Table`].\n\n    For rename, it a Set[Tuple[:class:`sqllineage.models.Table`, :class:`sqllineage.models.Table`]], with the first\n    table being original table before renaming and the latter after renaming.\n    \"\"\"\n\n    def __str__(self):\n        return \"\\n\".join(\n            f\"table {attr}: {sorted(getattr(self, attr), key=lambda x: str(x)) if getattr(self, attr) else '[]'}\"\n            for attr in [\"read\", \"write\", \"cte\", \"drop\", \"rename\"]\n        )\n\n    def __repr__(self):\n        return str(self)\n\n    @property\n    def read(self) -> Set[Table]:  # type: ignore\n        return {t for t in super().read if isinstance(t, DATASET_CLASSES)}\n\n    @property\n    def write(self) -> Set[Table]:  # type: ignore\n        return {t for t in super().write if isinstance(t, DATASET_CLASSES)}\n\n    @property\n    def drop(self) -> Set[Table]:\n        return self._property_getter(NodeTag.DROP)  # type: ignore\n\n    def add_drop(self, value) -> None:\n        self._property_setter(value, NodeTag.DROP)\n\n    @property\n    def rename(self) -> Set[Tuple[Table, Table]]:\n        return {\n            (src, tgt)\n            for src, tgt, attr in self.graph.edges(data=True)\n            if attr.get(\"type\") == EdgeType.RENAME\n        }\n\n    def add_rename(self, src: Table, tgt: Table) -> None:\n        self.graph.add_edge(src, tgt, type=EdgeType.RENAME)\n\n    @staticmethod\n    def of(holder: SubQueryLineageHolder):\n        stmt_holder = StatementLineageHolder()\n        stmt_holder.graph = holder.graph\n        return stmt_holder\n\n\nclass SQLLineageHolder(ColumnLineageMixin):\n    def __init__(self, graph: DiGraph):",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 280,
  "is_truncated": true
}