apexbase 1.25.0

"""
Tests for HTAP Gap Analysis v12 new features:
1. Transaction timeout protection (TxnManager 30s timeout + idle cleanup)
2. REINDEX command (full index rebuild)
3. PRAGMA commands (integrity_check / table_info / version / stats)
4. Composite multi-column indexes (CREATE INDEX ... ON t(c1, c2))
"""

import pytest
import tempfile
import shutil

from apexbase import ApexClient


# ============================================================================
# Fixtures
# ============================================================================

@pytest.fixture
def client():
    """Create a temporary client with a test table."""
    tmpdir = tempfile.mkdtemp(prefix="apexbase_v12_")
    c = ApexClient(dirpath=tmpdir)
    c.create_table("t1", {"name": "string", "age": "int", "city": "string", "score": "float"})
    c.use_table("t1")
    yield c
    c.close()
    shutil.rmtree(tmpdir, ignore_errors=True)


@pytest.fixture
def client_with_data(client):
    """Client with pre-loaded test data."""
    rows = []
    cities = ["Beijing", "Shanghai", "Guangzhou", "Shenzhen", "Hangzhou"]
    for i in range(100):
        rows.append({
            "name": f"user_{i}",
            "age": 20 + (i % 50),
            "city": cities[i % len(cities)],
            "score": 60.0 + (i % 40),
        })
    client.store(rows)
    return client


# ============================================================================
# 1. Transaction Timeout Protection
# ============================================================================

class TestTransactionTimeout:
    """Test transaction timeout protection (30s idle timeout)."""

    def test_txn_completes_within_timeout(self, client):
        """A transaction that completes quickly should succeed without timeout."""
        client.store([{"name": "Alice", "age": 25, "city": "NYC", "score": 90.0}])

        client.execute("BEGIN")
        client.execute("INSERT INTO t1 (name, age, city, score) VALUES ('Bob', 30, 'LA', 85.0)")
        client.execute("COMMIT")

        assert client.count_rows() == 2

    def test_multiple_rapid_transactions(self, client):
        """Multiple rapid transactions should all succeed (no stale timeout)."""
        client.store([{"name": "base", "age": 1, "city": "X", "score": 1.0}])

        for i in range(10):
            client.execute("BEGIN")
            client.execute(f"INSERT INTO t1 (name, age, city, score) VALUES ('txn_{i}', {i}, 'city_{i}', {float(i)})")
            client.execute("COMMIT")

        assert client.count_rows() == 11

    def test_txn_with_reads_refreshes_activity(self, client):
        """Reads within a transaction should refresh the activity timestamp."""
        client.store([{"name": "Alice", "age": 25, "city": "NYC", "score": 90.0}])

        client.execute("BEGIN")
        # Multiple reads should keep the transaction alive
        for _ in range(5):
            df = client.execute("SELECT * FROM t1").to_pandas()
            assert len(df) >= 1
        client.execute("INSERT INTO t1 (name, age, city, score) VALUES ('Bob', 30, 'LA', 85.0)")
        client.execute("COMMIT")

        assert client.count_rows() == 2

    def test_rollback_after_activity(self, client):
        """ROLLBACK should work even after multiple operations."""
        client.store([{"name": "Alice", "age": 25, "city": "NYC", "score": 90.0}])

        client.execute("BEGIN")
        client.execute("INSERT INTO t1 (name, age, city, score) VALUES ('Ghost', 99, 'X', 0.0)")
        # Read (refreshes activity)
        df = client.execute("SELECT * FROM t1").to_pandas()
        assert len(df) == 2  # read-your-writes
        client.execute("ROLLBACK")

        assert client.count_rows() == 1

    def test_new_txn_after_commit(self, client):
        """Starting a new transaction immediately after COMMIT should work."""
        client.store([{"name": "Alice", "age": 25, "city": "NYC", "score": 90.0}])

        client.execute("BEGIN")
        client.execute("INSERT INTO t1 (name, age, city, score) VALUES ('Bob', 30, 'LA', 85.0)")
        client.execute("COMMIT")

        client.execute("BEGIN")
        client.execute("INSERT INTO t1 (name, age, city, score) VALUES ('Charlie', 35, 'SF', 95.0)")
        client.execute("COMMIT")

        assert client.count_rows() == 3

    def test_new_txn_after_rollback(self, client):
        """Starting a new transaction after ROLLBACK should work cleanly."""
        client.store([{"name": "Alice", "age": 25, "city": "NYC", "score": 90.0}])

        client.execute("BEGIN")
        client.execute("INSERT INTO t1 (name, age, city, score) VALUES ('Ghost', 99, 'X', 0.0)")
        client.execute("ROLLBACK")

        client.execute("BEGIN")
        client.execute("INSERT INTO t1 (name, age, city, score) VALUES ('Bob', 30, 'LA', 85.0)")
        client.execute("COMMIT")

        assert client.count_rows() == 2
        df = client.execute("SELECT name FROM t1 ORDER BY name").to_pandas()
        assert list(df["name"]) == ["Alice", "Bob"]


