Repository: reata/sqllineage

Issue:
Missing column lineage from SELECT DISTINCT using non-validating dialect
Hi, 

I've tested to confirm two versions on an venv of `python 3.9.6`:
- the old one with: `sqllineage-1.3.7`
- the newest one: `sqllineage-1.4.1`

My SQL and code:
```
SQL = 
"""
WITH tb_dataraw_distinct AS 
(
  SELECT DISTINCT DATE(created_at,'Asia/Bangkok') AS created_date_key
        ,order_id
        ,city_id 
        ,deliveries_count
        ,category
        ,total_cod
        ,CASE WHEN total_cod > 0 THEN 1 ELSE 0 END AS is_cod
        ,last_order_status
        ,cancellation_reasons
  FROM `bi_dim_fact.tb_delivering_order_process`
 WHERE partition_date >= DATE_SUB(CURRENT_DATE('+7:00'), INTERVAL 7 DAY)
),
tb_source AS
(
  SELECT created_date_key
        ,city_id
        ,is_cod
        ,deliveries_count
        ,category
        ,last_order_status AS last_order_status_id 
        ,CASE WHEN last_order_status = 1 THEN  'cancel_by_sender_af_accepted_by_driver' 
              WHEN last_order_status = 2 THEN  'delivered'
              WHEN last_order_status = 5 THEN  'returned'
              WHEN last_order_status = 7 THEN  'cancel_by_driver'
              WHEN last_order_status = 24 THEN 'cancel_by_sender_bf_accepted_by_driver'
              WHEN last_order_status = 25 THEN 'not_find_driver'
              WHEN last_order_status = 26 THEN 'cancel_by_admin'
              WHEN last_order_status = 27 THEN 'complete_b_admin'END last_order_status            
        ,cancellation_reasons      
        ,COUNT(order_id) AS cancelled_requests
    FROM tb_dataraw_distinct AS od
  GROUP BY 1,2,3,4,5,6,7,8  
)
,
tb_history AS 
(
   SELECT *
     FROM `bi_dashboard.tb_bedelivery_cancelled_reasons` 
    WHERE created_date_key < DATE_SUB(CURRENT_DATE('+7:00'), INTERVAL 7 DAY)  
)
SELECT * FROM tb_source
UNION ALL
SELECT * FROM tb_history
"""
```
```
from sqllineage.runner import LineageRunner
print(LineageRunner(SQL).get_column_lineage(False))
```

**The output from the old version (1.3.7) is what I expected:**

[(Column: bi_dim_fact.tb_delivering_order_process.total_cod, Column: tb_dataraw_distinct.total_cod), (Column: bi_dashboard.tb_bedelivery_cancelled_reasons.*, Column: tb_history.*), (Column: bi_dim_fact.tb_delivering_order_process.cancellation_reasons, Column: tb_dataraw_distinct.cancellation_reasons, Column: tb_source.cancellation_reasons), (Column: bi_dim_fact.tb_delivering_order_process.order_id, Column: tb_dataraw_distinct.order_id, Column: tb_source.cancelled_requests), (Column: bi_dim_fact.tb_delivering_order_process.category, Column: tb_dataraw_distinct.category, Column: tb_source.category), (Column: bi_dim_fact.tb_delivering_order_process.city_id, Column: tb_dataraw_distinct.city_id, Column: tb_source.city_id), (Column: bi_dim_fact.tb_delivering_order_process.created_at, Column: tb_dataraw_distinct.created_date_key, Column: tb_source.created_date_key), (Column: bi_dim_fact.tb_delivering_order_process.deliveries_count, Column: tb_dataraw_distinct.deliveries_count, Column: tb_source.deliveries_count), (Column: bi_dim_fact.tb_delivering_order_process.total_cod, Column: tb_dataraw_distinct.is_cod, Column: tb_source.is_cod), (Column: bi_dim_fact.tb_delivering_order_process.last_order_status, Column: tb_dataraw_distinct.last_order_status, Column: tb_source.last_order_status), (Column: bi_dim_fact.tb_delivering_order_process.last_order_status, Column: tb_dataraw_distinct.last_order_status, Column: tb_source.last_order_status_id)]

