[WIP] Lineage for data and code

dataci/config.py

@@ -8,6 +8,7 @@
 import configparser
 import logging
 import os
+from datetime import datetime, timezone
 from pathlib import Path
 from textwrap import dedent
 from threading import Event as ThreadEvent
@@ -97,6 +98,7 @@ def load_config():
 LOG_DIR = None
 LOG_LEVEL = None
 STORAGE_BACKEND = None
+TIMEZONE = datetime.now(timezone.utc).astimezone().tzinfo
 
 # DataCI Trigger and Scheduler server
 SERVER_ADDRESS = '0.0.0.0'

dataci/db/init.py

@@ -17,19 +17,32 @@
 # Drop all tables
 with db_connection:
     db_connection.executescript("""
+    DROP TABLE IF EXISTS lineage;
+    DROP TABLE IF EXISTS run;
     DROP TABLE IF EXISTS dataset_tag;
     DROP TABLE IF EXISTS dataset;
     DROP TABLE IF EXISTS workflow_dag_node;
     DROP TABLE IF EXISTS stage_tag;
     DROP TABLE IF EXISTS stage;
     DROP TABLE IF EXISTS workflow_tag;
     DROP TABLE IF EXISTS workflow;
+    DROP TABLE IF EXISTS job;
     """)
 logger.info('Drop all tables.')
 
 # Create dataset table
 with db_connection:
     db_connection.executescript("""
+    CREATE TABLE job
+    (
+        workspace TEXT,
+        name      TEXT,
+        version   TEXT,
+        type      TEXT,
+        PRIMARY KEY (workspace, name, version, type),
+        UNIQUE (workspace, name, version, type)
+    );
+
     CREATE TABLE workflow
     (
         workspace   TEXT,
@@ -45,7 +58,9 @@
         script_filelist  TEXT,
         script_hash      TEXT,
         PRIMARY KEY (workspace, name, version),
-        UNIQUE (workspace, name, version)
+        UNIQUE (workspace, name, version),
+        FOREIGN KEY (workspace, name, version)
+            REFERENCES job (workspace, name, version)
     );
 
     CREATE TABLE workflow_tag
@@ -73,7 +88,9 @@
         script_filelist  TEXT,
         script_hash      TEXT,
         PRIMARY KEY (workspace, name, version),
-        UNIQUE (workspace, name, version)
+        UNIQUE (workspace, name, version),
+        FOREIGN KEY (workspace, name, version)
+            REFERENCES job (workspace, name, version)
     );
 
     CREATE TABLE stage_tag
@@ -133,5 +150,42 @@
         FOREIGN KEY (workspace, name, version)
             REFERENCES dataset (workspace, name, version)
     );
+
+    CREATE TABLE run
+    (
+        workspace     TEXT,
+        name          TEXT,
+        version       INTEGER,
+        status        TEXT,
+        job_workspace TEXT,
+        job_name      TEXT,
+        job_version   TEXT,
+        job_type      TEXT,
+        create_time   INTEGER,
+        update_time   INTEGER,
+        PRIMARY KEY (workspace, name, version),
+        UNIQUE (name, version),
+        FOREIGN KEY (workspace, name, version)
+            REFERENCES job (workspace, name, version),
+        FOREIGN KEY (job_workspace, job_name, job_version, job_type)
+            REFERENCES job (workspace, name, version, type)
+    );
+
+    CREATE TABLE lineage
+    (
+        upstream_workspace TEXT,
+        upstream_name      TEXT,
+        upstream_version   TEXT,
+        upstream_type      TEXT,
+        downstream_workspace TEXT,
+        downstream_name      TEXT,
+        downstream_version   TEXT,
+        downstream_type      TEXT,
+        PRIMARY KEY (upstream_workspace, upstream_name, upstream_version, upstream_type, downstream_workspace, downstream_name, downstream_version, downstream_type),
+        FOREIGN KEY (upstream_workspace, upstream_name, upstream_version, upstream_type)
+            REFERENCES job (workspace, name, version, type),
+        FOREIGN KEY (downstream_workspace, downstream_name, downstream_version, downstream_type)
+            REFERENCES job (workspace, name, version, type)
+    );
     """)
 logger.info('Create all tables.')