Repository: reata/sqllineage

Issue:
set target table column name from MetaDataProvider
**Is your feature request related to a problem? Please describe.**
sql: "insert into target select user_id, user_name from source"
target's column is id, name
column lineage 's target column is user_id, user_name from sql parse

**Describe the solution you'd like**
use MetaDataProvider, search target table's column ,write to Holder.
equal  "insert into target(id, name) select user_id, user_name from source"




Discussion:
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/reata/sqllineage/pull/557?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Comparison is base [(`06a474b`)](https://app.codecov.io/gh/reata/sqllineage/commit/06a474b5e7d6a26994db18f97cf9058246cd24bd?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) 99.50% compared to head [(`323ec83`)](https://app.codecov.io/gh/reata/sqllineage/pull/557?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) 99.50%.


<details><summary>Additional details and impacted files</summary>


```diff
@@           Coverage Diff           @@
##           master     #557   +/-   ##
=======================================
  Coverage   99.50%   99.50%           
=======================================
  Files          41       41           
  Lines        2211     2219    +8     
=======================================
+ Hits         2200     2208    +8     
  Misses         11       11           
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/reata/sqllineage/pull/557?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None).


Patch:
diff --git a/sqllineage/core/parser/sqlfluff/extractors/create_insert.py b/sqllineage/core/parser/sqlfluff/extractors/create_insert.py
index 7610bac..1873ab7 100644
--- a/sqllineage/core/parser/sqlfluff/extractors/create_insert.py
+++ b/sqllineage/core/parser/sqlfluff/extractors/create_insert.py
@@ -1,7 +1,7 @@
 from sqlfluff.core.parser import BaseSegment
 
 from sqllineage.core.holders import SubQueryLineageHolder
-from sqllineage.core.models import Path
+from sqllineage.core.models import Path, Table
 from sqllineage.core.parser.sqlfluff.extractors.base import BaseExtractor
 from sqllineage.core.parser.sqlfluff.extractors.select import SelectExtractor
 from sqllineage.core.parser.sqlfluff.models import SqlFluffColumn, SqlFluffTable
@@ -104,6 +104,15 @@ class CreateInsertExtractor(BaseExtractor):
                 if segment.type in ["table_reference", "object_reference"]:
                     write_obj = SqlFluffTable.of(segment)
                     holder.add_write(write_obj)
+                    # get target table columns from metadata if available
+                    if (
+                        isinstance(write_obj, Table)
+                        and self.metadata_provider
+                        and statement.type == "insert_statement"
+                    ):
+                        holder.add_write_column(
+                            *self.metadata_provider.get_table_columns(table=write_obj)
+                        )
                 elif segment.type == "literal":
                     if segment.raw.isnumeric():
                         # Special Handling for Spark Bucket Table DDL


Test patch:
diff --git a/tests/sql/column/test_metadata_target_column.py b/tests/sql/column/test_metadata_target_column.py
new file mode 100644
index 0000000..8361d0f
--- /dev/null
+++ b/tests/sql/column/test_metadata_target_column.py
@@ -0,0 +1,64 @@
+import pytest
+
+from sqllineage.core.metadata_provider import MetaDataProvider
+from sqllineage.utils.entities import ColumnQualifierTuple
+from ...helpers import assert_column_lineage_equal, generate_metadata_providers
+
+
+providers = generate_metadata_providers(
+    {
+        "ods.source_tab": ["day_id", "user_id", "name"],
+        "ods.target_tab": ["day_no", "user_code", "user_name"],
+    }
+)
+
+
+@pytest.mark.parametrize("provider", providers)
+def test_metadata_target_column(provider: MetaDataProvider):
+    sql = """insert into ods.target_tab select day_id as acct_id, user_id as xxx, name as yyy from ods.source_tab"""
+    assert_column_lineage_equal(
+        sql=sql,
+        column_lineages=[
+            (
+                ColumnQualifierTuple("name", "ods.source_tab"),
+                ColumnQualifierTuple("user_name", "ods.target_tab"),
+            ),
+            (
+                ColumnQualifierTuple("day_id", "ods.source_tab"),
+                ColumnQualifierTuple("day_no", "ods.target_tab"),
+            ),
+            (
+                ColumnQualifierTuple("user_id", "ods.source_tab"),
+                ColumnQualifierTuple("user_code", "ods.target_tab"),
+            ),
+        ],
+        metadata_provider=provider,
+        test_sqlparse=False,
+    )
+
+
+@pytest.mark.parametrize("provider", providers)
+def test_metadata_target_column_cte(provider: MetaDataProvider):
+    sql = """
+INSERT INTO ods.target_tab
+WITH cte_table AS (SELECT day_id as acct_id, user_id as xxx, name as yyy FROM ods.source_tab)
+SELECT acct_id, xxx, yyy FROM cte_table"""
+    assert_column_lineage_equal(
+        sql=sql,
+        column_lineages=[
+            (
+                ColumnQualifierTuple("user_id", "ods.source_tab"),
+                ColumnQualifierTuple("user_code", "ods.target_tab"),
+            ),
+            (
+                ColumnQualifierTuple("day_id", "ods.source_tab"),
+                ColumnQualifierTuple("day_no", "ods.target_tab"),
+            ),
+            (
+                ColumnQualifierTuple("name", "ods.source_tab"),
+                ColumnQualifierTuple("user_name", "ods.target_tab"),
+            ),
+        ],
+        metadata_provider=provider,
+        test_sqlparse=False,
+    )