Merge branch 'master' of https://github.com/apache/iotdb into improve-pipe-receiver-session-handling

Author: Caideyipi

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeReceiverAutoCreateDisabledIT.java

@@ -182,6 +182,43 @@ public void testAutoSplitHistoryTsFileWithDeletionWhenReceiverAutoCreateSchemaDi
         new HashSet<>(Arrays.asList("root.sg.aligned,true,null,INF,")));
   }
 
+  @Test
+  public void
+      testAutoSplitHistoryTsFileWithUnflushedAlignedDeletionWhenReceiverAutoCreateSchemaDisabled()
+          throws Exception {
+    TestUtils.executeNonQueries(
+        senderEnv,
+        Arrays.asList(
+            "create database root.sg",
+            "create aligned timeseries root.sg.aligned(s0 INT32, s1 INT64, s2 DOUBLE)",
+            "insert into root.sg.aligned(time, s0, s1, s2) aligned "
+                + "values(1, 1, 10, 1.0), (2, 2, 20, 2.0), (3, 3, 30, 3.0)",
+            "delete timeseries root.sg.aligned.s1"));
+
+    TestUtils.executeNonQuery(
+        senderEnv,
+        String.format(
+            "create pipe test with source ('inclusion'='all', 'source.history.enable'='true', 'source.realtime.mode'='batch') "
+                + "with sink ('sink'='iotdb-thrift-sink', 'sink.node-urls'='%s')",
+            receiverEnv.getDataNodeWrapper(0).getIpAndPortString()));
+
+    TestUtils.assertDataEventuallyOnEnv(
+        receiverEnv,
+        "select s0,s2 from root.sg.aligned",
+        "Time,root.sg.aligned.s0,root.sg.aligned.s2,",
+        new HashSet<>(Arrays.asList("1,1,1.0,", "2,2,2.0,", "3,3,3.0,")));
+    TestUtils.assertDataEventuallyOnEnv(
+        receiverEnv,
+        "count timeseries root.sg.aligned.*",
+        "count(timeseries),",
+        new HashSet<>(Arrays.asList("2,")));
+    TestUtils.assertDataEventuallyOnEnv(
+        receiverEnv,
+        "show devices root.sg.aligned",
+        "Device,IsAligned,Template,TTL(ms),",
+        new HashSet<>(Arrays.asList("root.sg.aligned,true,null,INF,")));
+  }
+
   private QueryResult queryForResult(final Statement statement, final String sql)
       throws SQLException {
     try (final ResultSet resultSet = statement.executeQuery(sql)) {

integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBDeletionTableIT.java

@@ -19,6 +19,8 @@
 
 package org.apache.iotdb.relational.it.db.it;
 
+import org.apache.iotdb.isession.ITableSession;
+import org.apache.iotdb.isession.SessionDataSet;
 import org.apache.iotdb.it.env.EnvFactory;
 import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
 import org.apache.iotdb.it.framework.IoTDBTestRunner;
@@ -40,6 +42,8 @@
 import java.nio.file.StandardOpenOption;
 import java.sql.Connection;
 import java.sql.Statement;
+import java.util.HashMap;
+import java.util.Map;
 
 @RunWith(IoTDBTestRunner.class)
 @Category({TableLocalStandaloneIT.class, TableClusterIT.class})
@@ -82,4 +86,94 @@ public void loadConfiguration() throws IOException {
       }
     }
   }
