feat(query): add history-based data lineage

.github/actions/test_logs/action.yml

@@ -4,6 +4,8 @@ runs:
   using: "composite"
   steps:
     - uses: ./.github/actions/setup_test
+      with:
+        artifacts: sqllogictests,meta,query
 
     - name: Install lsof
       shell: bash

src/common/tracing/src/config.rs

@@ -398,7 +398,8 @@ pub struct HistoryConfig {
 #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
 pub struct HistoryTableConfig {
     pub table_name: String,
-    pub retention: usize,
+    /// `None` is resolved to the table-specific policy by `init_history_tables`.
+    pub retention: Option<usize>,
     pub invisible: bool,
 }
 
@@ -415,7 +416,12 @@ impl Display for HistoryConfig {
             self.retention_interval,
             self.tables
                 .iter()
-                .map(|f| format!("{}({} hours)", f.table_name.clone(), f.retention))
+                .map(|f| match f.retention {
+                    Some(retention) => {
+                        format!("{}({} hours)", f.table_name, retention)
+                    }
+                    None => format!("{}(default)", f.table_name),
+                })
                 .join(", "),
             self.storage_params
                 .as_ref()
@@ -445,13 +451,21 @@ impl Default for HistoryTableConfig {
     fn default() -> Self {
         Self {
             table_name: "".to_string(),
-            retention: 168,
+            retention: None,
             invisible: false,
         }
     }
 }
 
 impl HistoryConfig {
+    pub fn is_table_enabled(&self, table_name: &str) -> bool {
+        self.on
+            && self
+                .tables
+                .iter()
+                .any(|table| table.table_name.eq_ignore_ascii_case(table_name))
+    }
+
     pub fn is_invisible(&self, table_name: &str) -> bool {
         self.tables
             .iter()

src/common/tracing/src/predefined_tables/history_tables.rs

@@ -22,39 +22,56 @@ use databend_common_exception::Result;
 use crate::HistoryConfig;
 
 const TABLES_TOML: &str = include_str!("./history_tables.toml");
+const DEFAULT_RETENTION_HOURS: u64 = 24 * 7;
+const LINEAGE_UNRESOLVED_TABLE: &str = "lineage_unresolved";
 
 #[derive(Debug)]
 pub struct HistoryTable {
     pub name: String,
     pub create: String,
-    pub transform: String,
-    pub delete: String,
+    pub transforms: Vec<String>,
+    pub delete: Option<String>,
 }
 
 impl HistoryTable {
-    pub fn create(predefined: PredefinedTable, retention: u64) -> Self {
+    pub fn create(predefined: PredefinedTable, retention: Option<u64>) -> Self {
+        let transforms = predefined.transforms();
         HistoryTable {
             name: predefined.name,
             create: predefined.create,
-            transform: predefined.transform,
-            delete: predefined
-                .delete
-                .replace("{retention_hours}", &retention.to_string()),
+            transforms,
+            delete: retention.map(|retention| {
+                predefined
+                    .delete
+                    .replace("{retention_hours}", &retention.to_string())
+            }),
         }
     }
 
-    pub fn assemble_log_history_transform(&self, stage_name: &str, batch_number: u64) -> String {
-        let mut transform = self.transform.clone();
-        transform = transform.replace("{stage_name}", stage_name);
-        transform = transform.replace("{batch_number}", &batch_number.to_string());
-        transform
+    pub fn assemble_log_history_transforms(
+        &self,
+        stage_name: &str,
+        batch_number: u64,
+    ) -> Vec<String> {
+        self.transforms
+            .iter()
+            .map(|transform| {
+                transform
+                    .replace("{stage_name}", stage_name)
+                    .replace("{batch_number}", &batch_number.to_string())
+            })
+            .collect()
     }
 
-    pub fn assemble_normal_transform(&self, begin: u64, end: u64) -> String {
-        let mut transform = self.transform.clone();
-        transform = transform.replace("{batch_begin}", &begin.to_string());
-        transform = transform.replace("{batch_end}", &end.to_string());
-        transform
+    pub fn assemble_normal_transforms(&self, begin: u64, end: u64) -> Vec<String> {
+        self.transforms
+            .iter()
+            .map(|transform| {
+                transform
+                    .replace("{batch_begin}", &begin.to_string())
+                    .replace("{batch_end}", &end.to_string())
+            })
+            .collect()
     }
 }
 
@@ -69,9 +86,19 @@ pub struct PredefinedTable {
     pub target: String,
     pub create: String,
     pub transform: String,
+    #[serde(default)]
+    pub additional_transforms: Vec<String>,
     pub delete: String,
 }
 
+impl PredefinedTable {
+    fn transforms(&self) -> Vec<String> {
+        std::iter::once(self.transform.clone())
+            .chain(self.additional_transforms.iter().cloned())
+            .collect()
+    }
+}
+
 pub fn init_history_tables(cfg: &HistoryConfig) -> Result<Vec<Arc<HistoryTable>>> {
     let predefined_tables: PredefinedTables =
         toml::from_str(TABLES_TOML).expect("Failed to parse toml");
@@ -93,11 +120,12 @@ pub fn init_history_tables(cfg: &HistoryConfig) -> Result<Vec<Arc<HistoryTable>>
             user_defined_log_history = true;
         }
         if let Some(predefined_table) = predefined_map.remove(&enable_table.table_name) {
-            let retention = enable_table.retention;
-            history_tables.push(Arc::new(HistoryTable::create(
-                predefined_table,
-                retention as u64,
-            )));
+            let retention = match enable_table.retention {
+                Some(retention) => Some(retention as u64),
+                None if enable_table.table_name == LINEAGE_UNRESOLVED_TABLE => None,
+                None => Some(DEFAULT_RETENTION_HOURS),
+            };
+            history_tables.push(Arc::new(HistoryTable::create(predefined_table, retention)));
         } else {
             return Err(ErrorCode::InvalidConfig(format!(
                 "Invalid history table name {}",
@@ -108,7 +136,7 @@ pub fn init_history_tables(cfg: &HistoryConfig) -> Result<Vec<Arc<HistoryTable>>
     if !user_defined_log_history {
         history_tables.push(Arc::new(HistoryTable::create(
             predefined_map.remove("log_history").unwrap(),
-            24 * 7,
+            Some(DEFAULT_RETENTION_HOURS),
         )));
     }
     Ok(history_tables)
@@ -135,3 +163,131 @@ pub fn get_all_history_table_names() -> Vec<String> {
         .map(|t| t.name)
         .collect()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_lineage_unresolved_uses_hash_pattern_merge() {
+        let tables: PredefinedTables = toml::from_str(TABLES_TOML).unwrap();
+        let lineage = tables
+            .tables
+            .iter()
+            .find(|table| table.name == "lineage_unresolved")
+            .unwrap();
+
+        assert!(
+            lineage
+                .create
+                .contains("column_lineage_hash STRING NOT NULL")
+        );
+        assert!(lineage.create.contains("source_to_target_columns MAP"));
+        assert!(lineage.create.contains("target_to_source_columns MAP"));
+        assert!(
+            lineage
+                .transform
+                .contains("MERGE INTO system_history.lineage_unresolved")
+        );
+        assert!(lineage.create.contains("source_catalog_type STRING"));
+        assert!(lineage.create.contains("target_catalog_type STRING"));
+        assert!(lineage.transform.contains("AS source_catalog_type"));
+        assert!(lineage.delete.contains("lineage_kind = 'DML'"));
+        assert!(
+            lineage
+                .transform
+                .contains("target.column_lineage_hash = source.column_lineage_hash")
+        );
+        assert!(lineage.transform.contains(
+            "PARTITION BY m['source']['lineage_key']::STRING, m['target']['lineage_key']::STRING, m['lineage_kind']::STRING, m['column_lineage_hash']::STRING"
+        ));
+        assert!(lineage.transform.contains(
+            "target.source_lineage_key = source.source_lineage_key AND target.target_lineage_key = source.target_lineage_key AND target.lineage_kind = source.lineage_kind"
+        ));
+        assert!(
+            lineage
+                .transform
+                .contains("WHEN MATCHED THEN UPDATE * WHEN NOT MATCHED THEN INSERT *")
+        );
+        assert!(
+            lineage
+                .transform
+                .contains("coalesce(m['operation']::STRING, 'UPSERT_EDGE') = 'UPSERT_EDGE'")
+        );
+        assert_eq!(lineage.additional_transforms.len(), 1);
+        assert!(lineage.additional_transforms[0].contains("DELETE_OBJECT"));
+        assert!(lineage.additional_transforms[0].contains("source_id IN"));
+        assert!(lineage.additional_transforms[0].contains("target_id IN"));
+    }
+
+    #[test]
+    fn test_lineage_same_batch_tombstone_replay_order() {
+        let tables: PredefinedTables = toml::from_str(TABLES_TOML).unwrap();
+        let lineage = tables
+            .tables
+            .into_iter()
+            .find(|table| table.name == LINEAGE_UNRESOLVED_TABLE)
+            .unwrap();
+        let history = HistoryTable::create(lineage, None);
+
+        // A same-batch tombstone must run after edge upserts. Replaying a batch repeats the same
+        // order, so the MERGE cannot resurrect an edge removed by the tombstone phase.
+        let first_attempt = history.assemble_normal_transforms(17, 23);
+        let replay = history.assemble_normal_transforms(17, 23);
+        assert_eq!(first_attempt, replay);
+        assert_eq!(first_attempt.len(), 2);
+        assert!(first_attempt[0].contains("MERGE INTO system_history.lineage_unresolved"));
+        assert!(first_attempt[0].contains("= 'UPSERT_EDGE'"));
+        assert!(first_attempt[1].starts_with("DELETE FROM system_history.lineage_unresolved"));
+        assert!(first_attempt[1].contains("= 'DELETE_OBJECT'"));
+        for phase in first_attempt {
+            assert!(phase.contains("batch_number >= 17"));
+            assert!(phase.contains("batch_number < 23"));
+        }
+    }
+
+    #[test]
+    fn test_lineage_retention_is_optional() {
+        let config = |table_name: &str, retention| HistoryConfig {
+            tables: vec![crate::HistoryTableConfig {
+                table_name: table_name.to_string(),
+                retention,
+                invisible: false,
+            }],
+            ..Default::default()
+        };
+
+        let lineage = init_history_tables(&config(LINEAGE_UNRESOLVED_TABLE, None)).unwrap();
+        assert!(
+            lineage
+                .iter()
+                .find(|table| table.name == LINEAGE_UNRESOLVED_TABLE)
+                .unwrap()
+                .delete
+                .is_none()
+        );
+
+        let lineage = init_history_tables(&config(LINEAGE_UNRESOLVED_TABLE, Some(24))).unwrap();
+        let delete = lineage
+            .iter()
+            .find(|table| table.name == LINEAGE_UNRESOLVED_TABLE)
+            .unwrap()
+            .delete
+            .as_ref()
+            .unwrap();
+        assert!(delete.contains("lineage_kind = 'DML'"));
+        assert!(delete.contains("subtract_hours(NOW(), 24)"));
+
+        let query = init_history_tables(&config("query_history", None)).unwrap();
+        assert!(
+            query
+                .iter()
+                .find(|table| table.name == "query_history")
+                .unwrap()
+                .delete
+                .as_ref()
+                .unwrap()
+                .contains("subtract_hours(NOW(), 168)")
+        );
+    }
+}

src/common/tracing/src/predefined_tables/history_tables.toml

@@ -33,3 +33,11 @@ target = "databend::log::access"
 create = "CREATE TABLE IF NOT EXISTS system_history.access_history (query_id VARCHAR NULL,  query_start TIMESTAMP NULL, user_name STRING NULL, base_objects_accessed VARIANT NULL, direct_objects_accessed VARIANT NULL, objects_modified VARIANT NULL, object_modified_by_ddl VARIANT NULL)"
 transform = "settings (timezone='Etc/UTC') INSERT INTO system_history.access_history FROM (SELECT m['query_id'], m['query_start'], m['user_name'], m['base_objects_accessed'], m['direct_objects_accessed'], m['objects_modified'], m['object_modified_by_ddl'] FROM (SELECT parse_json(message) as m FROM system_history.log_history WHERE target='databend::log::access' AND batch_number >= {batch_begin} AND batch_number < {batch_end}))"
 delete = "DELETE FROM system_history.access_history WHERE query_start < subtract_hours(NOW(), {retention_hours})"
+
+[[tables]]
+name = "lineage_unresolved"
+target = "databend::log::lineage"
+create = "CREATE TABLE IF NOT EXISTS system_history.lineage_unresolved (query_id STRING NULL, event_time TIMESTAMP NULL, query_kind STRING NULL, lineage_kind STRING NULL, column_lineage_hash STRING NOT NULL, source_lineage_key STRING NULL, source_address_kind STRING NULL, source_catalog_type STRING NULL, source_object_type STRING NULL, source_catalog STRING NULL, source_database STRING NULL, source_name STRING NULL, source_id UInt64 NULL, target_lineage_key STRING NULL, target_address_kind STRING NULL, target_catalog_type STRING NULL, target_object_type STRING NULL, target_catalog STRING NULL, target_database STRING NULL, target_name STRING NULL, target_id UInt64 NULL, source_to_target_columns MAP(STRING, ARRAY(STRING)) NOT NULL, target_to_source_columns MAP(STRING, ARRAY(STRING)) NOT NULL) CLUSTER BY LINEAR(event_time, target_lineage_key, source_lineage_key)"
+transform = "settings (timezone='Etc/UTC') MERGE INTO system_history.lineage_unresolved AS target USING (SELECT m['query_id']::STRING AS query_id, to_timestamp(m['event_time']::Int64) AS event_time, m['query_kind']::STRING AS query_kind, m['lineage_kind']::STRING AS lineage_kind, m['column_lineage_hash']::STRING AS column_lineage_hash, m['source']['lineage_key']::STRING AS source_lineage_key, m['source']['address_kind']::STRING AS source_address_kind, m['source']['catalog_type']::STRING AS source_catalog_type, m['source']['object_type']::STRING AS source_object_type, m['source']['catalog']::STRING AS source_catalog, m['source']['database']::STRING AS source_database, m['source']['name']::STRING AS source_name, m['source']['id']::UInt64 AS source_id, m['target']['lineage_key']::STRING AS target_lineage_key, m['target']['address_kind']::STRING AS target_address_kind, m['target']['catalog_type']::STRING AS target_catalog_type, m['target']['object_type']::STRING AS target_object_type, m['target']['catalog']::STRING AS target_catalog, m['target']['database']::STRING AS target_database, m['target']['name']::STRING AS target_name, m['target']['id']::UInt64 AS target_id, m['source_to_target_columns']::MAP(STRING, ARRAY(STRING)) AS source_to_target_columns, m['target_to_source_columns']::MAP(STRING, ARRAY(STRING)) AS target_to_source_columns FROM (SELECT parse_json(message) AS m FROM system_history.log_history WHERE target='databend::log::lineage' AND batch_number >= {batch_begin} AND batch_number < {batch_end}) WHERE coalesce(m['operation']::STRING, 'UPSERT_EDGE') = 'UPSERT_EDGE' QUALIFY row_number() OVER (PARTITION BY m['source']['lineage_key']::STRING, m['target']['lineage_key']::STRING, m['lineage_kind']::STRING, m['column_lineage_hash']::STRING ORDER BY m['event_time']::Int64 DESC, m['query_id']::STRING DESC) = 1) AS source ON target.source_lineage_key = source.source_lineage_key AND target.target_lineage_key = source.target_lineage_key AND target.lineage_kind = source.lineage_kind AND target.column_lineage_hash = source.column_lineage_hash WHEN MATCHED THEN UPDATE * WHEN NOT MATCHED THEN INSERT *;"
+additional_transforms = ["DELETE FROM system_history.lineage_unresolved WHERE (source_address_kind = 'ID' AND source_id IN (SELECT DISTINCT m['object_id']::UInt64 FROM (SELECT parse_json(message) AS m FROM system_history.log_history WHERE target='databend::log::lineage' AND batch_number >= {batch_begin} AND batch_number < {batch_end}) WHERE m['operation']::STRING = 'DELETE_OBJECT')) OR (target_address_kind = 'ID' AND target_id IN (SELECT DISTINCT m['object_id']::UInt64 FROM (SELECT parse_json(message) AS m FROM system_history.log_history WHERE target='databend::log::lineage' AND batch_number >= {batch_begin} AND batch_number < {batch_end}) WHERE m['operation']::STRING = 'DELETE_OBJECT'))"]
+delete = "DELETE FROM system_history.lineage_unresolved WHERE lineage_kind = 'DML' AND event_time < subtract_hours(NOW(), {retention_hours})"

src/meta/app/src/schema/catalog.rs

@@ -26,7 +26,7 @@ use crate::schema::catalog_id_ident;
 use crate::storage::StorageParams;
 use crate::tenant::Tenant;
 
-#[derive(Clone, Copy, PartialEq, Eq, Debug, Hash)]
+#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug, Hash)]
 pub enum CatalogType {
     Default = 1,
     Hive = 2,

src/query/catalog/src/table.rs

@@ -109,6 +109,15 @@ pub trait Table: Sync + Send {
 
     fn get_table_info(&self) -> &TableInfo;
 
+    /// Returns the source table whose data columns a stream exposes.
+    ///
+    /// Lineage intentionally passes through a stream to its source table.
+    /// Views are lineage boundaries and must not use this relation-level hook;
+    /// their output columns are annotated separately by the planner.
+    fn stream_source_table_info(&self) -> Option<&TableInfo> {
+        None
+    }
+
     fn get_data_source_info(&self) -> DataSourceInfo {
         DataSourceInfo::TableSource(self.get_table_info().clone())
     }

src/query/config/src/config.rs

@@ -2724,9 +2724,9 @@ pub struct HistoryLogTableConfig {
     /// Specifies the history table name
     pub table_name: String,
 
-    /// The retention period (in hours) for history logs.
-    /// Data older than this period will be deleted during retention tasks.
-    pub retention: usize,
+    /// The optional retention period (in hours) for history logs.
+    /// Unspecified tables use the default retention, except lineage, which is retained permanently.
+    pub retention: Option<usize>,
 
     /// Whether this history table is invisible for querying.
     /// Default is false.