Merge pull request #49 from tidesdb/0-9-6

addition of db stats, purge, and sync wal methods with tests; bumped …

Author: guycipher

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "tidesdb"
-version = "0.9.5"
+version = "0.9.6"
 description = "Official Python bindings for TidesDB - A high-performance embedded key-value storage engine"
 readme = "README.md"
 requires-python = ">=3.10"

src/tidesdb/__init__.py

@@ -16,6 +16,7 @@
     ColumnFamilyConfig,
     Stats,
     CacheStats,
+    DbStats,
     CompressionAlgorithm,
     SyncMode,
     LogLevel,
@@ -30,7 +31,7 @@
     COMMIT_HOOK_FUNC,
 )
 
-__version__ = "0.9.5"
+__version__ = "0.9.6"
 __all__ = [
     "TidesDB",
     "Transaction",
@@ -40,6 +41,7 @@
     "ColumnFamilyConfig",
     "Stats",
     "CacheStats",
+    "DbStats",
     "CompressionAlgorithm",
     "SyncMode",
     "LogLevel",

src/tidesdb/tidesdb.py

@@ -275,6 +275,28 @@ class _CCacheStats(Structure):
     ]
 
 
+class _CDbStats(Structure):
+    """C structure for tidesdb_db_stats_t."""
+
+    _fields_ = [
+        ("num_column_families", c_int),
+        ("total_memory", c_uint64),
+        ("available_memory", c_uint64),
+        ("resolved_memory_limit", c_size_t),
+        ("memory_pressure_level", c_int),
+        ("flush_pending_count", c_int),
+        ("total_memtable_bytes", c_int64),
+        ("total_immutable_count", c_int),
+        ("total_sstable_count", c_int),
+        ("total_data_size_bytes", c_uint64),
+        ("num_open_sstables", c_int),
+        ("global_seq", c_uint64),
+        ("txn_memory_bytes", c_int64),
+        ("compaction_queue_size", c_size_t),
+        ("flush_queue_size", c_size_t),
+    ]
+
+
 _lib.tidesdb_default_column_family_config.argtypes = []
 _lib.tidesdb_default_column_family_config.restype = _CColumnFamilyConfig
 
@@ -451,6 +473,18 @@ class _CCacheStats(Structure):
 _lib.tidesdb_delete_column_family.argtypes = [c_void_p, c_void_p]
 _lib.tidesdb_delete_column_family.restype = c_int
 
+_lib.tidesdb_purge_cf.argtypes = [c_void_p]
+_lib.tidesdb_purge_cf.restype = c_int
+
+_lib.tidesdb_purge.argtypes = [c_void_p]
+_lib.tidesdb_purge.restype = c_int
+
+_lib.tidesdb_sync_wal.argtypes = [c_void_p]
+_lib.tidesdb_sync_wal.restype = c_int
+
+_lib.tidesdb_get_db_stats.argtypes = [c_void_p, POINTER(_CDbStats)]
+_lib.tidesdb_get_db_stats.restype = c_int
+
 
 @dataclass
 class Config:
@@ -565,6 +599,27 @@ class CacheStats:
     num_partitions: int
 
 
+@dataclass
+class DbStats:
+    """Database-level aggregate statistics."""
+
+    num_column_families: int
+    total_memory: int
+    available_memory: int
+    resolved_memory_limit: int
+    memory_pressure_level: int
+    flush_pending_count: int
+    total_memtable_bytes: int
+    total_immutable_count: int
+    total_sstable_count: int
+    total_data_size_bytes: int
+    num_open_sstables: int
+    global_seq: int
+    txn_memory_bytes: int
+    compaction_queue_size: int
+    flush_queue_size: int
+
+
 @dataclass
 class CommitOp:
     """A single operation from a committed transaction batch."""
@@ -816,6 +871,48 @@ def is_compacting(self) -> bool:
         """Check if a compaction operation is in progress for this column family."""
         return bool(_lib.tidesdb_is_compacting(self._cf))
 
