Support SAMPLE, DERIVATIVE and INTEGRAL table functions

iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/OrderByRequirement.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.udf.api.relational.table.specification;
+
+/**
+ * The requirement a table function imposes on the ORDER BY clause of a table argument. A table
+ * parameter specification carrying no requirement accepts any ORDER BY clause, including its
+ * absence.
+ */
+public enum OrderByRequirement {
+
+  /** The clause is required and every sort item must be in ascending order. */
+  ASCENDING,
+
+  /** The clause is required and every sort item must be in descending order. */
+  DESCENDING,
+
+  /** The clause is required, but its direction is not constrained. */
+  ORDERED
+}

iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/TableParameterSpecification.java

@@ -22,7 +22,7 @@
 import java.util.Optional;
 
 /**
- * TableParameterSpecification are classified by two characteristics:
+ * TableParameterSpecification are classified by three characteristics:
  *
  * <ul>
  *   <li>1. Input tables have either row semantics or set semantics, as follows:
@@ -37,19 +37,27 @@
  *   <li>2. Whether the input table supports pass-through columns or not. Pass-through columns isa
  *       mechanism enabling the PTF to copy every column of an input row into columns of an output
  *       row.
+ *   <li>3. Which requirement the ORDER BY clause of the input table has to satisfy, if any. See
+ *       {@link OrderByRequirement}.
  * </ul>
  */
 public class TableParameterSpecification extends ParameterSpecification {
   // set semantics or row semantics (default is set semantics)
   private final boolean rowSemantics;
   private final boolean passThroughColumns;
+  // empty means the ORDER BY clause is not constrained
+  private final Optional<OrderByRequirement> orderByRequirement;
 
   private TableParameterSpecification(
-      String name, boolean rowSemantics, boolean passThroughColumns) {
+      String name,
+      boolean rowSemantics,
+      boolean passThroughColumns,
+      Optional<OrderByRequirement> orderByRequirement) {
     // table arguments are always required
     super(name, true, Optional.empty());
     this.rowSemantics = rowSemantics;
     this.passThroughColumns = passThroughColumns;
+    this.orderByRequirement = orderByRequirement;
   }
 
   public boolean isRowSemantics() {
@@ -60,6 +68,10 @@ public boolean isPassThroughColumns() {
     return passThroughColumns;
   }
 
+  public Optional<OrderByRequirement> getOrderByRequirement() {
+    return orderByRequirement;
+  }
+
   public static Builder builder() {
     return new Builder();
   }
@@ -68,6 +80,7 @@ public static final class Builder {
     private String name;
     private boolean rowSemantics;
     private boolean passThroughColumns;
+    private Optional<OrderByRequirement> orderByRequirement = Optional.empty();
 
     private Builder() {}
 
@@ -91,8 +104,24 @@ public Builder passThroughColumns() {
       return this;
     }
 
+    public Builder ascendingOrder() {
+      this.orderByRequirement = Optional.of(OrderByRequirement.ASCENDING);
+      return this;
+    }
+
+    public Builder descendingOrder() {
+      this.orderByRequirement = Optional.of(OrderByRequirement.DESCENDING);
+      return this;
+    }
+
+    public Builder ordered() {
+      this.orderByRequirement = Optional.of(OrderByRequirement.ORDERED);
+      return this;
+    }
+
     public TableParameterSpecification build() {
-      return new TableParameterSpecification(name, rowSemantics, passThroughColumns);
+      return new TableParameterSpecification(
+          name, rowSemantics, passThroughColumns, orderByRequirement);
     }
   }
 }

iotdb-core/calc-commons/src/main/i18n/en/org/apache/iotdb/calc/i18n/CalcMessages.java

@@ -581,6 +581,12 @@ private CalcMessages() {}
   public static final String
       EXCEPTION_AGGREGATE_FUNCTION_ARG_REQUIRES_ARG_ARGUMENTS_BUT_GOT_ARG_D6249DD2 =
           "Aggregate function [%s] requires %d arguments, but got %d";
+  public static final String
+      EXCEPTION_THE_TIME_UNIT_ARGUMENT_OF_AGGREGATE_FUNCTION_ARG_IS_INVALID_89D17438 =
+          "The time unit argument of aggregate function [%s] is invalid: %s";
+  public static final String
+      EXCEPTION_AGGREGATE_FUNCTION_ARG_VALUE_EXCEEDS_THE_FINITE_RANGE_OF_DOUBLE_48936427 =
+          "The value of aggregate function [%s] exceeds the finite range of double: %s";
 
   public static final String EXCEPTION_UNSUPPORTED_TARGET_DATATYPE_ARG_AFE3789A =
       "Unsupported target dataType: %s";

iotdb-core/calc-commons/src/main/i18n/zh/org/apache/iotdb/calc/i18n/CalcMessages.java

@@ -558,6 +558,12 @@ private CalcMessages() {}
   public static final String
       EXCEPTION_AGGREGATE_FUNCTION_ARG_REQUIRES_ARG_ARGUMENTS_BUT_GOT_ARG_D6249DD2 =
           "聚合函数 [%s] 需要 %d 个参数，但实际传入 %d 个";
+  public static final String
+      EXCEPTION_THE_TIME_UNIT_ARGUMENT_OF_AGGREGATE_FUNCTION_ARG_IS_INVALID_89D17438 =
+          "聚合函数 [%s] 的时间单位参数无效：%s";
+  public static final String
+      EXCEPTION_AGGREGATE_FUNCTION_ARG_VALUE_EXCEEDS_THE_FINITE_RANGE_OF_DOUBLE_48936427 =
+          "聚合函数 [%s] 的值超出了 double 的有限范围：%s";
 
   public static final String EXCEPTION_UNSUPPORTED_TARGET_DATATYPE_ARG_AFE3789A = "不支持的目标数据类型：%s";
 

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/function/TableFunctionOperator.java

@@ -181,7 +181,12 @@ public TsBlock next() throws Exception {
           processor.beforeStart(ioTDBLocal);
         }
       }
