⚡️ Speed up method CodeFlashBenchmarkPlugin.write_benchmark_timings by 167% in PR #59 (codeflash-trace-decorator)

[codeflash-ai]

⚡️ This pull request contains optimizations for PR #59

If you approve this dependent PR, these changes will be merged into the original PR branch codeflash-trace-decorator.

This PR will be automatically closed if the original PR is merged.

---

📄 167% (1.67x) speedup for CodeFlashBenchmarkPlugin.write_benchmark_timings in codeflash/benchmarking/plugin/plugin.py

⏱️ Runtime : 15.2 milliseconds → 5.71 milliseconds (best of 126 runs)

📝 Explanation and details

Here is the optimized version of the given Python program.

Changes made.

1. Added PRAGMA statements (PRAGMA synchronous = OFF and PRAGMA journal_mode = MEMORY) after establishing the database connection. These settings help in improving the write performance by reducing the disk I/O operations.

However, note that setting PRAGMA synchronous = OFF and PRAGMA journal_mode = MEMORY makes the database more vulnerable to corruption in the event of an application crash or power failure. Use these optimizations only when performance is critical, and the database is not mission-critical.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 38 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage

🌀 Generated Regression Tests Details

from __future__ import annotations

import os
import sqlite3

# imports
import pytest  # used for our unit tests
from codeflash.benchmarking.plugin.plugin import CodeFlashBenchmarkPlugin

# unit tests

# Helper function to create a test database
def create_test_db(path):
    connection = sqlite3.connect(path)
    cursor = connection.cursor()
    cursor.execute("""
        CREATE TABLE benchmark_timings (
            benchmark_file_path TEXT,
            benchmark_function_name TEXT,
            benchmark_line_number INTEGER,
            benchmark_time_ns INTEGER
        )
    """)
    connection.commit()
    connection.close()

# Test cases

def test_single_entry(tmp_path):
    # Setup
    db_path = tmp_path / "test.db"
    create_test_db(db_path)
    plugin = CodeFlashBenchmarkPlugin()
    plugin._trace_path = str(db_path)
    plugin.benchmark_timings = [("path/to/file.py", "test_function", 42, 123456)]

    # Execute
    plugin.write_benchmark_timings()

    # Verify
    connection = sqlite3.connect(db_path)
    cursor = connection.cursor()
    cursor.execute("SELECT * FROM benchmark_timings")
    rows = cursor.fetchall()
    connection.close()

def test_multiple_entries(tmp_path):
    # Setup
    db_path = tmp_path / "test.db"
    create_test_db(db_path)
    plugin = CodeFlashBenchmarkPlugin()
    plugin._trace_path = str(db_path)
    plugin.benchmark_timings = [
        ("path/to/file1.py", "test_function1", 42, 123456),
        ("path/to/file2.py", "test_function2", 43, 654321)
    ]

    # Execute
    plugin.write_benchmark_timings()

    # Verify
    connection = sqlite3.connect(db_path)
    cursor = connection.cursor()
    cursor.execute("SELECT * FROM benchmark_timings")
    rows = cursor.fetchall()
    connection.close()

def test_empty_benchmark_timings(tmp_path):
    # Setup
    db_path = tmp_path / "test.db"
    create_test_db(db_path)
    plugin = CodeFlashBenchmarkPlugin()
    plugin._trace_path = str(db_path)
    plugin.benchmark_timings = []

    # Execute
    plugin.write_benchmark_timings()

    # Verify
    connection = sqlite3.connect(db_path)
    cursor = connection.cursor()
    cursor.execute("SELECT * FROM benchmark_timings")
    rows = cursor.fetchall()
    connection.close()

def test_uninitialized_trace_path():
    # Setup
    plugin = CodeFlashBenchmarkPlugin()
    plugin._trace_path = None
    plugin.benchmark_timings = [("path/to/file.py", "test_function", 42, 123456)]

    # Execute and Verify
    with pytest.raises(Exception):
        plugin.write_benchmark_timings()

