Add LTTB downsampling as a built-in Table Model table function

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<String, String> 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<String, String> 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");
   }
 

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

@@ -72,6 +72,13 @@ 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. It is resolved
+   * once from the provider, which also gives the same value to every processor it creates, so the
+   * reservation and the enforced limit cannot diverge.
+   */
+  private final long estimatedMaxBufferedBytes;
+
   private TableFunctionDataProcessor processor;
   private PartitionState partitionState;
   private ListenableFuture<?> isBlocked;
@@ -82,6 +89,7 @@ public class TableFunctionOperator extends AbstractOperator implements ProcessOp
   public TableFunctionOperator(
       CommonOperatorContext operatorContext,
       TableFunctionProcessorProvider processorProvider,
+      long estimatedMaxBufferedBytes,
       Operator inputOperator,
       List<TSDataType> inputDataTypes,
       List<TSDataType> outputDataTypes,
@@ -101,6 +109,7 @@ public TableFunctionOperator(
         new PartitionRecognizer(
             partitionChannels, requiredChannels, passThroughChannels, inputDataTypes);
     this.isDeclaredAsPassThrough = isDeclaredAsPassThrough;
+    this.estimatedMaxBufferedBytes = estimatedMaxBufferedBytes;
     this.needPassThrough = properChannelCount != outputDataTypes.size();
     this.partitionState = null;
     this.properBlockBuilder = new TsBlockBuilder(outputDataTypes.subList(0, properChannelCount));
@@ -291,7 +300,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 +314,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

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/plan/planner/TableOperatorGenerator.java

@@ -108,6 +108,7 @@
 import org.apache.iotdb.commons.queryengine.plan.relational.function.BoundSignature;
 import org.apache.iotdb.commons.queryengine.plan.relational.function.FunctionKind;
 import org.apache.iotdb.commons.queryengine.plan.relational.function.ITableFunctionFactory;
+import org.apache.iotdb.commons.queryengine.plan.relational.function.TableFunctionProcessorMemoryBudget;
 import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ResolvedFunction;
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.OrderingScheme;
 import org.apache.iotdb.commons.queryengine.plan.relational.planner.SortOrder;
@@ -1684,6 +1685,12 @@ public Operator visitTableFunctionProcessor(TableFunctionProcessorNode node, C c
       return new TableFunctionOperator(
           operatorContext,
           processorProvider,
+          // The budget is taken from the provider itself, so the operator reserves exactly what the
+          // processors of this provider enforce. A user-defined function reports nothing.
+          processorProvider instanceof TableFunctionProcessorMemoryBudget
+              ? ((TableFunctionProcessorMemoryBudget) processorProvider)
+                  .getEstimatedMaxBufferedBytes()
+              : 0L,
           operator,
           inputDataTypes,
           outputDataTypes,

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java

@@ -1122,6 +1122,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(
@@ -2405,6 +2412,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(
@@ -2452,6 +2465,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(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java

@@ -23,6 +23,7 @@
 import org.apache.iotdb.commons.exception.IoTDBException;
 import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
 import org.apache.iotdb.commons.exception.SemanticException;
+import org.apache.iotdb.commons.i18n.CommonMessages;
 import org.apache.iotdb.commons.i18n.QueryMessages;
 import org.apache.iotdb.commons.queryengine.common.SessionInfo;
 import org.apache.iotdb.commons.queryengine.plan.relational.analyzer.NodeRef;
@@ -123,6 +124,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.commons.utils.FileUtils;
@@ -6033,13 +6035,15 @@ 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());
     }
 
     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)
@@ -6151,6 +6155,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<TableFunctionArgument> arguments,
+        List<ParameterSpecification> parameterSpecifications,
+        ImmutableMap.Builder<String, Argument> passedArguments) {
+      if (!TableBuiltinTableFunction.LTTB.getFunctionName().equalsIgnoreCase(functionName)) {
+        return;
+      }
+
+      Optional<TableFunctionArgument> nArgument =
+          findOptionalTableFunctionArgument(
+              arguments, parameterSpecifications, LTTBTableFunction.N_PARAMETER_NAME);
+      Optional<TableFunctionArgument> sizeArgument =
+          findOptionalTableFunctionArgument(
+              arguments, parameterSpecifications, LTTBTableFunction.SIZE_PARAMETER_NAME);
+      Optional<TableFunctionArgument> 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<TableFunctionArgument> arguments,
         List<ParameterSpecification> parameterSpecifications) {

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)

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java

@@ -3182,7 +3182,13 @@ private boolean canSplitTableFunctionProcessor(TableFunctionProcessorNode node)
     if (!isPartitionedGroup(node.getChild())) {
       return false;
     }
-    return !TableBuiltinTableFunction.FFT.getFunctionName().equalsIgnoreCase(node.getName());
+    // LTTB samples a whole partition and FFT transforms a whole partition, so neither may run once
+    // per fragment: a partitioned GroupNode does not make a partition complete within one branch.
+    // The function therefore has to run once, over the partitions merged in order. The split rule
+    // itself is a separate planner issue; the exclusion can be revisited once that is fixed.
+    String functionName = node.getName();
+    return !TableBuiltinTableFunction.FFT.getFunctionName().equalsIgnoreCase(functionName)
+        && !TableBuiltinTableFunction.LTTB.getFunctionName().equalsIgnoreCase(functionName);
   }
 
   private boolean isPartitionedGroup(PlanNode node) {

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/tvf/TableFunctionOperatorTest.java

@@ -371,6 +371,7 @@ private void appendRows(
           new TableFunctionOperator(
               operatorContext,
               provider,
+              0L,
               singleRowChild,
               Collections.singletonList(TSDataType.INT64),
               withPassThrough
@@ -444,6 +445,7 @@ public void process(
           new TableFunctionOperator(
               operatorContext,
               provider,
+              0L,
               twoPartitionChild,
               Collections.singletonList(TSDataType.INT64),
               Collections.singletonList(TSDataType.INT64),
@@ -516,6 +518,7 @@ public void beforeDestroy() {
           new TableFunctionOperator(
               operatorContext,
               provider,
+              0L,
               child,
               Collections.singletonList(TSDataType.INT64),
               Collections.singletonList(TSDataType.INT64),