diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/StatementFactory.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/StatementFactory.java
index ae1ac73fa..c5f2ff5f6 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/StatementFactory.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/StatementFactory.java
@@ -632,17 +632,19 @@ public class StatementFactory {
 
 		StatementBuilder<Select> select = createSelectAndOrder(selectors, tableName, filter, sort);
 
-		// TODO: Bind marker
-		if (query.getLimit() > 0) {
-			select.apply(it -> it.limit(Math.toIntExact(query.getLimit())));
-		}
-
 		if (query.isAllowFiltering()) {
 			select.apply(Select::allowFiltering);
 		}
 
 		select.onBuild(statementBuilder -> query.getPagingState().ifPresent(statementBuilder::setPagingState));
 
+		if (query.getLimit() > 0) {
+
+			int limit = Math.toIntExact(query.getLimit());
+			select.bind((statement, factory) -> factory.ifBoundOrInline(bindings -> statement.limit(bindings.bind(limit)),
+					() -> statement.limit(limit)));
+		}
+
 		query.getQueryOptions()
 				.ifPresent(it -> select.transform(statement -> QueryOptionsUtil.addQueryOptions(statement, it)));
 
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/StatementFactoryUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/StatementFactoryUnitTests.java
index 9d7e0bc70..e4932b13a 100644
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/StatementFactoryUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/StatementFactoryUnitTests.java
@@ -41,6 +41,7 @@ import org.springframework.data.cassandra.core.query.Criteria;
 import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.data.cassandra.core.query.Update;
 import org.springframework.data.cassandra.domain.Group;
+import org.springframework.data.domain.Limit;
 import org.springframework.data.domain.Sort;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
@@ -136,8 +137,8 @@ class StatementFactoryUnitTests {
 		assertThat(select.build(ParameterHandling.INLINE).getQuery()).isEqualTo("SELECT ttl(email) FROM group");
 	}
 
-	@Test // #1008
-	void shouldMapSelectQueryWithSortLimitAndAllowFiltering() {
+	@Test // GH-1008
+	void shouldMapSelectQueryWithSort() {
 
 		Query query = Query.empty().sort(Sort.by(DESC, "email", "age"));
 
@@ -148,6 +149,22 @@ class StatementFactoryUnitTests {
 				.isEqualTo("SELECT * FROM group ORDER BY email DESC,age DESC");
 	}
 
+	@Test // GH-1401
+	void shouldMapSelectQueryWithLimit() {
+
+		Query query = Query.query(Criteria.where("email").is("e@mail")).limit(Limit.of(10));
+
+		StatementBuilder<Select> select = statementFactory.select(query,
+				converter.getMappingContext().getRequiredPersistentEntity(Group.class));
+
+		assertThat(select.build(ParameterHandling.INLINE).getQuery())
+				.isEqualTo("SELECT * FROM group WHERE email='e@mail' LIMIT 10");
+
+		SimpleStatement statement = select.build(ParameterHandling.BY_INDEX);
+		assertThat(statement.getQuery()).isEqualTo("SELECT * FROM group WHERE email=? LIMIT ?");
+		assertThat(statement.getPositionalValues()).containsExactly("e@mail", 10);
+	}
+
 	@Test // DATACASS-343
 	void shouldMapSelectQueryWithSortByEmbeddedLimitAndAllowFiltering() {
 
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/PartTreeCassandraQueryUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/PartTreeCassandraQueryUnitTests.java
index ad4c03be4..f4c65c78c 100644
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/PartTreeCassandraQueryUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/PartTreeCassandraQueryUnitTests.java
@@ -242,12 +242,12 @@ class PartTreeCassandraQueryUnitTests {
 		assertThat(statement.getQuery()).isEqualTo("DELETE FROM person WHERE lastname=?");
 	}
 
-	@Test // DATACASS-512
+	@Test // DATACASS-512, GH-1401
 	void shouldCreateExistsQuery() {
 
 		SimpleStatement statement = deriveQueryFromMethod(Repo.class, "existsBy", new Class[0]);
 
-		assertThat(statement.getQuery()).isEqualTo("SELECT * FROM person LIMIT 1");
+		assertThat(statement.getQuery()).isEqualTo("SELECT * FROM person LIMIT ?");
 	}
 
 	private String deriveQueryFromMethod(String method, Object... args) {
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ReactivePartTreeCassandraQueryUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ReactivePartTreeCassandraQueryUnitTests.java
index ca410135f..b2315a709 100644
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ReactivePartTreeCassandraQueryUnitTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ReactivePartTreeCassandraQueryUnitTests.java
@@ -149,13 +149,13 @@ class ReactivePartTreeCassandraQueryUnitTests {
 		assertThat(statement.getQuery()).isEqualTo("DELETE FROM person WHERE lastname=?");
 	}
 
-	@Test // DATACASS-512
+	@Test // DATACASS-512, GH-1401
 	void shouldCreateExistsQuery() {
 
 		SimpleStatement statement = deriveQueryFromMethod(PartTreeCassandraQueryUnitTests.Repo.class, "existsBy",
 				new Class[0]);
 
-		assertThat(statement.getQuery()).isEqualTo("SELECT * FROM person LIMIT 1");
+		assertThat(statement.getQuery()).isEqualTo("SELECT * FROM person LIMIT ?");
 	}
 
 	private String deriveQueryFromMethod(String method, Object... args) {