# ============================================================================
# 2. REINDEX Command
# ============================================================================

class TestReindex:
    """Test REINDEX table command — full index rebuild."""

    def test_reindex_rebuilds_index(self, client_with_data):
        """REINDEX should rebuild indexes and queries still work correctly."""
        c = client_with_data
        c.execute("CREATE INDEX idx_city ON t1 (city)")

        # Verify index works before REINDEX
        r1 = c.execute("SELECT COUNT(*) as cnt FROM t1 WHERE city = 'Beijing'").to_dict()
        count_before = r1[0]["cnt"]

        # REINDEX
        result = c.execute("REINDEX t1")
        # Should return number of indexes rebuilt
        assert result is not None

        # Verify index still works after REINDEX
        r2 = c.execute("SELECT COUNT(*) as cnt FROM t1 WHERE city = 'Beijing'").to_dict()
        count_after = r2[0]["cnt"]
        assert count_before == count_after

    def test_reindex_no_indexes(self, client):
        """REINDEX on table with no indexes should return 0."""
        client.store([{"name": "Alice", "age": 25, "city": "NYC", "score": 90.0}])
        result = client.execute("REINDEX t1")
        assert result is not None

    def test_reindex_nonexistent_table(self, client):
        """REINDEX on non-existent table should error."""
        with pytest.raises(Exception):
            client.execute("REINDEX nonexistent_table")

    def test_reindex_after_insert(self, client):
        """REINDEX after INSERT should include new data in index."""
        client.store([{"name": "Alice", "age": 25, "city": "NYC", "score": 90.0}])
        client.execute("CREATE INDEX idx_city ON t1 (city)")

        # Insert new data
        client.execute("INSERT INTO t1 (name, age, city, score) VALUES ('Bob', 30, 'LA', 85.0)")

        # REINDEX
        client.execute("REINDEX t1")

        # Verify new data is in the index
        r = client.execute("SELECT * FROM t1 WHERE city = 'LA'").to_pandas()
        assert len(r) == 1
        assert r.iloc[0]["name"] == "Bob"

    def test_reindex_after_delete(self, client):
        """REINDEX after DELETE should not include deleted data in index."""
        client.store([
            {"name": "Alice", "age": 25, "city": "NYC", "score": 90.0},
            {"name": "Bob", "age": 30, "city": "LA", "score": 85.0},
            {"name": "Charlie", "age": 35, "city": "NYC", "score": 95.0},
        ])
        client.execute("CREATE INDEX idx_city ON t1 (city)")

        # Delete
        client.execute("DELETE FROM t1 WHERE name = 'Alice'")

        # REINDEX
        client.execute("REINDEX t1")

        # Verify deleted data is not in index results
        r = client.execute("SELECT * FROM t1 WHERE city = 'NYC'").to_pandas()
        assert len(r) == 1
        assert r.iloc[0]["name"] == "Charlie"

    def test_reindex_after_update(self, client):
        """REINDEX after UPDATE should reflect updated values in index."""
        client.store([
            {"name": "Alice", "age": 25, "city": "NYC", "score": 90.0},
            {"name": "Bob", "age": 30, "city": "LA", "score": 85.0},
        ])
        client.execute("CREATE INDEX idx_city ON t1 (city)")

        # Update city
        client.execute("UPDATE t1 SET city = 'SF' WHERE name = 'Alice'")

        # REINDEX
        client.execute("REINDEX t1")

        # Verify updated value is in index
        r_sf = client.execute("SELECT * FROM t1 WHERE city = 'SF'").to_pandas()
        assert len(r_sf) == 1
        assert r_sf.iloc[0]["name"] == "Alice"

        r_nyc = client.execute("SELECT * FROM t1 WHERE city = 'NYC'").to_pandas()
        assert len(r_nyc) == 0

    def test_reindex_multiple_indexes(self, client):
        """REINDEX should rebuild all indexes on the table."""
        client.store([
            {"name": "Alice", "age": 25, "city": "NYC", "score": 90.0},
            {"name": "Bob", "age": 30, "city": "LA", "score": 85.0},
        ])
        client.execute("CREATE INDEX idx_city ON t1 (city)")
        client.execute("CREATE INDEX idx_age ON t1 (age) USING BTREE")

        # REINDEX
        client.execute("REINDEX t1")

        # Verify both indexes work
        r1 = client.execute("SELECT * FROM t1 WHERE city = 'NYC'").to_pandas()
        assert len(r1) == 1

        r2 = client.execute("SELECT * FROM t1 WHERE age = 30").to_pandas()
        assert len(r2) == 1
        assert r2.iloc[0]["name"] == "Bob"

    def test_reindex_with_table_keyword(self, client):
        """REINDEX TABLE t1 syntax should also work."""
        client.store([{"name": "Alice", "age": 25, "city": "NYC", "score": 90.0}])
        client.execute("CREATE INDEX idx_city ON t1 (city)")
        result = client.execute("REINDEX TABLE t1")
        assert result is not None


