Add last_by and first_by aggregation support for table model

integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBMultiIDsWithAttributesTableIT.java

@@ -989,6 +989,67 @@ public void lastFirstMaxMinTest() {
     tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
   }
 
+  @Test
+  public void lastByFirstByTest() {
+    String[] expectedHeader1 = buildHeaders(13);
+    String[] expectedHeader2 = buildHeaders(14);
+
+    sql =
+        "select last_by(time,time),last_by(device,time),last_by(level,time),last_by(attr1,time),last_by(attr2,time),last_by(num,time),last_by(bignum,time),last_by(floatnum,time),last_by(str,time),last_by(bool,time),last_by(date,time),last_by(ts,time),last_by(stringv,time) from table0 where device='d2'";
+    retArray =
+        new String[] {
+          "1971-08-20T11:33:20.000Z,d2,l5,null,null,15,3147483648,235.213,watermelon,true,2023-01-01,null,null,",
+        };
+    tableResultSetEqualTest(sql, expectedHeader1, retArray, DATABASE_NAME);
+    sql =
+        "select last_by(time,time),last_by(device,time),last_by(level,time),last_by(attr1,time),last_by(attr2,time),last_by(num,time),last_by(bignum,time),last_by(floatnum,time),last_by(str,time),last_by(bool,time),last_by(date,time),last_by(ts,time),last_by(stringv,time),last_by(blob,time) from table0 where device='d2'";
+    retArray =
+        new String[] {
+          "1971-08-20T11:33:20.000Z,d2,l5,null,null,15,3147483648,235.213,watermelon,true,2023-01-01,null,null,null,",
+        };
+    tableResultSetEqualTest(sql, expectedHeader2, retArray, DATABASE_NAME);
+
+    sql =
+        "select last_by(time,time),last_by(time,device),last_by(time,level),last_by(time,attr1),last_by(time,attr2),last_by(time,num),last_by(time,bignum),last_by(time,floatnum),last_by(time,str),last_by(time,bool),last_by(time,date),last_by(time,ts),last_by(time,stringv) from table0 where device='d2'";
+    retArray =
+        new String[] {
+          "1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-04-26T17:46:40.000Z,1971-01-01T00:01:40.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-01-01T00:01:40.000Z,1971-01-01T00:01:40.000Z,",
+        };
+    tableResultSetEqualTest(sql, expectedHeader1, retArray, DATABASE_NAME);
+    sql =
+        "select last_by(time,time),last_by(time,device),last_by(time,level),last_by(time,attr1),last_by(time,attr2),last_by(time,num),last_by(time,bignum),last_by(time,floatnum),last_by(time,str),last_by(time,bool),last_by(time,date),last_by(time,ts),last_by(time,stringv),last_by(time,blob) from table0 where device='d2'";
+    retArray =
+        new String[] {
+          "1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-04-26T17:46:40.000Z,1971-01-01T00:01:40.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-01-01T00:01:40.000Z,1971-01-01T00:01:40.000Z,1970-01-01T00:00:00.080Z,",
+        };
+    tableResultSetEqualTest(sql, expectedHeader2, retArray, DATABASE_NAME);
+
+    String[] expectedHeader11 = buildHeaders(expectedHeader1.length * 2);
+    sql =
+        "select last_by(time,time),last_by(device,time),last_by(level,time),last_by(attr1,time),last_by(attr2,time),last_by(num,time),last_by(bignum,time),last_by(floatnum,time),last_by(str,time),last_by(bool,time),last_by(date,time),last_by(ts,time),last_by(stringv,time),last_by(time,time),last_by(time,device),last_by(time,level),last_by(time,attr1),last_by(time,attr2),last_by(time,num),last_by(time,bignum),last_by(time,floatnum),last_by(time,str),last_by(time,bool),last_by(time,date),last_by(time,ts),last_by(time,stringv) from table0 where device='d2'";
+    retArray =
+        new String[] {
+          "1971-08-20T11:33:20.000Z,d2,l5,null,null,15,3147483648,235.213,watermelon,true,2023-01-01,null,null,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-04-26T17:46:40.000Z,1971-01-01T00:01:40.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-08-20T11:33:20.000Z,1971-01-01T00:01:40.000Z,1971-01-01T00:01:40.000Z,",
+        };
+    tableResultSetEqualTest(sql, expectedHeader11, retArray, DATABASE_NAME);
+
+    sql =
+        "select first_by(time,time),first_by(device,time),first_by(level,time),first_by(attr1,time),first_by(attr2,time),first_by(num,time),first_by(bignum,time),first_by(floatnum,time),first_by(str,time),first_by(bool,time),first_by(date,time),first_by(ts,time),first_by(stringv,time) from table0 where device='d2' and time>80";
+    retArray =
+        new String[] {
+          "1970-01-01T00:00:00.100Z,d2,l5,null,null,8,2147483964,4654.231,papaya,true,null,null,null,",
+        };
+    tableResultSetEqualTest(sql, expectedHeader1, retArray, DATABASE_NAME);
+
+    sql =
+        "select first_by(time,time),first_by(time,device),first_by(time,level),first_by(time,attr1),first_by(time,attr2),first_by(time,num),first_by(time,bignum),first_by(time,floatnum),first_by(time,str),first_by(time,bool),first_by(time,date),first_by(time,ts),first_by(time,stringv) from table0 where device='d2' and time>80";
+    retArray =
+        new String[] {
+          "1970-01-01T00:00:00.100Z,1970-01-01T00:00:00.100Z,1970-01-01T00:00:00.100Z,1971-01-01T00:00:00.000Z,1971-01-01T00:00:00.000Z,1970-01-01T00:00:00.100Z,1970-01-01T00:00:00.100Z,1970-01-01T00:00:00.100Z,1970-01-01T00:00:00.100Z,1970-01-01T00:00:00.100Z,1971-08-20T11:33:20.000Z,1971-01-01T00:01:40.000Z,1971-01-01T00:01:40.000Z,",
+        };
+    tableResultSetEqualTest(sql, expectedHeader1, retArray, DATABASE_NAME);
+  }
+
   // ==================================================================
   // ============================ Join Test ===========================
   // ==================================================================
