From 4487c99f89da63a134b8bd7ef338b1c3abe59be6 Mon Sep 17 00:00:00 2001 From: Harsh Thakkar Date: Thu, 24 Sep 2026 05:49:50 +0530 Subject: [PATCH] fix: reject transform renames that collide column names Renaming a column onto the name of another existing column silently collapsed the two into one and discarded the source column's entire dataset, because the new CREATE TABLE column list was built with dict(new_column_pairs) and the earlier entry was overwritten. Detect duplicate target names and raise TransformError before any SQL is executed, leaving the table untouched. --- sqlite_utils/db.py | 20 ++++++++++++++++++++ tests/test_transform.py | 28 ++++++++++++++++++++++++++++ 2 files changed, 48 insertions(+) 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