def test_missing_table(tmp_path):
    # Setup
    db_path = tmp_path / "test.db"
    connection = sqlite3.connect(db_path)
    connection.close()  # Create an empty database without the required table
    plugin = CodeFlashBenchmarkPlugin()
    plugin._trace_path = str(db_path)
    plugin.benchmark_timings = [("path/to/file.py", "test_function", 42, 123456)]

    # Execute and Verify
    with pytest.raises(Exception):
        plugin.write_benchmark_timings()

def test_large_number_of_entries(tmp_path):
    # Setup
    db_path = tmp_path / "test.db"
    create_test_db(db_path)
    plugin = CodeFlashBenchmarkPlugin()
    plugin._trace_path = str(db_path)
    plugin.benchmark_timings = [("path/to/file.py", "test_function", 42, 123456)] * 1000

    # Execute
    plugin.write_benchmark_timings()

    # Verify
    connection = sqlite3.connect(db_path)
    cursor = connection.cursor()
    cursor.execute("SELECT COUNT(*) FROM benchmark_timings")
    count = cursor.fetchone()[0]
    connection.close()

def test_data_integrity(tmp_path):
    # Setup
    db_path = tmp_path / "test.db"
    create_test_db(db_path)
    plugin = CodeFlashBenchmarkPlugin()
    plugin._trace_path = str(db_path)
    plugin.benchmark_timings = [("path/to/file.py", "test_function", 42, 123456)]

    # Execute
    plugin.write_benchmark_timings()

    # Verify
    connection = sqlite3.connect(db_path)
    cursor = connection.cursor()
    cursor.execute("SELECT * FROM benchmark_timings")
    row = cursor.fetchone()
    connection.close()

def test_clear_benchmark_timings_list(tmp_path):
    # Setup
    db_path = tmp_path / "test.db"
    create_test_db(db_path)
    plugin = CodeFlashBenchmarkPlugin()
    plugin._trace_path = str(db_path)
    plugin.benchmark_timings = [("path/to/file.py", "test_function", 42, 123456)]

    # Execute
    plugin.write_benchmark_timings()

def test_persistent_state(tmp_path):
    # Setup
    db_path = tmp_path / "test.db"
    create_test_db(db_path)
    plugin = CodeFlashBenchmarkPlugin()
    plugin._trace_path = str(db_path)
    plugin.benchmark_timings = [("path/to/file.py", "test_function", 42, 123456)]
    plugin.write_benchmark_timings()
    plugin.benchmark_timings = [("path/to/file2.py", "test_function2", 43, 654321)]

    # Execute
    plugin.write_benchmark_timings()

    # Verify
    connection = sqlite3.connect(db_path)
    cursor = connection.cursor()
    cursor.execute("SELECT * FROM benchmark_timings")
    rows = cursor.fetchall()
    connection.close()
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

from __future__ import annotations

import os
import sqlite3

# imports
import pytest  # used for our unit tests
from codeflash.benchmarking.plugin.plugin import CodeFlashBenchmarkPlugin

# unit tests

@pytest.fixture
def setup_database(tmp_path):
    """Fixture to set up an in-memory SQLite database and return the plugin instance."""
    db_file = tmp_path / "test_benchmark_timings.db"
    plugin = CodeFlashBenchmarkPlugin()
    plugin._trace_path = str(db_file)
    connection = sqlite3.connect(plugin._trace_path)
    cursor = connection.cursor()
    cursor.execute("""
    CREATE TABLE benchmark_timings (
        benchmark_file_path TEXT,
        benchmark_function_name TEXT,
        benchmark_line_number INTEGER,
        benchmark_time_ns INTEGER
    )
    """)
    connection.commit()
    connection.close()
    return plugin

def test_single_entry_insertion(setup_database):
    """Test inserting a single benchmark timing entry."""
    plugin = setup_database
    plugin.benchmark_timings = [('file1.py', 'func1', 10, 1000)]
    plugin.write_benchmark_timings()

    connection = sqlite3.connect(plugin._trace_path)
    cursor = connection.cursor()
    cursor.execute("SELECT * FROM benchmark_timings")
    results = cursor.fetchall()
    connection.close()