@@ -1274,4 +1335,12 @@ public void fullOuterJoinTest2() {
             + "ORDER BY time, t1.device, t2.device";
     tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
   }
+
+  public static String[] buildHeaders(int length) {
+    String[] expectedHeader = new String[length];
+    for (int i = 0; i < length; i++) {
+      expectedHeader[i] = "_col" + i;
+    }
+    return expectedHeader;
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AccumulatorFactory.java

@@ -23,17 +23,18 @@
 import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedAccumulator;
 import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedAvgAccumulator;
 import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedCountAccumulator;
-import org.apache.iotdb.db.queryengine.plan.expression.Expression;
-import org.apache.iotdb.db.queryengine.plan.expression.binary.CompareBinaryExpression;
-import org.apache.iotdb.db.queryengine.plan.expression.leaf.ConstantOperand;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.SymbolReference;
 
 import org.apache.tsfile.enums.TSDataType;
 
 import java.util.List;
 import java.util.Map;
 
 import static com.google.common.base.Preconditions.checkState;
-import static org.apache.iotdb.db.queryengine.plan.relational.metadata.TableBuiltinAggregationFunction.LAST;
+import static org.apache.iotdb.commons.schema.table.TsTable.TIME_COLUMN_NAME;
+import static org.apache.iotdb.db.queryengine.plan.relational.metadata.TableBuiltinAggregationFunction.FIRST_BY;
+import static org.apache.iotdb.db.queryengine.plan.relational.metadata.TableBuiltinAggregationFunction.LAST_BY;
 
 public class AccumulatorFactory {
 
@@ -47,6 +48,23 @@ public static TableAccumulator createAccumulator(
     if (aggregationType == TAggregationType.UDAF) {
       // If UDAF accumulator receives raw input, it needs to check input's attribute
       throw new UnsupportedOperationException();
+    } else if ((LAST_BY.getFunctionName().equals(functionName)
+            || FIRST_BY.getFunctionName().equals(functionName))
+        && inputExpressions.size() > 1) {
+      boolean xIsTimeColumn = false;
+      boolean yIsTimeColumn = false;
+      if (isTimeColumn(inputExpressions.get(1))) {
+        yIsTimeColumn = true;
+      } else if (isTimeColumn(inputExpressions.get(0))) {
+        xIsTimeColumn = true;
+      }
+      if (LAST_BY.getFunctionName().equals(functionName)) {
+        return new LastByAccumulator(
+            inputDataTypes.get(0), inputDataTypes.get(1), xIsTimeColumn, yIsTimeColumn);
+      } else {
+        return new FirstByAccumulator(
+            inputDataTypes.get(0), inputDataTypes.get(1), xIsTimeColumn, yIsTimeColumn);
+      }
     } else {
       return createBuiltinAccumulator(
           aggregationType, inputDataTypes, inputExpressions, inputAttributes, ascending);
@@ -106,6 +124,10 @@ public static TableAccumulator createBuiltinAccumulator(
         return new MaxAccumulator(inputDataTypes.get(0));
       case MIN:
         return new MinAccumulator(inputDataTypes.get(0));
+      case LAST_BY:
+        return new LastByAccumulator(inputDataTypes.get(0), inputDataTypes.get(1), false, false);
+      case FIRST_BY:
+        return new FirstByAccumulator(inputDataTypes.get(0), inputDataTypes.get(1), false, false);
       default:
         throw new IllegalArgumentException("Invalid Aggregation function: " + aggregationType);
     }
@@ -219,34 +241,8 @@ public interface KeepEvaluator {
     boolean apply(long keep);
   }
 
-  public static KeepEvaluator initKeepEvaluator(Expression keepExpression) {
-    // We have checked semantic in FE,
-    // keep expression must be ConstantOperand or CompareBinaryExpression here
-    if (keepExpression instanceof ConstantOperand) {
-      return keep -> keep >= Long.parseLong(keepExpression.getExpressionString());
-    } else {
-      long constant =
-          Long.parseLong(
-              ((CompareBinaryExpression) keepExpression)
-                  .getRightExpression()
-                  .getExpressionString());
-      switch (keepExpression.getExpressionType()) {
-        case LESS_THAN:
-          return keep -> keep < constant;
-        case LESS_EQUAL:
-          return keep -> keep <= constant;
-        case GREATER_THAN:
-          return keep -> keep > constant;
-        case GREATER_EQUAL:
-          return keep -> keep >= constant;
-        case EQUAL_TO:
-          return keep -> keep == constant;
-        case NON_EQUAL:
-          return keep -> keep != constant;
-        default:
-          throw new IllegalArgumentException(
-              "unsupported expression type: " + keepExpression.getExpressionType());
-      }
-    }
+  public static boolean isTimeColumn(Expression expression) {
+    return expression instanceof SymbolReference
+        && TIME_COLUMN_NAME.equals(((SymbolReference) expression).getName());
   }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AvgAccumulator.java

@@ -138,7 +138,7 @@ public boolean hasFinalResult() {
 
   @Override
   public void addStatistics(Statistics[] statistics) {
-    if (statistics == null) {
+    if (statistics == null || statistics[0] == null) {
       return;
     }
     initResult = true;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/CountAccumulator.java

@@ -82,9 +82,10 @@ public boolean hasFinalResult() {
 
   @Override
   public void addStatistics(Statistics[] statistics) {
-    if (statistics[0] == null) {
+    if (statistics == null || statistics[0] == null) {
       return;
     }
+
     countState += statistics[0].getCount();
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/FirstAccumulator.java

@@ -196,12 +196,12 @@ public void evaluateFinal(ColumnBuilder columnBuilder) {
 
   @Override
   public boolean hasFinalResult() {
-    return initResult;
+    return false;
   }
 
   @Override
   public void addStatistics(Statistics[] statistics) {
-    if (statistics[0] == null) {
+    if (statistics == null || statistics[0] == null) {
       return;
     }
     switch (seriesDataType) {