diff --git a/spanner_orm/admin/update.py b/spanner_orm/admin/update.py
index 444e15d..617dc59 100644
--- a/spanner_orm/admin/update.py
+++ b/spanner_orm/admin/update.py
@@ -82,50 +82,6 @@ def ddl(self) -> str:
           parent=self._model.interleaved.table)
     return statement
 
-  def validate(self) -> None:
-    if not self._model.table:
-      raise error.SpannerError('New table has no name')
-
-    existing_model = metadata.SpannerMetadata.model(self._model.table)
-    if existing_model:
-      raise error.SpannerError('Table {} already exists'.format(
-          self._model.table))
-
-    if self._model.interleaved:
-      self._validate_parent()
-
-    self._validate_primary_keys()
-
-    if self._model.indexes.keys() - {index.Index.PRIMARY_INDEX}:
-      raise error.SpannerError(
-          'Secondary indexes cannot be created by CreateTable; use CreateIndex '
-          'in a separate migration.')
-
-  def _validate_parent(self) -> None:
-    """Verifies that the parent table information is valid."""
-    parent_primary_keys = self._model.interleaved.primary_keys
-    primary_keys = self._model.primary_keys
-
-    message = 'Table {} is not a child of parent table {}'.format(
-        self._model.table, self._model.interleaved.table)
-    for parent_key, key in zip(parent_primary_keys, primary_keys):
-      if parent_key != key:
-        raise error.SpannerError(message)
-    if len(parent_primary_keys) > len(primary_keys):
-      raise error.SpannerError(message)
-
-  def _validate_primary_keys(self) -> None:
-    """Verifies that the primary key data is valid."""
-    if not self._model.primary_keys:
-      raise error.SpannerError('Table {} has no primary key'.format(
-          self._model.table))
-
-    for key in self._model.primary_keys:
-      if key not in self._model.fields:
-        raise error.SpannerError(
-            'Table {} column {} in primary key but not in schema'.format(
-                self._model.table, key))
-
 
 class DropTable(SchemaUpdate):
   """Update for dropping an existing table."""
diff --git a/spanner_orm/tests/migrations_emulator_test.py b/spanner_orm/tests/migrations_emulator_test.py
index 1b1140f..d1b23ad 100644
--- a/spanner_orm/tests/migrations_emulator_test.py
+++ b/spanner_orm/tests/migrations_emulator_test.py
@@ -23,6 +23,7 @@
 from absl.testing import parameterized
 import spanner_orm
 from spanner_orm.admin import metadata
+from spanner_orm import error
 from spanner_orm.tests import models
 from spanner_orm.testlib.spanner_emulator import testlib as spanner_emulator_testlib
 
@@ -118,6 +119,108 @@ def _append_migrations(self, *migrations: str) -> None:
           )),
       )
 
+  @parameterized.named_parameters(
+      dict(
+          testcase_name='no_name',
+          create_table_migration="""
+              class _Foo(spanner_orm.Model):
+                key = spanner_orm.Field(spanner_orm.String, primary_key=True)
+
+              def upgrade():
+                return spanner_orm.CreateTable(_Foo)
+          """,
+          error_class=google_api_exceptions.InvalidArgument,
+      ),
+      dict(
+          testcase_name='no_primary_key',
+          create_table_migration="""
+              class _Foo(spanner_orm.Model):
+                __table__ = 'Foo'
+                value = spanner_orm.Field(spanner_orm.String)
+
+              def upgrade():
+                return spanner_orm.CreateTable(_Foo)
+          """,
+          error_class=error.ValidationError,
+          error_regex=r'An index must have at least one column',
+      ),
+      dict(
+          testcase_name='already_exists',
+          initial_migrations=("""
+              class _Foo(spanner_orm.Model):
+                __table__ = 'Foo'
+                key = spanner_orm.Field(spanner_orm.String, primary_key=True)
+
+              def upgrade():
+                return spanner_orm.CreateTable(_Foo)
+          """,),
+          create_table_migration="""
+              class _Foo(spanner_orm.Model):
+                __table__ = 'Foo'
+                key = spanner_orm.Field(spanner_orm.String, primary_key=True)
+
+              def upgrade():
+                return spanner_orm.CreateTable(_Foo)
+          """,
+          error_class=google_api_exceptions.FailedPrecondition,
+      ),
+      dict(
+          testcase_name='has_secondary_index',
+          create_table_migration="""
+              class _Foo(spanner_orm.Model):
+                __table__ = 'Foo'
+                key = spanner_orm.Field(spanner_orm.String, primary_key=True)
+                value = spanner_orm.Field(spanner_orm.String)
+                value_index = spanner_orm.Index(['value'])
+
+              def upgrade():
+                return spanner_orm.CreateTable(_Foo)
+          """,
+          error_class=RuntimeError,  # DO NOT MERGE
+      ),
+      dict(
+          testcase_name='mismatches_parent_primary_keys',
+          initial_migrations=("""
+              class _Parent(spanner_orm.Model):
+                __table__ = 'Parent'
+                parent_key = spanner_orm.Field(
+                    spanner_orm.String, primary_key=True)
+
+              def upgrade():
+                return spanner_orm.CreateTable(_Parent)
+          """,),
+          create_table_migration="""
+              class _Parent(spanner_orm.Model):
+                __table__ = 'Parent'
+                parent_key = spanner_orm.Field(
+                    spanner_orm.String, primary_key=True)
+
+              class _Child(spanner_orm.Model):
+                __table__ = 'Child'
+                __interleaved__ = _Parent
+                child_key = spanner_orm.Field(
+                    spanner_orm.String, primary_key=True)
+
+              def upgrade():
+                return spanner_orm.CreateTable(_Child)
+          """,
+          error_class=google_api_exceptions.FailedPrecondition,
+      ),
+  )
+  def test_create_table_error(
+      self,
+      *,
+      initial_migrations: Iterable[str] = (),
+      create_table_migration: str,
+      error_class: Type[Exception],
+      error_regex: str = '',
+  ):
+    self._append_migrations(*initial_migrations)
+    self.run_orm_migrations(self._migrations_dir)
+    self._append_migrations(create_table_migration)
+    with self.assertRaises(error_class):
+      self.run_orm_migrations(self._migrations_dir)
+
   def test_drop_interleaved_table(self):
     self._append_migrations(
         """