# ============================================================================
# 3. PRAGMA Commands
# ============================================================================

class TestPragmaIntegrityCheck:
    """Test PRAGMA integrity_check."""

    def test_integrity_check_healthy_table(self, client_with_data):
        """PRAGMA integrity_check on a healthy table should return all 'ok'."""
        c = client_with_data
        result = c.execute("PRAGMA integrity_check(t1)")
        df = result.to_pandas()

        assert len(df) > 0
        assert "check" in df.columns
        assert "status" in df.columns

        # All statuses should contain "ok"
        for _, row in df.iterrows():
            assert "ok" in row["status"].lower() or "FAIL" not in row["status"], \
                f"Check '{row['check']}' failed: {row['status']}"

    def test_integrity_check_with_indexes(self, client_with_data):
        """PRAGMA integrity_check should report index count."""
        c = client_with_data
        c.execute("CREATE INDEX idx_city ON t1 (city)")
        c.execute("CREATE INDEX idx_age ON t1 (age) USING BTREE")

        result = c.execute("PRAGMA integrity_check(t1)")
        df = result.to_pandas()

        # Find the indexes check row
        idx_row = df[df["check"] == "indexes"]
        assert len(idx_row) == 1
        assert "2 indexes" in idx_row.iloc[0]["status"]

    def test_integrity_check_reports_row_count(self, client_with_data):
        """PRAGMA integrity_check should report correct row count."""
        c = client_with_data
        result = c.execute("PRAGMA integrity_check(t1)")
        df = result.to_pandas()

        row_count_row = df[df["check"] == "row_count"]
        assert len(row_count_row) == 1
        assert "100 rows" in row_count_row.iloc[0]["status"]

    def test_integrity_check_reports_schema(self, client_with_data):
        """PRAGMA integrity_check should report schema column count."""
        c = client_with_data
        result = c.execute("PRAGMA integrity_check(t1)")
        df = result.to_pandas()

        schema_row = df[df["check"] == "schema_valid"]
        assert len(schema_row) == 1
        assert "columns" in schema_row.iloc[0]["status"]

    def test_integrity_check_nonexistent_table(self, client):
        """PRAGMA integrity_check on non-existent table should report FAIL."""
        result = client.execute("PRAGMA integrity_check(nonexistent)")
        df = result.to_pandas()
        # Should have file_exists check with FAIL
        file_row = df[df["check"] == "file_exists"]
        assert len(file_row) == 1
        assert "FAIL" in file_row.iloc[0]["status"]

    def test_integrity_check_empty_table(self, client):
        """PRAGMA integrity_check on empty table should still pass."""
        client.store([{"name": "temp", "age": 1, "city": "x", "score": 1.0}])
        client.execute("TRUNCATE TABLE t1")
        client.flush()

        result = client.execute("PRAGMA integrity_check(t1)")
        df = result.to_pandas()
        # File should exist and be readable
        file_row = df[df["check"] == "file_exists"]
        assert len(file_row) == 1
        assert "ok" in file_row.iloc[0]["status"].lower()


