python_benchmark.py

#!/usr/bin/env python3
"""
FILE SEARCH BENCHMARK — Python
Porównanie 5 metod wyszukiwania w plikach:
  1. Naive scan (os.walk + open)
  2. Concurrent ThreadPool scan
  3. mmap-based scan
  4. Inverted index (dict + set)
  5. Whoosh full-text index (BM25)
"""

import os
import re
import time
import mmap
import json
import math
import concurrent.futures
from pathlib import Path
from collections import defaultdict
from typing import List, Tuple, Dict

# ── Optional dependencies ──────────────────────────────────────────────
try:
    import whoosh.index as whoosh_index
    from whoosh.fields import Schema, TEXT, ID
    from whoosh.qparser import QueryParser
    from whoosh.query import And, Term
    WHOOSH_AVAILABLE = True
except ImportError:
    WHOOSH_AVAILABLE = False
    print("⚠️  whoosh not available — skipping method 5")

ROOT = Path("/home/claude/file_search_benchmark/testdata")
QUERIES = ["konie", "robot farma", "embedding wektor", "kernel linux"]

# ═══════════════════════════════════════════════════════════════════════
# Helper
# ═══════════════════════════════════════════════════════════════════════

def all_files(root: Path) -> List[Path]:
    return [p for p in root.rglob("*") if p.is_file()]

def fmt_ms(seconds: float) -> str:
    if seconds < 0.001:
        return f"{seconds*1_000_000:.1f} µs"
    elif seconds < 1:
        return f"{seconds*1000:.2f} ms"
    return f"{seconds:.3f} s"

# ═══════════════════════════════════════════════════════════════════════
# Method 1: Naive sequential scan
# ═══════════════════════════════════════════════════════════════════════

def naive_scan(query: str) -> Tuple[List[Path], float]:
    hits = []
    t0 = time.perf_counter()
    for path in ROOT.rglob("*"):
        if path.is_file():
            try:
                if query in path.read_text(errors="ignore"):
                    hits.append(path)
            except Exception:
                pass
    return hits, time.perf_counter() - t0

# ═══════════════════════════════════════════════════════════════════════
# Method 2: ThreadPool concurrent scan
# ═══════════════════════════════════════════════════════════════════════

def _check_file(args):
    path, query = args
    try:
        return path if query in path.read_text(errors="ignore") else None
    except Exception:
        return None

def thread_scan(files: List[Path], query: str) -> Tuple[List[Path], float]:
    t0 = time.perf_counter()
    args = [(p, query) for p in files]
    hits = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=os.cpu_count()) as ex:
        for result in ex.map(_check_file, args):
            if result:
                hits.append(result)
    return hits, time.perf_counter() - t0

# ═══════════════════════════════════════════════════════════════════════
# Method 3: mmap scan (byte-level, no heap copy)
# ═══════════════════════════════════════════════════════════════════════

def mmap_scan(files: List[Path], query: str) -> Tuple[List[Path], float]:
    query_bytes = query.encode()
    hits = []
    t0 = time.perf_counter()
    for path in files:
        try:
            with open(path, "rb") as f:
                size = os.path.getsize(path)
                if size == 0:
                    continue
                with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
                    if mm.find(query_bytes) != -1:
                        hits.append(path)
        except Exception:
            pass
    return hits, time.perf_counter() - t0

# ═══════════════════════════════════════════════════════════════════════
# Method 4: In-memory inverted index
# ═══════════════════════════════════════════════════════════════════════

class InvertedIndex:
    def __init__(self):
        self.index: Dict[str, set] = defaultdict(set)  # token -> {file_id}
        self.files: List[Path] = []

    def build(self, root: Path) -> float:
        t0 = time.perf_counter()
        pattern = re.compile(r'\W+')
        for path in root.rglob("*"):
            if not path.is_file():
                continue
            file_id = len(self.files)
            self.files.append(path)
            try:
                content = path.read_text(errors="ignore").lower()
                for token in pattern.split(content):
                    if len(token) >= 2:
                        self.index[token].add(file_id)
            except Exception:
                pass
        return time.perf_counter() - t0

    def search(self, query: str) -> Tuple[List[Path], float]:
        t0 = time.perf_counter()
        terms = query.lower().split()
        if not terms:
            return [], time.perf_counter() - t0

        result_ids = None
        for term in terms:
            postings = self.index.get(term, set())
            result_ids = postings if result_ids is None else result_ids & postings

        results = [self.files[i] for i in (result_ids or set())]
        return results, time.perf_counter() - t0

    def ram_estimate_kb(self) -> int:
        tokens_mem = sum(len(k) * 2 + 64 for k in self.index)
        postings_mem = sum(len(v) * 28 + 224 for v in self.index.values())
        files_mem = sum(len(str(p)) * 2 + 56 for p in self.files)
        return (tokens_mem + postings_mem + files_mem) // 1024

# ═══════════════════════════════════════════════════════════════════════
# Method 5: Whoosh full-text index (BM25)
# ═══════════════════════════════════════════════════════════════════════

WHOOSH_DIR = Path("/home/claude/file_search_benchmark/whoosh_index")

