Encode additional parameters in Component.data and store in 1 table

Author: blythed

.github/workflows/ci_code.yml

@@ -103,6 +103,7 @@ jobs:
 
     - name: Unit Testing
       run: |
+        sqlite3 test.db "create table t(f int); drop table t;"
         make unit_testing pytest_arguments="--cov=superduper --cov-report=xml" SUPERDUPER_CONFIG=test/configs/${{ matrix.config }}
 
     - name: Usecase Testing

.github/workflows/ci_plugins.yaml

@@ -95,6 +95,9 @@ jobs:
       - name: Plugin Testing
         run: |
           export PYTHONPATH=./
+          if [ "${{ matrix.plugin }}" = "sql" ]; then
+            sqlite3 test.db "create table t(f int); drop table t;"
+          fi
           if [ -d "plugins/${{ matrix.plugin }}/plugin_test" ]; then
             pytest --cov=superduper --cov-report=xml plugins/${{ matrix.plugin }}/plugin_test
           else
@@ -103,6 +106,9 @@ jobs:
 
       - name: Optionally run the base testing
         run: |
+          if [ "${{ matrix.plugin }}" = "sql" ]; then
+            sqlite3 test.db "create table t(f int); drop table t;"
+          fi
           SUPERDUPER_CONFIG="plugins/${{ matrix.plugin }}/plugin_test/config.yaml"
           if [ -f "$SUPERDUPER_CONFIG" ]; then
             echo "Running the base testing..."

CHANGELOG.md

@@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Add assertion to verify directory copy in FileSystemArtifactStore
 - Batch the Qdrant requests and add a retry to the config of Qdrant
 - Add use_component_cache to config
+- Save data in the `Component` table instead of in individual tables
 
 ### Bug fixes
 

plugins/qdrant/superduper_qdrant/qdrant.py

@@ -160,7 +160,6 @@ def _do_scroll(offset):
 
         return ids
 