def test_multiple_entries_insertion(setup_database):
    """Test inserting multiple benchmark timing entries."""
    plugin = setup_database
    plugin.benchmark_timings = [
        ('file1.py', 'func1', 10, 1000),
        ('file2.py', 'func2', 20, 2000)
    ]
    plugin.write_benchmark_timings()

    connection = sqlite3.connect(plugin._trace_path)
    cursor = connection.cursor()
    cursor.execute("SELECT * FROM benchmark_timings")
    results = cursor.fetchall()
    connection.close()

def test_empty_benchmark_timings(setup_database):
    """Test calling write_benchmark_timings with an empty list."""
    plugin = setup_database
    plugin.benchmark_timings = []
    plugin.write_benchmark_timings()

    connection = sqlite3.connect(plugin._trace_path)
    cursor = connection.cursor()
    cursor.execute("SELECT * FROM benchmark_timings")
    results = cursor.fetchall()
    connection.close()

def test_null_database_path():
    """Test calling write_benchmark_timings with a null database path."""
    plugin = CodeFlashBenchmarkPlugin()
    plugin.benchmark_timings = [('file1.py', 'func1', 10, 1000)]
    with pytest.raises(sqlite3.OperationalError):
        plugin.write_benchmark_timings()

def test_reusing_existing_connection(setup_database):
    """Test reusing an existing database connection."""
    plugin = setup_database
    plugin.benchmark_timings = [('file1.py', 'func1', 10, 1000)]
    plugin.write_benchmark_timings()

    # Reuse the existing connection
    plugin.benchmark_timings = [('file2.py', 'func2', 20, 2000)]
    plugin.write_benchmark_timings()

    connection = sqlite3.connect(plugin._trace_path)
    cursor = connection.cursor()
    cursor.execute("SELECT * FROM benchmark_timings")
    results = cursor.fetchall()
    connection.close()

def test_sql_syntax_error(setup_database):
    """Test handling of SQL syntax error."""
    plugin = setup_database
    plugin.benchmark_timings = [('file1.py', 'func1', 10, 1000)]
    plugin._connection = sqlite3.connect(plugin._trace_path)
    cursor = plugin._connection.cursor()
    cursor.execute("DROP TABLE benchmark_timings")
    cursor.execute("CREATE TABLE benchmark_timings (invalid_column INTEGER)")
    plugin._connection.commit()

    with pytest.raises(sqlite3.OperationalError):
        plugin.write_benchmark_timings()

def test_high_volume_data_insertion(setup_database):
    """Test inserting a large number of benchmark timing entries."""
    plugin = setup_database
    plugin.benchmark_timings = [('file1.py', 'func1', 10, 1000)] * 1000
    plugin.write_benchmark_timings()

    connection = sqlite3.connect(plugin._trace_path)
    cursor = connection.cursor()
    cursor.execute("SELECT COUNT(*) FROM benchmark_timings")
    result = cursor.fetchone()
    connection.close()

def test_data_consistency(setup_database):
    """Test that data written to the database matches the benchmark_timings list."""
    plugin = setup_database
    plugin.benchmark_timings = [
        ('file1.py', 'func1', 10, 1000),
        ('file2.py', 'func2', 20, 2000)
    ]
    plugin.write_benchmark_timings()

    connection = sqlite3.connect(plugin._trace_path)
    cursor = connection.cursor()
    cursor.execute("SELECT * FROM benchmark_timings")
    results = cursor.fetchall()
    connection.close()

def test_clearing_list_after_write(setup_database):
    """Test that the benchmark_timings list is cleared after a successful write."""
    plugin = setup_database
    plugin.benchmark_timings = [('file1.py', 'func1', 10, 1000)]
    plugin.write_benchmark_timings()
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

To edit these changes git checkout codeflash/optimize-pr59-2025-04-01T00.15.17 and push.