Repository: reata/sqllineage

Issue:
not supporting create table as in postgres dialect
```sql
CREATE TABLE bar AS
SELECT *
FROM foo
```

```shell
$ sqllineage -f test.sql --dialect=postgres
Traceback (most recent call last):
  File "/home/hujunwei/.pyenv/versions/3.8.6/bin/sqllineage", line 8, in <module>
    sys.exit(main())
  File "/home/hujunwei/.pyenv/versions/3.8.6/lib/python3.8/site-packages/sqllineage/cli.py", line 101, in main
    runner.print_table_lineage()
  File "/home/hujunwei/.pyenv/versions/3.8.6/lib/python3.8/site-packages/sqllineage/runner.py", line 168, in print_table_lineage
    print(str(self))
  File "/home/hujunwei/.pyenv/versions/3.8.6/lib/python3.8/site-packages/sqllineage/runner.py", line 22, in wrapper
    self._eval()
  File "/home/hujunwei/.pyenv/versions/3.8.6/lib/python3.8/site-packages/sqllineage/runner.py", line 177, in _eval
    self._stmt_holders = [analyzer.analyze(stmt) for stmt in self._stmt]
  File "/home/hujunwei/.pyenv/versions/3.8.6/lib/python3.8/site-packages/sqllineage/runner.py", line 177, in <listcomp>
    self._stmt_holders = [analyzer.analyze(stmt) for stmt in self._stmt]
  File "/home/hujunwei/.pyenv/versions/3.8.6/lib/python3.8/site-packages/sqllineage/core/parser/sqlfluff/analyzer.py", line 69, in analyze
    raise UnsupportedStatementException(
sqllineage.exceptions.UnsupportedStatementException: SQLLineage doesn't support analyzing statement type [create_table_as_statement] for SQL:CREATE TABLE bar AS
SELECT *
FROM foo
```

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

```diff
@@           Coverage Diff           @@
##           master     #397   +/-   ##
=======================================
  Coverage   99.39%   99.40%           
=======================================
  Files          42       42           
  Lines        1975     2008   +33     
=======================================
+ Hits         1963     1996   +33     
  Misses         12       12           
```


