diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowTVFIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowTVFIT.java index 0adddc0cf69ec..97996e77cec39 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowTVFIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowTVFIT.java @@ -23,6 +23,7 @@ import org.apache.iotdb.it.framework.IoTDBTestRunner; import org.apache.iotdb.itbase.category.TableClusterIT; import org.apache.iotdb.itbase.category.TableLocalStandaloneIT; +import org.apache.iotdb.itbase.runtime.ClusterTestResultSet; import org.apache.iotdb.rpc.TSStatusCode; import org.junit.AfterClass; @@ -38,9 +39,12 @@ import java.sql.Statement; import java.sql.Types; import java.time.Instant; +import java.util.ArrayList; import java.util.Arrays; import java.util.Comparator; +import java.util.HashSet; import java.util.List; +import java.util.Set; import static org.apache.iotdb.db.it.utils.TestUtils.*; import static org.junit.Assert.assertEquals; @@ -105,6 +109,14 @@ public class IoTDBWindowTVFIT { "INSERT INTO table5(time, factory_id, device_id, s1) VALUES (1970-01-01T08:00:00.002+08:00, 'F1', 'device_2', 20.0)", "INSERT INTO table5(time, factory_id, device_id, s1) VALUES (1970-01-01T08:00:00.011+08:00, 'F1', 'device_2', 25.0)", "INSERT INTO table5(time, factory_id, device_id, s1) VALUES (1970-01-01T08:00:00.003+08:00, 'F2', 'device_1', 30.0)", + "CREATE TABLE sample_empty(device_id STRING TAG, v DOUBLE FIELD)", + // derivative table function test data + "CREATE TABLE derivative_types(device_id STRING TAG, int_val INT32 FIELD, long_val INT64 FIELD, float_val FLOAT FIELD, double_val DOUBLE FIELD)", + "INSERT INTO derivative_types(time, device_id, int_val, long_val, float_val, double_val) VALUES (1, 'd1', 1, 10, 1.5, 1.25), (2, 'd1', 3, 30, 3.5, 3.75), (3, 'd1', 2, null, 2.5, null)", + "CREATE TABLE derivative_dup(device_id STRING TAG, v DOUBLE FIELD)", + "INSERT INTO derivative_dup(time, device_id, v) VALUES (1, 'd1', 1.0), (1, 'd2', 1.0), (2, 'd1', 2.0), (2, 'd2', 2.0), (3, 'd1', 3.0)", + "CREATE TABLE derivative_null_ts(device_id STRING TAG, v DOUBLE FIELD, ts TIMESTAMP FIELD)", + "INSERT INTO derivative_null_ts(time, device_id, v, ts) VALUES (1, 'd1', 1.0, 2021-01-01T00:00:01), (2, 'd1', 3.0, 2021-01-01T00:00:02), (3, 'd1', 100.0, null), (4, 'd1', 4.0, 2021-01-01T00:00:04)", "FLUSH", "CLEAR ATTRIBUTE CACHE", }; @@ -1469,7 +1481,7 @@ public void testM4CountWindowRejectsOrigin() { public void testM4DescendingOrderByRejected() { tableAssertTestFail( "SELECT * FROM M4(DATA => table1 PARTITION BY device_id ORDER BY time DESC, TIMECOL => 'time', SIZE => 10ms)", - "701: The ORDER BY clause of the DATA argument must sort the time column in ascending order.", + "701: The ORDER BY clause of the DATA argument must be in ascending order.", DATABASE_NAME); } @@ -1738,6 +1750,325 @@ public void testXCorrRejectsUnexpectedCalculationColumnCount() { DATABASE_NAME); } + @Test + public void testSampleFunctionReturnsAllRowsWhenNExceedsPartitionSize() { + String[] expectedHeader = new String[] {"device_id", "time", "s1", "s2", "s3"}; + String[] retArray = + new String[] { + "device_1,1970-01-01T00:00:00.001Z,15.0,12.0,OK,", + "device_1,1970-01-01T00:00:00.003Z,5.0,null,OK,", + "device_1,1970-01-01T00:00:00.006Z,30.0,null,WARN,", + "device_1,1970-01-01T00:00:00.009Z,10.0,null,OK,", + "device_1,1970-01-01T00:00:00.020Z,40.0,35.0,CRIT,", + "device_2,1970-01-01T00:00:00.002Z,8.0,8.0,OK,", + "device_2,1970-01-01T00:00:00.005Z,25.0,24.0,WARN,", + "device_2,1970-01-01T00:00:00.008Z,12.0,13.0,OK,", + "device_2,1970-01-01T00:00:00.011Z,18.0,19.0,OK,", + "device_2,1970-01-01T00:00:00.015Z,6.0,7.0,WARN," + }; + // N exceeds the rows of every partition: all rows are returned, partition columns come first + // and null values are kept as they are. + tableResultSetEqualTest( + "SELECT * FROM SAMPLE(DATA => table1 PARTITION BY device_id ORDER BY time, N => 100) " + + "ORDER BY device_id, time", + expectedHeader, + retArray, + DATABASE_NAME); + } + + @Test + public void testSampleFunctionWithoutPartitionBy() { + String[] expectedHeader = new String[] {"time", "device_id", "s1"}; + String[] retArray = + new String[] { + "1970-01-01T00:00:00.001Z,device_1,15.0,", + "1970-01-01T00:00:00.002Z,device_2,8.0,", + "1970-01-01T00:00:00.003Z,device_1,5.0,", + "1970-01-01T00:00:00.005Z,device_2,25.0,", + "1970-01-01T00:00:00.006Z,device_1,30.0,", + "1970-01-01T00:00:00.008Z,device_2,12.0,", + "1970-01-01T00:00:00.009Z,device_1,10.0,", + "1970-01-01T00:00:00.011Z,device_2,18.0,", + "1970-01-01T00:00:00.015Z,device_2,6.0,", + "1970-01-01T00:00:00.020Z,device_1,40.0," + }; + // Without PARTITION BY the whole table is a single partition, no extra column is produced and + // the input order of the columns is preserved. + tableResultSetEqualTest( + "SELECT * FROM SAMPLE(DATA => (SELECT time, device_id, s1 FROM table1) ORDER BY time, " + + "N => 100) ORDER BY time", + expectedHeader, + retArray, + DATABASE_NAME); + } + + @Test + public void testSampleFunctionWithMultiplePartitionColumns() { + String[] expectedHeader = new String[] {"factory_id", "device_id", "time", "s1"}; + String[] retArray = + new String[] { + "F1,device_1,1970-01-01T00:00:00.001Z,10.0,", + "F1,device_1,1970-01-01T00:00:00.005Z,15.0,", + "F1,device_1,1970-01-01T00:00:00.009Z,5.0,", + "F1,device_2,1970-01-01T00:00:00.002Z,20.0,", + "F1,device_2,1970-01-01T00:00:00.011Z,25.0,", + "F2,device_1,1970-01-01T00:00:00.003Z,30.0," + }; + // The output columns follow the order of PARTITION BY, not the original column order. + tableResultSetEqualTest( + "SELECT * FROM SAMPLE(DATA => table5 PARTITION BY (factory_id, device_id) ORDER BY time, " + + "N => 100) ORDER BY factory_id, device_id, time", + expectedHeader, + retArray, + DATABASE_NAME); + } + + @Test + public void testSampleFunctionKeepsAllColumnTypes() { + String[] expectedHeader = + new String[] { + "device_id", + "time", + "int_val", + "long_val", + "float_val", + "double_val", + "bool_val", + "str_val", + "blob_val", + "ts_val", + "date_val" + }; + String[] retArray = + new String[] { + "device1,2021-01-01T09:00:00.000Z,1,null,1.0,null,null,1,null,2021-01-01T09:00:00.000Z,null,", + "device1,2021-01-01T09:01:00.000Z,2,2,2.0,2.0,true,1,0x01,2021-01-01T09:00:00.000Z,2021-01-01,", + "device1,2021-01-01T09:02:00.000Z,3,3,3.0,3.0,null,2,0x02,2021-01-01T10:00:00.000Z,2021-01-02,", + "device1,2021-01-01T09:03:00.000Z,null,null,null,null,false,null,null,null,null,", + "device1,2021-01-01T09:04:00.000Z,null,null,null,null,null,null,null,null,null,", + "device1,2021-01-01T09:05:00.000Z,3,3,3.0,3.0,false,2,0x02,2021-01-01T10:00:00.000Z,2021-01-02," + }; + tableResultSetEqualTest( + "SELECT * FROM SAMPLE(DATA => multi_type PARTITION BY device_id ORDER BY time, N => 100) " + + "ORDER BY time", + expectedHeader, + retArray, + DATABASE_NAME); + } + + @Test + public void testSampleFunctionWhenNLessThanPartitionSize() { + String[] expectedHeader = new String[] {"device_id", "time", "s1", "s2", "s3"}; + Set validRows = + new HashSet<>( + Arrays.asList( + "device_1,1970-01-01T00:00:00.001Z,15.0,12.0,OK", + "device_1,1970-01-01T00:00:00.003Z,5.0,null,OK", + "device_1,1970-01-01T00:00:00.006Z,30.0,null,WARN", + "device_1,1970-01-01T00:00:00.009Z,10.0,null,OK", + "device_1,1970-01-01T00:00:00.020Z,40.0,35.0,CRIT")); + + List> rowsPerNode = + queryRowsPerNode( + "SELECT * FROM SAMPLE(DATA => (SELECT time, device_id, s1, s2, s3 FROM table1 " + + "WHERE device_id = 'device_1') PARTITION BY device_id ORDER BY time, N => 2)", + expectedHeader); + + // Each node may sample different rows. Check the row count, row integrity and relative order + // independently for every node instead of requiring identical samples across nodes. + assertTrue("no node results", !rowsPerNode.isEmpty()); + for (List rows : rowsPerNode) { + assertEquals(2, rows.size()); + Set sampledTimes = new HashSet<>(); + Instant previousTime = null; + for (String[] row : rows) { + String joinedRow = String.join(",", row); + assertTrue("unexpected sampled row: " + joinedRow, validRows.contains(joinedRow)); + assertTrue("duplicate sampled row: " + row[1], sampledTimes.add(row[1])); + Instant time = Instant.parse(row[1]); + if (previousTime != null) { + assertTrue("sampled rows are not in input order", previousTime.isBefore(time)); + } + previousTime = time; + } + } + } + + @Test + public void testSampleFunctionOnEmptyTable() { + tableResultSetEqualTest( + "SELECT * FROM SAMPLE(DATA => sample_empty PARTITION BY device_id ORDER BY time, N => 10)", + new String[] {"device_id", "time", "v"}, + new String[] {}, + DATABASE_NAME); + } + + @Test + public void testSampleFunctionInvalidArguments() { + tableAssertTestFail( + "SELECT * FROM SAMPLE(DATA => table1 PARTITION BY device_id ORDER BY time, N => 0)", + "N must be a positive integer.", + DATABASE_NAME); + tableAssertTestFail( + "SELECT * FROM SAMPLE(DATA => table1 PARTITION BY device_id ORDER BY time, N => -1)", + "N must be a positive integer.", + DATABASE_NAME); + tableAssertTestFail( + "SELECT * FROM SAMPLE(DATA => table1 PARTITION BY device_id ORDER BY time, N => 1.5)", + "Expected type INT32", + DATABASE_NAME); + tableAssertTestFail( + "SELECT * FROM SAMPLE(DATA => table1 PARTITION BY device_id ORDER BY time, N => null)", + "The parameter N cannot be actively filled with null.", + DATABASE_NAME); + tableAssertTestFail( + "SELECT * FROM SAMPLE(DATA => table1 PARTITION BY device_id ORDER BY time)", + "Missing required argument: N", + DATABASE_NAME); + } + + @Test + public void testDerivativeFunctionWithUnit() { + String[] expectedHeader = + new String[] {"stock_id", "time", "derivative(price)", "derivative(s1)"}; + String[] retArray = + new String[] { + "AAPL,2021-01-01T09:07:00.000Z,3.0,0.0,", + "AAPL,2021-01-01T09:09:00.000Z,-1.0,0.0,", + "TESL,2021-01-01T09:07:00.000Z,4.0,200.0,", + "TESL,2021-01-01T09:15:00.000Z,-1.75,32.5," + }; + // unit => 2m normalizes the change to (v2 - v1) / (delta time / 2 minutes), and the output + // time of a row is the time of the second point of the pair. + tableResultSetEqualWithTolerance( + "SELECT * FROM DERIVATIVE(DATA => bid PARTITION BY stock_id ORDER BY time, unit => 2m) " + + "ORDER BY stock_id, time", + expectedHeader, + retArray, + DATABASE_NAME); + } + + @Test + public void testDerivativeFunctionWithAllNumericTypes() { + String[] expectedHeader = + new String[] { + "device_id", + "time", + "derivative(int_val)", + "derivative(long_val)", + "derivative(float_val)", + "derivative(double_val)" + }; + String[] retArray = + new String[] { + "d1,1970-01-01T00:00:00.002Z,2000.0,20000.0,2000.0,2500.0,", + "d1,1970-01-01T00:00:00.003Z,-1000.0,null,-1000.0,null," + }; + // int32/int64/float/double columns all take part in the calculation, and a null on either side + // only nulls that column of the pair. + tableResultSetEqualWithTolerance( + "SELECT * FROM DERIVATIVE(DATA => derivative_types PARTITION BY device_id ORDER BY time) " + + "ORDER BY time", + expectedHeader, + retArray, + DATABASE_NAME); + } + + @Test + public void testDerivativeFunctionWhenTimestampIsDuplicated() { + String[] expectedHeader = new String[] {"time", "derivative(v)"}; + String[] retArray = + new String[] {"1970-01-01T00:00:00.002Z,1000.0,", "1970-01-01T00:00:00.003Z,1000.0,"}; + // Without PARTITION BY, d1 and d2 rows with the same timestamp are adjacent after sorting. The + // later row overwrites the earlier one, so the pair produces no output and the time never goes + // backwards. + tableResultSetEqualWithTolerance( + "SELECT * FROM DERIVATIVE(DATA => (SELECT time, v FROM derivative_dup) ORDER BY time) " + + "ORDER BY time", + expectedHeader, + retArray, + DATABASE_NAME); + } + + @Test + public void testDerivativeFunctionDropsNullTimestamp() { + String[] expectedHeader = new String[] {"device_id", "ts", "derivative(v)"}; + String[] retArray = + new String[] {"d1,2021-01-01T00:00:02.000Z,2.0,", "d1,2021-01-01T00:00:04.000Z,0.5,"}; + // The ORDER BY column does not have to be the time column of the table: here it is the + // nullable ts field, and the row whose ts is null is dropped before pairing. + tableResultSetEqualWithTolerance( + "SELECT * FROM DERIVATIVE(DATA => (SELECT device_id, v, ts FROM derivative_null_ts) " + + "PARTITION BY device_id ORDER BY ts) ORDER BY ts", + expectedHeader, + retArray, + DATABASE_NAME); + } + + @Test + public void testDerivativeFunctionInvalidArguments() { + tableAssertTestFail( + "SELECT * FROM DERIVATIVE(DATA => (SELECT time, v FROM derivative_dup))", + "Table argument with set semantics requires an ORDER BY clause.", + DATABASE_NAME); + tableAssertTestFail( + "SELECT * FROM DERIVATIVE(DATA => (SELECT time, v FROM derivative_dup) ORDER BY (time, v))", + "The ORDER BY clause of the DATA argument must contain exactly one TIMESTAMP column.", + DATABASE_NAME); + tableAssertTestFail( + "SELECT * FROM DERIVATIVE(DATA => (SELECT time, v FROM derivative_dup) ORDER BY v)", + "The type of the column [v] is not as expected.", + DATABASE_NAME); + tableAssertTestFail( + "SELECT * FROM DERIVATIVE(DATA => (SELECT time, v FROM derivative_dup) ORDER BY time DESC)", + "The ORDER BY clause of the DATA argument must be in ascending order.", + DATABASE_NAME); + tableAssertTestFail( + "SELECT * FROM DERIVATIVE(DATA => (SELECT time FROM derivative_dup) ORDER BY time)", + "No columns could be calculated.", + DATABASE_NAME); + tableAssertTestFail( + "SELECT * FROM DERIVATIVE(DATA => (SELECT time, device_id, v FROM derivative_dup) " + + "ORDER BY time)", + "Only column with double, float, int32, int64 can be calculated by the function, device_id is the STRING.", + DATABASE_NAME); + tableAssertTestFail( + "SELECT * FROM DERIVATIVE(DATA => bid PARTITION BY stock_id ORDER BY time, unit => 0)", + "should be a positive duration of at least 1", + DATABASE_NAME); + } + + private static List> queryRowsPerNode(String sql, String[] expectedHeader) { + List> rowsPerNode = new ArrayList<>(); + try (Connection connection = EnvFactory.getEnv().getTableConnection()) { + connection.setClientInfo("time_zone", "+00:00"); + try (Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE_NAME); + try (ResultSet clusterResultSet = statement.executeQuery(sql)) { + for (ResultSet resultSet : ClusterTestResultSet.getNodeResultSets(clusterResultSet)) { + List rows = new ArrayList<>(); + ResultSetMetaData metaData = resultSet.getMetaData(); + assertEquals(expectedHeader.length, metaData.getColumnCount()); + for (int i = 1; i <= metaData.getColumnCount(); i++) { + assertEquals(expectedHeader[i - 1], metaData.getColumnName(i)); + } + while (resultSet.next()) { + String[] row = new String[expectedHeader.length]; + for (int i = 1; i <= expectedHeader.length; i++) { + row[i - 1] = resultSet.getString(i); + } + rows.add(row); + } + rowsPerNode.add(rows); + } + } + } + } catch (SQLException e) { + fail(e.getMessage()); + } + return rowsPerNode; + } + private static void assertTimeWindowOrdering( String function, String[] expectedHeader, String[] expectedRows) { // A row can belong to several windows. Device/time no longer uniquely identifies a TVF row, diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/OrderByRequirement.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/OrderByRequirement.java new file mode 100644 index 0000000000000..55481ea24494c --- /dev/null +++ b/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 +} diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/TableParameterSpecification.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/TableParameterSpecification.java index a39af87dae15f..df86ae54d79bf 100644 --- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/specification/TableParameterSpecification.java +++ b/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: * *
    *
  • 1. Input tables have either row semantics or set semantics, as follows: @@ -37,19 +37,27 @@ *
  • 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. + *
  • 3. Which requirement the ORDER BY clause of the input table has to satisfy, if any. See + * {@link OrderByRequirement}. *
*/ 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; private TableParameterSpecification( - String name, boolean rowSemantics, boolean passThroughColumns) { + String name, + boolean rowSemantics, + boolean passThroughColumns, + Optional 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 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 = 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); } } } diff --git a/iotdb-core/calc-commons/src/main/i18n/en/org/apache/iotdb/calc/i18n/CalcMessages.java b/iotdb-core/calc-commons/src/main/i18n/en/org/apache/iotdb/calc/i18n/CalcMessages.java index f17a2af31e7dc..df1b628f47a96 100644 --- a/iotdb-core/calc-commons/src/main/i18n/en/org/apache/iotdb/calc/i18n/CalcMessages.java +++ b/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"; diff --git a/iotdb-core/calc-commons/src/main/i18n/zh/org/apache/iotdb/calc/i18n/CalcMessages.java b/iotdb-core/calc-commons/src/main/i18n/zh/org/apache/iotdb/calc/i18n/CalcMessages.java index 1afe27a34f68a..49b48f6a2d64d 100644 --- a/iotdb-core/calc-commons/src/main/i18n/zh/org/apache/iotdb/calc/i18n/CalcMessages.java +++ b/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"; diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/function/TableFunctionOperator.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/function/TableFunctionOperator.java index 0f954897e843e..1956d81e145ea 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/function/TableFunctionOperator.java +++ b/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 recordIterator = slice.getRequiredRecordIterator(requireRecordSnapshot); while (recordIterator.hasNext()) { processor.process( diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/AccumulatorFactory.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/AccumulatorFactory.java index d3f2b2534568e..293fc39162ef7 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/AccumulatorFactory.java +++ b/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); diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/TimeValueBufferBigArray.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/TimeValueBufferBigArray.java similarity index 97% rename from iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/TimeValueBufferBigArray.java rename to iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/TimeValueBufferBigArray.java index be8c882da6109..82a9f2d338f85 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/TimeValueBufferBigArray.java +++ b/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); diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/integral/AbstractGroupedIntegralAccumulator.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/integral/AbstractGroupedIntegralAccumulator.java new file mode 100644 index 0000000000000..876cea1c95f2b --- /dev/null +++ b/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() {} +} diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/integral/GroupedNaiveIntegralAccumulator.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/integral/GroupedNaiveIntegralAccumulator.java new file mode 100644 index 0000000000000..5ff6f71b1e313 --- /dev/null +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/integral/GroupedNaiveIntegralAccumulator.java @@ -0,0 +1,112 @@ +/* + * 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.AggregationMask; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.TimeValueBufferBigArray; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.integral.IntegralFunctionIntermediateStateCodec; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.integral.IntegralFunctionUtils; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.TimeValueBuffer; +import org.apache.iotdb.calc.plan.planner.memory.MemoryReservationManager; + +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.utils.RamUsageEstimator; + +/** Buffers the samples of each group and sorts them when the final result is evaluated. */ +public final class GroupedNaiveIntegralAccumulator extends AbstractGroupedIntegralAccumulator { + + private static final long INSTANCE_SIZE = + RamUsageEstimator.shallowSizeOfInstance(GroupedNaiveIntegralAccumulator.class); + + private final TimeValueBufferBigArray samples = new TimeValueBufferBigArray(); + private final MemoryReservationManager memoryReservationManager; + + public GroupedNaiveIntegralAccumulator( + TSDataType valueDataType, MemoryReservationManager memoryReservationManager) { + super(valueDataType); + this.memoryReservationManager = memoryReservationManager; + } + + @Override + public long getEstimatedSize() { + return INSTANCE_SIZE + samples.sizeOf(); + } + + @Override + public void setGroupCount(long groupCount) { + samples.ensureCapacity(groupCount); + } + + @Override + public void addInput(int[] groupIds, Column[] arguments, AggregationMask mask) { + initializeOrValidateTimeUnit(arguments[TIME_UNIT_ARGUMENT_INDEX]); + + Column valueColumn = arguments[VALUE_ARGUMENT_INDEX]; + Column timeColumn = arguments[TIME_ARGUMENT_INDEX]; + int selectedPositionCount = mask.getSelectedPositionCount(); + boolean selectAll = mask.isSelectAll(); + int[] positions = selectAll ? null : mask.getSelectedPositions(); + for (int i = 0; i < selectedPositionCount; i++) { + int position = selectAll ? i : positions[i]; + if (valueColumn.isNull(position) || timeColumn.isNull(position)) { + continue; + } + samples.add( + groupIds[position], timeColumn.getLong(position), readValue(valueColumn, position)); + } + } + + @Override + public void addIntermediate(int[] groupIds, Column argument) { + for (int position = 0; position < groupIds.length; position++) { + if (argument.isNull(position)) { + continue; + } + int groupId = groupIds[position]; + try (IntegralFunctionIntermediateStateCodec.DecodedState decoded = + IntegralFunctionIntermediateStateCodec.decode( + argument.getBinary(position), memoryReservationManager)) { + initializeOrValidateTimeUnit(decoded.getTimeUnit()); + samples.merge(groupId, decoded.getSamples()); + } + } + } + + @Override + public void evaluateIntermediate(int groupId, ColumnBuilder output) { + IntegralFunctionIntermediateStateCodec.encode( + timeUnit, samples.get(groupId), output, memoryReservationManager); + } + + @Override + public void evaluateFinal(int groupId, ColumnBuilder output) { + TimeValueBuffer buffer = samples.get(groupId); + output.writeDouble( + normalizeArea(buffer == null ? 0.0 : IntegralFunctionUtils.calculateArea(buffer))); + } + + @Override + public void reset() { + samples.reset(); + resetTimeUnit(); + } +} diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/integral/GroupedOrderedIntegralAccumulator.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/integral/GroupedOrderedIntegralAccumulator.java new file mode 100644 index 0000000000000..bd36918ffdd52 --- /dev/null +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/integral/GroupedOrderedIntegralAccumulator.java @@ -0,0 +1,165 @@ +/* + * 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.AggregationMask; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.array.DoubleBigArray; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.array.IntBigArray; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.array.LongBigArray; + +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.utils.RamUsageEstimator; + +/** + * Accumulates the trapezoidal area of each group in a single pass. Only used when the planner has + * proved that the input of every group arrives with ascending timestamps. + */ +public final class GroupedOrderedIntegralAccumulator extends AbstractGroupedIntegralAccumulator { + + private static final long INSTANCE_SIZE = + RamUsageEstimator.shallowSizeOfInstance(GroupedOrderedIntegralAccumulator.class); + + private final IntBigArray sampleCounts = new IntBigArray(); + // "start" is the last point whose value can no longer change, "end" is the newest point. The + // area of the segment between them is added only once a larger timestamp proves that "end" will + // not be overwritten by a later row of the same timestamp. + private final LongBigArray startTimes = new LongBigArray(); + private final DoubleBigArray startValues = new DoubleBigArray(); + private final LongBigArray endTimes = new LongBigArray(); + private final DoubleBigArray endValues = new DoubleBigArray(); + private final DoubleBigArray areas = new DoubleBigArray(); + + public GroupedOrderedIntegralAccumulator(TSDataType valueDataType) { + super(valueDataType); + } + + @Override + public long getEstimatedSize() { + return INSTANCE_SIZE + + sampleCounts.sizeOf() + + startTimes.sizeOf() + + startValues.sizeOf() + + endTimes.sizeOf() + + endValues.sizeOf() + + areas.sizeOf(); + } + + @Override + public void setGroupCount(long groupCount) { + sampleCounts.ensureCapacity(groupCount); + startTimes.ensureCapacity(groupCount); + startValues.ensureCapacity(groupCount); + endTimes.ensureCapacity(groupCount); + endValues.ensureCapacity(groupCount); + areas.ensureCapacity(groupCount); + } + + @Override + public void addInput(int[] groupIds, Column[] arguments, AggregationMask mask) { + initializeOrValidateTimeUnit(arguments[TIME_UNIT_ARGUMENT_INDEX]); + + Column valueColumn = arguments[VALUE_ARGUMENT_INDEX]; + Column timeColumn = arguments[TIME_ARGUMENT_INDEX]; + int selectedPositionCount = mask.getSelectedPositionCount(); + boolean selectAll = mask.isSelectAll(); + int[] positions = selectAll ? null : mask.getSelectedPositions(); + for (int i = 0; i < selectedPositionCount; i++) { + int position = selectAll ? i : positions[i]; + if (valueColumn.isNull(position) || timeColumn.isNull(position)) { + continue; + } + update(groupIds[position], timeColumn.getLong(position), readValue(valueColumn, position)); + } + } + + @Override + public void addIntermediate(int[] groupIds, Column argument) { + throw unsupportedIntermediate(); + } + + @Override + public void evaluateIntermediate(int groupId, ColumnBuilder output) { + throw unsupportedIntermediate(); + } + + @Override + public void evaluateFinal(int groupId, ColumnBuilder output) { + // fewer than two distinct timestamps have no time span, so the area is 0 + if (sampleCounts.get(groupId) < 2) { + output.writeDouble(0.0); + return; + } + // the newest point is final because there is no later input to overwrite it + double area = + areas.get(groupId) + + (startValues.get(groupId) + endValues.get(groupId)) + / 2.0 + * (endTimes.get(groupId) - startTimes.get(groupId)); + output.writeDouble(normalizeArea(area)); + } + + @Override + public void reset() { + sampleCounts.reset(); + startTimes.reset(); + startValues.reset(); + endTimes.reset(); + endValues.reset(); + areas.reset(); + resetTimeUnit(); + } + + private void update(int groupId, long time, double value) { + int sampleCount = sampleCounts.get(groupId); + if (sampleCount == 0) { + startTimes.set(groupId, time); + startValues.set(groupId, value); + endTimes.set(groupId, time); + endValues.set(groupId, value); + sampleCounts.set(groupId, 1); + return; + } + long endTime = endTimes.get(groupId); + if (time < endTime) { + throw orderedInputViolation(time, endTime); + } + if (time == endTime) { + // the later point overwrites the earlier one, so the segment ending at this timestamp is + // added only after its value is final + endValues.set(groupId, value); + return; + } + if (sampleCount >= 2) { + areas.set( + groupId, + areas.get(groupId) + + (startValues.get(groupId) + endValues.get(groupId)) + / 2.0 + * (endTime - startTimes.get(groupId))); + } + startTimes.set(groupId, endTime); + startValues.set(groupId, endValues.get(groupId)); + endTimes.set(groupId, time); + endValues.set(groupId, value); + sampleCounts.set(groupId, Math.incrementExact(sampleCount)); + } +} diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedNaiveDeltaAccumulator.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedNaiveDeltaAccumulator.java index 95d9bc99e5ee0..73e4502c291cb 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedNaiveDeltaAccumulator.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedNaiveDeltaAccumulator.java @@ -20,6 +20,7 @@ package org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.AggregationMask; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.TimeValueBufferBigArray; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.RateFunctionIntermediateStateCodec; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.RateFunctionType; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.RateFunctionValidation; diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedNaiveIncreaseAccumulator.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedNaiveIncreaseAccumulator.java index b7202bd6004b7..cfd15746ce9a9 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedNaiveIncreaseAccumulator.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedNaiveIncreaseAccumulator.java @@ -20,6 +20,7 @@ package org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.AggregationMask; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.TimeValueBufferBigArray; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.RateFunctionIntermediateStateCodec; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.RateFunctionType; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.RateFunctionValidation; diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedNaiveIrateAccumulator.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedNaiveIrateAccumulator.java index 534e09a2bd416..1285bda2ccd47 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedNaiveIrateAccumulator.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedNaiveIrateAccumulator.java @@ -20,6 +20,7 @@ package org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.AggregationMask; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.TimeValueBufferBigArray; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.RateFunctionIntermediateStateCodec; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.RateFunctionType; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.RateFunctionValidation; diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedNaiveRateAccumulator.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedNaiveRateAccumulator.java index 74919ee781062..01fb6745b8e37 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedNaiveRateAccumulator.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedNaiveRateAccumulator.java @@ -20,6 +20,7 @@ package org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.AggregationMask; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.TimeValueBufferBigArray; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.RateFunctionIntermediateStateCodec; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.RateFunctionType; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.RateFunctionValidation; diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/integral/AbstractIntegralTableAccumulator.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/integral/AbstractIntegralTableAccumulator.java new file mode 100644 index 0000000000000..a3921ad45f752 --- /dev/null +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/integral/AbstractIntegralTableAccumulator.java @@ -0,0 +1,87 @@ +/* + * 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.integral; + +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.TableAccumulator; +import org.apache.iotdb.commons.exception.SemanticException; + +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.enums.TSDataType; + +public abstract class AbstractIntegralTableAccumulator implements TableAccumulator { + + // Default time unit of integral() when the second argument is omitted + public static final String DEFAULT_INTEGRAL_TIME_UNIT = "1s"; + + 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 AbstractIntegralTableAccumulator(TSDataType valueDataType) { + this.valueDataType = valueDataType; + } + + protected final double readValue(Column column, int position) { + return IntegralFunctionUtils.readValue(valueDataType, column, position); + } + + /** + * The time unit has already been converted to ticks of the current timestamp precision in the + * analyzer, so only the first value has to be read. + */ + 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 static SemanticException invalidTimeUnit(String text) { + return IntegralFunctionUtils.invalidTimeUnit(text); + } + + protected static SemanticException orderedInputViolation(long time, long previousTime) { + return IntegralFunctionUtils.orderedInputViolation(time, previousTime); + } + + protected static UnsupportedOperationException unsupportedIntermediate() { + return IntegralFunctionUtils.unsupportedIntermediate(); + } +} diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/integral/IntegralFunctionIntermediateStateCodec.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/integral/IntegralFunctionIntermediateStateCodec.java new file mode 100644 index 0000000000000..ffe5d1e8fe818 --- /dev/null +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/integral/IntegralFunctionIntermediateStateCodec.java @@ -0,0 +1,179 @@ +/* + * 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.integral; + +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.TimeValueBuffer; +import org.apache.iotdb.calc.i18n.CalcMessages; +import org.apache.iotdb.calc.plan.planner.memory.MemoryReservationManager; +import org.apache.iotdb.commons.exception.SemanticException; + +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.common.conf.TSFileDescriptor; +import org.apache.tsfile.utils.Binary; + +import java.nio.BufferUnderflowException; +import java.nio.ByteBuffer; + +import static org.apache.iotdb.calc.utils.constant.SqlConstant.INTEGRAL; + +/** Serializes and merges the sampled points carried between partial and final integral(). */ +public final class IntegralFunctionIntermediateStateCodec { + + private static final int STATE_VERSION = 1; + private static final int SAMPLE_SIZE = Long.BYTES + Double.BYTES; + private static final int STATE_HEADER_SIZE = Integer.BYTES + Long.BYTES + Integer.BYTES; + + private IntegralFunctionIntermediateStateCodec() {} + + public static void encode( + long timeUnit, + TimeValueBuffer samples, + ColumnBuilder output, + MemoryReservationManager memoryReservationManager) { + if (samples == null || samples.isEmpty()) { + output.appendNull(); + return; + } + + long serializedSize = + Math.addExact(STATE_HEADER_SIZE, Math.multiplyExact((long) samples.size(), SAMPLE_SIZE)); + validateSerializedSize(serializedSize); + + memoryReservationManager.reserveMemoryImmediately(serializedSize); + try { + ByteBuffer target = ByteBuffer.allocate(Math.toIntExact(serializedSize)); + target.putInt(STATE_VERSION); + target.putLong(timeUnit); + target.putInt(samples.size()); + samples.writePayload(target); + output.writeBinary(new Binary(target.array())); + } finally { + // The output builder owns the serialized bytes after writeBinary returns. + memoryReservationManager.releaseMemoryCumulatively(serializedSize); + } + } + + public static DecodedState decode( + Binary binary, MemoryReservationManager memoryReservationManager) { + byte[] bytes = binary.getValues(); + if (bytes.length > TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes()) { + throw invalidState(); + } + try { + ByteBuffer source = ByteBuffer.wrap(bytes); + if (source.getInt() != STATE_VERSION) { + throw invalidState(); + } + long timeUnit = source.getLong(); + int sampleCount = source.getInt(); + if (sampleCount < 0 + || Math.addExact(STATE_HEADER_SIZE, Math.multiplyExact((long) sampleCount, SAMPLE_SIZE)) + != bytes.length) { + throw invalidState(); + } + + long transientSize = TimeValueBuffer.estimatedSizeForSampleCount(sampleCount); + memoryReservationManager.reserveMemoryImmediately(transientSize); + boolean decoded = false; + try { + TimeValueBuffer samples = new TimeValueBuffer(); + for (int index = 0; index < sampleCount; index++) { + long time = source.getLong(); + double value = source.getDouble(); + if (!Double.isFinite(value)) { + throw new SemanticException( + String.format( + CalcMessages + .EXCEPTION_AGGREGATE_FUNCTION_ARG_VALUE_EXCEEDS_THE_FINITE_RANGE_OF_DOUBLE_48936427, + INTEGRAL, + value)); + } + samples.add(time, value); + } + if (source.hasRemaining()) { + throw invalidState(); + } + decoded = true; + return new DecodedState(timeUnit, samples, memoryReservationManager, transientSize); + } finally { + if (!decoded) { + memoryReservationManager.releaseMemoryCumulatively(transientSize); + } + } + } catch (BufferUnderflowException | ArithmeticException exception) { + throw invalidState(); + } + } + + private static void validateSerializedSize(long serializedSize) { + int maxTsBlockSize = TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes(); + if (serializedSize > maxTsBlockSize) { + throw new SemanticException( + String.format( + CalcMessages + .EXCEPTION_INTERMEDIATE_STATE_FOR_AGGREGATE_FUNCTION_ARG_EXCEEDS_THE_MAXIMUM_TSBLOCK_SIZE_OF_ARG_BYTES_D53A5546, + INTEGRAL, + maxTsBlockSize)); + } + } + + private static SemanticException invalidState() { + return new SemanticException( + String.format( + CalcMessages.EXCEPTION_INVALID_INTERMEDIATE_STATE_FOR_AGGREGATE_FUNCTION_ARG_2999C30B, + INTEGRAL)); + } + + public static final class DecodedState implements AutoCloseable { + + private final long timeUnit; + private final TimeValueBuffer samples; + private final MemoryReservationManager memoryReservationManager; + private final long reservedBytes; + private boolean closed; + + private DecodedState( + long timeUnit, + TimeValueBuffer samples, + MemoryReservationManager memoryReservationManager, + long reservedBytes) { + this.timeUnit = timeUnit; + this.samples = samples; + this.memoryReservationManager = memoryReservationManager; + this.reservedBytes = reservedBytes; + } + + public long getTimeUnit() { + return timeUnit; + } + + public TimeValueBuffer getSamples() { + return samples; + } + + @Override + public void close() { + if (!closed) { + memoryReservationManager.releaseMemoryCumulatively(reservedBytes); + closed = true; + } + } + } +} diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/integral/IntegralFunctionUtils.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/integral/IntegralFunctionUtils.java new file mode 100644 index 0000000000000..4dadd063951f2 --- /dev/null +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/integral/IntegralFunctionUtils.java @@ -0,0 +1,131 @@ +/* + * 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.integral; + +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.TimeValueBuffer; +import org.apache.iotdb.calc.i18n.CalcMessages; +import org.apache.iotdb.commons.exception.SemanticException; + +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.enums.TSDataType; + +import static org.apache.iotdb.calc.utils.constant.SqlConstant.INTEGRAL; + +/** + * Helpers shared by the non-grouped and the grouped implementations of integral(), so that value + * reading, validation messages and the trapezoidal accumulation stay identical in both paths. + */ +public final class IntegralFunctionUtils { + + private IntegralFunctionUtils() {} + + public static double readValue(TSDataType valueDataType, Column column, int position) { + final double value = + switch (valueDataType) { + case INT32 -> column.getInt(position); + case INT64 -> column.getLong(position); + case FLOAT -> column.getFloat(position); + case DOUBLE -> column.getDouble(position); + default -> + throw new SemanticException( + String.format( + CalcMessages + .EXCEPTION_AGGREGATE_FUNCTION_ARG_DOES_NOT_SUPPORT_VALUE_TYPE_ARG_9DD7388D, + INTEGRAL, + valueDataType)); + }; + if (!Double.isFinite(value)) { + throw new SemanticException( + String.format( + CalcMessages + .EXCEPTION_AGGREGATE_FUNCTION_ARG_DOES_NOT_SUPPORT_NON_FINITE_VALUE_COL_ARG_AC2AAC62, + INTEGRAL, + value)); + } + return value; + } + + /** Sorts the samples by time and sums the trapezoidal area of adjacent distinct timestamps. */ + public static double calculateArea(TimeValueBuffer samples) { + // fewer than two effective points have no time span, so the area is 0 + if (samples == null || samples.size() < 2) { + return 0.0; + } + samples.sort(); + // "start" is the last point whose value can no longer change, "end" is the newest point. The + // area of the segment between them is added only once a larger timestamp proves that "end" + // will not be overwritten by a later sample of the same timestamp. + long startTime = samples.getTime(0); + double startValue = samples.getValue(0); + long endTime = startTime; + double endValue = startValue; + int effectiveSampleCount = 1; + double area = 0.0; + for (int index = 1; index < samples.size(); index++) { + long time = samples.getTime(index); + double value = samples.getValue(index); + if (time == endTime) { + // the later point overwrites the earlier one when timestamps are duplicated + endValue = value; + continue; + } + if (effectiveSampleCount >= 2) { + area += (startValue + endValue) / 2.0 * (endTime - startTime); + } + startTime = endTime; + startValue = endValue; + endTime = time; + endValue = value; + effectiveSampleCount++; + } + if (effectiveSampleCount < 2) { + return 0.0; + } + // the newest point is final because there is no later sample to overwrite it + return area + (startValue + endValue) / 2.0 * (endTime - startTime); + } + + public static SemanticException invalidTimeUnit(String text) { + return new SemanticException( + String.format( + CalcMessages + .EXCEPTION_THE_TIME_UNIT_ARGUMENT_OF_AGGREGATE_FUNCTION_ARG_IS_INVALID_89D17438, + INTEGRAL, + text)); + } + + public static SemanticException orderedInputViolation(long time, long previousTime) { + return new SemanticException( + String.format( + CalcMessages + .EXCEPTION_AGGREGATE_FUNCTION_ARG_EXPECTED_TIME_COL_IN_STRICTLY_ASCENDING_ORDER_BUT_GOT_ARG_AFTER_ARG_9289E0F9, + INTEGRAL, + time, + previousTime)); + } + + public static UnsupportedOperationException unsupportedIntermediate() { + return new UnsupportedOperationException( + String.format( + CalcMessages + .EXCEPTION_ORDERED_AGGREGATE_FUNCTION_ARG_DOES_NOT_SUPPORT_INTERMEDIATE_STATE_6B4B2B1B, + INTEGRAL)); + } +} diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/integral/NaiveIntegralAccumulator.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/integral/NaiveIntegralAccumulator.java new file mode 100644 index 0000000000000..2434a0735ccd5 --- /dev/null +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/integral/NaiveIntegralAccumulator.java @@ -0,0 +1,144 @@ +/* + * 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.integral; + +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.AggregationMask; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.TableAccumulator; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.TimeValueBuffer; +import org.apache.iotdb.calc.plan.planner.memory.MemoryReservationManager; + +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.statistics.Statistics; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.RamUsageEstimator; + +/** + * Buffers all samples, sorts them by time when the result is evaluated, and then accumulates the + * trapezoidal area. It is used when the input order is unknown and is the only implementation that + * supports intermediate states. + */ +public final class NaiveIntegralAccumulator extends AbstractIntegralTableAccumulator { + + private static final long INSTANCE_SIZE = + RamUsageEstimator.shallowSizeOfInstance(NaiveIntegralAccumulator.class); + + private final TimeValueBuffer samples = new TimeValueBuffer(); + private final MemoryReservationManager memoryReservationManager; + private long previousSamplesSize; + + public NaiveIntegralAccumulator( + TSDataType valueDataType, MemoryReservationManager memoryReservationManager) { + super(valueDataType); + this.memoryReservationManager = memoryReservationManager; + updateMemoryReservation(); + } + + @Override + public long getEstimatedSize() { + return INSTANCE_SIZE + samples.getEstimatedSize(); + } + + @Override + public TableAccumulator copy() { + return new NaiveIntegralAccumulator(valueDataType, memoryReservationManager); + } + + @Override + public void addInput(Column[] arguments, AggregationMask mask) { + + initializeOrValidateTimeUnit(arguments[TIME_UNIT_ARGUMENT_INDEX]); + + Column valueColumn = arguments[VALUE_ARGUMENT_INDEX]; + Column timeColumn = arguments[TIME_ARGUMENT_INDEX]; + int selectedPositionCount = mask.getSelectedPositionCount(); + boolean selectAll = mask.isSelectAll(); + int[] positions = selectAll ? null : mask.getSelectedPositions(); + for (int i = 0; i < selectedPositionCount; i++) { + int position = selectAll ? i : positions[i]; + if (valueColumn.isNull(position) || timeColumn.isNull(position)) { + continue; + } + samples.add(timeColumn.getLong(position), readValue(valueColumn, position)); + } + updateMemoryReservation(); + } + + @Override + public void addIntermediate(Column argument) { + for (int position = 0; position < argument.getPositionCount(); position++) { + if (argument.isNull(position)) { + continue; + } + decodeAndMerge(argument.getBinary(position)); + } + updateMemoryReservation(); + } + + @Override + public void evaluateIntermediate(ColumnBuilder output) { + IntegralFunctionIntermediateStateCodec.encode( + timeUnit, samples, output, memoryReservationManager); + } + + @Override + public void evaluateFinal(ColumnBuilder output) { + // timeUnit stays 0 when no input has ever been added + double area = IntegralFunctionUtils.calculateArea(samples); + output.writeDouble(timeUnit == 0 ? 0.0 : area / timeUnit); + } + + @Override + public boolean hasFinalResult() { + return false; + } + + @Override + public void addStatistics(Statistics[] statistics) { + throw new UnsupportedOperationException(); + } + + @Override + public void reset() { + resetTimeUnit(); + samples.reset(); + updateMemoryReservation(); + } + + private void decodeAndMerge(Binary binary) { + try (IntegralFunctionIntermediateStateCodec.DecodedState decoded = + IntegralFunctionIntermediateStateCodec.decode(binary, memoryReservationManager)) { + initializeOrValidateTimeUnit(decoded.getTimeUnit()); + samples.merge(decoded.getSamples()); + } + } + + private void updateMemoryReservation() { + long currentSize = samples.getEstimatedSize(); + long delta = currentSize - previousSamplesSize; + if (delta > 0) { + memoryReservationManager.reserveMemoryCumulatively(delta); + } else if (delta < 0) { + memoryReservationManager.releaseMemoryCumulatively(-delta); + } + previousSamplesSize = currentSize; + } +} diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/integral/OrderedIntegralAccumulator.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/integral/OrderedIntegralAccumulator.java new file mode 100644 index 0000000000000..e8f2614b1eacd --- /dev/null +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/integral/OrderedIntegralAccumulator.java @@ -0,0 +1,146 @@ +/* + * 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.integral; + +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.AggregationMask; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.TableAccumulator; + +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.statistics.Statistics; +import org.apache.tsfile.utils.RamUsageEstimator; + +/** + * Accumulates the trapezoidal area in a single pass. Only used when the planner has proved that the + * input of this group arrives with ascending timestamps. + */ +public final class OrderedIntegralAccumulator extends AbstractIntegralTableAccumulator { + + private static final long INSTANCE_SIZE = + RamUsageEstimator.shallowSizeOfInstance(OrderedIntegralAccumulator.class); + + private int sampleCount; + // "start" is the last point whose value can no longer change, "end" is the newest point. The + // area of the segment between them is added only once a larger timestamp proves that "end" will + // not be overwritten by a later row of the same timestamp. + private long startTime; + private double startValue; + private long endTime; + private double endValue; + private double area; + + public OrderedIntegralAccumulator(TSDataType valueDataType) { + super(valueDataType); + } + + @Override + public long getEstimatedSize() { + return INSTANCE_SIZE; + } + + @Override + public TableAccumulator copy() { + return new OrderedIntegralAccumulator(valueDataType); + } + + @Override + public void addInput(Column[] arguments, AggregationMask mask) { + initializeOrValidateTimeUnit(arguments[TIME_UNIT_ARGUMENT_INDEX]); + Column valueColumn = arguments[VALUE_ARGUMENT_INDEX]; + Column timeColumn = arguments[TIME_ARGUMENT_INDEX]; + int selectedPositionCount = mask.getSelectedPositionCount(); + boolean selectAll = mask.isSelectAll(); + int[] positions = selectAll ? null : mask.getSelectedPositions(); + for (int i = 0; i < selectedPositionCount; i++) { + int position = selectAll ? i : positions[i]; + if (valueColumn.isNull(position) || timeColumn.isNull(position)) { + continue; + } + double value = readValue(valueColumn, position); + long time = timeColumn.getLong(position); + if (sampleCount == 0) { + startTime = time; + startValue = value; + endTime = time; + endValue = value; + sampleCount = 1; + } else if (time < endTime) { + throw orderedInputViolation(time, endTime); + } else if (time == endTime) { + // the later point overwrites the earlier one, so the segment ending at this timestamp is + // added only after its value is final + endValue = value; + } else { + if (sampleCount >= 2) { + area += (startValue + endValue) / 2.0 * (endTime - startTime); + } + startTime = endTime; + startValue = endValue; + endTime = time; + endValue = value; + sampleCount++; + } + } + } + + @Override + public void addIntermediate(Column argument) { + throw unsupportedIntermediate(); + } + + @Override + public void evaluateIntermediate(ColumnBuilder columnBuilder) { + throw unsupportedIntermediate(); + } + + @Override + public void evaluateFinal(ColumnBuilder columnBuilder) { + // fewer than two distinct timestamps have no time span, so the area is 0 + if (sampleCount < 2) { + columnBuilder.writeDouble(0.0); + return; + } + // the newest point is final because there is no later input to overwrite it + double finalArea = area + (startValue + endValue) / 2.0 * (endTime - startTime); + columnBuilder.writeDouble(finalArea / timeUnit); + } + + @Override + public boolean hasFinalResult() { + return false; + } + + @Override + public void addStatistics(Statistics[] statistics) { + throw new UnsupportedOperationException(); + } + + @Override + public void reset() { + resetTimeUnit(); + sampleCount = 0; + startTime = 0; + startValue = 0.0; + endTime = 0; + endValue = 0.0; + area = 0.0; + } +} diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/rate/TimeValueBuffer.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/rate/TimeValueBuffer.java index f994dec16d8c4..733ff3d5cdd6a 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/rate/TimeValueBuffer.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/rate/TimeValueBuffer.java @@ -79,7 +79,7 @@ public long getEstimatedSize() { return estimatedSizeForCapacity(times.length); } - static long estimatedSizeForSampleCount(int sampleCount) { + public static long estimatedSizeForSampleCount(int sampleCount) { int capacity = INITIAL_CAPACITY; while (capacity < sampleCount) { capacity = Math.multiplyExact(capacity, 2); @@ -99,26 +99,30 @@ public void writePayload(ByteBuffer target) { } } - public void sortAndValidate(String functionName) { + public void sort() { if (sortedAndValidated) { return; } if (size > 1) { introSort(0, size, 2 * floorLog2(size)); - for (int index = 1; index < size; index++) { - if (times[index - 1] == times[index]) { - throw new SemanticException( - String.format( - CalcMessages - .EXCEPTION_AGGREGATE_FUNCTION_ARG_DOES_NOT_SUPPORT_DUPLICATE_TIME_COL_VALUES_IN_THE_SAME_AGGREGATION_GROUP_ARG_087A91BC, - functionName, - times[index])); - } - } } sortedAndValidated = true; } + public void sortAndValidate(String functionName) { + sort(); + for (int index = 1; index < size; index++) { + if (times[index - 1] == times[index]) { + throw new SemanticException( + String.format( + CalcMessages + .EXCEPTION_AGGREGATE_FUNCTION_ARG_DOES_NOT_SUPPORT_DUPLICATE_TIME_COL_VALUES_IN_THE_SAME_AGGREGATION_GROUP_ARG_087A91BC, + functionName, + times[index])); + } + } + } + private void ensureCapacity(int requiredCapacity) { if (requiredCapacity <= times.length) { return; diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/utils/constant/SqlConstant.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/utils/constant/SqlConstant.java index e08ddf253ff7e..b4e782e7b4893 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/utils/constant/SqlConstant.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/utils/constant/SqlConstant.java @@ -93,6 +93,7 @@ protected SqlConstant() { public static final String INCREASE = "increase"; public static final String IRATE = "irate"; public static final String DELTA = "delta"; + public static final String INTEGRAL = "integral"; // names of scalar functions public static final String DIFF = "diff"; diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index c0b6c32d3fce5..664e554cc5435 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java @@ -2189,6 +2189,10 @@ public final class DataNodeQueryMessages { "Second argument of Aggregate functions [%s] should be orderable"; public static final String AGGREGATE_FUNCTIONS_S_SHOULD_ONLY_HAVE_TWO_OR_THREE_ARGUMENTS = "Aggregate functions [%s] should only have two or three arguments"; + public static final String AGGREGATE_FUNCTIONS_S_SHOULD_ONLY_HAVE_ONE_TWO_OR_THREE_ARGUMENTS_C1ADA261 = + "Aggregate functions [%s] should only have one, two or three arguments"; + public static final String SECOND_ARGUMENT_OF_AGGREGATE_FUNCTIONS_S_SHOULD_BE_A_POSITIVE_DURATION_STRING_DDAEF6E8 = + "Second argument of Aggregate functions [%s] should be a positive duration string (e.g. '1s', '1m', '1h')"; public static final String AGGREGATE_FUNCTIONS_S_SHOULD_ONLY_HAVE_TWO_ARGUMENTS = "Aggregate functions [%s] should only have two arguments"; public static final String SECOND_ARGUMENT_OF_AGGREGATE_FUNCTIONS_S_SHOULD_BE_NUMBERIC_TYPE_AND_DO_NOT_USE = @@ -2412,6 +2416,12 @@ public final class DataNodeQueryMessages { "Invalid argument %s. Partitioning can not be specified for table argument with row semantics"; public static final String INVALID_ARGUMENT_S_ORDERING_CAN_NOT_BE_SPECIFIED_FOR_TABLE_ARGUMENT_WITH_ROW_SEMANTICS = "Invalid argument %s. Ordering can not be specified for table argument with row semantics"; + public static final String + EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_ARGUMENT_S_MUST_BE_IN_ASCENDING_ORDER_5E26E74A = + "The ORDER BY clause of the %s argument must be in ascending order."; + public static final String + EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_ARGUMENT_S_MUST_BE_IN_DESCENDING_ORDER_C2B688D7 = + "The ORDER BY clause of the %s argument must be in descending order."; public static final String INVALID_SCALAR_ARGUMENT_S_EXPECTED_TYPE_S_GOT_S = "Invalid scalar argument '%s'. Expected type %s, got %s"; public static final String INVALID_SCALAR_ARGUMENT_S_S = @@ -4925,4 +4935,7 @@ private DataNodeQueryMessages() {} public static final String EXCEPTION_ARG_ARG_MS_ARG_INVOCATIONS_ARG_APPLICATIONS_F056BD84 = "%s: %s ms, %s invocations, %s applications"; + + public static final String INVALID_SCALAR_ARGUMENT_S_VALUE_OUT_OF_RANGE_FOR_INT32_A865664C = + "Invalid scalar argument '%s'. Value out of range for INT32"; } diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index 522bf93d99ad3..4de1c9c9e0177 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java @@ -2362,6 +2362,12 @@ public final class DataNodeQueryMessages { public static final String AGGREGATE_FUNCTIONS_S_SHOULD_ONLY_HAVE_TWO_OR_THREE_ARGUMENTS = "聚合函数 [%s] 只能有两个或三个参数"; + public static final String AGGREGATE_FUNCTIONS_S_SHOULD_ONLY_HAVE_ONE_TWO_OR_THREE_ARGUMENTS_C1ADA261 = + + "聚合函数 [%s] 只能有一个、两个或三个参数"; + public static final String SECOND_ARGUMENT_OF_AGGREGATE_FUNCTIONS_S_SHOULD_BE_A_POSITIVE_DURATION_STRING_DDAEF6E8 = + + "聚合函数 [%s] 的第二个参数应为正的时间长度字符串（如 '1s'、'1m'、'1h'）"; public static final String AGGREGATE_FUNCTIONS_S_SHOULD_ONLY_HAVE_TWO_ARGUMENTS = "聚合函数 [%s] 只能有两个参数"; @@ -2697,6 +2703,8 @@ public final class DataNodeQueryMessages { public static final String INVALID_ARGUMENT_S_ORDERING_CAN_NOT_BE_SPECIFIED_FOR_TABLE_ARGUMENT_WITH_ROW_SEMANTICS = "无效参数 %s。不能为 row 语义的表参数指定 Ordering"; + public static final String EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_ARGUMENT_S_MUST_BE_IN_ASCENDING_ORDER_5E26E74A = "参数 %s 的 ORDER BY 子句必须为升序。"; + public static final String EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_ARGUMENT_S_MUST_BE_IN_DESCENDING_ORDER_C2B688D7 = "参数 %s 的 ORDER BY 子句必须为降序。"; public static final String INVALID_SCALAR_ARGUMENT_S_EXPECTED_TYPE_S_GOT_S = "无效标量参数 '%s'。期望类型 %s，实际为 %s"; @@ -5625,4 +5633,7 @@ private DataNodeQueryMessages() {} public static final String EXCEPTION_ARG_ARG_MS_ARG_INVOCATIONS_ARG_APPLICATIONS_F056BD84 = "%s：%s ms，调用 %s 次，应用 %s 次"; + + public static final String INVALID_SCALAR_ARGUMENT_S_VALUE_OUT_OF_RANGE_FOR_INT32_A865664C = + "无效标量参数 '%s'。值超出 INT32 的范围"; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/ExpressionAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/ExpressionAnalyzer.java index b95d90e0a275b..2d6e7d8eb2412 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/ExpressionAnalyzer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/ExpressionAnalyzer.java @@ -103,6 +103,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.metadata.OperatorNotFoundException; import org.apache.iotdb.db.queryengine.plan.relational.security.AccessControl; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.StackableAstVisitor; +import org.apache.iotdb.db.utils.DataNodeDateTimeUtils; import com.google.common.collect.HashMultimap; import com.google.common.collect.ImmutableList; @@ -113,6 +114,7 @@ import org.apache.tsfile.read.common.type.RowType; import org.apache.tsfile.read.common.type.TimestampType; import org.apache.tsfile.read.common.type.Type; +import org.apache.tsfile.utils.TimeDuration; import javax.annotation.Nullable; @@ -143,11 +145,13 @@ import static java.util.Collections.unmodifiableSet; import static java.util.Locale.ENGLISH; import static java.util.Objects.requireNonNull; +import static org.apache.iotdb.calc.execution.operator.source.relational.aggregation.integral.AbstractIntegralTableAccumulator.DEFAULT_INTEGRAL_TIME_UNIT; import static org.apache.iotdb.calc.plan.relational.metadata.CommonMetadataUtils.isCharType; import static org.apache.iotdb.calc.plan.relational.metadata.CommonMetadataUtils.isNumericType; import static org.apache.iotdb.calc.plan.relational.metadata.CommonMetadataUtils.isTwoTypeComparable; import static org.apache.iotdb.calc.utils.constant.SqlConstant.FIRST_AGGREGATION; import static org.apache.iotdb.calc.utils.constant.SqlConstant.FIRST_BY_AGGREGATION; +import static org.apache.iotdb.calc.utils.constant.SqlConstant.INTEGRAL; import static org.apache.iotdb.calc.utils.constant.SqlConstant.LAST_AGGREGATION; import static org.apache.iotdb.calc.utils.constant.SqlConstant.LAST_BY_AGGREGATION; import static org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.DereferenceExpression.isQualifiedAllFieldsReference; @@ -1174,6 +1178,51 @@ public Type visitFunctionCall(FunctionCall node, StackableAstVisitorContext integral(c1, '1s', time) + // integral(c1, time_unit) => integral(c1, time_unit, time) + if (argumentsNum == 1) { + node.getArguments().add(new StringLiteral(DEFAULT_INTEGRAL_TIME_UNIT)); + addTimeArgument(node.getArguments(), getActualTimeFieldName(relationType)); + } else if (argumentsNum == 2) { + addTimeArgument(node.getArguments(), getActualTimeFieldName(relationType)); + } else if (argumentsNum == 3) { + if (!checkArgumentIsTimestamp( + node.getArguments().get(2), (List) relationType.getVisibleFields())) { + throw new SemanticException( + String.format( + DataNodeQueryMessages + .THE_THIRD_ARGUMENT_OF_S_FUNCTION_MUST_BE_ACTUAL_TIME_NAME, + functionName)); + } + } else { + throw new SemanticException( + String.format( + DataNodeQueryMessages + .AGGREGATE_FUNCTIONS_S_SHOULD_ONLY_HAVE_ONE_TWO_OR_THREE_ARGUMENTS_C1ADA261, + functionName)); + } + // Normalize time_unit to ticks of the current timestamp precision once, so that the + // accumulator does not have to parse the duration string at execution time + Expression timeUnitArgument = node.getArguments().get(1); + if (!(timeUnitArgument instanceof LongLiteral)) { + if (!(timeUnitArgument instanceof StringLiteral)) { + throw new SemanticException( + String.format( + DataNodeQueryMessages + .SECOND_ARGUMENT_OF_AGGREGATE_FUNCTIONS_S_SHOULD_BE_A_POSITIVE_DURATION_STRING_DDAEF6E8, + functionName)); + } + node.getArguments() + .set( + 1, + new LongLiteral( + String.valueOf( + parseIntegralTimeUnit( + ((StringLiteral) timeUnitArgument).getValue(), functionName)))); + } + break; } List argumentTypes = getCallArgumentTypes(node.getArguments(), context); @@ -1327,6 +1376,23 @@ private String getActualTimeFieldName(RelationType relation) { .MISSING_VALID_TIME_COLUMN_THE_TABLE_MUST_CONTAIN_EITHER_A_COLUMN_WITH_THE_TIME_CATEGORY)); } + /** Parses the time unit and converts it to ticks of the current timestamp precision. */ + private long parseIntegralTimeUnit(String duration, String functionName) { + try { + TimeDuration timeUnit = DataNodeDateTimeUtils.constructTimeDuration(duration); + if (timeUnit.monthDuration == 0 && timeUnit.nonMonthDuration > 0) { + return timeUnit.nonMonthDuration; + } + } catch (IllegalArgumentException e) { + // fall through to report an invalid time unit + } + throw new SemanticException( + String.format( + DataNodeQueryMessages + .SECOND_ARGUMENT_OF_AGGREGATE_FUNCTIONS_S_SHOULD_BE_A_POSITIVE_DURATION_STRING_DDAEF6E8, + functionName)); + } + private Type analyzeMatchNumber( FunctionCall node, StackableAstVisitorContext context) { if (!node.getArguments().isEmpty()) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java index 5fe0b18156bbf..a5aa1b05cf586 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.exception.IoTDBException; import org.apache.iotdb.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.i18n.CommonMessages; import org.apache.iotdb.commons.i18n.QueryMessages; import org.apache.iotdb.commons.queryengine.common.SessionInfo; import org.apache.iotdb.commons.queryengine.plan.relational.analyzer.NodeRef; @@ -226,6 +227,7 @@ import org.apache.iotdb.udf.api.relational.table.argument.DescribedSchema; import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument; import org.apache.iotdb.udf.api.relational.table.argument.TableArgument; +import org.apache.iotdb.udf.api.relational.table.specification.OrderByRequirement; import org.apache.iotdb.udf.api.relational.table.specification.ParameterSpecification; import org.apache.iotdb.udf.api.relational.table.specification.ScalarParameterSpecification; import org.apache.iotdb.udf.api.relational.table.specification.TableParameterSpecification; @@ -6069,7 +6071,9 @@ private boolean isPartitionColumnsProvidedByProperSchema(String functionName) { || TableBuiltinTableFunction.LOWPASS.getFunctionName().equalsIgnoreCase(functionName) || TableBuiltinTableFunction.HIGHPASS.getFunctionName().equalsIgnoreCase(functionName) || TableBuiltinTableFunction.XCORR.getFunctionName().equalsIgnoreCase(functionName) - || TableBuiltinTableFunction.FFT.getFunctionName().equalsIgnoreCase(functionName); + || TableBuiltinTableFunction.FFT.getFunctionName().equalsIgnoreCase(functionName) + || TableBuiltinTableFunction.SAMPLE.getFunctionName().equalsIgnoreCase(functionName) + || TableBuiltinTableFunction.DERIVATIVE.getFunctionName().equalsIgnoreCase(functionName); } private void tryAppendFFTInternalArguments( @@ -6096,37 +6100,12 @@ private void tryAppendFFTInternalArguments( throw new SemanticException(QueryMessages.FFT_N_MUST_BE_POSITIVE_INTEGER); } - validateFFTOrderBySortOrder(arguments, parameterSpecifications); passedArguments.put( FFTTableFunction.SAMPLE_INTERVAL_SPECIFIED_PARAMETER_NAME, new ScalarArgument( org.apache.iotdb.udf.api.type.Type.BOOLEAN, sampleIntervalArgument.isPresent())); } - private void validateFFTOrderBySortOrder( - List arguments, - List parameterSpecifications) { - Optional dataArgument = - findOptionalTableFunctionArgument( - arguments, parameterSpecifications, FFTTableFunction.DATA_PARAMETER_NAME); - if (!dataArgument.isPresent() - || !(dataArgument.get().getValue() instanceof TableFunctionTableArgument)) { - return; - } - - Optional orderBy = - ((TableFunctionTableArgument) dataArgument.get().getValue()).getOrderBy(); - if (!orderBy.isPresent()) { - return; - } - - for (SortItem sortItem : orderBy.get().getSortItems()) { - if (sortItem.getOrdering() != SortItem.Ordering.ASCENDING) { - throw new SemanticException(QueryMessages.FFT_ORDER_BY_MUST_SORT_ASCENDING); - } - } - } - private void tryAppendM4ModeArgument( String functionName, List arguments, @@ -6171,38 +6150,11 @@ && containsTableFunctionArgument( QueryMessages .EXCEPTION_THE_ORIGIN_ARGUMENT_IS_ONLY_SUPPORTED_IN_TIME_WINDOW_MODE_B3F358B9); } - validateM4OrderBySortOrder(arguments, parameterSpecifications); passedArguments.put( M4TableFunction.WINDOW_MODE_PARAMETER_NAME, new ScalarArgument(org.apache.iotdb.udf.api.type.Type.BOOLEAN, isTimeWindow)); } - private void validateM4OrderBySortOrder( - List arguments, - List parameterSpecifications) { - Optional dataArgument = - findOptionalTableFunctionArgument( - arguments, parameterSpecifications, M4TableFunction.DATA_PARAMETER_NAME); - if (!dataArgument.isPresent() - || !(dataArgument.get().getValue() instanceof TableFunctionTableArgument)) { - return; - } - - Optional orderBy = - ((TableFunctionTableArgument) dataArgument.get().getValue()).getOrderBy(); - if (!orderBy.isPresent()) { - return; - } - - for (SortItem sortItem : orderBy.get().getSortItems()) { - if (sortItem.getOrdering() != SortItem.Ordering.ASCENDING) { - throw new SemanticException( - QueryMessages - .EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_DATA_ARGUMENT_MUST_SORT_THE_TIME_COLUMN_IN_ASCENDING_ORDER_C3FD0F6A); - } - } - } - private boolean containsTableFunctionArgument( List arguments, List parameterSpecifications, @@ -6467,6 +6419,14 @@ private ArgumentAnalysis analyzeTableArgument( .collect(toImmutableList()); } + Optional orderByRequirement = + argumentSpecification.getOrderByRequirement(); + if (orderByRequirement.isPresent() && tableArgument.getOrderBy().isEmpty()) { + throw new SemanticException( + CommonMessages + .EXCEPTION_TABLE_ARGUMENT_WITH_SET_SEMANTICS_REQUIRES_AN_ORDER_BY_CLAUSE_10C986D9); + } + // analyze ORDER BY if (tableArgument.getOrderBy().isPresent()) { if (argumentSpecification.isRowSemantics()) { @@ -6478,6 +6438,10 @@ private ArgumentAnalysis analyzeTableArgument( } OrderBy orderByExpression = tableArgument.getOrderBy().get(); analysisBuilder.withOrderBy(orderByExpression); + orderByRequirement.ifPresent( + order -> + validateOrderByDirection( + order, orderByExpression, argumentSpecification.getName())); orderByExpression.getSortItems().stream() .map(SortItem::getSortKey) .forEach( @@ -6523,6 +6487,30 @@ private ArgumentAnalysis analyzeTableArgument( Optional.of(analysisBuilder.build())); } + private static void validateOrderByDirection( + OrderByRequirement orderByRequirement, OrderBy orderBy, String argumentName) { + if (orderByRequirement == OrderByRequirement.ORDERED) { + // any direction is accepted, only the presence of the clause matters + return; + } + final SortItem.Ordering requiredOrdering = + orderByRequirement == OrderByRequirement.ASCENDING + ? SortItem.Ordering.ASCENDING + : SortItem.Ordering.DESCENDING; + for (SortItem sortItem : orderBy.getSortItems()) { + if (sortItem.getOrdering() != requiredOrdering) { + throw new SemanticException( + String.format( + requiredOrdering == SortItem.Ordering.ASCENDING + ? DataNodeQueryMessages + .EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_ARGUMENT_S_MUST_BE_IN_ASCENDING_ORDER_5E26E74A + : DataNodeQueryMessages + .EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_ARGUMENT_S_MUST_BE_IN_DESCENDING_ORDER_C2B688D7, + argumentName)); + } + } + } + private ArgumentAnalysis analyzeScalarArgument( Expression expression, ScalarParameterSpecification argumentSpecification) { if (expression instanceof TimeDurationLiteral) { @@ -6548,7 +6536,15 @@ private ArgumentAnalysis analyzeScalarArgument( constantValue = ((Binary) constantValue).getStringValue(TSFileConfig.STRING_CHARSET); } else if (argumentSpecification.getType().equals(org.apache.iotdb.udf.api.type.Type.INT32) && constantValue instanceof Long) { - constantValue = ((Long) constantValue).intValue(); + try { + constantValue = toIntExact((Long) constantValue); + } catch (ArithmeticException e) { + throw new SemanticException( + String.format( + DataNodeQueryMessages + .INVALID_SCALAR_ARGUMENT_S_VALUE_OUT_OF_RANGE_FOR_INT32_A865664C, + argumentSpecification.getName())); + } } else { throw new SemanticException( String.format( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java index 04dcc93b7312c..2f974257241a9 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java @@ -86,6 +86,7 @@ import static org.apache.iotdb.calc.plan.relational.metadata.CommonMetadataUtils.isDecimalType; import static org.apache.iotdb.calc.plan.relational.metadata.CommonMetadataUtils.isNumericType; +import static org.apache.iotdb.calc.plan.relational.metadata.CommonMetadataUtils.isSupportedMathNumericType; import static org.apache.iotdb.calc.transformation.dag.column.FailFunctionColumnTransformer.FAIL_FUNCTION_NAME; import static org.apache.tsfile.read.common.type.BlobType.BLOB; import static org.apache.tsfile.read.common.type.BooleanType.BOOLEAN; @@ -1464,6 +1465,15 @@ && isNumericType(argumentTypes.get(0)) DataNodeQueryMessages .EXCEPTION_AGGREGATE_FUNCTION_ARG_REQUIRES_4_ARGUMENTS_VALUE_TIME_WINDOW_START_WINDOW_END_FBEC794B); break; + case SqlConstant.INTEGRAL: + if (!isSupportedMathNumericType(argumentTypes.get(0))) { + throw new SemanticException( + String.format( + DataNodeQueryMessages + .AGGREGATE_FUNCTIONS_S_ONLY_SUPPORT_NUMERIC_DATA_TYPES_INT32_INT64_FLOAT_DOUBLE, + functionName)); + } + break; case SqlConstant.COUNT: break; default: @@ -1509,6 +1519,7 @@ && isNumericType(argumentTypes.get(0)) case SqlConstant.INCREASE: case SqlConstant.IRATE: case SqlConstant.DELTA: + case SqlConstant.INTEGRAL: return DOUBLE; case SqlConstant.APPROX_MOST_FREQUENT: return STRING; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java index 6f5f3c278ed3d..199ed70de7af5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java @@ -1613,7 +1613,9 @@ private boolean needAddPartitionColumn(String functionName) { && !TableBuiltinTableFunction.LOWPASS.getFunctionName().equalsIgnoreCase(functionName) && !TableBuiltinTableFunction.HIGHPASS.getFunctionName().equalsIgnoreCase(functionName) && !TableBuiltinTableFunction.XCORR.getFunctionName().equalsIgnoreCase(functionName) - && !TableBuiltinTableFunction.FFT.getFunctionName().equalsIgnoreCase(functionName); + && !TableBuiltinTableFunction.FFT.getFunctionName().equalsIgnoreCase(functionName) + && !TableBuiltinTableFunction.SAMPLE.getFunctionName().equalsIgnoreCase(functionName) + && !TableBuiltinTableFunction.DERIVATIVE.getFunctionName().equalsIgnoreCase(functionName); } private RelationPlan planExternalTsFileScan( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java index 1ead17632aaf1..bde58b0238af4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java @@ -156,6 +156,7 @@ import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Objects; import java.util.Optional; @@ -168,6 +169,7 @@ import static org.apache.iotdb.calc.utils.constant.SqlConstant.COUNT; import static org.apache.iotdb.calc.utils.constant.SqlConstant.DELTA; import static org.apache.iotdb.calc.utils.constant.SqlConstant.INCREASE; +import static org.apache.iotdb.calc.utils.constant.SqlConstant.INTEGRAL; import static org.apache.iotdb.calc.utils.constant.SqlConstant.IRATE; import static org.apache.iotdb.calc.utils.constant.SqlConstant.RATE; import static org.apache.iotdb.commons.partition.DataPartition.NOT_ASSIGNED; @@ -185,6 +187,27 @@ public class TableDistributedPlanGenerator implements PlanVisitor, TableDistributedPlanGenerator.PlanContext> { private static final Logger LOGGER = LoggerFactory.getLogger(TableDistributedPlanGenerator.class); + // Functions that keep per-partition state and therefore require every partition to be handled + // by a single processor. A partition is not guaranteed to be confined to one child of the + // distributed plan: PARTITION BY may use non-device columns, and a device itself may span + // regions. These functions are collected before being evaluated instead of being cloned for + // every child. + private static final Set NON_SPLITTABLE_TABLE_FUNCTIONS = + ImmutableSet.of( + TableBuiltinTableFunction.M4.getFunctionName(), + TableBuiltinTableFunction.FFT.getFunctionName(), + TableBuiltinTableFunction.FORECAST.getFunctionName(), + TableBuiltinTableFunction.PATTERN_MATCH.getFunctionName(), + TableBuiltinTableFunction.CLASSIFY.getFunctionName(), + TableBuiltinTableFunction.LOWPASS.getFunctionName(), + TableBuiltinTableFunction.HIGHPASS.getFunctionName(), + TableBuiltinTableFunction.XCORR.getFunctionName(), + TableBuiltinTableFunction.SAMPLE.getFunctionName(), + TableBuiltinTableFunction.DERIVATIVE.getFunctionName(), + TableBuiltinTableFunction.SESSION.getFunctionName(), + TableBuiltinTableFunction.VARIATION.getFunctionName(), + TableBuiltinTableFunction.CAPACITY.getFunctionName()); + private final MPPQueryContext queryContext; private final QueryId queryId; private final Analysis analysis; @@ -2270,17 +2293,28 @@ static boolean isInputOrderedByTimeAscending( List groupingKeys, OrderingScheme childOrdering) { String functionName = aggregation.getResolvedFunction().getSignature().getName(); - if (step != SINGLE - || childOrdering == null - || aggregation.getArguments().size() < 2 - || !(RATE.equalsIgnoreCase(functionName) - || INCREASE.equalsIgnoreCase(functionName) - || IRATE.equalsIgnoreCase(functionName) - || DELTA.equalsIgnoreCase(functionName))) { + if (step != SINGLE || childOrdering == null) { + return false; + } + + // integral(value, time_unit, time) takes the time column as the third argument, while the + // rate family takes it as the second one + final int timeArgumentIndex; + if (INTEGRAL.equalsIgnoreCase(functionName)) { + timeArgumentIndex = 2; + } else if (RATE.equalsIgnoreCase(functionName) + || INCREASE.equalsIgnoreCase(functionName) + || IRATE.equalsIgnoreCase(functionName) + || DELTA.equalsIgnoreCase(functionName)) { + timeArgumentIndex = 1; + } else { + return false; + } + if (aggregation.getArguments().size() <= timeArgumentIndex) { return false; } - Symbol timeSymbol = Symbol.from(aggregation.getArguments().get(1)); + Symbol timeSymbol = Symbol.from(aggregation.getArguments().get(timeArgumentIndex)); List orderBy = childOrdering.getOrderBy(); int timeIndex = orderBy.indexOf(timeSymbol); if (timeIndex < 0 || !childOrdering.getOrdering(timeSymbol).isAscending()) { @@ -3231,8 +3265,7 @@ public List visitTableFunctionProcessor( node.setChild(childrenNodes.get(0)); result = Collections.singletonList(node); } else if (!canSplitPushDown) { - OrderingScheme childOrdering = commonOrdering(childrenNodes); - node.setChild(mergeChildrenViaCollectOrMergeSort(childOrdering, childrenNodes)); + node.setChild(mergeChildrenForSingleProcessor(node, childrenNodes)); result = Collections.singletonList(node); } else { result = splitForEachChild(node, childrenNodes); @@ -3252,7 +3285,7 @@ private boolean canSplitTableFunctionProcessor(TableFunctionProcessorNode node) if (!isPartitionedGroup(node.getChild())) { return false; } - return !TableBuiltinTableFunction.FFT.getFunctionName().equalsIgnoreCase(node.getName()); + return !NON_SPLITTABLE_TABLE_FUNCTIONS.contains(node.getName().toLowerCase(Locale.ROOT)); } private boolean isPartitionedGroup(PlanNode node) { @@ -3340,6 +3373,49 @@ private PlanNode mergeChildrenViaCollectOrMergeSort( return collectNode; } + /** + * Merges the inputs of a table function processor that has to run once. The processor recognizes + * a partition only while its rows arrive consecutively, so the merged stream must be ordered by + * the GROUP clause: partition keys followed by the in-partition ordering. A merge sort is used + * when every child already carries that ordering; otherwise the merged stream is sorted + * explicitly, because concatenating the children may interleave the rows of a partition that + * spans several of them. + */ + private PlanNode mergeChildrenForSingleProcessor( + TableFunctionProcessorNode node, List childrenNodes) { + OrderingScheme requiredOrdering = null; + if (node.getChild() instanceof GroupNode) { + requiredOrdering = ((GroupNode) node.getChild()).getOrderingScheme(); + } + if (requiredOrdering == null || requiredOrdering.getOrderBy().isEmpty()) { + return mergeChildrenViaCollectOrMergeSort(commonOrdering(childrenNodes), childrenNodes); + } + if (orderedByAllChildren(requiredOrdering, childrenNodes)) { + return mergeChildrenViaCollectOrMergeSort(requiredOrdering, childrenNodes); + } + SortNode sortNode = + new SortNode( + queryId.genPlanNodeId(), + mergeChildrenViaCollectOrMergeSort(null, childrenNodes), + requiredOrdering, + false, + false); + nodeOrderingMap.put(sortNode.getPlanNodeId(), requiredOrdering); + return sortNode; + } + + private boolean orderedByAllChildren( + OrderingScheme requiredOrdering, List childrenNodes) { + int prefixSize = requiredOrdering.getOrderBy().size(); + for (PlanNode child : childrenNodes) { + if (!hasOrderingPrefix( + requiredOrdering, nodeOrderingMap.get(child.getPlanNodeId()), prefixSize)) { + return false; + } + } + return true; + } + private void processSortProperty( final DeviceTableScanNode deviceTableScanNode, final List resultTableScanNodeList, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushAggregationIntoTableScan.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushAggregationIntoTableScan.java index 984a711cc1e76..53dda25120f5e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushAggregationIntoTableScan.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushAggregationIntoTableScan.java @@ -51,6 +51,7 @@ import static org.apache.iotdb.calc.utils.constant.SqlConstant.DELTA; import static org.apache.iotdb.calc.utils.constant.SqlConstant.INCREASE; +import static org.apache.iotdb.calc.utils.constant.SqlConstant.INTEGRAL; import static org.apache.iotdb.calc.utils.constant.SqlConstant.IRATE; import static org.apache.iotdb.calc.utils.constant.SqlConstant.RATE; import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.TAG; @@ -152,7 +153,7 @@ private PushDownLevel calculatePushDownLevel( boolean hasProject = projectNode != null; Map assignments = hasProject ? projectNode.getAssignments().getMap() : null; - if (containsRateFunction(values)) { + if (containsOrderSensitiveFunction(values)) { return PushDownLevel.NOOP; } // calculate Function part @@ -270,7 +271,10 @@ private boolean isDateBinFunctionOfTime( } } - static boolean containsRateFunction(Collection aggregations) { + // These functions compute over the raw time-ordered sequence of each group, so they cannot be + // pre-aggregated inside AggregationTableScan. + static boolean containsOrderSensitiveFunction( + Collection aggregations) { return aggregations.stream() .map(aggregation -> aggregation.getResolvedFunction().getSignature().getName()) .anyMatch( @@ -278,7 +282,8 @@ static boolean containsRateFunction(Collection aggr RATE.equalsIgnoreCase(functionName) || INCREASE.equalsIgnoreCase(functionName) || IRATE.equalsIgnoreCase(functionName) - || DELTA.equalsIgnoreCase(functionName)); + || DELTA.equalsIgnoreCase(functionName) + || INTEGRAL.equalsIgnoreCase(functionName)); } private enum PushDownLevel { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/tvf/TableFunctionOperatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/tvf/TableFunctionOperatorTest.java index cf1117de40ac4..6606db879a830 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/tvf/TableFunctionOperatorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/tvf/TableFunctionOperatorTest.java @@ -518,10 +518,13 @@ public void beforeDestroy() { provider, child, Collections.singletonList(TSDataType.INT64), - Collections.singletonList(TSDataType.INT64), + // the extra output column is a pass-through column, which makes the operator cache + // the input slices until the partition ends, so closing it has retained memory to + // release and ramBytesUsed() has a measurable difference + Arrays.asList(TSDataType.INT64, TSDataType.INT64), 1, Collections.singletonList(0), - Collections.emptyList(), + Collections.singletonList(0), false, Collections.emptyList(), false, diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TableFunctionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TableFunctionTest.java index 637113026f4f0..09c16f3766d48 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TableFunctionTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TableFunctionTest.java @@ -626,7 +626,7 @@ public void testFFTRejectsInvalidArguments() { "Table argument with set semantics requires an ORDER BY clause."); assertAnalyzeFails( "SELECT * FROM FFT(DATA => table1 PARTITION BY tag1 ORDER BY time DESC, SAMPLE_INTERVAL => 1ms)", - "The ORDER BY clause of the DATA argument must sort the time column in ascending order."); + "The ORDER BY clause of the DATA argument must be in ascending order."); assertAnalyzeFails( "SELECT * FROM FFT(DATA => table1 PARTITION BY tag1 ORDER BY s1, SAMPLE_INTERVAL => 1ms)", "The ORDER BY clause of the DATA argument must contain exactly the time column specified by the TIMECOL argument."); @@ -647,6 +647,17 @@ public void testFFTRejectsInvalidArguments() { "No numeric columns found for FFT calculation."); } + @Test + public void testSampleSizeInt32Range() { + analyzeSQL( + "SELECT * FROM TABLE(SAMPLE(DATA => TABLE(table1), N => 10))", + TEST_MATADATA, + QUERY_CONTEXT); + assertAnalyzeFails( + "SELECT * FROM TABLE(SAMPLE(DATA => TABLE(table1), N => 4294967297))", + "Invalid scalar argument 'N'. Value out of range for INT32"); + } + private void assertAnalyzeFails(String sql, String message) { try { analyzeSQL(sql, TEST_MATADATA, QUERY_CONTEXT); @@ -838,8 +849,7 @@ public void testM4RejectsDescendingOrderBy() { fail(); } catch (SemanticException e) { assertEquals( - "The ORDER BY clause of the DATA argument must sort the time column in ascending order.", - e.getMessage()); + "The ORDER BY clause of the DATA argument must be in ascending order.", e.getMessage()); } } } diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java index 94ca9f9f43ab6..154692009516b 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -312,6 +312,15 @@ private CommonMessages() {} public static final String EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_DATA_ARGUMENT_MUST_CONTAIN_EXACTLY_THE_TIME_COLUMN_SPECIFIED_BY_THE_TIMECOL_ARGUMENT_4375BAE9 = "The ORDER BY clause of the DATA argument must contain exactly the time column specified by the TIMECOL argument."; + public static final String + EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_DATA_ARGUMENT_MUST_CONTAIN_EXACTLY_ONE_TIMESTAMP_COLUMN_6B40C59B = + "The ORDER BY clause of the DATA argument must contain exactly one TIMESTAMP column."; + public static final String + EXCEPTION_THE_TIME_COLUMN_OF_THE_DATA_ARGUMENT_MUST_BE_MONOTONICALLY_INCREASING_WITHIN_EACH_PARTITION_68B8C2B7 = + "The time column of the DATA argument must be monotonically increasing within each partition."; + public static final String + EXCEPTION_DERIVATIVE_FUNCTION_UNIT_SHOULD_BE_A_POSITIVE_DURATION_OF_AT_LEAST_ONE_43D975B7 = + "should be a positive duration of at least 1 %s"; public static final String EXCEPTION_UNSUPPORTED_M4_VALUE_TYPE_AF0EF286 = "Unsupported M4 value type: "; public static final String @@ -322,6 +331,8 @@ private CommonMessages() {} "subscription_cache_memory_usage_percentage must be in [0, 1], but was %s."; public static final String EXCEPTION_FILTER_FUNCTION_WPASS_VALIDATION = "the value of wpass should be in (0, 1)"; + public static final String EXCEPTION_SAMPLE_FUNCTION_N_VALIDATION = + "N must be a positive integer."; public static final String EXCEPTION_NO_CALCULATE_COLUMNS = "No columns could be calculated."; public static final String EXCEPTION_NOT_ALLOWED_COLUMNS = "Only column with double, float, int32, int64 can be calculated by the function, %s is the %s."; diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/QueryMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/QueryMessages.java index fc06f96c70271..0f2e293258fdb 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/QueryMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/QueryMessages.java @@ -562,7 +562,6 @@ private QueryMessages() {} public static final String EXCEPTION_INVALID_ARGUMENT_ARG_EXPECTED_SCALAR_ARGUMENT_4AD4E32F = "Invalid argument %s. Expected scalar argument"; public static final String EXCEPTION_THE_SLIDE_ARGUMENT_MUST_HAVE_THE_SAME_WINDOW_MODE_AS_THE_SIZE_ARGUMENT_419C9844 = "The SLIDE argument must have the same window mode as the SIZE argument."; public static final String EXCEPTION_THE_ORIGIN_ARGUMENT_IS_ONLY_SUPPORTED_IN_TIME_WINDOW_MODE_B3F358B9 = "The ORIGIN argument is only supported in time window mode."; - public static final String EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_DATA_ARGUMENT_MUST_SORT_THE_TIME_COLUMN_IN_ASCENDING_ORDER_C3FD0F6A = "The ORDER BY clause of the DATA argument must sort the time column in ascending order."; public static final String EXCEPTION_ARGUMENTS_SHOULD_NEVER_BE_EMPTY_WHEN_RESOLVING_M4_MODE_F982EA1C = "Arguments should never be empty when resolving M4 mode"; public static final String EXCEPTION_MISSING_REQUIRED_ARGUMENT_ARG_1308D0DC = "Missing required argument: %s"; public static final String EXCEPTION_UNKNOWN_ARGUMENT_ARG_F40E9394 = "Unknown argument: %s"; @@ -598,8 +597,6 @@ private QueryMessages() {} "The SAMPLE_INTERVAL argument of FFT must be a duration literal."; public static final String FFT_N_MUST_BE_POSITIVE_INTEGER = "The N argument of FFT must be a positive integer."; - public static final String FFT_ORDER_BY_MUST_SORT_ASCENDING = - "The ORDER BY clause of the DATA argument must sort the time column in ascending order."; public static final String EXCEPTION_FFT_LENGTH_MUST_BE_A_POSITIVE_INT_SIZED_VALUE_A000D3BB = "FFT length must be a positive int-sized value."; public static final String diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java index e05d6b32be658..4e7150cd0c92f 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -214,6 +214,9 @@ private CommonMessages() {} public static final String EXCEPTION_NO_COMPARABLE_COLUMNS_FOUND_FOR_M4_CALCULATION_4E5A3092 = "未找到可用于 M4 计算的可比较列。"; public static final String EXCEPTION_INVALID_SCALAR_ARGUMENT_SLIDE_SHOULD_BE_A_POSITIVE_VALUE_F019E091 = "无效的标量参数 SLIDE，应为正数"; public static final String EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_DATA_ARGUMENT_MUST_CONTAIN_EXACTLY_THE_TIME_COLUMN_SPECIFIED_BY_THE_TIMECOL_ARGUMENT_4375BAE9 = "DATA 参数的 ORDER BY 子句必须仅包含 TIMECOL 参数指定的时间列。"; + public static final String EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_DATA_ARGUMENT_MUST_CONTAIN_EXACTLY_ONE_TIMESTAMP_COLUMN_6B40C59B = "DATA 参数的 ORDER BY 子句必须仅包含一个 TIMESTAMP 列。"; + public static final String EXCEPTION_THE_TIME_COLUMN_OF_THE_DATA_ARGUMENT_MUST_BE_MONOTONICALLY_INCREASING_WITHIN_EACH_PARTITION_68B8C2B7 = "DATA 参数的时间列在每个分区内必须单调递增。"; + public static final String EXCEPTION_DERIVATIVE_FUNCTION_UNIT_SHOULD_BE_A_POSITIVE_DURATION_OF_AT_LEAST_ONE_43D975B7 = "应为正的时间长度，且至少为 1 %s"; public static final String EXCEPTION_UNSUPPORTED_M4_VALUE_TYPE_AF0EF286 = "不支持的 M4 值类型："; public static final String EXCEPTION_DISK_SPACE_WARNING_THRESHOLD_MUST_BE_IN_0_1_BUT_WAS_7B345766 = "disk_space_warning_threshold 必须在 [0, 1) 范围内，但实际为 "; public static final String @@ -225,6 +228,7 @@ private CommonMessages() {} LOG_USER_DATA_TRANSFER_ATTEMPT_TIME_ARG_INITIATOR_ARG_SOURCE_ARG_TARGET_ARG_PROTECTION_METHOD_ARG_RESULT_ARG_ERROR_ARG_D3E9A1DF = "用户数据传送尝试：时间=%d，发起者=%s，源端=%s，目标端=%s，保护方法=%s，结果=%s，错误=%s"; public static final String EXCEPTION_FILTER_FUNCTION_WPASS_VALIDATION = "wpass的取值范围应该位于(0, 1)"; + public static final String EXCEPTION_SAMPLE_FUNCTION_N_VALIDATION = "N 要求是正整数"; public static final String EXCEPTION_NO_CALCULATE_COLUMNS = "没有找到可以计算的列."; public static final String EXCEPTION_NOT_ALLOWED_COLUMNS = "只允许列类型为double, float, int32, int64参与函数计算, 当前列 %s 类型是 %s."; public static final String EXCEPTION_FILTER_FUNCTION_ROW_INDEX_EXCEED_MAXIMUM = "分区行数超过了最大限制"; diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/QueryMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/QueryMessages.java index e08448271bbb2..30692ea920d9d 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/QueryMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/QueryMessages.java @@ -563,7 +563,6 @@ private QueryMessages() {} public static final String EXCEPTION_INVALID_ARGUMENT_ARG_EXPECTED_SCALAR_ARGUMENT_4AD4E32F = "无效的参数 %s。期望标量参数"; public static final String EXCEPTION_THE_SLIDE_ARGUMENT_MUST_HAVE_THE_SAME_WINDOW_MODE_AS_THE_SIZE_ARGUMENT_419C9844 = "SLIDE 参数必须与 SIZE 参数具有相同的窗口模式。"; public static final String EXCEPTION_THE_ORIGIN_ARGUMENT_IS_ONLY_SUPPORTED_IN_TIME_WINDOW_MODE_B3F358B9 = "ORIGIN 参数仅在时间窗口模式下受支持。"; - public static final String EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_DATA_ARGUMENT_MUST_SORT_THE_TIME_COLUMN_IN_ASCENDING_ORDER_C3FD0F6A = "DATA 参数的 ORDER BY 子句必须按时间列升序排序。"; public static final String EXCEPTION_ARGUMENTS_SHOULD_NEVER_BE_EMPTY_WHEN_RESOLVING_M4_MODE_F982EA1C = "解析 M4 模式时参数不应为空"; public static final String EXCEPTION_MISSING_REQUIRED_ARGUMENT_ARG_1308D0DC = "缺少必要参数：%s"; public static final String EXCEPTION_UNKNOWN_ARGUMENT_ARG_F40E9394 = "未知参数：%s"; @@ -596,8 +595,6 @@ private QueryMessages() {} public static final String FFT_SAMPLE_INTERVAL_MUST_BE_DURATION_LITERAL = "FFT 的 SAMPLE_INTERVAL 参数必须是 duration 字面量。"; public static final String FFT_N_MUST_BE_POSITIVE_INTEGER = "FFT 的 N 参数必须是正整数。"; - public static final String FFT_ORDER_BY_MUST_SORT_ASCENDING = - "DATA 参数的 ORDER BY 子句必须按时间列升序排序。"; public static final String EXCEPTION_FFT_LENGTH_MUST_BE_A_POSITIVE_INT_SIZED_VALUE_A000D3BB = "FFT 长度必须是正数且不能超过 int 范围。"; public static final String diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/relational/function/TableBuiltinTableFunction.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/relational/function/TableBuiltinTableFunction.java index 410ca7c6ca4e1..b263e3c5ab651 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/relational/function/TableBuiltinTableFunction.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/relational/function/TableBuiltinTableFunction.java @@ -25,11 +25,13 @@ import org.apache.iotdb.commons.queryengine.plan.relational.function.tvf.PatternMatchTableFunction; import org.apache.iotdb.commons.udf.builtin.relational.tvf.CapacityTableFunction; import org.apache.iotdb.commons.udf.builtin.relational.tvf.CumulateTableFunction; +import org.apache.iotdb.commons.udf.builtin.relational.tvf.DerivativeTableFunction; import org.apache.iotdb.commons.udf.builtin.relational.tvf.FFTTableFunction; import org.apache.iotdb.commons.udf.builtin.relational.tvf.HOPTableFunction; import org.apache.iotdb.commons.udf.builtin.relational.tvf.HighPassTableFunction; import org.apache.iotdb.commons.udf.builtin.relational.tvf.LowPassTableFunction; import org.apache.iotdb.commons.udf.builtin.relational.tvf.M4TableFunction; +import org.apache.iotdb.commons.udf.builtin.relational.tvf.SampleTableFunction; import org.apache.iotdb.commons.udf.builtin.relational.tvf.SessionTableFunction; import org.apache.iotdb.commons.udf.builtin.relational.tvf.TumbleTableFunction; import org.apache.iotdb.commons.udf.builtin.relational.tvf.VariationTableFunction; @@ -56,7 +58,9 @@ public enum TableBuiltinTableFunction { CLASSIFY("classify"), LOWPASS("lowpass"), HIGHPASS("highpass"), - XCORR("xcorr"); + XCORR("xcorr"), + SAMPLE("sample"), + DERIVATIVE("derivative"); private final String functionName; private final boolean preservesInputOrder; @@ -133,6 +137,10 @@ public static TableFunction getBuiltinTableFunction(String functionName) { return new HighPassTableFunction(); case "xcorr": return new XCorrTableFunction(); + case "sample": + return new SampleTableFunction(); + case "derivative": + return new DerivativeTableFunction(); default: throw new UnsupportedOperationException( String.format(QueryMessages.UNSUPPORTED_TABLE_FUNCTION, functionName)); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/TableBuiltinAggregationFunction.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/TableBuiltinAggregationFunction.java index d232b7b77ef63..74e6bf87a6ff9 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/TableBuiltinAggregationFunction.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/TableBuiltinAggregationFunction.java @@ -71,7 +71,8 @@ public enum TableBuiltinAggregationFunction { RATE("rate"), INCREASE("increase"), IRATE("irate"), - DELTA("delta"); + DELTA("delta"), + INTEGRAL("integral"); private final String functionName; @@ -131,6 +132,7 @@ public static Type getIntermediateType(String name, List originalArgumentT case "increase": case "irate": case "delta": + case "integral": return BLOB; case "extreme": case "max": diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/DerivativeTableFunction.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/DerivativeTableFunction.java new file mode 100644 index 0000000000000..8f01200b188de --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/DerivativeTableFunction.java @@ -0,0 +1,346 @@ +/* + * 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.commons.udf.builtin.relational.tvf; + +import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.i18n.CommonMessages; +import org.apache.iotdb.commons.queryengine.utils.TimestampPrecisionUtils; +import org.apache.iotdb.udf.api.exception.UDFException; +import org.apache.iotdb.udf.api.relational.TableFunction; +import org.apache.iotdb.udf.api.relational.access.Record; +import org.apache.iotdb.udf.api.relational.table.MapTableFunctionHandle; +import org.apache.iotdb.udf.api.relational.table.TableFunctionAnalysis; +import org.apache.iotdb.udf.api.relational.table.TableFunctionHandle; +import org.apache.iotdb.udf.api.relational.table.TableFunctionProcessorProvider; +import org.apache.iotdb.udf.api.relational.table.argument.Argument; +import org.apache.iotdb.udf.api.relational.table.argument.DescribedSchema; +import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument; +import org.apache.iotdb.udf.api.relational.table.argument.TableArgument; +import org.apache.iotdb.udf.api.relational.table.processor.TableFunctionDataProcessor; +import org.apache.iotdb.udf.api.relational.table.specification.ParameterSpecification; +import org.apache.iotdb.udf.api.relational.table.specification.ScalarParameterSpecification; +import org.apache.iotdb.udf.api.relational.table.specification.TableParameterSpecification; +import org.apache.iotdb.udf.api.type.Type; + +import org.apache.tsfile.block.column.ColumnBuilder; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +public class DerivativeTableFunction implements TableFunction { + + public static final String DATA_PARAMETER_NAME = "DATA"; + public static final String UNIT_PARAMETER_NAME = "UNIT"; + + // duration literals are evaluated in the current timestamp precision, so the 1s default has to + // be expressed in that precision as well + private static final long DEFAULT_UNIT = + TimestampPrecisionUtils.currPrecision.convert(1, TimeUnit.SECONDS); + private static final String OUTPUT_TYPES_PROPERTY = "OUTPUT_TYPES"; + private static final String PARTITION_COLUMN_COUNT_PROPERTY = "PARTITION_COLUMN_COUNT"; + private static final String UNIT_PROPERTY = "UNIT"; + + @Override + public List getArgumentsSpecifications() { + return Arrays.asList( + TableParameterSpecification.builder() + .name(DATA_PARAMETER_NAME) + .setSemantics() + .ascendingOrder() + .build(), + ScalarParameterSpecification.builder() + .name(UNIT_PARAMETER_NAME) + .type(Type.INT64) + .defaultValue(DEFAULT_UNIT) + .addChecker( + object -> { + // a duration smaller than the smallest unit of the current timestamp precision + // has already been truncated to 0 when the argument was evaluated, so explain + // this case explicitly + if (object instanceof Long && (Long) object > 0) { + return null; + } + return String.format( + CommonMessages + .EXCEPTION_DERIVATIVE_FUNCTION_UNIT_SHOULD_BE_A_POSITIVE_DURATION_OF_AT_LEAST_ONE_43D975B7, + TimestampPrecisionUtils.TIMESTAMP_PRECISION); + }) + .build()); + } + + @Override + public TableFunctionAnalysis analyze(Map arguments) throws UDFException { + TableArgument tableArgument = (TableArgument) arguments.get(DATA_PARAMETER_NAME); + + // ORDER BY is mandatory here and must contain exactly one TIMESTAMP column: the derivative is + // taken with respect to this column, which is also emitted as the time of each output row. + List orderBy = checkAndGetOrderBy(tableArgument); + int timeColumnIndex = + WindowTVFUtils.findColumnIndex( + tableArgument, orderBy.get(0), Collections.singleton(Type.TIMESTAMP)); + + List partitionIndexes = WindowTVFUtils.getPartitionIndexes(tableArgument); + Set excludedIndexes = new HashSet<>(partitionIndexes); + excludedIndexes.add(timeColumnIndex); + + List outputTypes = new ArrayList<>(); + DescribedSchema.Builder schemaBuilder = new DescribedSchema.Builder(); + + // keep the partition columns with their original names and types, in PARTITION BY order + for (int partitionIndex : partitionIndexes) { + String columnName = + tableArgument + .getFieldNames() + .get(partitionIndex) + .orElseThrow( + () -> new SemanticException(CommonMessages.EXCEPTION_COLUMN_LACK_OF_NAME)); + Type columnType = tableArgument.getFieldTypes().get(partitionIndex); + outputTypes.add(columnType); + schemaBuilder.addField(columnName, columnType); + } + + // each output row corresponds to a pair of adjacent points and carries the time of the second + // point of the pair, so the time column is emitted as is. + String timeColumnName = + tableArgument + .getFieldNames() + .get(timeColumnIndex) + .orElseThrow(() -> new SemanticException(CommonMessages.EXCEPTION_COLUMN_LACK_OF_NAME)); + outputTypes.add(Type.TIMESTAMP); + schemaBuilder.addField(timeColumnName, Type.TIMESTAMP); + List calculationIndexes = + WindowTVFUtils.getCalculationIndexes( + tableArgument, + excludedIndexes, + columnName -> { + outputTypes.add(Type.DOUBLE); + schemaBuilder.addField(convertColumnName(columnName), Type.DOUBLE); + }); + if (calculationIndexes.isEmpty()) { + throw new SemanticException(CommonMessages.EXCEPTION_NO_CALCULATE_COLUMNS); + } + + // the output columns are in the same order as the required input columns: partition columns, + // the time column and one DOUBLE column per calculation column. The processor therefore only + // needs the output types and the number of partition columns to walk both of them. + MapTableFunctionHandle handle = + new MapTableFunctionHandle.Builder() + .addProperty(OUTPUT_TYPES_PROPERTY, WindowTVFUtils.joinTypes(outputTypes)) + .addProperty(PARTITION_COLUMN_COUNT_PROPERTY, partitionIndexes.size()) + .addProperty( + UNIT_PROPERTY, ((ScalarArgument) arguments.get(UNIT_PARAMETER_NAME)).getValue()) + .build(); + + List requiredColumns = new ArrayList<>(partitionIndexes); + requiredColumns.add(timeColumnIndex); + requiredColumns.addAll(calculationIndexes); + + return TableFunctionAnalysis.builder() + .properColumnSchema(schemaBuilder.build()) + .requireRecordSnapshot(false) + .requiredColumns(DATA_PARAMETER_NAME, requiredColumns) + .handle(handle) + .build(); + } + + private static List checkAndGetOrderBy(TableArgument tableArgument) { + List orderBy = tableArgument.getOrderBy(); + if (orderBy.isEmpty()) { + throw new SemanticException( + CommonMessages + .EXCEPTION_TABLE_ARGUMENT_WITH_SET_SEMANTICS_REQUIRES_AN_ORDER_BY_CLAUSE_10C986D9); + } + if (orderBy.size() != 1) { + throw new SemanticException( + CommonMessages + .EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_DATA_ARGUMENT_MUST_CONTAIN_EXACTLY_ONE_TIMESTAMP_COLUMN_6B40C59B); + } + return orderBy; + } + + private static String convertColumnName(String columnName) { + return String.format("derivative(%s)", columnName); + } + + @Override + public TableFunctionHandle createTableFunctionHandle() { + return new MapTableFunctionHandle(); + } + + @Override + public TableFunctionProcessorProvider getProcessorProvider( + TableFunctionHandle tableFunctionHandle) { + MapTableFunctionHandle handle = (MapTableFunctionHandle) tableFunctionHandle; + Type[] outputTypes = + WindowTVFUtils.parseTypes((String) handle.getProperty(OUTPUT_TYPES_PROPERTY)); + int partitionColumnCount = (Integer) handle.getProperty(PARTITION_COLUMN_COUNT_PROPERTY); + long unit = (Long) handle.getProperty(UNIT_PROPERTY); + return new TableFunctionProcessorProvider() { + @Override + public TableFunctionDataProcessor getDataProcessor() { + return new DerivativeDataProcessor(partitionColumnCount, outputTypes, unit); + } + }; + } + + /** + * Streaming processor over one partition. Each row is paired with the previous one, but the pair + * is not complete until the time of its second point is known to be final, because a later row + * may share that timestamp and overwrite its values. The newest point is therefore buffered and + * the pair ending at it is emitted when a larger timestamp arrives; {@code finish} emits the last + * pair. + * + *

The required input columns are the partition columns, the time column (the single ORDER BY + * column of DATA) and the calculation columns, in that order, which is also the order of the + * output columns. + */ + private static class DerivativeDataProcessor implements TableFunctionDataProcessor { + + private final Type[] outputTypes; + private final int partitionColumnCount; + private final int timeColumnIndex; + private final int calculationColumnCount; + private final long unit; + + // all rows of a partition share the same partition values, so they are captured from the first + // row and reused for every output row, including the one emitted in finish where no input row + // is available. + private Object[] partitionValues; + // "previous" is the finalized point right before "current". "current" is the newest point, + // whose values may still be overwritten by a later row with the same timestamp. + private Point previous; + private Point current; + + private DerivativeDataProcessor(int partitionColumnCount, Type[] outputTypes, long unit) { + this.outputTypes = outputTypes; + this.partitionColumnCount = partitionColumnCount; + this.timeColumnIndex = partitionColumnCount; + this.calculationColumnCount = outputTypes.length - partitionColumnCount - 1; + this.unit = unit; + } + + @Override + public void process( + Record input, + List properColumnBuilders, + ColumnBuilder passThroughIndexBuilder) { + // the ORDER BY column of DATA is not necessarily the implicit time column and may be null; + // such a row cannot be paired with anything and is dropped. + if (input.isNull(timeColumnIndex)) { + return; + } + long timestamp = input.getLong(timeColumnIndex); + + if (current == null) { + capturePartitionValues(input); + current = snapshot(input, timestamp); + return; + } + if (timestamp < current.timestamp) { + throw new SemanticException( + CommonMessages + .EXCEPTION_THE_TIME_COLUMN_OF_THE_DATA_ARGUMENT_MUST_BE_MONOTONICALLY_INCREASING_WITHIN_EACH_PARTITION_68B8C2B7); + } + if (timestamp == current.timestamp) { + // the same timestamp may appear across devices: the later row overwrites the earlier one, + // and the pair ending at this timestamp is not emitted until its value is final + current = snapshot(input, timestamp); + return; + } + + // a larger timestamp proves that no row can overwrite "current" anymore, so the pair ending + // at it is final and can be emitted + if (previous != null) { + emitDerivative(previous, current, properColumnBuilders); + } + previous = current; + current = snapshot(input, timestamp); + } + + @Override + public void finish( + List properColumnBuilders, ColumnBuilder passThroughIndexBuilder) { + if (previous != null && current != null) { + emitDerivative(previous, current, properColumnBuilders); + // clear the buffered point so that a repeated finish emits nothing + previous = null; + } + } + + private Point snapshot(Record input, long timestamp) { + Object[] values = new Object[calculationColumnCount]; + for (int i = 0; i < calculationColumnCount; i++) { + int columnIndex = timeColumnIndex + 1 + i; + // native boxed values of the calculation columns, null when the value is null + values[i] = input.isNull(columnIndex) ? null : input.getObject(columnIndex); + } + return new Point(timestamp, values); + } + + private void capturePartitionValues(Record input) { + partitionValues = new Object[partitionColumnCount]; + for (int i = 0; i < partitionColumnCount; i++) { + partitionValues[i] = + input.isNull(i) ? null : WindowTVFUtils.readValue(input, i, outputTypes[i]); + } + } + + private void emitDerivative(Point start, Point end, List properColumnBuilders) { + for (int i = 0; i < partitionColumnCount; i++) { + WindowTVFUtils.writeValue(properColumnBuilders.get(i), partitionValues[i], outputTypes[i]); + } + properColumnBuilders.get(timeColumnIndex).writeLong(end.timestamp); + + long timeDelta = end.timestamp - start.timestamp; + for (int i = 0; i < calculationColumnCount; i++) { + int columnIndex = timeColumnIndex + 1 + i; + ColumnBuilder builder = properColumnBuilders.get(columnIndex); + Object startValue = start.values[i]; + Object endValue = end.values[i]; + // a null on either side only makes this column null for this pair; pairing and time still + // advance normally. + if (startValue == null || endValue == null) { + builder.appendNull(); + } else { + double valueDelta = + ((Number) endValue).doubleValue() - ((Number) startValue).doubleValue(); + builder.writeDouble(valueDelta * unit / timeDelta); + } + } + } + + private static final class Point { + + private final long timestamp; + private final Object[] values; + + private Point(long timestamp, Object[] values) { + this.timestamp = timestamp; + this.values = values; + } + } + } +} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FFTTableFunction.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FFTTableFunction.java index ebdd089c1725c..253977ada2a8b 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FFTTableFunction.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FFTTableFunction.java @@ -100,7 +100,11 @@ public class FFTTableFunction implements TableFunction { @Override public List getArgumentsSpecifications() { return Arrays.asList( - TableParameterSpecification.builder().name(DATA_PARAMETER_NAME).setSemantics().build(), + TableParameterSpecification.builder() + .name(DATA_PARAMETER_NAME) + .setSemantics() + .ascendingOrder() + .build(), ScalarParameterSpecification.builder() .name(SAMPLE_INTERVAL_PARAMETER_NAME) .type(Type.INT64) @@ -201,8 +205,8 @@ public TableFunctionAnalysis analyze(Map arguments) throws UDF .getValue()) .addProperty(N_PARAMETER_NAME, transformLength) .addProperty(NORM_PARAMETER_NAME, norm) - .addProperty(PARTITION_TYPES_PROPERTY, joinTypes(partitionTypes)) - .addProperty(VALUE_TYPES_PROPERTY, joinTypes(valueTypes)) + .addProperty(PARTITION_TYPES_PROPERTY, WindowTVFUtils.joinTypes(partitionTypes)) + .addProperty(VALUE_TYPES_PROPERTY, WindowTVFUtils.joinTypes(valueTypes)) .addProperty(VALUE_NAMES_PROPERTY, encodeStrings(valueNames)) .build(); @@ -233,8 +237,10 @@ public TableFunctionProcessorProvider getProcessorProvider( long sampleInterval = (long) handle.getProperty(SAMPLE_INTERVAL_PARAMETER_NAME); long transformLength = (long) handle.getProperty(N_PARAMETER_NAME); String norm = (String) handle.getProperty(NORM_PARAMETER_NAME); - Type[] partitionTypes = parseTypes((String) handle.getProperty(PARTITION_TYPES_PROPERTY)); - Type[] valueTypes = parseTypes((String) handle.getProperty(VALUE_TYPES_PROPERTY)); + Type[] partitionTypes = + WindowTVFUtils.parseTypes((String) handle.getProperty(PARTITION_TYPES_PROPERTY)); + Type[] valueTypes = + WindowTVFUtils.parseTypes((String) handle.getProperty(VALUE_TYPES_PROPERTY)); String[] valueNames = decodeStrings((String) handle.getProperty(VALUE_NAMES_PROPERTY)); return new TableFunctionProcessorProvider() { @@ -293,29 +299,6 @@ private static void validateNorm(String norm) { } } - private static String joinTypes(List types) { - StringBuilder builder = new StringBuilder(); - for (int i = 0; i < types.size(); i++) { - if (i > 0) { - builder.append(','); - } - builder.append(types.get(i).name()); - } - return builder.toString(); - } - - private static Type[] parseTypes(String value) { - if (value.isEmpty()) { - return new Type[0]; - } - String[] values = value.split(","); - Type[] types = new Type[values.length]; - for (int i = 0; i < values.length; i++) { - types[i] = Type.valueOf(values[i]); - } - return types; - } - private static String encodeStrings(List values) { StringBuilder builder = new StringBuilder(); for (int i = 0; i < values.size(); i++) { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/M4TableFunction.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/M4TableFunction.java index 7b508fcb53d65..8f30b9eda5453 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/M4TableFunction.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/M4TableFunction.java @@ -75,7 +75,11 @@ public class M4TableFunction implements TableFunction { @Override public List getArgumentsSpecifications() { return Arrays.asList( - TableParameterSpecification.builder().name(DATA_PARAMETER_NAME).setSemantics().build(), + TableParameterSpecification.builder() + .name(DATA_PARAMETER_NAME) + .setSemantics() + .ascendingOrder() + .build(), ScalarParameterSpecification.builder() .name(TIMECOL_PARAMETER_NAME) .type(Type.STRING) diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/SampleTableFunction.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/SampleTableFunction.java new file mode 100644 index 0000000000000..30f130a93f2e7 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/SampleTableFunction.java @@ -0,0 +1,232 @@ +/* + * 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.commons.udf.builtin.relational.tvf; + +import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.i18n.CommonMessages; +import org.apache.iotdb.udf.api.exception.UDFException; +import org.apache.iotdb.udf.api.relational.TableFunction; +import org.apache.iotdb.udf.api.relational.access.Record; +import org.apache.iotdb.udf.api.relational.table.MapTableFunctionHandle; +import org.apache.iotdb.udf.api.relational.table.TableFunctionAnalysis; +import org.apache.iotdb.udf.api.relational.table.TableFunctionHandle; +import org.apache.iotdb.udf.api.relational.table.TableFunctionProcessorProvider; +import org.apache.iotdb.udf.api.relational.table.argument.Argument; +import org.apache.iotdb.udf.api.relational.table.argument.DescribedSchema; +import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument; +import org.apache.iotdb.udf.api.relational.table.argument.TableArgument; +import org.apache.iotdb.udf.api.relational.table.processor.TableFunctionDataProcessor; +import org.apache.iotdb.udf.api.relational.table.specification.ParameterSpecification; +import org.apache.iotdb.udf.api.relational.table.specification.ScalarParameterSpecification; +import org.apache.iotdb.udf.api.relational.table.specification.TableParameterSpecification; +import org.apache.iotdb.udf.api.type.Type; + +import org.apache.tsfile.block.column.ColumnBuilder; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ThreadLocalRandom; + +import static org.apache.iotdb.commons.i18n.CommonMessages.EXCEPTION_SAMPLE_FUNCTION_N_VALIDATION; + +public class SampleTableFunction implements TableFunction { + + public static final String DATA_PARAMETER_NAME = "DATA"; + public static final String SAMPLE_SIZE_PARAMETER_NAME = "N"; + + private static final String OUTPUT_TYPES_PROPERTY = "OUTPUT_TYPES"; + + @Override + public List getArgumentsSpecifications() { + return Arrays.asList( + TableParameterSpecification.builder().name(DATA_PARAMETER_NAME).setSemantics().build(), + ScalarParameterSpecification.builder() + .name(SAMPLE_SIZE_PARAMETER_NAME) + .type(Type.INT32) + .addChecker( + object -> { + if (object instanceof Number) { + int value = ((Number) object).intValue(); + if (value > 0) { + return null; + } + } + return EXCEPTION_SAMPLE_FUNCTION_N_VALIDATION; + }) + .build()); + } + + @Override + public TableFunctionAnalysis analyze(Map arguments) throws UDFException { + TableArgument tableArgument = (TableArgument) arguments.get(DATA_PARAMETER_NAME); + + // Sampling is row based: every row is a candidate, regardless of which column it comes from. + // All columns are therefore required from the source and kept in the output, with the + // partition columns moved to the front in PARTITION BY order; the remaining columns keep their + // original relative order. + List partitionIndexes = WindowTVFUtils.getPartitionIndexes(tableArgument); + Set partitionIndexSet = new HashSet<>(partitionIndexes); + + List outputIndexes = new ArrayList<>(tableArgument.size()); + List outputTypes = new ArrayList<>(tableArgument.size()); + DescribedSchema.Builder schemaBuilder = new DescribedSchema.Builder(); + + for (int partitionIndex : partitionIndexes) { + String columnName = + tableArgument + .getFieldNames() + .get(partitionIndex) + .orElseThrow( + () -> new SemanticException(CommonMessages.EXCEPTION_COLUMN_LACK_OF_NAME)); + Type columnType = tableArgument.getFieldTypes().get(partitionIndex); + outputIndexes.add(partitionIndex); + outputTypes.add(columnType); + schemaBuilder.addField(columnName, columnType); + } + + for (int i = 0; i < tableArgument.size(); i++) { + if (partitionIndexSet.contains(i)) { + continue; + } + Type columnType = tableArgument.getFieldTypes().get(i); + if (columnType == Type.OBJECT) { + throw new SemanticException(CommonMessages.OBJECT_TYPE_COLUMN_NOT_SUPPORTED); + } + outputIndexes.add(i); + outputTypes.add(columnType); + schemaBuilder.addField(tableArgument.getFieldNames().get(i), columnType); + } + + MapTableFunctionHandle handle = + new MapTableFunctionHandle.Builder() + .addProperty( + SAMPLE_SIZE_PARAMETER_NAME, + ((ScalarArgument) arguments.get(SAMPLE_SIZE_PARAMETER_NAME)).getValue()) + .addProperty(OUTPUT_TYPES_PROPERTY, WindowTVFUtils.joinTypes(outputTypes)) + .build(); + + return TableFunctionAnalysis.builder() + .properColumnSchema(schemaBuilder.build()) + .requireRecordSnapshot(false) + .requiredColumns(DATA_PARAMETER_NAME, outputIndexes) + .handle(handle) + .build(); + } + + @Override + public TableFunctionHandle createTableFunctionHandle() { + return new MapTableFunctionHandle(); + } + + @Override + public TableFunctionProcessorProvider getProcessorProvider( + TableFunctionHandle tableFunctionHandle) { + MapTableFunctionHandle handle = (MapTableFunctionHandle) tableFunctionHandle; + int sampleSize = (Integer) handle.getProperty(SAMPLE_SIZE_PARAMETER_NAME); + Type[] outputTypes = + WindowTVFUtils.parseTypes((String) handle.getProperty(OUTPUT_TYPES_PROPERTY)); + return new TableFunctionProcessorProvider() { + @Override + public TableFunctionDataProcessor getDataProcessor() { + return new SampleDataProcessor(sampleSize, outputTypes); + } + }; + } + + /** + * Reservoir sampling over whole rows. All rows of a partition are candidates: the first N rows + * fill the reservoir, and each following row replaces a uniformly chosen slot with probability N + * / (rows seen so far). Rows are buffered until the partition ends because the final sample is + * not known before then. + */ + private static class SampleDataProcessor implements TableFunctionDataProcessor { + + private final int sampleSize; + private final Type[] outputTypes; + private final List reservoir = new ArrayList<>(); + private long rowCount = 0; + + private SampleDataProcessor(int sampleSize, Type[] outputTypes) { + this.sampleSize = sampleSize; + this.outputTypes = outputTypes; + } + + @Override + public void process( + Record input, + List properColumnBuilders, + ColumnBuilder passThroughIndexBuilder) { + int slot = pickSlot(); + if (slot >= 0) { + SampledRow row = new SampledRow(rowCount, captureRow(input)); + if (slot == reservoir.size()) { + reservoir.add(row); + } else { + reservoir.set(slot, row); + } + } + rowCount++; + } + + /** Returns the reservoir slot the current row occupies, or -1 if the row is not kept. */ + private int pickSlot() { + if (rowCount < sampleSize) { + return (int) rowCount; + } + long candidate = ThreadLocalRandom.current().nextLong(rowCount + 1); + return candidate < sampleSize ? (int) candidate : -1; + } + + private Object[] captureRow(Record input) { + Object[] values = new Object[outputTypes.length]; + for (int i = 0; i < outputTypes.length; i++) { + values[i] = input.isNull(i) ? null : WindowTVFUtils.readValue(input, i, outputTypes[i]); + } + return values; + } + + @Override + public void finish( + List properColumnBuilders, ColumnBuilder passThroughIndexBuilder) { + // emit in input order so the ORDER BY of DATA still + // decides the relative order of the sampled rows. + reservoir.sort((left, right) -> Long.compare(left.position, right.position)); + for (SampledRow row : reservoir) { + for (int i = 0; i < outputTypes.length; i++) { + WindowTVFUtils.writeValue(properColumnBuilders.get(i), row.values[i], outputTypes[i]); + } + } + } + + private static class SampledRow { + private final long position; + private final Object[] values; + + private SampledRow(long position, Object[] values) { + this.position = position; + this.values = values; + } + } + } +} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/WindowTVFUtils.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/WindowTVFUtils.java index 2ba0c09572848..69f2470ebe34e 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/WindowTVFUtils.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/WindowTVFUtils.java @@ -175,17 +175,6 @@ public static List getCalculationIndexes( return calculationIndexes; } - public static String joinTypes(List types) { - StringBuilder builder = new StringBuilder(); - for (int i = 0; i < types.size(); i++) { - if (i > 0) { - builder.append(','); - } - builder.append(types.get(i).name()); - } - return builder.toString(); - } - /** check the order by column is the timeColumn */ public static int checkOrderByColumn( Map arguments, String dataParameterName, String timeParameterName) { @@ -203,6 +192,17 @@ public static int checkOrderByColumn( return timeColumnIndex; } + public static String joinTypes(List types) { + StringBuilder builder = new StringBuilder(); + for (int i = 0; i < types.size(); i++) { + if (i > 0) { + builder.append(','); + } + builder.append(types.get(i).name()); + } + return builder.toString(); + } + public static Type[] parseTypes(String value) { if (value.isEmpty()) { return new Type[0]; diff --git a/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift b/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift index 16b40a853541c..b593b34d71ca7 100644 --- a/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift +++ b/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift @@ -331,6 +331,7 @@ enum TAggregationType { INCREASE, IRATE, DELTA, + INTEGRAL } struct TShowConfigurationTemplateResp {