**And on the newest version there're some missing these `bi_dim_fact` dataset:**

[(Column: bi_dashboard.tb_bedelivery_cancelled_reasons.*, Column: tb_history.*), (Column: tb_dataraw_distinct.cancellation_reasons, Column: tb_source.cancellation_reasons), (Column: tb_dataraw_distinct.order_id, Column: tb_source.cancelled_requests), (Column: tb_dataraw_distinct.category, Column: tb_source.category), (Column: tb_dataraw_distinct.city_id, Column: tb_source.city_id), (Column: tb_dataraw_distinct.created_date_key, Column: tb_source.created_date_key), (Column: tb_dataraw_distinct.deliveries_count, Column: tb_source.deliveries_count), (Column: tb_dataraw_distinct.is_cod, Column: tb_source.is_cod), (Column: tb_dataraw_distinct.last_order_status, Column: tb_source.last_order_status), (Column: tb_dataraw_distinct.last_order_status, Column: tb_source.last_order_status_id)]


Discussion:
codecov[bot]: ## [Codecov](https://codecov.io/gh/reata/sqllineage/pull/365?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) Report
> Merging [#365](https://codecov.io/gh/reata/sqllineage/pull/365?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (c49b5ab) into [master](https://codecov.io/gh/reata/sqllineage/commit/db377efc71aed07bf29a94805662a5a0c82cdcea?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (db377ef) will **increase** coverage by `0.00%`.
> The diff coverage is `100.00%`.

```diff
@@           Coverage Diff           @@
##           master     #365   +/-   ##
=======================================
  Coverage   99.38%   99.38%           
=======================================
  Files          42       42           
  Lines        1964     1966    +2     
=======================================
+ Hits         1952     1954    +2     
  Misses         12       12           
```


| [Impacted Files](https://codecov.io/gh/reata/sqllineage/pull/365?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | Coverage Δ | |
|---|---|---|
| [sqllineage/core/parser/sqlparse/handlers/source.py](https://codecov.io/gh/reata/sqllineage/pull/365?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-c3FsbGluZWFnZS9jb3JlL3BhcnNlci9zcWxwYXJzZS9oYW5kbGVycy9zb3VyY2UucHk=) | `100.00% <100.00%> (ø)` | |



Patch:
diff --git a/sqllineage/core/parser/sqlparse/handlers/source.py b/sqllineage/core/parser/sqlparse/handlers/source.py
index 25e2691..761402e 100644
--- a/sqllineage/core/parser/sqlparse/handlers/source.py
+++ b/sqllineage/core/parser/sqlparse/handlers/source.py
@@ -47,6 +47,10 @@ class SourceHandler(SourceHandlerMixin, NextTokenBaseHandler):
         if token.normalized in ("UNION", "UNION ALL"):
             self.union_barriers.append((len(self.columns), len(self.tables)))
 
+        if self.column_flag is True and bool(token.normalized == "DISTINCT"):
+            # special handling for SELECT DISTINCT
+            return True
+
         if any(re.match(regex, token.normalized) for regex in self.SOURCE_TABLE_TOKENS):
             self.column_flag = False
             return True


Test patch:
diff --git a/tests/test_columns.py b/tests/test_columns.py
index d777bf4..1b085c0 100644
--- a/tests/test_columns.py
+++ b/tests/test_columns.py
@@ -52,6 +52,16 @@ FROM tab2 a
     )
 
 
+def test_select_distinct_column():
+    sql = """INSERT INTO tab1
+SELECT DISTINCT col1
+FROM tab2"""
+    assert_column_lineage_equal(
+        sql,
+        [(ColumnQualifierTuple("col1", "tab2"), ColumnQualifierTuple("col1", "tab1"))],
+    )
+
+
 def test_select_column_using_function():
     sql = """INSERT INTO tab1
 SELECT max(col1),