Fix filter parsing failure on date fields with non-default format

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java

@@ -70,9 +70,9 @@ public void supportSearchSargPushDown_timeRange() throws IOException {
         "source=opensearch-sql_test_index_bank"
             + "| where birthdate >= '2016-12-08 00:00:00.000000000' "
             + "and birthdate < '2018-11-09 00:00:00.000000000'";
-    var result = explainQueryToString(query);
-    String expected = loadExpectedPlan("explain_sarg_filter_push_time_range.json");
-    assertJsonEqualsIgnoreId(expected, result);
+    var result = explainQueryYaml(query);
+    String expected = loadExpectedPlan("explain_sarg_filter_push_time_range.yaml");
+    assertYamlEqualsIgnoreId(expected, result);
   }
 
   // Only for Calcite

integ-test/src/test/resources/expectedOutput/calcite/explain_filter_push_compare_timestamp_string.yaml

@@ -6,4 +6,4 @@ calcite:
           LogicalFilter(condition=[>($3, TIMESTAMP('2016-12-08 00:00:00.000000000':VARCHAR))])
             CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank]])
   physical: |
-    CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank]], PushDownContext=[[PROJECT->[account_number, firstname, address, birthdate, gender, city, lastname, balance, employer, state, age, email, male], FILTER->SEARCH($3, Sarg[('2016-12-08 00:00:00':VARCHAR..'2018-11-09 00:00:00':VARCHAR)]:VARCHAR), LIMIT->10000], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":10000,"timeout":"1m","query":{"range":{"birthdate":{"from":"2016-12-08T00:00:00.000Z","to":"2018-11-09T00:00:00.000Z","include_lower":false,"include_upper":false,"boost":1.0}}},"_source":{"includes":["account_number","firstname","address","birthdate","gender","city","lastname","balance","employer","state","age","email","male"],"excludes":[]}}, requestedTotalSize=10000, pageSize=null, startFrom=0)])
+    CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank]], PushDownContext=[[PROJECT->[account_number, firstname, address, birthdate, gender, city, lastname, balance, employer, state, age, email, male], FILTER->SEARCH($3, Sarg[('2016-12-08 00:00:00':VARCHAR..'2018-11-09 00:00:00':VARCHAR)]:VARCHAR), LIMIT->10000], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":10000,"timeout":"1m","query":{"range":{"birthdate":{"from":"2016-12-08T00:00:00.000Z","to":"2018-11-09T00:00:00.000Z","include_lower":false,"include_upper":false,"format":"date_time","boost":1.0}}},"_source":{"includes":["account_number","firstname","address","birthdate","gender","city","lastname","balance","employer","state","age","email","male"],"excludes":[]}}, requestedTotalSize=10000, pageSize=null, startFrom=0)])

integ-test/src/test/resources/expectedOutput/calcite/explain_filter_with_search.yaml

@@ -8,4 +8,4 @@ calcite:
               LogicalFilter(condition=[AND(>=($3, TIMESTAMP('2023-01-01 00:00:00':VARCHAR)), <($3, TIMESTAMP('2023-01-03 00:00:00':VARCHAR)))])
                 CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank]])
   physical: |
-    CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank]], PushDownContext=[[PROJECT->[birthdate], FILTER->SEARCH($0, Sarg[['2023-01-01 00:00:00':VARCHAR..'2023-01-03 00:00:00':VARCHAR); NULL AS FALSE]:VARCHAR), AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0},count()=COUNT()), PROJECT->[count(), span(birthdate,1d)], LIMIT->10000], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","query":{"bool":{"must":[{"range":{"birthdate":{"from":"2023-01-01T00:00:00.000Z","to":"2023-01-03T00:00:00.000Z","include_lower":true,"include_upper":false,"boost":1.0}}},{"exists":{"field":"birthdate","boost":1.0}}],"adjust_pure_negative":true,"boost":1.0}},"_source":{"includes":["birthdate"],"excludes":[]},"aggregations":{"composite_buckets":{"composite":{"size":1000,"sources":[{"span(birthdate,1d)":{"date_histogram":{"field":"birthdate","missing_bucket":false,"order":"asc","fixed_interval":"1d"}}}]}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])
+    CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank]], PushDownContext=[[PROJECT->[birthdate], FILTER->SEARCH($0, Sarg[['2023-01-01 00:00:00':VARCHAR..'2023-01-03 00:00:00':VARCHAR); NULL AS FALSE]:VARCHAR), AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0},count()=COUNT()), PROJECT->[count(), span(birthdate,1d)], LIMIT->10000], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","query":{"bool":{"must":[{"range":{"birthdate":{"from":"2023-01-01T00:00:00.000Z","to":"2023-01-03T00:00:00.000Z","include_lower":true,"include_upper":false,"format":"date_time","boost":1.0}}},{"exists":{"field":"birthdate","boost":1.0}}],"adjust_pure_negative":true,"boost":1.0}},"_source":{"includes":["birthdate"],"excludes":[]},"aggregations":{"composite_buckets":{"composite":{"size":1000,"sources":[{"span(birthdate,1d)":{"date_histogram":{"field":"birthdate","missing_bucket":false,"order":"asc","fixed_interval":"1d"}}}]}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])