+    def purge(self) -> None:
+        """
+        Force a synchronous flush and aggressive compaction for this column family.
+
+        Unlike flush_memtable() and compact() (which are non-blocking), purge blocks
+        until all flush and compaction I/O is complete.
+
+        Behavior:
+        1. Waits for any in-progress flush to complete
+        2. Force-flushes the active memtable (even if below threshold)
+        3. Waits for flush I/O to fully complete
+        4. Waits for any in-progress compaction to complete
+        5. Triggers synchronous compaction inline (bypasses the compaction queue)
+        6. Waits for any queued compaction to drain
+
+        Use cases:
+        - Before backup or checkpoint: ensure all data is on disk and compacted
+        - After bulk deletes: reclaim space immediately by compacting away tombstones
+        - Manual maintenance: force a clean state during a maintenance window
+        - Pre-shutdown: ensure all pending work is complete before closing
+        """
+        result = _lib.tidesdb_purge_cf(self._cf)
+        if result != TDB_SUCCESS:
+            raise TidesDBError.from_code(result, "failed to purge column family")
+
+    def sync_wal(self) -> None:
+        """
+        Force an immediate fsync of the active write-ahead log for this column family.
+
+        This is useful for explicit durability control when using SYNC_NONE or
+        SYNC_INTERVAL modes.
+
+        Use cases:
+        - Application-controlled durability: sync the WAL at specific points
+        - Pre-checkpoint: ensure all buffered WAL data is on disk
+        - Graceful shutdown: flush WAL buffers before closing
+        - Critical writes: force durability for specific high-value writes
+        """
+        result = _lib.tidesdb_sync_wal(self._cf)
+        if result != TDB_SUCCESS:
+            raise TidesDBError.from_code(result, "failed to sync WAL")
+
     def update_runtime_config(self, config: ColumnFamilyConfig, persist_to_disk: bool = True) -> None:
         """
         Update runtime-safe configuration settings for this column family.
@@ -1656,6 +1753,62 @@ def delete_column_family(self, cf: ColumnFamily) -> None:
         if result != TDB_SUCCESS:
             raise TidesDBError.from_code(result, "failed to delete column family")
 
+    def purge(self) -> None:
+        """
+        Force a synchronous flush and aggressive compaction for all column families,
+        then drain both the global flush and compaction queues.
+
+        Unlike flush_memtable() and compact() (which are non-blocking), purge blocks
+        until all work is complete across all column families.
+
+        Behavior:
+        1. Calls purge on each column family
+        2. Drains the global flush queue
+        3. Drains the global compaction queue
+
+        Raises:
+            TidesDBError: If purge fails for any column family
+        """
+        if self._closed:
+            raise TidesDBError("Database is closed")
+
+        result = _lib.tidesdb_purge(self._db)
+        if result != TDB_SUCCESS:
+            raise TidesDBError.from_code(result, "failed to purge database")
+
+    def get_db_stats(self) -> DbStats:
+        """
+        Get aggregate statistics across the entire database instance.
+
+        Returns:
+            DbStats instance with database-level statistics
+        """
+        if self._closed:
+            raise TidesDBError("Database is closed")
+
+        c_stats = _CDbStats()
+        result = _lib.tidesdb_get_db_stats(self._db, ctypes.byref(c_stats))
+        if result != TDB_SUCCESS:
+            raise TidesDBError.from_code(result, "failed to get database stats")
+
+        return DbStats(
+            num_column_families=c_stats.num_column_families,
+            total_memory=c_stats.total_memory,
+            available_memory=c_stats.available_memory,
+            resolved_memory_limit=c_stats.resolved_memory_limit,
+            memory_pressure_level=c_stats.memory_pressure_level,
+            flush_pending_count=c_stats.flush_pending_count,
+            total_memtable_bytes=c_stats.total_memtable_bytes,
+            total_immutable_count=c_stats.total_immutable_count,
+            total_sstable_count=c_stats.total_sstable_count,
+            total_data_size_bytes=c_stats.total_data_size_bytes,
+            num_open_sstables=c_stats.num_open_sstables,
+            global_seq=c_stats.global_seq,
+            txn_memory_bytes=c_stats.txn_memory_bytes,
+            compaction_queue_size=c_stats.compaction_queue_size,
+            flush_queue_size=c_stats.flush_queue_size,
+        )
+
     def __enter__(self) -> TidesDB:
         return self
 

tests/test_tidesdb.py

@@ -967,5 +967,177 @@ def test_config_with_max_memory_usage(self, temp_db_path):
         db.close()
 
 
+class TestPurgeCf:
+    """Tests for column family purge operations."""
+
+    def test_purge_cf_basic(self, db, cf):
+        """Test purging a column family with data."""
+        with db.begin_txn() as txn:
+            for i in range(50):
+                txn.put(cf, f"purge_key_{i}".encode(), f"purge_val_{i}".encode())
+            txn.commit()
+
+        cf.purge()
+
+        # Data should still be accessible after purge
+        with db.begin_txn() as txn:
+            for i in range(50):
+                value = txn.get(cf, f"purge_key_{i}".encode())
+                assert value == f"purge_val_{i}".encode()
+
+    def test_purge_cf_empty(self, db, cf):
+        """Test purging an empty column family succeeds."""
+        cf.purge()
+
+    def test_purge_cf_after_deletes(self, db, cf):
+        """Test purge after bulk deletes reclaims tombstones."""
+        with db.begin_txn() as txn:
+            for i in range(20):
+                txn.put(cf, f"del_key_{i}".encode(), f"del_val_{i}".encode())
+            txn.commit()
+
+        with db.begin_txn() as txn:
+            for i in range(20):
+                txn.delete(cf, f"del_key_{i}".encode())
+            txn.commit()
+
+        cf.purge()
+
+        # Deleted keys should not be found
+        with db.begin_txn() as txn:
+            for i in range(20):
+                with pytest.raises(tidesdb.TidesDBError):
+                    txn.get(cf, f"del_key_{i}".encode())
+
+
+class TestPurgeDb:
+    """Tests for database-level purge operations."""
+
+    def test_purge_db_basic(self, db):
+        """Test purging the entire database."""
+        db.create_column_family("purge_cf1")
+        db.create_column_family("purge_cf2")
+        cf1 = db.get_column_family("purge_cf1")
+        cf2 = db.get_column_family("purge_cf2")
+
+        with db.begin_txn() as txn:
+            txn.put(cf1, b"k1", b"v1")
+            txn.put(cf2, b"k2", b"v2")
+            txn.commit()
+
+        db.purge()
+
+        # Data should still be accessible
+        with db.begin_txn() as txn:
+            assert txn.get(cf1, b"k1") == b"v1"
+            assert txn.get(cf2, b"k2") == b"v2"
+
+        db.drop_column_family("purge_cf1")
+        db.drop_column_family("purge_cf2")
+
+    def test_purge_db_empty(self, db):
+        """Test purging a database with no column families succeeds."""
+        db.purge()
+
+
+class TestSyncWal:
+    """Tests for WAL sync operations."""
+
+    def test_sync_wal_basic(self, db, cf):
+        """Test syncing WAL after writes."""
+        with db.begin_txn() as txn:
+            txn.put(cf, b"sync_key", b"sync_val")
+            txn.commit()
+
+        cf.sync_wal()
+
+        # Data should be accessible after sync
+        with db.begin_txn() as txn:
+            assert txn.get(cf, b"sync_key") == b"sync_val"
+
+    def test_sync_wal_empty_cf(self, db, cf):
+        """Test syncing WAL on an empty column family."""
+        cf.sync_wal()
+
+    def test_sync_wal_after_batch(self, db, cf):
+        """Test syncing WAL after a batch of writes."""
+        with db.begin_txn() as txn:
+            for i in range(100):
+                txn.put(cf, f"batch_{i}".encode(), f"val_{i}".encode())
+            txn.commit()
+
+        cf.sync_wal()
+
+        with db.begin_txn() as txn:
+            for i in range(100):
+                value = txn.get(cf, f"batch_{i}".encode())
+                assert value == f"val_{i}".encode()
+
+
+class TestDbStats:
+    """Tests for database-level statistics."""
+
+    def test_db_stats_basic(self, db):
+        """Test getting database stats."""
+        stats = db.get_db_stats()
+        assert isinstance(stats, tidesdb.DbStats)
+        assert stats.num_column_families >= 0
+        assert stats.total_memory > 0
+        assert stats.resolved_memory_limit > 0
+        assert stats.memory_pressure_level >= 0
+        assert stats.global_seq >= 0
+
+    def test_db_stats_with_cf(self, db, cf):
+        """Test that db stats reflect column family count."""
+        stats = db.get_db_stats()
+        assert stats.num_column_families >= 1
+
+    def test_db_stats_after_writes(self, db, cf):
+        """Test db stats after writing data."""
+        with db.begin_txn() as txn:
+            for i in range(50):
+                txn.put(cf, f"stats_key_{i}".encode(), f"stats_val_{i}".encode())
+            txn.commit()
+
+        stats = db.get_db_stats()
+        assert stats.num_column_families >= 1
+        assert stats.total_memtable_bytes >= 0
+        assert stats.total_sstable_count >= 0
+        assert stats.total_data_size_bytes >= 0
+
+    def test_db_stats_all_fields_present(self, db):
+        """Test that all DbStats fields are populated."""
+        stats = db.get_db_stats()
+        assert isinstance(stats.num_column_families, int)
+        assert isinstance(stats.total_memory, int)
+        assert isinstance(stats.available_memory, int)
+        assert isinstance(stats.resolved_memory_limit, int)
+        assert isinstance(stats.memory_pressure_level, int)
+        assert isinstance(stats.flush_pending_count, int)
+        assert isinstance(stats.total_memtable_bytes, int)
+        assert isinstance(stats.total_immutable_count, int)
+        assert isinstance(stats.total_sstable_count, int)
+        assert isinstance(stats.total_data_size_bytes, int)
+        assert isinstance(stats.num_open_sstables, int)
+        assert isinstance(stats.global_seq, int)
+        assert isinstance(stats.txn_memory_bytes, int)
+        assert isinstance(stats.compaction_queue_size, int)
+        assert isinstance(stats.flush_queue_size, int)
+
+    def test_db_stats_after_purge(self, db, cf):
+        """Test db stats after purging."""
+        with db.begin_txn() as txn:
+            for i in range(20):
+                txn.put(cf, f"purge_stats_{i}".encode(), f"val_{i}".encode())
+            txn.commit()
+
+        db.purge()
+
+        stats = db.get_db_stats()
+        # After purge, flush and compaction queues should be drained
+        assert stats.flush_queue_size == 0
+        assert stats.compaction_queue_size == 0
+
+
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])