diff --git a/singer_sdk/connectors/sql.py b/singer_sdk/connectors/sql.py
index 02c8e8d16..e097a4ce9 100644
--- a/singer_sdk/connectors/sql.py
+++ b/singer_sdk/connectors/sql.py
@@ -464,7 +464,8 @@ def discover_catalog_entry(
                 th.Property(
                     name=column_name,
                     wrapped=th.CustomType(jsonschema_type),
-                    required=not is_nullable,
+                    nullable=is_nullable,
+                    required=column_name in key_properties if key_properties else False,
                 ),
             )
         schema = table_schema.to_dict()
diff --git a/singer_sdk/typing.py b/singer_sdk/typing.py
index 80e553574..7a412fe81 100644
--- a/singer_sdk/typing.py
+++ b/singer_sdk/typing.py
@@ -517,7 +517,7 @@ class Property(JSONTypeHelper[T], t.Generic[T]):
     """Generic Property. Should be nested within a `PropertiesList`."""
 
     # TODO: Make some of these arguments keyword-only. This is a breaking change.
-    def __init__(
+    def __init__(  # noqa: PLR0913
         self,
         name: str,
         wrapped: JSONTypeHelper[T] | type[JSONTypeHelper[T]],
@@ -527,6 +527,8 @@ def __init__(
         secret: bool | None = False,  # noqa: FBT002
         allowed_values: list[T] | None = None,
         examples: list[T] | None = None,
+        *,
+        nullable: bool | None = None,
     ) -> None:
         """Initialize Property object.
 
@@ -547,6 +549,7 @@ def __init__(
                 are permitted. This will define the type as an 'enum'.
             examples: Optional. A list of one or more sample values. These may be
                 displayed to the user as hints of the expected format of inputs.
+            nullable: If True, the property may be null.
         """
         self.name = name
         self.wrapped = wrapped
@@ -556,6 +559,7 @@ def __init__(
         self.secret = secret
         self.allowed_values = allowed_values or None
         self.examples = examples or None
+        self.nullable = nullable
 
     @property
     def type_dict(self) -> dict:  # type: ignore[override]
@@ -585,7 +589,7 @@ def to_dict(self) -> dict:
             A JSON Schema dictionary describing the object.
         """
         type_dict = self.type_dict
-        if self.optional:
+        if self.nullable or self.optional:
             type_dict = append_type(type_dict, "null")
         if self.default is not None:
             type_dict.update({"default": self.default})
diff --git a/tests/samples/conftest.py b/tests/samples/conftest.py
index b9ce33319..52c0857be 100644
--- a/tests/samples/conftest.py
+++ b/tests/samples/conftest.py
@@ -25,7 +25,14 @@ def _sqlite_sample_db(sqlite_connector):
         for t in range(3):
             conn.execute(sa.text(f"DROP TABLE IF EXISTS t{t}"))
             conn.execute(
-                sa.text(f"CREATE TABLE t{t} (c1 int PRIMARY KEY, c2 varchar(10))"),
+                sa.text(
+                    f"""
+                    CREATE TABLE t{t} (
+                        c1 int PRIMARY KEY NOT NULL,
+                        c2 varchar(10) NOT NULL
+                    )
+                    """
+                ),
             )
             for x in range(100):
                 conn.execute(
diff --git a/tests/samples/test_tap_sqlite.py b/tests/samples/test_tap_sqlite.py
index b5ed7b549..2c1094b75 100644
--- a/tests/samples/test_tap_sqlite.py
+++ b/tests/samples/test_tap_sqlite.py
@@ -80,6 +80,8 @@ def test_sqlite_discovery(sqlite_sample_tap: SQLTap):
 
     assert stream.metadata.root.table_key_properties == ["c1"]
     assert stream.primary_keys == ["c1"]
+    assert stream.schema["properties"]["c1"] == {"type": ["integer"]}
+    assert stream.schema["required"] == ["c1"]
 
 
 def test_sqlite_input_catalog(sqlite_sample_tap: SQLTap):
@@ -90,7 +92,7 @@ def test_sqlite_input_catalog(sqlite_sample_tap: SQLTap):
 
     for schema in [stream.schema, stream.stream_maps[0].transformed_schema]:
         assert len(schema["properties"]) == 2
-        assert schema["properties"]["c1"] == {"type": ["integer", "null"]}
+        assert schema["properties"]["c1"] == {"type": ["integer"]}
         assert schema["properties"]["c2"] == {"type": ["string", "null"]}
         assert stream.name == stream.tap_stream_id == "main-t1"