Add PPL Between functionality

docs/ppl-lang/PPL-Example-Commands.md

@@ -49,6 +49,8 @@ _- **Limitation: new field added by eval command with a function cannot be dropp
 - `source = table | where isempty(a)`
 - `source = table | where isblank(a)`
 - `source = table | where case(length(a) > 6, 'True' else 'False') = 'True'`
+- `source = table | where a between 1 and 4`
+- `source = table | where b not between '2024-09-10' and '2025-09-10'`
 
 ```sql
  source = table | eval status_category =

docs/ppl-lang/planning/ppl-between.md

@@ -0,0 +1,20 @@
+## between syntax proposal
+
+1. **Proposed syntax**
+    - `... | where expr1 [NOT] BETWEEN expr2 AND expr3`
+        - evaluate if expr1 is [not] in between expr2 and expr3 
+    - `... | where a between 1 and 4`
+    - `... | where b not between '2024-09-10' and '2025-09-10'`
+
+2. **Proposed impl**
+    - forward to sparks built-in function of between 
+
+### New syntax definition in ANTLR
+
+```ANTLR
+
+logicalExpression
+   ...
+   | expr1 = functionArg NOT? BETWEEN expr2 = functionArg AND expr3 = functionArg   # between
+
+```

docs/ppl-lang/ppl-where-command.md

@@ -41,6 +41,8 @@ PPL query:
 - `source = table | where isempty(a)`
 - `source = table | where isblank(a)`
 - `source = table | where case(length(a) > 6, 'True' else 'False') = 'True'`
+- `source = table | where a between 1 and 4`
+- `source = table | where b not between '2024-09-10' and '2025-09-10'`
 
 - `source = table | eval status_category =
       case(a >= 200 AND a < 300, 'Success',

integ-test/src/integration/scala/org/opensearch/flint/spark/ppl/FlintSparkPPLBetweenITSuite.scala

@@ -0,0 +1,65 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.flint.spark.ppl
+
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.streaming.StreamTest
+
+class FlintSparkPPLBetweenITSuite
+    extends QueryTest
+    with LogicalPlanTestUtils
+    with FlintPPLSuite
+    with StreamTest {
+
+  /** Test table and index name */
+  private val testTable = "spark_catalog.default.flint_ppl_test"
+
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+
+    // Create test table
+    createPartitionedStateCountryTable(testTable)
+  }
+
+  protected override def afterEach(): Unit = {
+    super.afterEach()
+    // Stop all streaming jobs if any
+    spark.streams.active.foreach { job =>
+      job.stop()
+      job.awaitTermination()
+    }
+  }
+
+  test("test between should return records between two values") {
+    val frame = sql(s"""
+                       | source = $testTable | where age between 20 and 30
+                       | """.stripMargin)
+
+    val results = frame.collect()
+    assert(results.length == 3)
+    assert(frame.columns.length == 6)
+
+    results.foreach(row => {
+      val age = row.getAs[Int]("age")
+      assert(age >= 20 && age <= 30, s"Age $age is not between 20 and 30")
+    })
+  }
+
+  test("test between should return records NOT between two values") {
+    val frame = sql(s"""
+                       | source = $testTable | where age NOT between 20 and 30
+                       | """.stripMargin)
+
+    val results = frame.collect()
+    assert(results.length == 1)
+    assert(frame.columns.length == 6)
+
+    results.foreach(row => {
+      val age = row.getAs[Int]("age")
+      assert(age < 20 || age > 30, s"Age $age is not between 20 and 30")
+    })
+  }
+}

ppl-spark-integration/src/main/antlr4/OpenSearchPPLLexer.g4

@@ -368,6 +368,7 @@ LIKE:                               'LIKE';
 ISNULL:                             'ISNULL';
 ISNOTNULL:                          'ISNOTNULL';
 ISPRESENT:                          'ISPRESENT';
+BETWEEN:                            'BETWEEN';
 
 // FLOWCONTROL FUNCTIONS
 IFNULL:                             'IFNULL';

ppl-spark-integration/src/main/antlr4/OpenSearchPPLParser.g4

@@ -371,6 +371,7 @@ logicalExpression
    | left = logicalExpression OR right = logicalExpression      # logicalOr
    | left = logicalExpression XOR right = logicalExpression     # logicalXor
    | booleanExpression                                          # booleanExpr
+   | expr1 = functionArg NOT? BETWEEN expr2 = functionArg AND expr3 = functionArg   # between
    ;
 
 comparisonExpression

ppl-spark-integration/src/main/java/org/opensearch/sql/ppl/CatalystQueryPlanVisitor.java

@@ -15,7 +15,9 @@
 import org.apache.spark.sql.catalyst.expressions.Descending$;
 import org.apache.spark.sql.catalyst.expressions.Exists$;
 import org.apache.spark.sql.catalyst.expressions.Expression;
