Repository: reata/sqllineage

Issue:
Add an Argument to Exclude SubQuery Column Node in Column Lineage Path
**Describe the bug**
now get_column_lineage's result include SubQuery

**To Reproduce**
For example:
```python
from sqllineage.runner import LineageRunner
v_sql = "insert into ta select b from (select b from tb union all select c from tc ) sub"
parse = LineageRunner(sql=v_sql)
for col_tuple in parse.get_column_lineage():
    print(col_tuple)
```
```
(Column: <default>.tb.b, Column: sub.b, Column: <default>.ta.b)
(Column: <default>.tc.c, Column: sub.b, Column: <default>.ta.b)
```

**Expected behavior**
```
(Column: <default>.tb.b, Column: <default>.ta.b)
(Column: <default>.tc.c, Column: <default>.ta.b)
```
**Python version (available via `python --version`)**
 - 3.10.13

**SQLLineage version (available via `sqllineage --version`):**
 - 1.4.9

**Additional context**
I will submit a PR later

Discussion:

Patch:
diff --git a/sqllineage/core/holders.py b/sqllineage/core/holders.py
index 688833f..eb5085d 100644
--- a/sqllineage/core/holders.py
+++ b/sqllineage/core/holders.py
@@ -13,7 +13,16 @@ DATASET_CLASSES = (Path, Table)
 
 
 class ColumnLineageMixin:
-    def get_column_lineage(self, exclude_subquery=True) -> Set[Tuple[Column, ...]]:
+    def get_column_lineage(
+        self, exclude_path_ending_in_subquery=True, exclude_subquery_columns=False
+    ) -> Set[Tuple[Column, ...]]:
+        """
+        :param exclude_path_ending_in_subquery:  exclude_subquery rename to exclude_path_ending_in_subquery
+               exclude column from SubQuery in the ending path
+        :param exclude_subquery_columns: exclude column from SubQuery in the path.
+
+        return a list of column tuple :class:`sqllineage.models.Column`
+        """
         self.graph: DiGraph  # For mypy attribute checking
         # filter all the column node in the graph
         column_nodes = [n for n in self.graph.nodes if isinstance(n, Column)]
@@ -25,7 +34,7 @@ class ColumnLineageMixin:
             for node, deg in column_graph.out_degree
             if isinstance(node, Column) and deg == 0
         }
-        if exclude_subquery:
+        if exclude_path_ending_in_subquery:
             target_columns = {
                 node for node in target_columns if isinstance(node.parent, Table)
             }
@@ -33,7 +42,14 @@ class ColumnLineageMixin:
         for source, target in itertools.product(source_columns, target_columns):
             simple_paths = list(nx.all_simple_paths(self.graph, source, target))
             for path in simple_paths:
-                columns.add(tuple(path))
+                if exclude_subquery_columns:
+                    path = [
+                        node for node in path if not isinstance(node.parent, SubQuery)
+                    ]
+                    if len(path) > 1:
+                        columns.add(tuple(path))
+                else:
+                    columns.add(tuple(path))
         return columns
 
 
diff --git a/sqllineage/core/models.py b/sqllineage/core/models.py
index fd96c60..8eb012f 100644
--- a/sqllineage/core/models.py
+++ b/sqllineage/core/models.py
@@ -58,7 +58,7 @@ class Table:
             self.raw_name = escape_identifier_name(table_name)
             if schema:
                 warnings.warn("Name is in schema.table format, schema param is ignored")
-        self.alias = kwargs.pop("alias", self.raw_name)
+        self.alias = escape_identifier_name(kwargs.pop("alias", self.raw_name))
 
     def __str__(self):
         return f"{self.schema}.{self.raw_name}"
diff --git a/sqllineage/core/parser/sqlfluff/extractors/base.py b/sqllineage/core/parser/sqlfluff/extractors/base.py
index 1f1ca36..275e3f1 100644
--- a/sqllineage/core/parser/sqlfluff/extractors/base.py
+++ b/sqllineage/core/parser/sqlfluff/extractors/base.py
@@ -156,7 +156,7 @@ class BaseExtractor:
                     )
                 if "." not in table_identifier.raw:
                     cte_dict = {s.alias: s for s in holder.cte}
