diff --git a/sqlite_utils/db.py b/sqlite_utils/db.py index c011d9bb0..46109dee7 100644 --- a/sqlite_utils/db.py +++ b/sqlite_utils/db.py @@ -2998,6 +2998,26 @@ def fk_with_renamed_columns(fk: ForeignKey) -> ForeignKey: new_column_pairs.append((new_name, type_)) copy_from_to[name] = new_name + duplicate_columns = sorted( + { + column_name + for column_name, _ in new_column_pairs + if sum( + 1 for other_name, _ in new_column_pairs if other_name == column_name + ) + > 1 + } + ) + if duplicate_columns: + raise TransformError( + "Cannot transform table '{}': renaming or retyping columns would result " + "in duplicate column name{} '{}'. No changes have been applied to this table.".format( + self.name, + "s" if len(duplicate_columns) > 1 else "", + "', '".join(duplicate_columns), + ) + ) + if existing_autoincrement: existing_autoincrement = resolve_casing( existing_autoincrement, existing_columns diff --git a/tests/test_transform.py b/tests/test_transform.py index 873871308..65871fa3f 100644 --- a/tests/test_transform.py +++ b/tests/test_transform.py @@ -254,6 +254,34 @@ def test_transform_rename_pk(fresh_db): ) +def test_transform_rename_cannot_lose_data_by_colliding_columns(fresh_db): + # Renaming one column onto the name of another used to silently collapse + # the two columns into one, discarding the source column's data entirely. + table = fresh_db.table("items") + table.insert({"a": 1, "b": 2}) + with pytest.raises(TransformError, match="duplicate column name"): + table.transform(rename={"b": "a"}) + # The failed transform must leave the table and its data untouched. + assert table.count == 1 + assert [(r["a"], r["b"]) for r in table.rows_where(order_by="rowid")] == [(1, 2)] + assert '"b"' in table.schema + + +def test_transform_rename_two_columns_to_same_name_errors(fresh_db): + table = fresh_db.table("items") + table.insert({"a": 1, "b": 2, "c": 3}) + with pytest.raises(TransformError, match="duplicate column name 'x'"): + table.transform(rename={"b": "x", "c": "x"}) + + +def test_transform_can_swap_two_column_names(fresh_db): + # A straight swap a<->b is legal and must not be blocked. + table = fresh_db.table("items") + table.insert({"a": 1, "b": 2}) + table.transform(rename={"a": "b", "b": "a"}) + assert [(r["a"], r["b"]) for r in table.rows_where(order_by="rowid")] == [(2, 1)] + + def test_transform_preserves_keyword_literal_defaults(fresh_db): # transform() used to requote keyword-literal defaults (DEFAULT TRUE became # DEFAULT 'TRUE'), so a default insert stored the text 'TRUE' instead of the