+import org.apache.spark.sql.catalyst.expressions.GreaterThanOrEqual;
 import org.apache.spark.sql.catalyst.expressions.InSubquery$;
+import org.apache.spark.sql.catalyst.expressions.LessThanOrEqual;
 import org.apache.spark.sql.catalyst.expressions.ListQuery$;
 import org.apache.spark.sql.catalyst.expressions.NamedExpression;
 import org.apache.spark.sql.catalyst.expressions.Predicate;
@@ -34,6 +36,7 @@
 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.BinaryExpression;
 import org.opensearch.sql.ast.expression.Case;
 import org.opensearch.sql.ast.expression.Compare;
@@ -829,5 +832,14 @@ public Expression visitExistsSubquery(ExistsSubquery node, CatalystPlanContext c
                 Option.empty());
             return context.getNamedParseExpressions().push(existsSubQuery);
         }
+
+        @Override
+        public Expression visitBetween(Between node, CatalystPlanContext context) {
+            Expression value = analyze(node.getValue(), context);
+            Expression lower = analyze(node.getLowerBound(), context);
+            Expression upper = analyze(node.getUpperBound(), context);
+            context.retainAllNamedParseExpressions(p -> p);
+            return context.getNamedParseExpressions().push(new org.apache.spark.sql.catalyst.expressions.And(new GreaterThanOrEqual(value, lower), new LessThanOrEqual(value, upper)));
+        }
     }
 }

ppl-spark-integration/src/main/java/org/opensearch/sql/ppl/parser/AstExpressionBuilder.java

@@ -16,6 +16,7 @@
 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.Compare;
 import org.opensearch.sql.ast.expression.DataType;
@@ -269,6 +270,12 @@ public UnresolvedExpression visitConvertedDataType(OpenSearchPPLParser.Converted
         return new Literal(ctx.getText(), DataType.STRING);
     }
 
+    @Override
+    public UnresolvedExpression visitBetween(OpenSearchPPLParser.BetweenContext ctx) {
+        UnresolvedExpression betweenExpr = new Between(visit(ctx.expr1),visit(ctx.expr2),visit(ctx.expr3));
+        return ctx.NOT() != null ? new Not(betweenExpr) : betweenExpr;
+    }
+
     private Function buildFunction(
             String functionName, List<OpenSearchPPLParser.FunctionArgContext> args) {
         return new Function(

ppl-spark-integration/src/test/scala/org/opensearch/flint/spark/ppl/PPLLogicalPlanBetweenExpressionTranslatorTestSuite.scala

@@ -0,0 +1,55 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.flint.spark.ppl
+
+import org.opensearch.flint.spark.ppl.PlaneUtils.plan
+import org.opensearch.sql.ppl.{CatalystPlanContext, CatalystQueryPlanVisitor}
+import org.scalatest.matchers.should.Matchers
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.analysis.{UnresolvedAttribute, UnresolvedRelation, UnresolvedStar}
+import org.apache.spark.sql.catalyst.expressions.{And, GreaterThanOrEqual, LessThanOrEqual, Literal}
+import org.apache.spark.sql.catalyst.plans.PlanTest
+import org.apache.spark.sql.catalyst.plans.logical._
+
+class PPLLogicalPlanBetweenExpressionTranslatorTestSuite
+    extends SparkFunSuite
+    with PlanTest
+    with LogicalPlanTestUtils
+    with Matchers {
+
+  private val planTransformer = new CatalystQueryPlanVisitor()
+  private val pplParser = new PPLSyntaxParser()
+
+  test("test between expression") {
+    // if successful build ppl logical plan and translate to catalyst logical plan
+    val context = new CatalystPlanContext
+    val logPlan = {
+      planTransformer.visit(
+        plan(
+          pplParser,
+          "source = table | where datetime_field between '2024-09-10' and  '2024-09-15'"),
+        context)
+    }
+    // SQL: SELECT * FROM table WHERE datetime_field BETWEEN '2024-09-10' AND '2024-09-15'
+    val star = Seq(UnresolvedStar(None))
+
+    val datetime_field = UnresolvedAttribute("datetime_field")
+    val tableRelation = UnresolvedRelation(Seq("table"))
+
+    val lowerBound = Literal("2024-09-10")
+    val upperBound = Literal("2024-09-15")
+    val betweenCondition = And(
+      GreaterThanOrEqual(datetime_field, lowerBound),
+      LessThanOrEqual(datetime_field, upperBound))
+
+    val filterPlan = Filter(betweenCondition, tableRelation)
+    val expectedPlan = Project(star, filterPlan)
+
+    comparePlans(expectedPlan, logPlan, false)
+  }
+
+}