Merge pull request #206 from lanl/issue198

allow users to overwrite existing schema, update er diagram writer and summary()

Author: jpulidojr

docs/index.rst

@@ -8,7 +8,6 @@ The Data Science Infrastructure Project (DSI)
 
 .. toctree::
    :maxdepth: 2
-   :caption: Contents:
 
    introduction
    installation
@@ -17,9 +16,15 @@ The Data Science Infrastructure Project (DSI)
    examples
    contributors
 
-Indices and tables
-==================
+Indices
+=======
 
 * :ref:`genindex`
 * :ref:`modindex`
-* :ref:`search`
+
+Contact Us
+==========
+
+For general inquiries or help, reach us at dsi-help (at) lanl.gov
+
+Bugs/Feature Requests: `DSI GitHub Requests <https://github.com/lanl/dsi/issues>`_

docs/introduction.rst

@@ -111,4 +111,11 @@ DSI Core
 
 DSI basic functionality is contained within the middleware known as the *core*.
 Users will leverage Core to employ Readers, Writers, and Backends to interact with their data.
-The two primary methods to achieve this are with the :ref:`python_api_label` or the :ref:`cli_api_label`
+The two primary methods to achieve this are with the :ref:`python_api_label` or the :ref:`cli_api_label`
+
+Contact Us
+----------
+
+For general inquiries or help, reach us at dsi-help (at) lanl.gov
+
+Bugs/Feature Requests: `DSI GitHub Requests <https://github.com/lanl/dsi/issues>`_

dsi/_version.py

@@ -1 +1 @@
-__version__ = "1.1.3"
+__version__ = "1.1.4"

dsi/backends/duckdb.py

@@ -175,6 +175,56 @@ def ingest_artifacts(self, collection, isVerbose=False):
         """
         artifacts = collection
 
+        self.cur.execute("BEGIN TRANSACTION")
+
+        if self.list() is not None and list(artifacts.keys()) == ["dsi_relations"]: 
+            pk_list = artifacts["dsi_relations"]["primary_key"]
+            fk_list = artifacts["dsi_relations"]["foreign_key"]
+            pk_tables = set(t[0] for t in pk_list)
+            fk_tables = set(t[0] for t in fk_list if t[0] != None)
+            all_schema_tables = pk_tables.union(fk_tables)
+            db_tables = [t[0] for t in self.list() if t[0] != "dsi_units"]
+
+            # check if tables from dsi_relations are all in the db
+            if all_schema_tables.issubset(set(db_tables)):
+                circ, _ = self.check_table_relations(all_schema_tables, artifacts["dsi_relations"])
+                if circ:
+                    return (ValueError, f"A complex schema with a circular dependency cannot be ingested into a DuckDB backend.")
+
+                drop_order = all_schema_tables
+                collect = self.process_artifacts()
+                if "dsi_relations" in collect.keys():
+                    curr_pk_tables = set(t[0] for t in collect["dsi_relations"]["primary_key"])
+                    curr_fk_tables = set(t[0] for t in collect["dsi_relations"]["foreign_key"] if t[0] != None)
+                    curr_schema_tables = curr_pk_tables.union(curr_fk_tables)
+
+                    # need to drop and reingest all tables in old schema and new schema
+                    all_schema_tables = all_schema_tables.union(curr_schema_tables)
+
+                    _, ord_tables1 = self.check_table_relations(all_schema_tables, collect["dsi_relations"])
+                    drop_order = ord_tables1
+                
+                for table in drop_order:
+                    self.cur.execute(f'DROP TABLE IF EXISTS "{table}";')
+                try:
+                    self.con.commit()
+                except Exception as e:
+                    self.cur.execute("ROLLBACK")
+                    self.cur.execute("CHECKPOINT")
+                    return (duckdb.Error, e)
+                
+                #do not reingest tables not in old or new schema as they will be the same
+                non_schema_tables = set(db_tables) - all_schema_tables 
+                for t in non_schema_tables:
+                    del collect[t]
+                
+                collect["dsi_relations"] = artifacts["dsi_relations"]
+                artifacts = collect
+
+            else:
+                print("WARNING: Complex schemas can only be ingested if all referenced data tables are loaded into DSI.")
+            
+
         table_order = artifacts.keys()
         if "dsi_relations" in artifacts.keys():
             circular, ordered_tables = self.check_table_relations(artifacts.keys(), artifacts["dsi_relations"])
@@ -184,10 +234,8 @@ def ingest_artifacts(self, collection, isVerbose=False):
             else:
                 table_order = list(reversed(ordered_tables)) # ingest primary key tables first then children
 
-        self.cur.execute("BEGIN TRANSACTION")
         if self.runTable:
-            runTable_create = "CREATE TABLE IF NOT EXISTS runTable " \
-            "(run_id INTEGER PRIMARY KEY, run_timestamp TEXT UNIQUE);"
+            runTable_create = "CREATE TABLE IF NOT EXISTS runTable (run_id INTEGER PRIMARY KEY, run_timestamp TEXT UNIQUE);"
             self.cur.execute(runTable_create)
 
             sequence_run_id = "CREATE SEQUENCE IF NOT EXISTS seq_run_id START 1;"
@@ -387,13 +435,16 @@ def notebook(self, interactive=False):
     def read_to_artifact(self):
         return self.process_artifacts()
 
-    def process_artifacts(self):
+    def process_artifacts(self, only_units_relations = False):
         """
         Reads data from the DuckDB database into a nested OrderedDict.
         Keys are table names, and values are OrderedDicts containing table data.
 
         If the database contains PK/FK relationships, they are stored in a special `dsi_relations` table.
 
