``Identity`` implies ``nullable=False``.

Altered the behavior of the :class:`_schema.Identity` construct such that
when applied to a :class:`_schema.Column`, it will automatically imply that
the value of :paramref:`_sql.Column.nullable` should default to ``False``,
in a similar manner as when the :paramref:`_sql.Column.primary_key`
parameter is set to ``True``.   This matches the default behavior of all
supporting databases where ``IDENTITY`` implies ``NOT NULL``.  The
PostgreSQL backend is the only one that supports adding ``NULL`` to an
``IDENTITY`` column, which is here supported by passing a ``True`` value
for the :paramref:`_sql.Column.nullable` parameter at the same time.

Fixes: #5775
Change-Id: I0516d506ff327cff35cda605e8897a27440e0373

2 files changed, 58 insertions, 3 deletions

diff --git a/test/sql/test_identity_column.py b/test/sql/test_identity_column.py
index 3ac97db92..1ce15f38c 100644
--- a/test/sql/test_identity_column.py
+++ b/test/sql/test_identity_column.py
@@ -102,7 +102,7 @@ class _IdentityDDLFixture(testing.AssertsCompiledSQL):
             CreateTable(t),
             "CREATE TABLE foo_table ("
             "foo INTEGER GENERATED ALWAYS AS IDENTITY (START "
-            "WITH 3) NOT NULL, UNIQUE (foo))",
+            "WITH 3), UNIQUE (foo))",
         )
 
     def test_autoincrement_true(self):
@@ -120,10 +120,41 @@ class _IdentityDDLFixture(testing.AssertsCompiledSQL):
         self.assert_compile(
             CreateTable(t),
             "CREATE TABLE foo_table ("
-            "foo INTEGER GENERATED ALWAYS AS IDENTITY (START WITH 3) NOT NULL"
+            "foo INTEGER GENERATED ALWAYS AS IDENTITY (START WITH 3)"
             ", PRIMARY KEY (foo))",
         )
 
+    def test_nullable_kwarg(self):
+        t = Table(
+            "t",
+            MetaData(),
+            Column("a", Integer(), Identity(), nullable=False),
+            Column("b", Integer(), Identity(), nullable=True),
+            Column("c", Integer(), Identity()),
+        )
+
+        is_(t.c.a.nullable, False)
+        is_(t.c.b.nullable, True)
+        is_(t.c.c.nullable, False)
+
+        nullable = ""
+        if getattr(self, "__dialect__", None) != "default" and testing.against(
+            "postgresql"
+        ):
+            nullable = " NULL"
+
+        self.assert_compile(
+            CreateTable(t),
+            (
+                "CREATE TABLE t ("
+                "a INTEGER GENERATED BY DEFAULT AS IDENTITY, "
+                "b INTEGER GENERATED BY DEFAULT AS IDENTITY%s, "
+                "c INTEGER GENERATED BY DEFAULT AS IDENTITY"
+                ")"
+            )
+            % nullable,
+        )
+
 
 class IdentityDDL(_IdentityDDLFixture, fixtures.TestBase):
     # this uses the connection dialect
@@ -167,7 +198,7 @@ class NotSupportingIdentityDDL(testing.AssertsCompiledSQL, fixtures.TestBase):
             Column("foo", Integer(), Identity("always", start=3)),
         )
         self.assert_compile(
-            CreateTable(t), "CREATE TABLE foo_table (foo INTEGER)"
+            CreateTable(t), "CREATE TABLE foo_table (foo INTEGER NOT NULL)"
         )
 
 
diff --git a/test/sql/test_metadata.py b/test/sql/test_metadata.py
index 9bf351f5c..17228dad6 100644
--- a/test/sql/test_metadata.py
+++ b/test/sql/test_metadata.py
@@ -1838,6 +1838,30 @@ class TableTest(fixtures.TestBase, AssertsCompiledSQL):
         assert not t2.c.x.nullable
         assert not t1.c.x.nullable
 
+    def test_pk_can_be_nullable(self):
+        m = MetaData()
+
+        t1 = Table(
+            "t1",
+            m,
+            Column("x", Integer, nullable=True),
+            PrimaryKeyConstraint("x"),
+        )
+
+        t2 = Table(
+            "t2", m, Column("x", Integer, primary_key=True, nullable=True)
+        )
+
+        eq_(list(t1.primary_key), [t1.c.x])
+
+        eq_(list(t2.primary_key), [t2.c.x])
+
+        assert t1.c.x.primary_key
+        assert t2.c.x.primary_key
+
+        assert t2.c.x.nullable
+        assert t1.c.x.nullable
+
     def test_must_exist(self):
         with testing.expect_raises_message(
             exc.InvalidRequestError, "Table 'foo' not defined"