diff --git a/docs/changelog.rst b/docs/changelog.rst
index 556e497..a68a9b6 100644
--- a/docs/changelog.rst
+++ b/docs/changelog.rst
@@ -10,6 +10,7 @@ Unreleased
 ----------
 
 - JSON output from the command-line tool no longer escapes non-ASCII characters, so ``sqlite-utils data.db "select '日本語' as text"`` now outputs ``[{"text": "日本語"}]``. This matches how values were already stored by ``insert`` and how CSV/TSV output already behaved. A new ``--ascii`` option restores the previous behavior of escaping non-ASCII characters, for output destinations that cannot handle UTF-8 - see :ref:`cli_query_json_ascii`. The option is available on the ``query``, ``rows``, ``search``, ``tables``, ``views``, ``triggers``, ``indexes`` and ``memory`` commands. The ``convert --multi --dry-run`` preview and ``plugins`` output also no longer escape non-ASCII characters. (:issue:`625`)
+- ``table.insert_all(..., pk=...)`` now raises ``InvalidColumns`` if ``pk=`` names columns that do not exist in an existing table. Previously this behaved inconsistently, with single-row inserts raising a ``KeyError`` while other row counts succeeded. (:issue:`732`)
 
 .. _v4_0rc3:
 
diff --git a/sqlite_utils/db.py b/sqlite_utils/db.py
index 0807892..68aee3e 100644
--- a/sqlite_utils/db.py
+++ b/sqlite_utils/db.py
@@ -4229,6 +4229,24 @@ class Table(Queryable):
         if hash_id:
             pk = hash_id
 
+        if pk and not hash_id and self.exists():
+            pk_cols = [pk] if isinstance(pk, str) else list(pk)
+            existing_columns = self.columns_dict
+            missing_pk_cols = [
+                col
+                for col in pk_cols
+                if resolve_casing(col, existing_columns) not in existing_columns
+            ]
+            if missing_pk_cols:
+                raise InvalidColumns(
+                    "Invalid primary key column{} {} for table {} with columns {}".format(
+                        "s" if len(missing_pk_cols) > 1 else "",
+                        missing_pk_cols,
+                        self.name,
+                        list(existing_columns),
+                    )
+                )
+
         if ignore and replace:
             raise ValueError("Use either ignore=True or replace=True, not both")
         all_columns = []
diff --git a/tests/test_create.py b/tests/test_create.py
index decefcf..7fbd7d5 100644
--- a/tests/test_create.py
+++ b/tests/test_create.py
@@ -3,6 +3,7 @@ from sqlite_utils.db import (
     Database,
     DescIndex,
     AlterError,
+    InvalidColumns,
     NoObviousTable,
     OperationalError,
     ForeignKey,
@@ -925,6 +926,21 @@ def test_insert_thousands_adds_extra_columns_after_first_100_with_alter(fresh_db
     assert rows == [{"i": 101, "word": None, "extra": "Should trigger ALTER"}]
 
 
+@pytest.mark.parametrize("num_rows", (0, 1, 2, 3, 10))
+def test_insert_all_pk_not_in_records_raises(fresh_db, num_rows):
+    # https://github.com/simonw/sqlite-utils/issues/732
+    fresh_db.conn.execute("CREATE TABLE t (a TEXT, b INT, PRIMARY KEY (a, b))")
+    rows = [{"a": "x{}".format(i), "b": i} for i in range(num_rows)]
+
+    with pytest.raises(InvalidColumns) as ex:
+        fresh_db["t"].insert_all(rows, pk="not_a_column", alter=True)
+
+    assert ex.value.args == (
+        "Invalid primary key column ['not_a_column'] for table t with columns ['a', 'b']",
+    )
+    assert fresh_db["t"].count == 0
+
+
 def test_insert_ignore(fresh_db):
     fresh_db["test"].insert({"id": 1, "bar": 2}, pk="id")
     # Should raise an error if we try this again