test_faiss_minimal.py

#!/usr/bin/env python3
"""
Minimal FAISS test to isolate segfault issue.
Tests each step individually to find the exact failure point.
"""

import sys
import numpy as np

print("=" * 60)
print("MINIMAL FAISS TEST - Step by step debugging")
print("=" * 60)
print(f"Python version: {sys.version}")
print()

# Test 1: Import numpy
print("Test 1: Import numpy...")
try:
    import numpy as np
    print(f"  ✓ numpy imported successfully (version {np.__version__})")
except Exception as e:
    print(f"  ❌ Failed: {e}")
    sys.exit(1)

# Test 2: Import faiss
print("\nTest 2: Import faiss...")
try:
    import faiss
    print(f"  ✓ faiss imported successfully")
except Exception as e:
    print(f"  ❌ Failed: {e}")
    sys.exit(1)

# Test 3: Create simple numpy array
print("\nTest 3: Create numpy array...")
try:
    test_data = np.random.rand(10, 128).astype('float32')
    print(f"  ✓ Created array with shape {test_data.shape}")
except Exception as e:
    print(f"  ❌ Failed: {e}")
    sys.exit(1)

# Test 4: Create FAISS index
print("\nTest 4: Create FAISS index...")
try:
    dimension = 128
    index = faiss.IndexFlatL2(dimension)
    print(f"  ✓ Created IndexFlatL2 with dimension {dimension}")
except Exception as e:
    print(f"  ❌ Failed: {e}")
    sys.exit(1)

# Test 5: Add vectors to index
print("\nTest 5: Add vectors to FAISS index...")
try:
    index.add(test_data)
    print(f"  ✓ Added {index.ntotal} vectors to index")
except Exception as e:
    print(f"  ❌ Failed: {e}")
    sys.exit(1)

# Test 6: Search index
print("\nTest 6: Search FAISS index...")
try:
    query = np.random.rand(1, 128).astype('float32')
    distances, indices = index.search(query, 5)
    print(f"  ✓ Search completed, found {len(indices[0])} results")
except Exception as e:
    print(f"  ❌ Failed: {e}")
    sys.exit(1)

# Test 7: Test with IndexFlatIP (what we actually use)
print("\nTest 7: Create IndexFlatIP...")
try:
    index_ip = faiss.IndexFlatIP(dimension)
    print(f"  ✓ Created IndexFlatIP")
except Exception as e:
    print(f"  ❌ Failed: {e}")
    sys.exit(1)

# Test 8: Normalize vectors (critical step)
print("\nTest 8: Normalize vectors with faiss.normalize_L2...")
try:
    test_data_copy = test_data.copy()
    faiss.normalize_L2(test_data_copy)
    print(f"  ✓ Normalized vectors")
except Exception as e:
    print(f"  ❌ Failed: {e}")
    import traceback
    traceback.print_exc()
    sys.exit(1)

# Test 9: Add normalized vectors to IndexFlatIP
print("\nTest 9: Add normalized vectors to IndexFlatIP...")
try:
    index_ip.add(test_data_copy)
    print(f"  ✓ Added {index_ip.ntotal} normalized vectors")
except Exception as e:
    print(f"  ❌ Failed: {e}")
    import traceback
    traceback.print_exc()
    sys.exit(1)

# Test 10: Write index to disk
print("\nTest 10: Write index to disk...")
try:
    faiss.write_index(index_ip, "test_index.faiss")
    print(f"  ✓ Index written to test_index.faiss")
except Exception as e:
    print(f"  ❌ Failed: {e}")
    import traceback
    traceback.print_exc()
    sys.exit(1)

# Test 11: Read index from disk
print("\nTest 11: Read index from disk...")
try:
    loaded_index = faiss.read_index("test_index.faiss")
    print(f"  ✓ Index loaded, contains {loaded_index.ntotal} vectors")
except Exception as e:
    print(f"  ❌ Failed: {e}")
    import traceback
    traceback.print_exc()
    sys.exit(1)

# Clean up
print("\nTest 12: Clean up test file...")
try:
    import os
    os.remove("test_index.faiss")
    print(f"  ✓ Test file removed")
except Exception as e:
    print(f"  ⚠️  Could not remove test file: {e}")

print("\n" + "=" * 60)
print("✅ ALL TESTS PASSED!")
print("=" * 60)
print("\nFAISS is working correctly on your system.")
print("The issue may be with:")
print("  - Specific data from the database")
print("  - Memory/size of actual embeddings")
print("  - Sentence transformers interaction")