+
+  // `show configuration` must display the in-memory effective value, not the raw config-file
+  // value. Several hot-reloaded parameters have setters that rewrite the loaded value (e.g.
+  // `if (x > 0) setX(x)` skips non-positive values; loadFixedSizeLimitForQuery rewrites `<=0`
+  // to a computed default). This test verifies those keys show the effective value after
+  // `load configuration`.
+  @Test
+  public void showConfigurationDisplaysEffectiveValue() throws Exception {
+    DataNodeWrapper dataNodeWrapper = EnvFactory.getEnv().getDataNodeWrapper(0);
+    String confPath =
+        dataNodeWrapper.getNodePath()
+            + File.separator
+            + "conf"
+            + File.separator
+            + "iotdb-system.properties";
+
+    // Case 1: non-positive guard params (`if (x > 0) setX(x)`) must display the effective
+    // default value, not the raw file value (0 / -1).
+    Map<String, String> afterGuardReload =
+        appendLinesLoadAndShow(
+            confPath,
+            "cte_buffer_size_in_bytes=0",
+            "max_rows_in_cte_buffer=-1",
+            "max_sub_task_num_for_information_table_scan=0");
+    Assert.assertEquals("131072", afterGuardReload.get("cte_buffer_size_in_bytes"));
+    Assert.assertEquals("1000", afterGuardReload.get("max_rows_in_cte_buffer"));
+    Assert.assertEquals("4", afterGuardReload.get("max_sub_task_num_for_information_table_scan"));
+
+    // Case 2: a valid value is not rewritten, so display must equal the file value (regression).
+    Map<String, String> afterValidReload =
+        appendLinesLoadAndShow(confPath, "cte_buffer_size_in_bytes=262144");
+    Assert.assertEquals("262144", afterValidReload.get("cte_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(confPath);
+    assertPositiveNonZero(defaultsReload, "sort_buffer_size_in_bytes");
+    assertPositiveNonZero(defaultsReload, "mods_cache_size_limit_per_fi_in_bytes");
+  }
+
+  // Appends the given lines to the config file, runs `load configuration`, fetches the
+  // `show configuration` result, and restores the file to its original length.
+  private Map<String, String> appendLinesLoadAndShow(String confPath, String... lines)
+      throws Exception {
+    long length = new File(confPath).length();
+    try {
+      if (lines.length > 0) {
+        try (FileWriter fileWriter = new FileWriter(confPath, true)) {
+          fileWriter.write(System.lineSeparator());
+          for (String line : lines) {
+            fileWriter.write(line);
+            fileWriter.write(System.lineSeparator());
+          }
+        }
+      }
+      try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+          Statement statement = connection.createStatement()) {
+        statement.execute("LOAD CONFIGURATION");
+      }
+      return fetchShowConfiguration();
+    } finally {
+      try (FileChannel fileChannel =
+          FileChannel.open(new File(confPath).toPath(), StandardOpenOption.WRITE)) {
+        fileChannel.truncate(length);
+      }
+    }
+  }
+
+  private Map<String, String> fetchShowConfiguration() throws Exception {
+    Map<String, String> result = new HashMap<>();
+    try (ITableSession tableSessionConnection = EnvFactory.getEnv().getTableSessionConnection()) {
+      SessionDataSet sessionDataSet =
+          tableSessionConnection.executeQueryStatement("show configuration");
+      SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
+      while (iterator.next()) {
+        String name = iterator.getString(1);
+        String value = iterator.isNull(2) ? null : iterator.getString(2);
+        result.put(name, value);
+      }
+    }
+    return result;
+  }
+
+  private static void assertPositiveNonZero(Map<String, String> configMap, String key) {
+    String value = configMap.get(key);
+    Assert.assertNotNull("show configuration is missing key: " + key, value);
+    long parsed = Long.parseLong(value);
+    Assert.assertTrue(
+        key + " should display a positive effective value, but was " + value, parsed > 0);
+  }
 }

integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadConfigurationTableIT.java

