diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLTTBTableFunctionIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLTTBTableFunctionIT.java new file mode 100644 index 0000000000000..ece266187316b --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLTTBTableFunctionIT.java @@ -0,0 +1,261 @@ +/* + * 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.relational.it.db.it; + +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.TableClusterIT; +import org.apache.iotdb.itbase.category.TableLocalStandaloneIT; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.sql.Connection; +import java.sql.Statement; + +import static org.apache.iotdb.db.it.utils.TestUtils.tableAssertTestFail; +import static org.apache.iotdb.db.it.utils.TestUtils.tableResultSetEqualTest; +import static org.junit.Assert.fail; + +@RunWith(IoTDBTestRunner.class) +@Category({TableLocalStandaloneIT.class, TableClusterIT.class}) +public class IoTDBLTTBTableFunctionIT { + + private static final String DATABASE_NAME = "lttb"; + + @BeforeClass + public static void setUp() throws Exception { + EnvFactory.getEnv().initClusterEnvironment(); + try (Connection connection = EnvFactory.getEnv().getTableConnection(); + Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE " + DATABASE_NAME); + statement.execute("USE " + DATABASE_NAME); + statement.execute( + "CREATE TABLE metrics(device STRING TAG, s1 DOUBLE FIELD, s2 DOUBLE FIELD)"); + statement.execute( + "INSERT INTO metrics(time, device, s1, s2) VALUES " + + "(0, 'd1', 0.0, 0.0)," + + "(1, 'd1', 1.0, 0.0)," + + "(2, 'd1', 8.0, null)," + + "(3, 'd1', 2.0, 9.0)," + + "(4, 'd1', 0.0, 1.0)," + + "(5, 'd1', 0.0, 0.0)," + + "(0, 'd2', null, null)," + + "(1, 'd2', null, null)"); + statement.execute( + "CREATE TABLE sparse(device STRING TAG, value DOUBLE FIELD, marker INT32 FIELD)"); + statement.execute( + "INSERT INTO sparse(time, device, value, marker) VALUES " + + "(0, 'd1', 0.0, 1), (1, 'd1', 1.0, 1), (2, 'd1', 2.0, 1)," + + "(3, 'd1', null, 1), (4, 'd1', null, 1), (5, 'd1', null, 1)," + + "(6, 'd1', 0.0, 1), (7, 'd1', 0.0, 1), (8, 'd1', 0.0, 1)"); + statement.execute("CREATE TABLE integral_values(device STRING TAG, value INT64 FIELD)"); + long baseline = 1L << 60; + long[] deltas = {0, 1, 8, 2, 0, 0}; + for (int i = 0; i < deltas.length; i++) { + statement.execute( + "INSERT INTO integral_values(time, device, value) VALUES (" + + i + + ", 'd1', " + + (baseline + deltas[i]) + + ")"); + } + statement.execute("FLUSH"); + } catch (Exception e) { + fail("insertData failed: " + e.getMessage()); + } + } + + @AfterClass + public static void tearDown() throws Exception { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testTargetCountPreservesEndpointsAndAlignsColumns() { + tableResultSetEqualTest( + "SELECT * FROM LTTB(DATA => metrics PARTITION BY device ORDER BY time, " + + "TIMECOL => 'time', N => 3) ORDER BY device, s1_time", + new String[] {"window_index", "device", "s1_time", "s1", "s2_time", "s2"}, + new String[] { + "0,d1,1970-01-01T00:00:00.000Z,0.0,1970-01-01T00:00:00.000Z,0.0,", + "0,d1,1970-01-01T00:00:00.002Z,8.0,1970-01-01T00:00:00.003Z,9.0,", + "0,d1,1970-01-01T00:00:00.005Z,0.0,1970-01-01T00:00:00.005Z,0.0,", + "0,d2,null,null,null,null," + }, + DATABASE_NAME); + } + + @Test + public void testCountWindowAndTimeWindowOutput() { + tableResultSetEqualTest( + "SELECT * FROM LTTB(DATA => (SELECT time, device, s1 FROM metrics WHERE device = 'd1') " + + "PARTITION BY device ORDER BY time, TIMECOL => 'time', SIZE => 2) " + + "ORDER BY window_index", + new String[] {"window_index", "device", "s1_time", "s1"}, + new String[] { + "0,d1,1970-01-01T00:00:00.001Z,1.0,", + "1,d1,1970-01-01T00:00:00.002Z,8.0,", + "2,d1,1970-01-01T00:00:00.004Z,0.0," + }, + DATABASE_NAME); + + tableResultSetEqualTest( + "SELECT * FROM LTTB(DATA => (SELECT time, device, s1 FROM metrics WHERE device = 'd1') " + + "PARTITION BY device ORDER BY time, TIMECOL => 'time', SIZE => 2ms, " + + "ORIGIN => 1970-01-01T00:00:00.000+00:00) ORDER BY window_start", + new String[] {"window_start", "window_end", "device", "s1_time", "s1"}, + new String[] { + "1970-01-01T00:00:00.000Z,1970-01-01T00:00:00.002Z,d1,1970-01-01T00:00:00.001Z,1.0,", + "1970-01-01T00:00:00.002Z,1970-01-01T00:00:00.004Z,d1,1970-01-01T00:00:00.002Z,8.0,", + "1970-01-01T00:00:00.004Z,1970-01-01T00:00:00.006Z,d1,1970-01-01T00:00:00.004Z,0.0," + }, + DATABASE_NAME); + } + + @Test + public void testDuplicateTimestampsAreRejectedWith701() { + // the same six d1 rows twice: (0, 1, 2, 3, 4, 5) each occurs twice in one partition + String duplicated = + "SELECT * FROM LTTB(DATA => (SELECT time, device, s1 FROM metrics WHERE device = 'd1' " + + "UNION ALL SELECT time, device, s1 FROM metrics WHERE device = 'd1') " + + "PARTITION BY device ORDER BY time, TIMECOL => 'time', "; + String expected = + "701: LTTB does not support duplicate timestamps, but timestamp 0 appears more than once" + + " in the same partition."; + + tableAssertTestFail(duplicated + "N => 3)", expected, DATABASE_NAME); + tableAssertTestFail(duplicated + "SIZE => 2)", expected, DATABASE_NAME); + tableAssertTestFail( + duplicated + "SIZE => 2ms, ORIGIN => 1970-01-01T00:00:00.000+00:00)", + expected, + DATABASE_NAME); + } + + @Test + public void testDuplicateTimestampsInDifferentPartitionsRemainValid() { + // each partition is unique on its own, so the query must succeed and keep both rows + tableResultSetEqualTest( + "SELECT * FROM LTTB(DATA => (SELECT time, device, s1 FROM metrics " + + "WHERE time = 0 AND (device = 'd1' OR device = 'd2')) " + + "PARTITION BY device ORDER BY time, TIMECOL => 'time', N => 3) " + + "ORDER BY device", + new String[] {"window_index", "device", "s1_time", "s1"}, + new String[] { + "0,d1,1970-01-01T00:00:00.000Z,0.0,", "0,d2,null,null,", + }, + DATABASE_NAME); + } + + @Test + public void testWindowSkipsNullBucketForLookahead() { + String data = + "DATA => (SELECT time, device, value FROM sparse) PARTITION BY device ORDER BY time, TIMECOL => 'time', "; + tableResultSetEqualTest( + "SELECT * FROM LTTB(" + data + "SIZE => 3) ORDER BY window_index", + new String[] {"window_index", "device", "value_time", "value"}, + new String[] { + "0,d1,1970-01-01T00:00:00.002Z,2.0,", + "1,d1,null,null,", + "2,d1,1970-01-01T00:00:00.006Z,0.0," + }, + DATABASE_NAME); + tableResultSetEqualTest( + "SELECT * FROM LTTB(" + data + "SIZE => 3ms) ORDER BY window_start", + new String[] {"window_start", "window_end", "device", "value_time", "value"}, + new String[] { + "1970-01-01T00:00:00.000Z,1970-01-01T00:00:00.003Z,d1,1970-01-01T00:00:00.002Z,2.0,", + "1970-01-01T00:00:00.003Z,1970-01-01T00:00:00.006Z,d1,null,null,", + "1970-01-01T00:00:00.006Z,1970-01-01T00:00:00.009Z,d1,1970-01-01T00:00:00.006Z,0.0," + }, + DATABASE_NAME); + } + + @Test + public void testNonFiniteValuesAreKeptInShortSeries() { + // three non-NULL input points whose values are NaN/Infinity for N => 3 + String nonFinite = + "SELECT * FROM LTTB(DATA => (SELECT time, device, s1 / 0.0 AS s1 FROM metrics " + + "WHERE device = 'd1' AND time <= 2) " + + "PARTITION BY device ORDER BY time, TIMECOL => 'time', "; + + tableResultSetEqualTest( + nonFinite + "N => 3) ORDER BY s1_time", + new String[] {"window_index", "device", "s1_time", "s1"}, + new String[] { + "0,d1,1970-01-01T00:00:00.000Z,NaN,", + "0,d1,1970-01-01T00:00:00.001Z,Infinity,", + "0,d1,1970-01-01T00:00:00.002Z,Infinity,", + }, + DATABASE_NAME); + } + + @Test + public void testInt64SamplingPreservesSmallPeaksAtLargeBaselines() { + String prefix = + "SELECT * FROM LTTB(DATA => integral_values PARTITION BY device ORDER BY time, " + + "TIMECOL => 'time', "; + tableResultSetEqualTest( + prefix + "N => 3) ORDER BY value_time", + new String[] {"window_index", "device", "value_time", "value"}, + new String[] { + "0,d1,1970-01-01T00:00:00.000Z,1152921504606846976,", + "0,d1,1970-01-01T00:00:00.002Z,1152921504606846984,", + "0,d1,1970-01-01T00:00:00.005Z,1152921504606846976," + }, + DATABASE_NAME); + tableResultSetEqualTest( + prefix + "SIZE => 2) ORDER BY window_index", + new String[] {"window_index", "device", "value_time", "value"}, + new String[] { + "0,d1,1970-01-01T00:00:00.001Z,1152921504606846977,", + "1,d1,1970-01-01T00:00:00.002Z,1152921504606846984,", + "2,d1,1970-01-01T00:00:00.004Z,1152921504606846976," + }, + DATABASE_NAME); + } + + @Test + public void testInvalidParameterCombinations() { + String prefix = + "SELECT * FROM LTTB(DATA => metrics PARTITION BY device ORDER BY time, TIMECOL => 'time', "; + tableAssertTestFail( + "SELECT * FROM LTTB(DATA => metrics PARTITION BY device ORDER BY time, TIMECOL => 'time')", + "701: Exactly one of the N and SIZE arguments must be specified for LTTB.", + DATABASE_NAME); + tableAssertTestFail( + prefix + "N => 3, SIZE => 2)", + "701: Exactly one of the N and SIZE arguments must be specified for LTTB.", + DATABASE_NAME); + tableAssertTestFail( + prefix + "N => 2)", "701: The N argument of LTTB must be at least 3.", DATABASE_NAME); + tableAssertTestFail( + prefix + "N => 3, SLIDE => 2)", + "701: The N argument of LTTB cannot be combined with the SLIDE or ORIGIN arguments.", + DATABASE_NAME); + tableAssertTestFail( + prefix + "SIZE => 2, ORIGIN => 1970-01-01T00:00:00.000+00:00)", + "701: The ORIGIN argument is only supported in time window mode.", + DATABASE_NAME); + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadConfigurationTableIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadConfigurationTableIT.java index f573426334087..0c3f413938b4b 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadConfigurationTableIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadConfigurationTableIT.java @@ -119,13 +119,19 @@ public void showConfigurationDisplaysEffectiveValue() throws Exception { // Case 2: a valid value is not rewritten, so display must equal the file value (regression). Map afterValidReload = - appendLinesLoadAndShow(dataNodeId, confPath, "cte_buffer_size_in_bytes=262144"); + appendLinesLoadAndShow( + dataNodeId, + confPath, + "cte_buffer_size_in_bytes=262144", + "lttb_buffer_size_in_bytes=262144"); Assert.assertEquals("262144", afterValidReload.get("cte_buffer_size_in_bytes")); + Assert.assertEquals("262144", afterValidReload.get("lttb_buffer_size_in_bytes")); // Case 3: params whose template default is 0 are always rewritten to a computed default, // so they must never display 0 (this was always wrong, even without a bad file value). Map defaultsReload = appendLinesLoadAndShow(dataNodeId, confPath); assertPositiveNonZero(defaultsReload, "sort_buffer_size_in_bytes"); + assertPositiveNonZero(defaultsReload, "lttb_buffer_size_in_bytes"); assertPositiveNonZero(defaultsReload, "mods_cache_size_limit_per_fi_in_bytes"); } diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/TableFunctionProcessorProvider.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/TableFunctionProcessorProvider.java index f86d631f5989a..9ca917070d7e5 100644 --- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/TableFunctionProcessorProvider.java +++ b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/table/TableFunctionProcessorProvider.java @@ -43,4 +43,16 @@ default TableFunctionLeafProcessor getSplitProcessor() { throw new UnsupportedOperationException( UdfApiMessages.TABLE_FUNCTION_DOES_NOT_PROCESS_LEAF_DATA); } + + /** + * The bytes the processors of this provider may hold in their own buffers while a partition is + * being processed, above what {@code process} returns. The engine adds this to the estimated + * memory of the {@linkplain TableFunctionDataProcessor} operator, so a table function that + * accumulates input (rather than emitting a result per record) is visible to memory management + * and can be back-pressured. The default is {@code 0} for table functions that keep no state + * across calls. + */ + default long getEstimatedMaxBufferedBytes() { + return 0L; + } } 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 fb9b90c9f604a..d8508a32e2d30 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 @@ -72,6 +72,9 @@ public class TableFunctionOperator extends AbstractOperator implements ProcessOp private final boolean isDeclaredAsPassThrough; private final IoTDBLocal ioTDBLocal; + /** Bytes the data processors may buffer internally while a partition is processed. */ + private final long estimatedMaxBufferedBytes; + private TableFunctionDataProcessor processor; private PartitionState partitionState; private ListenableFuture isBlocked; @@ -101,6 +104,7 @@ public TableFunctionOperator( new PartitionRecognizer( partitionChannels, requiredChannels, passThroughChannels, inputDataTypes); this.isDeclaredAsPassThrough = isDeclaredAsPassThrough; + this.estimatedMaxBufferedBytes = processorProvider.getEstimatedMaxBufferedBytes(); this.needPassThrough = properChannelCount != outputDataTypes.size(); this.partitionState = null; this.properBlockBuilder = new TsBlockBuilder(outputDataTypes.subList(0, properChannelCount)); @@ -291,7 +295,10 @@ public boolean isFinished() throws Exception { @Override public long calculateMaxPeekMemory() { + // A processor that buffers its input (LTTB) holds up to estimatedMaxBufferedBytes while the + // partition is processed, in addition to the input it reads and the block it builds. return inputOperator.calculateMaxPeekMemory() + + estimatedMaxBufferedBytes + Math.max(maxReturnSize, properBlockBuilder.getRetainedSizeInBytes()); } @@ -302,7 +309,9 @@ public long calculateMaxReturnSize() { @Override public long calculateRetainedSizeAfterCallingNext() { - return inputOperator.calculateRetainedSizeAfterCallingNext(); + // The buffers survive across next() calls until the partition completes, so they are retained + // here rather than charged to the peek of a single call. + return inputOperator.calculateRetainedSizeAfterCallingNext() + estimatedMaxBufferedBytes; } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java index 44971c718a104..cc8ab3714517d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java @@ -1100,6 +1100,13 @@ public void loadProperties(TrimProperties properties) throws BadNodeUrlException commonConfig.setCteBufferSize(cteBufferSizeInBytes); } + // The per-processor buffer budget of the LTTB table function. Like sort_buffer_size_in_bytes + // this is charged to the estimated memory of every fragment that runs the function, so the + // non-positive default resolves to a value derived from the operator memory of this node + // instead of one that would reject the query outright on a small-memory node. + loadFixedSizeLimitForQuery( + properties, "lttb_buffer_size_in_bytes", commonConfig::setLttbBufferSize); + // max number of rows for cte materialization int maxRowsInCteBuffer = Integer.parseInt( @@ -2392,6 +2399,12 @@ public synchronized void loadHotModifiedProps(TrimProperties properties) if (cteBufferSizeInBytes > 0) { commonConfig.setCteBufferSize(cteBufferSizeInBytes); } + // The per-processor buffer budget of the LTTB table function. Like sort_buffer_size_in_bytes + // this is charged to the estimated memory of every fragment that runs the function, so the + // non-positive default resolves to a value derived from the operator memory of this node + // instead of one that would reject the query outright on a small-memory node. + loadFixedSizeLimitForQuery( + properties, "lttb_buffer_size_in_bytes", commonConfig::setLttbBufferSize); // max number of rows for cte materialization int maxRowsInCteBuffer = Integer.parseInt( @@ -2439,6 +2452,8 @@ public synchronized void loadHotModifiedProps(TrimProperties properties) private void overlayEffectiveConfigurationValues() { ConfigurationFileUtils.updateAppliedProperties( "cte_buffer_size_in_bytes", Long.toString(commonConfig.getCteBufferSize())); + ConfigurationFileUtils.updateAppliedProperties( + "lttb_buffer_size_in_bytes", Long.toString(commonConfig.getLttbBufferSize())); ConfigurationFileUtils.updateAppliedProperties( "max_rows_in_cte_buffer", Integer.toString(commonConfig.getMaxRowsInCteBuffer())); ConfigurationFileUtils.updateAppliedProperties( 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 b5c394539f217..fc5cec448bd8a 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 @@ -22,6 +22,7 @@ import org.apache.iotdb.calc.plan.relational.metadata.CommonMetadataUtils; import org.apache.iotdb.commons.exception.IoTDBException; 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; @@ -122,6 +123,7 @@ import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; import org.apache.iotdb.commons.schema.table.column.TsTableColumnSchema; import org.apache.iotdb.commons.udf.builtin.relational.tvf.FFTTableFunction; +import org.apache.iotdb.commons.udf.builtin.relational.tvf.LTTBTableFunction; import org.apache.iotdb.commons.udf.builtin.relational.tvf.M4TableFunction; import org.apache.iotdb.commons.udf.utils.UDFDataTypeTransformer; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; @@ -5916,6 +5918,7 @@ private ArgumentsAnalysis analyzeArguments( } } tryAppendM4ModeArgument(functionName, arguments, parameterSpecifications, passedArguments); + tryAppendLTTBModeArgument(functionName, arguments, parameterSpecifications, passedArguments); tryAppendFFTInternalArguments( functionName, arguments, parameterSpecifications, passedArguments); return new ArgumentsAnalysis(passedArguments.buildOrThrow(), tableArgumentAnalyses.build()); @@ -5923,6 +5926,7 @@ private ArgumentsAnalysis analyzeArguments( private boolean isPartitionColumnsProvidedByProperSchema(String functionName) { return TableBuiltinTableFunction.M4.getFunctionName().equalsIgnoreCase(functionName) + || TableBuiltinTableFunction.LTTB.getFunctionName().equalsIgnoreCase(functionName) || TableBuiltinTableFunction.LOWPASS.getFunctionName().equalsIgnoreCase(functionName) || TableBuiltinTableFunction.HIGHPASS.getFunctionName().equalsIgnoreCase(functionName) || TableBuiltinTableFunction.XCORR.getFunctionName().equalsIgnoreCase(functionName) @@ -6034,6 +6038,87 @@ && containsTableFunctionArgument( new ScalarArgument(org.apache.iotdb.udf.api.type.Type.BOOLEAN, isTimeWindow)); } + /** + * Resolves the LTTB execution mode from the SQL literals. The mode (target-count vs. time + * window vs. count window) depends on whether {@code SIZE} is a duration literal, which is only + * visible at this level, so it is injected as the internal boolean argument {@link + * LTTBTableFunction#WINDOW_MODE_PARAMETER_NAME}. Parameter-combination checks that only need + * the presence of arguments are also performed here so they fail before analysis. + */ + private void tryAppendLTTBModeArgument( + String functionName, + List arguments, + List parameterSpecifications, + ImmutableMap.Builder passedArguments) { + if (!TableBuiltinTableFunction.LTTB.getFunctionName().equalsIgnoreCase(functionName)) { + return; + } + + Optional nArgument = + findOptionalTableFunctionArgument( + arguments, parameterSpecifications, LTTBTableFunction.N_PARAMETER_NAME); + Optional sizeArgument = + findOptionalTableFunctionArgument( + arguments, parameterSpecifications, LTTBTableFunction.SIZE_PARAMETER_NAME); + Optional slideArgument = + findOptionalTableFunctionArgument( + arguments, parameterSpecifications, LTTBTableFunction.SLIDE_PARAMETER_NAME); + boolean originSpecified = + containsTableFunctionArgument( + arguments, parameterSpecifications, LTTBTableFunction.ORIGIN_PARAMETER_NAME); + + if (nArgument.isPresent() == sizeArgument.isPresent()) { + throw new SemanticException( + CommonMessages + .EXCEPTION_EXACTLY_ONE_OF_THE_N_AND_SIZE_ARGUMENTS_MUST_BE_SPECIFIED_FOR_LTTB_54AF0733); + } + + boolean isTimeWindow = false; + if (nArgument.isPresent()) { + if (slideArgument.isPresent() || originSpecified) { + throw new SemanticException( + CommonMessages + .EXCEPTION_THE_N_ARGUMENT_OF_LTTB_CANNOT_BE_COMBINED_WITH_THE_SLIDE_OR_ORIGIN_ARGUMENTS_C2FF1FE1); + } + if (!(nArgument.get().getValue() instanceof Expression) + || nArgument.get().getValue() instanceof TimeDurationLiteral) { + throw new SemanticException( + CommonMessages.EXCEPTION_THE_N_ARGUMENT_OF_LTTB_MUST_BE_A_POSITIVE_INTEGER_21795D8E); + } + } else { + if (!(sizeArgument.get().getValue() instanceof Expression)) { + throw new SemanticException( + String.format( + QueryMessages.EXCEPTION_INVALID_ARGUMENT_ARG_EXPECTED_SCALAR_ARGUMENT_4AD4E32F, + LTTBTableFunction.SIZE_PARAMETER_NAME)); + } + isTimeWindow = sizeArgument.get().getValue() instanceof TimeDurationLiteral; + if (slideArgument.isPresent()) { + if (!(slideArgument.get().getValue() instanceof Expression)) { + throw new SemanticException( + String.format( + QueryMessages.EXCEPTION_INVALID_ARGUMENT_ARG_EXPECTED_SCALAR_ARGUMENT_4AD4E32F, + LTTBTableFunction.SLIDE_PARAMETER_NAME)); + } + boolean isTimeSlide = slideArgument.get().getValue() instanceof TimeDurationLiteral; + if (isTimeWindow != isTimeSlide) { + throw new SemanticException( + QueryMessages + .EXCEPTION_THE_SLIDE_ARGUMENT_MUST_HAVE_THE_SAME_WINDOW_MODE_AS_THE_SIZE_ARGUMENT_419C9844); + } + } + if (!isTimeWindow && originSpecified) { + throw new SemanticException( + QueryMessages + .EXCEPTION_THE_ORIGIN_ARGUMENT_IS_ONLY_SUPPORTED_IN_TIME_WINDOW_MODE_B3F358B9); + } + } + validateM4OrderBySortOrder(arguments, parameterSpecifications); + passedArguments.put( + LTTBTableFunction.WINDOW_MODE_PARAMETER_NAME, + new ScalarArgument(org.apache.iotdb.udf.api.type.Type.BOOLEAN, isTimeWindow)); + } + private void validateM4OrderBySortOrder( List arguments, List parameterSpecifications) { 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..f17d5d9a24e66 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 @@ -1610,6 +1610,7 @@ public RelationPlan visitTableFunctionInvocation(TableFunctionInvocation node, V private boolean needAddPartitionColumn(String functionName) { return !TableBuiltinTableFunction.M4.getFunctionName().equalsIgnoreCase(functionName) + && !TableBuiltinTableFunction.LTTB.getFunctionName().equalsIgnoreCase(functionName) && !TableBuiltinTableFunction.LOWPASS.getFunctionName().equalsIgnoreCase(functionName) && !TableBuiltinTableFunction.HIGHPASS.getFunctionName().equalsIgnoreCase(functionName) && !TableBuiltinTableFunction.XCORR.getFunctionName().equalsIgnoreCase(functionName) 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 00db7b95e4924..06b97fa7b7d69 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 @@ -2899,6 +2899,10 @@ private boolean canSplitTableFunctionProcessor(TableFunctionProcessorNode node) if (!isPartitionedGroup(node.getChild())) { return false; } + // LTTB needs whole partitions, which the partitioned GroupNode above this node already + // guarantees: every fragment receives complete partitions and the ordering is preserved, so a + // fragment-local sample is a valid standard LTTB result for that partition and no exclusion is + // needed. FFT is excluded because its transform over the whole partition is not splittable. return !TableBuiltinTableFunction.FFT.getFunctionName().equalsIgnoreCase(node.getName()); } 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..13a579a403adc 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 @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.queryengine.plan.relational.function.tvf.ForecastTableFunction; import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.JoinNode; import org.apache.iotdb.commons.udf.builtin.relational.tvf.FFTTableFunction; +import org.apache.iotdb.commons.udf.builtin.relational.tvf.LTTBTableFunction; import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan; import org.apache.iotdb.db.queryengine.plan.relational.planner.PlanTester; import org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern; @@ -842,4 +843,161 @@ public void testM4RejectsDescendingOrderBy() { e.getMessage()); } } + + // Aliases use the planner's actual symbol names so the same matcher also resolves them through + // the ExchangeNodes of the distributed plan (which bind symbols to their own names). + private static final ImmutableMap TABLE1_COLUMNS = + ImmutableMap.builder() + .put("time", "time") + .put("tag1_0", "tag1") + .put("tag2_1", "tag2") + .put("tag3_2", "tag3") + .put("attr1_3", "attr1") + .put("attr2_4", "attr2") + .put("s1_5", "s1") + .put("s2_6", "s2") + .put("s3_7", "s3") + .buildOrThrow(); + + private static final String LTTB_DATA_ARGUMENT = + "DATA => table1 PARTITION BY (tag1, tag2, tag3, attr1, attr2) ORDER BY time"; + + private static MapTableFunctionHandle.Builder lttbHandle(String mode) { + return new MapTableFunctionHandle.Builder() + .addProperty(LTTBTableFunction.MODE_PROPERTY, mode) + .addProperty( + LTTBTableFunction.PARTITION_TYPES_PROPERTY, "STRING,STRING,STRING,STRING,STRING") + .addProperty(LTTBTableFunction.PARTICIPANT_TYPES_PROPERTY, "INT64,INT64,DOUBLE"); + } + + private static TableFunctionProcessorMatcher.Builder lttbMatcher( + TableFunctionProcessorMatcher.Builder builder, String... windowColumns) { + ImmutableList.Builder properOutputs = ImmutableList.builder(); + properOutputs.add(windowColumns); + properOutputs.add( + "lttb_tag1", + "lttb_tag2", + "lttb_tag3", + "lttb_attr1", + "lttb_attr2", + "lttb_s1_time", + "lttb_s1", + "lttb_s2_time", + "lttb_s2", + "lttb_s3_time", + "lttb_s3"); + return builder + .name("lttb") + .properOutputs(properOutputs.build().toArray(new String[0])) + .requiredSymbols( + "time", "tag1_0", "tag2_1", "tag3_2", "attr1_3", "attr2_4", "s1_5", "s2_6", "s3_7"); + } + + @Test + public void testLTTBTargetCountMode() { + PlanTester planTester = new PlanTester(); + String sql = "SELECT * FROM LTTB(" + LTTB_DATA_ARGUMENT + ", TIMECOL => 'time', N => 100)"; + LogicalQueryPlan logicalQueryPlan = planTester.createPlan(sql); + PlanMatchPattern tableScan = tableScan("testdb.table1", TABLE1_COLUMNS); + + Consumer tableFunctionMatcher = + builder -> + lttbMatcher(builder, "window_index") + .handle( + lttbHandle("TARGET_COUNT") + .addProperty(LTTBTableFunction.N_PARAMETER_NAME, 100L) + .build()); + + // PARTITION BY several columns plans a GroupNode (a SortNode subclass) below the function + assertPlan( + logicalQueryPlan, anyTree(tableFunctionProcessor(tableFunctionMatcher, group(tableScan)))); + // LTTB is not mergeable: the whole partition is gathered and ordered before the function runs + assertPlan( + planTester.getFragmentPlan(0), + output( + tableFunctionProcessor( + tableFunctionMatcher, mergeSort(exchange(), exchange(), exchange())))); + } + + @Test + public void testLTTBTimeWindowMode() { + PlanTester planTester = new PlanTester(); + String sql = + "SELECT * FROM LTTB(" + + LTTB_DATA_ARGUMENT + + ", TIMECOL => 'time', SIZE => 1h, SLIDE => 30m, " + + "ORIGIN => 1970-01-01T00:00:00.000+00:00)"; + Consumer tableFunctionMatcher = + builder -> + lttbMatcher(builder, "window_start", "window_end") + .handle( + lttbHandle("TIME_WINDOW") + .addProperty(LTTBTableFunction.SIZE_PARAMETER_NAME, 3600000L) + .addProperty(LTTBTableFunction.SLIDE_PARAMETER_NAME, 1800000L) + .addProperty(LTTBTableFunction.ORIGIN_PARAMETER_NAME, 0L) + .build()); + assertPlan( + planTester.createPlan(sql), + anyTree( + tableFunctionProcessor( + tableFunctionMatcher, group(tableScan("testdb.table1", TABLE1_COLUMNS))))); + } + + @Test + public void testLTTBCountWindowMode() { + PlanTester planTester = new PlanTester(); + String sql = "SELECT * FROM LTTB(" + LTTB_DATA_ARGUMENT + ", TIMECOL => 'time', SIZE => 5)"; + Consumer tableFunctionMatcher = + builder -> + lttbMatcher(builder, "window_index") + .handle( + lttbHandle("COUNT_WINDOW") + .addProperty(LTTBTableFunction.SIZE_PARAMETER_NAME, 5L) + .addProperty(LTTBTableFunction.SLIDE_PARAMETER_NAME, 5L) + .build()); + assertPlan( + planTester.createPlan(sql), + anyTree( + tableFunctionProcessor( + tableFunctionMatcher, group(tableScan("testdb.table1", TABLE1_COLUMNS))))); + } + + @Test + public void testLTTBRejectsInvalidArguments() { + String data = "DATA => (SELECT time, tag1, s3 FROM table1) PARTITION BY tag1 ORDER BY time"; + assertAnalyzeFails( + "SELECT * FROM LTTB(" + data + ", TIMECOL => 'time')", + "Exactly one of the N and SIZE arguments must be specified for LTTB."); + assertAnalyzeFails( + "SELECT * FROM LTTB(" + data + ", TIMECOL => 'time', N => 10, SIZE => 1h)", + "Exactly one of the N and SIZE arguments must be specified for LTTB."); + assertAnalyzeFails( + "SELECT * FROM LTTB(" + data + ", TIMECOL => 'time', N => 2)", + "The N argument of LTTB must be at least 3."); + assertAnalyzeFails( + "SELECT * FROM LTTB(" + data + ", TIMECOL => 'time', N => 1h)", + "The N argument of LTTB must be a positive integer."); + assertAnalyzeFails( + "SELECT * FROM LTTB(" + + data + + ", TIMECOL => 'time', N => 10, ORIGIN => 1970-01-01T00:00:00.000+00:00)", + "The N argument of LTTB cannot be combined with the SLIDE or ORIGIN arguments."); + assertAnalyzeFails( + "SELECT * FROM LTTB(" + data + ", TIMECOL => 'time', SLIDE => 1h)", + "Exactly one of the N and SIZE arguments must be specified for LTTB."); + assertAnalyzeFails( + "SELECT * FROM LTTB(" + + data + + ", TIMECOL => 'time', SIZE => 5, ORIGIN => 1970-01-01T00:00:00.000+00:00)", + "The ORIGIN argument is only supported in time window mode."); + assertAnalyzeFails( + "SELECT * FROM LTTB(" + data + ", TIMECOL => 'time', SIZE => 1h, SLIDE => 5)", + "The SLIDE argument must have the same window mode as the SIZE argument."); + assertAnalyzeFails( + "SELECT * FROM LTTB(DATA => (SELECT time, tag1, s3 FROM table1) PARTITION BY tag1 ORDER BY time DESC, TIMECOL => 'time', N => 10)", + "The ORDER BY clause of the DATA argument must sort the time column in ascending order."); + assertAnalyzeFails( + "SELECT * FROM LTTB(DATA => (SELECT time, tag1, attr1, s3 FROM table1) PARTITION BY tag1 ORDER BY time, TIMECOL => 'time', N => 10)", + "Only column with double, float, int32, int64 can be calculated by the function, attr1 is the STRING."); + } } diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template index 866dde0a94e5b..783d23b7f6987 100644 --- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template +++ b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template @@ -1280,6 +1280,16 @@ sort_buffer_size_in_bytes=0 # Unit: bytes cte_buffer_size_in_bytes=131072 +# The per-processor buffer budget of the LTTB table function. It is charged to the estimated memory of every fragment that +# runs LTTB, so if lttb_buffer_size_in_bytes <= 0 a default value is used, default value = min(32MB, memory for query operators / +# query_thread_count / 2), matching sort_buffer_size_in_bytes. Raise it for queries with very large partitions or many overlapping +# windows. +# effectiveMode: hot_reload +# Datatype: long +# Privilege: SYSTEM +# Unit: bytes +lttb_buffer_size_in_bytes=0 + # Max rows for CTE materialization # effectiveMode: hot_reload 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 a6b958fd1420f..1014c2a686572 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 @@ -342,6 +342,37 @@ private CommonMessages() {} EXCEPTION_XCORR_REQUIRES_EXACTLY_TWO_CALCULATION_COLUMNS_BUT_FOUND_ARG_2FF8EB0C = "XCorr requires exactly two calculation columns, but found %d."; public static final String EXCEPTION_COLUMN_LACK_OF_NAME = "the column in table lack of the name"; + public static final String + EXCEPTION_EXACTLY_ONE_OF_THE_N_AND_SIZE_ARGUMENTS_MUST_BE_SPECIFIED_FOR_LTTB_54AF0733 = + "Exactly one of the N and SIZE arguments must be specified for LTTB."; + public static final String EXCEPTION_THE_N_ARGUMENT_OF_LTTB_MUST_BE_AT_LEAST_3_29CE2B87 = + "The N argument of LTTB must be at least 3."; + public static final String + EXCEPTION_THE_N_ARGUMENT_OF_LTTB_MUST_BE_A_POSITIVE_INTEGER_21795D8E = + "The N argument of LTTB must be a positive integer."; + public static final String + EXCEPTION_THE_N_ARGUMENT_OF_LTTB_CANNOT_BE_COMBINED_WITH_THE_SLIDE_OR_ORIGIN_ARGUMENTS_C2FF1FE1 = + "The N argument of LTTB cannot be combined with the SLIDE or ORIGIN arguments."; + public static final String + EXCEPTION_INVALID_SCALAR_ARGUMENT_SIZE_SHOULD_BE_A_POSITIVE_VALUE_3ECF76E0 = + "Invalid scalar argument SIZE, should be a positive value"; + public static final String + EXCEPTION_LTTB_BUFFERS_AT_MOST_ARG_ELIGIBLE_POINTS_PER_COLUMN_IN_ONE_PARTITION_OR_WINDOW_4C746D82 = + "LTTB buffers at most %d eligible points per column in one partition or window."; + public static final String + EXCEPTION_LTTB_DOES_NOT_SUPPORT_DUPLICATE_TIMESTAMPS_BUT_TIMESTAMP_ARG_APPEARS_MORE_THAN_ONCE_IN_THE_SAME_PARTITION_7713FEDD = + "LTTB does not support duplicate timestamps, but timestamp %s appears more than once in" + + " the same partition."; + public static final String EXCEPTION_UNSUPPORTED_LTTB_VALUE_TYPE_ARG_3E60F6FE = + "Unsupported LTTB value type: %s"; + public static final String + EXCEPTION_LTTB_COUNT_WINDOW_BOUNDARIES_EXCEED_THE_INT64_RANGE_5BC980CA = + "LTTB count window boundaries exceed the INT64 range."; + public static final String EXCEPTION_LTTB_WINDOW_BOUNDARIES_EXCEED_THE_TIMESTAMP_RANGE_7C0FC7E2 = + "LTTB window boundaries exceed the TIMESTAMP range."; + public static final String + EXCEPTION_LTTB_BUFFERS_EXCEED_THE_PER_PROCESSOR_MEMORY_LIMIT_OF_ARG_BYTES_954FEF00 = + "LTTB buffers exceed the per-processor memory limit of %d bytes."; public static final String EXCEPTION_SNAPSHOT_BUFFER_SIZE_MUST_NOT_EXCEED_ARG_BYTES_BUT_WAS_ARG_D1DA6F7E = "Snapshot buffer size must not exceed %d bytes, but was %d."; 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 c3af2dbe7e24a..a32d4c1df13cc 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 @@ -238,6 +238,36 @@ private CommonMessages() {} EXCEPTION_XCORR_REQUIRES_EXACTLY_TWO_CALCULATION_COLUMNS_BUT_FOUND_ARG_2FF8EB0C = "XCorr 要求必须正好有两列计算列,但实际找到 %d 列。"; public static final String EXCEPTION_COLUMN_LACK_OF_NAME = "表参数中列缺少名字"; + public static final String + EXCEPTION_EXACTLY_ONE_OF_THE_N_AND_SIZE_ARGUMENTS_MUST_BE_SPECIFIED_FOR_LTTB_54AF0733 = + "LTTB 必须且只能指定 N 和 SIZE 参数中的一个。"; + public static final String EXCEPTION_THE_N_ARGUMENT_OF_LTTB_MUST_BE_AT_LEAST_3_29CE2B87 = + "LTTB 的 N 参数必须至少为 3。"; + public static final String + EXCEPTION_THE_N_ARGUMENT_OF_LTTB_MUST_BE_A_POSITIVE_INTEGER_21795D8E = + "LTTB 的 N 参数必须是正整数。"; + public static final String + EXCEPTION_THE_N_ARGUMENT_OF_LTTB_CANNOT_BE_COMBINED_WITH_THE_SLIDE_OR_ORIGIN_ARGUMENTS_C2FF1FE1 = + "LTTB 的 N 参数不能与 SLIDE 或 ORIGIN 参数同时使用。"; + public static final String + EXCEPTION_INVALID_SCALAR_ARGUMENT_SIZE_SHOULD_BE_A_POSITIVE_VALUE_3ECF76E0 = + "无效的标量参数 SIZE,应为正数"; + public static final String + EXCEPTION_LTTB_BUFFERS_AT_MOST_ARG_ELIGIBLE_POINTS_PER_COLUMN_IN_ONE_PARTITION_OR_WINDOW_4C746D82 = + "LTTB 在单个分区或窗口内每列最多缓存 %d 个有效点。"; + public static final String + EXCEPTION_LTTB_DOES_NOT_SUPPORT_DUPLICATE_TIMESTAMPS_BUT_TIMESTAMP_ARG_APPEARS_MORE_THAN_ONCE_IN_THE_SAME_PARTITION_7713FEDD = + "LTTB 不支持重复的时间戳,但时间戳 %s 在同一分区中出现了多次。"; + public static final String EXCEPTION_UNSUPPORTED_LTTB_VALUE_TYPE_ARG_3E60F6FE = + "不支持的 LTTB 值类型:%s"; + public static final String + EXCEPTION_LTTB_COUNT_WINDOW_BOUNDARIES_EXCEED_THE_INT64_RANGE_5BC980CA = + "LTTB 计数窗口边界超出 INT64 的取值范围。"; + public static final String EXCEPTION_LTTB_WINDOW_BOUNDARIES_EXCEED_THE_TIMESTAMP_RANGE_7C0FC7E2 = + "LTTB 窗口边界超出 TIMESTAMP 的取值范围。"; + public static final String + EXCEPTION_LTTB_BUFFERS_EXCEED_THE_PER_PROCESSOR_MEMORY_LIMIT_OF_ARG_BYTES_954FEF00 = + "LTTB 缓冲区超过每个处理器 %d 字节的内存上限。"; public static final String EXCEPTION_SNAPSHOT_BUFFER_SIZE_MUST_NOT_EXCEED_ARG_BYTES_BUT_WAS_ARG_D1DA6F7E = "快照缓冲区大小不得超过 %d 字节,但实际为 %d。"; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java index cb5902986b8e8..653e616fc2f46 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java @@ -569,6 +569,9 @@ public class CommonConfig { /** Max number of rows for cte materialization */ private int maxRowsInCteBuffer = 1000; + /** The per-processor memory budget of the LTTB table function */ + private long lttbBufferSize = 64 * 1024 * 1024L; + CommonConfig() { // Empty constructor } @@ -3325,4 +3328,12 @@ public void setMaxRowsInCteBuffer(int maxRowsInCteBuffer) { public int getMaxRowsInCteBuffer() { return maxRowsInCteBuffer; } + + public void setLttbBufferSize(long lttbBufferSize) { + this.lttbBufferSize = lttbBufferSize; + } + + public long getLttbBufferSize() { + return lttbBufferSize; + } } 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 818747bb71494..69e7b38d53757 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 @@ -28,6 +28,7 @@ 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.LTTBTableFunction; 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.SessionTableFunction; @@ -49,6 +50,7 @@ public enum TableBuiltinTableFunction { VARIATION("variation"), CAPACITY("capacity"), M4("m4"), + LTTB("lttb"), FFT("fft"), FORECAST("forecast"), PATTERN_MATCH("pattern_match"), @@ -99,6 +101,8 @@ public static TableFunction getBuiltinTableFunction(String functionName) { return new CapacityTableFunction(); case "m4": return new M4TableFunction(); + case "lttb": + return new LTTBTableFunction(); case "fft": return new FFTTableFunction(); case "forecast": diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LTTBTableFunction.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LTTBTableFunction.java new file mode 100644 index 0000000000000..68793615c82b9 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LTTBTableFunction.java @@ -0,0 +1,1387 @@ +/* + * 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.conf.CommonDescriptor; +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 org.apache.tsfile.utils.RamUsageEstimator; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Deque; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.apache.iotdb.udf.api.relational.table.argument.ScalarArgumentChecker.POSITIVE_LONG_CHECKER; + +/** + * Largest-Triangle-Three-Buckets (LTTB) downsampling as a built-in Table Model table function. + * + *

The function shares its table-argument conventions with {@link M4TableFunction}: {@code DATA} + * has set semantics, {@code TIMECOL} names the {@code TIMESTAMP} ordering column, partition columns + * are echoed into the proper output, and every other numeric column is a participant column that is + * downsampled independently. Two mutually exclusive execution modes exist: + * + *

    + *
  • Target-count mode ({@code N}): the whole partition is buffered and the standard LTTB + * algorithm reduces each participant column to exactly {@code N} points (or all points when + * the column has no more than {@code N} eligible points). The output uses a fixed {@code + * window_index = 0}. + *
  • Window/bucket mode ({@code SIZE}, optional {@code SLIDE}/{@code ORIGIN}): windows + * are built exactly like M4 (time windows for duration literals, count windows for integers). + * {@code SIZE} defines a bucket, not the input range of a separate standard LTTB invocation. + * One anchor chain is maintained per partition and participant column: for the current bucket + * {@code A} is the point selected by the preceding bucket of that column ({@code B} in turn + * becomes the anchor of the following bucket), {@code B} ranges over the current bucket's + * eligible points, and {@code C} is the average point of the next bucket that carries data + * for that column (or the bucket's own last eligible point when there is none). Overlapping + * buckets ({@code SLIDE < SIZE}) do not restart the chain, and the state is independent + * between partitions and between participant columns. + *
+ * + *

Bucket mode emits one row per window that carries input, and at most one selected point per + * participant column in a row; a column with no eligible point in a window emits {@code NULL} in + * its place, and a window in which every participant column is {@code NULL} still emits a single + * all-{@code NULL} row (padding by output position). Bucket mode does not preserve the + * first and last eligible points: an endpoint used as {@code A} or as {@code C} is a triangle + * vertex, not a selected point, so it appears in the output only when it also wins its bucket. A + * bucket holding just one or two eligible points therefore still contributes a single row per + * column: with two points the areas are equal and the tie resolves to the earlier one. First/last + * preservation and the return-all-points case apply to target-count mode only. + * + *

{@code NULL} participant values are ignored per column; every non-null value, including {@code + * NaN} and the infinities, participates in the selection and is reproduced in the output. Ties in + * the triangle area are broken deterministically by the earliest timestamp (stable input order). + * + *

The timestamps of the input rows of one partition must be unique. As soon as two rows of the + * same partition carry the same timestamp the query fails with a semantic error ({@code 701}); the + * check runs at the partition input boundary, before per-column {@code NULL} filtering and before + * window expansion, so a row that several overlapping windows share is still a single input row. + * Equal timestamps in different partitions remain valid. Since the input is ordered by {@code + * TIMECOL}, equal timestamps are adjacent, and the check compares each row with the previous one + * and carries the last seen timestamp across input blocks. + */ +public class LTTBTableFunction implements TableFunction { + + public static final String DATA_PARAMETER_NAME = "DATA"; + public static final String TIMECOL_PARAMETER_NAME = "TIMECOL"; + public static final String N_PARAMETER_NAME = "N"; + public static final String SIZE_PARAMETER_NAME = "SIZE"; + public static final String SLIDE_PARAMETER_NAME = "SLIDE"; + public static final String ORIGIN_PARAMETER_NAME = "ORIGIN"; + + /** Internal boolean argument injected by the analyzer: {@code true} if SIZE is a duration. */ + public static final String WINDOW_MODE_PARAMETER_NAME = "__LTTB_WINDOW_MODE"; + + public static final long UNSPECIFIED_N = Long.MIN_VALUE; + public static final long UNSPECIFIED_SIZE = Long.MIN_VALUE; + public static final long UNSPECIFIED_SLIDE = Long.MIN_VALUE; + public static final long MIN_TARGET_COUNT = 3L; + + /** Hard cap on eligible points buffered per participant column in one partition or window. */ + public static final int MAX_BUFFERED_POINTS = 1 << 22; + + /** + * Shared budget for point buffers and selection state across all columns of one processor. It is + * the default of {@code lttb_buffer_size_in_bytes}; operators can tune the live value with that + * hot-reloadable configuration parameter for workloads with large partitions or many overlapping + * windows. + */ + public static final long DEFAULT_MAX_BUFFER_BYTES = 64L * 1024 * 1024; + + public static final String MODE_PROPERTY = "__LTTB_MODE"; + public static final String PARTITION_TYPES_PROPERTY = "__LTTB_PARTITION_TYPES"; + public static final String PARTICIPANT_TYPES_PROPERTY = "__LTTB_PARTICIPANT_TYPES"; + + private static final String OUTPUT_WINDOW_START_COLUMN = "window_start"; + private static final String OUTPUT_WINDOW_END_COLUMN = "window_end"; + private static final String OUTPUT_WINDOW_INDEX_COLUMN = "window_index"; + private static final String TIME_SUFFIX = "_time"; + private static final int INITIAL_CAPACITY = 64; + private static final int NOT_SELECTED = -1; + + /** Execution mode persisted in the {@link MapTableFunctionHandle}. */ + public enum Mode { + TARGET_COUNT, + TIME_WINDOW, + COUNT_WINDOW + } + + /** + * The live per-processor buffer budget, read from the hot-reloadable {@code + * lttb_buffer_size_in_bytes} configuration parameter. Values that are not positive fall back to + * {@link #DEFAULT_MAX_BUFFER_BYTES} so a misconfigured node still bounds memory instead of + * buffering without limit. Processors read this once at construction, so a reload applies to + * queries that start afterwards. + */ + public static long maxBufferBytes() { + long configured = CommonDescriptor.getInstance().getConfig().getLttbBufferSize(); + return configured > 0 ? configured : DEFAULT_MAX_BUFFER_BYTES; + } + + @Override + public List getArgumentsSpecifications() { + return Arrays.asList( + TableParameterSpecification.builder().name(DATA_PARAMETER_NAME).setSemantics().build(), + ScalarParameterSpecification.builder() + .name(TIMECOL_PARAMETER_NAME) + .type(Type.STRING) + .build(), + ScalarParameterSpecification.builder() + .name(N_PARAMETER_NAME) + .type(Type.INT64) + .defaultValue(UNSPECIFIED_N) + .build(), + ScalarParameterSpecification.builder() + .name(SIZE_PARAMETER_NAME) + .type(Type.INT64) + .defaultValue(UNSPECIFIED_SIZE) + .addChecker(POSITIVE_LONG_CHECKER) + .build(), + ScalarParameterSpecification.builder() + .name(SLIDE_PARAMETER_NAME) + .type(Type.INT64) + .defaultValue(UNSPECIFIED_SLIDE) + .addChecker(POSITIVE_LONG_CHECKER) + .build(), + ScalarParameterSpecification.builder() + .name(ORIGIN_PARAMETER_NAME) + .type(Type.TIMESTAMP) + .defaultValue(0L) + .build()); + } + + @Override + public TableFunctionAnalysis analyze(Map arguments) throws UDFException { + int timeColumnIndex = + WindowTVFUtils.checkOrderByColumn(arguments, DATA_PARAMETER_NAME, TIMECOL_PARAMETER_NAME); + TableArgument tableArgument = (TableArgument) arguments.get(DATA_PARAMETER_NAME); + List partitionIndexes = WindowTVFUtils.getPartitionIndexes(tableArgument); + Set excludedIndexes = new HashSet<>(partitionIndexes); + excludedIndexes.add(timeColumnIndex); + + long n = getLong(arguments, N_PARAMETER_NAME); + long size = getLong(arguments, SIZE_PARAMETER_NAME); + long slide = getLong(arguments, SLIDE_PARAMETER_NAME); + boolean nSpecified = n != UNSPECIFIED_N; + boolean sizeSpecified = size != UNSPECIFIED_SIZE; + boolean slideSpecified = slide != UNSPECIFIED_SLIDE; + if (nSpecified == sizeSpecified) { + throw new SemanticException( + CommonMessages + .EXCEPTION_EXACTLY_ONE_OF_THE_N_AND_SIZE_ARGUMENTS_MUST_BE_SPECIFIED_FOR_LTTB_54AF0733); + } + + Mode mode; + if (nSpecified) { + if (slideSpecified) { + throw new SemanticException( + CommonMessages + .EXCEPTION_THE_N_ARGUMENT_OF_LTTB_CANNOT_BE_COMBINED_WITH_THE_SLIDE_OR_ORIGIN_ARGUMENTS_C2FF1FE1); + } + if (n < MIN_TARGET_COUNT) { + throw new SemanticException( + CommonMessages.EXCEPTION_THE_N_ARGUMENT_OF_LTTB_MUST_BE_AT_LEAST_3_29CE2B87); + } + mode = Mode.TARGET_COUNT; + } else { + if (size <= 0) { + throw new UDFException( + CommonMessages + .EXCEPTION_INVALID_SCALAR_ARGUMENT_SIZE_SHOULD_BE_A_POSITIVE_VALUE_3ECF76E0); + } + if (!slideSpecified) { + slide = size; + } else if (slide <= 0) { + throw new UDFException( + CommonMessages + .EXCEPTION_INVALID_SCALAR_ARGUMENT_SLIDE_SHOULD_BE_A_POSITIVE_VALUE_F019E091); + } + boolean isTimeWindow = + arguments.containsKey(WINDOW_MODE_PARAMETER_NAME) + && (boolean) ((ScalarArgument) arguments.get(WINDOW_MODE_PARAMETER_NAME)).getValue(); + mode = isTimeWindow ? Mode.TIME_WINDOW : Mode.COUNT_WINDOW; + } + + DescribedSchema.Builder schemaBuilder = new DescribedSchema.Builder(); + if (mode == Mode.TIME_WINDOW) { + schemaBuilder + .addField(OUTPUT_WINDOW_START_COLUMN, Type.TIMESTAMP) + .addField(OUTPUT_WINDOW_END_COLUMN, Type.TIMESTAMP); + } else { + schemaBuilder.addField(OUTPUT_WINDOW_INDEX_COLUMN, Type.INT64); + } + + List partitionTypes = new ArrayList<>(); + for (int partitionIndex : partitionIndexes) { + Type type = tableArgument.getFieldTypes().get(partitionIndex); + partitionTypes.add(type); + schemaBuilder.addField(tableArgument.getFieldNames().get(partitionIndex).get(), type); + } + + // Rejects every non-numeric participant column with the shared "not allowed columns" message. + List participantIndexes = + WindowTVFUtils.getCalculationIndexes(tableArgument, excludedIndexes, null); + if (participantIndexes.isEmpty()) { + throw new SemanticException(CommonMessages.EXCEPTION_NO_CALCULATE_COLUMNS); + } + List participantTypes = new ArrayList<>(); + for (int participantIndex : participantIndexes) { + Type type = tableArgument.getFieldTypes().get(participantIndex); + String columnName = tableArgument.getFieldNames().get(participantIndex).get(); + participantTypes.add(type); + schemaBuilder.addField(columnName + TIME_SUFFIX, Type.TIMESTAMP); + schemaBuilder.addField(columnName, type); + } + + MapTableFunctionHandle.Builder handleBuilder = + new MapTableFunctionHandle.Builder() + .addProperty(MODE_PROPERTY, mode.name()) + .addProperty(PARTITION_TYPES_PROPERTY, WindowTVFUtils.joinTypes(partitionTypes)) + .addProperty(PARTICIPANT_TYPES_PROPERTY, WindowTVFUtils.joinTypes(participantTypes)); + if (mode == Mode.TARGET_COUNT) { + handleBuilder.addProperty(N_PARAMETER_NAME, n); + } else { + handleBuilder.addProperty(SIZE_PARAMETER_NAME, size).addProperty(SLIDE_PARAMETER_NAME, slide); + if (mode == Mode.TIME_WINDOW) { + handleBuilder.addProperty(ORIGIN_PARAMETER_NAME, getLong(arguments, ORIGIN_PARAMETER_NAME)); + } + } + + List requiredColumns = new ArrayList<>(); + requiredColumns.add(timeColumnIndex); + requiredColumns.addAll(partitionIndexes); + requiredColumns.addAll(participantIndexes); + + return TableFunctionAnalysis.builder() + .properColumnSchema(schemaBuilder.build()) + .requireRecordSnapshot(false) + .requiredColumns(DATA_PARAMETER_NAME, requiredColumns) + .handle(handleBuilder.build()) + .build(); + } + + private static long getLong(Map arguments, String name) { + return (long) ((ScalarArgument) arguments.get(name)).getValue(); + } + + @Override + public TableFunctionHandle createTableFunctionHandle() { + return new MapTableFunctionHandle(); + } + + @Override + public TableFunctionProcessorProvider getProcessorProvider( + TableFunctionHandle tableFunctionHandle) { + MapTableFunctionHandle handle = (MapTableFunctionHandle) tableFunctionHandle; + Mode mode = Mode.valueOf((String) handle.getProperty(MODE_PROPERTY)); + Type[] partitionTypes = + WindowTVFUtils.parseTypes((String) handle.getProperty(PARTITION_TYPES_PROPERTY)); + Type[] participantTypes = + WindowTVFUtils.parseTypes((String) handle.getProperty(PARTICIPANT_TYPES_PROPERTY)); + + return new TableFunctionProcessorProvider() { + @Override + public TableFunctionDataProcessor getDataProcessor() { + ParticipantColumn[] participantColumns = + createParticipantColumns(participantTypes, partitionTypes.length + 1); + switch (mode) { + case TARGET_COUNT: + return new TargetCountLTTBDataProcessor( + (long) handle.getProperty(N_PARAMETER_NAME), partitionTypes, participantColumns); + case TIME_WINDOW: + return new TimeWindowLTTBDataProcessor( + (long) handle.getProperty(SIZE_PARAMETER_NAME), + (long) handle.getProperty(SLIDE_PARAMETER_NAME), + (long) handle.getProperty(ORIGIN_PARAMETER_NAME), + partitionTypes, + participantColumns); + case COUNT_WINDOW: + return new CountWindowLTTBDataProcessor( + (long) handle.getProperty(SIZE_PARAMETER_NAME), + (long) handle.getProperty(SLIDE_PARAMETER_NAME), + partitionTypes, + participantColumns); + default: + throw new IllegalStateException(String.valueOf(mode)); + } + } + + /** + * Every LTTB processor buffers the points of the partition (or of the windows it keeps open) + * until it can select the output points, so the whole per-processor budget is reportable as + * memory the operator holds while the partition is processed. + */ + @Override + public long getEstimatedMaxBufferedBytes() { + return maxBufferBytes(); + } + }; + } + + private static ParticipantColumn[] createParticipantColumns(Type[] types, int firstInputIndex) { + ParticipantColumn[] columns = new ParticipantColumn[types.length]; + for (int i = 0; i < types.length; i++) { + columns[i] = new ParticipantColumn(firstInputIndex + i, types[i]); + } + return columns; + } + + // --------------------------------------------------------------------------------------------- + // LTTB core + // --------------------------------------------------------------------------------------------- + + /** + * Standard LTTB selection over an ordered point sequence. + * + *

When {@code count <= n} every index is returned. Otherwise the first and last points are + * retained and the {@code count - 2} intermediate points are divided into {@code n - 2} buckets; + * bucket {@code i} covers {@code [floor(i*(count-2)/(n-2))+1, floor((i+1)*(count-2)/(n-2))+1)}. + * For every bucket the candidate forming the largest triangle with the previously selected point + * and the average point of the next bucket wins (the last bucket uses the final point). Ties are + * broken by the earliest timestamp. + * + * @return ascending indexes of the selected points + */ + static int[] selectTargetCount(long[] times, double[] values, int count, long n) { + return selectTargetCount(times, values, null, count, n); + } + + private static int[] selectTargetCount( + long[] times, double[] values, long[] integralValues, int count, long n) { + if (count <= n) { + int[] all = new int[count]; + for (int i = 0; i < count; i++) { + all[i] = i; + } + return all; + } + + int[] selected = new int[(int) n]; + int selectedCount = 0; + selected[selectedCount++] = 0; + long bucketCount = n - 2; + long intermediate = count - 2L; + int anchor = 0; + for (long bucket = 0; bucket < bucketCount; bucket++) { + int currentStart = (int) (bucket * intermediate / bucketCount) + 1; + int currentEnd = (int) ((bucket + 1) * intermediate / bucketCount) + 1; + + long anchorTime = times[anchor]; + double anchorValue = values[anchor]; + long integralAnchor = integralValues == null ? 0 : integralValues[anchor]; + double avgTime; + double avgValue; + int nextEnd = + bucket == bucketCount - 1 ? count : (int) ((bucket + 2) * intermediate / bucketCount) + 1; + if (bucket == bucketCount - 1) { + avgTime = differenceAsDouble(times[count - 1], anchorTime); + avgValue = relativeValue(values, integralValues, count - 1, anchorValue, integralAnchor); + } else { + double sumTime = 0; + double sumValue = 0; + for (int i = currentEnd; i < nextEnd; i++) { + sumTime += differenceAsDouble(times[i], anchorTime); + sumValue += relativeValue(values, integralValues, i, anchorValue, integralAnchor); + } + int nextSize = nextEnd - currentEnd; + avgTime = sumTime / nextSize; + avgValue = sumValue / nextSize; + } + + int pick = + argMaxArea( + times, + values, + integralValues, + currentStart, + currentEnd, + anchorTime, + anchorValue, + integralAnchor, + avgTime, + avgValue); + if (pick == NOT_SELECTED) { + pick = + argMaxAreaExactly( + times, + values, + integralValues, + currentStart, + currentEnd, + anchorTime, + anchorValue, + integralAnchor, + times, + values, + integralValues, + currentEnd, + nextEnd); + } + selected[selectedCount++] = pick; + anchor = pick; + } + selected[selectedCount] = count - 1; + return selected; + } + + /** + * Area of the triangle formed by the candidate {@code index}, the anchor and the lookahead point, + * or {@code -1} when no finite area exists. Areas are non-negative, so {@code -1} unambiguously + * marks a candidate that cannot be scored. + * + *

A {@code NaN} or infinite participant value yields a non-finite (undefined) area rather than + * an overflow: no arithmetic can make it finite. Such a candidate never beats a finite one, so a + * bucket whose candidates are all non-finite keeps its earliest point. + */ + private static double boundedArea( + long[] times, + double[] values, + long[] integralValues, + int index, + long anchorTime, + double anchorValue, + long integralAnchor, + double nextTime, + double nextValue) { + double area = + Math.abs( + nextTime * relativeValue(values, integralValues, index, anchorValue, integralAnchor) + - differenceAsDouble(times[index], anchorTime) * nextValue); + return Double.isFinite(area) ? area : -1; + } + + /** Whether the participant value at {@code index} is representable as an exact decimal. */ + private static boolean hasExactValue(double[] values, long[] integralValues, int index) { + return integralValues != null || Double.isFinite(values[index]); + } + + private static int argMaxArea( + long[] times, + double[] values, + long[] integralValues, + int start, + int end, + long anchorTime, + double anchorValue, + long integralAnchor, + double nextTime, + double nextValue) { + // The exact fallback rebuilds the lookahead sum from the original points, so a non-finite + // average over representable points is still recoverable; only the anchor origin has to be + // representable, because it is converted to BigDecimal directly. A non-finite anchor value can + // only arise from non-finite input, which no arithmetic can repair. + boolean exactIsPossible = Double.isFinite(anchorValue); + int pick = start; + double maxArea = -1; + for (int i = start; i < end; i++) { + double area = + boundedArea( + times, + values, + integralValues, + i, + anchorTime, + anchorValue, + integralAnchor, + nextTime, + nextValue); + if (area < 0) { + if (!hasExactValue(values, integralValues, i)) { + // Non-finite participant: boundedArea scored it as the smallest area. + continue; + } + if (exactIsPossible) { + // Overflow on representable input: let the caller compare the whole bucket exactly. + return NOT_SELECTED; + } + } + if (area > maxArea) { + maxArea = area; + pick = i; + } + } + return pick; + } + + /** + * Overflow fallback for input values representable as exact decimals. Recomputes every candidate + * so an overflowing area cannot tie with or hide a larger one. Multiplying all areas by the + * lookahead point count lets us use sums instead of dividing by that count, keeping the + * comparison exact even for recurring averages. The normal path does not allocate decimal values. + * + *

A {@code NaN} or infinite participant cannot be represented by {@link BigDecimal}, so such a + * candidate is skipped and ties resolve to the earliest point. The caller only reaches this + * method when the anchor and the lookahead point are finite, so the {@link BigDecimal} + * conversions of those two origins and of the lookahead values are always well defined. + */ + private static int argMaxAreaExactly( + long[] times, + double[] values, + long[] integralValues, + int start, + int end, + long anchorTime, + double anchorValue, + long integralAnchor, + long[] nextTimes, + double[] nextValues, + long[] nextIntegralValues, + int nextStart, + int nextEnd) { + BigDecimal timeOrigin = BigDecimal.valueOf(anchorTime); + BigDecimal valueOrigin = + integralValues == null ? new BigDecimal(anchorValue) : BigDecimal.valueOf(integralAnchor); + BigDecimal sumTime = BigDecimal.ZERO; + BigDecimal sumValue = BigDecimal.ZERO; + for (int i = nextStart; i < nextEnd; i++) { + if (!hasExactValue(nextValues, nextIntegralValues, i)) { + continue; + } + sumTime = sumTime.add(BigDecimal.valueOf(nextTimes[i]).subtract(timeOrigin)); + sumValue = sumValue.add(exactValue(nextValues, nextIntegralValues, i).subtract(valueOrigin)); + } + int pick = start; + BigDecimal maxArea = null; + for (int i = start; i < end; i++) { + if (!hasExactValue(values, integralValues, i)) { + continue; + } + BigDecimal time = BigDecimal.valueOf(times[i]).subtract(timeOrigin); + BigDecimal value = exactValue(values, integralValues, i).subtract(valueOrigin); + BigDecimal area = sumTime.multiply(value).subtract(time.multiply(sumValue)).abs(); + if (maxArea == null || area.compareTo(maxArea) > 0) { + maxArea = area; + pick = i; + } + } + return pick; + } + + private static BigDecimal exactValue(double[] values, long[] integralValues, int index) { + // The constructor preserves the actual binary floating-point value, including tiny deltas. + return integralValues == null + ? new BigDecimal(values[index]) + : BigDecimal.valueOf(integralValues[index]); + } + + private static double relativeValue( + double[] values, long[] integralValues, int index, double origin, long integralOrigin) { + return integralValues == null + ? values[index] - origin + : differenceAsDouble(integralValues[index], integralOrigin); + } + + private static double differenceAsDouble(long value, long origin) { + long difference = value - origin; + // Subtract before converting so nearby integers retain their precision. For spans exceeding + // the long range, use floating-point subtraction instead of the overflowed long difference. + return ((value ^ origin) & (value ^ difference)) < 0 + ? (double) value - (double) origin + : (double) difference; + } + + // --------------------------------------------------------------------------------------------- + // Column access and buffering + // --------------------------------------------------------------------------------------------- + + /** Reservations include array capacity, including both old and new arrays during growth. */ + static final class BufferMemoryBudget { + private final long limit; + private long reservedBytes; + + BufferMemoryBudget(long limit) { + this.limit = limit; + } + + void reserve(long bytes) { + if (bytes > limit - reservedBytes) { + throw new SemanticException( + String.format( + CommonMessages + .EXCEPTION_LTTB_BUFFERS_EXCEED_THE_PER_PROCESSOR_MEMORY_LIMIT_OF_ARG_BYTES_954FEF00, + limit)); + } + reservedBytes += bytes; + } + + void release(long bytes) { + reservedBytes -= bytes; + } + } + + private static final class ParticipantColumn { + private final int inputIndex; + private final Type type; + + private ParticipantColumn(int inputIndex, Type type) { + this.inputIndex = inputIndex; + this.type = type; + } + + private boolean isIntegral() { + return type == Type.INT32 || type == Type.INT64; + } + + /** + * Appends the record's value to {@code buffer} if it is non-null. All non-null values are + * eligible, including {@code NaN} and the infinities: the definition only excludes {@code NULL} + * participants, and dropping non-finite values would silently turn populated input into an + * all-{@code NULL} result. + */ + private void collect(Record record, long time, PointBuffer buffer) { + if (record.isNull(inputIndex)) { + return; + } + switch (type) { + case INT32: + buffer.addIntegral(time, record.getInt(inputIndex)); + break; + case INT64: + buffer.addIntegral(time, record.getLong(inputIndex)); + break; + case FLOAT: + buffer.addFloating(time, record.getFloat(inputIndex)); + break; + case DOUBLE: + buffer.addFloating(time, record.getDouble(inputIndex)); + break; + default: + throw new IllegalArgumentException( + String.format( + CommonMessages.EXCEPTION_UNSUPPORTED_LTTB_VALUE_TYPE_ARG_3E60F6FE, type)); + } + } + + private void write( + ColumnBuilder timeBuilder, ColumnBuilder valueBuilder, PointBuffer buffer, int index) { + timeBuilder.writeLong(buffer.times[index]); + switch (type) { + case INT32: + valueBuilder.writeInt((int) buffer.integralValues[index]); + break; + case INT64: + valueBuilder.writeLong(buffer.integralValues[index]); + break; + case FLOAT: + valueBuilder.writeFloat((float) buffer.values[index]); + break; + case DOUBLE: + valueBuilder.writeDouble(buffer.values[index]); + break; + default: + throw new IllegalArgumentException( + String.format( + CommonMessages.EXCEPTION_UNSUPPORTED_LTTB_VALUE_TYPE_ARG_3E60F6FE, type)); + } + } + } + + /** + * Growable buffer of eligible points for one participant column. Integral columns additionally + * keep the exact {@code long} value so INT64 output is not rounded through {@code double}. + */ + static final class PointBuffer { + private static final long INSTANCE_SIZE = + RamUsageEstimator.shallowSizeOfInstance(PointBuffer.class); + + private final BufferMemoryBudget memoryBudget; + private long[] times; + private double[] values; + private long[] integralValues; + private int size; + + PointBuffer(boolean integral, BufferMemoryBudget memoryBudget) { + this.memoryBudget = memoryBudget; + if (memoryBudget != null) { + memoryBudget.reserve(INSTANCE_SIZE + arrayBytes(INITIAL_CAPACITY, integral)); + } + times = new long[INITIAL_CAPACITY]; + values = new double[INITIAL_CAPACITY]; + if (integral) { + integralValues = new long[INITIAL_CAPACITY]; + } + } + + private static long arrayBytes(int capacity, boolean integral) { + return RamUsageEstimator.sizeOfLongArray(capacity) + + RamUsageEstimator.sizeOfDoubleArray(capacity) + + (integral ? RamUsageEstimator.sizeOfLongArray(capacity) : 0); + } + + private void release() { + if (memoryBudget != null) { + memoryBudget.release(INSTANCE_SIZE + arrayBytes(times.length, integralValues != null)); + } + } + + void addFloating(long time, double value) { + ensureCapacity(); + times[size] = time; + values[size] = value; + size++; + } + + void addIntegral(long time, long value) { + ensureCapacity(); + times[size] = time; + values[size] = value; + integralValues[size] = value; + size++; + } + + private void ensureCapacity() { + if (size < times.length) { + return; + } + if (size >= MAX_BUFFERED_POINTS) { + throw new SemanticException( + String.format( + CommonMessages + .EXCEPTION_LTTB_BUFFERS_AT_MOST_ARG_ELIGIBLE_POINTS_PER_COLUMN_IN_ONE_PARTITION_OR_WINDOW_4C746D82, + MAX_BUFFERED_POINTS)); + } + int newCapacity = (int) Math.min((long) times.length * 2, MAX_BUFFERED_POINTS); + long oldArrayBytes = arrayBytes(times.length, integralValues != null); + if (memoryBudget != null) { + // Reserve before copying: the old arrays remain live until their replacements exist. + memoryBudget.reserve(arrayBytes(newCapacity, integralValues != null)); + } + times = Arrays.copyOf(times, newCapacity); + values = Arrays.copyOf(values, newCapacity); + if (integralValues != null) { + integralValues = Arrays.copyOf(integralValues, newCapacity); + } + if (memoryBudget != null) { + memoryBudget.release(oldArrayBytes); + } + } + + int size() { + return size; + } + + boolean isEmpty() { + return size == 0; + } + + long time(int index) { + return times[index]; + } + + double value(int index) { + return values[index]; + } + + double averageTime(long origin) { + double sum = 0; + for (int i = 0; i < size; i++) { + sum += differenceAsDouble(times[i], origin); + } + return sum / size; + } + + double averageValue(double origin, long integralOrigin) { + double sum = 0; + for (int i = 0; i < size; i++) { + sum += relativeValue(values, integralValues, i, origin, integralOrigin); + } + return sum / size; + } + } + + // --------------------------------------------------------------------------------------------- + // Processors + // --------------------------------------------------------------------------------------------- + + /** + * Partition-local timestamp uniqueness of the input rows. The input is ordered by {@code + * TIMECOL}, so a duplicate timestamp is always adjacent to its predecessor; the last accepted + * timestamp is kept in the instance so the rule also spans input blocks of the same partition. + */ + static final class TimestampUniquenessChecker { + private long previousTimestamp; + private boolean hasPreviousTimestamp; + + void check(long time) { + if (hasPreviousTimestamp && time == previousTimestamp) { + throw new SemanticException( + String.format( + CommonMessages + .EXCEPTION_LTTB_DOES_NOT_SUPPORT_DUPLICATE_TIMESTAMPS_BUT_TIMESTAMP_ARG_APPEARS_MORE_THAN_ONCE_IN_THE_SAME_PARTITION_7713FEDD, + time)); + } + previousTimestamp = time; + hasPreviousTimestamp = true; + } + } + + private abstract static class AbstractLTTBDataProcessor implements TableFunctionDataProcessor { + protected final Type[] partitionTypes; + protected final Object[] partitionValues; + protected final ParticipantColumn[] participantColumns; + private final TimestampUniquenessChecker timestampChecker = new TimestampUniquenessChecker(); + + protected AbstractLTTBDataProcessor( + Type[] partitionTypes, ParticipantColumn[] participantColumns) { + this.partitionTypes = partitionTypes; + this.partitionValues = new Object[partitionTypes.length]; + this.participantColumns = participantColumns; + } + + /** + * Rejects a repeated input timestamp of the partition. Runs at the partition input boundary, + * before per-column {@code NULL} filtering and before window expansion, so a row shared by + * several overlapping windows is still a single input row. + */ + protected final void checkTimestamp(long time) { + timestampChecker.check(time); + } + + protected final PointBuffer[] newBuffers(BufferMemoryBudget memoryBudget) { + PointBuffer[] buffers = new PointBuffer[participantColumns.length]; + for (int i = 0; i < buffers.length; i++) { + buffers[i] = new PointBuffer(participantColumns[i].isIntegral(), memoryBudget); + } + return buffers; + } + + /** Partition columns are constant within a partition; keep the first non-null value. */ + protected final void capturePartitionValues(Record input) { + for (int i = 0; i < partitionTypes.length; i++) { + int inputIndex = i + 1; + if (partitionValues[i] == null && !input.isNull(inputIndex)) { + partitionValues[i] = WindowTVFUtils.readValue(input, inputIndex, partitionTypes[i]); + } + } + } + + protected final void collect(Record input, long time, PointBuffer[] buffers) { + for (int i = 0; i < participantColumns.length; i++) { + participantColumns[i].collect(input, time, buffers[i]); + } + } + + /** + * Writes one output row. {@code windowColumnCount} leading builders were already filled by the + * caller; {@code selectedIndexes[i]} is {@link #NOT_SELECTED} when column {@code i} has no + * point. + */ + protected final void writeRow( + List builders, + int windowColumnCount, + PointBuffer[] buffers, + int[] selectedIndexes) { + int builderIndex = windowColumnCount; + for (int i = 0; i < partitionTypes.length; i++) { + WindowTVFUtils.writeValue( + builders.get(builderIndex++), partitionValues[i], partitionTypes[i]); + } + for (int i = 0; i < participantColumns.length; i++) { + ColumnBuilder timeBuilder = builders.get(builderIndex++); + ColumnBuilder valueBuilder = builders.get(builderIndex++); + if (selectedIndexes[i] == NOT_SELECTED) { + timeBuilder.appendNull(); + valueBuilder.appendNull(); + } else { + participantColumns[i].write(timeBuilder, valueBuilder, buffers[i], selectedIndexes[i]); + } + } + } + } + + /** Buffers the whole partition and applies the standard LTTB algorithm in {@link #finish}. */ + private static final class TargetCountLTTBDataProcessor extends AbstractLTTBDataProcessor { + private final long n; + private final PointBuffer[] buffers; + private final BufferMemoryBudget memoryBudget = new BufferMemoryBudget(maxBufferBytes()); + private boolean hasRows; + + private TargetCountLTTBDataProcessor( + long n, Type[] partitionTypes, ParticipantColumn[] participantColumns) { + super(partitionTypes, participantColumns); + this.n = n; + memoryBudget.reserve(RamUsageEstimator.sizeOfObjectArray(participantColumns.length)); + this.buffers = newBuffers(memoryBudget); + } + + @Override + public void process( + Record input, + List properColumnBuilders, + ColumnBuilder passThroughIndexBuilder) { + hasRows = true; + long time = input.getLong(0); + checkTimestamp(time); + capturePartitionValues(input); + collect(input, time, buffers); + } + + @Override + public void finish( + List properColumnBuilders, ColumnBuilder passThroughIndexBuilder) { + if (!hasRows) { + return; + } + long selectionBytes = + RamUsageEstimator.sizeOfObjectArray(participantColumns.length) + + RamUsageEstimator.sizeOfIntArray(participantColumns.length); + for (PointBuffer buffer : buffers) { + selectionBytes += RamUsageEstimator.sizeOfIntArray((int) Math.min(buffer.size, n)); + } + memoryBudget.reserve(selectionBytes); + int[][] selections = new int[participantColumns.length][]; + int rowCount = 1; + for (int i = 0; i < participantColumns.length; i++) { + PointBuffer buffer = buffers[i]; + selections[i] = + selectTargetCount(buffer.times, buffer.values, buffer.integralValues, buffer.size, n); + rowCount = Math.max(rowCount, selections[i].length); + } + + int[] selectedIndexes = new int[participantColumns.length]; + for (int row = 0; row < rowCount; row++) { + properColumnBuilders.get(0).writeLong(0L); + for (int i = 0; i < participantColumns.length; i++) { + selectedIndexes[i] = row < selections[i].length ? selections[i][row] : NOT_SELECTED; + } + writeRow(properColumnBuilders, 1, buffers, selectedIndexes); + } + } + + @Override + public void beforeDestroy() { + Arrays.fill(buffers, null); + memoryBudget.reservedBytes = 0; + } + } + + /** One window (= one LTTB bucket) worth of eligible points. */ + private abstract static class WindowState { + protected final PointBuffer[] buffers; + protected final PointBuffer[] nextNonemptyBuffers; + protected final long endExclusive; + protected int unresolvedColumns; + + private WindowState(PointBuffer[] buffers, long endExclusive) { + this.buffers = buffers; + this.nextNonemptyBuffers = new PointBuffer[buffers.length]; + this.endExclusive = endExclusive; + } + + protected abstract void writeWindowColumns(List builders); + + protected abstract int getWindowColumnCount(); + } + + private static final class TimeWindowState extends WindowState { + private final long windowStart; + + private TimeWindowState(PointBuffer[] buffers, long windowStart, long endExclusive) { + super(buffers, endExclusive); + this.windowStart = windowStart; + } + + @Override + protected void writeWindowColumns(List builders) { + builders.get(0).writeLong(windowStart); + builders.get(1).writeLong(endExclusive); + } + + @Override + protected int getWindowColumnCount() { + return 2; + } + } + + private static final class CountWindowState extends WindowState { + private final long windowIndex; + + private CountWindowState(PointBuffer[] buffers, long endExclusive, long windowIndex) { + super(buffers, endExclusive); + this.windowIndex = windowIndex; + } + + @Override + protected void writeWindowColumns(List builders) { + builders.get(0).writeLong(windowIndex); + } + + @Override + protected int getWindowColumnCount() { + return 1; + } + } + + /** + * Window/bucket mode. Windows are constructed like M4 and emitted in window order; a window is + * emitted only once the following window with data is closed (or in {@link #finish}), because the + * average point of that following window is the third vertex of the LTTB triangle. Anchors are + * chained per participant column across the windows of the partition; windows without data for a + * column leave its anchor untouched. + */ + private abstract static class AbstractWindowLTTBDataProcessor + extends AbstractLTTBDataProcessor { + private static final long WINDOW_INSTANCE_SIZE = + Math.max( + RamUsageEstimator.shallowSizeOfInstance(TimeWindowState.class), + RamUsageEstimator.shallowSizeOfInstance(CountWindowState.class)); + + protected final long size; + protected final long slide; + protected final Deque activeWindows = new ArrayDeque<>(); + private final Deque pendingWindows = new ArrayDeque<>(); + private final WindowState[] lastNonemptyWindows; + private final BufferMemoryBudget memoryBudget = new BufferMemoryBudget(maxBufferBytes()); + private final long windowMetadataBytes; + private final boolean[] hasAnchor; + private final long[] anchorTimes; + private final double[] anchorValues; + private final long[] integralAnchorValues; + + protected AbstractWindowLTTBDataProcessor( + long size, long slide, Type[] partitionTypes, ParticipantColumn[] participantColumns) { + super(partitionTypes, participantColumns); + this.size = size; + this.slide = slide; + this.windowMetadataBytes = + WINDOW_INSTANCE_SIZE + 2 * RamUsageEstimator.sizeOfObjectArray(participantColumns.length); + this.lastNonemptyWindows = new WindowState[participantColumns.length]; + this.hasAnchor = new boolean[participantColumns.length]; + this.anchorTimes = new long[participantColumns.length]; + this.anchorValues = new double[participantColumns.length]; + this.integralAnchorValues = new long[participantColumns.length]; + } + + protected final PointBuffer[] newWindowBuffers() { + // Even an all-NULL window allocates state, so reserve before creating any of its buffers. + memoryBudget.reserve(windowMetadataBytes); + return newBuffers(memoryBudget); + } + + private void releaseWindow(W window) { + for (PointBuffer buffer : window.buffers) { + buffer.release(); + } + memoryBudget.release(windowMetadataBytes); + } + + @Override + public final void process( + Record input, + List properColumnBuilders, + ColumnBuilder passThroughIndexBuilder) { + long time = input.getLong(0); + checkTimestamp(time); + capturePartitionValues(input); + processRecord(input, time, properColumnBuilders); + } + + protected abstract void processRecord( + Record input, long time, List properColumnBuilders); + + /** Close windows in order and link each column to its next nonempty, closed bucket. */ + protected final void emitClosedWindows(long position, List builders) { + while (!activeWindows.isEmpty() && activeWindows.peekFirst().endExclusive <= position) { + closeWindow(activeWindows.removeFirst()); + } + emitReadyWindows(builders, false); + } + + private void closeWindow(W window) { + for (int i = 0; i < participantColumns.length; i++) { + if (!window.buffers[i].isEmpty()) { + WindowState previous = lastNonemptyWindows[i]; + if (previous != null) { + previous.nextNonemptyBuffers[i] = window.buffers[i]; + previous.unresolvedColumns--; + } + lastNonemptyWindows[i] = window; + window.unresolvedColumns++; + } + } + pendingWindows.addLast(window); + } + + private void emitReadyWindows(List builders, boolean finishing) { + while (!pendingWindows.isEmpty() + && (finishing || pendingWindows.peekFirst().unresolvedColumns == 0)) { + W current = pendingWindows.removeFirst(); + emitWindow(current, builders); + releaseWindow(current); + } + } + + @Override + public final void finish( + List properColumnBuilders, ColumnBuilder passThroughIndexBuilder) { + while (!activeWindows.isEmpty()) { + closeWindow(activeWindows.removeFirst()); + } + emitReadyWindows(properColumnBuilders, true); + Arrays.fill(lastNonemptyWindows, null); + } + + @Override + public final void beforeDestroy() { + activeWindows.clear(); + pendingWindows.clear(); + Arrays.fill(lastNonemptyWindows, null); + // Also covers a reservation made by a window whose construction failed at the limit. + memoryBudget.reservedBytes = 0; + } + + private void emitWindow(W current, List builders) { + int[] selectedIndexes = new int[participantColumns.length]; + for (int i = 0; i < participantColumns.length; i++) { + PointBuffer bucket = current.buffers[i]; + if (bucket.isEmpty()) { + selectedIndexes[i] = NOT_SELECTED; + continue; + } + if (!hasAnchor[i]) { + // The first eligible point of the partition seeds the anchor, as in standard LTTB. + hasAnchor[i] = true; + anchorTimes[i] = bucket.time(0); + anchorValues[i] = bucket.value(0); + if (bucket.integralValues != null) { + integralAnchorValues[i] = bucket.integralValues[0]; + } + } + double nextTime; + double nextValue; + PointBuffer next = current.nextNonemptyBuffers[i]; + if (next != null) { + nextTime = next.averageTime(anchorTimes[i]); + nextValue = next.averageValue(anchorValues[i], integralAnchorValues[i]); + } else { + int last = bucket.size() - 1; + nextTime = differenceAsDouble(bucket.time(last), anchorTimes[i]); + nextValue = + relativeValue( + bucket.values, + bucket.integralValues, + last, + anchorValues[i], + integralAnchorValues[i]); + } + int pick = + argMaxArea( + bucket.times, + bucket.values, + bucket.integralValues, + 0, + bucket.size(), + anchorTimes[i], + anchorValues[i], + integralAnchorValues[i], + nextTime, + nextValue); + if (pick == NOT_SELECTED) { + PointBuffer lookahead = next == null ? bucket : next; + pick = + argMaxAreaExactly( + bucket.times, + bucket.values, + bucket.integralValues, + 0, + bucket.size(), + anchorTimes[i], + anchorValues[i], + integralAnchorValues[i], + lookahead.times, + lookahead.values, + lookahead.integralValues, + next == null ? bucket.size() - 1 : 0, + lookahead.size()); + } + selectedIndexes[i] = pick; + anchorTimes[i] = bucket.time(pick); + anchorValues[i] = bucket.value(pick); + if (bucket.integralValues != null) { + integralAnchorValues[i] = bucket.integralValues[pick]; + } + } + current.writeWindowColumns(builders); + writeRow(builders, current.getWindowColumnCount(), current.buffers, selectedIndexes); + } + } + + private static final class TimeWindowLTTBDataProcessor + extends AbstractWindowLTTBDataProcessor { + private final long origin; + private boolean nextWindowStartInitialized = false; + private boolean windowsExhausted; + private long nextWindowStart; + + private TimeWindowLTTBDataProcessor( + long size, + long slide, + long origin, + Type[] partitionTypes, + ParticipantColumn[] participantColumns) { + super(size, slide, partitionTypes, participantColumns); + this.origin = origin; + } + + @Override + protected void processRecord( + Record input, long time, List properColumnBuilders) { + emitClosedWindows(time, properColumnBuilders); + + if (!windowsExhausted) { + long firstCandidateStart = getFirstWindowStart(time); + if (!nextWindowStartInitialized || nextWindowStart < firstCandidateStart) { + nextWindowStart = firstCandidateStart; + nextWindowStartInitialized = true; + } + + while (!windowsExhausted && nextWindowStart <= time) { + long windowEnd; + try { + windowEnd = Math.addExact(nextWindowStart, size); + } catch (ArithmeticException e) { + throw windowBoundaryOverflow(); + } + activeWindows.addLast( + new TimeWindowState(newWindowBuffers(), nextWindowStart, windowEnd)); + // An unrepresentable next start means there are no more windows, but the current + // window can still be valid and must collect this and subsequent input rows. + if (nextWindowStart > Long.MAX_VALUE - slide) { + windowsExhausted = true; + } else { + nextWindowStart += slide; + } + } + } + + for (TimeWindowState window : activeWindows) { + collect(input, time, window.buffers); + } + } + + /** First aligned start whose window ends strictly after the current timestamp. */ + private long getFirstWindowStart(long time) { + try { + long distance = Math.subtractExact(Math.subtractExact(time, origin), size); + long steps = Math.addExact(Math.floorDiv(distance, slide), 1); + return Math.addExact(origin, Math.multiplyExact(steps, slide)); + } catch (ArithmeticException e) { + // Only intermediate arithmetic may overflow, e.g. time near Long.MIN_VALUE. + // Fall back to exact arithmetic before deciding whether the actual boundary is valid. + BigInteger step = BigInteger.valueOf(slide); + BigInteger[] quotientAndRemainder = + BigInteger.valueOf(time) + .subtract(BigInteger.valueOf(origin)) + .subtract(BigInteger.valueOf(size)) + .divideAndRemainder(step); + BigInteger steps = quotientAndRemainder[0]; + if (quotientAndRemainder[1].signum() >= 0) { + steps = steps.add(BigInteger.ONE); + } + BigInteger start = BigInteger.valueOf(origin).add(steps.multiply(step)); + if (start.compareTo(BigInteger.valueOf(Long.MAX_VALUE)) > 0) { + // The input lies in a gap and the next aligned window is beyond the time domain. + windowsExhausted = true; + return Long.MAX_VALUE; + } + try { + return start.longValueExact(); + } catch (ArithmeticException overflow) { + throw windowBoundaryOverflow(); + } + } + } + + private SemanticException windowBoundaryOverflow() { + return new SemanticException( + CommonMessages.EXCEPTION_LTTB_WINDOW_BOUNDARIES_EXCEED_THE_TIMESTAMP_RANGE_7C0FC7E2); + } + } + + private static final class CountWindowLTTBDataProcessor + extends AbstractWindowLTTBDataProcessor { + private long rowCount = 0; + private long nextWindowStart = 0; + private long nextWindowIndex = 0; + private boolean windowsExhausted; + + private CountWindowLTTBDataProcessor( + long size, long slide, Type[] partitionTypes, ParticipantColumn[] participantColumns) { + super(size, slide, partitionTypes, participantColumns); + } + + @Override + protected void processRecord( + Record input, long time, List properColumnBuilders) { + emitClosedWindows(rowCount, properColumnBuilders); + + // Input row positions advance one at a time, so no expired starts need to be skipped. + // Check the end only when a window is reached: a distant future end may be unrepresentable + // even though every window that actually receives input is valid. + if (!windowsExhausted && nextWindowStart == rowCount) { + long windowEnd; + try { + windowEnd = Math.addExact(nextWindowStart, size); + } catch (ArithmeticException e) { + throw new SemanticException( + CommonMessages + .EXCEPTION_LTTB_COUNT_WINDOW_BOUNDARIES_EXCEED_THE_INT64_RANGE_5BC980CA); + } + activeWindows.addLast( + new CountWindowState(newWindowBuffers(), windowEnd, nextWindowIndex++)); + if (nextWindowStart > Long.MAX_VALUE - slide) { + windowsExhausted = true; + } else { + nextWindowStart += slide; + } + } + + for (CountWindowState window : activeWindows) { + collect(input, time, window.buffers); + } + rowCount++; + } + } +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LTTBNumericBoundaryTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LTTBNumericBoundaryTest.java new file mode 100644 index 0000000000000..3fa9cc10264f6 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LTTBNumericBoundaryTest.java @@ -0,0 +1,341 @@ +/* + * 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.relational.access.Record; +import org.apache.iotdb.udf.api.relational.table.MapTableFunctionHandle; +import org.apache.iotdb.udf.api.relational.table.processor.TableFunctionDataProcessor; +import org.apache.iotdb.udf.api.type.Type; + +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.read.common.block.column.DoubleColumnBuilder; +import org.apache.tsfile.read.common.block.column.LongColumnBuilder; +import org.junit.Test; + +import java.lang.reflect.Proxy; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; + +public class LTTBNumericBoundaryTest { + + @Test + public void testInt64TargetCountPreservesPeakAtLargeBaselines() { + Long[][] values = {{0L}, {1L}, {8L}, {2L}, {0L}, {0L}}; + for (long offset : new long[] {0, 1L << 60, Long.MIN_VALUE, Long.MAX_VALUE - 8}) { + Column[] output = sample("TARGET_COUNT", 3, 3, 0, times(6), translate(values, offset)); + assertArrayEquals(new long[] {0, 2, 5}, longValues(output[1])); + assertArrayEquals(new long[] {offset, offset + 8, offset}, longValues(output[2])); + } + } + + @Test + public void testInt64TargetCountUsesRelativeNextBucketAverage() { + Long[][] values = {{0L}, {1L}, {8L}, {2L}, {0L}, {0L}, {9L}, {1L}, {3L}, {0L}}; + long[][] expected = {{0, 6, 9}, {0, 2, 6, 9}, {0, 2, 4, 6, 9}}; + for (long offset : new long[] {1L << 60, Long.MIN_VALUE, Long.MAX_VALUE - 9}) { + for (int n = 3; n <= 5; n++) { + Column[] output = sample("TARGET_COUNT", n, n, 0, times(10), translate(values, offset)); + assertArrayEquals(expected[n - 3], longValues(output[1])); + for (int row = 0; row < output[1].getPositionCount(); row++) { + int selectedTime = (int) output[1].getLong(row); + assertEquals(offset + values[selectedTime][0], output[2].getLong(row)); + } + } + } + } + + @Test + public void testInt64WindowsKeepIndependentAnchorsAcrossNullBuckets() { + Long[][] values = { + {0L, 0L}, {1L, 0L}, {2L, 0L}, + {null, 0L}, {null, 1L}, {null, 2L}, + {null, null}, {null, null}, {null, null}, + {0L, 0L}, {0L, 0L}, {0L, 0L} + }; + long[] offsets = {1L << 60, Long.MIN_VALUE}; + for (String mode : new String[] {"COUNT_WINDOW", "TIME_WINDOW"}) { + for (long slide : new long[] {1, 3, 4}) { + Column[] expected = sample(mode, 3, slide, 0, times(12), values); + Column[] actual = sample(mode, 3, slide, 0, times(12), translate(values, offsets)); + int firstValueColumn = mode.equals("TIME_WINDOW") ? 3 : 2; + for (int column = 0; column < actual.length; column++) { + assertEquals(expected[column].getPositionCount(), actual[column].getPositionCount()); + for (int row = 0; row < actual[column].getPositionCount(); row++) { + assertEquals(expected[column].isNull(row), actual[column].isNull(row)); + if (!actual[column].isNull(row)) { + long offset = + column >= firstValueColumn && (column - firstValueColumn) % 2 == 0 + ? offsets[(column - firstValueColumn) / 2] + : 0; + assertEquals(expected[column].getLong(row) + offset, actual[column].getLong(row)); + } + } + } + } + } + } + + @Test + public void testInt64DifferencesSpanningLongRangeDoNotWrap() { + Long[][] values = { + {Long.MIN_VALUE}, {0L}, {Long.MAX_VALUE}, {0L}, {Long.MIN_VALUE}, {Long.MIN_VALUE} + }; + Column[] target = sample("TARGET_COUNT", 3, 3, 0, times(6), values); + assertArrayEquals(new long[] {0, 2, 5}, longValues(target[1])); + for (String mode : new String[] {"COUNT_WINDOW", "TIME_WINDOW"}) { + Column[] output = sample(mode, 2, 2, 0, times(6), values); + assertArrayEquals(new long[] {1, 2, 4}, longValues(output[output.length - 2])); + } + } + + @Test + public void testFiniteDoublesPreserveSelectionsWhenAreasOverflow() { + assertFloatingScaleInvariant(new double[] {0, 4, 8, 2, 0, 0}); + } + + @Test + public void testFiniteDoublesPreserveSelectionsWhenLookaheadSumOverflows() { + assertFloatingScaleInvariant(new double[] {0, 4, 8, 7, 7, 0}); + } + + @Test + public void testFiniteDoublesPreserveSelectionsWhenDifferencesOverflow() { + assertFloatingScaleInvariant(new double[] {-8, 4, 8, 2, -8, -8}); + } + + @Test + public void testOverflowFallbackKeepsEarliestAreaTie() { + Column[] output = + sampleFloating("TARGET_COUNT", 3, new double[] {0, 1.6e308, 1.6e308, 0, 0, 0}); + assertArrayEquals(new long[] {0, 1, 5}, longValues(output[1])); + } + + @Test(timeout = 5000) + public void testCountWindowRejectsUnrepresentableEnd() { + SemanticException failure = + assertThrows( + SemanticException.class, + () -> + sample("COUNT_WINDOW", Long.MAX_VALUE, 1, 0, times(2), new Long[][] {{1L}, {2L}})); + assertEquals( + CommonMessages.EXCEPTION_LTTB_COUNT_WINDOW_BOUNDARIES_EXCEED_THE_INT64_RANGE_5BC980CA, + failure.getMessage()); + } + + @Test(timeout = 5000) + public void testCountWindowWithLargeSlideDoesNotCreateExtraWindow() { + Column[] output = + sample( + "COUNT_WINDOW", Long.MAX_VALUE, Long.MAX_VALUE, 0, times(2), new Long[][] {{1L}, {2L}}); + assertArrayEquals(new long[] {0}, longValues(output[0])); + assertArrayEquals(new long[] {0}, longValues(output[1])); + } + + private static void assertFloatingScaleInvariant(double[] values) { + double[] scaled = new double[values.length]; + for (int i = 0; i < values.length; i++) { + scaled[i] = values[i] * 2e307; + } + for (String mode : new String[] {"TARGET_COUNT", "COUNT_WINDOW", "TIME_WINDOW"}) { + for (int size : new int[] {3, 4}) { + Column[] expected = sampleFloating(mode, size, values); + Column[] actual = sampleFloating(mode, size, scaled); + int timeColumn = actual.length - 2; + assertArrayEquals( + mode + ": " + size, longValues(expected[timeColumn]), longValues(actual[timeColumn])); + for (int row = 0; row < actual[timeColumn].getPositionCount(); row++) { + assertEquals( + scaled[(int) actual[timeColumn].getLong(row)], + actual[timeColumn + 1].getDouble(row), + 0.0); + } + } + } + } + + private static Column[] sampleFloating(String mode, long size, double[] values) { + Double[][] rows = new Double[values.length][1]; + for (int i = 0; i < values.length; i++) { + rows[i][0] = values[i]; + } + return sample(mode, size, size, 0, times(values.length), rows, Type.DOUBLE); + } + + @Test(timeout = 5000) + public void testTimeWindowRejectsUnrepresentableEnd() { + assertWindowOverflow(2, 2, 0, Long.MAX_VALUE - 1); + assertWindowOverflow(1, 1, 0, Long.MAX_VALUE); + } + + @Test(timeout = 5000) + public void testTimeWindowRejectsUnrepresentableStart() { + assertWindowOverflow(3, 3, 0, Long.MIN_VALUE); + } + + @Test(timeout = 5000) + public void testTimeWindowAllowsIntermediateOverflow() { + for (long origin : new long[] {0, Long.MIN_VALUE, Long.MAX_VALUE - 1}) { + Column[] output = + sample("TIME_WINDOW", 2, 2, origin, new long[] {Long.MIN_VALUE}, new Long[][] {{7L}}); + assertEquals(1, output[0].getPositionCount()); + assertEquals(Long.MIN_VALUE, output[0].getLong(0)); + assertEquals(Long.MIN_VALUE + 2, output[1].getLong(0)); + assertEquals(Long.MIN_VALUE, output[2].getLong(0)); + assertEquals(7L, output[3].getLong(0)); + } + } + + @Test(timeout = 5000) + public void testTimeWindowStopsAdvancingPastLongMax() { + Column[] output = + sample( + "TIME_WINDOW", + 2, + 3, + Long.MAX_VALUE - 2, + new long[] {Long.MAX_VALUE - 2, Long.MAX_VALUE - 1, Long.MAX_VALUE}, + new Long[][] {{0L}, {8L}, {1L}}); + assertEquals(1, output[0].getPositionCount()); + assertEquals(Long.MAX_VALUE - 2, output[0].getLong(0)); + assertEquals(Long.MAX_VALUE, output[1].getLong(0)); + // Both candidate endpoints form zero-area triangles. The last input is in the following gap. + assertEquals(Long.MAX_VALUE - 2, output[2].getLong(0)); + } + + @Test(timeout = 5000) + public void testTimeWindowSkipsGapBeforeUnrepresentableNextStart() { + Column[] output = + sample( + "TIME_WINDOW", + 2, + 3, + Long.MAX_VALUE - 2, + new long[] {Long.MAX_VALUE}, + new Long[][] {{1L}}); + assertEquals(0, output[0].getPositionCount()); + } + + private static void assertWindowOverflow(long size, long slide, long origin, long time) { + SemanticException failure = + assertThrows( + SemanticException.class, + () -> + sample("TIME_WINDOW", size, slide, origin, new long[] {time}, new Long[][] {{1L}})); + assertEquals( + CommonMessages.EXCEPTION_LTTB_WINDOW_BOUNDARIES_EXCEED_THE_TIMESTAMP_RANGE_7C0FC7E2, + failure.getMessage()); + } + + private static long[] longValues(Column column) { + long[] values = new long[column.getPositionCount()]; + for (int i = 0; i < values.length; i++) { + values[i] = column.getLong(i); + } + return values; + } + + private static long[] times(int count) { + long[] times = new long[count]; + for (int i = 0; i < count; i++) { + times[i] = i; + } + return times; + } + + private static Long[][] translate(Long[][] values, long... offsets) { + Long[][] translated = new Long[values.length][offsets.length]; + for (int i = 0; i < values.length; i++) { + for (int j = 0; j < offsets.length; j++) { + translated[i][j] = values[i][j] == null ? null : Math.addExact(values[i][j], offsets[j]); + } + } + return translated; + } + + private static Column[] sample( + String mode, long size, long slide, long origin, long[] times, Long[][] values) { + return sample(mode, size, slide, origin, times, values, Type.INT64); + } + + private static Column[] sample( + String mode, long size, long slide, long origin, long[] times, Number[][] values, Type type) { + int columns = values[0].length; + MapTableFunctionHandle handle = + new MapTableFunctionHandle.Builder() + .addProperty(LTTBTableFunction.MODE_PROPERTY, mode) + .addProperty(LTTBTableFunction.PARTITION_TYPES_PROPERTY, "") + .addProperty( + LTTBTableFunction.PARTICIPANT_TYPES_PROPERTY, + String.join(",", Collections.nCopies(columns, type.name()))) + .addProperty(LTTBTableFunction.N_PARAMETER_NAME, size) + .addProperty(LTTBTableFunction.SIZE_PARAMETER_NAME, size) + .addProperty(LTTBTableFunction.SLIDE_PARAMETER_NAME, slide) + .addProperty(LTTBTableFunction.ORIGIN_PARAMETER_NAME, origin) + .build(); + TableFunctionDataProcessor processor = + new LTTBTableFunction().getProcessorProvider(handle).getDataProcessor(); + List builders = new ArrayList<>(); + int outputColumns = (mode.equals("TIME_WINDOW") ? 2 : 1) + 2 * columns; + for (int i = 0; i < outputColumns; i++) { + int firstValueColumn = mode.equals("TIME_WINDOW") ? 3 : 2; + builders.add( + type == Type.DOUBLE && i >= firstValueColumn && (i - firstValueColumn) % 2 == 0 + ? new DoubleColumnBuilder(null, 16) + : new LongColumnBuilder(null, 16)); + } + try { + for (int i = 0; i < times.length; i++) { + processor.process(record(times[i], values[i]), builders, null); + } + processor.finish(builders, null); + return builders.stream().map(ColumnBuilder::build).toArray(Column[]::new); + } finally { + processor.beforeDestroy(); + } + } + + private static Record record(long time, Number[] values) { + return (Record) + Proxy.newProxyInstance( + Record.class.getClassLoader(), + new Class[] {Record.class}, + (proxy, method, arguments) -> { + int index = (int) arguments[0]; + switch (method.getName()) { + case "getLong": + return index == 0 ? time : values[index - 1].longValue(); + case "getDouble": + return values[index - 1].doubleValue(); + case "isNull": + return index != 0 && values[index - 1] == null; + default: + throw new UnsupportedOperationException(method.getName()); + } + }); + } +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LTTBRegressionTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LTTBRegressionTest.java new file mode 100644 index 0000000000000..8468049cb3f64 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LTTBRegressionTest.java @@ -0,0 +1,278 @@ +/* + * 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.relational.access.Record; +import org.apache.iotdb.udf.api.relational.table.MapTableFunctionHandle; +import org.apache.iotdb.udf.api.relational.table.processor.TableFunctionDataProcessor; + +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.read.common.block.column.DoubleColumnBuilder; +import org.apache.tsfile.read.common.block.column.LongColumnBuilder; +import org.junit.Test; + +import java.lang.reflect.Proxy; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +public class LTTBRegressionTest { + private static final long NANOSECOND_EPOCH = 1_780_000_000_000_000_000L; + + @Test + public void testTargetCountIsInvariantUnderTimestampTranslation() { + double[] values = {0, 1, 8, 2, 0, 0, 9, 1, -3, 0}; + long[] times = new long[values.length]; + for (int i = 0; i < times.length; i++) { + times[i] = i; + } + for (int n : new int[] {3, 4, 5}) { + int[] expected = LTTBTableFunction.selectTargetCount(times, values, times.length, n); + for (long origin : new long[] {NANOSECOND_EPOCH, Long.MIN_VALUE, Long.MAX_VALUE - 10}) { + long[] shifted = Arrays.stream(times).map(time -> time + origin).toArray(); + assertArrayEquals( + expected, LTTBTableFunction.selectTargetCount(shifted, values, shifted.length, n)); + } + } + } + + @Test + public void testWindowsAreInvariantUnderTimestampTranslation() { + Double[][] rows = {{0.0}, {1.0}, {8.0}, {2.0}, {0.0}, {0.0}}; + for (String mode : new String[] {"COUNT_WINDOW", "TIME_WINDOW"}) { + for (long slide : new long[] {1, 2, 3}) { + List expected = sample(mode, 2, slide, 0, rows); + List actual = sample(mode, 2, slide, NANOSECOND_EPOCH, rows); + int timeIndex = actual.size() - 2; + assertEquals( + expected.get(timeIndex).getPositionCount(), actual.get(timeIndex).getPositionCount()); + for (int i = 0; i < actual.get(timeIndex).getPositionCount(); i++) { + assertEquals( + expected.get(timeIndex).build().getLong(i), + actual.get(timeIndex).build().getLong(i) - NANOSECOND_EPOCH); + assertEquals( + expected.get(timeIndex + 1).build().getDouble(i), + actual.get(timeIndex + 1).build().getDouble(i), + 0.0); + } + } + } + } + + @Test + public void testWindowsSkipNullBucketsIndependentlyPerColumn() { + Double[][] rows = { + {0.0, 0.0}, {1.0, 0.0}, {2.0, 0.0}, + {null, 0.0}, {null, 1.0}, {null, 2.0}, + {null, null}, {null, null}, {null, null}, + {0.0, 0.0}, {0.0, 0.0}, {0.0, 0.0} + }; + for (String mode : new String[] {"COUNT_WINDOW", "TIME_WINDOW"}) { + List out = sample(mode, 3, 3, 0, rows); + int timeIndex = mode.equals("TIME_WINDOW") ? 2 : 1; + assertEquals(4, out.get(0).getPositionCount()); + assertEquals(2, out.get(timeIndex).build().getLong(0)); + assertTrue(out.get(timeIndex).build().isNull(1)); + assertTrue(out.get(timeIndex).build().isNull(2)); + assertEquals(9, out.get(timeIndex).build().getLong(3)); + assertEquals(2, out.get(timeIndex + 2).build().getLong(0)); + assertEquals(5, out.get(timeIndex + 2).build().getLong(1)); + assertTrue(out.get(timeIndex + 2).build().isNull(2)); + // This column keeps its own anchor at (5, 2), so the final bucket chooses t=9. + assertEquals(9, out.get(timeIndex + 2).build().getLong(3)); + } + } + + @Test + public void testTrailingNullWindowsUseFinalNonemptyBucketFallback() { + Double[][] rows = {{0.0}, {5.0}, {0.0}, {null}, {null}, {null}}; + for (String mode : new String[] {"COUNT_WINDOW", "TIME_WINDOW"}) { + List out = sample(mode, 3, 3, 0, rows); + int timeIndex = out.size() - 2; + assertEquals(2, out.get(timeIndex).getPositionCount()); + assertEquals(1, out.get(timeIndex).build().getLong(0)); + assertTrue(out.get(timeIndex).build().isNull(1)); + } + } + + /** + * SIZE defines a bucket, not the input range of a separate standard LTTB invocation: the buckets + * share one anchor chain per partition and column, so an overlapping window continues from the + * point the preceding window selected instead of restarting at its own first point. Bucket {@code + * [1,4)} selecting {@code t=1} is only reachable with a chained anchor: with a fresh anchor at + * its own first point the largest triangle of that bucket would be {@code t=2} instead. Every + * window emits exactly one row, so the overlapping run yields one row per window. + */ + @Test + public void testOverlappingWindowsContinueTheAnchorChain() { + Double[][] rows = {{0.0}, {1.0}, {8.0}, {2.0}, {0.0}, {0.0}}; + + // Count buckets [0,3), [1,4), ..., [5,8) over six rows: one row per bucket. + List countWindow = sample("COUNT_WINDOW", 3, 1, 0, rows); + Column countPicks = countWindow.get(countWindow.size() - 2).build(); + assertEquals(6, countPicks.getPositionCount()); + assertEquals(2, countPicks.getLong(0)); + assertEquals(1, countPicks.getLong(1)); + assertEquals(2, countPicks.getLong(2)); + assertEquals(3, countPicks.getLong(3)); + + // Time buckets [0,3), [1,4), ... with SLIDE = 1 a row also opens the buckets that started + // before the first timestamp but have not ended yet, so eight buckets cover the six rows. The + // bucket [2,5) selecting t=2 and the next one selecting t=1 show the chain: [3,6) alone would + // pick t=3 or t=4, not the point its predecessor selected. + List timeWindow = sample("TIME_WINDOW", 3, 1, 0, rows); + Column timePicks = timeWindow.get(timeWindow.size() - 2).build(); + assertEquals(8, timePicks.getPositionCount()); + assertEquals(0, timePicks.getLong(0)); + assertEquals(2, timePicks.getLong(2)); + assertEquals(1, timePicks.getLong(3)); + } + + /** + * A bucket holds at most one selected point per participant column, so two eligible points in one + * bucket still emit a single row. The endpoints are not preserved: here the anchor and the third + * vertex are the two eligible points themselves, so the area is zero for both candidates and the + * tie resolves to the earliest, leaving the final point unreturned. + */ + @Test + public void testTwoEligiblePointsInOneBucketEmitOneRow() { + Double[][] rows = {{0.0}, {8.0}}; + for (String mode : new String[] {"COUNT_WINDOW", "TIME_WINDOW"}) { + List out = sample(mode, 3, 3, 0, rows); + int timeIndex = out.size() - 2; + assertEquals(1, out.get(timeIndex).getPositionCount()); + assertEquals(0, out.get(timeIndex).build().getLong(0)); + assertEquals(0.0, out.get(timeIndex + 1).build().getDouble(0), 0.0); + } + } + + /** + * A window emits one row per partition even when no participant column has an eligible point, so + * an all-NULL column produces NULL rows rather than no rows. Each row is padded per column. + */ + @Test + public void testWindowWithNoEligiblePointsStillEmitsARow() { + Double[][] rows = {{null}, {null}, {null}, {null}}; + for (String mode : new String[] {"COUNT_WINDOW", "TIME_WINDOW"}) { + List out = sample(mode, 2, 2, 0, rows); + int timeIndex = out.size() - 2; + assertEquals(2, out.get(timeIndex).getPositionCount()); + assertTrue(out.get(timeIndex).build().isNull(0)); + assertTrue(out.get(timeIndex + 1).build().isNull(0)); + assertTrue(out.get(timeIndex).build().isNull(1)); + assertTrue(out.get(timeIndex + 1).build().isNull(1)); + } + } + + @Test + public void testTargetColumnsShareMemoryBudget() { + TableFunctionDataProcessor processor = processor("TARGET_COUNT", 3, 3, 0, 8); + try { + for (int i = 0; i < 350_000; i++) { + // timestamps must be unique within the partition, so advance the input time + processor.process( + record(i, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0), Collections.emptyList(), null); + } + fail( + "Expected a controlled memory-limit exception before buffering 350000 rows in eight columns"); + } catch (SemanticException expected) { + // Match the same localized message as the window-memory regression suite. + assertEquals( + String.format( + CommonMessages + .EXCEPTION_LTTB_BUFFERS_EXCEED_THE_PER_PROCESSOR_MEMORY_LIMIT_OF_ARG_BYTES_954FEF00, + LTTBTableFunction.DEFAULT_MAX_BUFFER_BYTES), + expected.getMessage()); + } finally { + processor.beforeDestroy(); + } + } + + private static List sample( + String mode, long size, long slide, long origin, Double[][] rows) { + int columns = rows[0].length; + TableFunctionDataProcessor processor = processor(mode, size, slide, origin, columns); + List builders = new ArrayList<>(); + builders.add(new LongColumnBuilder(null, 16)); + if (mode.equals("TIME_WINDOW")) { + builders.add(new LongColumnBuilder(null, 16)); + } + for (int i = 0; i < columns; i++) { + builders.add(new LongColumnBuilder(null, 16)); + builders.add(new DoubleColumnBuilder(null, 16)); + } + try { + for (int i = 0; i < rows.length; i++) { + processor.process(record(origin + i, rows[i]), builders, null); + } + processor.finish(builders, null); + return builders; + } finally { + processor.beforeDestroy(); + } + } + + private static TableFunctionDataProcessor processor( + String mode, long size, long slide, long origin, int columns) { + MapTableFunctionHandle handle = + new MapTableFunctionHandle.Builder() + .addProperty(LTTBTableFunction.MODE_PROPERTY, mode) + .addProperty(LTTBTableFunction.PARTITION_TYPES_PROPERTY, "") + .addProperty( + LTTBTableFunction.PARTICIPANT_TYPES_PROPERTY, + String.join(",", Collections.nCopies(columns, "DOUBLE"))) + .addProperty(LTTBTableFunction.N_PARAMETER_NAME, size) + .addProperty(LTTBTableFunction.SIZE_PARAMETER_NAME, size) + .addProperty(LTTBTableFunction.SLIDE_PARAMETER_NAME, slide) + .addProperty(LTTBTableFunction.ORIGIN_PARAMETER_NAME, origin) + .build(); + return new LTTBTableFunction().getProcessorProvider(handle).getDataProcessor(); + } + + private static Record record(long time, Double... values) { + // A lightweight row avoids Mockito invocation bookkeeping in the memory-limit test. + return (Record) + Proxy.newProxyInstance( + Record.class.getClassLoader(), + new Class[] {Record.class}, + (proxy, method, arguments) -> { + int index = (int) arguments[0]; + switch (method.getName()) { + case "getLong": + return time; + case "getDouble": + return values[index - 1]; + case "isNull": + return index != 0 && values[index - 1] == null; + default: + throw new UnsupportedOperationException(method.getName()); + } + }); + } +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LTTBTableFunctionTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LTTBTableFunctionTest.java new file mode 100644 index 0000000000000..a19ef63097890 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LTTBTableFunctionTest.java @@ -0,0 +1,229 @@ +/* + * 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.table.TableFunctionAnalysis; +import org.apache.iotdb.udf.api.relational.table.argument.Argument; +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.type.Type; + +import org.junit.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +public class LTTBTableFunctionTest { + + @Test + public void testTargetCountPreservesEndpointsAndPeak() { + long[] times = {0, 1, 2, 3, 4, 5}; + double[] values = {0, 1, 8, 2, 0, 0}; + + assertArrayEquals( + new int[] {0, 2, 5}, LTTBTableFunction.selectTargetCount(times, values, 6, 3)); + } + + @Test + public void testTargetCountReturnsAllShortSeries() { + long[] times = {0, 1}; + double[] values = {1, 2}; + + assertArrayEquals(new int[] {0, 1}, LTTBTableFunction.selectTargetCount(times, values, 2, 3)); + } + + @Test + public void testTargetCountBreaksAreaTiesByStableInputOrder() { + long[] times = {0, 1, 2, 3, 4}; + double[] values = {0, 4, 4, 1, 0}; + + assertArrayEquals( + new int[] {0, 1, 4}, LTTBTableFunction.selectTargetCount(times, values, 5, 3)); + } + + @Test + public void testAnalyzeTargetCountOutputSchema() throws UDFException { + LTTBTableFunction function = new LTTBTableFunction(); + TableFunctionAnalysis analysis = function.analyze(targetCountArguments(3L)); + + assertEquals( + "window_index", analysis.getProperColumnSchema().get().getFields().get(0).getName().get()); + assertEquals( + "device", analysis.getProperColumnSchema().get().getFields().get(1).getName().get()); + assertEquals( + "s1_time", analysis.getProperColumnSchema().get().getFields().get(2).getName().get()); + assertEquals("s1", analysis.getProperColumnSchema().get().getFields().get(3).getName().get()); + assertEquals( + Arrays.asList(0, 1, 2), + analysis.getRequiredColumns().get(LTTBTableFunction.DATA_PARAMETER_NAME)); + } + + @Test + public void testAnalyzeRejectsTargetCountBelowThree() throws UDFException { + try { + new LTTBTableFunction().analyze(targetCountArguments(2L)); + fail(); + } catch (SemanticException e) { + // compare against the constant so the test also passes under -P with-zh-locale + assertEquals( + CommonMessages.EXCEPTION_THE_N_ARGUMENT_OF_LTTB_MUST_BE_AT_LEAST_3_29CE2B87, + e.getMessage()); + } + } + + @Test + public void testAnalyzeRejectsNAndSizeTogether() throws UDFException { + Map arguments = targetCountArguments(3L); + arguments.put(LTTBTableFunction.SIZE_PARAMETER_NAME, new ScalarArgument(Type.INT64, 2L)); + try { + new LTTBTableFunction().analyze(arguments); + fail(); + } catch (SemanticException e) { + assertEquals( + CommonMessages + .EXCEPTION_EXACTLY_ONE_OF_THE_N_AND_SIZE_ARGUMENTS_MUST_BE_SPECIFIED_FOR_LTTB_54AF0733, + e.getMessage()); + } + } + + @Test + public void testTargetCountKeepsNonFiniteValues() { + long[] times = {0, 1, 2}; + double[] values = {Double.NaN, Double.POSITIVE_INFINITY, Double.NEGATIVE_INFINITY}; + + // three eligible points with N => 3: every point must be returned, not discarded + assertArrayEquals( + new int[] {0, 1, 2}, LTTBTableFunction.selectTargetCount(times, values, 3, 3)); + } + + /** + * With {@code N < M} the algorithm computes triangle areas, so a non-finite participant reaches + * the area comparison. NaN and the infinities cannot form a finite area and {@link + * java.math.BigDecimal} cannot represent them, so such candidates must be skipped (earliest wins) + * instead of aborting the query with an unchecked {@code NumberFormatException}. + */ + @Test + public void testTargetCountSkipsNonFiniteCandidatesWhenSampling() { + long[] times = {0, 1, 2, 3, 4, 5}; + double[] values = {1.0, Double.NaN, Double.POSITIVE_INFINITY, 2.0, 3.0, 4.0}; + + int[] selected = LTTBTableFunction.selectTargetCount(times, values, 6, 3); + + // The first and last eligible points are always preserved. + assertEquals(0, selected[0]); + assertEquals(5, selected[selected.length - 1]); + assertEquals(3, selected.length); + for (int index : selected) { + assertTrue("selected index out of range: " + index, index >= 0 && index < 6); + } + } + + /** A bucket whose candidates are all non-finite keeps its earliest point, deterministically. */ + @Test + public void testTargetCountKeepsEarliestPointWhenAllCandidatesAreNonFinite() { + long[] times = {0, 1, 2, 3, 4, 5}; + double[] values = { + Double.NaN, + Double.NaN, + Double.POSITIVE_INFINITY, + Double.NaN, + Double.NEGATIVE_INFINITY, + Double.NaN + }; + + assertArrayEquals( + new int[] {0, 1, 5}, LTTBTableFunction.selectTargetCount(times, values, 6, 3)); + } + + @Test + public void testTimestampUniquenessAcceptsDistinctAndRejectsRepeatedTimestamps() { + LTTBTableFunction.TimestampUniquenessChecker checker = + new LTTBTableFunction.TimestampUniquenessChecker(); + checker.check(1L); + checker.check(2L); + checker.check(3L); + try { + checker.check(3L); + fail(); + } catch (SemanticException e) { + assertEquals( + String.format( + CommonMessages + .EXCEPTION_LTTB_DOES_NOT_SUPPORT_DUPLICATE_TIMESTAMPS_BUT_TIMESTAMP_ARG_APPEARS_MORE_THAN_ONCE_IN_THE_SAME_PARTITION_7713FEDD, + 3L), + e.getMessage()); + } + } + + /** The duplicate check must survive the block boundary: the state outlives one block's rows. */ + @Test + public void testTimestampUniquenessSpansInputBlocks() { + LTTBTableFunction.TimestampUniquenessChecker checker = + new LTTBTableFunction.TimestampUniquenessChecker(); + checker.check(7L); // last row of the first input block + try { + checker.check(7L); // first row of the second input block + fail(); + } catch (SemanticException e) { + assertEquals( + String.format( + CommonMessages + .EXCEPTION_LTTB_DOES_NOT_SUPPORT_DUPLICATE_TIMESTAMPS_BUT_TIMESTAMP_ARG_APPEARS_MORE_THAN_ONCE_IN_THE_SAME_PARTITION_7713FEDD, + 7L), + e.getMessage()); + } + } + + private static Map targetCountArguments(long n) { + Map arguments = new HashMap<>(); + arguments.put( + LTTBTableFunction.DATA_PARAMETER_NAME, + new TableArgument( + Arrays.asList(Optional.of("time"), Optional.of("device"), Optional.of("s1")), + Arrays.asList(Type.TIMESTAMP, Type.STRING, Type.DOUBLE), + Collections.singletonList("device"), + Collections.singletonList("time"), + false)); + arguments.put( + LTTBTableFunction.TIMECOL_PARAMETER_NAME, new ScalarArgument(Type.STRING, "time")); + arguments.put(LTTBTableFunction.N_PARAMETER_NAME, new ScalarArgument(Type.INT64, n)); + arguments.put( + LTTBTableFunction.SIZE_PARAMETER_NAME, + new ScalarArgument(Type.INT64, LTTBTableFunction.UNSPECIFIED_SIZE)); + arguments.put( + LTTBTableFunction.SLIDE_PARAMETER_NAME, + new ScalarArgument(Type.INT64, LTTBTableFunction.UNSPECIFIED_SLIDE)); + arguments.put(LTTBTableFunction.ORIGIN_PARAMETER_NAME, new ScalarArgument(Type.TIMESTAMP, 0L)); + arguments.put( + LTTBTableFunction.WINDOW_MODE_PARAMETER_NAME, new ScalarArgument(Type.BOOLEAN, false)); + return arguments; + } +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LTTBWindowMemoryTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LTTBWindowMemoryTest.java new file mode 100644 index 0000000000000..4bd6437856578 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LTTBWindowMemoryTest.java @@ -0,0 +1,223 @@ +/* + * 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.relational.access.Record; +import org.apache.iotdb.udf.api.relational.table.MapTableFunctionHandle; +import org.apache.iotdb.udf.api.relational.table.processor.TableFunctionDataProcessor; +import org.apache.iotdb.udf.api.type.Type; + +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.read.common.block.column.DoubleColumnBuilder; +import org.apache.tsfile.read.common.block.column.LongColumnBuilder; +import org.junit.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; +import static org.mockito.Mockito.withSettings; + +public class LTTBWindowMemoryTest { + + @Test + public void testCountWindowOverlapFailsBeforeExhaustingHeap() { + TableFunctionDataProcessor processor = processor(false, 1_000_000, 1, Type.DOUBLE); + try { + // Without the shared budget, 5000 rows retain over 280 MiB despite every individual + // window being far below MAX_BUFFERED_POINTS. This test also runs with a 128 MiB heap. + assertMemoryLimit( + () -> { + for (int i = 0; i < 5000; i++) { + // timestamps must be unique within the partition, so advance the input time + processor.process(record(i, 1.0), Collections.emptyList(), null); + } + }, + LTTBTableFunction.DEFAULT_MAX_BUFFER_BYTES); + } finally { + processor.beforeDestroy(); + } + } + + @Test + public void testTimeWindowCreationIsBoundedEvenForNullInput() { + TableFunctionDataProcessor processor = processor(true, 1_000_000, 1, Type.DOUBLE); + try { + // One row belongs to a million overlapping time windows. Initial buffer allocation + // must be charged even when the input column contains no eligible points. + assertMemoryLimit( + () -> processor.process(record(0, (Number) null), Collections.emptyList(), null), + LTTBTableFunction.DEFAULT_MAX_BUFFER_BYTES); + } finally { + processor.beforeDestroy(); + } + } + + @Test + public void testColumnsShareTheWindowBudget() { + TableFunctionDataProcessor processor = processor(false, 1_000_000, 1, Type.DOUBLE, Type.INT64); + try { + assertMemoryLimit( + () -> { + for (int i = 0; i < 1800; i++) { + // timestamps must be unique within the partition, so advance the input time + processor.process(record(i, 1.0, Long.MAX_VALUE), Collections.emptyList(), null); + } + }, + LTTBTableFunction.DEFAULT_MAX_BUFFER_BYTES); + } finally { + processor.beforeDestroy(); + } + } + + @Test + public void testArrayGrowthReservesOldAndNewIntegralArrays() { + LTTBTableFunction.PointBuffer buffer = + new LTTBTableFunction.PointBuffer(true, new LTTBTableFunction.BufferMemoryBudget(4096)); + for (int i = 0; i < 64; i++) { + buffer.addIntegral(i, Long.MAX_VALUE); + } + // The retained 128-point buffer fits in 4096 bytes, but the old 64-point arrays and + // replacement arrays coexist during growth. Reject before replacing any array. + assertMemoryLimit(() -> buffer.addIntegral(64, Long.MAX_VALUE), 4096); + assertEquals(64, buffer.size()); + assertEquals(63, buffer.time(63)); + } + + @Test + public void testInitialBuffersShareCapacityBudget() { + LTTBTableFunction.BufferMemoryBudget budget = new LTTBTableFunction.BufferMemoryBudget(4096); + new LTTBTableFunction.PointBuffer(true, budget); + new LTTBTableFunction.PointBuffer(true, budget); + assertMemoryLimit(() -> new LTTBTableFunction.PointBuffer(true, budget), 4096); + } + + @Test + public void testEmittedWindowsReleaseBudgetForLongStream() { + TableFunctionDataProcessor processor = processor(false, 1, 1, Type.DOUBLE); + List builders = builders(false); + long outputRows = 0; + try { + // Total allocations exceed the budget, but only two windows remain live at a time. + for (int row = 0; row < 80_000; row++) { + // timestamps must be unique within the partition, so advance the input time + processor.process(record(row, 42.0), builders, null); + if (row % 512 == 511) { + outputRows += builders.get(0).getPositionCount(); + builders = builders(false); + } + } + processor.finish(builders, null); + outputRows += builders.get(0).getPositionCount(); + assertEquals(80_000, outputRows); + assertEquals(42.0, builders.get(2).build().getDouble(0), 0.0); + } finally { + processor.beforeDestroy(); + } + } + + @Test + public void testWindowResultsArePreserved() { + for (boolean timeWindow : new boolean[] {false, true}) { + TableFunctionDataProcessor processor = processor(timeWindow, 2, 2, Type.DOUBLE); + List builders = builders(timeWindow); + double[] values = {0, 1, 8, 2, 0, 0}; + try { + for (int time = 0; time < values.length; time++) { + processor.process(record(time, values[time]), builders, null); + } + processor.finish(builders, null); + ColumnBuilder times = builders.get(timeWindow ? 2 : 1); + assertEquals(3, times.getPositionCount()); + assertEquals(1, times.build().getLong(0)); + assertEquals(2, times.build().getLong(1)); + assertEquals(4, times.build().getLong(2)); + } finally { + processor.beforeDestroy(); + } + } + } + + private static TableFunctionDataProcessor processor( + boolean timeWindow, long size, long slide, Type... types) { + MapTableFunctionHandle handle = + new MapTableFunctionHandle.Builder() + .addProperty( + LTTBTableFunction.MODE_PROPERTY, timeWindow ? "TIME_WINDOW" : "COUNT_WINDOW") + .addProperty(LTTBTableFunction.PARTITION_TYPES_PROPERTY, "") + .addProperty( + LTTBTableFunction.PARTICIPANT_TYPES_PROPERTY, + WindowTVFUtils.joinTypes(Arrays.asList(types))) + .addProperty(LTTBTableFunction.SIZE_PARAMETER_NAME, size) + .addProperty(LTTBTableFunction.SLIDE_PARAMETER_NAME, slide) + .addProperty(LTTBTableFunction.ORIGIN_PARAMETER_NAME, 0L) + .build(); + return new LTTBTableFunction().getProcessorProvider(handle).getDataProcessor(); + } + + private static List builders(boolean timeWindow) { + if (timeWindow) { + return Arrays.asList( + new LongColumnBuilder(null, 16), + new LongColumnBuilder(null, 16), + new LongColumnBuilder(null, 16), + new DoubleColumnBuilder(null, 16)); + } + return Arrays.asList( + new LongColumnBuilder(null, 16), + new LongColumnBuilder(null, 16), + new DoubleColumnBuilder(null, 16)); + } + + private static Record record(long time, Number... values) { + // Avoid retaining invocation histories in the long-stream memory regression test. + Record record = mock(Record.class, withSettings().stubOnly()); + when(record.getLong(0)).thenReturn(time); + for (int i = 0; i < values.length; i++) { + when(record.isNull(i + 1)).thenReturn(values[i] == null); + if (values[i] instanceof Long) { + when(record.getLong(i + 1)).thenReturn(values[i].longValue()); + } else if (values[i] != null) { + when(record.getDouble(i + 1)).thenReturn(values[i].doubleValue()); + } + } + return record; + } + + private static void assertMemoryLimit(Runnable action, long limit) { + try { + action.run(); + fail("Expected the shared LTTB window memory limit"); + } catch (SemanticException e) { + assertEquals( + String.format( + CommonMessages + .EXCEPTION_LTTB_BUFFERS_EXCEED_THE_PER_PROCESSOR_MEMORY_LIMIT_OF_ARG_BYTES_954FEF00, + limit), + e.getMessage()); + } + } +}