| [Impacted Files](https://app.codecov.io/gh/reata/sqllineage/pull/397?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | Coverage Δ | |
|---|---|---|
| [...parser/sqlfluff/extractors/dml\_insert\_extractor.py](https://app.codecov.io/gh/reata/sqllineage/pull/397?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-c3FsbGluZWFnZS9jb3JlL3BhcnNlci9zcWxmbHVmZi9leHRyYWN0b3JzL2RtbF9pbnNlcnRfZXh0cmFjdG9yLnB5) | `100.00% <ø> (ø)` | |
| [sqllineage/core/parser/sqlfluff/models.py](https://app.codecov.io/gh/reata/sqllineage/pull/397?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-c3FsbGluZWFnZS9jb3JlL3BhcnNlci9zcWxmbHVmZi9tb2RlbHMucHk=) | `100.00% <ø> (ø)` | |
| [sqllineage/core/parser/sqlparse/models.py](https://app.codecov.io/gh/reata/sqllineage/pull/397?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-c3FsbGluZWFnZS9jb3JlL3BhcnNlci9zcWxwYXJzZS9tb2RlbHMucHk=) | `100.00% <ø> (ø)` | |

... and [9 files with indirect coverage changes](https://app.codecov.io/gh/reata/sqllineage/pull/397/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)


reata: Is there any test case for column parent being optional?

Patch:
diff --git a/sqllineage/core/parser/sqlfluff/extractors/dml_insert_extractor.py b/sqllineage/core/parser/sqlfluff/extractors/dml_insert_extractor.py
index 538b6d9..f2859e4 100644
--- a/sqllineage/core/parser/sqlfluff/extractors/dml_insert_extractor.py
+++ b/sqllineage/core/parser/sqlfluff/extractors/dml_insert_extractor.py
@@ -22,6 +22,7 @@ class DmlInsertExtractor(LineageHolderExtractor):
     SUPPORTED_STMT_TYPES = [
         "insert_statement",
         "create_table_statement",
+        "create_table_as_statement",
         "create_view_statement",
         "update_statement",
         "copy_statement",
diff --git a/sqllineage/core/parser/sqlfluff/handlers/target.py b/sqllineage/core/parser/sqlfluff/handlers/target.py
index 3968412..b016023 100644
--- a/sqllineage/core/parser/sqlfluff/handlers/target.py
+++ b/sqllineage/core/parser/sqlfluff/handlers/target.py
@@ -80,7 +80,7 @@ class TargetHandler(ConditionalSegmentBaseHandler):
         :param segment: segment to be handled
         :param holder: 'SqlFluffSubQueryLineageHolder' to hold lineage
         """
-        if segment.type == "table_reference":
+        if segment.type in ["table_reference", "object_reference"]:
             write_obj = SqlFluffTable.of(segment)
             if self.prev_token_read:
                 holder.add_read(write_obj)
diff --git a/sqllineage/core/parser/sqlfluff/models.py b/sqllineage/core/parser/sqlfluff/models.py
index ecd4e4a..2ea94e7 100644
--- a/sqllineage/core/parser/sqlfluff/models.py
+++ b/sqllineage/core/parser/sqlfluff/models.py
@@ -192,7 +192,9 @@ class SqlFluffColumn(Column):
             ).get_column_lineage(exclude_subquery=False)
         ]
         source_columns = [
-            ColumnQualifierTuple(src_col.raw_name, src_col.parent.raw_name)
+            ColumnQualifierTuple(
+                src_col.raw_name, src_col.parent.raw_name if src_col.parent else None
+            )
             for src_col in src_cols
         ]
         return source_columns
diff --git a/sqllineage/core/parser/sqlparse/models.py b/sqllineage/core/parser/sqlparse/models.py
index 9f89b35..7dbd8e3 100644
--- a/sqllineage/core/parser/sqlparse/models.py
+++ b/sqllineage/core/parser/sqlparse/models.py
@@ -116,7 +116,10 @@ class SqlParseColumn(Column):
                     )
                 ]
                 source_columns = [
-                    ColumnQualifierTuple(src_col.raw_name, src_col.parent.raw_name)
+                    ColumnQualifierTuple(
+                        src_col.raw_name,
+                        src_col.parent.raw_name if src_col.parent else None,
+                    )
                     for src_col in src_cols
                 ]
             else:


Test patch:
diff --git a/tests/test_others_dialect_specific.py b/tests/test_others_dialect_specific.py
index ddbf51e..c7185f3 100644
--- a/tests/test_others_dialect_specific.py
+++ b/tests/test_others_dialect_specific.py
@@ -24,6 +24,22 @@ def test_create_select_without_as(dialect: str):
     )
 
 
+@pytest.mark.parametrize("dialect", ["duckdb", "greenplum", "postgres", "redshift"])
+def test_create_table_as_with_postgres_dialect(dialect: str):
+    """
+    sqlfluff postgres family dialects parse CTAS statement as "create_table_as_statement",
+    unlike "create_table_statement" in ansi dialect
+    """
+    assert_table_lineage_equal(
+        """CREATE TABLE bar AS
+SELECT *
+FROM foo""",
+        {"foo"},
+        {"bar"},
+        dialect,
+    )
+
+
 def test_create_using_serde():
     """
     https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL#LanguageManualDDL-RowFormats&SerDe
diff --git a/tests/test_sqlfluff.py b/tests/test_sqlfluff.py
index 975426e..dbc7fab 100644
--- a/tests/test_sqlfluff.py
+++ b/tests/test_sqlfluff.py
@@ -278,3 +278,25 @@ def test_insert_with_custom_columns():
         ],
         test_sqlparse=False,
     )
+
+
+def test_subquery_expression_without_source_table():
+    """
+    sqlparse implementation is capable of generating the correct result, however with redundant nodes in graph
+    """
+    assert_column_lineage_equal(
+        """INSERT INTO foo
+SELECT (SELECT col1 + col2 AS result) AS sum_result
+FROM bar""",
+        [
+            (
+                ColumnQualifierTuple("col1", "bar"),
+                ColumnQualifierTuple("sum_result", "foo"),
+            ),
+            (
+                ColumnQualifierTuple("col2", "bar"),
+                ColumnQualifierTuple("sum_result", "foo"),
+            ),
+        ],
+        test_sqlparse=False,
+    )