feat: add data lineage

scripts/ci/deploy/config/databend-query-node-1.toml

@@ -80,6 +80,9 @@ definition = "CREATE FUNCTION ping(STRING) RETURNS STRING LANGUAGE python HANDLE
 aggregate_spilling_memory_ratio = 60
 join_spilling_memory_ratio = 60
 
+[query.lineage]
+capture_enabled = true
+
 [log]
 
 [log.file]

scripts/ci/deploy/config/databend-query-node-2.toml

@@ -51,6 +51,9 @@ auth_type = "no_password"
 name = "ping"
 definition = "CREATE FUNCTION ping(STRING) RETURNS STRING LANGUAGE python HANDLER = 'ping' ADDRESS = 'http://0.0.0.0:8815'"
 
+[query.lineage]
+capture_enabled = true
+
 [log]
 
 [log.file]

scripts/ci/deploy/config/databend-query-node-3.toml

@@ -52,6 +52,9 @@ auth_type = "no_password"
 name = "ping"
 definition = "CREATE FUNCTION ping(STRING) RETURNS STRING LANGUAGE python HANDLER = 'ping' ADDRESS = 'http://0.0.0.0:8815'"
 
+[query.lineage]
+capture_enabled = true
+
 [log]
 
 [log.file]

scripts/ci/deploy/config/databend-query-node-hive.toml

@@ -44,6 +44,9 @@ auth_type = "no_password"
 name = "ping"
 definition = "CREATE FUNCTION ping(STRING) RETURNS STRING LANGUAGE python HANDLER = 'ping' ADDRESS = 'http://0.0.0.0:8815'"
 
+[query.lineage]
+capture_enabled = true
+
 [log]
 
 [log.file]

scripts/ci/deploy/config/databend-query-node-otlp-logs.toml

@@ -40,6 +40,9 @@ auth_type = "no_password"
 name = "ping"
 definition = "CREATE FUNCTION ping(STRING) RETURNS STRING LANGUAGE python HANDLER = 'ping' ADDRESS = 'http://0.0.0.0:8815'"
 
+[query.lineage]
+capture_enabled = true
+
 [log]
 
 [log.file]

scripts/ci/deploy/config/databend-query-node-system-managed.toml

@@ -51,6 +51,9 @@ definition = "CREATE FUNCTION ping(STRING) RETURNS STRING LANGUAGE python HANDLE
 type = "system_managed"
 node_group
 
+[query.lineage]
+capture_enabled = true
+
 [log]
 
 [log.file]

src/common/license/src/license.rs

@@ -37,6 +37,8 @@ pub enum Feature {
     StorageEncryption,
     #[serde(alias = "stream", alias = "STREAM")]
     Stream,
+    #[serde(alias = "lineage", alias = "LINEAGE")]
+    Lineage,
     #[serde(alias = "table_ref", alias = "TABLE_REF")]
     TableRef,
     #[serde(alias = "attach_table", alias = "ATTACH_TABLE")]
@@ -77,6 +79,7 @@ impl fmt::Display for Feature {
             Feature::ComputedColumn => write!(f, "computed_column"),
             Feature::StorageEncryption => write!(f, "storage_encryption"),
             Feature::Stream => write!(f, "stream"),
+            Feature::Lineage => write!(f, "lineage"),
             Feature::TableRef => write!(f, "table_ref"),
             Feature::AttacheTable => write!(f, "attach_table"),
             Feature::AmendTable => write!(f, "amend_table"),
@@ -211,6 +214,10 @@ mod tests {
             Feature::Stream,
             serde_json::from_str::<Feature>("\"Stream\"").unwrap()
         );
+        assert_eq!(
+            Feature::Lineage,
+            serde_json::from_str::<Feature>("\"LINEAGE\"").unwrap()
+        );
         assert_eq!(
             Feature::TableRef,
             serde_json::from_str::<Feature>("\"TableRef\"").unwrap()
@@ -270,6 +277,7 @@ mod tests {
                 Feature::ComputedColumn,
                 Feature::StorageEncryption,
                 Feature::Stream,
+                Feature::Lineage,
                 Feature::TableRef,
                 Feature::AttacheTable,
                 Feature::AmendTable,
@@ -281,7 +289,7 @@ mod tests {
         };
 
         assert_eq!(
-            "LicenseInfo{ type: enterprise, org: databend, tenants: [databend_tenant,foo], features: [amend_table,attach_table,computed_column,data_mask,license_info,private_task,row_access_policy,storage_encryption,stream,system_history,table_ref,vacuum,workload_group] }",
+            "LicenseInfo{ type: enterprise, org: databend, tenants: [databend_tenant,foo], features: [amend_table,attach_table,computed_column,data_mask,license_info,lineage,private_task,row_access_policy,storage_encryption,stream,system_history,table_ref,vacuum,workload_group] }",
             license_info.to_string()
         );
     }
@@ -297,6 +305,7 @@ mod tests {
             Feature::ComputedColumn,
             Feature::StorageEncryption,
             Feature::Stream,
+            Feature::Lineage,
             Feature::TableRef,
             Feature::AttacheTable,
             Feature::AmendTable,

src/meta/api/src/api_impl/auto_increment_api_test_suite.rs

@@ -128,6 +128,7 @@ impl AutoIncrementApiTestSuite {
                     table_name: tbl_name.to_string(),
                 },
                 table_meta: drop_table_meta(created_on),
+                lineage_updates: vec![],
                 as_dropped: true,
                 table_properties: None,
                 table_partition: None,

src/meta/api/src/api_impl/garbage_collection_api.rs

@@ -73,6 +73,7 @@ use log::info;
 use log::warn;
 
 use super::index_api::IndexApi;
+use super::lineage_api::append_delete_lineage_for_table_id_txn_ops;
 use crate::kv_app_error::KVAppError;
 use crate::kv_pb_api::KVPbApi;
 use crate::kv_pb_crud_api::KVPbCrudApi;
@@ -662,6 +663,13 @@ async fn gc_dropped_table_by_id(
         // 3)
 
         remove_index_for_dropped_table(kv_api, tenant, table_id_ident, &mut txn).await?;
+        append_delete_lineage_for_table_id_txn_ops(
+            kv_api,
+            tenant,
+            &mut txn,
+            table_id_ident.table_id,
+        )
+        .await?;
 
         // Count removed keys (approximate for DeleteByPrefix operations)
         let mut num_meta_keys_removed = 0;