Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion sqlite_utils/db.py
Original file line number Diff line number Diff line change
Expand Up @@ -2851,13 +2851,17 @@ def fk_with_renamed_columns(fk: ForeignKey) -> ForeignKey:
if keep_table:
sqls.append(f"DROP INDEX IF EXISTS {quote_identifier(index.name)};")
for col in index.columns:
if col in rename or col in drop:
if col in drop:
raise TransformError(
f"Index '{index.name}' column '{col}' is not in updated table '{self.name}'. "
f"You must manually drop this index prior to running this transformation "
f"and manually recreate the new index after running this transformation. "
f"The original index sql statement is: `{index_sql}`. No changes have been applied to this table."
)
elif col in rename:
index_sql = index_sql.replace(
quote_identifier(col), quote_identifier(rename[col])
)
sqls.append(index_sql)
return sqls

Expand Down
38 changes: 28 additions & 10 deletions tests/test_transform.py
Original file line number Diff line number Diff line change
Expand Up @@ -827,6 +827,19 @@ def test_transform_indexes(fresh_db, indexes, transform_params):
)


def test_transform_rename_column_with_index(fresh_db):
# https://github.com/simonw/sqlite-utils/issues/822
table = fresh_db["t"]
table.insert({"id": 1, "name": "Alice"}, pk="id")
table.create_index(["name"])
# Renaming an indexed column should not raise TransformError
table.transform(rename={"name": "full_name"})
assert [col.name for col in table.columns] == ["id", "full_name"]
# The index should be recreated on the new column name
assert len(table.indexes) == 1
assert table.indexes[0].columns == ["full_name"]


def test_transform_retains_indexes_with_foreign_keys(fresh_db):
dogs = fresh_db["dogs"]
owners = fresh_db["owners"]
Expand Down Expand Up @@ -855,22 +868,15 @@ def test_transform_retains_indexes_with_foreign_keys(fresh_db):
), f"Indexes before transform: {indexes_before_transform}\nIndexes after transform: {dogs.indexes}"


@pytest.mark.parametrize(
"transform_params",
[
{"rename": {"age": "dog_age"}},
{"drop": ["age"]},
],
)
def test_transform_with_indexes_errors(fresh_db, transform_params):
# Should error with a compound (name, age) index if age is renamed or dropped
def test_transform_with_indexes_errors(fresh_db):
# Should error with a compound (name, age) index if age is dropped
dogs = fresh_db["dogs"]
dogs.insert({"id": 1, "name": "Cleo", "age": 5}, pk="id")

dogs.create_index(["name", "age"])

with pytest.raises(TransformError) as excinfo:
dogs.transform(**transform_params)
dogs.transform(drop=["age"])

assert (
"Index 'idx_dogs_name_age' column 'age' is not in updated table 'dogs'. "
Expand All @@ -879,6 +885,18 @@ def test_transform_with_indexes_errors(fresh_db, transform_params):
)


def test_transform_rename_column_in_compound_index(fresh_db):
# https://github.com/simonw/sqlite-utils/issues/822
# Renaming a column in a compound index should update the index, not error
dogs = fresh_db["dogs"]
dogs.insert({"id": 1, "name": "Cleo", "age": 5}, pk="id")
dogs.create_index(["name", "age"])
dogs.transform(rename={"age": "dog_age"})
assert [col.name for col in dogs.columns] == ["id", "name", "dog_age"]
assert len(dogs.indexes) == 1
assert dogs.indexes[0].columns == ["name", "dog_age"]


def test_transform_with_unique_constraint_implicit_index(fresh_db):
dogs = fresh_db["dogs"]
# Create a table with a UNIQUE constraint on 'name', which creates an implicit index
Expand Down
Loading