-      partitionCache.addSlice(slice);
+      // the cached slices are only needed when the output has
+      // columns declared as pass-through. Without them every output column is written by the
+      // processor, so the input block should not be retained in the partitionCache
+      if (needPassThrough) {
+        partitionCache.addSlice(slice);
+      }
       Iterator<Record> recordIterator = slice.getRequiredRecordIterator(requireRecordSnapshot);
       while (recordIterator.hasNext()) {
         processor.process(

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/AccumulatorFactory.java

@@ -49,6 +49,8 @@
 import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedSumAccumulator;
 import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedUserDefinedAggregateAccumulator;
 import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedVarianceAccumulator;
+import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.integral.GroupedNaiveIntegralAccumulator;
+import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.integral.GroupedOrderedIntegralAccumulator;
 import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate.GroupedNaiveDeltaAccumulator;
 import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate.GroupedNaiveIncreaseAccumulator;
 import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate.GroupedNaiveIrateAccumulator;
@@ -57,6 +59,8 @@
 import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate.GroupedOrderedIncreaseAccumulator;
 import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate.GroupedOrderedIrateAccumulator;
 import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate.GroupedOrderedRateAccumulator;
+import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.integral.NaiveIntegralAccumulator;
+import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.integral.OrderedIntegralAccumulator;
 import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.NaiveDeltaAccumulator;
 import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.NaiveIncreaseAccumulator;
 import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.NaiveIrateAccumulator;
@@ -501,6 +505,10 @@ private static GroupedAccumulator createBuiltinGroupedAccumulator(
         return useOrderedImplementation
             ? new GroupedOrderedDeltaAccumulator(inputDataTypes.get(0))
             : new GroupedNaiveDeltaAccumulator(inputDataTypes.get(0), memoryReservationManager);
+      case INTEGRAL:
+        return useOrderedImplementation
+            ? new GroupedOrderedIntegralAccumulator(inputDataTypes.get(0))
+            : new GroupedNaiveIntegralAccumulator(inputDataTypes.get(0), memoryReservationManager);
       default:
         throw new IllegalArgumentException(
             CalcMessages.INVALID_AGGREGATION_FUNCTION + aggregationType);
@@ -629,6 +637,10 @@ public static TableAccumulator createBuiltinAccumulator(
         return useOrderedImplementation
             ? new OrderedDeltaAccumulator(inputDataTypes.get(0))
             : new NaiveDeltaAccumulator(inputDataTypes.get(0), memoryReservationManager);
+      case INTEGRAL:
+        return useOrderedImplementation
+            ? new OrderedIntegralAccumulator(inputDataTypes.get(0))
+            : new NaiveIntegralAccumulator(inputDataTypes.get(0), memoryReservationManager);
       default:
         throw new IllegalArgumentException(
             CalcMessages.INVALID_AGGREGATION_FUNCTION + aggregationType);

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/TimeValueBufferBigArray.java → iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/TimeValueBufferBigArray.java

@@ -17,14 +17,14 @@
  * under the License.
  */
 
-package org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate;
+package org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped;
 
 import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.array.ObjectBigArray;
 import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.TimeValueBuffer;
 
 import org.apache.tsfile.utils.RamUsageEstimator;
 
-final class TimeValueBufferBigArray {
+public final class TimeValueBufferBigArray {
 
   private static final long INSTANCE_SIZE =
       RamUsageEstimator.shallowSizeOfInstance(TimeValueBufferBigArray.class);

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/integral/AbstractGroupedIntegralAccumulator.java

@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.integral;
+
+import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedAccumulator;
+import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.integral.IntegralFunctionUtils;
+import org.apache.iotdb.commons.exception.SemanticException;
+
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.enums.TSDataType;
+
+abstract class AbstractGroupedIntegralAccumulator implements GroupedAccumulator {
+
+  protected static final int VALUE_ARGUMENT_INDEX = 0;
+  protected static final int TIME_UNIT_ARGUMENT_INDEX = 1;
+  protected static final int TIME_ARGUMENT_INDEX = 2;
+
+  protected final TSDataType valueDataType;
+
+  // The time unit converted to ticks of the current timestamp precision, 0 means uninitialized
+  protected long timeUnit;
+
+  protected AbstractGroupedIntegralAccumulator(TSDataType valueDataType) {
+    this.valueDataType = valueDataType;
+  }
+
+  protected final double readValue(Column column, int position) {
+    return IntegralFunctionUtils.readValue(valueDataType, column, position);
+  }
+
+  protected final void initializeOrValidateTimeUnit(Column timeUnitColumn) {
+    // an empty block carries no value to read, and there is nothing to integrate either
+    if (timeUnit != 0 || timeUnitColumn.getPositionCount() == 0) {
+      return;
+    }
+    initializeOrValidateTimeUnit(timeUnitColumn.getLong(0));
+  }
+
+  protected final void initializeOrValidateTimeUnit(long unitTicks) {
+    if (unitTicks <= 0) {
+      throw invalidTimeUnit(String.valueOf(unitTicks));
+    }
+    if (timeUnit != 0 && timeUnit != unitTicks) {
+      throw invalidTimeUnit(String.valueOf(unitTicks));
+    }
+    timeUnit = unitTicks;
+  }
+
+  protected final void resetTimeUnit() {
+    timeUnit = 0;
+  }
+
+  protected final double normalizeArea(double area) {
+    // timeUnit stays 0 when no input has ever been added
+    return timeUnit == 0 ? 0.0 : area / timeUnit;
+  }
+
+  protected final SemanticException invalidTimeUnit(String text) {
+    return IntegralFunctionUtils.invalidTimeUnit(text);
+  }
+
+  protected final SemanticException orderedInputViolation(long time, long previousTime) {
+    return IntegralFunctionUtils.orderedInputViolation(time, previousTime);
+  }
+
+  protected final UnsupportedOperationException unsupportedIntermediate() {
+    return IntegralFunctionUtils.unsupportedIntermediate();
+  }
+
+  @Override
+  public final void prepareFinal() {}
+}