diff --git a/core/src/main/java/org/opensearch/sql/ast/expression/Cast.java b/core/src/main/java/org/opensearch/sql/ast/expression/Cast.java
index 3acbe68d45..9121dbd87c 100644
--- a/core/src/main/java/org/opensearch/sql/ast/expression/Cast.java
+++ b/core/src/main/java/org/opensearch/sql/ast/expression/Cast.java
@@ -65,7 +65,7 @@ public class Cast extends UnresolvedExpression {
   private final UnresolvedExpression expression;
 
   /**
-   * Expression that represents ELSE statement result.
+   * Expression that represents name of the target type.
    */
   private final UnresolvedExpression convertedType;
 
diff --git a/core/src/main/java/org/opensearch/sql/executor/pagination/CanPaginateVisitor.java b/core/src/main/java/org/opensearch/sql/executor/pagination/CanPaginateVisitor.java
index 3164794abb..1256ef0b26 100644
--- a/core/src/main/java/org/opensearch/sql/executor/pagination/CanPaginateVisitor.java
+++ b/core/src/main/java/org/opensearch/sql/executor/pagination/CanPaginateVisitor.java
@@ -7,21 +7,53 @@
 
 import org.opensearch.sql.ast.AbstractNodeVisitor;
 import org.opensearch.sql.ast.Node;
+import org.opensearch.sql.ast.expression.Alias;
 import org.opensearch.sql.ast.expression.AllFields;
+import org.opensearch.sql.ast.expression.And;
+import org.opensearch.sql.ast.expression.Argument;
+import org.opensearch.sql.ast.expression.Between;
+import org.opensearch.sql.ast.expression.Case;
+import org.opensearch.sql.ast.expression.Cast;
+import org.opensearch.sql.ast.expression.Compare;
+import org.opensearch.sql.ast.expression.EqualTo;
+import org.opensearch.sql.ast.expression.Field;
+import org.opensearch.sql.ast.expression.Function;
+import org.opensearch.sql.ast.expression.HighlightFunction;
+import org.opensearch.sql.ast.expression.In;
+import org.opensearch.sql.ast.expression.Interval;
+import org.opensearch.sql.ast.expression.Literal;
+import org.opensearch.sql.ast.expression.Not;
+import org.opensearch.sql.ast.expression.Or;
+import org.opensearch.sql.ast.expression.QualifiedName;
+import org.opensearch.sql.ast.expression.RelevanceFieldList;
+import org.opensearch.sql.ast.expression.UnresolvedArgument;
+import org.opensearch.sql.ast.expression.UnresolvedAttribute;
+import org.opensearch.sql.ast.expression.When;
+import org.opensearch.sql.ast.expression.WindowFunction;
+import org.opensearch.sql.ast.expression.Xor;
+import org.opensearch.sql.ast.tree.Aggregation;
+import org.opensearch.sql.ast.tree.Filter;
+import org.opensearch.sql.ast.tree.Limit;
 import org.opensearch.sql.ast.tree.Project;
 import org.opensearch.sql.ast.tree.Relation;
+import org.opensearch.sql.ast.tree.Sort;
+import org.opensearch.sql.ast.tree.Values;
+import org.opensearch.sql.expression.function.BuiltinFunctionName;
 
 /**
  * Use this unresolved plan visitor to check if a plan can be serialized by PaginatedPlanCache.
- * If plan.accept(new CanPaginateVisitor(...)) returns true,
+ * If <pre>plan.accept(new CanPaginateVisitor(...))</pre> returns <em>true</em>,
  * then PaginatedPlanCache.convertToCursor will succeed. Otherwise, it will fail.
  * The purpose of this visitor is to activate legacy engine fallback mechanism.
- * Currently, the conditions are:
- * - only projection of a relation is supported.
- * - projection only has * (a.k.a. allFields).
- * - Relation only scans one table
- * - The table is an open search index.
- * So it accepts only queries like `select * from $index`
+ * Currently, V2 engine does not support queries with:
+ * - aggregation (GROUP BY clause or aggregation functions like min/max)
+ * - in memory aggregation (window function)
+ * - ORDER BY clause
+ * - LIMIT/OFFSET clause(s)
+ * - without FROM clause
+ * - JOIN
+ * - a subquery
+ * V2 also requires that the table being queried should be an OpenSearch index.
  * See PaginatedPlanCache.canConvertToCursor for usage.
  */
 public class CanPaginateVisitor extends AbstractNodeVisitor<Boolean, Object> {
@@ -36,22 +68,182 @@ public Boolean visitRelation(Relation node, Object context) {
     return Boolean.TRUE;
   }
 
+  private Boolean canPaginate(Node node, Object context) {
+    var childList = node.getChild();
+    if (childList != null) {
+      return childList.stream().allMatch(n -> n.accept(this, context));
+    }
+    return Boolean.TRUE;
+  }
+
+  // For queries with WHERE clause:
   @Override
-  public Boolean visitChildren(Node node, Object context) {
+  public Boolean visitFilter(Filter node, Object context) {
+    return canPaginate(node, context) && node.getCondition().accept(this, context);
+  }
+
+  // Queries with GROUP BY clause are not supported
+  @Override
+  public Boolean visitAggregation(Aggregation node, Object context) {
     return Boolean.FALSE;
   }
 
+  // Queries with ORDER BY clause are not supported
   @Override
-  public Boolean visitProject(Project node, Object context) {
-    // Allow queries with 'SELECT *' only. Those restriction could be removed, but consider
-    // in-memory aggregation performed by window function (see WindowOperator).
-    // SELECT max(age) OVER (PARTITION BY city) ...
-    var projections = node.getProjectList();
-    if (projections.size() != 1) {
+  public Boolean visitSort(Sort node, Object context) {
+    return Boolean.FALSE;
+  }
+
+  // Queries without FROM clause are not supported
+  @Override
+  public Boolean visitValues(Values node, Object context) {
+    return Boolean.FALSE;
+  }
+
+  // Queries with LIMIT clause are not supported
+  @Override
+  public Boolean visitLimit(Limit node, Object context) {
+    return Boolean.FALSE;
+  }
+
+  @Override
+  public Boolean visitLiteral(Literal node, Object context) {
+    return canPaginate(node, context);
+  }
+
+  @Override
+  public Boolean visitField(Field node, Object context) {
+    return canPaginate(node, context) && node.getFieldArgs().stream()
+        .allMatch(n -> n.accept(this, context));
+  }
+
+  @Override
+  public Boolean visitAlias(Alias node, Object context) {
+    return canPaginate(node, context) && node.getDelegated().accept(this, context);
+  }
+
+  @Override
+  public Boolean visitAllFields(AllFields node, Object context) {
+    return canPaginate(node, context);
+  }
+
+  @Override
+  public Boolean visitQualifiedName(QualifiedName node, Object context) {
+    return canPaginate(node, context);
+  }
+
+  @Override
+  public Boolean visitEqualTo(EqualTo node, Object context) {
+    return canPaginate(node, context);
+  }
+
+  @Override
+  public Boolean visitRelevanceFieldList(RelevanceFieldList node, Object context) {
+    return canPaginate(node, context);
+  }
+
+  @Override
+  public Boolean visitInterval(Interval node, Object context) {
+    return canPaginate(node, context);
+  }
+
+  @Override
+  public Boolean visitCompare(Compare node, Object context) {
+    return canPaginate(node, context);
+  }
+
+  @Override
+  public Boolean visitNot(Not node, Object context) {
+    return canPaginate(node, context);
+  }
+
+  @Override
+  public Boolean visitOr(Or node, Object context) {
+    return canPaginate(node, context);
+  }
+
+  @Override
+  public Boolean visitAnd(And node, Object context) {
+    return canPaginate(node, context);
+  }
+
+  @Override
+  public Boolean visitArgument(Argument node, Object context) {
+    return canPaginate(node, context);
+  }
+
+  @Override
+  public Boolean visitXor(Xor node, Object context) {
+    return canPaginate(node, context);
+  }
+
+  @Override
+  public Boolean visitFunction(Function node, Object context) {
+    // https://github.com/opensearch-project/sql/issues/1718
+    if (node.getFuncName()
+        .equalsIgnoreCase(BuiltinFunctionName.NESTED.getName().getFunctionName())) {
       return Boolean.FALSE;
     }
+    return canPaginate(node, context);
+  }
+
+  @Override
+  public Boolean visitIn(In node, Object context) {
+    return canPaginate(node, context) && node.getValueList().stream()
+        .allMatch(n -> n.accept(this, context));
+  }
+
+  @Override
+  public Boolean visitBetween(Between node, Object context) {
+    return canPaginate(node, context);
+  }
+
+  @Override
+  public Boolean visitCase(Case node, Object context) {
+    return canPaginate(node, context);
+  }
+
+  @Override
+  public Boolean visitWhen(When node, Object context) {
+    return canPaginate(node, context);
+  }
+
+  @Override
+  public Boolean visitCast(Cast node, Object context) {
+    return canPaginate(node, context) && node.getConvertedType().accept(this, context);
+  }
+
+  @Override
+  public Boolean visitHighlightFunction(HighlightFunction node, Object context) {
+    return canPaginate(node, context);
+  }
+
+  @Override
+  public Boolean visitUnresolvedArgument(UnresolvedArgument node, Object context) {
+    return canPaginate(node, context);
+  }
+
+  @Override
+  public Boolean visitUnresolvedAttribute(UnresolvedAttribute node, Object context) {
+    return canPaginate(node, context);
+  }
 
-    if (!(projections.get(0) instanceof AllFields)) {
+  @Override
+  public Boolean visitChildren(Node node, Object context) {
+    // for all not listed (= unchecked) - false
+    return Boolean.FALSE;
+  }
+
+  @Override
+  public Boolean visitWindowFunction(WindowFunction node, Object context) {
+    // don't support in-memory aggregation
+    // SELECT max(age) OVER (PARTITION BY city) ...
+    return Boolean.FALSE;
+  }
+
+  @Override
+  public Boolean visitProject(Project node, Object context) {
+    if (!node.getProjectList().stream().allMatch(n -> n.accept(this, context))) {
       return Boolean.FALSE;
     }
 
diff --git a/core/src/test/java/org/opensearch/sql/executor/pagination/CanPaginateVisitorTest.java b/core/src/test/java/org/opensearch/sql/executor/pagination/CanPaginateVisitorTest.java
index 02a0dbc05e..2535d1cb7a 100644
--- a/core/src/test/java/org/opensearch/sql/executor/pagination/CanPaginateVisitorTest.java
+++ b/core/src/test/java/org/opensearch/sql/executor/pagination/CanPaginateVisitorTest.java
@@ -5,20 +5,63 @@
 
 package org.opensearch.sql.executor.pagination;
 
+import static org.junit.jupiter.api.Assertions.assertAll;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 import static org.mockito.Mockito.withSettings;
+import static org.opensearch.sql.ast.dsl.AstDSL.agg;
+import static org.opensearch.sql.ast.dsl.AstDSL.aggregate;
+import static org.opensearch.sql.ast.dsl.AstDSL.alias;
+import static org.opensearch.sql.ast.dsl.AstDSL.allFields;
+import static org.opensearch.sql.ast.dsl.AstDSL.and;
+import static org.opensearch.sql.ast.dsl.AstDSL.argument;
+import static org.opensearch.sql.ast.dsl.AstDSL.between;
+import static org.opensearch.sql.ast.dsl.AstDSL.booleanLiteral;
+import static org.opensearch.sql.ast.dsl.AstDSL.caseWhen;
+import static org.opensearch.sql.ast.dsl.AstDSL.cast;
+import static org.opensearch.sql.ast.dsl.AstDSL.compare;
+import static org.opensearch.sql.ast.dsl.AstDSL.equalTo;
+import static org.opensearch.sql.ast.dsl.AstDSL.eval;
+import static org.opensearch.sql.ast.dsl.AstDSL.field;
+import static org.opensearch.sql.ast.dsl.AstDSL.filter;
+import static org.opensearch.sql.ast.dsl.AstDSL.function;
+import static org.opensearch.sql.ast.dsl.AstDSL.highlight;
+import static org.opensearch.sql.ast.dsl.AstDSL.in;
+import static org.opensearch.sql.ast.dsl.AstDSL.intLiteral;
+import static org.opensearch.sql.ast.dsl.AstDSL.intervalLiteral;
+import static org.opensearch.sql.ast.dsl.AstDSL.limit;
+import static org.opensearch.sql.ast.dsl.AstDSL.map;
+import static org.opensearch.sql.ast.dsl.AstDSL.not;
+import static org.opensearch.sql.ast.dsl.AstDSL.or;
+import static org.opensearch.sql.ast.dsl.AstDSL.project;
+import static org.opensearch.sql.ast.dsl.AstDSL.qualifiedName;
+import static org.opensearch.sql.ast.dsl.AstDSL.relation;
+import static org.opensearch.sql.ast.dsl.AstDSL.sort;
+import static org.opensearch.sql.ast.dsl.AstDSL.stringLiteral;
+import static org.opensearch.sql.ast.dsl.AstDSL.tableFunction;
+import static org.opensearch.sql.ast.dsl.AstDSL.unresolvedArg;
+import static org.opensearch.sql.ast.dsl.AstDSL.unresolvedAttr;
+import static org.opensearch.sql.ast.dsl.AstDSL.values;
+import static org.opensearch.sql.ast.dsl.AstDSL.when;
+import static org.opensearch.sql.ast.dsl.AstDSL.window;
+import static org.opensearch.sql.ast.dsl.AstDSL.xor;
 
 import java.util.List;
+import java.util.Map;
 import org.junit.jupiter.api.DisplayNameGeneration;
 import org.junit.jupiter.api.DisplayNameGenerator;
 import org.junit.jupiter.api.Test;
-import org.opensearch.sql.ast.dsl.AstDSL;
+import org.opensearch.sql.ast.Node;
+import org.opensearch.sql.ast.expression.Alias;
+import org.opensearch.sql.ast.expression.Argument;
+import org.opensearch.sql.ast.expression.DataType;
+import org.opensearch.sql.ast.expression.Literal;
+import org.opensearch.sql.ast.expression.RelevanceFieldList;
+import org.opensearch.sql.ast.expression.UnresolvedExpression;
 import org.opensearch.sql.ast.tree.Project;
 import org.opensearch.sql.ast.tree.Relation;
-import org.opensearch.sql.executor.pagination.CanPaginateVisitor;
 
 @DisplayNameGeneration(DisplayNameGenerator.ReplaceUnderscores.class)
 public class CanPaginateVisitorTest {
@@ -28,84 +71,307 @@ public class CanPaginateVisitorTest {
   @Test
   // select * from y
   public void accept_query_with_select_star_and_from() {
-    var plan = AstDSL.project(AstDSL.relation("dummy"), AstDSL.allFields());
+    var plan = project(relation("dummy"), allFields());
     assertTrue(plan.accept(visitor, null));
   }
 
   @Test
   // select x from y
-  public void reject_query_with_select_field_and_from() {
-    var plan = AstDSL.project(AstDSL.relation("dummy"), AstDSL.field("pewpew"));
-    assertFalse(plan.accept(visitor, null));
+  public void allow_query_with_select_field_and_from() {
+    var plan = project(relation("dummy"), field("pewpew"));
+    assertTrue(plan.accept(visitor, null));
+  }
+
+  @Test
+  // select x from y
+  public void visitUnresolvedAttribute() {
+    var plan = project(relation("dummy"), unresolvedAttr("pewpew"));
+    assertTrue(plan.accept(visitor, null));
+  }
+
+  @Test
+  // select x as z from y
+  public void allow_query_with_select_alias_and_from() {
+    var plan = project(relation("dummy"), alias("pew", field("pewpew"), "pew"));
+    assertTrue(plan.accept(visitor, null));
+  }
+
+  @Test
+  // select N from y
+  public void allow_query_with_select_literal_and_from() {
+    var plan = project(relation("dummy"), intLiteral(42));
+    assertTrue(plan.accept(visitor, null));
+  }
+
+  @Test
+  // select x.z from y
+  public void allow_query_with_select_qn_and_from() {
+    var plan = project(relation("dummy"), qualifiedName("field.subfield"));
+    assertTrue(plan.accept(visitor, null));
   }
 
   @Test
   // select x,z from y
-  public void reject_query_with_select_fields_and_from() {
-    var plan = AstDSL.project(AstDSL.relation("dummy"),
-        AstDSL.field("pewpew"), AstDSL.field("pewpew"));
-    assertFalse(plan.accept(visitor, null));
+  public void allow_query_with_select_fields_and_from() {
+    var plan = project(relation("dummy"), field("pewpew"), field("pewpew"));
+    assertTrue(plan.accept(visitor, null));
   }
 
   @Test
   // select x
   public void reject_query_without_from() {
-    var plan = AstDSL.project(AstDSL.values(List.of(AstDSL.intLiteral(1))),
-        AstDSL.alias("1",AstDSL.intLiteral(1)));
+    var plan = project(values(List.of(intLiteral(1))),
+        alias("1", intLiteral(1)));
     assertFalse(plan.accept(visitor, null));
   }
 
+  @Test
+  public void visitField() {
+    // test combinations of acceptable and not acceptable args for coverage
+    assertAll(
+        () -> assertFalse(project(relation("dummy"),
+                field(map("1", "2"), argument("name", intLiteral(0))))
+            .accept(visitor, null)),
+        () -> assertFalse(project(relation("dummy"),
+                field("field", new Argument("", new Literal(1, DataType.INTEGER) {
+                    @Override
+                    public List<UnresolvedExpression> getChild() {
+                      return List.of(map("1", "2"));
+                    }
+                })))
+            .accept(visitor, null))
+    );
+  }
+
+  @Test
+  public void visitAlias() {
+    // test combinations of acceptable and not acceptable args for coverage
+    assertAll(
+        () -> assertFalse(project(relation("dummy"),
+                alias("pew", map("1", "2"), "pew"))
+            .accept(visitor, null)),
+        () -> assertFalse(project(relation("dummy"), new Alias("pew", field("pew")) {
+              @Override
+              public List<? extends Node> getChild() {
+                return List.of(map("1", "2"));
+              }
+            })
+            .accept(visitor, null))
+    );
+  }
+
+  @Test
+  // select a = b
+  public void visitEqualTo() {
+    var plan = project(values(List.of(intLiteral(1))),
+        alias("1", equalTo(intLiteral(1), intLiteral(1))));
+    assertFalse(plan.accept(visitor, null));
+  }
+
+  @Test
+  // select interval
+  public void visitInterval() {
+    var plan = project(values(List.of(intLiteral(1))),
+        alias("1", intervalLiteral(intLiteral(1), DataType.INTEGER, "days")));
+    assertFalse(plan.accept(visitor, null));
+  }
+
+  @Test
+  // select a != b
+  public void visitCompare() {
+    var plan = project(values(List.of(intLiteral(1))),
+        alias("1", compare("!=", intLiteral(1), intLiteral(1))));
+    assertFalse(plan.accept(visitor, null));
+  }
+
+  @Test
+  // select NOT a
+  public void visitNot() {
+    var plan = project(values(List.of(intLiteral(1))),
+        alias("1", not(booleanLiteral(true))));
+    assertFalse(plan.accept(visitor, null));
+  }
+
+  @Test
+  // select a OR b
+  public void visitOr() {
+    var plan = project(values(List.of(intLiteral(1))),
+        alias("1", or(booleanLiteral(true), booleanLiteral(false))));
+    assertFalse(plan.accept(visitor, null));
+  }
+
+  @Test
+  // select a AND b
+  public void visitAnd() {
+    var plan = project(values(List.of(intLiteral(1))),
+        alias("1", and(booleanLiteral(true), booleanLiteral(false))));
+    assertFalse(plan.accept(visitor, null));
+  }
+
+  @Test
+  // select a XOR b
+  public void visitXor() {
+    var plan = project(values(List.of(intLiteral(1))),
+        alias("1", xor(booleanLiteral(true), booleanLiteral(false))));
+    assertFalse(plan.accept(visitor, null));
+  }
+
+  @Test
+  // select f()
+  public void visitFunction() {
+    var plan = project(values(List.of(intLiteral(1))),
+        function("func"));
+    assertFalse(plan.accept(visitor, null));
+  }
+
+  @Test
+  // select nested() ...
+  public void visitNested() {
+    var plan = project(values(List.of(intLiteral(1))),
+        function("nested"));
+    assertFalse(plan.accept(visitor, null));
+  }
+
+  @Test
+  // select a IN ()
+  public void visitIn() {
+    // test combinations of acceptable and not acceptable args for coverage
+    assertAll(
+        () -> assertFalse(project(values(List.of(intLiteral(1))), alias("1", in(field("a"))))
+            .accept(visitor, null)),
+        () -> assertFalse(project(values(List.of(intLiteral(1))),
+                alias("1", in(field("a"), map("1", "2"))))
+            .accept(visitor, null)),
+        () -> assertFalse(project(values(List.of(intLiteral(1))),
+                alias("1", in(map("1", "2"), field("a"))))
+            .accept(visitor, null))
+    );
+  }
+
+  @Test
+  // select a BETWEEN 1 AND 2
+  public void visitBetween() {
+    var plan = project(values(List.of(intLiteral(1))),
+        alias("1", between(field("a"), intLiteral(1), intLiteral(2))));
+    assertFalse(plan.accept(visitor, null));
+  }
+
+  @Test
+  // select a CASE 1 WHEN 2
+  public void visitCase() {
+    var plan = project(values(List.of(intLiteral(1))),
+        alias("1", caseWhen(intLiteral(1), when(intLiteral(3), intLiteral(4)))));
+    assertFalse(plan.accept(visitor, null));
+  }
+
+  @Test
+  // select CAST(a as TYPE)
+  public void visitCast() {
+    // test combinations of acceptable and not acceptable args for coverage
+    assertAll(
+        () -> assertFalse(project(values(List.of(intLiteral(1))),
+                alias("1", cast(intLiteral(2), stringLiteral("int"))))
+            .accept(visitor, null)),
+        () -> assertFalse(project(values(List.of(intLiteral(1))),
+                alias("1", cast(intLiteral(2), new Literal(1, DataType.INTEGER) {
+                  @Override
+                  public List<UnresolvedExpression> getChild() {
+                    return List.of(map("1", "2"));
+                  }
+                })))
+            .accept(visitor, null)),
+        () -> assertFalse(project(values(List.of(intLiteral(1))),
+                alias("1", cast(map("1", "2"), stringLiteral("int"))))
+            .accept(visitor, null))
+    );
+  }
+
+  @Test
+  public void visitArgument() {
+    var plan = project(relation("dummy"), field("pewpew", argument("name", intLiteral(0))));
+    assertTrue(plan.accept(visitor, null));
+  }
+
+  @Test
+  // source=x | eval a = b
+  public void reject_query_with_eval() {
+    var plan = project(eval(relation("dummy")));
+    assertFalse(plan.accept(visitor, null));
+  }
+
+  @Test
+  // select highlight("Body") from beer.stackexchange where
+  // simple_query_string(["Tags" ^ 1.5, "Title", "Body" 4.2], "taste")
+  // and Tags like "% % %" and Title like "%";
+  public void accept_query_with_highlight_and_relevance_func() {
+    var plan = project(
+        filter(
+            relation("beer.stackexchange"),
+            and(
+                and(
+                    function("like", qualifiedName("Tags"), stringLiteral("% % %")),
+                    function("like", qualifiedName("Title"), stringLiteral("%"))),
+                function("simple_query_string",
+                    unresolvedArg("fields",
+                        new RelevanceFieldList(Map.of("Title", 1.0F, "Body", 4.2F, "Tags", 1.5F))),
+                    unresolvedArg("query",
+                        stringLiteral("taste"))))),
+        alias("highlight(\"Body\")",
+            highlight(stringLiteral("Body"), Map.of())));
+    assertTrue(plan.accept(visitor, null));
+  }
+
   @Test
   // select * from y limit z
   public void reject_query_with_limit() {
-    var plan = AstDSL.project(AstDSL.limit(AstDSL.relation("dummy"), 1, 2), AstDSL.allFields());
+    var plan = project(limit(relation("dummy"), 1, 2), allFields());
     assertFalse(plan.accept(visitor, null));
   }
 
   @Test
   // select * from y where z
-  public void reject_query_with_where() {
-    var plan = AstDSL.project(AstDSL.filter(AstDSL.relation("dummy"),
-        AstDSL.booleanLiteral(true)), AstDSL.allFields());
-    assertFalse(plan.accept(visitor, null));
+  public void allow_query_with_where() {
+    var plan = project(filter(relation("dummy"),
+        booleanLiteral(true)), allFields());
+    assertTrue(plan.accept(visitor, null));
   }
 
   @Test
   // select * from y order by z
   public void reject_query_with_order_by() {
-    var plan = AstDSL.project(AstDSL.sort(AstDSL.relation("dummy"), AstDSL.field("1")),
-        AstDSL.allFields());
+    var plan = project(sort(relation("dummy"), field("1")),
+        allFields());
     assertFalse(plan.accept(visitor, null));
   }
 
   @Test
   // select * from y group by z
   public void reject_query_with_group_by() {
-    var plan = AstDSL.project(AstDSL.agg(
-        AstDSL.relation("dummy"), List.of(), List.of(), List.of(AstDSL.field("1")), List.of()),
-        AstDSL.allFields());
+    var plan = project(agg(
+        relation("dummy"), List.of(), List.of(), List.of(field("1")), List.of()),
+        allFields());
     assertFalse(plan.accept(visitor, null));
   }
 
   @Test
   // select agg(x) from y
   public void reject_query_with_aggregation_function() {
-    var plan = AstDSL.project(AstDSL.agg(
-        AstDSL.relation("dummy"),
-        List.of(AstDSL.alias("agg", AstDSL.aggregate("func", AstDSL.field("pewpew")))),
+    var plan = project(agg(
+        relation("dummy"),
+        List.of(alias("agg", aggregate("func", field("pewpew")))),
         List.of(), List.of(), List.of()),
-        AstDSL.allFields());
+        allFields());
     assertFalse(plan.accept(visitor, null));
   }
 
   @Test
   // select window(x) from y
   public void reject_query_with_window_function() {
-    var plan = AstDSL.project(AstDSL.relation("dummy"),
-        AstDSL.alias("pewpew",
-            AstDSL.window(
-                AstDSL.aggregate("func", AstDSL.field("pewpew")),
-                    List.of(AstDSL.qualifiedName("1")), List.of())));
+    var plan = project(relation("dummy"),
+        alias("pewpew",
+            window(
+                aggregate("func", field("pewpew")),
+                    List.of(qualifiedName("1")), List.of())));
     assertFalse(plan.accept(visitor, null));
   }
 
@@ -113,20 +379,49 @@ public void reject_query_with_window_function() {
   // select * from y, z
   public void reject_query_with_select_from_multiple_indices() {
     var plan = mock(Project.class);
-    when(plan.getChild()).thenReturn(List.of(AstDSL.relation("dummy"), AstDSL.relation("pummy")));
-    when(plan.getProjectList()).thenReturn(List.of(AstDSL.allFields()));
+    when(plan.getChild()).thenReturn(List.of(relation("dummy"), relation("pummy")));
+    when(plan.getProjectList()).thenReturn(List.of(allFields()));
     assertFalse(visitor.visitProject(plan, null));
   }
 
   @Test
   // unreal case, added for coverage only
   public void reject_project_when_relation_has_child() {
-    var relation = mock(Relation.class, withSettings().useConstructor(AstDSL.qualifiedName("42")));
-    when(relation.getChild()).thenReturn(List.of(AstDSL.relation("pewpew")));
+    var relation = mock(Relation.class, withSettings().useConstructor(qualifiedName("42")));
+    when(relation.getChild()).thenReturn(List.of(relation("pewpew")));
     when(relation.accept(visitor, null)).thenCallRealMethod();
     var plan = mock(Project.class);
     when(plan.getChild()).thenReturn(List.of(relation));
-    when(plan.getProjectList()).thenReturn(List.of(AstDSL.allFields()));
+    when(plan.getProjectList()).thenReturn(List.of(allFields()));
     assertFalse(visitor.visitProject((Project) plan, null));
   }
+
+  @Test
+  // test combinations of acceptable and not acceptable args for coverage
+  public void canPaginate() {
+    assertAll(
+        () -> assertFalse(project(values(List.of(intLiteral(1))),
+                function("func", intLiteral(1), intLiteral(1)))
+            .accept(visitor, null)),
+        () -> assertFalse(project(values(List.of(intLiteral(1))),
+                function("func", intLiteral(1), map("1", "2")))
+            .accept(visitor, null)),
+        () -> assertFalse(project(values(List.of(intLiteral(1))),
+                function("func", map("1", "2"), intLiteral(1)))
+            .accept(visitor, null))
+    );
+  }
+
+  @Test
+  // test combinations of acceptable and not acceptable args for coverage
+  public void visitFilter() {
+    assertAll(
+        () -> assertTrue(project(filter(relation("dummy"), booleanLiteral(true)))
+            .accept(visitor, null)),
+        () -> assertFalse(project(filter(relation("dummy"), map("1", "2")))
+            .accept(visitor, null)),
+        () -> assertFalse(project(filter(tableFunction(List.of("1", "2")), booleanLiteral(true)))
+            .accept(visitor, null))
+    );
+  }
 }
diff --git a/integ-test/src/test/java/org/opensearch/sql/legacy/CursorIT.java b/integ-test/src/test/java/org/opensearch/sql/legacy/CursorIT.java
index 5b9a583d04..4d62ddf306 100644
--- a/integ-test/src/test/java/org/opensearch/sql/legacy/CursorIT.java
+++ b/integ-test/src/test/java/org/opensearch/sql/legacy/CursorIT.java
@@ -13,6 +13,7 @@
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_DATE_TIME;
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_NESTED_SIMPLE;
 import static org.opensearch.sql.legacy.plugin.RestSqlAction.QUERY_API_ENDPOINT;
+import static org.opensearch.sql.util.TestUtils.verifyIsV2Cursor;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -123,7 +124,7 @@ public void validNumberOfPages() throws IOException {
     String selectQuery = StringUtils.format("SELECT firstname, state FROM %s", TEST_INDEX_ACCOUNT);
     JSONObject response = new JSONObject(executeFetchQuery(selectQuery, 50, JDBC));
     String cursor = response.getString(CURSOR);
-    verifyIsV1Cursor(cursor);
+    verifyIsV2Cursor(response);
 
     int pageCount = 1;
 
@@ -131,17 +132,21 @@ public void validNumberOfPages() throws IOException {
       response = executeCursorQuery(cursor);
       cursor = response.optString(CURSOR);
       if (!cursor.isEmpty()) {
-        verifyIsV1Cursor(cursor);
+        verifyIsV2Cursor(response);
       }
       pageCount++;
     }
 
+    // As of phase 1 of pagination feature implementation in V2, plugin returns an empty page at the
+    // end of scrolling
+    pageCount--;
+
     assertThat(pageCount, equalTo(20));
 
     // using random value here, with fetch size of 28 we should get 36 pages (ceil of 1000/28)
     response = new JSONObject(executeFetchQuery(selectQuery, 28, JDBC));
     cursor = response.getString(CURSOR);
-    verifyIsV1Cursor(cursor);
+    verifyIsV2Cursor(response);
     System.out.println(response);
     pageCount = 1;
 
@@ -149,10 +154,11 @@ public void validNumberOfPages() throws IOException {
       response = executeCursorQuery(cursor);
       cursor = response.optString(CURSOR);
       if (!cursor.isEmpty()) {
-        verifyIsV1Cursor(cursor);
+        verifyIsV2Cursor(response);
       }
       pageCount++;
     }
+
     assertThat(pageCount, equalTo(36));
   }
 
@@ -161,7 +167,7 @@ public void validNumberOfPages() throws IOException {
   public void validTotalResultWithAndWithoutPagination() throws IOException {
     // simple query - accounts index has 1000 docs, using higher limit to get all docs
     String selectQuery = StringUtils.format("SELECT firstname, state FROM %s ", TEST_INDEX_ACCOUNT);
-    verifyWithAndWithoutPaginationResponse(selectQuery + " LIMIT 2000", selectQuery, 80);
+    verifyWithAndWithoutPaginationResponse(selectQuery + " LIMIT 2000", selectQuery, 80, false);
   }
 
   @Test
@@ -169,7 +175,7 @@ public void validTotalResultWithAndWithoutPaginationWhereClause() throws IOExcep
     String selectQuery = StringUtils.format(
         "SELECT firstname, state FROM %s WHERE balance < 25000 AND age > 32", TEST_INDEX_ACCOUNT
     );
-    verifyWithAndWithoutPaginationResponse(selectQuery + " LIMIT 2000", selectQuery, 17);
+    verifyWithAndWithoutPaginationResponse(selectQuery + " LIMIT 2000", selectQuery, 17, false);
   }
 
   @Test
@@ -177,7 +183,7 @@ public void validTotalResultWithAndWithoutPaginationOrderBy() throws IOException
     String selectQuery = StringUtils.format(
         "SELECT firstname, state FROM %s ORDER BY balance DESC ", TEST_INDEX_ACCOUNT
     );
-    verifyWithAndWithoutPaginationResponse(selectQuery + " LIMIT 2000", selectQuery, 26);
+    verifyWithAndWithoutPaginationResponse(selectQuery + " LIMIT 2000", selectQuery, 26, true);
   }
 
   @Test
@@ -186,7 +192,7 @@ public void validTotalResultWithAndWithoutPaginationWhereAndOrderBy() throws IOE
         "SELECT firstname, state FROM %s WHERE balance < 25000 ORDER BY balance ASC ",
         TEST_INDEX_ACCOUNT
     );
-    verifyWithAndWithoutPaginationResponse(selectQuery + " LIMIT 2000", selectQuery, 80);
+    verifyWithAndWithoutPaginationResponse(selectQuery + " LIMIT 2000", selectQuery, 80, true);
 
   }
 
@@ -196,7 +202,7 @@ public void validTotalResultWithAndWithoutPaginationNested() throws IOException
     String selectQuery = StringUtils.format(
         "SELECT name, a.city, a.state FROM %s m , m.address as a ", TEST_INDEX_NESTED_SIMPLE
     );
-    verifyWithAndWithoutPaginationResponse(selectQuery + " LIMIT 2000", selectQuery, 1);
+    verifyWithAndWithoutPaginationResponse(selectQuery + " LIMIT 2000", selectQuery, 1, true);
   }
 
   @Test
@@ -210,6 +216,8 @@ public void noCursorWhenResultsLessThanFetchSize() throws IOException {
     assertFalse(response.has(CURSOR));
   }
 
+  @Ignore("Temporary deactivate the test until parameter substitution implemented in V2")
+  // Test was passing before, because such paging query was executed in V1, but now it is executed in V2
   @Test
   public void testCursorWithPreparedStatement() throws IOException {
     JSONObject response = executeJDBCRequest(String.format("{" +
@@ -336,12 +344,12 @@ public void testCursorCloseAPI() throws IOException {
         "SELECT firstname, state FROM %s WHERE balance > 100 and age < 40", TEST_INDEX_ACCOUNT);
     JSONObject result = new JSONObject(executeFetchQuery(selectQuery, 50, JDBC));
     String cursor = result.getString(CURSOR);
-    verifyIsV1Cursor(cursor);
+    verifyIsV2Cursor(result);
     // Retrieving next 10 pages out of remaining 19 pages
     for (int i = 0; i < 10; i++) {
       result = executeCursorQuery(cursor);
       cursor = result.optString(CURSOR);
-      verifyIsV1Cursor(cursor);
+      verifyIsV2Cursor(result);
     }
     //Closing the cursor
     JSONObject closeResp = executeCursorCloseQuery(cursor);
@@ -363,13 +371,12 @@ public void testCursorCloseAPI() throws IOException {
 
     JSONObject resp = new JSONObject(TestUtils.getResponseBody(response));
     assertThat(resp.getInt("status"), equalTo(404));
-    assertThat(resp.query("/error/reason"), equalTo("all shards failed"));
-    assertThat(resp.query("/error/caused_by/reason").toString(),
+    assertThat(resp.query("/error/reason").toString(), containsString("all shards failed"));
+    assertThat(resp.query("/error/details").toString(),
         containsString("No search context found"));
-    assertThat(resp.query("/error/type"), equalTo("search_phase_execution_exception"));
+    assertThat(resp.query("/error/type"), equalTo("SearchPhaseExecutionException"));
   }
 
-
   @Test
   public void invalidCursorIdNotDecodable() throws IOException {
     // could be either not decode-able
@@ -435,7 +442,8 @@ public void noPaginationWithNonJDBCFormat() throws IOException {
 
 
   public void verifyWithAndWithoutPaginationResponse(String sqlQuery, String cursorQuery,
-                                                     int fetch_size) throws IOException {
+                                                     int fetch_size, boolean shouldFallBackToV1)
+          throws IOException {
     // we are only checking here for schema and datarows
     JSONObject withoutCursorResponse = new JSONObject(executeFetchQuery(sqlQuery, 0, JDBC));
 
@@ -448,12 +456,22 @@ public void verifyWithAndWithoutPaginationResponse(String sqlQuery, String curso
     response.optJSONArray(DATAROWS).forEach(dataRows::put);
 
     String cursor = response.getString(CURSOR);
-    verifyIsV1Cursor(cursor);
+    if (shouldFallBackToV1) {
+      verifyIsV1Cursor(cursor);
+    } else {
+      verifyIsV2Cursor(response);
+    }
     while (!cursor.isEmpty()) {
       response = executeCursorQuery(cursor);
       response.optJSONArray(DATAROWS).forEach(dataRows::put);
       cursor = response.optString(CURSOR);
-      verifyIsV1Cursor(cursor);
+      if (shouldFallBackToV1) {
+        verifyIsV1Cursor(cursor);
+      } else {
+        if (response.has("cursor")) {
+          verifyIsV2Cursor(response);
+        }
+      }
     }
 
     verifySchema(withoutCursorResponse.optJSONArray(SCHEMA),
@@ -487,7 +505,7 @@ private void verifyIsV1Cursor(String cursor) {
     if (cursor.isEmpty()) {
       return;
     }
-    assertTrue("The cursor '" + cursor + "' is not from v1 engine.", cursor.startsWith("d:"));
+    assertTrue("The cursor '" + cursor.substring(0, 50) + "...' is not from v1 engine.", cursor.startsWith("d:"));
   }
 
   private String makeRequest(String query, String fetch_size) {
diff --git a/integ-test/src/test/java/org/opensearch/sql/sql/LegacyAPICompatibilityIT.java b/integ-test/src/test/java/org/opensearch/sql/sql/LegacyAPICompatibilityIT.java
index 1f85b2857f..adc40a24ec 100644
--- a/integ-test/src/test/java/org/opensearch/sql/sql/LegacyAPICompatibilityIT.java
+++ b/integ-test/src/test/java/org/opensearch/sql/sql/LegacyAPICompatibilityIT.java
@@ -17,6 +17,7 @@
 import java.io.IOException;
 import org.json.JSONObject;
 import org.junit.Assert;
+import org.junit.Ignore;
 import org.junit.Test;
 import org.opensearch.client.Request;
 import org.opensearch.client.RequestOptions;
diff --git a/integ-test/src/test/java/org/opensearch/sql/sql/PaginationFallbackIT.java b/integ-test/src/test/java/org/opensearch/sql/sql/PaginationFallbackIT.java
index 33d9c5f6a8..d443abc614 100644
--- a/integ-test/src/test/java/org/opensearch/sql/sql/PaginationFallbackIT.java
+++ b/integ-test/src/test/java/org/opensearch/sql/sql/PaginationFallbackIT.java
@@ -26,7 +26,7 @@ public void init() throws IOException {
   @Test
   public void testWhereClause() throws IOException {
     var response = executeQueryTemplate("SELECT * FROM %s WHERE 1 = 1", TEST_INDEX_ONLINE);
-    verifyIsV1Cursor(response);
+    verifyIsV2Cursor(response);
   }
 
   @Test
@@ -39,23 +39,22 @@ public void testSelectAll() throws IOException {
   public void testSelectWithOpenSearchFuncInFilter() throws IOException {
     var response = executeQueryTemplate(
         "SELECT * FROM %s WHERE `11` = match_phrase('96')", TEST_INDEX_ONLINE);
-    verifyIsV1Cursor(response);
+    verifyIsV2Cursor(response);
   }
 
   @Test
   public void testSelectWithHighlight() throws IOException {
     var response = executeQueryTemplate(
         "SELECT highlight(`11`) FROM %s WHERE match_query(`11`, '96')", TEST_INDEX_ONLINE);
-    // As of 2023-03-08, WHERE clause sends the query to legacy engine and legacy engine
-    // does not support highlight as an expression.
-    assertTrue(response.has("error"));
+
+    verifyIsV2Cursor(response);
   }
 
   @Test
   public void testSelectWithFullTextSearch() throws IOException {
     var response = executeQueryTemplate(
         "SELECT * FROM %s WHERE match_phrase(`11`, '96')", TEST_INDEX_ONLINE);
-    verifyIsV1Cursor(response);
+    verifyIsV2Cursor(response);
   }
 
   @Test
@@ -74,7 +73,7 @@ public void testSelectFromDataSource() throws IOException {
   @Test
   public void testSelectColumnReference() throws IOException {
     var response = executeQueryTemplate("SELECT `107` from %s", TEST_INDEX_ONLINE);
-    verifyIsV1Cursor(response);
+    verifyIsV2Cursor(response);
   }
 
   @Test
@@ -88,7 +87,7 @@ public void testSubquery() throws IOException {
   public void testSelectExpression() throws IOException {
     var response = executeQueryTemplate("SELECT 1 + 1 - `107` from %s",
         TEST_INDEX_ONLINE);
-    verifyIsV1Cursor(response);
+    verifyIsV2Cursor(response);
   }
 
   @Test
diff --git a/integ-test/src/test/java/org/opensearch/sql/sql/PaginationFilterIT.java b/integ-test/src/test/java/org/opensearch/sql/sql/PaginationFilterIT.java
new file mode 100644
index 0000000000..6ebc05efad
--- /dev/null
+++ b/integ-test/src/test/java/org/opensearch/sql/sql/PaginationFilterIT.java
@@ -0,0 +1,143 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+
+package org.opensearch.sql.sql;
+
+import com.carrotsearch.randomizedtesting.annotations.Name;
+import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import lombok.SneakyThrows;
+import org.json.JSONArray;
+import org.json.JSONObject;
+import org.junit.Test;
+import org.junit.jupiter.api.DisplayNameGeneration;
+import org.junit.jupiter.api.DisplayNameGenerator;
+import org.opensearch.sql.legacy.SQLIntegTestCase;
+import org.opensearch.sql.legacy.TestsConstants;
+
+/**
+ * Test pagination with `WHERE` clause using a parametrized test.
+ * See constructor {@link #PaginationFilterIT} for list of parameters
+ * and {@link #generateParameters} and {@link #STATEMENT_TO_NUM_OF_PAGES}
+ * to see how these parameters are generated.
+ */
+@DisplayNameGeneration(DisplayNameGenerator.ReplaceUnderscores.class)
+public class PaginationFilterIT extends SQLIntegTestCase {
+
+  /**
+   * Map of the OS-SQL statement sent to SQL-plugin, and the total number
+   * of expected hits (on all pages) from the filtered result
+   */
+  final private static Map<String, Integer> STATEMENT_TO_NUM_OF_PAGES = Map.of(
+      "SELECT * FROM " + TestsConstants.TEST_INDEX_ACCOUNT, 1000,
+      "SELECT * FROM " + TestsConstants.TEST_INDEX_ACCOUNT + " WHERE match(address, 'street')", 385,
+      "SELECT * FROM " + TestsConstants.TEST_INDEX_ACCOUNT + " WHERE match(address, 'street') AND match(city, 'Ola')", 1,
+      "SELECT firstname, lastname, highlight(address) FROM " + TestsConstants.TEST_INDEX_ACCOUNT + " WHERE match(address, 'street') AND match(state, 'OH')", 5,
+      "SELECT firstname, lastname, highlight('*') FROM " + TestsConstants.TEST_INDEX_ACCOUNT + " WHERE match(address, 'street') AND match(state, 'OH')", 5,
+      "SELECT * FROM " + TestsConstants.TEST_INDEX_BEER + " WHERE true", 60,
+      "SELECT * FROM " + TestsConstants.TEST_INDEX_BEER + " WHERE Id=10", 1,
+      "SELECT * FROM " + TestsConstants.TEST_INDEX_BEER + " WHERE Id + 5=15", 1,
+      "SELECT * FROM " + TestsConstants.TEST_INDEX_BANK, 7
+  );
+
+  private final String sqlStatement;
+
+  private final Integer totalHits;
+  private final Integer pageSize;
+
+  public PaginationFilterIT(@Name("statement") String sqlStatement,
+                            @Name("total_hits") Integer totalHits,
+                            @Name("page_size") Integer pageSize) {
+    this.sqlStatement = sqlStatement;
+    this.totalHits = totalHits;
+    this.pageSize = pageSize;
+  }
+
+  @Override
+  public void init() throws IOException {
+    initClient();
+    loadIndex(Index.ACCOUNT);
+    loadIndex(Index.BEER);
+    loadIndex(Index.BANK);
+  }
+
+  @ParametersFactory(argumentFormatting = "query = %1$s, total_hits = %2$d, page_size = %3$d")
+  public static Iterable<Object[]> generateParameters() {
+    List<Integer> pageSizes = List.of(5, 1000);
+    List<Object[]> testData = new ArrayList<Object[]>();
+
+    STATEMENT_TO_NUM_OF_PAGES.forEach((statement, totalHits) -> {
+      for (var pageSize : pageSizes) {
+        testData.add(new Object[] { statement, totalHits, pageSize });
+      }
+    });
+    return testData;
+  }
+
+  /**
+   * Test compares non-paginated results with paginated results
+   * To ensure that the pushdowns return the same number of hits even
+   * with filter WHERE pushed down
+   */
+  @Test
+  @SneakyThrows
+  public void test_pagination_with_where() {
+    // get non-paginated result for comparison
+    JSONObject nonPaginatedResponse = executeJdbcRequest(sqlStatement);
+    int totalResultsCount = nonPaginatedResponse.getInt("total");
+    JSONArray rows = nonPaginatedResponse.getJSONArray("datarows");
+    JSONArray schema = nonPaginatedResponse.getJSONArray("schema");
+    var testReportPrefix = String.format("query: %s; total hits: %d; page size: %d || ", sqlStatement, totalResultsCount, pageSize);
+    assertEquals(totalHits.intValue(), totalResultsCount);
+
+    var rowsPaged = new JSONArray();
+    var pagedSize = 0;
+    var responseCounter = 1;
+
+    // make first request - with a cursor
+    JSONObject paginatedResponse = new JSONObject(executeFetchQuery(sqlStatement, pageSize, "jdbc"));
+    this.logger.info(testReportPrefix + "<first response>");
+    do {
+      var cursor = paginatedResponse.has("cursor") ? paginatedResponse.getString("cursor") : null;
+      pagedSize += paginatedResponse.getInt("size");
+      var datarows = paginatedResponse.getJSONArray("datarows");
+      for (int i = 0; i < datarows.length(); i++) {
+        rowsPaged.put(datarows.get(i));
+      }
+
+      assertTrue(
+          "Paged response schema doesn't match to non-paged",
+          schema.similar(paginatedResponse.getJSONArray("schema")));
+
+      if (cursor != null) {
+        assertTrue(
+            testReportPrefix + "Cursor returned from legacy engine",
+            cursor.startsWith("n:"));
+
+        paginatedResponse = executeCursorQuery(cursor);
+
+        this.logger.info(testReportPrefix
+            + String.format("response %d/%d", responseCounter++, (totalResultsCount / pageSize) + 1));
+      } else {
+        break;
+      }
+    } while (true);
+    // last page expected results:
+    assertEquals(testReportPrefix + "Last page",
+        totalHits % pageSize, paginatedResponse.getInt("size"));
+    assertEquals(testReportPrefix + "Last page",
+        totalHits % pageSize, paginatedResponse.getJSONArray("datarows").length());
+
+    // compare paginated and non-paginated counts
+    assertEquals(testReportPrefix + "Paged responses returned an unexpected total",
+        totalResultsCount, pagedSize);
+    assertEquals(testReportPrefix + "Paged responses returned an unexpected rows count",
+        rows.length(), rowsPaged.length());
+  }
+}
diff --git a/integ-test/src/test/java/org/opensearch/sql/sql/PaginationIT.java b/integ-test/src/test/java/org/opensearch/sql/sql/PaginationIT.java
index 72ec20c679..46f8e72fb5 100644
--- a/integ-test/src/test/java/org/opensearch/sql/sql/PaginationIT.java
+++ b/integ-test/src/test/java/org/opensearch/sql/sql/PaginationIT.java
@@ -5,6 +5,8 @@
 
 package org.opensearch.sql.sql;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_CALCS;
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_ONLINE;
 import static org.opensearch.sql.legacy.plugin.RestSqlAction.EXPLAIN_API_ENDPOINT;
@@ -39,11 +41,11 @@ public void testSmallDataSet() throws IOException {
   }
 
   @Test
-  public void testLargeDataSetV1() throws IOException {
-    var v1query = "SELECT * from " + TEST_INDEX_ONLINE + " WHERE 1 = 1";
-    var v1response = new JSONObject(executeFetchQuery(v1query, 4, "jdbc"));
-    assertEquals(4, v1response.getInt("size"));
-    TestUtils.verifyIsV1Cursor(v1response);
+  public void testLargeDataSetV2WithWhere() throws IOException {
+    var query = "SELECT * from " + TEST_INDEX_ONLINE + " WHERE 1 = 1";
+    var response = new JSONObject(executeFetchQuery(query, 4, "jdbc"));
+    assertEquals(4, response.getInt("size"));
+    TestUtils.verifyIsV2Cursor(response);
   }
 
   @Test
diff --git a/integ-test/src/test/java/org/opensearch/sql/util/TestUtils.java b/integ-test/src/test/java/org/opensearch/sql/util/TestUtils.java
index 69f1649190..3281c172cb 100644
--- a/integ-test/src/test/java/org/opensearch/sql/util/TestUtils.java
+++ b/integ-test/src/test/java/org/opensearch/sql/util/TestUtils.java
@@ -7,6 +7,7 @@
 package org.opensearch.sql.util;
 
 import static com.google.common.base.Strings.isNullOrEmpty;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.opensearch.sql.executor.pagination.PlanSerializer.CURSOR_PREFIX;
 
@@ -856,8 +857,8 @@ private static void verifyCursor(JSONObject response, List<String> validCursorPr
       assertTrue("'cursor' property does not exist", response.has("cursor"));
 
       var cursor = response.getString("cursor");
-      assertTrue("'cursor' property is empty", !cursor.isEmpty());
-      assertTrue("The cursor '" + cursor + "' is not from " + engineName + " engine.",
+      assertFalse("'cursor' property is empty", cursor.isEmpty());
+      assertTrue("The cursor '" + cursor.substring(0, 50) + "...' is not from " + engineName + " engine.",
           validCursorPrefix.stream().anyMatch(cursor::startsWith));
     }
 
diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchQueryRequest.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchQueryRequest.java
index 45954a3871..2d035f8a5b 100644
--- a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchQueryRequest.java
+++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchQueryRequest.java
@@ -45,8 +45,6 @@ public class OpenSearchQueryRequest implements OpenSearchRequest {
    */
   private final SearchSourceBuilder sourceBuilder;
 
-
-
   /**
    * OpenSearchExprValueFactory.
    */
diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchRequest.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchRequest.java
index ee9da5b53b..cae988ea56 100644
--- a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchRequest.java
+++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchRequest.java
@@ -34,7 +34,7 @@ public interface OpenSearchRequest extends Writeable {
    *
    * @param searchAction search action.
    * @param scrollAction scroll search action.
-   * @return ElasticsearchResponse.
+   * @return OpenSearchResponse.
    */
   OpenSearchResponse search(Function<SearchRequest, SearchResponse> searchAction,
                             Function<SearchScrollRequest, SearchResponse> scrollAction);
@@ -47,8 +47,8 @@ OpenSearchResponse search(Function<SearchRequest, SearchResponse> searchAction,
   void clean(Consumer<String> cleanAction);
 
   /**
-   * Get the ElasticsearchExprValueFactory.
-   * @return ElasticsearchExprValueFactory.
+   * Get the OpenSearchExprValueFactory.
+   * @return OpenSearchExprValueFactory.
    */
   OpenSearchExprValueFactory getExprValueFactory();
 
diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchScrollRequest.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchScrollRequest.java
index 403626c610..6a5d001684 100644
--- a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchScrollRequest.java
+++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchScrollRequest.java
@@ -38,7 +38,12 @@
 @Getter
 @ToString
 public class OpenSearchScrollRequest implements OpenSearchRequest {
-  private final SearchRequest initialSearchRequest;
+
+  /**
+   * Search request used to initiate paged (scrolled) search. Not needed to get subsequent pages.
+   */
+  @EqualsAndHashCode.Exclude
+  private final transient SearchRequest initialSearchRequest;
   /** Scroll context timeout. */
   private final TimeValue scrollTimeout;
 
@@ -81,7 +86,7 @@ public OpenSearchScrollRequest(IndexName indexName,
         .scroll(scrollTimeout)
         .source(sourceBuilder);
 
-    includes =  sourceBuilder.fetchSource() == null
+    includes = sourceBuilder.fetchSource() == null
         ? List.of()
         : Arrays.asList(sourceBuilder.fetchSource().includes());
   }
@@ -96,6 +101,11 @@ public OpenSearchResponse search(Function<SearchRequest, SearchResponse> searchA
     if (isScroll()) {
       openSearchResponse = scrollAction.apply(scrollRequest());
     } else {
+      if (initialSearchRequest == null) {
+        // Probably a first page search (since there is no scroll set) called on a deserialized
+        // `OpenSearchScrollRequest`, which has no `initialSearchRequest`.
+        throw new UnsupportedOperationException("Misuse of OpenSearchScrollRequest");
+      }
       openSearchResponse = searchAction.apply(initialSearchRequest);
     }
 
@@ -154,7 +164,6 @@ public boolean hasAnotherBatch() {
 
   @Override
   public void writeTo(StreamOutput out) throws IOException {
-    initialSearchRequest.writeTo(out);
     out.writeTimeValue(scrollTimeout);
     out.writeString(scrollId);
     out.writeStringCollection(includes);
@@ -165,11 +174,11 @@ public void writeTo(StreamOutput out) throws IOException {
    * Constructs OpenSearchScrollRequest from serialized representation.
    * @param in stream to read data from.
    * @param engine OpenSearchSqlEngine to get node-specific context.
-   * @throws IOException thrown if reading from input {@param in} fails.
+   * @throws IOException thrown if reading from input {@code in} fails.
    */
   public OpenSearchScrollRequest(StreamInput in, OpenSearchStorageEngine engine)
       throws IOException {
-    initialSearchRequest = new SearchRequest(in);
+    initialSearchRequest = null;
     scrollTimeout = in.readTimeValue();
     scrollId = in.readString();
     includes = in.readStringList();
diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/request/OpenSearchScrollRequestTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/request/OpenSearchScrollRequestTest.java
index 69f38ee7f2..57ba6c201b 100644
--- a/opensearch/src/test/java/org/opensearch/sql/opensearch/request/OpenSearchScrollRequestTest.java
+++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/request/OpenSearchScrollRequestTest.java
@@ -6,9 +6,12 @@
 
 package org.opensearch.sql.opensearch.request;
 
+import static org.junit.jupiter.api.Assertions.assertAll;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 import static org.mockito.ArgumentMatchers.any;
@@ -21,7 +24,6 @@
 
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Function;
 import lombok.SneakyThrows;
 import org.apache.commons.lang3.reflect.FieldUtils;
 import org.apache.lucene.search.TotalHits;
@@ -53,12 +55,6 @@ class OpenSearchScrollRequestTest {
   public static final OpenSearchRequest.IndexName INDEX_NAME
       = new OpenSearchRequest.IndexName("test");
   public static final TimeValue SCROLL_TIMEOUT = TimeValue.timeValueMinutes(1);
-  @Mock
-  private Function<SearchRequest, SearchResponse> searchAction;
-
-  @Mock
-  private Function<SearchScrollRequest, SearchResponse> scrollAction;
-
   @Mock
   private SearchResponse searchResponse;
 
@@ -144,17 +140,12 @@ void search() {
     when(searchResponse.getHits()).thenReturn(searchHits);
     when(searchHits.getHits()).thenReturn(new SearchHit[] {searchHit});
 
-    Function<SearchScrollRequest, SearchResponse> scrollSearch = searchScrollRequest -> {
-      throw new AssertionError();
-    };
-    OpenSearchResponse openSearchResponse = request.search(searchRequest -> searchResponse,
-        scrollSearch);
-
-    assertFalse(openSearchResponse.isEmpty());
+    OpenSearchResponse response = request.search((sr) -> searchResponse, (sr) -> fail());
+    assertFalse(response.isEmpty());
   }
 
   @Test
-  void search_withoutContext() {
+  void search_without_context() {
     OpenSearchScrollRequest request = new OpenSearchScrollRequest(
         new OpenSearchRequest.IndexName("test"),
         TimeValue.timeValueMinutes(1),
@@ -162,13 +153,38 @@ void search_withoutContext() {
         factory
     );
 
-    when(searchAction.apply(any())).thenReturn(searchResponse);
     when(searchResponse.getHits()).thenReturn(searchHits);
     when(searchHits.getHits()).thenReturn(new SearchHit[] {searchHit});
 
-    OpenSearchResponse searchResponse = request.search(searchAction, scrollAction);
+    OpenSearchResponse response = request.search((sr) -> searchResponse, (sr) -> fail());
     verify(sourceBuilder, times(1)).fetchSource();
-    assertFalse(searchResponse.isEmpty());
+    assertFalse(response.isEmpty());
+  }
+
+  @Test
+  @SneakyThrows
+  void search_without_scroll_and_initial_request_should_throw() {
+    // Steps: serialize a not used request, deserialize it, then use
+    OpenSearchScrollRequest request = new OpenSearchScrollRequest(
+        new OpenSearchRequest.IndexName("test"),
+        TimeValue.timeValueMinutes(1),
+        sourceBuilder,
+        factory
+    );
+    var outStream = new BytesStreamOutput();
+    request.writeTo(outStream);
+    outStream.flush();
+    var inStream = new BytesStreamInput(outStream.bytes().toBytesRef().bytes);
+    var indexMock = mock(OpenSearchIndex.class);
+    var engine = mock(OpenSearchStorageEngine.class);
+    when(engine.getTable(any(), any())).thenReturn(indexMock);
+    var request2 = new OpenSearchScrollRequest(inStream, engine);
+    assertAll(
+        () -> assertFalse(request2.isScroll()),
+        () -> assertNull(request2.getInitialSearchRequest()),
+        () -> assertThrows(UnsupportedOperationException.class,
+            () -> request2.search(sr -> fail("search"), sr -> fail("scroll")))
+    );
   }
 
   @Test
@@ -180,12 +196,11 @@ void search_withoutIncludes() {
         factory
     );
 
-    when(searchAction.apply(any())).thenReturn(searchResponse);
     when(searchResponse.getHits()).thenReturn(searchHits);
     when(searchHits.getHits()).thenReturn(new SearchHit[] {searchHit});
 
-    OpenSearchResponse searchResponse = request.search(searchAction, scrollAction);
-    assertFalse(searchResponse.isEmpty());
+    OpenSearchResponse response = request.search((sr) -> searchResponse, (sr) -> fail());
+    assertFalse(response.isEmpty());
   }
 
   @Test
@@ -230,7 +245,7 @@ void no_clean_on_non_empty_response() {
     when(searchResponse.getHits()).thenReturn(
         new SearchHits(new SearchHit[1], new TotalHits(1, TotalHits.Relation.EQUAL_TO), 1F));
 
-    request.search((x) -> searchResponse, (x) -> searchResponse);
+    request.search((sr) -> searchResponse, (sr) -> searchResponse);
     assertEquals("scroll", request.getScrollId());
 
     request.clean((s) -> fail());
@@ -273,7 +288,7 @@ void serialize_deserialize_no_needClean() {
     var engine = mock(OpenSearchStorageEngine.class);
     when(engine.getTable(any(), any())).thenReturn(indexMock);
     var newRequest = new OpenSearchScrollRequest(inStream, engine);
-    assertEquals(request.getInitialSearchRequest(), newRequest.getInitialSearchRequest());
+    assertEquals(request, newRequest);
     assertEquals("", newRequest.getScrollId());
   }
 
@@ -296,7 +311,7 @@ void serialize_deserialize_needClean() {
     var engine = mock(OpenSearchStorageEngine.class);
     when(engine.getTable(any(), any())).thenReturn(indexMock);
     var newRequest = new OpenSearchScrollRequest(inStream, engine);
-    assertEquals(request.getInitialSearchRequest(), newRequest.getInitialSearchRequest());
+    assertEquals(request, newRequest);
     assertEquals("", newRequest.getScrollId());
   }
 
diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/OpenSearchDefaultImplementorTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/OpenSearchDefaultImplementorTest.java
index d7e5955491..f2b6a70a46 100644
--- a/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/OpenSearchDefaultImplementorTest.java
+++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/storage/OpenSearchDefaultImplementorTest.java
@@ -27,9 +27,6 @@ public class OpenSearchDefaultImplementorTest {
   @Mock
   OpenSearchClient client;
 
-  @Mock
-  Table table;
-
   @Test
   public void visitMachineLearning() {
     LogicalMLCommons node = Mockito.mock(LogicalMLCommons.class,