integ-test/src/test/resources/expectedOutput/calcite/explain_sarg_filter_push_time_range.yaml

@@ -0,0 +1,8 @@
+calcite:
+  logical: |
+    LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])
+      LogicalProject(account_number=[$0], firstname=[$1], address=[$2], birthdate=[$3], gender=[$4], city=[$5], lastname=[$6], balance=[$7], employer=[$8], state=[$9], age=[$10], email=[$11], male=[$12])
+        LogicalFilter(condition=[AND(>=($3, TIMESTAMP('2016-12-08 00:00:00.000000000':VARCHAR)), <($3, TIMESTAMP('2018-11-09 00:00:00.000000000':VARCHAR)))])
+          CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank]])
+  physical: |
+    CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank]], PushDownContext=[[PROJECT->[account_number, firstname, address, birthdate, gender, city, lastname, balance, employer, state, age, email, male], FILTER->SEARCH($3, Sarg[['2016-12-08 00:00:00':VARCHAR..'2018-11-09 00:00:00':VARCHAR)]:VARCHAR), LIMIT->10000], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":10000,"timeout":"1m","query":{"range":{"birthdate":{"from":"2016-12-08T00:00:00.000Z","to":"2018-11-09T00:00:00.000Z","include_lower":true,"include_upper":false,"format":"date_time","boost":1.0}}},"_source":{"includes":["account_number","firstname","address","birthdate","gender","city","lastname","balance","employer","state","age","email","male"],"excludes":[]}}, requestedTotalSize=10000, pageSize=null, startFrom=0)])

integ-test/src/test/resources/expectedOutput/calcite_no_pushdown/explain_sarg_filter_push_time_range.yaml

@@ -0,0 +1,10 @@
+calcite:
+  logical: |
+    LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])
+      LogicalProject(account_number=[$0], firstname=[$1], address=[$2], birthdate=[$3], gender=[$4], city=[$5], lastname=[$6], balance=[$7], employer=[$8], state=[$9], age=[$10], email=[$11], male=[$12])
+        LogicalFilter(condition=[AND(>=($3, TIMESTAMP('2016-12-08 00:00:00.000000000':VARCHAR)), <($3, TIMESTAMP('2018-11-09 00:00:00.000000000':VARCHAR)))])
+          CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank]])
+  physical: |
+    EnumerableLimit(fetch=[10000])
+      EnumerableCalc(expr#0..18=[{inputs}], expr#19=[Sarg[['2016-12-08 00:00:00':VARCHAR..'2018-11-09 00:00:00':VARCHAR)]:VARCHAR], expr#20=[SEARCH($t3, $t19)], proj#0..12=[{exprs}], $condition=[$t20])
+        CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_bank]])

integ-test/src/yamlRestTest/resources/rest-api-spec/test/issues/4490.yml

@@ -0,0 +1,53 @@
+setup:
+  - do:
+      indices.create:
+        index: test
+        body:
+          mappings:
+            properties:
+              "startTimeMillis":
+                type: date
+                format: epoch_millis
+
+  - do:
+      bulk:
+        index: test
+        refresh: true
+        body:
+          - '{"index": {"_id": "1"}}'
+          - '{"startTimeMillis": 539325296000}'
+          - '{"index": {"_id": "2"}}'
+          - '{"startTimeMillis": "1715126504378"}'
+---
+"handle epoch_millis format field with equal":
+  - skip:
+      features:
+        - headers
+        - allowed_warnings
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: source=test | eval example_time = STR_TO_DATE('1987-02-03 04:34:56', '%Y-%m-%d %H:%i:%S') | where startTimeMillis = example_time | fields startTimeMillis
+
+  - match: { total: 1 }
+  - match: {"schema": [{"name": "startTimeMillis", "type": "timestamp"}]}
+  - match: {"datarows": [["1987-02-03 04:34:56"]]}
+
+---
+"handle epoch_millis format field with comparison":
+  - skip:
+      features:
+        - headers
+        - allowed_warnings
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: source=test | eval example_time = STR_TO_DATE('1987-02-03 04:34:56', '%Y-%m-%d %H:%i:%S') | where startTimeMillis >= example_time | fields startTimeMillis
+
+  - match: { total: 2 }
+  - match: {"schema": [{"name": "startTimeMillis", "type": "timestamp"}]}
+  - match: {"datarows": [["1987-02-03 04:34:56"], ["2024-05-08 00:01:44.378"]]}