class TestPragmaTableInfo:
    """Test PRAGMA table_info."""

    def test_table_info_returns_columns(self, client_with_data):
        """PRAGMA table_info should return column info."""
        c = client_with_data
        result = c.execute("PRAGMA table_info(t1)")
        df = result.to_pandas()

        assert "cid" in df.columns
        assert "name" in df.columns
        assert "type" in df.columns
        assert len(df) >= 4  # _id, name, age, city, score (at least 4 user cols)

    def test_table_info_column_names(self, client_with_data):
        """PRAGMA table_info should list all column names."""
        c = client_with_data
        result = c.execute("PRAGMA table_info(t1)")
        df = result.to_pandas()

        col_names = set(df["name"].tolist())
        assert "name" in col_names
        assert "age" in col_names
        assert "city" in col_names
        assert "score" in col_names

    def test_table_info_cid_sequential(self, client_with_data):
        """PRAGMA table_info should have sequential cid values."""
        c = client_with_data
        result = c.execute("PRAGMA table_info(t1)")
        df = result.to_pandas()

        cids = sorted(df["cid"].tolist())
        assert cids == list(range(len(cids)))

    def test_table_info_nonexistent_table(self, client):
        """PRAGMA table_info on non-existent table should error."""
        with pytest.raises(Exception):
            client.execute("PRAGMA table_info(nonexistent)")

    def test_table_info_after_add_column(self):
        """PRAGMA table_info should reflect ALTER TABLE ADD COLUMN."""
        with tempfile.TemporaryDirectory() as tmpdir:
            c = ApexClient(tmpdir)
            c.create_table("t2", {"name": "string", "age": "int"})
            c.use_table("t2")
            c.store([{"name": "Alice", "age": 25}])

            c.execute("ALTER TABLE t2 ADD COLUMN email STRING")

            result = c.execute("PRAGMA table_info(t2)")
            df = result.to_pandas()
            col_names = set(df["name"].tolist())
            assert "email" in col_names
            c.close()


class TestPragmaVersion:
    """Test PRAGMA version."""

    def test_version_returns_string(self, client):
        """PRAGMA version should return a version string."""
        result = client.execute("PRAGMA version")
        df = result.to_pandas()

        assert len(df) == 1
        assert "version" in df.columns
        version_str = df.iloc[0]["version"]
        assert "ApexBase" in version_str

    def test_version_format(self, client):
        """PRAGMA version should contain version number."""
        result = client.execute("PRAGMA version")
        df = result.to_pandas()
        version_str = df.iloc[0]["version"]
        # Should match "ApexBase X.Y" format
        assert "1.0" in version_str or "ApexBase" in version_str


class TestPragmaStats:
    """Test PRAGMA stats."""

    def test_stats_after_analyze(self, client_with_data):
        """PRAGMA stats after ANALYZE should return table statistics."""
        c = client_with_data
        c.execute("ANALYZE t1")
        result = c.execute("PRAGMA stats(t1)")
        df = result.to_pandas()

        assert len(df) == 1
        assert "table" in df.columns
        assert "row_count" in df.columns
        assert "columns" in df.columns
        assert df.iloc[0]["row_count"] == 100

    def test_stats_without_analyze(self, client_with_data):
        """PRAGMA stats without prior ANALYZE should return scalar 0 or empty."""
        c = client_with_data
        result = c.execute("PRAGMA stats(t1)")
        # Without ANALYZE, stats may not be available
        # The result could be Scalar(0) or empty data
        assert result is not None

    def test_stats_column_count(self, client_with_data):
        """PRAGMA stats should report correct column count."""
        c = client_with_data
        c.execute("ANALYZE t1")
        result = c.execute("PRAGMA stats(t1)")
        df = result.to_pandas()

        assert len(df) == 1
        # Should have at least 4 user columns + _id
        assert df.iloc[0]["columns"] >= 4


class TestPragmaErrors:
    """Test PRAGMA error handling."""

    def test_unknown_pragma(self, client):
        """Unknown PRAGMA should raise an error."""
        with pytest.raises(Exception, match="Unknown PRAGMA|Unsupported|unknown"):
            client.execute("PRAGMA unknown_command")


