diff --git a/datamind/capabilities/ingest/service.py b/datamind/capabilities/ingest/service.py
index b04387e..1510c43 100644
--- a/datamind/capabilities/ingest/service.py
+++ b/datamind/capabilities/ingest/service.py
@@ -882,12 +882,25 @@ async def db_import_csv(
 
         # Sanitise column names: same rule as table names.
         safe_cols: list[str] = []
-        for raw in header:
+        used_cols: set[str] = set()
+        for index, raw in enumerate(header, start=1):
             col = raw.strip()
             if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]{0,63}", col):
                 # Fall back to col_<idx> if header is unusable.
-                col = f"col_{len(safe_cols) + 1}"
+                col = f"col_{index}"
+            if col in used_cols:
+                # SQL tables cannot contain duplicate column names. Keep the
+                # first header unchanged and give later occurrences a stable
+                # fallback name without losing their values in the row dict.
+                base = f"col_{index}"
+                col = base
+                suffix = 2
+                while col in used_cols:
+                    suffix_text = f"_{suffix}"
+                    col = f"{base[:64 - len(suffix_text)]}{suffix_text}"
+                    suffix += 1
             safe_cols.append(col)
+            used_cols.add(col)
 
         rows: list[dict[str, str]] = []
         for raw_row in reader:
diff --git a/datamind/tests/test_replace_receipts.py b/datamind/tests/test_replace_receipts.py
index ee4b2d6..285199d 100644
--- a/datamind/tests/test_replace_receipts.py
+++ b/datamind/tests/test_replace_receipts.py
@@ -77,3 +77,17 @@ async def test_append_retry_remains_deduplicated(database, tmp_path):
     assert second["results"][0]["status"] == "unchanged"
     assert second["revision"] == first["revision"]
     assert (await db.query_sql("SELECT amount FROM sales")).rows == [["200"]]
+
+
+@pytest.mark.asyncio
+async def test_csv_import_keeps_duplicate_headers_as_distinct_columns(database, tmp_path):
+    db, raw = database
+    source = tmp_path / "duplicate-columns.csv"
+    source.write_text("a,a\nfirst,second\n", encoding="utf-8")
+
+    receipt = await raw.get("db_import_csv").handler(
+        path=str(source), table="duplicate_columns", if_exists="replace"
+    )
+
+    assert receipt["columns"] == ["a", "col_2"]
+    assert (await db.query_sql("SELECT * FROM duplicate_columns")).rows == [["first", "second"]]