-                    cte = cte_dict.get(table_identifier.raw)
+                    cte = cte_dict.get(escape_identifier_name(table_identifier.raw))
                     if cte is not None:
                         # could reference CTE with or without alias
                         tables.append(
diff --git a/sqllineage/core/parser/sqlfluff/models.py b/sqllineage/core/parser/sqlfluff/models.py
index 6565f51..57ab368 100644
--- a/sqllineage/core/parser/sqlfluff/models.py
+++ b/sqllineage/core/parser/sqlfluff/models.py
@@ -195,7 +195,7 @@ class SqlFluffColumn(Column):
             lineage[0]
             for lineage in LineageRunner(
                 sub_segment.raw, dialect=SQLPARSE_DIALECT
-            ).get_column_lineage(exclude_subquery=False)
+            ).get_column_lineage(exclude_path_ending_in_subquery=False)
         ]
         source_columns = [
             ColumnQualifierTuple(
diff --git a/sqllineage/core/parser/sqlparse/handlers/source.py b/sqllineage/core/parser/sqlparse/handlers/source.py
index 8f5b77c..f6c62b4 100644
--- a/sqllineage/core/parser/sqlparse/handlers/source.py
+++ b/sqllineage/core/parser/sqlparse/handlers/source.py
@@ -27,6 +27,7 @@ from sqllineage.core.parser.sqlparse.utils import (
     is_values_clause,
 )
 from sqllineage.exceptions import SQLLineageException
+from sqllineage.utils.helpers import escape_identifier_name
 
 
 class SourceHandler(SourceHandlerMixin, NextTokenBaseHandler):
@@ -151,7 +152,9 @@ class SourceHandler(SourceHandlerMixin, NextTokenBaseHandler):
             else:
                 cte_dict = {s.alias: s for s in holder.cte}
                 if "." not in identifier.value:
-                    cte = cte_dict.get(identifier.get_real_name())
+                    cte = cte_dict.get(
+                        escape_identifier_name(identifier.get_real_name())
+                    )
                     if cte is not None:
                         # could reference CTE with or without alias
                         read = SqlParseSubQuery.of(
diff --git a/sqllineage/core/parser/sqlparse/models.py b/sqllineage/core/parser/sqlparse/models.py
index 6a49906..82a7c4a 100644
--- a/sqllineage/core/parser/sqlparse/models.py
+++ b/sqllineage/core/parser/sqlparse/models.py
@@ -110,7 +110,7 @@ class SqlParseColumn(Column):
                     lineage[0]
                     for lineage in LineageRunner(
                         token.value, dialect=SQLPARSE_DIALECT
-                    ).get_column_lineage(exclude_subquery=False)
+                    ).get_column_lineage(exclude_path_ending_in_subquery=False)
                 ]
                 source_columns = [
                     ColumnQualifierTuple(
diff --git a/sqllineage/runner.py b/sqllineage/runner.py
index 7db37f4..335a0ee 100644
--- a/sqllineage/runner.py
+++ b/sqllineage/runner.py
@@ -151,13 +151,17 @@ Target Tables:
         return sorted(self._sql_holder.intermediate_tables, key=lambda x: str(x))
 
     @lazy_method
-    def get_column_lineage(self, exclude_subquery=True) -> List[Tuple[Column, Column]]:
+    def get_column_lineage(
+        self, exclude_path_ending_in_subquery=True, exclude_subquery_columns=False
+    ) -> List[Tuple[Column, Column]]:
         """
         a list of column tuple :class:`sqllineage.models.Column`
         """
         # sort by target column, and then source column
         return sorted(
-            self._sql_holder.get_column_lineage(exclude_subquery),
+            self._sql_holder.get_column_lineage(
+                exclude_path_ending_in_subquery, exclude_subquery_columns
+            ),
             key=lambda x: (str(x[-1]), str(x[0])),
         )
 
diff --git a/sqllineagejs/package-lock.json b/sqllineagejs/package-lock.json
index 078ea99..da2c643 100644
--- a/sqllineagejs/package-lock.json
+++ b/sqllineagejs/package-lock.json
@@ -8449,9 +8449,9 @@
       "dev": true
     },
     "node_modules/follow-redirects": {
-      "version": "1.14.8",
-      "resolved": "https://registry.npmjs.org/follow-redirects/-/follow-redirects-1.14.8.tgz",
-      "integrity": "sha512-1x0S9UVJHsQprFcEC/qnNzBLcIxsjAV905f/UkQxbclCsoTWlacCNOpQa/anodLl2uaEKFhfWOvM2Qg77+15zA==",
+      "version": "1.15.4",
+      "resolved": "https://registry.npmjs.org/follow-redirects/-/follow-redirects-1.15.4.tgz",
+      "integrity": "sha512-Cr4D/5wlrb0z9dgERpUL3LrmPKVDsETIJhaCMeDfuFYcqa5bldGV6wBsAN6X/vxlXQtFBMrXdXxdL8CbDTGniw==",
       "dev": true,
       "funding": [
         {
@@ -25123,9 +25123,9 @@
       "dev": true
     },
     "follow-redirects": {
-      "version": "1.14.8",
-      "resolved": "https://registry.npmjs.org/follow-redirects/-/follow-redirects-1.14.8.tgz",
-      "integrity": "sha512-1x0S9UVJHsQprFcEC/qnNzBLcIxsjAV905f/UkQxbclCsoTWlacCNOpQa/anodLl2uaEKFhfWOvM2Qg77+15zA==",
+      "version": "1.15.4",
+      "resolved": "https://registry.npmjs.org/follow-redirects/-/follow-redirects-1.15.4.tgz",
+      "integrity": "sha512-Cr4D/5wlrb0z9dgERpUL3LrmPKVDsETIJhaCMeDfuFYcqa5bldGV6wBsAN6X/vxlXQtFBMrXdXxdL8CbDTGniw==",
       "dev": true
     },
     "for-each": {


Test patch:
diff --git a/tests/core/test_runner.py b/tests/core/test_runner.py
index ead0ef6..9659392 100644
--- a/tests/core/test_runner.py
+++ b/tests/core/test_runner.py
@@ -1,3 +1,4 @@
+from sqllineage.core.models import SubQuery
 from sqllineage.runner import LineageRunner
 from sqllineage.utils.constant import LineageLevel
 
@@ -22,3 +23,11 @@ def test_statements_trim_comment():
 def test_silent_mode():
     sql = "begin; select * from dual;"
     LineageRunner(sql, dialect="greenplum", silent_mode=True)._eval()
+
+
+def test_get_column_lineage_exclude_subquery_inpath():
+    v_sql = "insert into ta select b from (select b from tb union all select c from tc ) sub"
+    parse = LineageRunner(sql=v_sql)
+    for col_tuple in parse.get_column_lineage(exclude_subquery_columns=True):
+        for col in col_tuple:
+            assert not isinstance(col.parent, SubQuery)
diff --git a/tests/sql/column/test_column_select_column.py b/tests/sql/column/test_column_select_column.py
index f3aa3ea..c946aa4 100644
--- a/tests/sql/column/test_column_select_column.py
+++ b/tests/sql/column/test_column_select_column.py
@@ -96,6 +96,30 @@ FROM tab2 AS t"""
     )
 
 
+def test_select_column_with_table_qualifier_case_insensitive():
+    sql = """INSERT INTO tab1
+SELECT TA.col1
+FROM tab2 AS ta"""
+    assert_column_lineage_equal(
+        sql,
+        [(ColumnQualifierTuple("col1", "tab2"), ColumnQualifierTuple("col1", "tab1"))],
+    )
+    sql = """INSERT INTO tab1
+SELECT ta.col1
+FROM tab2 AS TA"""
+    assert_column_lineage_equal(
+        sql,
+        [(ColumnQualifierTuple("col1", "tab2"), ColumnQualifierTuple("col1", "tab1"))],
+    )
+    sql = """INSERT INTO tab1
+SELECT Ta.col1
+FROM tab2 AS tA"""
+    assert_column_lineage_equal(
+        sql,
+        [(ColumnQualifierTuple("col1", "tab2"), ColumnQualifierTuple("col1", "tab1"))],
+    )
+
+
 def test_select_columns():
     sql = """INSERT INTO tab1
 SELECT col1,
diff --git a/tests/sql/column/test_column_select_from_cte.py b/tests/sql/column/test_column_select_from_cte.py
index d892e44..17fa581 100644
--- a/tests/sql/column/test_column_select_from_cte.py
+++ b/tests/sql/column/test_column_select_from_cte.py
@@ -101,3 +101,27 @@ def test_cte_inside_insert_in_parenthesis():
         ],
         test_sqlparse=False,
     )
+
+
+def test_select_column_from_cte_case_insensitive():
+    sql = """WITH CTE1 AS (SELECT col1 FROM tab2)
+INSERT INTO tab1
+SELECT col1 FROM cte1"""
+    assert_column_lineage_equal(
+        sql,
+        [(ColumnQualifierTuple("col1", "tab2"), ColumnQualifierTuple("col1", "tab1"))],
+    )
+    sql = """WITH cte1 AS (SELECT col1 FROM tab2)
+INSERT INTO tab1
+SELECT col1 FROM CTE1"""
+    assert_column_lineage_equal(
+        sql,
+        [(ColumnQualifierTuple("col1", "tab2"), ColumnQualifierTuple("col1", "tab1"))],
+    )
+    sql = """WITH CTe1 AS (SELECT col1 FROM tab2)
+INSERT INTO tab1
+SELECT col1 FROM cTE1"""
+    assert_column_lineage_equal(
+        sql,
+        [(ColumnQualifierTuple("col1", "tab2"), ColumnQualifierTuple("col1", "tab1"))],
+    )
diff --git a/tests/sql/column/test_column_select_from_subquery.py b/tests/sql/column/test_column_select_from_subquery.py
index 7264e3e..389b0a0 100644
--- a/tests/sql/column/test_column_select_from_subquery.py
+++ b/tests/sql/column/test_column_select_from_subquery.py
@@ -126,3 +126,10 @@ FROM (SELECT col1 FROM tab2) DT"""
         sql,
         [(ColumnQualifierTuple("col1", "tab2"), ColumnQualifierTuple("col1", "tab1"))],
     )
+    sql = """INSERT INTO tab1
+SELECT Dt.col1
+FROM (SELECT col1 FROM tab2) dT"""
+    assert_column_lineage_equal(
+        sql,
+        [(ColumnQualifierTuple("col1", "tab2"), ColumnQualifierTuple("col1", "tab1"))],
+    )