@@ -44,6 +44,18 @@ public class StringUtils {
 
   static final int WILD_COMPARE_NO_MATCH = -1;
 
+  /**
+   * Maximum absolute {@link BigDecimal#scale()} accepted by {@link #consistentToString(BigDecimal)}
+   * before it falls back to scientific notation.
+   *
+   * <p>{@link BigDecimal#toPlainString()} expands the value to a non-scientific decimal string
+   * whose length grows with {@code |scale|}. A value such as {@code 1e1000000000} would therefore
+   * materialize a ~1GB {@link String}, causing {@link OutOfMemoryError} / denial of service on the
+   * client. Any scale beyond this bound is rejected for expansion and the caller gets the
+   * scientific form instead, which is a few dozen characters at most.
+   */
+  static final int MAX_PLAIN_STRING_SCALE = 10_000;
+
   static {
     for (int i = -128; i <= 127; i++) {
       allBytes[i - -128] = (byte) i;
@@ -65,9 +77,16 @@ public static String consistentToString(BigDecimal decimal) {
     if (decimal == null) {
       return null;
     }
+    // Guard against CVE-style DoS: BigDecimal values constructed from extreme scientific
+    // notation (e.g. "1e1000000000") have a huge |scale| and would otherwise cause
+    // toPlainString() to allocate a multi-GB String and OOM the JVM. Fall back to the
+    // scientific form, which is always short.
+    if (decimal.scale() > MAX_PLAIN_STRING_SCALE || decimal.scale() < -MAX_PLAIN_STRING_SCALE) {
+      return decimal.toString();
+    }
     if (toPlainStringMethod != null) {
       try {
-        return (String) toPlainStringMethod.invoke(decimal, null);
+        return (String) toPlainStringMethod.invoke(decimal, (Object[]) null);
       } catch (InvocationTargetException | IllegalAccessException e) {
         LOGGER.warn(JdbcMessages.CONSISTENT_TO_STRING_ERROR, e);
       }

iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/StringUtils.java

@@ -0,0 +1,74 @@
+/*
+ * 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.jdbc;
+
+import org.junit.Test;
+
+import java.math.BigDecimal;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+public class StringUtilsTest {
+
+  @Test
+  public void consistentToStringHandlesNull() {
+    assertNull(StringUtils.consistentToString(null));
+  }
+
+  @Test
+  public void consistentToStringExpandsNormalDecimal() {
+    assertEquals("123.45", StringUtils.consistentToString(new BigDecimal("123.45")));
+    // Plain-string form, not scientific notation.
+    assertEquals("100", StringUtils.consistentToString(new BigDecimal("1E+2")));
+  }
+
+  @Test
+  public void consistentToStringExpandsUpToLimit() {
+    // scale right at the boundary must still be expanded to plain form.
+    BigDecimal atLimit = new BigDecimal("1E+" + StringUtils.MAX_PLAIN_STRING_SCALE);
+    String out = StringUtils.consistentToString(atLimit);
+    assertEquals(StringUtils.MAX_PLAIN_STRING_SCALE + 1, out.length());
+    assertTrue(out.startsWith("1"));
+  }
+
+  /**
+   * Regression test for the BigDecimal DoS: {@code new BigDecimal("1e1000000000")} used to cause
+   * {@link BigDecimal#toPlainString()} to allocate a ~1GB string and OOM the JVM. The guard must
+   * refuse to expand such values and return the compact scientific form instead.
+   */
+  @Test
+  public void consistentToStringRefusesToExpandHugeScale() {
+    BigDecimal malicious = new BigDecimal("1e1000000000");
+    String out = StringUtils.consistentToString(malicious);
+    // Must be short (scientific notation), never the ~1e9-char plain form.
+    assertTrue("expected scientific form, got length " + out.length(), out.length() < 64);
+    assertTrue(out.contains("E") || out.contains("e"));
+  }
+
+  @Test
+  public void consistentToStringRefusesToExpandHugePositiveScale() {
+    // Very large positive scale -> also produces a long plain string; must be refused.
+    BigDecimal malicious = new BigDecimal("1e-1000000000");
+    String out = StringUtils.consistentToString(malicious);
+    assertTrue("expected scientific form, got length " + out.length(), out.length() < 64);
+  }
+}

iotdb-client/jdbc/src/test/java/org/apache/iotdb/jdbc/StringUtilsTest.java

@@ -915,6 +915,8 @@ public final class ProcedureMessages {
       "Start rollback Renaming table {}.{} on configNode";
   public static final String START_ROLLBACK_SET_PROPERTIES_TO_TABLE =
       "Start rollback set properties to table {}.{}";
+  public static final String STATE_MACHINE_PROCEDURE_EOF_STATE_SKIP_EXECUTION =
+      "StateMachineProcedure pid={} is scheduled with EOF state, skip execution: {}";
   public static final String STATE_STUCK_AT = "State stuck at ";
   public static final String STOPPIPEPROCEDUREV2_EXECUTEFROMCALCULATEINFOFORTASK =
       "StopPipeProcedureV2: executeFromCalculateInfoForTask({})";

iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java

@@ -913,6 +913,8 @@ public final class ProcedureMessages {
       "在 ConfigNode 上开始回滚重命名表 {}.{}";
   public static final String START_ROLLBACK_SET_PROPERTIES_TO_TABLE =
       "开始回滚设置表 {}.{} 的属性";
+  public static final String STATE_MACHINE_PROCEDURE_EOF_STATE_SKIP_EXECUTION =
+      "StateMachineProcedure pid={} 被 EOF 状态调度，跳过执行：{}";
   public static final String STATE_STUCK_AT = "状态卡在 ";
   public static final String STOPPIPEPROCEDUREV2_EXECUTEFROMCALCULATEINFOFORTASK =
       "StopPipeProcedureV2: executeFromCalculateInfoForTask({})";

iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java

@@ -146,10 +146,17 @@ protected Procedure<Env>[] execute(final Env env) throws InterruptedException {
     updateTimestamp();
     try {
       if (noMoreState() || isFailed()) {
-        return null;
+        return new Procedure[0];
       }
 
       TState state = getCurrentState();
+      if (state == null) {
+        LOG.warn(
+            ProcedureMessages.STATE_MACHINE_PROCEDURE_EOF_STATE_SKIP_EXECUTION, getProcId(), this);
+        stateFlow = Flow.NO_MORE_STATE;
+        setStateDeserialized(false);
+        return new Procedure[0];
+      }
 
       // init for the first execution
       if (states.isEmpty()) {
@@ -169,7 +176,9 @@ protected Procedure<Env>[] execute(final Env env) throws InterruptedException {
         subProcList.clear();
         return subProcedures;
       }
-      return (isWaiting() || isFailed() || noMoreState()) ? null : new Procedure[] {this};
+      return (isWaiting() || isFailed() || noMoreState())
+          ? new Procedure[0]
+          : new Procedure[] {this};
     } finally {
       updateTimestamp();
     }