-
     def _create_collection(self):
         measure = (
             self.measure.name

plugins/sql/plugin_test/config.yaml

@@ -1,4 +1,4 @@
-data_backend: sqlite://
+data_backend: sqlite://./test.db
 auto_schema: false
 force_apply: true
 json_native: false

plugins/sql/plugin_test/test_query.py

@@ -1,3 +1,4 @@
+import typing as t
 from test.utils.setup.fake_data import add_listeners, add_models, add_random_data
 
 import numpy as np
@@ -44,7 +45,7 @@ def test_renamings(db):
     add_random_data(db, n=5)
     add_models(db)
     add_listeners(db)
-    t = db["documents"]
+    t = db["documentz"]
     listener_uuid = [db.load('Listener', k).outputs for k in db.show("Listener")][0]
     q = t.select("id", "x", "y").outputs(listener_uuid.split('__', 1)[-1])
     data = q.execute()
@@ -62,13 +63,13 @@ def test_serialize_query(db):
 
 def test_get_data(db):
     add_random_data(db, n=5)
-    db["documents"].limit(2)
-    db.metadata.get_component("Table", "documents")
+    db["documentz"].limit(2)
+    db.metadata.get_component("Table", "documentz")
 
 
 def test_insert_select(db):
     add_random_data(db, n=5)
-    q = db["documents"].select("id", "x", "y").limit(2)
+    q = db["documentz"].select("id", "x", "y").limit(2)
     r = q.execute()
 
     assert len(r) == 2
@@ -77,7 +78,7 @@ def test_insert_select(db):
 
 def test_filter(db):
     add_random_data(db, n=5)
-    t = db["documents"]
+    t = db["documentz"]
     q = t.select("id", "y")
     r = q.execute()
     ys = [x["y"] for x in r]
@@ -88,17 +89,18 @@ def test_filter(db):
     assert len(r) == uq[1][0]
 
 
-class documents(Base):
+class documents_plugin(Base):
+    primary_id: t.ClassVar[str] = 'id'
     this: 'str'
 
 
 def test_select_using_ids(db):
-    db.create(documents)
+    db.create(documents_plugin)
 
-    table = db["documents"]
+    table = db["documents_plugin"]
     table.insert([{"this": f"is a test {i}", "id": str(i)} for i in range(4)])
 
-    basic_select = db['documents'].select()
+    basic_select = db['documents_plugin'].select()
 
     assert len(basic_select.execute()) == 4
     assert len(basic_select.subset(['1', '2'])) == 2
@@ -112,16 +114,16 @@ def my_func(this: str):
 
     my_func = ObjectModel('my_func', object=my_func)
 
-    db.create(documents)
+    db.create(documents_plugin)
 
-    table = db["documents"]
+    table = db["documents_plugin"]
     table.insert([{"this": f"is a test {i}", "id": str(i)} for i in range(4)])
 
     listener = Listener(
         'test',
         model=my_func,
         key='this',
-        select=db['documents'].select(),
+        select=db['documents_plugin'].select(),
     )
     db.apply(listener)
 

plugins/sql/superduper_sql/data_backend.py

@@ -359,7 +359,10 @@ class SQLDatabackend(IbisDataBackend):
 
     def __init__(self, uri, plugin, flavour=None):
         super().__init__(uri, plugin, flavour)
-        self._create_sqlalchemy_engine()
+        if 'sqlite://./' in uri:
+            self._create_sqlalchemy_engine(uri.replace('./', '//'))
+        else:
+            self._create_sqlalchemy_engine(uri)
         self.sm = sessionmaker(bind=self.alchemy_engine)
 
     @property
@@ -374,6 +377,8 @@ def update(self, table, condition, key, value):
         with self.sm() as session:
             metadata = MetaData()
 
+            assert table in self.list_tables()
+
             metadata.reflect(bind=session.bind)
             table = Table(table, metadata, autoload_with=session.bind)
 
@@ -422,16 +427,16 @@ def delete(self, table, condition):
         except NoSuchTableError:
             raise exceptions.NotFound("Table", table)
 
-    def _create_sqlalchemy_engine(self):
+    def _create_sqlalchemy_engine(self, uri):
         with self.connection_manager.get_connection() as conn:
-            self.alchemy_engine = create_engine(self.uri, creator=lambda: conn.con)
+            self.alchemy_engine = create_engine(uri, creator=lambda: conn.con)
             if not self._test_engine():
                 logging.warn(
                     "Unable to reuse the ibis connection "
                     "to create the SQLAlchemy engine. "
                     "Creating a new connection with the URI."
                 )
-                self.alchemy_engine = create_engine(self.uri)
+                self.alchemy_engine = create_engine(uri)
 
     def _test_engine(self):
         """Test the engine."""

superduper/backends/base/data_backend.py

@@ -1,3 +1,4 @@
+import copy
 import functools
 import hashlib
 import json
@@ -643,11 +644,7 @@ def delete(self, table, condition):
         :param table: The table to delete from.
         :param condition: The condition to delete.
         """
-        r_table = self._get_with_component_identifier('Table', table)
-        if r_table is None:
-            raise exceptions.NotFound("Table", table)
-
-        if not r_table['is_component']:
+        if table not in {'Deployment', 'Component'}:
             pid = self.primary_id(table)
             if pid in condition:
                 docs = self.get_many(table, condition[pid])
@@ -660,13 +657,16 @@ def delete(self, table, condition):
             if 'uuid' in condition:
                 docs = self.get_many(table, '*', condition['uuid'])
             elif 'identifier' in condition:
-                docs = self.get_many(table, condition['identifier'], '*')
+                assert 'component' in condition
+                docs = self.get_many(
+                    table, condition['component'], condition['identifier'], '*'
+                )
                 docs = self._do_filter(docs, condition)
             else:
-                docs = self.get_many(table, '*', '*')
+                docs = self.get_many(table, '*', '*', '*')
                 docs = self._do_filter(docs, condition)
             for r in docs:
-                del self[table, r['identifier'], r['uuid']]
+                del self[table, r['component'], r['identifier'], r['uuid']]
 
     def drop_table(self, table):
         """Drop data from table.
@@ -735,9 +735,7 @@ def replace(self, table, condition, r):
         :param condition: The condition to update.
         :param r: The document to replace.
         """
-        r_table = self._get_with_component_identifier('Table', table)
-
-        if not r_table['is_component']:
+        if table not in {'Component', 'Deployment'}:
             pid = self.primary_id(table)
             if pid in condition:
                 docs = self.get_many(table, condition[pid])
@@ -749,18 +747,21 @@ def replace(self, table, condition, r):
                 self[table, s[pid]] = r
         else:
             if 'uuid' in condition:
-                s = self.get_many(table, '*', condition['uuid'])[0]
-                self[table, s['identifier'], condition['uuid']] = r
+                s = self.get_many(table, '*', '*', condition['uuid'])[0]
+                self[table, s['component'], s['identifier'], condition['uuid']] = r
             elif 'identifier' in condition:
-                docs = self.get_many(table, condition['identifier'], '*')
+                assert 'component' in condition
+                docs = self.get_many(
+                    table, condition['component'], condition['identifier'], '*'
+                )
                 docs = self._do_filter(docs, condition)
                 for s in docs:
-                    self[table, s['identifier'], s['uuid']] = r
+                    self[table, s['component'], s['identifier'], s['uuid']] = r
             else:
-                docs = self.get_many(table, '*', '*')
+                docs = self.get_many(table, '*', '*', '*')
                 docs = self._do_filter(docs, condition)
                 for s in docs:
-                    self[table, s['identifier'], s['uuid']] = r
+                    self[table, s['component'], s['identifier'], s['uuid']] = r
 
     def update(self, table, condition, key, value):
         """Update data in the database.
@@ -770,9 +771,7 @@ def update(self, table, condition, key, value):
         :param key: The key to update.
         :param value: The value to update.
         """
-        r_table = self._get_with_component_identifier('Table', table)
-
-        if not r_table['is_component']:
+        if table not in {'Component', 'Deployment'}:
             pid = self.primary_id(table)
             if pid in condition:
                 docs = self.get_many(table, '*', condition[pid]) + self.get_many(
@@ -788,19 +787,21 @@ def update(self, table, condition, key, value):
             if 'uuid' in condition:
                 s = self.get_many(table, '*', condition['uuid'])[0]
                 s[key] = value
-                self[table, s['identifier'], condition['uuid']] = s
+                self[table, s['component'], s['identifier'], condition['uuid']] = s
             elif 'identifier' in condition:
-                docs = self.get_many(table, condition['identifier'], '*')
+                docs = self.get_many(
+                    table, condition['component'], condition['identifier'], '*'
+                )
                 docs = self._do_filter(docs, condition)
                 for s in docs:
                     s[key] = value
-                    self[table, s['identifier'], s['uuid']] = s
+                    self[table, s['component'], s['identifier'], s['uuid']] = s
             else:
-                docs = self.get_many(table, '*', '*')
+                docs = self.get_many(table, '*', '*', '*')
                 docs = self._do_filter(docs, condition)
                 for s in docs:
                     s[key] = value
-                    self[table, s['identifier'], s['uuid']] = s
+                    self[table, s['component'], s['identifier'], s['uuid']] = s
 
     @abstractmethod
     def keys(self, *pattern) -> t.List[t.Tuple[str, str, str]]:
@@ -882,9 +883,11 @@ def insert(self, table, documents):
         except exceptions.NotFound:
             pid = None
 
-        if ('uuid' == pid or not pid) and "uuid" in documents[0]:
+        if table in {'Component', 'Deployment'}:
             for r in documents:
-                self[table, r['identifier'], r['uuid']] = r
+                self[table, r['component'], r['identifier'], r['uuid']] = copy.deepcopy(
+                    r
+                )
                 ids.append(r['uuid'])
         elif pid:
             pid = self.primary_id(table)
@@ -944,13 +947,7 @@ def do_test(r):
                         return False
                 return True
 
-        tables = self.get_many('Table', query.table, '*')
-        if not tables:
-            raise exceptions.NotFound("Table", query.table)
-
-        is_component = max(tables, key=lambda x: x['version'])['is_component']
-
-        if not is_component:
+        if query.table not in {'Component', 'Deployment'}:
             pid = self.primary_id(query.table)
 
             if pid in filter_kwargs:
@@ -966,41 +963,47 @@ def do_test(r):
         else:
 
             if not filter_kwargs:
-                keys = self.keys(query.table, '*', '*')
+                keys = self.keys(query.table, '*', '*', '*')
                 docs = [self[k] for k in keys]
             elif set(filter_kwargs.keys()) == {'uuid'}:
-                keys = self.keys(query.table, '*', filter_kwargs['uuid']['value'])
+                keys = self.keys(query.table, '*', '*', filter_kwargs['uuid']['value'])
                 docs = [self[k] for k in keys]
             elif set(filter_kwargs.keys()) == {'identifier'}:
                 assert filter_kwargs['identifier']['op'] == '=='
-
                 keys = self.keys(query.table, filter_kwargs['identifier']['value'], '*')
                 docs = [self[k] for k in keys]
-            elif set(filter_kwargs.keys()) == {'identifier', 'uuid'}:
+            elif set(filter_kwargs.keys()) == {'component', 'identifier', 'uuid'}:
                 assert filter_kwargs['identifier']['op'] == '=='
                 assert filter_kwargs['uuid']['op'] == '=='
+                assert filter_kwargs['component']['op'] == '=='
 
                 r = self[
                     query.table,
+                    filter_kwargs['component']['value'],
                     filter_kwargs['identifier']['value'],
                     filter_kwargs['uuid']['value'],
                 ]
                 if r is None:
                     docs = []
                 else:
                     docs = [r]
-            elif set(filter_kwargs.keys()) == {'identifier', 'version'}:
+            elif set(filter_kwargs.keys()) == {'component', 'identifier', 'version'}:
                 assert filter_kwargs['identifier']['op'] == '=='
+                assert filter_kwargs['component']['op'] == '=='
                 assert filter_kwargs['version']['op'] == '=='
 
-                keys = self.keys(query.table, filter_kwargs['identifier']['value'], '*')
+                keys = self.keys(
+                    query.table,
+                    filter_kwargs['component']['value'],
+                    filter_kwargs['identifier']['value'],
+                    '*',
+                )
                 docs = [self[k] for k in keys]
                 docs = [
                     r for r in docs if r['version'] == filter_kwargs['version']['value']
                 ]
             else:
-
-                keys = self.keys(query.table, '*', '*')
+                keys = self.keys(query.table, '*', '*', '*')
                 docs = [self[k] for k in keys]
                 docs = [r for r in docs if do_test(r)]
 
@@ -1011,4 +1014,4 @@ def do_test(r):
             cols = query.decomposition.select.args
             for i, r in enumerate(docs):
                 docs[i] = {k: v for k, v in r.items() if k in cols}
-        return docs
+        return [copy.deepcopy(r) for r in docs]