opensearch/src/main/java/org/opensearch/sql/opensearch/request/PredicateAnalyzer.java

@@ -1274,11 +1274,9 @@ public QueryExpression notLike(LiteralExpression literal) {
     @Override
     public QueryExpression equals(LiteralExpression literal) {
       Object value = literal.value();
-      if (value instanceof GregorianCalendar) {
+      if (literal.isDateTime()) {
         builder =
-            boolQuery()
-                .must(addFormatIfNecessary(literal, rangeQuery(getFieldReference()).gte(value)))
-                .must(addFormatIfNecessary(literal, rangeQuery(getFieldReference()).lte(value)));
+            addFormatIfNecessary(literal, rangeQuery(getFieldReference()).gte(value).lte(value));
       } else {
         builder = termQuery(getFieldReferenceForTermQuery(), value);
       }
@@ -1288,7 +1286,7 @@ public QueryExpression equals(LiteralExpression literal) {
     @Override
     public QueryExpression notEquals(LiteralExpression literal) {
       Object value = literal.value();
-      if (value instanceof GregorianCalendar) {
+      if (literal.isDateTime()) {
         builder =
             boolQuery()
                 .should(addFormatIfNecessary(literal, rangeQuery(getFieldReference()).gt(value)))
@@ -1399,8 +1397,12 @@ public QueryExpression notIn(LiteralExpression literal) {
 
     @Override
     public QueryExpression equals(Object point, boolean isTimeStamp) {
-      builder =
-          termQuery(getFieldReferenceForTermQuery(), convertEndpointValue(point, isTimeStamp));
+      Object value = convertEndpointValue(point, isTimeStamp);
+      if (isTimeStamp) {
+        builder = rangeQuery(getFieldReference()).gte(value).lte(value).format("date_time");
+      } else {
+        builder = termQuery(getFieldReferenceForTermQuery(), value);
+      }
       return this;
     }
 
@@ -1419,6 +1421,7 @@ public QueryExpression between(Range<?> range, boolean isTimeStamp) {
           range.upperBoundType() == BoundType.CLOSED
               ? rangeQueryBuilder.lte(upperBound)
               : rangeQueryBuilder.lt(upperBound);
+      if (isTimeStamp) rangeQueryBuilder.format("date_time");
       builder = rangeQueryBuilder;
       return this;
     }
@@ -1511,7 +1514,7 @@ public List<RexNode> getUnAnalyzableNodes() {
    */
   private static RangeQueryBuilder addFormatIfNecessary(
       LiteralExpression literal, RangeQueryBuilder rangeQueryBuilder) {
-    if (literal.value() instanceof GregorianCalendar) {
+    if (literal.isDateTime()) {
       rangeQueryBuilder.format("date_time");
     }
     return rangeQueryBuilder;
@@ -1634,7 +1637,7 @@ Object value() {
         return doubleValue();
       } else if (isBoolean()) {
         return booleanValue();
-      } else if (isTimestamp()) {
+      } else if (isTimestamp() || isDate()) {
         return timestampValueForPushDown(RexLiteral.stringValue(literal));
       } else if (isString()) {
         return RexLiteral.stringValue(literal);
@@ -1676,10 +1679,21 @@ public boolean isTimestamp() {
       return false;
     }
 
+    public boolean isDate() {
+      if (literal.getType() instanceof ExprSqlType exprSqlType) {
+        return exprSqlType.getUdt() == ExprUDT.EXPR_DATE;
+      }
+      return false;
+    }
+
     public boolean isIp() {
       return literal.getType() instanceof ExprIPType;
     }
 
+    public boolean isDateTime() {
+      return rawValue() instanceof GregorianCalendar || isTimestamp() || isDate();
+    }
+
     long longValue() {
       return ((Number) literal.getValue()).longValue();
     }