table.transform(rename=...) now preserves indexes, closes #822

docs/changelog.rst

@@ -12,6 +12,7 @@ Unreleased
 - New ``table.checks``, ``table.column_checks`` and ``table.table_checks`` introspection properties expose column-level and table-level ``CHECK`` constraints. (:issue:`834`)
 - ``table.transform()`` now preserves ``CHECK`` constraints, including comments within their expressions. Renaming a column rewrites identifier references in checks without changing string literals or function names. Dropping a column drops a check owned by that column, and raises ``TransformError`` if a remaining check depends on it. (:issue:`762`)
 - ``table.transform()`` now preserves comments immediately before or after column definitions. These comments move with the column if it is renamed or reordered, and are removed if the column is dropped. (:issue:`762`)
+- ``table.transform(rename=...)`` now preserves explicit indexes on renamed columns by dropping and recreating those indexes against the new column names. Previously this raised a ``TransformError``. (:issue:`822`)
 - ``table.transform()`` now works for tables that are referenced by views. Previously the ``ALTER TABLE ... RENAME TO`` step raised ``no such table`` if a view referenced the table being transformed. View definitions are left unchanged - see :ref:`python_api_transform_views`. This also fixes a bug where ``transform(keep_table=...)`` silently rewrote dependent views to point at the frozen backup table instead of the live one. (:issue:`831`)
 - ``table.default_values`` now unescapes doubled single quotes in string defaults, so a default such as ``'O''Brien'`` is returned as ``"O'Brien"``. Thanks, `ikatyal2110 <https://github.com/ikatyal2110>`__. (`#811 <https://github.com/simonw/sqlite-utils/pull/811>`__)
 - ``table.default_values`` now decodes unquoted ``TRUE``, ``FALSE`` and ``NULL`` default literals as ``True``, ``False`` and ``None`` respectively. (:issue:`836`)

sqlite_utils/db.py

@@ -2947,6 +2947,80 @@ class Table(Queryable):
             new_cols=", ".join(quote_identifier(col) for col in new_cols),
         )
         sqls.append(copy_sql)
+        # Capture indexes before the old table is changed. Simple indexes that
+        # reference renamed columns are recreated from structured PRAGMA
+        # metadata instead of editing their stored CREATE INDEX SQL.
+        index_drop_sqls = []
+        index_create_sqls = []
+        xindexes_by_name = {index.name: index for index in self.xindexes}
+        for index in self.indexes:
+            if index.origin == "pk":
+                continue
+            index_sql = self.db.execute(
+                """SELECT sql FROM sqlite_master WHERE type = 'index' AND name = :index_name;""",
+                {"index_name": index.name},
+            ).fetchall()[0][0]
+            if index_sql is None:
+                raise TransformError(
+                    f"Index '{index.name}' on table '{self.name}' does not have a "
+                    "CREATE INDEX statement. You must manually drop this index prior to running this "
+                    "transformation and manually recreate the new index after running this transformation."
+                )
+            dropped_index_column = next(
+                (column for column in index.columns if column in drop), None
+            )
+            renamed_index_column = next(
+                (column for column in index.columns if column in rename), None
+            )
+            if dropped_index_column is not None:
+                raise TransformError(
+                    f"Index '{index.name}' column '{dropped_index_column}' 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."
+                )
+            xindex = xindexes_by_name[index.name]
+            indexed_columns = sorted(
+                (column for column in xindex.columns if column.key),
+                key=lambda column: column.seqno,
+            )
+            if (rename or drop) and (
+                index.partial or any(column.name is None for column in indexed_columns)
+            ):
+                raise TransformError(
+                    f"Index '{index.name}' is a partial or expression index, so it "
+                    f"cannot be safely recreated while columns are renamed or dropped. "
+                    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."
+                )
+            if renamed_index_column is not None:
+                columns_sql = []
+                for column in indexed_columns:
+                    assert column.name is not None
+                    column_sql = quote_identifier(
+                        rename.get(column.name) or column.name
+                    )
+                    if column.coll and column.coll.upper() != "BINARY":
+                        column_sql += f" COLLATE {quote_identifier(column.coll)}"
+                    if column.desc:
+                        column_sql += " DESC"
+                    columns_sql.append(column_sql)
+                index_sql = "CREATE {unique}INDEX {index_name} ON {table_name} ({columns})".format(
+                    unique="UNIQUE " if index.unique else "",
+                    index_name=quote_identifier(index.name),
+                    table_name=quote_identifier(self.name),
+                    columns=", ".join(columns_sql),
+                )
+                index_drop_sqls.append(
+                    f"DROP INDEX IF EXISTS {quote_identifier(index.name)};"
+                )
+            elif keep_table:
+                index_drop_sqls.append(
+                    f"DROP INDEX IF EXISTS {quote_identifier(index.name)};"
+                )
+            index_create_sqls.append(index_sql)
+        sqls.extend(index_drop_sqls)
         # Drop (or keep) the old table, then rename the new one into place.
         # Since SQLite 3.25 ALTER TABLE ... RENAME TO rewrites references to
         # the renamed table in every view definition, which fails if a view