+        `only_units_relations` : bool, default=False
+            **USERS SHOULD IGNORE THIS FLAG.** Used internally by duckdb.py.
+
         `return` : OrderedDict
             A nested OrderedDict containing all data from the DuckDB database.
         """
@@ -404,20 +455,22 @@ def process_artifacts(self):
                                      SELECT table_name FROM information_schema.tables
                                      WHERE table_schema = 'main' AND table_type = 'BASE TABLE'
                                      """).fetchall()
-        for item in tableList:
-            tableName = self.duckdb_compatible_name(item[0])
 
-            tableInfo = self.cur.execute(f"PRAGMA table_info({tableName});").fetchdf()
-            colDict = OrderedDict((self.duckdb_compatible_name(col), []) for col in tableInfo['name'])
+        if only_units_relations == False:
+            for item in tableList:
+                tableName = self.duckdb_compatible_name(item[0])
+
+                tableInfo = self.cur.execute(f"PRAGMA table_info({tableName});").fetchdf()
+                colDict = OrderedDict((self.duckdb_compatible_name(col), []) for col in tableInfo['name'])
 
-            data = self.cur.execute(f"SELECT * FROM {tableName};").fetchall()
-            for row in data:
-                for colName, val in zip(colDict.keys(), row):
-                    if val == "NULL":
-                        colDict[colName].append(None)
-                    else:
-                        colDict[colName].append(val)
-            artifact[tableName] = colDict
+                data = self.cur.execute(f"SELECT * FROM {tableName};").fetchall()
+                for row in data:
+                    for colName, val in zip(colDict.keys(), row):
+                        if val == "NULL":
+                            colDict[colName].append(None)
+                        else:
+                            colDict[colName].append(val)
+                artifact[tableName] = colDict
 
         pk_list = []
         fkData = self.cur.execute(f"""
@@ -743,6 +796,8 @@ def list(self):
                                      SELECT table_name FROM information_schema.tables
                                      WHERE table_schema = 'main' AND table_type = 'BASE TABLE'
                                      """).fetchall()
+        if not tableList:
+            return None
         tableList = [self.duckdb_compatible_name(table[0]) for table in tableList]
 
         info_list = []
@@ -839,12 +894,13 @@ def summary_helper(self, table_name):
         col_info = self.cur.execute(f"PRAGMA table_info({table_name})").fetchall()
 
         numeric_types = {'INTEGER', 'REAL', 'FLOAT', 'NUMERIC', 'DECIMAL', 'DOUBLE', 'BIGINT'}
