Simulate PostgreSQL Vacuum to Reclaim Space in Python

import shutil
import os

class VacuumCleaner:
    """Simulates PostgreSQL-style vacuum reclaiming dead space in a file."""
    
    def __init__(self, filepath, fill_ratio=0.7, dead_marker="[DELETED]"):
        self.filepath = filepath
        self.fill_ratio = fill_ratio
        self.dead_marker = dead_marker
    
    def create_mock_data(self):
        """Create a mock data file with some dead rows marked as deleted."""
        records = [
            "Alice,32",
            f"{self.dead_marker} Bob,45",
            "Charlie,28",
            f"{self.dead_marker} Diana,51",
            "Eve,39"
        ]
        with open(self.filepath, "w") as f:
            for record in records:
                f.write(record + "\n")
    
    def analyze(self):
        """Report live vs dead space before vacuum."""
        dead_bytes = 0
        live_bytes = 0
        with open(self.filepath, "r") as f:
            for line in f:
                if self.dead_marker in line:
                    dead_bytes += len(line.encode("utf-8"))
                else:
                    live_bytes += len(line.encode("utf-8"))
        return live_bytes, dead_bytes
    
    def vacuum(self):
        """Rewrite file, keeping only live rows, then shrink to actual size."""
        tmp_path = self.filepath + ".vacuum_tmp"
        with open(self.filepath, "r") as src, open(tmp_path, "w") as dst:
            for line in src:
                if self.dead_marker not in line:
                    dst.write(line)
        # Atomically replace original, then reclaim disk space
        os.replace(tmp_path, self.filepath)
        
        # Simulate filesystem reclaiming unused blocks (not needed in reality, just a mock)
        with open(self.filepath, "ab") as f:
            f.truncate()


if __name__ == "__main__":
    test_file = "mock_data.txt"
    cleaner = VacuumCleaner(test_file)
    cleaner.create_mock_data()
    
    live_before, dead_before = cleaner.analyze()
    size_before = os.path.getsize(test_file)
    
    cleaner.vacuum()
    
    live_after, dead_after = cleaner.analyze()
    size_after = os.path.getsize(test_file)
    
    print(f"Before vacuum: {live_before} live bytes, {dead_before} dead bytes, {size_before} total bytes")
    print(f"After vacuum:  {live_after} live bytes, {dead_after} dead bytes, {size_after} total bytes")
    print(f"Reclaimed: {dead_before} bytes of dead space")
    
    # Clean up
    os.remove(test_file)