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

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);
+  }
+}

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-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;
+  }
 }

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

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(