-        headers = ['column', 'type', 'min', 'max', 'avg', 'std_dev']
+        headers = ['column', 'type', 'unique', 'min', 'max', 'avg', 'std_dev']
         rows = []
 
         for col in col_info:
             col_name = col[1]
             col_type = col[2].upper()
+            unique_vals = self.cur.execute(f"SELECT COUNT(DISTINCT {col_name}) FROM {table_name};").fetchone()[0]
             is_primary = col[5] > 0
             display_name = f"{col_name}*" if is_primary else col_name
 
@@ -863,7 +919,7 @@ def summary_helper(self, table_name):
 
             if avg_val != None and std_dev == None:
                 std_dev = 0
-            rows.append([display_name, col_type, min_val, max_val, avg_val, std_dev])
+            rows.append([display_name, col_type, unique_vals, min_val, max_val, avg_val, std_dev])
 
         return headers, rows
 
@@ -1007,7 +1063,7 @@ def visit(node):
         if any(visit(node) for node in list(graph.keys())):
             return True, None  # Circular dependency detected
 
-        # Step 3: Order tables from least dependencies to most (if no circular dependencies)
+        # Order tables from least dependencies to most (if no circular dependencies)
         in_degree = {table: 0 for table in tables}
         for child in graph:
             for parent in graph[child]:

dsi/backends/sqlite.py

@@ -183,6 +183,80 @@ def ingest_artifacts(self, collection, isVerbose=False):
         #     self.cur.execute("PRAGMA FOREIGN KEYS = ON;")
         #     self.con.commit()
 
+        #there are tables in db, only dsi_relations to be ingested
+        if self.list() is not None and list(artifacts.keys()) == ["dsi_relations"]: 
+            pk_list = artifacts["dsi_relations"]["primary_key"]
+            fk_list = artifacts["dsi_relations"]["foreign_key"]
+            pk_tables = set(t[0] for t in pk_list)
+            fk_tables = set(t[0] for t in fk_list if t[0] != None)
+            all_schema_tables = pk_tables.union(fk_tables)
+            db_tables = [t[0] for t in self.list() if t[0] != "dsi_units"]
+            # check if tables from dsi_relations are all in the db
+            if all_schema_tables.issubset(set(db_tables)):
+                for table in all_schema_tables:
+                    pk_cols = list({v for k, v in pk_list if k == table})
+                    fk_dict = {fks[1]: pks for pks, fks, in zip(pk_list, fk_list) if fks[0] == table}
+                    
+                    #1. create new table using existing table's columns but with new relations
+                        # use consistent naming scheme. ex: table1 --> table1_dsi_temmp
+                    create_stmt = f"CREATE TABLE {table}_dsi_temp ("
+                    table_info = self.cur.execute(f"PRAGMA table_info({table});").fetchall()
+                    for col in table_info:
+                        if col[1] in pk_cols:
+                            create_stmt += f"{col[1]} {col[2]} PRIMARY KEY, "
+                        else:
+                            create_stmt += f"{col[1]} {col[2]}, "
+                    
+                    if fk_dict:
+                        fk_stmt = ""
+                        for k, v in fk_dict.items():
+                            if k not in create_stmt:
+                                msg = f"Input schema references a nonexistent column, {k}, in the foreign_key section of {table}"
+                                raise ValueError(msg)
+                            fk_stmt += f"FOREIGN KEY ({k}) REFERENCES {v[0]}({v[1]}), "
+                        create_stmt += fk_stmt
+                    create_stmt = create_stmt[:-2] + ");"
+
+                    #2. move all data from existing to new table
+                    try:
+                        self.cur.execute(create_stmt)
+                        self.cur.execute(f"INSERT INTO {table}_dsi_temp SELECT * FROM {table};")
+                    except sqlite3.Error as e:
+                        self.con.rollback()
+                        return (sqlite3.Error, e)
+                
+                # check if other non-schema tables have constraints -- if yes, make temp table, move data, and drop old table with schema tables
+                other_tables = list(set(db_tables) - all_schema_tables)
+                for table in other_tables:
+                    table_info = self.cur.execute(f"PRAGMA table_info({table});").fetchall()
+                    if any(col[5] > 0 for col in table_info) or len(self.cur.execute(f"PRAGMA foreign_key_list({table});").fetchall()) > 0:
+                        create_stmt = f"CREATE TABLE {table}_dsi_temp ("
+                        create_stmt += ", ".join(f"{col[1]} {col[2]}" for col in table_info)
+                        create_stmt += ");"
+                        try:
+                            self.cur.execute(create_stmt)
+                            self.cur.execute(f"INSERT INTO {table}_dsi_temp SELECT * FROM {table};")
+                        except sqlite3.Error as e:
+                            self.con.rollback()
+                            return (sqlite3.Error, e)
+                        all_schema_tables.add(table)
+
+                # reaching here means no errors with new schema --- SQLITE doesnt care about order of tables deleted
+                # drop all tables in schema and rename temp tables to original names
+                for table in all_schema_tables:
+                    self.cur.execute(f'DROP TABLE IF EXISTS "{table}";')
+                    self.cur.execute(f'ALTER TABLE {table}_dsi_temp RENAME TO {table};')
+                
+                try:
+                    self.con.commit()
+                except Exception as e:
+                    self.con.rollback()
+                    return (sqlite3.Error, e)
+            
+                return #early return so dont make any other changes to db
+            else:
+                print("WARNING: Complex schemas can only be ingested if all referenced data tables are loaded into DSI.")
+            
         if self.runTable:
             runTable_create = "CREATE TABLE IF NOT EXISTS runTable (run_id INTEGER PRIMARY KEY AUTOINCREMENT, run_timestamp TEXT UNIQUE);"
             self.cur.execute(runTable_create)