# ============================================================================
# 4. Composite Multi-Column Index
# ============================================================================

class TestCompositeIndexCreation:
    """Test CREATE INDEX with multiple columns."""

    def test_create_composite_index(self, client_with_data):
        """CREATE INDEX on (city, age) should succeed."""
        c = client_with_data
        result = c.execute("CREATE INDEX idx_city_age ON t1 (city, age)")
        assert result is not None

    def test_create_composite_index_btree(self, client_with_data):
        """CREATE INDEX ... USING BTREE on multiple columns should succeed."""
        c = client_with_data
        result = c.execute("CREATE INDEX idx_city_age ON t1 (city, age) USING BTREE")
        assert result is not None

    def test_create_composite_index_unique(self, client_with_data):
        """CREATE UNIQUE INDEX on multiple columns should succeed."""
        c = client_with_data
        result = c.execute("CREATE UNIQUE INDEX idx_name_city ON t1 (name, city)")
        assert result is not None

    def test_create_composite_index_if_not_exists(self, client_with_data):
        """CREATE INDEX IF NOT EXISTS on composite should not error on duplicate."""
        c = client_with_data
        c.execute("CREATE INDEX idx_city_age ON t1 (city, age)")
        result = c.execute("CREATE INDEX IF NOT EXISTS idx_city_age ON t1 (city, age)")
        assert result is not None

    def test_create_composite_index_duplicate_error(self, client_with_data):
        """CREATE INDEX without IF NOT EXISTS should error on duplicate composite index."""
        c = client_with_data
        c.execute("CREATE INDEX idx_city_age ON t1 (city, age)")
        with pytest.raises(Exception):
            c.execute("CREATE INDEX idx_city_age ON t1 (city, age)")

    def test_create_composite_index_nonexistent_column(self, client_with_data):
        """CREATE INDEX with non-existent column should error."""
        c = client_with_data
        with pytest.raises(Exception):
            c.execute("CREATE INDEX idx_bad ON t1 (city, nonexistent_col)")

    def test_create_composite_index_on_empty_table(self, client):
        """CREATE INDEX on empty table with schema should succeed."""
        client.store([{"name": "temp", "age": 1, "city": "x", "score": 1.0}])
        client.execute("TRUNCATE TABLE t1")
        result = client.execute("CREATE INDEX idx_city_age ON t1 (city, age)")
        assert result is not None

    def test_drop_composite_index(self, client_with_data):
        """DROP INDEX on composite index should succeed."""
        c = client_with_data
        c.execute("CREATE INDEX idx_city_age ON t1 (city, age)")
        result = c.execute("DROP INDEX idx_city_age ON t1")
        assert result is not None


class TestCompositeIndexQuery:
    """Test that composite indexes return correct query results."""

    def test_composite_index_equality_filter(self, client_with_data):
        """WHERE on composite index columns should return correct results."""
        c = client_with_data
        c.execute("CREATE INDEX idx_city_age ON t1 (city, age)")

        result = c.execute("SELECT * FROM t1 WHERE city = 'Beijing' AND age = 20")
        df = result.to_pandas()
        # All results should match both conditions
        for _, row in df.iterrows():
            assert row["city"] == "Beijing"
            assert row["age"] == 20

    def test_composite_index_correctness_vs_scan(self, client_with_data):
        """Composite index results should match full-scan results."""
        c = client_with_data

        # Full scan (no index)
        r_scan = c.execute(
            "SELECT name FROM t1 WHERE city = 'Shanghai' AND age = 21 ORDER BY name"
        ).to_pandas()

        # Create composite index
        c.execute("CREATE INDEX idx_city_age ON t1 (city, age)")

        # Index-accelerated query
        r_idx = c.execute(
            "SELECT name FROM t1 WHERE city = 'Shanghai' AND age = 21 ORDER BY name"
        ).to_pandas()

        assert len(r_scan) == len(r_idx)
        assert r_scan["name"].tolist() == r_idx["name"].tolist()

    def test_composite_index_with_order_by(self, client_with_data):
        """Composite index + ORDER BY should return correctly ordered results."""
        c = client_with_data
        c.execute("CREATE INDEX idx_city_age ON t1 (city, age)")

        result = c.execute(
            "SELECT * FROM t1 WHERE city = 'Beijing' ORDER BY age"
        ).to_pandas()
        ages = result["age"].tolist()
        assert ages == sorted(ages)

    def test_composite_index_with_limit(self, client_with_data):
        """Composite index + LIMIT should respect the limit."""
        c = client_with_data
        c.execute("CREATE INDEX idx_city_age ON t1 (city, age)")

        result = c.execute("SELECT * FROM t1 WHERE city = 'Beijing' LIMIT 3")
        df = result.to_pandas()
        assert len(df) <= 3


