fix(checkers): cover identity columns in column_value_at_risk + seque…

src/pgsleuth/checkers/column_value_at_risk.py

@@ -10,6 +10,11 @@
 `bigserial`, `GENERATED ... AS IDENTITY`, or a manually-OWNED-BY sequence)
 qualifies. Non-PK serial-backed columns can overflow the same way; the type
 of column doesn't matter to the underlying failure mode.
+
+The pg_depend filter accepts deptype IN ('a', 'i', 'I'): SERIAL and manual
+OWNED BY use 'a' (AUTO); identity columns use 'i' (INTERNAL, PG10-11) or
+'I' (INTERNAL_AUTO, PG12+). Filtering only on 'a' silently drops every
+identity column, which is the modern recommended replacement for SERIAL.
 """
 
 from __future__ import annotations
@@ -38,7 +43,7 @@
 JOIN pg_namespace   ns_seq ON ns_seq.oid = seq.relnamespace
 JOIN pg_depend      dep    ON dep.objid = seq.oid
                           AND dep.classid = 'pg_class'::regclass
-                          AND dep.deptype = 'a'
+                          AND dep.deptype IN ('a', 'i', 'I')
 JOIN pg_class       tbl    ON tbl.oid = dep.refobjid
 JOIN pg_namespace   ns_tbl ON ns_tbl.oid = tbl.relnamespace
 JOIN pg_attribute   col    ON col.attrelid = tbl.oid

src/pgsleuth/checkers/sequence_drift.py

@@ -38,7 +38,8 @@
 JOIN pg_namespace   tn  ON tn.oid = t.relnamespace
 JOIN pg_attribute   a   ON a.attrelid = t.oid AND a.attnum = d.refobjsubid
 WHERE s.relkind = 'S'
-  AND d.deptype = 'a'
+  -- 'a' = SERIAL / manual OWNED BY; 'i' = identity on PG10-11; 'I' = identity on PG12+.
+  AND d.deptype IN ('a', 'i', 'I')
   AND t.relkind IN ('r', 'p')
   {schema_filter}
 ORDER BY sn.nspname, s.relname;

tests/checkers/test_column_value_at_risk.py

@@ -71,3 +71,32 @@ def test_clean_for_bigint_at_realistic_usage(ctx, conn, schema):
 
     issues = [i for i in ColumnValueAtRisk().run(ctx) if i.object_name.startswith(schema)]
     assert issues == []
+
+
+def test_flags_identity_column_near_max(ctx, conn, schema):
+    """GENERATED ... AS IDENTITY columns use a different pg_depend.deptype
+    ('i' on PG10-11, 'I' on PG12+) than SERIAL ('a'). Both must be picked up.
+    """
+    with conn.cursor() as cur:
+        cur.execute("CREATE TABLE t (id integer GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY)")
+        cur.execute(f"SELECT pg_get_serial_sequence('{schema}.t', 'id')")
+        seq = cur.fetchone()[0]
+        cur.execute(f"SELECT setval('{seq}', 1700000000)")  # ~79% of int4 max
+
+    issues = [i for i in ColumnValueAtRisk().run(ctx) if i.object_name.startswith(schema)]
+    assert len(issues) == 1
+    assert issues[0].object_name.endswith(".t.id")
+    assert float(issues[0].extra["ratio"]) > 0.70
+
+
+def test_flags_generated_always_identity(ctx, conn, schema):
+    """GENERATED ALWAYS uses the same dependency machinery as BY DEFAULT."""
+    with conn.cursor() as cur:
+        cur.execute("CREATE TABLE t (id integer GENERATED ALWAYS AS IDENTITY PRIMARY KEY)")
+        cur.execute(f"SELECT pg_get_serial_sequence('{schema}.t', 'id')")
+        seq = cur.fetchone()[0]
+        cur.execute(f"SELECT setval('{seq}', 1700000000)")
+
+    issues = [i for i in ColumnValueAtRisk().run(ctx) if i.object_name.startswith(schema)]
+    assert len(issues) == 1
+    assert issues[0].object_name.endswith(".t.id")

tests/checkers/test_sequence_drift.py

@@ -23,3 +23,22 @@ def test_flags_sequence_drift(ctx, conn, schema):
     assert "t_id_seq" in issues[0].object_name
     assert issues[0].extra["max_value"] == "1001"
     assert int(issues[0].extra["next_value"]) <= 1001
+
+
+def test_flags_drift_on_identity_column(ctx, conn, schema):
+    """Identity columns use pg_depend.deptype 'i'/'I' (not 'a' like SERIAL).
+    Drift on them must still be flagged.
+    """
+    with conn.cursor() as cur:
+        cur.execute(
+            "CREATE TABLE t (id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, name text)"
+        )
+        cur.execute(f"SELECT pg_get_serial_sequence('{schema}.t', 'id')")
+        seq = cur.fetchone()[0]
+        cur.execute("INSERT INTO t (id, name) VALUES (1000, 'a'), (1001, 'b')")
+        cur.execute(f"SELECT setval('{seq}', 1, false)")
+
+    issues = [i for i in SequenceDrift().run(ctx) if i.object_name.startswith(schema)]
+    assert len(issues) == 1
+    assert issues[0].extra["max_value"] == "1001"
+    assert int(issues[0].extra["next_value"]) <= 1001