@@ -831,6 +905,8 @@ def list(self):
         Return a list of all tables and their dimensions from this SQLite backend
         """
         tableList = self.cur.execute("SELECT name FROM sqlite_master WHERE type ='table';").fetchall()
+        if not tableList:
+            return None
         tableList = [self.sqlite_compatible_name(table[0]) for table in tableList if table[0] != "sqlite_sequence"]
 
         info_list = []
@@ -916,12 +992,13 @@ def summary_helper(self, table_name):
         col_info = self.cur.execute(f"PRAGMA table_info({table_name})").fetchall()
 
         numeric_types = {'INTEGER', 'REAL', 'FLOAT', 'NUMERIC', 'DECIMAL', 'DOUBLE'}
-        headers = ['column', 'type', 'min', 'max', 'avg', 'std_dev']
+        headers = ['column', 'type', 'unique', 'min', 'max', 'avg', 'std_dev']
         rows = []
 
         for col in col_info:
             col_name = col[1]
             col_type = col[2].upper()
+            unique_vals = self.cur.execute(f"SELECT COUNT(DISTINCT {col_name}) FROM {table_name};").fetchone()[0]
             is_primary = col[5] > 0
             display_name = f"{col_name}*" if is_primary else col_name
 
@@ -955,7 +1032,7 @@ def summary_helper(self, table_name):
 
             if avg_val != None and std_dev == None:
                 std_dev = 0
-            rows.append([display_name, col_type, min_val, max_val, avg_val, std_dev])
+            rows.append([display_name, col_type, unique_vals, min_val, max_val, avg_val, std_dev])
 
         return headers, rows
 

dsi/core.py

@@ -485,6 +485,8 @@ def artifact_handler(self, interaction_type, query = None, **kwargs):
                     if self.debug_level != 0:
                         self.logger.error(f"Error ingesting data in {original_file} @ line {return_line_number} due to {errorMessage[1]}")
                     if self.user_wrapper:
+                        if errorMessage[1].startswith("A complex schema"):
+                            raise errorMessage[0](errorMessage[1])
                         raise errorMessage[0](f"Error ingesting data due to {errorMessage[1]}")
                     else:
                         raise errorMessage[0](f"Error ingesting data in {original_file} @ line {return_line_number} due to {errorMessage[1]}")