class TestCompositeIndexDML:
    """Test that composite indexes stay in sync with DML operations."""

    def test_insert_updates_composite_index(self, client):
        """SQL INSERT after composite index creation should be found."""
        client.store([{"name": "Alice", "age": 25, "city": "NYC", "score": 90.0}])
        client.execute("CREATE INDEX idx_city_age ON t1 (city, age)")

        client.execute("INSERT INTO t1 (name, age, city, score) VALUES ('Bob', 30, 'LA', 85.0)")

        result = client.execute("SELECT * FROM t1 WHERE city = 'LA' AND age = 30")
        df = result.to_pandas()
        assert len(df) == 1
        assert df.iloc[0]["name"] == "Bob"

    def test_delete_updates_composite_index(self, client):
        """SQL DELETE should remove entries from composite index."""
        client.store([
            {"name": "Alice", "age": 25, "city": "NYC", "score": 90.0},
            {"name": "Bob", "age": 30, "city": "LA", "score": 85.0},
            {"name": "Charlie", "age": 25, "city": "NYC", "score": 95.0},
        ])
        client.execute("CREATE INDEX idx_city_age ON t1 (city, age)")

        client.execute("DELETE FROM t1 WHERE name = 'Alice'")

        result = client.execute("SELECT * FROM t1 WHERE city = 'NYC' AND age = 25")
        df = result.to_pandas()
        assert len(df) == 1
        assert df.iloc[0]["name"] == "Charlie"

    def test_update_updates_composite_index(self, client):
        """SQL UPDATE should update composite index entries."""
        client.store([
            {"name": "Alice", "age": 25, "city": "NYC", "score": 90.0},
            {"name": "Bob", "age": 30, "city": "LA", "score": 85.0},
        ])
        client.execute("CREATE INDEX idx_city_age ON t1 (city, age)")

        client.execute("UPDATE t1 SET city = 'SF', age = 28 WHERE name = 'Alice'")

        # Old composite key should not match
        r_old = client.execute("SELECT * FROM t1 WHERE city = 'NYC' AND age = 25")
        assert len(r_old.to_pandas()) == 0

        # New composite key should match
        r_new = client.execute("SELECT * FROM t1 WHERE city = 'SF' AND age = 28")
        df_new = r_new.to_pandas()
        assert len(df_new) == 1
        assert df_new.iloc[0]["name"] == "Alice"

    def test_reindex_composite_index(self, client):
        """REINDEX should correctly rebuild composite indexes."""
        client.store([
            {"name": "Alice", "age": 25, "city": "NYC", "score": 90.0},
            {"name": "Bob", "age": 30, "city": "LA", "score": 85.0},
            {"name": "Charlie", "age": 25, "city": "NYC", "score": 95.0},
        ])
        client.execute("CREATE INDEX idx_city_age ON t1 (city, age)")

        # Modify data
        client.execute("INSERT INTO t1 (name, age, city, score) VALUES ('Dave', 40, 'SF', 75.0)")
        client.execute("DELETE FROM t1 WHERE name = 'Bob'")

        # REINDEX
        client.execute("REINDEX t1")

        # Verify results
        r = client.execute("SELECT * FROM t1 WHERE city = 'NYC' AND age = 25")
        df = r.to_pandas()
        assert len(df) == 2  # Alice + Charlie
        names = sorted(df["name"].tolist())
        assert names == ["Alice", "Charlie"]

        # Verify deleted row not found
        r2 = client.execute("SELECT * FROM t1 WHERE city = 'LA' AND age = 30")
        assert len(r2.to_pandas()) == 0

        # Verify new row found
        r3 = client.execute("SELECT * FROM t1 WHERE city = 'SF' AND age = 40")
        assert len(r3.to_pandas()) == 1