def build_whoosh_index(root: Path) -> float:
    if not WHOOSH_AVAILABLE:
        return 0.0
    WHOOSH_DIR.mkdir(exist_ok=True)
    schema = Schema(path=ID(stored=True), content=TEXT(stored=False))
    ix = whoosh_index.create_in(str(WHOOSH_DIR), schema)
    writer = ix.writer()
    t0 = time.perf_counter()
    for path in root.rglob("*"):
        if path.is_file():
            try:
                content = path.read_text(errors="ignore")
                writer.add_document(path=str(path), content=content)
            except Exception:
                pass
    writer.commit()
    return time.perf_counter() - t0

def whoosh_search(query: str) -> Tuple[List[str], float]:
    if not WHOOSH_AVAILABLE:
        return [], 0.0
    ix = whoosh_index.open_dir(str(WHOOSH_DIR))
    t0 = time.perf_counter()
    with ix.searcher() as searcher:
        q = QueryParser("content", ix.schema).parse(query)
        results = searcher.search(q, limit=None)
        hits = [r["path"] for r in results]
    return hits, time.perf_counter() - t0

# ═══════════════════════════════════════════════════════════════════════
# Main benchmark runner
# ═══════════════════════════════════════════════════════════════════════

def main():
    print("\n╔══════════════════════════════════════════════════════════════╗")
    print("║        FILE SEARCH BENCHMARK — Python                        ║")
    print("╚══════════════════════════════════════════════════════════════╝\n")

    # Collect file list once
    files = all_files(ROOT)
    print(f"📁 Total files: {len(files)}")

    # ── Build inverted index ──────────────────────────────────────────
    print("⚙️  Building Python inverted index...")
    idx = InvertedIndex()
    idx_build_time = idx.build(ROOT)
    print(f"   ✅ Built in {fmt_ms(idx_build_time)} | tokens: {len(idx.index)} | RAM ≈ {idx.ram_estimate_kb()} KB")

    # ── Build Whoosh index ────────────────────────────────────────────
    whoosh_build_time = 0.0
    if WHOOSH_AVAILABLE:
        if not WHOOSH_DIR.exists():
            print("⚙️  Building Whoosh BM25 index...")
            whoosh_build_time = build_whoosh_index(ROOT)
            print(f"   ✅ Built in {fmt_ms(whoosh_build_time)}")
        else:
            print("ℹ️  Whoosh index already exists, reusing")
    
    print()

    results_data = []

    for query in QUERIES:
        print("─────────────────────────────────────────────────────────────")
        print(f"🔎 Query: \"{query}\"")

        # 1. Naive
        r1, t1 = naive_scan(query)
        print(f"  [1] Naive scan:          {fmt_ms(t1):>10}  | hits: {len(r1)}")

        # 2. ThreadPool
        r2, t2 = thread_scan(files, query)
        print(f"  [2] ThreadPool scan:     {fmt_ms(t2):>10}  | hits: {len(r2)}")

        # 3. mmap (first term only for multi-word)
        first_term = query.split()[0]
        r3, t3 = mmap_scan(files, first_term)
        print(f"  [3] mmap scan:           {fmt_ms(t3):>10}  | hits: {len(r3)} (1st term)")

        # 4. Inverted index
        r4, t4 = idx.search(query)
        print(f"  [4] Inverted index:      {fmt_ms(t4):>10}  | hits: {len(r4)}")

        # 5. Whoosh
        if WHOOSH_AVAILABLE:
            r5, t5 = whoosh_search(query)
            print(f"  [5] Whoosh BM25:         {fmt_ms(t5):>10}  | hits: {len(r5)}")
        else:
            t5 = 0

        if t4 > 0:
            speedup_naive = t1 / t4
            speedup_thread = t2 / t4
            print(f"  📊 Index speedup: {speedup_naive:.0f}× vs naive | {speedup_thread:.0f}× vs threads")

        row = {
            "query": query,
            "naive_ms": round(t1 * 1000, 3),
            "thread_ms": round(t2 * 1000, 3),
            "mmap_ms": round(t3 * 1000, 3),
            "index_ms": round(t4 * 1000, 6),
            "whoosh_ms": round(t5 * 1000, 3) if WHOOSH_AVAILABLE else None,
            "hits_naive": len(r1),
            "hits_index": len(r4),
        }
        results_data.append(row)

    print("\n═══════════════════════════════════════════════════════════════")
    print(f"📋 Index: {len(idx.index)} tokens | {len(idx.files)} files | {idx.ram_estimate_kb()} KB RAM")
    print(f"   Build time: {fmt_ms(idx_build_time)}")

    output = {
        "lang": "python",
        "total_files": len(files),
        "index_build_ms": round(idx_build_time * 1000, 3),
        "index_ram_kb": idx.ram_estimate_kb(),
        "whoosh_build_ms": round(whoosh_build_time * 1000, 3),
        "results": results_data,
    }
    output_path = Path("/home/claude/file_search_benchmark/python_results.json")
    output_path.write_text(json.dumps(output, indent=2, ensure_ascii=False))
    print(f"✅ Results saved to python_results.json")
    return output

if __name__ == "__main__":
    main()