@@ -2976,29 +3050,7 @@ class Table(Queryable):
             )
         )
         # Re-add existing indexes
-        for index in self.indexes:
-            if index.origin != "pk":
-                index_sql = self.db.execute(
-                    """SELECT sql FROM sqlite_master WHERE type = 'index' AND name = :index_name;""",
-                    {"index_name": index.name},
-                ).fetchall()[0][0]
-                if index_sql is None:
-                    raise TransformError(
-                        f"Index '{index.name}' on table '{self.name}' does not have a "
-                        "CREATE INDEX statement. You must manually drop this index prior to running this "
-                        "transformation and manually recreate the new index after running this transformation."
-                    )
-                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:
-                        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."
-                        )
-                sqls.append(index_sql)
+        sqls.extend(index_create_sqls)
         return sqls
 
     def extract(

tests/test_transform.py

@@ -893,22 +893,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.table("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'. "
@@ -917,6 +910,73 @@ def test_transform_with_indexes_errors(fresh_db, transform_params):
     )
 
 
+@pytest.mark.parametrize(
+    ("table_name", "index_name"),
+    (("name", "idx_name"), ("t", "name")),
+)
+def test_transform_rename_column_with_index(fresh_db, table_name, index_name):
+    # https://github.com/simonw/sqlite-utils/issues/822
+    # Use the same name for the table, column and index to ensure only the
+    # indexed column changes.
+    table = fresh_db.table(table_name)
+    table.insert({"id": 1, "name": "Cleo"}, pk="id")
+    table.create_index(["name"], index_name=index_name)
+
+    sqls = table.transform_sql(rename={"name": "full_name"}, tmp_suffix="suffix")
+    drop_index_sql = f'DROP INDEX IF EXISTS "{index_name}";'
+    assert drop_index_sql in sqls
+    assert sqls.index(drop_index_sql) < sqls.index(f'DROP TABLE "{table_name}";')
+
+    table.transform(rename={"name": "full_name"})
+
+    assert [column.name for column in table.columns] == ["id", "full_name"]
+    assert [(index.name, index.columns) for index in table.indexes] == [
+        (index_name, ["full_name"])
+    ]
+
+
+def test_transform_recreates_renamed_index_from_metadata(fresh_db):
+    table = fresh_db.table("t")
+    table.insert({"alpha": "one", "beta": "two"})
+    # Deliberately use unquoted SQL and index details that need to survive the
+    # reconstruction. Renaming both columns also guards against cascading
+    # string substitutions.
+    fresh_db.execute(
+        "CREATE UNIQUE INDEX swap_idx ON t(alpha COLLATE NOCASE DESC, beta)"
+    )
+
+    table.transform(rename={"alpha": "beta", "beta": "alpha"})
+
+    assert table.columns_dict == {"beta": str, "alpha": str}
+    assert [(index.name, index.unique, index.columns) for index in table.indexes] == [
+        ("swap_idx", 1, ["beta", "alpha"])
+    ]
+    key_columns = [column for column in table.xindexes[0].columns if column.key]
+    assert [(column.name, column.desc, column.coll) for column in key_columns] == [
+        ("beta", 1, "NOCASE"),
+        ("alpha", 0, "BINARY"),
+    ]
+
+
+@pytest.mark.parametrize(
+    "index_sql",
+    (
+        "CREATE INDEX idx_t_name ON t(lower(name))",
+        "CREATE INDEX idx_t_name ON t(name) WHERE name IS NOT NULL",
+    ),
+)
+def test_transform_rename_complex_index_errors(fresh_db, index_sql):
+    table = fresh_db.table("t")
+    table.insert({"id": 1, "name": "Cleo"}, pk="id")
+    fresh_db.execute(index_sql)
+
+    with pytest.raises(TransformError, match="partial or expression index"):
+        table.transform(rename={"name": "full_name"})
+
+    assert table.columns_dict == {"id": int, "name": str}
+    assert [index.name for index in table.indexes] == ["idx_t_name"]
+
+
 def test_transform_with_unique_constraint_implicit_index(fresh_db):
     dogs = fresh_db.table("dogs")
     # Create a table with a UNIQUE constraint on 'name', which creates an implicit index