class TestCompositeIndexThreeColumns:
    """Test composite index with 3 columns."""

    def test_three_column_composite_index(self):
        """CREATE INDEX on 3 columns should work."""
        with tempfile.TemporaryDirectory() as tmpdir:
            c = ApexClient(tmpdir)
            c.create_table("t3", {"a": "string", "b": "int", "c": "string"})
            c.use_table("t3")
            c.store([
                {"a": "x", "b": 1, "c": "p"},
                {"a": "x", "b": 1, "c": "q"},
                {"a": "x", "b": 2, "c": "p"},
                {"a": "y", "b": 1, "c": "p"},
            ])
            c.execute("CREATE INDEX idx_abc ON t3 (a, b, c)")

            result = c.execute("SELECT * FROM t3 WHERE a = 'x' AND b = 1 AND c = 'p'")
            df = result.to_pandas()
            assert len(df) == 1

            result2 = c.execute("SELECT * FROM t3 WHERE a = 'x' AND b = 1")
            df2 = result2.to_pandas()
            assert len(df2) == 2
            c.close()


# ============================================================================
# Integration: Cross-Feature Tests
# ============================================================================

class TestCrossFeatureIntegration:
    """Tests combining multiple v12 features together."""

    def test_pragma_integrity_after_reindex(self, client_with_data):
        """PRAGMA integrity_check after REINDEX should still pass."""
        c = client_with_data
        c.execute("CREATE INDEX idx_city ON t1 (city)")
        c.execute("REINDEX t1")

        result = c.execute("PRAGMA integrity_check(t1)")
        df = result.to_pandas()
        for _, row in df.iterrows():
            assert "FAIL" not in row["status"]

    def test_reindex_composite_then_pragma(self, client_with_data):
        """REINDEX composite index then verify with PRAGMA integrity_check."""
        c = client_with_data
        c.execute("CREATE INDEX idx_city_age ON t1 (city, age)")
        c.execute("REINDEX t1")

        result = c.execute("PRAGMA integrity_check(t1)")
        df = result.to_pandas()

        idx_row = df[df["check"] == "indexes"]
        assert len(idx_row) == 1
        assert "1 indexes" in idx_row.iloc[0]["status"]

    def test_txn_with_composite_index(self, client):
        """Transaction operations with composite index should work."""
        client.store([{"name": "Alice", "age": 25, "city": "NYC", "score": 90.0}])
        client.execute("CREATE INDEX idx_city_age ON t1 (city, age)")

        client.execute("BEGIN")
        client.execute("INSERT INTO t1 (name, age, city, score) VALUES ('Bob', 30, 'LA', 85.0)")
        client.execute("COMMIT")

        result = client.execute("SELECT * FROM t1 WHERE city = 'LA' AND age = 30")
        df = result.to_pandas()
        assert len(df) == 1
        assert df.iloc[0]["name"] == "Bob"

    def test_fast_txn_batch_insert_updates_composite_index(self, client):
        """Fast transaction batch commit keeps every composite-index entry current."""
        client.store([{"name": "Alice", "age": 25, "city": "NYC", "score": 90.0}])
        client.execute("CREATE INDEX idx_city_age ON t1 (city, age)")

        client.execute("BEGIN")
        client.execute("INSERT INTO t1 (name, age, city, score) VALUES ('Bob', 30, 'LA', 85.0)")
        client.execute("INSERT INTO t1 (name, age, city, score) VALUES ('Cara', 31, 'LA', 80.0)")
        client.execute("COMMIT")

        for name, age in (("Bob", 30), ("Cara", 31)):
            df = client.execute(
                f"SELECT * FROM t1 WHERE city = 'LA' AND age = {age}"
            ).to_pandas()
            assert len(df) == 1
            assert df.iloc[0]["name"] == name

    def test_pragma_table_info_after_operations(self, client):
        """PRAGMA table_info should work after various DML operations."""
        client.store([{"name": "Alice", "age": 25, "city": "NYC", "score": 90.0}])
        client.execute("INSERT INTO t1 (name, age, city, score) VALUES ('Bob', 30, 'LA', 85.0)")
        client.execute("DELETE FROM t1 WHERE name = 'Alice'")

        result = client.execute("PRAGMA table_info(t1)")
        df = result.to_pandas()
        col_names = set(df["name"].tolist())
        assert "name" in col_names
        assert "age" in col_names
        assert "city" in col_names
        assert "score" in col_names