diff --git a/singer_sdk/streams/sql.py b/singer_sdk/streams/sql.py
index 7a3fe3886..d5fb52219 100644
--- a/singer_sdk/streams/sql.py
+++ b/singer_sdk/streams/sql.py
@@ -5,8 +5,6 @@
 import abc
 import typing as t
 
-import sqlalchemy
-
 import singer_sdk.helpers._catalog as catalog
 from singer_sdk._singerlib import CatalogEntry, MetadataMapping
 from singer_sdk.connectors import SQLConnector
@@ -195,12 +193,7 @@ def get_records(self, context: dict | None) -> t.Iterable[dict[str, t.Any]]:
 
             start_val = self.get_starting_replication_key_value(context)
             if start_val:
-                query = query.where(
-                    sqlalchemy.text(":replication_key >= :start_val").bindparams(
-                        replication_key=replication_key_col,
-                        start_val=start_val,
-                    ),
-                )
+                query = query.where(replication_key_col >= start_val)
 
         if self.ABORT_AT_RECORD_COUNT is not None:
             # Limit record count to one greater than the abort threshold. This ensures
diff --git a/tests/samples/conftest.py b/tests/samples/conftest.py
index e101d1845..60f277412 100644
--- a/tests/samples/conftest.py
+++ b/tests/samples/conftest.py
@@ -34,7 +34,11 @@ def _sqlite_sample_db(sqlite_connector):
 
 
 @pytest.fixture
-def sqlite_sample_tap(_sqlite_sample_db, sqlite_sample_db_config) -> SQLiteTap:
+def sqlite_sample_tap(
+    _sqlite_sample_db,
+    sqlite_sample_db_config,
+    sqlite_sample_db_state,
+) -> SQLiteTap:
     _ = _sqlite_sample_db
     catalog_obj = Catalog.from_dict(
         _get_tap_catalog(SQLiteTap, config=sqlite_sample_db_config, select_all=True),
@@ -51,7 +55,11 @@ def sqlite_sample_tap(_sqlite_sample_db, sqlite_sample_db_config) -> SQLiteTap:
     t2.key_properties = ["c1"]
     t2.replication_key = "c1"
     t2.replication_method = "INCREMENTAL"
-    return SQLiteTap(config=sqlite_sample_db_config, catalog=catalog_obj.to_dict())
+    return SQLiteTap(
+        config=sqlite_sample_db_config,
+        catalog=catalog_obj.to_dict(),
+        state=sqlite_sample_db_state,
+    )
 
 
 @pytest.fixture
@@ -68,3 +76,13 @@ def path_to_sample_data_db(tmp_path: Path) -> Path:
 def sqlite_sample_db_config(path_to_sample_data_db: str) -> dict:
     """Get configuration dictionary for target-csv."""
     return {"path_to_db": str(path_to_sample_data_db)}
+
+
+@pytest.fixture
+def sqlite_sample_db_state() -> dict:
+    """Get configuration dictionary for target-csv."""
+    return {
+        "bookmarks": {
+            "main-t0": {"replication_key": "c1", "replication_key_value": 55},
+        },
+    }