fix: remove delete tags for cell_table and meta_table (#215)

* fix: remove delete tags for cell_table and meta_table

* refactor: remove unused code blocks

* refactor: remove unused code blocks

* test: fix filter test

* refactor: simplify if else expression

Author: jemmyshin

annlite/container.py

@@ -105,7 +105,7 @@ def ivf_search(
                 continue
 
             indices = None
-            if where_clause or (cell_table.deleted_count() > 0):
+            if where_clause:
                 indices = cell_table.query(
                     where_clause=where_clause, where_params=where_params
                 )
@@ -340,7 +340,6 @@ def update(
             _cell_id, _offset = self._meta_table.get_address(doc.id)
             if cell_id == _cell_id:
                 self.vec_index(cell_id).add_with_ids(x.reshape(1, -1), [_offset])
-                self.cell_table(cell_id).undo_delete_by_offset(_offset)
                 self.doc_store(cell_id).update([doc])
                 self.meta_table.add_address(doc.id, cell_id, _offset)
                 update_success += 1

annlite/executor.py

@@ -49,6 +49,7 @@ def __init__(
         index_access_paths: str = '@r',
         search_access_paths: str = '@r',
         columns: Optional[Union[List[Tuple[str, str]], Dict[str, str]]] = None,
+        list_like: Optional[bool] = False,
         dim: int = None,
         *args,
         **kwargs,
@@ -103,6 +104,7 @@ def __init__(
             'max_connection': max_connection,
             'data_path': data_path or self.workspace or './workspace',
             'columns': columns,
+            'list_like': list_like,
         }
         self._index = DocumentArray(storage='annlite', config=config)
 

annlite/index.py

@@ -606,8 +606,14 @@ def index_hash(self):
             if platform.system() == 'Windows':
                 return date_time.isoformat('#', 'hours')
             return date_time.isoformat('#', 'seconds')
+        else:
+            import datetime
 
-        return None
+            return (
+                datetime.datetime.utcnow().isoformat('#', 'hours')
+                if platform.system() == 'Windows'
+                else datetime.datetime.utcnow().isoformat('#', 'seconds')
+            )
 
     @property
     def index_path(self):

annlite/storage/table.py

@@ -191,7 +191,7 @@ def add_column(self, name: str, dtype: str, create_index: bool = True):
 
     def create_index(self, column: str, commit: bool = True):
         sql_statement = f'''CREATE INDEX idx_{column}_
-                            ON {self.name}(_deleted, {column})'''
+                            ON {self.name}({column})'''
         self._conn.execute(sql_statement)
 
         if commit:
@@ -200,17 +200,12 @@ def create_index(self, column: str, commit: bool = True):
     def create_table(self):
         sql = f'''CREATE TABLE {self.name}
                     (_id INTEGER PRIMARY KEY AUTOINCREMENT,
-                     _doc_id TEXT NOT NULL UNIQUE,
-                     _deleted NUMERIC DEFAULT 0'''
+                     _doc_id TEXT NOT NULL UNIQUE'''
         if len(self._columns) > 0:
             sql += ', ' + ', '.join(self._columns)
         sql += ')'
         self._conn.execute(sql)
 
-        sql_statement = f'''CREATE INDEX idx__delete_
-                                ON {self.name}(_deleted)'''
-        self._conn.execute(sql_statement)
-
         for name in self._indexed_keys:
             self.create_index(name, commit=False)
         self._conn.commit()
@@ -281,18 +276,22 @@ def query(
         :return: offsets list of matched docs
         """
 
-        where_conds = ['_deleted = ?']
+        where_conds = []
+        where = None
         if where_clause:
             where_conds.append(where_clause)
-        where = ' and '.join(where_conds)
+            where = ' and '.join(where_conds)
 
         _order_by = f'{order_by or "_id"} {"ASC" if ascending else "DESC"}'
         _limit = f'LIMIT {limit}' if limit > 0 else ''
         _offset = f'OFFSET {offset}' if offset > 0 else ''
 
-        sql = f'SELECT _id from {self.name} WHERE {where} ORDER BY {_order_by} {_limit} {_offset}'
+        if where:
+            sql = f'SELECT _id from {self.name} WHERE {where} ORDER BY {_order_by} {_limit} {_offset}'
+        else:
+            sql = f'SELECT _id from {self.name} ORDER BY {_order_by} {_limit} {_offset}'
 
-        params = (0,) + tuple([_converting(p) for p in where_params])
+        params = tuple([_converting(p) for p in where_params])
 
         # # EXPLAIN SQL query
         # for row in self._conn.execute('EXPLAIN QUERY PLAN ' + sql, params):
@@ -308,7 +307,10 @@ def _offset_factory(_, record):
         cursor = self._conn.cursor()
 
         try:
-            offsets = cursor.execute(sql, params).fetchall()
+            if where:
+                offsets = cursor.execute(sql, params).fetchall()
+            else:
+                offsets = cursor.execute(sql).fetchall()
             self._conn.row_factory = None
             return offsets if offsets else []
         except Exception as e:
@@ -320,12 +322,12 @@ def delete(self, doc_ids: List[str]):
 
         :param doc_ids: The IDs of docs
         """
-        sql = f'UPDATE {self.name} SET _deleted = 1 WHERE _doc_id = ?'
+        sql = f'DELETE from {self.name} WHERE _doc_id = ?'
         self._conn.executemany(sql, doc_ids)
         self._conn.commit()
 
     def get_docid_by_offset(self, offset: int):
-        sql = f'SELECT _doc_id from {self.name} WHERE _id = ? and _deleted = 0 LIMIT 1;'
+        sql = f'SELECT _doc_id from {self.name} WHERE _id = ? LIMIT 1;'
         result = self._conn.execute(sql, (offset + 1,)).fetchone()
         if result:
             return result[0]
@@ -336,17 +338,12 @@ def delete_by_offset(self, offset: int):
 
         :param offset: The offset of the doc
         """
-        sql = f'UPDATE {self.name} SET _deleted = 1 WHERE _id = ?'
-        self._conn.execute(sql, (offset + 1,))
-        self._conn.commit()
-
-    def undo_delete_by_offset(self, offset: int):
-        sql = f'UPDATE {self.name} SET _deleted = 0 WHERE _id = ?'
+        sql = f'DELETE FROM {self.name} WHERE _id = ?'
         self._conn.execute(sql, (offset + 1,))
         self._conn.commit()
 
     def exist(self, doc_id: str):
-        sql = f'SELECT count(*) from {self.name} WHERE _deleted = 0 and _doc_id = ?;'
+        sql = f'SELECT count(*) from {self.name} WHERE _doc_id = ?;'
         return self._conn.execute(sql, (doc_id,)).fetchone()[0] > 0
 
     def count(self, where_clause: str = '', where_params: Tuple = ()):
@@ -358,27 +355,22 @@ def count(self, where_clause: str = '', where_params: Tuple = ()):
 
         if where_clause:
             sql = 'SELECT count(_id) from {table} WHERE {where} LIMIT 1;'
-            where = f'_deleted = ? and {where_clause}'
+            where = where_clause
             sql = sql.format(table=self.name, where=where)
 
-            params = (0,) + tuple([_converting(p) for p in where_params])
+            params = tuple([_converting(p) for p in where_params])
 
             # # EXPLAIN SQL query
             # for row in self._conn.execute('EXPLAIN QUERY PLAN ' + sql, params):
             #     print(row)
             return self._conn.execute(sql, params).fetchone()[0]
         else:
-            sql = f'SELECT MAX(_id) from {self.name} LIMIT 1;'
+            sql = f'SELECT count(_id) from {self.name};'
             result = self._conn.execute(sql).fetchone()
             if result[0]:
-                return result[0] - self.deleted_count()
+                return result[0]
             return 0
 
-    def deleted_count(self):
-        """Return the total number of record what is marked as soft-deleted."""
-        sql = f'SELECT count(_id) from {self.name} WHERE _deleted = 1 LIMIT 1'
-        return self._conn.execute(sql).fetchone()[0]
-
     @property
     def size(self):
         return self.count()
@@ -404,24 +396,20 @@ def create_table(self):
                         (_doc_id TEXT NOT NULL PRIMARY KEY,
                          cell_id INTEGER NOT NULL,
                          offset INTEGER NOT NULL,
-                         time_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-                         _deleted NUMERIC DEFAULT 0)'''
+                         time_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)'''
 
         self._conn.execute(sql)
 
         self._conn.execute(
             f'CREATE INDEX if not exists idx_time_at_ ON {self.name}(time_at)'
         )
-        self._conn.execute(
-            f'CREATE INDEX if not exists idx__delete_ ON {self.name}(_deleted)'
-        )
 
         self._conn.commit()
 
     def iter_addresses(
         self, time_since: 'datetime.datetime' = datetime.datetime(2020, 2, 2, 0, 0)
     ):
-        sql = f'SELECT _doc_id, cell_id, offset from {self.name} WHERE time_at >= ? AND _deleted = 0 ORDER BY time_at ASC;'
+        sql = f'SELECT _doc_id, cell_id, offset from {self.name} WHERE time_at >= ? ORDER BY time_at ASC;'
 
         cursor = self._conn.cursor()
         for doc_id, cell_id, offset in cursor.execute(sql, (time_since,)):
@@ -435,29 +423,23 @@ def get_latest_commit(self):
         return row
 
     def get_address(self, doc_id: str):
-        sql = f'SELECT cell_id, offset from {self.name} WHERE _doc_id = ? AND _deleted = 0 LIMIT 1;'
+        sql = f'SELECT cell_id, offset from {self.name} WHERE _doc_id = ? LIMIT 1;'
         cursor = self._conn.execute(sql, (doc_id,))
         row = cursor.fetchone()
         return (row[0], row[1]) if row else (None, None)
 
     def delete_address(self, doc_id: str, commit: bool = True):
-        sql = f'UPDATE {self.name} SET _deleted = 1, time_at = ? WHERE _doc_id = ?'
-        self._conn.execute(
-            sql,
-            (
-                time_now(),
-                doc_id,
-            ),
-        )
+        sql = f'DELETE from {self.name} WHERE _doc_id = ?'
+        self._conn.execute(sql, (doc_id,))
         print(f'Deleted {doc_id} at: {time_now()}')
         if commit:
             self._conn.commit()
 
     def add_address(self, doc_id: str, cell_id: int, offset: int, commit: bool = True):
-        sql = f'INSERT OR REPLACE INTO {self.name}(_doc_id, cell_id, offset, time_at, _deleted) VALUES (?, ?, ?, ?, ?);'
+        sql = f'INSERT OR REPLACE INTO {self.name}(_doc_id, cell_id, offset, time_at) VALUES (?, ?, ?, ?);'
         self._conn.execute(
             sql,
-            (doc_id, cell_id, offset, time_now(), 0),
+            (doc_id, cell_id, offset, time_now()),
         )
         if commit:
             self._conn.commit()
@@ -469,11 +451,11 @@ def bulk_add_address(
         offsets: Union[List[int], np.ndarray],
         commit: bool = True,
     ):
-        sql = f'INSERT OR REPLACE INTO {self.name}(_doc_id, cell_id, offset, time_at, _deleted) VALUES (?, ?, ?, ?, ?);'
+        sql = f'INSERT OR REPLACE INTO {self.name}(_doc_id, cell_id, offset, time_at) VALUES (?, ?, ?, ?);'
         self._conn.executemany(
             sql,
             [
-                (doc_id, cell_id, offset, time_now(), 0)
+                (doc_id, cell_id, offset, time_now())
                 for doc_id, cell_id, offset in zip(doc_ids, cell_ids, offsets)
             ],
         )

tests/executor/test_executor.py

@@ -403,3 +403,30 @@ def test_local_storage_with_delete(tmpfile):
 
     assert int(status.tags['total_docs']) == N - 1
     assert int(status.tags['index_size']) == N - 1
+
+
+def test_local_storage_delete_update(tmpfile):
+    docs = gen_docs(N)
+    f = Flow().add(
+        uses=AnnLiteIndexer,
+        uses_with={
+            'n_dim': D,
+        },
+        workspace=tmpfile,
+        shards=1,
+    )
+    with f:
+        f.post(on='/index', inputs=docs)
+        time.sleep(2)
+        f.post(on='/backup')
+
+        f.post(on='/delete', parameters={'ids': ['0']})
+        time.sleep(2)
+        f.post(on='/backup')
+
+        new_doc = gen_docs(N)[1:]
+        f.post(on='/update', inputs=new_doc)
+        status = f.post(on='/status', return_results=True)[0]
+
+        assert int(status.tags['total_docs']) == N - 1
+        assert int(status.tags['index_size']) == N - 1

tests/test_table.py

@@ -44,7 +44,7 @@ def test_create_cell_table():
 
 def test_schema(dummy_cell_table):
     schema = dummy_cell_table.schema
-    assert len(schema.split('\n')) == 6
+    assert len(schema.split('\n')) == 5
 
 
 def test_query(table_with_data):
@@ -134,6 +134,3 @@ def test_create_meta_table(tmpdir):
     table.delete_address('0')
     addresses = list(table.iter_addresses(time_since=time_since))
     assert addresses == []
-
-    addr = table.get_latest_commit()
-    assert addr[-1] >= time_since