Support refering to implicit @timestamp field in time-based aggregations

Signed-off-by: Yuanchun Shen <yuanchu@amazon.com>

Author: yuancu

core/src/main/java/org/opensearch/sql/ast/expression/Span.java

@@ -7,6 +7,7 @@
 
 import com.google.common.collect.ImmutableList;
 import java.util.List;
+import javax.annotation.Nullable;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
@@ -19,13 +20,16 @@
 @RequiredArgsConstructor
 @ToString
 public class Span extends UnresolvedExpression {
-  private final UnresolvedExpression field;
+  @Nullable private final UnresolvedExpression field;
   private final UnresolvedExpression value;
   private final SpanUnit unit;
 
   @Override
   public List<UnresolvedExpression> getChild() {
-    return ImmutableList.of(field, value);
+    if (field == null) {
+      return ImmutableList.of(value);
+    }
+    return List.of(field, value);
   }
 
   @Override

core/src/main/java/org/opensearch/sql/calcite/CalciteRexNodeVisitor.java

@@ -46,6 +46,7 @@
 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.In;
 import org.opensearch.sql.ast.expression.Interval;
@@ -67,6 +68,7 @@
 import org.opensearch.sql.ast.expression.subquery.InSubquery;
 import org.opensearch.sql.ast.expression.subquery.ScalarSubquery;
 import org.opensearch.sql.ast.tree.UnresolvedPlan;
+import org.opensearch.sql.calcite.plan.OpenSearchConstants;
 import org.opensearch.sql.calcite.utils.OpenSearchTypeFactory;
 import org.opensearch.sql.calcite.utils.PlanUtils;
 import org.opensearch.sql.common.utils.StringUtils;
@@ -349,7 +351,19 @@ public RexNode visitAlias(Alias node, CalcitePlanContext context) {
 
   @Override
   public RexNode visitSpan(Span node, CalcitePlanContext context) {
-    RexNode field = analyze(node.getField(), context);
+    RexNode field;
+    if (node.getField() != null) {
+      field = analyze(node.getField(), context);
+    } else {
+      try {
+        field = referenceImplicitTimestampField(context);
+      } catch (IllegalArgumentException e) {
+        throw new SemanticCheckException(
+            "SPAN operation requires an explicit field or an implicit '@timestamp' field, but"
+                + " '@timestamp' was not found in the input schema.",
+            e);
+      }
+    }
     RexNode value = analyze(node.getValue(), context);
     SpanUnit unit = node.getUnit();
     RexBuilder rexBuilder = context.relBuilder.getRexBuilder();
@@ -359,6 +373,11 @@ public RexNode visitSpan(Span node, CalcitePlanContext context) {
         context.rexBuilder, BuiltinFunctionName.SPAN, field, value, unitNode);
   }
 
+  private RexNode referenceImplicitTimestampField(CalcitePlanContext context) {
+    return analyze(
+        new Field(new QualifiedName(OpenSearchConstants.IMPLICIT_FIELD_TIMESTAMP)), context);
+  }
+
   private boolean isTimeBased(SpanUnit unit) {
     return !(unit == NONE || unit == UNKNOWN);
   }

core/src/main/java/org/opensearch/sql/calcite/plan/OpenSearchConstants.java

@@ -20,6 +20,8 @@ public interface OpenSearchConstants {
 
   String METADATA_FIELD_ROUTING = "_routing";
 
+  String IMPLICIT_FIELD_TIMESTAMP = "@timestamp";
+
   java.util.Map<String, ExprType> METADATAFIELD_TYPE_MAP =
       Map.of(
           METADATA_FIELD_ID, ExprCoreType.STRING,

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

@@ -14,6 +14,7 @@
 import static org.opensearch.sql.util.MatcherUtils.rows;
 import static org.opensearch.sql.util.MatcherUtils.schema;
 import static org.opensearch.sql.util.MatcherUtils.verifyDataRows;
+import static org.opensearch.sql.util.MatcherUtils.verifyErrorMessageContains;
 import static org.opensearch.sql.util.MatcherUtils.verifySchema;
 import static org.opensearch.sql.util.MatcherUtils.verifySchemaInOrder;
 
@@ -24,6 +25,8 @@
 import org.junit.Ignore;
 import org.junit.jupiter.api.Test;
 import org.opensearch.client.Request;
+import org.opensearch.sql.common.utils.StringUtils;
+import org.opensearch.sql.exception.SemanticCheckException;
 import org.opensearch.sql.ppl.PPLIntegTestCase;
 
 public class CalcitePPLAggregationIT extends PPLIntegTestCase {
@@ -38,6 +41,7 @@ public void init() throws Exception {
     loadIndex(Index.CALCS);
     loadIndex(Index.DATE_FORMATS);
     loadIndex(Index.DATA_TYPE_NUMERIC);
+    loadIndex(Index.BIG5);
   }
 
   @Test
@@ -515,6 +519,26 @@ public void testCountBySpanForCustomFormats() throws IOException {
     verifyDataRows(actual, rows(1, "00:00:00"), rows(1, "12:00:00"));
   }
 
+  // Only available in v3 with Calcite
+  @Test
+  public void testSpanByImplicitTimestamp() throws IOException {
+    JSONObject result = executeQuery("source=big5 | stats count() by span(1d) as span");
+    verifySchema(result, schema("count()", "bigint"), schema("span", "timestamp"));
+    verifyDataRows(result, rows(1, "2023-01-02 00:00:00"));
+
+    Throwable t =
+        assertThrowsWithReplace(
+            SemanticCheckException.class,
+            () ->
+                executeQuery(
+                    StringUtils.format(
+                        "source=%s | stats count() by span(5m)", TEST_INDEX_DATE_FORMATS)));
+    verifyErrorMessageContains(
+        t,
+        "SPAN operation requires an explicit field or an implicit '@timestamp' field, but"
+            + " '@timestamp' was not found in the input schema.");
+  }
+
   @Test
   public void testCountDistinct() throws IOException {
     JSONObject actual =

ppl/src/main/antlr/OpenSearchPPLParser.g4

@@ -381,7 +381,7 @@ bySpanClause
    ;
 
 spanClause
-   : SPAN LT_PRTHS fieldExpression COMMA value = literalValue (unit = timespanUnit)? RT_PRTHS
+   : SPAN LT_PRTHS (fieldExpression COMMA)? value = literalValue (unit = timespanUnit)? RT_PRTHS
    ;
 
 sortbyClause

ppl/src/main/java/org/opensearch/sql/ppl/parser/AstExpressionBuilder.java

@@ -523,7 +523,8 @@ public UnresolvedExpression visitBySpanClause(BySpanClauseContext ctx) {
   @Override
   public UnresolvedExpression visitSpanClause(SpanClauseContext ctx) {
     String unit = ctx.unit != null ? ctx.unit.getText() : "";
-    return new Span(visit(ctx.fieldExpression()), visit(ctx.value), SpanUnit.of(unit));
+    var field = ctx.fieldExpression() != null ? visit(ctx.fieldExpression()) : null;
+    return new Span(field, visit(ctx.value), SpanUnit.of(unit));
   }
 
   @Override

ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java

@@ -553,9 +553,13 @@ public String visitAggregateFunction(AggregateFunction node, String context) {
 
     @Override
     public String visitSpan(Span node, String context) {
-      String field = analyze(node.getField(), context);
+      String field = node.getField() != null ? analyze(node.getField(), context) : null;
       String value = analyze(node.getValue(), context);
-      return StringUtils.format("span(%s, %s %s)", field, value, node.getUnit().getName());
+      if (field != null) {
+        return StringUtils.format("span(%s, %s %s)", field, value, node.getUnit().getName());
+      } else {
+        return StringUtils.format("span(%s %s)", value, node.getUnit().getName());
+      }
     }
 
     @Override